feat(promql): route evaluated fragments to their canonical owner (#10038)

* feat(promql): route evaluated chains independently by canonical key

Signed-off-by: discord9 <discord9@163.com>

* test(promql): verify canonical routing key semantics

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): release closed-output chains and yield routing producers

Signed-off-by: discord9 <discord9@163.com>

* test(promql): check closed-chain budget before releasing the child

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): account histogram slices and verify routing completion

Signed-off-by: discord9 <discord9@163.com>

* fix(promql): finalize contiguous series at repartition owners

Wait for successful producer EOF before publishing exact complete metric-name/tag identities. Reuse the existing sample merge for position ordering, complementary fields, and conflicting-cell errors; recompute one final close per owner identity.

Keep N producer tasks and M outputs. Admit nonwaiting queue payloads through the query pool, reserve merge workspace, and retain cancellation, failure, memory-release and unread-owner coverage.

Signed-off-by: discord9 <discord9@163.com>

* feat(promql): publish completed repartition identities incrementally

Signed-off-by: discord9 <discord9@163.com>

* test(promql): expose histogram amplification in public repartition

Signed-off-by: discord9 <discord9@163.com>

* fix(datatypes): encode only referenced histogram table rows

Signed-off-by: discord9 <discord9@163.com>

* test(promql): wait for queued retention before draining a failed read

Signed-off-by: discord9 <discord9@163.com>

* perf(promql): bound identity rescans and share repeated histogram rows

Signed-off-by: discord9 <discord9@163.com>

* docs(query): scope the evaluated validator to one chain

Signed-off-by: discord9 <discord9@163.com>

---------

Signed-off-by: discord9 <discord9@163.com>

src/common/query/src/series_frame.rs

@@ -571,6 +571,13 @@ impl SequenceValidator {
 
 /// Cross-chunk validation state for evaluated chains.
 ///
+/// This is a **chain-local** validator. It checks exactly three things: that a chunk does not change
+/// identity while its chain is open, that a chain closes exactly once, and that the positions of an
+/// open chain keep increasing across adjacent chunks. It keeps at most the one open chain, so it
+/// never certifies that a complete identity does not recur later in the stream; that global
+/// non-recurrence is a producer contract enforced where identities are completed (the repartition
+/// owner), not here.
+///
 /// `is_last` closes one chunk chain and nothing more. It is not the end of a label set: a range
 /// function or a label rewrite can make two input series show the same complete labels, and each of
 /// them closes itself, so a closed chain may be followed by another chain whose labels happen to be

src/common/query/src/series_frame/tests.rs

@@ -2185,10 +2185,10 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
     struct Expected {
         time: i64,
         float_bits: Option<u64>,
-        hist_index: Option<u32>,
+        hist_sum: Option<f64>,
     }
-    // Table rows: 0 => sum 10.5, 1 => sum -9.5, 2 => sum 7.25. Row-local indices stay
-    // unchanged, so equal indices in different rows keep the same sum.
+    // Table rows: 0 => sum 10.5, 1 => sum -9.5, 2 => sum 7.25. Each frame row compacts
+    // referenced histogram rows densely, so the same local index can carry different sums.
     let nan_bits = f64::from_bits(0x7ff8_0000_0000_0042).to_bits();
     let rows: [(Vec<Expected>, bool); 4] = [
         (
@@ -2196,12 +2196,12 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
                 Expected {
                     time: 10,
                     float_bits: Some(1.0_f64.to_bits()),
-                    hist_index: Some(1),
+                    hist_sum: Some(-9.5),
                 },
                 Expected {
                     time: 20,
                     float_bits: None,
-                    hist_index: None,
+                    hist_sum: None,
                 },
             ],
             false,
@@ -2210,7 +2210,7 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
             vec![Expected {
                 time: 30,
                 float_bits: Some(nan_bits),
-                hist_index: Some(0),
+                hist_sum: Some(10.5),
             }],
             false,
         ),
@@ -2218,7 +2218,7 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
             vec![Expected {
                 time: 40,
                 float_bits: Some((-0.0_f64).to_bits()),
-                hist_index: Some(0),
+                hist_sum: Some(10.5),
             }],
             false,
         ),
@@ -2226,12 +2226,11 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
             vec![Expected {
                 time: 50,
                 float_bits: Some(crate::prometheus::PROMETHEUS_STALE_NAN_BITS),
-                hist_index: Some(2),
+                hist_sum: Some(7.25),
             }],
             true,
         ),
     ];
-    let sums = [10.5_f64, -9.5, 7.25];
     for (expected, last) in rows {
         let chunk = decoded.next().await.unwrap().unwrap();
         assert_eq!(chunk.primary_key.as_slice(), b"series-key");
@@ -2250,12 +2249,12 @@ async fn multirow_ipc_roundtrip_preserves_cross_chunk_variable_segments() {
                 SampleRef::Missing => assert_eq!(expected.float_bits, None),
                 SampleRef::Histogram(_) => panic!("expected a float field"),
             }
-            match expected.hist_index {
-                Some(index_value) => {
+            match expected.hist_sum {
+                Some(sum) => {
                     assert!(hist.indices().is_valid(index));
-                    assert_eq!(hist.indices().value(index), index_value);
+                    assert_eq!(hist.indices().value(index), 0);
                     let view = hist.view_at(index).expect("histogram sample present");
-                    assert_eq!(view.sum.to_bits(), sums[index_value as usize].to_bits());
+                    assert_eq!(view.sum.to_bits(), sum.to_bits());
                     assert_eq!(view.schema, -7);
                     assert_eq!(view.reset_hint, CounterResetHint::Unrecognized(37));
                     assert_eq!(view.custom_values, [1.5, 4.5]);

src/common/query/src/series_frame/tests/evaluated.rs

@@ -1286,8 +1286,8 @@ async fn mixed_payload_roundtrips_without_dropping_either_kind() {
     assert_eq!(mixed.histograms().len(), 4);
     assert_eq!(
         mixed.histograms().batch().len(),
-        2,
-        "the referenced table survives whole"
+        1,
+        "only the referenced histogram row survives compaction"
     );
     match field.sample(0) {
         SampleRef::Float(value) => assert_eq!(value.to_bits(), 1.5_f64.to_bits()),
@@ -1297,10 +1297,12 @@ async fn mixed_payload_roundtrips_without_dropping_either_kind() {
         SampleRef::Histogram(histogram) => {
             assert_eq!(histogram.count.to_bits(), 17.25_f64.to_bits());
             assert_eq!(histogram.sum.to_bits(), (-9.5_f64).to_bits());
+            assert_eq!(histogram.start_timestamp, Some(-123));
         }
         _ => panic!("position 1 was a histogram"),
     }
     assert!(matches!(field.sample(2), SampleRef::Missing));
+    assert!(mixed.histograms().view_at(2).is_none());
     match field.sample(3) {
         // A NaN float stays a float value, and its histogram slot stays empty.
         SampleRef::Float(value) => assert!(value.is_nan()),

src/common/query/src/series_frame/tests/lifecycle.rs

@@ -346,9 +346,9 @@ async fn streaming_round_trip_preserves_sliced_multifield_payload_and_owners() {
     assert_eq!(mixed_seen, [Some(1.25_f64.to_bits()), None, None]);
 }
 
-/// One row of one sample, with every array a nonzero-offset slice of a larger owner: the decoded
-/// chunk keeps the source buffers by pointer, and the values stay readable and bit-exact after
-/// every source owner is dropped.
+/// One row of one sample, with payload arrays as nonzero-offset slices of larger owners: the
+/// decoded chunk keeps the float and time buffers by pointer, and values stay readable and bit-exact
+/// after every source owner is dropped.
 #[tokio::test]
 async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners() {
     let schema = Arc::new(SeriesSchema {
@@ -388,9 +388,9 @@ async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners()
     let index_owner = UInt32Array::from(vec![Some(3), Some(0)]);
     let indices = index_owner.slice(1, 1);
     let hist = Samples::try_from_histogram(indices.clone(), sliced_table.clone()).unwrap();
-    let mixed_float_owner = Float64Array::from(vec![Some(0.0), Some(-0.0)]);
+    let mixed_float_owner = Float64Array::from(vec![Some(0.0), None]);
     let mixed_floats = mixed_float_owner.slice(1, 1);
-    let mixed_index_owner = UInt32Array::from(vec![Some(0), None]);
+    let mixed_index_owner = UInt32Array::from(vec![None, Some(0)]);
     let mixed_indices = mixed_index_owner.slice(1, 1);
     let mixed = Samples::try_from_mixed(
         mixed_floats.clone(),
@@ -409,9 +409,7 @@ async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners()
 
     let input_times_ptr = chunk.time.as_buffer().as_ptr();
     let input_float_ptr = float.values().as_ptr();
-    let input_hist_indices_ptr = indices.values().as_ptr();
     let input_mixed_float_ptr = mixed_floats.values().as_ptr();
-    let input_mixed_indices_ptr = mixed_indices.values().as_ptr();
 
     let input: SendableRawSeriesStream = Box::pin(RawFixture {
         schema: schema.clone(),
@@ -438,18 +436,9 @@ async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners()
     else {
         panic!("expected histogram")
     };
-    assert_eq!(
-        decoded_hist.indices().values().as_ptr(),
-        input_hist_indices_ptr
-    );
-    assert!(
-        decoded_hist
-            .batch()
-            .as_arrow()
-            .to_data()
-            .ptr_eq(&sliced_table.as_arrow().to_data()),
-        "a single row keeps the sliced histogram table by pointer"
-    );
+    assert!(decoded_hist.indices().is_valid(0));
+    assert_eq!(decoded_hist.indices().value(0), 0);
+    assert_histogram(decoded_hist.batch().view(0).unwrap(), -9.5, Some(456));
     let DenseSamplesRef::Mixed(decoded_mixed) = output[0].fields.get(2).unwrap().dense().unwrap()
     else {
         panic!("expected mixed")
@@ -458,17 +447,13 @@ async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners()
         decoded_mixed.floats().values().as_ptr(),
         input_mixed_float_ptr
     );
-    assert_eq!(
-        decoded_mixed.histograms().indices().values().as_ptr(),
-        input_mixed_indices_ptr
-    );
-    assert!(
-        decoded_mixed
-            .histograms()
-            .batch()
-            .as_arrow()
-            .to_data()
-            .ptr_eq(&sliced_table.as_arrow().to_data())
+    assert!(decoded_mixed.floats().is_null(0));
+    assert!(decoded_mixed.histograms().indices().is_valid(0));
+    assert_eq!(decoded_mixed.histograms().indices().value(0), 0);
+    assert_histogram(
+        decoded_mixed.histograms().batch().view(0).unwrap(),
+        -9.5,
+        Some(456),
     );
     assert_eq!(output[0].time.as_buffer().as_ptr(), input_times_ptr);
 
@@ -506,8 +491,8 @@ async fn single_row_sliced_payload_keeps_zero_copy_buffers_and_releases_owners()
         panic!("expected histogram")
     };
     assert_histogram(value, -9.5, Some(456));
-    match chunk.fields.get(2).unwrap().sample(0) {
-        SampleRef::Float(value) => assert_eq!(value.to_bits(), (-0.0_f64).to_bits()),
-        SampleRef::Missing | SampleRef::Histogram(_) => panic!("expected mixed float"),
-    }
+    let SampleRef::Histogram(value) = chunk.fields.get(2).unwrap().sample(0) else {
+        panic!("expected mixed histogram")
+    };
+    assert_histogram(value, -9.5, Some(456));
 }

src/datatypes/src/series/frame.rs

@@ -73,7 +73,7 @@
 #[cfg(test)]
 mod tests;
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 
 use arrow::array::{
@@ -81,7 +81,7 @@ use arrow::array::{
     StructArray, UInt32Array, UInt64Array,
 };
 use arrow::buffer::{Buffer, NullBuffer, OffsetBuffer, ScalarBuffer};
-use arrow::compute::concat;
+use arrow::compute::{concat, take};
 use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef};
 use arrow::record_batch::{RecordBatch, RecordBatchOptions};
 use serde::{Deserialize, Serialize};
@@ -1457,13 +1457,43 @@ fn histogram_charge(values: &HistogramSamples) -> Result<usize> {
     )
 }
 
+fn compact_histogram(values: &HistogramSamples) -> Result<(ArrayRef, UInt32Array)> {
+    let indices = values.indices();
+    // One encoded entry per *distinct* referenced row: a segment that references the same row many
+    // times would otherwise copy that row once per reference, so a small table plus a long run of
+    // equal indices would still grow the payload with every position.
+    let mut carried: HashMap<u32, u32> = HashMap::new();
+    let mut selected = Vec::new();
+    let mut remapped = Vec::with_capacity(indices.len());
+    for index in 0..indices.len() {
+        if indices.is_valid(index) {
+            let original = indices.value(index);
+            let local_index = match carried.get(&original) {
+                Some(local_index) => *local_index,
+                None => {
+                    let local_index = u32::try_from(selected.len())
+                        .map_err(|_| err("too many referenced histograms in one frame segment"))?;
+                    selected.push(original);
+                    carried.insert(original, local_index);
+                    local_index
+                }
+            };
+            remapped.push(Some(local_index));
+        } else {
+            remapped.push(None);
+        }
+    }
+    let selected = UInt32Array::from(selected);
+    let table = arrow(take(values.batch().as_arrow(), &selected, None))?;
+    Ok((table, UInt32Array::from(remapped)))
+}
+
 /// Encodes buffered rows as one batch with direct N-row Arrow array construction.
 ///
-/// Row-local histogram indices stay exactly as encoded: physical tables are concatenated under
-/// cumulative row list offsets, so no index is remapped. Each histogram-bearing row carries its
-/// complete table, including unreferenced entries; many segments referencing one large table can
-/// therefore repeat that table in the encoded batch. A single constituent child array is reused
-/// by clone (sharing its buffers); multiple constituents are concatenated, which copies.
+/// Each histogram segment carries each *distinct* table row it references exactly once, and its
+/// local indices are remapped into that compact table, so the payload grows with the rows a segment
+/// actually names rather than with its positions. A single constituent child array is reused by
+/// clone (sharing its buffers); multiple constituents are concatenated, which copies.
 fn encode_rows(schema: &SchemaRef, shape: FrameShape, rows: &[FrameRow]) -> Result<RecordBatch> {
     let count = rows.len();
     let DataType::Struct(time_fields) = schema.field(2).data_type() else {
@@ -1590,20 +1620,22 @@ fn encode_rows(schema: &SchemaRef, shape: FrameShape, rows: &[FrameRow]) -> Resu
                 DenseSamplesRef::Histogram(values) => {
                     float_lengths.push(0);
                     float_valid.push(false);
-                    index_lengths.push(values.len());
-                    index_parts.push(Arc::new(values.indices().clone()));
-                    table_lengths.push(values.batch().len());
-                    table_parts.push(Arc::new(values.batch().as_arrow().clone()));
+                    let (table, indices) = compact_histogram(values)?;
+                    index_lengths.push(indices.len());
+                    index_parts.push(Arc::new(indices));
+                    table_lengths.push(table.len());
+                    table_parts.push(table);
                     histogram_valid.push(true);
                 }
                 DenseSamplesRef::Mixed(values) => {
                     float_lengths.push(values.floats().len());
                     float_parts.push(Arc::new(values.floats().clone()));
                     float_valid.push(true);
-                    index_lengths.push(values.histograms().len());
-                    index_parts.push(Arc::new(values.histograms().indices().clone()));
-                    table_lengths.push(values.histograms().batch().len());
-                    table_parts.push(Arc::new(values.histograms().batch().as_arrow().clone()));
+                    let (table, indices) = compact_histogram(values.histograms())?;
+                    index_lengths.push(indices.len());
+                    index_parts.push(Arc::new(indices));
+                    table_lengths.push(table.len());
+                    table_parts.push(table);
                     histogram_valid.push(true);
                 }
             }

src/datatypes/src/series/frame/tests.rs

@@ -231,11 +231,6 @@ fn assert_histogram_with_start(
     );
 }
 
-fn assert_unreferenced_histogram(histogram: crate::series::histogram::HistogramView<'_>) {
-    assert_eq!(histogram.schema, -7);
-    assert_histogram_with_start(histogram, 90.5, None);
-}
-
 fn assert_float_bits(actual: &Samples, expected: &[Option<f64>]) {
     assert_eq!(actual.len(), expected.len());
     for (index, expected) in expected.iter().enumerate() {
@@ -347,8 +342,7 @@ fn frame_histogram_lists(batch: &RecordBatch, field: usize) -> (ListArray, ListA
     (indices, tables)
 }
 
-/// Every row is expected to resolve index 0 inside its own table only, with the whole table
-/// retained and the other index entry null.
+/// Each row's referenced histogram is preserved in its compact row-local table.
 fn assert_row_local_histogram_rows(rows: &[RawSeriesChunk]) {
     assert_eq!(rows.len(), 2);
     assert_eq!(rows[0].time.as_buffer().as_ref(), &[1_i64, 2]);
@@ -357,23 +351,20 @@ fn assert_row_local_histogram_rows(rows: &[RawSeriesChunk]) {
     let DenseSamplesRef::Histogram(first) = rows[0].fields.get(0).unwrap().dense().unwrap() else {
         panic!("expected histogram samples");
     };
-    assert_eq!(first.batch().len(), 2);
+    assert_eq!(first.batch().len(), 1);
     assert_eq!(first.indices().len(), 2);
     assert_eq!(first.indices().value(0), 0);
     assert!(!first.indices().is_valid(1));
     assert_histogram(first.view_at(0).unwrap(), -9.5);
-    assert_unreferenced_histogram(first.batch().view(1).unwrap());
 
     let DenseSamplesRef::Histogram(second) = rows[1].fields.get(0).unwrap().dense().unwrap() else {
         panic!("expected histogram samples");
     };
-    assert_eq!(second.batch().len(), 2);
+    assert_eq!(second.batch().len(), 1);
     assert_eq!(second.indices().len(), 2);
     assert!(!second.indices().is_valid(0));
     assert_eq!(second.indices().value(1), 0);
-    // A global index remap would resolve index 0 into the first row's table instead.
     assert_histogram(second.view_at(1).unwrap(), 42.25);
-    assert_histogram_with_start(second.batch().view(1).unwrap(), 142.25, None);
 }
 
 #[test]
@@ -431,23 +422,10 @@ fn sliced_histogram_frames_densify_without_duplicating_shared_table() {
         DenseSamplesRef::Histogram(histograms) => histograms.batch(),
         _ => panic!("expected histogram samples"),
     };
-    assert!(!Arc::ptr_eq(&first_table, second_table));
-    let first_values = first_table
-        .as_arrow()
-        .column(2)
-        .as_any()
-        .downcast_ref::<Float64Array>()
-        .unwrap();
-    let second_values = second_table
-        .as_arrow()
-        .column(2)
-        .as_any()
-        .downcast_ref::<Float64Array>()
-        .unwrap();
-    assert_eq!(
-        first_values.values().as_ptr(),
-        second_values.values().as_ptr()
-    );
+    assert_eq!(first_table.len(), 1);
+    assert_eq!(second_table.len(), 1);
+    assert_histogram(first_table.view(0).unwrap(), -9.5);
+    assert_histogram(second_table.view(0).unwrap(), -9.5);
 
     // The default budget groups the same 64 rows into one multirow batch; every row still decodes
     // to its own slice of the shared table with bit-exact values.
@@ -490,21 +468,9 @@ fn sliced_histogram_frames_densify_without_duplicating_shared_table() {
         panic!("expected dense histogram samples");
     };
     assert_eq!(histograms.len(), 64);
-    assert_eq!(histograms.batch().len(), first_table.len());
     for position in 0..64 {
         assert_histogram(histograms.view_at(position).unwrap(), -9.5);
     }
-    let dense_values = histograms
-        .batch()
-        .as_arrow()
-        .column(2)
-        .as_any()
-        .downcast_ref::<Float64Array>()
-        .unwrap();
-    assert_eq!(
-        dense_values.values().as_ptr(),
-        first_values.values().as_ptr()
-    );
 }
 
 #[test]
@@ -2730,7 +2696,7 @@ fn rejects_malformed_histogram_and_payload_encodings() {
                 DataType::List(item) => item.clone(),
                 _ => unreachable!(),
             },
-            OffsetBuffer::new(ScalarBuffer::from(vec![0_i32, 1, 1])),
+            OffsetBuffer::new(ScalarBuffer::from(vec![0_i32, 1])),
             Arc::new(arrow::array::Int32Array::from(vec![1])),
             None,
         )
@@ -2740,8 +2706,8 @@ fn rejects_malformed_histogram_and_payload_encodings() {
                 DataType::List(item) => item.clone(),
                 _ => unreachable!(),
             },
-            OffsetBuffer::new(ScalarBuffer::from(vec![0_i32, 2, 5])),
-            Arc::new(Float64Array::from(vec![3.25, 7.5, 3.0, 7.0, 8.0])),
+            OffsetBuffer::new(ScalarBuffer::from(vec![0_i32, 2])),
+            Arc::new(Float64Array::from(vec![3.25, 7.5])),
             None,
         )
         .unwrap();
@@ -3165,67 +3131,19 @@ fn sliced_canonical_mixed_histogram_round_trip_keeps_buffers_alive() {
         .as_any()
         .downcast_ref::<UInt32Array>()
         .unwrap();
-    for (packed_indices, source_indices) in [
-        (packed_mixed_indices, source_indices),
-        (packed_hist_indices, &histogram_indices),
-    ] {
-        assert_eq!(
-            packed_indices.values().as_ptr(),
-            source_indices.values().as_ptr()
-        );
-        assert_eq!(
-            packed_indices.nulls().unwrap().buffer().as_ptr(),
-            source_indices.nulls().unwrap().buffer().as_ptr()
-        );
-        assert_eq!(
-            packed_indices.nulls().unwrap().offset(),
-            source_indices.nulls().unwrap().offset()
-        );
-    }
+    assert_eq!(
+        packed_mixed_indices,
+        &UInt32Array::from(vec![Some(0), None])
+    );
+    assert_eq!(
+        packed_hist_indices,
+        &UInt32Array::from(vec![Some(0), Some(0)]),
+        "both positions name the same distinct row"
+    );
     let packed_mixed_table = frame_histogram_table(packed, 0);
     let packed_hist_table = frame_histogram_table(packed, 1);
-    for packed_table in [&packed_mixed_table, &packed_hist_table] {
-        let packed_sum = packed_table
-            .column(2)
-            .as_any()
-            .downcast_ref::<Float64Array>()
-            .unwrap();
-        assert_eq!(packed_sum.values().as_ptr(), sliced_sum.values().as_ptr());
-        let packed_positive = packed_table
-            .column(10)
-            .as_any()
-            .downcast_ref::<ListArray>()
-            .unwrap();
-        assert_eq!(packed_positive.offset(), sliced_positive.offset());
-        assert_eq!(
-            packed_positive.value_offsets().as_ptr(),
-            sliced_positive.value_offsets().as_ptr()
-        );
-        assert_eq!(
-            packed_positive.value_offsets()[sliced_table_offset],
-            sliced_positive.value_offsets()[sliced_table_offset]
-        );
-        assert_eq!(
-            packed_positive.value_offsets()[sliced_table_offset + 1],
-            sliced_positive.value_offsets()[sliced_table_offset + 1]
-        );
-        assert_eq!(
-            packed_positive
-                .values()
-                .as_any()
-                .downcast_ref::<Float64Array>()
-                .unwrap()
-                .values()
-                .as_ptr(),
-            sliced_positive
-                .values()
-                .as_any()
-                .downcast_ref::<Float64Array>()
-                .unwrap()
-                .values()
-                .as_ptr()
-        );
-    }
+    assert_eq!(packed_mixed_table.len(), 1);
+    assert_eq!(packed_hist_table.len(), 1);
     let decoded = unpack_raw_series_frame(
         &schema,
         FrameLabelMode::Key,
@@ -3245,30 +3163,14 @@ fn sliced_canonical_mixed_histogram_round_trip_keeps_buffers_alive() {
     let decoded_time_values = decoded[0].time.as_buffer();
     assert_eq!(decoded_time_values.as_ptr(), packed_times.values().as_ptr());
     assert_eq!(decoded_time_values, source_times);
-    for (decoded_indices, source_indices) in [
-        (decoded_mixed.histograms().indices(), source_indices),
-        (decoded_histogram.indices(), &histogram_indices),
-    ] {
-        assert_eq!(
-            decoded_indices.values().as_ptr(),
-            source_indices.values().as_ptr()
-        );
-        assert_eq!(
-            decoded_indices.nulls().unwrap().buffer().as_ptr(),
-            source_indices.nulls().unwrap().buffer().as_ptr()
-        );
-        assert_eq!(
-            decoded_indices.nulls().unwrap().offset(),
-            source_indices.nulls().unwrap().offset()
-        );
-    }
     assert_eq!(
-        decoded_mixed.histograms().indices().values().as_ptr(),
-        indices_owner.values().as_ptr().wrapping_add(1)
+        decoded_mixed.histograms().indices(),
+        &UInt32Array::from(vec![Some(0), None])
     );
     assert_eq!(
-        decoded_histogram.indices().values().as_ptr(),
-        histogram_indices_owner.values().as_ptr().wrapping_add(1)
+        decoded_histogram.indices(),
+        &UInt32Array::from(vec![Some(0), Some(0)]),
+        "both positions name the one distinct row"
     );
     assert_eq!(decoded_mixed.histograms().indices().value(0), 0);
     assert!(!decoded_mixed.histograms().indices().is_valid(1));
@@ -3281,84 +3183,16 @@ fn sliced_canonical_mixed_histogram_round_trip_keeps_buffers_alive() {
         float_owner.nulls().unwrap().buffer().as_ptr()
     );
     assert_eq!(decoded_mixed.floats().nulls().unwrap().offset(), 1);
-    assert_eq!(decoded_histogram.indices().value(0), 1);
-    assert_eq!(decoded_histogram.indices().value(1), 1);
-    assert_eq!(decoded_histogram.batch().len(), 2);
-    let decoded_table = decoded_histogram.batch().as_arrow();
-    let decoded_sum = decoded_table
-        .column(2)
-        .as_any()
-        .downcast_ref::<Float64Array>()
-        .unwrap();
-    assert_eq!(decoded_sum.values().as_ptr(), sliced_sum.values().as_ptr());
-    let decoded_mixed_table = decoded_mixed.histograms().batch().as_arrow();
-    for decoded_table_view in [decoded_mixed_table, decoded_table] {
-        let primitive = decoded_table_view
-            .column(2)
-            .as_any()
-            .downcast_ref::<Float64Array>()
-            .unwrap();
-        assert_eq!(
-            primitive.values().as_ptr(),
-            source_sum.values().as_ptr().wrapping_add(1)
-        );
-        let positive = decoded_table_view
-            .column(10)
-            .as_any()
-            .downcast_ref::<ListArray>()
-            .unwrap();
-        assert_eq!(positive.offset(), source_positive.offset());
-        assert_eq!(
-            positive.value_offsets()[sliced_table_offset],
-            source_positive.value_offsets()[1]
-        );
-        assert_eq!(
-            positive.value_offsets().as_ptr(),
-            sliced_positive.value_offsets().as_ptr()
-        );
-        assert_eq!(
-            positive.value_offsets()[sliced_table_offset + 1],
-            source_positive.value_offsets()[2]
-        );
-        assert_eq!(
-            positive
-                .values()
-                .as_any()
-                .downcast_ref::<Float64Array>()
-                .unwrap()
-                .values()
-                .as_ptr(),
-            source_positive
-                .values()
-                .as_any()
-                .downcast_ref::<Float64Array>()
-                .unwrap()
-                .values()
-                .as_ptr()
-        );
-    }
-    let decoded_positive_buckets = decoded_table
-        .column(10)
-        .as_any()
-        .downcast_ref::<ListArray>()
-        .unwrap()
-        .values()
-        .as_any()
-        .downcast_ref::<Float64Array>()
-        .unwrap();
-    let source_positive_buckets = source_table
-        .column(10)
-        .as_any()
-        .downcast_ref::<ListArray>()
-        .unwrap()
-        .values()
-        .as_any()
-        .downcast_ref::<Float64Array>()
-        .unwrap();
-    assert_eq!(
-        decoded_positive_buckets.values().as_ptr(),
-        source_positive_buckets.values().as_ptr()
+    assert_eq!(decoded_histogram.indices().value(0), 0);
+    assert_eq!(decoded_histogram.indices().value(1), 0);
+    assert_eq!(decoded_histogram.batch().len(), 1);
+    assert_histogram_with_start(
+        decoded_mixed.histograms().batch().view(0).unwrap(),
+        -9.5,
+        Some(-123),
     );
+    // Both positions name the one distinct row of the compact table.
+    assert_histogram_with_start(decoded_histogram.batch().view(0).unwrap(), 90.5, None);
 
     drop(packed_mixed_table);
     drop(packed_hist_table);
@@ -3372,11 +3206,13 @@ fn sliced_canonical_mixed_histogram_round_trip_keeps_buffers_alive() {
     drop(chunk);
     drop(frames);
 
-    assert_eq!(decoded_table.len(), 2);
-    assert_histogram(decoded_mixed.histograms().batch().view(0).unwrap(), -9.5);
-    assert_unreferenced_histogram(decoded_mixed.histograms().batch().view(1).unwrap());
-    assert_histogram(decoded_histogram.batch().view(0).unwrap(), -9.5);
-    assert_unreferenced_histogram(decoded_histogram.batch().view(1).unwrap());
+    assert_eq!(decoded_histogram.batch().len(), 1);
+    assert_histogram_with_start(
+        decoded_mixed.histograms().batch().view(0).unwrap(),
+        -9.5,
+        Some(-123),
+    );
+    assert_histogram_with_start(decoded_histogram.batch().view(0).unwrap(), 90.5, None);
     assert!(matches!(
         decoded[0].fields.get(0).unwrap().sample(0),
         SampleRef::Histogram(_)
@@ -3884,6 +3720,160 @@ fn byte_budget_flushes_before_exceeding_and_retries_the_pending_row() {
     assert!(offset_buffer(&[i32::MAX as usize, 1]).is_err());
 }
 
+#[test]
+fn histogram_tables_encode_only_referenced_rows_and_remap_indices() {
+    let schema = test_schema(&["histogram", "mixed", "all_null"]);
+    let table = histogram_batch();
+    let indices = UInt32Array::from(vec![Some(1), None, Some(0), Some(1)]);
+    let mixed = Samples::try_from_mixed(
+        Float64Array::from(vec![None, Some(12.5), None, None]),
+        crate::series::samples::HistogramSamples::try_new(indices.clone(), table.clone()).unwrap(),
+    )
+    .unwrap();
+    let fields = SeriesFields::try_from_vec(vec![
+        Samples::try_from_histogram(indices.clone(), table.clone()).unwrap(),
+        mixed,
+        Samples::try_from_histogram(UInt32Array::new_null(4), table.clone()).unwrap(),
+    ])
+    .unwrap();
+    let chunk = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(vec![0x71_u8]),
+        None,
+        RawTimestamps::try_new(vec![1_i64, 2, 3, 4].into()).unwrap(),
+        fields,
+        true,
+    )
+    .unwrap();
+    let packed = pack_raw_series_chunk(
+        &schema,
+        FrameLabelMode::Key,
+        Some(&dense_description()),
+        &chunk,
+    )
+    .unwrap();
+    let (encoded_indices, _) = frame_histogram_lists(&packed[0], 0);
+    let encoded_indices = encoded_indices
+        .value(0)
+        .as_any()
+        .downcast_ref::<UInt32Array>()
+        .unwrap()
+        .clone();
+    // Source indices are [1, null, 0, 1]: row 1 is named first and then row 0, and the second
+    // reference to row 1 reuses the first encoded row.
+    assert_eq!(
+        encoded_indices,
+        UInt32Array::from(vec![Some(0), None, Some(1), Some(0)])
+    );
+    assert_eq!(
+        frame_histogram_table(&packed[0], 0).len(),
+        2,
+        "one encoded row per distinct referenced row, not per reference"
+    );
+    assert_eq!(frame_histogram_table(&packed[0], 1).len(), 2);
+    assert_eq!(frame_histogram_table(&packed[0], 2).len(), 0);
+
+    let decoded = decode_batches(&schema, FrameLabelMode::Key, &packed);
+    let DenseSamplesRef::Histogram(histograms) = decoded[0].fields.get(0).unwrap().dense().unwrap()
+    else {
+        panic!("histogram representation was normalized");
+    };
+    assert_eq!(
+        histograms.indices(),
+        &UInt32Array::from(vec![Some(0), None, Some(1), Some(0)])
+    );
+    assert_histogram_with_start(histograms.view_at(0).unwrap(), 90.5, None);
+    assert_histogram_with_start(histograms.view_at(2).unwrap(), -9.5, Some(-123));
+    assert_histogram_with_start(histograms.view_at(3).unwrap(), 90.5, None);
+
+    let DenseSamplesRef::Mixed(mixed) = decoded[0].fields.get(1).unwrap().dense().unwrap() else {
+        panic!("mixed representation was normalized");
+    };
+    assert_eq!(mixed.histograms().indices(), histograms.indices());
+    for (index, expected) in [None, Some(12.5_f64), None, None].into_iter().enumerate() {
+        match (
+            mixed.floats().is_valid(index),
+            mixed.floats().value(index),
+            expected,
+        ) {
+            (false, _, None) => {}
+            (true, actual, Some(expected)) => {
+                assert_eq!(actual.to_bits(), expected.to_bits(), "position {index}");
+            }
+            _ => panic!("unexpected mixed float at position {index}"),
+        }
+    }
+    assert!(mixed.floats().is_null(0));
+    assert!(mixed.histograms().indices().is_valid(0));
+    assert!(mixed.floats().is_valid(1));
+    assert!(!mixed.histograms().indices().is_valid(1));
+    assert_histogram_with_start(mixed.histograms().view_at(0).unwrap(), 90.5, None);
+    assert_histogram_with_start(mixed.histograms().view_at(2).unwrap(), -9.5, Some(-123));
+    assert_histogram_with_start(mixed.histograms().view_at(3).unwrap(), 90.5, None);
+
+    let DenseSamplesRef::Histogram(all_null) = decoded[0].fields.get(2).unwrap().dense().unwrap()
+    else {
+        panic!("all-null histogram representation was normalized");
+    };
+    assert_eq!(all_null.indices().len(), 4);
+    assert!((0..4).all(|index| !all_null.indices().is_valid(index)));
+    assert_eq!(all_null.batch().len(), 0);
+}
+
+#[test]
+fn repeated_references_to_one_histogram_row_encode_it_once() {
+    let schema = test_schema(&["histogram"]);
+    let table = histogram_batch();
+    // Every position names the same row: the encoded payload must grow with distinct named rows,
+    // not with positions.
+    let positions = 10_000;
+    let indices = UInt32Array::from(vec![Some(1); positions]);
+    let chunk = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(vec![0x33_u8]),
+        None,
+        RawTimestamps::try_new((0..positions as i64).collect::<Vec<_>>().into()).unwrap(),
+        SeriesFields::one(Samples::try_from_histogram(indices, table).unwrap()),
+        true,
+    )
+    .unwrap();
+    let packed = pack_raw_series_chunk(
+        &schema,
+        FrameLabelMode::Key,
+        Some(&dense_description()),
+        &chunk,
+    )
+    .unwrap();
+    let (encoded_indices, _) = frame_histogram_lists(&packed[0], 0);
+    let encoded = encoded_indices
+        .value(0)
+        .as_any()
+        .downcast_ref::<UInt32Array>()
+        .unwrap()
+        .clone();
+    assert_eq!(encoded.len(), positions);
+    assert!(
+        (0..positions).all(|position| encoded.value(position) == 0),
+        "every position reuses the one encoded row"
+    );
+    assert_eq!(
+        frame_histogram_table(&packed[0], 0).len(),
+        1,
+        "one distinct referenced row, regardless of how often it is referenced"
+    );
+
+    let decoded = decode_batches(&schema, FrameLabelMode::Key, &packed);
+    let DenseSamplesRef::Histogram(histograms) = decoded[0].fields.get(0).unwrap().dense().unwrap()
+    else {
+        panic!("histogram representation was normalized");
+    };
+    assert_eq!(histograms.batch().len(), 1);
+    // Row 1 of the fixture is the one with sum 90.5 and no start timestamp.
+    for position in [0, positions / 2, positions - 1] {
+        assert_histogram_with_start(histograms.view_at(position).unwrap(), 90.5, None);
+    }
+}
+
 #[test]
 fn multirow_histogram_rows_keep_row_local_indices_and_tables() {
     let schema = test_schema(&["histogram"]);
@@ -3932,16 +3922,16 @@ fn multirow_histogram_rows_keep_row_local_indices_and_tables() {
     assert_eq!(
         encoded_second.value(1),
         0,
-        "row-local indices must not be remapped"
+        "indices must be remapped within their row's compact table"
     );
-    // The second row references its own table through a nonzero slice of the concatenated child.
-    assert_eq!(tables.value_offsets(), [0, 2, 4]);
+    // Each row's sole referenced table entry is encoded independently.
+    assert_eq!(tables.value_offsets(), [0, 1, 2]);
     let concatenated = tables
         .values()
         .as_any()
         .downcast_ref::<StructArray>()
         .unwrap();
-    assert_eq!(concatenated.len(), 4);
+    assert_eq!(concatenated.len(), 2);
 
     let decoded = decode_batches(
         &schema,
@@ -3950,8 +3940,8 @@ fn multirow_histogram_rows_keep_row_local_indices_and_tables() {
     );
     assert_row_local_histogram_rows(&decoded);
 
-    // A real Arrow IPC round trip keeps the schema, the whole tables, the null index entries and
-    // the bit-exact payloads.
+    // A real Arrow IPC round trip keeps the schema, compact row tables, null index entries and
+    // the bit-exact referenced payloads.
     let mut ipc = Vec::new();
     {
         let mut writer = StreamWriter::try_new(&mut ipc, &packed[0].schema()).unwrap();

src/promql/src/extension_plan.rs

@@ -26,6 +26,7 @@ mod raw_series_source;
 mod scalar_calculate;
 mod series_divide;
 mod series_gather;
+mod series_repartition;
 #[cfg(test)]
 mod test_util;
 mod union_distinct_on;
@@ -56,6 +57,7 @@ pub use range_manipulate::{RangeManipulate, RangeManipulateExec, RangeManipulate
 pub use raw_series_source::{RawSeriesSource, RawSeriesSourceExec};
 pub use scalar_calculate::ScalarCalculate;
 pub use series_divide::{SeriesDivide, SeriesDivideExec, SeriesDivideStream};
+pub use series_repartition::SeriesRepartitionExec;
 pub use union_distinct_on::{UnionDistinctOn, UnionDistinctOnExec, UnionDistinctOnStream};
 
 pub type Millisecond = <TimestampMillisecondType as ArrowPrimitiveType>::Native;

src/promql/src/native/execute.rs

@@ -155,7 +155,7 @@ pub(super) fn evaluated_bytes(chunk: &EvaluatedChunk) -> usize {
 /// allocation however many chunks index into it, so it is not a per-chunk cost at all. Dropping it
 /// here is therefore only half the story -- it must not become a figure the operand never charges.
 /// [`unbilled_tables`] is the other half, and the operand bills each table once through it.
-fn operand_chunk_bytes(series: &Series) -> usize {
+pub(crate) fn operand_chunk_bytes(series: &Series) -> usize {
     series_bytes(series)
         .saturating_sub(fields_bytes(&series.fields))
         .saturating_add(sample_slice_bytes(&series.fields))
@@ -243,7 +243,7 @@ fn unbilled_tables(series: &Series, billed: &HashSet<usize>) -> (Vec<usize>, usi
 ///
 /// Existing values and histogram tables remain shared with the fragments, whose estimates already
 /// account for their payloads.
-fn merge_workspace_bytes(rows: usize, fragments: usize, fields: usize) -> usize {
+pub(crate) fn merge_workspace_bytes(rows: usize, fragments: usize, fields: usize) -> usize {
     let sorted_rows = rows.saturating_mul(
         std::mem::size_of::<(usize, usize, usize)>() + std::mem::size_of::<usize>(),
     );