Native hash joins lose the streamed side's order when it arrives sorted from outside the native plan #6787

Describe the bug

A native hash join can return its streamed side's rows out of order while Spark relies on that order, so a sort Spark removed as redundant is lost and sortWithinPartitions returns unsorted partitions.

Spark's HashJoin.outputOrdering is the streamed side's ordering, and CometHashJoinExec and CometBroadcastHashJoinExec copy it from the Spark join (op.outputOrdering in operators.scala). When the streamed side already arrives partitioned and sorted on the key, with no exchange in between, RemoveRedundantSorts drops a sort above the join on that key.

Natively the join builds on the left, so a BuildRight join becomes a DataFusion join whose probe side is the streamed side. DataFusion keeps the probe side's order for unmatched rows only when that input declares an ordering: HashJoinExec passes self.right.output_ordering().is_some() (datafusion-physical-plan 55.1.0, joins/hash_join/exec.rs:1574) to append_right_indices, which otherwise appends the unmatched probe rows after the matched ones in each batch. A sorted input that reaches the native plan from the JVM goes through ScanExec, which declares no ordering, so for a left outer join building on the right the unmatched rows come out after the matched ones.

When the sort runs natively in the same plan as the join, DataFusion sees its ordering and the output stays sorted.

Steps to reproduce

Parquet tables big as (i % 100, i) for 10000 rows and small as (i * 10, i) for 10 rows, with spark.comet.convert.inMemoryCache.enabled=true, spark.sql.adaptive.enabled=false, spark.sql.autoBroadcastJoinThreshold=-1 and spark.sql.shuffle.partitions=2:

val streamed = spark.table("big").repartition(2, $"_1").sortWithinPartitions("_1").cache()
streamed.count()
val build = spark.table("small")
val df = streamed
  .join(build.hint("shuffle_hash"), streamed("_1") === build("_1"), "left_outer")
  .select(streamed("_1").as("k"), streamed("_2").as("v"), build("_2").as("w"))
  .sortWithinPartitions("k")
df.queryExecution.toRdd
  .mapPartitions { it =>
    val keys = it.map(_.getInt(0)).toArray
    Iterator(keys.sameElements(keys.sorted))
  }
  .collect()
// Array(false, false); the rows match Spark's

The plan has no sort above the join:

CometProject [k, v, w]
+- CometHashJoin [_1], [_1], LeftOuter, BuildRight
   :- CometSparkColumnarToColumnar
   :  +- InMemoryTableScan [_1, _2]
   +- CometExchange hashpartitioning(_1, 2)

build.hint("broadcast") gives the same result with CometBroadcastHashJoin. Reproduced on main (9876862) with Spark 4.1. branch-1.1 copies the Spark join's output ordering the same way.

Expected behavior

Every partition sorted by k, as in Spark.

Additional context

It needs a sorted streamed side that reaches the native join from outside the native plan. Here that is a cached table read through spark.comet.convert.inMemoryCache.enabled; I have not checked which other inputs reach it, such as Comet's native cache scan or another JVM operator below a native join. A fix could have the two hash join execs report no output ordering for a join type and build side whose unmatched probe rows DataFusion may reorder, unless the probe child is known to keep its order in the native plan, or have the scan declare the ordering it receives.

Found while working on #6770.