This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 585125b12 Optimize C++ batch decode and output fast paths (#898)
585125b12 is described below
commit 585125b12621251410b86c916dc292cfda24d402
Author: Colin Lee <[email protected]>
AuthorDate: Thu Aug 6 10:54:12 2026 +0800
Optimize C++ batch decode and output fast paths (#898)
* Optimize C++ batch decode fast paths
* Fix reader test iterator lifetime
---
cpp/src/common/tsblock/tsblock.h | 22 ++++
.../common/tsblock/vector/fixed_length_vector.h | 7 +
cpp/src/encoding/gorilla_decoder.h | 68 +++++++++-
cpp/src/encoding/ts2diff_decoder.h | 37 ++++++
cpp/src/reader/aligned_chunk_reader.cc | 24 ++++
cpp/test/common/tsblock/tslock_test.cc | 37 ++++++
cpp/test/encoding/encoding_coverage_test.cc | 35 +++++
cpp/test/encoding/gorilla_codec_test.cc | 143 +++++++++++++++++++++
cpp/test/reader/tsfile_reader_test.cc | 92 +++++++++++++
9 files changed, 464 insertions(+), 1 deletion(-)
diff --git a/cpp/src/common/tsblock/tsblock.h b/cpp/src/common/tsblock/tsblock.h
index 7959f7c36..056186211 100644
--- a/cpp/src/common/tsblock/tsblock.h
+++ b/cpp/src/common/tsblock/tsblock.h
@@ -174,6 +174,28 @@ class RowAppender {
}
}
+ FORCE_INLINE bool can_bulk_append_fixed(uint32_t slot_index,
+ uint32_t elem_size) const {
+ ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count());
+ Vector* vec = tsblock_->vectors_[slot_index];
+ TSDataType datatype = vec->get_vector_type();
+ if (datatype == STRING || datatype == TEXT || datatype == BLOB) {
+ return false;
+ }
+ return static_cast<FixedLengthVector*>(vec)->get_type_len() ==
+ elem_size;
+ }
+
+ FORCE_INLINE void bulk_append_fixed(uint32_t slot_index, const char*
values,
+ uint32_t count) {
+ ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count());
+ Vector* vec = tsblock_->vectors_[slot_index];
+ ASSERT(vec->get_vector_type() != STRING &&
+ vec->get_vector_type() != TEXT &&
+ vec->get_vector_type() != BLOB);
+ static_cast<FixedLengthVector*>(vec)->append_batch(values, count);
+ }
+
FORCE_INLINE void append_null(uint32_t slot_index) {
Vector* vec = tsblock_->vectors_[slot_index];
vec->set_null(tsblock_->row_count_ - 1);
diff --git a/cpp/src/common/tsblock/vector/fixed_length_vector.h
b/cpp/src/common/tsblock/vector/fixed_length_vector.h
index df20c97d9..eb5596343 100644
--- a/cpp/src/common/tsblock/vector/fixed_length_vector.h
+++ b/cpp/src/common/tsblock/vector/fixed_length_vector.h
@@ -49,6 +49,13 @@ class FixedLengthVector : public Vector {
values_.append_fixed_value(value, len);
}
+ FORCE_INLINE void append_batch(const char* values, uint32_t count) {
+ values_.append_fixed_value(values, count * type_len_);
+ add_row_nums(count);
+ }
+
+ FORCE_INLINE uint32_t get_type_len() const { return type_len_; }
+
// cppcheck-suppress missingOverride
FORCE_INLINE char* read(uint32_t* __restrict len, bool* __restrict null,
uint32_t rowid) OVERRIDE {
diff --git a/cpp/src/encoding/gorilla_decoder.h
b/cpp/src/encoding/gorilla_decoder.h
index c8c76d965..ccbbdef88 100644
--- a/cpp/src/encoding/gorilla_decoder.h
+++ b/cpp/src/encoding/gorilla_decoder.h
@@ -19,8 +19,13 @@
#ifndef ENCODING_GORILLA_DECODER_H
#define ENCODING_GORILLA_DECODER_H
+#include <algorithm>
#include <climits>
+#if defined(_MSC_VER)
+#include <intrin.h>
+#endif
+
#include "common/allocator/byte_stream.h"
#include "decoder.h"
#include "encode_utils.h"
@@ -30,6 +35,23 @@
namespace storage {
+FORCE_INLINE int gorilla_count_leading_zeros_nonzero(uint64_t value) {
+#if defined(__GNUC__) || defined(__clang__)
+ return __builtin_clzll(value);
+#elif defined(_MSC_VER)
+ unsigned long index;
+ _BitScanReverse64(&index, value);
+ return 63 - static_cast<int>(index);
+#else
+ int count = 0;
+ while ((value & (UINT64_C(1) << 63)) == 0) {
+ value <<= 1;
+ ++count;
+ }
+ return count;
+#endif
+}
+
// ── Raw-pointer bit reader ────────────────────────────────────────────────
// Operates directly on a contiguous byte array, bypassing ByteStream's
// per-byte read_buf() overhead (atomic loads, page boundary checks, memcpy).
@@ -91,6 +113,37 @@ struct GorillaBitReader {
return bit;
}
+ // Consume up to max_count consecutive zero control bits. The first one bit
+ // remains unread so the normal control decoder can handle the following
+ // changed value. This turns long repeated-value runs into one leading-zero
+ // count per reservoir instead of one read_next() call per value.
+ FORCE_INLINE int consume_zero_bits(int max_count) {
+ int consumed = 0;
+ while (consumed < max_count) {
+ if (UNLIKELY(!refill_if_empty())) {
+ break;
+ }
+
+ const int available_bits = bits;
+ const uint64_t aligned =
+ available_bits == 64 ? buffer : buffer << (64 -
available_bits);
+ const int zero_bits =
+ aligned == 0 ? available_bits
+ : gorilla_count_leading_zeros_nonzero(aligned);
+ const int remaining = max_count - consumed;
+ const int take = std::min(zero_bits, remaining);
+ bits -= take;
+ consumed += take;
+
+ // A one bit follows the consumed zeros. Leave it in the reservoir
+ // for read_control_bits(), or stop once the requested limit is
met.
+ if (zero_bits < available_bits || consumed == max_count) {
+ break;
+ }
+ }
+ return consumed;
+ }
+
FORCE_INLINE uint64_t read_long(int n) {
if (UNLIKELY(n < 0 || n > 64)) {
invalid = true;
@@ -460,8 +513,21 @@ class GorillaDecoder : public Decoder {
// Main batch loop
while (actual < capacity && has_next_) {
- out[actual++] =
+ const Output decoded =
GorillaDecodeOutput<T, Output>::convert(stored_value_);
+ const int repeated = r.consume_zero_bits(capacity - actual - 1);
+ const int run_length = repeated + 1;
+ // Include the current value in the bulk fill. Besides removing a
+ // scalar store and a second position update for every run, this
+ // gives the compiler one contiguous range to vectorize. Integer
+ // and floating-point outputs are copied without arithmetic, so
+ // special IEEE-754 bit patterns remain unchanged.
+ std::fill_n(out + actual, run_length, decoded);
+ actual += run_length;
+
+ // Prime the next value even when the repeated run exactly fills
the
+ // caller's buffer. This preserves the scalar decoder invariant
that
+ // stored_value_ is the next value to return on the following call.
if (UNLIKELY(!GorillaRawOps<T>::read_next(
r, stored_value_, stored_leading_zeros_,
stored_trailing_zeros_))) {
diff --git a/cpp/src/encoding/ts2diff_decoder.h
b/cpp/src/encoding/ts2diff_decoder.h
index 224d7402e..206b7f559 100644
--- a/cpp/src/encoding/ts2diff_decoder.h
+++ b/cpp/src/encoding/ts2diff_decoder.h
@@ -648,6 +648,43 @@ inline int
TS2DIFFDecoder<int64_t>::read_batch_int64(int64_t* out, int capacity,
int64_t prev = first_value_;
int32_t i = 0;
+ // An evenly spaced timestamp block has no packed residual data. Build
+ // the arithmetic progression directly instead of entering the generic
+ // bit-extraction path (whose SIMD guard requires readable input
bytes).
+ if (bit_width_ == 0) {
+#ifdef ENABLE_SIMD
+ if (remaining >= 4) {
+ int64_t value1 = prev + delta_min_;
+ int64_t value2 = value1 + delta_min_;
+ int64_t value3 = value2 + delta_min_;
+ int64_t value4 = value3 + delta_min_;
+ simde__m256i values =
+ simde_mm256_set_epi64x(value4, value3, value2, value1);
+
+ simde__m256i step = simde_mm256_set1_epi64x(delta_min_);
+ step = simde_mm256_add_epi64(step, step);
+ step = simde_mm256_add_epi64(step, step);
+
+ for (; i + 3 < remaining; i += 4) {
+ simde_mm256_storeu_si256(
+ reinterpret_cast<simde__m256i*>(out + actual), values);
+ actual += 4;
+ values = simde_mm256_add_epi64(values, step);
+ }
+ prev = out[actual - 1];
+ }
+#endif
+
+ for (; i < remaining; ++i) {
+ prev += delta_min_;
+ out[actual++] = prev;
+ }
+
+ first_value_ = prev;
+ current_index_ = 0;
+ continue;
+ }
+
#ifdef ENABLE_SIMD
// SIMD path: decode 4 INT64 values at a time
for (; i + 3 < remaining; i += 4) {
diff --git a/cpp/src/reader/aligned_chunk_reader.cc
b/cpp/src/reader/aligned_chunk_reader.cc
index 97c469288..a140c1e98 100644
--- a/cpp/src/reader/aligned_chunk_reader.cc
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -922,6 +922,30 @@ int AlignedChunkReader::decode_tv_batch(ByteStream&
time_in,
}
}
+ // Dense fixed-width batches are already laid out exactly as the two
+ // destination vectors expect. Appending them row by row would issue
+ // two tiny memcpy calls per row (time + value), plus virtual dispatch
+ // and bookkeeping. Copy each column once instead. Integral value
+ // filters still need the scalar satisfy(time, value) check unless the
+ // decoder proved the whole block passes, so those batches retain the
+ // fallback below.
+ const bool needs_integral_value_filter =
+ std::is_integral<T>::value && filter != nullptr && !block_all_pass;
+ if (pass_count == time_count && nonnull_count == time_count &&
+ !needs_integral_value_filter &&
+ row_appender.can_bulk_append_fixed(0, sizeof(int64_t)) &&
+ row_appender.can_bulk_append_fixed(1, sizeof(T))) {
+ row_appender.bulk_append_fixed(0,
+ reinterpret_cast<const
char*>(times),
+ static_cast<uint32_t>(time_count));
+ row_appender.bulk_append_fixed(
+ 1, reinterpret_cast<const char*>(values),
+ static_cast<uint32_t>(time_count));
+ row_appender.add_rows(static_cast<uint32_t>(time_count));
+ cur_value_index += time_count;
+ continue;
+ }
+
int val_idx = 0;
for (int i = 0; i < time_count; ++i) {
cur_value_index++;
diff --git a/cpp/test/common/tsblock/tslock_test.cc
b/cpp/test/common/tsblock/tslock_test.cc
index 750585aaf..c58b368f5 100644
--- a/cpp/test/common/tsblock/tslock_test.cc
+++ b/cpp/test/common/tsblock/tslock_test.cc
@@ -63,6 +63,43 @@ TEST(TsBlockTest, ColAppender_AddRowAndAppend) {
EXPECT_EQ(col_appender.get_col_row_count(), 50);
}
+TEST(TsBlockTest, RowAppenderBulkAppendFixedAtExactCapacity) {
+ TupleDesc tuple_desc;
+ tuple_desc.push_back(ColumnSchema("time", TIMESTAMP, UNCOMPRESSED, PLAIN));
+ tuple_desc.push_back(ColumnSchema("value", FLOAT, UNCOMPRESSED, PLAIN));
+ TsBlock ts_block(&tuple_desc, 4);
+ ASSERT_EQ(ts_block.init(), E_OK);
+ RowAppender row_appender(&ts_block);
+
+ const int64_t times[] = {101, 103, 107, 109};
+ const float values[] = {1.25f, 2.5f, 3.75f, 5.0f};
+ ASSERT_TRUE(row_appender.can_bulk_append_fixed(0, sizeof(int64_t)));
+ ASSERT_TRUE(row_appender.can_bulk_append_fixed(1, sizeof(float)));
+ EXPECT_FALSE(row_appender.can_bulk_append_fixed(1, sizeof(double)));
+
+ row_appender.bulk_append_fixed(0, reinterpret_cast<const char*>(times), 4);
+ row_appender.bulk_append_fixed(1, reinterpret_cast<const char*>(values),
4);
+ row_appender.add_rows(4);
+
+ EXPECT_EQ(ts_block.get_row_count(), 4u);
+ EXPECT_EQ(row_appender.remaining(), 0u);
+ EXPECT_EQ(ts_block.get_vector(0)->get_row_num(), 4u);
+ EXPECT_EQ(ts_block.get_vector(1)->get_row_num(), 4u);
+
+ ColIterator time_iter(0, &ts_block);
+ ColIterator value_iter(1, &ts_block);
+ for (uint32_t i = 0; i < 4; ++i) {
+ uint32_t len = 0;
+ EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), times[i]);
+ EXPECT_EQ(len, sizeof(int64_t));
+ EXPECT_FLOAT_EQ(*reinterpret_cast<float*>(value_iter.read(&len)),
+ values[i]);
+ EXPECT_EQ(len, sizeof(float));
+ time_iter.next();
+ value_iter.next();
+ }
+}
+
TEST(TsBlockTest, RowIterator_ReadAndNext) {
TupleDesc tuple_desc;
ColumnSchema col1("test_col1", INT32, SNAPPY, RLE);
diff --git a/cpp/test/encoding/encoding_coverage_test.cc
b/cpp/test/encoding/encoding_coverage_test.cc
index 6970b9387..5d6ef813e 100644
--- a/cpp/test/encoding/encoding_coverage_test.cc
+++ b/cpp/test/encoding/encoding_coverage_test.cc
@@ -330,6 +330,41 @@ TEST(EncodingCoverage, TS2DIFFBatchInt64MultipleBlocks) {
for (int i = 0; i < N; i++) EXPECT_EQ(out[i], values[i]) << "i=" << i;
}
+TEST(EncodingCoverage, TS2DIFFBatchInt64EvenlySpacedNegativeDelta) {
+ TS2DIFFEncoder<int64_t> enc;
+ common::ByteStream s(8192, common::MOD_DEFAULT);
+ // Full encoder blocks contain 127 residuals, so this covers repeated SIMD
+ // groups, the 3-value scalar tail, and a final partial block.
+ const int N = 389;
+ std::vector<int64_t> values(N);
+ for (int i = 0; i < N; i++) {
+ values[i] = INT64_C(9000000000000) - static_cast<int64_t>(i) * 29;
+ ASSERT_EQ(enc.encode(values[i], s), common::E_OK);
+ }
+ ASSERT_EQ(enc.flush(s), common::E_OK);
+
+ uint32_t total = s.total_size();
+ std::vector<uint8_t> buf(total);
+ uint32_t got = 0;
+ s.read_buf(buf.data(), total, got);
+ common::ByteStream wrapped(common::MOD_DEFAULT);
+ wrapped.wrap_from((const char*)buf.data(), total);
+
+ TS2DIFFDecoder<int64_t> dec;
+ std::vector<int64_t> out(N);
+ int total_decoded = 0;
+ while (dec.has_remaining(wrapped) && total_decoded < N) {
+ int actual = 0;
+ ASSERT_EQ(dec.read_batch_int64(out.data() + total_decoded,
+ N - total_decoded, actual, wrapped),
+ common::E_OK);
+ if (actual == 0) break;
+ total_decoded += actual;
+ }
+ EXPECT_EQ(total_decoded, N);
+ for (int i = 0; i < N; i++) EXPECT_EQ(out[i], values[i]) << "i=" << i;
+}
+
// ── Plain encoder: encode_batch fast paths for each type ───────────────
TEST(EncodingCoverage, PlainEncoderBatchAllTypes) {
PlainEncoder enc;
diff --git a/cpp/test/encoding/gorilla_codec_test.cc
b/cpp/test/encoding/gorilla_codec_test.cc
index 039a9a4f0..ccf0ae1da 100644
--- a/cpp/test/encoding/gorilla_codec_test.cc
+++ b/cpp/test/encoding/gorilla_codec_test.cc
@@ -324,6 +324,108 @@ TEST_F(GorillaCodecTest, FloatBatchDecode) {
}
}
+TEST_F(GorillaCodecTest, FloatBatchDecodeLongRepeatedRuns) {
+ storage::FloatGorillaEncoder encoder;
+ common::ByteStream stream(1024, common::MOD_DEFAULT);
+ const int N = 600;
+ std::vector<float> expected(N);
+ for (int i = 0; i < N; i++) {
+ if (i < 258) {
+ // Two complete 129-row batches. The repeated run crosses several
+ // 64-bit reservoirs and ends exactly at a batch boundary.
+ expected[i] = 17.25f;
+ } else if (i < 517) {
+ expected[i] = -3.5f;
+ } else {
+ expected[i] = static_cast<float>(i - 517) * 0.125f;
+ }
+ ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder.flush(stream), common::E_OK);
+
+ const uint32_t total = stream.total_size();
+ std::vector<uint8_t> encoded(total);
+ uint32_t got = 0;
+ stream.read_buf(encoded.data(), total, got);
+ ASSERT_EQ(got, total);
+
+ common::ByteStream wrapped(common::MOD_DEFAULT);
+ wrapped.wrap_from(reinterpret_cast<const char*>(encoded.data()), total);
+ storage::FloatGorillaDecoder decoder;
+ std::vector<float> decoded(N);
+ int total_decoded = 0;
+ while (total_decoded < N) {
+ const int capacity = std::min(129, N - total_decoded);
+ int actual = 0;
+ ASSERT_EQ(decoder.read_batch_float(decoded.data() + total_decoded,
+ capacity, actual, wrapped),
+ common::E_OK);
+ ASSERT_EQ(actual, capacity);
+ total_decoded += actual;
+ }
+
+ for (int i = 0; i < N; i++) {
+ EXPECT_EQ(common::float_to_int(decoded[i]),
+ common::float_to_int(expected[i]))
+ << "i=" << i;
+ }
+}
+
+TEST_F(GorillaCodecTest, FloatBatchRepeatedFillPreservesBitPatterns) {
+ constexpr uint32_t NAN_BITS = 0x7FC12345U;
+ float payload_nan = 0;
+ std::memcpy(&payload_nan, &NAN_BITS, sizeof(payload_nan));
+ ASSERT_TRUE(std::isnan(payload_nan));
+ ASSERT_NE(common::float_to_int(payload_nan),
+ common::float_to_int(GORILLA_ENCODING_ENDING_FLOAT));
+
+ const int N = 520;
+ std::vector<float> expected(N);
+ for (int i = 0; i < N; i++) {
+ if (i < 257) {
+ expected[i] = payload_nan;
+ } else if (i < 514) {
+ expected[i] = -0.0f;
+ } else {
+ expected[i] = static_cast<float>(i - 514) + 0.25f;
+ }
+ }
+
+ storage::FloatGorillaEncoder encoder;
+ common::ByteStream stream(1024, common::MOD_DEFAULT);
+ for (float value : expected) {
+ ASSERT_EQ(encoder.encode(value, stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder.flush(stream), common::E_OK);
+
+ const uint32_t total = stream.total_size();
+ std::vector<uint8_t> encoded(total);
+ uint32_t got = 0;
+ stream.read_buf(encoded.data(), total, got);
+ ASSERT_EQ(got, total);
+
+ common::ByteStream wrapped(common::MOD_DEFAULT);
+ wrapped.wrap_from(reinterpret_cast<const char*>(encoded.data()), total);
+ storage::FloatGorillaDecoder decoder;
+ std::vector<float> decoded(N);
+ int total_decoded = 0;
+ while (total_decoded < N) {
+ const int capacity = std::min(129, N - total_decoded);
+ int actual = 0;
+ ASSERT_EQ(decoder.read_batch_float(decoded.data() + total_decoded,
+ capacity, actual, wrapped),
+ common::E_OK);
+ ASSERT_EQ(actual, capacity);
+ total_decoded += actual;
+ }
+
+ for (int i = 0; i < N; i++) {
+ EXPECT_EQ(common::float_to_int(decoded[i]),
+ common::float_to_int(expected[i]))
+ << "i=" << i;
+ }
+}
+
TEST_F(GorillaCodecTest, FloatBatchDecodeUnwrappedInput) {
storage::FloatGorillaEncoder encoder;
common::ByteStream stream(1024, common::MOD_DEFAULT);
@@ -418,6 +520,47 @@ TEST_F(GorillaCodecTest, DoubleBatchDecodeOneValueAtATime)
{
}
}
+TEST_F(GorillaCodecTest, Int64BatchDecodeLongRepeatedRuns) {
+ storage::LongGorillaEncoder encoder;
+ common::ByteStream stream(1024, common::MOD_DEFAULT);
+ const int N = 513;
+ std::vector<int64_t> expected(N);
+ for (int i = 0; i < N; i++) {
+ if (i < 333) {
+ expected[i] = 0x123456789ABCDEFLL;
+ } else if (i < 500) {
+ expected[i] = -9876543210LL;
+ } else {
+ expected[i] = static_cast<int64_t>(i) * 17 - 9;
+ }
+ ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder.flush(stream), common::E_OK);
+
+ const uint32_t total = stream.total_size();
+ std::vector<uint8_t> encoded(total);
+ uint32_t got = 0;
+ stream.read_buf(encoded.data(), total, got);
+ ASSERT_EQ(got, total);
+
+ common::ByteStream wrapped(common::MOD_DEFAULT);
+ wrapped.wrap_from(reinterpret_cast<const char*>(encoded.data()), total);
+ storage::LongGorillaDecoder decoder;
+ std::vector<int64_t> decoded(N);
+ int total_decoded = 0;
+ while (total_decoded < N) {
+ const int capacity = std::min(73, N - total_decoded);
+ int actual = 0;
+ ASSERT_EQ(decoder.read_batch_int64(decoded.data() + total_decoded,
+ capacity, actual, wrapped),
+ common::E_OK);
+ ASSERT_EQ(actual, capacity);
+ total_decoded += actual;
+ }
+
+ EXPECT_EQ(decoded, expected);
+}
+
TEST_F(GorillaCodecTest, DoubleBatchScalarAndSkipInterleave) {
storage::DoubleGorillaEncoder encoder;
common::ByteStream stream(1024, common::MOD_DEFAULT);
diff --git a/cpp/test/reader/tsfile_reader_test.cc
b/cpp/test/reader/tsfile_reader_test.cc
index cdb6d33cf..df4c642a8 100644
--- a/cpp/test/reader/tsfile_reader_test.cc
+++ b/cpp/test/reader/tsfile_reader_test.cc
@@ -23,6 +23,7 @@
#include <cmath>
#include <map>
+#include <numeric>
#include <random>
#include <unordered_map>
#include <vector>
@@ -1511,6 +1512,97 @@ TEST_F(TsFileReaderTest,
AlignedSchemaReportsValueDataType) {
reader.close();
}
+TEST_F(TsFileReaderTest,
+ AlignedFloatBatchCopyPreservesDenseNullAndFilteredAlignment) {
+ const std::string device = "root.dev_aligned_float_batch";
+ MeasurementSchema schema("v0", FLOAT, GORILLA, UNCOMPRESSED);
+ ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, schema),
+ E_OK);
+
+ const int row_count = 300;
+ const int null_row = 150;
+ auto schemas = std::make_shared<std::vector<MeasurementSchema>>(1, schema);
+ Tablet tablet(device, schemas, row_count);
+ for (int row = 0; row < row_count; ++row) {
+ ASSERT_EQ(tablet.add_timestamp(row, 10000 + row * 3), E_OK);
+ if (row != null_row) {
+ ASSERT_EQ(tablet.add_value(row, 0u, row + 0.25f), 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<StringArrayDeviceID>(device);
+
+ auto scan_and_check = [&](storage::Filter* filter,
+ const std::vector<int>& expected_rows) {
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ ASSERT_EQ(io_reader.alloc_ssi(device_id, "v0", ssi, pa, filter), E_OK);
+ ASSERT_NE(ssi, nullptr);
+
+ size_t expected_index = 0;
+ while (true) {
+ common::TsBlock* block = nullptr;
+ int ret = ssi->get_next(block, /*alloc_tsblock=*/true, filter);
+ if (ret == E_NO_MORE_DATA) break;
+ ASSERT_EQ(ret, E_OK);
+ ASSERT_NE(block, nullptr);
+
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator value_iter(1, block);
+ while (!time_iter.end()) {
+ ASSERT_LT(expected_index, expected_rows.size());
+ const int expected_row = expected_rows[expected_index++];
+ uint32_t len = 0;
+ bool is_null = false;
+
EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)),
+ 10000 + expected_row * 3);
+ EXPECT_EQ(len, sizeof(int64_t));
+ char* value = value_iter.read(&len, &is_null);
+ if (expected_row == null_row) {
+ EXPECT_TRUE(is_null);
+ EXPECT_EQ(value, nullptr);
+ } else {
+ ASSERT_FALSE(is_null);
+ ASSERT_NE(value, nullptr);
+ EXPECT_EQ(len, sizeof(float));
+ EXPECT_FLOAT_EQ(*reinterpret_cast<float*>(value),
+ expected_row + 0.25f);
+ }
+ time_iter.next();
+ value_iter.next();
+ }
+ }
+ ssi->revert_tsblock();
+ }
+ EXPECT_EQ(expected_index, expected_rows.size());
+ io_reader.revert_ssi(ssi);
+ };
+
+ std::vector<int> all_rows(row_count);
+ std::iota(all_rows.begin(), all_rows.end(), 0);
+ scan_and_check(/*filter=*/nullptr, all_rows);
+
+ std::vector<int64_t> selected_times;
+ std::vector<int> selected_rows;
+ for (int row = 0; row < row_count; row += 11) {
+ selected_rows.push_back(row);
+ selected_times.push_back(10000 + row * 3);
+ }
+ selected_rows.push_back(null_row);
+ selected_times.push_back(10000 + null_row * 3);
+ std::sort(selected_rows.begin(), selected_rows.end());
+ std::sort(selected_times.begin(), selected_times.end());
+ storage::TimeIn time_filter(selected_times, /*not_in=*/false);
+ scan_and_check(&time_filter, selected_rows);
+}
+
namespace storage {
class TsFileReaderMetaArenaTest {
public:
