Skip to content

[Java][CPP] Preserve SPRINTZ extreme transitions and block boundaries #1023

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ColinLeeo wants to merge 1 commit into
base: develop
Choose a base branch
from
Open

[Java][CPP] Preserve SPRINTZ extreme transitions and block boundaries #1023

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 18 additions & 7 deletions cpp/src/encoding/double_sprintz_decoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,9 +100,8 @@ class DoubleSprintzDecoder : public SprintzDecoder {
}

bool has_remaining(const common::ByteStream& input) override {
uint32_t min_length = sizeof(uint32_t) + 1;
return (is_block_read_ && current_count_ < decode_size_) ||
input.remaining_size() >= min_length;
input.has_remaining();
}

protected:
Expand All @@ -126,6 +125,14 @@ class DoubleSprintzDecoder : public SprintzDecoder {
return ret;
}
}
// Java's Gorilla decoder eagerly consumes a byte after a
// byte-aligned terminator. Match it before the next Sprintz block.
if (decoder.bits_left_ == 0) {
uint8_t padding;
uint32_t read_len;
if (RET_FAIL(input.read_buf(&padding, 1, read_len))) return ret;
}

} else {
decode_size_ = block_size_ + 1;
common::SerializationUtil::read_double(pre_value_, input);
Expand Down Expand Up @@ -156,15 +163,17 @@ class DoubleSprintzDecoder : public SprintzDecoder {
if (predict_scheme_ == "delta") {
uint64_t prev_bits;
std::memcpy(&prev_bits, &current_buffer_[0], sizeof(prev_bits));
int64_t corrected0 =
convert_buffer_[0] + static_cast<int64_t>(prev_bits);
int64_t corrected0 = static_cast<int64_t>(
static_cast<uint64_t>(convert_buffer_[0]) + prev_bits);
convert_buffer_[0] = corrected0;
double d0;
std::memcpy(&d0, &corrected0, sizeof(corrected0));
current_buffer_[1] = d0;

for (int i = 1; i < block_size_; ++i) {
convert_buffer_[i] += convert_buffer_[i - 1];
convert_buffer_[i] = static_cast<int64_t>(
static_cast<uint64_t>(convert_buffer_[i]) +
static_cast<uint64_t>(convert_buffer_[i - 1]));
int64_t bits = convert_buffer_[i];
double di;
std::memcpy(&di, &bits, sizeof(bits));
Expand All @@ -177,7 +186,8 @@ class DoubleSprintzDecoder : public SprintzDecoder {
std::memcpy(&prev_bits, &current_buffer_[0], sizeof(prev_bits));
int64_t p = fire_pred_.predict(prev_bits);
int64_t e0 = convert_buffer_[0];
int64_t corrected0 = p + e0;
int64_t corrected0 = static_cast<int64_t>(
static_cast<uint64_t>(p) + static_cast<uint64_t>(e0));
convert_buffer_[0] = corrected0;
double d0;
std::memcpy(&d0, &corrected0, sizeof(corrected0));
Expand All @@ -190,7 +200,8 @@ class DoubleSprintzDecoder : public SprintzDecoder {
sizeof(prev_bits_i));
int64_t err = convert_buffer_[i];
int64_t pred = fire_pred_.predict(prev_bits_i);
int64_t corrected = pred + err;
int64_t corrected = static_cast<int64_t>(
static_cast<uint64_t>(pred) + static_cast<uint64_t>(err));
convert_buffer_[i] = corrected;
double di;
std::memcpy(&di, &corrected, sizeof(corrected));
Expand Down
37 changes: 32 additions & 5 deletions cpp/src/encoding/double_sprintz_encoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,13 +79,31 @@ class DoubleSprintzEncoder : public SprintzEncoder {
values_.push_back(value);

if (values_.size() == (size_t)block_size_ + 1) {
const std::vector<double> raw_values(values_);
unsafe_residual_ = false;
fire_pred_.reset();
for (int i = 1; i <= block_size_; ++i) {
convert_buffer_[i - 1] = predict(values_[i], values_[i - 1]);
}
bit_pack();
is_first_cached_ = false;
values_.clear();
if (unsafe_residual_) {
// Reuse the existing raw fallback format; retain the ninth
// value as the start of the next block.
common::SerializationUtil::
write_int_little_endian_padded_on_bit_width(
block_size_ | 0x80, byte_cache_, 1);
DoubleGorillaEncoder encoder;
for (int i = 0; i < block_size_; ++i) {
if (RET_FAIL(encoder.encode(raw_values[i], byte_cache_)))
return ret;
}
if (RET_FAIL(encoder.flush(byte_cache_))) return ret;
values_.assign(1, raw_values.back());
is_first_cached_ = true;
} else {
bit_pack();
is_first_cached_ = false;
values_.clear();
}
group_num_++;
if (group_num_ == group_max_) {
if (RET_FAIL(flush(out_stream))) return ret;
Expand Down Expand Up @@ -151,19 +169,28 @@ class DoubleSprintzEncoder : public SprintzEncoder {
int64_t prev_bits = common::double_to_long(prev);
int64_t raw_pred = 0;
if (predict_method_ == "delta") {
raw_pred = curr_bits - prev_bits;
raw_pred = static_cast<int64_t>(static_cast<uint64_t>(curr_bits) -
static_cast<uint64_t>(prev_bits));
} else if (predict_method_ == "fire") {
int64_t pred = fire_pred_.predict(prev_bits);
int64_t err = curr_bits - pred;
int64_t err = static_cast<int64_t>(
static_cast<uint64_t>(curr_bits) - static_cast<uint64_t>(pred));
fire_pred_.train(prev_bits, curr_bits, err);
raw_pred = err;
} else {
ASSERT(false);
}
// The historical mapping is only reversible within this range.
const int64_t limit = int64_t(1) << 62;
if (raw_pred <= -limit || raw_pred >= limit) {
unsafe_residual_ = true;
return 0;
}
return (raw_pred <= 0) ? -2 * raw_pred : 2 * raw_pred - 1;
}

private:
bool unsafe_residual_ = false;
std::vector<double> values_;
std::vector<int64_t> convert_buffer_;
std::shared_ptr<Int64Packer> packer_;
Expand Down
52 changes: 25 additions & 27 deletions cpp/src/encoding/fire.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,16 +64,21 @@ class IntFire : public Fire<int> {

int predict(int value) override {
int alpha = accumulator_ >> learn_shift_;
int diff = static_cast<int>((static_cast<int64_t>(alpha) * delta_)) >>
int diff = static_cast<int32_t>(static_cast<uint32_t>(alpha) *
static_cast<uint32_t>(delta_)) >>
bit_width_;

return value + diff;
return static_cast<int32_t>(static_cast<uint32_t>(value) +
static_cast<uint32_t>(diff));
}

void train(int pre, int val, int err) override {
int gradient = err > 0 ? -delta_ : delta_;
accumulator_ -= gradient;
delta_ = val - pre;
uint32_t gradient = err > 0
? uint32_t(0) - static_cast<uint32_t>(delta_)
: static_cast<uint32_t>(delta_);
accumulator_ = static_cast<int32_t>(
static_cast<uint32_t>(accumulator_) - gradient);
delta_ = static_cast<int32_t>(static_cast<uint32_t>(val) -
static_cast<uint32_t>(pre));
}
};

Expand All @@ -92,30 +97,23 @@ class LongFire : public Fire<int64_t> {

int64_t predict(int64_t value) override {
int64_t alpha = accumulator_ >> learn_shift_;
int64_t diff = safe_mul_shift(alpha, delta_, bit_width_);
return value + diff;
// Java wraps the product to 64 bits before the arithmetic shift.
int64_t diff = static_cast<int64_t>(static_cast<uint64_t>(alpha) *
static_cast<uint64_t>(delta_)) >>
bit_width_;
return static_cast<int64_t>(static_cast<uint64_t>(value) +
static_cast<uint64_t>(diff));
}

void train(int64_t pre, int64_t val, int64_t err) override {
int64_t gradient = err > 0 ? -delta_ : delta_;
accumulator_ -= gradient;
delta_ = val - pre;
}

private:
/** (alpha * delta_) >> shift without signed overflow; both args are
* int64_t. */
static int64_t safe_mul_shift(int64_t alpha, int64_t delta, int shift) {
#if defined(__SIZEOF_INT128__) && __SIZEOF_INT128__ >= 16
__int128 product = static_cast<__int128>(alpha) * delta;
return static_cast<int64_t>(product >> shift);
#else
/* Portable fallback: use double for product. Exact for |alpha|,|delta|
* < 2^53. */
double prod = static_cast<double>(alpha) * static_cast<double>(delta);
double div = static_cast<double>(1LL << shift);
return static_cast<int64_t>(prod / div);
#endif
uint64_t gradient = err > 0
? uint64_t(0) - static_cast<uint64_t>(delta_)
: static_cast<uint64_t>(delta_);
accumulator_ =
static_cast<int32_t>(static_cast<uint32_t>(accumulator_) -
static_cast<uint32_t>(gradient));
delta_ = static_cast<int64_t>(static_cast<uint64_t>(val) -
static_cast<uint64_t>(pre));
}
};

Expand Down
25 changes: 18 additions & 7 deletions cpp/src/encoding/float_sprintz_decoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,9 +86,8 @@ class FloatSprintzDecoder : public SprintzDecoder {
}

bool has_remaining(const common::ByteStream& input) override {
size_t min_length = sizeof(uint32_t) + 1;
return (is_block_read_ && current_count_ < decode_size_) ||
input.remaining_size() >= min_length;
input.has_remaining();
}

int read_float(float& ret_value, common::ByteStream& input) override {
Expand Down Expand Up @@ -127,6 +126,14 @@ class FloatSprintzDecoder : public SprintzDecoder {
return ret;
}
}
// Java's Gorilla decoder eagerly consumes a byte after a
// byte-aligned terminator. Match it before the next Sprintz block.
if (decoder.bits_left_ == 0) {
uint8_t padding;
uint32_t read_len;
if (RET_FAIL(input.read_buf(&padding, 1, read_len))) return ret;
}

} else {
// packed block
decode_size_ = block_size_ + 1;
Expand Down Expand Up @@ -162,8 +169,8 @@ class FloatSprintzDecoder : public SprintzDecoder {
std::memcpy(&prev_bits, &current_buffer_[0], sizeof(prev_bits));
// Java: convertBuffer[0] = convertBuffer[0] +
// Float.floatToIntBits(preValue);
int32_t corrected0 =
convert_buffer_[0] + static_cast<int32_t>(prev_bits);
int32_t corrected0 = static_cast<int32_t>(
static_cast<uint32_t>(convert_buffer_[0]) + prev_bits);
convert_buffer_[0] = corrected0;
// Java: currentBuffer[1] = Float.intBitsToFloat(convertBuffer[0]);
float f0;
Expand All @@ -172,7 +179,9 @@ class FloatSprintzDecoder : public SprintzDecoder {

for (int i = 1; i < block_size_; ++i) {
// Java: convertBuffer[i] += convertBuffer[i - 1];
convert_buffer_[i] += convert_buffer_[i - 1];
convert_buffer_[i] = static_cast<int32_t>(
static_cast<uint32_t>(convert_buffer_[i]) +
static_cast<uint32_t>(convert_buffer_[i - 1]));
int32_t bits = convert_buffer_[i];
float fi;
std::memcpy(&fi, &bits, sizeof(bits));
Expand All @@ -186,7 +195,8 @@ class FloatSprintzDecoder : public SprintzDecoder {
// Java: int p = firePred.predict(Float.floatToIntBits(preValue));
int32_t p = fire_pred_.predict(prev_bits);
int32_t e0 = convert_buffer_[0];
int32_t corrected0 = p + e0;
int32_t corrected0 = static_cast<int32_t>(
static_cast<uint32_t>(p) + static_cast<uint32_t>(e0));
convert_buffer_[0] = corrected0;
float f0;
std::memcpy(&f0, &corrected0, sizeof(corrected0));
Expand All @@ -201,7 +211,8 @@ class FloatSprintzDecoder : public SprintzDecoder {
sizeof(prev_bits_i));
int32_t err = convert_buffer_[i];
int32_t pred = fire_pred_.predict(prev_bits_i);
int32_t corrected = pred + err;
int32_t corrected = static_cast<int32_t>(
static_cast<uint32_t>(pred) + static_cast<uint32_t>(err));
convert_buffer_[i] = corrected;
float fi;
std::memcpy(&fi, &corrected, sizeof(corrected));
Expand Down
37 changes: 32 additions & 5 deletions cpp/src/encoding/float_sprintz_encoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,13 +76,31 @@ class FloatSprintzEncoder : public SprintzEncoder {
values_.push_back(value);

if (values_.size() == (size_t)block_size_ + 1) {
const std::vector<float> raw_values(values_);
unsafe_residual_ = false;
fire_pred_.reset();
for (int i = 1; i <= block_size_; ++i) {
convert_buffer_[i - 1] = predict(values_[i], values_[i - 1]);
}
bit_pack();
is_first_cached_ = false;
values_.clear();
if (unsafe_residual_) {
// Reuse the existing raw fallback format; retain the ninth
// value as the start of the next block.
common::SerializationUtil::
write_int_little_endian_padded_on_bit_width(
block_size_ | 0x80, byte_cache_, 1);
FloatGorillaEncoder encoder;
for (int i = 0; i < block_size_; ++i) {
if (RET_FAIL(encoder.encode(raw_values[i], byte_cache_)))
return ret;
}
if (RET_FAIL(encoder.flush(byte_cache_))) return ret;
values_.assign(1, raw_values.back());
is_first_cached_ = true;
} else {
bit_pack();
is_first_cached_ = false;
values_.clear();
}
group_num_++;
if (group_num_ == group_max_) {
if (RET_FAIL(flush(out_stream))) return ret;
Expand Down Expand Up @@ -149,20 +167,29 @@ class FloatSprintzEncoder : public SprintzEncoder {
int32_t prev_bits = common::float_to_int(prev_value);
int32_t raw_pred = 0;
if (predict_method_ == "delta") {
raw_pred = curr_bits - prev_bits;
raw_pred = static_cast<int32_t>(static_cast<uint32_t>(curr_bits) -
static_cast<uint32_t>(prev_bits));
} else if (predict_method_ == "fire") {
int32_t pred = fire_pred_.predict(prev_bits);
int32_t err = curr_bits - pred;
int32_t err = static_cast<int32_t>(
static_cast<uint32_t>(curr_bits) - static_cast<uint32_t>(pred));
fire_pred_.train(prev_bits, curr_bits, err);
raw_pred = err;
} else {
// unsupported
ASSERT(false);
}
// The historical mapping is only reversible within this range.
const int32_t limit = int32_t(1) << 30;
if (raw_pred <= -limit || raw_pred >= limit) {
unsafe_residual_ = true;
return 0;
}
return (raw_pred <= 0) ? -2 * raw_pred : 2 * raw_pred - 1;
}

private:
bool unsafe_residual_ = false;
std::vector<float> values_;
std::vector<int32_t> convert_buffer_;
std::shared_ptr<Int32Packer> packer_;
Expand Down
17 changes: 9 additions & 8 deletions cpp/src/encoding/int32_sprintz_decoder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,9 +56,8 @@ class Int32SprintzDecoder : public SprintzDecoder {
}

bool has_remaining(const common::ByteStream& in) override {
uint32_t min_len = sizeof(int32_t) + 1;
return (is_block_read_ && current_count_ < block_size_) ||
in.remaining_size() >= min_len;
return (is_block_read_ && current_count_ < decode_size_) ||
in.has_remaining();
}

int read_boolean(bool& ret_value, common::ByteStream& in) override {
Expand Down Expand Up @@ -103,9 +102,8 @@ class Int32SprintzDecoder : public SprintzDecoder {
}

bool has_next(common::ByteStream& input) {
uint32_t min_length = sizeof(int32_t) + 1;
return (is_block_read_ && current_count_ < block_size_) ||
input.remaining_size() >= min_length;
return (is_block_read_ && current_count_ < decode_size_) ||
input.has_remaining();
}

protected:
Expand Down Expand Up @@ -166,14 +164,17 @@ class Int32SprintzDecoder : public SprintzDecoder {

if (predict_scheme_ == "delta") {
for (size_t i = 1; i < current_buffer_.size(); ++i) {
current_buffer_[i] += current_buffer_[i - 1];
current_buffer_[i] = static_cast<int32_t>(
static_cast<uint32_t>(current_buffer_[i]) +
static_cast<uint32_t>(current_buffer_[i - 1]));
}
} else if (predict_scheme_ == "fire") {
fire_pred_.reset();
for (int i = 1; i <= block_size_; ++i) {
int32_t pred = fire_pred_.predict(current_buffer_[i - 1]);
int32_t err = current_buffer_[i];
current_buffer_[i] = pred + err;
current_buffer_[i] = static_cast<int32_t>(
static_cast<uint32_t>(pred) + static_cast<uint32_t>(err));
fire_pred_.train(current_buffer_[i - 1], current_buffer_[i],
err);
}
Expand Down
Loading
Loading