Use shared value fetching entry point for bucket collectors

src/aggregation/bucket/histogram/histogram.rs

@@ -492,11 +492,7 @@ impl<B: BucketIdSlot, const SOURCE_CONTAINS_MULTIVALUES: bool> SegmentAggregatio
         let get_bucket_pos = |val| get_bucket_pos_f64(val, interval, offset) as i64;
 
         let accessor = &mut agg_data.column_block_accessor;
-        if SOURCE_CONTAINS_MULTIVALUES {
-            accessor.fetch_block_with_missing_unique_per_doc(docs, &*req.accessor, None, false);
-        } else {
-            accessor.fetch_block(docs, &*req.accessor);
-        }
+        accessor.fetch_block_with_missing_unique_per_doc(docs, &*req.accessor, None, false);
         // Known single-valued sources compile out deduplication; otherwise check the loaded batch.
         let multivalued = SOURCE_CONTAINS_MULTIVALUES && accessor.is_batch_multivalued();
         // Document IDs are needed for child collection and multivalued deduplication.

src/aggregation/bucket/range.rs

@@ -283,16 +283,12 @@ impl<B: SubAggBuffer, const SOURCE_CONTAINS_MULTIVALUES: bool> SegmentAggregatio
         agg_data: &mut AggregationsSegmentCtx,
     ) -> crate::Result<()> {
         let accessor = &mut agg_data.column_block_accessor;
-        if SOURCE_CONTAINS_MULTIVALUES {
-            accessor.fetch_block_with_missing_unique_per_doc(
-                docs,
-                &*self.req_data.accessor,
-                None,
-                false,
-            );
-        } else {
-            accessor.fetch_block(docs, &*self.req_data.accessor);
-        }
+        accessor.fetch_block_with_missing_unique_per_doc(
+            docs,
+            &*self.req_data.accessor,
+            None,
+            false,
+        );
 
         let buckets = &mut self.parent_buckets[parent_bucket_id as usize];
 

src/aggregation/value_source/block_accessor.rs

@@ -41,8 +41,13 @@ pub(crate) struct ColumnBlockAccessor {
 }
 
 impl ColumnBlockAccessor {
+    /// Fetches a block without deduplicating values within each document.
+    ///
+    /// Bucket aggregations on multivalued sources should use
+    /// [`Self::fetch_block_with_missing_unique_per_doc`] instead, so duplicate values do not
+    /// inflate document counts.
     #[inline]
-    pub(crate) fn fetch_block<S: ValueSource + ?Sized>(&mut self, docs: &[DocId], source: &S) {
+    fn fetch_block<S: ValueSource + ?Sized>(&mut self, docs: &[DocId], source: &S) {
         self.cardinality = source.load_block(
             docs,
             &mut self.val_cache,