| /* |
| * 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_DECODER_H |
| #define ENCODING_TS2DIFF_DECODER_H |
| |
| #include <sys/types.h> |
| |
| #include <cmath> |
| #include <cstddef> |
| #include <cstring> |
| #include <vector> |
| |
| #include "common/allocator/alloc_base.h" |
| #include "common/allocator/byte_stream.h" |
| #include "decoder.h" |
| #include "ts2diff_wire_format.h" |
| #include "utils/util_define.h" |
| |
| #ifdef ENABLE_SIMD |
| #include "simde/x86/avx2.h" |
| #endif |
| |
| namespace storage { |
| |
| // ============================================================================ |
| // SIMD batch decode helpers (INT32) |
| // ============================================================================ |
| #ifdef ENABLE_SIMD |
| |
| // Decode 4 INT32 values from bit-packed data using SIMD gather + shift. |
| // @in: pointer to the start of packed bit data for the block |
| // @bit_width: bits per delta value |
| // @delta_min: minimum delta offset for this block |
| // @index: current position within the block (0-based, among write_index_ |
| // deltas) |
| // @base: the previous reconstructed value (for prefix-sum) |
| // @out: output array (4 values written) |
| // Returns: the last reconstructed value (new base for next group) |
| static inline int32_t simd_decode_4_i32(const uint8_t* in, int32_t bit_width, |
| int32_t delta_min, int32_t index, |
| int32_t base, int32_t out[4]) { |
| static const simde__m128i SHUF_REV4 = simde_mm_setr_epi8( |
| 3, 2, 1, 0, 7, 6, 5, 4, 11, 10, 9, 8, 15, 14, 13, 12); |
| |
| const simde__m128i VMIN4 = simde_mm_set1_epi32(delta_min); |
| |
| int32_t pos0 = index * bit_width; |
| int32_t pos[4] = {pos0, pos0 + bit_width, pos0 + 2 * bit_width, |
| pos0 + 3 * bit_width}; |
| int32_t bidx[4] = {pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, pos[3] >> 3}; |
| int32_t off[4] = {pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7}; |
| |
| simde__m128i IDX = simde_mm_setr_epi32(bidx[0], bidx[1], bidx[2], bidx[3]); |
| simde__m128i OFF = simde_mm_setr_epi32(off[0], off[1], off[2], off[3]); |
| |
| simde__m128i V4; |
| |
| if (bit_width <= 16) { |
| int rshift = 32 - bit_width; |
| simde__m128i w32_le = simde_mm_i32gather_epi32((const int*)in, IDX, 1); |
| simde__m128i w32_be = simde_mm_shuffle_epi8(w32_le, SHUF_REV4); |
| simde__m128i U32 = simde_mm_sllv_epi32(w32_be, OFF); |
| simde__m128i RS32 = simde_mm_set1_epi32(rshift); |
| V4 = simde_mm_srlv_epi32(U32, RS32); |
| } else { |
| static const simde__m256i SHUF_REV8 = simde_mm256_setr_epi8( |
| 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, |
| 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); |
| int rshift = 64 - bit_width; |
| simde__m256i w64_le = |
| simde_mm256_i32gather_epi64((const int64_t*)in, IDX, 1); |
| simde__m256i w64_be = simde_mm256_shuffle_epi8(w64_le, SHUF_REV8); |
| simde__m256i OFF64 = simde_mm256_cvtepu32_epi64(OFF); |
| simde__m256i U64 = simde_mm256_sllv_epi64(w64_be, OFF64); |
| simde__m256i V64 = |
| simde_mm256_srl_epi64(U64, simde_mm_cvtsi32_si128(rshift)); |
| simde__m256i perm = simde_mm256_setr_epi32(0, 2, 4, 6, 0, 0, 0, 0); |
| simde__m256i comp = simde_mm256_permutevar8x32_epi32(V64, perm); |
| V4 = simde_mm256_castsi256_si128(comp); |
| } |
| |
| // Add delta_min |
| V4 = simde_mm_add_epi32(V4, VMIN4); |
| |
| // Prefix sum to reconstruct absolute values |
| simde__m128i t; |
| t = simde_mm_slli_si128(V4, 4); |
| V4 = simde_mm_add_epi32(V4, t); |
| t = simde_mm_slli_si128(V4, 8); |
| V4 = simde_mm_add_epi32(V4, t); |
| |
| // Add base |
| simde__m128i C4 = simde_mm_set1_epi32(base); |
| V4 = simde_mm_add_epi32(V4, C4); |
| |
| simde_mm_storeu_si128((simde__m128i*)out, V4); |
| return out[3]; |
| } |
| |
| // Decode 4 INT64 values from bit-packed data using SIMD. |
| static inline int64_t simd_decode_4_i64(const uint8_t* in, int32_t bit_width, |
| int64_t delta_min, int32_t index, |
| int64_t base, int64_t out[4]) { |
| static const simde__m256i SHUF_REV8 = simde_mm256_setr_epi8( |
| 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, |
| 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); |
| |
| const simde__m256i VMIN4 = simde_mm256_set1_epi64x(delta_min); |
| |
| int32_t pos0 = index * bit_width; |
| int32_t pos[4] = {pos0, pos0 + bit_width, pos0 + 2 * bit_width, |
| pos0 + 3 * bit_width}; |
| int32_t bidx[4] = {pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, pos[3] >> 3}; |
| int32_t off[4] = {pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7}; |
| |
| simde__m128i IDX = simde_mm_setr_epi32(bidx[0], bidx[1], bidx[2], bidx[3]); |
| |
| int rshift = 64 - bit_width; |
| simde__m256i w64_le = |
| simde_mm256_i32gather_epi64((const int64_t*)in, IDX, 1); |
| simde__m256i w64_be = simde_mm256_shuffle_epi8(w64_le, SHUF_REV8); |
| simde__m256i OFF64 = simde_mm256_cvtepu32_epi64( |
| simde_mm_setr_epi32(off[0], off[1], off[2], off[3])); |
| simde__m256i U64 = simde_mm256_sllv_epi64(w64_be, OFF64); |
| simde__m256i V64 = |
| simde_mm256_srl_epi64(U64, simde_mm_cvtsi32_si128(rshift)); |
| |
| // Add delta_min |
| V64 = simde_mm256_add_epi64(V64, VMIN4); |
| |
| // Prefix sum (64-bit, 4 lanes) |
| simde__m256i t; |
| // shift by 8 bytes = 1 lane |
| t = simde_mm256_slli_si256(V64, 8); |
| V64 = simde_mm256_add_epi64(V64, t); |
| // cross-lane: add lane[1] to lane[2] and lane[3] |
| // Extract high 128 bits, add broadcast of element[1] to both elements |
| int64_t tmp_buf[4]; |
| simde_mm256_storeu_si256((simde__m256i*)tmp_buf, V64); |
| tmp_buf[2] += tmp_buf[1]; |
| tmp_buf[3] += tmp_buf[1]; |
| V64 = simde_mm256_loadu_si256((const simde__m256i*)tmp_buf); |
| |
| // Add base |
| simde__m256i C4 = simde_mm256_set1_epi64x(base); |
| V64 = simde_mm256_add_epi64(V64, C4); |
| |
| simde_mm256_storeu_si256((simde__m256i*)out, V64); |
| return out[3]; |
| } |
| |
| #endif // ENABLE_SIMD |
| |
| // ============================================================================ |
| // Scalar batch decode helpers |
| // ============================================================================ |
| |
| // Scalar: extract one value from bit-packed data. |
| // @data: pointer to packed bits (NOT advanced; caller handles position) |
| // @bit_pos: bit offset from start of data |
| // @bit_width: bits per value |
| static inline int64_t scalar_read_bits(const uint8_t* data, int32_t bit_pos, |
| int32_t bit_width) { |
| int64_t value = 0; |
| int bits = bit_width; |
| int byte_idx = bit_pos >> 3; |
| int bit_offset = bit_pos & 7; |
| int bits_avail = 8 - bit_offset; |
| |
| while (bits > 0) { |
| if (bits >= bits_avail) { |
| uint8_t d = data[byte_idx] & ((1 << bits_avail) - 1); |
| value = (value << bits_avail) | d; |
| bits -= bits_avail; |
| byte_idx++; |
| bits_avail = 8; |
| } else { |
| uint8_t d = |
| (data[byte_idx] >> (bits_avail - bits)) & ((1 << bits) - 1); |
| value = (value << bits) | d; |
| bits = 0; |
| } |
| } |
| return value; |
| } |
| |
| namespace ts2diff_java_detail { |
| |
| inline bool bitmap_marked(const std::vector<uint8_t>& bm, int idx) { |
| if (bm.empty()) { |
| return false; |
| } |
| size_t byte_idx = static_cast<size_t>(idx / 8); |
| if (byte_idx >= bm.size()) { |
| return false; |
| } |
| return (bm[byte_idx] & static_cast<uint8_t>(1u << (idx % 8))) != 0; |
| } |
| |
| // Page-level FLOAT/DOUBLE metadata, parsed exactly once per page. |
| // Layout (see cpp/docs/ts2diff-float-double-wire-format.md): |
| // form 1: [maxPointNumber varint] |
| // form 2: [Integer.MAX_VALUE][count][scaled bitmap][maxPointNumber] |
| // form 3: [Integer.MAX_VALUE-1][count][scaled bitmap][raw bitmap] |
| // [maxPointNumber] |
| // A leading 0x00 byte is the normal encoding of maxPointNumber = 0 (the |
| // Java Ts2Diff builder default), not a legacy marker. maxPointNumber is |
| // whatever the varint carries: Java applies no upper bound, and a value so |
| // large that Math.pow overflows simply yields maxPointValue = +inf, which |
| // decodes every value through the raw-bits form. |
| // |
| // Inputs that do not match this grammar are a format error (E_DECODE_ERR); |
| // a stream that ends mid-field propagates the underlying read error. |
| struct PageMeta { |
| bool has_scaled_bm = false; |
| bool has_raw_bm = false; |
| std::vector<uint8_t> scaled_bm; |
| std::vector<uint8_t> raw_bm; |
| int max_point_number = 0; |
| int page_value_count = 0; |
| }; |
| |
| inline int read_page_meta(common::ByteStream& in, PageMeta& meta) { |
| int ret = common::E_OK; |
| uint32_t tag = 0; |
| if (RET_FAIL(common::SerializationUtil::read_var_uint(tag, in))) { |
| return ret; |
| } |
| if (tag == FLAG_SCALED_VALUE_OVERFLOW || |
| tag == FLAG_ORIGINAL_VALUE_OVERFLOW) { |
| uint32_t count = 0; |
| if (RET_FAIL(common::SerializationUtil::read_var_uint(count, in))) { |
| return ret; |
| } |
| if (count == 0 || count > 0x7FFFFFFFu) { |
| return common::E_DECODE_ERR; |
| } |
| const int bm_len = static_cast<int>(count) / 8 + 1; |
| // Guard the allocation: a tiny corrupt page must not trigger a |
| // bitmap-sized allocation the stream cannot possibly back. Form 3 |
| // carries two bitmaps of bm_len bytes each, so the required byte |
| // count is bm_len * bitmaps, not bm_len. |
| const uint64_t bitmaps = tag == FLAG_ORIGINAL_VALUE_OVERFLOW ? 2 : 1; |
| if (static_cast<uint64_t>(bm_len) * bitmaps > |
| static_cast<uint64_t>(in.remaining_size())) { |
| return common::E_DECODE_ERR; |
| } |
| meta.has_scaled_bm = true; |
| meta.scaled_bm.resize(static_cast<size_t>(bm_len), 0); |
| uint32_t read_len = 0; |
| if (RET_FAIL(in.read_buf(meta.scaled_bm.data(), |
| static_cast<uint32_t>(bm_len), read_len))) { |
| return ret; |
| } |
| if (read_len != static_cast<uint32_t>(bm_len)) { |
| return common::E_DECODE_ERR; |
| } |
| if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW) { |
| meta.has_raw_bm = true; |
| meta.raw_bm.resize(static_cast<size_t>(bm_len), 0); |
| if (RET_FAIL(in.read_buf(meta.raw_bm.data(), |
| static_cast<uint32_t>(bm_len), |
| read_len))) { |
| return ret; |
| } |
| if (read_len != static_cast<uint32_t>(bm_len)) { |
| return common::E_DECODE_ERR; |
| } |
| } |
| meta.page_value_count = static_cast<int>(count); |
| uint32_t mpn = 0; |
| if (RET_FAIL(common::SerializationUtil::read_var_uint(mpn, in))) { |
| return ret; |
| } |
| meta.max_point_number = static_cast<int>(mpn); |
| } else { |
| meta.max_point_number = static_cast<int>(tag); |
| meta.page_value_count = 0; // unknown until the blocks are decoded |
| } |
| return common::E_OK; |
| } |
| |
| } // namespace ts2diff_java_detail |
| |
| // ============================================================================ |
| // TS2DIFFDecoder template |
| // ============================================================================ |
| template <typename T> |
| class TS2DIFFDecoder : public Decoder { |
| public: |
| TS2DIFFDecoder() { reset(); } |
| ~TS2DIFFDecoder() override {} |
| |
| void reset() override { |
| write_index_ = -1; |
| bits_left_ = 0; |
| stored_value_ = 0; |
| buffer_ = 0; |
| delta_min_ = 0; |
| first_value_ = 0; |
| previous_value_ = 0; |
| bit_width_ = 0; |
| current_index_ = 0; |
| header_peeked_ = false; |
| read_error_ = common::E_OK; |
| max_values_ = -1; |
| } |
| |
| FORCE_INLINE bool has_remaining(const common::ByteStream& buffer) override { |
| if (buffer.has_remaining()) return true; |
| return header_peeked_ || bits_left_ != 0 || |
| (current_index_ <= write_index_ && write_index_ != -1 && |
| current_index_ != 0); |
| } |
| |
| // Reads the 4+4 byte block header. A failure latches read_error_ so |
| // the value-returning decode() path can surface it, and is returned |
| // directly for the batch entry points. |
| // |
| // writeIndex is bounded by the bytes the stream can actually supply, |
| // not by the encoder's BLOCK_DEFAULT_SIZE: Java exposes block-size |
| // constructors, so blocks larger than 129 values are valid input. |
| int read_header(common::ByteStream& in) { |
| int32_t write_index = 0; |
| int32_t bit_width = 0; |
| int ret = common::SerializationUtil::read_i32(write_index, in); |
| if (ret == common::E_OK) { |
| ret = common::SerializationUtil::read_i32(bit_width, in); |
| } |
| if (ret != common::E_OK) { |
| read_error_ = ret; |
| return ret; |
| } |
| const int64_t block_bytes_64 = |
| (static_cast<int64_t>(write_index) * bit_width + 7) / 8; |
| // The value-count bound is computed in int64: write_index + 1 |
| // overflows for write_index == INT32_MAX and would wrap negative, |
| // silently passing the comparison. |
| if (write_index < 0 || bit_width < 0 || |
| bit_width > (int)sizeof(T) * 8 || |
| (max_values_ >= 0 && static_cast<int64_t>(write_index) + 1 > |
| static_cast<int64_t>(max_values_)) || |
| block_bytes_64 > static_cast<int64_t>(in.remaining_size())) { |
| read_error_ = common::E_DECODE_ERR; |
| return common::E_DECODE_ERR; |
| } |
| write_index_ = write_index; |
| bit_width_ = bit_width; |
| return common::E_OK; |
| } |
| |
| void set_max_values(int max_values) { max_values_ = max_values; } |
| |
| // If empty, cache 8 bits from in_stream to 'buffer_'. |
| void read_byte_if_empty(common::ByteStream& in) { |
| if (bits_left_ == 0) { |
| uint32_t read_len = 0; |
| in.read_buf(&buffer_, 1, read_len); |
| if (read_len != 0) { |
| bits_left_ = 8; |
| } |
| } |
| } |
| |
| int64_t read_long(int bits, common::ByteStream& in) { |
| int64_t value = 0; |
| while (bits > 0) { |
| read_byte_if_empty(in); |
| // The stream ended with bits still owed, so the page is |
| // truncated or desynced. read_header's availability check |
| // grants slack for the delta_min/first_value fields, so a |
| // truncated page can reach here. |
| if (bits_left_ == 0 && !in.has_remaining()) { |
| read_error_ = common::E_DECODE_ERR; |
| current_index_ = 0; |
| write_index_ = 0; |
| break; |
| } |
| if (bits > bits_left_ || bits == 8) { |
| // Take only the bits_left_ "least significant" bits. |
| uint8_t d = (uint8_t)(buffer_ & ((1 << bits_left_) - 1)); |
| value = (value << bits_left_) + (d & 0xFF); |
| bits -= bits_left_; |
| bits_left_ = 0; |
| } else { |
| // Shift to correct position and take only least significant |
| // bits. |
| uint8_t d = |
| (uint8_t)((((uint8_t)buffer_) >> (bits_left_ - bits)) & |
| ((1 << bits) - 1)); |
| value = (value << bits) + (d & 0xFF); |
| bits_left_ -= bits; |
| bits = 0; |
| } |
| if (bits <= 0 && current_index_ == 0) { |
| break; |
| } |
| } |
| return value; |
| } |
| |
| T decode(common::ByteStream& in); |
| int read_boolean(bool& ret_value, common::ByteStream& in) override; |
| int read_int32(int32_t& ret_value, common::ByteStream& in) override; |
| int read_int64(int64_t& ret_value, common::ByteStream& in) override; |
| int read_float(float& ret_value, common::ByteStream& in) override; |
| int read_double(double& ret_value, common::ByteStream& in) override; |
| int read_String(common::String& ret_value, common::PageArena& pa, |
| common::ByteStream& in) override; |
| |
| int read_batch_int32(int32_t* out, int capacity, int& actual, |
| common::ByteStream& in) override; |
| int read_batch_int64(int64_t* out, int capacity, int& actual, |
| common::ByteStream& in) override; |
| int skip_int32(int count, int& skipped, common::ByteStream& in) override; |
| int skip_int64(int count, int& skipped, common::ByteStream& in) override; |
| |
| bool peek_next_block_range_int64(common::ByteStream& in, int64_t& block_min, |
| int64_t& block_max, |
| int& block_count) override; |
| int skip_peeked_block_int64(common::ByteStream& in, int& skipped) override; |
| |
| public: |
| T first_value_; |
| T previous_value_; |
| T stored_value_; |
| T delta_min_; |
| uint8_t buffer_; |
| int bits_left_; |
| int bit_width_; |
| int write_index_; |
| int current_index_; |
| bool header_peeked_; |
| // Sticky: a block header or fixed-field read failed. decode() returns |
| // a value rather than an error code, so its callers consult this. |
| int read_error_{common::E_OK}; |
| int max_values_{-1}; |
| }; |
| |
| // ============================================================================ |
| // Per-value decode (unchanged) |
| // ============================================================================ |
| |
| template <> |
| inline int32_t TS2DIFFDecoder<int32_t>::decode(common::ByteStream& in) { |
| int32_t ret_value = stored_value_; |
| if (UNLIKELY(current_index_ == 0)) { |
| if (read_header(in) != common::E_OK) { |
| // Poison the block state so callers stop; value is undefined |
| // for corrupt input. |
| write_index_ = 0; |
| current_index_ = 0; |
| return ret_value; |
| } |
| int fixed_ret = common::SerializationUtil::read_i32(delta_min_, in); |
| if (fixed_ret == common::E_OK) { |
| fixed_ret = common::SerializationUtil::read_i32(first_value_, in); |
| } |
| if (fixed_ret != common::E_OK) { |
| // Surface the stream's own error rather than relabelling a |
| // short read as a format violation. |
| read_error_ = fixed_ret; |
| return ret_value; |
| } |
| ret_value = first_value_; |
| bits_left_ = 0; |
| buffer_ = 0; |
| if (write_index_ == 0) { |
| current_index_ = 0; |
| } else { |
| current_index_ = 1; |
| } |
| return ret_value; |
| } |
| // although it seems we are reading an int64, bit_width_ guarantees |
| // that it does not overflow int32 |
| stored_value_ = read_long(bit_width_, in); |
| ret_value = stored_value_ + first_value_ + delta_min_; |
| if (current_index_++ >= write_index_) { |
| current_index_ = 0; |
| bits_left_ = 0; |
| } |
| first_value_ = ret_value; |
| return ret_value; |
| } |
| |
| template <> |
| inline int64_t TS2DIFFDecoder<int64_t>::decode(common::ByteStream& in) { |
| int64_t ret_value = stored_value_; |
| if (UNLIKELY(current_index_ == 0)) { |
| if (read_header(in) != common::E_OK) { |
| write_index_ = 0; |
| current_index_ = 0; |
| return ret_value; |
| } |
| int fixed_ret = common::SerializationUtil::read_i64(delta_min_, in); |
| if (fixed_ret == common::E_OK) { |
| fixed_ret = common::SerializationUtil::read_i64(first_value_, in); |
| } |
| if (fixed_ret != common::E_OK) { |
| // Surface the stream's own error rather than relabelling a |
| // short read as a format violation. |
| read_error_ = fixed_ret; |
| return ret_value; |
| } |
| ret_value = first_value_; |
| if (write_index_ == 0) { |
| current_index_ = 0; |
| } else { |
| current_index_ = 1; |
| } |
| return ret_value; |
| } |
| stored_value_ = (int64_t)read_long(bit_width_, in); |
| ret_value = stored_value_ + first_value_ + delta_min_; |
| first_value_ = ret_value; |
| if (current_index_++ >= write_index_) { |
| current_index_ = 0; |
| bits_left_ = 0; |
| } |
| return ret_value; |
| } |
| |
| // ============================================================================ |
| // Batch decode: INT32 |
| // Decodes one full block (up to 129 values) per call using SIMD when enabled. |
| // ============================================================================ |
| |
| template <> |
| inline int TS2DIFFDecoder<int32_t>::read_batch_int32(int32_t* out, int capacity, |
| int& actual, |
| common::ByteStream& in) { |
| actual = 0; |
| |
| while (actual < capacity && has_remaining(in)) { |
| // If we are mid-block (current_index_ != 0), finish it per-value. |
| if (current_index_ != 0) { |
| while (actual < capacity && current_index_ != 0 && |
| has_remaining(in)) { |
| out[actual++] = decode(in); |
| } |
| continue; |
| } |
| |
| // Start of a new block — read header |
| int hdr_ret = read_header(in); |
| if (hdr_ret != common::E_OK) { |
| return hdr_ret; |
| } |
| int fixed_ret = common::SerializationUtil::read_i32(delta_min_, in); |
| if (fixed_ret == common::E_OK) { |
| fixed_ret = common::SerializationUtil::read_i32(first_value_, in); |
| } |
| if (fixed_ret != common::E_OK) { |
| read_error_ = fixed_ret; |
| return fixed_ret; |
| } |
| bits_left_ = 0; |
| buffer_ = 0; |
| |
| // Output first_value |
| if (actual >= capacity) { |
| // Must consume first_value next time; set state for per-value path |
| current_index_ = 0; |
| // We already consumed the header; push first_value as stored |
| // and let the next call to decode() handle it. |
| // Actually, we need to handle this: rewind is not possible. |
| // So we output first_value and accept going 1 over capacity. |
| } |
| out[actual++] = first_value_; |
| |
| if (write_index_ == 0) { |
| // Block has only first_value, no deltas |
| current_index_ = 0; |
| continue; |
| } |
| |
| int32_t remaining = write_index_; |
| if (actual + remaining > capacity) { |
| // Block won't fit in output. Fall back to per-value decode. |
| // Stream is at packed data start; bits_left_/buffer_ are reset. |
| current_index_ = 1; |
| continue; |
| } |
| if (!in.is_wrapped()) { |
| // SIMD/scalar block decode below requires a contiguous wrapped |
| // buffer. For a paged ByteStream, drop down to per-value |
| // decode the same way the doesn't-fit branch does. |
| current_index_ = 1; |
| continue; |
| } |
| |
| // Full block decode. Validate against corrupt headers before |
| // advancing the read position — a bogus bit_width_ or write_index_ |
| // could compute a block_bytes that overflows the int32_t multiply |
| // or runs past the wrapped buffer. |
| if (UNLIKELY(write_index_ < 0 || bit_width_ < 0 || bit_width_ > 32)) { |
| return common::E_DECODE_ERR; |
| } |
| const int64_t block_bytes_64 = |
| (static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8; |
| if (UNLIKELY(block_bytes_64 > |
| static_cast<int64_t>(in.remaining_size()))) { |
| return common::E_DECODE_ERR; |
| } |
| if (UNLIKELY(block_bytes_64 > INT32_MAX)) { |
| // This path indexes the packed buffer with int32 offsets. A |
| // block that large is still valid input, so fall back to the |
| // per-value path rather than rejecting it. |
| current_index_ = 1; |
| continue; |
| } |
| const int32_t block_bytes = static_cast<int32_t>(block_bytes_64); |
| const uint8_t* blk_ptr = |
| (const uint8_t*)in.get_wrapped_buf() + in.read_pos(); |
| in.wrapped_buf_advance_read_pos(static_cast<uint32_t>(block_bytes)); |
| |
| int32_t prev = first_value_; |
| int32_t i = 0; |
| |
| #ifdef ENABLE_SIMD |
| // SIMD path: decode 8 values at a time (2 groups of 4) |
| for (; i + 7 < remaining; i += 8) { |
| int32_t need_bytes = ((i + 7) * bit_width_ + bit_width_ + 7) / 8 + |
| (bit_width_ > 16 ? 8 : 4); |
| if (need_bytes > block_bytes) break; |
| |
| int32_t grp_out[8]; |
| prev = simd_decode_4_i32(blk_ptr, bit_width_, delta_min_, i, prev, |
| grp_out); |
| prev = simd_decode_4_i32(blk_ptr, bit_width_, delta_min_, i + 4, |
| prev, grp_out + 4); |
| |
| memcpy(out + actual, grp_out, 8 * sizeof(int32_t)); |
| actual += 8; |
| } |
| #endif |
| |
| // Scalar tail |
| int32_t bit_pos = i * bit_width_; |
| for (; i < remaining; ++i) { |
| int64_t delta = scalar_read_bits(blk_ptr, bit_pos, bit_width_); |
| bit_pos += bit_width_; |
| int32_t val = (int32_t)delta + prev + delta_min_; |
| prev = val; |
| out[actual++] = val; |
| } |
| |
| // Block done, reset state |
| first_value_ = prev; |
| current_index_ = 0; |
| } |
| |
| return common::E_OK; |
| } |
| |
| // ============================================================================ |
| // Batch decode: INT64 |
| // ============================================================================ |
| |
| template <> |
| inline int TS2DIFFDecoder<int64_t>::read_batch_int64(int64_t* out, int capacity, |
| int& actual, |
| common::ByteStream& in) { |
| actual = 0; |
| |
| while (actual < capacity && has_remaining(in)) { |
| // If mid-block, finish per-value |
| if (current_index_ != 0) { |
| while (actual < capacity && current_index_ != 0 && |
| has_remaining(in)) { |
| out[actual++] = decode(in); |
| } |
| continue; |
| } |
| |
| // Start of a new block |
| if (!header_peeked_) { |
| const int hdr_ret = read_header(in); |
| if (hdr_ret != common::E_OK) { |
| return hdr_ret; |
| } |
| int fixed_ret = common::SerializationUtil::read_i64(delta_min_, in); |
| if (fixed_ret == common::E_OK) { |
| fixed_ret = |
| common::SerializationUtil::read_i64(first_value_, in); |
| } |
| if (fixed_ret != common::E_OK) { |
| read_error_ = fixed_ret; |
| return fixed_ret; |
| } |
| bits_left_ = 0; |
| buffer_ = 0; |
| } |
| header_peeked_ = false; |
| |
| out[actual++] = first_value_; |
| |
| if (write_index_ == 0) { |
| current_index_ = 0; |
| continue; |
| } |
| |
| int32_t remaining = write_index_; |
| if (actual + remaining > capacity) { |
| // Block won't fit in output. Fall back to per-value decode. |
| // Stream is at packed data start; bits_left_/buffer_ are reset. |
| current_index_ = 1; |
| continue; |
| } |
| if (!in.is_wrapped()) { |
| // SIMD/scalar block decode below requires a contiguous wrapped |
| // buffer. Page-backed ByteStreams must use the per-value path. |
| current_index_ = 1; |
| continue; |
| } |
| |
| // Validate against corrupt headers (see int32 path). |
| if (UNLIKELY(write_index_ < 0 || bit_width_ < 0 || bit_width_ > 64)) { |
| return common::E_DECODE_ERR; |
| } |
| const int64_t block_bytes_64 = |
| (static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8; |
| if (UNLIKELY(block_bytes_64 > |
| static_cast<int64_t>(in.remaining_size()))) { |
| return common::E_DECODE_ERR; |
| } |
| if (UNLIKELY(block_bytes_64 > INT32_MAX)) { |
| // See the int32 path: int32 packed-buffer offsets, so a block |
| // this large falls back to per-value decode instead of being |
| // rejected. |
| current_index_ = 1; |
| continue; |
| } |
| const int32_t block_bytes = static_cast<int32_t>(block_bytes_64); |
| // Direct pointer into the wrapped ByteStream buffer. |
| const uint8_t* blk_ptr = |
| (const uint8_t*)in.get_wrapped_buf() + in.read_pos(); |
| in.wrapped_buf_advance_read_pos(static_cast<uint32_t>(block_bytes)); |
| |
| int64_t prev = first_value_; |
| int32_t i = 0; |
| |
| // An evenly spaced timestamp block has no packed residual data. Build |
| // the arithmetic progression directly instead of entering the generic |
| // bit-extraction path (whose SIMD guard requires readable input bytes). |
| if (bit_width_ == 0) { |
| #ifdef ENABLE_SIMD |
| if (remaining >= 4) { |
| int64_t value1 = prev + delta_min_; |
| int64_t value2 = value1 + delta_min_; |
| int64_t value3 = value2 + delta_min_; |
| int64_t value4 = value3 + delta_min_; |
| simde__m256i values = |
| simde_mm256_set_epi64x(value4, value3, value2, value1); |
| |
| simde__m256i step = simde_mm256_set1_epi64x(delta_min_); |
| step = simde_mm256_add_epi64(step, step); |
| step = simde_mm256_add_epi64(step, step); |
| |
| for (; i + 3 < remaining; i += 4) { |
| simde_mm256_storeu_si256( |
| reinterpret_cast<simde__m256i*>(out + actual), values); |
| actual += 4; |
| values = simde_mm256_add_epi64(values, step); |
| } |
| prev = out[actual - 1]; |
| } |
| #endif |
| |
| for (; i < remaining; ++i) { |
| prev += delta_min_; |
| out[actual++] = prev; |
| } |
| |
| first_value_ = prev; |
| current_index_ = 0; |
| continue; |
| } |
| |
| #ifdef ENABLE_SIMD |
| // SIMD path: decode 4 INT64 values at a time |
| for (; i + 3 < remaining; i += 4) { |
| int32_t need_bytes = |
| ((i + 3) * bit_width_ + bit_width_ + 7) / 8 + 8; |
| if (need_bytes > block_bytes) break; |
| |
| int64_t grp_out[4]; |
| prev = simd_decode_4_i64(blk_ptr, bit_width_, delta_min_, i, prev, |
| grp_out); |
| memcpy(out + actual, grp_out, 4 * sizeof(int64_t)); |
| actual += 4; |
| } |
| #endif |
| |
| // Scalar tail |
| int32_t bit_pos = i * bit_width_; |
| for (; i < remaining; ++i) { |
| int64_t delta = scalar_read_bits(blk_ptr, bit_pos, bit_width_); |
| bit_pos += bit_width_; |
| int64_t val = delta + prev + delta_min_; |
| prev = val; |
| out[actual++] = val; |
| } |
| |
| first_value_ = prev; |
| current_index_ = 0; |
| } |
| |
| return common::E_OK; |
| } |
| |
| // ============================================================================ |
| // Skip: INT32 — read header only, jump over packed data |
| // ============================================================================ |
| |
| template <> |
| inline int TS2DIFFDecoder<int32_t>::skip_int32(int count, int& skipped, |
| common::ByteStream& in) { |
| skipped = 0; |
| |
| // If mid-block, finish current block per-value |
| while (skipped < count && current_index_ != 0 && has_remaining(in)) { |
| decode(in); |
| ++skipped; |
| } |
| |
| while (skipped < count && has_remaining(in)) { |
| int32_t wi = 0, bw = 0, dm = 0, fv = 0; |
| // The header reads must succeed as a group: a truncated stream |
| // would otherwise leave stale stack values in wi/bw. |
| int rc = common::SerializationUtil::read_i32(wi, in); |
| if (rc == common::E_OK) { |
| rc = common::SerializationUtil::read_i32(bw, in); |
| } |
| if (rc == common::E_OK) { |
| rc = common::SerializationUtil::read_i32(dm, in); |
| } |
| if (rc == common::E_OK) { |
| rc = common::SerializationUtil::read_i32(fv, in); |
| } |
| if (rc != common::E_OK) { |
| return rc; |
| } |
| |
| // Same availability bound as read_header (writeIndex is not |
| // capped by BLOCK_DEFAULT_SIZE); computed in int64 so a corrupt |
| // header cannot overflow the multiply. |
| const int64_t skip_bytes_64 = (static_cast<int64_t>(wi) * bw + 7) / 8; |
| if (wi < 0 || bw < 0 || bw > 32 || |
| skip_bytes_64 > static_cast<int64_t>(in.remaining_size())) { |
| return common::E_DECODE_ERR; |
| } |
| const int32_t skip_bytes = static_cast<int32_t>(skip_bytes_64); |
| |
| int32_t block_vals = wi + 1; |
| bits_left_ = 0; |
| buffer_ = 0; |
| |
| if (count - skipped >= block_vals) { |
| // Whole-block fast path: jump over packed body. |
| in.wrapped_buf_advance_read_pos(skip_bytes); |
| skipped += block_vals; |
| current_index_ = 0; |
| write_index_ = -1; |
| } else { |
| // Partial block: reinstate decoder state as if we'd just |
| // emitted first_value_ from decode(), bump skipped by 1, |
| // then per-value decode the remaining count, leaving the |
| // rest of the block intact for the next decode() call. |
| write_index_ = wi; |
| bit_width_ = bw; |
| delta_min_ = dm; |
| first_value_ = fv; |
| current_index_ = (wi == 0) ? 0 : 1; |
| ++skipped; |
| while (skipped < count && current_index_ != 0 && |
| has_remaining(in)) { |
| decode(in); |
| ++skipped; |
| } |
| } |
| } |
| |
| return common::E_OK; |
| } |
| |
| // ============================================================================ |
| // Skip: INT64 |
| // ============================================================================ |
| |
| template <> |
| inline int TS2DIFFDecoder<int64_t>::skip_int64(int count, int& skipped, |
| common::ByteStream& in) { |
| skipped = 0; |
| |
| while (skipped < count && current_index_ != 0 && has_remaining(in)) { |
| decode(in); |
| ++skipped; |
| } |
| |
| while (skipped < count && has_remaining(in)) { |
| int32_t wi = 0, bw = 0; |
| int64_t dm = 0, fv = 0; |
| // The header reads must succeed as a group: a truncated stream |
| // would otherwise leave stale stack values in wi/bw. |
| int rc = common::SerializationUtil::read_i32(wi, in); |
| if (rc == common::E_OK) { |
| rc = common::SerializationUtil::read_i32(bw, in); |
| } |
| if (rc == common::E_OK) { |
| rc = common::SerializationUtil::read_i64(dm, in); |
| } |
| if (rc == common::E_OK) { |
| rc = common::SerializationUtil::read_i64(fv, in); |
| } |
| if (rc != common::E_OK) { |
| return rc; |
| } |
| |
| // Same availability bound as read_header (writeIndex is not |
| // capped by BLOCK_DEFAULT_SIZE); computed in int64 so a corrupt |
| // header cannot overflow the multiply. |
| const int64_t skip_bytes_64 = (static_cast<int64_t>(wi) * bw + 7) / 8; |
| if (wi < 0 || bw < 0 || bw > 64 || |
| skip_bytes_64 > static_cast<int64_t>(in.remaining_size())) { |
| return common::E_DECODE_ERR; |
| } |
| const int32_t skip_bytes = static_cast<int32_t>(skip_bytes_64); |
| |
| int32_t block_vals = wi + 1; |
| bits_left_ = 0; |
| buffer_ = 0; |
| |
| if (count - skipped >= block_vals) { |
| in.wrapped_buf_advance_read_pos(skip_bytes); |
| skipped += block_vals; |
| current_index_ = 0; |
| write_index_ = -1; |
| } else { |
| write_index_ = wi; |
| bit_width_ = bw; |
| delta_min_ = dm; |
| first_value_ = fv; |
| current_index_ = (wi == 0) ? 0 : 1; |
| ++skipped; |
| while (skipped < count && current_index_ != 0 && |
| has_remaining(in)) { |
| decode(in); |
| ++skipped; |
| } |
| } |
| } |
| |
| return common::E_OK; |
| } |
| |
| // ============================================================================ |
| // Block-level filter check: peek header and compute value range |
| // ============================================================================ |
| |
| template <> |
| inline bool TS2DIFFDecoder<int64_t>::peek_next_block_range_int64( |
| common::ByteStream& in, int64_t& block_min, int64_t& block_max, |
| int& block_count) { |
| // Precondition: this must only be driven on the monotonically-increasing |
| // time column (all callers invoke it via time_decoder_). The block_min = |
| // first_value / block_max = next-block-first_value shortcuts below rely on |
| // that ordering; an unsorted INT64 *value* column that happens to use |
| // TS_2DIFF would get a wrong range here, so it must never be called on a |
| // value decoder (value decoders decode normally and never call this). |
| if (current_index_ != 0 || !has_remaining(in)) return false; |
| |
| if (read_header(in) != common::E_OK) { |
| // Reports "no next block with a usable range", so the caller uses |
| // the regular per-value path. This is deliberately a bool: an |
| // error code would convert to true and make callers act on an |
| // unset range. |
| return false; |
| } |
| int fixed_ret = common::SerializationUtil::read_i64(delta_min_, in); |
| if (fixed_ret == common::E_OK) { |
| fixed_ret = common::SerializationUtil::read_i64(first_value_, in); |
| } |
| if (fixed_ret != common::E_OK) { |
| read_error_ = fixed_ret; |
| return false; |
| } |
| bits_left_ = 0; |
| buffer_ = 0; |
| |
| block_min = first_value_; |
| block_count = write_index_ + 1; |
| |
| // Look-ahead: since timestamps are monotonically increasing, the true |
| // block_max is the last timestamp, which equals next block's first_value_. |
| // The next block header starts at read_pos + packed_bytes. first_value_ is |
| // at offset 16 within the header |
| // (write_index_(4)+bit_width_(4)+delta_min_(8)). We read it via raw pointer |
| // so the stream position is not consumed. |
| // int64 arithmetic, matching skip_peeked_block_int64: read_header bounds |
| // the packed size by the stream length only, so write_index_ * bit_width_ |
| // can exceed INT32_MAX on a huge page. |
| const int64_t packed_bytes = |
| (static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8; |
| if (in.remaining_size() >= static_cast<uint64_t>(packed_bytes) + 24) { |
| char* next_fv_ptr = |
| in.get_wrapped_buf() + in.read_pos() + packed_bytes + 16; |
| block_max = (int64_t)common::SerializationUtil::read_ui64(next_fv_ptr); |
| } else { |
| // Last block in page: fall back to conservative estimate. |
| if (write_index_ == 0 || bit_width_ == 0) { |
| block_max = first_value_ + (int64_t)write_index_ * delta_min_; |
| } else if (bit_width_ >= 63) { |
| block_max = INT64_MAX; |
| } else { |
| int64_t max_delta = delta_min_ + ((1LL << bit_width_) - 1); |
| block_max = first_value_ + (int64_t)write_index_ * max_delta; |
| } |
| } |
| |
| header_peeked_ = true; |
| return true; |
| } |
| |
| template <> |
| inline int TS2DIFFDecoder<int64_t>::skip_peeked_block_int64( |
| common::ByteStream& in, int& skipped) { |
| skipped = write_index_ + 1; |
| // int64 arithmetic: read_header admits blocks whose packed size is |
| // only bounded by the stream length, and write_index_ * bit_width_ |
| // can exceed INT32_MAX on a huge page. |
| const int64_t skip_bytes_64 = |
| (static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8; |
| in.wrapped_buf_advance_read_pos(static_cast<uint64_t>(skip_bytes_64)); |
| header_peeked_ = false; |
| bits_left_ = 0; |
| buffer_ = 0; |
| current_index_ = 0; |
| write_index_ = -1; |
| return common::E_OK; |
| } |
| |
| // INT32 specialization: not applicable (timestamps are always INT64) |
| template <> |
| inline bool TS2DIFFDecoder<int32_t>::peek_next_block_range_int64( |
| common::ByteStream& in, int64_t& block_min, int64_t& block_max, |
| int& block_count) { |
| return false; |
| } |
| |
| template <> |
| inline int TS2DIFFDecoder<int32_t>::skip_peeked_block_int64( |
| common::ByteStream& in, int& skipped) { |
| return common::E_NOT_SUPPORT; |
| } |
| |
| // ============================================================================ |
| // Default (unsupported type) batch/skip — fall back to base class |
| // ============================================================================ |
| |
| template <> |
| inline int TS2DIFFDecoder<int32_t>::read_batch_int64(int64_t* out, int capacity, |
| int& actual, |
| common::ByteStream& in) { |
| return Decoder::read_batch_int64(out, capacity, actual, in); |
| } |
| |
| template <> |
| inline int TS2DIFFDecoder<int32_t>::skip_int64(int count, int& skipped, |
| common::ByteStream& in) { |
| return Decoder::skip_int64(count, skipped, in); |
| } |
| |
| template <> |
| inline int TS2DIFFDecoder<int64_t>::read_batch_int32(int32_t* out, int capacity, |
| int& actual, |
| common::ByteStream& in) { |
| return Decoder::read_batch_int32(out, capacity, actual, in); |
| } |
| |
| template <> |
| inline int TS2DIFFDecoder<int64_t>::skip_int32(int count, int& skipped, |
| common::ByteStream& in) { |
| return Decoder::skip_int32(count, skipped, in); |
| } |
| |
| // ============================================================================ |
| // Float / Double wrapper decoders (unchanged) |
| // ============================================================================ |
| |
| class FloatTS2DIFFDecoder : public TS2DIFFDecoder<int32_t> { |
| public: |
| FloatTS2DIFFDecoder() = default; |
| // PageReader invokes reset() at every page boundary; the page-level |
| // FLOAT/DOUBLE metadata (maxPointNumber, page-wide bitmaps, page value |
| // position) is parsed once per page and survives block transitions. |
| void reset() override { |
| TS2DIFFDecoder<int32_t>::reset(); |
| page_meta_parsed_ = false; |
| max_point_value_ = 1.0; |
| page_pos_ = 0; |
| page_value_count_ = 0; |
| scaled_bm_.clear(); |
| raw_bm_.clear(); |
| } |
| float decode(common::ByteStream& in) { |
| int32_t value_int = TS2DIFFDecoder<int32_t>::decode(in); |
| return common::int_to_float(value_int); |
| } |
| |
| int read_boolean(bool& ret_value, common::ByteStream& in) override; |
| int read_int32(int32_t& ret_value, common::ByteStream& in) override; |
| int read_int64(int64_t& ret_value, common::ByteStream& in) override; |
| int read_float(float& ret_value, common::ByteStream& in) override; |
| int read_double(double& ret_value, common::ByteStream& in) override; |
| |
| int read_batch_float(float* out, int capacity, int& actual, |
| common::ByteStream& in) override; |
| int read_batch_int32(int32_t* out, int capacity, int& actual, |
| common::ByteStream& in) override; |
| |
| private: |
| // Parses the page metadata on the first value of a page. Returns |
| // E_OK and leaves the stream positioned at the first integer block. |
| int ensure_page_meta(common::ByteStream& in) { |
| if (page_meta_parsed_) { |
| return common::E_OK; |
| } |
| ts2diff_java_detail::PageMeta meta; |
| int ret = ts2diff_java_detail::read_page_meta(in, meta); |
| if (RET_FAIL(ret)) { |
| return ret; |
| } |
| max_point_value_ = |
| meta.max_point_number <= 0 |
| ? 1.0 |
| : std::pow(10.0, static_cast<double>(meta.max_point_number)); |
| page_value_count_ = meta.page_value_count; |
| if (page_value_count_ > 0) { |
| TS2DIFFDecoder<int32_t>::set_max_values(page_value_count_); |
| } |
| scaled_bm_ = std::move(meta.scaled_bm); |
| raw_bm_ = std::move(meta.raw_bm); |
| page_pos_ = 0; |
| page_meta_parsed_ = true; |
| return common::E_OK; |
| } |
| |
| float value_at(int32_t value_int) const { |
| if (!raw_bm_.empty() && |
| ts2diff_java_detail::bitmap_marked(raw_bm_, page_pos_)) { |
| return common::int_to_float(value_int); |
| } |
| const bool use_scaled = |
| scaled_bm_.empty() || |
| ts2diff_java_detail::bitmap_marked(scaled_bm_, page_pos_); |
| const double divisor = use_scaled ? max_point_value_ : 1.0; |
| return static_cast<float>(static_cast<double>(value_int) / divisor); |
| } |
| |
| double max_point_value_{1.0}; |
| int page_pos_{0}; |
| int page_value_count_{0}; |
| bool page_meta_parsed_{false}; |
| std::vector<uint8_t> scaled_bm_; |
| std::vector<uint8_t> raw_bm_; |
| }; |
| |
| class DoubleTS2DIFFDecoder : public TS2DIFFDecoder<int64_t> { |
| public: |
| DoubleTS2DIFFDecoder() = default; |
| void reset() override { |
| TS2DIFFDecoder<int64_t>::reset(); |
| page_meta_parsed_ = false; |
| max_point_value_ = 1.0; |
| page_pos_ = 0; |
| page_value_count_ = 0; |
| scaled_bm_.clear(); |
| raw_bm_.clear(); |
| } |
| double decode(common::ByteStream& in) { |
| int64_t value_long = TS2DIFFDecoder<int64_t>::decode(in); |
| return common::long_to_double(value_long); |
| } |
| |
| int read_boolean(bool& ret_value, common::ByteStream& in) override; |
| int read_int32(int32_t& ret_value, common::ByteStream& in) override; |
| int read_int64(int64_t& ret_value, common::ByteStream& in) override; |
| int read_float(float& ret_value, common::ByteStream& in) override; |
| int read_double(double& ret_value, common::ByteStream& in) override; |
| |
| int read_batch_double(double* out, int capacity, int& actual, |
| common::ByteStream& in) override; |
| int read_batch_int64(int64_t* out, int capacity, int& actual, |
| common::ByteStream& in) override; |
| |
| private: |
| int ensure_page_meta(common::ByteStream& in) { |
| if (page_meta_parsed_) { |
| return common::E_OK; |
| } |
| ts2diff_java_detail::PageMeta meta; |
| int ret = ts2diff_java_detail::read_page_meta(in, meta); |
| if (RET_FAIL(ret)) { |
| return ret; |
| } |
| max_point_value_ = |
| meta.max_point_number <= 0 |
| ? 1.0 |
| : std::pow(10.0, static_cast<double>(meta.max_point_number)); |
| page_value_count_ = meta.page_value_count; |
| if (page_value_count_ > 0) { |
| TS2DIFFDecoder<int64_t>::set_max_values(page_value_count_); |
| } |
| scaled_bm_ = std::move(meta.scaled_bm); |
| raw_bm_ = std::move(meta.raw_bm); |
| page_pos_ = 0; |
| page_meta_parsed_ = true; |
| return common::E_OK; |
| } |
| |
| double value_at(int64_t value_long) const { |
| if (!raw_bm_.empty() && |
| ts2diff_java_detail::bitmap_marked(raw_bm_, page_pos_)) { |
| return common::long_to_double(value_long); |
| } |
| const bool use_scaled = |
| scaled_bm_.empty() || |
| ts2diff_java_detail::bitmap_marked(scaled_bm_, page_pos_); |
| const double divisor = use_scaled ? max_point_value_ : 1.0; |
| return static_cast<double>(value_long) / divisor; |
| } |
| |
| double max_point_value_{1.0}; |
| int page_pos_{0}; |
| int page_value_count_{0}; |
| bool page_meta_parsed_{false}; |
| std::vector<uint8_t> scaled_bm_; |
| std::vector<uint8_t> raw_bm_; |
| }; |
| |
| typedef TS2DIFFDecoder<int32_t> IntTS2DIFFDecoder; |
| typedef TS2DIFFDecoder<int64_t> LongTS2DIFFDecoder; |
| |
| // wrap as Decoder interface |
| template <> |
| FORCE_INLINE int IntTS2DIFFDecoder::read_boolean(bool& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int IntTS2DIFFDecoder::read_int32(int32_t& ret_value, |
| common::ByteStream& in) { |
| ret_value = decode(in); |
| return read_error_; |
| } |
| template <> |
| FORCE_INLINE int IntTS2DIFFDecoder::read_int64(int64_t& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int IntTS2DIFFDecoder::read_float(float& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int IntTS2DIFFDecoder::read_double(double& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int IntTS2DIFFDecoder::read_String(common::String& ret_value, |
| common::PageArena& pa, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int LongTS2DIFFDecoder::read_boolean(bool& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int LongTS2DIFFDecoder::read_int32(int32_t& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int LongTS2DIFFDecoder::read_int64(int64_t& ret_value, |
| common::ByteStream& in) { |
| ret_value = decode(in); |
| return read_error_; |
| } |
| template <> |
| FORCE_INLINE int LongTS2DIFFDecoder::read_float(float& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int LongTS2DIFFDecoder::read_double(double& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| template <> |
| FORCE_INLINE int LongTS2DIFFDecoder::read_String(common::String& ret_value, |
| common::PageArena& pa, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_boolean(bool& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_int32(int32_t& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_int64(int64_t& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_float(float& ret_value, |
| common::ByteStream& in) { |
| int ret = common::E_OK; |
| if (RET_FAIL(ensure_page_meta(in))) { |
| return ret; |
| } |
| int32_t value_int = TS2DIFFDecoder<int32_t>::decode(in); |
| if (TS2DIFFDecoder<int32_t>::read_error_ != common::E_OK) { |
| return TS2DIFFDecoder<int32_t>::read_error_; |
| } |
| ret_value = value_at(value_int); |
| page_pos_++; |
| return common::E_OK; |
| } |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_double(double& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| // Page-level metadata is parsed once, then the integer blocks are decoded |
| // with the integer batch decoder (SIMD fast path where available) and the |
| // page-wide bitmaps are applied afterwards using the page position. |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_batch_float(float* out, int capacity, |
| int& actual, |
| common::ByteStream& in) { |
| int ret = common::E_OK; |
| actual = 0; |
| if (RET_FAIL(ensure_page_meta(in))) { |
| return ret; |
| } |
| constexpr int kIntsPerBatch = 128; |
| int32_t ints[kIntsPerBatch]; |
| while (actual < capacity) { |
| int block_actual = 0; |
| const int want = capacity - actual; |
| const int request = want < kIntsPerBatch ? want : kIntsPerBatch; |
| if (RET_FAIL(TS2DIFFDecoder<int32_t>::read_batch_int32( |
| ints, request, block_actual, in))) { |
| return ret; |
| } |
| if (block_actual == 0) { |
| break; |
| } |
| for (int i = 0; i < block_actual; i++) { |
| out[actual + i] = value_at(ints[i]); |
| page_pos_++; |
| } |
| actual += block_actual; |
| } |
| return common::E_OK; |
| } |
| FORCE_INLINE int FloatTS2DIFFDecoder::read_batch_int32(int32_t* out, |
| int capacity, |
| int& actual, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_boolean(bool& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_int32(int32_t& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_int64(int64_t& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_float(float& ret_value, |
| common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_double(double& ret_value, |
| common::ByteStream& in) { |
| int ret = common::E_OK; |
| if (RET_FAIL(ensure_page_meta(in))) { |
| return ret; |
| } |
| int64_t value_long = TS2DIFFDecoder<int64_t>::decode(in); |
| if (TS2DIFFDecoder<int64_t>::read_error_ != common::E_OK) { |
| return TS2DIFFDecoder<int64_t>::read_error_; |
| } |
| ret_value = value_at(value_long); |
| page_pos_++; |
| return common::E_OK; |
| } |
| // See FloatTS2DIFFDecoder::read_batch_float for the layout rationale. |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_batch_double( |
| double* out, int capacity, int& actual, common::ByteStream& in) { |
| int ret = common::E_OK; |
| actual = 0; |
| if (RET_FAIL(ensure_page_meta(in))) { |
| return ret; |
| } |
| constexpr int kIntsPerBatch = 128; |
| int64_t ints[kIntsPerBatch]; |
| while (actual < capacity) { |
| int block_actual = 0; |
| const int want = capacity - actual; |
| const int request = want < kIntsPerBatch ? want : kIntsPerBatch; |
| if (RET_FAIL(TS2DIFFDecoder<int64_t>::read_batch_int64( |
| ints, request, block_actual, in))) { |
| return ret; |
| } |
| if (block_actual == 0) { |
| break; |
| } |
| for (int i = 0; i < block_actual; i++) { |
| out[actual + i] = value_at(ints[i]); |
| page_pos_++; |
| } |
| actual += block_actual; |
| } |
| return common::E_OK; |
| } |
| FORCE_INLINE int DoubleTS2DIFFDecoder::read_batch_int64( |
| int64_t* out, int capacity, int& actual, common::ByteStream& in) { |
| ASSERT(false); |
| return common::E_NOT_SUPPORT; |
| } |
| |
| } // end namespace storage |
| #endif // ENCODING_TS2DIFF_DECODER_H |
