viirya commented on code in PR #6835: URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4251867497
########## 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, the doc reads well now. ########## 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: Thanks, and good catch on the right join. I was only thinking about the null-aware anti join reporting `[false, false]`, and missed that DataFusion reports every right join as keeping its probe order even though it only keeps it when the probe declares one. Stopping at any hash join that isn't an inner join is the right call. ########## 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: Thanks for adding both. -- 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]
