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());