diff --git a/CHANGELOG.md b/CHANGELOG.md index 8fc9c254..85d40d54 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All significant changes to this project will be documented in this file. ### New features * Add `HllUnion::serialize` to serialize union results directly. +* Add `TupleUnion::serialize` to serialize union results without cloning summaries, which avoids per-entry allocations for heap-backed summaries. * Add immutable `XorFilter` behind the `xor` feature, with 8-, 16-, and 32-bit fingerprints, construction from values with automatic deduplication or borrowed distinct precomputed hashes, bounded seeded retries and compact temporary storage, optional cardinality metadata for imported filters, standard serialization for all three widths (Java-compatible for 8/16-bit filters), and validated imports for application-defined storage. ### Improvements diff --git a/benchmarks/tuple/mod.rs b/benchmarks/tuple/mod.rs index 4839176b..32c8b7f2 100644 --- a/benchmarks/tuple/mod.rs +++ b/benchmarks/tuple/mod.rs @@ -15,8 +15,10 @@ // specific language governing permissions and limitations // under the License. +mod serde; mod update; pub fn register(runner: &mut benchmarks::Runner) { + serde::register(runner); update::register(runner); } diff --git a/benchmarks/tuple/serde.rs b/benchmarks/tuple/serde.rs new file mode 100644 index 00000000..1fcdff07 --- /dev/null +++ b/benchmarks/tuple/serde.rs @@ -0,0 +1,162 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::hint::black_box; + +use benchmarks::Bencher; +use benchmarks::Runner; +use datasketches::codec::SketchBytes; +use datasketches::codec::SketchSlice; +use datasketches::error::Error; +use datasketches::tuple::DefaultUnionPolicy; +use datasketches::tuple::DefaultUpdatePolicy; +use datasketches::tuple::SummaryCombinePolicy; +use datasketches::tuple::SummaryPolicy; +use datasketches::tuple::SummaryUpdatePolicy; +use datasketches::tuple::TupleSketchBuilder; +use datasketches::tuple::TupleSummaryValue; +use datasketches::tuple::TupleUnion; +use datasketches::tuple::TupleUnionBuilder; + +const ROWS: [u64; 4] = [0, 4, 100, 80_000]; + +fn u64_union(rows: u64) -> TupleUnion> { + let mut sketch = TupleSketchBuilder::new(DefaultUpdatePolicy::::default()) + .build() + .unwrap(); + for value in 0..rows { + sketch.update(value, 1); + } + let mut union = TupleUnionBuilder::new(DefaultUnionPolicy::::default()) + .build() + .unwrap(); + union.update(&sketch).unwrap(); + union +} + +fn serialize_union(bencher: Bencher, rows: u64) { + let union = u64_union(rows); + bencher.bench_local(|| black_box(&union).serialize(ORDERED)); +} + +fn serialize_union_via_sketch(bencher: Bencher, rows: u64) { + let union = u64_union(rows); + bencher.bench_local(|| black_box(&union).to_sketch(ORDERED).serialize()); +} + +/// Heap-backed summary holding a few short labels per key. +#[derive(Clone, Default)] +struct Labels(Vec); + +impl TupleSummaryValue for Labels { + fn serialize_size(&self) -> usize { + 4 + self.0.iter().map(|label| 4 + label.len()).sum::() + } + + fn serialize_value(&self, bytes: &mut SketchBytes) { + bytes.write_u32_le(self.0.len() as u32); + for label in &self.0 { + bytes.write_u32_le(label.len() as u32); + bytes.write(label.as_bytes()); + } + } + + fn deserialize_value(_cursor: &mut SketchSlice<'_>) -> Result { + unimplemented!("only serialization is benchmarked") + } +} + +#[derive(Clone, Copy)] +struct LabelsPolicy; + +impl SummaryPolicy for LabelsPolicy { + type Summary = Labels; + + fn create(&self) -> Self::Summary { + Labels::default() + } +} + +impl SummaryUpdatePolicy for LabelsPolicy { + fn update(&self, summary: &mut Self::Summary, value: u64) { + for i in 0..4 { + summary.0.push(format!("label-{value:08}-{i}")); + } + } +} + +impl SummaryCombinePolicy for LabelsPolicy { + fn combine(&self, summary: &mut Self::Summary, other: &Self::Summary) { + summary.0.extend(other.0.iter().cloned()); + } +} + +fn labels_union(rows: u64) -> TupleUnion { + let mut sketch = TupleSketchBuilder::new(LabelsPolicy).build().unwrap(); + for value in 0..rows { + sketch.update(value, value); + } + let mut union = TupleUnionBuilder::new(LabelsPolicy).build().unwrap(); + union.update(&sketch).unwrap(); + union +} + +fn serialize_labels_union(bencher: Bencher, rows: u64) { + let union = labels_union(rows); + bencher.bench_local(|| black_box(&union).serialize(ORDERED)); +} + +fn serialize_labels_union_via_sketch(bencher: Bencher, rows: u64) { + let union = labels_union(rows); + bencher.bench_local(|| black_box(&union).to_sketch(ORDERED).serialize()); +} + +pub fn register(runner: &mut Runner) { + for rows in ROWS { + register_order::(runner, rows); + register_order::(runner, rows); + } +} + +fn register_order(runner: &mut Runner, rows: u64) { + let order = if ORDERED { "ordered" } else { "unordered" }; + let rounds = if rows >= 100 { 500 } else { 10_000 }; + runner.compare( + &format!("tuple::serde::u64::{order}::{rows}"), + rounds, + &[ + ("via_compact_sketch", &|bencher| { + serialize_union_via_sketch::(bencher, rows) + }), + ("direct", &|bencher| { + serialize_union::(bencher, rows) + }), + ], + ); + runner.compare( + &format!("tuple::serde::labels::{order}::{rows}"), + rounds, + &[ + ("via_compact_sketch", &|bencher| { + serialize_labels_union_via_sketch::(bencher, rows) + }), + ("direct", &|bencher| { + serialize_labels_union::(bencher, rows) + }), + ], + ); +} diff --git a/datasketches/src/thetafamily/common/mod.rs b/datasketches/src/thetafamily/common/mod.rs index dbc273ee..358fb756 100644 --- a/datasketches/src/thetafamily/common/mod.rs +++ b/datasketches/src/thetafamily/common/mod.rs @@ -33,6 +33,12 @@ pub(super) trait SketchEntry { fn hash(&self) -> u64; } +impl SketchEntry for &E { + fn hash(&self) -> u64 { + E::hash(self) + } +} + pub(super) trait KeySketch: Copy { fn metadata(self) -> sketch_state::ThetaFamilySketchMetadata; diff --git a/datasketches/src/thetafamily/common/union.rs b/datasketches/src/thetafamily/common/union.rs index 4ad3b466..b9e5d1db 100644 --- a/datasketches/src/thetafamily/common/union.rs +++ b/datasketches/src/thetafamily/common/union.rs @@ -113,21 +113,37 @@ where where E: Clone, { - let Some(result_theta) = self.result_theta else { + let Some(theta) = self.result_theta else { return CompactSketchState::empty(self.table.seed_hash()); }; + let entries = self.retained_entries(theta).cloned().collect(); + self.finish_compaction(entries, theta, ordered) + } - let mut theta = result_theta.min(self.table.retention_theta()); - let mut retained_entries = if result_theta >= self.table.retention_theta() { - self.table.iter_entries().cloned().collect::>() - } else { - self.table - .iter_entries() - .filter(|entry| entry.hash() < theta) - .cloned() - .collect::>() + /// Returns the union as canonical compact-sketch state that borrows the retained entries. + #[cfg(feature = "tuple")] + pub fn to_borrowed_compact_sketch_state(&self, ordered: bool) -> CompactSketchState<&E> { + let Some(theta) = self.result_theta else { + return CompactSketchState::empty(self.table.seed_hash()); }; + let entries = self.retained_entries(theta).collect(); + self.finish_compaction(entries, theta, ordered) + } + fn retained_entries(&self, theta: u64) -> impl Iterator { + let all_retained = theta >= self.table.retention_theta(); + self.table + .iter_entries() + .filter(move |entry| all_retained || entry.hash() < theta) + } + + fn finish_compaction( + &self, + mut retained_entries: Vec, + result_theta: u64, + ordered: bool, + ) -> CompactSketchState { + let mut theta = result_theta.min(self.table.retention_theta()); let nominal_num = 1usize << self.table.lg_nom_size(); if retained_entries.len() > nominal_num { let (_, kth, _) = diff --git a/datasketches/src/thetafamily/tuple/sketch.rs b/datasketches/src/thetafamily/tuple/sketch.rs index 0fee8218..286cf597 100644 --- a/datasketches/src/thetafamily/tuple/sketch.rs +++ b/datasketches/src/thetafamily/tuple/sketch.rs @@ -22,6 +22,7 @@ //! created by a [`SummaryPolicy`] and updated through one or more [`SummaryUpdatePolicy`] //! implementations. +use std::borrow::Borrow; use std::hash::Hash; use std::slice; @@ -442,6 +443,60 @@ where } } +/// Serializes compact Tuple state whose retained entries are owned or borrowed. +pub(super) fn serialize_compact_state(compact_state: &CompactSketchState) -> Vec +where + S: TupleSummaryValue, + E: Borrow>, +{ + if compact_state.is_empty() { + return EMPTY_SKETCH_BYTES.to_vec(); + } + + let retained_entries = compact_state.retained_entries(); + let is_estimation_mode = compact_state.is_estimation_mode(); + let pre_longs = if is_estimation_mode { + 3 + } else if retained_entries.len() == 1 { + 1 + } else { + 2 + }; + let entries_size: usize = retained_entries + .iter() + .map(|entry| 8 + entry.borrow().summary().serialize_size()) + .sum(); + let mut bytes = SketchBytes::with_capacity(8 * pre_longs as usize + entries_size); + + bytes.write_u8(pre_longs); + bytes.write_u8(SERIAL_VERSION); + bytes.write_u8(Family::TUPLE.id); + bytes.write_u8(SKETCH_TYPE); + bytes.write_u8(0); // unused + + let mut flags = FLAGS_IS_READ_ONLY | FLAGS_IS_COMPACT; + if compact_state.is_ordered() { + flags |= FLAGS_IS_ORDERED; + } + bytes.write_u8(flags); + bytes.write_u16_le(compact_state.seed_hash()); + + if pre_longs > 1 { + bytes.write_u32_le(retained_entries.len() as u32); + bytes.write_u32_le(0); // unused + } + if is_estimation_mode { + bytes.write_u64_le(compact_state.theta()); + } + + for entry in retained_entries { + let entry = entry.borrow(); + bytes.write_u64_le(entry.hash()); + entry.summary().serialize_value(&mut bytes); + } + bytes.into_bytes() +} + /// Compact (immutable) Tuple sketch. /// /// This is the serialization-friendly form: a compact array of retained hash-summary pairs plus @@ -546,16 +601,6 @@ impl CompactTupleSketch { self.compact_state.retained_entries_capacity() * size_of::>() } - fn preamble_longs(&self) -> u8 { - if self.is_estimation_mode() { - 3 - } else if self.num_retained() == 1 { - 1 - } else { - 2 - } - } - /// Serializes this sketch into the compact Tuple binary format. /// /// The serialized data can be read by other DataSketches implementations when both @@ -579,44 +624,7 @@ impl CompactTupleSketch { where S: TupleSummaryValue, { - if self.is_empty() { - return EMPTY_SKETCH_BYTES.to_vec(); - } - - let retained_entries = self.retained_entries(); - let pre_longs = self.preamble_longs(); - let entries_size: usize = retained_entries - .iter() - .map(|entry| 8 + entry.summary().serialize_size()) - .sum(); - let mut bytes = SketchBytes::with_capacity(8 * pre_longs as usize + entries_size); - - bytes.write_u8(pre_longs); - bytes.write_u8(SERIAL_VERSION); - bytes.write_u8(Family::TUPLE.id); - bytes.write_u8(SKETCH_TYPE); - bytes.write_u8(0); // unused - - let mut flags = FLAGS_IS_READ_ONLY | FLAGS_IS_COMPACT; - if self.is_ordered() { - flags |= FLAGS_IS_ORDERED; - } - bytes.write_u8(flags); - bytes.write_u16_le(self.seed_hash()); - - if pre_longs > 1 { - bytes.write_u32_le(retained_entries.len() as u32); - bytes.write_u32_le(0); // unused - } - if self.is_estimation_mode() { - bytes.write_u64_le(self.theta64()); - } - - for entry in retained_entries { - bytes.write_u64_le(entry.hash()); - entry.summary().serialize_value(&mut bytes); - } - bytes.into_bytes() + serialize_compact_state(&self.compact_state) } /// Deserializes a compact Tuple sketch using the default seed. diff --git a/datasketches/src/thetafamily/tuple/union.rs b/datasketches/src/thetafamily/tuple/union.rs index 029aeba7..fc74f7a5 100644 --- a/datasketches/src/thetafamily/tuple/union.rs +++ b/datasketches/src/thetafamily/tuple/union.rs @@ -29,8 +29,10 @@ use crate::thetacommon::constants::DEFAULT_LG_K; use crate::thetacommon::union::UnionState; use crate::tuple::hash_table::TupleEntry; use crate::tuple::policy::SummaryCombinePolicy; +use crate::tuple::serialization::TupleSummaryValue; use crate::tuple::sketch::CompactTupleSketch; use crate::tuple::sketch::TupleSketchView; +use crate::tuple::sketch::serialize_compact_state; /// Union (set OR) of Tuple sketches. /// @@ -107,6 +109,43 @@ where CompactTupleSketch::from_compact_state(self.state.to_compact_sketch_state(ordered)) } + /// Serializes the union result into the compact Tuple binary format. + /// + /// The result is the same as `self.to_sketch(ordered).serialize()`: the same retained entries, + /// theta, and seed hash, and identical bytes when `ordered` is `true`. Summaries are encoded in + /// place instead of being cloned into an intermediate sketch, so `P::Summary` does not need to + /// implement [`Clone`]. + /// + /// # Examples + /// + /// ``` + /// use datasketches::tuple::CompactTupleSketch; + /// use datasketches::tuple::DefaultUnionPolicy; + /// use datasketches::tuple::DefaultUpdatePolicy; + /// use datasketches::tuple::TupleSketchBuilder; + /// use datasketches::tuple::TupleUnionBuilder; + /// + /// let mut sketch = TupleSketchBuilder::new(DefaultUpdatePolicy::::default()) + /// .build() + /// .unwrap(); + /// sketch.update("apple", 1); + /// + /// let mut union = TupleUnionBuilder::new(DefaultUnionPolicy::::default()) + /// .build() + /// .unwrap(); + /// union.update(&sketch).unwrap(); + /// + /// let bytes = union.serialize(true); + /// let result = CompactTupleSketch::::deserialize(&bytes).unwrap(); + /// assert_eq!(result.num_retained(), 1); + /// ``` + pub fn serialize(&self, ordered: bool) -> Vec + where + P::Summary: TupleSummaryValue, + { + serialize_compact_state(&self.state.to_borrowed_compact_sketch_state(ordered)) + } + /// Resets the union to its initial empty state. pub fn reset(&mut self) { self.state.reset(); diff --git a/tests-integration/tests/serde_tests/tuple.rs b/tests-integration/tests/serde_tests/tuple.rs index 3a71e25a..2c10af80 100644 --- a/tests-integration/tests/serde_tests/tuple.rs +++ b/tests-integration/tests/serde_tests/tuple.rs @@ -15,13 +15,24 @@ // specific language governing permissions and limitations // under the License. +use std::fmt::Debug; use std::fs; +use std::io; use std::path::PathBuf; +use datasketches::codec::SketchBytes; +use datasketches::codec::SketchSlice; +use datasketches::error::Error; use datasketches::error::ErrorKind; use datasketches::tuple::CompactTupleSketch; +use datasketches::tuple::DefaultUnionPolicy; use datasketches::tuple::DefaultUpdatePolicy; +use datasketches::tuple::SummaryCombinePolicy; +use datasketches::tuple::SummaryPolicy; +use datasketches::tuple::SummaryUpdatePolicy; use datasketches::tuple::TupleSketchBuilder; +use datasketches::tuple::TupleSummaryValue; +use datasketches::tuple::TupleUnionBuilder; use googletest::assert_that; use googletest::prelude::each; use googletest::prelude::eq; @@ -214,3 +225,169 @@ fn declared_entry_payload_is_checked_before_allocating() { assert!(CompactTupleSketch::::deserialize(&bytes).is_err()); } + +/// Asserts that two serialized compact Tuple sketches hold the same result. Unordered images may +/// list the same entries in a different order. +fn assert_same_compact_result(actual: &[u8], expected: &[u8]) +where + S: TupleSummaryValue + PartialEq + Debug, +{ + let actual = CompactTupleSketch::::deserialize(actual).unwrap(); + let expected = CompactTupleSketch::::deserialize(expected).unwrap(); + assert_eq!(actual.is_empty(), expected.is_empty()); + assert_eq!(actual.is_ordered(), expected.is_ordered()); + assert_eq!(actual.theta64(), expected.theta64()); + assert_eq!(actual.seed_hash(), expected.seed_hash()); + + fn sorted_entries(sketch: &CompactTupleSketch) -> Vec<(u64, &S)> { + let mut entries: Vec<_> = sketch + .iter() + .map(|entry| (entry.hash(), entry.summary())) + .collect(); + entries.sort_unstable_by_key(|(hash, _)| *hash); + entries + } + assert_eq!(sorted_entries(&actual), sorted_entries(&expected)); +} + +#[test] +fn union_serialization_matches_compact_result() { + let union_builder = || TupleUnionBuilder::new(DefaultUnionPolicy::::default()); + let empty = union_builder().build().unwrap(); + for ordered in [true, false] { + assert_eq!( + empty.serialize(ordered), + empty.to_sketch(ordered).serialize() + ); + } + + // Covers exact mode with zero, one and many entries, trimming to nominal capacity, a sampled + // union, and a non-empty result whose entries were all screened by theta. + for (lg_k, sampling_probability, count) in [ + (5, 1.0, 0), + (5, 1.0, 1), + (5, 1.0, 20), + (5, 1.0, 10_000), + (8, 0.5, 10_000), + (5, 1e-12, 10), + ] { + let mut a = TupleSketchBuilder::new(DefaultUpdatePolicy::::default()) + .build() + .unwrap(); + let mut b = TupleSketchBuilder::new(DefaultUpdatePolicy::::default()) + .build() + .unwrap(); + for value in 0..count { + a.update(value, 1u64); + b.update(value + count / 2, 2u64); + } + let mut union = union_builder() + .lg_k(lg_k) + .sampling_probability(sampling_probability) + .build() + .unwrap(); + union.update(&a).unwrap(); + union.update(&b).unwrap(); + + for ordered in [true, false] { + let bytes = union.serialize(ordered); + let expected = union.to_sketch(ordered).serialize(); + if ordered { + assert_eq!(bytes, expected); + } + assert_same_compact_result::(&bytes, &expected); + } + } +} + +/// Variable-size, heap-backed summary: the tags recorded for a key. +#[derive(Debug, Default, Clone, PartialEq)] +struct Tags(Vec); + +impl TupleSummaryValue for Tags { + fn serialize_size(&self) -> usize { + 4 + self.0.iter().map(|tag| 4 + tag.len()).sum::() + } + + fn serialize_value(&self, bytes: &mut SketchBytes) { + bytes.write_u32_le(self.0.len() as u32); + for tag in &self.0 { + bytes.write_u32_le(tag.len() as u32); + bytes.write(tag.as_bytes()); + } + } + + fn deserialize_value(cursor: &mut SketchSlice<'_>) -> Result { + let malformed = |_: io::Error| Error::new(ErrorKind::InvalidData, "malformed tags"); + let count = cursor.read_u32_le().map_err(malformed)?; + let mut tags = Vec::new(); + for _ in 0..count { + let mut tag = vec![0; cursor.read_u32_le().map_err(malformed)? as usize]; + cursor.read_exact(&mut tag).map_err(malformed)?; + tags.push( + String::from_utf8(tag) + .map_err(|_| Error::new(ErrorKind::InvalidData, "tag is not UTF-8"))?, + ); + } + Ok(Tags(tags)) + } +} + +#[derive(Debug, Clone, Copy)] +struct TagsPolicy; + +impl SummaryPolicy for TagsPolicy { + type Summary = Tags; + + fn create(&self) -> Self::Summary { + Tags::default() + } +} + +impl SummaryUpdatePolicy<&str> for TagsPolicy { + fn update(&self, summary: &mut Self::Summary, value: &str) { + summary.0.push(value.to_string()); + } +} + +impl SummaryCombinePolicy for TagsPolicy { + fn combine(&self, summary: &mut Self::Summary, other: &Self::Summary) { + summary.0.extend(other.0.iter().cloned()); + } +} + +#[test] +fn union_serialization_with_heap_backed_summaries_supports_later_merges() { + let sketch = |values: std::ops::Range, tag: &str| { + let mut sketch = TupleSketchBuilder::new(TagsPolicy).lg_k(5).build().unwrap(); + for value in values { + sketch.update(value, tag); + } + sketch + }; + let union_builder = || TupleUnionBuilder::new(TagsPolicy).lg_k(5); + + let mut union = union_builder().build().unwrap(); + union.update(&sketch(0..1_000, "first")).unwrap(); + union + .update(&sketch(500..1_500, "a longer second tag")) + .unwrap(); + let later = sketch(1_000..2_000, "third"); + + for ordered in [true, false] { + let bytes = union.serialize(ordered); + let expected = union.to_sketch(ordered).serialize(); + assert_same_compact_result::(&bytes, &expected); + + let restored = CompactTupleSketch::::deserialize(&bytes).unwrap(); + let mut from_bytes = union_builder().build().unwrap(); + from_bytes.update(&restored).unwrap(); + from_bytes.update(&later).unwrap(); + + let mut from_sketch = union_builder().build().unwrap(); + from_sketch.update(&union.to_sketch(ordered)).unwrap(); + from_sketch.update(&later).unwrap(); + + assert_eq!(from_bytes.serialize(true), from_sketch.serialize(true)); + } +}