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 206b7f559..1175cf4b9 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -239,19 +239,123 @@ inline bool looks_like_ts2diff_header(common::ByteStream& in) { 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; + } + if (shift > 28) { + return common::E_TSFILE_CORRUPTED; + } + 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; + } + if (RET_FAIL(common::SerializationUtil::read_i64(h.first_value, in))) { + return ret; + } + } else { + int32_t dm = 0; + int32_t fv = 0; + if (RET_FAIL(common::SerializationUtil::read_i32(dm, in))) { + return ret; + } + if (RET_FAIL(common::SerializationUtil::read_i32(fv, in))) { + return ret; + } + h.delta_min = dm; + h.first_value = fv; + } + h.ready = true; + return common::E_OK; +} + inline int consume_float_double_ts2diff_prefix( - common::ByteStream& in, bool& is_legacy_raw, 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 || @@ -263,7 +367,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)) { @@ -278,25 +381,55 @@ 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; } - // Distinguish Java maxPointNumber prefix from legacy raw C++ block. + // 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 // ============================================================================ @@ -319,6 +452,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 { @@ -403,6 +537,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}; }; // ============================================================================ @@ -413,9 +550,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; @@ -442,9 +585,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; @@ -939,6 +1086,21 @@ inline int TS2DIFFDecoder::skip_int32(int count, int& skipped, 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); @@ -952,15 +1114,13 @@ 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. + // 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: @@ -971,11 +1131,22 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder { int segment_size_{0}; std::vector underflow_bm_; std::vector overflow_bm_; + bool page_first_segment_{true}; }; 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); @@ -989,15 +1160,12 @@ 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. + // skip_int32/skip_int64 are likewise unsupported here (see the + // float decoder note). + return Decoder::read_batch_double(out, capacity, actual, in); } private: @@ -1008,6 +1176,7 @@ class DoubleTS2DIFFDecoder : public TS2DIFFDecoder { int segment_size_{0}; std::vector underflow_bm_; std::vector overflow_bm_; + bool page_first_segment_{true}; }; typedef TS2DIFFDecoder IntTS2DIFFDecoder; @@ -1106,16 +1275,33 @@ 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) { + 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_point_number_, underflow_bm_, - overflow_bm_, segment_size_))) { + in, is_legacy_raw_, max_pn_present, max_point_number_, + underflow_bm_, overflow_bm_, segment_size_, page_first_segment_, + false, preload))) { return ret; } - max_point_value_ = - max_point_number_ <= 0 - ? 1.0 - : std::pow(10.0, static_cast(max_point_number_)); + // 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_)); + } + // 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_) { @@ -1167,16 +1353,31 @@ 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) { + 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_point_number_, underflow_bm_, - overflow_bm_, segment_size_))) { + in, is_legacy_raw_, max_pn_present, max_point_number_, + underflow_bm_, overflow_bm_, segment_size_, page_first_segment_, + true, preload))) { return ret; } - max_point_value_ = - max_point_number_ <= 0 - ? 1.0 - : std::pow(10.0, static_cast(max_point_number_)); + // 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_)); + } + 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_) { 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 fb997103c..07fdd94a3 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; @@ -480,4 +523,538 @@ TEST(FloatTS2DIFFEncoderResetTest, ResetClearsUnderflowFlags) { } } -} // namespace storage +// 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)); +} + +// ============================================================================ +// 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