From c9d160f6a1b339980cf1c03290347472eafed045 Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 21 Oct 2026 14:53:33 +0800 Subject: [PATCH] fix(table): declare the distribution a raw series frame already has `SeriesScan` guarantees every series arrives whole in one partition - its own docs say so, and both modes enforce it in different ways: the two-phase collector places a series by its tsid, and the legacy divider only emits a batch once it holds all of a series' rows. But the leaf built for a raw frame declared `UnknownPartitioning`, so nothing on the plan said it. That mattered to a consumer holding one window per series: with the distribution unspecified an optimizer is free to round-robin the partition, and two halves of one series would each look final to their own stream. So the leaf now declares `Hash(read_key)`. `read_key` is the primary key as a binary column, so hashing it groups exactly the rows of one series and splits nothing; `FusedSeriesExec` asks for the same key and is satisfied without a shuffle. The three tests that pinned `UnknownPartitioning` now pin the key instead. Signed-off-by: discord9 --- src/table/src/table/scan.rs | 16 +++++++++++++++- src/table/src/table/scan/series_tests.rs | 8 +++++--- 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/src/table/src/table/scan.rs b/src/table/src/table/scan.rs index d6ac26c0a21..ef8439bd8e9 100644 --- a/src/table/src/table/scan.rs +++ b/src/table/src/table/scan.rs @@ -599,9 +599,23 @@ impl RegionScanExec { ordering, )?; let partition_count = scanner.properties().num_partitions().max(1); + // Every series arrives whole in one partition, and `read_key` is its primary key, so the + // key says exactly that: hashing it groups the rows of one series and splits nothing. It + // is declared rather than left unspecified because a consumer that holds one window per + // series has to be told the series is whole - an optimizer free to round-robin the + // partition would hand it two halves of one series, each marked final by its own stream, + // and both would answer half the truth without saying so. + // The name is the one `raw_series_frame_schema` gives the column. + let partitioning = match Column::new_with_schema("read_key", &arrow_schema) { + Ok(read_key) => Partitioning::Hash(vec![Arc::new(read_key) as _], partition_count), + // The schema this leaf builds always has the column; losing it would mean the frame + // schema changed shape, and claiming a distribution over a column that is not there + // would be worse than saying nothing. + Err(_) => Partitioning::UnknownPartitioning(partition_count), + }; let properties = Arc::new(PlanProperties::new( EquivalenceProperties::new(arrow_schema.clone()), - Partitioning::UnknownPartitioning(partition_count), + partitioning, EmissionType::Incremental, Boundedness::Bounded, )); diff --git a/src/table/src/table/scan/series_tests.rs b/src/table/src/table/scan/series_tests.rs index 19209a50af3..01c1189533c 100644 --- a/src/table/src/table/scan/series_tests.rs +++ b/src/table/src/table/scan/series_tests.rs @@ -699,9 +699,11 @@ async fn native_raw_frame_executes_and_decodes_bound_data() { }) .unwrap(); assert!(native_bytes > 0); + // One series arrives whole in one partition, and `read_key` says which series a row belongs + // to, so the leaf declares that as its distribution instead of leaving it unspecified. assert!(matches!( plan.properties().output_partitioning(), - Partitioning::UnknownPartitioning(1) + Partitioning::Hash(columns, 1) if columns.len() == 1 )); assert!(plan.properties().output_ordering().is_none()); } @@ -1287,7 +1289,7 @@ async fn invalid_partition_and_nonempty_children_are_rejected() { .unwrap(); assert!(matches!( plan.properties().output_partitioning(), - Partitioning::UnknownPartitioning(2) + Partitioning::Hash(columns, 2) if columns.len() == 1 )); assert!( plan.execute(1, context()) @@ -1309,7 +1311,7 @@ async fn invalid_partition_and_nonempty_children_are_rejected() { .unwrap(); assert!(matches!( changed_plan.properties().output_partitioning(), - Partitioning::UnknownPartitioning(3) + Partitioning::Hash(columns, 3) if columns.len() == 1 )); assert!(plan.execute(0, context()).is_err()); assert!(plan.execute(2, context()).is_err());