This is an automated email from the ASF dual-hosted git repository. ColinLeeo pushed a commit to branch perf/gorilla-wide-bit-reader in repository https://gitbox.apache.org/repos/asf/tsfile.git
commit 98269f30dec2dc6afb315d951af611f7974b9274 Author: ColinLee <[email protected]> AuthorDate: Thu Jul 23 11:23:43 2026 +0800 Optimize Gorilla batch floating-point decoding --- cpp/src/encoding/gorilla_decoder.h | 388 ++++++++++++++++++-------------- cpp/test/encoding/gorilla_codec_test.cc | 163 ++++++++++++++ 2 files changed, 382 insertions(+), 169 deletions(-) diff --git a/cpp/src/encoding/gorilla_decoder.h b/cpp/src/encoding/gorilla_decoder.h index e1e490105..c52a4d59a 100644 --- a/cpp/src/encoding/gorilla_decoder.h +++ b/cpp/src/encoding/gorilla_decoder.h @@ -33,71 +33,106 @@ namespace storage { // ── 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). +// The 64-bit reservoir amortizes bounds checks and refill work across up to +// eight encoded bytes. Valid bits are kept right-aligned in buffer; bits is +// the number of unread low bits. struct GorillaBitReader { const uint8_t* data; uint32_t pos; // next byte index to load uint32_t data_len; // total bytes - int bits; // remaining bits in cur_byte (0..8) - uint8_t cur_byte; - // Set once a load was attempted on an empty input, or once read_bit / - // read_long ran out of bits mid-value. Without this, a truncated page - // would spin read_long() forever (bits stays 0, n -= 0 makes no - // progress) and read_bit() would execute a negative shift via - // (cur_byte >> (bits - 1)). + int bits; // remaining bits in buffer (0..64) + uint64_t buffer; + // Set once a read cannot be satisfied from the encoded input. bool exhausted = false; + bool invalid = false; - FORCE_INLINE void load_byte_if_empty() { - if (bits == 0) { - if (pos < data_len) { - cur_byte = data[pos++]; - bits = 8; - } else { - exhausted = true; - } + FORCE_INLINE bool refill_if_empty() { + if (bits != 0) { + return true; + } + if (UNLIKELY(pos >= data_len)) { + exhausted = true; + return false; + } + + uint32_t available = data_len - pos; + if (LIKELY(available >= sizeof(uint64_t))) { + // Explicit byte assembly is alignment-safe and portable; optimizing + // compilers recognize it as one load plus a byte swap on + // little-endian targets. + const uint8_t* src = data + pos; + buffer = (static_cast<uint64_t>(src[0]) << 56) | + (static_cast<uint64_t>(src[1]) << 48) | + (static_cast<uint64_t>(src[2]) << 40) | + (static_cast<uint64_t>(src[3]) << 32) | + (static_cast<uint64_t>(src[4]) << 24) | + (static_cast<uint64_t>(src[5]) << 16) | + (static_cast<uint64_t>(src[6]) << 8) | + static_cast<uint64_t>(src[7]); + pos += sizeof(uint64_t); + bits = 64; + } else { + buffer = 0; + do { + buffer = (buffer << 8) | data[pos++]; + bits += 8; + } while (pos < data_len); } + return true; } FORCE_INLINE bool read_bit() { - if (UNLIKELY(bits == 0)) { - exhausted = true; + if (UNLIKELY(!refill_if_empty())) { return false; } - bool bit = ((cur_byte >> (bits - 1)) & 1) == 1; + bool bit = ((buffer >> (bits - 1)) & 1) != 0; bits--; - load_byte_if_empty(); return bit; } - FORCE_INLINE int64_t read_long(int n) { - int64_t value = 0; - while (n > 0) { - if (UNLIKELY(bits == 0)) { - // Input drained mid-value; bail so the outer loop in - // read_control_bits / batch_decode_raw doesn't spin. - exhausted = true; - return value; - } - if (n > bits || n == 8) { - value = (value << bits) + (cur_byte & ((1 << bits) - 1)); - n -= bits; - bits = 0; - } else { - value = - (value << n) + ((cur_byte >> (bits - n)) & ((1 << n) - 1)); - bits -= n; - n = 0; + FORCE_INLINE uint64_t read_long(int n) { + if (UNLIKELY(n < 0 || n > 64)) { + invalid = true; + return 0; + } + + if (n == 0) { + return 0; + } + if (UNLIKELY(!refill_if_empty())) { + return 0; + } + + if (LIKELY(n <= bits)) { + bits -= n; + if (n == 64) { + return buffer; } - load_byte_if_empty(); + return (buffer >> bits) & ((uint64_t{1} << n) - 1); } - return value; + + // A request is at most 64 bits, so after consuming the current + // reservoir it can cross into at most one full 64-bit refill. + int first_bits = bits; + uint64_t value = buffer & ((uint64_t{1} << first_bits) - 1); + int remaining = n - first_bits; + bits = 0; + if (UNLIKELY(!refill_if_empty() || bits < remaining)) { + exhausted = true; + return 0; + } + + bits -= remaining; + uint64_t tail = (buffer >> bits) & ((uint64_t{1} << remaining) - 1); + return (value << remaining) | tail; } FORCE_INLINE uint8_t read_control_bits(int max_bits) { uint8_t value = 0x00; for (int i = 0; i < max_bits; i++) { value <<= 1; - if (exhausted) break; + if (UNLIKELY(exhausted || invalid)) break; if (read_bit()) { value |= 0x01; } else { @@ -112,20 +147,21 @@ struct GorillaBitReader { template <typename T> struct GorillaRawOps { - static FORCE_INLINE T read_next(GorillaBitReader& r, T& stored_value, - int& stored_leading_zeros, - int& stored_trailing_zeros); + static FORCE_INLINE bool read_next(GorillaBitReader& r, T& stored_value, + int& stored_leading_zeros, + int& stored_trailing_zeros); }; template <> struct GorillaRawOps<int32_t> { static constexpr int VALUE_BITS = VALUE_BITS_LENGTH_32BIT; - static FORCE_INLINE int32_t read_next(GorillaBitReader& r, - int32_t& stored_value, - int& stored_leading_zeros, - int& stored_trailing_zeros) { + static FORCE_INLINE bool read_next(GorillaBitReader& r, + int32_t& stored_value, + int& stored_leading_zeros, + int& stored_trailing_zeros) { uint8_t ctrl = r.read_control_bits(2); + if (UNLIKELY(r.exhausted || r.invalid)) return false; switch (ctrl) { case 3: { stored_leading_zeros = @@ -133,21 +169,32 @@ struct GorillaRawOps<int32_t> { uint8_t sig = (uint8_t)r.read_long(MEANINGFUL_XOR_BITS_LENGTH_32BIT); sig++; + if (UNLIKELY(r.exhausted || + stored_leading_zeros + sig > VALUE_BITS)) { + r.invalid = !r.exhausted; + return false; + } stored_trailing_zeros = VALUE_BITS - sig - stored_leading_zeros; } // fallthrough case 2: { - int32_t xor_value = (int32_t)r.read_long( - VALUE_BITS - stored_leading_zeros - stored_trailing_zeros); - xor_value = static_cast<uint32_t>(xor_value) - << stored_trailing_zeros; - stored_value ^= xor_value; + int meaningful = + VALUE_BITS - stored_leading_zeros - stored_trailing_zeros; + if (UNLIKELY(meaningful <= 0 || meaningful > VALUE_BITS)) { + r.invalid = true; + return false; + } + uint32_t xor_value = + static_cast<uint32_t>(r.read_long(meaningful)); + if (UNLIKELY(r.exhausted || r.invalid)) return false; + xor_value <<= stored_trailing_zeros; + stored_value ^= static_cast<int32_t>(xor_value); } // fallthrough default: - return stored_value; + return true; } - return stored_value; + return true; } }; @@ -155,11 +202,12 @@ template <> struct GorillaRawOps<int64_t> { static constexpr int VALUE_BITS = VALUE_BITS_LENGTH_64BIT; - static FORCE_INLINE int64_t read_next(GorillaBitReader& r, - int64_t& stored_value, - int& stored_leading_zeros, - int& stored_trailing_zeros) { + static FORCE_INLINE bool read_next(GorillaBitReader& r, + int64_t& stored_value, + int& stored_leading_zeros, + int& stored_trailing_zeros) { uint8_t ctrl = r.read_control_bits(2); + if (UNLIKELY(r.exhausted || r.invalid)) return false; switch (ctrl) { case 3: { stored_leading_zeros = @@ -167,21 +215,53 @@ struct GorillaRawOps<int64_t> { uint8_t sig = (uint8_t)r.read_long(MEANINGFUL_XOR_BITS_LENGTH_64BIT); sig++; + if (UNLIKELY(r.exhausted || + stored_leading_zeros + sig > VALUE_BITS)) { + r.invalid = !r.exhausted; + return false; + } stored_trailing_zeros = VALUE_BITS - sig - stored_leading_zeros; } // fallthrough case 2: { - int64_t xor_value = r.read_long( - VALUE_BITS - stored_leading_zeros - stored_trailing_zeros); - xor_value = static_cast<uint64_t>(xor_value) - << stored_trailing_zeros; - stored_value ^= xor_value; + int meaningful = + VALUE_BITS - stored_leading_zeros - stored_trailing_zeros; + if (UNLIKELY(meaningful <= 0 || meaningful > VALUE_BITS)) { + r.invalid = true; + return false; + } + uint64_t xor_value = r.read_long(meaningful); + if (UNLIKELY(r.exhausted || r.invalid)) return false; + xor_value <<= stored_trailing_zeros; + stored_value ^= static_cast<int64_t>(xor_value); } // fallthrough default: - return stored_value; + return true; } - return stored_value; + return true; + } +}; + +template <typename Stored, typename Output> +struct GorillaDecodeOutput; + +template <typename T> +struct GorillaDecodeOutput<T, T> { + static FORCE_INLINE T convert(T value) { return value; } +}; + +template <> +struct GorillaDecodeOutput<int32_t, float> { + static FORCE_INLINE float convert(int32_t value) { + return common::int_to_float(value); + } +}; + +template <> +struct GorillaDecodeOutput<int64_t, double> { + static FORCE_INLINE double convert(int64_t value) { + return common::long_to_double(value); } }; @@ -210,20 +290,24 @@ class GorillaDecoder : public Decoder { return buffer.has_remaining() || has_next(); } - // If empty, cache 8 bits from in_stream to 'buffer_'. + // If empty, cache 8 bits from in_stream to 'buffer_'. The batch path may + // leave more than 8 prefetched bits here; scalar reads consume those first. void flush_byte_if_empty(common::ByteStream& in) { if (bits_left_ == 0) { + uint8_t next_byte = 0; uint32_t read_len = 0; - in.read_buf(&buffer_, 1, read_len); - bits_left_ = 8; + in.read_buf(&next_byte, 1, read_len); + buffer_ = next_byte; + bits_left_ = static_cast<int>(read_len) * 8; } } // Reads the next bit and returns true if the next bit is 1, otherwise 0. bool read_bit(common::ByteStream& in) { + flush_byte_if_empty(in); + if (UNLIKELY(bits_left_ == 0)) return false; bool bit = ((buffer_ >> (bits_left_ - 1)) & 1) == 1; bits_left_--; - flush_byte_if_empty(in); return bit; } @@ -233,26 +317,24 @@ class GorillaDecoder : public Decoder { * @bits: How many next bits are reader from the stream * return: long value that was reader from the stream */ - int64_t read_long(int bits, common::ByteStream& in) { - int64_t value = 0; + uint64_t read_long(int bits, common::ByteStream& in) { + uint64_t value = 0; while (bits > 0) { - if (bits > bits_left_ || bits == 8) { - // Take only the bits_left_ "least significant" bits. - uint8_t d = (uint8_t)(buffer_ & ((1 << bits_left_) - 1)); - value = (value << bits_left_) + (d & 0xFF); - bits -= bits_left_; - bits_left_ = 0; + flush_byte_if_empty(in); + if (UNLIKELY(bits_left_ == 0)) return value; + + int take = bits < bits_left_ ? bits : bits_left_; + uint64_t chunk; + if (take == 64) { + chunk = buffer_; } else { - // Shift to correct position and take only least significant - // bits. - uint8_t d = - (uint8_t)((((uint8_t)buffer_) >> (bits_left_ - bits)) & - ((1 << bits) - 1)); - value = (value << bits) + (d & 0xFF); - bits_left_ -= bits; - bits = 0; + chunk = (buffer_ >> (bits_left_ - take)) & + ((uint64_t{1} << take) - 1); + value <<= take; } - flush_byte_if_empty(in); + value |= chunk; + bits_left_ -= take; + bits -= take; } return value; } @@ -301,7 +383,8 @@ class GorillaDecoder : public Decoder { // // batch_decode_raw replicates this logic using GorillaBitReader on the // wrapped contiguous buffer, then syncs state back to ByteStream. - int batch_decode_raw(T* out, int capacity, int& actual, T ending, + template <typename Output> + int batch_decode_raw(Output* out, int capacity, int& actual, T ending, common::ByteStream& in) { int ret = common::E_OK; actual = 0; @@ -327,57 +410,55 @@ class GorillaDecoder : public Decoder { r.pos = 0; r.data_len = remain; r.bits = bits_left_; - r.cur_byte = buffer_; + r.buffer = buffer_; // Bootstrap first value if needed (mirrors decode()'s first-call path) if (UNLIKELY(!first_value_was_read_)) { if (r.bits == 0 && r.pos >= r.data_len) goto done; - r.load_byte_if_empty(); stored_value_ = (T)r.read_long(GorillaRawOps<T>::VALUE_BITS); - if (UNLIKELY(r.exhausted)) { + if (UNLIKELY(r.exhausted || r.invalid)) { // Page truncated before the first value finished; refuse to // emit a partially-decoded sentinel. first_value_was_read_ = false; - ret = common::E_BUF_NOT_ENOUGH; + ret = r.invalid ? common::E_TSFILE_CORRUPTED + : common::E_BUF_NOT_ENOUGH; goto done; } first_value_was_read_ = true; - // Save the first value before cache_next mutates stored_value_ + // Save the first value before cache_next mutates stored_value_. T first_value = stored_value_; // cache_next: read_next then check ending - GorillaRawOps<T>::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + if (UNLIKELY(!GorillaRawOps<T>::read_next( + r, stored_value_, stored_leading_zeros_, + stored_trailing_zeros_))) { + ret = r.invalid ? common::E_TSFILE_CORRUPTED + : common::E_BUF_NOT_ENOUGH; goto done; } - if (stored_value_ == ending) { - has_next_ = false; - } else { - has_next_ = true; - } + has_next_ = stored_value_ != ending; // Output the first value - out[actual++] = first_value; + out[actual++] = + GorillaDecodeOutput<T, Output>::convert(first_value); if (!has_next_ || actual >= capacity) goto done; } // Main batch loop while (actual < capacity && has_next_) { - out[actual++] = stored_value_; - GorillaRawOps<T>::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + out[actual++] = + GorillaDecodeOutput<T, Output>::convert(stored_value_); + if (UNLIKELY(!GorillaRawOps<T>::read_next( + r, stored_value_, stored_leading_zeros_, + stored_trailing_zeros_))) { + ret = r.invalid ? common::E_TSFILE_CORRUPTED + : common::E_BUF_NOT_ENOUGH; goto done; } - if (stored_value_ == ending) { - has_next_ = false; - } + has_next_ = stored_value_ != ending; } done: // Sync bit-reader state back - buffer_ = r.cur_byte; + buffer_ = r.buffer; bits_left_ = r.bits; in.wrapped_buf_advance_read_pos(r.pos); return ret; @@ -408,22 +489,23 @@ class GorillaDecoder : public Decoder { r.pos = 0; r.data_len = remain; r.bits = bits_left_; - r.cur_byte = buffer_; + r.buffer = buffer_; if (UNLIKELY(!first_value_was_read_)) { if (r.bits == 0 && r.pos >= r.data_len) goto done; - r.load_byte_if_empty(); stored_value_ = (T)r.read_long(GorillaRawOps<T>::VALUE_BITS); - if (UNLIKELY(r.exhausted)) { + if (UNLIKELY(r.exhausted || r.invalid)) { first_value_was_read_ = false; - ret = common::E_BUF_NOT_ENOUGH; + ret = r.invalid ? common::E_TSFILE_CORRUPTED + : common::E_BUF_NOT_ENOUGH; goto done; } first_value_was_read_ = true; - GorillaRawOps<T>::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + if (UNLIKELY(!GorillaRawOps<T>::read_next( + r, stored_value_, stored_leading_zeros_, + stored_trailing_zeros_))) { + ret = r.invalid ? common::E_TSFILE_CORRUPTED + : common::E_BUF_NOT_ENOUGH; goto done; } if (stored_value_ == ending) { @@ -438,10 +520,11 @@ class GorillaDecoder : public Decoder { while (skipped < count && has_next_) { skipped++; - GorillaRawOps<T>::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + if (UNLIKELY(!GorillaRawOps<T>::read_next( + r, stored_value_, stored_leading_zeros_, + stored_trailing_zeros_))) { + ret = r.invalid ? common::E_TSFILE_CORRUPTED + : common::E_BUF_NOT_ENOUGH; goto done; } if (stored_value_ == ending) { @@ -450,17 +533,18 @@ class GorillaDecoder : public Decoder { } done: - buffer_ = r.cur_byte; + buffer_ = r.buffer; bits_left_ = r.bits; in.wrapped_buf_advance_read_pos(r.pos); return ret; } - int batch_decode_fallback(T* out, int capacity, int& actual, T ending, + template <typename Output> + int batch_decode_fallback(Output* out, int capacity, int& actual, T ending, common::ByteStream& in) { actual = 0; while (actual < capacity && has_remaining(in)) { - out[actual++] = decode(in); + out[actual++] = GorillaDecodeOutput<T, Output>::convert(decode(in)); } return common::E_OK; } @@ -483,7 +567,7 @@ class GorillaDecoder : public Decoder { int bits_left_; bool first_value_was_read_; bool has_next_; - uint8_t buffer_; + uint64_t buffer_; }; template <> @@ -558,9 +642,7 @@ template <> FORCE_INLINE int32_t GorillaDecoder<int32_t>::cache_next(common::ByteStream& in) { read_next(in); - if (stored_value_ == GORILLA_ENCODING_ENDING_INTEGER) { - has_next_ = false; - } + has_next_ = stored_value_ != GORILLA_ENCODING_ENDING_INTEGER; return stored_value_; } @@ -568,9 +650,7 @@ template <> FORCE_INLINE int64_t GorillaDecoder<int64_t>::cache_next(common::ByteStream& in) { read_next(in); - if (stored_value_ == GORILLA_ENCODING_ENDING_LONG) { - has_next_ = false; - } + has_next_ = stored_value_ != GORILLA_ENCODING_ENDING_LONG; return stored_value_; } @@ -615,30 +695,15 @@ class FloatGorillaDecoder : public GorillaDecoder<int32_t> { int32_t cache_next(common::ByteStream& in) override { read_next(in); - if (stored_value_ == - common::float_to_int(GORILLA_ENCODING_ENDING_FLOAT)) { - has_next_ = false; - } + has_next_ = stored_value_ != + common::float_to_int(GORILLA_ENCODING_ENDING_FLOAT); return stored_value_; } int read_batch_float(float* out, int capacity, int& actual, common::ByteStream& in) override { int32_t ending = common::float_to_int(GORILLA_ENCODING_ENDING_FLOAT); - actual = 0; - while (actual < capacity && has_remaining(in)) { - int32_t buf[129]; - int batch = std::min(129, capacity - actual); - int buf_actual = 0; - int ret = batch_decode_raw(buf, batch, buf_actual, ending, in); - if (ret != common::E_OK) return ret; - if (buf_actual == 0) break; - for (int i = 0; i < buf_actual; i++) { - out[actual + i] = common::int_to_float(buf[i]); - } - actual += buf_actual; - } - return common::E_OK; + return batch_decode_raw(out, capacity, actual, ending, in); } int skip_float(int count, int& skipped, common::ByteStream& in) override { @@ -662,30 +727,15 @@ class DoubleGorillaDecoder : public GorillaDecoder<int64_t> { int64_t cache_next(common::ByteStream& in) override { read_next(in); - if (stored_value_ == - common::double_to_long(GORILLA_ENCODING_ENDING_DOUBLE)) { - has_next_ = false; - } + has_next_ = stored_value_ != + common::double_to_long(GORILLA_ENCODING_ENDING_DOUBLE); return stored_value_; } int read_batch_double(double* out, int capacity, int& actual, common::ByteStream& in) override { int64_t ending = common::double_to_long(GORILLA_ENCODING_ENDING_DOUBLE); - actual = 0; - while (actual < capacity && has_remaining(in)) { - int64_t buf[129]; - int batch = std::min(129, capacity - actual); - int buf_actual = 0; - int ret = batch_decode_raw(buf, batch, buf_actual, ending, in); - if (ret != common::E_OK) return ret; - if (buf_actual == 0) break; - for (int i = 0; i < buf_actual; i++) { - out[actual + i] = common::long_to_double(buf[i]); - } - actual += buf_actual; - } - return common::E_OK; + return batch_decode_raw(out, capacity, actual, ending, in); } int skip_double(int count, int& skipped, common::ByteStream& in) override { diff --git a/cpp/test/encoding/gorilla_codec_test.cc b/cpp/test/encoding/gorilla_codec_test.cc index 945451088..5f271c979 100644 --- a/cpp/test/encoding/gorilla_codec_test.cc +++ b/cpp/test/encoding/gorilla_codec_test.cc @@ -322,6 +322,31 @@ TEST_F(GorillaCodecTest, FloatBatchDecode) { } } +TEST_F(GorillaCodecTest, FloatBatchDecodeUnwrappedInput) { + storage::FloatGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + const int N = 127; + std::vector<float> expected(N); + for (int i = 0; i < N; i++) { + expected[i] = std::sin(i * 0.125f) * 23.0f; + ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK); + } + encoder.flush(stream); + + storage::FloatGorillaDecoder decoder; + std::vector<float> actual(N); + int actual_count = 0; + ASSERT_EQ(decoder.read_batch_float(actual.data(), actual.size(), + actual_count, stream), + common::E_OK); + ASSERT_EQ(actual_count, N); + for (int i = 0; i < N; i++) { + EXPECT_EQ(common::float_to_int(actual[i]), + common::float_to_int(expected[i])) + << "i=" << i; + } +} + TEST_F(GorillaCodecTest, DoubleBatchDecode) { storage::DoubleGorillaEncoder encoder; common::ByteStream stream(1024, common::MOD_DEFAULT); @@ -359,6 +384,144 @@ TEST_F(GorillaCodecTest, DoubleBatchDecode) { } } +TEST_F(GorillaCodecTest, DoubleBatchDecodeOneValueAtATime) { + storage::DoubleGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + const int N = 512; + std::vector<double> expected(N); + for (int i = 0; i < N; i++) { + expected[i] = (i % 9 == 0) ? 42.0 : std::sin(i * 0.03125) * 1000.0; + ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK); + } + encoder.flush(stream); + + uint32_t total = stream.total_size(); + std::vector<uint8_t> buf(total); + uint32_t got = 0; + stream.read_buf(buf.data(), total, got); + ASSERT_EQ(got, total); + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from((const char*)buf.data(), total); + + storage::DoubleGorillaDecoder decoder; + for (int i = 0; i < N; i++) { + double decoded = 0; + int actual = 0; + ASSERT_EQ(decoder.read_batch_double(&decoded, 1, actual, wrapped), + common::E_OK) + << "i=" << i; + ASSERT_EQ(actual, 1) << "i=" << i; + EXPECT_EQ(common::double_to_long(decoded), + common::double_to_long(expected[i])) + << "i=" << i; + } +} + +TEST_F(GorillaCodecTest, DoubleBatchScalarAndSkipInterleave) { + storage::DoubleGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + const int N = 400; + std::vector<double> expected(N); + for (int i = 0; i < N; i++) { + expected[i] = (i / 7) * 0.125 + std::cos(i * 0.017); + ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK); + } + encoder.flush(stream); + + uint32_t total = stream.total_size(); + std::vector<uint8_t> buf(total); + uint32_t got = 0; + stream.read_buf(buf.data(), total, got); + ASSERT_EQ(got, total); + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from((const char*)buf.data(), total); + + storage::DoubleGorillaDecoder decoder; + int cursor = 0; + for (; cursor < 7; cursor++) { + double decoded = 0; + ASSERT_EQ(decoder.read_double(decoded, wrapped), common::E_OK); + EXPECT_EQ(common::double_to_long(decoded), + common::double_to_long(expected[cursor])); + } + ASSERT_TRUE(decoder.has_remaining(wrapped)) + << "remaining=" << wrapped.remaining_size() + << " bits_left=" << decoder.bits_left_ + << " has_next=" << decoder.has_next_; + + std::vector<double> batch(113); + int actual = 0; + ASSERT_EQ( + decoder.read_batch_double(batch.data(), batch.size(), actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, static_cast<int>(batch.size())); + for (int i = 0; i < actual; i++, cursor++) { + EXPECT_EQ(common::double_to_long(batch[i]), + common::double_to_long(expected[cursor])); + } + + for (int i = 0; i < 5; i++, cursor++) { + double decoded = 0; + ASSERT_EQ(decoder.read_double(decoded, wrapped), common::E_OK); + EXPECT_EQ(common::double_to_long(decoded), + common::double_to_long(expected[cursor])); + } + + int skipped = 0; + ASSERT_EQ(decoder.skip_double(137, skipped, wrapped), common::E_OK); + ASSERT_EQ(skipped, 137); + cursor += skipped; + + std::vector<double> tail(N - cursor); + actual = 0; + ASSERT_EQ( + decoder.read_batch_double(tail.data(), tail.size(), actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, static_cast<int>(tail.size())); + for (int i = 0; i < actual; i++, cursor++) { + EXPECT_EQ(common::double_to_long(tail[i]), + common::double_to_long(expected[cursor])); + } + EXPECT_EQ(cursor, N); +} + +TEST_F(GorillaCodecTest, DoubleBatchDecodeFullWidthXor) { + const uint64_t patterns[] = { + 0x0000000000000000ULL, 0xFFFFFFFFFFFFFFFFULL, 0x0123456789ABCDEFULL, + 0xFEDCBA9876543210ULL, 0x8000000000000001ULL, 0x7FEFFFFFFFFFFFFFULL, + }; + const int N = sizeof(patterns) / sizeof(patterns[0]); + std::vector<double> expected(N); + + storage::DoubleGorillaEncoder encoder; + common::ByteStream stream(1024, common::MOD_DEFAULT); + for (int i = 0; i < N; i++) { + memcpy(&expected[i], &patterns[i], sizeof(double)); + ASSERT_EQ(encoder.encode(expected[i], stream), common::E_OK); + } + encoder.flush(stream); + + uint32_t total = stream.total_size(); + std::vector<uint8_t> buf(total); + uint32_t got = 0; + stream.read_buf(buf.data(), total, got); + ASSERT_EQ(got, total); + common::ByteStream wrapped(common::MOD_DEFAULT); + wrapped.wrap_from((const char*)buf.data(), total); + + storage::DoubleGorillaDecoder decoder; + std::vector<double> decoded(N); + int actual = 0; + ASSERT_EQ(decoder.read_batch_double(decoded.data(), N, actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, N); + for (int i = 0; i < N; i++) { + uint64_t decoded_bits = 0; + memcpy(&decoded_bits, &decoded[i], sizeof(double)); + EXPECT_EQ(decoded_bits, patterns[i]) << "i=" << i; + } +} + TEST_F(GorillaCodecTest, Int32BatchSkip) { storage::IntGorillaEncoder encoder; common::ByteStream stream(1024, common::MOD_DEFAULT);
