Migrate data_plane to the restructured sketchlib API

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<u8> {
         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)),

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::{

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() {

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;

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<Self, Box<dyn std::error::Error>> {
         Ok(Self {
-            inner: CountMinSketch::deserialize_msgpack(buffer)
+            inner: CountMinSketch::from_msgpack(buffer)
                 .map_err(|e| -> Box<dyn std::error::Error> { e.to_string().into() })?,
         })
     }
@@ -309,7 +309,7 @@ impl SerializableToSink for CountMinSketchAccumulator {
     }
 
     fn serialize_to_bytes(&self) -> Vec<u8> {
-        self.inner.serialize_msgpack().unwrap_or_default()
+        self.inner.to_msgpack().unwrap_or_default()
     }
 }
 

data_plane/src/precompute_engine/operators/count_min_sketch_with_heap_accumulator.rs

@@ -2,22 +2,22 @@ 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 {
     pub inner: CountMinSketchWithHeap,
 }
 
 // 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<u8> {
-        self.inner.serialize_msgpack().unwrap_or_default()
+        self.inner.to_msgpack().unwrap_or_default()
     }
 }
 

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<Self, Box<dyn std::error::Error>> {
         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<u8> {
-        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<Vec<f64>>, 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);

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<Self, Box<dyn std::error::Error>> {
         Ok(Self {
-            inner: KllSketch::deserialize_msgpack(buffer)
+            inner: KllSketch::from_msgpack(buffer)
                 .map_err(|e| -> Box<dyn std::error::Error> { e.to_string().into() })?,
         })
     }
@@ -207,7 +207,7 @@ impl SerializableToSink for DatasketchesKLLAccumulator {
     }
 
     fn serialize_to_bytes(&self) -> Vec<u8> {
-        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);