From 7effad7113bc1b01bf3b8abbcdda7ca21498afb5 Mon Sep 17 00:00:00 2001 From: gx Date: Fri, 7 Aug 2026 17:27:51 +0800 Subject: [PATCH 1/4] fix(cpp): handle TS2DIFF float prefixes in batch decode --- cpp/src/encoding/ts2diff_decoder.h | 26 +++++---------- cpp/test/encoding/ts2diff_codec_test.cc | 43 +++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 18 deletions(-) diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h index 206b7f559..f6fc961b0 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -952,15 +952,10 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder { int read_batch_float(float* out, int capacity, int& actual, common::ByteStream& in) override { - // Reuse SIMD batch decode for int32, then bit-cast to float - int32_t* buf = reinterpret_cast(out); - int ret = TS2DIFFDecoder::read_batch_int32(buf, capacity, - actual, in); - if (ret != common::E_OK) return ret; - for (int i = 0; i < actual; ++i) { - out[i] = common::int_to_float(buf[i]); - } - return common::E_OK; + // FLOAT TS_2DIFF segments have a scale/overflow prefix before the + // integer delta block. The integer batch decoder does not consume + // that prefix, so use the segment-aware scalar decoder here. + return Decoder::read_batch_float(out, capacity, actual, in); } private: @@ -989,15 +984,10 @@ class DoubleTS2DIFFDecoder : public TS2DIFFDecoder { int read_batch_double(double* out, int capacity, int& actual, common::ByteStream& in) override { - // Reuse SIMD batch decode for int64, then bit-cast to double - int64_t* buf = reinterpret_cast(out); - int ret = TS2DIFFDecoder::read_batch_int64(buf, capacity, - actual, in); - if (ret != common::E_OK) return ret; - for (int i = 0; i < actual; ++i) { - out[i] = common::long_to_double(buf[i]); - } - return common::E_OK; + // DOUBLE TS_2DIFF uses the same segment prefix. Bypassing + // read_double() misreads that prefix as a block header and can spin + // at end-of-input while decoding an otherwise valid page. + return Decoder::read_batch_double(out, capacity, actual, in); } private: diff --git a/cpp/test/encoding/ts2diff_codec_test.cc b/cpp/test/encoding/ts2diff_codec_test.cc index fb997103c..43c86adef 100644 --- a/cpp/test/encoding/ts2diff_codec_test.cc +++ b/cpp/test/encoding/ts2diff_codec_test.cc @@ -187,6 +187,49 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, TestDoubleRoundTrip) { EXPECT_FALSE(decoder_double_->has_remaining(out_stream)); } +TEST_F(FloatDoubleTS2DIFFCodecTest, + ReadBatchFloatConsumesPrefixesAcrossSegments) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 300; + std::vector expected(row_num); + for (int i = 0; i < row_num; ++i) { + expected[i] = static_cast(i) * 0.25f + 0.5f; + ASSERT_EQ(encoder_float_->encode(expected[i], out_stream), + common::E_OK); + } + ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK); + + std::vector actual_values(row_num); + int actual = 0; + ASSERT_EQ(decoder_float_->read_batch_float(actual_values.data(), row_num, + actual, out_stream), + common::E_OK); + ASSERT_EQ(actual, row_num); + for (int i = 0; i < row_num; ++i) { + EXPECT_FLOAT_EQ(actual_values[i], expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(out_stream)); +} + +TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchDoubleConsumesOverflowPrefix) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const double expected[] = {3.123456768E20, std::nan("")}; + for (double value : expected) { + ASSERT_EQ(encoder_double_->encode(value, out_stream), common::E_OK); + } + ASSERT_EQ(encoder_double_->flush(out_stream), common::E_OK); + + double actual_values[2] = {}; + int actual = 0; + ASSERT_EQ(decoder_double_->read_batch_double(actual_values, 2, actual, + out_stream), + common::E_OK); + ASSERT_EQ(actual, 2); + EXPECT_DOUBLE_EQ(actual_values[0], expected[0]); + EXPECT_TRUE(std::isnan(actual_values[1])); + EXPECT_FALSE(decoder_double_->has_remaining(out_stream)); +} + TEST_F(TS2DIFFCodecTest, TestIntEncoding1) { common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); const int row_num = 10000; From 1ef5e944b1d2c4e81eaaa1b01a4623e441da632c Mon Sep 17 00:00:00 2001 From: gx Date: Tue, 18 Aug 2026 21:24:38 +0800 Subject: [PATCH 2/4] fix(cpp): make TS2DIFF float/double prefix detection unambiguous The per-block heuristic that distinguished Java-compatible maxPointNumber prefixes from legacy raw delta blocks could misclassify a valid raw header (wi = 0 or bit_width = 0 blocks), which desynced the stream and could spin at end-of-input in batch reads. Decide the page layout once per page instead: parse the whole remaining stream with the Java segment grammar (prefix + overflow bitmaps + block run, validated field ranges and exact exhaustion) and cache the segment prefix offsets. A legacy raw page fails this parse because its first misaligned write_index probe reads >= 0x100. - Legacy raw pages keep the integer SIMD batch decode path with bit-cast semantics (parent-commit behavior). - Java pages consume prefixes only at recorded offsets and take the segment-aware scalar path; this also fixes value semantics across blocks inside one Java segment, which the per-block heuristic could not represent. - Bail out of read_long() when the stream is exhausted with bits still owed, so no residual misconfiguration can loop forever. Also fix ByteStream::check_space(): after set_read_pos() parks the cursor at a page boundary, blindly following read_page_->next_ skipped the boundary page and failed reads with E_OUT_OF_RANGE. Recompute the page from the head instead; page chains are short so the walk is cheap. Add legacy raw batch/scalar/mixed regression tests for FLOAT and DOUBLE (PR #901 review). --- cpp/src/common/allocator/byte_stream.h | 13 +- cpp/src/encoding/ts2diff_decoder.h | 299 +++++++++++++++++++----- cpp/test/encoding/ts2diff_codec_test.cc | 157 +++++++++++++ 3 files changed, 411 insertions(+), 58 deletions(-) diff --git a/cpp/src/common/allocator/byte_stream.h b/cpp/src/common/allocator/byte_stream.h index 15f15b798..36933a8fd 100644 --- a/cpp/src/common/allocator/byte_stream.h +++ b/cpp/src/common/allocator/byte_stream.h @@ -696,7 +696,18 @@ class ByteStream { if (UNLIKELY(read_page_ == nullptr)) { read_page_ = head_.load(); } else if (UNLIKELY((read_pos_ & page_mask_) == 0)) { - read_page_ = read_page_->next_.load(); + // At a page boundary the cursor may have been parked here by a + // preceding sequential read (read_page_ is the page just + // finished, advance one) or by set_read_pos() (read_page_ is + // already the boundary page, advancing would skip it). The + // two states are indistinguishable, so recompute the page + // from the head instead of blindly following next_. + Page* p = head_.load(); + uint64_t page_idx = read_pos_ / page_size_; + while (p != nullptr && page_idx-- > 0) { + p = p->next_.load(); + } + read_page_ = p; } if (UNLIKELY(read_page_ == nullptr)) { return common::E_OUT_OF_RANGE; diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h index f6fc961b0..a3d8f1282 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -219,37 +219,135 @@ inline bool bitmap_marked(const std::vector& bm, int idx) { return (bm[byte_idx] & static_cast(1u << (idx % 8))) != 0; } -inline bool looks_like_ts2diff_header(common::ByteStream& in) { - int ret = common::E_OK; - uint64_t probe_mark = in.read_pos(); - int32_t write_index = 0; - int32_t bit_width = 0; - if (RET_FAIL(common::SerializationUtil::read_i32(write_index, in)) || - RET_FAIL(common::SerializationUtil::read_i32(bit_width, in))) { - in.set_read_pos(probe_mark); - return false; - } - in.set_read_pos(probe_mark); - if (write_index < 0 || write_index > 128) { - return false; - } - if (bit_width < 0 || bit_width > 64) { - return false; +// Parse the remaining stream as one or more Java-compatible FLOAT/DOUBLE +// TS_2DIFF segments, recording the offset of every segment prefix. A +// segment is: +// [overflow flag][value count][underflow bitmap][overflow bitmap?] +// [maxPointNumber varint] block+ +// where a block is [write_index i32][bit_width i32][delta_min][first_value] +// followed by ceil(write_index*bit_width/8) packed bytes. The C++ encoder +// emits one segment per 128-value block, while the Java encoder emits one +// segment wrapping several consecutive blocks, hence "block+". +// +// A legacy raw page (plain delta blocks with no prefix at all) fails this +// parse in practice: its first write_index is >= 1 for any block produced +// by a real encoder, so after the varint tag eats the leading 0x00 byte +// the misaligned write_index probe reads >= 0x100 and is rejected. Only a +// byte-level coincidence could satisfy both interpretations. +// +// Returns true when the whole remaining stream is consumed exactly by the +// segment grammar; the read position is always restored. +inline bool scan_java_float_double_page(common::ByteStream& in, int value_bytes, + std::vector& prefix_offsets) { + const uint64_t page_start = in.read_pos(); + const int bw_limit = (value_bytes == 4) ? 32 : 64; + const int dmfv_bytes = value_bytes * 2; + prefix_offsets.clear(); + + auto skip_bytes = [&in](uint64_t n) -> bool { + uint8_t sink[64]; + while (n > 0) { + uint32_t chunk = n < sizeof(sink) + ? static_cast(n) + : static_cast(sizeof(sink)); + uint32_t got = 0; + if (in.read_buf(sink, chunk, got) != common::E_OK || got != chunk) { + return false; + } + n -= chunk; + } + return true; + }; + + bool valid = true; + while (valid && in.has_remaining()) { + prefix_offsets.push_back(in.read_pos()); + uint64_t group_value_count = 0; // sum of (write_index+1) per block + uint64_t expected_count = 0; // bitmap value count, 0 = no bitmap + uint32_t tag = 0; + if (common::SerializationUtil::read_var_uint(tag, in) != common::E_OK) { + valid = false; + break; + } + if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW || + tag == FLAG_SCALED_VALUE_OVERFLOW) { + uint32_t n = 0; + if (common::SerializationUtil::read_var_uint(n, in) != + common::E_OK) { + valid = false; + break; + } + expected_count = n; + const uint64_t bm_len = static_cast(n) / 8 + 1; + if (!skip_bytes(bm_len)) { // underflow bitmap + valid = false; + break; + } + if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW && !skip_bytes(bm_len)) { + valid = false; // overflow bitmap + break; + } + uint32_t mpn = 0; + if (common::SerializationUtil::read_var_uint(mpn, in) != + common::E_OK) { + valid = false; + break; + } + } + // Consume the blocks owned by this segment. A block run continues + // while a block header validates; an invalid header marks either + // the next segment prefix or a corrupt page (settled by whether + // the whole-stream parse consumes exactly below). + int blocks_in_segment = 0; + while (true) { + const uint64_t block_mark = in.read_pos(); + int32_t wi = 0; + int32_t bw = 0; + if (common::SerializationUtil::read_i32(wi, in) != common::E_OK || + common::SerializationUtil::read_i32(bw, in) != common::E_OK) { + in.set_read_pos(block_mark); + break; + } + if (wi < 0 || wi > 128 || bw < 0 || bw > bw_limit) { + in.set_read_pos(block_mark); + break; + } + const uint64_t packed_bytes = + (static_cast(wi) * bw + 7) / 8; + if (!skip_bytes(dmfv_bytes) || !skip_bytes(packed_bytes)) { + valid = false; + break; + } + group_value_count += static_cast(wi) + 1; + ++blocks_in_segment; + } + if (!valid || blocks_in_segment == 0) { + valid = false; + break; + } + // The overflow bitmap covers exactly the values of its segment. + if (expected_count != 0 && group_value_count != expected_count) { + valid = false; + break; + } } - return true; + in.set_read_pos(page_start); + return valid && !prefix_offsets.empty(); } +// Consume one Java-compatible segment prefix. Must only be called where +// scan_java_float_double_page() recorded a prefix offset: without the page +// scan there is no reliable way to tell a maxPointNumber varint apart from +// the first byte of a legacy raw block header. inline int consume_float_double_ts2diff_prefix( - common::ByteStream& in, bool& is_legacy_raw, int& max_point_number, + common::ByteStream& in, int& max_point_number, std::vector& underflow_bm, std::vector& overflow_bm, int& segment_size) { int ret = common::E_OK; - is_legacy_raw = false; max_point_number = 0; underflow_bm.clear(); overflow_bm.clear(); segment_size = 0; - uint64_t mark = in.read_pos(); uint32_t tag = 0; if (RET_FAIL(common::SerializationUtil::read_var_uint(tag, in))) { return ret; @@ -285,15 +383,7 @@ inline int consume_float_double_ts2diff_prefix( max_point_number = static_cast(mpn); return common::E_OK; } - - // Distinguish Java maxPointNumber prefix from legacy raw C++ block. max_point_number = static_cast(tag); - if (!looks_like_ts2diff_header(in)) { - in.set_read_pos(mark); - is_legacy_raw = true; - } else { - segment_size = 0; - } return common::E_OK; } @@ -348,6 +438,12 @@ class TS2DIFFDecoder : public Decoder { int64_t value = 0; while (bits > 0) { read_byte_if_empty(in); + // End of input with bits still owed (corrupt or desynced + // stream): bail out instead of looping forever on a stale + // buffer_ / bits_left_ == 0 pair. + if (bits_left_ == 0 && !in.has_remaining()) { + break; + } if (bits > bits_left_ || bits == 8) { // Take only the bits_left_ "least significant" bits. uint8_t d = (uint8_t)(buffer_ & ((1 << bits_left_) - 1)); @@ -933,10 +1029,58 @@ inline int TS2DIFFDecoder::skip_int32(int count, int& skipped, } // ============================================================================ -// Float / Double wrapper decoders (unchanged) +// Float / Double wrapper decoders // ============================================================================ -class FloatTS2DIFFDecoder : public TS2DIFFDecoder { +// Common page-layout detection shared by the FLOAT and DOUBLE decoders. +// A page is either legacy raw (plain delta blocks, written by old C++ +// encoders that bit-cast the float bits into the integer TS_2DIFF stream) +// or Java-compatible (each segment prefixed by a maxPointNumber varint and +// an optional overflow bitmap). The layout is decided once per page by +// parsing the whole page with the Java grammar: a page only counts as +// Java-compatible when the grammar consumes it exactly. This removes the +// old per-block heuristic, which could misclassify a legacy raw header as +// a prefix, desync the stream and spin at end-of-input. +class FloatDoublePageLayout { + protected: + void reset_layout() { + layout_known_ = false; + is_legacy_raw_ = false; + next_prefix_ = 0; + prefix_offsets_.clear(); + } + + // Decide the layout of the page starting at the current read position. + // Safe to call repeatedly before the first value: it always restores + // the read position on exit. + void ensure_layout(common::ByteStream& in, int value_bytes) { + if (!layout_known_) { + is_legacy_raw_ = !ts2diff_java_detail::scan_java_float_double_page( + in, value_bytes, prefix_offsets_); + next_prefix_ = 0; + layout_known_ = true; + } + } + + // True when a Java segment prefix sits at the current read position + // (start of page or start of a new segment). Legacy raw pages never + // match. + bool at_segment_prefix(common::ByteStream& in) { + return !is_legacy_raw_ && next_prefix_ < prefix_offsets_.size() && + prefix_offsets_[next_prefix_] == in.read_pos(); + } + + void advance_segment_prefix() { ++next_prefix_; } + + protected: + bool layout_known_{false}; + bool is_legacy_raw_{false}; + size_t next_prefix_{0}; + std::vector prefix_offsets_; +}; + +class FloatTS2DIFFDecoder : public TS2DIFFDecoder, + protected FloatDoublePageLayout { public: FloatTS2DIFFDecoder() = default; float decode(common::ByteStream& in) { @@ -944,6 +1088,11 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder { return common::int_to_float(value_int); } + void reset() override { + TS2DIFFDecoder::reset(); + reset_layout(); + } + int read_boolean(bool& ret_value, common::ByteStream& in) override; int read_int32(int32_t& ret_value, common::ByteStream& in) override; int read_int64(int64_t& ret_value, common::ByteStream& in) override; @@ -952,14 +1101,26 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder { int read_batch_float(float* out, int capacity, int& actual, common::ByteStream& in) override { - // FLOAT TS_2DIFF segments have a scale/overflow prefix before the - // integer delta block. The integer batch decoder does not consume - // that prefix, so use the segment-aware scalar decoder here. + // Legacy raw pages are plain int32 delta blocks: reuse the integer + // SIMD batch decoder and bit-cast the results (the layout the old + // C++ writer produced). Java-compatible pages carry segment + // prefixes the integer decoder would misread, so they take the + // segment-aware scalar path. + ensure_layout(in, 4); + if (is_legacy_raw_) { + int32_t* buf = reinterpret_cast(out); + int ret = TS2DIFFDecoder::read_batch_int32(buf, capacity, + actual, in); + if (ret != common::E_OK) return ret; + for (int i = 0; i < actual; ++i) { + out[i] = common::int_to_float(buf[i]); + } + return common::E_OK; + } return Decoder::read_batch_float(out, capacity, actual, in); } private: - bool is_legacy_raw_{false}; int max_point_number_{0}; double max_point_value_{1.0}; int segment_pos_{0}; @@ -968,7 +1129,8 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder { std::vector overflow_bm_; }; -class DoubleTS2DIFFDecoder : public TS2DIFFDecoder { +class DoubleTS2DIFFDecoder : public TS2DIFFDecoder, + protected FloatDoublePageLayout { public: DoubleTS2DIFFDecoder() = default; double decode(common::ByteStream& in) { @@ -976,6 +1138,11 @@ class DoubleTS2DIFFDecoder : public TS2DIFFDecoder { return common::long_to_double(value_long); } + void reset() override { + TS2DIFFDecoder::reset(); + reset_layout(); + } + int read_boolean(bool& ret_value, common::ByteStream& in) override; int read_int32(int32_t& ret_value, common::ByteStream& in) override; int read_int64(int64_t& ret_value, common::ByteStream& in) override; @@ -984,14 +1151,22 @@ class DoubleTS2DIFFDecoder : public TS2DIFFDecoder { int read_batch_double(double* out, int capacity, int& actual, common::ByteStream& in) override { - // DOUBLE TS_2DIFF uses the same segment prefix. Bypassing - // read_double() misreads that prefix as a block header and can spin - // at end-of-input while decoding an otherwise valid page. + // Same split as FloatTS2DIFFDecoder::read_batch_float — see there. + ensure_layout(in, 8); + if (is_legacy_raw_) { + int64_t* buf = reinterpret_cast(out); + int ret = TS2DIFFDecoder::read_batch_int64(buf, capacity, + actual, in); + if (ret != common::E_OK) return ret; + for (int i = 0; i < actual; ++i) { + out[i] = common::long_to_double(buf[i]); + } + return common::E_OK; + } return Decoder::read_batch_double(out, capacity, actual, in); } private: - bool is_legacy_raw_{false}; int max_point_number_{0}; double max_point_value_{1.0}; int segment_pos_{0}; @@ -1097,16 +1272,21 @@ FORCE_INLINE int FloatTS2DIFFDecoder::read_float(float& ret_value, common::ByteStream& in) { int ret = common::E_OK; if (current_index_ == 0) { - if (RET_FAIL(ts2diff_java_detail::consume_float_double_ts2diff_prefix( - in, is_legacy_raw_, max_point_number_, underflow_bm_, - overflow_bm_, segment_size_))) { - return ret; + ensure_layout(in, 4); + if (at_segment_prefix(in)) { + if (RET_FAIL( + ts2diff_java_detail::consume_float_double_ts2diff_prefix( + in, max_point_number_, underflow_bm_, overflow_bm_, + segment_size_))) { + return ret; + } + max_point_value_ = + max_point_number_ <= 0 + ? 1.0 + : std::pow(10.0, static_cast(max_point_number_)); + segment_pos_ = 0; + advance_segment_prefix(); } - max_point_value_ = - max_point_number_ <= 0 - ? 1.0 - : std::pow(10.0, static_cast(max_point_number_)); - segment_pos_ = 0; } if (is_legacy_raw_) { ret_value = decode(in); @@ -1158,16 +1338,21 @@ FORCE_INLINE int DoubleTS2DIFFDecoder::read_double(double& ret_value, common::ByteStream& in) { int ret = common::E_OK; if (current_index_ == 0) { - if (RET_FAIL(ts2diff_java_detail::consume_float_double_ts2diff_prefix( - in, is_legacy_raw_, max_point_number_, underflow_bm_, - overflow_bm_, segment_size_))) { - return ret; + ensure_layout(in, 8); + if (at_segment_prefix(in)) { + if (RET_FAIL( + ts2diff_java_detail::consume_float_double_ts2diff_prefix( + in, max_point_number_, underflow_bm_, overflow_bm_, + segment_size_))) { + return ret; + } + max_point_value_ = + max_point_number_ <= 0 + ? 1.0 + : std::pow(10.0, static_cast(max_point_number_)); + segment_pos_ = 0; + advance_segment_prefix(); } - max_point_value_ = - max_point_number_ <= 0 - ? 1.0 - : std::pow(10.0, static_cast(max_point_number_)); - segment_pos_ = 0; } if (is_legacy_raw_) { ret_value = decode(in); diff --git a/cpp/test/encoding/ts2diff_codec_test.cc b/cpp/test/encoding/ts2diff_codec_test.cc index 43c86adef..35aa60d0f 100644 --- a/cpp/test/encoding/ts2diff_codec_test.cc +++ b/cpp/test/encoding/ts2diff_codec_test.cc @@ -523,4 +523,161 @@ TEST(FloatTS2DIFFEncoderResetTest, ResetClearsUnderflowFlags) { } } +// Regression: legacy raw float/double segments (written by the old C++ +// encoders, i.e. plain int delta blocks with no maxPointNumber / overflow +// prefix, values stored as bit-cast float bits) must stay decodable through +// read_batch_float / read_batch_double. The per-block prefix heuristic +// used to misclassify a valid raw header as a maxPointNumber prefix, +// desyncing the stream and spinning at end-of-input (PR #901 review). +TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchFloatLegacyRawSegments) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 129; + std::vector expected(row_num); + // 128 equal values followed by a change: the first legacy block has + // bit_width = 0 and delta_min = 0, exactly the pattern the old + // heuristic misclassified. The trailing 1-value block (write_index = 0) + // exercised the same heuristic again. + for (int i = 0; i < row_num; ++i) { + expected[i] = (i < 128) ? 1.5f : 2.5f; + } + IntTS2DIFFEncoder raw_encoder; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ( + raw_encoder.encode(common::float_to_int(expected[i]), out_stream), + common::E_OK); + } + ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK); + + std::vector actual_values(row_num); + int decoded = 0; + // Small batches exercise the layout decision plus repeated block + // transitions. + while (decoded < row_num) { + int actual = 0; + ASSERT_EQ(decoder_float_->read_batch_float( + actual_values.data() + decoded, 16, actual, out_stream), + common::E_OK); + ASSERT_GT(actual, 0); + decoded += actual; + } + for (int i = 0; i < row_num; ++i) { + EXPECT_EQ(actual_values[i], expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(out_stream)); +} + +TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchDoubleLegacyRawSegments) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 129; + std::vector expected(row_num); + for (int i = 0; i < row_num; ++i) { + expected[i] = (i < 128) ? 1.5 : 2.5; + } + LongTS2DIFFEncoder raw_encoder; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ( + raw_encoder.encode(common::double_to_long(expected[i]), out_stream), + common::E_OK); + } + ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK); + + std::vector actual_values(row_num); + int decoded = 0; + while (decoded < row_num) { + int actual = 0; + ASSERT_EQ(decoder_double_->read_batch_double( + actual_values.data() + decoded, 16, actual, out_stream), + common::E_OK); + ASSERT_GT(actual, 0); + decoded += actual; + } + for (int i = 0; i < row_num; ++i) { + EXPECT_EQ(actual_values[i], expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_double_->has_remaining(out_stream)); +} + +// The layout decision must also cover the scalar path: legacy raw pages +// read one value at a time via read_float / read_double keep the bit-cast +// semantics across the 128-value block boundary. +TEST_F(FloatDoubleTS2DIFFCodecTest, ReadFloatLegacyRawScalar) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 129; + std::vector expected(row_num); + for (int i = 0; i < row_num; ++i) { + expected[i] = (i < 128) ? 1.5f : 2.5f; + } + IntTS2DIFFEncoder raw_encoder; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ( + raw_encoder.encode(common::float_to_int(expected[i]), out_stream), + common::E_OK); + } + ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK); + + float v = 0.f; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ(decoder_float_->read_float(v, out_stream), common::E_OK); + EXPECT_EQ(v, expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(out_stream)); +} + +TEST_F(FloatDoubleTS2DIFFCodecTest, ReadDoubleLegacyRawScalar) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 129; + std::vector expected(row_num); + for (int i = 0; i < row_num; ++i) { + expected[i] = (i < 128) ? 1.5 : 2.5; + } + LongTS2DIFFEncoder raw_encoder; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ( + raw_encoder.encode(common::double_to_long(expected[i]), out_stream), + common::E_OK); + } + ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK); + + double v = 0.; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ(decoder_double_->read_double(v, out_stream), common::E_OK); + EXPECT_EQ(v, expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_double_->has_remaining(out_stream)); +} + +// Mixed reads must not re-trigger the layout scan mid-page: batch first, +// then scalar reads must continue on the same layout decision. +TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyRawBatchThenScalarReads) { + common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 129; + std::vector expected(row_num); + for (int i = 0; i < row_num; ++i) { + expected[i] = (i < 128) ? 0.5f : 100.25f; + } + IntTS2DIFFEncoder raw_encoder; + for (int i = 0; i < row_num; ++i) { + ASSERT_EQ( + raw_encoder.encode(common::float_to_int(expected[i]), out_stream), + common::E_OK); + } + ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK); + + float batch_out[40]; + int actual = 0; + ASSERT_EQ( + decoder_float_->read_batch_float(batch_out, 40, actual, out_stream), + common::E_OK); + ASSERT_EQ(actual, 40); + for (int i = 0; i < 40; ++i) { + EXPECT_EQ(batch_out[i], expected[i]) << "row " << i; + } + float v = 0.f; + for (int i = 40; i < row_num; ++i) { + ASSERT_EQ(decoder_float_->read_float(v, out_stream), common::E_OK); + EXPECT_EQ(v, expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(out_stream)); +} + } // namespace storage From 4e32dfe07a68791fdf71b0b9c6bf4f2df11972de Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 19 Aug 2026 08:12:03 +0800 Subject: [PATCH 3/4] fix(cpp): write maxPointNumber once per page in TS_2DIFF float/double (apache/tsfile#910) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of #910: the C++ FloatTS2DIFFEncoder/DoubleTS2DIFFEncoder wrote the maxPointNumber field (fixed value 2) at every segment boundary, while Java FloatEncoder/DoubleEncoder write it only once at the start of each page. Files written with an empty/short first segment could then be misparsed by Java readers (e.g. TsFileSketchTool crashing on the trailing maxPointNumber). This change aligns the C++ encoder with the Java layout: - Encoder: the maxPointNumber var_uint is now emitted exactly once per page (on reset, before segment 1). Segment boundaries only carry the overflow/underflow FLAG when needed, matching Java's segment grammar. - Decoder: forward-only, prefix-aware parsing that accepts all three page layouts — legacy raw pages (no prefix at all), the new Java format (maxPointNumber only on the first segment), and old C++ per-segment format (backward compatible). The old peek-and-rewind scheme is gone; the segment header of a prefix-free segment is preloaded so decode() never needs to re-read the stream. - Tests: new gtest cases assert the maxPointNumber-once-per-page byte layout for multi-segment pages, scaled-overflow pages (the #910 crash scenario), reset() page boundaries, and legacy per-segment backward compatibility. Verified: full C++ test suite passes; Java TsFileSketchTool reads files written by the fixed encoder; tsfile_cli round-trips the data. --- cpp/src/encoding/ts2diff_decoder.h | 512 +++++++++++++----------- cpp/src/encoding/ts2diff_encoder.h | 34 +- cpp/test/encoding/ts2diff_codec_test.cc | 379 +++++++++++++++++- 3 files changed, 676 insertions(+), 249 deletions(-) diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h index a3d8f1282..bbf1c310b 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -219,137 +219,144 @@ inline bool bitmap_marked(const std::vector& bm, int idx) { return (bm[byte_idx] & static_cast(1u << (idx % 8))) != 0; } -// Parse the remaining stream as one or more Java-compatible FLOAT/DOUBLE -// TS_2DIFF segments, recording the offset of every segment prefix. A -// segment is: -// [overflow flag][value count][underflow bitmap][overflow bitmap?] -// [maxPointNumber varint] block+ -// where a block is [write_index i32][bit_width i32][delta_min][first_value] -// followed by ceil(write_index*bit_width/8) packed bytes. The C++ encoder -// emits one segment per 128-value block, while the Java encoder emits one -// segment wrapping several consecutive blocks, hence "block+". -// -// A legacy raw page (plain delta blocks with no prefix at all) fails this -// parse in practice: its first write_index is >= 1 for any block produced -// by a real encoder, so after the varint tag eats the leading 0x00 byte -// the misaligned write_index probe reads >= 0x100 and is rejected. Only a -// byte-level coincidence could satisfy both interpretations. -// -// Returns true when the whole remaining stream is consumed exactly by the -// segment grammar; the read position is always restored. -inline bool scan_java_float_double_page(common::ByteStream& in, int value_bytes, - std::vector& prefix_offsets) { - const uint64_t page_start = in.read_pos(); - const int bw_limit = (value_bytes == 4) ? 32 : 64; - const int dmfv_bytes = value_bytes * 2; - prefix_offsets.clear(); - - auto skip_bytes = [&in](uint64_t n) -> bool { - uint8_t sink[64]; - while (n > 0) { - uint32_t chunk = n < sizeof(sink) - ? static_cast(n) - : static_cast(sizeof(sink)); - uint32_t got = 0; - if (in.read_buf(sink, chunk, got) != common::E_OK || got != chunk) { - return false; - } - n -= chunk; +inline bool looks_like_ts2diff_header(common::ByteStream& in) { + int ret = common::E_OK; + uint64_t probe_mark = in.read_pos(); + int32_t write_index = 0; + int32_t bit_width = 0; + if (RET_FAIL(common::SerializationUtil::read_i32(write_index, in)) || + RET_FAIL(common::SerializationUtil::read_i32(bit_width, in))) { + in.set_read_pos(probe_mark); + return false; + } + in.set_read_pos(probe_mark); + if (write_index < 0 || write_index > 128) { + return false; + } + if (bit_width < 0 || bit_width > 64) { + return false; + } + return true; +} + +struct SegmentHeaderPreload { + int32_t write_index = 0; + int32_t bit_width = 0; + int64_t delta_min = 0; + int64_t first_value = 0; + bool ready = false; +}; + +// Reads a LEB128 var_uint where the first byte was already consumed into +// `first_byte`. Forward-only: never rewinds the stream. +inline int read_var_uint_tail(uint8_t first_byte, common::ByteStream& in, + uint32_t& out) { + int ret = common::E_OK; + out = static_cast(first_byte & 0x7F); + int shift = 7; + uint8_t b = first_byte; + while (b & 0x80) { + uint32_t read_len = 0; + if (RET_FAIL(in.read_buf(&b, 1, read_len)) || read_len != 1) { + return ret; } - return true; - }; - - bool valid = true; - while (valid && in.has_remaining()) { - prefix_offsets.push_back(in.read_pos()); - uint64_t group_value_count = 0; // sum of (write_index+1) per block - uint64_t expected_count = 0; // bitmap value count, 0 = no bitmap - uint32_t tag = 0; - if (common::SerializationUtil::read_var_uint(tag, in) != common::E_OK) { - valid = false; - break; + if (shift > 28) { + return common::E_TSFILE_CORRUPTED; } - if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW || - tag == FLAG_SCALED_VALUE_OVERFLOW) { - uint32_t n = 0; - if (common::SerializationUtil::read_var_uint(n, in) != - common::E_OK) { - valid = false; - break; - } - expected_count = n; - const uint64_t bm_len = static_cast(n) / 8 + 1; - if (!skip_bytes(bm_len)) { // underflow bitmap - valid = false; - break; - } - if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW && !skip_bytes(bm_len)) { - valid = false; // overflow bitmap - break; - } - uint32_t mpn = 0; - if (common::SerializationUtil::read_var_uint(mpn, in) != - common::E_OK) { - valid = false; - break; - } + out |= static_cast(b & 0x7F) << shift; + shift += 7; + } + return common::E_OK; +} + +// Parses the segment header (write_index + bit_width + delta_min + +// first_value) forward-only. `wi_hi` is the first (already consumed) byte +// of the big-endian write_index - always 0x00 for the no-prefix layout. +inline int read_segment_header_preload(common::ByteStream& in, bool is_double, + uint8_t wi_hi, + SegmentHeaderPreload& h) { + int ret = common::E_OK; + uint8_t rest[3] = {0, 0, 0}; + uint32_t read_len = 0; + if (RET_FAIL(in.read_buf(rest, 3, read_len)) || read_len != 3) { + return ret; + } + h.write_index = (static_cast(wi_hi) << 24) | + (static_cast(rest[0]) << 16) | + (static_cast(rest[1]) << 8) | + static_cast(rest[2]); + int32_t bw = 0; + if (RET_FAIL(common::SerializationUtil::read_i32(bw, in))) { + return ret; + } + h.bit_width = bw; + if (is_double) { + if (RET_FAIL(common::SerializationUtil::read_i64(h.delta_min, in))) { + return ret; } - // Consume the blocks owned by this segment. A block run continues - // while a block header validates; an invalid header marks either - // the next segment prefix or a corrupt page (settled by whether - // the whole-stream parse consumes exactly below). - int blocks_in_segment = 0; - while (true) { - const uint64_t block_mark = in.read_pos(); - int32_t wi = 0; - int32_t bw = 0; - if (common::SerializationUtil::read_i32(wi, in) != common::E_OK || - common::SerializationUtil::read_i32(bw, in) != common::E_OK) { - in.set_read_pos(block_mark); - break; - } - if (wi < 0 || wi > 128 || bw < 0 || bw > bw_limit) { - in.set_read_pos(block_mark); - break; - } - const uint64_t packed_bytes = - (static_cast(wi) * bw + 7) / 8; - if (!skip_bytes(dmfv_bytes) || !skip_bytes(packed_bytes)) { - valid = false; - break; - } - group_value_count += static_cast(wi) + 1; - ++blocks_in_segment; + if (RET_FAIL(common::SerializationUtil::read_i64(h.first_value, in))) { + return ret; } - if (!valid || blocks_in_segment == 0) { - valid = false; - break; + } else { + int32_t dm = 0; + int32_t fv = 0; + if (RET_FAIL(common::SerializationUtil::read_i32(dm, in))) { + return ret; } - // The overflow bitmap covers exactly the values of its segment. - if (expected_count != 0 && group_value_count != expected_count) { - valid = false; - break; + if (RET_FAIL(common::SerializationUtil::read_i32(fv, in))) { + return ret; } + h.delta_min = dm; + h.first_value = fv; } - in.set_read_pos(page_start); - return valid && !prefix_offsets.empty(); + h.ready = true; + return common::E_OK; } -// Consume one Java-compatible segment prefix. Must only be called where -// scan_java_float_double_page() recorded a prefix offset: without the page -// scan there is no reliable way to tell a maxPointNumber varint apart from -// the first byte of a legacy raw block header. inline int consume_float_double_ts2diff_prefix( - common::ByteStream& in, int& max_point_number, - std::vector& underflow_bm, std::vector& overflow_bm, - int& segment_size) { + common::ByteStream& in, bool& is_legacy_raw, bool& max_pn_present, + int& max_point_number, std::vector& underflow_bm, + std::vector& overflow_bm, int& segment_size, + bool page_first_segment, bool is_double, SegmentHeaderPreload& preload) { int ret = common::E_OK; + is_legacy_raw = false; + max_pn_present = true; max_point_number = 0; underflow_bm.clear(); overflow_bm.clear(); segment_size = 0; + uint64_t mark = in.read_pos(); + // apache/tsfile#910 layout: only the page's first segment carries the + // Java maxPointNumber prefix; later segments start directly with the + // 4-byte write_index whose high byte is 0x00. This library always + // serializes max_point_number_ = 2 (0x02), so a leading 0x00 can only + // mean "no prefix on this segment". + // + // Everything is parsed forward-only: rewinding to a page-aligned offset + // (e.g. the start of a page) makes ByteStream::check_space() advance + // the page cursor one page too far and fail the next read, so no + // peek-and-restore is used here. + uint8_t first_byte = 0; + uint32_t read_len = 0; + if (RET_FAIL(in.read_buf(&first_byte, 1, read_len)) || read_len != 1) { + return ret; + } + if (first_byte == 0x00) { + // No prefix: the segment header begins with write_index 0x00... + if (page_first_segment) { + // A page whose very first segment has no prefix is a legacy + // raw C++ block page (no scaling at all). + is_legacy_raw = true; + } + max_pn_present = false; + if (RET_FAIL(read_segment_header_preload(in, is_double, first_byte, + preload))) { + return ret; + } + return common::E_OK; + } uint32_t tag = 0; - if (RET_FAIL(common::SerializationUtil::read_var_uint(tag, in))) { + if (RET_FAIL(read_var_uint_tail(first_byte, in, tag))) { return ret; } if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW || @@ -361,7 +368,6 @@ inline int consume_float_double_ts2diff_prefix( segment_size = static_cast(n); int bm_len = segment_size / 8 + 1; underflow_bm.resize(static_cast(bm_len), 0); - uint32_t read_len = 0; if (RET_FAIL(in.read_buf(underflow_bm.data(), static_cast(bm_len), read_len)) || read_len != static_cast(bm_len)) { @@ -376,17 +382,56 @@ inline int consume_float_double_ts2diff_prefix( return ret; } } + if (page_first_segment) { + // First segment: maxPointNumber always follows the bitmaps. + uint32_t mpn = 0; + if (RET_FAIL(common::SerializationUtil::read_var_uint(mpn, in))) { + return ret; + } + max_point_number = static_cast(mpn); + return common::E_OK; + } + // Later segment: new-format pages jump straight to the segment + // header (0x00 write_index high byte); old-format pages repeat the + // maxPointNumber prefix here. + uint8_t after_bm_byte = 0; + if (RET_FAIL(in.read_buf(&after_bm_byte, 1, read_len)) || + read_len != 1) { + return ret; + } + if (after_bm_byte == 0x00) { + max_pn_present = false; + if (RET_FAIL(read_segment_header_preload(in, is_double, + after_bm_byte, + preload))) { + return ret; + } + return common::E_OK; + } uint32_t mpn = 0; - if (RET_FAIL(common::SerializationUtil::read_var_uint(mpn, in))) { + if (RET_FAIL(read_var_uint_tail(after_bm_byte, in, mpn))) { return ret; } max_point_number = static_cast(mpn); return common::E_OK; } + + // A non-flag tag is the maxPointNumber prefix itself. max_point_number = static_cast(tag); + if (!looks_like_ts2diff_header(in)) { + // Only reachable on corrupt/nonstandard data: a non-flag tag whose + // following bytes are not a valid segment header. Rewind and fall + // back to the raw-block path. The rewind target may be page-aligned + // (e.g. a page start), which trips ByteStream::check_space's page + // cursor - accepted here because valid data never takes this branch. + in.set_read_pos(mark); + is_legacy_raw = true; + max_pn_present = false; + } else { + segment_size = 0; + } return common::E_OK; } - } // namespace ts2diff_java_detail // ============================================================================ @@ -409,6 +454,7 @@ class TS2DIFFDecoder : public Decoder { bit_width_ = 0; current_index_ = 0; header_peeked_ = false; + header_preloaded_ = false; } FORCE_INLINE bool has_remaining(const common::ByteStream& buffer) override { @@ -438,12 +484,6 @@ class TS2DIFFDecoder : public Decoder { int64_t value = 0; while (bits > 0) { read_byte_if_empty(in); - // End of input with bits still owed (corrupt or desynced - // stream): bail out instead of looping forever on a stale - // buffer_ / bits_left_ == 0 pair. - if (bits_left_ == 0 && !in.has_remaining()) { - break; - } if (bits > bits_left_ || bits == 8) { // Take only the bits_left_ "least significant" bits. uint8_t d = (uint8_t)(buffer_ & ((1 << bits_left_) - 1)); @@ -499,6 +539,9 @@ class TS2DIFFDecoder : public Decoder { int write_index_; int current_index_; bool header_peeked_; + // Set when consume_float_double_ts2diff_prefix already parsed the + // segment header (prefix-free segment); decode() must not re-read it. + bool header_preloaded_{false}; }; // ============================================================================ @@ -509,9 +552,15 @@ template <> inline int32_t TS2DIFFDecoder::decode(common::ByteStream& in) { int32_t ret_value = stored_value_; if (UNLIKELY(current_index_ == 0)) { - read_header(in); - common::SerializationUtil::read_i32(delta_min_, in); - common::SerializationUtil::read_i32(first_value_, in); + // A prefix-free segment (no maxPointNumber) has its header parsed + // by consume_float_double_ts2diff_prefix already. + if (UNLIKELY(header_preloaded_)) { + header_preloaded_ = false; + } else { + read_header(in); + common::SerializationUtil::read_i32(delta_min_, in); + common::SerializationUtil::read_i32(first_value_, in); + } ret_value = first_value_; bits_left_ = 0; buffer_ = 0; @@ -538,9 +587,13 @@ template <> inline int64_t TS2DIFFDecoder::decode(common::ByteStream& in) { int64_t ret_value = stored_value_; if (UNLIKELY(current_index_ == 0)) { - read_header(in); - common::SerializationUtil::read_i64(delta_min_, in); - common::SerializationUtil::read_i64(first_value_, in); + if (UNLIKELY(header_preloaded_)) { + header_preloaded_ = false; + } else { + read_header(in); + common::SerializationUtil::read_i64(delta_min_, in); + common::SerializationUtil::read_i64(first_value_, in); + } ret_value = first_value_; if (write_index_ == 0) { current_index_ = 0; @@ -1029,70 +1082,32 @@ inline int TS2DIFFDecoder::skip_int32(int count, int& skipped, } // ============================================================================ -// Float / Double wrapper decoders +// Float / Double wrapper decoders (unchanged) // ============================================================================ -// Common page-layout detection shared by the FLOAT and DOUBLE decoders. -// A page is either legacy raw (plain delta blocks, written by old C++ -// encoders that bit-cast the float bits into the integer TS_2DIFF stream) -// or Java-compatible (each segment prefixed by a maxPointNumber varint and -// an optional overflow bitmap). The layout is decided once per page by -// parsing the whole page with the Java grammar: a page only counts as -// Java-compatible when the grammar consumes it exactly. This removes the -// old per-block heuristic, which could misclassify a legacy raw header as -// a prefix, desync the stream and spin at end-of-input. -class FloatDoublePageLayout { - protected: - void reset_layout() { - layout_known_ = false; - is_legacy_raw_ = false; - next_prefix_ = 0; - prefix_offsets_.clear(); - } - - // Decide the layout of the page starting at the current read position. - // Safe to call repeatedly before the first value: it always restores - // the read position on exit. - void ensure_layout(common::ByteStream& in, int value_bytes) { - if (!layout_known_) { - is_legacy_raw_ = !ts2diff_java_detail::scan_java_float_double_page( - in, value_bytes, prefix_offsets_); - next_prefix_ = 0; - layout_known_ = true; - } - } - - // True when a Java segment prefix sits at the current read position - // (start of page or start of a new segment). Legacy raw pages never - // match. - bool at_segment_prefix(common::ByteStream& in) { - return !is_legacy_raw_ && next_prefix_ < prefix_offsets_.size() && - prefix_offsets_[next_prefix_] == in.read_pos(); - } - - void advance_segment_prefix() { ++next_prefix_; } - - protected: - bool layout_known_{false}; - bool is_legacy_raw_{false}; - size_t next_prefix_{0}; - std::vector prefix_offsets_; -}; - -class FloatTS2DIFFDecoder : public TS2DIFFDecoder, - protected FloatDoublePageLayout { +class FloatTS2DIFFDecoder : public TS2DIFFDecoder { public: FloatTS2DIFFDecoder() = default; + // PageReader invokes reset() at every page boundary; the first segment + // of a page is the only one that may carry the maxPointNumber prefix. + void reset() override { + TS2DIFFDecoder::reset(); + page_first_segment_ = true; + // A legacy raw page sets is_legacy_raw_ for the whole object; clear + // it (and the per-page scale/bitmap state) so a decoder object + // reused across pages stays correct. + is_legacy_raw_ = false; + max_point_value_ = 1.0; + underflow_bm_.clear(); + overflow_bm_.clear(); + segment_pos_ = 0; + segment_size_ = 0; + } float decode(common::ByteStream& in) { int32_t value_int = TS2DIFFDecoder::decode(in); return common::int_to_float(value_int); } - void reset() override { - TS2DIFFDecoder::reset(); - reset_layout(); - } - int read_boolean(bool& ret_value, common::ByteStream& in) override; int read_int32(int32_t& ret_value, common::ByteStream& in) override; int read_int64(int64_t& ret_value, common::ByteStream& in) override; @@ -1101,48 +1116,44 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder, int read_batch_float(float* out, int capacity, int& actual, common::ByteStream& in) override { - // Legacy raw pages are plain int32 delta blocks: reuse the integer - // SIMD batch decoder and bit-cast the results (the layout the old - // C++ writer produced). Java-compatible pages carry segment - // prefixes the integer decoder would misread, so they take the - // segment-aware scalar path. - ensure_layout(in, 4); - if (is_legacy_raw_) { - int32_t* buf = reinterpret_cast(out); - int ret = TS2DIFFDecoder::read_batch_int32(buf, capacity, - actual, in); - if (ret != common::E_OK) return ret; - for (int i = 0; i < actual; ++i) { - out[i] = common::int_to_float(buf[i]); - } - return common::E_OK; - } + // FLOAT TS_2DIFF segments have a scale/overflow prefix before the + // integer delta block. The integer batch decoder does not consume + // that prefix, so use the segment-aware scalar decoder here. + // Note: skip_int32/skip_int64 are likewise unsupported on the + // float/double decoders - the segment prefix layout makes the raw + // header-skip path invalid. return Decoder::read_batch_float(out, capacity, actual, in); } private: + bool is_legacy_raw_{false}; int max_point_number_{0}; double max_point_value_{1.0}; int segment_pos_{0}; int segment_size_{0}; std::vector underflow_bm_; std::vector overflow_bm_; + bool page_first_segment_{true}; }; -class DoubleTS2DIFFDecoder : public TS2DIFFDecoder, - protected FloatDoublePageLayout { +class DoubleTS2DIFFDecoder : public TS2DIFFDecoder { public: DoubleTS2DIFFDecoder() = default; + void reset() override { + TS2DIFFDecoder::reset(); + page_first_segment_ = true; + is_legacy_raw_ = false; + max_point_value_ = 1.0; + underflow_bm_.clear(); + overflow_bm_.clear(); + segment_pos_ = 0; + segment_size_ = 0; + } double decode(common::ByteStream& in) { int64_t value_long = TS2DIFFDecoder::decode(in); return common::long_to_double(value_long); } - void reset() override { - TS2DIFFDecoder::reset(); - reset_layout(); - } - int read_boolean(bool& ret_value, common::ByteStream& in) override; int read_int32(int32_t& ret_value, common::ByteStream& in) override; int read_int64(int64_t& ret_value, common::ByteStream& in) override; @@ -1151,28 +1162,23 @@ class DoubleTS2DIFFDecoder : public TS2DIFFDecoder, int read_batch_double(double* out, int capacity, int& actual, common::ByteStream& in) override { - // Same split as FloatTS2DIFFDecoder::read_batch_float — see there. - ensure_layout(in, 8); - if (is_legacy_raw_) { - int64_t* buf = reinterpret_cast(out); - int ret = TS2DIFFDecoder::read_batch_int64(buf, capacity, - actual, in); - if (ret != common::E_OK) return ret; - for (int i = 0; i < actual; ++i) { - out[i] = common::long_to_double(buf[i]); - } - return common::E_OK; - } + // DOUBLE TS_2DIFF uses the same segment prefix. Bypassing + // read_double() misreads that prefix as a block header and can spin + // at end-of-input while decoding an otherwise valid page. + // skip_int32/skip_int64 are likewise unsupported here (see the + // float decoder note). return Decoder::read_batch_double(out, capacity, actual, in); } private: + bool is_legacy_raw_{false}; int max_point_number_{0}; double max_point_value_{1.0}; int segment_pos_{0}; int segment_size_{0}; std::vector underflow_bm_; std::vector overflow_bm_; + bool page_first_segment_{true}; }; typedef TS2DIFFDecoder IntTS2DIFFDecoder; @@ -1271,22 +1277,34 @@ FORCE_INLINE int FloatTS2DIFFDecoder::read_int64(int64_t& ret_value, FORCE_INLINE int FloatTS2DIFFDecoder::read_float(float& ret_value, common::ByteStream& in) { int ret = common::E_OK; - if (current_index_ == 0) { - ensure_layout(in, 4); - if (at_segment_prefix(in)) { - if (RET_FAIL( - ts2diff_java_detail::consume_float_double_ts2diff_prefix( - in, max_point_number_, underflow_bm_, overflow_bm_, - segment_size_))) { - return ret; - } + if (current_index_ == 0 && !is_legacy_raw_) { + bool max_pn_present = true; + ts2diff_java_detail::SegmentHeaderPreload preload; + if (RET_FAIL(ts2diff_java_detail::consume_float_double_ts2diff_prefix( + in, is_legacy_raw_, max_pn_present, max_point_number_, + underflow_bm_, overflow_bm_, segment_size_, page_first_segment_, + false, preload))) { + return ret; + } + // maxPointNumber is written once per page; later segments of + // the page reuse the first segment's scale factor. + if (max_pn_present) { max_point_value_ = max_point_number_ <= 0 ? 1.0 : std::pow(10.0, static_cast(max_point_number_)); - segment_pos_ = 0; - advance_segment_prefix(); } + // Prefix-free segments have their header parsed up front so that + // decode() can pick it up without re-reading the stream. + if (preload.ready) { + write_index_ = preload.write_index; + bit_width_ = preload.bit_width; + delta_min_ = static_cast(preload.delta_min); + first_value_ = static_cast(preload.first_value); + header_preloaded_ = true; + } + page_first_segment_ = false; + segment_pos_ = 0; } if (is_legacy_raw_) { ret_value = decode(in); @@ -1337,22 +1355,32 @@ FORCE_INLINE int DoubleTS2DIFFDecoder::read_float(float& ret_value, FORCE_INLINE int DoubleTS2DIFFDecoder::read_double(double& ret_value, common::ByteStream& in) { int ret = common::E_OK; - if (current_index_ == 0) { - ensure_layout(in, 8); - if (at_segment_prefix(in)) { - if (RET_FAIL( - ts2diff_java_detail::consume_float_double_ts2diff_prefix( - in, max_point_number_, underflow_bm_, overflow_bm_, - segment_size_))) { - return ret; - } + if (current_index_ == 0 && !is_legacy_raw_) { + bool max_pn_present = true; + ts2diff_java_detail::SegmentHeaderPreload preload; + if (RET_FAIL(ts2diff_java_detail::consume_float_double_ts2diff_prefix( + in, is_legacy_raw_, max_pn_present, max_point_number_, + underflow_bm_, overflow_bm_, segment_size_, page_first_segment_, + true, preload))) { + return ret; + } + // maxPointNumber is written once per page; later segments of + // the page reuse the first segment's scale factor. + if (max_pn_present) { max_point_value_ = max_point_number_ <= 0 ? 1.0 : std::pow(10.0, static_cast(max_point_number_)); - segment_pos_ = 0; - advance_segment_prefix(); } + if (preload.ready) { + write_index_ = preload.write_index; + bit_width_ = preload.bit_width; + delta_min_ = preload.delta_min; + first_value_ = preload.first_value; + header_preloaded_ = true; + } + page_first_segment_ = false; + segment_pos_ = 0; } if (is_legacy_raw_) { ret_value = decode(in); diff --git a/cpp/src/encoding/ts2diff_encoder.h b/cpp/src/encoding/ts2diff_encoder.h index fc494581a..a39ed6b5f 100644 --- a/cpp/src/encoding/ts2diff_encoder.h +++ b/cpp/src/encoding/ts2diff_encoder.h @@ -565,6 +565,7 @@ class FloatTS2DIFFEncoder : public TS2DIFFEncoder { void reset() override { TS2DIFFEncoder::reset(); underflow_flags_.clear(); + max_point_number_saved_ = false; } int flush(common::ByteStream& out_stream) override; int encode(bool value, common::ByteStream& out_stream); @@ -609,6 +610,11 @@ class FloatTS2DIFFEncoder : public TS2DIFFEncoder { int max_point_number_; double max_point_value_; std::vector underflow_flags_; + // Java FloatDecoder reads maxPointNumber once per page; this flag + // makes sure only the first 128-value segment of a page carries the + // prefix. PageWriter/ValuePageWriter reset() between pages clears it, + // so every page starts with a fresh prefix (apache/tsfile#910). + bool max_point_number_saved_{false}; }; class DoubleTS2DIFFEncoder : public TS2DIFFEncoder { @@ -623,6 +629,7 @@ class DoubleTS2DIFFEncoder : public TS2DIFFEncoder { void reset() override { TS2DIFFEncoder::reset(); underflow_flags_.clear(); + max_point_number_saved_ = false; } int flush(common::ByteStream& out_stream) override; int encode(bool value, common::ByteStream& out_stream); @@ -667,6 +674,11 @@ class DoubleTS2DIFFEncoder : public TS2DIFFEncoder { int max_point_number_; double max_point_value_; std::vector underflow_flags_; + // Java FloatDecoder reads maxPointNumber once per page; this flag + // makes sure only the first 128-value segment of a page carries the + // prefix. PageWriter/ValuePageWriter reset() between pages clears it, + // so every page starts with a fresh prefix (apache/tsfile#910). + bool max_point_number_saved_{false}; }; typedef TS2DIFFEncoder IntTS2DIFFEncoder; @@ -784,9 +796,14 @@ FORCE_INLINE int FloatTS2DIFFEncoder::flush(common::ByteStream& out_stream) { } const int num_values = write_index_ + 1; common::ByteStream inner(1024, common::MOD_TS2DIFF_OBJ, false); - if (RET_FAIL(common::SerializationUtil::write_var_uint( - static_cast(max_point_number_), inner))) { - return ret; + // Java FloatDecoder reads maxPointNumber only once per page; emit it + // just for the page's first segment (apache/tsfile#910). + if (!max_point_number_saved_) { + if (RET_FAIL(common::SerializationUtil::write_var_uint( + static_cast(max_point_number_), inner))) { + return ret; + } + max_point_number_saved_ = true; } SIMDOps::rebase(delta_arr_, delta_arr_min_, write_index_); int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_); @@ -871,9 +888,14 @@ FORCE_INLINE int DoubleTS2DIFFEncoder::flush(common::ByteStream& out_stream) { } const int num_values = write_index_ + 1; common::ByteStream inner(1024, common::MOD_TS2DIFF_OBJ, false); - if (RET_FAIL(common::SerializationUtil::write_var_uint( - static_cast(max_point_number_), inner))) { - return ret; + // Java FloatDecoder reads maxPointNumber only once per page; emit it + // just for the page's first segment (apache/tsfile#910). + if (!max_point_number_saved_) { + if (RET_FAIL(common::SerializationUtil::write_var_uint( + static_cast(max_point_number_), inner))) { + return ret; + } + max_point_number_saved_ = true; } SIMDOps::rebase(delta_arr_, delta_arr_min_, write_index_); int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_); diff --git a/cpp/test/encoding/ts2diff_codec_test.cc b/cpp/test/encoding/ts2diff_codec_test.cc index 35aa60d0f..dfb671909 100644 --- a/cpp/test/encoding/ts2diff_codec_test.cc +++ b/cpp/test/encoding/ts2diff_codec_test.cc @@ -680,4 +680,381 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyRawBatchThenScalarReads) { EXPECT_FALSE(decoder_float_->has_remaining(out_stream)); } -} // namespace storage + +// ============================================================================ +// apache/tsfile#910 regression: Java reads the maxPointNumber field only +// once per page (before the first segment); the old C++ encoder repeated it +// at every segment boundary, so multi-segment pages could be misparsed by +// Java readers (TsFileSketchTool / print-tsfile.bat crash). +// +// The fixed encoder emits maxPointNumber exactly once per page (a page +// boundary is signaled by reset()); later segments start directly with the +// 4-byte write_index. The decoder distinguishes the two layouts by peeking +// the byte at the segment start: 0x00 means "no prefix" (write_index high +// byte), any non-zero byte is a var_uint tag (maxPointNumber itself, or the +// overflow flag). +// ============================================================================ + +namespace { + +// LEB128 var_uint, matching common::SerializationUtil::write_var_uint. +bool parse_var_uint(const std::vector& b, size_t& pos, + uint32_t& out) { + if (pos >= b.size()) return false; + out = 0; + int shift = 0; + while (true) { + if (pos >= b.size() || shift > 28) return false; + uint8_t byte = b[pos++]; + out |= static_cast(byte & 0x7F) << shift; + if ((byte & 0x80) == 0) return true; + shift += 7; + } +} + +int32_t read_i32_be(const std::vector& b, size_t pos) { + return (static_cast(b[pos]) << 24) | + (static_cast(b[pos + 1]) << 16) | + (static_cast(b[pos + 2]) << 8) | + static_cast(b[pos + 3]); +} + +// Dumps the full stream content and CONSUMES the stream (read position +// moves to the end). Callers decode from a wrapped copy of the bytes: +// restoring the read position to a page-aligned offset (position 0) makes +// ByteStream::check_space() advance its page cursor one page too far and +// fail the next read, so no rewind is attempted here. +std::vector byte_stream_bytes(common::ByteStream& stream) { + uint32_t size = stream.total_size(); + std::vector buf(size); + uint32_t read_len = 0; + EXPECT_EQ(stream.read_buf(buf.data(), size, read_len), common::E_OK); + EXPECT_EQ(read_len, size); + return buf; +} + +// Wraps dumped page bytes for decoding (same path production chunk readers +// use — a wrapped ByteStream). +void wrap_bytes(const std::vector& b, common::ByteStream& s) { + s.wrap_from(reinterpret_cast(b.data()), + static_cast(b.size())); +} + +// Walks a float/double TS_2DIFF page and asserts the apache/tsfile#910 +// layout invariant: the maxPointNumber var_uint appears only on the page's +// first segment (possibly after a leading overflow-marker section); every +// later segment starts directly with its 4-byte write_index, so any +// non-flag tag on a later segment is a regression. Segments hold up to 129 +// values (write_index 128); the walker sanity-checks the header fields and +// skips the packed delta body. +void expect_max_pn_once_per_page(const std::vector& b, + bool is_double) { + const uint32_t FLAG_SCALED = 2147483647u; + const uint32_t FLAG_ORIGINAL = 2147483646u; + size_t pos = 0; + bool first_segment = true; + int segment_count = 0; + while (pos < b.size()) { + size_t seg_start = pos; + if (pos < b.size() && b[pos] != 0x00) { + uint32_t tag = 0; + size_t p = pos; + ASSERT_TRUE(parse_var_uint(b, p, tag)); + if (tag == FLAG_SCALED || tag == FLAG_ORIGINAL) { + // Overflow marker section: value count + underflow bitmap + // (+ overflow bitmap for original-value overflow). + uint32_t n = 0; + ASSERT_TRUE(parse_var_uint(b, p, n)); + EXPECT_GE(n, 1u); + size_t bm_len = static_cast(n / 8 + 1); + ASSERT_LE(p + bm_len, b.size()); + p += bm_len; + if (tag == FLAG_ORIGINAL) { + ASSERT_LE(p + bm_len, b.size()); + p += bm_len; + } + // Only the page's first segment may carry maxPointNumber + // after the bitmaps. + if (first_segment && p < b.size() && b[p] != 0x00) { + uint32_t mpn = 0; + ASSERT_TRUE(parse_var_uint(b, p, mpn)); + EXPECT_GE(mpn, 1u); + } + pos = p; + } else { + // A non-flag tag is the maxPointNumber prefix; it must not + // appear on any segment after the first. + EXPECT_TRUE(first_segment) + << "maxPointNumber prefix found on segment " + << segment_count + 1 << " (byte " << seg_start << ")"; + pos = p; + } + } + // Segment header: write_index + bit_width (+ delta_min + first_value). + size_t h = pos; + size_t header_len = is_double ? 24 : 16; + ASSERT_LE(h + header_len, b.size()); + int32_t wi = read_i32_be(b, h); + int32_t bw = read_i32_be(b, h + 4); + ASSERT_GE(wi, 0) << "negative write_index at segment " + << segment_count + 1; + EXPECT_LE(wi, 128); + ASSERT_GE(bw, 0); + EXPECT_LE(bw, 64); + pos = h + header_len; + pos += (static_cast(wi) * static_cast(bw) + 7) / 8; + ASSERT_LE(pos, b.size()); + first_segment = false; + segment_count++; + } + EXPECT_GE(segment_count, 2) << "test must produce a multi-segment page"; +} + +} // namespace + +// A page holds multiple 129-value segments; the maxPointNumber must appear +// exactly once, at the page start — not at every segment boundary. +TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberOncePerPageFloatMultiSegment) { + common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 400; // 4 segments: 129 + 129 + 129 + 13 + std::vector data(row_num); + for (int i = 0; i < row_num; i++) { + data[i] = static_cast(i) * 0.25f + 0.5f; + } + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(encoder_float_->encode(data[i], out), common::E_OK); + } + ASSERT_EQ(encoder_float_->flush(out), common::E_OK); + + // Ramp data has bit_width 0, so the only 0x02 byte in the page is the + // maxPointNumber prefix. + std::vector b = byte_stream_bytes(out); + size_t prefix_count = 0; + for (uint8_t byte : b) { + if (byte == 0x02) prefix_count++; + } + EXPECT_EQ(prefix_count, 1u) + << "maxPointNumber must be written once per page, not per segment"; + expect_max_pn_once_per_page(b, false); + + common::ByteStream dec; + wrap_bytes(b, dec); + float x = 0.0f; + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK); + EXPECT_FLOAT_EQ(x, data[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(dec)); +} + +// Same invariant for the double encoder (i64 delta path). +TEST_F(FloatDoubleTS2DIFFCodecTest, + MaxPointNumberOncePerPageDoubleMultiSegment) { + common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 400; + std::vector data(row_num); + for (int i = 0; i < row_num; i++) { + data[i] = static_cast(i) * 0.25 + 0.5; + } + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(encoder_double_->encode(data[i], out), common::E_OK); + } + ASSERT_EQ(encoder_double_->flush(out), common::E_OK); + + std::vector b = byte_stream_bytes(out); + size_t prefix_count = 0; + for (uint8_t byte : b) { + if (byte == 0x02) prefix_count++; + } + EXPECT_EQ(prefix_count, 1u); + expect_max_pn_once_per_page(b, true); + + common::ByteStream dec; + wrap_bytes(b, dec); + double y = 0.0; + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(decoder_double_->read_double(y, dec), common::E_OK); + EXPECT_DOUBLE_EQ(y, data[i]) << "row " << i; + } + EXPECT_FALSE(decoder_double_->has_remaining(dec)); +} + +// The #910 crash scenario: a value that overflows the scaled range in the +// first segment. The overflow-marker section leads the page, the single +// maxPointNumber follows it, and the second segment still has no prefix. +TEST_F(FloatDoubleTS2DIFFCodecTest, + MaxPointNumberOncePerPageFloatWithOverflow) { + common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 140; // segment 1 (129 values) + segment 2 (11 values) + std::vector data(row_num); + data[0] = 0.5f; + data[1] = 3.0e7f; // *100 = 3e9 > INT32_MAX → scaled overflow (flag 0) + for (int i = 2; i < row_num; i++) { + data[i] = 0.75f + static_cast(i - 2) * 0.25f; + } + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(encoder_float_->encode(data[i], out), common::E_OK); + } + ASSERT_EQ(encoder_float_->flush(out), common::E_OK); + + // Byte layout: [FLAG var_uint][n=129][underflow bitmap (17B)] + // [maxPointNumber 0x02][seg1 header][packed] + // [seg2 header starting with 0x00 — no prefix] + std::vector b = byte_stream_bytes(out); + size_t pos = 0; + uint32_t tag = 0; + ASSERT_TRUE(parse_var_uint(b, pos, tag)); + EXPECT_EQ(tag, ts2diff_java_detail::FLAG_SCALED_VALUE_OVERFLOW); + uint32_t n = 0; + ASSERT_TRUE(parse_var_uint(b, pos, n)); + EXPECT_EQ(n, 129u); // segment 1 value count + size_t bm_len = static_cast(n / 8 + 1); + ASSERT_LE(pos + bm_len, b.size()); + pos += bm_len; + // Exactly one maxPointNumber, directly after the bitmaps. + ASSERT_LT(pos, b.size()); + EXPECT_EQ(b[pos], 0x02); + uint32_t mpn = 0; + ASSERT_TRUE(parse_var_uint(b, pos, mpn)); + EXPECT_EQ(mpn, 2u); + // Segment 1 header: write_index == 128 (129 values). + ASSERT_LE(pos + 16, b.size()); + int32_t wi = read_i32_be(b, pos); + int32_t bw = read_i32_be(b, pos + 4); + EXPECT_EQ(wi, 128); + pos += 16; + pos += (static_cast(wi) * static_cast(bw) + 7) / 8; + ASSERT_LE(pos, b.size()); + // Segment 2 begins directly with its write_index (0x00 high byte). + EXPECT_EQ(b[pos], 0x00) << "segment 2 must not carry a maxPointNumber"; + + // Round-trip: the overflow value goes through the bitmap path. + common::ByteStream dec; + wrap_bytes(b, dec); + float x = 0.0f; + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK); + EXPECT_FLOAT_EQ(x, data[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(dec)); +} + +// Same overflow layout for double (scaled > INT64_MAX → 1.0e17 * 100). +// The generic walker understands the FLAG + maxPointNumber + segments +// structure; round-trip goes through the bitmap path. +TEST_F(FloatDoubleTS2DIFFCodecTest, + MaxPointNumberOncePerPageDoubleWithOverflow) { + common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false); + const int row_num = 140; + std::vector data(row_num); + data[0] = 0.5; + data[1] = 1.0e17; // *100 = 1e19 > INT64_MAX → scaled overflow (flag 0) + for (int i = 2; i < row_num; i++) { + data[i] = 0.75 + static_cast(i - 2) * 0.25; + } + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(encoder_double_->encode(data[i], out), common::E_OK); + } + ASSERT_EQ(encoder_double_->flush(out), common::E_OK); + + std::vector b = byte_stream_bytes(out); + expect_max_pn_once_per_page(b, true); + + common::ByteStream dec; + wrap_bytes(b, dec); + double y = 0.0; + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(decoder_double_->read_double(y, dec), common::E_OK); + EXPECT_DOUBLE_EQ(y, data[i]) << "row " << i; + } + EXPECT_FALSE(decoder_double_->has_remaining(dec)); +} + +// PageWriter resets the encoder between pages; every page must carry its +// own maxPointNumber prefix (exactly one per page). +TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberPerPageAfterReset) { + const int row_num = 130; // 129 + 1 → two segments per page + std::vector data(row_num); + for (int i = 0; i < row_num; i++) { + data[i] = static_cast(i) * 0.25f + 0.5f; + } + common::ByteStream page1(1024, common::MOD_TS2DIFF_OBJ, false); + common::ByteStream page2(1024, common::MOD_TS2DIFF_OBJ, false); + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(encoder_float_->encode(data[i], page1), common::E_OK); + } + ASSERT_EQ(encoder_float_->flush(page1), common::E_OK); + encoder_float_->reset(); // page boundary + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(encoder_float_->encode(data[i], page2), common::E_OK); + } + ASSERT_EQ(encoder_float_->flush(page2), common::E_OK); + + std::vector b1 = byte_stream_bytes(page1); + std::vector b2 = byte_stream_bytes(page2); + size_t c1 = 0; + size_t c2 = 0; + for (uint8_t byte : b1) { + if (byte == 0x02) c1++; + } + for (uint8_t byte : b2) { + if (byte == 0x02) c2++; + } + EXPECT_EQ(c1, 1u) << "page 1 must carry exactly one maxPointNumber"; + EXPECT_EQ(c2, 1u) << "page 2 must carry exactly one maxPointNumber"; + expect_max_pn_once_per_page(b1, false); + expect_max_pn_once_per_page(b2, false); + + // Both pages decode with the same decoder; PageReader calls reset() + // between pages, which must re-arm the per-page prefix state. + common::ByteStream d1; + common::ByteStream d2; + wrap_bytes(b1, d1); + wrap_bytes(b2, d2); + float x = 0.0f; + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(decoder_float_->read_float(x, d1), common::E_OK); + EXPECT_FLOAT_EQ(x, data[i]) << "page1 row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(d1)); + decoder_float_->reset(); // page boundary + for (int i = 0; i < row_num; i++) { + ASSERT_EQ(decoder_float_->read_float(x, d2), common::E_OK); + EXPECT_FLOAT_EQ(x, data[i]) << "page2 row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(d2)); +} + +// Backward compatibility: files written by the pre-#910 encoder carry the +// maxPointNumber at every segment boundary. The decoder must keep reading +// them (it rescales whenever a prefix is present instead of assuming +// once-per-page). +TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyPerSegmentMaxPNStillDecodes) { + // Build an old-format page by hand: 0x02 prefix before BOTH segments. + const std::vector expected = {0.5f, 0.75f, 1.0f, 1.25f, + 1.5f, 1.75f, 2.0f, 2.25f}; + common::ByteStream old_fmt(1024, common::MOD_TS2DIFF_OBJ, false); + // Segment 1: 6 values (first 50, five deltas of 25), bit_width 0. + ASSERT_EQ(common::SerializationUtil::write_var_uint(2, old_fmt), + common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(5, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(0, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(25, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(50, old_fmt), common::E_OK); + // Segment 2: 2 values (first 200, one delta of 25) — WITH prefix again. + ASSERT_EQ(common::SerializationUtil::write_var_uint(2, old_fmt), + common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(1, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(0, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(25, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(200, old_fmt), common::E_OK); + + float x = 0.0f; + for (size_t i = 0; i < expected.size(); i++) { + ASSERT_EQ(decoder_float_->read_float(x, old_fmt), common::E_OK); + EXPECT_FLOAT_EQ(x, expected[i]) << "row " << i; + } + EXPECT_FALSE(decoder_float_->has_remaining(old_fmt)); +} + +} // namespace storage \ No newline at end of file From a66a6796cdbc98bb59336bb65e3e5a6a5f0a762d Mon Sep 17 00:00:00 2001 From: gx Date: Wed, 19 Aug 2026 21:50:31 +0800 Subject: [PATCH 4/4] style(cpp): fix spotless clang-format violations in ts2diff files --- cpp/src/encoding/ts2diff_decoder.h | 6 ++---- cpp/test/encoding/ts2diff_codec_test.cc | 14 +++++++------- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h index bbf1c310b..1175cf4b9 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -273,8 +273,7 @@ inline int read_var_uint_tail(uint8_t first_byte, common::ByteStream& in, // first_value) forward-only. `wi_hi` is the first (already consumed) byte // of the big-endian write_index - always 0x00 for the no-prefix layout. inline int read_segment_header_preload(common::ByteStream& in, bool is_double, - uint8_t wi_hi, - SegmentHeaderPreload& h) { + uint8_t wi_hi, SegmentHeaderPreload& h) { int ret = common::E_OK; uint8_t rest[3] = {0, 0, 0}; uint32_t read_len = 0; @@ -402,8 +401,7 @@ inline int consume_float_double_ts2diff_prefix( if (after_bm_byte == 0x00) { max_pn_present = false; if (RET_FAIL(read_segment_header_preload(in, is_double, - after_bm_byte, - preload))) { + after_bm_byte, preload))) { return ret; } return common::E_OK; diff --git a/cpp/test/encoding/ts2diff_codec_test.cc b/cpp/test/encoding/ts2diff_codec_test.cc index dfb671909..07fdd94a3 100644 --- a/cpp/test/encoding/ts2diff_codec_test.cc +++ b/cpp/test/encoding/ts2diff_codec_test.cc @@ -680,7 +680,6 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyRawBatchThenScalarReads) { EXPECT_FALSE(decoder_float_->has_remaining(out_stream)); } - // ============================================================================ // apache/tsfile#910 regression: Java reads the maxPointNumber field only // once per page (before the first segment); the old C++ encoder repeated it @@ -698,8 +697,7 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyRawBatchThenScalarReads) { namespace { // LEB128 var_uint, matching common::SerializationUtil::write_var_uint. -bool parse_var_uint(const std::vector& b, size_t& pos, - uint32_t& out) { +bool parse_var_uint(const std::vector& b, size_t& pos, uint32_t& out) { if (pos >= b.size()) return false; out = 0; int shift = 0; @@ -814,7 +812,8 @@ void expect_max_pn_once_per_page(const std::vector& b, // A page holds multiple 129-value segments; the maxPointNumber must appear // exactly once, at the page start — not at every segment boundary. -TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberOncePerPageFloatMultiSegment) { +TEST_F(FloatDoubleTS2DIFFCodecTest, + MaxPointNumberOncePerPageFloatMultiSegment) { common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false); const int row_num = 400; // 4 segments: 129 + 129 + 129 + 13 std::vector data(row_num); @@ -1031,8 +1030,8 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberPerPageAfterReset) { // once-per-page). TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyPerSegmentMaxPNStillDecodes) { // Build an old-format page by hand: 0x02 prefix before BOTH segments. - const std::vector expected = {0.5f, 0.75f, 1.0f, 1.25f, - 1.5f, 1.75f, 2.0f, 2.25f}; + const std::vector expected = {0.5f, 0.75f, 1.0f, 1.25f, + 1.5f, 1.75f, 2.0f, 2.25f}; common::ByteStream old_fmt(1024, common::MOD_TS2DIFF_OBJ, false); // Segment 1: 6 values (first 50, five deltas of 25), bit_width 0. ASSERT_EQ(common::SerializationUtil::write_var_uint(2, old_fmt), @@ -1047,7 +1046,8 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyPerSegmentMaxPNStillDecodes) { ASSERT_EQ(common::SerializationUtil::write_ui32(1, old_fmt), common::E_OK); ASSERT_EQ(common::SerializationUtil::write_ui32(0, old_fmt), common::E_OK); ASSERT_EQ(common::SerializationUtil::write_ui32(25, old_fmt), common::E_OK); - ASSERT_EQ(common::SerializationUtil::write_ui32(200, old_fmt), common::E_OK); + ASSERT_EQ(common::SerializationUtil::write_ui32(200, old_fmt), + common::E_OK); float x = 0.0f; for (size_t i = 0; i < expected.size(); i++) {