perf(cpp): speed up wide-table reads (RLE SIMD, batch index loading, BOUNDARY bulk)

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_;

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

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<uint8_t> 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<uint8_t> pack_buf(bit_width_ + 8, 0);
             uint32_t read_len = 0;
             input.read_buf(reinterpret_cast<char*>(pack_buf.data()), bit_width_,
                            read_len);

cpp/src/encoding/float_rle_decoder.h

@@ -20,6 +20,7 @@
 #ifndef ENCODING_FLOAT_RLE_DECODER_H
 #define ENCODING_FLOAT_RLE_DECODER_H
 
+#include <algorithm>
 #include <cmath>
 #include <cstdint>
 #include <vector>
@@ -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;

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<uint8_t> 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<uint8_t> pack_buf(bit_width_ + 8, 0);
             uint32_t read_len = 0;
             input.read_buf(reinterpret_cast<char*>(pack_buf.data()), bit_width_,
                            read_len);

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 <immintrin.h>
+#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.,