This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 218bbdf76 fix(cpp): preserve Gorilla sentinel values in batch reads
(#883)
218bbdf76 is described below
commit 218bbdf7652ab79d806f53f6fa86111d99618d99
Author: Colin Lee <[email protected]>
AuthorDate: Mon Aug 3 09:52:03 2026 +0800
fix(cpp): preserve Gorilla sentinel values in batch reads (#883)
* fix(cpp): preserve Gorilla sentinel values in aligned reads
* fix(cpp): handle Gorilla sentinels across chunk readers
---
cpp/src/encoding/decoder.h | 103 ++++++++++++++++++
cpp/src/reader/aligned_chunk_reader.cc | 180 ++++++++++++++++----------------
cpp/src/reader/chunk_reader.cc | 83 +++++++--------
cpp/test/encoding/gorilla_codec_test.cc | 42 ++++++++
cpp/test/reader/tsfile_reader_test.cc | 119 +++++++++++++++++++++
5 files changed, 393 insertions(+), 134 deletions(-)
diff --git a/cpp/src/encoding/decoder.h b/cpp/src/encoding/decoder.h
index 70b9cde93..035023f8c 100644
--- a/cpp/src/encoding/decoder.h
+++ b/cpp/src/encoding/decoder.h
@@ -155,6 +155,52 @@ class Decoder {
return common::E_OK;
}
+ // Some encodings use an otherwise valid value as an end marker. Gorilla,
+ // for example, uses canonical NaN for FLOAT/DOUBLE and the minimum value
+ // for INT32/INT64. The regular batch API must honor that marker because it
+ // only receives a capacity. Callers that know the exact value count from
+ // page metadata can use these methods to disambiguate an embedded marker
+ // from the physical end marker.
+ int read_exact_int32(int32_t* out, int count, common::ByteStream& in) {
+ return read_exact_impl(out, count, in, &Decoder::read_batch_int32,
+ &Decoder::read_int32);
+ }
+
+ int read_exact_int64(int64_t* out, int count, common::ByteStream& in) {
+ return read_exact_impl(out, count, in, &Decoder::read_batch_int64,
+ &Decoder::read_int64);
+ }
+
+ int read_exact_float(float* out, int count, common::ByteStream& in) {
+ return read_exact_impl(out, count, in, &Decoder::read_batch_float,
+ &Decoder::read_float);
+ }
+
+ int read_exact_double(double* out, int count, common::ByteStream& in) {
+ return read_exact_impl(out, count, in, &Decoder::read_batch_double,
+ &Decoder::read_double);
+ }
+
+ int skip_exact_int32(int count, common::ByteStream& in) {
+ return skip_exact_impl<int32_t>(count, in, &Decoder::skip_int32,
+ &Decoder::read_int32);
+ }
+
+ int skip_exact_int64(int count, common::ByteStream& in) {
+ return skip_exact_impl<int64_t>(count, in, &Decoder::skip_int64,
+ &Decoder::read_int64);
+ }
+
+ int skip_exact_float(int count, common::ByteStream& in) {
+ return skip_exact_impl<float>(count, in, &Decoder::skip_float,
+ &Decoder::read_float);
+ }
+
+ int skip_exact_double(int count, common::ByteStream& in) {
+ return skip_exact_impl<double>(count, in, &Decoder::skip_double,
+ &Decoder::read_double);
+ }
+
// Block-level filter pushdown for TS_2DIFF-encoded INT64 columns.
//
// TS_2DIFF stores values in self-contained "blocks": a header (value
@@ -188,6 +234,63 @@ class Decoder {
virtual int skip_peeked_block_int64(common::ByteStream& in, int& skipped) {
return common::E_NOT_SUPPORT;
}
+
+ private:
+ template <typename T>
+ int read_exact_impl(T* out, int count, common::ByteStream& in,
+ int (Decoder::*read_batch)(T*, int, int&,
+ common::ByteStream&),
+ int (Decoder::*read_one)(T&, common::ByteStream&)) {
+ if (count < 0 || (count > 0 && out == nullptr)) {
+ return common::E_INVALID_ARG;
+ }
+
+ int actual = 0;
+ while (actual < count) {
+ int batch_actual = 0;
+ int ret = (this->*read_batch)(out + actual, count - actual,
+ batch_actual, in);
+ if (ret != common::E_OK) return ret;
+ if (batch_actual < 0 || batch_actual > count - actual) {
+ return common::E_TSFILE_CORRUPTED;
+ }
+ actual += batch_actual;
+ if (actual == count) return common::E_OK;
+
+ if ((ret = (this->*read_one)(out[actual], in)) != common::E_OK) {
+ return ret;
+ }
+ ++actual;
+ }
+ return common::E_OK;
+ }
+
+ template <typename T>
+ int skip_exact_impl(int count, common::ByteStream& in,
+ int (Decoder::*skip_batch)(int, int&,
+ common::ByteStream&),
+ int (Decoder::*read_one)(T&, common::ByteStream&)) {
+ if (count < 0) return common::E_INVALID_ARG;
+
+ int skipped = 0;
+ while (skipped < count) {
+ int batch_skipped = 0;
+ int ret = (this->*skip_batch)(count - skipped, batch_skipped, in);
+ if (ret != common::E_OK) return ret;
+ if (batch_skipped < 0 || batch_skipped > count - skipped) {
+ return common::E_TSFILE_CORRUPTED;
+ }
+ skipped += batch_skipped;
+ if (skipped == count) return common::E_OK;
+
+ T ignored;
+ if ((ret = (this->*read_one)(ignored, in)) != common::E_OK) {
+ return ret;
+ }
+ ++skipped;
+ }
+ return common::E_OK;
+ }
};
} // end namespace storage
diff --git a/cpp/src/reader/aligned_chunk_reader.cc
b/cpp/src/reader/aligned_chunk_reader.cc
index 795ec6b80..97c469288 100644
--- a/cpp/src/reader/aligned_chunk_reader.cc
+++ b/cpp/src/reader/aligned_chunk_reader.cc
@@ -784,40 +784,39 @@ int AlignedChunkReader::i32_DECODE_TYPED_TV_INTO_TSBLOCK(
}
namespace {
-// Type-dispatched value batch read / skip for decode_tv_batch<T>. Overload
-// resolution on the value pointer type selects the matching Decoder method, so
-// the four fixed-width value types share one decode loop.
-FORCE_INLINE int read_value_batch_typed(Decoder* d, int32_t* out, int cap,
- int& actual, ByteStream& in) {
- return d->read_batch_int32(out, cap, actual, in);
+// Type-dispatched exact read / skip for decode_tv_batch<T>. Overload
+// resolution on the value pointer selects the matching Decoder method.
+FORCE_INLINE int read_value_exact_typed(Decoder* d, int32_t* out, int count,
+ ByteStream& in) {
+ return d->read_exact_int32(out, count, in);
}
-FORCE_INLINE int read_value_batch_typed(Decoder* d, int64_t* out, int cap,
- int& actual, ByteStream& in) {
- return d->read_batch_int64(out, cap, actual, in);
+FORCE_INLINE int read_value_exact_typed(Decoder* d, int64_t* out, int count,
+ ByteStream& in) {
+ return d->read_exact_int64(out, count, in);
}
-FORCE_INLINE int read_value_batch_typed(Decoder* d, float* out, int cap,
- int& actual, ByteStream& in) {
- return d->read_batch_float(out, cap, actual, in);
+FORCE_INLINE int read_value_exact_typed(Decoder* d, float* out, int count,
+ ByteStream& in) {
+ return d->read_exact_float(out, count, in);
}
-FORCE_INLINE int read_value_batch_typed(Decoder* d, double* out, int cap,
- int& actual, ByteStream& in) {
- return d->read_batch_double(out, cap, actual, in);
+FORCE_INLINE int read_value_exact_typed(Decoder* d, double* out, int count,
+ ByteStream& in) {
+ return d->read_exact_double(out, count, in);
}
-FORCE_INLINE int skip_value_typed(Decoder* d, int32_t*, int n, int& skipped,
- ByteStream& in) {
- return d->skip_int32(n, skipped, in);
+FORCE_INLINE int skip_value_exact_typed(Decoder* d, int32_t*, int count,
+ ByteStream& in) {
+ return d->skip_exact_int32(count, in);
}
-FORCE_INLINE int skip_value_typed(Decoder* d, int64_t*, int n, int& skipped,
- ByteStream& in) {
- return d->skip_int64(n, skipped, in);
+FORCE_INLINE int skip_value_exact_typed(Decoder* d, int64_t*, int count,
+ ByteStream& in) {
+ return d->skip_exact_int64(count, in);
}
-FORCE_INLINE int skip_value_typed(Decoder* d, float*, int n, int& skipped,
- ByteStream& in) {
- return d->skip_float(n, skipped, in);
+FORCE_INLINE int skip_value_exact_typed(Decoder* d, float*, int count,
+ ByteStream& in) {
+ return d->skip_exact_float(count, in);
}
-FORCE_INLINE int skip_value_typed(Decoder* d, double*, int n, int& skipped,
- ByteStream& in) {
- return d->skip_double(n, skipped, in);
+FORCE_INLINE int skip_value_exact_typed(Decoder* d, double*, int count,
+ ByteStream& in) {
+ return d->skip_exact_double(count, in);
}
} // namespace
@@ -864,17 +863,8 @@ int AlignedChunkReader::decode_tv_batch(ByteStream&
time_in,
}
cur_value_index += block_count;
if (nonnull > 0) {
- // skip_* may legitimately fail (truncated page) or
- // short-read (corrupt bitmap vs. data); both must
abort
- // the loop rather than silently desync the value
- // decoder.
- int sk = 0;
- if (RET_FAIL(skip_value_typed(value_decoder_, values,
- nonnull, sk, value_in)))
{
- break;
- }
- if (sk != nonnull) {
- ret = E_TSFILE_CORRUPTED;
+ if (RET_FAIL(skip_value_exact_typed(
+ value_decoder_, values, nonnull, value_in))) {
break;
}
}
@@ -916,14 +906,8 @@ int AlignedChunkReader::decode_tv_batch(ByteStream&
time_in,
if (pass_count == 0) {
if (nonnull_count > 0) {
- int skipped = 0;
- if (RET_FAIL(skip_value_typed(value_decoder_, values,
- nonnull_count, skipped,
- value_in))) {
- break;
- }
- if (skipped != nonnull_count) {
- ret = E_TSFILE_CORRUPTED;
+ if (RET_FAIL(skip_value_exact_typed(value_decoder_, values,
+ nonnull_count, value_in)))
{
break;
}
}
@@ -931,11 +915,9 @@ int AlignedChunkReader::decode_tv_batch(ByteStream&
time_in,
continue;
}
- int value_count = 0;
if (nonnull_count > 0) {
- if (RET_FAIL(read_value_batch_typed(value_decoder_, values,
- nonnull_count, value_count,
- value_in))) {
+ if (RET_FAIL(read_value_exact_typed(value_decoder_, values,
+ nonnull_count, value_in))) {
break;
}
}
@@ -1704,7 +1686,6 @@ int
AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx,
uint32_t elem_size = common::get_data_type_size(dt);
pps.predecoded_values.resize(static_cast<size_t>(nonnull_total) *
elem_size);
- int actual = 0;
switch (dt) {
case common::BOOLEAN: {
bool* out = reinterpret_cast<bool*>(pps.predecoded_values.data());
@@ -1714,39 +1695,38 @@ int
AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx,
return ret;
}
}
- actual = nonnull_total;
break;
}
case common::INT32:
case common::DATE:
- if (RET_FAIL(col->decoder->read_batch_int32(
+ if (RET_FAIL(col->decoder->read_exact_int32(
reinterpret_cast<int32_t*>(pps.predecoded_values.data()),
- nonnull_total, actual, in))) {
+ nonnull_total, in))) {
cleanup();
return ret;
}
break;
case common::INT64:
case common::TIMESTAMP:
- if (RET_FAIL(col->decoder->read_batch_int64(
+ if (RET_FAIL(col->decoder->read_exact_int64(
reinterpret_cast<int64_t*>(pps.predecoded_values.data()),
- nonnull_total, actual, in))) {
+ nonnull_total, in))) {
cleanup();
return ret;
}
break;
case common::FLOAT:
- if (RET_FAIL(col->decoder->read_batch_float(
+ if (RET_FAIL(col->decoder->read_exact_float(
reinterpret_cast<float*>(pps.predecoded_values.data()),
- nonnull_total, actual, in))) {
+ nonnull_total, in))) {
cleanup();
return ret;
}
break;
case common::DOUBLE:
- if (RET_FAIL(col->decoder->read_batch_double(
+ if (RET_FAIL(col->decoder->read_exact_double(
reinterpret_cast<double*>(pps.predecoded_values.data()),
- nonnull_total, actual, in))) {
+ nonnull_total, in))) {
cleanup();
return ret;
}
@@ -1755,7 +1735,7 @@ int
AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx,
cleanup();
return E_NOT_SUPPORT;
}
- pps.predecoded_count = actual;
+ pps.predecoded_count = nonnull_total;
cleanup();
return E_OK;
}
@@ -2372,35 +2352,39 @@ int
AlignedChunkReader::decompress_and_parse_value_page(ValueColumnState& col,
break;
}
out[i] = v;
+ actual++;
}
- actual = nonnull_total;
break;
}
case common::INT32:
case common::DATE:
- rret = col.decoder->read_batch_int32(
+ rret = col.decoder->read_exact_int32(
reinterpret_cast<int32_t*>(
col.pending_decoded_values.data()),
- nonnull_total, actual, col.in);
+ nonnull_total, col.in);
+ if (rret == common::E_OK) actual = nonnull_total;
break;
case common::INT64:
case common::TIMESTAMP:
- rret = col.decoder->read_batch_int64(
+ rret = col.decoder->read_exact_int64(
reinterpret_cast<int64_t*>(
col.pending_decoded_values.data()),
- nonnull_total, actual, col.in);
+ nonnull_total, col.in);
+ if (rret == common::E_OK) actual = nonnull_total;
break;
case common::FLOAT:
- rret = col.decoder->read_batch_float(
+ rret = col.decoder->read_exact_float(
reinterpret_cast<float*>(
col.pending_decoded_values.data()),
- nonnull_total, actual, col.in);
+ nonnull_total, col.in);
+ if (rret == common::E_OK) actual = nonnull_total;
break;
case common::DOUBLE:
- rret = col.decoder->read_batch_double(
+ rret = col.decoder->read_exact_double(
reinterpret_cast<double*>(
col.pending_decoded_values.data()),
- nonnull_total, actual, col.in);
+ nonnull_total, col.in);
+ if (rret == common::E_OK) actual = nonnull_total;
break;
default:
rret = common::E_OUT_OF_RANGE;
@@ -2524,11 +2508,9 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock*
ret_tsblock,
}
}
- // Skip values if no rows pass time filter. Skip/read errors and
- // short reads (decoder returned fewer values than the bitmap
- // promised) must abort; otherwise the input stream is left
- // mid-value and later batches would decode garbage from
- // misaligned bytes.
+ // Skip values if no rows pass time filter. The aligned bitmap
+ // supplies the exact count, including values equal to a codec's
+ // in-band terminator.
if (pass_count == 0 && cb.nonnull_count > 0) {
int dret = common::E_OK;
int sk = 0;
@@ -2543,21 +2525,29 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock*
ret_tsblock,
}
case common::INT32:
case common::DATE:
- dret = col->decoder->skip_int32(cb.nonnull_count, sk,
- col->in);
+ dret = skip_value_exact_typed(
+ col->decoder, static_cast<int32_t*>(nullptr),
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK) sk = cb.nonnull_count;
break;
case common::INT64:
case common::TIMESTAMP:
- dret = col->decoder->skip_int64(cb.nonnull_count, sk,
- col->in);
+ dret = skip_value_exact_typed(
+ col->decoder, static_cast<int64_t*>(nullptr),
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK) sk = cb.nonnull_count;
break;
case common::FLOAT:
- dret = col->decoder->skip_float(cb.nonnull_count, sk,
- col->in);
+ dret = skip_value_exact_typed(
+ col->decoder, static_cast<float*>(nullptr),
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK) sk = cb.nonnull_count;
break;
case common::DOUBLE:
- dret = col->decoder->skip_double(cb.nonnull_count, sk,
- col->in);
+ dret = skip_value_exact_typed(
+ col->decoder, static_cast<double*>(nullptr),
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK) sk = cb.nonnull_count;
break;
case common::STRING:
case common::TEXT:
@@ -2621,25 +2611,33 @@ int AlignedChunkReader::multi_DECODE_TV_BATCH(TsBlock*
ret_tsblock,
}
case common::INT32:
case common::DATE:
- dret = col->decoder->read_batch_int32(
+ dret = col->decoder->read_exact_int32(
reinterpret_cast<int32_t*>(cb.val_buf),
- cb.nonnull_count, cb.val_count, col->in);
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK)
+ cb.val_count = cb.nonnull_count;
break;
case common::INT64:
case common::TIMESTAMP:
- dret = col->decoder->read_batch_int64(
+ dret = col->decoder->read_exact_int64(
reinterpret_cast<int64_t*>(cb.val_buf),
- cb.nonnull_count, cb.val_count, col->in);
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK)
+ cb.val_count = cb.nonnull_count;
break;
case common::FLOAT:
- dret = col->decoder->read_batch_float(
+ dret = col->decoder->read_exact_float(
reinterpret_cast<float*>(cb.val_buf),
- cb.nonnull_count, cb.val_count, col->in);
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK)
+ cb.val_count = cb.nonnull_count;
break;
case common::DOUBLE:
- dret = col->decoder->read_batch_double(
+ dret = col->decoder->read_exact_double(
reinterpret_cast<double*>(cb.val_buf),
- cb.nonnull_count, cb.val_count, col->in);
+ cb.nonnull_count, col->in);
+ if (dret == common::E_OK)
+ cb.val_count = cb.nonnull_count;
break;
case common::STRING:
case common::TEXT:
diff --git a/cpp/src/reader/chunk_reader.cc b/cpp/src/reader/chunk_reader.cc
index ce8d87a58..b21b1b4c5 100644
--- a/cpp/src/reader/chunk_reader.cc
+++ b/cpp/src/reader/chunk_reader.cc
@@ -479,7 +479,10 @@ int ChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in,
ByteStream& value_in,
if (!filter->satisfy_start_end_time(block_min, block_max)) {
int skipped = 0;
time_decoder_->skip_peeked_block_int64(time_in, skipped);
- value_decoder_->skip_int32(block_count, skipped, value_in);
+ if (RET_FAIL(value_decoder_->skip_exact_int32(block_count,
+ value_in))) {
+ break;
+ }
continue;
}
if (filter->contain_start_end_time(block_min, block_max)) {
@@ -489,7 +492,6 @@ int ChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in,
ByteStream& value_in,
}
int time_count = 0;
- int value_count = 0;
if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch,
time_count, time_in))) {
@@ -505,20 +507,15 @@ int ChunkReader::i32_DECODE_TV_BATCH(ByteStream& time_in,
ByteStream& value_in,
}
if (pass_count == 0) {
- int skipped = 0;
- value_decoder_->skip_int32(time_count, skipped, value_in);
+ if (RET_FAIL(
+ value_decoder_->skip_exact_int32(time_count, value_in))) {
+ break;
+ }
continue;
}
- if (RET_FAIL(value_decoder_->read_batch_int32(values, time_count,
- value_count, value_in)))
{
- break;
- }
- // Time and value chunks are written in lock-step; any discrepancy
- // means the file is truncated or corrupted. Reading uninitialised
- // values[i] would silently surface garbage as decoded rows.
- if (value_count != time_count) {
- ret = E_TSFILE_CORRUPTED;
+ if (RET_FAIL(value_decoder_->read_exact_int32(values, time_count,
+ value_in))) {
break;
}
@@ -568,7 +565,10 @@ int ChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in,
ByteStream& value_in,
if (!filter->satisfy_start_end_time(block_min, block_max)) {
int skipped = 0;
time_decoder_->skip_peeked_block_int64(time_in, skipped);
- value_decoder_->skip_int64(block_count, skipped, value_in);
+ if (RET_FAIL(value_decoder_->skip_exact_int64(block_count,
+ value_in))) {
+ break;
+ }
continue;
}
if (filter->contain_start_end_time(block_min, block_max)) {
@@ -578,7 +578,6 @@ int ChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in,
ByteStream& value_in,
}
int time_count = 0;
- int value_count = 0;
if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch,
time_count, time_in))) {
@@ -594,17 +593,15 @@ int ChunkReader::i64_DECODE_TV_BATCH(ByteStream& time_in,
ByteStream& value_in,
}
if (pass_count == 0) {
- int skipped = 0;
- value_decoder_->skip_int64(time_count, skipped, value_in);
+ if (RET_FAIL(
+ value_decoder_->skip_exact_int64(time_count, value_in))) {
+ break;
+ }
continue;
}
- if (RET_FAIL(value_decoder_->read_batch_int64(values, time_count,
- value_count, value_in)))
{
- break;
- }
- if (value_count != time_count) {
- ret = E_TSFILE_CORRUPTED;
+ if (RET_FAIL(value_decoder_->read_exact_int64(values, time_count,
+ value_in))) {
break;
}
@@ -655,7 +652,10 @@ int ChunkReader::float_DECODE_TV_BATCH(ByteStream& time_in,
if (!filter->satisfy_start_end_time(block_min, block_max)) {
int skipped = 0;
time_decoder_->skip_peeked_block_int64(time_in, skipped);
- value_decoder_->skip_float(block_count, skipped, value_in);
+ if (RET_FAIL(value_decoder_->skip_exact_float(block_count,
+ value_in))) {
+ break;
+ }
continue;
}
if (filter->contain_start_end_time(block_min, block_max)) {
@@ -665,7 +665,6 @@ int ChunkReader::float_DECODE_TV_BATCH(ByteStream& time_in,
}
int time_count = 0;
- int value_count = 0;
if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch,
time_count, time_in))) {
@@ -681,17 +680,15 @@ int ChunkReader::float_DECODE_TV_BATCH(ByteStream&
time_in,
}
if (pass_count == 0) {
- int skipped = 0;
- value_decoder_->skip_float(time_count, skipped, value_in);
+ if (RET_FAIL(
+ value_decoder_->skip_exact_float(time_count, value_in))) {
+ break;
+ }
continue;
}
- if (RET_FAIL(value_decoder_->read_batch_float(values, time_count,
- value_count, value_in)))
{
- break;
- }
- if (value_count != time_count) {
- ret = E_TSFILE_CORRUPTED;
+ if (RET_FAIL(value_decoder_->read_exact_float(values, time_count,
+ value_in))) {
break;
}
@@ -738,7 +735,10 @@ int ChunkReader::double_DECODE_TV_BATCH(ByteStream&
time_in,
if (!filter->satisfy_start_end_time(block_min, block_max)) {
int skipped = 0;
time_decoder_->skip_peeked_block_int64(time_in, skipped);
- value_decoder_->skip_double(block_count, skipped,
value_in);
+ if (RET_FAIL(value_decoder_->skip_exact_double(block_count,
+ value_in)))
{
+ break;
+ }
continue;
}
if (filter->contain_start_end_time(block_min, block_max)) {
@@ -748,7 +748,6 @@ int ChunkReader::double_DECODE_TV_BATCH(ByteStream& time_in,
}
int time_count = 0;
- int value_count = 0;
if (RET_FAIL(time_decoder_->read_batch_int64(times, eff_batch,
time_count, time_in))) {
@@ -764,17 +763,15 @@ int ChunkReader::double_DECODE_TV_BATCH(ByteStream&
time_in,
}
if (pass_count == 0) {
- int skipped = 0;
- value_decoder_->skip_double(time_count, skipped, value_in);
+ if (RET_FAIL(
+ value_decoder_->skip_exact_double(time_count, value_in))) {
+ break;
+ }
continue;
}
- if (RET_FAIL(value_decoder_->read_batch_double(
- values, time_count, value_count, value_in))) {
- break;
- }
- if (value_count != time_count) {
- ret = E_TSFILE_CORRUPTED;
+ if (RET_FAIL(value_decoder_->read_exact_double(values, time_count,
+ value_in))) {
break;
}
diff --git a/cpp/test/encoding/gorilla_codec_test.cc
b/cpp/test/encoding/gorilla_codec_test.cc
index b1fe72136..039a9a4f0 100644
--- a/cpp/test/encoding/gorilla_codec_test.cc
+++ b/cpp/test/encoding/gorilla_codec_test.cc
@@ -486,6 +486,48 @@ TEST_F(GorillaCodecTest,
DoubleBatchScalarAndSkipInterleave) {
EXPECT_EQ(cursor, N);
}
+TEST_F(GorillaCodecTest, DoubleExactReadAndSkipPreserveEmbeddedNaN) {
+ const double expected[] = {1.25, std::nan(""), 2.5, std::nan(""), 5.0};
+ const int count = sizeof(expected) / sizeof(expected[0]);
+
+ storage::DoubleGorillaEncoder encoder;
+ common::ByteStream stream(1024, common::MOD_DEFAULT);
+ for (double value : expected) {
+ ASSERT_EQ(encoder.encode(value, stream), common::E_OK);
+ }
+ encoder.flush(stream);
+
+ const uint32_t total = stream.total_size();
+ std::vector<uint8_t> buf(total);
+ uint32_t got = 0;
+ stream.read_buf(buf.data(), total, got);
+ ASSERT_EQ(got, total);
+
+ common::ByteStream read_stream(common::MOD_DEFAULT);
+ read_stream.wrap_from(reinterpret_cast<const char*>(buf.data()), total);
+ storage::DoubleGorillaDecoder read_decoder;
+ double actual[count];
+ ASSERT_EQ(read_decoder.read_exact_double(actual, count, read_stream),
+ common::E_OK);
+ for (int i = 0; i < count; i++) {
+ if (std::isnan(expected[i])) {
+ EXPECT_TRUE(std::isnan(actual[i])) << "i=" << i;
+ } else {
+ EXPECT_DOUBLE_EQ(actual[i], expected[i]) << "i=" << i;
+ }
+ }
+
+ common::ByteStream skip_stream(common::MOD_DEFAULT);
+ skip_stream.wrap_from(reinterpret_cast<const char*>(buf.data()), total);
+ storage::DoubleGorillaDecoder skip_decoder;
+ ASSERT_EQ(skip_decoder.skip_exact_double(3, skip_stream), common::E_OK);
+ double tail[2];
+ ASSERT_EQ(skip_decoder.read_exact_double(tail, 2, skip_stream),
+ common::E_OK);
+ EXPECT_TRUE(std::isnan(tail[0]));
+ EXPECT_DOUBLE_EQ(tail[1], 5.0);
+}
+
TEST_F(GorillaCodecTest, DoubleBatchDecodeFullWidthXor) {
const uint64_t patterns[] = {
0x0000000000000000ULL, 0xFFFFFFFFFFFFFFFFULL, 0x0123456789ABCDEFULL,
diff --git a/cpp/test/reader/tsfile_reader_test.cc
b/cpp/test/reader/tsfile_reader_test.cc
index f7df9c8c9..cdb6d33cf 100644
--- a/cpp/test/reader/tsfile_reader_test.cc
+++ b/cpp/test/reader/tsfile_reader_test.cc
@@ -21,6 +21,7 @@
#include <gtest/gtest.h>
#include <sys/stat.h>
+#include <cmath>
#include <map>
#include <random>
#include <unordered_map>
@@ -1343,6 +1344,124 @@ TEST_F(TsFileReaderTest,
MultiValueAlignedWideChunkParallelDecode) {
io_reader.revert_ssi(ssi);
}
+// Regression: Gorilla uses the canonical NaN bit pattern as its stream
+// terminator. When the same bit pattern appears as a value, count-aware reads
+// must preserve it and every following value instead of stopping early.
+TEST_F(TsFileReaderTest, GorillaDoubleNaNPreservedByChunkReaders) {
+ const std::string aligned_device = "root.dev_aligned_gorilla_nan";
+ const std::string non_aligned_device = "root.dev_gorilla_nan";
+ std::vector<MeasurementSchema> schema_vec;
+ schema_vec.emplace_back("v0", DOUBLE, GORILLA, UNCOMPRESSED);
+ schema_vec.emplace_back("v1", DOUBLE, GORILLA, UNCOMPRESSED);
+ ASSERT_EQ(
+ tsfile_writer_->register_timeseries(non_aligned_device, schema_vec[0]),
+ E_OK);
+ {
+ std::vector<MeasurementSchema*> registered;
+ for (const auto& schema : schema_vec) {
+ registered.push_back(new MeasurementSchema(schema));
+ }
+ ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(aligned_device,
+ registered),
+ E_OK);
+ }
+
+ constexpr int kRowCount = 32;
+ constexpr int kNaNRow = 7;
+ Tablet aligned_tablet(
+ aligned_device,
+ std::make_shared<std::vector<MeasurementSchema>>(schema_vec),
+ kRowCount);
+ auto non_aligned_schema =
+ std::make_shared<std::vector<MeasurementSchema>>(1, schema_vec[0]);
+ Tablet non_aligned_tablet(non_aligned_device, non_aligned_schema,
+ kRowCount);
+ for (int row = 0; row < kRowCount; row++) {
+ ASSERT_EQ(aligned_tablet.add_timestamp(row, row), E_OK);
+ ASSERT_EQ(non_aligned_tablet.add_timestamp(row, row), E_OK);
+ const double v0 = row == kNaNRow ? std::nan("") : 1000.0 + row;
+ ASSERT_EQ(aligned_tablet.add_value(row, 0u, v0), E_OK);
+ ASSERT_EQ(aligned_tablet.add_value(row, 1u, 2000.0 + row), E_OK);
+ ASSERT_EQ(non_aligned_tablet.add_value(row, 0u, v0), E_OK);
+ }
+ ASSERT_EQ(tsfile_writer_->write_tablet(non_aligned_tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->write_tablet_aligned(aligned_tablet), E_OK);
+ ASSERT_EQ(tsfile_writer_->flush(), E_OK);
+ ASSERT_EQ(tsfile_writer_->close(), E_OK);
+
+ storage::TsFileIOReader io_reader;
+ ASSERT_EQ(io_reader.init(file_name_), E_OK);
+
+ auto check_values = [&](storage::TsFileSeriesScanIterator* ssi,
+ bool has_v1) {
+ common::TsBlock* block = nullptr;
+ ASSERT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true), E_OK);
+ ASSERT_NE(block, nullptr);
+ ASSERT_EQ(block->get_row_count(), kRowCount);
+
+ {
+ common::ColIterator time_iter(0, block);
+ common::ColIterator v0_iter(1, block);
+ std::unique_ptr<common::ColIterator> v1_iter;
+ if (has_v1) {
+ v1_iter.reset(new common::ColIterator(2, block));
+ }
+ for (int row = 0; row < kRowCount; row++) {
+ uint32_t len = 0;
+ const int64_t time =
+ *reinterpret_cast<int64_t*>(time_iter.read(&len));
+ const double v0 =
+ *reinterpret_cast<double*>(v0_iter.read(&len));
+ EXPECT_EQ(time, row);
+ if (row == kNaNRow) {
+ EXPECT_TRUE(std::isnan(v0));
+ } else {
+ EXPECT_DOUBLE_EQ(v0, 1000.0 + row);
+ }
+ if (has_v1) {
+ const double v1 =
+ *reinterpret_cast<double*>(v1_iter->read(&len));
+ EXPECT_DOUBLE_EQ(v1, 2000.0 + row);
+ v1_iter->next();
+ }
+ time_iter.next();
+ v0_iter.next();
+ }
+ }
+ ssi->revert_tsblock();
+
+ block = nullptr;
+ EXPECT_EQ(ssi->get_next(block, /*alloc_tsblock=*/true),
E_NO_MORE_DATA);
+ };
+
+ {
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ auto device_id =
+ std::make_shared<StringArrayDeviceID>(non_aligned_device);
+ ASSERT_EQ(io_reader.alloc_ssi(device_id, "v0", ssi, pa,
+ /*time_filter=*/nullptr),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+ check_values(ssi, /*has_v1=*/false);
+ io_reader.revert_ssi(ssi);
+ }
+
+ {
+ storage::TsFileSeriesScanIterator* ssi = nullptr;
+ common::PageArena pa;
+ pa.init(512, common::MOD_TSFILE_READER);
+ auto device_id = std::make_shared<StringArrayDeviceID>(aligned_device);
+ ASSERT_EQ(io_reader.alloc_multi_ssi(device_id, {"v0", "v1"}, ssi, pa,
+ /*time_filter=*/nullptr),
+ E_OK);
+ ASSERT_NE(ssi, nullptr);
+ check_values(ssi, /*has_v1=*/true);
+ io_reader.revert_ssi(ssi);
+ }
+}
+
// Regression: AlignedTimeseriesIndex::get_data_type() returns the time column
// type (VECTOR), which the schema accessor used to surface verbatim — every
// aligned column came back as VECTOR instead of its real INT32/FLOAT/etc.
