Fix CardinalityCollector deserialization: use deserialize_byte_buf

The Serialize impl uses serializer.serialize_bytes() which hints the
serializer to use its bytes-optimized path. The Deserialize impl was
using Vec<u8>::deserialize() which goes through deserialize_seq(),
a different deserialization path. While the wire formats are theoretically
identical in postcard for u8 sequences, using deserialize_byte_buf
explicitly pairs with serialize_bytes and eliminates any potential
mismatch in binary serializers.

The visitor implements visit_bytes, visit_borrowed_bytes, visit_byte_buf,
and visit_seq to handle all possible deserialization paths gracefully.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: congx4

src/aggregation/metric/cardinality.rs

@@ -341,7 +341,43 @@ impl Serialize for CardinalityCollector {
 
 impl<'de> Deserialize<'de> for CardinalityCollector {
     fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
-        let bytes: Vec<u8> = Deserialize::deserialize(deserializer)?;
+        struct HllBytesVisitor;
+
+        impl<'de> serde::de::Visitor<'de> for HllBytesVisitor {
+            type Value = Vec<u8>;
+
+            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+                f.write_str("HLL sketch bytes")
+            }
+
+            fn visit_bytes<E: serde::de::Error>(self, v: &[u8]) -> Result<Vec<u8>, E> {
+                Ok(v.to_vec())
+            }
+
+            fn visit_borrowed_bytes<E: serde::de::Error>(
+                self,
+                v: &'de [u8],
+            ) -> Result<Vec<u8>, E> {
+                Ok(v.to_vec())
+            }
+
+            fn visit_byte_buf<E: serde::de::Error>(self, v: Vec<u8>) -> Result<Vec<u8>, E> {
+                Ok(v)
+            }
+
+            fn visit_seq<A: serde::de::SeqAccess<'de>>(
+                self,
+                mut seq: A,
+            ) -> Result<Vec<u8>, A::Error> {
+                let mut bytes = Vec::with_capacity(seq.size_hint().unwrap_or(0));
+                while let Some(byte) = seq.next_element()? {
+                    bytes.push(byte);
+                }
+                Ok(bytes)
+            }
+        }
+
+        let bytes = deserializer.deserialize_byte_buf(HllBytesVisitor)?;
         let sketch = HllSketch::deserialize(&bytes).map_err(serde::de::Error::custom)?;
         Ok(Self { sketch, salt: 0 })
     }
@@ -558,6 +594,218 @@ mod tests {
         assert_eq!(original_estimate, 2.0);
     }
 
+    #[test]
+    fn cardinality_collector_postcard_roundtrip() {
+        use super::CardinalityCollector;
+
+        let mut collector = CardinalityCollector::default();
+        collector.insert("hello");
+        collector.insert("world");
+        collector.insert("hello");
+
+        let original_estimate = collector.clone().finalize().unwrap();
+
+        let serialized = postcard::to_allocvec(&collector).expect("postcard serialize failed");
+        let deserialized: CardinalityCollector =
+            postcard::from_bytes(&serialized).expect("postcard deserialize failed");
+
+        let roundtrip_estimate = deserialized.finalize().unwrap();
+        assert_eq!(original_estimate, roundtrip_estimate);
+        assert_eq!(original_estimate, 2.0);
+    }
+
+    #[test]
+    fn cardinality_collector_postcard_bytes_fidelity() {
+        use super::CardinalityCollector;
+
+        let mut collector = CardinalityCollector::default();
+        for i in 0..10u64 {
+            collector.insert(i);
+        }
+
+        let hll_bytes = collector.sketch.serialize();
+        println!("HLL bytes len: {}, first 16: {:?}", hll_bytes.len(), &hll_bytes[..16.min(hll_bytes.len())]);
+
+        let postcard_bytes = postcard::to_allocvec(&collector).unwrap();
+        println!("Postcard bytes len: {}", postcard_bytes.len());
+
+        let deserialized: CardinalityCollector = postcard::from_bytes(&postcard_bytes).unwrap();
+        let hll_bytes_after = deserialized.sketch.serialize();
+        println!(
+            "HLL bytes after roundtrip len: {}, first 16: {:?}",
+            hll_bytes_after.len(),
+            &hll_bytes_after[..16.min(hll_bytes_after.len())]
+        );
+
+        assert_eq!(hll_bytes, hll_bytes_after, "HLL bytes should be identical after Postcard roundtrip");
+    }
+
+    #[test]
+    fn cardinality_collector_postcard_roundtrip_large() {
+        use super::CardinalityCollector;
+
+        let mut collector = CardinalityCollector::default();
+        for i in 0..1000u64 {
+            collector.insert(i);
+        }
+
+        let original_estimate = collector.clone().finalize().unwrap();
+        assert!((original_estimate - 1000.0).abs() < 50.0);
+
+        let serialized = postcard::to_allocvec(&collector).expect("postcard serialize failed");
+        println!(
+            "Large HLL sketch serialized to {} postcard bytes",
+            serialized.len()
+        );
+        let deserialized: CardinalityCollector =
+            postcard::from_bytes(&serialized).expect("postcard deserialize failed");
+
+        let roundtrip_estimate = deserialized.finalize().unwrap();
+        assert_eq!(original_estimate, roundtrip_estimate);
+    }
+
+    #[test]
+    fn cardinality_intermediate_metric_result_postcard_roundtrip() {
+        use super::CardinalityCollector;
+        use crate::aggregation::intermediate_agg_result::IntermediateMetricResult;
+
+        let mut collector = CardinalityCollector::default();
+        collector.insert("hello");
+        collector.insert("world");
+
+        let intermediate = IntermediateMetricResult::Cardinality(collector);
+        let serialized =
+            postcard::to_allocvec(&intermediate).expect("postcard serialize failed");
+        let deserialized: IntermediateMetricResult =
+            postcard::from_bytes(&serialized).expect("postcard deserialize failed");
+
+        match deserialized {
+            IntermediateMetricResult::Cardinality(c) => {
+                assert_eq!(c.finalize().unwrap(), 2.0);
+            }
+            _ => panic!("expected Cardinality variant"),
+        }
+    }
+
+    #[test]
+    fn cardinality_postcard_multisegment_roundtrip() {
+        use crate::aggregation::agg_req::Aggregations;
+        use crate::aggregation::collector::AggregationCollector;
+        use crate::aggregation::intermediate_agg_result::IntermediateAggregationResults;
+        use crate::aggregation::AggContextParams;
+        use crate::collector::{Collector, SegmentCollector};
+        use crate::query::AllQuery;
+
+        let segment_and_terms = vec![
+            vec!["terma"],
+            vec!["termb"],
+            vec!["termc"],
+            vec!["terma"],
+        ];
+        let index = get_test_index_from_terms(false, &segment_and_terms).unwrap();
+
+        let agg_req: Aggregations = serde_json::from_value(json!({
+            "cardinality": {
+                "cardinality": {
+                    "field": "string_id",
+                }
+            },
+        }))
+        .unwrap();
+
+        let collector = AggregationCollector::from_aggs(
+            agg_req,
+            AggContextParams::new(Default::default(), index.tokenizers().clone()),
+        );
+
+        let reader = index.reader().unwrap();
+        let searcher = reader.searcher();
+
+        let segments = searcher.segment_readers();
+        assert!(
+            segments.len() > 1,
+            "Need multiple segments for this test, got {}",
+            segments.len()
+        );
+
+        // Collect from each segment individually and serialize via Postcard
+        let serialized_results: Vec<Vec<u8>> = segments
+            .iter()
+            .enumerate()
+            .map(|(ord, segment_reader)| {
+                let mut segment_collector = collector
+                    .for_segment(ord as u32, segment_reader)
+                    .unwrap();
+                for doc in segment_reader.doc_ids_alive() {
+                    segment_collector.collect(doc, 0.0);
+                }
+                let fruit = segment_collector.harvest().unwrap();
+                postcard::to_allocvec(&fruit).expect("postcard serialize should work")
+            })
+            .collect();
+
+        // Deserialize and merge (this is what quickwit does)
+        let merged: IntermediateAggregationResults = serialized_results
+            .iter()
+            .map(|bytes| {
+                postcard::from_bytes::<IntermediateAggregationResults>(bytes)
+                    .expect("postcard deserialize should work")
+            })
+            .fold(None, |acc: Option<IntermediateAggregationResults>, fruits| {
+                match acc {
+                    Some(mut merged) => {
+                        merged.merge_fruits(fruits).unwrap();
+                        Some(merged)
+                    }
+                    None => Some(fruits),
+                }
+            })
+            .unwrap();
+
+        // Verify the merged result can be serialized again
+        let _final_bytes =
+            postcard::to_allocvec(&merged).expect("final postcard serialize should work");
+    }
+
+    #[test]
+    fn cardinality_full_intermediate_agg_results_postcard_roundtrip() {
+        use super::CardinalityCollector;
+        use crate::aggregation::intermediate_agg_result::{
+            IntermediateAggregationResult, IntermediateAggregationResults,
+            IntermediateMetricResult,
+        };
+
+        let mut collector = CardinalityCollector::default();
+        collector.insert("hello");
+        collector.insert("world");
+
+        let mut results = IntermediateAggregationResults::default();
+        results
+            .push(
+                "test_card".to_string(),
+                IntermediateAggregationResult::Metric(
+                    IntermediateMetricResult::Cardinality(collector),
+                ),
+            )
+            .unwrap();
+
+        let serialized =
+            postcard::to_allocvec(&results).expect("postcard serialize failed");
+        let deserialized: IntermediateAggregationResults =
+            postcard::from_bytes(&serialized).expect("postcard deserialize failed");
+
+        let result = deserialized
+            .aggs_res
+            .get("test_card")
+            .expect("missing key");
+        match result {
+            IntermediateAggregationResult::Metric(IntermediateMetricResult::Cardinality(c)) => {
+                assert_eq!(c.clone().finalize().unwrap(), 2.0);
+            }
+            _ => panic!("expected Cardinality variant"),
+        }
+    }
+
     #[test]
     fn cardinality_collector_merge() {
         use super::CardinalityCollector;