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(count, in, &Decoder::skip_int32, + &Decoder::read_int32); + } + + int skip_exact_int64(int count, common::ByteStream& in) { + return skip_exact_impl(count, in, &Decoder::skip_int64, + &Decoder::read_int64); + } + + int skip_exact_float(int count, common::ByteStream& in) { + return skip_exact_impl(count, in, &Decoder::skip_float, + &Decoder::read_float); + } + + int skip_exact_double(int count, common::ByteStream& in) { + return skip_exact_impl(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 + 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 + 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. 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. 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(nonnull_total) * elem_size); - int actual = 0; switch (dt) { case common::BOOLEAN: { bool* out = reinterpret_cast(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(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(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(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(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( 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( 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( 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( 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(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(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(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(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(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(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(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(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 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(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(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 #include +#include #include #include #include @@ -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 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 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>(schema_vec), + kRowCount); + auto non_aligned_schema = + std::make_shared>(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 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(time_iter.read(&len)); + const double v0 = + *reinterpret_cast(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(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(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(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.