/*
 * 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.
 */

#include "aligned_chunk_reader.h"

#include <algorithm>

#include "compress/compressor_factory.h"
#include "encoding/decoder_factory.h"

using namespace common;
namespace storage {

int AlignedChunkReader::init(ReadFile* read_file, String m_name,
                             TSDataType data_type, Filter* time_filter) {
    read_file_ = read_file;
    measurement_name_.shallow_copy_from(m_name);
    time_decoder_ = DecoderFactory::alloc_time_decoder();
    value_decoder_ = nullptr;
    time_compressor_ = nullptr;
    value_compressor_ = nullptr;
    time_filter_ = time_filter;
    time_uncompressed_buf_ = nullptr;
    value_uncompressed_buf_ = nullptr;
    if (IS_NULL(time_decoder_)) {
        return E_OOM;
    }
    return E_OK;
}

void AlignedChunkReader::reset() {
    time_chunk_meta_ = nullptr;
    value_chunk_meta_ = nullptr;
    time_chunk_header_.reset();
    value_chunk_header_.reset();
    cur_time_page_header_.reset();
    cur_value_page_header_.reset();

    char* file_data_buf = time_in_stream_.get_wrapped_buf();
    if (file_data_buf != nullptr) {
        mem_free(file_data_buf);
    }
    time_in_stream_.reset();
    file_data_buf = value_in_stream_.get_wrapped_buf();
    if (file_data_buf != nullptr) {
        mem_free(file_data_buf);
    }
    value_in_stream_.reset();
    file_data_time_buf_size_ = 0;
    file_data_value_buf_size_ = 0;
    time_chunk_visit_offset_ = 0;
    value_chunk_visit_offset_ = 0;

    // Free leftover uncompressed buffers from the previous chunk.
    if (time_uncompressed_buf_ != nullptr && time_compressor_ != nullptr) {
        time_compressor_->after_uncompress(time_uncompressed_buf_);
        time_uncompressed_buf_ = nullptr;
    }

    // Multi-value reset
    for (auto* col : value_columns_) {
        // Free uncompressed buffer before resetting.
        if (col->uncompressed_buf != nullptr && col->compressor != nullptr) {
            col->compressor->after_uncompress(col->uncompressed_buf);
            col->uncompressed_buf = nullptr;
        }
        char* buf = col->in_stream.get_wrapped_buf();
        if (buf != nullptr) mem_free(buf);
        col->in_stream.reset();
        col->in.reset();
        col->chunk_header.reset();
        col->cur_page_header.reset();
        col->file_data_buf_size = 0;
        col->chunk_visit_offset = 0;
        col->notnull_bitmap.clear();
        col->cur_value_index = -1;
        col->chunk_meta = nullptr;
        // Note: decoder/compressor are NOT freed here — they are reused by
        // alloc_compressor_and_decoder() in load_by_aligned_meta_multi().
    }
}

void AlignedChunkReader::destroy() {
    if (time_decoder_ != nullptr) {
        time_decoder_->~Decoder();
        DecoderFactory::free(time_decoder_);
        time_decoder_ = nullptr;
    }
    if (value_decoder_ != nullptr) {
        value_decoder_->~Decoder();
        DecoderFactory::free(value_decoder_);
        value_decoder_ = nullptr;
    }
    if (time_compressor_ != nullptr) {
        time_compressor_->~Compressor();
        CompressorFactory::free(time_compressor_);
        time_compressor_ = nullptr;
    }
    if (value_compressor_ != nullptr) {
        value_compressor_->~Compressor();
        CompressorFactory::free(value_compressor_);
        value_compressor_ = nullptr;
    }
    char* buf = time_in_stream_.get_wrapped_buf();
    if (buf != nullptr) {
        mem_free(buf);
        time_in_stream_.clear_wrapped_buf();
    }
    cur_time_page_header_.reset();
    buf = value_in_stream_.get_wrapped_buf();
    if (buf != nullptr) {
        mem_free(buf);
        value_in_stream_.clear_wrapped_buf();
    }
    cur_value_page_header_.reset();
    chunk_header_.~ChunkHeader();

    // Multi-value destroy
    for (auto* col : value_columns_) {
        if (col->decoder != nullptr) {
            col->decoder->~Decoder();
            DecoderFactory::free(col->decoder);
            col->decoder = nullptr;
        }
        if (col->compressor != nullptr) {
            col->compressor->~Compressor();
            CompressorFactory::free(col->compressor);
            col->compressor = nullptr;
        }
        buf = col->in_stream.get_wrapped_buf();
        if (buf != nullptr) {
            mem_free(buf);
            col->in_stream.clear_wrapped_buf();
        }
        col->cur_page_header.reset();
        delete col;
    }
    value_columns_.clear();
    if (decode_pool_ != nullptr) {
        delete decode_pool_;
        decode_pool_ = nullptr;
    }
}

int AlignedChunkReader::load_by_aligned_meta(ChunkMeta* time_chunk_meta,
                                             ChunkMeta* value_chunk_meta) {
    int ret = E_OK;
    time_chunk_meta_ = time_chunk_meta;
    value_chunk_meta_ = value_chunk_meta;
#if DEBUG_SE
    std::cout << "AlignedChunkReader::load_by_meta, meta=" << *time_chunk_meta
              << ", " << *value_chunk_meta << std::endl;
#endif
    /* ================ deserialize time_chunk_header ================*/
    // TODO configurable
    file_data_time_buf_size_ = 1024;
    file_data_value_buf_size_ = 1024;
    int32_t ret_read_len = 0;
    char* time_file_data_buf =
        (char*)mem_alloc(file_data_time_buf_size_, MOD_CHUNK_READER);
    if (IS_NULL(time_file_data_buf)) {
        return E_OOM;
    }
    ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
                           time_file_data_buf, file_data_time_buf_size_,
                           ret_read_len);
    if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
        ret = E_TSFILE_CORRUPTED;
        LOGE("file corrupted, ret=" << ret << ", offset="
                                    << time_chunk_meta_->offset_of_chunk_header_
                                    << "read_len=" << ret_read_len);
        mem_free(time_file_data_buf);
    }
    if (IS_SUCC(ret)) {
        time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
        if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
        } else {
            time_chunk_visit_offset_ = time_in_stream_.read_pos();
        }
    }
    /* ================ deserialize value_chunk_header ================*/
    ret_read_len = 0;
    char* value_file_data_buf =
        (char*)mem_alloc(file_data_value_buf_size_, MOD_CHUNK_READER);
    if (IS_NULL(value_file_data_buf)) {
        return E_OOM;
    }
    ret = read_file_->read(value_chunk_meta_->offset_of_chunk_header_,
                           value_file_data_buf, file_data_value_buf_size_,
                           ret_read_len);
    if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
        ret = E_TSFILE_CORRUPTED;
        LOGE("file corrupted, ret="
             << ret << ", offset=" << value_chunk_meta_->offset_of_chunk_header_
             << "read_len=" << ret_read_len);
        mem_free(value_file_data_buf);
    }
    if (IS_SUCC(ret)) {
        value_in_stream_.wrap_from(value_file_data_buf, ret_read_len);
        if (RET_FAIL(value_chunk_header_.deserialize_from(value_in_stream_))) {
        } else if (RET_FAIL(alloc_compressor_and_decoder(
                       time_decoder_, time_compressor_,
                       time_chunk_header_.encoding_type_,
                       time_chunk_header_.data_type_,
                       time_chunk_header_.compression_type_))) {
        } else if (RET_FAIL(alloc_compressor_and_decoder(
                       value_decoder_, value_compressor_,
                       value_chunk_header_.encoding_type_,
                       value_chunk_header_.data_type_,
                       value_chunk_header_.compression_type_))) {
        } else {
            value_chunk_visit_offset_ = value_in_stream_.read_pos();
#if DEBUG_SE
            std::cout << "AlignedChunkReader::load_by_meta, time_chunk_header="
                      << time_chunk_header_
                      << ", value_chunk_header=" << value_chunk_header_
                      << std::endl;
#endif
        }
    }
    return ret;
}

int AlignedChunkReader::alloc_compressor_and_decoder(
    storage::Decoder*& decoder, storage::Compressor*& compressor,
    TSEncoding encoding, TSDataType data_type, CompressionType compression) {
    if (decoder != nullptr) {
        decoder->reset();
    } else {
        decoder = DecoderFactory::alloc_value_decoder(encoding, data_type);
        if (IS_NULL(decoder)) {
            return E_OOM;
        }
    }

    if (compressor != nullptr) {
        compressor->reset(false);
    } else {
        compressor = CompressorFactory::alloc_compressor(compression);
        if (compressor == nullptr) {
            return E_OOM;
        }
    }
    return E_OK;
}

int AlignedChunkReader::get_next_page(TsBlock* ret_tsblock,
                                      Filter* oneshoot_filter, PageArena& pa) {
    if (multi_value_mode_) {
        return get_next_page_multi(ret_tsblock, oneshoot_filter, pa);
    }
    int ret = E_OK;
    Filter* filter =
        (oneshoot_filter != nullptr ? oneshoot_filter : time_filter_);
    bool pt = prev_time_page_not_finish();
    bool pv = prev_value_page_not_finish();
    if (pt && pv) {
        ret = decode_time_value_buf_into_tsblock(ret_tsblock, oneshoot_filter,
                                                 &pa);
        return ret;
    }
    if (!pt && !pv) {
        while (IS_SUCC(ret)) {
            if (RET_FAIL(get_cur_page_header(
                    time_chunk_meta_, time_in_stream_, cur_time_page_header_,
                    time_chunk_visit_offset_, time_chunk_header_))) {
            } else if (RET_FAIL(get_cur_page_header(
                           value_chunk_meta_, value_in_stream_,
                           cur_value_page_header_, value_chunk_visit_offset_,
                           value_chunk_header_))) {
            } else if (cur_page_statisify_filter(filter)) {
                break;
            } else if (RET_FAIL(skip_cur_page())) {
            }
            if (!has_more_data()) {
                ret = E_NO_MORE_DATA;
                break;
            }
        }
        if (IS_SUCC(ret)) {
            ret = decode_cur_time_page_data() || decode_cur_value_page_data();
        }
    }
    if (IS_SUCC(ret)) {
        ret = decode_time_value_buf_into_tsblock(ret_tsblock, oneshoot_filter,
                                                 &pa);
    }
    return ret;
}

int AlignedChunkReader::get_cur_page_header(ChunkMeta*& chunk_meta,
                                            common::ByteStream& in_stream,
                                            PageHeader& cur_page_header,
                                            uint32_t& chunk_visit_offset,
                                            ChunkHeader& chunk_header,
                                            int32_t* override_buf_size) {
    int ret = E_OK;
    bool retry = true;
    int cur_page_header_serialized_size = 0;
    // TODO： configurable
    int retry_read_want_size = 1024;
    if (chunk_visit_offset - chunk_header.serialized_size_ >=
        chunk_header.data_size_) {
        cur_page_header.reset();
        return E_OK;
    }

    do {
        in_stream.mark_read_pos();
        cur_page_header.reset();
        ret = cur_page_header.deserialize_from(
            in_stream, !chunk_has_only_one_page(chunk_header),
            chunk_header.data_type_);
        cur_page_header_serialized_size = in_stream.get_mark_len();
        if (deserialize_buf_not_enough(ret) && retry) {
            retry = false;
            retry_read_want_size += 1024;
            int32_t& file_data_buf_size =
                override_buf_size != nullptr ? *override_buf_size
                : chunk_header.data_type_ == common::VECTOR
                    ? file_data_time_buf_size_
                    : file_data_value_buf_size_;
            // do not shrink buffer for page header, otherwise, the buffer is
            // most likely to grow back when reading page data
            if (E_OK == read_from_file_and_rewrap(
                            in_stream, chunk_meta, chunk_visit_offset,
                            file_data_buf_size, retry_read_want_size, false)) {
                continue;
            }
        }
        break;
    } while (true);
    if (IS_SUCC(ret)) {
        // visit a header
        chunk_visit_offset += cur_page_header_serialized_size;
    }
#if DEBUG_SE
    std::cout << "get_cur_page_header, ret=" << ret << ", retry=" << retry
              << ", cur_page_header=" << cur_page_header
              << ", chunk_meta->offset_of_chunk_header_="
              << chunk_meta->offset_of_chunk_header_
              << ", cur_page_header_serialized_size="
              << cur_page_header_serialized_size << std::endl;
#endif
    return ret;
}

// reader at least @want_size bytes from file and wrap the buffer into
// @in_stream_
int AlignedChunkReader::read_from_file_and_rewrap(
    common::ByteStream& in_stream_, ChunkMeta*& chunk_meta,
    uint32_t& chunk_visit_offset, int32_t& file_data_buf_size, int want_size,
    bool may_shrink) {
    int ret = E_OK;
    const int DEFAULT_READ_SIZE = 4096;  // may use page_size + page_header_size
    char* file_data_buf = in_stream_.get_wrapped_buf();
    int offset = chunk_meta->offset_of_chunk_header_ + chunk_visit_offset;
    int read_size =
        (want_size < DEFAULT_READ_SIZE ? DEFAULT_READ_SIZE : want_size);
    if (file_data_buf_size < read_size ||
        (may_shrink && read_size < file_data_buf_size / 10)) {
        file_data_buf = (char*)mem_realloc(file_data_buf, read_size);
        if (IS_NULL(file_data_buf)) {
            return E_OOM;
        }
        file_data_buf_size = read_size;
    }
    int ret_read_len = 0;
    if (RET_FAIL(
            read_file_->read(offset, file_data_buf, read_size, ret_read_len))) {
    } else {
        in_stream_.wrap_from(file_data_buf, ret_read_len);
#ifdef DEBUG_SE
        std::cout << "file offset = " << offset << " len = " << ret_read_len
                  << std::endl;
        DEBUG_hex_dump_buf("wrapped buf = ", file_data_buf, 256);
#endif
    }
    return ret;
}

bool AlignedChunkReader::cur_page_statisify_filter(Filter* filter) {
    bool value_satisfy = filter == nullptr ||
                         cur_value_page_header_.statistic_ == nullptr ||
                         filter->satisfy(cur_value_page_header_.statistic_);
    bool time_satisfy = filter == nullptr ||
                        cur_time_page_header_.statistic_ == nullptr ||
                        filter->satisfy(cur_time_page_header_.statistic_);
    return time_satisfy && value_satisfy;
}

int AlignedChunkReader::skip_cur_page() {
    int ret = E_OK;
    // visit a page tv data
    time_chunk_visit_offset_ += cur_time_page_header_.compressed_size_;
    time_in_stream_.wrapped_buf_advance_read_pos(
        cur_time_page_header_.compressed_size_);
    value_chunk_visit_offset_ += cur_value_page_header_.compressed_size_;
    value_in_stream_.wrapped_buf_advance_read_pos(
        cur_value_page_header_.compressed_size_);
    return ret;
}

int AlignedChunkReader::decode_cur_time_page_data() {
    int ret = E_OK;

    // Step 1: make sure we load the whole page data in @in_stream_
    if (time_in_stream_.remaining_size() <
        cur_time_page_header_.compressed_size_) {
        // std::cout << "decode_cur_page_data. in_stream_.remaining_size="<<
        // in_stream_.remaining_size() << ", cur_page_header_.compressed_size_="
        // << cur_page_header_.compressed_size_ << std::endl;
        if (RET_FAIL(read_from_file_and_rewrap(
                time_in_stream_, time_chunk_meta_, time_chunk_visit_offset_,
                file_data_time_buf_size_,
                cur_time_page_header_.compressed_size_))) {
        }
    }

    char* time_compressed_buf = nullptr;
    char* time_uncompressed_buf = nullptr;
    uint32_t time_compressed_buf_size = 0;
    uint32_t time_uncompressed_buf_size = 0;

    // Step 2: do uncompress
    if (IS_SUCC(ret)) {
        time_compressed_buf =
            time_in_stream_.get_wrapped_buf() + time_in_stream_.read_pos();
#ifdef DEBUG_SE
        std::cout << "AlignedChunkReader::decode_cur_page_data,time_in_stream_."
                     "get_wrapped_buf="
                  << (void*)(time_in_stream_.get_wrapped_buf())
                  << ", time_in_stream_.read_pos=" << time_in_stream_.read_pos()
                  << std::endl;
#endif
        time_compressed_buf_size = cur_time_page_header_.compressed_size_;
        time_in_stream_.wrapped_buf_advance_read_pos(time_compressed_buf_size);
        time_chunk_visit_offset_ += time_compressed_buf_size;
        if (RET_FAIL(time_compressor_->reset(false))) {
        } else if (RET_FAIL(time_compressor_->uncompress(
                       time_compressed_buf, time_compressed_buf_size,
                       time_uncompressed_buf, time_uncompressed_buf_size))) {
        } else {
            time_uncompressed_buf_ = time_uncompressed_buf;
        }
#ifdef DEBUG_SE
        DEBUG_hex_dump_buf(
            "AlignedChunkReader reader, time_uncompressed buf = ",
            time_uncompressed_buf, time_uncompressed_buf_size);
#endif
        if (ret != E_OK || time_uncompressed_buf_size !=
                               cur_time_page_header_.uncompressed_size_) {
            ret = E_TSFILE_CORRUPTED;
            ASSERT(false);
        }
    }

    time_decoder_->reset();
#ifdef DEBUG_SE
    DEBUG_hex_dump_buf("AlignedChunkReader reader, time_buf = ", time_buf,
                       time_buf_size);
#endif
    time_in_.wrap_from(time_uncompressed_buf_, time_uncompressed_buf_size);
    return ret;
}

int AlignedChunkReader::decode_cur_value_page_data() {
    int ret = E_OK;

    // Step 1: make sure we load the whole page data in @in_stream_
    if (value_in_stream_.remaining_size() <
        cur_value_page_header_.compressed_size_) {
        // std::cout << "decode_cur_page_data. in_stream_.remaining_size="<<
        // in_stream_.remaining_size() << ", cur_page_header_.compressed_size_="
        // << cur_page_header_.compressed_size_ << std::endl;
        if (RET_FAIL(read_from_file_and_rewrap(
                value_in_stream_, value_chunk_meta_, value_chunk_visit_offset_,
                file_data_value_buf_size_,
                cur_value_page_header_.compressed_size_))) {
        }
    }

    char* value_compressed_buf = nullptr;
    char* value_uncompressed_buf = nullptr;
    uint32_t value_compressed_buf_size = 0;
    uint32_t value_uncompressed_buf_size = 0;
    char* value_buf = nullptr;
    uint32_t value_buf_size = 0;

    if (cur_value_page_header_.compressed_size_ == 0) {
        value_in_.wrap_from(value_buf, 0);
        return E_OK;
    }

    // Step 2: do uncompress
    if (IS_SUCC(ret)) {
        value_compressed_buf =
            value_in_stream_.get_wrapped_buf() + value_in_stream_.read_pos();
        value_compressed_buf_size = cur_value_page_header_.compressed_size_;
        value_in_stream_.wrapped_buf_advance_read_pos(
            value_compressed_buf_size);
        value_chunk_visit_offset_ += value_compressed_buf_size;
        if (RET_FAIL(value_compressor_->reset(false))) {
        } else if (RET_FAIL(value_compressor_->uncompress(
                       value_compressed_buf, value_compressed_buf_size,
                       value_uncompressed_buf, value_uncompressed_buf_size))) {
        } else {
            value_uncompressed_buf_ = value_uncompressed_buf;
        }
#ifdef DEBUG_SE
        DEBUG_hex_dump_buf(
            "AlignedChunkReader reader, value_uncompressed buf = ",
            value_uncompressed_buf, value_uncompressed_buf_size);
#endif
        if (ret != E_OK || value_uncompressed_buf_size !=
                               cur_value_page_header_.uncompressed_size_) {
            ret = E_TSFILE_CORRUPTED;
            ASSERT(false);
        }
    }
    // Step 3: get value_buf
    if (IS_SUCC(ret)) {
        uint32_t value_uncompressed_buf_offset = 0;
        value_page_data_num_ =
            SerializationUtil::read_ui32(value_uncompressed_buf);
        value_uncompressed_buf_offset += sizeof(uint32_t);
        value_page_col_notnull_bitmap_.resize((value_page_data_num_ + 7) / 8);
        for (unsigned char& i : value_page_col_notnull_bitmap_) {
            i = *(value_uncompressed_buf + value_uncompressed_buf_offset);
            value_uncompressed_buf_offset++;
        }
        cur_value_index = -1;
        value_buf = value_uncompressed_buf + value_uncompressed_buf_offset;
        value_buf_size =
            value_uncompressed_buf_size - value_uncompressed_buf_offset;
    }
    value_decoder_->reset();
#ifdef DEBUG_SE
    DEBUG_hex_dump_buf("AlignedChunkReader reader, value_buf = ", value_buf,
                       value_buf_size);
#endif
    value_in_.wrap_from(value_buf, value_buf_size);
    return ret;
}

int AlignedChunkReader::decode_time_value_buf_into_tsblock(
    TsBlock*& ret_tsblock, Filter* filter, common::PageArena* pa) {
    int ret = common::E_OK;
    ret = decode_tv_buf_into_tsblock_by_datatype(time_in_, value_in_,
                                                 ret_tsblock, filter, pa);
    // if we return during @decode_tv_buf_into_tsblock, we should keep
    // @uncompressed_buf_ valid until all TV pairs are decoded.
    if (ret != E_OVERFLOW) {
        if (time_uncompressed_buf_ != nullptr) {
            time_compressor_->after_uncompress(time_uncompressed_buf_);
            time_uncompressed_buf_ = nullptr;
        }
        if (value_uncompressed_buf_ != nullptr) {
            value_compressor_->after_uncompress(value_uncompressed_buf_);
            value_uncompressed_buf_ = nullptr;
        }
        if (!prev_value_page_not_finish()) {
            value_in_.reset();
        }
        if (!prev_time_page_not_finish()) {
            time_in_.reset();
        }
        value_page_col_notnull_bitmap_.clear();
        value_page_col_notnull_bitmap_.shrink_to_fit();
    } else {
        ret = E_OK;
    }
    return ret;
}

#define DECODE_TYPED_TV_INTO_TSBLOCK(CppType, ReadType, time_in, value_in,     \
                                     row_appender)                             \
    do {                                                                       \
        uint32_t mask = 1 << 7;                                                \
        int64_t time = 0;                                                      \
        CppType value;                                                         \
        while (time_decoder_->has_remaining(time_in)) {                        \
            cur_value_index++;                                                 \
            if (value_page_col_notnull_bitmap_.empty() ||                      \
                ((value_page_col_notnull_bitmap_[cur_value_index / 8] &        \
                  0xFF) &                                                      \
                 (mask >> (cur_value_index % 8))) == 0) {                      \
                ret = time_decoder_->read_int64(time, time_in);                \
                if (ret != E_OK) {                                             \
                    break;                                                     \
                }                                                              \
                if (UNLIKELY(!row_appender.add_row())) {                       \
                    ret = E_OVERFLOW;                                          \
                    break;                                                     \
                }                                                              \
                row_appender.append(0, (char*)&time, sizeof(time));            \
                row_appender.append_null(1);                                   \
                continue;                                                      \
            }                                                                  \
            assert(value_decoder_->has_remaining(value_in));                   \
            if (!value_decoder_->has_remaining(value_in)) {                    \
                return common::E_DATA_INCONSISTENCY;                           \
            }                                                                  \
            if (UNLIKELY(!row_appender.add_row())) {                           \
                ret = E_OVERFLOW;                                              \
                cur_value_index--;                                             \
                break;                                                         \
            } else if (RET_FAIL(time_decoder_->read_int64(time, time_in))) {   \
            } else if (RET_FAIL(value_decoder_->read_##ReadType(value,         \
                                                                value_in))) {  \
            } else if (filter != nullptr && !filter->satisfy(time, value)) {   \
                row_appender.backoff_add_row();                                \
                continue;                                                      \
            } else {                                                           \
                /*std::cout << "decoder: time=" << time << ", value=" << value \
                 * << std::endl;*/                                             \
                row_appender.append(0, (char*)&time, sizeof(time));            \
                row_appender.append(1, (char*)&value, sizeof(value));          \
            }                                                                  \
        }                                                                      \
    } while (false)

int AlignedChunkReader::i32_DECODE_TYPED_TV_INTO_TSBLOCK(
    ByteStream& time_in, ByteStream& value_in, RowAppender& row_appender,
    Filter* filter) {
    int ret = E_OK;
    uint32_t mask = 1 << 7;
    int64_t time = 0;
    int32_t value;
    while (time_decoder_->has_remaining(time_in)) {
        cur_value_index++;
        if (value_page_col_notnull_bitmap_.empty() ||
            ((value_page_col_notnull_bitmap_[cur_value_index / 8] & 0xFF) &
             (mask >> (cur_value_index % 8))) == 0) {
            ret = time_decoder_->read_int64(time, time_in);
            if (ret != E_OK) {
                break;
            }
            if (UNLIKELY(!row_appender.add_row())) {
                ret = E_OVERFLOW;
                break;
            }
            row_appender.append(0, (char*)&time, sizeof(time));
            row_appender.append_null(1);
            continue;
        }
        assert(value_decoder_->has_remaining(value_in));
        if (!value_decoder_->has_remaining(value_in)) {
            return common::E_DATA_INCONSISTENCY;
        }
        if (UNLIKELY(!row_appender.add_row())) {
            ret = E_OVERFLOW;
            cur_value_index--;
            break;
        } else if (RET_FAIL(time_decoder_->read_int64(time, time_in))) {
        } else if (RET_FAIL(value_decoder_->read_int32(value, value_in))) {
        } else if (filter != nullptr && !filter->satisfy(time, value)) {
            row_appender.backoff_add_row();
            continue;
        } else {
            /*std::cout << "decoder: time=" << time << ", value=" << value
             * << std::endl;*/
            row_appender.append(0, (char*)&time, sizeof(time));
            row_appender.append(1, (char*)&value, sizeof(value));
        }
    }
    return ret;
}

int AlignedChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in,
                                            ByteStream& value_in,
                                            RowAppender& row_appender,
                                            Filter* filter) {
    int ret = E_OK;
    const int BATCH = 129;
    int64_t times[BATCH];
    int32_t values[BATCH];
    const uint32_t null_mask_base = 1 << 7;

    while (time_decoder_->has_remaining(time_in)) {
        if (row_appender.remaining() < (uint32_t)BATCH) {
            ret = E_OVERFLOW;
            break;
        }

        // Block-level time filter check
        bool block_all_pass = false;
        if (filter != nullptr) {
            int64_t block_min, block_max;
            int block_count;
            if (time_decoder_->peek_next_block_range_int64(
                    time_in, block_min, block_max, block_count)) {
                if (!filter->satisfy_start_end_time(block_min, block_max)) {
                    int skipped = 0;
                    time_decoder_->skip_peeked_block_int64(time_in, skipped);
                    int nonnull = 0;
                    for (int i = 0; i < block_count; ++i) {
                        int vi = cur_value_index + 1 + i;
                        if (!value_page_col_notnull_bitmap_.empty() &&
                            ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                             (null_mask_base >> (vi % 8))) != 0) {
                            ++nonnull;
                        }
                    }
                    cur_value_index += block_count;
                    if (nonnull > 0) {
                        int sk = 0;
                        value_decoder_->skip_int32(nonnull, sk, value_in);
                    }
                    continue;
                }
                if (filter->contain_start_end_time(block_min, block_max)) {
                    block_all_pass = true;
                }
            }
        }

        int time_count = 0;
        if (RET_FAIL(time_decoder_->read_batch_int64(
                times, BATCH, time_count, time_in))) {
            break;
        }
        if (time_count == 0) break;

        bool is_null[BATCH];
        int nonnull_count = 0;
        for (int i = 0; i < time_count; ++i) {
            int vi = cur_value_index + 1 + i;
            if (value_page_col_notnull_bitmap_.empty() ||
                ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                 (null_mask_base >> (vi % 8))) == 0) {
                is_null[i] = true;
            } else {
                is_null[i] = false;
                ++nonnull_count;
            }
        }

        bool time_mask[BATCH];
        int pass_count = time_count;
        if (filter != nullptr && !block_all_pass) {
            pass_count = filter->satisfy_batch_time(times, time_count,
                                                     time_mask);
        }

        if (pass_count == 0) {
            if (nonnull_count > 0) {
                int skipped = 0;
                value_decoder_->skip_int32(nonnull_count, skipped, value_in);
            }
            cur_value_index += time_count;
            continue;
        }

        int value_count = 0;
        if (nonnull_count > 0) {
            if (RET_FAIL(value_decoder_->read_batch_int32(
                    values, nonnull_count, value_count, value_in))) {
                break;
            }
        }

        int val_idx = 0;
        for (int i = 0; i < time_count; ++i) {
            cur_value_index++;
            if (filter != nullptr && !block_all_pass && !time_mask[i]) {
                if (!is_null[i]) ++val_idx;
                continue;
            }
            if (is_null[i]) {
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append_null(1);
            } else {
                int32_t val = values[val_idx++];
                if (filter != nullptr && !block_all_pass &&
                    !filter->satisfy(times[i], (int64_t)val)) {
                    continue;
                }
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append(1, (char*)&val, sizeof(int32_t));
            }
        }
        if (ret != E_OK) break;
    }
    return ret;
}

int AlignedChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in,
                                            ByteStream& value_in,
                                            RowAppender& row_appender,
                                            Filter* filter) {
    int ret = E_OK;
    const int BATCH = 129;
    int64_t times[BATCH];
    int64_t values[BATCH];
    const uint32_t null_mask_base = 1 << 7;

    while (time_decoder_->has_remaining(time_in)) {
        if (row_appender.remaining() < (uint32_t)BATCH) {
            ret = E_OVERFLOW;
            break;
        }

        // Block-level time filter check: skip entire block if out of range
        bool block_all_pass = false;
        if (filter != nullptr) {
            int64_t block_min, block_max;
            int block_count;
            if (time_decoder_->peek_next_block_range_int64(
                    time_in, block_min, block_max, block_count)) {
                if (!filter->satisfy_start_end_time(block_min, block_max)) {
                    int skipped = 0;
                    time_decoder_->skip_peeked_block_int64(time_in, skipped);
                    int nonnull = 0;
                    for (int i = 0; i < block_count; ++i) {
                        int vi = cur_value_index + 1 + i;
                        if (!value_page_col_notnull_bitmap_.empty() &&
                            ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                             (null_mask_base >> (vi % 8))) != 0) {
                            ++nonnull;
                        }
                    }
                    cur_value_index += block_count;
                    if (nonnull > 0) {
                        int sk = 0;
                        value_decoder_->skip_int64(nonnull, sk, value_in);
                    }
                    continue;
                }
                if (filter->contain_start_end_time(block_min, block_max)) {
                    block_all_pass = true;
                }
            }
        }

        int time_count = 0;
        if (RET_FAIL(time_decoder_->read_batch_int64(
                times, BATCH, time_count, time_in))) {
            break;
        }
        if (time_count == 0) break;

        bool is_null[BATCH];
        int nonnull_count = 0;
        for (int i = 0; i < time_count; ++i) {
            int vi = cur_value_index + 1 + i;
            if (value_page_col_notnull_bitmap_.empty() ||
                ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                 (null_mask_base >> (vi % 8))) == 0) {
                is_null[i] = true;
            } else {
                is_null[i] = false;
                ++nonnull_count;
            }
        }

        bool time_mask[BATCH];
        int pass_count = time_count;
        if (filter != nullptr && !block_all_pass) {
            pass_count = filter->satisfy_batch_time(times, time_count,
                                                     time_mask);
        }

        if (pass_count == 0) {
            if (nonnull_count > 0) {
                int skipped = 0;
                value_decoder_->skip_int64(nonnull_count, skipped, value_in);
            }
            cur_value_index += time_count;
            continue;
        }

        int value_count = 0;
        if (nonnull_count > 0) {
            if (RET_FAIL(value_decoder_->read_batch_int64(
                    values, nonnull_count, value_count, value_in))) {
                break;
            }
        }

        int val_idx = 0;
        for (int i = 0; i < time_count; ++i) {
            cur_value_index++;
            if (filter != nullptr && !block_all_pass && !time_mask[i]) {
                if (!is_null[i]) ++val_idx;
                continue;
            }
            if (is_null[i]) {
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append_null(1);
            } else {
                int64_t val = values[val_idx++];
                if (filter != nullptr && !block_all_pass &&
                    !filter->satisfy(times[i], val)) {
                    continue;
                }
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append(1, (char*)&val, sizeof(int64_t));
            }
        }
        if (ret != E_OK) break;
    }
    return ret;
}

int AlignedChunkReader::float_DECODE_TV_BATCH(ByteStream& time_in,
                                               ByteStream& value_in,
                                               RowAppender& row_appender,
                                               Filter* filter) {
    int ret = E_OK;
    const int BATCH = 129;
    int64_t times[BATCH];
    float values[BATCH];
    const uint32_t null_mask_base = 1 << 7;

    while (time_decoder_->has_remaining(time_in)) {
        if (row_appender.remaining() < (uint32_t)BATCH) {
            ret = E_OVERFLOW;
            break;
        }

        // Block-level time filter check
        bool block_all_pass = false;
        if (filter != nullptr) {
            int64_t block_min, block_max;
            int block_count;
            if (time_decoder_->peek_next_block_range_int64(
                    time_in, block_min, block_max, block_count)) {
                if (!filter->satisfy_start_end_time(block_min, block_max)) {
                    int skipped = 0;
                    time_decoder_->skip_peeked_block_int64(time_in, skipped);
                    int nonnull = 0;
                    for (int i = 0; i < block_count; ++i) {
                        int vi = cur_value_index + 1 + i;
                        if (!value_page_col_notnull_bitmap_.empty() &&
                            ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                             (null_mask_base >> (vi % 8))) != 0) {
                            ++nonnull;
                        }
                    }
                    cur_value_index += block_count;
                    if (nonnull > 0) {
                        int sk = 0;
                        value_decoder_->skip_float(nonnull, sk, value_in);
                    }
                    continue;
                }
                if (filter->contain_start_end_time(block_min, block_max)) {
                    block_all_pass = true;
                }
            }
        }

        int time_count = 0;
        if (RET_FAIL(time_decoder_->read_batch_int64(
                times, BATCH, time_count, time_in))) {
            break;
        }
        if (time_count == 0) break;

        bool is_null[BATCH];
        int nonnull_count = 0;
        for (int i = 0; i < time_count; ++i) {
            int vi = cur_value_index + 1 + i;
            if (value_page_col_notnull_bitmap_.empty() ||
                ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                 (null_mask_base >> (vi % 8))) == 0) {
                is_null[i] = true;
            } else {
                is_null[i] = false;
                ++nonnull_count;
            }
        }

        bool time_mask[BATCH];
        int pass_count = time_count;
        if (filter != nullptr && !block_all_pass) {
            pass_count = filter->satisfy_batch_time(times, time_count,
                                                     time_mask);
        }

        if (pass_count == 0) {
            if (nonnull_count > 0) {
                int skipped = 0;
                value_decoder_->skip_float(nonnull_count, skipped, value_in);
            }
            cur_value_index += time_count;
            continue;
        }

        int value_count = 0;
        if (nonnull_count > 0) {
            if (RET_FAIL(value_decoder_->read_batch_float(
                    values, nonnull_count, value_count, value_in))) {
                break;
            }
        }

        int val_idx = 0;
        for (int i = 0; i < time_count; ++i) {
            cur_value_index++;
            if (filter != nullptr && !block_all_pass && !time_mask[i]) {
                if (!is_null[i]) ++val_idx;
                continue;
            }
            if (is_null[i]) {
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append_null(1);
            } else {
                float val = values[val_idx++];
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append(1, (char*)&val, sizeof(float));
            }
        }
        if (ret != E_OK) break;
    }
    return ret;
}

int AlignedChunkReader::double_DECODE_TV_BATCH(ByteStream& time_in,
                                               ByteStream& value_in,
                                               RowAppender& row_appender,
                                               Filter* filter) {
    int ret = E_OK;
    const int BATCH = 129;
    int64_t times[BATCH];
    double values[BATCH];
    const uint32_t null_mask_base = 1 << 7;

    while (time_decoder_->has_remaining(time_in)) {
        if (row_appender.remaining() < (uint32_t)BATCH) {
            ret = E_OVERFLOW;
            break;
        }

        // Block-level time filter check
        bool block_all_pass = false;
        if (filter != nullptr) {
            int64_t block_min, block_max;
            int block_count;
            if (time_decoder_->peek_next_block_range_int64(
                    time_in, block_min, block_max, block_count)) {
                if (!filter->satisfy_start_end_time(block_min, block_max)) {
                    int skipped = 0;
                    time_decoder_->skip_peeked_block_int64(time_in, skipped);
                    int nonnull = 0;
                    for (int i = 0; i < block_count; ++i) {
                        int vi = cur_value_index + 1 + i;
                        if (!value_page_col_notnull_bitmap_.empty() &&
                            ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                             (null_mask_base >> (vi % 8))) != 0) {
                            ++nonnull;
                        }
                    }
                    cur_value_index += block_count;
                    if (nonnull > 0) {
                        int sk = 0;
                        value_decoder_->skip_double(nonnull, sk, value_in);
                    }
                    continue;
                }
                if (filter->contain_start_end_time(block_min, block_max)) {
                    block_all_pass = true;
                }
            }
        }

        int time_count = 0;
        if (RET_FAIL(time_decoder_->read_batch_int64(
                times, BATCH, time_count, time_in))) {
            break;
        }
        if (time_count == 0) break;

        bool is_null[BATCH];
        int nonnull_count = 0;
        for (int i = 0; i < time_count; ++i) {
            int vi = cur_value_index + 1 + i;
            if (value_page_col_notnull_bitmap_.empty() ||
                ((value_page_col_notnull_bitmap_[vi / 8] & 0xFF) &
                 (null_mask_base >> (vi % 8))) == 0) {
                is_null[i] = true;
            } else {
                is_null[i] = false;
                ++nonnull_count;
            }
        }

        bool time_mask[BATCH];
        int pass_count = time_count;
        if (filter != nullptr && !block_all_pass) {
            pass_count = filter->satisfy_batch_time(times, time_count,
                                                     time_mask);
        }

        if (pass_count == 0) {
            if (nonnull_count > 0) {
                int skipped = 0;
                value_decoder_->skip_double(nonnull_count, skipped, value_in);
            }
            cur_value_index += time_count;
            continue;
        }

        int value_count = 0;
        if (nonnull_count > 0) {
            if (RET_FAIL(value_decoder_->read_batch_double(
                    values, nonnull_count, value_count, value_in))) {
                break;
            }
        }

        int val_idx = 0;
        for (int i = 0; i < time_count; ++i) {
            cur_value_index++;
            if (filter != nullptr && !block_all_pass && !time_mask[i]) {
                if (!is_null[i]) ++val_idx;
                continue;
            }
            if (is_null[i]) {
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append_null(1);
            } else {
                double val = values[val_idx++];
                if (UNLIKELY(!row_appender.add_row())) {
                    ret = E_OVERFLOW;
                    break;
                }
                row_appender.append(0, (char*)&times[i], sizeof(int64_t));
                row_appender.append(1, (char*)&val, sizeof(double));
            }
        }
        if (ret != E_OK) break;
    }
    return ret;
}

int AlignedChunkReader::decode_tv_buf_into_tsblock_by_datatype(
    ByteStream& time_in, ByteStream& value_in, TsBlock* ret_tsblock,
    Filter* filter, common::PageArena* pa) {
    int ret = E_OK;
    RowAppender row_appender(ret_tsblock);
    switch (value_chunk_header_.data_type_) {
        case common::BOOLEAN:
            DECODE_TYPED_TV_INTO_TSBLOCK(bool, boolean, time_in_, value_in_,
                                         row_appender);
            break;
        case common::DATE:
        case common::INT32:
            ret = i32_DECODE_TV_BATCH(time_in_, value_in_,
                                      row_appender, filter);
            break;
        case common::TIMESTAMP:
        case common::INT64:
            ret = i64_DECODE_TV_BATCH(time_in_, value_in_,
                                      row_appender, filter);
            break;
        case common::FLOAT:
            ret = float_DECODE_TV_BATCH(time_in_, value_in_,
                                        row_appender, filter);
            break;
        case common::DOUBLE:
            ret = double_DECODE_TV_BATCH(time_in_, value_in_,
                                         row_appender, filter);
            break;
        case common::STRING:
        case common::BLOB:
        case common::TEXT:
            ret = STRING_DECODE_TYPED_TV_INTO_TSBLOCK(
                time_in, value_in, row_appender, *pa, filter);
            break;
        default:
            ret = E_NOT_SUPPORT;
            ASSERT(false);
    }
    if (ret_tsblock->get_row_count() == 0 && ret == E_OK) {
        ret = E_NO_MORE_DATA;
    }
    return ret;
}

int AlignedChunkReader::STRING_DECODE_TYPED_TV_INTO_TSBLOCK(
    ByteStream& time_in, ByteStream& value_in, RowAppender& row_appender,
    PageArena& pa, Filter* filter) {
    int ret = E_OK;
    int64_t time = 0;
    common::String value;
    uint32_t mask = 1 << 7;
    while (time_decoder_->has_remaining(time_in)) {
        cur_value_index++;
        bool should_read_data = true;
        if (value_page_col_notnull_bitmap_.empty() ||
            ((value_page_col_notnull_bitmap_[cur_value_index / 8] & 0xFF) &
             (mask >> (cur_value_index % 8))) == 0) {
            should_read_data = false;
        }

        if (should_read_data) {
            assert(value_decoder_->has_remaining(value_in));
            if (!value_decoder_->has_remaining(value_in)) {
                return E_DATA_INCONSISTENCY;
            }
        }

        if (UNLIKELY(!row_appender.add_row())) {
            ret = E_OVERFLOW;
            cur_value_index--;
            break;
        } else if (RET_FAIL(time_decoder_->read_int64(time, time_in))) {
        } else if (should_read_data &&
                   RET_FAIL(value_decoder_->read_String(value, pa, value_in))) {
        } else if (filter != nullptr && !filter->satisfy(time, value)) {
            row_appender.backoff_add_row();
            continue;
        } else {
            row_appender.append(0, (char*)&time, sizeof(time));
            if (!should_read_data) {
                row_appender.append_null(1);
            } else {
                row_appender.append(1, value.buf_, value.len_);
            }
        }
    }
    return ret;
}

// ══════════════════════════════════════════════════════════════════════════
//  Multi-value AlignedChunkReader implementation
// ══════════════════════════════════════════════════════════════════════════

int AlignedChunkReader::load_by_aligned_meta_multi(
    ChunkMeta* time_chunk_meta,
    const std::vector<ChunkMeta*>& value_metas) {
    int ret = E_OK;
    multi_value_mode_ = true;
    time_chunk_meta_ = time_chunk_meta;

    // ── Load time chunk header ──
    file_data_time_buf_size_ = 1024;
    int32_t ret_read_len = 0;
    char* time_file_data_buf =
        (char*)mem_alloc(file_data_time_buf_size_, MOD_CHUNK_READER);
    if (IS_NULL(time_file_data_buf)) return E_OOM;

    ret = read_file_->read(time_chunk_meta_->offset_of_chunk_header_,
                           time_file_data_buf, file_data_time_buf_size_,
                           ret_read_len);
    if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
        ret = E_TSFILE_CORRUPTED;
        mem_free(time_file_data_buf);
        return ret;
    }
    if (IS_SUCC(ret)) {
        time_in_stream_.wrap_from(time_file_data_buf, ret_read_len);
        if (RET_FAIL(time_chunk_header_.deserialize_from(time_in_stream_))) {
            return ret;
        }
        time_chunk_visit_offset_ = time_in_stream_.read_pos();
    }

    // Alloc time decoder/compressor
    if (IS_SUCC(ret)) {
        if (RET_FAIL(alloc_compressor_and_decoder(
                time_decoder_, time_compressor_,
                time_chunk_header_.encoding_type_,
                time_chunk_header_.data_type_,
                time_chunk_header_.compression_type_))) {
            return ret;
        }
    }

    // ── Load each value column ──
    // Reuse existing ValueColumnState objects if count matches (reset() already
    // cleared their internal state).  Otherwise, recreate.
    if (value_columns_.size() != value_metas.size()) {
        for (auto* p : value_columns_) delete p;
        value_columns_.clear();
        value_columns_.reserve(value_metas.size());
        for (size_t c = 0; c < value_metas.size(); c++) {
            value_columns_.push_back(new ValueColumnState);
        }
    }
    for (size_t c = 0; c < value_metas.size() && IS_SUCC(ret); c++) {
        auto* col = value_columns_[c];
        col->chunk_meta = value_metas[c];
        col->file_data_buf_size = 1024;
        ret_read_len = 0;
        char* vbuf =
            (char*)mem_alloc(col->file_data_buf_size, MOD_CHUNK_READER);
        if (IS_NULL(vbuf)) return E_OOM;

        ret = read_file_->read(col->chunk_meta->offset_of_chunk_header_, vbuf,
                               col->file_data_buf_size, ret_read_len);
        if (IS_SUCC(ret) && ret_read_len < ChunkHeader::MIN_SERIALIZED_SIZE) {
            ret = E_TSFILE_CORRUPTED;
            mem_free(vbuf);
            break;
        }
        if (IS_SUCC(ret)) {
            col->in_stream.wrap_from(vbuf, ret_read_len);
            if (RET_FAIL(
                    col->chunk_header.deserialize_from(col->in_stream))) {
                break;
            }
            col->chunk_visit_offset = col->in_stream.read_pos();
            if (RET_FAIL(alloc_compressor_and_decoder(
                    col->decoder, col->compressor,
                    col->chunk_header.encoding_type_,
                    col->chunk_header.data_type_,
                    col->chunk_header.compression_type_))) {
                break;
            }
        }
    }

    // Create thread pool for parallel decode when we have multiple columns.
    if (IS_SUCC(ret) && value_columns_.size() > 1 &&
        decode_pool_ == nullptr) {
        int nthreads = std::min((int)value_columns_.size(), 4);
        decode_pool_ = new common::DecodeThreadPool(nthreads);
    }

    return ret;
}

bool AlignedChunkReader::has_more_data_multi() const {
    if (prev_time_page_not_finish() || prev_any_value_page_not_finish_multi()) {
        return true;
    }
    if (time_chunk_visit_offset_ - time_chunk_header_.serialized_size_ <
        time_chunk_header_.data_size_) {
        return true;
    }
    for (const auto* col : value_columns_) {
        if (col->chunk_visit_offset - col->chunk_header.serialized_size_ <
            col->chunk_header.data_size_) {
            return true;
        }
    }
    return false;
}

bool AlignedChunkReader::prev_any_value_page_not_finish_multi() const {
    for (const auto* col : value_columns_) {
        if ((col->decoder && col->decoder->has_remaining(col->in)) ||
            col->in.has_remaining()) {
            return true;
        }
    }
    return false;
}

int AlignedChunkReader::get_next_page_multi(TsBlock* ret_tsblock,
                                             Filter* oneshoot_filter,
                                             PageArena& pa) {
    int ret = E_OK;
    Filter* filter =
        (oneshoot_filter != nullptr ? oneshoot_filter : time_filter_);

    bool pt = prev_time_page_not_finish();
    bool pv = prev_any_value_page_not_finish_multi();
    if (pt && pv) {
        ret = decode_time_value_buf_into_tsblock_multi(ret_tsblock, filter,
                                                        &pa);
        return ret;
    }
    if (!pt && !pv) {
        while (IS_SUCC(ret)) {
            // Get time page header
            if (RET_FAIL(get_cur_page_header(
                    time_chunk_meta_, time_in_stream_, cur_time_page_header_,
                    time_chunk_visit_offset_, time_chunk_header_))) {
                break;
            }
            // Get each value column's page header
            for (size_t c = 0; c < value_columns_.size() && IS_SUCC(ret);
                 c++) {
                auto* col = value_columns_[c];
                if (RET_FAIL(get_cur_page_header(
                        col->chunk_meta, col->in_stream, col->cur_page_header,
                        col->chunk_visit_offset, col->chunk_header,
                        &col->file_data_buf_size))) {
                }
            }
            if (IS_FAIL(ret)) break;

            if (cur_page_statisify_filter_multi(filter)) {
                break;
            }
            if (RET_FAIL(skip_cur_page_multi())) break;
            if (!has_more_data()) {
                ret = E_NO_MORE_DATA;
                break;
            }
        }
        if (IS_SUCC(ret)) {
            ret = decode_cur_time_page_data();
            if (IS_SUCC(ret)) {
                ret = decode_cur_value_pages_multi();
            }
        }
    }
    if (IS_SUCC(ret)) {
        ret = decode_time_value_buf_into_tsblock_multi(ret_tsblock, filter,
                                                        &pa);
    }
    return ret;
}

bool AlignedChunkReader::cur_page_statisify_filter_multi(Filter* filter) {
    bool time_satisfy = filter == nullptr ||
                        cur_time_page_header_.statistic_ == nullptr ||
                        filter->satisfy(cur_time_page_header_.statistic_);
    return time_satisfy;
}

int AlignedChunkReader::skip_cur_page_multi() {
    time_chunk_visit_offset_ += cur_time_page_header_.compressed_size_;
    time_in_stream_.wrapped_buf_advance_read_pos(
        cur_time_page_header_.compressed_size_);
    for (auto* col : value_columns_) {
        col->chunk_visit_offset += col->cur_page_header.compressed_size_;
        col->in_stream.wrapped_buf_advance_read_pos(
            col->cur_page_header.compressed_size_);
    }
    return E_OK;
}

int AlignedChunkReader::decode_cur_value_pages_multi() {
    int ret = E_OK;
    // Phase 1: Serial IO — ensure each column's page data is in memory.
    for (size_t c = 0; c < value_columns_.size() && IS_SUCC(ret); c++) {
        ret = ensure_value_page_loaded(*value_columns_[c]);
    }
    if (IS_FAIL(ret)) return ret;

    // Phase 2: Parallel CPU — decompress + parse bitmap + reset decoder.
    if (value_columns_.size() > 1 && decode_pool_ != nullptr) {
        std::vector<int> col_rets(value_columns_.size(), E_OK);
        for (size_t c = 0; c < value_columns_.size(); c++) {
            auto* col = value_columns_[c];
            int* col_ret = &col_rets[c];
            decode_pool_->submit([col, col_ret] {
                *col_ret = decompress_and_parse_value_page(*col);
            });
        }
        decode_pool_->wait_all();
        for (size_t c = 0; c < col_rets.size(); c++) {
            if (IS_FAIL(col_rets[c])) return col_rets[c];
        }
    } else {
        for (size_t c = 0; c < value_columns_.size() && IS_SUCC(ret); c++) {
            ret = decompress_and_parse_value_page(*value_columns_[c]);
        }
    }
    return ret;
}

int AlignedChunkReader::decode_cur_value_page_data_for(ValueColumnState& col) {
    int ret = E_OK;

    // Step 1: ensure full page data is loaded
    if (col.in_stream.remaining_size() <
        col.cur_page_header.compressed_size_) {
        if (RET_FAIL(read_from_file_and_rewrap(
                col.in_stream, col.chunk_meta, col.chunk_visit_offset,
                col.file_data_buf_size,
                col.cur_page_header.compressed_size_))) {
            return ret;
        }
    }

    if (col.cur_page_header.compressed_size_ == 0) {
        col.in.wrap_from(nullptr, 0);
        return E_OK;
    }

    // Step 2: uncompress
    char* compressed_buf =
        col.in_stream.get_wrapped_buf() + col.in_stream.read_pos();
    uint32_t compressed_size = col.cur_page_header.compressed_size_;
    col.in_stream.wrapped_buf_advance_read_pos(compressed_size);
    col.chunk_visit_offset += compressed_size;

    char* uncompressed_buf = nullptr;
    uint32_t uncompressed_size = 0;
    if (RET_FAIL(col.compressor->reset(false))) {
        return ret;
    }
    if (RET_FAIL(col.compressor->uncompress(compressed_buf, compressed_size,
                                             uncompressed_buf,
                                             uncompressed_size))) {
        return ret;
    }
    col.uncompressed_buf = uncompressed_buf;

    if (uncompressed_size != col.cur_page_header.uncompressed_size_) {
        return E_TSFILE_CORRUPTED;
    }

    // Step 3: parse bitmap + value data
    uint32_t offset = 0;
    uint32_t data_num = SerializationUtil::read_ui32(uncompressed_buf);
    offset += sizeof(uint32_t);
    col.notnull_bitmap.resize((data_num + 7) / 8);
    for (size_t i = 0; i < col.notnull_bitmap.size(); i++) {
        col.notnull_bitmap[i] = *(uncompressed_buf + offset);
        offset++;
    }
    col.cur_value_index = -1;

    char* value_buf = uncompressed_buf + offset;
    uint32_t value_buf_size = uncompressed_size - offset;
    col.decoder->reset();
    col.in.wrap_from(value_buf, value_buf_size);
    return ret;
}

int AlignedChunkReader::ensure_value_page_loaded(ValueColumnState& col) {
    int ret = E_OK;
    if (col.in_stream.remaining_size() <
        col.cur_page_header.compressed_size_) {
        if (RET_FAIL(read_from_file_and_rewrap(
                col.in_stream, col.chunk_meta, col.chunk_visit_offset,
                col.file_data_buf_size,
                col.cur_page_header.compressed_size_))) {
            return ret;
        }
    }
    return ret;
}

int AlignedChunkReader::decompress_and_parse_value_page(
    ValueColumnState& col) {
    int ret = E_OK;

    if (col.cur_page_header.compressed_size_ == 0) {
        col.in.wrap_from(nullptr, 0);
        return E_OK;
    }

    // Decompress
    char* compressed_buf =
        col.in_stream.get_wrapped_buf() + col.in_stream.read_pos();
    uint32_t compressed_size = col.cur_page_header.compressed_size_;
    col.in_stream.wrapped_buf_advance_read_pos(compressed_size);
    col.chunk_visit_offset += compressed_size;

    char* uncompressed_buf = nullptr;
    uint32_t uncompressed_size = 0;
    if (RET_FAIL(col.compressor->reset(false))) {
        return ret;
    }
    if (RET_FAIL(col.compressor->uncompress(compressed_buf, compressed_size,
                                             uncompressed_buf,
                                             uncompressed_size))) {
        return ret;
    }
    col.uncompressed_buf = uncompressed_buf;

    if (uncompressed_size != col.cur_page_header.uncompressed_size_) {
        return E_TSFILE_CORRUPTED;
    }

    // Parse bitmap + value data
    uint32_t offset = 0;
    uint32_t data_num = SerializationUtil::read_ui32(uncompressed_buf);
    offset += sizeof(uint32_t);
    col.notnull_bitmap.resize((data_num + 7) / 8);
    for (size_t i = 0; i < col.notnull_bitmap.size(); i++) {
        col.notnull_bitmap[i] = *(uncompressed_buf + offset);
        offset++;
    }
    col.cur_value_index = -1;

    char* value_buf = uncompressed_buf + offset;
    uint32_t value_buf_size = uncompressed_size - offset;
    col.decoder->reset();
    col.in.wrap_from(value_buf, value_buf_size);
    return ret;
}

int AlignedChunkReader::decode_time_value_buf_into_tsblock_multi(
    TsBlock*& ret_tsblock, Filter* filter, PageArena* pa) {
    int ret = E_OK;
    RowAppender row_appender(ret_tsblock);
    ret = multi_DECODE_TV_BATCH(ret_tsblock, row_appender, filter, pa);

    // Release uncompressed buffers if pages are done
    if (ret != E_OVERFLOW) {
        if (time_uncompressed_buf_ != nullptr) {
            time_compressor_->after_uncompress(time_uncompressed_buf_);
            time_uncompressed_buf_ = nullptr;
        }
        for (auto* col : value_columns_) {
            if (col->uncompressed_buf != nullptr) {
                col->compressor->after_uncompress(col->uncompressed_buf);
                col->uncompressed_buf = nullptr;
            }
            if (!(col->decoder && col->decoder->has_remaining(col->in)) &&
                !col->in.has_remaining()) {
                col->in.reset();
            }
            col->notnull_bitmap.clear();
            col->notnull_bitmap.shrink_to_fit();
        }
        if (!prev_time_page_not_finish()) {
            time_in_.reset();
        }
    } else {
        ret = E_OK;
    }
    return ret;
}

int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock* ret_tsblock,
                                               RowAppender& row_appender,
                                               Filter* filter,
                                               PageArena* pa) {
    int ret = E_OK;
    const int BATCH = 129;
    int64_t times[BATCH];
    const uint32_t null_mask_base = 1 << 7;
    const uint32_t num_cols = value_columns_.size();


    while (time_decoder_->has_remaining(time_in_)) {
        if (row_appender.remaining() < (uint32_t)BATCH) {
            ret = E_OVERFLOW;
            break;
        }

        // ── Phase 1: Decode a batch of timestamps ──
        int time_count = 0;
        if (RET_FAIL(time_decoder_->read_batch_int64(times, BATCH, time_count,
                                                      time_in_))) {
            break;
        }
        if (time_count == 0) break;

        // ── Phase 2: Apply time filter ──
        bool time_mask[BATCH];
        bool block_all_pass = (filter == nullptr);
        int pass_count = time_count;
        if (!block_all_pass) {
            pass_count =
                filter->satisfy_batch_time(times, time_count, time_mask);
        }

        // ── Phase 3: Per-column null check + value decode ──
        // For each column, compute null flags and decode non-null values.
        // We store decoded values in column-specific buffers.
        // Max 8 bytes per value, 129 values per batch.
        struct ColBatch {
            bool is_null[BATCH];
            int nonnull_count;
            // Value buffer — up to 129 * 8 bytes = 1032 bytes on stack
            char val_buf[BATCH * 8];
            int val_count;
        };
        // Allocate on heap if many columns, stack for small counts
        std::vector<ColBatch> col_batches(num_cols);

        for (uint32_t c = 0; c < num_cols; c++) {
            auto* col = value_columns_[c];
            auto& cb = col_batches[c];
            cb.nonnull_count = 0;
            cb.val_count = 0;
            for (int i = 0; i < time_count; i++) {
                int vi = col->cur_value_index + 1 + i;
                if (col->notnull_bitmap.empty() ||
                    ((col->notnull_bitmap[vi / 8] & 0xFF) &
                     (null_mask_base >> (vi % 8))) == 0) {
                    cb.is_null[i] = true;
                } else {
                    cb.is_null[i] = false;
                    cb.nonnull_count++;
                }
            }

            // Skip values if no rows pass time filter
            if (pass_count == 0 && cb.nonnull_count > 0) {
                switch (col->chunk_header.data_type_) {
                    case common::INT32:
                    case common::DATE: {
                        int sk = 0;
                        col->decoder->skip_int32(cb.nonnull_count, sk, col->in);
                        break;
                    }
                    case common::INT64:
                    case common::TIMESTAMP: {
                        int sk = 0;
                        col->decoder->skip_int64(cb.nonnull_count, sk, col->in);
                        break;
                    }
                    case common::FLOAT: {
                        int sk = 0;
                        col->decoder->skip_float(cb.nonnull_count, sk, col->in);
                        break;
                    }
                    case common::DOUBLE: {
                        int sk = 0;
                        col->decoder->skip_double(cb.nonnull_count, sk, col->in);
                        break;
                    }
                    default:
                        // STRING etc - fall through to value decode
                        break;
                }
                cb.nonnull_count = 0;  // already skipped
            }

            // Decode non-null values
            if (cb.nonnull_count > 0) {
                switch (col->chunk_header.data_type_) {
                    case common::INT32:
                    case common::DATE:
                        col->decoder->read_batch_int32(
                            reinterpret_cast<int32_t*>(cb.val_buf),
                            cb.nonnull_count, cb.val_count, col->in);
                        break;
                    case common::INT64:
                    case common::TIMESTAMP:
                        col->decoder->read_batch_int64(
                            reinterpret_cast<int64_t*>(cb.val_buf),
                            cb.nonnull_count, cb.val_count, col->in);
                        break;
                    case common::FLOAT:
                        col->decoder->read_batch_float(
                            reinterpret_cast<float*>(cb.val_buf),
                            cb.nonnull_count, cb.val_count, col->in);
                        break;
                    case common::DOUBLE:
                        col->decoder->read_batch_double(
                            reinterpret_cast<double*>(cb.val_buf),
                            cb.nonnull_count, cb.val_count, col->in);
                        break;
                    default:
                        // STRING handled below in scatter loop
                        break;
                }
            }
        }

        // ── Phase 4: Skip if no rows pass ──
        if (pass_count == 0) {
            for (uint32_t c = 0; c < num_cols; c++) {
                value_columns_[c]->cur_value_index += time_count;
            }
            continue;
        }

        // ── Phase 5: Scatter into TsBlock ──

        // Fast path: all rows pass filter AND all columns have no nulls
        // → batch memcpy directly into Vector buffers.
        if (pass_count == time_count) {
            bool all_nonnull = true;
            for (uint32_t c = 0; c < num_cols; c++) {
                if (col_batches[c].nonnull_count != time_count) {
                    all_nonnull = false;
                    break;
                }
            }
            if (all_nonnull) {
                // Batch append time column
                common::Vector* time_vec = ret_tsblock->get_vector(0);
                time_vec->get_value_data().append_fixed_value(
                    (const char*)times,
                    static_cast<uint32_t>(time_count) * sizeof(int64_t));
                // Batch append each value column
                for (uint32_t c = 0; c < num_cols; c++) {
                    auto& cb = col_batches[c];
                    auto* col = value_columns_[c];
                    uint32_t elem_size = common::get_data_type_size(
                        col->chunk_header.data_type_);
                    common::Vector* vec = ret_tsblock->get_vector(c + 1);
                    vec->get_value_data().append_fixed_value(
                        cb.val_buf,
                        static_cast<uint32_t>(cb.val_count) * elem_size);
                    col->cur_value_index += time_count;
                }
                row_appender.add_rows(static_cast<uint32_t>(time_count));
                continue;
            }
        }

        // Slow path: per-row scatter (has filter or has nulls)
        std::vector<int> val_idx(num_cols, 0);

        for (int i = 0; i < time_count; i++) {
            bool passes = block_all_pass || time_mask[i];

            if (!passes) {
                for (uint32_t c = 0; c < num_cols; c++) {
                    value_columns_[c]->cur_value_index++;
                    if (!col_batches[c].is_null[i]) val_idx[c]++;
                }
                continue;
            }

            if (UNLIKELY(!row_appender.add_row())) {
                ret = E_OVERFLOW;
                break;
            }

            row_appender.append(0, (char*)&times[i], sizeof(int64_t));

            for (uint32_t c = 0; c < num_cols; c++) {
                value_columns_[c]->cur_value_index++;
                auto& cb = col_batches[c];
                auto* col = value_columns_[c];

                if (cb.is_null[i]) {
                    row_appender.append_null(c + 1);
                } else {
                    uint32_t elem_size =
                        common::get_data_type_size(col->chunk_header.data_type_);
                    row_appender.append(c + 1,
                                        cb.val_buf + val_idx[c] * elem_size,
                                        elem_size);
                    val_idx[c]++;
                }
            }
        }
        if (ret != E_OK) break;
    }
    return ret;
}

}  // end namespace storage