From 7bafa5092d4456c5335edbe335575815cb1badee Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 5 Nov 2026 16:37:49 +0800 Subject: [PATCH 1/7] perf(cpp): speed up wide-table reads RLE decoding: - Add read_batch_int32/int64 to Int32/Int64RleDecoder: RLE runs fill the output directly, bit-packed groups memcpy out of the already-unpacked buffer; Float/DoubleRLEDecoder get read_batch_float/double. - Replace the per-byte read_ui8 copy in read_bit_packing_buffer with one read_buf call. - Decode 59/61/62/63-bit groups through a 128-bit byte buffer instead of bit-by-bit (NaN raw-bit values widen a column to 63 bits). SIMD: - New common/simd_dispatch.h plus __attribute__((target("avx2"))) twins of the int64/int32 bit-unpack kernels, selected at runtime with __builtin_cpu_supports. Baseline builds never add -mavx2, so simde was emitting its portable fallback and the "SIMD" path measured ~6x slower than real AVX2. Index loading: - get_timeseries_indexes now collects the device's leaf entries, reads the whole contiguous timeseries-index range with one pread and deserializes each index straight into the caller's arena, instead of a binary search, pread, temp-arena decode and clone per measurement. On a 20-feature x 5214-stock x 240-min/month wide table, a 5-feature x 4500-stock x 1-month query drops from 5.98s to ~1.35s. --- cpp/src/common/simd_dispatch.h | 58 ++++++++++++ cpp/src/encoding/float_rle_decoder.h | 49 ++++++++++ cpp/src/encoding/int32_packer.h | 94 +++++++++++++++++- cpp/src/encoding/int32_rle_decoder.h | 61 ++++++++++-- cpp/src/encoding/int64_packer.h | 137 ++++++++++++++++++++++----- cpp/src/encoding/int64_rle_decoder.h | 67 +++++++++++-- cpp/src/file/tsfile_io_reader.cc | 99 ++++++++++++------- 7 files changed, 495 insertions(+), 70 deletions(-) create mode 100644 cpp/src/common/simd_dispatch.h diff --git a/cpp/src/common/simd_dispatch.h b/cpp/src/common/simd_dispatch.h new file mode 100644 index 000000000..866349858 --- /dev/null +++ b/cpp/src/common/simd_dispatch.h @@ -0,0 +1,58 @@ +/* + * 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 COMMON_SIMD_DISPATCH_H +#define COMMON_SIMD_DISPATCH_H + +#include "common/global.h" + +// Runtime SIMD dispatch. +// +// The library is built for the portable x86-64 baseline (SSE2 only), so a +// kernel written with simde's AVX2/SSSE3 API is compiled into simde's portable +// fallback — measured ~6x slower than real AVX2 for the RLE bit-unpack. To run +// the fast instructions without raising the binary's ISA floor (an -mavx2 +// binary faults on older CPUs), such kernels get a second copy annotated with +// __attribute__((target("avx2"))) that uses raw intrinsics, and call sites pick +// it at runtime via cpu_has_avx2(). +// +// Only GCC/Clang on x86 provide both pieces; elsewhere callers keep the +// portable path. + +#if defined(__x86_64__) || defined(__i386__) +#if defined(__GNUC__) || defined(__clang__) +#define TSFILE_X86_AVX2_DISPATCH 1 +#define TSFILE_TARGET_AVX2 __attribute__((target("avx2"))) +#endif +#endif + +namespace common { + +#ifdef TSFILE_X86_AVX2_DISPATCH +// True when the running CPU implements AVX2. Cached after the first call; +// glibc/libgcc initialize the CPU feature bits before any regular function +// runs, so no explicit __builtin_cpu_init() is required here. +inline bool cpu_has_avx2() { + static const bool has_avx2 = __builtin_cpu_supports("avx2") != 0; + return has_avx2; +} +#endif + +} // namespace common + +#endif // COMMON_SIMD_DISPATCH_H diff --git a/cpp/src/encoding/float_rle_decoder.h b/cpp/src/encoding/float_rle_decoder.h index 6cdea9825..8084fe69c 100644 --- a/cpp/src/encoding/float_rle_decoder.h +++ b/cpp/src/encoding/float_rle_decoder.h @@ -20,6 +20,7 @@ #ifndef ENCODING_FLOAT_RLE_DECODER_H #define ENCODING_FLOAT_RLE_DECODER_H +#include #include #include #include @@ -75,6 +76,30 @@ class FloatRLEDecoder : public Decoder { page_pos_++; return ret; } + int read_batch_float(float* out, int capacity, int& actual, + common::ByteStream& in) override { + actual = 0; + int ret = ensure_page_meta(in); + if (RET_FAIL(ret)) { + return ret; + } + int32_t tmp[64]; + while (actual < capacity && int_decoder_.has_remaining(in)) { + const int chunk = std::min(capacity - actual, 64); + int got = 0; + if (RET_FAIL(int_decoder_.read_batch_int32(tmp, chunk, got, in))) { + return ret; + } + if (got == 0) { + break; + } + for (int i = 0; i < got; i++) { + out[actual++] = value_at(tmp[i]); + page_pos_++; + } + } + return ret; + } int read_double(double& /*ret_value*/, common::ByteStream& /*in*/) override { return common::E_TYPE_NOT_MATCH; @@ -170,6 +195,30 @@ class DoubleRLEDecoder : public Decoder { page_pos_++; return ret; } + int read_batch_double(double* out, int capacity, int& actual, + common::ByteStream& in) override { + actual = 0; + int ret = ensure_page_meta(in); + if (RET_FAIL(ret)) { + return ret; + } + int64_t tmp[64]; + while (actual < capacity && int_decoder_.has_remaining(in)) { + const int chunk = std::min(capacity - actual, 64); + int got = 0; + if (RET_FAIL(int_decoder_.read_batch_int64(tmp, chunk, got, in))) { + return ret; + } + if (got == 0) { + break; + } + for (int i = 0; i < got; i++) { + out[actual++] = value_at(tmp[i]); + page_pos_++; + } + } + return ret; + } int read_String(common::String& /*ret_value*/, common::PageArena& /*pa*/, common::ByteStream& /*in*/) override { return common::E_TYPE_NOT_MATCH; diff --git a/cpp/src/encoding/int32_packer.h b/cpp/src/encoding/int32_packer.h index ecb42b248..41db56489 100644 --- a/cpp/src/encoding/int32_packer.h +++ b/cpp/src/encoding/int32_packer.h @@ -24,14 +24,92 @@ #include "encoder.h" +#ifdef ENABLE_SIMD +#include "common/simd_dispatch.h" +#include "simde/x86/avx2.h" +#ifdef TSFILE_X86_AVX2_DISPATCH +#include +#endif +#endif + namespace storage { +#ifdef ENABLE_SIMD +// SIMD decode of 4 big-endian bit-packed INT32 values. Uses the 64-bit gather +// so all widths 1..32 fit without a width guard; the trailing permute packs +// the low 32 bits of each 64-bit lane into four contiguous INT32 lanes. +static inline void rle_simd_unpack_4_i32(const uint8_t* in, int32_t width, + int32_t index, int32_t out[4]) { + static const simde__m256i SHUF_REV8 = simde_mm256_setr_epi8( + 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, + 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); + const int32_t pos0 = index * width; + const int32_t pos[4] = {pos0, pos0 + width, pos0 + 2 * width, + pos0 + 3 * width}; + const int32_t bidx[4] = {pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, + pos[3] >> 3}; + const int32_t off[4] = {pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7}; + const simde__m128i IDX = + simde_mm_setr_epi32(bidx[0], bidx[1], bidx[2], bidx[3]); + const int32_t rshift = 64 - width; + simde__m256i w64_le = + simde_mm256_i32gather_epi64((const int64_t*)in, IDX, 1); + simde__m256i w64_be = simde_mm256_shuffle_epi8(w64_le, SHUF_REV8); + simde__m256i OFF64 = simde_mm256_cvtepu32_epi64( + simde_mm_setr_epi32(off[0], off[1], off[2], off[3])); + simde__m256i U64 = simde_mm256_sllv_epi64(w64_be, OFF64); + simde__m256i V64 = + simde_mm256_srl_epi64(U64, simde_mm_cvtsi32_si128(rshift)); + const simde__m256i perm = simde_mm256_setr_epi32(0, 2, 4, 6, 0, 0, 0, 0); + simde__m256i comp = simde_mm256_permutevar8x32_epi32(V64, perm); + simde_mm_storeu_si128((simde__m128i*)out, + simde_mm256_castsi256_si128(comp)); +} + +#ifdef TSFILE_X86_AVX2_DISPATCH +// Native-AVX2 twin of rle_simd_unpack_4_i32; see int64_packer.h for why the +// target attribute is required (simde otherwise emits its portable fallback). +TSFILE_TARGET_AVX2 +static inline void rle_avx2_unpack_4_i32(const uint8_t* in, int32_t width, + int32_t index, int32_t out[4]) { + const __m256i SHUF_REV8 = + _mm256_setr_epi8(7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, + 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); + const int32_t pos0 = index * width; + const int32_t pos[4] = {pos0, pos0 + width, pos0 + 2 * width, + pos0 + 3 * width}; + const __m128i IDX = + _mm_setr_epi32(pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, pos[3] >> 3); + const int32_t rshift = 64 - width; + __m256i w64_le = _mm256_i32gather_epi64((const long long*)in, IDX, 1); + __m256i w64_be = _mm256_shuffle_epi8(w64_le, SHUF_REV8); + __m256i OFF64 = _mm256_cvtepu32_epi64( + _mm_setr_epi32(pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7)); + __m256i U64 = _mm256_sllv_epi64(w64_be, OFF64); + __m256i V64 = _mm256_srl_epi64(U64, _mm_cvtsi32_si128(rshift)); + const __m256i perm = _mm256_setr_epi32(0, 2, 4, 6, 0, 0, 0, 0); + __m256i comp = _mm256_permutevar8x32_epi32(V64, perm); + _mm_storeu_si128((__m128i*)out, _mm256_castsi256_si128(comp)); +} +#endif // TSFILE_X86_AVX2_DISPATCH +#endif + class Int32Packer { private: int width_; +#ifdef TSFILE_X86_AVX2_DISPATCH + // Resolved once per packer (one per bit-width pattern) so the hot + // unpack loop pays no thread-safe-static guard on every group. + bool use_avx2_; +#endif public: - Int32Packer(int width_) { this->width_ = width_; } +#ifdef TSFILE_X86_AVX2_DISPATCH + Int32Packer(int width_) + : width_(width_), use_avx2_(common::cpu_has_avx2()) {} +#else + Int32Packer(int width_) : width_(width_) {} +#endif ~Int32Packer() { destroy(); } void destroy() { /* do nothing for IntPacker */ @@ -101,6 +179,20 @@ class Int32Packer { */ void unpack_8values(const unsigned char buf[], int offset, int32_t values[]) { +#ifdef ENABLE_SIMD + // Every width 1..32 fits within the 8-byte gather, so no width guard + // is needed here (unlike the 64-bit packer). +#ifdef TSFILE_X86_AVX2_DISPATCH + if (use_avx2_) { + rle_avx2_unpack_4_i32(buf + offset, width_, 0, values); + rle_avx2_unpack_4_i32(buf + offset, width_, 4, values + 4); + return; + } +#endif + rle_simd_unpack_4_i32(buf + offset, width_, 0, values); + rle_simd_unpack_4_i32(buf + offset, width_, 4, values + 4); + return; +#endif int byte_idx = offset; uint64_t buffer = 0; // total bits which have reader from 'buf' to 'buffer'. i.e., diff --git a/cpp/src/encoding/int32_rle_decoder.h b/cpp/src/encoding/int32_rle_decoder.h index ef7b6f095..f840b583b 100644 --- a/cpp/src/encoding/int32_rle_decoder.h +++ b/cpp/src/encoding/int32_rle_decoder.h @@ -20,6 +20,8 @@ #ifndef ENCODING_INT32RLE_DECODER_H #define ENCODING_INT32RLE_DECODER_H +#include +#include #include #include "common/allocator/alloc_base.h" @@ -73,6 +75,49 @@ class Int32RleDecoder : public Decoder { int read_int32(int32_t& ret_value, common::ByteStream& in) override { return read_int(ret_value, in); } + int read_batch_int32(int32_t* out, int capacity, int& actual, + common::ByteStream& in) override { + actual = 0; + int ret = common::E_OK; + while (actual < capacity && has_remaining(in)) { + if (!is_length_and_bitwidth_readed_) { + if (RET_FAIL(read_length_and_bitwidth(in))) { + return ret; + } + } + if (current_count_ == 0) { + uint32_t header_value = 0; + if (RET_FAIL(common::SerializationUtil::read_var_uint( + header_value, byte_cache_))) { + return ret; + } + if (header_value & 1) { + if (RET_FAIL(call_read_bit_packing_buffer(header_value))) { + return ret; + } + } else { + if (RET_FAIL(call_read_rle_run(header_value))) { + return ret; + } + } + } + const int n = std::min(current_count_, capacity - actual); + if (is_rle_run_) { + std::fill(out + actual, out + actual + n, rle_value_); + } else { + const int32_t* src = + current_buffer_ + (bitpacking_num_ - current_count_); + std::memcpy(out + actual, src, + static_cast(n) * sizeof(int32_t)); + } + actual += n; + current_count_ -= n; + if (!has_next_package()) { + is_length_and_bitwidth_readed_ = false; + } + } + return ret; + } int read_int64(int64_t& ret_value, common::ByteStream& in) override { return common::E_TYPE_NOT_MATCH; } @@ -210,13 +255,17 @@ class Int32RleDecoder : public Decoder { if (bytes_to_read > (int)byte_cache_.remaining_size()) { bytes_to_read = byte_cache_.remaining_size(); } - std::vector bytes(bytes_to_read); + // +8 zero padding: the SIMD unpack gathers 8 bytes per value and may + // read up to 7 bytes past the last group's end for small bit widths. + std::vector bytes(bytes_to_read + 8, 0); - for (int i = 0; i < bytes_to_read; i++) { - if (RET_FAIL(common::SerializationUtil::read_ui8(bytes[i], - byte_cache_))) { - return ret; - } + uint32_t got = 0; + if (RET_FAIL(byte_cache_.read_buf( + bytes.data(), static_cast(bytes_to_read), got))) { + return ret; + } + if (got != static_cast(bytes_to_read)) { + return common::E_PARTIAL_READ; } // save all int values in currentBuffer diff --git a/cpp/src/encoding/int64_packer.h b/cpp/src/encoding/int64_packer.h index 23267206f..f260f0d81 100644 --- a/cpp/src/encoding/int64_packer.h +++ b/cpp/src/encoding/int64_packer.h @@ -24,14 +24,92 @@ #include "encoder.h" +#ifdef ENABLE_SIMD +#include "common/simd_dispatch.h" +#include "simde/x86/avx2.h" +#ifdef TSFILE_X86_AVX2_DISPATCH +#include +#endif +#endif + namespace storage { +#ifdef ENABLE_SIMD +// SIMD decode of 4 big-endian bit-packed INT64 values. Mirrors the TS_2DIFF +// helper but skips delta reconstruction: RLE packs raw values, so the gathered +// and shifted bits are already the final value (negative values are always +// stored with the full 64-bit width, so the logical shift is lossless). +// `in` points at the start of the packed group; `index` is the first value's +// position within the group (0 or 4 for an 8-value group). +static inline void rle_simd_unpack_4_i64(const uint8_t* in, int32_t width, + int32_t index, int64_t out[4]) { + static const simde__m256i SHUF_REV8 = simde_mm256_setr_epi8( + 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, + 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); + const int32_t pos0 = index * width; + const int32_t pos[4] = {pos0, pos0 + width, pos0 + 2 * width, + pos0 + 3 * width}; + const int32_t bidx[4] = {pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, + pos[3] >> 3}; + const int32_t off[4] = {pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7}; + const simde__m128i IDX = + simde_mm_setr_epi32(bidx[0], bidx[1], bidx[2], bidx[3]); + const int32_t rshift = 64 - width; + simde__m256i w64_le = + simde_mm256_i32gather_epi64((const int64_t*)in, IDX, 1); + simde__m256i w64_be = simde_mm256_shuffle_epi8(w64_le, SHUF_REV8); + simde__m256i OFF64 = simde_mm256_cvtepu32_epi64( + simde_mm_setr_epi32(off[0], off[1], off[2], off[3])); + simde__m256i U64 = simde_mm256_sllv_epi64(w64_be, OFF64); + simde__m256i V64 = + simde_mm256_srl_epi64(U64, simde_mm_cvtsi32_si128(rshift)); + simde_mm256_storeu_si256((simde__m256i*)out, V64); +} + +#ifdef TSFILE_X86_AVX2_DISPATCH +// Same kernel with real AVX2 intrinsics. simde picks its native path only when +// the whole translation unit is compiled with -mavx2, which the library is not +// (portable baseline), so a target-attributed copy is needed to actually emit +// vpgatherqq/vpshufb. Selected once per packer via common::cpu_has_avx2(). +TSFILE_TARGET_AVX2 +static inline void rle_avx2_unpack_4_i64(const uint8_t* in, int32_t width, + int32_t index, int64_t out[4]) { + const __m256i SHUF_REV8 = + _mm256_setr_epi8(7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, + 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); + const int32_t pos0 = index * width; + const int32_t pos[4] = {pos0, pos0 + width, pos0 + 2 * width, + pos0 + 3 * width}; + const __m128i IDX = + _mm_setr_epi32(pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, pos[3] >> 3); + const int32_t rshift = 64 - width; + __m256i w64_le = _mm256_i32gather_epi64((const long long*)in, IDX, 1); + __m256i w64_be = _mm256_shuffle_epi8(w64_le, SHUF_REV8); + __m256i OFF64 = _mm256_cvtepu32_epi64( + _mm_setr_epi32(pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7)); + __m256i U64 = _mm256_sllv_epi64(w64_be, OFF64); + __m256i V64 = _mm256_srl_epi64(U64, _mm_cvtsi32_si128(rshift)); + _mm256_storeu_si256((__m256i*)out, V64); +} +#endif // TSFILE_X86_AVX2_DISPATCH +#endif + class Int64Packer { private: int width_; +#ifdef TSFILE_X86_AVX2_DISPATCH + // Resolved once per packer (one per bit-width pattern) so the hot + // unpack loop pays no thread-safe-static guard on every group. + bool use_avx2_; +#endif public: - Int64Packer(int width_) { this->width_ = width_; } +#ifdef TSFILE_X86_AVX2_DISPATCH + Int64Packer(int width_) + : width_(width_), use_avx2_(common::cpu_has_avx2()) {} +#else + Int64Packer(int width_) : width_(width_) {} +#endif ~Int64Packer() { destroy(); } void destroy() { /* do nothing for IntPacker */ @@ -102,35 +180,44 @@ class Int64Packer { */ void unpack_8values(const unsigned char buf[], int offset, int64_t values[]) { +#ifdef ENABLE_SIMD + // Widths 59/61/62/63 can start at a bit offset that needs a ninth + // byte for the gather; decode those with the scalar path below. + if (width_ <= 58 || width_ == 60 || width_ == 64) { +#ifdef TSFILE_X86_AVX2_DISPATCH + if (use_avx2_) { + rle_avx2_unpack_4_i64(buf + offset, width_, 0, values); + rle_avx2_unpack_4_i64(buf + offset, width_, 4, values + 4); + return; + } +#endif + rle_simd_unpack_4_i64(buf + offset, width_, 0, values); + rle_simd_unpack_4_i64(buf + offset, width_, 4, values + 4); + return; + } +#endif + // Scalar fallback for widths that can cross a ninth byte (59/61/62/63). + // Decode whole bytes into a 128-bit buffer — ~8x fewer iterations than + // the old bit-by-bit loop — then shift out one width-wide value at a + // time. Widths up to 63 need up to 63 + 7 = 70 buffered bits, which the + // 64-bit int32-style buffer cannot hold; hence 128 bits. int byte_idx = offset; - int value_idx = 0; - int left_bits = 8; + unsigned __int128 buffer = 0; int total_bits = 0; - + int value_idx = 0; while (value_idx < 8) { - values[value_idx] = 0; - total_bits = 0; - while (total_bits < width_) { - if (width_ - total_bits >= left_bits) { - values[value_idx] <<= left_bits; - values[value_idx] |= static_cast( - buf[byte_idx] & ((1 << left_bits) - 1)); - total_bits += left_bits; - byte_idx++; - left_bits = 8; - } else { - int t = width_ - total_bits; - values[value_idx] <<= t; - values[value_idx] |= static_cast( - (buf[byte_idx] & ((1 << left_bits) - 1)) >> - (left_bits - t)); - left_bits -= t; - total_bits += t; - } + buffer = (buffer << 8) | (buf[byte_idx] & 0xFF); + byte_idx++; + total_bits += 8; + } + while (total_bits >= width_ && value_idx < 8) { + values[value_idx] = + static_cast(buffer >> (total_bits - width_)); + value_idx++; + total_bits -= width_; + buffer &= (((unsigned __int128)1 << total_bits) - 1); } - - value_idx++; } } diff --git a/cpp/src/encoding/int64_rle_decoder.h b/cpp/src/encoding/int64_rle_decoder.h index bef72bbed..1fdd13990 100644 --- a/cpp/src/encoding/int64_rle_decoder.h +++ b/cpp/src/encoding/int64_rle_decoder.h @@ -20,6 +20,8 @@ #ifndef ENCODING_INT64RLE_DECODER_H #define ENCODING_INT64RLE_DECODER_H +#include +#include #include #include "common/allocator/alloc_base.h" @@ -71,6 +73,55 @@ class Int64RleDecoder : public Decoder { int read_int64(int64_t& ret_value, common::ByteStream& in) override { return read_int(ret_value, in); } + int read_batch_int64(int64_t* out, int capacity, int& actual, + common::ByteStream& in) override { + actual = 0; + int ret = common::E_OK; + while (actual < capacity && has_remaining(in)) { + if (!is_length_and_bitwidth_readed_) { + if (RET_FAIL(read_length_and_bitwidth(in))) { + return ret; + } + } + if (current_count_ == 0) { + // Low bit = 0 => RLE run, = 1 => bit-packing group (same + // header dispatch as read_int). + uint32_t header_value = 0; + if (RET_FAIL(common::SerializationUtil::read_var_uint( + header_value, byte_cache_))) { + return ret; + } + if (header_value & 1) { + if (RET_FAIL(call_read_bit_packing_buffer(header_value))) { + return ret; + } + } else { + if (RET_FAIL(call_read_rle_run(header_value))) { + return ret; + } + } + } + const int n = std::min(current_count_, capacity - actual); + if (is_rle_run_) { + // The whole run holds the same value; a broadcast fill is + // auto-vectorized by the compiler. + std::fill(out + actual, out + actual + n, rle_value_); + } else { + // Bit-packing groups are already unpacked into current_buffer_ + // by the packer, so the remaining values are a straight copy. + const int64_t* src = + current_buffer_ + (bitpacking_num_ - current_count_); + std::memcpy(out + actual, src, + static_cast(n) * sizeof(int64_t)); + } + actual += n; + current_count_ -= n; + if (!has_next_package()) { + is_length_and_bitwidth_readed_ = false; + } + } + return ret; + } int read_float(float& ret_value, common::ByteStream& in) override { return common::E_TYPE_NOT_MATCH; } @@ -209,13 +260,17 @@ class Int64RleDecoder : public Decoder { if (bytes_to_read > (int)byte_cache_.remaining_size()) { bytes_to_read = byte_cache_.remaining_size(); } - std::vector bytes(bytes_to_read); + // +8 zero padding: the SIMD unpack gathers 8 bytes per value and may + // read up to 7 bytes past the last group's end for small bit widths. + std::vector bytes(bytes_to_read + 8, 0); - for (int i = 0; i < bytes_to_read; i++) { - if (RET_FAIL(common::SerializationUtil::read_ui8(bytes[i], - byte_cache_))) { - return ret; - } + uint32_t got = 0; + if (RET_FAIL(byte_cache_.read_buf( + bytes.data(), static_cast(bytes_to_read), got))) { + return ret; + } + if (got != static_cast(bytes_to_read)) { + return common::E_PARTIAL_READ; } // save all int values in currentBuffer diff --git a/cpp/src/file/tsfile_io_reader.cc b/cpp/src/file/tsfile_io_reader.cc index 7c2b1897b..1d8f54ad0 100644 --- a/cpp/src/file/tsfile_io_reader.cc +++ b/cpp/src/file/tsfile_io_reader.cc @@ -904,8 +904,6 @@ int TsFileIOReader::get_timeseries_indexes( int ret = E_OK; std::shared_ptr device_index_entry; int64_t device_ie_end_offset = 0; - std::shared_ptr measurement_index_entry; - int64_t measurement_ie_end_offset = 0; if (RET_FAIL(load_device_index_entry( std::make_shared(device_id), device_index_entry, device_ie_end_offset))) { @@ -936,48 +934,85 @@ int TsFileIOReader::get_timeseries_indexes( } bool is_aligned = is_aligned_device(top_node); - TimeseriesIndex* timeseries_index = nullptr; + TimeseriesIndex* time_timeseries_index = nullptr; if (is_aligned) { - if (RET_FAIL( - get_time_column_metadata(top_node, timeseries_index, pa))) { + if (RET_FAIL(get_time_column_metadata(top_node, time_timeseries_index, + pa))) { return ret; } } - int64_t idx = 0; - for (const auto& measurement_name : measurement_names) { - timeseries_indexs[idx] = nullptr; - ret = load_measurement_index_entry(measurement_name, top_node, - measurement_index_entry, - measurement_ie_end_offset, pa); - if (ret == E_MEASUREMENT_NOT_EXIST || ret == E_NOT_EXIST) { - ret = E_OK; - idx++; - continue; - } - if (RET_FAIL(ret)) { - return ret; - } + // Collect every measurement leaf once and read the whole (contiguous) + // timeseries-index range in a single pread, then deserialize each index + // directly into `pa`. The previous path did a binary search + a separate + // pread + a deserialize-into-temp-arena + clone per measurement (4500x for + // a wide table), which dominated index loading. + std::vector, int64_t>> + leaf_entries; + if (RET_FAIL(get_all_leaf(top_node, leaf_entries, pa))) { + return ret; + } - ret = do_load_timeseries_index( - measurement_name, measurement_index_entry->get_offset(), - measurement_ie_end_offset, pa, timeseries_indexs[idx], is_aligned); - if (ret == E_NOT_EXIST) { - ret = E_OK; - idx++; - continue; + std::unordered_map name_to_index; + if (!leaf_entries.empty()) { + const int64_t ts_start_offset = + leaf_entries.front().first->get_offset(); + const int64_t ts_end_offset = leaf_entries.back().second; + const int32_t ts_read_size = (int32_t)(ts_end_offset - ts_start_offset); + int32_t ts_read_len = 0; + char* ts_buf = (char*)pa.alloc(ts_read_size); + if (IS_NULL(ts_buf)) { + return E_OOM; } - if (RET_FAIL(ret)) { + if (RET_FAIL(read_file_->read(ts_start_offset, ts_buf, ts_read_size, + ts_read_len))) { return ret; + } else if (ts_read_len != ts_read_size) { + return E_FILE_READ_ERR; } - if (is_aligned) { - AlignedTimeseriesIndex* aligned_timeseries_index = - dynamic_cast(timeseries_indexs[idx]); - if (aligned_timeseries_index) { - aligned_timeseries_index->time_ts_idx_ = timeseries_index; + ByteStream bs; + bs.wrap_from(ts_buf, ts_read_size); + while (bs.has_remaining()) { + const uint64_t relative_start = bs.read_pos(); + void* ts_idx_buf = pa.alloc(sizeof(TimeseriesIndex)); + if (IS_NULL(ts_idx_buf)) { + return E_OOM; + } + TimeseriesIndex* ts_idx = new (ts_idx_buf) TimeseriesIndex; + if (RET_FAIL(ts_idx->deserialize_from(bs, &pa))) { + return ret; + } + if (bs.read_pos() < relative_start || + bs.read_pos() - relative_start > UINT32_MAX) { + return E_OVERFLOW; + } + ts_idx->set_metadata_range( + ts_start_offset + relative_start, + static_cast(bs.read_pos() - relative_start)); + if (ts_idx->get_measurement_name().len_ == 0) continue; + const std::string name = + ts_idx->get_measurement_name().to_std_string(); + if (is_aligned) { + void* aligned_buf = pa.alloc(sizeof(AlignedTimeseriesIndex)); + if (IS_NULL(aligned_buf)) { + return E_OOM; + } + AlignedTimeseriesIndex* aligned_ts_idx = + new (aligned_buf) AlignedTimeseriesIndex; + aligned_ts_idx->value_ts_idx_ = ts_idx; + aligned_ts_idx->time_ts_idx_ = time_timeseries_index; + name_to_index[name] = aligned_ts_idx; + } else { + name_to_index[name] = ts_idx; } } + } + int64_t idx = 0; + for (const auto& measurement_name : measurement_names) { + const auto it = name_to_index.find(measurement_name); + timeseries_indexs[idx] = + (it != name_to_index.end()) ? it->second : nullptr; idx++; } return ret; From 05e19aef04b69d75ce5ae519f2c5defe02ab8e3e Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 5 Nov 2026 17:48:34 +0800 Subject: [PATCH 2/7] perf(cpp): bulk-copy contiguous BOUNDARY pages in aligned reader A time range that covers only part of a chunk marked every page BOUNDARY, so get_next_page_multi_impl fell into the row-by-row scatter loop: one RowAppender::append per value plus a per-timestamp filter re-check. That slow path measured 8.6% of a 1-week wide-table query, and made queries that read less data (1 week) as slow as a full-month query. A BOUNDARY page under a contiguous time filter (TimeBetween with not_between=false) has a fully matching [row_begin, row_end] range with no interior holes, so it can use the same bulk memcpy path as FULL_PASS: the page is decoded whole, so predecoded_values is row-indexed at row_begin. - Add Filter::is_contiguous_time_filter() (default false; TimeBetween returns !not_). - Let can_bulk accept a BOUNDARY page when that predicate holds and no value column carries nulls (predecoded_count == row_count, the condition that keeps predecoded_values row-aligned). - Discrete filters (TimeIn, NOT BETWEEN) keep the slow path. 1-week 5f x 4500-stock query: 1.63s -> ~1.05s. --- cpp/src/reader/aligned_chunk_reader.cc | 18 +++++++++++++----- cpp/src/reader/filter/filter.h | 8 ++++++++ cpp/src/reader/filter/time_operator.h | 2 ++ 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index a0a0b2306..3d63205e0 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -2194,12 +2194,20 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, int32_t remaining_in_page = page_time_count_ - page_time_cursor_; uint32_t budget = row_appender.remaining(); - // Fast path: FULL_PASS page, no nulls in any value column, types - // match destination, budget > 0. Bulk-memcpys up to + // Fast path: FULL_PASS page, or a BOUNDARY page under a contiguous + // time filter (e.g. TimeBetween) whose whole [row_begin, row_end] + // range matches — no nulls in any value column, types match + // destination, budget > 0. Bulk-memcpys up to // min(budget, remaining_in_page) rows from page_time_cursor_; tail // pages of an SSI tsblock still take the memcpy path instead of - // falling into the row-by-row scatter loop. - bool can_bulk = page_info.pass_type == PagePassType::FULL_PASS && + // falling into the row-by-row scatter loop. Discrete filters (TimeIn, + // NOT BETWEEN) keep the slow path because the compressed range can + // still contain unmatched timestamps. + const bool boundary_contiguous = + page_info.pass_type == PagePassType::BOUNDARY && + (filter == nullptr || filter->is_contiguous_time_filter()); + bool can_bulk = (page_info.pass_type == PagePassType::FULL_PASS || + boundary_contiguous) && remaining_in_page > 0 && budget > 0; if (can_bulk) { for (uint32_t c = 0; c < num_cols; c++) { @@ -2209,7 +2217,7 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, if (dt == common::STRING || dt == common::TEXT || dt == common::BLOB || ret_tsblock->get_vector(c + 1)->get_vector_type() != dt || - pps.predecoded_count != page_time_count_) { + pps.predecoded_count != pps.row_count) { can_bulk = false; break; } diff --git a/cpp/src/reader/filter/filter.h b/cpp/src/reader/filter/filter.h index e53992308..3fe7eb680 100644 --- a/cpp/src/reader/filter/filter.h +++ b/cpp/src/reader/filter/filter.h @@ -64,6 +64,14 @@ class Filter { return nullptr; } + // True when this time filter selects one contiguous interval: inside a + // BOUNDARY page's [row_begin, row_end] range every timestamp matches (e.g. + // TimeBetween, one-sided Gt/Lt), so the range can be bulk-copied. Discrete + // sets (TimeIn) or negated forms (NOT BETWEEN / NotEq) may contain holes + // and must still be checked per value. Defaults to false; only explicitly + // contiguous time filters override it to true. + virtual bool is_contiguous_time_filter() const { return false; } + // Batch time filter: evaluate time filter on an array of timestamps. // Writes true/false into @mask for each element. // Returns the number of elements that passed (mask[i] == true). diff --git a/cpp/src/reader/filter/time_operator.h b/cpp/src/reader/filter/time_operator.h index f972a4259..b48ca7fba 100644 --- a/cpp/src/reader/filter/time_operator.h +++ b/cpp/src/reader/filter/time_operator.h @@ -52,6 +52,8 @@ class TimeBetween : public Filter { FilterType get_filter_type() { return type_; } + bool is_contiguous_time_filter() const override { return !not_; } + private: int64_t value1_; int64_t value2_; From a74b8af50d9b20d0cc26fa209b49e2451d2b5964 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Thu, 5 Nov 2026 18:45:34 +0800 Subject: [PATCH 3/7] perf(cpp): avoid pattern memcpy and dispatch TS_2DIFF SIMD to AVX2 - read_length_and_bitwidth: wrap byte_cache_ directly over the caller's wrapped buffer instead of copying the whole RLE pattern into tmp_buf_; fall back to the copy for paged or short streams. Removes a per-pattern memcpy of the encoded payload. - TS_2DIFF: add __attribute__((target("avx2"))) twins of simd_decode_4_i64/i32 and dispatch through common::cpu_has_avx2(), so the gather/shuffle kernels run as real AVX2 instead of simde's portable fallback in baseline builds (the library does not compile with -mavx2). TS_2DIFF round-trip (smooth / jumpy / small / constant columns) and RLE round-trip both pass; 1-day/1-week wide-table queries ~8-10% faster. --- cpp/src/encoding/int32_rle_decoder.h | 26 +++-- cpp/src/encoding/int64_rle_decoder.h | 20 +++- cpp/src/encoding/ts2diff_decoder.h | 141 +++++++++++++++++++++++++-- 3 files changed, 168 insertions(+), 19 deletions(-) diff --git a/cpp/src/encoding/int32_rle_decoder.h b/cpp/src/encoding/int32_rle_decoder.h index f840b583b..09e631878 100644 --- a/cpp/src/encoding/int32_rle_decoder.h +++ b/cpp/src/encoding/int32_rle_decoder.h @@ -280,6 +280,16 @@ class Int32RleDecoder : public Decoder { if (RET_FAIL( common::SerializationUtil::read_var_uint(length_, buffer))) { return common::E_PARTIAL_READ; + } + // Prefer a read-only view over the caller's buffer instead of copying + // the whole pattern into tmp_buf_. `buffer` is the current value + // page's ByteStream and outlives the pattern decode; byte_cache_ only + // reads from it. Fall back to the copy when the stream is paged or + // shorter than the pattern. + if (buffer.is_wrapped() && buffer.remaining_size() >= length_) { + byte_cache_.wrap_from(buffer.get_wrapped_buf() + buffer.read_pos(), + static_cast(length_)); + buffer.wrapped_buf_advance_read_pos(length_); } else { if (tmp_buf_) { common::mem_free(tmp_buf_); @@ -297,15 +307,15 @@ class Int32RleDecoder : public Decoder { ret = common::E_PARTIAL_READ; } byte_cache_.wrap_from((char*)tmp_buf_, length_); - is_length_and_bitwidth_readed_ = true; - uint8_t tmp_bit_width; - common::SerializationUtil::read_ui8(tmp_bit_width, byte_cache_); - bit_width_ = tmp_bit_width; - if (packer_ != nullptr) { - delete packer_; - } - init_packer(); } + is_length_and_bitwidth_readed_ = true; + uint8_t tmp_bit_width; + common::SerializationUtil::read_ui8(tmp_bit_width, byte_cache_); + bit_width_ = tmp_bit_width; + if (packer_ != nullptr) { + delete packer_; + } + init_packer(); return ret; } diff --git a/cpp/src/encoding/int64_rle_decoder.h b/cpp/src/encoding/int64_rle_decoder.h index 1fdd13990..0d8fa94f4 100644 --- a/cpp/src/encoding/int64_rle_decoder.h +++ b/cpp/src/encoding/int64_rle_decoder.h @@ -285,6 +285,16 @@ class Int64RleDecoder : public Decoder { if (RET_FAIL( common::SerializationUtil::read_var_uint(length_, buffer))) { return common::E_PARTIAL_READ; + } + // Prefer a read-only view over the caller's buffer instead of copying + // the whole pattern into tmp_buf_. `buffer` is the current value + // page's ByteStream and outlives the pattern decode; byte_cache_ only + // reads from it. Fall back to the copy when the stream is paged or + // shorter than the pattern. + if (buffer.is_wrapped() && buffer.remaining_size() >= length_) { + byte_cache_.wrap_from(buffer.get_wrapped_buf() + buffer.read_pos(), + static_cast(length_)); + buffer.wrapped_buf_advance_read_pos(length_); } else { if (tmp_buf_) { common::mem_free(tmp_buf_); @@ -302,12 +312,12 @@ class Int64RleDecoder : public Decoder { ret = common::E_PARTIAL_READ; } byte_cache_.wrap_from((char*)tmp_buf_, length_); - is_length_and_bitwidth_readed_ = true; - uint8_t tmp_bit_width; - common::SerializationUtil::read_ui8(tmp_bit_width, byte_cache_); - bit_width_ = tmp_bit_width; - init_packer(); } + is_length_and_bitwidth_readed_ = true; + uint8_t tmp_bit_width; + common::SerializationUtil::read_ui8(tmp_bit_width, byte_cache_); + bit_width_ = tmp_bit_width; + init_packer(); return ret; } diff --git a/cpp/src/encoding/ts2diff_decoder.h b/cpp/src/encoding/ts2diff_decoder.h index 721656b9c..5031c9226 100644 --- a/cpp/src/encoding/ts2diff_decoder.h +++ b/cpp/src/encoding/ts2diff_decoder.h @@ -34,7 +34,11 @@ #include "utils/util_define.h" #ifdef ENABLE_SIMD +#include "common/simd_dispatch.h" #include "simde/x86/avx2.h" +#ifdef TSFILE_X86_AVX2_DISPATCH +#include +#endif #endif namespace storage { @@ -53,9 +57,10 @@ namespace storage { // @base: the previous reconstructed value (for prefix-sum) // @out: output array (4 values written) // Returns: the last reconstructed value (new base for next group) -static inline int32_t simd_decode_4_i32(const uint8_t* in, int32_t bit_width, - int32_t delta_min, int32_t index, - int32_t base, int32_t out[4]) { +static inline int32_t simd_decode_4_i32_simde(const uint8_t* in, + int32_t bit_width, + int32_t delta_min, int32_t index, + int32_t base, int32_t out[4]) { static const simde__m128i SHUF_REV4 = simde_mm_setr_epi8( 3, 2, 1, 0, 7, 6, 5, 4, 11, 10, 9, 8, 15, 14, 13, 12); @@ -114,10 +119,77 @@ static inline int32_t simd_decode_4_i32(const uint8_t* in, int32_t bit_width, return out[3]; } +#ifdef TSFILE_X86_AVX2_DISPATCH +// Native-AVX2 twin of simd_decode_4_i32_simde (see simd_decode_4_i64_avx2). +TSFILE_TARGET_AVX2 +static inline int32_t simd_decode_4_i32_avx2(const uint8_t* in, + int32_t bit_width, + int32_t delta_min, int32_t index, + int32_t base, int32_t out[4]) { + const __m128i SHUF_REV4 = + _mm_setr_epi8(3, 2, 1, 0, 7, 6, 5, 4, 11, 10, 9, 8, 15, 14, 13, 12); + const __m128i VMIN4 = _mm_set1_epi32(delta_min); + int32_t pos0 = index * bit_width; + int32_t pos[4] = {pos0, pos0 + bit_width, pos0 + 2 * bit_width, + pos0 + 3 * bit_width}; + int32_t bidx[4] = {pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, pos[3] >> 3}; + int32_t off[4] = {pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7}; + __m128i IDX = _mm_setr_epi32(bidx[0], bidx[1], bidx[2], bidx[3]); + __m128i OFF = _mm_setr_epi32(off[0], off[1], off[2], off[3]); + __m128i V4; + if (bit_width <= 16) { + int rshift = 32 - bit_width; + __m128i w32_le = _mm_i32gather_epi32((const int*)in, IDX, 1); + __m128i w32_be = _mm_shuffle_epi8(w32_le, SHUF_REV4); + __m128i U32 = _mm_sllv_epi32(w32_be, OFF); + __m128i RS32 = _mm_set1_epi32(rshift); + V4 = _mm_srlv_epi32(U32, RS32); + } else { + const __m256i SHUF_REV8 = _mm256_setr_epi8( + 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, + 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); + int rshift = 64 - bit_width; + __m256i w64_le = _mm256_i32gather_epi64((const long long*)in, IDX, 1); + __m256i w64_be = _mm256_shuffle_epi8(w64_le, SHUF_REV8); + __m256i OFF64 = _mm256_cvtepu32_epi64(OFF); + __m256i U64 = _mm256_sllv_epi64(w64_be, OFF64); + __m256i V64 = _mm256_srl_epi64(U64, _mm_cvtsi32_si128(rshift)); + __m256i perm = _mm256_setr_epi32(0, 2, 4, 6, 0, 0, 0, 0); + __m256i comp = _mm256_permutevar8x32_epi32(V64, perm); + V4 = _mm256_castsi256_si128(comp); + } + // Add delta_min + V4 = _mm_add_epi32(V4, VMIN4); + // Prefix sum to reconstruct absolute values + __m128i t = _mm_slli_si128(V4, 4); + V4 = _mm_add_epi32(V4, t); + t = _mm_slli_si128(V4, 8); + V4 = _mm_add_epi32(V4, t); + // Add base + __m128i C4 = _mm_set1_epi32(base); + V4 = _mm_add_epi32(V4, C4); + _mm_storeu_si128((__m128i*)out, V4); + return out[3]; +} +#endif // TSFILE_X86_AVX2_DISPATCH + +static inline int32_t simd_decode_4_i32(const uint8_t* in, int32_t bit_width, + int32_t delta_min, int32_t index, + int32_t base, int32_t out[4]) { +#ifdef TSFILE_X86_AVX2_DISPATCH + if (common::cpu_has_avx2()) { + return simd_decode_4_i32_avx2(in, bit_width, delta_min, index, base, + out); + } +#endif + return simd_decode_4_i32_simde(in, bit_width, delta_min, index, base, out); +} + // Decode 4 INT64 values from bit-packed data using SIMD. -static inline int64_t simd_decode_4_i64(const uint8_t* in, int32_t bit_width, - int64_t delta_min, int32_t index, - int64_t base, int64_t out[4]) { +static inline int64_t simd_decode_4_i64_simde(const uint8_t* in, + int32_t bit_width, + int64_t delta_min, int32_t index, + int64_t base, int64_t out[4]) { static const simde__m256i SHUF_REV8 = simde_mm256_setr_epi8( 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); @@ -166,6 +238,63 @@ static inline int64_t simd_decode_4_i64(const uint8_t* in, int32_t bit_width, return out[3]; } +#ifdef TSFILE_X86_AVX2_DISPATCH +// Native-AVX2 twin of simd_decode_4_i64_simde. simde only emits native AVX2 +// when the whole translation unit is compiled with -mavx2; the library targets +// the portable baseline, so a target-attributed copy is needed for the gather +// and shuffle to run as real AVX2 instructions. +TSFILE_TARGET_AVX2 +static inline int64_t simd_decode_4_i64_avx2(const uint8_t* in, + int32_t bit_width, + int64_t delta_min, int32_t index, + int64_t base, int64_t out[4]) { + const __m256i SHUF_REV8 = + _mm256_setr_epi8(7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8, + 7, 6, 5, 4, 3, 2, 1, 0, 15, 14, 13, 12, 11, 10, 9, 8); + const __m256i VMIN4 = _mm256_set1_epi64x(delta_min); + int32_t pos0 = index * bit_width; + int32_t pos[4] = {pos0, pos0 + bit_width, pos0 + 2 * bit_width, + pos0 + 3 * bit_width}; + int32_t bidx[4] = {pos[0] >> 3, pos[1] >> 3, pos[2] >> 3, pos[3] >> 3}; + int32_t off[4] = {pos[0] & 7, pos[1] & 7, pos[2] & 7, pos[3] & 7}; + __m128i IDX = _mm_setr_epi32(bidx[0], bidx[1], bidx[2], bidx[3]); + int rshift = 64 - bit_width; + __m256i w64_le = _mm256_i32gather_epi64((const long long*)in, IDX, 1); + __m256i w64_be = _mm256_shuffle_epi8(w64_le, SHUF_REV8); + __m256i OFF64 = + _mm256_cvtepu32_epi64(_mm_setr_epi32(off[0], off[1], off[2], off[3])); + __m256i U64 = _mm256_sllv_epi64(w64_be, OFF64); + __m256i V64 = _mm256_srl_epi64(U64, _mm_cvtsi32_si128(rshift)); + // Add delta_min + V64 = _mm256_add_epi64(V64, VMIN4); + // Prefix sum (64-bit, 4 lanes) + __m256i t = _mm256_slli_si256(V64, 8); + V64 = _mm256_add_epi64(V64, t); + int64_t tmp_buf[4]; + _mm256_storeu_si256((__m256i*)tmp_buf, V64); + tmp_buf[2] += tmp_buf[1]; + tmp_buf[3] += tmp_buf[1]; + V64 = _mm256_loadu_si256((const __m256i*)tmp_buf); + // Add base + __m256i C4 = _mm256_set1_epi64x(base); + V64 = _mm256_add_epi64(V64, C4); + _mm256_storeu_si256((__m256i*)out, V64); + return out[3]; +} +#endif // TSFILE_X86_AVX2_DISPATCH + +static inline int64_t simd_decode_4_i64(const uint8_t* in, int32_t bit_width, + int64_t delta_min, int32_t index, + int64_t base, int64_t out[4]) { +#ifdef TSFILE_X86_AVX2_DISPATCH + if (common::cpu_has_avx2()) { + return simd_decode_4_i64_avx2(in, bit_width, delta_min, index, base, + out); + } +#endif + return simd_decode_4_i64_simde(in, bit_width, delta_min, index, base, out); +} + #endif // ENABLE_SIMD // ============================================================================ From d20987b54c619bbda2becd168c1bd1b109928e5f Mon Sep 17 00:00:00 2001 From: ColinLee Date: Fri, 6 Nov 2026 15:00:07 +0800 Subject: [PATCH 4/7] perf(cpp): reuse one TsBlock across devices in device-ordered scans A device-ordered scan creates a SingleDeviceTsBlockReader for every device it visits - and with it a TsFileSeriesScanIterator and an output TsBlock plus one buffer per column - then throws all of it away. On a wide table every device rebuilds the same time+value layout, so the allocation and first-touch cost of those buffers shows up as system time and page faults: 20 devices of a 4501-column table churn several GB of anonymous memory per query. Let the enclosing DeviceOrderedTsBlockReader own one ReusableTsBlock and hand it to each device's iterator through a small factory. The factory keeps the block when the column count, value types and row capacity match, refreshing the column names in place, and only rebuilds it when the layout actually changes. Iterators without a factory keep the previous allocate-per-series behaviour, so the prepared-series and tree paths are untouched. --- .../block/device_ordered_tsblock_reader.cc | 1 + .../block/device_ordered_tsblock_reader.h | 4 ++ .../block/single_device_tsblock_reader.cc | 17 +++-- .../block/single_device_tsblock_reader.h | 70 ++++++++++++++++++- cpp/src/reader/tsfile_series_scan_iterator.cc | 32 ++++++++- cpp/src/reader/tsfile_series_scan_iterator.h | 18 +++++ 6 files changed, 135 insertions(+), 7 deletions(-) diff --git a/cpp/src/reader/block/device_ordered_tsblock_reader.cc b/cpp/src/reader/block/device_ordered_tsblock_reader.cc index 5c30d9bc6..0ccde9144 100644 --- a/cpp/src/reader/block/device_ordered_tsblock_reader.cc +++ b/cpp/src/reader/block/device_ordered_tsblock_reader.cc @@ -69,6 +69,7 @@ int DeviceOrderedTsBlockReader::has_next(bool& has_next) { current_reader_ = new SingleDeviceTsBlockReader( task, block_size_, metadata_querier_, tsfile_io_reader_, time_filter_, field_filter_); + current_reader_->set_reusable_block(&reusable_block_); if (current_reader_ == nullptr) { return common::E_OOM; } diff --git a/cpp/src/reader/block/device_ordered_tsblock_reader.h b/cpp/src/reader/block/device_ordered_tsblock_reader.h index 8018725c9..0190d11a8 100644 --- a/cpp/src/reader/block/device_ordered_tsblock_reader.h +++ b/cpp/src/reader/block/device_ordered_tsblock_reader.h @@ -58,6 +58,10 @@ class DeviceOrderedTsBlockReader : public TsBlockReader { Filter* field_filter_ = nullptr; int remaining_offset_ = 0; int remaining_limit_ = -1; + // Output block shared by the per-device readers this iterator creates. + // Each device otherwise allocates (and faults in) its own block and column + // buffers; the layout is identical across devices, so one block is enough. + ReusableTsBlock reusable_block_; }; } // namespace storage diff --git a/cpp/src/reader/block/single_device_tsblock_reader.cc b/cpp/src/reader/block/single_device_tsblock_reader.cc index 619201e63..19e6dd16d 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.cc +++ b/cpp/src/reader/block/single_device_tsblock_reader.cc @@ -30,7 +30,12 @@ namespace storage { namespace { const char* kTimeOnlyContextName = "__time_only_aligned_context__"; + +common::TsBlock* reusable_block_factory(void* ctx, common::TupleDesc& desc, + uint32_t max_rows) { + return static_cast(ctx)->acquire(desc, max_rows); } +} // namespace SingleDeviceTsBlockReader::SingleDeviceTsBlockReader( DeviceQueryTask* device_query_task, uint32_t block_size, @@ -283,9 +288,9 @@ int SingleDeviceTsBlockReader::init_internal(DeviceQueryTask* device_query_task, dense_row_count_ >= 0) ? remaining_offset_ : 0; - int ctx_ret = - ctx->init(device_query_task_, meas_names, time_filter, pos_list, - pa_, aligned_field_filter_, ssi_offset, -1); + int ctx_ret = ctx->init(device_query_task_, meas_names, time_filter, + pos_list, pa_, aligned_field_filter_, + ssi_offset, -1, reusable_block_); if (ctx_ret == common::E_OK) { if (ssi_offset > 0) { row_offset_pushed_to_ssi_ = true; @@ -1209,7 +1214,8 @@ int VectorMeasurementColumnContext::init( DeviceQueryTask* device_query_task, const std::vector& measurement_names, Filter* time_filter, std::vector>& pos_in_result, common::PageArena& pa, - AlignedFieldFilter* field_filter, int ssi_offset, int ssi_limit) { + AlignedFieldFilter* field_filter, int ssi_offset, int ssi_limit, + ReusableTsBlock* reusable_block) { int ret = common::E_OK; pos_in_result_ = pos_in_result; column_names_ = measurement_names; @@ -1218,6 +1224,9 @@ int VectorMeasurementColumnContext::init( time_filter, field_filter))) { return ret; } + if (reusable_block != nullptr) { + ssi_->set_tsblock_factory(reusable_block_factory, reusable_block); + } ssi_->set_row_range(ssi_offset, ssi_limit); if (RET_FAIL(get_next_tsblock(true))) { return ret; diff --git a/cpp/src/reader/block/single_device_tsblock_reader.h b/cpp/src/reader/block/single_device_tsblock_reader.h index fb592a6c4..6081a3234 100644 --- a/cpp/src/reader/block/single_device_tsblock_reader.h +++ b/cpp/src/reader/block/single_device_tsblock_reader.h @@ -33,6 +33,66 @@ class DeviceQueryTask; class MeasurementColumnContext; class IdColumnContext; +// One output TsBlock shared by every device a reader visits. A device-ordered +// scan builds the same time+value layout for each series; allocating a fresh +// block (and faulting in its per-column buffers) per device is a large share of +// the query cost once the table is wide, so the layout is kept and only rebuilt +// when the column count or value types change. +class ReusableTsBlock { + public: + ReusableTsBlock() {} + ~ReusableTsBlock() { release(); } + ReusableTsBlock(const ReusableTsBlock&) = delete; + ReusableTsBlock& operator=(const ReusableTsBlock&) = delete; + + // Returns a block matching `wanted` (time + value columns), reusing the + // previous buffers when the shape matches. The caller resets it before use. + common::TsBlock* acquire(common::TupleDesc& wanted, uint32_t max_rows) { + bool rebuild = block_ == nullptr || max_rows_ != max_rows || + desc_.get_column_count() != wanted.get_column_count(); + if (!rebuild) { + for (uint32_t i = 0; i < wanted.get_column_count(); i++) { + if (desc_[i].data_type_ != wanted[i].data_type_) { + rebuild = true; + break; + } + } + } + if (rebuild) { + release(); + for (uint32_t i = 0; i < wanted.get_column_count(); i++) { + desc_.push_back(wanted[i]); + } + max_rows_ = max_rows; + block_ = new common::TsBlock(&desc_, max_rows_); + if (block_->init() != common::E_OK) { + delete block_; + block_ = nullptr; + } + return block_; + } + // Same layout: refresh names in place, the block points at desc_. + for (uint32_t i = 0; i < wanted.get_column_count(); i++) { + desc_[i] = wanted[i]; + } + return block_; + } + + void release() { + if (block_ != nullptr) { + delete block_; + block_ = nullptr; + } + desc_.release(); + max_rows_ = 0; + } + + private: + common::TupleDesc desc_; + common::TsBlock* block_ = nullptr; + uint32_t max_rows_ = 0; +}; + class SingleDeviceTsBlockReader : public TsBlockReader { public: explicit SingleDeviceTsBlockReader(DeviceQueryTask* device_query_task, @@ -51,6 +111,9 @@ class SingleDeviceTsBlockReader : public TsBlockReader { int row_limit); void close() override; + // Borrow a block that can be reused across devices (see ReusableTsBlock). + void set_reusable_block(ReusableTsBlock* block) { reusable_block_ = block; } + int get_remaining_offset() const { return remaining_offset_; } int get_remaining_limit() const { return remaining_limit_; } int32_t get_dense_row_count() const { return dense_row_count_; } @@ -103,6 +166,10 @@ class SingleDeviceTsBlockReader : public TsBlockReader { SelectionMask aligned_filter_mask_; uint32_t aligned_filter_next_row_ = 0; bool aligned_filter_mask_valid_ = false; + // Output block owned by the enclosing device-ordered reader and handed to + // this device's aligned multi-value reader, so the per-column buffers + // survive across devices. Null when this reader owns nothing (standalone). + ReusableTsBlock* reusable_block_ = nullptr; }; class MeasurementColumnContext { @@ -205,7 +272,8 @@ class VectorMeasurementColumnContext final : public MeasurementColumnContext { Filter* time_filter, std::vector>& pos_in_result, common::PageArena& pa, AlignedFieldFilter* field_filter = nullptr, - int ssi_offset = 0, int ssi_limit = -1); + int ssi_offset = 0, int ssi_limit = -1, + ReusableTsBlock* reusable_block = nullptr); int get_next_tsblock(bool alloc_mem) override; int get_current_time(int64_t& time) override; int get_current_value(char*& value, uint32_t& len) override; diff --git a/cpp/src/reader/tsfile_series_scan_iterator.cc b/cpp/src/reader/tsfile_series_scan_iterator.cc index 948d1a553..adc1bf130 100644 --- a/cpp/src/reader/tsfile_series_scan_iterator.cc +++ b/cpp/src/reader/tsfile_series_scan_iterator.cc @@ -363,8 +363,7 @@ int TsFileSeriesScanIterator::get_next(TsBlock*& ret_tsblock, bool alloc, } if (IS_SUCC(ret)) { if (alloc && ret_tsblock == nullptr) { - ret_tsblock = - is_multi_value_ ? alloc_tsblock_multi() : alloc_tsblock(); + ret_tsblock = alloc_or_reuse_tsblock(); } ret = chunk_reader_->get_next_page(ret_tsblock, filter, *data_pa_, min_time_hint, row_offset_, @@ -540,6 +539,35 @@ int TsFileSeriesScanIterator::init_chunk_reader_multi() { return ret; } +TsBlock* TsFileSeriesScanIterator::alloc_or_reuse_tsblock() { + if (!is_multi_value_ || tsblock_factory_ == nullptr) { + return is_multi_value_ ? alloc_tsblock_multi() : alloc_tsblock(); + } + auto* acr = static_cast(chunk_reader_); + // Same layout alloc_tsblock_multi() would build: time + one column per + // value measurement of the series now loaded. + tuple_desc_.reset(); + tuple_desc_.push_back(ColumnSchema("time", common::INT64, + common::INVALID_COMPRESSION, + common::INVALID_ENCODING)); + const uint32_t num_cols = acr->get_value_column_count(); + for (uint32_t c = 0; c < num_cols; c++) { + ChunkHeader& ch = acr->get_value_chunk_header(c); + tuple_desc_.push_back(ColumnSchema(ch.measurement_name_, ch.data_type_, + ch.compression_type_, + ch.encoding_type_)); + } + common::TsBlock* block = + tsblock_factory_(tsblock_factory_ctx_, tuple_desc_, max_block_rows_); + if (block != nullptr) { + block->reset(); + return block; + } + // The factory could not match this layout; a fresh block is allocated and + // owned by this iterator (destroyed by revert_tsblock()/destroy()). + return alloc_tsblock_multi(); +} + TsBlock* TsFileSeriesScanIterator::alloc_tsblock() { ChunkHeader& ch = chunk_reader_->get_chunk_header(); diff --git a/cpp/src/reader/tsfile_series_scan_iterator.h b/cpp/src/reader/tsfile_series_scan_iterator.h index 19c261694..53c63058c 100644 --- a/cpp/src/reader/tsfile_series_scan_iterator.h +++ b/cpp/src/reader/tsfile_series_scan_iterator.h @@ -96,6 +96,20 @@ class TsFileSeriesScanIterator { max_block_rows_ = max_block_rows; } + // Optional factory for the output TsBlock. A device-ordered scan reads one + // series at a time and would otherwise allocate (and fault in) a fresh + // block plus all its per-column buffers for every device. The factory is + // handed the exact layout the block must have and may return a matching + // block that is reused across series; returning nullptr makes the iterator + // allocate a new one. The returned block must already be reset(). + using TsBlockFactory = common::TsBlock* (*)(void* ctx, + common::TupleDesc& desc, + uint32_t max_block_rows); + void set_tsblock_factory(TsBlockFactory factory, void* ctx) { + tsblock_factory_ = factory; + tsblock_factory_ctx_ = ctx; + } + /** Current row offset/limit after chunk/page skip; used to sync with QDS * for single-path. */ int get_row_offset() const { return row_offset_; } @@ -215,6 +229,8 @@ class TsFileSeriesScanIterator { ChunkMeta* time_cm, const std::vector& value_cms); common::TsBlock* alloc_tsblock(); common::TsBlock* alloc_tsblock_multi(); + // Reuse a block supplied by tsblock_factory_, or allocate a new one. + common::TsBlock* alloc_or_reuse_tsblock(); private: RandomAccessReadFile* read_file_; @@ -247,6 +263,8 @@ class TsFileSeriesScanIterator { uint32_t max_block_rows_; int row_offset_; int row_limit_; + TsBlockFactory tsblock_factory_ = nullptr; + void* tsblock_factory_ctx_ = nullptr; }; } // end namespace storage From ed1d73323d83feb23c84c6a3e78073b2f35a2a7f Mon Sep 17 00:00:00 2001 From: ColinLee Date: Fri, 6 Nov 2026 15:00:20 +0800 Subject: [PATCH 5/7] perf(cpp): decode aligned value pages into the TsBlock and harden bulk copies When the whole page plan of a chunk fits into the current TsBlock, value pages are decoded straight into the destination column buffer instead of a staging buffer that is copied over afterwards. Each page's destination is a prefix sum of the retained rows, so only pages whose retained range starts at row 0 and covers the whole page are written in place; pages with nulls, partial ranges or offset pushdown keep the staging path, and the two can be mixed within a chunk. TsBlock columns store non-null values compactly (RowIterator advances offsets only for non-null rows), so the write cursor is the column's byte size rather than a row index. Also fixes two bugs in the same bulk path: - a BOUNDARY page bounded by a FIELD predicate mask can still reject rows inside [row_begin, row_end), so only contiguous time filters may bulk copy; - a column without a value page for this page decodes to an empty slot whose predecoded count equals its zero row count, which passed the bulk check and memcpy'd from a null buffer (SIGSEGV in WriteDataWithEmptyField and in the aligned field-filter regression suite). --- cpp/src/common/container/byte_buffer.h | 8 + cpp/src/reader/aligned_chunk_reader.cc | 201 ++++++++++++++++++++++--- cpp/src/reader/aligned_chunk_reader.h | 27 ++++ 3 files changed, 214 insertions(+), 22 deletions(-) diff --git a/cpp/src/common/container/byte_buffer.h b/cpp/src/common/container/byte_buffer.h index 3f4954093..a3b2c525f 100644 --- a/cpp/src/common/container/byte_buffer.h +++ b/cpp/src/common/container/byte_buffer.h @@ -133,6 +133,14 @@ class ByteBuffer { FORCE_INLINE uint32_t get_data_size() const { return real_data_size_; } + // Declare bytes that were written directly through get_data() without going + // through append_*(). Used when a caller decodes straight into this buffer + // at a known offset. + FORCE_INLINE void set_data_size(uint32_t size) { + ASSERT(size <= reserved_size_); + real_data_size_ = size; + } + private: char* data_; uint8_t variable_type_len_; diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index 3d63205e0..24e616740 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -117,6 +117,7 @@ void AlignedChunkReader::reset() { col->chunk_meta = nullptr; for (auto& pps : col->per_page_state) { pps.predecode_pa.destroy(); + pps.direct_values = nullptr; } col->per_page_state.clear(); col->pending_decoded_values.clear(); @@ -202,6 +203,7 @@ void AlignedChunkReader::destroy() { } for (auto& pps : col->per_page_state) { pps.predecode_pa.destroy(); + pps.direct_values = nullptr; } col->per_page_state.clear(); col->pending_decoded_values.clear(); @@ -1687,6 +1689,11 @@ int AlignedChunkReader::build_page_plan(Filter* filter, int& row_offset) { page_plan_built_ = IS_SUCC(ret); if (page_plan_built_) { + int32_t out_row = 0; + for (auto& page : chunk_pages_) { + page.out_row_begin = out_row; + out_row += page.row_end - page.row_begin; + } per_page_times_.assign(chunk_pages_.size(), std::vector{}); for (auto* col : value_columns_) { col->per_page_state.clear(); @@ -1701,6 +1708,9 @@ void AlignedChunkReader::release_current_page_state() { page_all_times_.clear(); page_time_count_ = 0; page_time_cursor_ = 0; + direct_page_end_ = 0; + direct_tsblock_ = nullptr; + direct_base_row_ = 0; for (auto* col : value_columns_) { if (col->uncompressed_buf != nullptr && col->compressor != nullptr) { col->compressor->after_uncompress(col->uncompressed_buf); @@ -1711,6 +1721,7 @@ void AlignedChunkReader::release_current_page_state() { col->in.reset(); for (auto& pps : col->per_page_state) { pps.predecode_pa.destroy(); + pps.direct_values = nullptr; } col->per_page_state.clear(); col->pending_decoded_values.clear(); @@ -1734,6 +1745,7 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, pps.predecoded_strings.clear(); pps.predecoded_read_pos = 0; pps.predecoded_count = 0; + pps.direct_values = nullptr; pps.predecode_pa.destroy(); if (page_info.value_compressed_sizes[col_idx] == 0) { @@ -1848,11 +1860,28 @@ 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); + char* values_dst = nullptr; + // Direct decode: prepare_direct_decode() reserved this page's output rows + // in the destination Vector before decoding started, and a page without + // nulls maps one decoded value to one row, so the value stream can be + // written in place. Pages that carry nulls, or whose retained range is only + // part of the page, keep their own staging buffer (the row/scatter code + // copies those). + if (page_idx < direct_page_end_ && page_info.row_begin == 0 && + page_info.row_end == static_cast(data_num) && + nonnull_total == static_cast(data_num)) { + common::Vector* vec = direct_tsblock_->get_vector(col_idx + 1); + values_dst = + vec->get_value_data().get_data() + direct_col_off_[col_idx]; + pps.direct_values = values_dst; + } else { + pps.predecoded_values.resize(static_cast(nonnull_total) * + elem_size); + values_dst = pps.predecoded_values.data(); + } switch (dt) { case common::BOOLEAN: { - bool* out = reinterpret_cast(pps.predecoded_values.data()); + bool* out = reinterpret_cast(values_dst); for (int i = 0; i < nonnull_total; i++) { if (RET_FAIL(col->decoder->read_boolean(out[i], in))) { cleanup(); @@ -1864,8 +1893,8 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, case common::INT32: case common::DATE: if (RET_FAIL(col->decoder->read_exact_int32( - reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, in))) { + reinterpret_cast(values_dst), nonnull_total, + in))) { cleanup(); return ret; } @@ -1873,24 +1902,23 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, case common::INT64: case common::TIMESTAMP: if (RET_FAIL(col->decoder->read_exact_int64( - reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, in))) { + reinterpret_cast(values_dst), nonnull_total, + in))) { cleanup(); return ret; } break; case common::FLOAT: if (RET_FAIL(col->decoder->read_exact_float( - reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, in))) { + reinterpret_cast(values_dst), nonnull_total, in))) { cleanup(); return ret; } break; case common::DOUBLE: if (RET_FAIL(col->decoder->read_exact_double( - reinterpret_cast(pps.predecoded_values.data()), - nonnull_total, in))) { + reinterpret_cast(values_dst), nonnull_total, + in))) { cleanup(); return ret; } @@ -1900,10 +1928,105 @@ int AlignedChunkReader::decode_value_page_for_slot(uint32_t col_idx, return E_NOT_SUPPORT; } pps.predecoded_count = nonnull_total; + if (page_idx < direct_page_end_) { + // Values of this page occupy space in the destination column whether + // they were written in place or will be appended from this buffer. + direct_col_off_[col_idx] += + static_cast(nonnull_total) * elem_size; + } cleanup(); return E_OK; } +// Decide how many leading pages of the plan can be decoded straight into the +// destination TsBlock's column buffers. The destination address of a row must +// be fixed before decode starts, because parallel decode runs ahead of the +// scatter loop, so the pages written in place must all be scattered within the +// same call — that is exactly the prefix of pages whose output rows still fit +// in the rows available in the current TsBlock (the scatter consumes pages in +// order and only swaps blocks once the budget is exhausted). +// - no FIELD filter: a page-level mask can drop rows and shift the output; +// - a contiguous time filter: a page's retained rows stay one contiguous run; +// - FULL_PASS pages starting at row 0. A page whose decoded stream starts +// before its output slice (partial range / offset pushdown) would write +// before the destination base, so the direct prefix stops there; +// - fixed-length columns whose type matches the destination Vector. +// In lazy (single-thread) mode pages are decoded during the scatter itself, so +// a partial prefix cannot be used: the block may be swapped between pages +// before they are decoded. There, the whole plan must fit. +bool AlignedChunkReader::prepare_direct_decode(TsBlock* tsblock, + Filter* filter) { + direct_page_end_ = 0; + direct_tsblock_ = nullptr; + direct_base_row_ = 0; + direct_col_off_.clear(); + if (field_filter_ != nullptr || tsblock == nullptr || + chunk_pages_.empty()) { + return false; + } + if (filter != nullptr && !filter->is_contiguous_time_filter()) { + return false; + } + if (has_variable_length_value_column()) { + return false; + } + const uint32_t num_cols = value_columns_.size(); + if (num_cols == 0 || tsblock->get_column_count() < num_cols + 1) { + return false; + } + for (uint32_t c = 0; c < num_cols; c++) { + if (value_columns_[c]->chunk_header.data_type_ != + tsblock->get_vector(c + 1)->get_vector_type()) { + return false; + } + } + const int64_t base_row = static_cast(tsblock->get_row_count()); + const int64_t available = + static_cast(tsblock->get_max_row_count()) - base_row; + size_t end = 0; + for (size_t i = 0; i < chunk_pages_.size(); i++) { + const auto& page = chunk_pages_[i]; + // A page whose retained range starts at row 0 and covers the whole page + // (checked again after decode, once the row count is known) writes its + // values in order. That includes conservatively classified BOUNDARY + // pages: when the time predicate is contiguous, [row_begin, row_end) + // holds exactly the matching rows, so an all-covering range is + // equivalent to FULL_PASS even without page statistics. + if (page.row_begin != 0) { + break; + } + const int64_t page_end = + static_cast(page.out_row_begin) + page.row_end; + if (page_end > available) { + break; + } + end = i + 1; + } + if (end == 0) { + return false; + } + // decode_all_planned_pages() keeps a partial prefix only when it decodes + // every page up front; mirror that decision here. +#ifdef ENABLE_THREADS + const bool eager_decode = + decode_pool_ != nullptr && value_columns_.size() > 1; +#else + const bool eager_decode = false; +#endif + if (!eager_decode && end != chunk_pages_.size()) { + return false; + } + direct_page_end_ = end; + direct_base_row_ = static_cast(base_row); + direct_tsblock_ = tsblock; + direct_col_off_.assign(num_cols, 0); + for (uint32_t c = 0; c < num_cols; c++) { + direct_col_off_[c] = + tsblock->get_vector(c + 1)->get_value_data().get_data_size(); + } + return true; +} + // Multi-thread path: one task per value column, each decoding all non-SKIP // pages of that column serially. Time pages dispatched as worker-bucketed // strided tasks using a separate decoder/compressor for each task (filled @@ -2061,6 +2184,7 @@ void AlignedChunkReader::release_page_slot(size_t page_idx) { pps.predecode_pa.destroy(); pps.predecoded_count = 0; pps.predecoded_read_pos = 0; + pps.direct_values = nullptr; } } @@ -2133,6 +2257,7 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, if (RET_FAIL(build_page_plan(filter, row_offset))) { return ret; } + prepare_direct_decode(ret_tsblock, filter); if (RET_FAIL(decode_all_planned_pages())) { return ret; } @@ -2203,8 +2328,13 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, // falling into the row-by-row scatter loop. Discrete filters (TimeIn, // NOT BETWEEN) keep the slow path because the compressed range can // still contain unmatched timestamps. + // A BOUNDARY page may be bounded by the time filter or by a FIELD + // predicate mask. Only the former keeps every row of + // [row_begin, row_end): a field mask can still reject rows inside that + // range, so those pages must stay on the row-by-row path. const bool boundary_contiguous = page_info.pass_type == PagePassType::BOUNDARY && + field_filter_ == nullptr && (filter == nullptr || filter->is_contiguous_time_filter()); bool can_bulk = (page_info.pass_type == PagePassType::FULL_PASS || boundary_contiguous) && @@ -2217,7 +2347,14 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, if (dt == common::STRING || dt == common::TEXT || dt == common::BLOB || ret_tsblock->get_vector(c + 1)->get_vector_type() != dt || - pps.predecoded_count != pps.row_count) { + pps.predecoded_count != pps.row_count || + // The decoded buffer must cover every row about to be + // copied. A column without a value page for this page (a + // missing aligned measurement) decodes to an empty slot + // with row_count == 0 while the plan still expects rows; + // those rows are all null and the row-by-row path marks + // them as such. + pps.row_count < static_cast(remaining_in_page)) { can_bulk = false; break; } @@ -2246,10 +2383,20 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, uint32_t elem_size = common::get_data_type_size(col->chunk_header.data_type_); common::Vector* vec = ret_tsblock->get_vector(c + 1); - vec->get_value_data().append_fixed_value( - pps.predecoded_values.data() + - static_cast(page_time_cursor_) * elem_size, - bulk_count * elem_size); + // A directly decoded page already sits in the destination + // buffer; only declare its bytes. Everything else is appended, + // which also keeps the column's compact layout (null rows take + // no space) in sync. + if (pps.direct_values == nullptr) { + vec->get_value_data().append_fixed_value( + pps.values() + + static_cast(page_time_cursor_) * elem_size, + static_cast(bulk_count) * elem_size); + } else { + vec->get_value_data().set_data_size( + vec->get_value_data().get_data_size() + + static_cast(bulk_count) * elem_size); + } vec->add_row_nums(bulk_count); } row_appender.add_rows(bulk_count); @@ -2332,12 +2479,22 @@ int AlignedChunkReader::get_next_page_multi_impl(TsBlock* ret_tsblock, } else { uint32_t elem_size = common::get_data_type_size( col->chunk_header.data_type_); - row_appender.append( - c + 1, - pps.predecoded_values.data() + - static_cast(pps.predecoded_read_pos++) * - elem_size, - elem_size); + if (pps.direct_values != nullptr) { + // Decode already placed this value in the destination + // buffer; only declare it, do not copy bytes onto + // themselves. + common::Vector* vec = ret_tsblock->get_vector(c + 1); + vec->get_value_data().set_data_size( + vec->get_value_data().get_data_size() + elem_size); + pps.predecoded_read_pos++; + } else { + row_appender.append( + c + 1, + pps.values() + + static_cast(pps.predecoded_read_pos++) * + elem_size, + elem_size); + } } } page_time_cursor_++; diff --git a/cpp/src/reader/aligned_chunk_reader.h b/cpp/src/reader/aligned_chunk_reader.h index 54c82799f..b64e3223c 100644 --- a/cpp/src/reader/aligned_chunk_reader.h +++ b/cpp/src/reader/aligned_chunk_reader.h @@ -49,6 +49,9 @@ struct ChunkPageInfo { uint32_t time_uncompressed_size = 0; int32_t row_begin = 0; // inclusive int32_t row_end = 0; // exclusive + // Row index of this page's first retained row in the chunk reader's output + // stream. Filled by build_page_plan() once the whole plan is known. + int32_t out_row_begin = 0; std::vector value_file_offsets; std::vector value_compressed_sizes; std::vector value_uncompressed_sizes; @@ -65,6 +68,17 @@ struct PageDecodedState { common::PageArena predecode_pa; int32_t predecoded_count = 0; int32_t predecoded_read_pos = 0; + // Non-null when the page was decoded straight into the destination TsBlock + // column buffer (see AlignedChunkReader::prepare_direct_decode()). Aliases + // memory owned by the Vector; predecoded_values stays empty in that case. + char* direct_values = nullptr; + + // Where this slot's fixed-length values live: the direct view when present, + // otherwise the slot's own buffer. + char* values() { + return direct_values != nullptr ? direct_values + : predecoded_values.data(); + } }; // Per-value-column state for multi-value AlignedChunkReader. @@ -274,6 +288,7 @@ class AlignedChunkReader : public IChunkReader { common::RowAppender& row_appender, Filter* filter, common::PageArena* pa); int build_page_plan(Filter* filter, int& row_offset); + bool prepare_direct_decode(common::TsBlock* tsblock, Filter* filter); void build_filter_column_mask(); int evaluate_page_field_mask(size_t page_idx); int decode_output_columns_for_page(size_t page_idx); @@ -346,6 +361,18 @@ class AlignedChunkReader : public IChunkReader { SelectionMask current_page_field_mask_; bool current_page_field_mask_valid_ = false; + // Direct decode: pages [0, direct_page_end_) are decoded straight into the + // destination Vector memory instead of a per-page staging buffer. See + // prepare_direct_decode(). + size_t direct_page_end_ = 0; + int32_t direct_base_row_ = 0; + common::TsBlock* direct_tsblock_ = nullptr; + // Per value column: byte offset in the Vector's buffer where the next + // decoded value goes. TsBlock columns store non-null values compactly + // (RowIterator advances offsets only for non-null rows), so this advances + // by the non-null count of each page, not by its row index. + std::vector direct_col_off_; + #ifdef ENABLE_THREADS common::ThreadPool* decode_pool_ = nullptr; // borrowed, not owned // Per-task time decoder + compressor pool for parallel time-page decode. From 7da143133845f4efaa3279c0b49b2cf755ee8e44 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Fri, 6 Nov 2026 16:05:18 +0800 Subject: [PATCH 6/7] fix(cpp): release direct-decode offsets and pad SIMD unpack buffers Two problems found by running the suite under AddressSanitizer. AlignedChunkReader is placement-new'd into an arena and torn down with destroy() + mem_free(), so its destructor never runs. prepare_direct_decode() keeps a std::vector of per-column write offsets, and its heap buffer was never released; every reader leaked it (44 LeakSanitizer reports). Free it in destroy() like the other heap-owning members. The AVX2 bit-unpack gathers 8 bytes per value, so the last group of a buffer may read up to 7 bytes past its end. The RLE decoders already zero-pad their input by 8 bytes for that reason, but the four Sprintz decoders allocate exactly bit_width_ bytes and the packer tests pass exact-size buffers, so both read out of bounds (3 AddressSanitizer SEGVs, one of them on the production FloatSprintzDecoder path). Pad those buffers, document the contract on unpack_8values(), and make the multi-block tests pass the packed length rather than the padded vector size. --- cpp/src/encoding/double_sprintz_decoder.h | 4 +++- cpp/src/encoding/float_sprintz_decoder.h | 4 +++- cpp/src/encoding/int32_packer.h | 5 +++++ cpp/src/encoding/int32_sprintz_decoder.h | 4 +++- cpp/src/encoding/int64_packer.h | 5 +++++ cpp/src/encoding/int64_sprintz_decoder.h | 4 +++- cpp/src/reader/aligned_chunk_reader.cc | 2 ++ cpp/test/encoding/int32_packer_test.cc | 26 +++++++++++++--------- cpp/test/encoding/int64_packer_test.cc | 27 ++++++++++++++--------- 9 files changed, 55 insertions(+), 26 deletions(-) diff --git a/cpp/src/encoding/double_sprintz_decoder.h b/cpp/src/encoding/double_sprintz_decoder.h index 57f4b0b23..c189192e1 100644 --- a/cpp/src/encoding/double_sprintz_decoder.h +++ b/cpp/src/encoding/double_sprintz_decoder.h @@ -130,7 +130,9 @@ class DoubleSprintzDecoder : public SprintzDecoder { decode_size_ = block_size_ + 1; common::SerializationUtil::read_double(pre_value_, input); current_buffer_[0] = pre_value_; - std::vector pack_buf(bit_width_); + // +8 zero bytes: the SIMD unpack gathers 8 bytes per value and + // may read up to 7 bytes past the packed group. + std::vector pack_buf(bit_width_ + 8, 0); uint32_t read_len = 0; input.read_buf(reinterpret_cast(pack_buf.data()), bit_width_, read_len); diff --git a/cpp/src/encoding/float_sprintz_decoder.h b/cpp/src/encoding/float_sprintz_decoder.h index e3842eaaf..c97cc1551 100644 --- a/cpp/src/encoding/float_sprintz_decoder.h +++ b/cpp/src/encoding/float_sprintz_decoder.h @@ -133,7 +133,9 @@ class FloatSprintzDecoder : public SprintzDecoder { common::SerializationUtil::read_float(pre_value_, input); current_buffer_[0] = pre_value_; // read packed data - std::vector pack_buf(bit_width_); + // +8 zero bytes: the SIMD unpack gathers 8 bytes per value and + // may read up to 7 bytes past the packed group. + std::vector pack_buf(bit_width_ + 8, 0); uint32_t read_len = 0; input.read_buf(reinterpret_cast(pack_buf.data()), bit_width_, read_len); diff --git a/cpp/src/encoding/int32_packer.h b/cpp/src/encoding/int32_packer.h index 41db56489..3ff0b81b7 100644 --- a/cpp/src/encoding/int32_packer.h +++ b/cpp/src/encoding/int32_packer.h @@ -176,6 +176,11 @@ class Int32Packer { * @param offset - offset of first byte to be decoded in buf * @param values - decoded result , the length of 'values' should be @{link * IntPacker#NUM_OF_INTS} + * + * The SIMD paths gather 8 bytes per value, so the buffer must hold at + * least 8 readable bytes past the packed group; callers allocate that + * padding (the RLE decoders zero it, the Sprintz decoders size their + * block buffer with +8). */ void unpack_8values(const unsigned char buf[], int offset, int32_t values[]) { diff --git a/cpp/src/encoding/int32_sprintz_decoder.h b/cpp/src/encoding/int32_sprintz_decoder.h index a7c92eede..81c5127f0 100644 --- a/cpp/src/encoding/int32_sprintz_decoder.h +++ b/cpp/src/encoding/int32_sprintz_decoder.h @@ -136,7 +136,9 @@ class Int32SprintzDecoder : public SprintzDecoder { pre_value_ = tmp_prev_value; current_buffer_[0] = pre_value_; - std::vector pack_buf(bit_width_); + // +8 zero bytes: the SIMD unpack gathers 8 bytes per value and + // may read up to 7 bytes past the packed group. + std::vector pack_buf(bit_width_ + 8, 0); uint32_t read_len = 0; input.read_buf(reinterpret_cast(pack_buf.data()), bit_width_, read_len); diff --git a/cpp/src/encoding/int64_packer.h b/cpp/src/encoding/int64_packer.h index f260f0d81..95d6eab7b 100644 --- a/cpp/src/encoding/int64_packer.h +++ b/cpp/src/encoding/int64_packer.h @@ -177,6 +177,11 @@ class Int64Packer { * @param offset - offset of first byte to be decoded in buf * @param values - decoded result , the length of 'values' should be @{link * IntPacker#NUM_OF_INTS} + * + * The SIMD paths gather 8 bytes per value, so the buffer must hold at + * least 8 readable bytes past the packed group; callers allocate that + * padding (the RLE decoders zero it, the Sprintz decoders size their + * block buffer with +8). */ void unpack_8values(const unsigned char buf[], int offset, int64_t values[]) { diff --git a/cpp/src/encoding/int64_sprintz_decoder.h b/cpp/src/encoding/int64_sprintz_decoder.h index 7b0827688..245984670 100644 --- a/cpp/src/encoding/int64_sprintz_decoder.h +++ b/cpp/src/encoding/int64_sprintz_decoder.h @@ -135,7 +135,9 @@ class Int64SprintzDecoder : public SprintzDecoder { current_buffer_[0] = pre_value_; // Read packed buffer - std::vector pack_buf(bit_width_); + // +8 zero bytes: the SIMD unpack gathers 8 bytes per value and + // may read up to 7 bytes past the packed group. + std::vector pack_buf(bit_width_ + 8, 0); uint32_t read_len = 0; input.read_buf(reinterpret_cast(pack_buf.data()), bit_width_, read_len); diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index 24e616740..9fcc40240 100644 --- a/cpp/src/reader/aligned_chunk_reader.cc +++ b/cpp/src/reader/aligned_chunk_reader.cc @@ -139,6 +139,7 @@ void AlignedChunkReader::destroy() { // LeakSanitizer doesn't flag the (rather large) ChunkPageInfo buffers. std::vector{}.swap(chunk_pages_); std::vector{}.swap(page_all_times_); + std::vector{}.swap(direct_col_off_); if (time_uncompressed_buf_ != nullptr && time_compressor_ != nullptr) { time_compressor_->after_uncompress(time_uncompressed_buf_); time_uncompressed_buf_ = nullptr; @@ -1711,6 +1712,7 @@ void AlignedChunkReader::release_current_page_state() { direct_page_end_ = 0; direct_tsblock_ = nullptr; direct_base_row_ = 0; + direct_col_off_.clear(); for (auto* col : value_columns_) { if (col->uncompressed_buf != nullptr && col->compressor != nullptr) { col->compressor->after_uncompress(col->uncompressed_buf); diff --git a/cpp/test/encoding/int32_packer_test.cc b/cpp/test/encoding/int32_packer_test.cc index 1b072383f..3f4e151a3 100644 --- a/cpp/test/encoding/int32_packer_test.cc +++ b/cpp/test/encoding/int32_packer_test.cc @@ -31,7 +31,8 @@ TEST(IntPackerTest, SequentialValues) { for (int i = 0; i < 8; ++i) arr[i] = i; Int32Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int32_t res[8] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -50,7 +51,8 @@ TEST(IntPackerStressTest, PackUnpackRandomPositiveValues) { std::vector pre_values; std::vector buffer; pre_values.reserve(total_values); - buffer.resize(count * width); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + buffer.resize(count * width + 8, 0); int idx = 0; std::srand(12345); // Optional: deterministic seed for (int i = 0; i < count; ++i) { @@ -68,8 +70,7 @@ TEST(IntPackerStressTest, PackUnpackRandomPositiveValues) { } std::vector res(total_values); - packer.unpack_all_values(buffer.data(), static_cast(buffer.size()), - res.data()); + packer.unpack_all_values(buffer.data(), count * width, res.data()); std::string diff_msg; for (int i = 0; i < total_values; ++i) { if (res[i] != pre_values[i]) { @@ -87,7 +88,8 @@ TEST(Int32PackerTest, AllZeroValues) { int32_t arr[NUM_OF_INTS] = {0}; Int32Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int32_t res[NUM_OF_INTS] = { 1}; // initialize non-zero to catch failures @@ -104,7 +106,8 @@ TEST(Int32PackerTest, BoundaryWidthOneAlternating) { int32_t arr[NUM_OF_INTS] = {0, 1, 0, 1, 0, 1, 0, 1}; Int32Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int32_t res[NUM_OF_INTS] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -127,7 +130,8 @@ TEST(Int32PackerTest, MaxWidth32Random) { } Int32Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int32_t res[NUM_OF_INTS] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -150,7 +154,8 @@ TEST(Int32PackerTest, AllNegative32Random) { } Int32Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int32_t res[NUM_OF_INTS] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -167,7 +172,7 @@ TEST(Int32PackerTest, UnpackAllValuesMultipleBlocks) { const int blocks = 10; Int32Packer packer(width); std::vector orig(blocks * NUM_OF_INTS); - std::vector buf(blocks * width); + std::vector buf(blocks * width + 8, 0); // Fill orig with pattern: block * 16 + index // Example: block 0 = [0,1,...,7], block 1 = [16,17,...,23], etc. @@ -182,8 +187,7 @@ TEST(Int32PackerTest, UnpackAllValuesMultipleBlocks) { std::vector res(blocks * NUM_OF_INTS, 0); // Unpack all blocks at once - packer.unpack_all_values(buf.data(), static_cast(buf.size()), - res.data()); + packer.unpack_all_values(buf.data(), blocks * width, res.data()); // Verify each unpacked value matches the original sequence for (size_t i = 0; i < orig.size(); ++i) { diff --git a/cpp/test/encoding/int64_packer_test.cc b/cpp/test/encoding/int64_packer_test.cc index 846f9a0c9..2e463f7f3 100644 --- a/cpp/test/encoding/int64_packer_test.cc +++ b/cpp/test/encoding/int64_packer_test.cc @@ -32,7 +32,8 @@ TEST(Int64PackerTest, SequentialValues) { for (int i = 0; i < 8; ++i) arr[i] = i; Int64Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int64_t res[8] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -49,7 +50,7 @@ TEST(Int64PackerTest, PackUnpackSingleBatchRandomPositiveLongs) { Int64Packer packer(byte_count); std::vector pre_values; - std::vector buffer(count * byte_count); + std::vector buffer(count * byte_count + 8, 0); pre_values.reserve(total_values); int idx = 0; @@ -72,8 +73,7 @@ TEST(Int64PackerTest, PackUnpackSingleBatchRandomPositiveLongs) { } std::vector result(total_values); - packer.unpack_all_values(buffer.data(), static_cast(buffer.size()), - result.data()); + packer.unpack_all_values(buffer.data(), count * byte_count, result.data()); for (int i = 0; i < total_values; ++i) { ASSERT_EQ(result[i], pre_values[i]) << "Mismatch at index " << i; @@ -124,6 +124,8 @@ TEST(Int64PackerTest, PackAllManualBitWidth) { std::string packed_data = oss.str(); ASSERT_EQ(static_cast(packed_data.size()), 2 * bp_bit_width); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + packed_data.append(8, '\0'); // Decode int64_t read_array[16] = {0}; @@ -143,7 +145,8 @@ TEST(Int64PackerTest, AllZeroValues) { int64_t arr[NUM_OF_INTS] = {0}; Int64Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int64_t res[NUM_OF_INTS] = { 1}; // initialize non-zero to catch failures @@ -160,7 +163,8 @@ TEST(Int64PackerTest, BoundaryWidthOneAlternating) { int64_t arr[NUM_OF_INTS] = {0, 1, 0, 1, 0, 1, 0, 1}; Int64Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int64_t res[NUM_OF_INTS] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -183,7 +187,8 @@ TEST(Int64PackerTest, MaxWidth64Random) { } Int64Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int64_t res[NUM_OF_INTS] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -206,7 +211,8 @@ TEST(Int64PackerTest, AllNegative64Random) { } Int64Packer packer(width); const int bufSize = NUM_OF_INTS * width / 8; - std::vector buf(bufSize, 0); + // +8: the SIMD unpack may read up to 7 bytes past the last group. + std::vector buf(bufSize + 8, 0); packer.pack_8values(arr, 0, buf.data()); int64_t res[NUM_OF_INTS] = {0}; packer.unpack_8values(buf.data(), 0, res); @@ -223,7 +229,7 @@ TEST(Int64PackerTest, UnpackAllValuesMultipleBlocks) { const int blocks = 10; Int64Packer packer(width); std::vector orig(blocks * NUM_OF_INTS); - std::vector buf(blocks * width); + std::vector buf(blocks * width + 8, 0); // Fill orig with pattern: block * 16 + index // Example: block 0 = [0,1,...,7], block 1 = [16,17,...,23], etc. @@ -238,8 +244,7 @@ TEST(Int64PackerTest, UnpackAllValuesMultipleBlocks) { std::vector res(blocks * NUM_OF_INTS, 0); // Unpack all blocks at once - packer.unpack_all_values(buf.data(), static_cast(buf.size()), - res.data()); + packer.unpack_all_values(buf.data(), blocks * width, res.data()); // Verify each unpacked value matches the original sequence for (size_t i = 0; i < orig.size(); ++i) { From a9268ae712c0e2c1097275c3b9f92412fde31e02 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Fri, 6 Nov 2026 16:19:03 +0800 Subject: [PATCH 7/7] fix(cpp): fall back to the portable unpack loop without __int128 MSVC has no 128-bit integer, so the wide-buffer scalar path used for the RLE bit widths that can cross a ninth byte (59/61/62/63) failed to compile ("error C4235: nonstandard extension used: '__int128'"). Keep the 128-bit path for compilers that define __SIZEOF_INT128__ and restore the portable bit-by-bit loop otherwise. --- cpp/src/encoding/int64_packer.h | 38 ++++++++++++++++++++++++++++++--- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/cpp/src/encoding/int64_packer.h b/cpp/src/encoding/int64_packer.h index 95d6eab7b..d9c0ef589 100644 --- a/cpp/src/encoding/int64_packer.h +++ b/cpp/src/encoding/int64_packer.h @@ -202,10 +202,11 @@ class Int64Packer { } #endif // Scalar fallback for widths that can cross a ninth byte (59/61/62/63). +#if defined(__SIZEOF_INT128__) // Decode whole bytes into a 128-bit buffer — ~8x fewer iterations than - // the old bit-by-bit loop — then shift out one width-wide value at a - // time. Widths up to 63 need up to 63 + 7 = 70 buffered bits, which the - // 64-bit int32-style buffer cannot hold; hence 128 bits. + // the bit-by-bit loop — then shift out one width-wide value at a time. + // Widths up to 63 need up to 63 + 7 = 70 buffered bits, which a 64-bit + // buffer cannot hold; hence 128 bits. int byte_idx = offset; unsigned __int128 buffer = 0; int total_bits = 0; @@ -224,6 +225,37 @@ class Int64Packer { buffer &= (((unsigned __int128)1 << total_bits) - 1); } } +#else + // MSVC has no 128-bit integer; fall back to the portable bit-by-bit + // loop used before the wide-buffer variant. + int byte_idx = offset; + int value_idx = 0; + int left_bits = 8; + int total_bits = 0; + while (value_idx < 8) { + values[value_idx] = 0; + total_bits = 0; + while (total_bits < width_) { + if (width_ - total_bits >= left_bits) { + values[value_idx] <<= left_bits; + values[value_idx] |= static_cast( + buf[byte_idx] & ((1 << left_bits) - 1)); + total_bits += left_bits; + byte_idx++; + left_bits = 8; + } else { + int t = width_ - total_bits; + values[value_idx] <<= t; + values[value_idx] |= static_cast( + (buf[byte_idx] & ((1 << left_bits) - 1)) >> + (left_bits - t)); + left_bits -= t; + total_bits += t; + } + } + value_idx++; + } +#endif } /**