Optimize C++ batch decode and output fast paths

cpp/src/common/tsblock/tsblock.h

@@ -174,6 +174,28 @@ class RowAppender {
         }
     }
 
+    FORCE_INLINE bool can_bulk_append_fixed(uint32_t slot_index,
+                                            uint32_t elem_size) const {
+        ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count());
+        Vector* vec = tsblock_->vectors_[slot_index];
+        TSDataType datatype = vec->get_vector_type();
+        if (datatype == STRING || datatype == TEXT || datatype == BLOB) {
+            return false;
+        }
+        return static_cast<FixedLengthVector*>(vec)->get_type_len() ==
+               elem_size;
+    }
+
+    FORCE_INLINE void bulk_append_fixed(uint32_t slot_index, const char* values,
+                                        uint32_t count) {
+        ASSERT(slot_index < tsblock_->tuple_desc_->get_column_count());
+        Vector* vec = tsblock_->vectors_[slot_index];
+        ASSERT(vec->get_vector_type() != STRING &&
+               vec->get_vector_type() != TEXT &&
+               vec->get_vector_type() != BLOB);
+        static_cast<FixedLengthVector*>(vec)->append_batch(values, count);
+    }
+
     FORCE_INLINE void append_null(uint32_t slot_index) {
         Vector* vec = tsblock_->vectors_[slot_index];
         vec->set_null(tsblock_->row_count_ - 1);

cpp/src/common/tsblock/vector/fixed_length_vector.h

@@ -49,6 +49,13 @@ class FixedLengthVector : public Vector {
         values_.append_fixed_value(value, len);
     }
 
+    FORCE_INLINE void append_batch(const char* values, uint32_t count) {
+        values_.append_fixed_value(values, count * type_len_);
+        add_row_nums(count);
+    }
+
+    FORCE_INLINE uint32_t get_type_len() const { return type_len_; }
+
     // cppcheck-suppress missingOverride
     FORCE_INLINE char* read(uint32_t* __restrict len, bool* __restrict null,
                             uint32_t rowid) OVERRIDE {

cpp/src/encoding/gorilla_decoder.h

@@ -19,8 +19,13 @@
 #ifndef ENCODING_GORILLA_DECODER_H
 #define ENCODING_GORILLA_DECODER_H
 
+#include <algorithm>
 #include <climits>
 
+#if defined(_MSC_VER)
+#include <intrin.h>
+#endif
+
 #include "common/allocator/byte_stream.h"
 #include "decoder.h"
 #include "encode_utils.h"
@@ -30,6 +35,23 @@
 
 namespace storage {
 
+FORCE_INLINE int gorilla_count_leading_zeros_nonzero(uint64_t value) {
+#if defined(__GNUC__) || defined(__clang__)
+    return __builtin_clzll(value);
+#elif defined(_MSC_VER)
+    unsigned long index;
+    _BitScanReverse64(&index, value);
+    return 63 - static_cast<int>(index);
+#else
+    int count = 0;
+    while ((value & (UINT64_C(1) << 63)) == 0) {
+        value <<= 1;
+        ++count;
+    }
+    return count;
+#endif
+}
+
 // ── Raw-pointer bit reader ────────────────────────────────────────────────
 // Operates directly on a contiguous byte array, bypassing ByteStream's
 // per-byte read_buf() overhead (atomic loads, page boundary checks, memcpy).
@@ -91,6 +113,37 @@ struct GorillaBitReader {
         return bit;
     }
 
+    // Consume up to max_count consecutive zero control bits. The first one bit
+    // remains unread so the normal control decoder can handle the following
+    // changed value. This turns long repeated-value runs into one leading-zero
+    // count per reservoir instead of one read_next() call per value.
+    FORCE_INLINE int consume_zero_bits(int max_count) {
+        int consumed = 0;
+        while (consumed < max_count) {
+            if (UNLIKELY(!refill_if_empty())) {
+                break;
+            }
+
+            const int available_bits = bits;
+            const uint64_t aligned =
+                available_bits == 64 ? buffer : buffer << (64 - available_bits);
+            const int zero_bits =
+                aligned == 0 ? available_bits
+                             : gorilla_count_leading_zeros_nonzero(aligned);
+            const int remaining = max_count - consumed;
+            const int take = std::min(zero_bits, remaining);
+            bits -= take;
+            consumed += take;
+
+            // A one bit follows the consumed zeros. Leave it in the reservoir
+            // for read_control_bits(), or stop once the requested limit is met.
+            if (zero_bits < available_bits || consumed == max_count) {
+                break;
+            }
+        }
+        return consumed;
+    }
+
     FORCE_INLINE uint64_t read_long(int n) {
         if (UNLIKELY(n < 0 || n > 64)) {
             invalid = true;
@@ -460,8 +513,21 @@ class GorillaDecoder : public Decoder {
 
         // Main batch loop
         while (actual < capacity && has_next_) {
-            out[actual++] =
+            const Output decoded =
                 GorillaDecodeOutput<T, Output>::convert(stored_value_);
+            const int repeated = r.consume_zero_bits(capacity - actual - 1);
+            const int run_length = repeated + 1;
+            // Include the current value in the bulk fill. Besides removing a
+            // scalar store and a second position update for every run, this
+            // gives the compiler one contiguous range to vectorize. Integer
+            // and floating-point outputs are copied without arithmetic, so
+            // special IEEE-754 bit patterns remain unchanged.
+            std::fill_n(out + actual, run_length, decoded);
+            actual += run_length;
+
+            // Prime the next value even when the repeated run exactly fills the
+            // caller's buffer. This preserves the scalar decoder invariant that
+            // stored_value_ is the next value to return on the following call.
             if (UNLIKELY(!GorillaRawOps<T>::read_next(
                     r, stored_value_, stored_leading_zeros_,
                     stored_trailing_zeros_))) {

cpp/src/encoding/ts2diff_decoder.h

@@ -648,6 +648,43 @@ inline int TS2DIFFDecoder<int64_t>::read_batch_int64(int64_t* out, int capacity,
         int64_t prev = first_value_;
         int32_t i = 0;
 
+        // An evenly spaced timestamp block has no packed residual data. Build
+        // the arithmetic progression directly instead of entering the generic
+        // bit-extraction path (whose SIMD guard requires readable input bytes).
+        if (bit_width_ == 0) {
+#ifdef ENABLE_SIMD
+            if (remaining >= 4) {
+                int64_t value1 = prev + delta_min_;
+                int64_t value2 = value1 + delta_min_;
+                int64_t value3 = value2 + delta_min_;
+                int64_t value4 = value3 + delta_min_;
+                simde__m256i values =
+                    simde_mm256_set_epi64x(value4, value3, value2, value1);
+
+                simde__m256i step = simde_mm256_set1_epi64x(delta_min_);
+                step = simde_mm256_add_epi64(step, step);
+                step = simde_mm256_add_epi64(step, step);
+
+                for (; i + 3 < remaining; i += 4) {
+                    simde_mm256_storeu_si256(
+                        reinterpret_cast<simde__m256i*>(out + actual), values);
+                    actual += 4;
+                    values = simde_mm256_add_epi64(values, step);
+                }
+                prev = out[actual - 1];
+            }
+#endif
+
+            for (; i < remaining; ++i) {
+                prev += delta_min_;
+                out[actual++] = prev;
+            }
+
+            first_value_ = prev;
+            current_index_ = 0;
+            continue;
+        }
+
 #ifdef ENABLE_SIMD
         // SIMD path: decode 4 INT64 values at a time
         for (; i + 3 < remaining; i += 4) {

cpp/src/reader/aligned_chunk_reader.cc

@@ -922,6 +922,30 @@ int AlignedChunkReader::decode_tv_batch(ByteStream& time_in,
             }
         }
 
+        // Dense fixed-width batches are already laid out exactly as the two
+        // destination vectors expect. Appending them row by row would issue
+        // two tiny memcpy calls per row (time + value), plus virtual dispatch
+        // and bookkeeping. Copy each column once instead. Integral value
+        // filters still need the scalar satisfy(time, value) check unless the
+        // decoder proved the whole block passes, so those batches retain the
+        // fallback below.
+        const bool needs_integral_value_filter =
+            std::is_integral<T>::value && filter != nullptr && !block_all_pass;
+        if (pass_count == time_count && nonnull_count == time_count &&
+            !needs_integral_value_filter &&
+            row_appender.can_bulk_append_fixed(0, sizeof(int64_t)) &&
+            row_appender.can_bulk_append_fixed(1, sizeof(T))) {
+            row_appender.bulk_append_fixed(0,
+                                           reinterpret_cast<const char*>(times),
+                                           static_cast<uint32_t>(time_count));
+            row_appender.bulk_append_fixed(
+                1, reinterpret_cast<const char*>(values),
+                static_cast<uint32_t>(time_count));
+            row_appender.add_rows(static_cast<uint32_t>(time_count));
+            cur_value_index += time_count;
+            continue;
+        }
+
         int val_idx = 0;
         for (int i = 0; i < time_count; ++i) {
             cur_value_index++;

cpp/test/common/tsblock/tslock_test.cc

@@ -63,6 +63,43 @@ TEST(TsBlockTest, ColAppender_AddRowAndAppend) {
     EXPECT_EQ(col_appender.get_col_row_count(), 50);
 }
 
+TEST(TsBlockTest, RowAppenderBulkAppendFixedAtExactCapacity) {
+    TupleDesc tuple_desc;
+    tuple_desc.push_back(ColumnSchema("time", TIMESTAMP, UNCOMPRESSED, PLAIN));
+    tuple_desc.push_back(ColumnSchema("value", FLOAT, UNCOMPRESSED, PLAIN));
+    TsBlock ts_block(&tuple_desc, 4);
+    ASSERT_EQ(ts_block.init(), E_OK);
+    RowAppender row_appender(&ts_block);
+
+    const int64_t times[] = {101, 103, 107, 109};
+    const float values[] = {1.25f, 2.5f, 3.75f, 5.0f};
+    ASSERT_TRUE(row_appender.can_bulk_append_fixed(0, sizeof(int64_t)));
+    ASSERT_TRUE(row_appender.can_bulk_append_fixed(1, sizeof(float)));
+    EXPECT_FALSE(row_appender.can_bulk_append_fixed(1, sizeof(double)));
+
+    row_appender.bulk_append_fixed(0, reinterpret_cast<const char*>(times), 4);
+    row_appender.bulk_append_fixed(1, reinterpret_cast<const char*>(values), 4);
+    row_appender.add_rows(4);
+
+    EXPECT_EQ(ts_block.get_row_count(), 4u);
+    EXPECT_EQ(row_appender.remaining(), 0u);
+    EXPECT_EQ(ts_block.get_vector(0)->get_row_num(), 4u);
+    EXPECT_EQ(ts_block.get_vector(1)->get_row_num(), 4u);
+
+    ColIterator time_iter(0, &ts_block);
+    ColIterator value_iter(1, &ts_block);
+    for (uint32_t i = 0; i < 4; ++i) {
+        uint32_t len = 0;
+        EXPECT_EQ(*reinterpret_cast<int64_t*>(time_iter.read(&len)), times[i]);
+        EXPECT_EQ(len, sizeof(int64_t));
+        EXPECT_FLOAT_EQ(*reinterpret_cast<float*>(value_iter.read(&len)),
+                        values[i]);
+        EXPECT_EQ(len, sizeof(float));
+        time_iter.next();
+        value_iter.next();
+    }
+}
+
 TEST(TsBlockTest, RowIterator_ReadAndNext) {
     TupleDesc tuple_desc;
     ColumnSchema col1("test_col1", INT32, SNAPPY, RLE);

cpp/test/encoding/encoding_coverage_test.cc

@@ -330,6 +330,41 @@ TEST(EncodingCoverage, TS2DIFFBatchInt64MultipleBlocks) {
     for (int i = 0; i < N; i++) EXPECT_EQ(out[i], values[i]) << "i=" << i;
 }
 
+TEST(EncodingCoverage, TS2DIFFBatchInt64EvenlySpacedNegativeDelta) {
+    TS2DIFFEncoder<int64_t> enc;
+    common::ByteStream s(8192, common::MOD_DEFAULT);
+    // Full encoder blocks contain 127 residuals, so this covers repeated SIMD
+    // groups, the 3-value scalar tail, and a final partial block.
+    const int N = 389;
+    std::vector<int64_t> values(N);
+    for (int i = 0; i < N; i++) {
+        values[i] = INT64_C(9000000000000) - static_cast<int64_t>(i) * 29;
+        ASSERT_EQ(enc.encode(values[i], s), common::E_OK);
+    }
+    ASSERT_EQ(enc.flush(s), common::E_OK);
+
+    uint32_t total = s.total_size();
+    std::vector<uint8_t> buf(total);
+    uint32_t got = 0;
+    s.read_buf(buf.data(), total, got);
+    common::ByteStream wrapped(common::MOD_DEFAULT);
+    wrapped.wrap_from((const char*)buf.data(), total);
+
+    TS2DIFFDecoder<int64_t> dec;
+    std::vector<int64_t> out(N);
+    int total_decoded = 0;
+    while (dec.has_remaining(wrapped) && total_decoded < N) {
+        int actual = 0;
+        ASSERT_EQ(dec.read_batch_int64(out.data() + total_decoded,
+                                       N - total_decoded, actual, wrapped),
+                  common::E_OK);
+        if (actual == 0) break;
+        total_decoded += actual;
+    }
+    EXPECT_EQ(total_decoded, N);
+    for (int i = 0; i < N; i++) EXPECT_EQ(out[i], values[i]) << "i=" << i;
+}
+
 // ── Plain encoder: encode_batch fast paths for each type ───────────────
 TEST(EncodingCoverage, PlainEncoderBatchAllTypes) {
     PlainEncoder enc;