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/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/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_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/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 ecb42b248..3ff0b81b7 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 */ @@ -98,9 +176,28 @@ 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[]) { +#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..09e631878 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 @@ -231,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_); @@ -248,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/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 23267206f..d9c0ef589 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 */ @@ -99,18 +177,64 @@ 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[]) { +#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). +#if defined(__SIZEOF_INT128__) + // Decode whole bytes into a 128-bit buffer — ~8x fewer iterations than + // 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; + int value_idx = 0; + while (value_idx < 8) { + while (total_bits < width_) { + 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); + } + } +#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; @@ -129,9 +253,9 @@ class Int64Packer { total_bits += t; } } - value_idx++; } +#endif } /** diff --git a/cpp/src/encoding/int64_rle_decoder.h b/cpp/src/encoding/int64_rle_decoder.h index bef72bbed..0d8fa94f4 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 @@ -230,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_); @@ -247,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/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/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 // ============================================================================ 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; diff --git a/cpp/src/reader/aligned_chunk_reader.cc b/cpp/src/reader/aligned_chunk_reader.cc index a0a0b2306..9fcc40240 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(); @@ -138,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; @@ -202,6 +204,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 +1690,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 +1709,10 @@ 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; + direct_col_off_.clear(); for (auto* col : value_columns_) { if (col->uncompressed_buf != nullptr && col->compressor != nullptr) { col->compressor->after_uncompress(col->uncompressed_buf); @@ -1711,6 +1723,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 +1747,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 +1862,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 +1895,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 +1904,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 +1930,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 +2186,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 +2259,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; } @@ -2194,12 +2321,25 @@ 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. + // 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) && remaining_in_page > 0 && budget > 0; if (can_bulk) { for (uint32_t c = 0; c < num_cols; c++) { @@ -2209,7 +2349,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 != page_time_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; } @@ -2238,10 +2385,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); @@ -2324,12 +2481,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. 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/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_; 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 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) {