diff --git a/cpp/src/common/tsblock/tsblock.h b/cpp/src/common/tsblock/tsblock.h index 7959f7c36..056186211 100644 --- a/cpp/src/common/tsblock/tsblock.h +++ b/cpp/src/common/tsblock/tsblock.h @@ -174,6 +174,28 @@ class RowAppender { } } + FORCE_INLINE bool can_bulk_append_fixed(uint32_t slot_index, + uint32_t elem_size) const { + ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count()); + Vector* vec = tsblock_->vectors_[slot_index]; + TSDataType datatype = vec->get_vector_type(); + if (datatype == STRING || datatype == TEXT || datatype == BLOB) { + return false; + } + return static_cast(vec)->get_type_len() == + elem_size; + } + + FORCE_INLINE void bulk_append_fixed(uint32_t slot_index, const char* values, + uint32_t count) { + ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count()); + Vector* vec = tsblock_->vectors_[slot_index]; + ASSERT(vec->get_vector_type() != STRING && + vec->get_vector_type() != TEXT && + vec->get_vector_type() != BLOB); + static_cast(vec)->append_batch(values, count); + } + FORCE_INLINE void append_null(uint32_t slot_index) { Vector* vec = tsblock_->vectors_[slot_index]; vec->set_null(tsblock_->row_count_ - 1); diff --git a/cpp/src/common/tsblock/vector/fixed_length_vector.h b/cpp/src/common/tsblock/vector/fixed_length_vector.h index df20c97d9..eb5596343 100644 --- a/cpp/src/common/tsblock/vector/fixed_length_vector.h +++ b/cpp/src/common/tsblock/vector/fixed_length_vector.h @@ -49,6 +49,13 @@ class FixedLengthVector : public Vector { values_.append_fixed_value(value, len); } + FORCE_INLINE void append_batch(const char* values, uint32_t count) { + values_.append_fixed_value(values, count * type_len_); + add_row_nums(count); + } + + FORCE_INLINE uint32_t get_type_len() const { return type_len_; } + // cppcheck-suppress missingOverride FORCE_INLINE char* read(uint32_t* __restrict len, bool* __restrict null, uint32_t rowid) OVERRIDE { diff --git a/cpp/src/encoding/gorilla_decoder.h b/cpp/src/encoding/gorilla_decoder.h index c8c76d965..ccbbdef88 100644 --- a/cpp/src/encoding/gorilla_decoder.h +++ b/cpp/src/encoding/gorilla_decoder.h @@ -19,8 +19,13 @@ #ifndef ENCODING_GORILLA_DECODER_H #define ENCODING_GORILLA_DECODER_H +#include #include +#if defined(_MSC_VER) +#include +#endif + #include "common/allocator/byte_stream.h" #include "decoder.h" #include "encode_utils.h" @@ -30,6 +35,23 @@ namespace storage { +FORCE_INLINE int gorilla_count_leading_zeros_nonzero(uint64_t value) { +#if defined(__GNUC__) || defined(__clang__) + return __builtin_clzll(value); +#elif defined(_MSC_VER) + unsigned long index; + _BitScanReverse64(&index, value); + return 63 - static_cast(index); +#else + int count = 0; + while ((value & (UINT64_C(1) << 63)) == 0) { + value <<= 1; + ++count; + } + return count; +#endif +} + // ── Raw-pointer bit reader ──────────────────────────────────────────────── // Operates directly on a contiguous byte array, bypassing ByteStream's // per-byte read_buf() overhead (atomic loads, page boundary checks, memcpy). @@ -91,6 +113,37 @@ struct GorillaBitReader { return bit; } + // Consume up to max_count consecutive zero control bits. The first one bit + // remains unread so the normal control decoder can handle the following + // changed value. This turns long repeated-value runs into one leading-zero + // count per reservoir instead of one read_next() call per value. + FORCE_INLINE int consume_zero_bits(int max_count) { + int consumed = 0; + while (consumed < max_count) { + if (UNLIKELY(!refill_if_empty())) { + break; + } + + const int available_bits = bits; + const uint64_t aligned = + available_bits == 64 ? buffer : buffer << (64 - available_bits); + const int zero_bits = + aligned == 0 ? available_bits + : gorilla_count_leading_zeros_nonzero(aligned); + const int remaining = max_count - consumed; + const int take = std::min(zero_bits, remaining); + bits -= take; + consumed += take; + + // A one bit follows the consumed zeros. Leave it in the reservoir + // for read_control_bits(), or stop once the requested limit is met. + if (zero_bits < available_bits || consumed == max_count) { + break; + } + } + return consumed; + } + FORCE_INLINE uint64_t read_long(int n) { if (UNLIKELY(n < 0 || n > 64)) { invalid = true; @@ -460,8 +513,21 @@ class GorillaDecoder : public Decoder { // Main batch loop while (actual < capacity && has_next_) { - out[actual++] = + const Output decoded = GorillaDecodeOutput::convert(stored_value_); + const int repeated = r.consume_zero_bits(capacity - actual - 1); + const int run_length = repeated + 1; + // Include the current value in the bulk fill. Besides removing a + // scalar store and a second position update for every run, this + // gives the compiler one contiguous range to vectorize. Integer + // and floating-point outputs are copied without arithmetic, so + // special IEEE-754 bit patterns remain unchanged. + std::fill_n(out + actual, run_length, decoded); + actual += run_length; + + // Prime the next value even when the repeated run exactly fills the + // caller's buffer. This preserves the scalar decoder invariant that + // stored_value_ is the next value to return on the following call. if (UNLIKELY(!GorillaRawOps::read_next( r, stored_value_, stored_leading_zeros_, stored_trailing_zeros_))) { diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h index 224d7402e..206b7f559 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -648,6 +648,43 @@ inline int TS2DIFFDecoder::read_batch_int64(int64_t* out, int capacity, int64_t prev = first_value_; int32_t i = 0; + // An evenly spaced timestamp block has no packed residual data. Build + // the arithmetic progression directly instead of entering the generic + // bit-extraction path (whose SIMD guard requires readable input bytes). + if (bit_width_ == 0) { +#ifdef ENABLE_SIMD + if (remaining >= 4) { + int64_t value1 = prev + delta_min_; + int64_t value2 = value1 + delta_min_; + int64_t value3 = value2 + delta_min_; + int64_t value4 = value3 + delta_min_; + simde__m256i values = + simde_mm256_set_epi64x(value4, value3, value2, value1); + + simde__m256i step = simde_mm256_set1_epi64x(delta_min_); + step = simde_mm256_add_epi64(step, step); + step = simde_mm256_add_epi64(step, step); + + for (; i + 3 < remaining; i += 4) { + simde_mm256_storeu_si256( + reinterpret_cast(out + actual), values); + actual += 4; + values = simde_mm256_add_epi64(values, step); + } + prev = out[actual - 1]; + } +#endif + + for (; i < remaining; ++i) { + prev += delta_min_; + out[actual++] = prev; + } + + first_value_ = prev; + current_index_ = 0; + continue; + } + #ifdef ENABLE_SIMD // SIMD path: decode 4 INT64 values at a time for (; i + 3 < remaining; i += 4) { diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index 97c469288..a140c1e98 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -922,6 +922,30 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in, } } + // Dense fixed-width batches are already laid out exactly as the two + // destination vectors expect. Appending them row by row would issue + // two tiny memcpy calls per row (time + value), plus virtual dispatch + // and bookkeeping. Copy each column once instead. Integral value + // filters still need the scalar satisfy(time, value) check unless the + // decoder proved the whole block passes, so those batches retain the + // fallback below. + const bool needs_integral_value_filter = + std::is_integral::value && filter != nullptr && !block_all_pass; + if (pass_count == time_count && nonnull_count == time_count && + !needs_integral_value_filter && + row_appender.can_bulk_append_fixed(0, sizeof(int64_t)) && + row_appender.can_bulk_append_fixed(1, sizeof(T))) { + row_appender.bulk_append_fixed(0, + reinterpret_cast(times), + static_cast(time_count)); + row_appender.bulk_append_fixed( + 1, reinterpret_cast(values), + static_cast(time_count)); + row_appender.add_rows(static_cast(time_count)); + cur_value_index += time_count; + continue; + } + int val_idx = 0; for (int i = 0; i < time_count; ++i) { cur_value_index++; diff --git a/cpp/test/common/tsblock/tslock_test.cc b/cpp/test/common/tsblock/tslock_test.cc index 750585aaf..c58b368f5 100644 --- a/cpp/test/common/tsblock/tslock_test.cc +++ b/cpp/test/common/tsblock/tslock_test.cc @@ -63,6 +63,43 @@ TEST(TsBlockTest, ColAppender_AddRowAndAppend) { EXPECT_EQ(col_appender.get_col_row_count(), 50); } +TEST(TsBlockTest, RowAppenderBulkAppendFixedAtExactCapacity) { + TupleDesc tuple_desc; + tuple_desc.push_back(ColumnSchema("time", TIMESTAMP, UNCOMPRESSED, PLAIN)); + tuple_desc.push_back(ColumnSchema("value", FLOAT, UNCOMPRESSED, PLAIN)); + TsBlock ts_block(&tuple_desc, 4); + ASSERT_EQ(ts_block.init(), E_OK); + RowAppender row_appender(&ts_block); + + const int64_t times[] = {101, 103, 107, 109}; + const float values[] = {1.25f, 2.5f, 3.75f, 5.0f}; + ASSERT_TRUE(row_appender.can_bulk_append_fixed(0, sizeof(int64_t))); + ASSERT_TRUE(row_appender.can_bulk_append_fixed(1, sizeof(float))); + EXPECT_FALSE(row_appender.can_bulk_append_fixed(1, sizeof(double))); + + row_appender.bulk_append_fixed(0, reinterpret_cast(times), 4); + row_appender.bulk_append_fixed(1, reinterpret_cast(values), 4); + row_appender.add_rows(4); + + EXPECT_EQ(ts_block.get_row_count(), 4u); + EXPECT_EQ(row_appender.remaining(), 0u); + EXPECT_EQ(ts_block.get_vector(0)->get_row_num(), 4u); + EXPECT_EQ(ts_block.get_vector(1)->get_row_num(), 4u); + + ColIterator time_iter(0, &ts_block); + ColIterator value_iter(1, &ts_block); + for (uint32_t i = 0; i < 4; ++i) { + uint32_t len = 0; + EXPECT_EQ(*reinterpret_cast(time_iter.read(&len)), times[i]); + EXPECT_EQ(len, sizeof(int64_t)); + EXPECT_FLOAT_EQ(*reinterpret_cast(value_iter.read(&len)), + values[i]); + EXPECT_EQ(len, sizeof(float)); + time_iter.next(); + value_iter.next(); + } +} + TEST(TsBlockTest, RowIterator_ReadAndNext) { TupleDesc tuple_desc; ColumnSchema col1("test_col1", INT32, SNAPPY, RLE); diff --git a/cpp/test/encoding/encoding_coverage_test.cc b/cpp/test/encoding/encoding_coverage_test.cc index 6970b9387..5d6ef813e 100644 --- a/cpp/test/encoding/encoding_coverage_test.cc +++ b/cpp/test/encoding/encoding_coverage_test.cc @@ -330,6 +330,41 @@ TEST(EncodingCoverage, TS2DIFFBatchInt64MultipleBlocks) { for (int i = 0; i < N; i++) EXPECT_EQ(out[i], values[i]) << "i=" << i; } +TEST(EncodingCoverage, TS2DIFFBatchInt64EvenlySpacedNegativeDelta) { + TS2DIFFEncoder enc; + common::ByteStream s(8192, common::MOD_DEFAULT); + // Full encoder blocks contain 127 residuals, so this covers repeated SIMD + // groups, the 3-value scalar tail, and a final partial block. + const int N = 389; + std::vector values(N); + for (int i = 0; i < N; i++) { + values[i] = INT64_C(9000000000000) - static_cast(i) * 29; + ASSERT_EQ(enc.encode(values[i], s), common::E_OK); + } + ASSERT_EQ(enc.flush(s), common::E_OK); + + uint32_t total = s.total_size(); + std::vector buf(total); + uint32_t got = 0; + s.read_buf(buf.data(), total, got); + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from((const char*)buf.data(), total); + + TS2DIFFDecoder dec; + std::vector out(N); + int total_decoded = 0; + while (dec.has_remaining(wrapped) && total_decoded < N) { + int actual = 0; + ASSERT_EQ(dec.read_batch_int64(out.data() + total_decoded, + N - total_decoded, actual, wrapped), + common::E_OK); + if (actual == 0) break; + total_decoded += actual; + } + EXPECT_EQ(total_decoded, N); + for (int i = 0; i < N; i++) EXPECT_EQ(out[i], values[i]) << "i=" << i; +} + // ── Plain encoder: encode_batch fast paths for each type ─────────────── TEST(EncodingCoverage, PlainEncoderBatchAllTypes) { PlainEncoder enc; diff --git a/cpp/test/encoding/gorilla_codec_test.cc b/cpp/test/encoding/gorilla_codec_test.cc index 039a9a4f0..ccf0ae1da 100644 --- a/cpp/test/encoding/gorilla_codec_test.cc +++ b/cpp/test/encoding/gorilla_codec_test.cc @@ -324,6 +324,108 @@ TEST_F(GorillaCodecTest, FloatBatchDecode) { } } +TEST_F(GorillaCodecTest, FloatBatchDecodeLongRepeatedRuns) { + storage::FloatGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + const int N = 600; + std::vector expected(N); + for (int i = 0; i < N; i++) { + if (i < 258) { + // Two complete 129-row batches. The repeated run crosses several + // 64-bit reservoirs and ends exactly at a batch boundary. + expected[i] = 17.25f; + } else if (i < 517) { + expected[i] = -3.5f; + } else { + expected[i] = static_cast(i - 517) * 0.125f; + } + ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK); + } + ASSERT_EQ(encoder.flush(stream), common::E_OK); + + const uint32_t total = stream.total_size(); + std::vector encoded(total); + uint32_t got = 0; + stream.read_buf(encoded.data(), total, got); + ASSERT_EQ(got, total); + + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from(reinterpret_cast(encoded.data()), total); + storage::FloatGorillaDecoder decoder; + std::vector decoded(N); + int total_decoded = 0; + while (total_decoded < N) { + const int capacity = std::min(129, N - total_decoded); + int actual = 0; + ASSERT_EQ(decoder.read_batch_float(decoded.data() + total_decoded, + capacity, actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, capacity); + total_decoded += actual; + } + + for (int i = 0; i < N; i++) { + EXPECT_EQ(common::float_to_int(decoded[i]), + common::float_to_int(expected[i])) + << "i=" << i; + } +} + +TEST_F(GorillaCodecTest, FloatBatchRepeatedFillPreservesBitPatterns) { + constexpr uint32_t NAN_BITS = 0x7FC12345U; + float payload_nan = 0; + std::memcpy(&payload_nan, &NAN_BITS, sizeof(payload_nan)); + ASSERT_TRUE(std::isnan(payload_nan)); + ASSERT_NE(common::float_to_int(payload_nan), + common::float_to_int(GORILLA_ENCODING_ENDING_FLOAT)); + + const int N = 520; + std::vector expected(N); + for (int i = 0; i < N; i++) { + if (i < 257) { + expected[i] = payload_nan; + } else if (i < 514) { + expected[i] = -0.0f; + } else { + expected[i] = static_cast(i - 514) + 0.25f; + } + } + + storage::FloatGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + for (float value : expected) { + ASSERT_EQ(encoder.encode(value, stream), common::E_OK); + } + ASSERT_EQ(encoder.flush(stream), common::E_OK); + + const uint32_t total = stream.total_size(); + std::vector encoded(total); + uint32_t got = 0; + stream.read_buf(encoded.data(), total, got); + ASSERT_EQ(got, total); + + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from(reinterpret_cast(encoded.data()), total); + storage::FloatGorillaDecoder decoder; + std::vector decoded(N); + int total_decoded = 0; + while (total_decoded < N) { + const int capacity = std::min(129, N - total_decoded); + int actual = 0; + ASSERT_EQ(decoder.read_batch_float(decoded.data() + total_decoded, + capacity, actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, capacity); + total_decoded += actual; + } + + for (int i = 0; i < N; i++) { + EXPECT_EQ(common::float_to_int(decoded[i]), + common::float_to_int(expected[i])) + << "i=" << i; + } +} + TEST_F(GorillaCodecTest, FloatBatchDecodeUnwrappedInput) { storage::FloatGorillaEncoder encoder; common::ByteStream stream(1024, common::MOD_DEFAULT); @@ -418,6 +520,47 @@ TEST_F(GorillaCodecTest, DoubleBatchDecodeOneValueAtATime) { } } +TEST_F(GorillaCodecTest, Int64BatchDecodeLongRepeatedRuns) { + storage::LongGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + const int N = 513; + std::vector expected(N); + for (int i = 0; i < N; i++) { + if (i < 333) { + expected[i] = 0x123456789ABCDEFLL; + } else if (i < 500) { + expected[i] = -9876543210LL; + } else { + expected[i] = static_cast(i) * 17 - 9; + } + ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK); + } + ASSERT_EQ(encoder.flush(stream), common::E_OK); + + const uint32_t total = stream.total_size(); + std::vector encoded(total); + uint32_t got = 0; + stream.read_buf(encoded.data(), total, got); + ASSERT_EQ(got, total); + + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from(reinterpret_cast(encoded.data()), total); + storage::LongGorillaDecoder decoder; + std::vector decoded(N); + int total_decoded = 0; + while (total_decoded < N) { + const int capacity = std::min(73, N - total_decoded); + int actual = 0; + ASSERT_EQ(decoder.read_batch_int64(decoded.data() + total_decoded, + capacity, actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, capacity); + total_decoded += actual; + } + + EXPECT_EQ(decoded, expected); +} + TEST_F(GorillaCodecTest, DoubleBatchScalarAndSkipInterleave) { storage::DoubleGorillaEncoder encoder; common::ByteStream stream(1024, common::MOD_DEFAULT); diff --git a/cpp/test/reader/tsfile_reader_test.cc b/cpp/test/reader/tsfile_reader_test.cc index cdb6d33cf..df4c642a8 100644 --- a/cpp/test/reader/tsfile_reader_test.cc +++ b/cpp/test/reader/tsfile_reader_test.cc @@ -23,6 +23,7 @@ #include #include +#include #include #include #include @@ -1511,6 +1512,97 @@ TEST_F(TsFileReaderTest, AlignedSchemaReportsValueDataType) { reader.close(); } +TEST_F(TsFileReaderTest, + AlignedFloatBatchCopyPreservesDenseNullAndFilteredAlignment) { + const std::string device = "root.dev_aligned_float_batch"; + MeasurementSchema schema("v0", FLOAT, GORILLA, UNCOMPRESSED); + ASSERT_EQ(tsfile_writer_->register_aligned_timeseries(device, schema), + E_OK); + + const int row_count = 300; + const int null_row = 150; + auto schemas = std::make_shared>(1, schema); + Tablet tablet(device, schemas, row_count); + for (int row = 0; row < row_count; ++row) { + ASSERT_EQ(tablet.add_timestamp(row, 10000 + row * 3), E_OK); + if (row != null_row) { + ASSERT_EQ(tablet.add_value(row, 0u, row + 0.25f), E_OK); + } + } + ASSERT_EQ(tsfile_writer_->write_tablet_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 device_id = std::make_shared(device); + + auto scan_and_check = [&](storage::Filter* filter, + const std::vector& expected_rows) { + storage::TsFileSeriesScanIterator* ssi = nullptr; + common::PageArena pa; + pa.init(512, common::MOD_TSFILE_READER); + ASSERT_EQ(io_reader.alloc_ssi(device_id, "v0", ssi, pa, filter), E_OK); + ASSERT_NE(ssi, nullptr); + + size_t expected_index = 0; + while (true) { + common::TsBlock* block = nullptr; + int ret = ssi->get_next(block, /*alloc_tsblock=*/true, filter); + if (ret == E_NO_MORE_DATA) break; + ASSERT_EQ(ret, E_OK); + ASSERT_NE(block, nullptr); + + { + common::ColIterator time_iter(0, block); + common::ColIterator value_iter(1, block); + while (!time_iter.end()) { + ASSERT_LT(expected_index, expected_rows.size()); + const int expected_row = expected_rows[expected_index++]; + uint32_t len = 0; + bool is_null = false; + EXPECT_EQ(*reinterpret_cast(time_iter.read(&len)), + 10000 + expected_row * 3); + EXPECT_EQ(len, sizeof(int64_t)); + char* value = value_iter.read(&len, &is_null); + if (expected_row == null_row) { + EXPECT_TRUE(is_null); + EXPECT_EQ(value, nullptr); + } else { + ASSERT_FALSE(is_null); + ASSERT_NE(value, nullptr); + EXPECT_EQ(len, sizeof(float)); + EXPECT_FLOAT_EQ(*reinterpret_cast(value), + expected_row + 0.25f); + } + time_iter.next(); + value_iter.next(); + } + } + ssi->revert_tsblock(); + } + EXPECT_EQ(expected_index, expected_rows.size()); + io_reader.revert_ssi(ssi); + }; + + std::vector all_rows(row_count); + std::iota(all_rows.begin(), all_rows.end(), 0); + scan_and_check(/*filter=*/nullptr, all_rows); + + std::vector selected_times; + std::vector selected_rows; + for (int row = 0; row < row_count; row += 11) { + selected_rows.push_back(row); + selected_times.push_back(10000 + row * 3); + } + selected_rows.push_back(null_row); + selected_times.push_back(10000 + null_row * 3); + std::sort(selected_rows.begin(), selected_rows.end()); + std::sort(selected_times.begin(), selected_times.end()); + storage::TimeIn time_filter(selected_times, /*not_in=*/false); + scan_and_check(&time_filter, selected_rows); +} + namespace storage { class TsFileReaderMetaArenaTest { public: