Google Git
Sign in
apache/tsfile/refs/heads/colin_tsfile_dataframe_init/./cpp/src/encoding/ts2diff_encoder.h
blob: b2b219b55669c2128b26ed2127c05e16bd976384 [file]
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* License); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#ifndef ENCODING_TS2DIFF_ENCODER_H
#define ENCODING_TS2DIFF_ENCODER_H
#include <sys/types.h>
#include "common/allocator/alloc_base.h"
#include "common/allocator/byte_stream.h"
#include "encoder.h"
#ifdef ENABLE_SIMD
#include "simde/x86/avx2.h"
#endif
namespace storage {
template <typename T>
struct SIMDOps;
template <>
struct SIMDOps<int32_t> {
#ifdef ENABLE_SIMD
static void rebase(int32_t* arr, int32_t min_val, size_t size) {
const simde__m128i min_vec = simde_mm_set1_epi32(min_val);
size_t i = 0;
for (; i + 3 < size; i += 4) {
simde__m128i vec = simde_mm_loadu_si128(
reinterpret_cast<const simde__m128i*>(arr + i));
vec = simde_mm_sub_epi32(vec, min_vec);
simde_mm_storeu_si128(reinterpret_cast<simde__m128i*>(arr + i),
vec);
}
for (; i < size; ++i) {
arr[i] -= min_val;
}
}
#else
static void rebase(int32_t* arr, int32_t min_val, size_t size) {
for (size_t i = 0; i < size; ++i) {
arr[i] -= min_val;
}
}
#endif
};
template <>
struct SIMDOps<int64_t> {
#ifdef ENABLE_SIMD
static void rebase(int64_t* arr, int64_t min_val, size_t size) {
const simde__m256i min_vec = simde_mm256_set1_epi64x(min_val);
size_t i = 0;
for (; i + 3 < size; i += 4) {
simde__m256i vec = simde_mm256_loadu_si256(
reinterpret_cast<const simde__m256i*>(arr + i));
vec = simde_mm256_sub_epi64(vec, min_vec);
simde_mm256_storeu_si256(reinterpret_cast<simde__m256i*>(arr + i),
vec);
}
for (; i < size; ++i) {
arr[i] -= min_val;
}
}
#else
static void rebase(int64_t* arr, int64_t min_val, size_t size) {
for (size_t i = 0; i < size; ++i) {
arr[i] -= min_val;
}
}
#endif
};
template <typename T>
class TS2DIFFEncoder : public Encoder {
public:
TS2DIFFEncoder() { init(); }
~TS2DIFFEncoder() { destroy(); }
void reset() override { write_index_ = -1; }
void init() {
block_size_ = 128;
// block_size_ = 16;
delta_arr_ = (T*)common::mem_alloc(sizeof(T) * block_size_,
common::MOD_TS2DIFF_OBJ);
write_index_ = -1;
bits_left_ = 8;
buffer_ = 0;
delta_arr_min_ = 0;
delta_arr_max_ = 0;
first_value_ = 0;
previous_value_ = 0;
}
void destroy() override {
if (delta_arr_ != nullptr) {
common::mem_free(delta_arr_);
delta_arr_ = nullptr;
}
}
void write_bits(int64_t value, int bits, common::ByteStream& out_stream) {
while (bits > 0) {
int shift = bits - bits_left_;
if (shift >= 0) {
buffer_ |=
(uint8_t)((value >> shift) & ((1 << bits_left_) - 1));
bits -= bits_left_;
bits_left_ = 0;
} else {
shift = bits_left_ - bits;
buffer_ |= (uint8_t)(value << shift);
bits_left_ -= bits;
bits = 0;
}
flush_byte_if_full(out_stream);
}
}
void flush_remaining(common::ByteStream& out_stream) {
// FIXME bits_left_ != 0 does not means something to be flushed. (=8)
if (bits_left_ != 0 && bits_left_ != 8) {
bits_left_ = 0;
flush_byte_if_full(out_stream);
}
}
void flush_byte_if_full(common::ByteStream& out_stream) {
if (bits_left_ == 0) {
out_stream.write_buf(&buffer_, 1);
buffer_ = 0;
bits_left_ = 8;
}
}
void rebase_arr(int i) { delta_arr_[i] = delta_arr_[i] - delta_arr_min_; }
int cal_bit_width(T n) {
int bit_width = 0;
while (n > 0) {
bit_width++;
n >>= 1;
}
return bit_width;
}
// Batch bit-pack `count` values (each `bit_width` bits, MSB-first within
// byte) into a single contiguous buffer and write it to out_stream in one
// call. Avoids the per-byte write_buf overhead of the scalar write_bits
// loop.
//
// Returns 0 on success, -1 if bit_width > 56 (accumulator overflow risk;
// caller should fall back to write_bits + flush_remaining).
template <typename U>
static int pack_bits_msb(const U* values, int count, int bit_width,
common::ByteStream& out_stream) {
if (count <= 0 || bit_width <= 0) return 0;
if (bit_width > 56) return -1; // fall back
size_t total_bytes = ((size_t)count * (size_t)bit_width + 7) / 8;
std::vector<uint8_t> buf(total_bytes, 0);
uint64_t accum = 0;
int bits_in_accum = 0;
size_t pos = 0;
const uint64_t mask = (1ULL << bit_width) - 1;
for (int i = 0; i < count; i++) {
uint64_t v = static_cast<uint64_t>(values[i]) & mask;
accum = (accum << bit_width) | v;
bits_in_accum += bit_width;
while (bits_in_accum >= 8) {
buf[pos++] = static_cast<uint8_t>(accum >> (bits_in_accum - 8));
bits_in_accum -= 8;
}
if (bits_in_accum > 0) {
accum &= ((1ULL << bits_in_accum) - 1);
} else {
accum = 0;
}
}
if (bits_in_accum > 0) {
buf[pos++] = static_cast<uint8_t>(accum << (8 - bits_in_accum));
}
out_stream.write_buf(buf.data(), pos);
return 0;
}
int do_encode(T value, common::ByteStream& out_stream);
int encode(bool value, common::ByteStream& out_stream) override;
int encode(int32_t value, common::ByteStream& out_stream) override;
int encode(int64_t value, common::ByteStream& out_stream) override;
int encode(float value, common::ByteStream& out_stream) override;
int encode(double value, common::ByteStream& out_stream) override;
int encode(common::String value, common::ByteStream& out_stream) override;
int encode_batch(const int32_t* values, uint32_t count,
common::ByteStream& out_stream) override;
int encode_batch(const int64_t* values, uint32_t count,
common::ByteStream& out_stream) override;
int flush(common::ByteStream& out_stream) override;
int get_max_byte_size() override {
// The meaning of 24 is: index(4)+width(4)+minDeltaBase(8)+firstValue(8)
return 24 + write_index_ * 8;
}
public:
int block_size_;
T* delta_arr_;
T first_value_;
T previous_value_;
T delta_arr_min_;
T delta_arr_max_;
uint8_t buffer_;
int bits_left_;
int write_index_;
};
template <typename T>
int TS2DIFFEncoder<T>::do_encode(T value, common::ByteStream& out_stream) {
if (write_index_ == -1) {
first_value_ = value;
previous_value_ = first_value_;
write_index_++;
return common::E_OK;
}
// Calculate the delta between the current value and the previous_value_
T delta = value - previous_value_;
previous_value_ = value;
if (write_index_ == 0) {
delta_arr_max_ = delta;
delta_arr_min_ = delta;
}
if (delta > delta_arr_max_) {
delta_arr_max_ = delta;
}
if (delta < delta_arr_min_) {
delta_arr_min_ = delta;
}
delta_arr_[write_index_] = delta;
write_index_++;
if (write_index_ >= block_size_) {
return flush(out_stream);
}
return common::E_OK;
}
template <>
inline int TS2DIFFEncoder<int32_t>::flush(common::ByteStream& out_stream) {
int ret = common::E_OK;
if (write_index_ == -1) {
return common::E_OK;
}
// Subtract the minimum value for each delta_arr_ item
SIMDOps<int32_t>::rebase(delta_arr_, delta_arr_min_, write_index_);
// Calculate the bit length of each value to writer
int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_);
// writer header
common::SerializationUtil::write_ui32(write_index_, out_stream);
common::SerializationUtil::write_ui32(bit_width, out_stream);
common::SerializationUtil::write_ui32(delta_arr_min_, out_stream);
common::SerializationUtil::write_ui32(first_value_, out_stream);
// writer data — batched bit-pack + single write_buf for the common case;
// fall back to per-bit path for the rare wide bit_width.
if (pack_bits_msb(delta_arr_, write_index_, bit_width, out_stream) != 0) {
for (int i = 0; i < write_index_; i++) {
write_bits(delta_arr_[i], bit_width, out_stream);
}
flush_remaining(out_stream);
}
reset();
return ret;
}
template <>
inline int TS2DIFFEncoder<int64_t>::flush(common::ByteStream& out_stream) {
int ret = common::E_OK;
if (write_index_ == -1) {
return common::E_OK;
}
// Subtract the minimum value for each delta_arr_ item
SIMDOps<int64_t>::rebase(delta_arr_, delta_arr_min_, write_index_);
// Calculate the bit length of each value to writer
int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_);
// writer header
common::SerializationUtil::write_i32(write_index_, out_stream);
common::SerializationUtil::write_i32(bit_width, out_stream);
common::SerializationUtil::write_i64(delta_arr_min_, out_stream);
common::SerializationUtil::write_i64(first_value_, out_stream);
// writer data — batched bit-pack + single write_buf for the common case;
// fall back to per-bit path for the rare wide bit_width (>56).
if (pack_bits_msb(delta_arr_, write_index_, bit_width, out_stream) != 0) {
for (int i = 0; i < write_index_; i++) {
write_bits(delta_arr_[i], bit_width, out_stream);
}
flush_remaining(out_stream);
}
reset(); // 语义，writeIndex=-1;
return ret;
}
// ============================================================================
// Batch encode: INT32
// Adjacent-difference removes sequential dependency; SIMD for delta + min/max.
// ============================================================================
template <>
inline int TS2DIFFEncoder<int32_t>::encode_batch(
const int32_t* values, uint32_t count, common::ByteStream& out_stream) {
int ret = common::E_OK;
uint32_t offset = 0;
while (offset < count) {
// Start of new block: store first_value
if (write_index_ == -1) {
first_value_ = values[offset];
previous_value_ = first_value_;
write_index_ = 0;
offset++;
continue;
}
// How many deltas fit in current block
uint32_t space = static_cast<uint32_t>(block_size_) - write_index_;
uint32_t batch = std::min(count - offset, space);
// ── Adjacent difference: delta[i] = values[i] - values[i-1] ──
// First delta uses previous_value_
delta_arr_[write_index_] = values[offset] - previous_value_;
uint32_t i = 1;
#ifdef ENABLE_SIMD
// SIMD: 4 adjacent differences at a time
for (; i + 3 < batch; i += 4) {
simde__m128i cur = simde_mm_loadu_si128(
reinterpret_cast<const simde__m128i*>(values + offset + i));
simde__m128i prv = simde_mm_loadu_si128(
reinterpret_cast<const simde__m128i*>(values + offset + i - 1));
simde__m128i diff = simde_mm_sub_epi32(cur, prv);
simde_mm_storeu_si128(
reinterpret_cast<simde__m128i*>(delta_arr_ + write_index_ + i),
diff);
}
#endif
for (; i < batch; i++) {
delta_arr_[write_index_ + i] =
values[offset + i] - values[offset + i - 1];
}
previous_value_ = values[offset + batch - 1];
// ── Min/max of new deltas ──
int32_t local_min = delta_arr_[write_index_];
int32_t local_max = delta_arr_[write_index_];
uint32_t j = 1;
#ifdef ENABLE_SIMD
if (batch >= 5) {
simde__m128i vmin = simde_mm_set1_epi32(local_min);
simde__m128i vmax = vmin;
for (; j + 3 < batch; j += 4) {
simde__m128i v =
simde_mm_loadu_si128(reinterpret_cast<const simde__m128i*>(
delta_arr_ + write_index_ + j));
vmin = simde_mm_min_epi32(vmin, v);
vmax = simde_mm_max_epi32(vmax, v);
}
// Horizontal reduce
int32_t tmp[4];
simde_mm_storeu_si128(reinterpret_cast<simde__m128i*>(tmp), vmin);
for (int k = 0; k < 4; k++)
if (tmp[k] < local_min) local_min = tmp[k];
simde_mm_storeu_si128(reinterpret_cast<simde__m128i*>(tmp), vmax);
for (int k = 0; k < 4; k++)
if (tmp[k] > local_max) local_max = tmp[k];
}
#endif
for (; j < batch; j++) {
int32_t d = delta_arr_[write_index_ + j];
if (d < local_min) local_min = d;
if (d > local_max) local_max = d;
}
// Merge with block min/max
if (write_index_ == 0) {
delta_arr_min_ = local_min;
delta_arr_max_ = local_max;
} else {
if (local_min < delta_arr_min_) delta_arr_min_ = local_min;
if (local_max > delta_arr_max_) delta_arr_max_ = local_max;
}
write_index_ += batch;
offset += batch;
if (write_index_ >= block_size_) {
if (RET_FAIL(flush(out_stream))) return ret;
}
}
return ret;
}
// ============================================================================
// Batch encode: INT64
// ============================================================================
template <>
inline int TS2DIFFEncoder<int64_t>::encode_batch(
const int64_t* values, uint32_t count, common::ByteStream& out_stream) {
int ret = common::E_OK;
uint32_t offset = 0;
while (offset < count) {
if (write_index_ == -1) {
first_value_ = values[offset];
previous_value_ = first_value_;
write_index_ = 0;
offset++;
continue;
}
uint32_t space = static_cast<uint32_t>(block_size_) - write_index_;
uint32_t batch = std::min(count - offset, space);
// Adjacent difference
delta_arr_[write_index_] = values[offset] - previous_value_;
uint32_t i = 1;
#ifdef ENABLE_SIMD
// SIMD: 2 adjacent differences at a time (128-bit, native NEON)
for (; i + 1 < batch; i += 2) {
simde__m128i cur = simde_mm_loadu_si128(
reinterpret_cast<const simde__m128i*>(values + offset + i));
simde__m128i prv = simde_mm_loadu_si128(
reinterpret_cast<const simde__m128i*>(values + offset + i - 1));
simde__m128i diff = simde_mm_sub_epi64(cur, prv);
simde_mm_storeu_si128(
reinterpret_cast<simde__m128i*>(delta_arr_ + write_index_ + i),
diff);
}
#endif
for (; i < batch; i++) {
delta_arr_[write_index_ + i] =
values[offset + i] - values[offset + i - 1];
}
previous_value_ = values[offset + batch - 1];
// Min/max (scalar — no efficient 64-bit SIMD min/max before AVX-512)
int64_t local_min = delta_arr_[write_index_];
int64_t local_max = delta_arr_[write_index_];
for (uint32_t j = 1; j < batch; j++) {
int64_t d = delta_arr_[write_index_ + j];
if (d < local_min) local_min = d;
if (d > local_max) local_max = d;
}
if (write_index_ == 0) {
delta_arr_min_ = local_min;
delta_arr_max_ = local_max;
} else {
if (local_min < delta_arr_min_) delta_arr_min_ = local_min;
if (local_max > delta_arr_max_) delta_arr_max_ = local_max;
}
write_index_ += batch;
offset += batch;
if (write_index_ >= block_size_) {
if (RET_FAIL(flush(out_stream))) return ret;
}
}
return ret;
}
// Default: unsupported types fall back to base class loop
template <typename T>
int TS2DIFFEncoder<T>::encode_batch(const int32_t* values, uint32_t count,
common::ByteStream& out) {
return Encoder::encode_batch(values, count, out);
}
template <typename T>
int TS2DIFFEncoder<T>::encode_batch(const int64_t* values, uint32_t count,
common::ByteStream& out) {
return Encoder::encode_batch(values, count, out);
}
class FloatTS2DIFFEncoder : public TS2DIFFEncoder<int32_t> {
public:
int do_encode(float value, common::ByteStream& out_stream) {
int32_t value_int = common::float_to_int(value);
return TS2DIFFEncoder<int32_t>::do_encode(value_int, out_stream);
}
int encode(bool value, common::ByteStream& out_stream);
int encode(int32_t value, common::ByteStream& out_stream);
int encode(int64_t value, common::ByteStream& out_stream);
int encode(float value, common::ByteStream& out_stream);
int encode(double value, common::ByteStream& out_stream);
};
class DoubleTS2DIFFEncoder : public TS2DIFFEncoder<int64_t> {
public:
int do_encode(double value, common::ByteStream& out_stream) {
int64_t value_long = common::double_to_long(value);
return TS2DIFFEncoder<int64_t>::do_encode(value_long, out_stream);
}
int encode(bool value, common::ByteStream& out_stream);
int encode(int32_t value, common::ByteStream& out_stream);
int encode(int64_t value, common::ByteStream& out_stream);
int encode(float value, common::ByteStream& out_stream);
int encode(double value, common::ByteStream& out_stream);
};
typedef TS2DIFFEncoder<int32_t> IntTS2DIFFEncoder;
typedef TS2DIFFEncoder<int64_t> LongTS2DIFFEncoder;
// wrap as Encoder
template <>
FORCE_INLINE int IntTS2DIFFEncoder::encode(bool value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int IntTS2DIFFEncoder::encode(int32_t value,
common::ByteStream& out) {
return do_encode(value, out);
}
template <>
FORCE_INLINE int IntTS2DIFFEncoder::encode(int64_t value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int IntTS2DIFFEncoder::encode(float value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int IntTS2DIFFEncoder::encode(double value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int IntTS2DIFFEncoder::encode(common::String value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int LongTS2DIFFEncoder::encode(bool value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int LongTS2DIFFEncoder::encode(int32_t value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int LongTS2DIFFEncoder::encode(int64_t value,
common::ByteStream& out) {
return do_encode(value, out);
}
template <>
FORCE_INLINE int LongTS2DIFFEncoder::encode(float value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int LongTS2DIFFEncoder::encode(double value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
template <>
FORCE_INLINE int LongTS2DIFFEncoder::encode(common::String value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int FloatTS2DIFFEncoder::encode(bool value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int FloatTS2DIFFEncoder::encode(int32_t value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int FloatTS2DIFFEncoder::encode(int64_t value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int FloatTS2DIFFEncoder::encode(float value,
common::ByteStream& out) {
return do_encode(value, out);
}
FORCE_INLINE int FloatTS2DIFFEncoder::encode(double value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int DoubleTS2DIFFEncoder::encode(bool value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int DoubleTS2DIFFEncoder::encode(int32_t value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int DoubleTS2DIFFEncoder::encode(int64_t value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int DoubleTS2DIFFEncoder::encode(float value,
common::ByteStream& out) {
return common::E_TYPE_NOT_MATCH;
}
FORCE_INLINE int DoubleTS2DIFFEncoder::encode(double value,
common::ByteStream& out) {
return do_encode(value, out);
}
} // end namespace storage
#endif // ENCODING_TS2DIFF_ENCODER_H