From 7effad7113bc1b01bf3b8abbcdda7ca21498afb5 Mon Sep 17 00:00:00 2001 From: gx Date: Fri, 7 Aug 2026 17:27:51 +0800 Subject: [PATCH 1/8] 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/8] 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/8] 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/8] 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++) { From 5a84843937ff3e93a5c46a6dbf7159a0260d1844 Mon Sep 17 00:00:00 2001 From: gx Date: Sat, 22 Aug 2026 18:59:11 +0800 Subject: [PATCH 5/8] fix(windows): open file paths as UTF-8 via _wopen (port of vendored patch) CRT ::open interprets bytes in the active code page; UTF-8 paths with non-ASCII characters fail with E_FILE_OPEN_ERR (28) on machines where the 8.3-shortpath / ACP-transcode workarounds unavailable (8dot3 disabled on the volume, or ACP cannot represent the characters). file_internal::open_utf8 converts UTF-8 -> wide chars -> _wopen, same as the vendored TsFileCpp tree. Applied to ReadFile::open, WriteFile, and RestorableTsFileIOWriter's self-check reader. --- cpp/src/file/read_file.cc | 3 +- cpp/src/file/restorable_tsfile_io_writer.cc | 3 +- cpp/src/file/utf8_file_open.h | 69 +++++++++++++++++++++ cpp/src/file/write_file.cc | 3 +- 4 files changed, 75 insertions(+), 3 deletions(-) create mode 100644 cpp/src/file/utf8_file_open.h diff --git a/cpp/src/file/read_file.cc b/cpp/src/file/read_file.cc index 7d41d7095..e790ac722 100644 --- a/cpp/src/file/read_file.cc +++ b/cpp/src/file/read_file.cc @@ -34,6 +34,7 @@ ssize_t pread(int fd, void* buf, size_t count, uint64_t offset); #include "common/logger/elog.h" #include "common/tsfile_common.h" +#include "file/utf8_file_open.h" #include "utils/util_define.h" // ssize_t and other platform-compat shims using namespace common; @@ -54,7 +55,7 @@ int ReadFile::open(const std::string& file_path) { #ifdef _WIN32 flags |= O_BINARY; #endif - fd_ = ::open(file_path_.c_str(), flags); + fd_ = file_internal::open_utf8(file_path_, flags); if (fd_ < 0) { std::cerr << "open file " << file_path << " error: " << strerror(errno) << " (errno " << errno << ")" << std::endl; diff --git a/cpp/src/file/restorable_tsfile_io_writer.cc b/cpp/src/file/restorable_tsfile_io_writer.cc index a1fc53402..a0b0d7313 100644 --- a/cpp/src/file/restorable_tsfile_io_writer.cc +++ b/cpp/src/file/restorable_tsfile_io_writer.cc @@ -18,6 +18,7 @@ */ #include "file/restorable_tsfile_io_writer.h" +#include "file/utf8_file_open.h" #include @@ -96,7 +97,7 @@ struct SelfCheckReader { #ifdef _WIN32 fd_ = ::_open(path.c_str(), _O_RDONLY | _O_BINARY); #else - fd_ = ::open(path.c_str(), O_RDONLY); + fd_ = file_internal::open_utf8(path, O_RDONLY); #endif if (fd_ < 0) { return E_FILE_OPEN_ERR; diff --git a/cpp/src/file/utf8_file_open.h b/cpp/src/file/utf8_file_open.h new file mode 100644 index 000000000..7148fe592 --- /dev/null +++ b/cpp/src/file/utf8_file_open.h @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include +#include + +#ifdef _WIN32 +#include +#include +#else +#include +#endif + +namespace storage { +namespace file_internal { + +inline int open_utf8(const std::string& path, int flags, int mode = 0) { +#ifdef _WIN32 + if (path.find('\0') != std::string::npos || path.size() > INT_MAX) { + errno = EINVAL; + return -1; + } + if (path.empty()) { + errno = ENOENT; + return -1; + } + const int size = MultiByteToWideChar(CP_UTF8, MB_ERR_INVALID_CHARS, + path.data(), + static_cast(path.size()), nullptr, + 0); + if (size <= 0) { + errno = EINVAL; + return -1; + } + std::wstring wide_path(static_cast(size), L'\0'); + if (MultiByteToWideChar(CP_UTF8, MB_ERR_INVALID_CHARS, path.data(), + static_cast(path.size()), &wide_path[0], + size) != size) { + errno = EINVAL; + return -1; + } + return ::_wopen(wide_path.c_str(), flags, mode); +#else + return ::open(path.c_str(), flags, mode); +#endif +} + +} // namespace file_internal +} // namespace storage diff --git a/cpp/src/file/write_file.cc b/cpp/src/file/write_file.cc index 68ac127ac..c2eb58ce2 100644 --- a/cpp/src/file/write_file.cc +++ b/cpp/src/file/write_file.cc @@ -18,6 +18,7 @@ */ #include "write_file.h" +#include "file/utf8_file_open.h" #include #include @@ -59,7 +60,7 @@ int WriteFile::do_create(int flags, mode_t mode) { flags |= O_BINARY; #endif // TODO make sure no same file exists - fd_ = ::open(path_.c_str(), flags, mode); + fd_ = file_internal::open_utf8(path_, flags, mode); if (fd_ < 0) { // log_err("open file error, path=%s, errno=%d", path_.c_str(), errno); ret = E_FILE_OPEN_ERR; From 6147431bcfbec19b880419f92446312e259aacf2 Mon Sep 17 00:00:00 2001 From: gx Date: Sat, 22 Aug 2026 19:26:48 +0800 Subject: [PATCH 6/8] fix: place utf8_file_open.h after the tsfile headers in restorable writer windows.h from utf8_file_open.h before decoder_factory.h made INT32/ DATE/DOUBLE ambiguous with using-namespace common in the decoder switch. --- cpp/src/file/restorable_tsfile_io_writer.cc | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cpp/src/file/restorable_tsfile_io_writer.cc b/cpp/src/file/restorable_tsfile_io_writer.cc index a0b0d7313..347caece8 100644 --- a/cpp/src/file/restorable_tsfile_io_writer.cc +++ b/cpp/src/file/restorable_tsfile_io_writer.cc @@ -18,7 +18,6 @@ */ #include "file/restorable_tsfile_io_writer.h" -#include "file/utf8_file_open.h" #include @@ -45,6 +44,7 @@ ssize_t pread(int fd, void* buf, size_t count, uint64_t offset); #include #endif +#include "file/utf8_file_open.h" using namespace common; namespace storage { From d377de649d4a75d31ea3bac44896cd7e25bde2b9 Mon Sep 17 00:00:00 2001 From: gx Date: Sat, 22 Aug 2026 20:40:02 +0800 Subject: [PATCH 7/8] fix(reader): report the file's actual encoding/compression in schema get_timeseries_schema built MeasurementSchema with the 2-arg ctor, whose encoding/compression are library defaults (DOUBLE->GORILLA, LZ4) rather than what the file stores. Take both from the first ChunkMeta of the timeseries (chunk metadata is deserialized from the file), falling back to defaults when no chunk metadata is available. --- cpp/src/reader/tsfile_reader.cc | 31 ++++++++++++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc index fb5f8fd92..865f925c6 100644 --- a/cpp/src/reader/tsfile_reader.cc +++ b/cpp/src/reader/tsfile_reader.cc @@ -438,9 +438,34 @@ int TsFileReader::get_timeseries_schema( dt = aligned->value_ts_idx_->get_data_type(); } } - MeasurementSchema ms( - timeseries_index->get_measurement_name().to_std_string(), dt); - result.push_back(ms); + // Report the encoding/compression the file actually stores + // (from the chunk metadata), not library defaults. The 2-arg + // MeasurementSchema ctor fills get_value_encoder(dt) / + // get_default_compressor(), which mislabels e.g. a TS_2DIFF + // UNCOMPRESSED column as GORILLA/LZ4. + common::TSEncoding enc = common::INVALID_ENCODING; + common::CompressionType comp = common::INVALID_COMPRESSION; + auto* chunk_meta_list = timeseries_index->get_chunk_meta_list(); + if (chunk_meta_list != nullptr && chunk_meta_list->size() > 0) { + ChunkMeta* first_meta = chunk_meta_list->front(); + if (first_meta != nullptr) { + enc = first_meta->encoding_; + comp = first_meta->compression_type_; + } + } + if (enc == common::INVALID_ENCODING || + comp == common::INVALID_COMPRESSION) { + // No chunk metadata available: fall back to defaults. + MeasurementSchema ms( + timeseries_index->get_measurement_name().to_std_string(), + dt); + result.push_back(ms); + } else { + MeasurementSchema ms( + timeseries_index->get_measurement_name().to_std_string(), + dt, enc, comp); + result.push_back(ms); + } } } return E_OK; From 927e43e4a1624855004967b48877c398d4227621 Mon Sep 17 00:00:00 2001 From: gx Date: Sat, 22 Aug 2026 20:53:59 +0800 Subject: [PATCH 8/8] fix(reader): schema reads real encoding/compression from chunk header bytes ChunkMeta entries from the metadata index carry only offsets (C++ deserialization never fills encoding_/compression_type_, unlike Java), so the previous attempt read uninitialized memory. Now: read 256 bytes at the first chunk's offset_of_chunk_header_ and deserialize the ChunkHeader (encoding/compression live there). Adds TsFileIOReader::get_read_file(). --- cpp/src/file/tsfile_io_reader.h | 4 +++ cpp/src/reader/tsfile_reader.cc | 43 +++++++++++++++++++++++++-------- 2 files changed, 37 insertions(+), 10 deletions(-) diff --git a/cpp/src/file/tsfile_io_reader.h b/cpp/src/file/tsfile_io_reader.h index dc9069b5c..960532c46 100644 --- a/cpp/src/file/tsfile_io_reader.h +++ b/cpp/src/file/tsfile_io_reader.h @@ -90,6 +90,10 @@ class TsFileIOReader { std::string get_file_path() const { return read_file_->file_path(); } + // Raw read access for callers that need file bytes (e.g. parsing chunk + // headers at offsets from chunk metadata). + ReadFile* get_read_file() const { return read_file_; } + TsFileMeta* get_tsfile_meta() { load_tsfile_meta_if_necessary(); return &tsfile_meta_; diff --git a/cpp/src/reader/tsfile_reader.cc b/cpp/src/reader/tsfile_reader.cc index 865f925c6..3312e2024 100644 --- a/cpp/src/reader/tsfile_reader.cc +++ b/cpp/src/reader/tsfile_reader.cc @@ -20,6 +20,9 @@ #include +#include "file/read_file.h" +#include "common/tsfile_common.h" +#include "common/allocator/byte_stream.h" #include "common/schema.h" #include "filter/time_operator.h" #include "tsfile_executor.h" @@ -438,19 +441,39 @@ int TsFileReader::get_timeseries_schema( dt = aligned->value_ts_idx_->get_data_type(); } } - // Report the encoding/compression the file actually stores - // (from the chunk metadata), not library defaults. The 2-arg - // MeasurementSchema ctor fills get_value_encoder(dt) / - // get_default_compressor(), which mislabels e.g. a TS_2DIFF - // UNCOMPRESSED column as GORILLA/LZ4. + // Report the encoding/compression the file actually stores, + // not library defaults. The 2-arg MeasurementSchema ctor fills + // get_value_encoder(dt) / get_default_compressor(), which + // mislabels e.g. a TS_2DIFF UNCOMPRESSED column as GORILLA/LZ4. + // ChunkMeta entries deserialized by the metadata index carry + // only offsets (their encoding_ fields are uninitialized), so + // read the first chunk's header bytes from the file at that + // offset — the ChunkHeader serialization carries the real + // encoding/compression. common::TSEncoding enc = common::INVALID_ENCODING; common::CompressionType comp = common::INVALID_COMPRESSION; auto* chunk_meta_list = timeseries_index->get_chunk_meta_list(); - if (chunk_meta_list != nullptr && chunk_meta_list->size() > 0) { - ChunkMeta* first_meta = chunk_meta_list->front(); - if (first_meta != nullptr) { - enc = first_meta->encoding_; - comp = first_meta->compression_type_; + if (chunk_meta_list != nullptr && chunk_meta_list->size() > 0 && + chunk_meta_list->front() != nullptr) { + const int64_t chunk_header_offset = + chunk_meta_list->front()->offset_of_chunk_header_; + ReadFile* rf = tsfile_executor_->get_tsfile_io_reader() + ->get_read_file(); + if (rf != nullptr && chunk_header_offset >= 0) { + char buf[256]; + int32_t read_len = 0; + if (rf->read(chunk_header_offset, buf, sizeof(buf), + read_len) == E_OK && + read_len > 0) { + common::ByteStream in( + read_len, common::MOD_TSFILE_READER, false); + in.wrap_from(buf, read_len); + ChunkHeader ch; + if (ch.deserialize_from(in) == E_OK) { + enc = ch.encoding_type_; + comp = ch.compression_type_; + } + } } } if (enc == common::INVALID_ENCODING ||