viirya commented on code in PR #6835: URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4243690680
########## 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: Thanks for spelling out what sees the order above the join. `SortExec` with a fetch is one more consumer. `compute_properties` takes `extract_common_sort_prefix` from its input, and TopK uses `common_sort_prefix` to stop early. So if the order doesn't hold, a native TopK above the join whose keys start with the declared order would return the wrong rows, not only wrong groups. Could the doc mention it too? ########## 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: This trusts the order Spark reports for the probe side whenever the native plan declares none. NOT IN is the case we know about where that claim is wrong, and #6810 takes care of it. But any other native operator that doesn't keep the order Spark reports for it would hit the same problem, and the result would now be wrong rather than just out of order. Would it work for `with_declared_order` to declare only when the order clearly comes from outside the native plan? For example, it could walk down from the probe root, require every node on the path to report `maintains_input_order()` for its child, and stop at a `ScanExec` or `ShuffleScanExec` leaf. Anything else would leave the plan as it is. The null-aware `HashJoinExec(LeftAnti)` reports `[false, false]`, so the walk would have stopped there even without #6810. Some of Comet's pass-through operators may need `maintains_input_order` implemented for the walk to get through them. ########## 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: The new aggregate test in `CometInMemoryCacheSuite` runs with batches of 8, so it reaches the split output path through the groups. The order tests here and in the cache suite still have no residual join condition and no null probe keys. Could you add a case with a condition, and one with null keys on the streamed side sorted with `desc_nulls_last`? Both go through parts of the in-order path the current data doesn't reach. #6810 already has a `descending, nulls last streamed order with null keys` test that could be ported. -- 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]
