This is an automated email from the ASF dual-hosted git repository.
ColinLeeo pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/tsfile.git
The following commit(s) were added to refs/heads/develop by this push:
new 7d44338c1 Optimize Gorilla batch floating-point decoding (#873)
7d44338c1 is described below
commit 7d44338c1d6ec1906bae52522276d666a5ec9bea
Author: Colin Lee <[email protected]>
AuthorDate: Thu Jul 23 20:15:41 2026 +0800
Optimize Gorilla batch floating-point decoding (#873)
---
cpp/src/encoding/gorilla_decoder.h | 486 ++++++++++++++++++++------------
cpp/test/encoding/gorilla_codec_test.cc | 200 +++++++++++++
2 files changed, 499 insertions(+), 187 deletions(-)
diff --git a/cpp/src/encoding/gorilla_decoder.h
b/cpp/src/encoding/gorilla_decoder.h
index e1e490105..c8c76d965 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);
}
};
@@ -203,6 +283,7 @@ class GorillaDecoder : public Decoder {
first_value_was_read_ = false;
has_next_ = false;
buffer_ = 0;
+ read_status_ = common::E_OK;
}
FORCE_INLINE bool has_next() { return has_next_; }
@@ -210,20 +291,31 @@ class GorillaDecoder : public Decoder {
return buffer.has_remaining() || has_next();
}
- // If empty, cache 8 bits from in_stream to 'buffer_'.
- void flush_byte_if_empty(common::ByteStream& in) {
+ // 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.
+ bool flush_byte_if_empty(common::ByteStream& in) {
+ if (UNLIKELY(read_status_ != common::E_OK)) {
+ return false;
+ }
if (bits_left_ == 0) {
+ uint8_t next_byte = 0;
uint32_t read_len = 0;
- in.read_buf(&buffer_, 1, read_len);
+ in.read_buf(&next_byte, 1, read_len);
+ if (UNLIKELY(read_len == 0)) {
+ read_status_ = common::E_BUF_NOT_ENOUGH;
+ return false;
+ }
+ buffer_ = next_byte;
bits_left_ = 8;
}
+ return true;
}
// Reads the next bit and returns true if the next bit is 1, otherwise 0.
bool read_bit(common::ByteStream& in) {
+ if (UNLIKELY(!flush_byte_if_empty(in))) return false;
bool bit = ((buffer_ >> (bits_left_ - 1)) & 1) == 1;
bits_left_--;
- flush_byte_if_empty(in);
return bit;
}
@@ -233,26 +325,29 @@ 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) {
+ if (UNLIKELY(bits < 0 || bits > 64)) {
+ read_status_ = common::E_TSFILE_CORRUPTED;
+ return 0;
+ }
+
+ 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_;
+ if (UNLIKELY(!flush_byte_if_empty(in))) return value;
+
+ int take = bits < bits_left_ ? bits : bits_left_;
+ if (take == 64) {
+ // A read is at most 64 bits, so this is necessarily the only
+ // iteration. Return directly to avoid an undefined shift by
64.
bits_left_ = 0;
- } 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;
+ return buffer_;
}
- flush_byte_if_empty(in);
+
+ uint64_t chunk =
+ (buffer_ >> (bits_left_ - take)) & ((uint64_t{1} << take) - 1);
+ value = (value << take) | chunk;
+ bits_left_ -= take;
+ bits -= take;
}
return value;
}
@@ -301,10 +396,14 @@ 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;
+ if (UNLIKELY(read_status_ != common::E_OK)) {
+ return read_status_;
+ }
// Bootstrap below would unconditionally write out[0]; guard the
// zero-capacity edge case so callers can probe without writing.
if (capacity <= 0) {
@@ -327,57 +426,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;
@@ -387,6 +484,9 @@ class GorillaDecoder : public Decoder {
common::ByteStream& in) {
int ret = common::E_OK;
skipped = 0;
+ if (UNLIKELY(read_status_ != common::E_OK)) {
+ return read_status_;
+ }
// Bootstrap below would consume first_value_ even when count == 0,
// advancing the stream past data the caller didn't ask to skip.
if (count <= 0) {
@@ -408,22 +508,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 +539,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 +552,22 @@ 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);
+ T value = decode(in);
+ if (UNLIKELY(read_status_ != common::E_OK)) {
+ return read_status_;
+ }
+ out[actual++] = GorillaDecodeOutput<T, Output>::convert(value);
}
return common::E_OK;
}
@@ -470,6 +577,9 @@ class GorillaDecoder : public Decoder {
skipped = 0;
while (skipped < count && has_remaining(in)) {
decode(in);
+ if (UNLIKELY(read_status_ != common::E_OK)) {
+ return read_status_;
+ }
skipped++;
}
return common::E_OK;
@@ -483,34 +593,51 @@ class GorillaDecoder : public Decoder {
int bits_left_;
bool first_value_was_read_;
bool has_next_;
- uint8_t buffer_;
+ uint64_t buffer_;
+ int read_status_;
};
template <>
FORCE_INLINE int32_t
GorillaDecoder<int32_t>::read_next(common::ByteStream& in) {
uint8_t control_bits = read_next_control_bit(2, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
uint8_t significant_bits = 0;
- int32_t xor_value = 0;
switch (control_bits) {
case 3: // case '11': use new leading and trailing zeros
stored_leading_zeros_ =
(int)read_long(LEADING_ZERO_BITS_LENGTH_32BIT,
in); // todo: int or int32_t?
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
significant_bits =
(uint8_t)read_long(MEANINGFUL_XOR_BITS_LENGTH_32BIT, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
significant_bits++;
+ if (UNLIKELY(stored_leading_zeros_ + significant_bits >
+ VALUE_BITS_LENGTH_32BIT)) {
+ read_status_ = common::E_TSFILE_CORRUPTED;
+ return stored_value_;
+ }
stored_trailing_zeros_ = VALUE_BITS_LENGTH_32BIT -
significant_bits - stored_leading_zeros_;
// missing break is intentional, we want to overflow to next one
case 2: // case '10': use stored leading and trailing zeros
- xor_value = (int32_t)read_long(VALUE_BITS_LENGTH_32BIT -
- stored_leading_zeros_ -
- stored_trailing_zeros_,
- in);
- xor_value = static_cast<uint32_t>(xor_value)
- << stored_trailing_zeros_;
- stored_value_ ^= xor_value;
+ {
+ int meaningful = VALUE_BITS_LENGTH_32BIT - stored_leading_zeros_ -
+ stored_trailing_zeros_;
+ if (UNLIKELY(meaningful <= 0 ||
+ meaningful > VALUE_BITS_LENGTH_32BIT)) {
+ read_status_ = common::E_TSFILE_CORRUPTED;
+ return stored_value_;
+ }
+ uint32_t xor_value =
+ static_cast<uint32_t>(read_long(meaningful, in));
+ if (UNLIKELY(read_status_ != common::E_OK)) {
+ return stored_value_;
+ }
+ xor_value <<= stored_trailing_zeros_;
+ stored_value_ ^= static_cast<int32_t>(xor_value);
+ }
// missing break is intentional, we want to overflow to next one
default: // case '0': use stored value
return stored_value_;
@@ -522,29 +649,40 @@ template <>
FORCE_INLINE int64_t
GorillaDecoder<int64_t>::read_next(common::ByteStream& in) {
uint8_t control_bits = read_next_control_bit(2, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
uint8_t significant_bits = 0;
- int64_t xor_value = 0;
switch (control_bits) {
case 3: { // case '11': use new leading and trailing zeros
stored_leading_zeros_ =
(int)read_long(LEADING_ZERO_BITS_LENGTH_64BIT,
in); // todo: int or int32_t?
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
significant_bits =
(uint8_t)read_long(MEANINGFUL_XOR_BITS_LENGTH_64BIT, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
significant_bits++;
+ if (UNLIKELY(stored_leading_zeros_ + significant_bits >
+ VALUE_BITS_LENGTH_64BIT)) {
+ read_status_ = common::E_TSFILE_CORRUPTED;
+ return stored_value_;
+ }
stored_trailing_zeros_ = VALUE_BITS_LENGTH_64BIT -
significant_bits - stored_leading_zeros_;
// missing break is intentional, we want to overflow to next one
}
case 2: { // case '10': use stored leading and trailing zeros
- xor_value =
- read_long(VALUE_BITS_LENGTH_64BIT - stored_leading_zeros_ -
- stored_trailing_zeros_,
- in);
- xor_value = static_cast<uint64_t>(xor_value)
- << stored_trailing_zeros_;
- stored_value_ ^= xor_value;
+ int meaningful = VALUE_BITS_LENGTH_64BIT - stored_leading_zeros_ -
+ stored_trailing_zeros_;
+ if (UNLIKELY(meaningful <= 0 ||
+ meaningful > VALUE_BITS_LENGTH_64BIT)) {
+ read_status_ = common::E_TSFILE_CORRUPTED;
+ return stored_value_;
+ }
+ uint64_t xor_value = read_long(meaningful, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return stored_value_;
+ xor_value <<= stored_trailing_zeros_;
+ stored_value_ ^= static_cast<int64_t>(xor_value);
// missing break is intentional, we want to overflow to next one
}
default: { // case '0': use stored value
@@ -558,8 +696,8 @@ 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;
+ if (LIKELY(read_status_ == common::E_OK)) {
+ has_next_ = stored_value_ != GORILLA_ENCODING_ENDING_INTEGER;
}
return stored_value_;
}
@@ -568,8 +706,8 @@ 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;
+ if (LIKELY(read_status_ == common::E_OK)) {
+ has_next_ = stored_value_ != GORILLA_ENCODING_ENDING_LONG;
}
return stored_value_;
}
@@ -578,8 +716,8 @@ template <>
FORCE_INLINE int32_t GorillaDecoder<int32_t>::decode(common::ByteStream& in) {
int32_t ret_value = stored_value_;
if (UNLIKELY(!first_value_was_read_)) {
- flush_byte_if_empty(in);
stored_value_ = (int32_t)read_long(VALUE_BITS_LENGTH_32BIT, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return ret_value;
first_value_was_read_ = true;
ret_value = stored_value_;
}
@@ -591,8 +729,8 @@ template <>
FORCE_INLINE int64_t GorillaDecoder<int64_t>::decode(common::ByteStream& in) {
int64_t ret_value = stored_value_;
if (UNLIKELY(!first_value_was_read_)) {
- flush_byte_if_empty(in);
stored_value_ = read_long(VALUE_BITS_LENGTH_64BIT, in);
+ if (UNLIKELY(read_status_ != common::E_OK)) return ret_value;
first_value_was_read_ = true;
ret_value = stored_value_;
}
@@ -615,9 +753,9 @@ 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;
+ if (LIKELY(read_status_ == common::E_OK)) {
+ has_next_ = stored_value_ !=
+ common::float_to_int(GORILLA_ENCODING_ENDING_FLOAT);
}
return stored_value_;
}
@@ -625,20 +763,7 @@ class FloatGorillaDecoder : public GorillaDecoder<int32_t>
{
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,9 +787,9 @@ 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;
+ if (LIKELY(read_status_ == common::E_OK)) {
+ has_next_ = stored_value_ !=
+ common::double_to_long(GORILLA_ENCODING_ENDING_DOUBLE);
}
return stored_value_;
}
@@ -672,20 +797,7 @@ class DoubleGorillaDecoder : public
GorillaDecoder<int64_t> {
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 {
@@ -760,7 +872,7 @@ template <>
FORCE_INLINE int IntGorillaDecoder::read_int32(int32_t& ret_value,
common::ByteStream& in) {
ret_value = decode(in);
- return common::E_OK;
+ return read_status_;
}
template <>
FORCE_INLINE int IntGorillaDecoder::read_int64(int64_t& ret_value,
@@ -803,7 +915,7 @@ template <>
FORCE_INLINE int LongGorillaDecoder::read_int64(int64_t& ret_value,
common::ByteStream& in) {
ret_value = decode(in);
- return common::E_OK;
+ return read_status_;
}
template <>
FORCE_INLINE int LongGorillaDecoder::read_float(float& ret_value,
@@ -842,7 +954,7 @@ FORCE_INLINE int FloatGorillaDecoder::read_int64(int64_t&
ret_value,
FORCE_INLINE int FloatGorillaDecoder::read_float(float& ret_value,
common::ByteStream& in) {
ret_value = decode(in);
- return common::E_OK;
+ return read_status_;
}
FORCE_INLINE int FloatGorillaDecoder::read_double(double& ret_value,
common::ByteStream& in) {
@@ -872,7 +984,7 @@ FORCE_INLINE int DoubleGorillaDecoder::read_float(float&
ret_value,
FORCE_INLINE int DoubleGorillaDecoder::read_double(double& ret_value,
common::ByteStream& in) {
ret_value = decode(in);
- return common::E_OK;
+ return read_status_;
}
} // end namespace storage
diff --git a/cpp/test/encoding/gorilla_codec_test.cc
b/cpp/test/encoding/gorilla_codec_test.cc
index 945451088..b1fe72136 100644
--- a/cpp/test/encoding/gorilla_codec_test.cc
+++ b/cpp/test/encoding/gorilla_codec_test.cc
@@ -18,6 +18,8 @@
*/
#include <gtest/gtest.h>
+#include <cmath>
+#include <cstring>
#include <limits>
#include "encoding/gorilla_decoder.h"
@@ -322,6 +324,30 @@ 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(), N, 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 +385,180 @@ 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(reinterpret_cast<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(reinterpret_cast<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()
+ << " has_next=" << decoder.has_next();
+
+ std::vector<double> batch(113);
+ int actual = 0;
+ ASSERT_EQ(
+ decoder.read_batch_double(batch.data(), static_cast<int>(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(), static_cast<int>(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++) {
+ std::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(reinterpret_cast<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;
+ std::memcpy(&decoded_bits, &decoded[i], sizeof(double));
+ EXPECT_EQ(decoded_bits, patterns[i]) << "i=" << i;
+ }
+}
+
+TEST_F(GorillaCodecTest, DoubleBatchTruncatedUnwrappedInputReturnsError) {
+ storage::DoubleGorillaEncoder encoder;
+ common::ByteStream encoded_stream(1024, common::MOD_DEFAULT);
+ const int N = 128;
+ for (int i = 0; i < N; i++) {
+ double value = std::sin(i * 0.03125) * 1000.0 + i * 0.125;
+ ASSERT_EQ(encoder.encode(value, encoded_stream), common::E_OK);
+ }
+ ASSERT_EQ(encoder.flush(encoded_stream), common::E_OK);
+
+ uint32_t total = encoded_stream.total_size();
+ ASSERT_GT(total, 1u);
+ std::vector<uint8_t> encoded(total);
+ uint32_t got = 0;
+ encoded_stream.read_buf(encoded.data(), total, got);
+ ASSERT_EQ(got, total);
+
+ common::ByteStream decode_input(1024, common::MOD_DEFAULT);
+ ASSERT_EQ(decode_input.write_buf(encoded.data(), total - 1), common::E_OK);
+ storage::DoubleGorillaDecoder decoder;
+ std::vector<double> decoded(N);
+ int actual = -1;
+ EXPECT_EQ(
+ decoder.read_batch_double(decoded.data(), N, actual, decode_input),
+ common::E_BUF_NOT_ENOUGH);
+ EXPECT_LT(actual, N);
+
+ common::ByteStream skip_input(1024, common::MOD_DEFAULT);
+ ASSERT_EQ(skip_input.write_buf(encoded.data(), total - 1), common::E_OK);
+ storage::DoubleGorillaDecoder skip_decoder;
+ int skipped = -1;
+ EXPECT_EQ(skip_decoder.skip_double(N, skipped, skip_input),
+ common::E_BUF_NOT_ENOUGH);
+ EXPECT_LT(skipped, N);
+}
+
TEST_F(GorillaCodecTest, Int32BatchSkip) {
storage::IntGorillaEncoder encoder;
common::ByteStream stream(1024, common::MOD_DEFAULT);
