From 455c0b503e918111fe017d19b168272da4155fb8 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 30 Jul 2026 14:32:44 +0800 Subject: [PATCH 1/2] fix(cpp): preserve Gorilla sentinel values in aligned reads --- cpp/src/reader/aligned_chunk_reader.cc | 148 +++++++++++++++++++------ cpp/test/reader/tsfile_reader_test.cc | 94 ++++++++++++++++ 2 files changed, 209 insertions(+), 33 deletions(-) diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index 795ec6b80..1cb18751e 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -803,6 +803,18 @@ FORCE_INLINE int read_value_batch_typed(Decoder* d, double* out, int cap, int& actual, ByteStream& in) { return d->read_batch_double(out, cap, actual, in); } +FORCE_INLINE int read_value_typed(Decoder* d, int32_t& out, ByteStream& in) { + return d->read_int32(out, in); +} +FORCE_INLINE int read_value_typed(Decoder* d, int64_t& out, ByteStream& in) { + return d->read_int64(out, in); +} +FORCE_INLINE int read_value_typed(Decoder* d, float& out, ByteStream& in) { + return d->read_float(out, in); +} +FORCE_INLINE int read_value_typed(Decoder* d, double& out, ByteStream& in) { + return d->read_double(out, in); +} FORCE_INLINE int skip_value_typed(Decoder* d, int32_t*, int n, int& skipped, ByteStream& in) { return d->skip_int32(n, skipped, in); @@ -819,6 +831,56 @@ FORCE_INLINE int skip_value_typed(Decoder* d, double*, int n, int& skipped, ByteStream& in) { return d->skip_double(n, skipped, in); } + +// Aligned pages carry an authoritative non-null count in their bitmap. Some +// encodings also use an otherwise valid fixed-width value as their stream +// terminator (for example, Gorilla uses canonical NaN). The regular batch +// API stops before returning that value. When the bitmap still promises a +// value, consume it through the scalar API and then resume the fast batch +// path. This preserves embedded sentinel values without changing the +// decoder's general streaming contract. +template +FORCE_INLINE int read_value_batch_exact_typed(Decoder* d, T* out, int expected, + int& actual, ByteStream& in) { + actual = 0; + while (actual < expected) { + int batch_actual = 0; + int ret = read_value_batch_typed(d, out + actual, expected - actual, + batch_actual, in); + if (ret != E_OK) return ret; + if (batch_actual < 0 || batch_actual > expected - actual) { + return E_TSFILE_CORRUPTED; + } + actual += batch_actual; + if (actual == expected) return E_OK; + + if ((ret = read_value_typed(d, out[actual], in)) != E_OK) return ret; + ++actual; + } + return E_OK; +} + +template +FORCE_INLINE int skip_value_exact_typed(Decoder* d, T* type_tag, int expected, + int& skipped, ByteStream& in) { + skipped = 0; + while (skipped < expected) { + int batch_skipped = 0; + int ret = skip_value_typed(d, type_tag, expected - skipped, + batch_skipped, in); + if (ret != E_OK) return ret; + if (batch_skipped < 0 || batch_skipped > expected - skipped) { + return E_TSFILE_CORRUPTED; + } + skipped += batch_skipped; + if (skipped == expected) return E_OK; + + T ignored; + if ((ret = read_value_typed(d, ignored, in)) != E_OK) return ret; + ++skipped; + } + return E_OK; +} } // namespace // Unified aligned time+value page decode for fixed-width value types @@ -869,8 +931,9 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, // the loop rather than silently desync the value // decoder. int sk = 0; - if (RET_FAIL(skip_value_typed(value_decoder_, values, - nonnull, sk, value_in))) { + if (RET_FAIL(skip_value_exact_typed(value_decoder_, + values, nonnull, sk, + value_in))) { break; } if (sk != nonnull) { @@ -917,9 +980,9 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, if (pass_count == 0) { if (nonnull_count > 0) { int skipped = 0; - if (RET_FAIL(skip_value_typed(value_decoder_, values, - nonnull_count, skipped, - value_in))) { + if (RET_FAIL(skip_value_exact_typed(value_decoder_, values, + nonnull_count, skipped, + value_in))) { break; } if (skipped != nonnull_count) { @@ -933,9 +996,9 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, int value_count = 0; if (nonnull_count > 0) { - if (RET_FAIL(read_value_batch_typed(value_decoder_, values, - nonnull_count, value_count, - value_in))) { + if (RET_FAIL(read_value_batch_exact_typed(value_decoder_, values, + nonnull_count, + value_count, value_in))) { break; } } @@ -1719,7 +1782,8 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, } case common::INT32: case common::DATE: - if (RET_FAIL(col->decoder->read_batch_int32( + if (RET_FAIL(read_value_batch_exact_typed( + col->decoder, reinterpret_cast(pps.predecoded_values.data()), nonnull_total, actual, in))) { cleanup(); @@ -1728,7 +1792,8 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, break; case common::INT64: case common::TIMESTAMP: - if (RET_FAIL(col->decoder->read_batch_int64( + if (RET_FAIL(read_value_batch_exact_typed( + col->decoder, reinterpret_cast(pps.predecoded_values.data()), nonnull_total, actual, in))) { cleanup(); @@ -1736,7 +1801,8 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, } break; case common::FLOAT: - if (RET_FAIL(col->decoder->read_batch_float( + if (RET_FAIL(read_value_batch_exact_typed( + col->decoder, reinterpret_cast(pps.predecoded_values.data()), nonnull_total, actual, in))) { cleanup(); @@ -1744,7 +1810,8 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, } break; case common::DOUBLE: - if (RET_FAIL(col->decoder->read_batch_double( + if (RET_FAIL(read_value_batch_exact_typed( + col->decoder, reinterpret_cast(pps.predecoded_values.data()), nonnull_total, actual, in))) { cleanup(); @@ -1755,6 +1822,11 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, cleanup(); return E_NOT_SUPPORT; } + // Exact fixed-width reads must fill the bitmap-promised value buffer. + if (actual != nonnull_total) { + cleanup(); + return E_TSFILE_CORRUPTED; + } pps.predecoded_count = actual; cleanup(); return E_OK; @@ -2378,26 +2450,30 @@ int AlignedChunkReader::decompress_and_parse_value_page(ValueColumnState& col, } case common::INT32: case common::DATE: - rret = col.decoder->read_batch_int32( + rret = read_value_batch_exact_typed( + col.decoder, reinterpret_cast( col.pending_decoded_values.data()), nonnull_total, actual, col.in); break; case common::INT64: case common::TIMESTAMP: - rret = col.decoder->read_batch_int64( + rret = read_value_batch_exact_typed( + col.decoder, reinterpret_cast( col.pending_decoded_values.data()), nonnull_total, actual, col.in); break; case common::FLOAT: - rret = col.decoder->read_batch_float( + rret = read_value_batch_exact_typed( + col.decoder, reinterpret_cast( col.pending_decoded_values.data()), nonnull_total, actual, col.in); break; case common::DOUBLE: - rret = col.decoder->read_batch_double( + rret = read_value_batch_exact_typed( + col.decoder, reinterpret_cast( col.pending_decoded_values.data()), nonnull_total, actual, col.in); @@ -2524,11 +2600,9 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock* ret_tsblock, } } - // Skip values if no rows pass time filter. Skip/read errors and - // short reads (decoder returned fewer values than the bitmap - // promised) must abort; otherwise the input stream is left - // mid-value and later batches would decode garbage from - // misaligned bytes. + // Skip values if no rows pass time filter. The aligned bitmap + // supplies the exact count, including values equal to a codec's + // in-band terminator. if (pass_count == 0 && cb.nonnull_count > 0) { int dret = common::E_OK; int sk = 0; @@ -2543,21 +2617,25 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock* ret_tsblock, } case common::INT32: case common::DATE: - dret = col->decoder->skip_int32(cb.nonnull_count, sk, - col->in); + dret = skip_value_exact_typed( + col->decoder, static_cast(nullptr), + cb.nonnull_count, sk, col->in); break; case common::INT64: case common::TIMESTAMP: - dret = col->decoder->skip_int64(cb.nonnull_count, sk, - col->in); + dret = skip_value_exact_typed( + col->decoder, static_cast(nullptr), + cb.nonnull_count, sk, col->in); break; case common::FLOAT: - dret = col->decoder->skip_float(cb.nonnull_count, sk, - col->in); + dret = skip_value_exact_typed( + col->decoder, static_cast(nullptr), + cb.nonnull_count, sk, col->in); break; case common::DOUBLE: - dret = col->decoder->skip_double(cb.nonnull_count, sk, - col->in); + dret = skip_value_exact_typed( + col->decoder, static_cast(nullptr), + cb.nonnull_count, sk, col->in); break; case common::STRING: case common::TEXT: @@ -2621,23 +2699,27 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock* ret_tsblock, } case common::INT32: case common::DATE: - dret = col->decoder->read_batch_int32( + dret = read_value_batch_exact_typed( + col->decoder, reinterpret_cast(cb.val_buf), cb.nonnull_count, cb.val_count, col->in); break; case common::INT64: case common::TIMESTAMP: - dret = col->decoder->read_batch_int64( + dret = read_value_batch_exact_typed( + col->decoder, reinterpret_cast(cb.val_buf), cb.nonnull_count, cb.val_count, col->in); break; case common::FLOAT: - dret = col->decoder->read_batch_float( + dret = read_value_batch_exact_typed( + col->decoder, reinterpret_cast(cb.val_buf), cb.nonnull_count, cb.val_count, col->in); break; case common::DOUBLE: - dret = col->decoder->read_batch_double( + dret = read_value_batch_exact_typed( + col->decoder, reinterpret_cast(cb.val_buf), cb.nonnull_count, cb.val_count, col->in); break; diff --git a/cpp/test/reader/tsfile_reader_test.cc b/cpp/test/reader/tsfile_reader_test.cc index f7df9c8c9..524fe9ac7 100644 --- a/cpp/test/reader/tsfile_reader_test.cc +++ b/cpp/test/reader/tsfile_reader_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -1343,6 +1344,99 @@ TEST_F(TsFileReaderTest, MultiValueAlignedWideChunkParallelDecode) { io_reader.revert_ssi(ssi); } +// Regression: Gorilla also uses the canonical NaN bit pattern as its stream +// terminator, so its regular batch API stops when the same bit pattern appears +// as a value. An aligned page's bitmap supplies the exact non-null count; the +// parallel reader must use it to preserve the NaN and every following value. +TEST_F(TsFileReaderTest, MultiValueAlignedParallelPreservesDoubleGorillaNaN) { +#ifdef ENABLE_THREADS + struct ParallelReadGuard { + explicit ParallelReadGuard(bool saved) : saved_(saved) {} + ~ParallelReadGuard() { + common::g_config_value_.parallel_read_enabled_ = saved_; + } + bool saved_; + } guard(common::g_config_value_.parallel_read_enabled_); + common::g_config_value_.parallel_read_enabled_ = true; + ASSERT_NE(common::g_thread_pool_, nullptr); + + const std::string device = "root.dev_parallel_double_gorilla_nan"; + std::vector schema_vec; + schema_vec.emplace_back("v0", DOUBLE, GORILLA, UNCOMPRESSED); + schema_vec.emplace_back("v1", DOUBLE, GORILLA, UNCOMPRESSED); + { + std::vector registered; + for (const auto& schema : schema_vec) { + registered.push_back(new MeasurementSchema(schema)); + } + ASSERT_EQ( + tsfile_writer_->register_aligned_timeseries(device, registered), + E_OK); + } + + constexpr int kRowCount = 32; + constexpr int kNaNRow = 7; + Tablet tablet(device, + std::make_shared>(schema_vec), + kRowCount); + for (int row = 0; row < kRowCount; row++) { + ASSERT_EQ(tablet.add_timestamp(row, row), E_OK); + const double v0 = row == kNaNRow ? std::nan("") : 1000.0 + row; + ASSERT_EQ(tablet.add_value(row, 0u, v0), E_OK); + ASSERT_EQ(tablet.add_value(row, 1u, 2000.0 + row), E_OK); + } + ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->flush(), E_OK); + ASSERT_EQ(tsfile_writer_->close(), E_OK); + + storage::TsFileIOReader io_reader; + ASSERT_EQ(io_reader.init(file_name_), E_OK); + auto device_id = std::make_shared(device); + storage::TsFileSeriesScanIterator* ssi = nullptr; + common::PageArena pa; + pa.init(512, common::MOD_TSFILE_READER); + ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, {"v0", "v1"}, ssi, pa, + /*time_filter=*/nullptr), + E_OK); + ASSERT_NE(ssi, nullptr); + + common::TsBlock* block = nullptr; + const int read_ret = ssi->get_next(block, /*alloc_tsblock=*/true); + ASSERT_EQ(read_ret, E_OK); + ASSERT_NE(block, nullptr); + ASSERT_EQ(block->get_row_count(), kRowCount); + { + common::ColIterator time_iter(0, block); + common::ColIterator v0_iter(1, block); + common::ColIterator v1_iter(2, block); + for (int row = 0; row < kRowCount; row++) { + uint32_t len = 0; + const int64_t time = + *reinterpret_cast(time_iter.read(&len)); + const double v0 = *reinterpret_cast(v0_iter.read(&len)); + const double v1 = *reinterpret_cast(v1_iter.read(&len)); + EXPECT_EQ(time, row); + if (row == kNaNRow) { + EXPECT_TRUE(std::isnan(v0)); + } else { + EXPECT_DOUBLE_EQ(v0, 1000.0 + row); + } + EXPECT_DOUBLE_EQ(v1, 2000.0 + row); + time_iter.next(); + v0_iter.next(); + v1_iter.next(); + } + } + ssi->revert_tsblock(); + + block = nullptr; + EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), E_NO_MORE_DATA); + io_reader.revert_ssi(ssi); +#else + GTEST_SKIP() << "parallel reader support is disabled in this build"; +#endif +} + // Regression: AlignedTimeseriesIndex::get_data_type() returns the time column // type (VECTOR), which the schema accessor used to surface verbatim — every // aligned column came back as VECTOR instead of its real INT32/FLOAT/etc. From 1bdbcd857d058ffda7f2d73a26b212504701deb4 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 30 Jul 2026 16:35:11 +0800 Subject: [PATCH 2/2] fix(cpp): handle Gorilla sentinels across chunk readers --- cpp/src/encoding/decoder.h | 103 ++++++++++ cpp/src/reader/aligned_chunk_reader.cc | 238 ++++++++---------------- cpp/src/reader/chunk_reader.cc | 83 ++++----- cpp/test/encoding/gorilla_codec_test.cc | 42 +++++ cpp/test/reader/tsfile_reader_test.cc | 157 +++++++++------- 5 files changed, 353 insertions(+), 270 deletions(-) diff --git a/cpp/src/encoding/decoder.h b/cpp/src/encoding/decoder.h index 70b9cde93..035023f8c 100644 --- a/cpp/src/encoding/decoder.h +++ b/cpp/src/encoding/decoder.h @@ -155,6 +155,52 @@ class Decoder { return common::E_OK; } + // Some encodings use an otherwise valid value as an end marker. Gorilla, + // for example, uses canonical NaN for FLOAT/DOUBLE and the minimum value + // for INT32/INT64. The regular batch API must honor that marker because it + // only receives a capacity. Callers that know the exact value count from + // page metadata can use these methods to disambiguate an embedded marker + // from the physical end marker. + int read_exact_int32(int32_t* out, int count, common::ByteStream& in) { + return read_exact_impl(out, count, in, &Decoder::read_batch_int32, + &Decoder::read_int32); + } + + int read_exact_int64(int64_t* out, int count, common::ByteStream& in) { + return read_exact_impl(out, count, in, &Decoder::read_batch_int64, + &Decoder::read_int64); + } + + int read_exact_float(float* out, int count, common::ByteStream& in) { + return read_exact_impl(out, count, in, &Decoder::read_batch_float, + &Decoder::read_float); + } + + int read_exact_double(double* out, int count, common::ByteStream& in) { + return read_exact_impl(out, count, in, &Decoder::read_batch_double, + &Decoder::read_double); + } + + int skip_exact_int32(int count, common::ByteStream& in) { + return skip_exact_impl(count, in, &Decoder::skip_int32, + &Decoder::read_int32); + } + + int skip_exact_int64(int count, common::ByteStream& in) { + return skip_exact_impl(count, in, &Decoder::skip_int64, + &Decoder::read_int64); + } + + int skip_exact_float(int count, common::ByteStream& in) { + return skip_exact_impl(count, in, &Decoder::skip_float, + &Decoder::read_float); + } + + int skip_exact_double(int count, common::ByteStream& in) { + return skip_exact_impl(count, in, &Decoder::skip_double, + &Decoder::read_double); + } + // Block-level filter pushdown for TS_2DIFF-encoded INT64 columns. // // TS_2DIFF stores values in self-contained "blocks": a header (value @@ -188,6 +234,63 @@ class Decoder { virtual int skip_peeked_block_int64(common::ByteStream& in, int& skipped) { return common::E_NOT_SUPPORT; } + + private: + template + int read_exact_impl(T* out, int count, common::ByteStream& in, + int (Decoder::*read_batch)(T*, int, int&, + common::ByteStream&), + int (Decoder::*read_one)(T&, common::ByteStream&)) { + if (count < 0 || (count > 0 && out == nullptr)) { + return common::E_INVALID_ARG; + } + + int actual = 0; + while (actual < count) { + int batch_actual = 0; + int ret = (this->*read_batch)(out + actual, count - actual, + batch_actual, in); + if (ret != common::E_OK) return ret; + if (batch_actual < 0 || batch_actual > count - actual) { + return common::E_TSFILE_CORRUPTED; + } + actual += batch_actual; + if (actual == count) return common::E_OK; + + if ((ret = (this->*read_one)(out[actual], in)) != common::E_OK) { + return ret; + } + ++actual; + } + return common::E_OK; + } + + template + int skip_exact_impl(int count, common::ByteStream& in, + int (Decoder::*skip_batch)(int, int&, + common::ByteStream&), + int (Decoder::*read_one)(T&, common::ByteStream&)) { + if (count < 0) return common::E_INVALID_ARG; + + int skipped = 0; + while (skipped < count) { + int batch_skipped = 0; + int ret = (this->*skip_batch)(count - skipped, batch_skipped, in); + if (ret != common::E_OK) return ret; + if (batch_skipped < 0 || batch_skipped > count - skipped) { + return common::E_TSFILE_CORRUPTED; + } + skipped += batch_skipped; + if (skipped == count) return common::E_OK; + + T ignored; + if ((ret = (this->*read_one)(ignored, in)) != common::E_OK) { + return ret; + } + ++skipped; + } + return common::E_OK; + } }; } // end namespace storage diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index 1cb18751e..97c469288 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -784,102 +784,39 @@ int AlignedChunkReader::i32_DECODE_TYPED_TV_INTO_TSBLOCK( } namespace { -// Type-dispatched value batch read / skip for decode_tv_batch. Overload -// resolution on the value pointer type selects the matching Decoder method, so -// the four fixed-width value types share one decode loop. -FORCE_INLINE int read_value_batch_typed(Decoder* d, int32_t* out, int cap, - int& actual, ByteStream& in) { - return d->read_batch_int32(out, cap, actual, in); +// Type-dispatched exact read / skip for decode_tv_batch. Overload +// resolution on the value pointer selects the matching Decoder method. +FORCE_INLINE int read_value_exact_typed(Decoder* d, int32_t* out, int count, + ByteStream& in) { + return d->read_exact_int32(out, count, in); } -FORCE_INLINE int read_value_batch_typed(Decoder* d, int64_t* out, int cap, - int& actual, ByteStream& in) { - return d->read_batch_int64(out, cap, actual, in); +FORCE_INLINE int read_value_exact_typed(Decoder* d, int64_t* out, int count, + ByteStream& in) { + return d->read_exact_int64(out, count, in); } -FORCE_INLINE int read_value_batch_typed(Decoder* d, float* out, int cap, - int& actual, ByteStream& in) { - return d->read_batch_float(out, cap, actual, in); +FORCE_INLINE int read_value_exact_typed(Decoder* d, float* out, int count, + ByteStream& in) { + return d->read_exact_float(out, count, in); } -FORCE_INLINE int read_value_batch_typed(Decoder* d, double* out, int cap, - int& actual, ByteStream& in) { - return d->read_batch_double(out, cap, actual, in); +FORCE_INLINE int read_value_exact_typed(Decoder* d, double* out, int count, + ByteStream& in) { + return d->read_exact_double(out, count, in); } -FORCE_INLINE int read_value_typed(Decoder* d, int32_t& out, ByteStream& in) { - return d->read_int32(out, in); +FORCE_INLINE int skip_value_exact_typed(Decoder* d, int32_t*, int count, + ByteStream& in) { + return d->skip_exact_int32(count, in); } -FORCE_INLINE int read_value_typed(Decoder* d, int64_t& out, ByteStream& in) { - return d->read_int64(out, in); +FORCE_INLINE int skip_value_exact_typed(Decoder* d, int64_t*, int count, + ByteStream& in) { + return d->skip_exact_int64(count, in); } -FORCE_INLINE int read_value_typed(Decoder* d, float& out, ByteStream& in) { - return d->read_float(out, in); +FORCE_INLINE int skip_value_exact_typed(Decoder* d, float*, int count, + ByteStream& in) { + return d->skip_exact_float(count, in); } -FORCE_INLINE int read_value_typed(Decoder* d, double& out, ByteStream& in) { - return d->read_double(out, in); -} -FORCE_INLINE int skip_value_typed(Decoder* d, int32_t*, int n, int& skipped, - ByteStream& in) { - return d->skip_int32(n, skipped, in); -} -FORCE_INLINE int skip_value_typed(Decoder* d, int64_t*, int n, int& skipped, - ByteStream& in) { - return d->skip_int64(n, skipped, in); -} -FORCE_INLINE int skip_value_typed(Decoder* d, float*, int n, int& skipped, - ByteStream& in) { - return d->skip_float(n, skipped, in); -} -FORCE_INLINE int skip_value_typed(Decoder* d, double*, int n, int& skipped, - ByteStream& in) { - return d->skip_double(n, skipped, in); -} - -// Aligned pages carry an authoritative non-null count in their bitmap. Some -// encodings also use an otherwise valid fixed-width value as their stream -// terminator (for example, Gorilla uses canonical NaN). The regular batch -// API stops before returning that value. When the bitmap still promises a -// value, consume it through the scalar API and then resume the fast batch -// path. This preserves embedded sentinel values without changing the -// decoder's general streaming contract. -template -FORCE_INLINE int read_value_batch_exact_typed(Decoder* d, T* out, int expected, - int& actual, ByteStream& in) { - actual = 0; - while (actual < expected) { - int batch_actual = 0; - int ret = read_value_batch_typed(d, out + actual, expected - actual, - batch_actual, in); - if (ret != E_OK) return ret; - if (batch_actual < 0 || batch_actual > expected - actual) { - return E_TSFILE_CORRUPTED; - } - actual += batch_actual; - if (actual == expected) return E_OK; - - if ((ret = read_value_typed(d, out[actual], in)) != E_OK) return ret; - ++actual; - } - return E_OK; -} - -template -FORCE_INLINE int skip_value_exact_typed(Decoder* d, T* type_tag, int expected, - int& skipped, ByteStream& in) { - skipped = 0; - while (skipped < expected) { - int batch_skipped = 0; - int ret = skip_value_typed(d, type_tag, expected - skipped, - batch_skipped, in); - if (ret != E_OK) return ret; - if (batch_skipped < 0 || batch_skipped > expected - skipped) { - return E_TSFILE_CORRUPTED; - } - skipped += batch_skipped; - if (skipped == expected) return E_OK; - - T ignored; - if ((ret = read_value_typed(d, ignored, in)) != E_OK) return ret; - ++skipped; - } - return E_OK; +FORCE_INLINE int skip_value_exact_typed(Decoder* d, double*, int count, + ByteStream& in) { + return d->skip_exact_double(count, in); } } // namespace @@ -926,18 +863,8 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, } cur_value_index += block_count; if (nonnull > 0) { - // skip_* may legitimately fail (truncated page) or - // short-read (corrupt bitmap vs. data); both must abort - // the loop rather than silently desync the value - // decoder. - int sk = 0; - if (RET_FAIL(skip_value_exact_typed(value_decoder_, - values, nonnull, sk, - value_in))) { - break; - } - if (sk != nonnull) { - ret = E_TSFILE_CORRUPTED; + if (RET_FAIL(skip_value_exact_typed( + value_decoder_, values, nonnull, value_in))) { break; } } @@ -979,14 +906,8 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, if (pass_count == 0) { if (nonnull_count > 0) { - int skipped = 0; if (RET_FAIL(skip_value_exact_typed(value_decoder_, values, - nonnull_count, skipped, - value_in))) { - break; - } - if (skipped != nonnull_count) { - ret = E_TSFILE_CORRUPTED; + nonnull_count, value_in))) { break; } } @@ -994,11 +915,9 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, continue; } - int value_count = 0; if (nonnull_count > 0) { - if (RET_FAIL(read_value_batch_exact_typed(value_decoder_, values, - nonnull_count, - value_count, value_in))) { + if (RET_FAIL(read_value_exact_typed(value_decoder_, values, + nonnull_count, value_in))) { break; } } @@ -1767,7 +1686,6 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, uint32_t elem_size = common::get_data_type_size(dt); pps.predecoded_values.resize(static_cast(nonnull_total) * elem_size); - int actual = 0; switch (dt) { case common::BOOLEAN: { bool* out = reinterpret_cast(pps.predecoded_values.data()); @@ -1777,43 +1695,38 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, return ret; } } - actual = nonnull_total; break; } case common::INT32: case common::DATE: - if (RET_FAIL(read_value_batch_exact_typed( - col->decoder, + if (RET_FAIL(col->decoder->read_exact_int32( reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, actual, in))) { + nonnull_total, in))) { cleanup(); return ret; } break; case common::INT64: case common::TIMESTAMP: - if (RET_FAIL(read_value_batch_exact_typed( - col->decoder, + if (RET_FAIL(col->decoder->read_exact_int64( reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, actual, in))) { + nonnull_total, in))) { cleanup(); return ret; } break; case common::FLOAT: - if (RET_FAIL(read_value_batch_exact_typed( - col->decoder, + if (RET_FAIL(col->decoder->read_exact_float( reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, actual, in))) { + nonnull_total, in))) { cleanup(); return ret; } break; case common::DOUBLE: - if (RET_FAIL(read_value_batch_exact_typed( - col->decoder, + if (RET_FAIL(col->decoder->read_exact_double( reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, actual, in))) { + nonnull_total, in))) { cleanup(); return ret; } @@ -1822,12 +1735,7 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, cleanup(); return E_NOT_SUPPORT; } - // Exact fixed-width reads must fill the bitmap-promised value buffer. - if (actual != nonnull_total) { - cleanup(); - return E_TSFILE_CORRUPTED; - } - pps.predecoded_count = actual; + pps.predecoded_count = nonnull_total; cleanup(); return E_OK; } @@ -2444,39 +2352,39 @@ int AlignedChunkReader::decompress_and_parse_value_page(ValueColumnState& col, break; } out[i] = v; + actual++; } - actual = nonnull_total; break; } case common::INT32: case common::DATE: - rret = read_value_batch_exact_typed( - col.decoder, + rret = col.decoder->read_exact_int32( reinterpret_cast( col.pending_decoded_values.data()), - nonnull_total, actual, col.in); + nonnull_total, col.in); + if (rret == common::E_OK) actual = nonnull_total; break; case common::INT64: case common::TIMESTAMP: - rret = read_value_batch_exact_typed( - col.decoder, + rret = col.decoder->read_exact_int64( reinterpret_cast( col.pending_decoded_values.data()), - nonnull_total, actual, col.in); + nonnull_total, col.in); + if (rret == common::E_OK) actual = nonnull_total; break; case common::FLOAT: - rret = read_value_batch_exact_typed( - col.decoder, + rret = col.decoder->read_exact_float( reinterpret_cast( col.pending_decoded_values.data()), - nonnull_total, actual, col.in); + nonnull_total, col.in); + if (rret == common::E_OK) actual = nonnull_total; break; case common::DOUBLE: - rret = read_value_batch_exact_typed( - col.decoder, + rret = col.decoder->read_exact_double( reinterpret_cast( col.pending_decoded_values.data()), - nonnull_total, actual, col.in); + nonnull_total, col.in); + if (rret == common::E_OK) actual = nonnull_total; break; default: rret = common::E_OUT_OF_RANGE; @@ -2619,23 +2527,27 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock* ret_tsblock, case common::DATE: dret = skip_value_exact_typed( col->decoder, static_cast(nullptr), - cb.nonnull_count, sk, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) sk = cb.nonnull_count; break; case common::INT64: case common::TIMESTAMP: dret = skip_value_exact_typed( col->decoder, static_cast(nullptr), - cb.nonnull_count, sk, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) sk = cb.nonnull_count; break; case common::FLOAT: dret = skip_value_exact_typed( col->decoder, static_cast(nullptr), - cb.nonnull_count, sk, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) sk = cb.nonnull_count; break; case common::DOUBLE: dret = skip_value_exact_typed( col->decoder, static_cast(nullptr), - cb.nonnull_count, sk, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) sk = cb.nonnull_count; break; case common::STRING: case common::TEXT: @@ -2699,29 +2611,33 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock* ret_tsblock, } case common::INT32: case common::DATE: - dret = read_value_batch_exact_typed( - col->decoder, + dret = col->decoder->read_exact_int32( reinterpret_cast(cb.val_buf), - cb.nonnull_count, cb.val_count, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) + cb.val_count = cb.nonnull_count; break; case common::INT64: case common::TIMESTAMP: - dret = read_value_batch_exact_typed( - col->decoder, + dret = col->decoder->read_exact_int64( reinterpret_cast(cb.val_buf), - cb.nonnull_count, cb.val_count, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) + cb.val_count = cb.nonnull_count; break; case common::FLOAT: - dret = read_value_batch_exact_typed( - col->decoder, + dret = col->decoder->read_exact_float( reinterpret_cast(cb.val_buf), - cb.nonnull_count, cb.val_count, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) + cb.val_count = cb.nonnull_count; break; case common::DOUBLE: - dret = read_value_batch_exact_typed( - col->decoder, + dret = col->decoder->read_exact_double( reinterpret_cast(cb.val_buf), - cb.nonnull_count, cb.val_count, col->in); + cb.nonnull_count, col->in); + if (dret == common::E_OK) + cb.val_count = cb.nonnull_count; break; case common::STRING: case common::TEXT: diff --git a/cpp/src/reader/chunk_reader.cc b/cpp/src/reader/chunk_reader.cc index ce8d87a58..b21b1b4c5 100644 --- a/cpp/src/reader/chunk_reader.cc +++ b/cpp/src/reader/chunk_reader.cc @@ -479,7 +479,10 @@ int ChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in, ByteStream& value_in, if (!filter->satisfy_start_end_time(block_min, block_max)) { int skipped = 0; time_decoder_->skip_peeked_block_int64(time_in, skipped); - value_decoder_->skip_int32(block_count, skipped, value_in); + if (RET_FAIL(value_decoder_->skip_exact_int32(block_count, + value_in))) { + break; + } continue; } if (filter->contain_start_end_time(block_min, block_max)) { @@ -489,7 +492,6 @@ int ChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in, ByteStream& value_in, } int time_count = 0; - int value_count = 0; if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch, time_count, time_in))) { @@ -505,20 +507,15 @@ int ChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in, ByteStream& value_in, } if (pass_count == 0) { - int skipped = 0; - value_decoder_->skip_int32(time_count, skipped, value_in); + if (RET_FAIL( + value_decoder_->skip_exact_int32(time_count, value_in))) { + break; + } continue; } - if (RET_FAIL(value_decoder_->read_batch_int32(values, time_count, - value_count, value_in))) { - break; - } - // Time and value chunks are written in lock-step; any discrepancy - // means the file is truncated or corrupted. Reading uninitialised - // values[i] would silently surface garbage as decoded rows. - if (value_count != time_count) { - ret = E_TSFILE_CORRUPTED; + if (RET_FAIL(value_decoder_->read_exact_int32(values, time_count, + value_in))) { break; } @@ -568,7 +565,10 @@ int ChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in, ByteStream& value_in, if (!filter->satisfy_start_end_time(block_min, block_max)) { int skipped = 0; time_decoder_->skip_peeked_block_int64(time_in, skipped); - value_decoder_->skip_int64(block_count, skipped, value_in); + if (RET_FAIL(value_decoder_->skip_exact_int64(block_count, + value_in))) { + break; + } continue; } if (filter->contain_start_end_time(block_min, block_max)) { @@ -578,7 +578,6 @@ int ChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in, ByteStream& value_in, } int time_count = 0; - int value_count = 0; if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch, time_count, time_in))) { @@ -594,17 +593,15 @@ int ChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in, ByteStream& value_in, } if (pass_count == 0) { - int skipped = 0; - value_decoder_->skip_int64(time_count, skipped, value_in); + if (RET_FAIL( + value_decoder_->skip_exact_int64(time_count, value_in))) { + break; + } continue; } - if (RET_FAIL(value_decoder_->read_batch_int64(values, time_count, - value_count, value_in))) { - break; - } - if (value_count != time_count) { - ret = E_TSFILE_CORRUPTED; + if (RET_FAIL(value_decoder_->read_exact_int64(values, time_count, + value_in))) { break; } @@ -655,7 +652,10 @@ int ChunkReader::float_DECODE_TV_BATCH(ByteStream& time_in, if (!filter->satisfy_start_end_time(block_min, block_max)) { int skipped = 0; time_decoder_->skip_peeked_block_int64(time_in, skipped); - value_decoder_->skip_float(block_count, skipped, value_in); + if (RET_FAIL(value_decoder_->skip_exact_float(block_count, + value_in))) { + break; + } continue; } if (filter->contain_start_end_time(block_min, block_max)) { @@ -665,7 +665,6 @@ int ChunkReader::float_DECODE_TV_BATCH(ByteStream& time_in, } int time_count = 0; - int value_count = 0; if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch, time_count, time_in))) { @@ -681,17 +680,15 @@ int ChunkReader::float_DECODE_TV_BATCH(ByteStream& time_in, } if (pass_count == 0) { - int skipped = 0; - value_decoder_->skip_float(time_count, skipped, value_in); + if (RET_FAIL( + value_decoder_->skip_exact_float(time_count, value_in))) { + break; + } continue; } - if (RET_FAIL(value_decoder_->read_batch_float(values, time_count, - value_count, value_in))) { - break; - } - if (value_count != time_count) { - ret = E_TSFILE_CORRUPTED; + if (RET_FAIL(value_decoder_->read_exact_float(values, time_count, + value_in))) { break; } @@ -738,7 +735,10 @@ int ChunkReader::double_DECODE_TV_BATCH(ByteStream& time_in, if (!filter->satisfy_start_end_time(block_min, block_max)) { int skipped = 0; time_decoder_->skip_peeked_block_int64(time_in, skipped); - value_decoder_->skip_double(block_count, skipped, value_in); + if (RET_FAIL(value_decoder_->skip_exact_double(block_count, + value_in))) { + break; + } continue; } if (filter->contain_start_end_time(block_min, block_max)) { @@ -748,7 +748,6 @@ int ChunkReader::double_DECODE_TV_BATCH(ByteStream& time_in, } int time_count = 0; - int value_count = 0; if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch, time_count, time_in))) { @@ -764,17 +763,15 @@ int ChunkReader::double_DECODE_TV_BATCH(ByteStream& time_in, } if (pass_count == 0) { - int skipped = 0; - value_decoder_->skip_double(time_count, skipped, value_in); + if (RET_FAIL( + value_decoder_->skip_exact_double(time_count, value_in))) { + break; + } continue; } - if (RET_FAIL(value_decoder_->read_batch_double( - values, time_count, value_count, value_in))) { - break; - } - if (value_count != time_count) { - ret = E_TSFILE_CORRUPTED; + if (RET_FAIL(value_decoder_->read_exact_double(values, time_count, + value_in))) { break; } diff --git a/cpp/test/encoding/gorilla_codec_test.cc b/cpp/test/encoding/gorilla_codec_test.cc index b1fe72136..039a9a4f0 100644 --- a/cpp/test/encoding/gorilla_codec_test.cc +++ b/cpp/test/encoding/gorilla_codec_test.cc @@ -486,6 +486,48 @@ TEST_F(GorillaCodecTest, DoubleBatchScalarAndSkipInterleave) { EXPECT_EQ(cursor, N); } +TEST_F(GorillaCodecTest, DoubleExactReadAndSkipPreserveEmbeddedNaN) { + const double expected[] = {1.25, std::nan(""), 2.5, std::nan(""), 5.0}; + const int count = sizeof(expected) / sizeof(expected[0]); + + storage::DoubleGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + for (double value : expected) { + ASSERT_EQ(encoder.encode(value, stream), common::E_OK); + } + encoder.flush(stream); + + const uint32_t total = stream.total_size(); + std::vector buf(total); + uint32_t got = 0; + stream.read_buf(buf.data(), total, got); + ASSERT_EQ(got, total); + + common::ByteStream read_stream(common::MOD_DEFAULT); + read_stream.wrap_from(reinterpret_cast(buf.data()), total); + storage::DoubleGorillaDecoder read_decoder; + double actual[count]; + ASSERT_EQ(read_decoder.read_exact_double(actual, count, read_stream), + common::E_OK); + for (int i = 0; i < count; i++) { + if (std::isnan(expected[i])) { + EXPECT_TRUE(std::isnan(actual[i])) << "i=" << i; + } else { + EXPECT_DOUBLE_EQ(actual[i], expected[i]) << "i=" << i; + } + } + + common::ByteStream skip_stream(common::MOD_DEFAULT); + skip_stream.wrap_from(reinterpret_cast(buf.data()), total); + storage::DoubleGorillaDecoder skip_decoder; + ASSERT_EQ(skip_decoder.skip_exact_double(3, skip_stream), common::E_OK); + double tail[2]; + ASSERT_EQ(skip_decoder.read_exact_double(tail, 2, skip_stream), + common::E_OK); + EXPECT_TRUE(std::isnan(tail[0])); + EXPECT_DOUBLE_EQ(tail[1], 5.0); +} + TEST_F(GorillaCodecTest, DoubleBatchDecodeFullWidthXor) { const uint64_t patterns[] = { 0x0000000000000000ULL, 0xFFFFFFFFFFFFFFFFULL, 0x0123456789ABCDEFULL, diff --git a/cpp/test/reader/tsfile_reader_test.cc b/cpp/test/reader/tsfile_reader_test.cc index 524fe9ac7..cdb6d33cf 100644 --- a/cpp/test/reader/tsfile_reader_test.cc +++ b/cpp/test/reader/tsfile_reader_test.cc @@ -1344,97 +1344,122 @@ TEST_F(TsFileReaderTest, MultiValueAlignedWideChunkParallelDecode) { io_reader.revert_ssi(ssi); } -// Regression: Gorilla also uses the canonical NaN bit pattern as its stream -// terminator, so its regular batch API stops when the same bit pattern appears -// as a value. An aligned page's bitmap supplies the exact non-null count; the -// parallel reader must use it to preserve the NaN and every following value. -TEST_F(TsFileReaderTest, MultiValueAlignedParallelPreservesDoubleGorillaNaN) { -#ifdef ENABLE_THREADS - struct ParallelReadGuard { - explicit ParallelReadGuard(bool saved) : saved_(saved) {} - ~ParallelReadGuard() { - common::g_config_value_.parallel_read_enabled_ = saved_; - } - bool saved_; - } guard(common::g_config_value_.parallel_read_enabled_); - common::g_config_value_.parallel_read_enabled_ = true; - ASSERT_NE(common::g_thread_pool_, nullptr); - - const std::string device = "root.dev_parallel_double_gorilla_nan"; +// Regression: Gorilla uses the canonical NaN bit pattern as its stream +// terminator. When the same bit pattern appears as a value, count-aware reads +// must preserve it and every following value instead of stopping early. +TEST_F(TsFileReaderTest, GorillaDoubleNaNPreservedByChunkReaders) { + const std::string aligned_device = "root.dev_aligned_gorilla_nan"; + const std::string non_aligned_device = "root.dev_gorilla_nan"; std::vector schema_vec; schema_vec.emplace_back("v0", DOUBLE, GORILLA, UNCOMPRESSED); schema_vec.emplace_back("v1", DOUBLE, GORILLA, UNCOMPRESSED); + ASSERT_EQ( + tsfile_writer_->register_timeseries(non_aligned_device, schema_vec[0]), + E_OK); { std::vector registered; for (const auto& schema : schema_vec) { registered.push_back(new MeasurementSchema(schema)); } - ASSERT_EQ( - tsfile_writer_->register_aligned_timeseries(device, registered), - E_OK); + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(aligned_device, + registered), + E_OK); } constexpr int kRowCount = 32; constexpr int kNaNRow = 7; - Tablet tablet(device, - std::make_shared>(schema_vec), - kRowCount); + Tablet aligned_tablet( + aligned_device, + std::make_shared>(schema_vec), + kRowCount); + auto non_aligned_schema = + std::make_shared>(1, schema_vec[0]); + Tablet non_aligned_tablet(non_aligned_device, non_aligned_schema, + kRowCount); for (int row = 0; row < kRowCount; row++) { - ASSERT_EQ(tablet.add_timestamp(row, row), E_OK); + ASSERT_EQ(aligned_tablet.add_timestamp(row, row), E_OK); + ASSERT_EQ(non_aligned_tablet.add_timestamp(row, row), E_OK); const double v0 = row == kNaNRow ? std::nan("") : 1000.0 + row; - ASSERT_EQ(tablet.add_value(row, 0u, v0), E_OK); - ASSERT_EQ(tablet.add_value(row, 1u, 2000.0 + row), E_OK); + ASSERT_EQ(aligned_tablet.add_value(row, 0u, v0), E_OK); + ASSERT_EQ(aligned_tablet.add_value(row, 1u, 2000.0 + row), E_OK); + ASSERT_EQ(non_aligned_tablet.add_value(row, 0u, v0), E_OK); } - ASSERT_EQ(tsfile_writer_->write_tablet_aligned(tablet), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet(non_aligned_tablet), E_OK); + ASSERT_EQ(tsfile_writer_->write_tablet_aligned(aligned_tablet), E_OK); ASSERT_EQ(tsfile_writer_->flush(), E_OK); ASSERT_EQ(tsfile_writer_->close(), E_OK); storage::TsFileIOReader io_reader; ASSERT_EQ(io_reader.init(file_name_), E_OK); - auto device_id = std::make_shared(device); - storage::TsFileSeriesScanIterator* ssi = nullptr; - common::PageArena pa; - pa.init(512, common::MOD_TSFILE_READER); - ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, {"v0", "v1"}, ssi, pa, - /*time_filter=*/nullptr), - E_OK); - ASSERT_NE(ssi, nullptr); - common::TsBlock* block = nullptr; - const int read_ret = ssi->get_next(block, /*alloc_tsblock=*/true); - ASSERT_EQ(read_ret, E_OK); - ASSERT_NE(block, nullptr); - ASSERT_EQ(block->get_row_count(), kRowCount); - { - common::ColIterator time_iter(0, block); - common::ColIterator v0_iter(1, block); - common::ColIterator v1_iter(2, block); - for (int row = 0; row < kRowCount; row++) { - uint32_t len = 0; - const int64_t time = - *reinterpret_cast(time_iter.read(&len)); - const double v0 = *reinterpret_cast(v0_iter.read(&len)); - const double v1 = *reinterpret_cast(v1_iter.read(&len)); - EXPECT_EQ(time, row); - if (row == kNaNRow) { - EXPECT_TRUE(std::isnan(v0)); - } else { - EXPECT_DOUBLE_EQ(v0, 1000.0 + row); + auto check_values = [&](storage::TsFileSeriesScanIterator* ssi, + bool has_v1) { + common::TsBlock* block = nullptr; + ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), E_OK); + ASSERT_NE(block, nullptr); + ASSERT_EQ(block->get_row_count(), kRowCount); + + { + common::ColIterator time_iter(0, block); + common::ColIterator v0_iter(1, block); + std::unique_ptr v1_iter; + if (has_v1) { + v1_iter.reset(new common::ColIterator(2, block)); + } + for (int row = 0; row < kRowCount; row++) { + uint32_t len = 0; + const int64_t time = + *reinterpret_cast(time_iter.read(&len)); + const double v0 = + *reinterpret_cast(v0_iter.read(&len)); + EXPECT_EQ(time, row); + if (row == kNaNRow) { + EXPECT_TRUE(std::isnan(v0)); + } else { + EXPECT_DOUBLE_EQ(v0, 1000.0 + row); + } + if (has_v1) { + const double v1 = + *reinterpret_cast(v1_iter->read(&len)); + EXPECT_DOUBLE_EQ(v1, 2000.0 + row); + v1_iter->next(); + } + time_iter.next(); + v0_iter.next(); } - EXPECT_DOUBLE_EQ(v1, 2000.0 + row); - time_iter.next(); - v0_iter.next(); - v1_iter.next(); } + ssi->revert_tsblock(); + + block = nullptr; + EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), E_NO_MORE_DATA); + }; + + { + storage::TsFileSeriesScanIterator* ssi = nullptr; + common::PageArena pa; + pa.init(512, common::MOD_TSFILE_READER); + auto device_id = + std::make_shared(non_aligned_device); + ASSERT_EQ(io_reader.alloc_ssi(device_id, "v0", ssi, pa, + /*time_filter=*/nullptr), + E_OK); + ASSERT_NE(ssi, nullptr); + check_values(ssi, /*has_v1=*/false); + io_reader.revert_ssi(ssi); } - ssi->revert_tsblock(); - block = nullptr; - EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), E_NO_MORE_DATA); - io_reader.revert_ssi(ssi); -#else - GTEST_SKIP() << "parallel reader support is disabled in this build"; -#endif + { + storage::TsFileSeriesScanIterator* ssi = nullptr; + common::PageArena pa; + pa.init(512, common::MOD_TSFILE_READER); + auto device_id = std::make_shared(aligned_device); + ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, {"v0", "v1"}, ssi, pa, + /*time_filter=*/nullptr), + E_OK); + ASSERT_NE(ssi, nullptr); + check_values(ssi, /*has_v1=*/true); + io_reader.revert_ssi(ssi); + } } // Regression: AlignedTimeseriesIndex::get_data_type() returns the time column