From fda8d90e01959f8a96b2f249bb07751068ea6152 Mon Sep 17 00:00:00 2001 From: Nash Aschenbath Date: Wed, 4 Nov 2026 20:17:58 +0800 Subject: [PATCH] perf(tuple): serialize union results without cloning summaries Add TupleUnion::serialize. It prepares the union result over borrowed entries and encodes the summaries in place, so serializing no longer clones every retained summary into an intermediate CompactTupleSketch, and the summary type does not need to implement Clone. The union result preparation (theta filtering, trimming to nominal capacity, ordering) is shared between the cloning and the borrowing paths, and CompactTupleSketch::serialize uses the same encoder. Closes #321 Generated-by: Claude Code (Claude Opus 5.5) Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + benchmarks/tuple/mod.rs | 1 + benchmarks/tuple/serde.rs | 128 ++++++++++++++ datasketches/src/thetafamily/common/mod.rs | 6 + datasketches/src/thetafamily/common/union.rs | 21 ++- datasketches/src/thetafamily/tuple/sketch.rs | 104 ++++++----- datasketches/src/thetafamily/tuple/union.rs | 39 ++++ tests-integration/tests/serde_tests/tuple.rs | 177 +++++++++++++++++++ 8 files changed, 427 insertions(+), 50 deletions(-) create mode 100644 benchmarks/tuple/serde.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f254ac5..308123f6 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` and `XorFilterBuilder` behind the `xor` feature, with 8-, 16-, and 32-bit fingerprints, consuming builder configuration with complete-input construction, 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 raw parts for application-defined storage. ### Improvements diff --git a/benchmarks/tuple/mod.rs b/benchmarks/tuple/mod.rs index 1dea33be..13c2486b 100644 --- a/benchmarks/tuple/mod.rs +++ b/benchmarks/tuple/mod.rs @@ -15,4 +15,5 @@ // specific language governing permissions and limitations // under the License. +mod serde; mod update; diff --git a/benchmarks/tuple/serde.rs b/benchmarks/tuple/serde.rs new file mode 100644 index 00000000..96028c6b --- /dev/null +++ b/benchmarks/tuple/serde.rs @@ -0,0 +1,128 @@ +// 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 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; +use divan::Bencher; +use divan::black_box; + +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 +} + +#[divan::bench(args = ROWS, consts = [true, false])] +fn serialize_union(bencher: Bencher, rows: u64) { + let union = u64_union(rows); + bencher.bench_local(|| black_box(&union).serialize(ORDERED)); +} + +#[divan::bench(args = ROWS, consts = [true, false])] +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 +} + +#[divan::bench(args = ROWS, consts = [true, false])] +fn serialize_labels_union(bencher: Bencher, rows: u64) { + let union = labels_union(rows); + bencher.bench_local(|| black_box(&union).serialize(ORDERED)); +} + +#[divan::bench(args = ROWS, consts = [true, false])] +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()); +} 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..f7152d9c 100644 --- a/datasketches/src/thetafamily/common/union.rs +++ b/datasketches/src/thetafamily/common/union.rs @@ -113,18 +113,35 @@ where where E: Clone, { + self.compact_sketch_state_with(ordered, E::clone) + } + + /// Returns the union as canonical compact-sketch state that borrows the retained entries. + /// + /// Theta filtering, trimming to nominal capacity, and ordering match + /// [`to_compact_sketch_state`](Self::to_compact_sketch_state). + #[cfg(feature = "tuple")] + pub fn to_borrowed_compact_sketch_state(&self, ordered: bool) -> CompactSketchState<&E> { + self.compact_sketch_state_with(ordered, |entry| entry) + } + + fn compact_sketch_state_with<'a, T: SketchEntry>( + &'a self, + ordered: bool, + project: impl FnMut(&'a E) -> T, + ) -> CompactSketchState { let Some(result_theta) = self.result_theta else { return CompactSketchState::empty(self.table.seed_hash()); }; 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::>() + self.table.iter_entries().map(project).collect::>() } else { self.table .iter_entries() .filter(|entry| entry.hash() < theta) - .cloned() + .map(project) .collect::>() }; 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)); + } +}