diff --git a/Cargo.lock b/Cargo.lock index 70e8b5c23..259030d7c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -363,12 +363,10 @@ dependencies = [ [[package]] name = "asap_sketchlib" -version = "0.1.0" +version = "0.2.2" dependencies = [ "bytes", "prost", - "prost-build", - "protoc-bin-vendored", "rand 0.9.4", "rmp-serde", "serde", @@ -827,7 +825,7 @@ dependencies = [ "clap 4.6.1", "criterion-plot", "is-terminal", - "itertools 0.10.5", + "itertools", "num-traits", "once_cell", "oorandom", @@ -848,7 +846,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" dependencies = [ "cast", - "itertools 0.10.5", + "itertools", ] [[package]] @@ -1051,7 +1049,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1756,7 +1754,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi 0.5.2", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -1774,15 +1772,6 @@ dependencies = [ "either", ] -[[package]] -name = "itertools" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" -dependencies = [ - "either", -] - [[package]] name = "itoa" version = "1.0.18" @@ -2490,7 +2479,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck 0.5.0", - "itertools 0.14.0", + "itertools", "log", "multimap", "once_cell", @@ -2510,7 +2499,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools", "proc-macro2", "quote", "syn 2.0.117", @@ -2961,7 +2950,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3392,7 +3381,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4204,7 +4193,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] diff --git a/data_plane/benches/sketch_db.rs b/data_plane/benches/sketch_db.rs index e5064143b..f2c62e32f 100644 --- a/data_plane/benches/sketch_db.rs +++ b/data_plane/benches/sketch_db.rs @@ -25,8 +25,8 @@ use std::collections::{BTreeMap, BTreeSet}; use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion, Throughput}; -use asap_sketchlib::sketches::ddsketch::DdSketch; -use asap_sketchlib::sketches::hll::{HllSketch, HllVariant}; +use asap_sketchlib::DdSketch; +use asap_sketchlib::{HllSketch, HllVariant}; use asap_sketchlib::proto::sketchlib::{ sketch_envelope, DdSketchState, HllVariant as ProtoVariant, HyperLogLogState, KllState, SketchEnvelope, @@ -91,6 +91,7 @@ fn encode_hll(distinct_items: usize, precision: u32) -> Vec { hip_kxq0: sk.hip_kxq0, hip_kxq1: sk.hip_kxq1, hip_est: sk.hip_est, + registers_sparse: None, }; SketchEnvelope { sketch_state: Some(sketch_envelope::SketchState::Hll(state)), diff --git a/data_plane/examples/sketch_db_diag.rs b/data_plane/examples/sketch_db_diag.rs index ae7e06ba7..46ebe6761 100644 --- a/data_plane/examples/sketch_db_diag.rs +++ b/data_plane/examples/sketch_db_diag.rs @@ -20,7 +20,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::time::Instant; use asap_sketchlib::proto::sketchlib::{sketch_envelope, DdSketchState, SketchEnvelope}; -use asap_sketchlib::sketches::ddsketch::DdSketch; +use asap_sketchlib::DdSketch; use prost::Message; use data_plane::storage_engines::sketch_db::data::{ diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 7675b9d4a..12b19fafd 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -36,6 +36,7 @@ use asap_otel_proto::tonic::collector::metrics::v1::{ use asap_otel_proto::tonic::common::v1::any_value::Value as AnyValueVariant; use asap_otel_proto::tonic::metrics::v1::number_data_point::Value as NumberValue; use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope}; +use asap_sketchlib::MessagePackCodec; use axum::{body::Bytes, extract::State, routing::post, Json, Router}; use flate2::read::GzDecoder; use prost::Message; @@ -1631,8 +1632,8 @@ fn sketch_kind_handle_for( // decode AND the heap is non-empty; otherwise stay with // vanilla `CountSketch`. if dp.encoding == ENCODING_MSGPACK { - use asap_sketchlib::sketches::countminsketch_topk::CountMinSketchWithHeap; - if let Ok(cms) = CountMinSketchWithHeap::deserialize_msgpack(&dp.sketch) { + use asap_sketchlib::CountMinSketchWithHeap; + if let Ok(cms) = CountMinSketchWithHeap::from_msgpack(&dp.sketch) { if !cms.topk_heap_items().is_empty() { return SketchKindHandle::CountSketchWithHeap; } @@ -1649,8 +1650,8 @@ fn sketch_kind_handle_for( // CmsWithHeap so ASAP-tier `topk` can read the heap. // Otherwise stay with vanilla `CountMin`. if dp.encoding == ENCODING_MSGPACK { - use asap_sketchlib::sketches::countminsketch_topk::CountMinSketchWithHeap; - if let Ok(cms) = CountMinSketchWithHeap::deserialize_msgpack(&dp.sketch) { + use asap_sketchlib::CountMinSketchWithHeap; + if let Ok(cms) = CountMinSketchWithHeap::from_msgpack(&dp.sketch) { if !cms.topk_heap_items().is_empty() { return SketchKindHandle::CmsWithHeap; } @@ -2549,8 +2550,8 @@ mod dispatcher_tests { use super::*; use crate::storage_engines::types::AggregateCore; use crate::precompute_engine::operators::{DDSketchAccumulator, HllSketchAccumulator}; - use asap_sketchlib::sketches::ddsketch::DdSketch; - use asap_sketchlib::sketches::hll::HllVariant; + use asap_sketchlib::DdSketch; + use asap_sketchlib::HllVariant; #[test] fn apply_modified_otlp_delta_bytes_ddsketch_round_trip() { diff --git a/data_plane/src/precompute_engine/ingest_handler.rs b/data_plane/src/precompute_engine/ingest_handler.rs index cd3aceb34..51603d166 100644 --- a/data_plane/src/precompute_engine/ingest_handler.rs +++ b/data_plane/src/precompute_engine/ingest_handler.rs @@ -172,7 +172,7 @@ mod tests { use crate::drivers::ingest::otel::{apply_modified_otlp_delta_bytes, SketchKind}; use crate::precompute_engine::operators::DDSketchAccumulator; use asap_otel_proto::sketchlib::v1::{DdSketchBucketDelta, DdSketchDelta as PbDelta}; - use asap_sketchlib::sketches::ddsketch::DdSketch; + use asap_sketchlib::DdSketch; use prost::Message; const ENCODING_PROTO_DELTA: i32 = 2; diff --git a/data_plane/src/precompute_engine/operators/count_min_sketch_accumulator.rs b/data_plane/src/precompute_engine/operators/count_min_sketch_accumulator.rs index fca7ca229..85c74591a 100644 --- a/data_plane/src/precompute_engine/operators/count_min_sketch_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/count_min_sketch_accumulator.rs @@ -2,13 +2,13 @@ use crate::storage_engines::types::{ AggregateCore, AggregationType, KeyByLabelValues, MergeableAccumulator, MultipleSubpopulationAggregate, SerializableToSink, }; -use asap_sketchlib::sketches::countminsketch::{CountMinSketch, CountMinSketchDelta}; +use asap_sketchlib::{CountMinSketch, CountMinSketchDelta, MessagePackCodec}; use serde_json::Value; use std::collections::HashMap; use promql_utilities::query_logics::enums::Statistic; -/// Count-Min Sketch accumulator — wraps asap_sketchlib::sketches::CountMinSketch. +/// Count-Min Sketch accumulator — wraps asap_sketchlib::CountMinSketch. /// Core struct, update/merge/serde logic live in `asap_sketchlib::sketches`. /// This file retains QE-specific trait impls, legacy deserializers, and JSON output. #[derive(Debug, Clone)] @@ -69,7 +69,7 @@ impl CountMinSketchAccumulator { /// uses — this method is the modified-OTLP entrypoint for PR I). pub fn from_msgpack_bytes(buffer: &[u8]) -> Result> { Ok(Self { - inner: CountMinSketch::deserialize_msgpack(buffer) + inner: CountMinSketch::from_msgpack(buffer) .map_err(|e| -> Box { e.to_string().into() })?, }) } @@ -309,7 +309,7 @@ impl SerializableToSink for CountMinSketchAccumulator { } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } diff --git a/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs b/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs index 698e16b3a..2db5379a7 100644 --- a/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs @@ -2,14 +2,14 @@ use crate::storage_engines::types::{ AggregateCore, AggregationType, KeyByLabelValues, MergeableAccumulator, MultipleSubpopulationAggregate, SerializableToSink, }; -use asap_sketchlib::sketches::countminsketch_topk::{CmsHeapItem, CountMinSketchWithHeap}; +use asap_sketchlib::{CmsHeapItem, CountMinSketchWithHeap, MessagePackCodec}; use serde_json::Value; use std::collections::HashMap; use promql_utilities::query_logics::enums::Statistic; -/// Count-Min Sketch with Heap accumulator — wraps `asap_sketchlib::sketches::CountMinSketchWithHeap`. -/// Core struct, update/merge/serde logic live in `asap_sketchlib::sketches::countminsketch_topk`. +/// Count-Min Sketch with Heap accumulator — wraps `asap_sketchlib::CountMinSketchWithHeap`. +/// Core struct, update/merge/serde logic live in `asap_sketchlib::message_pack_format::portable::countminsketch_topk`. /// This file retains QE-specific trait impls, legacy deserializers, and JSON output. #[derive(Debug, Clone)] pub struct CountMinSketchWithHeapAccumulator { @@ -17,7 +17,7 @@ pub struct CountMinSketchWithHeapAccumulator { } // Re-export HeapItem so existing code using CountMinSketchWithHeapAccumulator::HeapItem still works. -pub use asap_sketchlib::sketches::countminsketch_topk::CmsHeapItem as HeapItemReexport; +pub use asap_sketchlib::CmsHeapItem as HeapItemReexport; impl CountMinSketchWithHeapAccumulator { pub fn new(row_num: usize, col_num: usize, heap_size: usize) -> Self { @@ -123,7 +123,7 @@ impl SerializableToSink for CountMinSketchWithHeapAccumulator { } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } diff --git a/data_plane/src/precompute_engine/operators/count_sketch_accumulator.rs b/data_plane/src/precompute_engine/operators/count_sketch_accumulator.rs index 25e3b673a..c94bb9f35 100644 --- a/data_plane/src/precompute_engine/operators/count_sketch_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/count_sketch_accumulator.rs @@ -1,4 +1,4 @@ -//! Count Sketch accumulator — wraps `asap_sketchlib::sketches::countsketch::CountSketch`. +//! Count Sketch accumulator — wraps `asap_sketchlib::CountSketch`. //! //! This is the concrete accumulator reached from the modified-OTLP //! `Metric.data = CountSketch{…}` hot path (PR C-CountSketch). Its @@ -20,7 +20,7 @@ //! round-trip works end-to-end without that richer query surface. use crate::storage_engines::types::{AggregateCore, AggregationType, KeyByLabelValues, SerializableToSink}; -use asap_sketchlib::sketches::countsketch::{CountSketch, CountSketchDelta}; +use asap_sketchlib::{CountSketch, CountSketchDelta, MessagePackCodec}; use serde_json::Value; use std::collections::HashMap; @@ -44,7 +44,7 @@ impl CountSketchAccumulator { /// `CountSketch` struct — PR I parity entrypoint. pub fn from_msgpack_bytes(buffer: &[u8]) -> Result> { Ok(Self { - inner: CountSketch::deserialize_msgpack(buffer) + inner: CountSketch::from_msgpack(buffer) .map_err(|e| format!("deserialize CountSketch msgpack: {e}"))?, }) } @@ -213,7 +213,7 @@ impl SerializableToSink for CountSketchAccumulator { } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } @@ -278,7 +278,7 @@ impl AggregateCore for CountSketchAccumulator { // but not a heavy-hitter answer). Hash compatibility note: // this relies on the agent and backend using the // sketchlib HashSpec; sketchlib-go's `portableHashSpec` - // is the canonical seed list, and `asap_sketchlib::sketches::CountSketch` + // is the canonical seed list, and `asap_sketchlib::CountSketch` // hashes against the same spec. match statistic { Statistic::Topk | Statistic::Count => { @@ -314,7 +314,7 @@ impl AggregateCore for CountSketchAccumulator { /// /// Hash compatibility with the agent is via the sketchlib hash /// spec; the agent's `sketchlib-go::CountSketch` and the -/// backend's `asap_sketchlib::sketches::countsketch::CountSketch` must use +/// backend's `asap_sketchlib::CountSketch` must use /// the same seed list (sketchlib's `portableHashSpec` / /// `default_hash_spec`). fn count_sketch_query_key(matrix: &Vec>, key: &str) -> f64 { @@ -519,7 +519,7 @@ mod tests { 2, 3, ); - let bytes = original.serialize_msgpack().unwrap(); + let bytes = original.to_msgpack().unwrap(); let acc = CountSketchAccumulator::from_msgpack_bytes(&bytes).expect("decode ok"); assert_eq!(acc.inner.rows, 2); assert_eq!(acc.inner.cols, 3); diff --git a/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs b/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs index a3a626b23..67be4eac0 100644 --- a/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/datasketches_kll_accumulator.rs @@ -2,7 +2,7 @@ use crate::storage_engines::types::{ AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, SingleSubpopulationAggregate, }; -use asap_sketchlib::sketches::kll::KllSketch; +use asap_sketchlib::{KllSketch, MessagePackCodec}; use base64::{engine::general_purpose, Engine as _}; use serde_json::Value; use std::collections::HashMap; @@ -12,7 +12,7 @@ use tracing::debug; use promql_utilities::query_logics::enums::Statistic; -/// KLL sketch accumulator — wraps asap_sketchlib::sketches::KllSketch. +/// KLL sketch accumulator — wraps asap_sketchlib::KllSketch. /// Core struct, update/merge/serde logic live in `asap_sketchlib::sketches`. /// This file retains QE-specific trait impls and JSON output. pub struct DatasketchesKLLAccumulator { @@ -46,7 +46,7 @@ impl DatasketchesKLLAccumulator { /// serializes its full internal state to msgpack. pub fn from_msgpack_bytes(buffer: &[u8]) -> Result> { Ok(Self { - inner: KllSketch::deserialize_msgpack(buffer) + inner: KllSketch::from_msgpack(buffer) .map_err(|e| -> Box { e.to_string().into() })?, }) } @@ -207,7 +207,7 @@ impl SerializableToSink for DatasketchesKLLAccumulator { } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } @@ -556,6 +556,9 @@ mod tests { levels: vec![0, 64], items: items.clone(), coin: None, + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), }; let bytes = state.encode_to_vec(); @@ -596,6 +599,9 @@ mod tests { levels: vec![0, 64], items, coin: None, + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), }; let env = SketchEnvelope { sketch_state: Some(sketch_envelope::SketchState::Kll(state)), @@ -636,6 +642,9 @@ mod tests { levels: Vec::new(), items: Vec::new(), coin: None, + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), }; let bytes = state.encode_to_vec(); let result = DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(&bytes); @@ -655,6 +664,9 @@ mod tests { levels: vec![0, 5, 10], items: vec![1.0, 2.0, 3.0, 4.0, 5.0], coin: None, + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), }; let bytes = state.encode_to_vec(); let result = DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(&bytes); diff --git a/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs b/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs index 145f1543f..f11b8c20c 100644 --- a/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/dd_sketch_accumulator.rs @@ -1,4 +1,4 @@ -//! DDSketch accumulator — wraps `asap_sketchlib::sketches::ddsketch::DdSketch`. +//! DDSketch accumulator — wraps `asap_sketchlib::DdSketch`. //! //! Concrete accumulator reached from the modified-OTLP //! `Metric.data = DDSketch{…}` hot path (PR C-CountSketch follow-up). @@ -12,7 +12,7 @@ //! works end-to-end without that richer query surface. use crate::storage_engines::types::{AggregateCore, AggregationType, KeyByLabelValues, SerializableToSink}; -use asap_sketchlib::sketches::ddsketch::{DdSketch, DdSketchDelta}; +use asap_sketchlib::{DdSketch, DdSketchDelta, MessagePackCodec}; use serde_json::Value; use std::collections::HashMap; @@ -36,7 +36,7 @@ impl DDSketchAccumulator { /// `DdSketch` struct — PR I parity entrypoint. pub fn from_msgpack_bytes(buffer: &[u8]) -> Result> { Ok(Self { - inner: DdSketch::deserialize_msgpack(buffer) + inner: DdSketch::from_msgpack(buffer) .map_err(|e| format!("deserialize DdSketch msgpack: {e}"))?, }) } @@ -142,7 +142,7 @@ impl SerializableToSink for DDSketchAccumulator { } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } @@ -354,7 +354,7 @@ mod tests { #[test] fn test_from_msgpack_bytes_round_trip() { let original = DdSketch::from_raw(0.01, vec![5, 10, 15, 20], -2, 50, 150.0, 0.25, 8.0); - let bytes = original.serialize_msgpack().unwrap(); + let bytes = original.to_msgpack().unwrap(); let acc = DDSketchAccumulator::from_msgpack_bytes(&bytes).expect("decode ok"); assert_eq!(acc.inner.alpha, 0.01); assert_eq!(acc.inner.store_counts, vec![5, 10, 15, 20]); diff --git a/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs b/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs index c86f1ae84..9f2cf4982 100644 --- a/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs +++ b/data_plane/src/precompute_engine/operators/edge_runtime_adapter.rs @@ -104,8 +104,8 @@ pub fn unwrap_envelope_state( /// Result of [`reconstruct_via_runtime`] — backend uses the inner /// `asap_sketchlib::sketches::*` to construct its own accumulator. pub enum ReconstructedSketch { - /// Reconstructed [`asap_sketchlib::sketches::DdSketch`] state. - DdSketch(asap_sketchlib::sketches::ddsketch::DdSketch), + /// Reconstructed [`asap_sketchlib::DdSketch`] state. + DdSketch(asap_sketchlib::DdSketch), /// Reconstructed KLL — asap-precompute-rs's [`KLLWrapper`] owns /// the high-throughput `asap_sketchlib::sketches::kll::KLL` /// internally; backend's KLL accumulator wraps the wire-format- @@ -203,12 +203,12 @@ pub fn reconstruct_via_runtime( } } -/// Snapshot a backend-side `asap_sketchlib::sketches::DdSketch` through +/// Snapshot a backend-side `asap_sketchlib::DdSketch` through /// asap-precompute-rs's `Sketch` trait — the canonical encode path /// shared with the agent runtime. Used by the round-trip test /// (`tests/edge_runtime_adapter.rs`). pub fn snapshot_ddsketch_via_runtime( - sk: &asap_sketchlib::sketches::ddsketch::DdSketch, + sk: &asap_sketchlib::DdSketch, ) -> Result, Box> { let mut wrapper = DDSketchWrapper::new(sk.alpha); // Bridge into the wrapper by merging in the existing sketch. @@ -238,7 +238,7 @@ pub fn snapshot_ddsketch_via_runtime( /// `asap_sketchlib::proto::sketchlib::*` types. Lives here so the /// backend's existing accumulators don't need to import the wrapper /// internals. -pub fn encode_ddsketch_envelope(sk: &asap_sketchlib::sketches::ddsketch::DdSketch) -> Vec { +pub fn encode_ddsketch_envelope(sk: &asap_sketchlib::DdSketch) -> Vec { use asap_sketchlib::proto::sketchlib::{ sketch_envelope, DdSketchState, SketchEnvelope as ProtoEnvelope, }; @@ -271,15 +271,15 @@ pub fn encode_ddsketch_envelope(sk: &asap_sketchlib::sketches::ddsketch::DdSketc /// Merge two `DdSketch` instances by routing through asap-precompute-rs's /// runtime `Sketch::merge`. The result is byte-identical to -/// `asap_sketchlib::sketches::DdSketch::merge_refs(&[a, b])` because +/// `asap_sketchlib::DdSketch::merge_refs(&[a, b])` because /// both paths call the same underlying merge logic. /// /// Used by the cross-runtime parity test /// (`tests/edge_runtime_adapter.rs::ddsketch_merge_via_runtime_matches_native`). pub fn merge_ddsketches_via_runtime( - a: &asap_sketchlib::sketches::ddsketch::DdSketch, - b: &asap_sketchlib::sketches::ddsketch::DdSketch, -) -> Result> { + a: &asap_sketchlib::DdSketch, + b: &asap_sketchlib::DdSketch, +) -> Result> { if (a.alpha - b.alpha).abs() > f64::EPSILON { return Err(format!( "merge_ddsketches_via_runtime: alpha mismatch ({} vs {})", @@ -335,7 +335,7 @@ mod tests { /// asap-precompute-rs's wrapper produces an envelope; the runtime /// adapter's reconstruction returns a backend-shaped - /// `asap_sketchlib::sketches::DdSketch` whose serialized bytes + /// `asap_sketchlib::DdSketch` whose serialized bytes /// (re-encoded through the same envelope shape) match the /// original. #[test] diff --git a/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs b/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs index 2e32b8d13..f35b44fda 100644 --- a/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/hll_sketch_accumulator.rs @@ -1,4 +1,4 @@ -//! HLL accumulator — wraps `asap_sketchlib::sketches::hll::HllSketch`. +//! HLL accumulator — wraps `asap_sketchlib::HllSketch`. //! //! Concrete accumulator reached from the modified-OTLP //! `Metric.data = HLLSketch{…}` hot path (PR C-CountSketch follow-up). @@ -12,7 +12,7 @@ //! store round-trip works end-to-end without that richer query surface. use crate::storage_engines::types::{AggregateCore, AggregationType, KeyByLabelValues, SerializableToSink}; -use asap_sketchlib::sketches::hll::{HllSketch, HllSketchDelta, HllVariant}; +use asap_sketchlib::{HllSketch, HllSketchDelta, HllVariant, MessagePackCodec}; use serde_json::Value; use std::collections::HashMap; @@ -36,7 +36,7 @@ impl HllSketchAccumulator { /// `HllSketch` struct — PR I parity entrypoint. pub fn from_msgpack_bytes(buffer: &[u8]) -> Result> { Ok(Self { - inner: HllSketch::deserialize_msgpack(buffer) + inner: HllSketch::from_msgpack(buffer) .map_err(|e| format!("deserialize HllSketch msgpack: {e}"))?, }) } @@ -152,7 +152,7 @@ impl SerializableToSink for HllSketchAccumulator { } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } @@ -234,7 +234,7 @@ impl AggregateCore for HllSketchAccumulator { /// (linear-counting) and large-range (32-bit space) corrections /// from the original Flajolet et al. paper. /// -/// Inlined here rather than added as a method on `asap_sketchlib::sketches::HllSketch` +/// Inlined here rather than added as a method on `asap_sketchlib::HllSketch` /// because the existing `asap_sketchlib::asap` types only expose merge / /// serialize today; adding a query method there would force a /// cross-crate change. @@ -296,6 +296,7 @@ mod tests { hip_kxq0, hip_kxq1, hip_est, + registers_sparse: None, }; state.encode_to_vec() } @@ -352,6 +353,7 @@ mod tests { hip_kxq0: 0.0, hip_kxq1: 0.0, hip_est: 0.0, + registers_sparse: None, }; let env = SketchEnvelope { sketch_state: Some(sketch_envelope::SketchState::Hll(state)), @@ -441,7 +443,7 @@ mod tests { 2.5, 42.0, ); - let bytes = original.serialize_msgpack().unwrap(); + let bytes = original.to_msgpack().unwrap(); let acc = HllSketchAccumulator::from_msgpack_bytes(&bytes).expect("decode ok"); assert_eq!(acc.inner.variant, HllVariant::Hip); assert_eq!(acc.inner.precision, 3); diff --git a/data_plane/src/precompute_engine/operators/hydra_kll_accumulator.rs b/data_plane/src/precompute_engine/operators/hydra_kll_accumulator.rs index 23fe32902..d53081398 100644 --- a/data_plane/src/precompute_engine/operators/hydra_kll_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/hydra_kll_accumulator.rs @@ -5,13 +5,13 @@ use crate::{ }, KeyByLabelValues, }; -use asap_sketchlib::sketches::hydra_kll::HydraKllSketch; +use asap_sketchlib::{HydraKllSketch, MessagePackCodec}; use base64::{engine::general_purpose, Engine as _}; use std::collections::HashMap; use promql_utilities::query_logics::enums::Statistic; -/// HydraKLL sketch accumulator — wraps asap_sketchlib::sketches::HydraKllSketch. +/// HydraKLL sketch accumulator — wraps asap_sketchlib::HydraKllSketch. /// Core struct, update/merge/serde logic live in `asap_sketchlib::sketches`. /// This file retains QE-specific trait impls and JSON output. #[derive(Debug, Clone)] @@ -43,13 +43,13 @@ impl HydraKllSketchAccumulator { impl SerializableToSink for HydraKllSketchAccumulator { fn serialize_to_json(&self) -> serde_json::Value { // Mirror Python implementation: {"sketch": base64_encoded_string} - let sketch_bytes = self.inner.serialize_msgpack().unwrap_or_default(); + let sketch_bytes = self.inner.to_msgpack().unwrap_or_default(); let sketch_b64 = general_purpose::STANDARD.encode(&sketch_bytes); serde_json::json!({ "sketch": sketch_b64 }) } fn serialize_to_bytes(&self) -> Vec { - self.inner.serialize_msgpack().unwrap_or_default() + self.inner.to_msgpack().unwrap_or_default() } } diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 9be157698..670347cab 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -1248,7 +1248,7 @@ mod tests { use crate::precompute_engine::operators::datasketches_kll_accumulator::DatasketchesKLLAccumulator; use crate::precompute_engine::operators::multiple_sum_accumulator::MultipleSumAccumulator; use crate::precompute_engine::operators::sum_accumulator::SumAccumulator; - use asap_sketchlib::sketches::kll::KllSketch; + use asap_sketchlib::KllSketch; use asap_types::enums::{AggregationType, WindowType}; fn make_agg_config( @@ -2283,7 +2283,7 @@ aggregations: // ----------------------------------------------------------------------- use crate::precompute_engine::operators::DDSketchAccumulator; - use asap_sketchlib::sketches::ddsketch::DdSketch; + use asap_sketchlib::DdSketch; /// Build a fresh DDSketch holding `vals` so each test has a real, /// non-empty sketch to push through `process_accumulator_input`. diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index ce6beb9db..9e5ac51c4 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -2573,7 +2573,7 @@ mod asap_tier_classify_tests { use asap_sketchlib::proto::sketchlib::{ sketch_envelope, HllVariant as ProtoVariant, HyperLogLogState, SketchEnvelope, }; - use asap_sketchlib::sketches::hll::{HllSketch, HllVariant}; + use asap_sketchlib::{HllSketch, HllVariant}; use prost::Message; let mut sk = HllSketch::new(HllVariant::Regular, precision); for i in 0..distinct { @@ -2585,7 +2585,8 @@ mod asap_tier_classify_tests { registers: sk.registers.clone(), hip_kxq0: sk.hip_kxq0, hip_kxq1: sk.hip_kxq1, - hip_est: sk.hip_est}; + hip_est: sk.hip_est, + registers_sparse: None}; let env = SketchEnvelope { sketch_state: Some(sketch_envelope::SketchState::Hll(state)), ..Default::default() @@ -3619,7 +3620,8 @@ mod outer_agg_integration_tests { use crate::storage_engines::sketch_db::index::{ AccuracyBound, Capability, SketchConfig, SketchEncoding, SketchStore, SketchInstanceMetadata, SketchKindHandle, SketchSampleState}; - use asap_sketchlib::sketches::ddsketch::DdSketch; + use asap_sketchlib::DdSketch; + use asap_sketchlib::MessagePackCodec; use std::collections::{BTreeMap, BTreeSet}; fn build_engine_with_index(idx: Arc) -> ASAPQueryEngine { @@ -3637,7 +3639,7 @@ mod outer_agg_integration_tests { for v in values { sk.update(*v); } - sk.serialize_msgpack().expect("ddsketch msgpack serialization") + sk.to_msgpack().expect("ddsketch msgpack serialization") } fn dd_meta_for(sid: u64, metric: &str, group_by: &[&str]) -> SketchInstanceMetadata { diff --git a/data_plane/src/storage_engines/sketch_db/query/decoders.rs b/data_plane/src/storage_engines/sketch_db/query/decoders.rs index 9aa32f18b..0471f9e5f 100644 --- a/data_plane/src/storage_engines/sketch_db/query/decoders.rs +++ b/data_plane/src/storage_engines/sketch_db/query/decoders.rs @@ -15,9 +15,10 @@ //! `ASAPTierError::DeserializeFailure` so the engine router falls over //! to archive cleanly. -use asap_sketchlib::sketches::countminsketch::CountMinSketch; -use asap_sketchlib::sketches::countminsketch_topk::CountMinSketchWithHeap; -use asap_sketchlib::sketches::countsketch::CountSketch; +use asap_sketchlib::CountMinSketch; +use asap_sketchlib::CountMinSketchWithHeap; +use asap_sketchlib::CountSketch; +use asap_sketchlib::MessagePackCodec; /// Decode a `CountMinSketch` from the modified-OTLP wire bytes. /// MSGPACK path round-trips `CountMinSketch::deserialize_msgpack`; @@ -91,7 +92,7 @@ pub fn decode_cms_from_proto(buffer: &[u8]) -> Result { /// format). Mirrors /// `CountMinSketchAccumulator::from_msgpack_bytes`. pub fn decode_cms_from_msgpack(buffer: &[u8]) -> Result { - CountMinSketch::deserialize_msgpack(buffer) + CountMinSketch::from_msgpack(buffer) .map_err(|e| format!("deserialize CountMinSketch msgpack: {e}")) } @@ -166,7 +167,7 @@ pub fn decode_cs_from_proto(buffer: &[u8]) -> Result { /// Decode a `CountSketch` from msgpack bytes (sketch-core wire format). pub fn decode_cs_from_msgpack(buffer: &[u8]) -> Result { - CountSketch::deserialize_msgpack(buffer) + CountSketch::from_msgpack(buffer) .map_err(|e| format!("deserialize CountSketch msgpack: {e}")) } @@ -174,8 +175,8 @@ pub fn decode_cs_from_msgpack(buffer: &[u8]) -> Result { /// `CountMinSketch` wire bytes when the gateway/precompute layer /// marked the sid as CmsWithHeap (heap embedded in the /// `CountMinSketchWithHeapSerialized` outer wrapper). Delegates to -/// `asap_sketchlib::sketches::CountMinSketchWithHeap::deserialize_msgpack`. +/// `asap_sketchlib::CountMinSketchWithHeap::deserialize_msgpack`. pub fn decode_cms_with_heap_from_msgpack(buffer: &[u8]) -> Result { - CountMinSketchWithHeap::deserialize_msgpack(buffer) + CountMinSketchWithHeap::from_msgpack(buffer) .map_err(|e| format!("deserialize CountMinSketchWithHeap msgpack: {e}")) } diff --git a/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs b/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs index 2400e913a..c1f7de834 100644 --- a/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs +++ b/data_plane/src/storage_engines/sketch_db/query/delta_apply.rs @@ -30,9 +30,10 @@ //! `Full`. For cumulative mode this means the cumulative answer //! starts at the first Full in the range, not at `t0`. -use asap_sketchlib::sketches::ddsketch::DdSketch; -use asap_sketchlib::sketches::hll::HllSketch; -use asap_sketchlib::sketches::kll::KllSketch; +use asap_sketchlib::DdSketch; +use asap_sketchlib::HllSketch; +use asap_sketchlib::KllSketch; +use asap_sketchlib::MessagePackCodec; use crate::storage_engines::sketch_db::index::{SketchEncoding, SketchSampleState}; @@ -60,7 +61,7 @@ fn decode_full( Ok(RollingState::Dd(sk)) } (DeltaSketchKind::DDSketch, SketchEncoding::MsgpackFull) => { - let sk = DdSketch::deserialize_msgpack(bytes) + let sk = DdSketch::from_msgpack(bytes) .map_err(|e| format!("deserialize DDSketch msgpack: {e}"))?; Ok(RollingState::Dd(sk)) } @@ -69,7 +70,7 @@ fn decode_full( Ok(RollingState::Hll(sk)) } (DeltaSketchKind::Hll, SketchEncoding::MsgpackFull) => { - let sk = HllSketch::deserialize_msgpack(bytes) + let sk = HllSketch::from_msgpack(bytes) .map_err(|e| format!("deserialize HllSketch msgpack: {e}"))?; Ok(RollingState::Hll(sk)) } @@ -78,7 +79,7 @@ fn decode_full( Ok(RollingState::Kll(sk)) } (DeltaSketchKind::Kll, SketchEncoding::MsgpackFull) => { - let sk = KllSketch::deserialize_msgpack(bytes) + let sk = KllSketch::from_msgpack(bytes) .map_err(|e| format!("deserialize KllSketch msgpack: {e}"))?; Ok(RollingState::Kll(sk)) } @@ -147,7 +148,7 @@ impl RollingState { // MsgpackDelta for HLL isn't a sparse encoding; // it's a serialized HllSketch fragment, mergeable // via `HllSketch::merge`. - let other = HllSketch::deserialize_msgpack(bytes) + let other = HllSketch::from_msgpack(bytes) .map_err(|e| format!("deserialize HllSketch (delta-as-msgpack): {e}"))?; sk.merge(&other) .map_err(|e| format!("merge HLL delta: {e}"))?; @@ -366,7 +367,7 @@ fn hll_from_proto(buffer: &[u8]) -> Result { use asap_sketchlib::proto::sketchlib::{ sketch_envelope, HllVariant as ProtoVariant, HyperLogLogState, SketchEnvelope, }; - use asap_sketchlib::sketches::hll::HllVariant; + use asap_sketchlib::HllVariant; use prost::Message; let state = match SketchEnvelope::decode(buffer) { Ok(env) => match env.sketch_state { @@ -416,7 +417,7 @@ fn hll_from_proto(buffer: &[u8]) -> Result { /// `(index, value)` updates, `register = max(register, value)`). fn apply_hll_proto_delta(sk: &mut HllSketch, buffer: &[u8]) -> Result<(), String> { use asap_otel_proto::sketchlib::v1::HllDelta as PbDelta; - use asap_sketchlib::sketches::hll::HllSketchDelta; + use asap_sketchlib::HllSketchDelta; use prost::Message; let pb = PbDelta::decode(buffer).map_err(|e| format!("decode HLLDelta: {e}"))?; diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index f68b31415..6593c9747 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -52,11 +52,12 @@ use std::collections::BTreeMap; use std::sync::Arc; -use asap_sketchlib::sketches::countminsketch::CountMinSketch; -use asap_sketchlib::sketches::countsketch::CountSketch; -use asap_sketchlib::sketches::ddsketch::DdSketch; -use asap_sketchlib::sketches::hll::HllSketch; -use asap_sketchlib::sketches::kll::KllSketch; +use asap_sketchlib::CountMinSketch; +use asap_sketchlib::CountSketch; +use asap_sketchlib::DdSketch; +use asap_sketchlib::HllSketch; +use asap_sketchlib::KllSketch; +use asap_sketchlib::MessagePackCodec; use crate::storage_engines::sketch_db::query::decoders::{ decode_cms_from_msgpack, decode_cms_from_proto, decode_cms_with_heap_from_msgpack, @@ -1160,7 +1161,7 @@ fn decode_ddsketch(sid: u64, state: &SketchSampleState) -> Result DdSketch::deserialize_msgpack(&state.bytes).map_err(|e| { + SketchEncoding::MsgpackFull => DdSketch::from_msgpack(&state.bytes).map_err(|e| { ASAPTierError::DeserializeFailure { sid, encoding: state.encoding, @@ -1192,7 +1193,7 @@ fn decode_kll(sid: u64, state: &SketchSampleState) -> Result KllSketch::deserialize_msgpack(&state.bytes).map_err(|e| { + SketchEncoding::MsgpackFull => KllSketch::from_msgpack(&state.bytes).map_err(|e| { ASAPTierError::DeserializeFailure { sid, encoding: state.encoding, @@ -1221,7 +1222,7 @@ fn decode_hll(sid: u64, state: &SketchSampleState) -> Result HllSketch::deserialize_msgpack(&state.bytes).map_err(|e| { + SketchEncoding::MsgpackFull => HllSketch::from_msgpack(&state.bytes).map_err(|e| { ASAPTierError::DeserializeFailure { sid, encoding: state.encoding, @@ -1311,7 +1312,7 @@ fn HllSketch_from_sketchlib_proto_bytes(buffer: &[u8]) -> Result match env.sketch_state { @@ -1362,11 +1363,11 @@ fn HllSketch_from_sketchlib_proto_bytes(buffer: &[u8]) -> Result Option { - CountMinSketch::deserialize_msgpack(buffer).ok() + CountMinSketch::from_msgpack(buffer).ok() } #[allow(dead_code)] fn _unused_count_sketch_kept_for_future_topk(buffer: &[u8]) -> Option { - CountSketch::deserialize_msgpack(buffer).ok() + CountSketch::from_msgpack(buffer).ok() } /// Decode a sid's per-window frequency sketch and emit a per-window diff --git a/data_plane/src/storage_engines/sketch_db/query/tests.rs b/data_plane/src/storage_engines/sketch_db/query/tests.rs index 808f84099..1b2d795fe 100644 --- a/data_plane/src/storage_engines/sketch_db/query/tests.rs +++ b/data_plane/src/storage_engines/sketch_db/query/tests.rs @@ -12,8 +12,9 @@ use std::collections::{BTreeMap, BTreeSet}; -use asap_sketchlib::sketches::ddsketch::DdSketch; -use asap_sketchlib::sketches::hll::{HllSketch, HllVariant}; +use asap_sketchlib::DdSketch; +use asap_sketchlib::MessagePackCodec; +use asap_sketchlib::{HllSketch, HllVariant}; use crate::storage_engines::sketch_db::query::{SketchReducer, ASAPTierError}; use crate::storage_engines::sketch_db::index::{ @@ -81,6 +82,7 @@ fn encode_hll(sk: &HllSketch) -> Vec { hip_kxq0: sk.hip_kxq0, hip_kxq1: sk.hip_kxq1, hip_est: sk.hip_est, + registers_sparse: None, }; let env = SketchEnvelope { sketch_state: Some(sketch_envelope::SketchState::Hll(state)), @@ -463,7 +465,7 @@ fn multi_series_one_per_label_value() { // TODO-1 tests — CMS-with-heap top-k. // --------------------------------------------------------------------------- -use asap_sketchlib::sketches::countminsketch_topk::CountMinSketchWithHeap; +use asap_sketchlib::CountMinSketchWithHeap; fn cms_heap_meta(sid: u64) -> SketchInstanceMetadata { let cfg = SketchConfig::CountMin { rows: 4, cols: 256 }; @@ -535,7 +537,7 @@ fn cms_with_heap_topk_returns_top_items() { cms.update(k, 1.0); } } - let bytes = cms.serialize_msgpack().expect("serialize cms with heap"); + let bytes = cms.to_msgpack().expect("serialize cms with heap"); idx.append_sample(sid, BTreeMap::new(), (1000, 1010), msgpack_full(bytes)); let reducer = SketchReducer::new(&idx); @@ -576,7 +578,7 @@ fn cms_without_heap_returns_missing_heap() { // sketch-kind side before decoding bytes. let mut cms = CountMinSketchWithHeap::new(4, 256, 20); cms.update("foo", 1.0); - let bytes = cms.serialize_msgpack().expect("serialize"); + let bytes = cms.to_msgpack().expect("serialize"); idx.append_sample(sid, BTreeMap::new(), (1000, 1010), msgpack_full(bytes)); let reducer = SketchReducer::new(&idx); @@ -696,7 +698,7 @@ fn hll_cumulative_full_plus_one_delta() { for i in 500..1000 { sk2.update(format!("user-{i}").as_bytes()); } - let bytes2 = sk2.serialize_msgpack().expect("serialize HLL msgpack"); + let bytes2 = sk2.to_msgpack().expect("serialize HLL msgpack"); let delta_sample = SketchSampleState { bytes: bytes2, encoding: SketchEncoding::MsgpackDelta, diff --git a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs index 9fd1c2faa..b5d79475c 100644 --- a/data_plane/tests/e2e_controller_plans_and_backend_serves.rs +++ b/data_plane/tests/e2e_controller_plans_and_backend_serves.rs @@ -53,6 +53,7 @@ use asap_sketchlib::proto::sketchlib::{ CountMinState, CountSketchState, CounterType, DdSketchState, HllVariant as ProtoHllVariant, HyperLogLogState, KllState, }; +use asap_sketchlib::MessagePackCodec; use control_plane::types::SketchType; use prost::Message; @@ -428,6 +429,9 @@ fn build_kll_state(k: u32, items: Vec) -> KllState { levels: Vec::new(), items, coin: None, + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), } } @@ -493,6 +497,7 @@ fn build_hll_state(precision: u32, registers: Vec) -> HyperLogLogState { hip_kxq0: 0.0, hip_kxq1: 0.0, hip_est: 0.0, + registers_sparse: None, } } @@ -1470,14 +1475,14 @@ async fn controller_plan_to_query_full_roundtrip_count_min_sketch() { /// with the supplied `(key, count)` pairs. Returns the bytes ready /// for the OTLP DP's `sketch` field with `encoding=MSGPACK`. fn build_heap_bearing_msgpack(rows: usize, cols: usize, top_k: usize, items: &[(&str, u64)]) -> Vec { - use asap_sketchlib::sketches::countminsketch_topk::CountMinSketchWithHeap; + use asap_sketchlib::CountMinSketchWithHeap; let mut cms = CountMinSketchWithHeap::new(rows, cols, top_k); for (key, count) in items { for _ in 0..*count { cms.update(key, 1.0); } } - cms.serialize_msgpack() + cms.to_msgpack() .expect("CountMinSketchWithHeap::serialize_msgpack should not fail") } diff --git a/data_plane/tests/e2e_modified_otlp_sketch_path.rs b/data_plane/tests/e2e_modified_otlp_sketch_path.rs index 14edf5c2b..1030b733b 100644 --- a/data_plane/tests/e2e_modified_otlp_sketch_path.rs +++ b/data_plane/tests/e2e_modified_otlp_sketch_path.rs @@ -36,6 +36,7 @@ use asap_sketchlib::proto::sketchlib::{ CountMinState, CountSketchState, CounterType, DdSketchState, HllVariant as ProtoHllVariant, HyperLogLogState, KllState, }; +use asap_sketchlib::MessagePackCodec; use asap_types::aggregation_config::AggregationConfig; use asap_types::enums::{AggregationType, WindowType}; use prost::Message; @@ -580,6 +581,9 @@ fn build_kll_state(k: u32, items: Vec) -> KllState { levels: Vec::new(), items, coin: None, + offset: 0.0, + value_scale: 0, + residuals: Vec::new(), } } @@ -937,6 +941,7 @@ fn build_hll_state(precision: u32, registers: Vec) -> HyperLogLogState { hip_kxq0: 0.0, hip_kxq1: 0.0, hip_est: 0.0, + registers_sparse: None, } } @@ -1185,11 +1190,11 @@ async fn e2e_count_min_sketch_msgpack_modified_otlp_path() { // is what the Go producer (sketchlib-go) will emit once PR I's matching // Go-side work lands. let mut cms = - asap_sketchlib::sketches::countminsketch::CountMinSketch::new(rows as usize, cols as usize); + asap_sketchlib::CountMinSketch::new(rows as usize, cols as usize); cms.update("user_a", 1.0); cms.update("user_b", 1.0); cms.update("user_a", 1.0); - let sketch_bytes = cms.serialize_msgpack().expect("serialize CMS msgpack"); + let sketch_bytes = cms.to_msgpack().expect("serialize CMS msgpack"); let client = reqwest::Client::new(); let req = build_count_min_msgpack_export_request( @@ -1202,13 +1207,13 @@ async fn e2e_count_min_sketch_msgpack_modified_otlp_path() { // Watermark advance using an empty msgpack sketch. let empty = - asap_sketchlib::sketches::countminsketch::CountMinSketch::new(rows as usize, cols as usize); + asap_sketchlib::CountMinSketch::new(rows as usize, cols as usize); let watermark_req = build_count_min_msgpack_export_request( metric_name, service_label, 2_000_000_000, empty - .serialize_msgpack() + .to_msgpack() .expect("serialize empty CMS msgpack"), ); post_otlp_http(&client, otlp_http_port, watermark_req).await; diff --git a/data_plane/tests/edge_runtime_consumes_precompute_rs.rs b/data_plane/tests/edge_runtime_consumes_precompute_rs.rs index 78a1d99f9..9d71194a5 100644 --- a/data_plane/tests/edge_runtime_consumes_precompute_rs.rs +++ b/data_plane/tests/edge_runtime_consumes_precompute_rs.rs @@ -239,7 +239,7 @@ fn kll_envelope_structural_assertions() { #[test] #[ignore = "blocked on ASAPCollector#243 HLL/CS/CMS byte parity"] fn hll_envelope_round_trip_through_backend_adapter() { - use asap_sketchlib::sketches::HllVariant; + use asap_sketchlib::HllVariant; let mut w = HLLWrapper::new(HllVariant::Regular, 12); for i in 0..1000u32 { w.update(&i.to_le_bytes());