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(src[0]) << 56) | + (static_cast(src[1]) << 48) | + (static_cast(src[2]) << 40) | + (static_cast(src[3]) << 32) | + (static_cast(src[4]) << 24) | + (static_cast(src[5]) << 16) | + (static_cast(src[6]) << 8) | + static_cast(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 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 { 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 { 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(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(r.read_long(meaningful)); + if (UNLIKELY(r.exhausted || r.invalid)) return false; + xor_value <<= stored_trailing_zeros; + stored_value ^= static_cast(xor_value); } // fallthrough default: - return stored_value; + return true; } - return stored_value; + return true; } }; @@ -155,11 +202,12 @@ template <> struct GorillaRawOps { 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 { 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(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(xor_value); } // fallthrough default: - return stored_value; + return true; } - return stored_value; + return true; + } +}; + +template +struct GorillaDecodeOutput; + +template +struct GorillaDecodeOutput { + static FORCE_INLINE T convert(T value) { return value; } +}; + +template <> +struct GorillaDecodeOutput { + static FORCE_INLINE float convert(int32_t value) { + return common::int_to_float(value); + } +}; + +template <> +struct GorillaDecodeOutput { + 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 + 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::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::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + if (UNLIKELY(!GorillaRawOps::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::convert(first_value); if (!has_next_ || actual >= capacity) goto done; } // Main batch loop while (actual < capacity && has_next_) { - out[actual++] = stored_value_; - GorillaRawOps::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + out[actual++] = + GorillaDecodeOutput::convert(stored_value_); + if (UNLIKELY(!GorillaRawOps::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::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::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + if (UNLIKELY(!GorillaRawOps::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::read_next(r, stored_value_, stored_leading_zeros_, - stored_trailing_zeros_); - if (UNLIKELY(r.exhausted)) { - ret = common::E_BUF_NOT_ENOUGH; + if (UNLIKELY(!GorillaRawOps::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 + 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::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::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(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(read_long(meaningful, in)); + if (UNLIKELY(read_status_ != common::E_OK)) { + return stored_value_; + } + xor_value <<= stored_trailing_zeros_; + stored_value_ ^= static_cast(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::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(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(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::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::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::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::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 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 { 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 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 { 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 +#include +#include #include #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 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 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 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 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(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 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 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(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 batch(113); + int actual = 0; + ASSERT_EQ( + decoder.read_batch_double(batch.data(), static_cast(batch.size()), + actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, static_cast(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 tail(N - cursor); + actual = 0; + ASSERT_EQ(decoder.read_batch_double( + tail.data(), static_cast(tail.size()), actual, wrapped), + common::E_OK); + ASSERT_EQ(actual, static_cast(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 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 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(buf.data()), total); + + storage::DoubleGorillaDecoder decoder; + std::vector 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 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 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);