pmcgleenon commented on code in PR #204: URL: https://github.com/apache/datasketches-rust/pull/204#discussion_r3847071822
########## datasketches/src/req/compactor.rs: ########## @@ -0,0 +1,604 @@ +// 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. + +//! Compactor implementation for REQ sketch levels. +//! +//! Each level in the REQ sketch uses a compactor to maintain a bounded set of items +//! with deterministic compaction when capacity is exceeded. + +use super::RankAccuracy; +use super::value::ReqValue; +use crate::error::Error; + +fn nearest_even(value: f32) -> u32 { + ((value / 2.0).round() as u32) << 1 +} + +/// A compactor maintains items at a specific level of the REQ sketch. +/// +/// When the compactor reaches its nominal capacity, it performs compaction +/// by keeping approximately half the items and promoting the rest to the next level. +#[derive(Debug, Clone)] +pub(super) struct Compactor<T> { + /// Current items in the compactor + items: Vec<T>, + /// Whether items are currently sorted + is_sorted: bool, + /// State for deterministic compaction + state: u64, + /// Reusable scratch buffer for compaction operations + scratch_buffer: Vec<T>, + + /// Actual section size (rounded to integer) + section_size: u32, + /// Number of sections in this compactor + num_sections: u8, + /// The level of this compactor (0 = base level) + lg_weight: u8, + + /// Whether this compactor is configured for high rank accuracy + rank_accuracy: RankAccuracy, + /// Raw section size (may be fractional) + section_size_raw: f32, + /// Random bit for compaction + coin: bool, +} + +impl<T> Compactor<T> +where + T: Clone + ReqValue, +{ + /// Creates a new compactor for the given level. + /// + /// # Arguments + /// * `lg_weight` - The level (log weight) of this compactor + /// * `k` - The k parameter from the parent sketch + /// * `rank_accuracy` - Rank accuracy configuration + pub(super) fn new(lg_weight: u8, k: u16, rank_accuracy: RankAccuracy) -> Self { + let section_size_raw = k as f32; + let section_size = nearest_even(section_size_raw); + let num_sections = 3u8; + + let nominal: usize = (2 * section_size * num_sections as u32) as usize; + + Self { + items: Vec::with_capacity(nominal), + is_sorted: true, + state: 0, + scratch_buffer: Vec::with_capacity(nominal / 2 + 8), + + section_size, + num_sections, + lg_weight, + + rank_accuracy, + section_size_raw, + coin: false, + } + } + + /// Returns the number of items currently in this compactor. + pub(super) fn num_items(&self) -> u32 { + self.items.len() as u32 + } + + /// Returns the nominal capacity of this compactor. + pub(super) fn nominal_capacity(&self) -> u32 { + 2 * self.section_size * self.num_sections as u32 + } + + /// Returns whether the items are currently sorted. + pub(super) fn is_sorted(&self) -> bool { + self.is_sorted + } + + /// Appends an item to this compactor. + #[inline(always)] + pub(super) fn append(&mut self, item: T) { + self.items.push(item); + if self.items.len() > 1 { + self.is_sorted = false; + } + } + + /// Merges items from another compactor into this one. + pub(super) fn merge(&mut self, other: &Self) { + self.state |= other.state; + self.items.extend_from_slice(&other.items); + if !other.items.is_empty() { + self.is_sorted = false; + } + // OR-ing the schedule counters can advance state past several doubling + // thresholds at once. Loop until no more doublings are needed (C++: + // req_compactor_impl.hpp:250 — `while (ensure_enough_sections()) {}`). + while self.ensure_enough_sections() {} + } + + /// Counts the items at-or-below (`inclusive`) or strictly below `item`. + /// + /// Uses binary search when this compactor is sorted, and a linear scan + /// otherwise. This lets [`ReqSketch::rank`](super::ReqSketch::rank) sum + /// per-level weights directly without first building a sorted view. + pub(super) fn count_below(&self, item: &T, inclusive: bool) -> usize { + if self.is_sorted { + if inclusive { + self.items.partition_point(|x| x.total_cmp(item).is_le()) + } else { + self.items.partition_point(|x| x.total_cmp(item).is_lt()) + } + } else { + self.items + .iter() + .filter(|x| { + let ord = x.total_cmp(item); + if inclusive { ord.is_le() } else { ord.is_lt() } + }) + .count() + } + } + + /// Merges pre-sorted items into this compactor. + /// Merges sorted items into this compactor using scratch buffer to avoid allocation. + /// Both this compactor's items and the input must be sorted. + #[inline(always)] + pub(super) fn merge_sorted(&mut self, items: &[T]) { + if items.is_empty() { + return; + } + + if self.items.is_empty() { + self.items.extend_from_slice(items); + self.is_sorted = true; + return; + } + + // Ensure sorted on both inputs by contract + let total = self.items.len() + items.len(); + self.scratch_buffer.clear(); + if self.scratch_buffer.capacity() < total { + self.scratch_buffer + .reserve(total - self.scratch_buffer.capacity()); + } + + let (mut i, mut j) = (0usize, 0usize); + let (a, b) = (&self.items, items); + + // Two-pointer merge into scratch buffer + while i < a.len() && j < b.len() { + if a[i].total_cmp(&b[j]).is_le() { + self.scratch_buffer.push(a[i].clone()); + i += 1; + } else { + self.scratch_buffer.push(b[j].clone()); + j += 1; + } + } + + // Add remaining elements + if i < a.len() { + self.scratch_buffer.extend_from_slice(&a[i..]); + } + if j < b.len() { + self.scratch_buffer.extend_from_slice(&b[j..]); + } + + // Swap scratch buffer with items (zero-copy) + self.items.clear(); + std::mem::swap(&mut self.items, &mut self.scratch_buffer); + self.is_sorted = true; + } + + /// Sorts the items in this compactor if not already sorted. + #[inline(always)] + pub(super) fn sort(&mut self) { + if !self.is_sorted { + // Use unstable sort for better performance (stable not needed for REQ sketch) + self.items.sort_unstable_by(|a, b| a.total_cmp(b)); + self.is_sorted = true; + } + } + + /// Compacts into the provided output buffer without allocating. + /// Writes promoted items into `out` and removes the compacted range in-place via `copy_within + + /// truncate`. + #[inline(always)] + pub(super) fn compact_into(&mut self, _rank_accuracy: RankAccuracy, out: &mut Vec<T>) { + if self.items.is_empty() { + out.clear(); + return; + } + + // Sort entire buffer (C++ sorts full buffer before compaction) + self.sort(); + + // Calculate sections to compact based on state + let secs_to_compact = + ((!self.state).trailing_zeros() + 1).min(self.num_sections as u32) as u8; + let compaction_range = self.compute_compaction_range(secs_to_compact); + + // Must have at least 2 items to compact + if compaction_range.1 <= compaction_range.0 || (compaction_range.1 - compaction_range.0) < 2 + { + out.clear(); + return; + } + + if (self.state & 1) == 1 { + self.coin = !self.coin; // flip coin for odd states + } else { + self.coin = rand::random::<bool>(); // random coin flip for even states + } + let odds = self.coin; + + // Build promoted items directly into output buffer (no alloc) + out.clear(); + let (start, end) = compaction_range; + let mut i = start + if odds { 1 } else { 0 }; + while i < end { + out.push(self.items[i].clone()); // TODO: use Copy fast-path for numeric types + i += 2; + } + + // Remove the compacted range in-place by rotating elements left + let removed = end - start; + if end < self.items.len() { + // Use rotate_left to move tail elements to fill the gap + self.items[start..].rotate_left(removed); + } + self.items.truncate(self.items.len() - removed); + + // Update state, then ensure enough sections (C++ order) + self.state += 1; + self.ensure_enough_sections(); + } + + /// Returns an iterator over the items in this compactor. + pub(super) fn iter(&self) -> impl Iterator<Item = &T> { + self.items.iter() + } + + /// Returns a slice of items for zero-allocation iteration. + pub(super) fn items_slice(&self) -> &[T] { + &self.items + } + + /// Returns the weight (2^lg_weight) for items in this compactor. + pub(super) fn weight(&self) -> u64 { + 1u64 << self.lg_weight + } + + // Private helper methods + + fn ensure_enough_sections(&mut self) -> bool { + let ssr = self.section_size_raw / (2.0_f32).sqrt(); Review Comment: thanks, I've used it here https://github.com/apache/datasketches-rust/pull/204/changes/d5f9d2f08890160b4372f62ca0fa74d8c27e3296 ########## datasketches/src/req/compactor.rs: ########## @@ -0,0 +1,604 @@ +// 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. + +//! Compactor implementation for REQ sketch levels. +//! +//! Each level in the REQ sketch uses a compactor to maintain a bounded set of items +//! with deterministic compaction when capacity is exceeded. + +use super::RankAccuracy; +use super::value::ReqValue; +use crate::error::Error; + +fn nearest_even(value: f32) -> u32 { + ((value / 2.0).round() as u32) << 1 +} + +/// A compactor maintains items at a specific level of the REQ sketch. +/// +/// When the compactor reaches its nominal capacity, it performs compaction +/// by keeping approximately half the items and promoting the rest to the next level. +#[derive(Debug, Clone)] +pub(super) struct Compactor<T> { + /// Current items in the compactor + items: Vec<T>, + /// Whether items are currently sorted + is_sorted: bool, + /// State for deterministic compaction + state: u64, + /// Reusable scratch buffer for compaction operations + scratch_buffer: Vec<T>, + + /// Actual section size (rounded to integer) + section_size: u32, + /// Number of sections in this compactor + num_sections: u8, + /// The level of this compactor (0 = base level) + lg_weight: u8, + + /// Whether this compactor is configured for high rank accuracy + rank_accuracy: RankAccuracy, + /// Raw section size (may be fractional) + section_size_raw: f32, + /// Random bit for compaction + coin: bool, +} + +impl<T> Compactor<T> +where + T: Clone + ReqValue, +{ + /// Creates a new compactor for the given level. + /// + /// # Arguments + /// * `lg_weight` - The level (log weight) of this compactor + /// * `k` - The k parameter from the parent sketch + /// * `rank_accuracy` - Rank accuracy configuration + pub(super) fn new(lg_weight: u8, k: u16, rank_accuracy: RankAccuracy) -> Self { + let section_size_raw = k as f32; + let section_size = nearest_even(section_size_raw); + let num_sections = 3u8; + + let nominal: usize = (2 * section_size * num_sections as u32) as usize; + + Self { + items: Vec::with_capacity(nominal), + is_sorted: true, + state: 0, + scratch_buffer: Vec::with_capacity(nominal / 2 + 8), + + section_size, + num_sections, + lg_weight, + + rank_accuracy, + section_size_raw, + coin: false, + } + } + + /// Returns the number of items currently in this compactor. + pub(super) fn num_items(&self) -> u32 { + self.items.len() as u32 + } + + /// Returns the nominal capacity of this compactor. + pub(super) fn nominal_capacity(&self) -> u32 { + 2 * self.section_size * self.num_sections as u32 + } + + /// Returns whether the items are currently sorted. + pub(super) fn is_sorted(&self) -> bool { + self.is_sorted + } + + /// Appends an item to this compactor. + #[inline(always)] + pub(super) fn append(&mut self, item: T) { + self.items.push(item); + if self.items.len() > 1 { + self.is_sorted = false; + } + } + + /// Merges items from another compactor into this one. + pub(super) fn merge(&mut self, other: &Self) { + self.state |= other.state; + self.items.extend_from_slice(&other.items); + if !other.items.is_empty() { + self.is_sorted = false; + } + // OR-ing the schedule counters can advance state past several doubling + // thresholds at once. Loop until no more doublings are needed (C++: + // req_compactor_impl.hpp:250 — `while (ensure_enough_sections()) {}`). + while self.ensure_enough_sections() {} + } + + /// Counts the items at-or-below (`inclusive`) or strictly below `item`. + /// + /// Uses binary search when this compactor is sorted, and a linear scan + /// otherwise. This lets [`ReqSketch::rank`](super::ReqSketch::rank) sum + /// per-level weights directly without first building a sorted view. + pub(super) fn count_below(&self, item: &T, inclusive: bool) -> usize { + if self.is_sorted { + if inclusive { + self.items.partition_point(|x| x.total_cmp(item).is_le()) + } else { + self.items.partition_point(|x| x.total_cmp(item).is_lt()) + } + } else { + self.items + .iter() + .filter(|x| { + let ord = x.total_cmp(item); + if inclusive { ord.is_le() } else { ord.is_lt() } + }) + .count() + } + } + + /// Merges pre-sorted items into this compactor. + /// Merges sorted items into this compactor using scratch buffer to avoid allocation. + /// Both this compactor's items and the input must be sorted. + #[inline(always)] + pub(super) fn merge_sorted(&mut self, items: &[T]) { + if items.is_empty() { + return; + } + + if self.items.is_empty() { + self.items.extend_from_slice(items); + self.is_sorted = true; + return; + } + + // Ensure sorted on both inputs by contract + let total = self.items.len() + items.len(); + self.scratch_buffer.clear(); + if self.scratch_buffer.capacity() < total { + self.scratch_buffer + .reserve(total - self.scratch_buffer.capacity()); + } + + let (mut i, mut j) = (0usize, 0usize); + let (a, b) = (&self.items, items); + + // Two-pointer merge into scratch buffer + while i < a.len() && j < b.len() { + if a[i].total_cmp(&b[j]).is_le() { + self.scratch_buffer.push(a[i].clone()); + i += 1; + } else { + self.scratch_buffer.push(b[j].clone()); + j += 1; + } + } + + // Add remaining elements + if i < a.len() { + self.scratch_buffer.extend_from_slice(&a[i..]); + } + if j < b.len() { + self.scratch_buffer.extend_from_slice(&b[j..]); + } + + // Swap scratch buffer with items (zero-copy) + self.items.clear(); + std::mem::swap(&mut self.items, &mut self.scratch_buffer); + self.is_sorted = true; + } + + /// Sorts the items in this compactor if not already sorted. + #[inline(always)] + pub(super) fn sort(&mut self) { + if !self.is_sorted { + // Use unstable sort for better performance (stable not needed for REQ sketch) + self.items.sort_unstable_by(|a, b| a.total_cmp(b)); + self.is_sorted = true; + } + } + + /// Compacts into the provided output buffer without allocating. + /// Writes promoted items into `out` and removes the compacted range in-place via `copy_within + + /// truncate`. + #[inline(always)] + pub(super) fn compact_into(&mut self, _rank_accuracy: RankAccuracy, out: &mut Vec<T>) { + if self.items.is_empty() { + out.clear(); + return; + } + + // Sort entire buffer (C++ sorts full buffer before compaction) + self.sort(); + + // Calculate sections to compact based on state + let secs_to_compact = + ((!self.state).trailing_zeros() + 1).min(self.num_sections as u32) as u8; + let compaction_range = self.compute_compaction_range(secs_to_compact); + + // Must have at least 2 items to compact + if compaction_range.1 <= compaction_range.0 || (compaction_range.1 - compaction_range.0) < 2 + { + out.clear(); + return; + } + + if (self.state & 1) == 1 { + self.coin = !self.coin; // flip coin for odd states + } else { + self.coin = rand::random::<bool>(); // random coin flip for even states + } + let odds = self.coin; + + // Build promoted items directly into output buffer (no alloc) + out.clear(); + let (start, end) = compaction_range; + let mut i = start + if odds { 1 } else { 0 }; + while i < end { + out.push(self.items[i].clone()); // TODO: use Copy fast-path for numeric types + i += 2; + } + + // Remove the compacted range in-place by rotating elements left + let removed = end - start; + if end < self.items.len() { + // Use rotate_left to move tail elements to fill the gap + self.items[start..].rotate_left(removed); + } + self.items.truncate(self.items.len() - removed); + + // Update state, then ensure enough sections (C++ order) + self.state += 1; + self.ensure_enough_sections(); + } + + /// Returns an iterator over the items in this compactor. + pub(super) fn iter(&self) -> impl Iterator<Item = &T> { + self.items.iter() + } + + /// Returns a slice of items for zero-allocation iteration. + pub(super) fn items_slice(&self) -> &[T] { + &self.items + } + + /// Returns the weight (2^lg_weight) for items in this compactor. + pub(super) fn weight(&self) -> u64 { + 1u64 << self.lg_weight + } + + // Private helper methods + + fn ensure_enough_sections(&mut self) -> bool { + let ssr = self.section_size_raw / (2.0_f32).sqrt(); + let ne = nearest_even(ssr); + + const MIN_K: u32 = 4; // matches datasketches-cpp Review Comment: Good catch, updated https://github.com/apache/datasketches-rust/pull/204/changes/d5f9d2f08890160b4372f62ca0fa74d8c27e3296 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
