fix: keep a streamed side sorted outside the native plan in order through outer hash joins

docs/source/contributor-guide/adding_a_new_operator.md

@@ -357,6 +357,29 @@ case class CometProjectExec(
 }
 ```
 
+#### Reporting an Output Ordering
+
+An operator that overrides `outputOrdering`, as `CometFilterExec` does above with
+`child.outputOrdering`, makes a claim that Spark acts on: it drops a sort above the operator that
+the order satisfies. Report what the Spark operator reports and nothing more. `SampleExec`, for
+example, reports no order, so `CometSampleExec` must not report its child's.
+
+The claim has to hold at run time as well. A native hash join declares the order Spark reports for
+its streamed side to DataFusion, and an aggregate above the join may then run in sorted mode, or a
+top-k sort may stop reading early, on the strength of that order. If the order is false, the
+result is wrong groups or wrong rows, not only rows out of order. So an operator that reports an
+ordering must keep it natively, and the native operator must say so by returning `true` from
+`maintains_input_order` for that child, as DataFusion's filter and projection do. The planner
+declares the order through an operator only when the operator says it keeps it. An operator that
+cannot keep the order natively must not report it, or has to be added to `mayReportFalseOrder` in
+`operators.scala`, which makes the hash join serde send no order over it.
+
+A leaf that reports an ordering is where the order starts, so the planner has to know that the
+native operator produces its rows in that order. It knows that of an input from the JVM and of a
+scan of files, which reads the files of each partition one after the other, and the native range
+declares its order itself through `SortedInputExec`. A new leaf that reports an ordering must
+declare it natively the same way.
+
 #### Plan Identity and Exchange Reuse
 
 Spark's `ReuseExchangeAndSubquery` identifies equivalent plans through canonicalization and may

docs/source/user-guide/latest/compatibility/operators.md

@@ -91,6 +91,27 @@ rows by the grouping keys only, and Spark and Comet can leave rows with equal ke
 orders, so a group with more than one candidate value can return a different value than Spark.
 Both results are valid under Spark's semantics for these functions.
 
+## Hash Join Output Order
+
+Spark reports the sort order of a hash join's streamed side as the join's own, and drops a sort
+above the join that this order satisfies. Comet's native hash joins keep that order where Comet can
+show that it holds. That includes a left or right outer join over a streamed side that was sorted
+outside the native plan, such as a sorted relation read from the [in-memory cache](#in-memory-cache).
+
+Comet passes the order to the native join as sort keys, and an outer join needs at least the first
+key to be one that Comet can express natively. Declaring any leading part of the order keeps the
+rows in place, so the join does not need the rest. An outer join whose first sort key Comet cannot
+express falls back to Spark and records a fallback reason (see
+[Understanding Comet Plans](../understanding-comet-plans.md)), because Comet cannot tell whether
+an operator above the join relies on the order.
+
+Where the order Spark reports might not hold, Comet declares nothing and an outer join can return
+its unmatched rows out of order. The output of a `NOT IN` subquery or of a nested loop join
+([#6898](https://github.com/apache/datafusion-comet/issues/6898)) is one such case, and so is a
+cache or checkpoint that one of them wrote. Comet also declares nothing when it cannot follow the
+order back to its source, as with an RDD checkpoint, and over some caches written by an outer hash
+join whose order does hold ([#6922](https://github.com/apache/datafusion-comet/issues/6922)).
+
 ## Window Functions
 
 Comet runs `WindowExec` natively and it is enabled by default (`spark.comet.exec.window.enabled`). A broad set of

native/core/src/execution/operators/mod.rs

@@ -41,5 +41,7 @@ mod csv_scan;
 pub mod projection;
 mod scan;
 mod shuffle_scan;
+mod sorted_input;
 pub use csv_scan::init_csv_datasource_exec;
 pub use shuffle_scan::ShuffleScanExec;
+pub use sorted_input::SortedInputExec;

native/core/src/execution/operators/sorted_input.rs

@@ -0,0 +1,135 @@
+// 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::{LexOrdering, PhysicalExpr};
+use datafusion::physical_plan::execution_plan::CardinalityEffect;
+use datafusion::physical_plan::metrics::MetricsSet;
+use datafusion::physical_plan::{
+    apply_expression_roots, DisplayAs, DisplayFormatType, ExecutionPlan, ExecutionPlanProperties,
+    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, and only when the rows reach
+/// that input from the JVM through operators that keep their order. 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, a sort aggregate takes the order as already
+/// satisfied, and a top-k sort whose keys start with it stops reading early. A wrong order would
+/// therefore give wrong groups or wrong rows, not only rows out of order.
+#[derive(Debug)]
+pub struct SortedInputExec {
+    input: Arc<dyn ExecutionPlan>,
+    ordering: LexOrdering,
+    cache: Arc<PlanProperties>,
+}
+
+impl SortedInputExec {
+    pub fn new(input: Arc<dyn ExecutionPlan>, ordering: LexOrdering) -> Self {
+        let input_properties = input.properties();
+        // Start from the input's own properties, so the constants and equivalence classes it
+        // knows of stay visible to the operators above.
+        let mut equivalence_properties = input.equivalence_properties().clone();
+        equivalence_properties.add_ordering(ordering.iter().cloned());
+        let cache = Arc::new(PlanProperties::new(
+            equivalence_properties,
+            input_properties.output_partitioning().clone(),
+            input_properties.emission_type,
+            input_properties.boundedness,
+        ));
+        Self {
+            input,
+            ordering,
+            cache,
+        }
+    }
+}
+
+impl DisplayAs for SortedInputExec {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut Formatter) -> std::fmt::Result {
+        write!(f, "SortedInputExec: ordering=[{}]", self.ordering)
+    }
+}
+
+impl ExecutionPlan for SortedInputExec {
+    fn name(&self) -> &str {
+        "SortedInputExec"
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        &self.cache
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![&self.input]
+    }
+
+    fn apply_expressions(
+        &self,
+        f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
+        apply_expression_roots(self.ordering.iter().map(|sort| Arc::clone(&sort.expr)), f)
+    }
+
+    fn maintains_input_order(&self) -> Vec<bool> {
+        vec![true]
+    }
+
+    /// The rows pass through, so the input's metrics are this operator's, and a range whose root
+    /// this wraps still reports its output rows.
+    fn metrics(&self) -> Option<MetricsSet> {
+        self.input.metrics()
+    }
+
+    fn cardinality_effect(&self) -> CardinalityEffect {
+        CardinalityEffect::Equal
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        mut children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        if children.len() != 1 {
+            return internal_err!("SortedInputExec requires one child");
+        }
+        Ok(Arc::new(Self::new(
+            children.remove(0),
+            self.ordering.clone(),
+        )))
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        context: Arc<TaskContext>,
+    ) -> Result<SendableRecordBatchStream> {
+        self.input.execute(partition, context)
+    }
+}