feat: MessagePack encoding parity for modified-OTLP sketch hot path (PR I)

asap-common/dependencies/rs/asap_otel_proto/proto/opentelemetry/proto/metrics/v1/metrics.proto

@@ -384,6 +384,11 @@ enum DDSketchEncoding {
   // github.com/DataDog/sketches-go/ddsketch/pb/sketchpb.DDSketch message.
   DDSKETCH_ENCODING_PROTO = 1;
   DDSKETCH_ENCODING_PROTO_DELTA = 2;
+  // DDSKETCH_ENCODING_MSGPACK indicates the sketch bytes are the MessagePack
+  // serialization of the cross-language sketch-core `DdSketch` struct
+  // (rmp_serde encoding on Rust, matching struct on Go via sketchlib-go).
+  DDSKETCH_ENCODING_MSGPACK = 3;
+  DDSKETCH_ENCODING_MSGPACK_DELTA = 4;
 }
 
 // KLLSketch represents the type of a metric that encodes measurements using the KLL quantile sketch.
@@ -415,6 +420,10 @@ message KLLSketchDataPoint {
 enum KLLSketchEncoding {
   KLL_SKETCH_ENCODING_UNSPECIFIED = 0;
   KLL_SKETCH_ENCODING_PROTO = 1;
+  // KLL_SKETCH_ENCODING_MSGPACK indicates the sketch bytes are the MessagePack
+  // serialization of the cross-language sketch-core `KllSketch` struct.
+  KLL_SKETCH_ENCODING_MSGPACK = 3;
+  KLL_SKETCH_ENCODING_MSGPACK_DELTA = 4;
 }
 
 // CountSketch represents the type of a metric that encodes frequency estimations using CountSketch.
@@ -446,6 +455,10 @@ enum CountSketchEncoding {
   COUNT_SKETCH_ENCODING_UNSPECIFIED = 0;
   COUNT_SKETCH_ENCODING_PROTO = 1;
   COUNT_SKETCH_ENCODING_DELTA = 2;
+  // COUNT_SKETCH_ENCODING_MSGPACK indicates the sketch bytes are the MessagePack
+  // serialization of the cross-language sketch-core `CountSketch` struct.
+  COUNT_SKETCH_ENCODING_MSGPACK = 3;
+  COUNT_SKETCH_ENCODING_MSGPACK_DELTA = 4;
 }
 
 // CountMinSketch represents the type of a metric encoding frequency estimations using CountMinSketch.
@@ -477,6 +490,11 @@ enum CountMinSketchEncoding {
   COUNT_MIN_SKETCH_ENCODING_UNSPECIFIED = 0;
   COUNT_MIN_SKETCH_ENCODING_PROTO = 1;
   COUNT_MIN_SKETCH_ENCODING_DELTA = 2;
+  // COUNT_MIN_SKETCH_ENCODING_MSGPACK indicates the sketch bytes are the
+  // MessagePack serialization of the cross-language sketch-core
+  // `CountMinSketch` wire struct.
+  COUNT_MIN_SKETCH_ENCODING_MSGPACK = 3;
+  COUNT_MIN_SKETCH_ENCODING_MSGPACK_DELTA = 4;
 }
 
 // HLLSketch represents the type of a metric that encodes cardinality estimations using HyperLogLog.
@@ -508,6 +526,11 @@ enum HLLSketchEncoding {
   HLL_SKETCH_ENCODING_UNSPECIFIED = 0;
   HLL_SKETCH_ENCODING_PROTO = 1;
   HLL_SKETCH_ENCODING_DELTA = 2;
+  // HLL_SKETCH_ENCODING_MSGPACK indicates the sketch bytes are the
+  // MessagePack serialization of the cross-language sketch-core
+  // `HllSketch` struct (registers + variant + precision + HIP state).
+  HLL_SKETCH_ENCODING_MSGPACK = 3;
+  HLL_SKETCH_ENCODING_MSGPACK_DELTA = 4;
 }
 
 // AggregationTemporality defines how a metric aggregator reports aggregated

asap-query-engine/src/drivers/ingest/otel.rs

@@ -701,43 +701,57 @@ fn decode_modified_otlp_sketch_bytes(
         DatasketchesKLLAccumulator, HllSketchAccumulator,
     };
 
-    // The encoding value is the raw i32 from the proto enum. We only
-    // accept ENCODING_PROTO (= 1) for now; ENCODING_PROTO_DELTA (= 2)
-    // and ENCODING_MSGPACK / ENCODING_MSGPACK_DELTA (PR I) are routed
-    // to a "not yet implemented" Err so the caller falls through to
-    // the §5.2 fallback.
+    // The encoding value is the raw i32 from the per-sketch encoding
+    // enum. All five sketch variants share the same wire tag layout for
+    // tags 1–4, so we can match on one set of constants here:
+    //
+    //   1  — ENCODING_PROTO          (PR B / PR C: sketchlib proto-encoded)
+    //   2  — ENCODING_PROTO_DELTA    (delta transmission; deferred — falls
+    //                                 through to §5.2 fallback)
+    //   3  — ENCODING_MSGPACK        (PR I: cross-language sketch-core
+    //                                 msgpack wire format — this dispatcher)
+    //   4  — ENCODING_MSGPACK_DELTA  (deferred same as PROTO_DELTA)
     const ENCODING_PROTO: i32 = 1;
-
-    if encoding != ENCODING_PROTO {
-        return Err(format!(
+    const ENCODING_MSGPACK: i32 = 3;
+
+    match encoding {
+        ENCODING_PROTO => match kind {
+            SketchKind::CountMin => Ok(Box::new(
+                CountMinSketchAccumulator::from_sketchlib_proto_bytes(bytes)?,
+            )),
+            SketchKind::CountSketch => Ok(Box::new(
+                CountSketchAccumulator::from_sketchlib_proto_bytes(bytes)?,
+            )),
+            SketchKind::Kll => Ok(Box::new(
+                DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(bytes)?,
+            )),
+            SketchKind::DdSketch => Ok(Box::new(DDSketchAccumulator::from_sketchlib_proto_bytes(
+                bytes,
+            )?)),
+            SketchKind::Hll => Ok(Box::new(HllSketchAccumulator::from_sketchlib_proto_bytes(
+                bytes,
+            )?)),
+        },
+        ENCODING_MSGPACK => match kind {
+            SketchKind::CountMin => Ok(Box::new(CountMinSketchAccumulator::from_msgpack_bytes(
+                bytes,
+            )?)),
+            SketchKind::CountSketch => {
+                Ok(Box::new(CountSketchAccumulator::from_msgpack_bytes(bytes)?))
+            }
+            SketchKind::Kll => Ok(Box::new(DatasketchesKLLAccumulator::from_msgpack_bytes(
+                bytes,
+            )?)),
+            SketchKind::DdSketch => Ok(Box::new(DDSketchAccumulator::from_msgpack_bytes(bytes)?)),
+            SketchKind::Hll => Ok(Box::new(HllSketchAccumulator::from_msgpack_bytes(bytes)?)),
+        },
+        _ => Err(format!(
             "modified-OTLP sketch encoding {encoding} not yet supported \
-             (only ENCODING_PROTO = 1 is wired today; deltas tracked in PR C, \
-             msgpack tracked in PR I)"
+             (PROTO = 1 and MSGPACK = 3 are wired; PROTO_DELTA = 2 and \
+             MSGPACK_DELTA = 4 are deferred — caller falls through to §5.2 \
+             fallback)"
         )
-        .into());
-    }
-
-    match kind {
-        SketchKind::CountMin => {
-            let acc = CountMinSketchAccumulator::from_sketchlib_proto_bytes(bytes)?;
-            Ok(Box::new(acc))
-        }
-        SketchKind::CountSketch => {
-            let acc = CountSketchAccumulator::from_sketchlib_proto_bytes(bytes)?;
-            Ok(Box::new(acc))
-        }
-        SketchKind::Kll => {
-            let acc = DatasketchesKLLAccumulator::from_sketchlib_proto_bytes(bytes)?;
-            Ok(Box::new(acc))
-        }
-        SketchKind::DdSketch => {
-            let acc = DDSketchAccumulator::from_sketchlib_proto_bytes(bytes)?;
-            Ok(Box::new(acc))
-        }
-        SketchKind::Hll => {
-            let acc = HllSketchAccumulator::from_sketchlib_proto_bytes(bytes)?;
-            Ok(Box::new(acc))
-        }
+        .into()),
     }
 }
 

asap-query-engine/src/precompute_operators/count_min_sketch_accumulator.rs

@@ -68,6 +68,18 @@ impl CountMinSketchAccumulator {
         })
     }
 
+    /// Decode from the modified OTLP wire format's
+    /// `CountMinSketchDataPoint.sketch` bytes when
+    /// `encoding = COUNT_MIN_SKETCH_ENCODING_MSGPACK`. The bytes are the
+    /// MessagePack serialization of the cross-language sketch-core
+    /// `CountMinSketch` wire struct (same format the legacy Arroyo path
+    /// 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)?,
+        })
+    }
+
     /// Decode from the modified OTLP wire format's
     /// `CountMinSketchDataPoint.sketch` bytes — i.e. the protobuf-encoded
     /// `asap_sketchlib::proto::sketchlib::CountMinState` message used by

asap-query-engine/src/precompute_operators/count_sketch_accumulator.rs

@@ -37,6 +37,18 @@ impl CountSketchAccumulator {
         }
     }
 
+    /// Decode from the modified OTLP wire format's
+    /// `CountSketchDataPoint.sketch` bytes when
+    /// `encoding = COUNT_SKETCH_ENCODING_MSGPACK`. The bytes are the
+    /// MessagePack serialization of the cross-language sketch-core
+    /// `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)
+                .map_err(|e| format!("deserialize CountSketch msgpack: {e}"))?,
+        })
+    }
+
     /// Decode from the modified OTLP wire format's
     /// `CountSketchDataPoint.sketch` bytes — the protobuf-encoded
     /// `asap_sketchlib::proto::sketchlib::CountSketchState` message
@@ -301,4 +313,24 @@ mod tests {
         let result = cs.merge_with(&cms);
         assert!(result.is_err());
     }
+
+    #[test]
+    fn test_from_msgpack_bytes_round_trip() {
+        let original = CountSketch::from_legacy_matrix(
+            vec![vec![1.0, -2.0, 3.0], vec![-4.0, 5.0, -6.0]],
+            2,
+            3,
+        );
+        let bytes = original.serialize_msgpack();
+        let acc = CountSketchAccumulator::from_msgpack_bytes(&bytes).expect("decode ok");
+        assert_eq!(acc.inner.row_num, 2);
+        assert_eq!(acc.inner.col_num, 3);
+        assert_eq!(acc.inner.sketch(), original.sketch());
+    }
+
+    #[test]
+    fn test_from_msgpack_bytes_rejects_garbage() {
+        let result = CountSketchAccumulator::from_msgpack_bytes(b"not valid msgpack");
+        assert!(result.is_err());
+    }
 }

asap-query-engine/src/precompute_operators/datasketches_kll_accumulator.rs

@@ -70,6 +70,21 @@ impl DatasketchesKLLAccumulator {
         })
     }
 
+    /// Decode from the modified OTLP wire format's
+    /// `KLLSketchDataPoint.sketch` bytes when
+    /// `encoding = KLL_SKETCH_ENCODING_MSGPACK`. The bytes are the
+    /// MessagePack serialization of the cross-language sketch-core
+    /// `KllSketch` struct — PR I parity entrypoint. Unlike the
+    /// `_ENCODING_PROTO` path (which does lossy statistical
+    /// reconstruction via `update()` replay), the msgpack path is a
+    /// bit-identical round-trip because sketch-core's `KllSketch`
+    /// 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)?,
+        })
+    }
+
     /// Decode from the modified OTLP wire format's
     /// `KLLSketchDataPoint.sketch` bytes — the protobuf-encoded
     /// `asap_sketchlib::proto::sketchlib::KllState` message that

asap-query-engine/src/precompute_operators/dd_sketch_accumulator.rs

@@ -29,6 +29,18 @@ impl DDSketchAccumulator {
         }
     }
 
+    /// Decode from the modified OTLP wire format's
+    /// `DDSketchDataPoint.sketch` bytes when
+    /// `encoding = DDSKETCH_ENCODING_MSGPACK`. The bytes are the
+    /// MessagePack serialization of the cross-language sketch-core
+    /// `DdSketch` struct — PR I parity entrypoint.
+    pub fn from_msgpack_bytes(buffer: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
+        Ok(Self {
+            inner: DdSketch::deserialize_msgpack(buffer)
+                .map_err(|e| format!("deserialize DdSketch msgpack: {e}"))?,
+        })
+    }
+
     /// Decode from the modified OTLP wire format's
     /// `DDSketchDataPoint.sketch` bytes — the protobuf-encoded
     /// `asap_sketchlib::proto::sketchlib::DDSketchState` message that
@@ -206,4 +218,22 @@ mod tests {
         let cs = CountSketchAccumulator::new(2, 3);
         assert!(dd.merge_with(&cs).is_err());
     }
+
+    #[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();
+        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]);
+        assert_eq!(acc.inner.store_offset, -2);
+        assert_eq!(acc.inner.count, 50);
+        assert_eq!(acc.inner.sum, 150.0);
+    }
+
+    #[test]
+    fn test_from_msgpack_bytes_rejects_garbage() {
+        let result = DDSketchAccumulator::from_msgpack_bytes(b"not valid msgpack");
+        assert!(result.is_err());
+    }
 }

asap-query-engine/src/precompute_operators/hll_sketch_accumulator.rs

@@ -29,6 +29,18 @@ impl HllSketchAccumulator {
         }
     }
 
+    /// Decode from the modified OTLP wire format's
+    /// `HLLSketchDataPoint.sketch` bytes when
+    /// `encoding = HLL_SKETCH_ENCODING_MSGPACK`. The bytes are the
+    /// MessagePack serialization of the cross-language sketch-core
+    /// `HllSketch` struct — PR I parity entrypoint.
+    pub fn from_msgpack_bytes(buffer: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
+        Ok(Self {
+            inner: HllSketch::deserialize_msgpack(buffer)
+                .map_err(|e| format!("deserialize HllSketch msgpack: {e}"))?,
+        })
+    }
+
     /// Decode from the modified OTLP wire format's
     /// `HLLSketchDataPoint.sketch` bytes — the protobuf-encoded
     /// `asap_sketchlib::proto::sketchlib::HyperLogLogState` message
@@ -258,4 +270,28 @@ mod tests {
         let cs = CountSketchAccumulator::new(2, 3);
         assert!(hll.merge_with(&cs).is_err());
     }
+
+    #[test]
+    fn test_from_msgpack_bytes_round_trip() {
+        let original = HllSketch::from_raw(
+            HllVariant::Hip,
+            3,
+            vec![0, 1, 2, 3, 4, 5, 6, 7],
+            1.5,
+            2.5,
+            42.0,
+        );
+        let bytes = original.serialize_msgpack();
+        let acc = HllSketchAccumulator::from_msgpack_bytes(&bytes).expect("decode ok");
+        assert_eq!(acc.inner.variant, HllVariant::Hip);
+        assert_eq!(acc.inner.precision, 3);
+        assert_eq!(acc.inner.registers, vec![0, 1, 2, 3, 4, 5, 6, 7]);
+        assert_eq!(acc.inner.hip_kxq0, 1.5);
+    }
+
+    #[test]
+    fn test_from_msgpack_bytes_rejects_garbage() {
+        let result = HllSketchAccumulator::from_msgpack_bytes(b"not valid msgpack");
+        assert!(result.is_err());
+    }
 }

asap-query-engine/src/stores/simple_map_store/per_key.rs

@@ -347,6 +347,21 @@ impl SimpleMapStorePerKey {
         })
     }
 
+    /// Total number of times the insert hot path has blocked on
+    /// flusher back-pressure (i.e. observed sealed memory at or above
+    /// `hard_cap_bytes` and entered `wait_for_memory_under`'s wait
+    /// loop). Returns 0 when persistence is disabled. Monotonic.
+    ///
+    /// Used as a timing-independent signal in back-pressure tests —
+    /// wall-clock thresholds are flaky on fast CI runners where a
+    /// flush tick completes in under a millisecond.
+    pub fn back_pressure_wait_count(&self) -> u64 {
+        match self.persistence.as_ref() {
+            Some(state) => state.flusher.back_pressure_wait_count(),
+            None => 0,
+        }
+    }
+
     /// Collect diagnostic info about store contents.
     pub fn diagnostic_info(&self) -> super::StoreDiagnostics {
         use super::{AggregationDiagnostic, StoreDiagnostics};

asap-query-engine/src/stores/simple_map_store/persistence/cache.rs

@@ -36,7 +36,7 @@ impl PartCache {
                     .weigher(|_k: &PartId, v: &LoadedPart| -> u32 {
                         // Weight = sum of mmap'd bytes. Moka's weigher
                         // returns u32, so clamp large parts.
-                        let len = v.meta.data_len as u64 + v.meta.index_len as u64;
+                        let len = v.meta.data_len + v.meta.index_len;
                         len.min(u32::MAX as u64) as u32
                     })
                     .max_capacity(byte_budget)