fix(cubesql): Fix SortPushDown pushing sort over joins

LogicalPlan::Join and CrossJoin do not preserve the ordering semantically
When planned as HashJoin it will output batches in same order as they are coming from right stream
But both Join and CrossJoin will have same partitioning as right input (even when repartition_joins disabled), and these partitions can be collected in arbitrary order by CoalescePartitions
See https://github.com/apache/datafusion/blob/7.0.0/datafusion/src/physical_plan/hash_join.rs#L282-L284
See https://github.com/apache/datafusion/blob/7.0.0/datafusion/src/physical_plan/cross_join.rs#L141-L143

Also, Substrait says that for both Join and Cross Product
> Orderedness is empty post operation
See https://substrait.io/relations/logical_relations/#join-operation

Author: mcheshkov

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__sort_push_down__tests__sort_down_cross_join_sort_left.snap

@@ -3,9 +3,9 @@ source: cubesql/src/compile/engine/df/optimizers/sort_push_down.rs
 expression: optimize(&plan)
 ---
 Projection: #j1.c1, #j2.c2
-  CrossJoin:
-    Sort: #j1.c1 ASC NULLS LAST
+  Sort: #j1.c1 ASC NULLS LAST
+    CrossJoin:
       Projection: #j1.key, #j1.c1
         TableScan: j1 projection=None
-    Projection: #j2.key, #j2.c2
-      TableScan: j2 projection=None
+      Projection: #j2.key, #j2.c2
+        TableScan: j2 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__sort_push_down__tests__sort_down_join_sort_left.snap

@@ -3,9 +3,9 @@ source: cubesql/src/compile/engine/df/optimizers/sort_push_down.rs
 expression: optimize(&plan)
 ---
 Projection: #j1.c1, #j2.c2
-  Inner Join: #j1.key = #j2.key
-    Sort: #j1.c1 ASC NULLS LAST
+  Sort: #j1.c1 ASC NULLS LAST
+    Inner Join: #j1.key = #j2.key
       Projection: #j1.key, #j1.c1
         TableScan: j1 projection=None
-    Projection: #j2.key, #j2.c2
-      TableScan: j2 projection=None
+      Projection: #j2.key, #j2.c2
+        TableScan: j2 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/sort_push_down.rs

@@ -3,15 +3,13 @@ use std::{collections::HashMap, sync::Arc};
 use datafusion::{
     error::{DataFusionError, Result},
     logical_plan::{
-        plan::{
-            Aggregate, CrossJoin, Distinct, Join, Limit, Projection, Sort, Subquery, Union, Window,
-        },
+        plan::{Aggregate, Distinct, Limit, Projection, Sort, Subquery, Union, Window},
         Column, DFSchema, Expr, Filter, LogicalPlan,
     },
     optimizer::optimizer::{OptimizerConfig, OptimizerRule},
 };
 
-use super::utils::{get_schema_columns, is_column_expr, plan_has_projections, rewrite};
+use super::utils::{is_column_expr, plan_has_projections, rewrite};
 
 /// Sort Push Down optimizer rule pushes ORDER BY clauses consisting of specific,
 /// mostly simple, expressions down the plan, all the way to the Projection
@@ -167,97 +165,6 @@ fn sort_push_down(
                 optimizer_config,
             )
         }
-        LogicalPlan::Join(Join {
-            left,
-            right,
-            on,
-            join_type,
-            join_constraint,
-            schema,
-            null_equals_null,
-        }) => {
-            // DataFusion preserves the sorting of the joined plans, prioritizing left side.
-            // Taking this into account, we can push Sort down the left plan if Sort references
-            // columns just from the left side.
-            // TODO: check if this is still the case with multiple target partitions
-            if let Some(some_sort_expr) = &sort_expr {
-                let left_columns = get_schema_columns(left.schema());
-                if some_sort_expr.iter().all(|expr| {
-                    if let Expr::Sort { expr, .. } = expr {
-                        if let Expr::Column(column) = expr.as_ref() {
-                            return left_columns.contains(column);
-                        }
-                    }
-                    false
-                }) {
-                    return Ok(LogicalPlan::Join(Join {
-                        left: Arc::new(sort_push_down(
-                            optimizer,
-                            left,
-                            sort_expr,
-                            optimizer_config,
-                        )?),
-                        right: Arc::new(sort_push_down(optimizer, right, None, optimizer_config)?),
-                        on: on.clone(),
-                        join_type: *join_type,
-                        join_constraint: *join_constraint,
-                        schema: schema.clone(),
-                        null_equals_null: *null_equals_null,
-                    }));
-                }
-            }
-
-            issue_sort(
-                sort_expr,
-                LogicalPlan::Join(Join {
-                    left: Arc::new(sort_push_down(optimizer, left, None, optimizer_config)?),
-                    right: Arc::new(sort_push_down(optimizer, right, None, optimizer_config)?),
-                    on: on.clone(),
-                    join_type: *join_type,
-                    join_constraint: *join_constraint,
-                    schema: schema.clone(),
-                    null_equals_null: *null_equals_null,
-                }),
-            )
-        }
-        LogicalPlan::CrossJoin(CrossJoin {
-            left,
-            right,
-            schema,
-        }) => {
-            // See `LogicalPlan::Join` notes above.
-            if let Some(some_sort_expr) = &sort_expr {
-                let left_columns = get_schema_columns(left.schema());
-                if some_sort_expr.iter().all(|expr| {
-                    if let Expr::Sort { expr, .. } = expr {
-                        if let Expr::Column(column) = expr.as_ref() {
-                            return left_columns.contains(column);
-                        }
-                    }
-                    false
-                }) {
-                    return Ok(LogicalPlan::CrossJoin(CrossJoin {
-                        left: Arc::new(sort_push_down(
-                            optimizer,
-                            left,
-                            sort_expr,
-                            optimizer_config,
-                        )?),
-                        right: Arc::new(sort_push_down(optimizer, right, None, optimizer_config)?),
-                        schema: schema.clone(),
-                    }));
-                }
-            }
-
-            issue_sort(
-                sort_expr,
-                LogicalPlan::CrossJoin(CrossJoin {
-                    left: Arc::new(sort_push_down(optimizer, left, None, optimizer_config)?),
-                    right: Arc::new(sort_push_down(optimizer, right, None, optimizer_config)?),
-                    schema: schema.clone(),
-                }),
-            )
-        }
         LogicalPlan::Union(Union {
             inputs,
             schema,