/*
 * 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;
}

// Reconstruct with modulo-2^64 arithmetic, as in the SIMD additions. A
// residual plus the previous value can overflow INT64 before a negative
// minimum delta brings the final value back into range.
static inline int64_t ts2diff_restore_int64(int64_t residual, int64_t previous,
                                            int64_t minimum_delta) {
    const uint64_t bits = static_cast<uint64_t>(residual) +
                          static_cast<uint64_t>(previous) +
                          static_cast<uint64_t>(minimum_delta);
    int64_t value;
    std::memcpy(&value, &bits, sizeof(value));
    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 = ts2diff_restore_int64(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
        // Each SIMD lane gathers eight bytes. Widths 59, 61, 62 and 63 can
        // start at a bit offset that requires a ninth byte; decode those
        // blocks with the scalar path below to preserve the low bits.
        const bool simd_width_fits =
            bit_width_ <= 58 || bit_width_ == 60 || bit_width_ == 64;
        for (; simd_width_fits && 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 = ts2diff_restore_int64(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