Merge branch 'main' into fix/exact-hash-table-accounting-primitive

Author: mohitgurav20

datafusion/core/tests/sql/joins.rs

@@ -521,33 +521,6 @@ async fn asof_join_all_match_directions_across_batches() -> Result<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn asof_join_coerces_equality_and_match_types() -> Result<()> {
-    let ctx = SessionContext::new();
-    let batches = ctx
-        .sql(
-            "SELECT t.id, p.price \
-             FROM (VALUES (CAST(1 AS INT), CAST(4 AS INT), 7)) t(k, ts, id) \
-             ASOF JOIN \
-             (VALUES (CAST(1 AS BIGINT), CAST(2 AS BIGINT), 20)) p(k, ts, price) \
-             MATCH_CONDITION (t.ts >= p.ts) ON t.k = p.k",
-        )
-        .await?
-        .collect()
-        .await?;
-    assert_batches_eq!(
-        [
-            "+----+-------+",
-            "| id | price |",
-            "+----+-------+",
-            "| 7  | 20    |",
-            "+----+-------+",
-        ],
-        &batches
-    );
-    Ok(())
-}
-
 #[tokio::test]
 async fn asof_join_broadcasts_multi_partition_right_input() -> Result<()> {
     let config = SessionConfig::new().with_target_partitions(4);
@@ -639,28 +612,6 @@ async fn asof_join_projection_drops_output_ordering() -> Result<()> {
     Ok(())
 }
 
-#[tokio::test]
-async fn asof_join_explain_names_equality_and_match_conditions() -> Result<()> {
-    let ctx = SessionContext::new();
-    register_asof_test_tables(&ctx)?;
-    let batches = ctx
-        .sql(
-            "EXPLAIN SELECT t.trade_id, p.price FROM trades t \
-             ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts) \
-             ON t.symbol = p.symbol",
-        )
-        .await?
-        .collect()
-        .await?;
-    let explain = arrow::util::pretty::pretty_format_batches(&batches)?.to_string();
-    assert_contains!(explain.as_str(), "AsOf Join: match=[t.ts >= p.ts]");
-    assert_contains!(explain.as_str(), "on=[t.symbol = p.symbol]");
-    assert_contains!(explain.as_str(), "AsOfJoinExec:");
-    assert_contains!(explain.as_str(), "on=[(symbol = symbol)]");
-    assert_contains!(explain.as_str(), "match=[ts >= ts]");
-    Ok(())
-}
-
 #[tokio::test]
 async fn asof_join_rejects_unbounded_inputs_during_physical_planning() -> Result<()> {
     let ctx = SessionContext::new();
@@ -698,7 +649,7 @@ async fn asof_join_rejects_unbounded_inputs_during_physical_planning() -> Result
 }
 
 #[tokio::test]
-async fn asof_join_using_preserves_key_access_and_unparser_round_trips() -> Result<()> {
+async fn asof_join_using_unparser_round_trips() -> Result<()> {
     let ctx = SessionContext::new();
     register_asof_test_tables(&ctx)?;
     let df = ctx
@@ -707,45 +658,11 @@ async fn asof_join_using_preserves_key_access_and_unparser_round_trips() -> Resu
              MATCH_CONDITION (t.ts >= p.ts) USING (symbol)",
         )
         .await?;
-    assert_eq!(
-        df.schema()
-            .fields()
-            .iter()
-            .map(|field| field.name())
-            .collect::<Vec<_>>(),
-        vec!["ts", "trade_id", "symbol", "ts", "price"]
-    );
     let sql = plan_to_sql(df.logical_plan())?.to_string();
     assert!(sql.contains("ASOF JOIN"));
     assert!(sql.contains("MATCH_CONDITION"));
     assert!(sql.contains("USING(symbol)"), "unexpected SQL: {sql}");
     ctx.sql(&sql).await?;
-
-    let batches = ctx
-        .sql(
-            "SELECT t.trade_id, t.symbol AS left_symbol, p.symbol AS right_symbol \
-             FROM trades t ASOF JOIN prices p \
-             MATCH_CONDITION (t.ts >= p.ts) USING (symbol) \
-             ORDER BY t.trade_id",
-        )
-        .await?
-        .collect()
-        .await?;
-    assert_batches_eq!(
-        [
-            "+----------+-------------+--------------+",
-            "| trade_id | left_symbol | right_symbol |",
-            "+----------+-------------+--------------+",
-            "| 1        | A           |              |",
-            "| 2        | A           | A            |",
-            "| 3        | A           | A            |",
-            "| 4        | B           | B            |",
-            "| 5        | B           | B            |",
-            "| 6        |             |              |",
-            "+----------+-------------+--------------+",
-        ],
-        &batches
-    );
     Ok(())
 }
 
@@ -784,20 +701,3 @@ async fn asof_join_unparser_preserves_right_preselection() -> Result<()> {
     }
     Ok(())
 }
-
-#[tokio::test]
-async fn asof_join_rejects_invalid_contracts() -> Result<()> {
-    let ctx = SessionContext::new();
-    register_asof_test_tables(&ctx)?;
-    for sql in [
-        "SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts = p.ts) ON t.symbol = p.symbol",
-        "SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (p.ts >= t.ts) ON t.symbol = p.symbol",
-        "SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts) ON t.symbol > p.symbol",
-        "SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (1 >= p.ts) ON t.symbol = p.symbol",
-        "SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts) ON 1 = 1",
-        "SELECT * FROM trades t ASOF JOIN prices p MATCH_CONDITION (t.ts >= p.ts AND t.ts > p.ts) ON t.symbol = p.symbol",
-    ] {
-        assert!(ctx.sql(sql).await.is_err(), "query should fail: {sql}");
-    }
-    Ok(())
-}

datafusion/physical-expr/src/aggregate.rs

@@ -61,7 +61,7 @@ use datafusion_functions_aggregate_common::accumulator::{
 };
 use datafusion_functions_aggregate_common::order::AggregateOrderSensitivity;
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
-use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
+use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
 
 #[derive(Debug, Clone)]
 struct AggregateHumanDisplay {
@@ -265,12 +265,18 @@ impl AggregateExprBuilder {
         assert_or_internal_err!(!args.is_empty(), "args should not be empty");
 
         // An order-insensitive aggregate ignores its ORDER BY, so drop it here.
-        // Everything derived from `order_bys` below, such as the ordering fields
-        // in the aggregate's state, then agrees that there is no ordering.
+        // Otherwise drop sort keys that repeat an earlier one: a repeated key can
+        // never break a tie the earlier one left (`ORDER BY b ASC, b DESC` orders
+        // like `ORDER BY b ASC`), and accumulators build their ordering with
+        // `LexOrdering`, which drops such repeats. Everything derived from
+        // `order_bys` below, such as the ordering fields in the aggregate's
+        // state, then matches the sort keys the accumulators compare.
         let order_bys = if fun.order_sensitivity().is_insensitive() {
             vec![]
         } else {
-            order_bys
+            LexOrdering::new(order_bys)
+                .map(Vec::from)
+                .unwrap_or_default()
         };
 
         let ordering_types = order_bys