fix(cubestore): Фvoid empty SortPreservingMerge for empty sort_on (cube-js#11160)

A rolling-window pre-aggregation joins a generated time-series with the
rollup table on an inequality (and, under Tesseract, scalar window-bound
subqueries), which lowers to a join with an empty equi-join `on`. The empty
`on` propagates as an empty `sort_on` to the index scan, which then builds a
SortPreservingMergeExec with no sort expressions. Under DataFusion 46 such a
merge errors at execution with "Sort expressions cannot be empty for streaming
merge" once a worker has more than one partition to merge.

Author: waralexrom

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -54,6 +54,7 @@ pub fn sql_tests(prefix: &str) -> Vec<(&'static str, TestFn)> {
         t("float_merge", float_merge),
         t("join", join),
         t("filtered_join", filtered_join),
+        t("cross_join_empty_sort_on", cross_join_empty_sort_on),
         t("three_tables_join", three_tables_join),
         t(
             "three_tables_join_with_filter",
@@ -874,6 +875,48 @@ async fn join(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
     Ok(())
 }
 
+/// Reproduces CORE-593: a rolling-window pre-aggregation generates a cross/range join (empty
+/// equi-join `on`) over a rollup table plus GROUP BY + ORDER BY. The empty `on` propagates as an
+/// empty `sort_on` to the index scan, which builds a SortPreservingMergeExec with no sort
+/// expressions. With DataFusion 46 such a merge errors at execution ("Sort expressions cannot be
+/// empty for streaming merge") -- but only when the worker has more than one partition to merge,
+/// so the table must hold several chunks.
+async fn cross_join_empty_sort_on(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
+    let _ = service.exec_query("CREATE SCHEMA foo").await?;
+    let _ = service
+        .exec_query("CREATE TABLE foo.t (source text, n int)")
+        .await?;
+
+    // Separate inserts produce separate chunks, so the index scan merges more than one partition.
+    service
+        .exec_query("INSERT INTO foo.t (source, n) VALUES ('a', 1), ('b', 2)")
+        .await?;
+    service
+        .exec_query("INSERT INTO foo.t (source, n) VALUES ('a', 3)")
+        .await?;
+    service
+        .exec_query("INSERT INTO foo.t (source, n) VALUES ('c', 4)")
+        .await?;
+
+    let result = service
+        .exec_query(
+            "SELECT q.source, sum(q.n) FROM foo.t q \
+             CROSS JOIN (SELECT 1 AS x UNION ALL SELECT 2 AS x) series \
+             GROUP BY q.source ORDER BY q.source",
+        )
+        .await?;
+
+    assert_eq!(
+        to_rows(&result),
+        vec![
+            vec![TableValue::String("a".to_string()), TableValue::Int(8)],
+            vec![TableValue::String("b".to_string()), TableValue::Int(4)],
+            vec![TableValue::String("c".to_string()), TableValue::Int(8)],
+        ]
+    );
+    Ok(())
+}
+
 async fn filtered_join(service: Box<dyn SqlClient>) -> Result<(), CubeError> {
     let _ = service.exec_query("CREATE SCHEMA foo").await?;
 

rust/cubestore/cubestore-sql-tests/tests/migration.rs

@@ -30,6 +30,8 @@ fn main() {
         "repartition_multi_node_consistency".to_string(),
         "--skip".to_string(),
         "rolling_window_no_aggregates".to_string(),
+        "--skip".to_string(),
+        "cross_join_empty_sort_on".to_string(),
     ];
     run_sql_tests("migration", extra_args, move |test_name, test_fn| {
         let r = Builder::new_current_thread()

rust/cubestore/cubestore/src/queryplanner/planning.rs

@@ -1464,15 +1464,21 @@ fn pick_index(
 
     let schema = Arc::new(schema);
     let create_snapshot = |index: &IdRow<Index>| {
-        let index_sort_on = sort_on.map(|sc| {
-            index
-                .get_row()
-                .columns()
-                .iter()
-                .take(sc.0.len())
-                .map(|c| c.get_name().clone())
-                .collect::<Vec<_>>()
-        });
+        // An empty join `on` (a cross/range join, as rolling-window queries produce) yields an
+        // empty `sort_on`. Normalize it to `None` here so every consumer of `IndexSnapshot::sort_on`
+        // (scan construction and `required_input_ordering`) consistently falls back to the index
+        // sort key instead of describing an empty ordering.
+        let index_sort_on = sort_on
+            .map(|sc| {
+                index
+                    .get_row()
+                    .columns()
+                    .iter()
+                    .take(sc.0.len())
+                    .map(|c| c.get_name().clone())
+                    .collect::<Vec<_>>()
+            })
+            .filter(|cols| !cols.is_empty());
         IndexSnapshot {
             index: index.clone(),
             partitions: Vec::new(), // filled with results of `pick_partitions` later.