| /* |
| * 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 <cmath> |
| #include <limits> |
| #include <vector> |
| |
| #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. |
| // |
| // Result codes: |
| // E_OK → written successfully. |
| // -1 → caller must fall back to write_bits + flush_remaining because |
| // bit_width exceeds the safe accumulator width. |
| // any other non-zero value → real write_buf error; the caller must |
| // propagate it instead of treating the flush as successful. |
| 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 common::E_OK; |
| 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)); |
| } |
| // Surface write failures. Previously the return code was dropped on |
| // the floor and flush() returned E_OK, then reset() wiped the |
| // encoder state — the on-disk page ended up missing its delta block |
| // but the caller thought the data was safe. |
| return out_stream.write_buf(buf.data(), pos); |
| } |
| |
| 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_); |
| // Header writes can fail too (back-pressure / OOM on the underlying |
| // stream); a half-written header followed by reset() leaves the page |
| // corrupted but the caller thinking the data was flushed. |
| if (RET_FAIL( |
| common::SerializationUtil::write_ui32(write_index_, out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL( |
| common::SerializationUtil::write_ui32(bit_width, out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_ui32(delta_arr_min_, |
| out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL( |
| common::SerializationUtil::write_ui32(first_value_, out_stream))) { |
| return ret; |
| } |
| // writer data — batched bit-pack + single write_buf for the common case; |
| // fall back to per-bit path for the rare wide bit_width. |
| const int pack_ret = |
| pack_bits_msb(delta_arr_, write_index_, bit_width, out_stream); |
| if (pack_ret == -1) { |
| for (int i = 0; i < write_index_; i++) { |
| write_bits(delta_arr_[i], bit_width, out_stream); |
| } |
| flush_remaining(out_stream); |
| } else if (pack_ret != common::E_OK) { |
| // Real write failure — don't clear encoder state so the higher |
| // layer can detect the page is poisoned. |
| return pack_ret; |
| } |
| 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_); |
| // Header writes can fail too — see int32 specialization for rationale. |
| if (RET_FAIL( |
| common::SerializationUtil::write_i32(write_index_, out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_i32(bit_width, out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL( |
| common::SerializationUtil::write_i64(delta_arr_min_, out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL( |
| common::SerializationUtil::write_i64(first_value_, out_stream))) { |
| return ret; |
| } |
| // 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). |
| const int pack_ret = |
| pack_bits_msb(delta_arr_, write_index_, bit_width, out_stream); |
| if (pack_ret == -1) { |
| for (int i = 0; i < write_index_; i++) { |
| write_bits(delta_arr_[i], bit_width, out_stream); |
| } |
| flush_remaining(out_stream); |
| } else if (pack_ret != common::E_OK) { |
| return pack_ret; |
| } |
| 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: |
| FloatTS2DIFFEncoder() : max_point_number_(2), max_point_value_(100.0) {} |
| int do_encode(float value, common::ByteStream& out_stream) { |
| int32_t value_int = convert_float_to_int(value); |
| return TS2DIFFEncoder<int32_t>::do_encode(value_int, out_stream); |
| } |
| // PageWriter resets the encoder between pages without going through a |
| // successful flush() (e.g. when the prior page was aborted). The base |
| // reset() only clears write_index_; underflow_flags_ would otherwise |
| // leak the prior page's overflow markers into the next page's bitmap. |
| void reset() override { |
| TS2DIFFEncoder<int32_t>::reset(); |
| underflow_flags_.clear(); |
| } |
| int flush(common::ByteStream& out_stream) override; |
| 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); |
| |
| private: |
| int32_t convert_float_to_int(float value) { |
| const double scaled = static_cast<double>(value) * max_point_value_; |
| if (scaled > static_cast<double>(std::numeric_limits<int32_t>::max()) || |
| scaled < static_cast<double>(std::numeric_limits<int32_t>::min())) { |
| if (std::isnan(value) || |
| value > |
| static_cast<float>(std::numeric_limits<int32_t>::max()) || |
| value < |
| static_cast<float>(std::numeric_limits<int32_t>::min())) { |
| underflow_flags_.push_back(-1); |
| return common::float_to_int(value); |
| } |
| underflow_flags_.push_back(0); |
| return static_cast<int32_t>(std::lround(value)); |
| } |
| if (std::isnan(value)) { |
| underflow_flags_.push_back(-1); |
| return common::float_to_int(value); |
| } |
| underflow_flags_.push_back(1); |
| return static_cast<int32_t>(std::lround(scaled)); |
| } |
| bool has_overflow() const { |
| for (int8_t f : underflow_flags_) { |
| if (f != 1) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| private: |
| int max_point_number_; |
| double max_point_value_; |
| std::vector<int8_t> underflow_flags_; |
| }; |
| |
| class DoubleTS2DIFFEncoder : public TS2DIFFEncoder<int64_t> { |
| public: |
| DoubleTS2DIFFEncoder() : max_point_number_(2), max_point_value_(100.0) {} |
| int do_encode(double value, common::ByteStream& out_stream) { |
| int64_t value_long = convert_double_to_long(value); |
| return TS2DIFFEncoder<int64_t>::do_encode(value_long, out_stream); |
| } |
| // See FloatTS2DIFFEncoder::reset for rationale — the prior page's |
| // overflow markers must not bleed into the next. |
| void reset() override { |
| TS2DIFFEncoder<int64_t>::reset(); |
| underflow_flags_.clear(); |
| } |
| int flush(common::ByteStream& out_stream) override; |
| 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); |
| |
| private: |
| int64_t convert_double_to_long(double value) { |
| const double scaled = value * max_point_value_; |
| if (scaled > static_cast<double>(std::numeric_limits<int64_t>::max()) || |
| scaled < static_cast<double>(std::numeric_limits<int64_t>::min())) { |
| if (std::isnan(value) || |
| value > |
| static_cast<double>(std::numeric_limits<int64_t>::max()) || |
| value < |
| static_cast<double>(std::numeric_limits<int64_t>::min())) { |
| underflow_flags_.push_back(-1); |
| return common::double_to_long(value); |
| } |
| underflow_flags_.push_back(0); |
| return static_cast<int64_t>(std::llround(value)); |
| } |
| if (std::isnan(value)) { |
| underflow_flags_.push_back(-1); |
| return common::double_to_long(value); |
| } |
| underflow_flags_.push_back(1); |
| return static_cast<int64_t>(std::llround(scaled)); |
| } |
| bool has_overflow() const { |
| for (int8_t f : underflow_flags_) { |
| if (f != 1) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| private: |
| int max_point_number_; |
| double max_point_value_; |
| std::vector<int8_t> underflow_flags_; |
| }; |
| |
| 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); |
| } |
| |
| // Keep float/double TS_2DIFF page layout compatible with Java. |
| FORCE_INLINE int FloatTS2DIFFEncoder::flush(common::ByteStream& out_stream) { |
| int ret = common::E_OK; |
| if (write_index_ == -1) { |
| return common::E_OK; |
| } |
| const int num_values = write_index_ + 1; |
| common::ByteStream inner(1024, common::MOD_TS2DIFF_OBJ, false); |
| if (RET_FAIL(common::SerializationUtil::write_var_uint( |
| static_cast<uint32_t>(max_point_number_), inner))) { |
| return ret; |
| } |
| SIMDOps<int32_t>::rebase(delta_arr_, delta_arr_min_, write_index_); |
| int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_); |
| if (RET_FAIL(common::SerializationUtil::write_ui32( |
| static_cast<uint32_t>(write_index_), inner))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_ui32( |
| static_cast<uint32_t>(bit_width), inner))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_ui32( |
| static_cast<uint32_t>(delta_arr_min_), inner))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_ui32( |
| static_cast<uint32_t>(first_value_), inner))) { |
| return ret; |
| } |
| for (int i = 0; i < write_index_; i++) { |
| write_bits(delta_arr_[i], bit_width, inner); |
| } |
| flush_remaining(inner); |
| |
| const bool overflow = has_overflow(); |
| if (overflow) { |
| std::vector<uint8_t> underflow_bitmap( |
| static_cast<size_t>(num_values / 8 + 1), 0); |
| std::vector<uint8_t> overflow_bitmap( |
| static_cast<size_t>(num_values / 8 + 1), 0); |
| bool has_original_value_overflow = false; |
| for (int i = 0; i < num_values; i++) { |
| int8_t f = underflow_flags_[static_cast<size_t>(i)]; |
| if (f == 1) { |
| underflow_bitmap[static_cast<size_t>(i / 8)] |= |
| static_cast<uint8_t>(1u << (i % 8)); |
| } else if (f == -1) { |
| has_original_value_overflow = true; |
| overflow_bitmap[static_cast<size_t>(i / 8)] |= |
| static_cast<uint8_t>(1u << (i % 8)); |
| } |
| } |
| constexpr uint32_t FLAG_SCALED_VALUE_OVERFLOW = |
| 2147483647u; // Integer.MAX_VALUE |
| constexpr uint32_t FLAG_ORIGINAL_VALUE_OVERFLOW = |
| 2147483646u; // Integer.MAX_VALUE - 1 |
| if (RET_FAIL(common::SerializationUtil::write_var_uint( |
| has_original_value_overflow ? FLAG_ORIGINAL_VALUE_OVERFLOW |
| : FLAG_SCALED_VALUE_OVERFLOW, |
| out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_var_uint( |
| static_cast<uint32_t>(num_values), out_stream))) { |
| return ret; |
| } |
| const uint32_t bm_len = static_cast<uint32_t>(num_values / 8 + 1); |
| if (RET_FAIL(out_stream.write_buf(underflow_bitmap.data(), bm_len))) { |
| return ret; |
| } |
| if (has_original_value_overflow && |
| RET_FAIL(out_stream.write_buf(overflow_bitmap.data(), bm_len))) { |
| return ret; |
| } |
| } |
| if (RET_FAIL(merge_byte_stream(out_stream, inner, true))) { |
| return ret; |
| } |
| // Defer encoder-state wipe until after every write into out_stream has |
| // committed. An earlier reset() let a mid-flush failure leave |
| // write_index_ at -1, so the next flush() short-circuited at the top |
| // and the data was silently lost. |
| underflow_flags_.clear(); |
| TS2DIFFEncoder<int32_t>::reset(); |
| return ret; |
| } |
| |
| FORCE_INLINE int DoubleTS2DIFFEncoder::flush(common::ByteStream& out_stream) { |
| int ret = common::E_OK; |
| if (write_index_ == -1) { |
| return common::E_OK; |
| } |
| const int num_values = write_index_ + 1; |
| common::ByteStream inner(1024, common::MOD_TS2DIFF_OBJ, false); |
| if (RET_FAIL(common::SerializationUtil::write_var_uint( |
| static_cast<uint32_t>(max_point_number_), inner))) { |
| return ret; |
| } |
| SIMDOps<int64_t>::rebase(delta_arr_, delta_arr_min_, write_index_); |
| int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_); |
| if (RET_FAIL(common::SerializationUtil::write_i32(write_index_, inner))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_i32(bit_width, inner))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_i64(delta_arr_min_, inner))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_i64(first_value_, inner))) { |
| return ret; |
| } |
| for (int i = 0; i < write_index_; i++) { |
| write_bits(delta_arr_[i], bit_width, inner); |
| } |
| flush_remaining(inner); |
| |
| const bool overflow = has_overflow(); |
| if (overflow) { |
| std::vector<uint8_t> underflow_bitmap( |
| static_cast<size_t>(num_values / 8 + 1), 0); |
| std::vector<uint8_t> overflow_bitmap( |
| static_cast<size_t>(num_values / 8 + 1), 0); |
| bool has_original_value_overflow = false; |
| for (int i = 0; i < num_values; i++) { |
| int8_t f = underflow_flags_[static_cast<size_t>(i)]; |
| if (f == 1) { |
| underflow_bitmap[static_cast<size_t>(i / 8)] |= |
| static_cast<uint8_t>(1u << (i % 8)); |
| } else if (f == -1) { |
| has_original_value_overflow = true; |
| overflow_bitmap[static_cast<size_t>(i / 8)] |= |
| static_cast<uint8_t>(1u << (i % 8)); |
| } |
| } |
| constexpr uint32_t FLAG_SCALED_VALUE_OVERFLOW = |
| 2147483647u; // Integer.MAX_VALUE |
| constexpr uint32_t FLAG_ORIGINAL_VALUE_OVERFLOW = |
| 2147483646u; // Integer.MAX_VALUE - 1 |
| if (RET_FAIL(common::SerializationUtil::write_var_uint( |
| has_original_value_overflow ? FLAG_ORIGINAL_VALUE_OVERFLOW |
| : FLAG_SCALED_VALUE_OVERFLOW, |
| out_stream))) { |
| return ret; |
| } |
| if (RET_FAIL(common::SerializationUtil::write_var_uint( |
| static_cast<uint32_t>(num_values), out_stream))) { |
| return ret; |
| } |
| const uint32_t bm_len = static_cast<uint32_t>(num_values / 8 + 1); |
| if (RET_FAIL(out_stream.write_buf(underflow_bitmap.data(), bm_len))) { |
| return ret; |
| } |
| if (has_original_value_overflow && |
| RET_FAIL(out_stream.write_buf(overflow_bitmap.data(), bm_len))) { |
| return ret; |
| } |
| } |
| if (RET_FAIL(merge_byte_stream(out_stream, inner, true))) { |
| return ret; |
| } |
| // Same deferred-reset rationale as FloatTS2DIFFEncoder::flush — keeping |
| // write_index_ live until every committed write succeeds avoids the |
| // "next flush returns E_OK on lost data" pattern. |
| underflow_flags_.clear(); |
| TS2DIFFEncoder<int64_t>::reset(); |
| return ret; |
| } |
| |
| } // end namespace storage |
| #endif // ENCODING_TS2DIFF_ENCODER_H |
