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 <discord9@163.com>

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

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