andygrove commented on code in PR #6835: URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4248176107
########## native/core/src/execution/operators/sorted_input.rs: ########## @@ -0,0 +1,123 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! An input that declares the order its rows arrive in. + +use std::fmt::Formatter; +use std::sync::Arc; + +use datafusion::common::tree_node::TreeNodeRecursion; +use datafusion::common::{internal_err, Result}; +use datafusion::execution::TaskContext; +use datafusion::physical_expr::{EquivalenceProperties, LexOrdering, PhysicalExpr}; +use datafusion::physical_plan::execution_plan::CardinalityEffect; +use datafusion::physical_plan::{ + apply_expression_roots, DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, + SendableRecordBatchStream, +}; + +/// Passes `input` through unchanged and declares to DataFusion the order its rows arrive in. +/// +/// Spark's plan knows the order of an input that enters the native plan from the JVM, a relation +/// sorted before it was cached for instance, but the `ScanExec` it arrives through declares none, +/// and DataFusion's hash join keeps the unmatched probe rows of a right outer join in place only +/// when the probe input declares an order. Otherwise it moves them after the matched rows of each +/// batch, which loses an order Spark has already dropped a sort for. The planner puts this only +/// on that join's probe input, with the order Spark reports for it, but the join keeps the order +/// in its own output, so the operators above it in the native plan see it too: an aggregate +/// grouped by the key runs in sorted, streaming mode, and a sort aggregate takes the order as +/// already satisfied. The order therefore has to be one Spark guarantees, since a wrong one gives +/// wrong groups rather than only rows out of order. Review Comment: Added in 2f95eefb36. The doc now names a top-k sort whose keys start with the declared order, which stops reading early, so a wrong order gives wrong rows as well as wrong groups. ########## native/core/src/execution/planner.rs: ########## @@ -2937,6 +2951,27 @@ impl PhysicalPlanner { Ok(Arc::new(prepared)) } + /// `plan`, declaring `sort_orders` as the order its rows arrive in, unless it already + /// declares one. Spark's plan knows the order of an input that enters the native plan from + /// the JVM, but the `ScanExec` it arrives through declares none. + fn with_declared_order( + &self, + plan: Arc<dyn ExecutionPlan>, + sort_orders: &[Expr], + ) -> Result<Arc<dyn ExecutionPlan>, ExecutionError> { + if plan.output_ordering().is_some() { Review Comment: Done in 2f95eefb36. `with_declared_order` now walks down from the probe input through operators that report `maintains_input_order()` for one child, and declares the order only when it reaches a `ScanExec` or `ShuffleScanExec`. The walk can't take every operator at its word, though. DataFusion reports every right hash join as keeping its probe order, but `HashJoinExec` only keeps it when the probe declares an order (`right_side_ordered`), and nothing on this path declares one. A right join below whose probe got no declaration would pass the walk while moving its unmatched rows, so the walk stops at any hash join that isn't an inner join. Comet's filter and projection already forward `maintains_input_order`, so the cache and `CometSparkToColumnar` shapes get through without more work. The planner tests cover a projection on the path, the null-aware anti join, and a right join below. With the walk in place, @dwsmith1983's NOT IN repro groups correctly on this branch, so I added it now rather than after the rebase. ########## spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala: ########## @@ -2254,4 +2254,112 @@ class CometJoinSuite extends CometTestBase { } } } + + test("outer hash joins keep a streamed side sorted outside the native plan in order") { + // Spark drops the sort above the join because the join reports the streamed side's order, + // and the streamed side arrives sorted through a JVM input, which declares no order to + // DataFusion unless the planner declares it on the join's probe input; see + // CometHashJoin.doConvert. CometInMemoryCacheSuite covers the native cache scan. + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "2", + CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "false", + CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED.key -> "true") { Review Comment: Added both in 2f95eefb36. The outer test here now runs each shape with and without `build._2 > 3` as a residual condition, and a new test ports the descending, nulls last case with null keys from #6810 for both shapes, checking each partition with an `InterpretedOrdering` over the key. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
