This is an automated email from the ASF dual-hosted git repository.
tisonkun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datasketches-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 34f3041 perf(cpc): encode and decode payloads without intermediate
states (#229)
34f3041 is described below
commit 34f3041e669737a3aeb7169985050b4a0097d9ab
Author: tison <[email protected]>
AuthorDate: Wed Aug 26 18:07:50 2026 +0800
perf(cpc): encode and decode payloads without intermediate states (#229)
---
CHANGELOG.md | 1 +
datasketches/src/cpc/compression.rs | 729 ++++++------------------------------
datasketches/src/cpc/sketch.rs | 260 ++++++++++---
3 files changed, 318 insertions(+), 672 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 269fa4d..eba1dae 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ All significant changes to this project will be documented in
this file.
### Performance improvements
* Reduce T-Digest allocation overhead and retained memory across updates,
compression, merges, serialization, deserialization, and freezing while
preserving the serialized format.
+* Reduce CPC serialization and deserialization allocations by encoding
directly into the output buffer and decoding directly from the input payload.
### Bug fixes
diff --git a/datasketches/src/cpc/compression.rs
b/datasketches/src/cpc/compression.rs
index db07f8e..4cefe56 100644
--- a/datasketches/src/cpc/compression.rs
+++ b/datasketches/src/cpc/compression.rs
@@ -17,545 +17,127 @@
use std::cmp::Ordering;
-use crate::cpc::CpcSketch;
-use crate::cpc::Flavor;
-use crate::cpc::compression_data::COLUMN_PERMUTATIONS_FOR_DECODING;
-use crate::cpc::compression_data::COLUMN_PERMUTATIONS_FOR_ENCODING;
+use crate::codec::SketchBytes;
use crate::cpc::compression_data::DECODING_TABLES_FOR_HIGH_ENTROPY_BYTE;
use crate::cpc::compression_data::ENCODING_TABLES_FOR_HIGH_ENTROPY_BYTE;
use crate::cpc::compression_data::LENGTH_LIMITED_UNARY_DECODING_TABLE65;
use crate::cpc::compression_data::LENGTH_LIMITED_UNARY_ENCODING_TABLE65;
-use crate::cpc::determine_correct_offset;
-use crate::cpc::determine_flavor;
-use crate::cpc::pair_table::PairTable;
use crate::error::Error;
-#[derive(Default)]
-pub(super) struct CompressedState {
- pub(super) table_data: Vec<u32>,
- pub(super) table_data_words: usize,
- // can be different from the number of entries in the sketch in hybrid mode
- pub(super) table_num_entries: u32,
- pub(super) window_data: Vec<u32>,
- pub(super) window_data_words: usize,
-}
+pub(super) fn encode_pairs(pairs: &[u32], lg_k: u8, output: &mut SketchBytes)
-> usize {
+ let num_pairs = pairs.len() as u32;
+ let num_base_bits =
+ golomb_choose_number_of_base_bits((1 << lg_k) + num_pairs,
u64::from(num_pairs));
+ let mut bits = BitWriter::new(output);
+ let golomb_lo_mask = (1 << num_base_bits) - 1;
+ let mut predicted_row_index = 0;
+ let mut predicted_col_index = 0;
-impl CompressedState {
- pub fn compress(&mut self, source: &CpcSketch) {
- match source.flavor() {
- Flavor::Empty => {
- // do nothing
- }
- Flavor::Sparse => {
- self.compress_sparse_flavor(source);
- debug_assert!(self.window_data.is_empty(), "window is not
expected");
- debug_assert!(!self.table_data.is_empty(), "table is
expected");
- }
- Flavor::Hybrid => {
- self.compress_hybrid_flavor(source);
- debug_assert!(self.window_data.is_empty(), "window is not
expected");
- debug_assert!(!self.table_data.is_empty(), "table is
expected");
- }
- Flavor::Pinned => {
- self.compress_pinned_flavor(source);
- debug_assert!(!self.window_data.is_empty(), "window is
expected");
- }
- Flavor::Sliding => {
- self.compress_sliding_flavor(source);
- debug_assert!(!self.window_data.is_empty(), "window is
expected");
- }
+ for &row_col in pairs {
+ let row_index = row_col >> 6;
+ let col_index = row_col & 63;
+ if row_index != predicted_row_index {
+ predicted_col_index = 0;
}
- }
-
- fn compress_sparse_flavor(&mut self, source: &CpcSketch) {
- debug_assert!(source.sliding_window.is_empty());
- let mut pairs = source.surprising_value_table().unwrapping_get_items();
- pairs.sort_unstable();
- self.compress_surprising_values(&pairs, source.lg_k());
- }
+ assert!(row_index >= predicted_row_index);
+ assert!(col_index >= predicted_col_index);
- fn compress_hybrid_flavor(&mut self, source: &CpcSketch) {
- debug_assert!(!source.sliding_window.is_empty());
- debug_assert_eq!(source.window_offset, 0);
-
- let k = 1 << source.lg_k();
- let mut pairs = source.surprising_value_table().unwrapping_get_items();
- pairs.sort_unstable();
- let num_pairs_from_table = pairs.len();
- let num_pairs_from_window = (source.num_coupons() as usize) -
num_pairs_from_table;
-
- let all_pairs_len = num_pairs_from_table + num_pairs_from_window;
- let mut all_pairs = vec![0; all_pairs_len];
- // tricky read pairs from sliding_window
- {
- // The empty space that this leaves at the beginning of the output
array will be filled
- // later.
- let mut idx = num_pairs_from_table;
- for row_index in 0..k {
- let mut byte = source.sliding_window[row_index];
- while byte != 0 {
- let col_index = byte.trailing_zeros();
- byte ^= 1 << col_index; // erase the 1
- all_pairs[idx] = ((row_index << 6) as u32) | col_index;
- idx += 1;
- }
- }
- assert_eq!(idx, all_pairs_len);
- }
- // two-way merge of pairs_from_table and pairs_from_window into
all_pairs
- {
- let mut final_idx = 0;
- let mut table_idx = 0;
- let mut window_idx = num_pairs_from_table;
-
- while final_idx < all_pairs_len {
- if table_idx < num_pairs_from_table
- && (window_idx >= all_pairs_len || pairs[table_idx] <=
all_pairs[window_idx])
- {
- all_pairs[final_idx] = pairs[table_idx];
- table_idx += 1;
- } else {
- all_pairs[final_idx] = all_pairs[window_idx];
- window_idx += 1;
- }
- final_idx += 1;
- }
- }
+ let y_delta = row_index - predicted_row_index;
+ let x_delta = col_index - predicted_col_index;
+ predicted_row_index = row_index;
+ predicted_col_index = col_index + 1;
- self.compress_surprising_values(&all_pairs, source.lg_k());
+ let code_info = LENGTH_LIMITED_UNARY_ENCODING_TABLE65[x_delta as
usize];
+ bits.write(u64::from(code_info & 0xfff), (code_info >> 12) as u8);
+ bits.write_unary(u64::from(y_delta >> num_base_bits));
+ bits.write(u64::from(y_delta & golomb_lo_mask), num_base_bits);
}
- fn compress_pinned_flavor(&mut self, source: &CpcSketch) {
- self.compress_sliding_window(&source.sliding_window, source.lg_k(),
source.num_coupons());
- let mut pairs = source.surprising_value_table().unwrapping_get_items();
- if !pairs.is_empty() {
- // Here we subtract 8 from the column indices. Because they are
stored in the low 6 bits
- // of each row_col pair, and because no column index is less than
8 for a "Pinned"
- // sketch, we can simply subtract 8 from the pairs themselves.
-
- // shift the columns over by 8 positions before compressing
(because of the window)
- for pair in &mut pairs {
- assert!(*pair & 63 >= 8, "pair column index is less than 8:
{pair}");
- *pair -= 8;
- }
+ bits.pad(10u8.saturating_sub(num_base_bits));
+ bits.finish()
+}
- pairs.sort_unstable();
- self.compress_surprising_values(&pairs, source.lg_k());
- }
+pub(super) fn encode_window(
+ window: &[u8],
+ lg_k: u8,
+ num_coupons: u32,
+ output: &mut SketchBytes,
+) -> usize {
+ let pseudo_phase = determine_pseudo_phase(lg_k, num_coupons);
+ let encoding_table = &ENCODING_TABLES_FOR_HIGH_ENTROPY_BYTE[pseudo_phase
as usize];
+ let mut bits = BitWriter::new(output);
+ for &byte in window {
+ let code_info = encoding_table[byte as usize];
+ bits.write(u64::from(code_info & 0xfff), (code_info >> 12) as u8);
}
+ bits.pad(11);
+ bits.finish()
+}
- // Complicated by the existence of both a left fringe and a right fringe.
- fn compress_sliding_flavor(&mut self, source: &CpcSketch) {
- self.compress_sliding_window(&source.sliding_window, source.lg_k(),
source.num_coupons());
- let mut pairs = source.surprising_value_table().unwrapping_get_items();
- if !pairs.is_empty() {
- // Here we apply a complicated transformation to the column
indices, which
- // changes the implied ordering of the pairs, so we must do it
before sorting.
-
- let pseudo_phase = determine_pseudo_phase(source.lg_k(),
source.num_coupons());
- let permutation = &COLUMN_PERMUTATIONS_FOR_ENCODING[pseudo_phase
as usize];
- let offset = source.window_offset;
- debug_assert!(offset <= 56);
- for pair in &mut pairs {
- let row_col = *pair;
- let row = row_col >> 6;
- let mut col = (row_col & 63) as u8;
- // first rotate the columns into a canonical configuration:
- // new = ((old - (offset+8)) + 64) mod 64
- col = (col + 56 - offset) & 63;
- debug_assert!(col < 56);
- // then apply the permutation
- col = permutation[col as usize];
- *pair = (row << 6) | (col as u32);
- }
+struct BitWriter<'a> {
+ output: &'a mut SketchBytes,
+ buffer: u64,
+ buffered_bits: u8,
+ words_written: usize,
+}
- pairs.sort_unstable();
- self.compress_surprising_values(&pairs, source.lg_k());
+impl<'a> BitWriter<'a> {
+ fn new(output: &'a mut SketchBytes) -> Self {
+ Self {
+ output,
+ buffer: 0,
+ buffered_bits: 0,
+ words_written: 0,
}
}
- fn compress_surprising_values(&mut self, pairs: &[u32], lg_k: u8) {
- let k = 1 << lg_k;
- let num_pairs = pairs.len() as u32;
- let num_base_bits = golomb_choose_number_of_base_bits(k + num_pairs,
num_pairs as u64);
- let table_len = safe_length_for_compressed_pair_buf(k, num_pairs,
num_base_bits);
- self.table_data.resize(table_len, 0);
-
- let compressed_surprising_values =
self.low_level_compress_pairs(pairs, num_base_bits);
-
- // At this point we could free the unused portion of the compression
output buffer,
- // but it is not necessary if it is temporary
- // Note: realloc caused strange timing spikes for lgK = 11 and 12.
-
- self.table_data_words = compressed_surprising_values;
- self.table_num_entries = num_pairs;
+ fn write(&mut self, value: u64, count: u8) {
+ self.buffer |= value << self.buffered_bits;
+ self.buffered_bits += count;
+ self.flush_full_word();
}
- fn compress_sliding_window(&mut self, window: &[u8], lg_k: u8,
num_coupons: u32) {
- let k = 1 << lg_k;
- let window_buf_len = safe_length_for_compressed_window_buf(k);
- self.window_data.resize(window_buf_len, 0);
- let pseudo_phase = determine_pseudo_phase(lg_k, num_coupons);
- let data_words = self.low_level_compress_bytes(
- window,
- k,
- &ENCODING_TABLES_FOR_HIGH_ENTROPY_BYTE[pseudo_phase as usize],
- );
-
- // At this point we could free the unused portion of the compression
output buffer,
- // but it is not necessary if it is temporary
- // Note: realloc caused strange timing spikes for lgK = 11 and 12.
-
- self.window_data_words = data_words;
- }
-
- /// Returns the number of compressed words that were actually used.
- ///
- /// It is the caller's responsibility to ensure that the window_data is
long enough.
- fn low_level_compress_bytes(
- &mut self,
- byte_array: &[u8],
- num_bytes_to_encode: u32,
- encoding_table: &[u16],
- ) -> usize {
- // bits are packed into this first, then are flushed to window_data
- let mut bitbuf = 0;
- // number of bits currently in bitbuf; must be between 0 and 31
- let mut bufbits = 0;
- let mut next_word_index = 0;
-
- for byte_index in 0..num_bytes_to_encode {
- let code_info = encoding_table[byte_array[byte_index as usize] as
usize];
- let code_val = (code_info & 0xfff) as u64;
- let code_len = (code_info >> 12) as u8;
- bitbuf |= code_val << bufbits;
- bufbits += code_len;
- maybe_flush_bitbuf(
- &mut bitbuf,
- &mut bufbits,
- &mut self.window_data,
- &mut next_word_index,
- );
- }
-
- // Pad the bitstream with 11 zero-bits so that the decompressor's
12-bit peek can't overrun
- // its input.
- bufbits += 11;
- maybe_flush_bitbuf(
- &mut bitbuf,
- &mut bufbits,
- &mut self.window_data,
- &mut next_word_index,
- );
-
- if bufbits > 0 {
- // We are done encoding now, so we flush the bit buffer.
- debug_assert!(bufbits < 32);
- self.window_data[next_word_index] = (bitbuf & 0xffffffff) as u32;
- next_word_index += 1;
-
- // not really necessary unset since no more use
- //bitbuf = 0;
- //bufbits = 0;
+ fn write_unary(&mut self, value: u64) {
+ let mut remaining = value;
+ while remaining >= 16 {
+ self.pad(16);
+ remaining -= 16;
}
-
- next_word_index
+ self.write(1 << remaining, (remaining + 1) as u8);
}
- /// Returns the number of table_data actually used.
- ///
- /// Here "pairs" refers to row/column pairs that specify the positions of
surprising values in
- /// the bit matrix.
- fn low_level_compress_pairs(&mut self, pairs: &[u32], num_base_bits: u8)
-> usize {
- let mut bitbuf = 0;
- let mut bufbits = 0;
- let mut next_word_index = 0;
- let golomb_lo_mask = ((1 << num_base_bits) - 1) as u64;
- let mut predicted_row_index = 0;
- let mut predicted_col_index = 0;
-
- for pair_index in 0..pairs.len() {
- let row_col = pairs[pair_index];
- let row_index = row_col >> 6;
- let col_index = row_col & 63;
-
- if row_index != predicted_row_index {
- predicted_col_index = 0;
- }
-
- assert!(row_index >= predicted_row_index);
- assert!(col_index >= predicted_col_index);
-
- let y_delta = row_index - predicted_row_index;
- let x_delta = col_index - predicted_col_index;
-
- predicted_row_index = row_index;
- predicted_col_index = col_index + 1;
-
- let code_info = LENGTH_LIMITED_UNARY_ENCODING_TABLE65[x_delta as
usize];
- let code_val = (code_info & 0xfff) as u64;
- let code_len = (code_info >> 12) as u8;
- bitbuf |= code_val << bufbits;
- bufbits += code_len;
-
- maybe_flush_bitbuf(
- &mut bitbuf,
- &mut bufbits,
- &mut self.table_data,
- &mut next_word_index,
- );
-
- let golomb_lo = (y_delta as u64) & golomb_lo_mask;
- let golomb_hi = (y_delta as u64) >> num_base_bits;
- write_unary(
- &mut self.table_data,
- &mut next_word_index,
- &mut bitbuf,
- &mut bufbits,
- golomb_hi,
- );
-
- bitbuf |= golomb_lo << bufbits;
- bufbits += num_base_bits;
- maybe_flush_bitbuf(
- &mut bitbuf,
- &mut bufbits,
- &mut self.table_data,
- &mut next_word_index,
- );
- }
-
- // Pad the bitstream so that the decompressor's 12-bit peek can't
overrun its input.
- let padding = 10u8.saturating_sub(num_base_bits);
- bufbits += padding;
- maybe_flush_bitbuf(
- &mut bitbuf,
- &mut bufbits,
- &mut self.table_data,
- &mut next_word_index,
- );
-
- if bufbits > 0 {
- // We are done encoding now, so we flush the bit buffer
- assert!(bufbits < 32);
- self.table_data[next_word_index] = (bitbuf & 0xffffffff) as u32;
- next_word_index += 1;
-
- // not really necessary unset since no more use
- //bitbuf = 0;
- //bufbits = 0;
- }
-
- next_word_index
+ fn pad(&mut self, count: u8) {
+ self.buffered_bits += count;
+ self.flush_full_word();
}
-}
-
-pub(super) struct UncompressedState {
- pub(super) table: PairTable,
- pub(super) window: Vec<u8>,
-}
-impl CompressedState {
- pub fn uncompress(&self, lg_k: u8, num_coupons: u32) ->
Result<UncompressedState, Error> {
- match determine_flavor(lg_k, num_coupons) {
- Flavor::Empty => Ok(UncompressedState {
- table: PairTable::new(2, lg_k + 6),
- window: vec![],
- }),
- Flavor::Sparse => self.uncompress_sparse_flavor(lg_k),
- Flavor::Hybrid => self.uncompress_hybrid_flavor(lg_k),
- Flavor::Pinned => self.uncompress_pinned_flavor(lg_k, num_coupons),
- Flavor::Sliding => self.uncompress_sliding_flavor(lg_k,
num_coupons),
+ fn flush_full_word(&mut self) {
+ if self.buffered_bits >= 32 {
+ self.output.write_u32_le(self.buffer as u32);
+ self.words_written += 1;
+ self.buffer >>= 32;
+ self.buffered_bits -= 32;
}
}
- fn uncompress_sparse_flavor(&self, lg_k: u8) -> Result<UncompressedState,
Error> {
- debug_assert!(self.window_data.is_empty(), "window is not expected");
-
- let pairs = uncompress_surprising_values(
- &self.table_data,
- self.table_data_words,
- self.table_num_entries,
- lg_k,
- )?;
-
- Ok(UncompressedState {
- table: PairTable::from_slots(lg_k, self.table_num_entries, pairs)?,
- window: vec![],
- })
- }
-
- fn uncompress_hybrid_flavor(&self, lg_k: u8) -> Result<UncompressedState,
Error> {
- debug_assert!(self.window_data.is_empty(), "window is not expected");
-
- let mut pairs = uncompress_surprising_values(
- &self.table_data,
- self.table_data_words,
- self.table_num_entries,
- lg_k,
- )?;
-
- // In the hybrid flavor, some of these pairs actually belong in the
window, so we will
- // separate them out, moving the "true" pairs to the bottom of the
array.
- let k = 1 << lg_k;
- let mut window = vec![0u8; k]; // important: zero the memory
- let mut next_true_pair = 0;
- for i in 0..self.table_num_entries {
- let row_col = pairs[i as usize];
- let col = row_col & 63;
- if col < 8 {
- let row = row_col >> 6;
- window[row as usize] |= 1 << col; // set the window bit
- } else {
- pairs[next_true_pair as usize] = row_col;
- next_true_pair += 1;
- }
+ fn finish(mut self) -> usize {
+ if self.buffered_bits > 0 {
+ debug_assert!(self.buffered_bits < 32);
+ self.output.write_u32_le(self.buffer as u32);
+ self.words_written += 1;
}
-
- Ok(UncompressedState {
- table: PairTable::from_slots(lg_k, next_true_pair, pairs)?,
- window,
- })
- }
-
- fn uncompress_pinned_flavor(
- &self,
- lg_k: u8,
- num_coupons: u32,
- ) -> Result<UncompressedState, Error> {
- let mut window = vec![];
- uncompress_sliding_window(
- &self.window_data,
- self.window_data_words,
- &mut window,
- lg_k,
- num_coupons,
- )?;
- let num_pairs = self.table_num_entries;
- let table = if num_pairs == 0 {
- PairTable::new(2, lg_k + 6)
- } else {
- let mut pairs = uncompress_surprising_values(
- &self.table_data,
- self.table_data_words,
- num_pairs,
- lg_k,
- )?;
- // undo the compressor's 8-column shift
- for i in 0..num_pairs {
- let i = i as usize;
- if (pairs[i] & 63) >= 56 {
- return Err(Error::deserial(format!(
- "CPC pinned table pair column index is invalid: {}",
- pairs[i]
- )));
- }
- pairs[i] += 8;
- }
- PairTable::from_slots(lg_k, num_pairs, pairs)?
- };
- Ok(UncompressedState { table, window })
- }
-
- fn uncompress_sliding_flavor(
- &self,
- lg_k: u8,
- num_coupons: u32,
- ) -> Result<UncompressedState, Error> {
- let mut window = vec![];
- uncompress_sliding_window(
- &self.window_data,
- self.window_data_words,
- &mut window,
- lg_k,
- num_coupons,
- )?;
- let num_pairs = self.table_num_entries;
- let table = if num_pairs == 0 {
- PairTable::new(2, lg_k + 6)
- } else {
- let mut pairs = uncompress_surprising_values(
- &self.table_data,
- self.table_data_words,
- num_pairs,
- lg_k,
- )?;
- let pseudo_phase = determine_pseudo_phase(lg_k, num_coupons);
- let permutation = &COLUMN_PERMUTATIONS_FOR_DECODING[pseudo_phase
as usize];
- let offset = determine_correct_offset(lg_k, num_coupons);
- if offset > 56 {
- return Err(Error::deserial(format!(
- "CPC sliding window offset is invalid: {offset}"
- )));
- }
-
- for i in 0..num_pairs {
- let i = i as usize;
- let row_col = pairs[i];
- let row = row_col >> 6;
- let col = (row_col & 63) as usize;
- // first undo the permutation
- if col >= permutation.len() {
- return Err(Error::deserial(format!(
- "CPC sliding table pair column index is invalid: {}",
- pairs[i]
- )));
- }
- let mut col = permutation[col];
- // then undo the rotation: old = (new + (offset+8)) mod 64
- col = (col + (offset + 8)) & 63;
- pairs[i] = (row << 6) | (col as u32);
- }
- PairTable::from_slots(lg_k, num_pairs, pairs)?
- };
- Ok(UncompressedState { table, window })
+ self.words_written
}
}
-fn uncompress_surprising_values(
- data: &[u32],
- data_words: usize,
- num_pairs: u32,
- lg_k: u8,
-) -> Result<Vec<u32>, Error> {
+pub(super) fn decode_pairs(data: &[u8], num_pairs: u32, lg_k: u8) ->
Result<Vec<u32>, Error> {
if num_pairs == 0 {
return Ok(vec![]);
}
let k = 1 << lg_k;
let mut pairs = vec![0; num_pairs as usize];
let num_base_bits = golomb_choose_number_of_base_bits(k + num_pairs,
num_pairs as u64);
- low_level_uncompress_pairs(&mut pairs, num_pairs, k, num_base_bits, data,
data_words)?;
- Ok(pairs)
-}
-
-fn uncompress_sliding_window(
- data: &[u32],
- data_words: usize,
- window: &mut Vec<u8>,
- lg_k: u8,
- num_coupons: u32,
-) -> Result<(), Error> {
- let k = 1 << lg_k;
- window.resize(k, 0);
- let pseudo_phase = determine_pseudo_phase(lg_k, num_coupons);
- low_level_uncompress_bytes(
- window,
- k as u32,
- data,
- data_words,
- &DECODING_TABLES_FOR_HIGH_ENTROPY_BYTE[pseudo_phase as usize],
- )
-}
-
-fn low_level_uncompress_pairs(
- pairs: &mut [u32],
- num_pairs_to_decode: u32,
- k: u32,
- num_base_bits: u8,
- compressed_words: &[u32],
- num_compressed_words: usize,
-) -> Result<(), Error> {
- let mut bits = BitReader::new(compressed_words, num_compressed_words)?;
+ let mut bits = BitReader::new(data);
let golomb_lo_mask = (1 << num_base_bits) - 1;
let mut predicted_row_index = 0u32;
let mut predicted_col_index = 0u32;
@@ -565,7 +147,7 @@ fn low_level_uncompress_pairs(
// y_delta_hi (unary)
// y_delta_lo (basebits)
- for pair_index in 0..num_pairs_to_decode {
+ for pair_index in 0..num_pairs {
let peek12 = bits.peek(12)?;
let lookup = LENGTH_LIMITED_UNARY_DECODING_TABLE65[peek12 as usize];
let code_word_length = (lookup >> 8) as u8;
@@ -602,58 +184,52 @@ fn low_level_uncompress_pairs(
predicted_row_index = row_index;
predicted_col_index = col_index + 1;
}
- Ok(())
+ Ok(pairs)
}
-fn low_level_uncompress_bytes(
- byte_array: &mut [u8],
- num_bytes_to_decode: u32,
- compressed_words: &[u32],
- num_compressed_words: usize,
- decoding_table: &[u16],
-) -> Result<(), Error> {
- let mut bits = BitReader::new(compressed_words, num_compressed_words)?;
+pub(super) fn decode_window(data: &[u8], lg_k: u8, num_coupons: u32) ->
Result<Vec<u8>, Error> {
+ let mut window = vec![0; 1 << lg_k];
+ let pseudo_phase = determine_pseudo_phase(lg_k, num_coupons);
+ let decoding_table = &DECODING_TABLES_FOR_HIGH_ENTROPY_BYTE[pseudo_phase
as usize];
+ let mut bits = BitReader::new(data);
- for byte_index in 0..num_bytes_to_decode {
+ for byte in &mut window {
// These 12 bits will include an entire Huffman codeword.
let peek12 = bits.peek(12)?;
let lookup = decoding_table[peek12 as usize];
let code_word_length = (lookup >> 8) as u8;
- let decoded_byte = (lookup & 0xff) as u8;
- byte_array[byte_index as usize] = decoded_byte;
+ *byte = (lookup & 0xff) as u8;
bits.consume(code_word_length);
}
- Ok(())
+ Ok(window)
}
struct BitReader<'a> {
- words: &'a [u32],
- next_word: usize,
+ bytes: &'a [u8],
+ next_byte: usize,
buffer: u64,
buffered_bits: u8,
}
impl<'a> BitReader<'a> {
- fn new(words: &'a [u32], num_words: usize) -> Result<Self, Error> {
- let words = words
- .get(..num_words)
- .ok_or_else(|| Error::deserial("CPC compressed word count exceeds
payload"))?;
- Ok(Self {
- words,
- next_word: 0,
+ fn new(bytes: &'a [u8]) -> Self {
+ Self {
+ bytes,
+ next_byte: 0,
buffer: 0,
buffered_bits: 0,
- })
+ }
}
fn fill(&mut self, needed: u8) -> Result<(), Error> {
if self.buffered_bits < needed {
let word = self
- .words
- .get(self.next_word)
+ .bytes
+ .get(self.next_byte..self.next_byte + size_of::<u32>())
.ok_or_else(|| Error::deserial("CPC compressed stream is
truncated"))?;
- self.buffer |= u64::from(*word) << self.buffered_bits;
- self.next_word += 1;
+ let word = u32::from_le_bytes(word.try_into().unwrap());
+ self.buffer |= u64::from(word) << self.buffered_bits;
+ self.next_byte += size_of::<u32>();
self.buffered_bits += 32;
}
Ok(())
@@ -698,7 +274,7 @@ impl<'a> BitReader<'a> {
}
}
-fn determine_pseudo_phase(lg_k: u8, num_coupons: u32) -> u8 {
+pub(super) fn determine_pseudo_phase(lg_k: u8, num_coupons: u32) -> u8 {
let k = 1u64 << lg_k;
let num_coupons = u64::from(num_coupons);
// This mid-range logic produces pseudo-phases. They are used to select
encoding tables.
@@ -735,82 +311,6 @@ fn determine_pseudo_phase(lg_k: u8, num_coupons: u32) ->
u8 {
}
}
-fn write_unary(
- compressed_words: &mut [u32],
- next_word_index: &mut usize,
- bitbuf: &mut u64,
- bufbits: &mut u8,
- value: u64,
-) {
- assert!(*bufbits <= 31);
-
- let mut remaining = value;
- while remaining >= 16 {
- remaining -= 16;
- // Here we output 16 zeros, but we don't need to physically write them
into bitbuf
- // because it already contains zeros in that region.
- *bufbits += 16; // Record the fact that 16 bits of output have
occurred.
- maybe_flush_bitbuf(bitbuf, bufbits, compressed_words, next_word_index);
- }
-
- let the_unary_code = 1 << remaining;
- *bitbuf |= the_unary_code << *bufbits;
- *bufbits += (remaining + 1) as u8;
- maybe_flush_bitbuf(bitbuf, bufbits, compressed_words, next_word_index);
-}
-
-fn maybe_flush_bitbuf(
- bitbuf: &mut u64,
- bufbits: &mut u8,
- word: &mut [u32],
- word_index: &mut usize,
-) {
- if *bufbits >= 32 {
- word[*word_index] = (*bitbuf & 0xffffffff) as u32;
- *word_index += 1;
- *bitbuf >>= 32;
- *bufbits -= 32;
- }
-}
-
-// Explanation of padding: we write
-// 1) xdelta (huffman, provides at least 1 bit, requires 12-bit lookahead)
-// 2) ydeltaGolombHi (unary, provides at least 1 bit, requires 8-bit lookahead)
-// 3) ydeltaGolombLo (straight B bits).
-// So the 12-bit lookahead is the tight constraint, but there are at least (2
+ B) bits emitted,
-// so we would be safe with max (0, 10 - B) bits of padding at the end of the
bitstream.
-fn safe_length_for_compressed_window_buf(k: u32) -> usize {
- // 11 bits of padding, due to 12-bit lookahead, with 1 bit certainly
present.
- let bits = 12 * k + 11;
- divide_longs_rounding_up(bits as usize, 32)
-}
-
-fn safe_length_for_compressed_pair_buf(k: u32, num_pairs: u32, num_base_bits:
u8) -> usize {
- // Long ybits = k + numPairs; // simpler and safer UB
- // The following tighter UB on ybits is based on page 198
- // of the textbook "Managing Gigabytes" by Witten, Moffat, and Bell.
- // Notice that if numBaseBits == 0 it coincides with (k + numPairs).
-
- let k = k as usize;
- let num_pairs = num_pairs as usize;
- let num_base_bits = num_base_bits as usize;
-
- let ybits = num_pairs * (1 + num_base_bits) + (k >> num_base_bits);
- let xbits = 12 * (num_pairs);
- let padding = 10usize.saturating_sub(num_base_bits);
- divide_longs_rounding_up(xbits + ybits + padding, 32)
-}
-
-fn divide_longs_rounding_up(x: usize, y: usize) -> usize {
- debug_assert_ne!(y, 0);
- let quotient = x / y;
- if quotient * y == x {
- quotient
- } else {
- quotient + 1
- }
-}
-
/// Returns an integer that is between zero and ceil(log_2(k)) - 1, inclusive.
fn golomb_choose_number_of_base_bits(k: u32, count: u64) -> u8 {
debug_assert!(k > 0);
@@ -841,9 +341,10 @@ fn floor_log2_of_long(x: u64) -> u8 {
#[cfg(test)]
mod tests {
- use super::CompressedState;
+ use super::decode_pairs;
use super::determine_pseudo_phase;
- use super::uncompress_surprising_values;
+ use super::encode_pairs;
+ use crate::codec::SketchBytes;
#[test]
fn pseudo_phase_handles_maximum_lg_k() {
@@ -853,15 +354,11 @@ mod tests {
#[test]
fn pair_decoder_rejects_empty_table_sentinel() {
- let mut compressed = CompressedState::default();
- compressed.compress_surprising_values(&[u32::MAX], 26);
-
- let error = uncompress_surprising_values(
- &compressed.table_data,
- compressed.table_data_words,
- 1,
- 26,
- );
+ let mut compressed = SketchBytes::with_capacity(16);
+ encode_pairs(&[u32::MAX], 26, &mut compressed);
+ let compressed = compressed.into_bytes();
+
+ let error = decode_pairs(&compressed, 1, 26);
assert_eq!(
error.unwrap_err().message(),
"CPC pair uses the reserved empty-table sentinel"
diff --git a/datasketches/src/cpc/sketch.rs b/datasketches/src/cpc/sketch.rs
index 2ae9164..96a48fb 100644
--- a/datasketches/src/cpc/sketch.rs
+++ b/datasketches/src/cpc/sketch.rs
@@ -29,7 +29,13 @@ use crate::cpc::DEFAULT_LG_K;
use crate::cpc::Flavor;
use crate::cpc::MAX_LG_K;
use crate::cpc::MIN_LG_K;
-use crate::cpc::compression::CompressedState;
+use crate::cpc::compression::decode_pairs;
+use crate::cpc::compression::decode_window;
+use crate::cpc::compression::determine_pseudo_phase;
+use crate::cpc::compression::encode_pairs;
+use crate::cpc::compression::encode_window;
+use crate::cpc::compression_data::COLUMN_PERMUTATIONS_FOR_DECODING;
+use crate::cpc::compression_data::COLUMN_PERMUTATIONS_FOR_ENCODING;
use crate::cpc::count_bits_set_in_matrix;
use crate::cpc::determine_correct_offset;
use crate::cpc::determine_flavor;
@@ -468,15 +474,93 @@ impl CpcSketch {
impl CpcSketch {
/// Serializes this `CpcSketch` to bytes.
pub fn serialize(&self) -> Vec<u8> {
- let mut bytes = SketchBytes::with_capacity(256);
-
- let mut compressed = CompressedState::default();
- compressed.compress(self);
-
+ let flavor = self.flavor();
let has_hip = !self.merge_flag;
- let has_table = !compressed.table_data.is_empty();
- let has_window = !compressed.window_data.is_empty();
+ let has_window = matches!(flavor, Flavor::Pinned | Flavor::Sliding);
+ let mut pairs = match flavor {
+ Flavor::Empty => vec![],
+ Flavor::Sparse => {
+ debug_assert!(self.sliding_window.is_empty());
+ self.surprising_value_table().unwrapping_get_items()
+ }
+ Flavor::Hybrid => {
+ debug_assert!(!self.sliding_window.is_empty());
+ debug_assert_eq!(self.window_offset, 0);
+
+ let mut table_pairs =
self.surprising_value_table().unwrapping_get_items();
+ table_pairs.sort_unstable();
+ let num_table_pairs = table_pairs.len();
+ let mut all_pairs = vec![0; self.num_coupons as usize];
+
+ let mut index = num_table_pairs;
+ for (row_index, &window_byte) in
self.sliding_window.iter().enumerate() {
+ let mut window_byte = window_byte;
+ while window_byte != 0 {
+ let col_index = window_byte.trailing_zeros();
+ window_byte ^= 1 << col_index;
+ all_pairs[index] = ((row_index << 6) as u32) |
col_index;
+ index += 1;
+ }
+ }
+ assert_eq!(index, all_pairs.len());
+
+ let mut table_index = 0;
+ let mut window_index = num_table_pairs;
+ for final_index in 0..all_pairs.len() {
+ if table_index < num_table_pairs
+ && (window_index >= all_pairs.len()
+ || table_pairs[table_index] <=
all_pairs[window_index])
+ {
+ all_pairs[final_index] = table_pairs[table_index];
+ table_index += 1;
+ } else {
+ all_pairs[final_index] = all_pairs[window_index];
+ window_index += 1;
+ }
+ }
+ all_pairs
+ }
+ Flavor::Pinned => {
+ let mut pairs =
self.surprising_value_table().unwrapping_get_items();
+ for pair in &mut pairs {
+ assert!(*pair & 63 >= 8, "pair column index is less than
8: {pair}");
+ *pair -= 8;
+ }
+ pairs
+ }
+ Flavor::Sliding => {
+ let mut pairs =
self.surprising_value_table().unwrapping_get_items();
+ let pseudo_phase = determine_pseudo_phase(self.lg_k,
self.num_coupons);
+ let permutation =
&COLUMN_PERMUTATIONS_FOR_ENCODING[pseudo_phase as usize];
+ debug_assert!(self.window_offset <= 56);
+ for pair in &mut pairs {
+ let row = *pair >> 6;
+ let col = ((*pair & 63) as u8 + 56 - self.window_offset) &
63;
+ debug_assert!(col < 56);
+ *pair = (row << 6) | u32::from(permutation[col as usize]);
+ }
+ pairs
+ }
+ };
+ pairs.sort_unstable();
+
+ let table_num_entries = pairs.len() as u32;
+ let mut payload = SketchBytes::with_capacity(if self.is_empty() { 0 }
else { 256 });
+ let window_words = has_window.then(|| {
+ encode_window(
+ &self.sliding_window,
+ self.lg_k,
+ self.num_coupons,
+ &mut payload,
+ )
+ });
+ let table_words =
+ (!pairs.is_empty()).then(|| encode_pairs(&pairs, self.lg_k, &mut
payload));
+ let payload = payload.into_bytes();
+
+ let has_table = table_words.is_some();
let preamble_ints = make_preamble_ints(self.num_coupons, has_hip,
has_table, has_window);
+ let mut bytes = SketchBytes::with_capacity(40 + payload.len());
bytes.write_u8(preamble_ints);
bytes.write_u8(SERIAL_VERSION);
bytes.write_u8(Family::CPC.id);
@@ -493,35 +577,26 @@ impl CpcSketch {
bytes.write_u32_le(self.num_coupons);
if has_table && has_window {
// if there is no window it is the same as number of coupons
- bytes.write_u32_le(compressed.table_num_entries);
+ bytes.write_u32_le(table_num_entries);
// HIP values can be in two different places in the sequence
of fields
// this is the first HIP decision point
if has_hip {
self.write_hip(&mut bytes);
}
}
- if has_table {
- debug_assert!(compressed.table_data_words <= u32::MAX as
usize);
- bytes.write_u32_le(compressed.table_data_words as u32);
+ if let Some(table_words) = table_words {
+ debug_assert!(table_words <= u32::MAX as usize);
+ bytes.write_u32_le(table_words as u32);
}
- if has_window {
- debug_assert!(compressed.window_data_words <= u32::MAX as
usize);
- bytes.write_u32_le(compressed.window_data_words as u32);
+ if let Some(window_words) = window_words {
+ debug_assert!(window_words <= u32::MAX as usize);
+ bytes.write_u32_le(window_words as u32);
}
// this is the second HIP decision point
if has_hip && !(has_table && has_window) {
self.write_hip(&mut bytes);
}
- if has_window {
- for i in 0..compressed.window_data_words {
- bytes.write_u32_le(compressed.window_data[i]);
- }
- }
- if has_table {
- for i in 0..compressed.table_data_words {
- bytes.write_u32_le(compressed.table_data[i]);
- }
- }
+ bytes.write(&payload);
}
bytes.into_bytes()
}
@@ -561,8 +636,10 @@ impl CpcSketch {
let has_table = flags & (1 << FLAG_HAS_TABLE) != 0;
let has_window = flags & (1 << FLAG_HAS_WINDOW) != 0;
- let mut compressed = CompressedState::default();
let mut num_coupons = 0;
+ let mut table_num_entries = 0;
+ let mut table_data_words = 0;
+ let mut window_data_words = 0;
let mut kxp = 0.0;
let mut hip_est_accum = 0.0;
@@ -571,7 +648,7 @@ impl CpcSketch {
.read_u32_le()
.map_err(insufficient_data("num_coupons"))?;
if has_table && has_window {
- compressed.table_num_entries = cursor
+ table_num_entries = cursor
.read_u32_le()
.map_err(insufficient_data("table_num_entries"))?;
if has_hip {
@@ -582,13 +659,13 @@ impl CpcSketch {
}
}
if has_table {
- compressed.table_data_words = cursor
+ table_data_words = cursor
.read_u32_le()
.map_err(insufficient_data("table_data_words"))?
as usize;
}
if has_window {
- compressed.window_data_words = cursor
+ window_data_words = cursor
.read_u32_le()
.map_err(insufficient_data("window_data_words"))?
as usize;
@@ -599,24 +676,8 @@ impl CpcSketch {
.read_f64_le()
.map_err(insufficient_data("hip_est_accum"))?;
}
- if has_window {
- for _ in 0..compressed.window_data_words {
- let word = cursor
- .read_u32_le()
- .map_err(insufficient_data("window_data"))?;
- compressed.window_data.push(word);
- }
- }
- if has_table {
- for _ in 0..compressed.table_data_words {
- let word = cursor
- .read_u32_le()
- .map_err(insufficient_data("table_data"))?;
- compressed.table_data.push(word);
- }
- }
if !has_window {
- compressed.table_num_entries = num_coupons;
+ table_num_entries = num_coupons;
}
}
@@ -668,41 +729,128 @@ impl CpcSketch {
}
// The number of stored table entries can never exceed the number of
coupons.
- if compressed.table_num_entries > num_coupons {
+ if table_num_entries > num_coupons {
return Err(Error::deserial(format!(
"table_num_entries ({}) exceeds num_coupons ({})",
- compressed.table_num_entries, num_coupons
+ table_num_entries, num_coupons
)));
}
// A pair requires at least one bit to encode, so the declared number
of table entries can
// never exceed the number of bits available in the table data. This
also bounds the size
// of the allocation made while decoding, rejecting corrupt inputs
that claim an enormous
// entry count backed by only a few data words.
- if (compressed.table_num_entries as usize) >
compressed.table_data_words.saturating_mul(32)
- {
+ if (table_num_entries as usize) > table_data_words.saturating_mul(32) {
return Err(Error::deserial(format!(
"table_num_entries ({}) exceeds capacity of table data ({}
words)",
- compressed.table_num_entries, compressed.table_data_words
+ table_num_entries, table_data_words
)));
}
let k = 1usize << lg_k;
- if has_window && compressed.window_data_words.saturating_mul(32) < k {
+ if has_window && window_data_words.saturating_mul(32) < k {
return Err(Error::deserial(format!(
"window data ({} words) is too short for lg_k = {lg_k}",
- compressed.window_data_words
+ window_data_words
)));
}
- let uncompressed = compressed.uncompress(lg_k, num_coupons)?;
+ let window_data_bytes = window_data_words.checked_mul(4).ok_or_else(||
{
+ Error::deserial("CPC window data word count overflows payload
length")
+ })?;
+ let table_data_bytes = table_data_words
+ .checked_mul(4)
+ .ok_or_else(|| Error::deserial("CPC table data word count
overflows payload length"))?;
+ let payload_bytes = window_data_bytes
+ .checked_add(table_data_bytes)
+ .ok_or_else(|| Error::deserial("CPC payload length overflows"))?;
+ let payload = cursor
+ .remaining()
+ .get(..payload_bytes)
+ .ok_or_else(|| Error::deserial("insufficient data for CPC
compressed payload"))?;
+ let (window_data, table_data) = payload.split_at(window_data_bytes);
+ let (table, window) = match flavor {
+ Flavor::Empty => (PairTable::new(2, lg_k + 6), vec![]),
+ Flavor::Sparse => {
+ debug_assert!(window_data.is_empty(), "window is not
expected");
+ let pairs = decode_pairs(table_data, table_num_entries, lg_k)?;
+ (
+ PairTable::from_slots(lg_k, table_num_entries, pairs)?,
+ vec![],
+ )
+ }
+ Flavor::Hybrid => {
+ debug_assert!(window_data.is_empty(), "window is not
expected");
+ let mut pairs = decode_pairs(table_data, table_num_entries,
lg_k)?;
+ let mut window = vec![0u8; 1 << lg_k];
+ let mut next_true_pair = 0;
+ for index in 0..table_num_entries as usize {
+ let row_col = pairs[index];
+ let col = row_col & 63;
+ if col < 8 {
+ window[(row_col >> 6) as usize] |= 1 << col;
+ } else {
+ pairs[next_true_pair as usize] = row_col;
+ next_true_pair += 1;
+ }
+ }
+ (PairTable::from_slots(lg_k, next_true_pair, pairs)?, window)
+ }
+ Flavor::Pinned => {
+ let window = decode_window(window_data, lg_k, num_coupons)?;
+ let table = if table_num_entries == 0 {
+ PairTable::new(2, lg_k + 6)
+ } else {
+ let mut pairs = decode_pairs(table_data,
table_num_entries, lg_k)?;
+ for pair in &mut pairs {
+ if (*pair & 63) >= 56 {
+ return Err(Error::deserial(format!(
+ "CPC pinned table pair column index is
invalid: {pair}"
+ )));
+ }
+ *pair += 8;
+ }
+ PairTable::from_slots(lg_k, table_num_entries, pairs)?
+ };
+ (table, window)
+ }
+ Flavor::Sliding => {
+ let window = decode_window(window_data, lg_k, num_coupons)?;
+ let table = if table_num_entries == 0 {
+ PairTable::new(2, lg_k + 6)
+ } else {
+ let mut pairs = decode_pairs(table_data,
table_num_entries, lg_k)?;
+ let pseudo_phase = determine_pseudo_phase(lg_k,
num_coupons);
+ let permutation =
&COLUMN_PERMUTATIONS_FOR_DECODING[pseudo_phase as usize];
+ let offset = determine_correct_offset(lg_k, num_coupons);
+ if offset > 56 {
+ return Err(Error::deserial(format!(
+ "CPC sliding window offset is invalid: {offset}"
+ )));
+ }
+ for pair in &mut pairs {
+ let row = *pair >> 6;
+ let col = (*pair & 63) as usize;
+ if col >= permutation.len() {
+ return Err(Error::deserial(format!(
+ "CPC sliding table pair column index is
invalid: {pair}"
+ )));
+ }
+ let col = (permutation[col] + offset + 8) & 63;
+ *pair = (row << 6) | u32::from(col);
+ }
+ PairTable::from_slots(lg_k, table_num_entries, pairs)?
+ };
+ (table, window)
+ }
+ };
Ok(CpcSketch {
lg_k,
seed,
seed_hash,
first_interesting_column,
num_coupons,
- surprising_value_table: Some(uncompressed.table),
+ surprising_value_table: Some(table),
window_offset: determine_correct_offset(lg_k, num_coupons),
- sliding_window: uncompressed.window,
+ sliding_window: window,
merge_flag: !has_hip,
kxp,
hip_est_accum,
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
