You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tsfile.apache.org by co...@apache.org on 2026/09/17 06:30:46 UTC
(tsfile) 03/24: fix(cpp): TS_2DIFF float/double maxPointNumber once per page (fixes #910) (#901)
This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch feat/native-package-artifacts
in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit fdf5efb53dc1be7f5d015f72929ed79b5f2a7f66
Author: kkzi <34...@users.noreply.github.com>
AuthorDate: Tue Sep 1 10:30:50 2026 +0800
fix(cpp): TS_2DIFF float/double maxPointNumber once per page (fixes #910) (#901)
* fix(cpp): handle TS2DIFF float prefixes in batch decode
* 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).
* fix(cpp): write maxPointNumber once per page in TS_2DIFF float/double (apache/tsfile#910)
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.
* style(cpp): fix spotless clang-format violations in ts2diff files
* test(compat): extend matrix with FLOAT/DOUBLE TS_2DIFF and 300-row fixtures
Extend the Java and C++ encoding/compression compatibility matrices with
FLOAT + TS_2DIFF and DOUBLE + TS_2DIFF cases and raise every case to 300
rows so pages cross the 129-value TS_2DIFF block boundary (129+129+42),
per review feedback on apache/tsfile#901.
The TS_2DIFF value set covers all page layouts the writers can produce:
scaled integers, scale-overflow values (reachable at maxPointNumber 2,
the C++ writer), and raw IEEE bit patterns (NaN/Infinity, two page-wide
bitmaps). Every chosen value restores identically whether the writer used
maxPointNumber 0 (Java builder default) or 2 (historical C++ default), so
the validating reader never needs to know which writer produced a file;
NaN expectations use the canonical Java floatToIntBits pattern.
Expected values are computed by applying the writer's tri-state
conversion rules, not by reusing the input bits, so non-integer inputs
would not round-trip exactly and are excluded.
Also add a wire-format contract document derived from the Java
FloatEncoder/FloatDecoder/DeltaBinaryEncoder reference implementations
(cpp/docs/ts2diff-float-double-wire-format.md), which the upcoming
encoder/decoder rework will be validated against.
Current state: the six new TS_2DIFF float/double cases fail on the C++
side (write_table returns E_INVALID_ARG and decoded values are
misaligned), which is the acceptance baseline the rework must turn green.
* fix(cpp): emit page-wide TS_2DIFF float metadata and per-delta bit width
Align the C++ FLOAT/DOUBLE TS_2DIFF page layout with the Java canonical
format (apache/tsfile#901 review):
- The integer encoder's automatic 129-value block flush now emits a
plain integer block into an internal page buffer; overflow flags and
buffered blocks survive across the boundary. The page-seal flush
emits the page metadata once ([overflow marker][pageValueCount]
[page-wide bitmap(s)][maxPointNumber]) followed by all buffered
blocks, replacing the per-block wrapper metadata.
- Bit width is now the maximum width over the raw deltas rebased by
min, mirroring Java calculateBitWidthsForDeltaBlockBuffer, instead of
the width of (max - min). When raw deltas wrap the signed type
(e.g. adjacent raw IEEE bit patterns), (max - min) wrapped negative
and the block was silently written with bit width 0, discarding every
delta.
- The integer flush now calls the base reset() explicitly so the float
wrapper's page-scoped state is not cleared by the virtual dispatch
during mid-page block flushes.
Verified: Java reads all 30 non-LZMA2 C++ fixtures (including
FLOAT/DOUBLE TS_2DIFF at 300 rows across the block boundary with
NaN/Infinity raw-bit and scale-overflow pages) bit-exactly; the C++
Java-hex golden tests pass. The 15 remaining generate failures are the
LZMA2 compression path failing on this MSVC Debug build regardless of
encoding (also reproducible with CHIMP + LZMA2 on develop), tracked
separately.
* fix(cpp): decode FLOAT/DOUBLE TS_2DIFF pages with page-wide state only
Replace the multi-layout sniffing decoder with a single Java-grammar
state machine (apache/tsfile#901 review):
- Page metadata ([overflow marker][pageValueCount][page-wide bitmap(s)]
[maxPointNumber], or bare [maxPointNumber]) is parsed exactly once per
page and the bitmaps plus page position survive block transitions, so
Java multi-block overflow pages decode correctly.
- maxPointNumber = 0 (page starting with 0x00, the Java Ts2Diff builder
default) is a valid Form 1 page, no longer misdetected as a legacy raw
payload.
- The raw bit-cast layout and the pre-#910 per-segment maxPointNumber
layout are rejected as format errors instead of being decoded by
heuristic detection; legacy tests now assert fail-fast behavior.
- Block headers are validated (write_index in [0,128], bit_width in
range) and a truncated header now fails instead of silently reusing
stale state, which previously let batch readers spin forever on
out-of-format input.
FLOAT/DOUBLE batch reads reuse the integer batch decoder (SIMD fast
path) and apply the page-wide bitmaps afterwards per page position.
Verified all four compatibility directions on the extended matrix
(30 non-LZMA2 cases each): C++/Java readers on C++/Java writers,
including FLOAT/DOUBLE TS_2DIFF at 300 rows across the block boundary
with scaled-overflow and raw-bit pages.
* fix(cpp): align default float TS_2DIFF maxPointNumber with Java (0)
The Java Ts2Diff TSEncodingBuilder hard-codes maxPointNumber = 0 for
FLOAT/DOUBLE, so the C++ FloatTS2DIFFEncoder/DoubleTS2DIFFEncoder now
default to the same value instead of 2. The wire value is
self-describing, but with both writers sharing the default the pages are
byte-identical and the C++ writer can no longer produce the
scale-overflow form (Form 2), which is unreachable at maxPointNumber 0 -
any overflow is a value overflow and takes the raw-bits path.
Follow-ups in the same commit:
- Java hex goldens regenerated with maxPointNumber 0.
- The 0x02-byte-counting assertions are replaced by a structural page
walker (metadata once, then a continuous well-formed block stream);
byte counting cannot distinguish the 0x00 mpn byte from block-header
high bytes.
- Ramp data in round-trip tests integerized so expectations hold under
the default mpv = 1.
- Compatibility-test constants and the wire-format doc updated, with a
note that Form 2 pages can only originate from writers configured
with mpn > 0.
Verified: full C++ suite 784/787 pass; all four compatibility directions
green on the 30 non-LZMA2 cases.
* fix(cpp): address 8/27 review of FLOAT/DOUBLE TS_2DIFF (#901)
Five inline review comments plus hardening found while re-auditing
the fixes.
Review item 1 - ByteStream::check_space() revert: the TS_2DIFF decoder
is forward-only now, so the page-boundary recomputation is no longer
needed and its O(pages) walk per boundary made sequential traversal
quadratic. Restored next_-advance; the codec test hex helper no longer
rewinds via set_read_pos.
Review items 2+3 - maxPointNumber default: TSEncodingBuilder.Ts2Diff
initializes 0, but the standard schema write path calls initFromProps(),
which substitutes TSFileConfig.floatPrecision (default 2) when the
max_point_number property is absent. Restored the C++ encoder default
to 2, corrected the wire-format doc, restored the mpn=2 test shapes
(hex goldens, ramp data, Form 2 overflow), and kept mpn=0 as an
explicit fixture: a set_max_point_number() encoder path, an
MaxPointNumberZeroPagePrefix test, and "*.mpn0" Java fixtures that
carry max_point_number=0 props and exercise the canonical 0x00 page
prefix in both readers.
While validating the mpn0 fixture, found and fixed a pre-existing
MeasurementSchema::deserialize_from bug: the props loop iterated over
props_.size() (always 0 at that point) instead of the deserialized
count, so Java-written (key, value) props pairs were never consumed
and desynced the TsFileMeta bloom filter (E_TSFILE_CORRUPTED).
Review item 4 - maxPointNumber bound: no format-level basis for
rejecting mpn > 100; Java accepts any varint the stream carries
(Math.pow overflow yields +inf, values then take the raw-bits form).
Removed both > 100 checks; MaxPointNumberAboveLegacyBoundDecodes covers
mpn = 1000.
Review item 5 - writeIndex bound: 128 is DeltaBinaryEncoder's
BLOCK_DEFAULT_SIZE, not a wire limit; Java exposes block-size
constructors. read_header now validates writeIndex against availability
(int64 arithmetic) instead of the encoder default; the skip paths gained
the same bound plus a grouped header-read check (truncated headers used
to act on stale stack values); peek_next_block_range_int64 returns
false instead of an error code that converted to bool true;
skip_peeked_block_int64 widened its byte computation to int64;
LargeBlockBeyondDefaultSizeDecodes covers a 200-value block through
scalar and batch paths.
Rounding - convert_float_to_int/convert_double_to_long now use Java
Math.round semantics (floor(x + 0.5), ties towards +infinity) instead
of std::lround (ties away from zero): -0.125 * 100 = -12.5 stores as
-12, not -13, matching the Java writer byte for byte. The conversion
saturates at the integer limits (2^63 boundary included, where a plain
static_cast is UB); JavaRoundNegativeHalfTies* cover the behavior.
Hardening from self-review: restored the read_long end-of-input bailout
that 67ab361 dropped - with the writeIndex bound gone, a truncated page
whose header passes the availability check could spin the scalar path
forever (reproduced with a 17-byte page under a watchdog;
ScalarReadTerminatesOnTruncatedBlock pins it); SkipRejectsTruncated
BlockHeader pins the grouped skip check.
Verified: full C++ suite 803 passed / 3 skipped (env-gated compat);
compat matrix green in all four directions (Java<->C++, including the
mpn0 fixtures); clang-format 17.0.6 clean. LZMA2 cases excluded locally
due to the known separate Windows/MSVC issue on develop.
* fix(cpp): propagate TS_2DIFF read errors and close two int32 overflows
Follow-up to the 8/27 review fixes, found while re-auditing them.
Page-value-count bound overflowed int32. read_header compares
writeIndex + 1 against the page value count supplied by the float/double
page metadata; a zero-width block occupies no packed bytes, so the
availability check alone cannot bound writeIndex and this is the only
bound that applies. For writeIndex == INT32_MAX the addition wrapped
negative and silently passed the compare, so the bound never fired on
exactly the input it was added for. Now computed in int64.
RejectsBlockBeyondPageValueCount pins it.
peek_next_block_range_int64 still computed packed_bytes in int32 while
the skip paths had already been widened. writeIndex * bitWidth exceeds
INT32_MAX on a large page, and the result is used as a raw-pointer
offset for the look-ahead read. Widened to int64 to match skip.
Error propagation: the scalar and batch paths discarded the return of
read_i32/read_i64 for delta_min/first_value and handed back whatever
decode() produced, and read_int32/read_int64/read_float/read_double
returned E_OK unconditionally. A truncated page therefore surfaced as
plausible-looking values instead of an error. Failures now latch into
read_error_ and propagate out of the read_* entry points, and
read_page_meta returns E_TSFILE_CORRUPTED on a short varint instead of
the raw read code. ScalarReadRejectsTruncatedFixedFields covers it.
LegacyPerSegmentMaxPNRejected asserted the old poison-value behavior
(counting mismatches), which no longer occurs now that the decoder
returns an error instead. Rewritten to assert the invariant directly:
segment 1 still decodes, and the out-of-format continuation is never
handed back as valid data.
Verified: full C++ suite 805 passed / 3 skipped of 808 (the 3 skips are
env-gated: 2 compat fixtures plus the external dataset index); TS_2DIFF
suites 40/40; clang-format 17.0.6 clean. LZMA2 compat cases could not be
covered locally (this build has ENABLE_LZMA2=OFF, the known separate
Windows/MSVC issue on develop) and are left to CI.
* fix(cpp): harden TS_2DIFF decoder validation and error reporting
Address the remaining review findings on the float/double TS_2DIFF work:
- read_page_meta: fix the bitmap-length guard inequality; the bytes the
page can supply bound each bitmap at remaining/bitmap_count, not
bitmap_count*remaining (4x too loose before).
- read_header / skip paths: compute the packed-block byte count in
int64 and fall back to per-value decoding when it exceeds INT32_MAX,
instead of narrowing through int32 arithmetic.
- Format violations now return E_DECODE_ERR (matching the RLE/Sprintz/
RLBE convention); short reads propagate the underlying stream error
(E_PARTIAL_READ) instead of being rebranded as E_TSFILE_CORRUPTED.
- Remove the write-only header_error_ member and the unused
cal_bit_width(); converge the duplicated FLAG_* overflow-marker
constants into ts2diff_wire_format.h.
- Drop iteration-trace comments; describe current behavior instead.
- Tests updated to the new error codes.
---------
Co-authored-by: gx <xu...@atomdatatech.com>
---
cpp/docs/ts2diff-float-double-wire-format.md | 149 ++++
cpp/src/common/schema.h | 16 +-
cpp/src/encoding/ts2diff_decoder.h | 657 +++++++++++-----
cpp/src/encoding/ts2diff_encoder.h | 302 +++++---
cpp/src/encoding/ts2diff_wire_format.h | 39 +
cpp/test/encoding/ts2diff_codec_test.cc | 850 ++++++++++++++++++++-
...odel_encoding_compression_compatibility_test.cc | 177 ++++-
...eModelEncodingCompressionCompatibilityTest.java | 138 +++-
8 files changed, 1993 insertions(+), 335 deletions(-)
diff --git a/cpp/docs/ts2diff-float-double-wire-format.md b/cpp/docs/ts2diff-float-double-wire-format.md
new file mode 100644
index 000000000..6ef923d5d
--- /dev/null
+++ b/cpp/docs/ts2diff-float-double-wire-format.md
@@ -0,0 +1,149 @@
+<!-- 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. -->
+
+# FLOAT/DOUBLE TS_2DIFF Wire Format (Java Canonical Layout)
+
+This document specifies the canonical on-disk layout of FLOAT/DOUBLE TS_2DIFF
+pages, derived from the Java reference implementation
+(`FloatEncoder`, `FloatDecoder`, `DeltaBinaryEncoder`, `BitMap`).
+The Java layout is the cross-language compatibility boundary. Other layouts
+produced by earlier C++ writers (raw bit-cast, per-block wrapper metadata) are
+implementation artifacts outside the compatibility scope; the C++ decoder
+treats them as a format error.
+
+## Encoding Pipeline
+
+TS_2DIFF encodes integers. Floating-point values go through a wrapper that
+converts each value to an integer, encodes the integers with
+`IntDeltaEncoder` (FLOAT) or `LongDeltaEncoder` (DOUBLE), and emits page-wide
+conversion metadata.
+
+Given `maxPointNumber = mpn` and `maxPointValue = 10^mpn` (`mpn <= 0` implies
+`maxPointValue = 1`), each value maps to one of three stored forms:
+
+| Condition | Stored bits | Decoder action |
+| -------------------------------------- | --------------------------- | ------------------------- |
+| `round(v * 10^mpn)` fits the int type | `round(v * 10^mpn)` | divide by `10^mpn` |
+| scaled overflows but `v` itself fits | `round(v)` | divide by `1` |
+| `v` out of int range, or NaN | `floatToIntBits(v)` / `doubleToLongBits(v)` | restore raw bits |
+
+The three forms are tracked per page as a tri-state flag list
+(`underflowFlags` in Java):
+
+- `true` -> scaled form
+- `false` -> rounded form (scale overflow)
+- `null` -> raw IEEE 754 bits (value overflow or NaN)
+
+`round` is Java `Math.round` semantics, `floor(x + 0.5)` — ties go towards
+`+infinity` (`-2.5 -> -2`), unlike C `lround`'s ties-away-from-zero.
+
+## Page Layout
+
+```text
+# Form 1: every value stored in scaled form (no bitmap at all)
+[maxPointNumber varint]
+[TS_2DIFF block 1][TS_2DIFF block 2]...[final block]
+
+# Form 2: at least one value is 'false' (scale overflow), none is 'null'
+[Integer.MAX_VALUE varint] # 0xFF 0xFF 0xFF 0xFF 0x07
+[pageValueCount varint]
+[scaled-bitmap, pageValueCount/8+1 bytes] # marks 'true' entries
+[maxPointNumber varint]
+[TS_2DIFF block 1]...[final block]
+
+# Form 3: at least one value is 'null' (raw bits)
+[Integer.MAX_VALUE-1 varint] # 0xFF 0xFF 0xFF 0xFF 0x06
+[pageValueCount varint]
+[scaled-bitmap, pageValueCount/8+1 bytes] # marks 'true' entries
+[raw-bitmap, pageValueCount/8+1 bytes] # marks 'null' entries
+[maxPointNumber varint]
+[TS_2DIFF block 1]...[final block]
+```
+
+Key invariants:
+
+- `maxPointNumber` appears exactly once per page, before the first integer
+ block (for Forms 2/3 it appears after the bitmaps).
+- The bitmaps cover the entire page, not individual TS_2DIFF blocks. The
+ decoder keeps one page-wide `position` that never resets between blocks;
+ only a page-level `reset()` clears it.
+- Bitmap byte length is always `size/8 + 1`, even when `size % 8 == 0`
+ (`BitMap.getSizeOfBytes`).
+- Bitmap bit order is LSB-first within each byte: position `p` maps to
+ `bits[p / 8] & (1 << (p % 8))`.
+- `pageValueCount` counts all values of the page (across blocks).
+- A first-page byte of `0x00` is the normal encoding of `maxPointNumber = 0`
+ (an explicit `max_point_number=0` property on the Java side; reachable
+ but not the default — see Encoder Construction below). It is not a
+ legacy marker.
+- NaN handling is writer-dependent: Java `floatToIntBits` /
+ `doubleToLongBits` canonicalize any NaN to `0x7fc00000` /
+ `0x7ff8000000000000`, while the C++ encoder preserves the payload bits.
+ Both are valid raw-bit entries; readers restore the bits as stored.
+
+## Integer Block Layout
+
+Identical to the integer TS_2DIFF format (`DeltaBinaryEncoder`):
+
+```text
+[writeIndex int32 BE] # number of deltas in this block
+[bitWidth int32 BE]
+[block-specific header] # first value; min delta
+[packed data] # writeIndex * bitWidth bits
+```
+
+A block stores `writeIndex + 1` values (first value + writeIndex deltas).
+`BLOCK_DEFAULT_SIZE = 128` is only `DeltaBinaryEncoder`'s default buffer
+size, not a wire-format limit: Java exposes block-size constructors
+(`IntDeltaEncoder(int)`), so `writeIndex` is bounded only by the declared
+page value count and the packed bytes available. A 300-value page from the
+default encoder produces blocks of 129, 129, 42 values.
+
+## Encoder Construction
+
+The Java `TSEncodingBuilder.Ts2Diff` field initializes `maxPointNumber = 0`,
+but the standard schema write path (`MeasurementSchema.getValueEncoder`)
+always calls `initFromProps()`, which replaces it with the schema's
+`max_point_number` property or, when the property is absent, with
+`TSFileConfig.floatPrecision` (current default `2`). A writer that
+explicitly sets `max_point_number = 0` produces Form 1 pages starting with
+`0x00`. The C++ `FloatTS2DIFFEncoder` / `DoubleTS2DIFFEncoder` default to
+`2`, matching the standard Java schema path. The value stored in the
+stream is self-describing, so files written with other `maxPointNumber`
+values remain readable.
+
+## Decoder State Machine
+
+Per page, exactly once, the decoder reads the leading marker:
+
+1. Read varint `tag`.
+2. `tag == Integer.MAX_VALUE` -> read `count` varint, `count/8+1` bytes
+ scaled-bitmap, then varint `maxPointNumber` (Form 2).
+3. `tag == Integer.MAX_VALUE-1` -> additionally read a second
+ `count/8+1` bytes raw-bitmap (Form 3).
+4. Otherwise `tag` itself is `maxPointNumber` (Form 1); `mpn <= 0` means
+ `maxPointValue = 1`.
+
+Then values are decoded from the integer blocks. For value at page position
+`p`:
+
+- raw-bitmap (if present) marks `p` -> `intBitsToFloat` / `longBitsToDouble`
+- else scaled-bitmap (if present) marks `p` -> `value / 10^mpn`
+- else -> `value / 1`
+
+Any input that does not conform to this grammar (for example, an integer
+TS_2DIFF block header where the page metadata is expected) is a format error.
diff --git a/cpp/src/common/schema.h b/cpp/src/common/schema.h
index 7094d3db7..ed59b27b1 100644
--- a/cpp/src/common/schema.h
+++ b/cpp/src/common/schema.h
@@ -134,19 +134,27 @@ struct MeasurementSchema {
encoding_ = static_cast<common::TSEncoding>(encoding);
compression_type_ =
static_cast<common::CompressionType>(compression_type);
- uint32_t props_size;
+ uint32_t props_size = 0;
if (ret == common::E_OK) {
if (RET_FAIL(
common::SerializationUtil::read_ui32(props_size, in))) {
- for (uint32_t i = 0; i < props_.size(); ++i) {
+ } else {
+ // Java MeasurementSchema.serializeTo writes an int count
+ // followed by count (key, value) string pairs. Consuming
+ // them (rather than just the count) keeps the cursor
+ // aligned for whatever follows the schema in the stream
+ // (apache/tsfile#901: a schema carrying max_point_number
+ // props desynced the TsFileMeta bloom filter). A corrupt
+ // count terminates on the first short read.
+ for (uint32_t i = 0; IS_SUCC(ret) && i < props_size; ++i) {
std::string key, value;
if (RET_FAIL(
common::SerializationUtil::read_str(key, in))) {
} else if (RET_FAIL(common::SerializationUtil::read_str(
value, in))) {
+ } else {
+ props_.insert(std::make_pair(key, value));
}
- props_.insert(std::make_pair(key, value));
- if (IS_FAIL(ret)) break;
}
}
}
diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h
index 206b7f559..ab172e051 100644
--- a/cpp/src/encoding/ts2diff_decoder.h
+++ b/cpp/src/encoding/ts2diff_decoder.h
@@ -30,6 +30,7 @@
#include "common/allocator/alloc_base.h"
#include "common/allocator/byte_stream.h"
#include "decoder.h"
+#include "ts2diff_wire_format.h"
#include "utils/util_define.h"
#ifdef ENABLE_SIMD
@@ -202,12 +203,6 @@ static inline int64_t scalar_read_bits(const uint8_t* data, int32_t bit_pos,
namespace ts2diff_java_detail {
-// Java float/double TS_2DIFF overflow page markers.
-constexpr uint32_t FLAG_ORIGINAL_VALUE_OVERFLOW =
- 2147483646u; // Integer.MAX_VALUE - 1
-constexpr uint32_t FLAG_SCALED_VALUE_OVERFLOW =
- 2147483647u; // Integer.MAX_VALUE
-
inline bool bitmap_marked(const std::vector<uint8_t>& bm, int idx) {
if (bm.empty()) {
return false;
@@ -219,80 +214,85 @@ inline bool bitmap_marked(const std::vector<uint8_t>& bm, int idx) {
return (bm[byte_idx] & static_cast<uint8_t>(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;
- }
- return true;
-}
+// Page-level FLOAT/DOUBLE metadata, parsed exactly once per page.
+// Layout (see cpp/docs/ts2diff-float-double-wire-format.md):
+// form 1: [maxPointNumber varint]
+// form 2: [Integer.MAX_VALUE][count][scaled bitmap][maxPointNumber]
+// form 3: [Integer.MAX_VALUE-1][count][scaled bitmap][raw bitmap]
+// [maxPointNumber]
+// A leading 0x00 byte is the normal encoding of maxPointNumber = 0 (the
+// Java Ts2Diff builder default), not a legacy marker. maxPointNumber is
+// whatever the varint carries: Java applies no upper bound, and a value so
+// large that Math.pow overflows simply yields maxPointValue = +inf, which
+// decodes every value through the raw-bits form.
+//
+// Inputs that do not match this grammar are a format error (E_DECODE_ERR);
+// a stream that ends mid-field propagates the underlying read error.
+struct PageMeta {
+ bool has_scaled_bm = false;
+ bool has_raw_bm = false;
+ std::vector<uint8_t> scaled_bm;
+ std::vector<uint8_t> raw_bm;
+ int max_point_number = 0;
+ int page_value_count = 0;
+};
-inline int consume_float_double_ts2diff_prefix(
- common::ByteStream& in, bool& is_legacy_raw, int& max_point_number,
- std::vector<uint8_t>& underflow_bm, std::vector<uint8_t>& overflow_bm,
- int& segment_size) {
+inline int read_page_meta(common::ByteStream& in, PageMeta& meta) {
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;
}
- if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW ||
- tag == FLAG_SCALED_VALUE_OVERFLOW) {
- uint32_t n = 0;
- if (RET_FAIL(common::SerializationUtil::read_var_uint(n, in))) {
+ if (tag == FLAG_SCALED_VALUE_OVERFLOW ||
+ tag == FLAG_ORIGINAL_VALUE_OVERFLOW) {
+ uint32_t count = 0;
+ if (RET_FAIL(common::SerializationUtil::read_var_uint(count, in))) {
return ret;
}
- segment_size = static_cast<int>(n);
- int bm_len = segment_size / 8 + 1;
- underflow_bm.resize(static_cast<size_t>(bm_len), 0);
+ if (count == 0 || count > 0x7FFFFFFFu) {
+ return common::E_DECODE_ERR;
+ }
+ const int bm_len = static_cast<int>(count) / 8 + 1;
+ // Guard the allocation: a tiny corrupt page must not trigger a
+ // bitmap-sized allocation the stream cannot possibly back. Form 3
+ // carries two bitmaps of bm_len bytes each, so the required byte
+ // count is bm_len * bitmaps, not bm_len.
+ const uint64_t bitmaps = tag == FLAG_ORIGINAL_VALUE_OVERFLOW ? 2 : 1;
+ if (static_cast<uint64_t>(bm_len) * bitmaps >
+ static_cast<uint64_t>(in.remaining_size())) {
+ return common::E_DECODE_ERR;
+ }
+ meta.has_scaled_bm = true;
+ meta.scaled_bm.resize(static_cast<size_t>(bm_len), 0);
uint32_t read_len = 0;
- if (RET_FAIL(in.read_buf(underflow_bm.data(),
- static_cast<uint32_t>(bm_len), read_len)) ||
- read_len != static_cast<uint32_t>(bm_len)) {
+ if (RET_FAIL(in.read_buf(meta.scaled_bm.data(),
+ static_cast<uint32_t>(bm_len), read_len))) {
return ret;
}
+ if (read_len != static_cast<uint32_t>(bm_len)) {
+ return common::E_DECODE_ERR;
+ }
if (tag == FLAG_ORIGINAL_VALUE_OVERFLOW) {
- overflow_bm.resize(static_cast<size_t>(bm_len), 0);
- if (RET_FAIL(in.read_buf(overflow_bm.data(),
+ meta.has_raw_bm = true;
+ meta.raw_bm.resize(static_cast<size_t>(bm_len), 0);
+ if (RET_FAIL(in.read_buf(meta.raw_bm.data(),
static_cast<uint32_t>(bm_len),
- read_len)) ||
- read_len != static_cast<uint32_t>(bm_len)) {
+ read_len))) {
return ret;
}
+ if (read_len != static_cast<uint32_t>(bm_len)) {
+ return common::E_DECODE_ERR;
+ }
}
+ meta.page_value_count = static_cast<int>(count);
uint32_t mpn = 0;
if (RET_FAIL(common::SerializationUtil::read_var_uint(mpn, in))) {
return ret;
}
- max_point_number = static_cast<int>(mpn);
- return common::E_OK;
- }
-
- // Distinguish Java maxPointNumber prefix from legacy raw C++ block.
- max_point_number = static_cast<int>(tag);
- if (!looks_like_ts2diff_header(in)) {
- in.set_read_pos(mark);
- is_legacy_raw = true;
+ meta.max_point_number = static_cast<int>(mpn);
} else {
- segment_size = 0;
+ meta.max_point_number = static_cast<int>(tag);
+ meta.page_value_count = 0; // unknown until the blocks are decoded
}
return common::E_OK;
}
@@ -319,6 +319,8 @@ class TS2DIFFDecoder : public Decoder {
bit_width_ = 0;
current_index_ = 0;
header_peeked_ = false;
+ read_error_ = common::E_OK;
+ max_values_ = -1;
}
FORCE_INLINE bool has_remaining(const common::ByteStream& buffer) override {
@@ -328,11 +330,44 @@ class TS2DIFFDecoder : public Decoder {
current_index_ != 0);
}
- void read_header(common::ByteStream& in) {
- common::SerializationUtil::read_i32(write_index_, in);
- common::SerializationUtil::read_i32(bit_width_, in);
+ // Reads the 4+4 byte block header. A failure latches read_error_ so
+ // the value-returning decode() path can surface it, and is returned
+ // directly for the batch entry points.
+ //
+ // writeIndex is bounded by the bytes the stream can actually supply,
+ // not by the encoder's BLOCK_DEFAULT_SIZE: Java exposes block-size
+ // constructors, so blocks larger than 129 values are valid input.
+ int read_header(common::ByteStream& in) {
+ int32_t write_index = 0;
+ int32_t bit_width = 0;
+ int ret = common::SerializationUtil::read_i32(write_index, in);
+ if (ret == common::E_OK) {
+ ret = common::SerializationUtil::read_i32(bit_width, in);
+ }
+ if (ret != common::E_OK) {
+ read_error_ = ret;
+ return ret;
+ }
+ const int64_t block_bytes_64 =
+ (static_cast<int64_t>(write_index) * bit_width + 7) / 8;
+ // The value-count bound is computed in int64: write_index + 1
+ // overflows for write_index == INT32_MAX and would wrap negative,
+ // silently passing the comparison.
+ if (write_index < 0 || bit_width < 0 ||
+ bit_width > (int)sizeof(T) * 8 ||
+ (max_values_ >= 0 && static_cast<int64_t>(write_index) + 1 >
+ static_cast<int64_t>(max_values_)) ||
+ block_bytes_64 > static_cast<int64_t>(in.remaining_size())) {
+ read_error_ = common::E_DECODE_ERR;
+ return common::E_DECODE_ERR;
+ }
+ write_index_ = write_index;
+ bit_width_ = bit_width;
+ return common::E_OK;
}
+ void set_max_values(int max_values) { max_values_ = max_values; }
+
// If empty, cache 8 bits from in_stream to 'buffer_'.
void read_byte_if_empty(common::ByteStream& in) {
if (bits_left_ == 0) {
@@ -348,6 +383,16 @@ class TS2DIFFDecoder : public Decoder {
int64_t value = 0;
while (bits > 0) {
read_byte_if_empty(in);
+ // The stream ended with bits still owed, so the page is
+ // truncated or desynced. read_header's availability check
+ // grants slack for the delta_min/first_value fields, so a
+ // truncated page can reach here.
+ if (bits_left_ == 0 && !in.has_remaining()) {
+ read_error_ = common::E_DECODE_ERR;
+ current_index_ = 0;
+ write_index_ = 0;
+ 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));
@@ -403,6 +448,10 @@ class TS2DIFFDecoder : public Decoder {
int write_index_;
int current_index_;
bool header_peeked_;
+ // Sticky: a block header or fixed-field read failed. decode() returns
+ // a value rather than an error code, so its callers consult this.
+ int read_error_{common::E_OK};
+ int max_values_{-1};
};
// ============================================================================
@@ -413,9 +462,23 @@ template <>
inline int32_t TS2DIFFDecoder<int32_t>::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);
+ if (read_header(in) != common::E_OK) {
+ // Poison the block state so callers stop; value is undefined
+ // for corrupt input.
+ write_index_ = 0;
+ current_index_ = 0;
+ return ret_value;
+ }
+ int fixed_ret = common::SerializationUtil::read_i32(delta_min_, in);
+ if (fixed_ret == common::E_OK) {
+ fixed_ret = common::SerializationUtil::read_i32(first_value_, in);
+ }
+ if (fixed_ret != common::E_OK) {
+ // Surface the stream's own error rather than relabelling a
+ // short read as a format violation.
+ read_error_ = fixed_ret;
+ return ret_value;
+ }
ret_value = first_value_;
bits_left_ = 0;
buffer_ = 0;
@@ -442,9 +505,21 @@ template <>
inline int64_t TS2DIFFDecoder<int64_t>::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 (read_header(in) != common::E_OK) {
+ write_index_ = 0;
+ current_index_ = 0;
+ return ret_value;
+ }
+ int fixed_ret = common::SerializationUtil::read_i64(delta_min_, in);
+ if (fixed_ret == common::E_OK) {
+ fixed_ret = common::SerializationUtil::read_i64(first_value_, in);
+ }
+ if (fixed_ret != common::E_OK) {
+ // Surface the stream's own error rather than relabelling a
+ // short read as a format violation.
+ read_error_ = fixed_ret;
+ return ret_value;
+ }
ret_value = first_value_;
if (write_index_ == 0) {
current_index_ = 0;
@@ -485,9 +560,18 @@ inline int TS2DIFFDecoder<int32_t>::read_batch_int32(int32_t* out, int capacity,
}
// Start of a new block — read header
- read_header(in);
- common::SerializationUtil::read_i32(delta_min_, in);
- common::SerializationUtil::read_i32(first_value_, in);
+ int hdr_ret = read_header(in);
+ if (hdr_ret != common::E_OK) {
+ return hdr_ret;
+ }
+ int fixed_ret = common::SerializationUtil::read_i32(delta_min_, in);
+ if (fixed_ret == common::E_OK) {
+ fixed_ret = common::SerializationUtil::read_i32(first_value_, in);
+ }
+ if (fixed_ret != common::E_OK) {
+ read_error_ = fixed_ret;
+ return fixed_ret;
+ }
bits_left_ = 0;
buffer_ = 0;
@@ -528,14 +612,22 @@ inline int TS2DIFFDecoder<int32_t>::read_batch_int32(int32_t* out, int capacity,
// could compute a block_bytes that overflows the int32_t multiply
// or runs past the wrapped buffer.
if (UNLIKELY(write_index_ < 0 || bit_width_ < 0 || bit_width_ > 32)) {
- return common::E_TSFILE_CORRUPTED;
+ return common::E_DECODE_ERR;
}
- int64_t block_bytes_64 =
+ const int64_t block_bytes_64 =
(static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8;
- if (UNLIKELY(block_bytes_64 > in.remaining_size())) {
- return common::E_TSFILE_CORRUPTED;
+ if (UNLIKELY(block_bytes_64 >
+ static_cast<int64_t>(in.remaining_size()))) {
+ return common::E_DECODE_ERR;
}
- int32_t block_bytes = static_cast<int32_t>(block_bytes_64);
+ if (UNLIKELY(block_bytes_64 > INT32_MAX)) {
+ // This path indexes the packed buffer with int32 offsets. A
+ // block that large is still valid input, so fall back to the
+ // per-value path rather than rejecting it.
+ current_index_ = 1;
+ continue;
+ }
+ const int32_t block_bytes = static_cast<int32_t>(block_bytes_64);
const uint8_t* blk_ptr =
(const uint8_t*)in.get_wrapped_buf() + in.read_pos();
in.wrapped_buf_advance_read_pos(static_cast<uint32_t>(block_bytes));
@@ -601,9 +693,19 @@ inline int TS2DIFFDecoder<int64_t>::read_batch_int64(int64_t* out, int capacity,
// Start of a new block
if (!header_peeked_) {
- read_header(in);
- common::SerializationUtil::read_i64(delta_min_, in);
- common::SerializationUtil::read_i64(first_value_, in);
+ const int hdr_ret = read_header(in);
+ if (hdr_ret != common::E_OK) {
+ return hdr_ret;
+ }
+ int fixed_ret = common::SerializationUtil::read_i64(delta_min_, in);
+ if (fixed_ret == common::E_OK) {
+ fixed_ret =
+ common::SerializationUtil::read_i64(first_value_, in);
+ }
+ if (fixed_ret != common::E_OK) {
+ read_error_ = fixed_ret;
+ return fixed_ret;
+ }
bits_left_ = 0;
buffer_ = 0;
}
@@ -632,14 +734,22 @@ inline int TS2DIFFDecoder<int64_t>::read_batch_int64(int64_t* out, int capacity,
// Validate against corrupt headers (see int32 path).
if (UNLIKELY(write_index_ < 0 || bit_width_ < 0 || bit_width_ > 64)) {
- return common::E_TSFILE_CORRUPTED;
+ return common::E_DECODE_ERR;
}
- int64_t block_bytes_64 =
+ const int64_t block_bytes_64 =
(static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8;
- if (UNLIKELY(block_bytes_64 > in.remaining_size())) {
- return common::E_TSFILE_CORRUPTED;
+ if (UNLIKELY(block_bytes_64 >
+ static_cast<int64_t>(in.remaining_size()))) {
+ return common::E_DECODE_ERR;
+ }
+ if (UNLIKELY(block_bytes_64 > INT32_MAX)) {
+ // See the int32 path: int32 packed-buffer offsets, so a block
+ // this large falls back to per-value decode instead of being
+ // rejected.
+ current_index_ = 1;
+ continue;
}
- int32_t block_bytes = static_cast<int32_t>(block_bytes_64);
+ const int32_t block_bytes = static_cast<int32_t>(block_bytes_64);
// Direct pointer into the wrapped ByteStream buffer.
const uint8_t* blk_ptr =
(const uint8_t*)in.get_wrapped_buf() + in.read_pos();
@@ -733,11 +843,32 @@ inline int TS2DIFFDecoder<int32_t>::skip_int32(int count, int& skipped,
}
while (skipped < count && has_remaining(in)) {
- int32_t wi, bw, dm, fv;
- common::SerializationUtil::read_i32(wi, in);
- common::SerializationUtil::read_i32(bw, in);
- common::SerializationUtil::read_i32(dm, in);
- common::SerializationUtil::read_i32(fv, in);
+ int32_t wi = 0, bw = 0, dm = 0, fv = 0;
+ // The header reads must succeed as a group: a truncated stream
+ // would otherwise leave stale stack values in wi/bw.
+ int rc = common::SerializationUtil::read_i32(wi, in);
+ if (rc == common::E_OK) {
+ rc = common::SerializationUtil::read_i32(bw, in);
+ }
+ if (rc == common::E_OK) {
+ rc = common::SerializationUtil::read_i32(dm, in);
+ }
+ if (rc == common::E_OK) {
+ rc = common::SerializationUtil::read_i32(fv, in);
+ }
+ if (rc != common::E_OK) {
+ return rc;
+ }
+
+ // Same availability bound as read_header (writeIndex is not
+ // capped by BLOCK_DEFAULT_SIZE); computed in int64 so a corrupt
+ // header cannot overflow the multiply.
+ const int64_t skip_bytes_64 = (static_cast<int64_t>(wi) * bw + 7) / 8;
+ if (wi < 0 || bw < 0 || bw > 32 ||
+ skip_bytes_64 > static_cast<int64_t>(in.remaining_size())) {
+ return common::E_DECODE_ERR;
+ }
+ const int32_t skip_bytes = static_cast<int32_t>(skip_bytes_64);
int32_t block_vals = wi + 1;
bits_left_ = 0;
@@ -745,7 +876,6 @@ inline int TS2DIFFDecoder<int32_t>::skip_int32(int count, int& skipped,
if (count - skipped >= block_vals) {
// Whole-block fast path: jump over packed body.
- int32_t skip_bytes = (wi * bw + 7) / 8;
in.wrapped_buf_advance_read_pos(skip_bytes);
skipped += block_vals;
current_index_ = 0;
@@ -787,19 +917,39 @@ inline int TS2DIFFDecoder<int64_t>::skip_int64(int count, int& skipped,
}
while (skipped < count && has_remaining(in)) {
- int32_t wi, bw;
- int64_t dm, fv;
- common::SerializationUtil::read_i32(wi, in);
- common::SerializationUtil::read_i32(bw, in);
- common::SerializationUtil::read_i64(dm, in);
- common::SerializationUtil::read_i64(fv, in);
+ int32_t wi = 0, bw = 0;
+ int64_t dm = 0, fv = 0;
+ // The header reads must succeed as a group: a truncated stream
+ // would otherwise leave stale stack values in wi/bw.
+ int rc = common::SerializationUtil::read_i32(wi, in);
+ if (rc == common::E_OK) {
+ rc = common::SerializationUtil::read_i32(bw, in);
+ }
+ if (rc == common::E_OK) {
+ rc = common::SerializationUtil::read_i64(dm, in);
+ }
+ if (rc == common::E_OK) {
+ rc = common::SerializationUtil::read_i64(fv, in);
+ }
+ if (rc != common::E_OK) {
+ return rc;
+ }
+
+ // Same availability bound as read_header (writeIndex is not
+ // capped by BLOCK_DEFAULT_SIZE); computed in int64 so a corrupt
+ // header cannot overflow the multiply.
+ const int64_t skip_bytes_64 = (static_cast<int64_t>(wi) * bw + 7) / 8;
+ if (wi < 0 || bw < 0 || bw > 64 ||
+ skip_bytes_64 > static_cast<int64_t>(in.remaining_size())) {
+ return common::E_DECODE_ERR;
+ }
+ const int32_t skip_bytes = static_cast<int32_t>(skip_bytes_64);
int32_t block_vals = wi + 1;
bits_left_ = 0;
buffer_ = 0;
if (count - skipped >= block_vals) {
- int32_t skip_bytes = (wi * bw + 7) / 8;
in.wrapped_buf_advance_read_pos(skip_bytes);
skipped += block_vals;
current_index_ = 0;
@@ -838,9 +988,21 @@ inline bool TS2DIFFDecoder<int64_t>::peek_next_block_range_int64(
// value decoder (value decoders decode normally and never call this).
if (current_index_ != 0 || !has_remaining(in)) return false;
- read_header(in);
- common::SerializationUtil::read_i64(delta_min_, in);
- common::SerializationUtil::read_i64(first_value_, in);
+ if (read_header(in) != common::E_OK) {
+ // Reports "no next block with a usable range", so the caller uses
+ // the regular per-value path. This is deliberately a bool: an
+ // error code would convert to true and make callers act on an
+ // unset range.
+ return false;
+ }
+ int fixed_ret = common::SerializationUtil::read_i64(delta_min_, in);
+ if (fixed_ret == common::E_OK) {
+ fixed_ret = common::SerializationUtil::read_i64(first_value_, in);
+ }
+ if (fixed_ret != common::E_OK) {
+ read_error_ = fixed_ret;
+ return false;
+ }
bits_left_ = 0;
buffer_ = 0;
@@ -853,8 +1015,12 @@ inline bool TS2DIFFDecoder<int64_t>::peek_next_block_range_int64(
// at offset 16 within the header
// (write_index_(4)+bit_width_(4)+delta_min_(8)). We read it via raw pointer
// so the stream position is not consumed.
- int32_t packed_bytes = (write_index_ * bit_width_ + 7) / 8;
- if (in.remaining_size() >= (uint32_t)packed_bytes + 24) {
+ // int64 arithmetic, matching skip_peeked_block_int64: read_header bounds
+ // the packed size by the stream length only, so write_index_ * bit_width_
+ // can exceed INT32_MAX on a huge page.
+ const int64_t packed_bytes =
+ (static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8;
+ if (in.remaining_size() >= static_cast<uint64_t>(packed_bytes) + 24) {
char* next_fv_ptr =
in.get_wrapped_buf() + in.read_pos() + packed_bytes + 16;
block_max = (int64_t)common::SerializationUtil::read_ui64(next_fv_ptr);
@@ -878,8 +1044,12 @@ template <>
inline int TS2DIFFDecoder<int64_t>::skip_peeked_block_int64(
common::ByteStream& in, int& skipped) {
skipped = write_index_ + 1;
- int32_t skip_bytes = (write_index_ * bit_width_ + 7) / 8;
- in.wrapped_buf_advance_read_pos(skip_bytes);
+ // int64 arithmetic: read_header admits blocks whose packed size is
+ // only bounded by the stream length, and write_index_ * bit_width_
+ // can exceed INT32_MAX on a huge page.
+ const int64_t skip_bytes_64 =
+ (static_cast<int64_t>(write_index_) * bit_width_ + 7) / 8;
+ in.wrapped_buf_advance_read_pos(static_cast<uint64_t>(skip_bytes_64));
header_peeked_ = false;
bits_left_ = 0;
buffer_ = 0;
@@ -939,6 +1109,18 @@ inline int TS2DIFFDecoder<int64_t>::skip_int32(int count, int& skipped,
class FloatTS2DIFFDecoder : public TS2DIFFDecoder<int32_t> {
public:
FloatTS2DIFFDecoder() = default;
+ // PageReader invokes reset() at every page boundary; the page-level
+ // FLOAT/DOUBLE metadata (maxPointNumber, page-wide bitmaps, page value
+ // position) is parsed once per page and survives block transitions.
+ void reset() override {
+ TS2DIFFDecoder<int32_t>::reset();
+ page_meta_parsed_ = false;
+ max_point_value_ = 1.0;
+ page_pos_ = 0;
+ page_value_count_ = 0;
+ scaled_bm_.clear();
+ raw_bm_.clear();
+ }
float decode(common::ByteStream& in) {
int32_t value_int = TS2DIFFDecoder<int32_t>::decode(in);
return common::int_to_float(value_int);
@@ -951,31 +1133,69 @@ class FloatTS2DIFFDecoder : public TS2DIFFDecoder<int32_t> {
int read_double(double& ret_value, common::ByteStream& in) override;
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<int32_t*>(out);
- int ret = TS2DIFFDecoder<int32_t>::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]);
+ common::ByteStream& in) override;
+ int read_batch_int32(int32_t* out, int capacity, int& actual,
+ common::ByteStream& in) override;
+
+ private:
+ // Parses the page metadata on the first value of a page. Returns
+ // E_OK and leaves the stream positioned at the first integer block.
+ int ensure_page_meta(common::ByteStream& in) {
+ if (page_meta_parsed_) {
+ return common::E_OK;
+ }
+ ts2diff_java_detail::PageMeta meta;
+ int ret = ts2diff_java_detail::read_page_meta(in, meta);
+ if (RET_FAIL(ret)) {
+ return ret;
}
+ max_point_value_ =
+ meta.max_point_number <= 0
+ ? 1.0
+ : std::pow(10.0, static_cast<double>(meta.max_point_number));
+ page_value_count_ = meta.page_value_count;
+ if (page_value_count_ > 0) {
+ TS2DIFFDecoder<int32_t>::set_max_values(page_value_count_);
+ }
+ scaled_bm_ = std::move(meta.scaled_bm);
+ raw_bm_ = std::move(meta.raw_bm);
+ page_pos_ = 0;
+ page_meta_parsed_ = true;
return common::E_OK;
}
- private:
- bool is_legacy_raw_{false};
- int max_point_number_{0};
+ float value_at(int32_t value_int) const {
+ if (!raw_bm_.empty() &&
+ ts2diff_java_detail::bitmap_marked(raw_bm_, page_pos_)) {
+ return common::int_to_float(value_int);
+ }
+ const bool use_scaled =
+ scaled_bm_.empty() ||
+ ts2diff_java_detail::bitmap_marked(scaled_bm_, page_pos_);
+ const double divisor = use_scaled ? max_point_value_ : 1.0;
+ return static_cast<float>(static_cast<double>(value_int) / divisor);
+ }
+
double max_point_value_{1.0};
- int segment_pos_{0};
- int segment_size_{0};
- std::vector<uint8_t> underflow_bm_;
- std::vector<uint8_t> overflow_bm_;
+ int page_pos_{0};
+ int page_value_count_{0};
+ bool page_meta_parsed_{false};
+ std::vector<uint8_t> scaled_bm_;
+ std::vector<uint8_t> raw_bm_;
};
class DoubleTS2DIFFDecoder : public TS2DIFFDecoder<int64_t> {
public:
DoubleTS2DIFFDecoder() = default;
+ void reset() override {
+ TS2DIFFDecoder<int64_t>::reset();
+ page_meta_parsed_ = false;
+ max_point_value_ = 1.0;
+ page_pos_ = 0;
+ page_value_count_ = 0;
+ scaled_bm_.clear();
+ raw_bm_.clear();
+ }
double decode(common::ByteStream& in) {
int64_t value_long = TS2DIFFDecoder<int64_t>::decode(in);
return common::long_to_double(value_long);
@@ -988,26 +1208,53 @@ class DoubleTS2DIFFDecoder : public TS2DIFFDecoder<int64_t> {
int read_double(double& ret_value, common::ByteStream& in) override;
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<int64_t*>(out);
- int ret = TS2DIFFDecoder<int64_t>::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]);
+ common::ByteStream& in) override;
+ int read_batch_int64(int64_t* out, int capacity, int& actual,
+ common::ByteStream& in) override;
+
+ private:
+ int ensure_page_meta(common::ByteStream& in) {
+ if (page_meta_parsed_) {
+ return common::E_OK;
+ }
+ ts2diff_java_detail::PageMeta meta;
+ int ret = ts2diff_java_detail::read_page_meta(in, meta);
+ if (RET_FAIL(ret)) {
+ return ret;
+ }
+ max_point_value_ =
+ meta.max_point_number <= 0
+ ? 1.0
+ : std::pow(10.0, static_cast<double>(meta.max_point_number));
+ page_value_count_ = meta.page_value_count;
+ if (page_value_count_ > 0) {
+ TS2DIFFDecoder<int64_t>::set_max_values(page_value_count_);
}
+ scaled_bm_ = std::move(meta.scaled_bm);
+ raw_bm_ = std::move(meta.raw_bm);
+ page_pos_ = 0;
+ page_meta_parsed_ = true;
return common::E_OK;
}
- private:
- bool is_legacy_raw_{false};
- int max_point_number_{0};
+ double value_at(int64_t value_long) const {
+ if (!raw_bm_.empty() &&
+ ts2diff_java_detail::bitmap_marked(raw_bm_, page_pos_)) {
+ return common::long_to_double(value_long);
+ }
+ const bool use_scaled =
+ scaled_bm_.empty() ||
+ ts2diff_java_detail::bitmap_marked(scaled_bm_, page_pos_);
+ const double divisor = use_scaled ? max_point_value_ : 1.0;
+ return static_cast<double>(value_long) / divisor;
+ }
+
double max_point_value_{1.0};
- int segment_pos_{0};
- int segment_size_{0};
- std::vector<uint8_t> underflow_bm_;
- std::vector<uint8_t> overflow_bm_;
+ int page_pos_{0};
+ int page_value_count_{0};
+ bool page_meta_parsed_{false};
+ std::vector<uint8_t> scaled_bm_;
+ std::vector<uint8_t> raw_bm_;
};
typedef TS2DIFFDecoder<int32_t> IntTS2DIFFDecoder;
@@ -1024,7 +1271,7 @@ template <>
FORCE_INLINE int IntTS2DIFFDecoder::read_int32(int32_t& ret_value,
common::ByteStream& in) {
ret_value = decode(in);
- return common::E_OK;
+ return read_error_;
}
template <>
FORCE_INLINE int IntTS2DIFFDecoder::read_int64(int64_t& ret_value,
@@ -1067,7 +1314,7 @@ template <>
FORCE_INLINE int LongTS2DIFFDecoder::read_int64(int64_t& ret_value,
common::ByteStream& in) {
ret_value = decode(in);
- return common::E_OK;
+ return read_error_;
}
template <>
FORCE_INLINE int LongTS2DIFFDecoder::read_float(float& ret_value,
@@ -1106,37 +1353,15 @@ 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 (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;
- }
- max_point_value_ =
- max_point_number_ <= 0
- ? 1.0
- : std::pow(10.0, static_cast<double>(max_point_number_));
- segment_pos_ = 0;
- }
- if (is_legacy_raw_) {
- ret_value = decode(in);
- return common::E_OK;
+ if (RET_FAIL(ensure_page_meta(in))) {
+ return ret;
}
int32_t value_int = TS2DIFFDecoder<int32_t>::decode(in);
- if (!overflow_bm_.empty() &&
- ts2diff_java_detail::bitmap_marked(overflow_bm_, segment_pos_)) {
- ret_value = common::int_to_float(value_int);
- } else {
- bool use_scaled = true;
- if (!underflow_bm_.empty()) {
- use_scaled =
- ts2diff_java_detail::bitmap_marked(underflow_bm_, segment_pos_);
- }
- const double divisor = use_scaled ? max_point_value_ : 1.0;
- ret_value =
- static_cast<float>(static_cast<double>(value_int) / divisor);
+ if (TS2DIFFDecoder<int32_t>::read_error_ != common::E_OK) {
+ return TS2DIFFDecoder<int32_t>::read_error_;
}
- segment_pos_++;
+ ret_value = value_at(value_int);
+ page_pos_++;
return common::E_OK;
}
FORCE_INLINE int FloatTS2DIFFDecoder::read_double(double& ret_value,
@@ -1144,6 +1369,45 @@ FORCE_INLINE int FloatTS2DIFFDecoder::read_double(double& ret_value,
ASSERT(false);
return common::E_NOT_SUPPORT;
}
+// Page-level metadata is parsed once, then the integer blocks are decoded
+// with the integer batch decoder (SIMD fast path where available) and the
+// page-wide bitmaps are applied afterwards using the page position.
+FORCE_INLINE int FloatTS2DIFFDecoder::read_batch_float(float* out, int capacity,
+ int& actual,
+ common::ByteStream& in) {
+ int ret = common::E_OK;
+ actual = 0;
+ if (RET_FAIL(ensure_page_meta(in))) {
+ return ret;
+ }
+ constexpr int kIntsPerBatch = 128;
+ int32_t ints[kIntsPerBatch];
+ while (actual < capacity) {
+ int block_actual = 0;
+ const int want = capacity - actual;
+ const int request = want < kIntsPerBatch ? want : kIntsPerBatch;
+ if (RET_FAIL(TS2DIFFDecoder<int32_t>::read_batch_int32(
+ ints, request, block_actual, in))) {
+ return ret;
+ }
+ if (block_actual == 0) {
+ break;
+ }
+ for (int i = 0; i < block_actual; i++) {
+ out[actual + i] = value_at(ints[i]);
+ page_pos_++;
+ }
+ actual += block_actual;
+ }
+ return common::E_OK;
+}
+FORCE_INLINE int FloatTS2DIFFDecoder::read_batch_int32(int32_t* out,
+ int capacity,
+ int& actual,
+ common::ByteStream& in) {
+ ASSERT(false);
+ return common::E_NOT_SUPPORT;
+}
FORCE_INLINE int DoubleTS2DIFFDecoder::read_boolean(bool& ret_value,
common::ByteStream& in) {
ASSERT(false);
@@ -1167,38 +1431,51 @@ 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 (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;
- }
- max_point_value_ =
- max_point_number_ <= 0
- ? 1.0
- : std::pow(10.0, static_cast<double>(max_point_number_));
- segment_pos_ = 0;
- }
- if (is_legacy_raw_) {
- ret_value = decode(in);
- return common::E_OK;
+ if (RET_FAIL(ensure_page_meta(in))) {
+ return ret;
}
int64_t value_long = TS2DIFFDecoder<int64_t>::decode(in);
- if (!overflow_bm_.empty() &&
- ts2diff_java_detail::bitmap_marked(overflow_bm_, segment_pos_)) {
- ret_value = common::long_to_double(value_long);
- } else {
- bool use_scaled = true;
- if (!underflow_bm_.empty()) {
- use_scaled =
- ts2diff_java_detail::bitmap_marked(underflow_bm_, segment_pos_);
+ if (TS2DIFFDecoder<int64_t>::read_error_ != common::E_OK) {
+ return TS2DIFFDecoder<int64_t>::read_error_;
+ }
+ ret_value = value_at(value_long);
+ page_pos_++;
+ return common::E_OK;
+}
+// See FloatTS2DIFFDecoder::read_batch_float for the layout rationale.
+FORCE_INLINE int DoubleTS2DIFFDecoder::read_batch_double(
+ double* out, int capacity, int& actual, common::ByteStream& in) {
+ int ret = common::E_OK;
+ actual = 0;
+ if (RET_FAIL(ensure_page_meta(in))) {
+ return ret;
+ }
+ constexpr int kIntsPerBatch = 128;
+ int64_t ints[kIntsPerBatch];
+ while (actual < capacity) {
+ int block_actual = 0;
+ const int want = capacity - actual;
+ const int request = want < kIntsPerBatch ? want : kIntsPerBatch;
+ if (RET_FAIL(TS2DIFFDecoder<int64_t>::read_batch_int64(
+ ints, request, block_actual, in))) {
+ return ret;
}
- const double divisor = use_scaled ? max_point_value_ : 1.0;
- ret_value = static_cast<double>(value_long) / divisor;
+ if (block_actual == 0) {
+ break;
+ }
+ for (int i = 0; i < block_actual; i++) {
+ out[actual + i] = value_at(ints[i]);
+ page_pos_++;
+ }
+ actual += block_actual;
}
- segment_pos_++;
return common::E_OK;
}
+FORCE_INLINE int DoubleTS2DIFFDecoder::read_batch_int64(
+ int64_t* out, int capacity, int& actual, common::ByteStream& in) {
+ ASSERT(false);
+ return common::E_NOT_SUPPORT;
+}
} // end namespace storage
#endif // ENCODING_TS2DIFF_DECODER_H
diff --git a/cpp/src/encoding/ts2diff_encoder.h b/cpp/src/encoding/ts2diff_encoder.h
index fc494581a..de75fd2b7 100644
--- a/cpp/src/encoding/ts2diff_encoder.h
+++ b/cpp/src/encoding/ts2diff_encoder.h
@@ -24,11 +24,13 @@
#include <cmath>
#include <limits>
+#include <type_traits>
#include <vector>
#include "common/allocator/alloc_base.h"
#include "common/allocator/byte_stream.h"
#include "encoder.h"
+#include "ts2diff_wire_format.h"
#ifdef ENABLE_SIMD
#include "simde/x86/avx2.h"
@@ -157,11 +159,27 @@ class TS2DIFFEncoder : public Encoder {
void rebase_arr(int i) { delta_arr_[i] = delta_arr_[i] - delta_arr_min_; }
- int cal_bit_width(T n) {
+ // Java-compatible bit width: the maximum width over the *rebased*
+ // deltas, not the width of (max - min). When the raw deltas wrap
+ // (e.g. two adjacent raw IEEE bit patterns whose difference
+ // overflows the integer type), (max - min) itself wraps negative and
+ // its width would be 0, silently discarding every delta in the
+ // block. Mirrors Java calculateBitWidthsForDeltaBlockBuffer, which
+ // widens each rebased delta individually.
+ int cal_bit_width_rebased() {
+ typedef typename std::make_unsigned<T>::type UT;
+ UT max_rebased = 0;
+ for (int i = 0; i < write_index_; i++) {
+ UT rebased = static_cast<UT>(delta_arr_[i]) -
+ static_cast<UT>(delta_arr_min_);
+ if (rebased > max_rebased) {
+ max_rebased = rebased;
+ }
+ }
int bit_width = 0;
- while (n > 0) {
+ while (max_rebased > 0) {
bit_width++;
- n >>= 1;
+ max_rebased >>= 1;
}
return bit_width;
}
@@ -208,10 +226,9 @@ class TS2DIFFEncoder : public Encoder {
if (bits_in_accum > 0) {
buf[pos++] = static_cast<uint8_t>(accum << (8 - bits_in_accum));
}
- // Surface write failures. Previously the return code was dropped on
- // the floor and flush() returned E_OK, then reset() wiped the
- // encoder state — the on-disk page ended up missing its delta block
- // but the caller thought the data was safe.
+ // The write status must reach the caller: a dropped return code
+ // here would let flush() report E_OK for a page whose delta block
+ // never made it to disk.
return out_stream.write_buf(buf.data(), pos);
}
@@ -284,10 +301,12 @@ inline int TS2DIFFEncoder<int32_t>::flush(common::ByteStream& out_stream) {
if (write_index_ == -1) {
return common::E_OK;
}
+ // Bit width over the raw deltas rebased by min (computed before the
+ // array itself is rebased, so cal_bit_width_rebased subtracts min
+ // exactly once).
+ int bit_width = cal_bit_width_rebased();
// Subtract the minimum value for each delta_arr_ item
SIMDOps<int32_t>::rebase(delta_arr_, delta_arr_min_, write_index_);
- // Calculate the bit length of each value to writer
- int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_);
// Header writes can fail too (back-pressure / OOM on the underlying
// stream); a half-written header followed by reset() leaves the page
// corrupted but the caller thinking the data was flushed.
@@ -321,7 +340,10 @@ inline int TS2DIFFEncoder<int32_t>::flush(common::ByteStream& out_stream) {
// layer can detect the page is poisoned.
return pack_ret;
}
- reset();
+ // Base reset only (write_index_ = -1). This is a virtual call so a
+ // float wrapper encoder can keep its page-scoped state (overflow
+ // flags, buffered blocks) alive across the per-block flush.
+ TS2DIFFEncoder<int32_t>::reset();
return ret;
}
@@ -331,10 +353,11 @@ inline int TS2DIFFEncoder<int64_t>::flush(common::ByteStream& out_stream) {
if (write_index_ == -1) {
return common::E_OK;
}
+ // Bit width over the raw deltas rebased by min; see the int32
+ // specialization for ordering rationale.
+ int bit_width = cal_bit_width_rebased();
// Subtract the minimum value for each delta_arr_ item
SIMDOps<int64_t>::rebase(delta_arr_, delta_arr_min_, write_index_);
- // Calculate the bit length of each value to writer
- int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_);
// Header writes can fail too — see int32 specialization for rationale.
if (RET_FAIL(
common::SerializationUtil::write_i32(write_index_, out_stream))) {
@@ -363,7 +386,9 @@ inline int TS2DIFFEncoder<int64_t>::flush(common::ByteStream& out_stream) {
} else if (pack_ret != common::E_OK) {
return pack_ret;
}
- reset(); // 语义，writeIndex=-1;
+ // Base reset only — see the int32 specialization for the virtual-call
+ // rationale.
+ TS2DIFFEncoder<int64_t>::reset();
return ret;
}
@@ -551,20 +576,56 @@ int TS2DIFFEncoder<T>::encode_batch(const int64_t* values, uint32_t count,
return Encoder::encode_batch(values, count, out);
}
+// Java Math.round(x) == floor(x + 0.5) (ties go towards +infinity),
+// unlike std::lround's ties-away-from-zero (-2.5 -> -2, not -3).
+// Math.round saturates at the integer limits; a plain static_cast of the
+// boundary value (e.g. INT64_MAX rounding up from below, or 2^63 passing
+// the <= (double)INT64_MAX gate) would be undefined behaviour, so the
+// conversion clamps first. The 64-bit upper bound is compared against
+// 2^63 rather than (double)INT64_MAX, which rounds up to 2^63 and would
+// let exactly 2^63 through; the lower limits are exactly representable.
+template <typename Int>
+inline Int java_round(double x) {
+ const double r = std::floor(x + 0.5);
+ if (sizeof(Int) < 8) {
+ if (r > static_cast<double>(std::numeric_limits<Int>::max())) {
+ return std::numeric_limits<Int>::max();
+ }
+ } else if (r >= 9223372036854775808.0) { // 2^63
+ return std::numeric_limits<Int>::max();
+ }
+ if (r < static_cast<double>(std::numeric_limits<Int>::min())) {
+ return std::numeric_limits<Int>::min();
+ }
+ return static_cast<Int>(r);
+}
+
class FloatTS2DIFFEncoder : public TS2DIFFEncoder<int32_t> {
public:
- FloatTS2DIFFEncoder() : max_point_number_(2), max_point_value_(100.0) {}
+ FloatTS2DIFFEncoder()
+ : max_point_number_(2), // Java Ts2Diff builder default via
+ // initFromProps -> TSFileConfig
+ // floatPrecision
+ max_point_value_(100.0),
+ page_blocks_(1024, common::MOD_TS2DIFF_OBJ, false) {}
int do_encode(float value, common::ByteStream& out_stream) {
int32_t value_int = convert_float_to_int(value);
return TS2DIFFEncoder<int32_t>::do_encode(value_int, out_stream);
}
+ // The wire value is self-describing; tests use this to exercise
+ // maxPointNumber = 0 pages (Form 1 leading 0x00 byte).
+ void set_max_point_number(int mpn) {
+ max_point_number_ = mpn;
+ max_point_value_ = mpn <= 0 ? 1.0 : std::pow(10.0, mpn);
+ }
// PageWriter resets the encoder between pages without going through a
// successful flush() (e.g. when the prior page was aborted). The base
- // reset() only clears write_index_; underflow_flags_ would otherwise
- // leak the prior page's overflow markers into the next page's bitmap.
+ // reset() only clears write_index_; underflow_flags_ and the buffered
+ // complete blocks would otherwise leak into the next page.
void reset() override {
TS2DIFFEncoder<int32_t>::reset();
underflow_flags_.clear();
+ page_blocks_.reset();
}
int flush(common::ByteStream& out_stream) override;
int encode(bool value, common::ByteStream& out_stream);
@@ -587,14 +648,14 @@ class FloatTS2DIFFEncoder : public TS2DIFFEncoder<int32_t> {
return common::float_to_int(value);
}
underflow_flags_.push_back(0);
- return static_cast<int32_t>(std::lround(value));
+ return java_round<int32_t>(value);
}
if (std::isnan(value)) {
underflow_flags_.push_back(-1);
return common::float_to_int(value);
}
underflow_flags_.push_back(1);
- return static_cast<int32_t>(std::lround(scaled));
+ return java_round<int32_t>(scaled);
}
bool has_overflow() const {
for (int8_t f : underflow_flags_) {
@@ -609,20 +670,38 @@ class FloatTS2DIFFEncoder : public TS2DIFFEncoder<int32_t> {
int max_point_number_;
double max_point_value_;
std::vector<int8_t> underflow_flags_;
+ // Java FloatEncoder emits the page metadata (maxPointNumber and, when
+ // needed, the page-wide overflow bitmaps) once per page, followed by a
+ // continuous sequence of integer TS_2DIFF blocks. The integer encoder
+ // flushes a complete 129-value block on its own whenever it fills, so
+ // those blocks are buffered here and emitted, metadata first, when the
+ // page is sealed.
+ common::ByteStream page_blocks_;
};
class DoubleTS2DIFFEncoder : public TS2DIFFEncoder<int64_t> {
public:
- DoubleTS2DIFFEncoder() : max_point_number_(2), max_point_value_(100.0) {}
+ DoubleTS2DIFFEncoder()
+ : max_point_number_(2), // Java Ts2Diff builder default via
+ // initFromProps -> TSFileConfig
+ // floatPrecision
+ max_point_value_(100.0),
+ page_blocks_(1024, common::MOD_TS2DIFF_OBJ, false) {}
int do_encode(double value, common::ByteStream& out_stream) {
int64_t value_long = convert_double_to_long(value);
return TS2DIFFEncoder<int64_t>::do_encode(value_long, out_stream);
}
+ // See FloatTS2DIFFEncoder::set_max_point_number.
+ void set_max_point_number(int mpn) {
+ max_point_number_ = mpn;
+ max_point_value_ = mpn <= 0 ? 1.0 : std::pow(10.0, mpn);
+ }
// See FloatTS2DIFFEncoder::reset for rationale — the prior page's
- // overflow markers must not bleed into the next.
+ // overflow markers and buffered blocks must not bleed into the next.
void reset() override {
TS2DIFFEncoder<int64_t>::reset();
underflow_flags_.clear();
+ page_blocks_.reset();
}
int flush(common::ByteStream& out_stream) override;
int encode(bool value, common::ByteStream& out_stream);
@@ -645,14 +724,14 @@ class DoubleTS2DIFFEncoder : public TS2DIFFEncoder<int64_t> {
return common::double_to_long(value);
}
underflow_flags_.push_back(0);
- return static_cast<int64_t>(std::llround(value));
+ return java_round<int64_t>(value);
}
if (std::isnan(value)) {
underflow_flags_.push_back(-1);
return common::double_to_long(value);
}
underflow_flags_.push_back(1);
- return static_cast<int64_t>(std::llround(scaled));
+ return java_round<int64_t>(scaled);
}
bool has_overflow() const {
for (int8_t f : underflow_flags_) {
@@ -667,6 +746,9 @@ class DoubleTS2DIFFEncoder : public TS2DIFFEncoder<int64_t> {
int max_point_number_;
double max_point_value_;
std::vector<int8_t> underflow_flags_;
+ // See FloatTS2DIFFEncoder::page_blocks_ — buffered complete integer
+ // blocks, emitted with the page metadata when the page is sealed.
+ common::ByteStream page_blocks_;
};
typedef TS2DIFFEncoder<int32_t> IntTS2DIFFEncoder;
@@ -776,66 +858,55 @@ FORCE_INLINE int DoubleTS2DIFFEncoder::encode(double value,
return do_encode(value, out);
}
-// Keep float/double TS_2DIFF page layout compatible with Java.
+// Java FloatEncoder page layout:
+// no overflow: [maxPointNumber varint][block 1][block 2]...
+// overflow: [overflow marker varint][pageValueCount varint]
+// [page-wide bitmap(s)][maxPointNumber varint][blocks...]
+// The integer encoder triggers flush() whenever a 129-value block fills
+// (write_index_ == block_size_, reachable only from do_encode); those
+// blocks are buffered in page_blocks_ without any float metadata. When
+// the page is sealed (write_index_ < block_size_), the trailing block is
+// appended and the page metadata plus all buffered blocks are emitted.
FORCE_INLINE int FloatTS2DIFFEncoder::flush(common::ByteStream& out_stream) {
int ret = common::E_OK;
- if (write_index_ == -1) {
- return common::E_OK;
- }
- 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<uint32_t>(max_point_number_), inner))) {
- return ret;
- }
- SIMDOps<int32_t>::rebase(delta_arr_, delta_arr_min_, write_index_);
- int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_);
- if (RET_FAIL(common::SerializationUtil::write_ui32(
- static_cast<uint32_t>(write_index_), inner))) {
- return ret;
- }
- if (RET_FAIL(common::SerializationUtil::write_ui32(
- static_cast<uint32_t>(bit_width), inner))) {
- return ret;
- }
- if (RET_FAIL(common::SerializationUtil::write_ui32(
- static_cast<uint32_t>(delta_arr_min_), inner))) {
- return ret;
- }
- if (RET_FAIL(common::SerializationUtil::write_ui32(
- static_cast<uint32_t>(first_value_), inner))) {
- return ret;
+ const bool block_flush = (write_index_ == block_size_);
+ if (block_flush) {
+ // Complete-block flush from do_encode: plain integer block, no
+ // float wrapper metadata, overflow flags must survive for the
+ // page-wide bitmap.
+ return TS2DIFFEncoder<int32_t>::flush(page_blocks_);
+ }
+ if (write_index_ != -1) {
+ // Page-seal flush: append the trailing (partial) integer block.
+ if (RET_FAIL(TS2DIFFEncoder<int32_t>::flush(page_blocks_))) {
+ return ret;
+ }
}
- for (int i = 0; i < write_index_; i++) {
- write_bits(delta_arr_[i], bit_width, inner);
+ if (underflow_flags_.empty()) {
+ // Empty page (nothing encoded, no blocks buffered).
+ return common::E_OK;
}
- flush_remaining(inner);
-
- const bool overflow = has_overflow();
- if (overflow) {
- std::vector<uint8_t> underflow_bitmap(
+ const int num_values = static_cast<int>(underflow_flags_.size());
+ if (has_overflow()) {
+ std::vector<uint8_t> scaled_bitmap(
static_cast<size_t>(num_values / 8 + 1), 0);
- std::vector<uint8_t> overflow_bitmap(
+ std::vector<uint8_t> raw_bits_bitmap(
static_cast<size_t>(num_values / 8 + 1), 0);
- bool has_original_value_overflow = false;
+ bool has_raw_bits = false;
for (int i = 0; i < num_values; i++) {
int8_t f = underflow_flags_[static_cast<size_t>(i)];
if (f == 1) {
- underflow_bitmap[static_cast<size_t>(i / 8)] |=
+ scaled_bitmap[static_cast<size_t>(i / 8)] |=
static_cast<uint8_t>(1u << (i % 8));
} else if (f == -1) {
- has_original_value_overflow = true;
- overflow_bitmap[static_cast<size_t>(i / 8)] |=
+ has_raw_bits = true;
+ raw_bits_bitmap[static_cast<size_t>(i / 8)] |=
static_cast<uint8_t>(1u << (i % 8));
}
}
- constexpr uint32_t FLAG_SCALED_VALUE_OVERFLOW =
- 2147483647u; // Integer.MAX_VALUE
- constexpr uint32_t FLAG_ORIGINAL_VALUE_OVERFLOW =
- 2147483646u; // Integer.MAX_VALUE - 1
if (RET_FAIL(common::SerializationUtil::write_var_uint(
- has_original_value_overflow ? FLAG_ORIGINAL_VALUE_OVERFLOW
- : FLAG_SCALED_VALUE_OVERFLOW,
+ has_raw_bits ? ts2diff_java_detail::FLAG_ORIGINAL_VALUE_OVERFLOW
+ : ts2diff_java_detail::FLAG_SCALED_VALUE_OVERFLOW,
out_stream))) {
return ret;
}
@@ -844,81 +915,68 @@ FORCE_INLINE int FloatTS2DIFFEncoder::flush(common::ByteStream& out_stream) {
return ret;
}
const uint32_t bm_len = static_cast<uint32_t>(num_values / 8 + 1);
- if (RET_FAIL(out_stream.write_buf(underflow_bitmap.data(), bm_len))) {
+ if (RET_FAIL(out_stream.write_buf(scaled_bitmap.data(), bm_len))) {
return ret;
}
- if (has_original_value_overflow &&
- RET_FAIL(out_stream.write_buf(overflow_bitmap.data(), bm_len))) {
+ if (has_raw_bits &&
+ RET_FAIL(out_stream.write_buf(raw_bits_bitmap.data(), bm_len))) {
return ret;
}
}
- if (RET_FAIL(merge_byte_stream(out_stream, inner, true))) {
+ // maxPointNumber sits right before the first integer block in every
+ // page layout.
+ if (RET_FAIL(common::SerializationUtil::write_var_uint(
+ static_cast<uint32_t>(max_point_number_), out_stream))) {
+ return ret;
+ }
+ if (RET_FAIL(common::merge_byte_stream(out_stream, page_blocks_))) {
return ret;
}
- // Defer encoder-state wipe until after every write into out_stream has
- // committed. An earlier reset() let a mid-flush failure leave
- // write_index_ at -1, so the next flush() short-circuited at the top
- // and the data was silently lost.
+ // The encoder state is wiped only after every write into out_stream has
+ // committed: clearing it earlier would leave write_index_ at -1 after a
+ // mid-flush failure, so the retrying flush() would short-circuit at the
+ // top and drop the data.
underflow_flags_.clear();
- TS2DIFFEncoder<int32_t>::reset();
+ page_blocks_.reset();
return ret;
}
+// See FloatTS2DIFFEncoder::flush for the page layout rationale.
FORCE_INLINE int DoubleTS2DIFFEncoder::flush(common::ByteStream& out_stream) {
int ret = common::E_OK;
- if (write_index_ == -1) {
- return common::E_OK;
- }
- 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<uint32_t>(max_point_number_), inner))) {
- return ret;
- }
- SIMDOps<int64_t>::rebase(delta_arr_, delta_arr_min_, write_index_);
- int bit_width = cal_bit_width(delta_arr_max_ - delta_arr_min_);
- if (RET_FAIL(common::SerializationUtil::write_i32(write_index_, inner))) {
- return ret;
+ const bool block_flush = (write_index_ == block_size_);
+ if (block_flush) {
+ return TS2DIFFEncoder<int64_t>::flush(page_blocks_);
}
- if (RET_FAIL(common::SerializationUtil::write_i32(bit_width, inner))) {
- return ret;
- }
- if (RET_FAIL(common::SerializationUtil::write_i64(delta_arr_min_, inner))) {
- return ret;
- }
- if (RET_FAIL(common::SerializationUtil::write_i64(first_value_, inner))) {
- return ret;
+ if (write_index_ != -1) {
+ if (RET_FAIL(TS2DIFFEncoder<int64_t>::flush(page_blocks_))) {
+ return ret;
+ }
}
- for (int i = 0; i < write_index_; i++) {
- write_bits(delta_arr_[i], bit_width, inner);
+ if (underflow_flags_.empty()) {
+ return common::E_OK;
}
- flush_remaining(inner);
-
- const bool overflow = has_overflow();
- if (overflow) {
- std::vector<uint8_t> underflow_bitmap(
+ const int num_values = static_cast<int>(underflow_flags_.size());
+ if (has_overflow()) {
+ std::vector<uint8_t> scaled_bitmap(
static_cast<size_t>(num_values / 8 + 1), 0);
- std::vector<uint8_t> overflow_bitmap(
+ std::vector<uint8_t> raw_bits_bitmap(
static_cast<size_t>(num_values / 8 + 1), 0);
- bool has_original_value_overflow = false;
+ bool has_raw_bits = false;
for (int i = 0; i < num_values; i++) {
int8_t f = underflow_flags_[static_cast<size_t>(i)];
if (f == 1) {
- underflow_bitmap[static_cast<size_t>(i / 8)] |=
+ scaled_bitmap[static_cast<size_t>(i / 8)] |=
static_cast<uint8_t>(1u << (i % 8));
} else if (f == -1) {
- has_original_value_overflow = true;
- overflow_bitmap[static_cast<size_t>(i / 8)] |=
+ has_raw_bits = true;
+ raw_bits_bitmap[static_cast<size_t>(i / 8)] |=
static_cast<uint8_t>(1u << (i % 8));
}
}
- constexpr uint32_t FLAG_SCALED_VALUE_OVERFLOW =
- 2147483647u; // Integer.MAX_VALUE
- constexpr uint32_t FLAG_ORIGINAL_VALUE_OVERFLOW =
- 2147483646u; // Integer.MAX_VALUE - 1
if (RET_FAIL(common::SerializationUtil::write_var_uint(
- has_original_value_overflow ? FLAG_ORIGINAL_VALUE_OVERFLOW
- : FLAG_SCALED_VALUE_OVERFLOW,
+ has_raw_bits ? ts2diff_java_detail::FLAG_ORIGINAL_VALUE_OVERFLOW
+ : ts2diff_java_detail::FLAG_SCALED_VALUE_OVERFLOW,
out_stream))) {
return ret;
}
@@ -927,22 +985,26 @@ FORCE_INLINE int DoubleTS2DIFFEncoder::flush(common::ByteStream& out_stream) {
return ret;
}
const uint32_t bm_len = static_cast<uint32_t>(num_values / 8 + 1);
- if (RET_FAIL(out_stream.write_buf(underflow_bitmap.data(), bm_len))) {
+ if (RET_FAIL(out_stream.write_buf(scaled_bitmap.data(), bm_len))) {
return ret;
}
- if (has_original_value_overflow &&
- RET_FAIL(out_stream.write_buf(overflow_bitmap.data(), bm_len))) {
+ if (has_raw_bits &&
+ RET_FAIL(out_stream.write_buf(raw_bits_bitmap.data(), bm_len))) {
return ret;
}
}
- if (RET_FAIL(merge_byte_stream(out_stream, inner, true))) {
+ if (RET_FAIL(common::SerializationUtil::write_var_uint(
+ static_cast<uint32_t>(max_point_number_), out_stream))) {
+ return ret;
+ }
+ if (RET_FAIL(common::merge_byte_stream(out_stream, page_blocks_))) {
return ret;
}
// Same deferred-reset rationale as FloatTS2DIFFEncoder::flush — keeping
// write_index_ live until every committed write succeeds avoids the
// "next flush returns E_OK on lost data" pattern.
underflow_flags_.clear();
- TS2DIFFEncoder<int64_t>::reset();
+ page_blocks_.reset();
return ret;
}
diff --git a/cpp/src/encoding/ts2diff_wire_format.h b/cpp/src/encoding/ts2diff_wire_format.h
new file mode 100644
index 000000000..bd1536c92
--- /dev/null
+++ b/cpp/src/encoding/ts2diff_wire_format.h
@@ -0,0 +1,39 @@
+/*
+ * 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.
+ */
+
+#ifndef ENCODING_TS2DIFF_WIRE_FORMAT_H
+#define ENCODING_TS2DIFF_WIRE_FORMAT_H
+
+#include <cstdint>
+
+namespace storage {
+namespace ts2diff_java_detail {
+
+// FLOAT/DOUBLE TS_2DIFF page markers, shared by the encoder and the decoder
+// so a single definition describes the wire format. See
+// cpp/docs/ts2diff-float-double-wire-format.md.
+constexpr uint32_t FLAG_ORIGINAL_VALUE_OVERFLOW =
+ 2147483646u; // Integer.MAX_VALUE - 1
+constexpr uint32_t FLAG_SCALED_VALUE_OVERFLOW =
+ 2147483647u; // Integer.MAX_VALUE
+
+} // namespace ts2diff_java_detail
+} // namespace storage
+
+#endif // ENCODING_TS2DIFF_WIRE_FORMAT_H
diff --git a/cpp/test/encoding/ts2diff_codec_test.cc b/cpp/test/encoding/ts2diff_codec_test.cc
index fb997103c..ee299f38d 100644
--- a/cpp/test/encoding/ts2diff_codec_test.cc
+++ b/cpp/test/encoding/ts2diff_codec_test.cc
@@ -97,14 +97,16 @@ class FloatDoubleTS2DIFFCodecTest : public ::testing::Test {
DoubleTS2DIFFDecoder* decoder_double_{nullptr};
};
+// Reads the whole stream into a hex string. Both call sites pass a
+// freshly written stream (read position 0), so a plain sequential read is
+// enough — set_read_pos() to rewind would park the cursor at a page
+// boundary and misbehave in check_space().
static std::string byte_stream_to_hex(common::ByteStream& stream) {
- uint32_t mark = stream.read_pos();
uint32_t size = stream.total_size();
std::vector<uint8_t> 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);
- stream.set_read_pos(mark);
std::ostringstream oss;
for (uint32_t i = 0; i < size; i++) {
@@ -122,7 +124,7 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, TestFloatRoundTrip) {
const int row_num = 1000;
std::vector<float> data(row_num);
for (int i = 0; i < row_num; i++) {
- data[i] = static_cast<float>(i) * 0.25f + 0.50f;
+ data[i] = static_cast<float>(i) * 2.0f + 1.0f;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_float_->encode(data[i], out_stream), common::E_OK);
@@ -146,6 +148,8 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, TestFloatJavaDefaultHexCompatibility) {
}
EXPECT_EQ(encoder_float_->flush(out_stream), common::E_OK);
+ // Golden bytes from the Java FloatEncoder + TS_2DIFF writer at the
+ // default maxPointNumber = 2 (TSFileConfig.floatPrecision).
const std::string expected_hex =
"FE FF FF FF 07 02 00 03 02 00 00 00 01 00 00 00 00 1E 38 8A AA 61 87 "
"75 56";
@@ -161,6 +165,8 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, TestDoubleJavaDefaultHexCompatibility) {
}
EXPECT_EQ(encoder_double_->flush(out_stream), common::E_OK);
+ // Golden bytes from the Java DoubleEncoder + TS_2DIFF writer at the
+ // default maxPointNumber = 2 (TSFileConfig.floatPrecision).
const std::string expected_hex =
"FE FF FF FF 07 02 00 03 02 00 00 00 01 00 00 00 00 3B C7 11 55 3D "
"D4 27 08 44 30 EE AA C2 2B D8 F8";
@@ -172,7 +178,7 @@ TEST_F(FloatDoubleTS2DIFFCodecTest, TestDoubleRoundTrip) {
const int row_num = 800;
std::vector<double> data(row_num);
for (int i = 0; i < row_num; i++) {
- data[i] = static_cast<double>(i) * 0.25 + 0.5;
+ data[i] = static_cast<double>(i) * 2.0 + 1.0;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_double_->encode(data[i], out_stream), common::E_OK);
@@ -187,6 +193,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<float> expected(row_num);
+ for (int i = 0; i < row_num; ++i) {
+ expected[i] = static_cast<float>(i) * 2.0f + 1.0f;
+ ASSERT_EQ(encoder_float_->encode(expected[i], out_stream),
+ common::E_OK);
+ }
+ ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK);
+
+ std::vector<float> 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 +529,797 @@ 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
+// Legacy raw TS_2DIFF pages (pre-#796 C++ writer output: plain int delta
+// blocks over bit-cast float bits, no maxPointNumber / bitmap prefix) are
+// outside the cross-language format: the Java reader never supported them.
+// The decoder treats such input as a format error instead of guessing the
+// layout: it must fail fast rather than hang at end-of-input or silently
+// return misdecoded values.
+//
+// A raw block starts with the 4-byte big-endian write_index whose high
+// byte is 0x00; to the page-metadata parser that is a valid
+// maxPointNumber = 0 (Form 1), so detection happens at the block level:
+// the following bytes are not a consistent block stream and the batch
+// reader must terminate with an error rather than loop forever.
+TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchFloatLegacyRawSegments) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const int row_num = 129;
+ std::vector<float> 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);
+
+ std::vector<float> actual_values(row_num);
+ int actual = 0;
+ // Must terminate (no end-of-input spin) and report a format error;
+ // never return E_OK with garbage values.
+ const int rc = decoder_float_->read_batch_float(
+ actual_values.data(), row_num, actual, out_stream);
+ ASSERT_NE(rc, common::E_OK);
+}
+
+TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchDoubleLegacyRawSegments) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const int row_num = 129;
+ std::vector<double> 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<double> actual_values(row_num);
+ int actual = 0;
+ const int rc = decoder_double_->read_batch_double(
+ actual_values.data(), row_num, actual, out_stream);
+ ASSERT_NE(rc, common::E_OK);
+}
+
+TEST_F(FloatDoubleTS2DIFFCodecTest, ReadFloatLegacyRawScalar) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const int row_num = 129;
+ std::vector<float> 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;
+ // Scalar path: may return a bounded number of values, but must not
+ // spin forever; assert that reading the full stream terminates and
+ // does not report success for all rows of a known-invalid layout.
+ int ok_rows = 0;
+ int rc = common::E_OK;
+ for (int i = 0; i < row_num; ++i) {
+ rc = decoder_float_->read_float(v, out_stream);
+ if (rc != common::E_OK) break;
+ ok_rows++;
+ }
+ // Termination is the contract; the values themselves are undefined
+ // for this out-of-format input.
+ SUCCEED();
+}
+
+TEST_F(FloatDoubleTS2DIFFCodecTest, ReadDoubleLegacyRawScalar) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const int row_num = 129;
+ std::vector<double> 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.;
+ int rc = common::E_OK;
+ for (int i = 0; i < row_num; ++i) {
+ rc = decoder_double_->read_double(v, out_stream);
+ if (rc != common::E_OK) break;
+ }
+ SUCCEED();
+}
+
+// Mixed reads on a legacy raw page: the batch reader must fail fast; the
+// scalar reader after it must also terminate.
+TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyRawBatchThenScalarReads) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const int row_num = 129;
+ std::vector<float> expected(row_num);
+ for (int i = 0; i < row_num; ++i) {
+ expected[i] = (i < 128) ? 1.0f : 100.0f;
+ }
+ 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;
+ const int rc =
+ decoder_float_->read_batch_float(batch_out, 40, actual, out_stream);
+ ASSERT_NE(rc, common::E_OK);
+ float v = 0.f;
+ for (int i = 0; i < row_num; ++i) {
+ if (decoder_float_->read_float(v, out_stream) != common::E_OK) break;
+ }
+ SUCCEED();
+}
+// ============================================================================
+// 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<uint8_t>& 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<uint32_t>(byte & 0x7F) << shift;
+ if ((byte & 0x80) == 0) return true;
+ shift += 7;
+ }
+}
+
+int32_t read_i32_be(const std::vector<uint8_t>& b, size_t pos) {
+ return (static_cast<int32_t>(b[pos]) << 24) |
+ (static_cast<int32_t>(b[pos + 1]) << 16) |
+ (static_cast<int32_t>(b[pos + 2]) << 8) |
+ static_cast<int32_t>(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<uint8_t> byte_stream_bytes(common::ByteStream& stream) {
+ uint32_t size = stream.total_size();
+ std::vector<uint8_t> 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<uint8_t>& b, common::ByteStream& s) {
+ s.wrap_from(reinterpret_cast<const char*>(b.data()),
+ static_cast<int32_t>(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.
+// Structural page walker for the canonical Java layout: parses the page
+// metadata once, then walks the integer block stream. Verifies that the
+// metadata (and thus maxPointNumber) appears exactly once per page and
+// that every block header is well formed.
+void expect_max_pn_once_per_page(const std::vector<uint8_t>& b,
+ bool is_double) {
+ const uint32_t FLAG_SCALED = 2147483647u;
+ const uint32_t FLAG_ORIGINAL = 2147483646u;
+ size_t pos = 0;
+ size_t header_len = is_double ? 24 : 16;
+
+ // Page metadata, exactly once at the start.
+ ASSERT_FALSE(b.empty());
+ uint32_t tag = 0;
+ ASSERT_TRUE(parse_var_uint(b, pos, tag));
+ if (tag == FLAG_SCALED || tag == FLAG_ORIGINAL) {
+ // Forms 2/3: [marker][count][bitmap(s)][maxPointNumber].
+ uint32_t n = 0;
+ ASSERT_TRUE(parse_var_uint(b, pos, n));
+ EXPECT_GE(n, 1u);
+ size_t bm_len = static_cast<size_t>(n / 8 + 1);
+ ASSERT_LE(pos + bm_len, b.size());
+ pos += bm_len;
+ if (tag == FLAG_ORIGINAL) {
+ ASSERT_LE(pos + bm_len, b.size());
+ pos += bm_len;
+ }
+ uint32_t mpn = 0;
+ ASSERT_TRUE(parse_var_uint(b, pos, mpn));
+ EXPECT_EQ(mpn, 2u) << "default encoder writes maxPointNumber = 2";
+ } else {
+ // Form 1: the leading varint IS the maxPointNumber (0x02 = 2).
+ EXPECT_EQ(tag, 2u) << "default encoder writes maxPointNumber = 2";
+ }
+
+ // Continuous integer block stream with no float metadata between
+ // blocks.
+ int segment_count = 0;
+ while (pos < b.size()) {
+ ASSERT_LE(pos + header_len, b.size());
+ int32_t wi = read_i32_be(b, pos);
+ int32_t bw = read_i32_be(b, pos + 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 += header_len;
+ pos += (static_cast<size_t>(wi) * static_cast<size_t>(bw) + 7) / 8;
+ ASSERT_LE(pos, b.size());
+ segment_count++;
+ }
+ EXPECT_GE(segment_count, 1);
+}
+
+} // 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<float> data(row_num);
+ for (int i = 0; i < row_num; i++) {
+ data[i] = static_cast<float>(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);
+
+ // The default maxPointNumber is 2 (TSFileConfig.floatPrecision), so
+ // the page starts with the single 0x02 mpn byte; the structural scan
+ // below verifies the prefix appears exactly once.
+ std::vector<uint8_t> b = byte_stream_bytes(out);
+ ASSERT_FALSE(b.empty());
+ EXPECT_EQ(b[0], 0x02) << "page must start with the maxPointNumber=2 byte";
+ 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<double> data(row_num);
+ for (int i = 0; i < row_num; i++) {
+ data[i] = static_cast<double>(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<uint8_t> b = byte_stream_bytes(out);
+ ASSERT_FALSE(b.empty());
+ EXPECT_EQ(b[0], 0x02) << "page must start with the maxPointNumber=2 byte";
+ 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<float> 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<float>(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][pageValueCount=140][page-wide
+ // underflow bitmap (18B)][maxPointNumber 0x02][block 1 header][packed]
+ // [block 2 header starting with 0x00 — no prefix]
+ std::vector<uint8_t> 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, 140u); // page-wide bitmap covers every value in the page
+ size_t bm_len = static_cast<size_t>(n / 8 + 1);
+ ASSERT_LE(pos + bm_len, b.size());
+ pos += bm_len;
+ // Exactly one maxPointNumber, directly after the bitmap.
+ ASSERT_LT(pos, b.size());
+ EXPECT_EQ(b[pos], 0x02) << "maxPointNumber = 2 byte";
+ 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<size_t>(wi) * static_cast<size_t>(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<double> 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<double>(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<uint8_t> 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<float> data(row_num);
+ for (int i = 0; i < row_num; i++) {
+ data[i] = static_cast<float>(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<uint8_t> b1 = byte_stream_bytes(page1);
+ std::vector<uint8_t> b2 = byte_stream_bytes(page2);
+ ASSERT_FALSE(b1.empty());
+ ASSERT_FALSE(b2.empty());
+ EXPECT_EQ(b1[0], 0x02) << "page 1 must start with maxPointNumber=2";
+ EXPECT_EQ(b2[0], 0x02) << "page 2 must start with maxPointNumber=2";
+ 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));
+}
+
+// Explicit maxPointNumber = 0 (an explicit max_point_number property on
+// the Java side): a canonical Form 1 page starting with the 0x00 byte.
+// At mpn = 0 every finite in-range value takes the scaled form, so the
+// page has no bitmap; the 0x00 byte is the maxPointNumber varint itself,
+// not a legacy marker.
+TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberZeroPagePrefix) {
+ const int row_num = 140; // 129 + 11: crosses the block boundary
+ std::vector<float> data(row_num);
+ for (int i = 0; i < row_num; i++) {
+ data[i] = static_cast<float>(i) * 2.0f + 1.0f;
+ }
+ common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
+ encoder_float_->set_max_point_number(0);
+ 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);
+
+ std::vector<uint8_t> b = byte_stream_bytes(out);
+ ASSERT_FALSE(b.empty());
+ EXPECT_EQ(b[0], 0x00) << "mpn=0 page must start with the 0x00 byte";
+ // Form 1: the 0x00 byte is consumed as maxPointNumber, and the rest
+ // must be a well-formed block stream (the walker asserts mpn == 0).
+ size_t pos = 0;
+ uint32_t mpn = 999;
+ ASSERT_TRUE(parse_var_uint(b, pos, mpn));
+ EXPECT_EQ(mpn, 0u);
+
+ 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 for double.
+ std::vector<double> data_d(row_num);
+ for (int i = 0; i < row_num; i++) {
+ data_d[i] = static_cast<double>(i) * 2.0 + 1.0;
+ }
+ common::ByteStream out_d(1024, common::MOD_TS2DIFF_OBJ, false);
+ encoder_double_->set_max_point_number(0);
+ for (int i = 0; i < row_num; i++) {
+ ASSERT_EQ(encoder_double_->encode(data_d[i], out_d), common::E_OK);
+ }
+ ASSERT_EQ(encoder_double_->flush(out_d), common::E_OK);
+
+ std::vector<uint8_t> bd = byte_stream_bytes(out_d);
+ ASSERT_FALSE(bd.empty());
+ EXPECT_EQ(bd[0], 0x00) << "mpn=0 page must start with the 0x00 byte";
+ common::ByteStream dec_d;
+ wrap_bytes(bd, dec_d);
+ double y = 0.0;
+ for (int i = 0; i < row_num; i++) {
+ ASSERT_EQ(decoder_double_->read_double(y, dec_d), common::E_OK);
+ EXPECT_DOUBLE_EQ(y, data_d[i]) << "row " << i;
+ }
+ EXPECT_FALSE(decoder_double_->has_remaining(dec_d));
+}
+
+// The pre-#910 C++ encoder repeated the maxPointNumber at every segment
+// boundary. That layout was never produced by the Java writer and is now
+// outside the compatibility boundary: the decoder parses page metadata
+// exactly once, so a hand-built old-format page fails the block-header
+// validation instead of being silently misdecoded.
+TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyPerSegmentMaxPNRejected) {
+ // Build an old-format page by hand: 0x02 prefix before BOTH segments.
+ const std::vector<float> 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);
+
+ // Segment 1 (a valid single-block Form 1 page) decodes. The trailing
+ // 0x02 of the second segment prefix is then read as a block header and
+ // fails validation, so the decode loop terminates (no end-of-input
+ // spin). The invariant is that the out-of-format continuation is
+ // never handed back as valid data: reading stops before the segment-2
+ // values, either by returning an error or by yielding a mismatch.
+ float x = 0.0f;
+ size_t good = 0;
+ for (size_t i = 0; i < expected.size(); i++) {
+ if (decoder_float_->read_float(x, old_fmt) != common::E_OK) break;
+ if (std::fabs(x - expected[i]) > 1e-6f) break;
+ good++;
+ }
+ EXPECT_LT(good, expected.size())
+ << "out-of-format continuation must not decode to the expected values";
+ EXPECT_GE(good, 1u) << "segment 1 is a valid page and must still decode";
+}
+
+// A truncated page whose block header passes the availability check (the
+// check grants slack for the delta_min/first_value fields) but whose
+// packed bits never arrive: the scalar read path must terminate instead
+// of spinning on a stale buffer.
+TEST_F(FloatDoubleTS2DIFFCodecTest, ScalarReadTerminatesOnTruncatedBlock) {
+ // [mpn=2][wi=1][bw=8][dm=0][fv=42] - after the header, remaining=8
+ // covers the 1 packed byte, then dm+fv consume it all.
+ const unsigned char page[] = {
+ 0x02, 0x00, 0x00, 0x00, 0x01, // write_index = 1
+ 0x00, 0x00, 0x00, 0x08, // bit_width = 8
+ 0x00, 0x00, 0x00, 0x00, // delta_min
+ 0x00, 0x00, 0x00, 0x2a, // first_value
+ };
+ common::ByteStream dec;
+ dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
+ FloatTS2DIFFDecoder decoder;
+ float x = 0.0f;
+ ASSERT_EQ(decoder.read_float(x, dec), common::E_OK);
+ EXPECT_FLOAT_EQ(x, 0.42f); // first_value 42 / maxPointValue 100
+ // The stream is exhausted mid-block; further reads must return
+ // (garbage is acceptable - the caller stops on has_remaining) but
+ // never hang.
+ for (int i = 0; i < 3; i++) {
+ decoder.read_float(x, dec);
+ }
+ EXPECT_FALSE(decoder.has_remaining(dec));
+ SUCCEED();
+}
+
+TEST_F(FloatDoubleTS2DIFFCodecTest, RejectsBlockBeyondPageValueCount) {
+ // [overflow marker][count=1][bitmap][mpn=0][wi=INT_MAX][bw=0]...
+ // A zero-width block has no packed bytes, so availability alone cannot
+ // bound wi; the page value count must provide the bound.
+ const unsigned char page[] = {
+ 0xff, 0xff, 0xff, 0xff, 0x07, 0x01, 0x00, 0x00, 0x7f,
+ 0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
+ };
+ common::ByteStream dec;
+ dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
+ FloatTS2DIFFDecoder decoder;
+ float value = 0.0f;
+ // A block header that violates the format is a decode error, matching
+ // the RLE/Sprintz/RLBE convention.
+ EXPECT_EQ(decoder.read_float(value, dec), common::E_DECODE_ERR);
+}
+
+TEST_F(FloatDoubleTS2DIFFCodecTest, ScalarReadRejectsTruncatedFixedFields) {
+ // The page metadata is valid, but the block is missing one byte of its
+ // fixed first-value field.
+ const unsigned char page[] = {
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+ };
+ common::ByteStream dec;
+ dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
+ FloatTS2DIFFDecoder decoder;
+ float value = 0.0f;
+ // A short read propagates the stream's own error rather than being
+ // rebranded as a format violation.
+ EXPECT_EQ(decoder.read_float(value, dec), common::E_PARTIAL_READ);
+}
+
+// A page whose trailing block header is truncated mid-field: the skip
+// paths must reject it via the grouped header-read check instead of
+// acting on stale stack values.
+TEST_F(TS2DIFFCodecTest, SkipRejectsTruncatedBlockHeader) {
+ // [wi=1][bw=8] then only 4 of the 8 dm/fv bytes.
+ const unsigned char page[] = {
+ 0x00, 0x00, 0x00, 0x01, // write_index = 1
+ 0x00, 0x00, 0x00, 0x08, // bit_width = 8
+ 0x00, 0x00, 0x00, 0x00, // delta_min
+ 0x00, 0x00, 0x00, // first_value truncated to 3 bytes
+ };
+ common::ByteStream dec;
+ dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
+ IntTS2DIFFDecoder decoder;
+ int skipped = 0;
+ const int rc = decoder.skip_int32(10, skipped, dec);
+ // Truncated header: the underlying short-read error propagates.
+ EXPECT_EQ(rc, common::E_PARTIAL_READ);
+}
+
+// Java Math.round semantics: floor(x + 0.5), ties towards +infinity.
+// -0.125 * 100 = -12.5 exactly (0.125 is a binary fraction), and the tie
+// must store as -12 (std::lround would give -13), so the round trip
+// returns -0.12 rather than -0.13.
+TEST_F(FloatDoubleTS2DIFFCodecTest, JavaRoundNegativeHalfTiesFloat) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const float data[] = {-0.125f, -0.375f, 0.125f, 0.375f};
+ const float expect[] = {-0.12f, -0.37f, 0.13f, 0.38f};
+ for (float v : data) {
+ ASSERT_EQ(encoder_float_->encode(v, out_stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK);
+
+ float x = 0.0f;
+ for (int i = 0; i < 4; i++) {
+ ASSERT_EQ(decoder_float_->read_float(x, out_stream), common::E_OK);
+ EXPECT_FLOAT_EQ(x, expect[i]) << "value " << i;
+ }
+}
+
+TEST_F(FloatDoubleTS2DIFFCodecTest, JavaRoundNegativeHalfTiesDouble) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ const double data[] = {-2.5, -7.5, 2.5, 7.5}; // mpn=0: scaled == value
+ const double expect[] = {-2.0, -7.0, 3.0, 8.0};
+ encoder_double_->set_max_point_number(0);
+ for (double v : data) {
+ ASSERT_EQ(encoder_double_->encode(v, out_stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder_double_->flush(out_stream), common::E_OK);
+
+ double y = 0.0;
+ for (int i = 0; i < 4; i++) {
+ ASSERT_EQ(decoder_double_->read_double(y, out_stream), common::E_OK);
+ EXPECT_DOUBLE_EQ(y, expect[i]) << "value " << i;
+ }
+}
+
+// writeIndex is bounded by availability, not BLOCK_DEFAULT_SIZE: a
+// hand-built block larger than 129 values (Java exposes block-size
+// constructors) must decode through both the scalar and batch paths.
+TEST_F(TS2DIFFCodecTest, LargeBlockBeyondDefaultSizeDecodes) {
+ const int kCount = 200; // wi = 199 > 128
+ const int32_t first = 1000;
+ const int32_t delta = 3;
+ const int32_t bit_width = 3;
+
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ ASSERT_EQ(common::SerializationUtil::write_ui32(kCount - 1, out_stream),
+ common::E_OK);
+ ASSERT_EQ(common::SerializationUtil::write_ui32(bit_width, out_stream),
+ common::E_OK);
+ // delta_min: constant deltas of 3 rebase to 0 (raw - min), so the
+ // decoder reconstructs value += stored(0) + delta_min(3) per step.
+ ASSERT_EQ(common::SerializationUtil::write_ui32(delta, out_stream),
+ common::E_OK);
+ ASSERT_EQ(common::SerializationUtil::write_ui32(first, out_stream),
+ common::E_OK);
+ // 199 * 3 bits, MSB-packed, all-zero bytes == rebased delta 0. The
+ // packed body is ceil(bits / 8) = 75 bytes.
+ const int packed_bytes = ((kCount - 1) * bit_width + 7) / 8; // 75
+ for (int i = 0; i < packed_bytes; i++) {
+ ASSERT_EQ(common::SerializationUtil::write_ui8(0, out_stream),
+ common::E_OK);
+ }
+
+ // Copy the encoded page into a plain buffer; wrap two read streams
+ // over it (get_wrapped_buf is only valid for wrap_from streams).
+ std::vector<uint8_t> page(out_stream.total_size());
+ uint32_t read_len = 0;
+ ASSERT_EQ(
+ out_stream.read_buf(page.data(), out_stream.total_size(), read_len),
+ common::E_OK);
+ ASSERT_EQ(read_len, page.size());
+
+ // Scalar path.
+ common::ByteStream dec1;
+ dec1.wrap_from(reinterpret_cast<const char*>(page.data()),
+ static_cast<int32_t>(page.size()));
+ IntTS2DIFFDecoder dec_scalar;
+ int32_t v = 0;
+ ASSERT_EQ(dec_scalar.read_int32(v, dec1), common::E_OK);
+ EXPECT_EQ(v, first);
+ for (int i = 1; i < kCount; i++) {
+ ASSERT_EQ(dec_scalar.read_int32(v, dec1), common::E_OK);
+ EXPECT_EQ(v, first + i * delta) << "row " << i;
+ }
+ EXPECT_FALSE(dec_scalar.has_remaining(dec1));
+
+ // Batch path.
+ common::ByteStream dec2;
+ dec2.wrap_from(reinterpret_cast<const char*>(page.data()),
+ static_cast<int32_t>(page.size()));
+ IntTS2DIFFDecoder dec_batch;
+ std::vector<int32_t> out(kCount);
+ int actual = 0;
+ ASSERT_EQ(dec_batch.read_batch_int32(out.data(), kCount, actual, dec2),
+ common::E_OK);
+ ASSERT_EQ(actual, kCount);
+ for (int i = 0; i < kCount; i++) {
+ EXPECT_EQ(out[i], first + i * delta) << "row " << i;
+ }
+}
+
+// maxPointNumber has no format-level upper bound: a page written with
+// mpn = 1000 (Java Math.pow overflows to +inf for the decoder, values
+// then divide by inf) stays readable; the decoder must accept the varint
+// instead of rejecting it as corrupt.
+TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberAboveLegacyBoundDecodes) {
+ common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
+ encoder_float_->set_max_point_number(1000);
+ const float data[] = {1.0f, 2.0f, 3.0f};
+ for (float v : data) {
+ ASSERT_EQ(encoder_float_->encode(v, out_stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK);
+
+ // The encoder stores round(v * 10^1000); the double product overflows
+ // to +inf for every value, so every entry takes the raw-bits form
+ // (Form 3). The decode restores the exact bits.
+ std::vector<uint8_t> page(out_stream.total_size());
+ uint32_t read_len = 0;
+ ASSERT_EQ(
+ out_stream.read_buf(page.data(), out_stream.total_size(), read_len),
+ common::E_OK);
+ ASSERT_EQ(read_len, page.size());
+ common::ByteStream dec;
+ dec.wrap_from(reinterpret_cast<const char*>(page.data()),
+ static_cast<int32_t>(page.size()));
+ float x = 0.0f;
+ for (int i = 0; i < 3; i++) {
+ ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK);
+ EXPECT_EQ(common::float_to_int(x), common::float_to_int(data[i]))
+ << "row " << i;
+ }
+}
+
} // namespace storage
diff --git a/cpp/test/reader/table_view/table_model_encoding_compression_compatibility_test.cc b/cpp/test/reader/table_view/table_model_encoding_compression_compatibility_test.cc
index 5db8266b0..c52cd0515 100644
--- a/cpp/test/reader/table_view/table_model_encoding_compression_compatibility_test.cc
+++ b/cpp/test/reader/table_view/table_model_encoding_compression_compatibility_test.cc
@@ -21,6 +21,7 @@
#include <cctype>
#include <cerrno>
+#include <cmath>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
@@ -65,7 +66,9 @@ const char* const kTableName = "compat_table";
const char* const kTagColumn = "device";
const char* const kValueColumn = "value";
const char* const kTagValue = "compat_device";
-constexpr int kRowCount = 32;
+// Crosses the 129-value TS_2DIFF block boundary (300 -> 129+129+42), so
+// page-wide bitmaps must survive block transitions. Applied to every case.
+constexpr int kRowCount = 300;
const int32_t kIntValues[] = {
0,
@@ -123,6 +126,46 @@ const uint64_t kDoubleBits[] = {
UINT64_C(0x405edd2f1a9fbe77), UINT64_C(0xc05edd2f1a9fbe77),
};
+// FLOAT/DOUBLE + TS_2DIFF converts values to fixed-point with
+// maxPointNumber (10^mpn). The expected value is computed by applying the
+// same tri-state conversion the writer performs and then decoding it back,
+// so expectations reflect the encoding rules rather than the raw input
+// bits. Every chosen value has the same expected value whether the writer
+// used maxPointNumber 0 (explicit property; the "*.mpn0" Java fixtures) or
+// 2 (the TSFileConfig.floatPrecision default shared with the C++ writer),
+// so the validating reader never needs to know which writer produced the
+// file. The values cover all page layouts the writers can produce:
+// - plain integers (scaled form; at mpn = 0 every finite in-range value
+// is scaled, so pages written at mpn = 0 contain no bitmap)
+// - 1.5e9f / 1e18: the scaled product overflows while round(v) still
+// fits -> scale-overflow form, single page-wide bitmap (reachable only
+// at mpn > 0, i.e. the default writer configuration)
+// - NaN and +/-Infinity (stored as raw IEEE bits -> two page-wide
+// bitmaps; Java floatToIntBits canonicalizes NaN)
+const uint32_t kTs2DiffFloatBits[] = {
+ 0x00000000U, 0x3f800000U, 0xbf800000U, 0x461c4000U,
+ 0xc61c4000U, 0x4eb2d05eU, 0x7f800000U, 0xff800000U,
+ 0x7fc00000U, 0x3f800000U, 0x00000000U, 0x4a742400U,
+ 0xca742400U, 0x4e6e6b28U, 0x3f800000U, 0x00000000U,
+};
+
+const uint64_t kTs2DiffDoubleBits[] = {
+ UINT64_C(0x0000000000000000), UINT64_C(0x3ff0000000000000),
+ UINT64_C(0xbff0000000000000), UINT64_C(0x40c3880000000000),
+ UINT64_C(0xc0c3880000000000), UINT64_C(0x43abc16d674ec800),
+ UINT64_C(0x7ff0000000000000), UINT64_C(0xfff0000000000000),
+ UINT64_C(0x7ff8000000000000), UINT64_C(0x3ff0000000000000),
+ UINT64_C(0x0000000000000000), UINT64_C(0x41f0000000000000),
+ UINT64_C(0xc1cd6f3458800000), UINT64_C(0x430c6bf526340000),
+ UINT64_C(0x3ff0000000000000), UINT64_C(0x0000000000000000),
+};
+
+// maxPointNumber used by the C++ FloatTS2DIFFEncoder/DoubleTS2DIFFEncoder,
+// matching the standard Java schema write path (TSEncodingBuilder.Ts2Diff
+// initFromProps -> TSFileConfig.floatPrecision, current default 2).
+constexpr int kTs2DiffMaxPointNumber = 2;
+constexpr double kTs2DiffMaxPointValue = 100.0;
+
const int32_t kDateValues[] = {
19700101, 19991231, 20000229, 20240229,
20380119, 20500615, 19690720, 19800106,
@@ -292,6 +335,12 @@ uint32_t FloatBits(float value) {
return bits;
}
+float BitsToFloat(uint32_t bits) {
+ float value;
+ std::memcpy(&value, &bits, sizeof(value));
+ return value;
+}
+
float FloatValue(int row) {
uint32_t bits =
kFloatBits[row % (sizeof(kFloatBits) / sizeof(kFloatBits[0]))];
@@ -300,12 +349,26 @@ float FloatValue(int row) {
return value;
}
+float Ts2DiffFloatValue(int row) {
+ uint32_t bits = kTs2DiffFloatBits[row % (sizeof(kTs2DiffFloatBits) /
+ sizeof(kTs2DiffFloatBits[0]))];
+ float value;
+ std::memcpy(&value, &bits, sizeof(value));
+ return value;
+}
+
uint64_t DoubleBits(double value) {
uint64_t bits;
std::memcpy(&bits, &value, sizeof(bits));
return bits;
}
+double BitsToDouble(uint64_t bits) {
+ double value;
+ std::memcpy(&value, &bits, sizeof(value));
+ return value;
+}
+
double DoubleValue(int row) {
uint64_t bits =
kDoubleBits[row % (sizeof(kDoubleBits) / sizeof(kDoubleBits[0]))];
@@ -314,6 +377,94 @@ double DoubleValue(int row) {
return value;
}
+double Ts2DiffDoubleValue(int row) {
+ uint64_t bits = kTs2DiffDoubleBits[row % (sizeof(kTs2DiffDoubleBits) /
+ sizeof(kTs2DiffDoubleBits[0]))];
+ double value;
+ std::memcpy(&value, &bits, sizeof(value));
+ return value;
+}
+
+bool IsTs2DiffFloatCase(const FixtureCase& fixture_case) {
+ return fixture_case.encoding == TS_2DIFF &&
+ (fixture_case.data_type == FLOAT ||
+ fixture_case.data_type == DOUBLE);
+}
+
+bool StringEndsWith(const std::string& value, const std::string& suffix) {
+ return value.size() >= suffix.size() &&
+ value.compare(value.size() - suffix.size(), suffix.size(), suffix) ==
+ 0;
+}
+
+// maxPointValue of the writer configuration for this fixture: the
+// "*.mpn0" Java fixtures carry an explicit max_point_number=0 property,
+// all others use the shared default (see kTs2DiffMaxPointNumber).
+double Ts2DiffMaxPointValue(const FixtureCase& fixture_case) {
+ return StringEndsWith(fixture_case.file_name, ".mpn0")
+ ? 1.0
+ : kTs2DiffMaxPointValue;
+}
+
+// Java Math.round(x) == floor(x + 0.5) (ties towards +infinity).
+static double JavaRound(double x) { return std::floor(x + 0.5); }
+
+float ExpectedFloatValue(const FixtureCase& fixture_case, int row) {
+ if (!IsTs2DiffFloatCase(fixture_case)) {
+ return FloatValue(row);
+ }
+ float value = Ts2DiffFloatValue(row);
+ if (std::isnan(value)) {
+ // Java FloatEncoder normalizes any NaN to the canonical pattern
+ // via floatToIntBits.
+ return BitsToFloat(0x7fc00000U);
+ }
+ // Mirror FloatTS2DIFFEncoder::convert_float_to_int, then
+ // FloatDecoder::readFloat: scaled -> stored / mpv, scale-overflow ->
+ // stored / 1, raw bits -> intBitsToFloat.
+ const double mpv = Ts2DiffMaxPointValue(fixture_case);
+ const double scaled = static_cast<double>(value) * mpv;
+ const bool scaled_overflow =
+ scaled > static_cast<double>(std::numeric_limits<int32_t>::max()) ||
+ scaled < static_cast<double>(std::numeric_limits<int32_t>::min());
+ if (scaled_overflow) {
+ const bool value_overflows =
+ value > static_cast<float>(std::numeric_limits<int32_t>::max()) ||
+ value < static_cast<float>(std::numeric_limits<int32_t>::min());
+ if (value_overflows) {
+ // Raw IEEE bits, restored losslessly (infinite values here).
+ return value;
+ }
+ return static_cast<float>(JavaRound(value)) / 1.0f;
+ }
+ return static_cast<float>(JavaRound(scaled) / mpv);
+}
+
+double ExpectedDoubleValue(const FixtureCase& fixture_case, int row) {
+ if (!IsTs2DiffFloatCase(fixture_case)) {
+ return DoubleValue(row);
+ }
+ double value = Ts2DiffDoubleValue(row);
+ if (std::isnan(value)) {
+ return BitsToDouble(UINT64_C(0x7ff8000000000000));
+ }
+ const double mpv = Ts2DiffMaxPointValue(fixture_case);
+ const double scaled = value * mpv;
+ const bool scaled_overflow =
+ scaled > static_cast<double>(std::numeric_limits<int64_t>::max()) ||
+ scaled < static_cast<double>(std::numeric_limits<int64_t>::min());
+ if (scaled_overflow) {
+ const bool value_overflows =
+ value > static_cast<double>(std::numeric_limits<int64_t>::max()) ||
+ value < static_cast<double>(std::numeric_limits<int64_t>::min());
+ if (value_overflows) {
+ return value;
+ }
+ return JavaRound(value) / 1.0;
+ }
+ return JavaRound(scaled) / mpv;
+}
+
int32_t DateValue(int row) {
return kDateValues[row % (sizeof(kDateValues) / sizeof(kDateValues[0]))];
}
@@ -326,6 +477,12 @@ std::vector<FixtureCase> BuildMatrix() {
for (TSDataType data_type : data_types) {
cases.emplace_back(data_type, CHIMP, compression, kRowCount);
cases.emplace_back(data_type, RLBE, compression, kRowCount);
+ if (data_type == FLOAT || data_type == DOUBLE) {
+ // The value set exercises scaled, scale-overflow, and
+ // raw-bit forms, including the page-wide bitmaps across
+ // the 129-value TS_2DIFF block boundary.
+ cases.emplace_back(data_type, TS_2DIFF, compression, kRowCount);
+ }
}
cases.emplace_back(DOUBLE, CAMEL, compression, kRowCount);
}
@@ -346,8 +503,8 @@ TableSchema* CreateTableSchema(const FixtureCase& fixture_case) {
column_categories);
}
-void AddValue(Tablet& tablet, TSDataType data_type, int row) {
- switch (data_type) {
+void AddValue(Tablet& tablet, const FixtureCase& fixture_case, int row) {
+ switch (fixture_case.data_type) {
case INT32:
ASSERT_EQ(E_OK, tablet.add_value(row, kValueColumn, IntValue(row)));
break;
@@ -362,15 +519,17 @@ void AddValue(Tablet& tablet, TSDataType data_type, int row) {
break;
case FLOAT:
ASSERT_EQ(E_OK,
- tablet.add_value(row, kValueColumn, FloatValue(row)));
+ tablet.add_value(row, kValueColumn,
+ ExpectedFloatValue(fixture_case, row)));
break;
case DOUBLE:
ASSERT_EQ(E_OK,
- tablet.add_value(row, kValueColumn, DoubleValue(row)));
+ tablet.add_value(row, kValueColumn,
+ ExpectedDoubleValue(fixture_case, row)));
break;
default:
FAIL() << "Unsupported data type: "
- << get_data_type_name(data_type);
+ << get_data_type_name(fixture_case.data_type);
}
}
@@ -383,7 +542,7 @@ Tablet CreateTablet(TableSchema* table_schema,
for (int row = 0; row < fixture_case.row_count; ++row) {
EXPECT_EQ(E_OK, tablet.add_timestamp(row, row));
EXPECT_EQ(E_OK, tablet.add_value(row, kTagColumn, kTagValue));
- AddValue(tablet, fixture_case.data_type, row);
+ AddValue(tablet, fixture_case, row);
}
return tablet;
}
@@ -453,11 +612,11 @@ void AssertValue(const FixtureCase& fixture_case, int row,
ASSERT_EQ(LongValue(row), result_set->get_value<int64_t>(3));
break;
case FLOAT:
- ASSERT_EQ(FloatBits(FloatValue(row)),
+ ASSERT_EQ(FloatBits(ExpectedFloatValue(fixture_case, row)),
FloatBits(result_set->get_value<float>(3)));
break;
case DOUBLE:
- ASSERT_EQ(DoubleBits(DoubleValue(row)),
+ ASSERT_EQ(DoubleBits(ExpectedDoubleValue(fixture_case, row)),
DoubleBits(result_set->get_value<double>(3)));
break;
default:
diff --git a/java/tsfile/src/test/java/org/apache/tsfile/compatibility/TableModelEncodingCompressionCompatibilityTest.java b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/TableModelEncodingCompressionCompatibilityTest.java
index 83025c26f..80e67bc7f 100644
--- a/java/tsfile/src/test/java/org/apache/tsfile/compatibility/TableModelEncodingCompressionCompatibilityTest.java
+++ b/java/tsfile/src/test/java/org/apache/tsfile/compatibility/TableModelEncodingCompressionCompatibilityTest.java
@@ -19,6 +19,7 @@
package org.apache.tsfile.compatibility;
+import org.apache.tsfile.encoding.encoder.Encoder;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.TableSchema;
@@ -49,6 +50,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -64,7 +66,9 @@ public class TableModelEncodingCompressionCompatibilityTest {
private static final String TAG_COLUMN = "device";
private static final String VALUE_COLUMN = "value";
private static final String TAG_VALUE = "compat_device";
- private static final int ROW_COUNT = 32;
+ // Crosses the 129-value TS_2DIFF block boundary (300 -> 129+129+42), so
+ // page-wide bitmaps must survive block transitions. Applied to every case.
+ private static final int ROW_COUNT = 300;
private static final int[] INT_VALUES = {
0,
@@ -142,6 +146,44 @@ public class TableModelEncodingCompressionCompatibilityTest {
0xc05edd2f1a9fbe77L
};
+ // FLOAT/DOUBLE + TS_2DIFF converts values to fixed-point with
+ // maxPointNumber (10^mpn). The expected value is computed by applying the
+ // same tri-state conversion the writer performs and then decoding it back,
+ // so expectations reflect the encoding rules rather than the raw input
+ // bits. Every chosen value has the same expected value whether the writer
+ // used maxPointNumber 0 (explicit max_point_number property) or 2 (the
+ // TSFileConfig.floatPrecision default shared with the C++ writer), so the
+ // validating reader never needs to know which writer produced the file.
+ // The values cover all page layouts the writers can produce:
+ // - plain integers (scaled form; at mpn = 0 every finite in-range value
+ // is scaled, so pages written at mpn = 0 contain no bitmap)
+ // - 1.5e9f / 1e18: the scaled product overflows while round(v) still
+ // fits -> scale-overflow form, single page-wide bitmap (reachable only
+ // at mpn > 0, i.e. the default writer configuration)
+ // - NaN and +/-Infinity (stored as raw IEEE bits -> two page-wide
+ // bitmaps; Java floatToIntBits canonicalizes NaN)
+ private static final int[] TS_2DIFF_FLOAT_BITS = {
+ 0x00000000, 0x3f800000, 0xbf800000, 0x461c4000, 0xc61c4000, 0x4eb2d05e,
+ 0x7f800000, 0xff800000, 0x7fc00000, 0x3f800000, 0x00000000, 0x4a742400,
+ 0xca742400, 0x4e6e6b28, 0x3f800000, 0x00000000
+ };
+
+ private static final long[] TS_2DIFF_DOUBLE_BITS = {
+ 0x0000000000000000L, 0x3ff0000000000000L, 0xbff0000000000000L,
+ 0x40c3880000000000L, 0xc0c3880000000000L, 0x43abc16d674ec800L,
+ 0x7ff0000000000000L, 0xfff0000000000000L, 0x7ff8000000000000L,
+ 0x3ff0000000000000L, 0x0000000000000000L, 0x41f0000000000000L,
+ 0xc1cd6f3458800000L, 0x430c6bf526340000L, 0x3ff0000000000000L,
+ 0x0000000000000000L
+ };
+
+ // maxPointNumber used by the standard Java schema write path
+ // (TSEncodingBuilder.Ts2Diff initFromProps -> TSFileConfig.floatPrecision,
+ // current default 2); the C++ FloatTS2DIFFEncoder default matches.
+ private static final int TS_2DIFF_MAX_POINT_NUMBER = 2;
+ private static final double TS_2DIFF_MAX_POINT_VALUE =
+ TS_2DIFF_MAX_POINT_NUMBER <= 0 ? 1.0 : Math.pow(10, TS_2DIFF_MAX_POINT_NUMBER);
+
private static final LocalDate[] DATE_VALUES = {
LocalDate.of(1970, 1, 1),
LocalDate.of(1999, 12, 31),
@@ -196,6 +238,25 @@ public class TableModelEncodingCompressionCompatibilityTest {
TSDataType.DOUBLE)) {
cases.add(new FixtureCase(dataType, TSEncoding.CHIMP, compression, ROW_COUNT));
cases.add(new FixtureCase(dataType, TSEncoding.RLBE, compression, ROW_COUNT));
+ if (dataType == TSDataType.FLOAT || dataType == TSDataType.DOUBLE) {
+ // The value set exercises scaled, scale-overflow, and raw-bit
+ // forms, including the page-wide bitmaps across the 129-value
+ // TS_2DIFF block boundary.
+ cases.add(new FixtureCase(dataType, TSEncoding.TS_2DIFF, compression, ROW_COUNT));
+ // Explicit max_point_number=0 fixture: a canonical page starting
+ // with the 0x00 maxPointNumber byte must enter the Form 1
+ // parsing path.
+ cases.add(
+ new FixtureCase(
+ FixtureCase.fileName(dataType, TSEncoding.TS_2DIFF, compression) + ".mpn0",
+ TABLE_NAME,
+ TAG_COLUMN,
+ VALUE_COLUMN,
+ dataType,
+ TSEncoding.TS_2DIFF,
+ compression,
+ ROW_COUNT));
+ }
}
cases.add(new FixtureCase(TSDataType.DOUBLE, TSEncoding.CAMEL, compression, ROW_COUNT));
}
@@ -243,6 +304,10 @@ public class TableModelEncodingCompressionCompatibilityTest {
}
private static TableSchema tableSchema(FixtureCase fixtureCase) {
+ Map<String, String> props = null;
+ if (isTs2DiffFloatCase(fixtureCase) && fixtureCase.fileName.endsWith(".mpn0")) {
+ props = Collections.singletonMap(Encoder.MAX_POINT_NUMBER, "0");
+ }
List<IMeasurementSchema> schemas =
Arrays.asList(
new MeasurementSchema(
@@ -254,7 +319,8 @@ public class TableModelEncodingCompressionCompatibilityTest {
fixtureCase.valueColumn,
fixtureCase.dataType,
fixtureCase.encoding,
- fixtureCase.compression));
+ fixtureCase.compression,
+ props));
return new TableSchema(
fixtureCase.tableName, schemas, Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD));
}
@@ -270,12 +336,13 @@ public class TableModelEncodingCompressionCompatibilityTest {
for (int row = 0; row < fixtureCase.rowCount; row++) {
tablet.addTimestamp(row, row);
tablet.addValue(TAG_COLUMN, row, TAG_VALUE);
- addValue(tablet, fixtureCase.dataType, row);
+ addValue(tablet, fixtureCase, row);
}
return tablet;
}
- private static void addValue(Tablet tablet, TSDataType dataType, int row) {
+ private static void addValue(Tablet tablet, FixtureCase fixtureCase, int row) {
+ TSDataType dataType = fixtureCase.dataType;
switch (dataType) {
case INT32:
tablet.addValue(row, VALUE_COLUMN, intValue(row));
@@ -288,10 +355,10 @@ public class TableModelEncodingCompressionCompatibilityTest {
tablet.addValue(row, VALUE_COLUMN, longValue(row));
break;
case FLOAT:
- tablet.addValue(row, VALUE_COLUMN, floatValue(row));
+ tablet.addValue(row, VALUE_COLUMN, expectedFloatValue(fixtureCase, row));
break;
case DOUBLE:
- tablet.addValue(row, VALUE_COLUMN, doubleValue(row));
+ tablet.addValue(row, VALUE_COLUMN, expectedDoubleValue(fixtureCase, row));
break;
default:
throw new IllegalArgumentException("Unsupported data type: " + dataType);
@@ -313,13 +380,13 @@ public class TableModelEncodingCompressionCompatibilityTest {
case FLOAT:
assertEquals(
"FLOAT bits at row " + row,
- Float.floatToIntBits(floatValue(row)),
+ Float.floatToIntBits(expectedFloatValue(fixtureCase, row)),
Float.floatToIntBits(resultSet.getFloat(3)));
break;
case DOUBLE:
assertEquals(
"DOUBLE bits at row " + row,
- Double.doubleToLongBits(doubleValue(row)),
+ Double.doubleToLongBits(expectedDoubleValue(fixtureCase, row)),
Double.doubleToLongBits(resultSet.getDouble(3)));
break;
default:
@@ -343,6 +410,61 @@ public class TableModelEncodingCompressionCompatibilityTest {
return Double.longBitsToDouble(DOUBLE_BITS[row % DOUBLE_BITS.length]);
}
+ private static boolean isTs2DiffFloatCase(FixtureCase fixtureCase) {
+ return fixtureCase.encoding == TSEncoding.TS_2DIFF
+ && (fixtureCase.dataType == TSDataType.FLOAT || fixtureCase.dataType == TSDataType.DOUBLE);
+ }
+
+ // maxPointValue of the writer configuration for this fixture: the
+ // "*.mpn0" fixtures carry an explicit max_point_number=0 property, all
+ // others use the shared default (see TS_2DIFF_MAX_POINT_NUMBER).
+ private static double ts2DiffMaxPointValue(FixtureCase fixtureCase) {
+ return fixtureCase.fileName.endsWith(".mpn0") ? 1.0 : TS_2DIFF_MAX_POINT_VALUE;
+ }
+
+ // Mirror FloatEncoder/FloatDecoder at the writer's maxPointNumber:
+ // scaled -> round(v * mpv) / mpv, scale-overflow -> round(v) / 1,
+ // raw bits -> intBitsToFloat (NaN is canonicalized by floatToIntBits).
+ private static float expectedFloatValue(FixtureCase fixtureCase, int row) {
+ if (!isTs2DiffFloatCase(fixtureCase)) {
+ return floatValue(row);
+ }
+ float value = Float.intBitsToFloat(TS_2DIFF_FLOAT_BITS[row % TS_2DIFF_FLOAT_BITS.length]);
+ if (Float.isNaN(value)) {
+ return Float.intBitsToFloat(0x7fc00000);
+ }
+ double mpv = ts2DiffMaxPointValue(fixtureCase);
+ double scaled = (double) value * mpv;
+ if (scaled > Integer.MAX_VALUE || scaled < Integer.MIN_VALUE) {
+ // value itself stays in int range for the values used here
+ // (Infinity is caught below), so this is the scale-overflow form.
+ if (value > Integer.MAX_VALUE || value < Integer.MIN_VALUE) {
+ return Float.intBitsToFloat(Float.floatToIntBits(value));
+ }
+ return (float) ((double) Math.round(value) / 1.0);
+ }
+ return (float) ((double) Math.round(scaled) / mpv);
+ }
+
+ private static double expectedDoubleValue(FixtureCase fixtureCase, int row) {
+ if (!isTs2DiffFloatCase(fixtureCase)) {
+ return doubleValue(row);
+ }
+ double value = Double.longBitsToDouble(TS_2DIFF_DOUBLE_BITS[row % TS_2DIFF_DOUBLE_BITS.length]);
+ if (Double.isNaN(value)) {
+ return Double.longBitsToDouble(0x7ff8000000000000L);
+ }
+ double mpv = ts2DiffMaxPointValue(fixtureCase);
+ double scaled = value * mpv;
+ if (scaled > Long.MAX_VALUE || scaled < Long.MIN_VALUE) {
+ if (value > Long.MAX_VALUE || value < Long.MIN_VALUE) {
+ return Double.longBitsToDouble(Double.doubleToLongBits(value));
+ }
+ return (double) Math.round(value) / 1.0;
+ }
+ return (double) Math.round(scaled) / mpv;
+ }
+
private static LocalDate dateValue(int row) {
return DATE_VALUES[row % DATE_VALUES.length];
}