dwsmith1983 commented on PR #6835:
URL:
https://github.com/apache/datafusion-comet/pull/6835#issuecomment-6111648685
> @dwsmith1983 you worked on the native hash join ordering in #6810 and
#6785, so this touches code you know. If you have time, would you be able to
review it? Thanks.
Declaring the order on the probe input is the better fix for the two outer
shapes. DataFusion's hash join only checks whether the probe declares an order
(self.right.output_ordering().is_some()) before keeping unmatched probe rows in
place, so the declaration costs nothing, where #6810 sorts the whole streamed
side whenever it arrives from the JVM. With the in-memory cache on by default
since #5634 that sort now runs with default settings, so I'd rather this
approach covers those shapes.
The declared order goes past the join. The doc on SortedInputExec says the
order is declared on the probe "and nowhere else", but a Right join keeps its
probe's ordering in its own output, so everything above it in the native plan
sees it too. An aggregate grouped by the key switches to sorted, streaming
mode, and the sort aggregate's ordering_satisfy check treats the order as
already there. That is a win when Spark's claim holds. It also means a claim
that is wrong turns into wrong groups rather than only wrong order. Could the
comment say so, and could one test put an aggregate on the key above the join?
NOT IN is not covered, and it is a claim that does not hold. Spark reports
the streamed side's order for a null-aware anti join, but Comet runs it with
Spark's streamed side as DataFusion's build side, and DataFusion concatenates
the build batches in reverse, so its output is not in that order. Declaring an
order cannot fix that; it needs a sort. Under this PR an outer join above it
would wrap that output in SortedInputExec with the reported order.
On this branch it gives wrong groups. A cache repartitioned on k and sorted
within partitions, filtered with k NOT IN (...), then a broadcast left outer
join on k and a GROUP BY k, all in one native plan: the native plan puts
SortedInputExec: ordering=[col_0@0 ASC] over the null-aware LeftAnti join, and
both aggregates run with ordering_mode=Sorted. With 2000 rows over 40 keys it
returns 75 rows instead of 37, with 29 keys split into two groups whose counts
add up to the right one. It also fails at the default batch sizes, as long as a
partition spans more than one batch. main and #6810 return the right rows for
the same query. One trap if you add a test: dense keys take DataFusion's
perfect hash build path, which keeps the batches in order and hides the bug, so
the keys need to be spread out ((i % 40) * 1000 is enough).
#6810 already has the NOT IN sort and its tests. Since this branch turns
that case from wrong order into wrong groups, the sort needs to be in before or
with this change. I'll narrow #6810 to the NOT IN case now so it can land
first, using field 12 for now or field 11 if you'd rather this PR rebase onto
it. Once both are in, the outer-shape sort in #6810 goes away in favour of the
declaration here. Let me know if you agree with narrowing #6810.
Tests.
- Both new tests use left_outer with BuildRight. RightOuter with BuildLeft
takes the path that is not swapped (planner.rs around the BuildLeft branch) and
has no test.
- collect().toSet hides duplicate rows, which is the failure mode above.
Comparing the rows as a multiset would catch it.
- A planner test for with_declared_order: it wraps the right side, and it
skips the wrap when the input already declares an order.
- The fallback when the order cannot be expressed natively has no test.
- The repo asks for a Spark SQL suite verdict on planner and serde changes,
and the PR tier only ran 4.1.
The branch also conflicts with main in CometJoinSuite.scala.
```scala
test("NOT IN below an outer hash join over a sorted cache keeps each group
once") {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
SQLConf.SHUFFLE_PARTITIONS.key -> "2",
SQLConf.COLUMN_BATCH_SIZE.key -> "8",
CometConf.COMET_BATCH_SIZE.key -> "8",
CometConf.COMET_EXEC_IN_MEMORY_CACHE_ENABLED.key -> "true") {
// Sparse keys keep DataFusion off its perfect hash build path, which
keeps batch order.
withParquetTable((0 until 2000).map(i => ((i % 40) * 1000, i)), "big")
{
withParquetTable(Seq(1000, 10000, 20000, 100000).map(x =>
Tuple1(Option(x))), "excl") {
withParquetTable((0 until 40 by 2).map(k => (k * 1000, k)),
"small") {
try {
spark
.table("big")
.toDF("k", "v")
.repartition(2, $"k")
.sortWithinPartitions("k")
.cache()
.createOrReplaceTempView("cached")
spark.table("cached").count()
val query =
"""SELECT /*+ BROADCAST(b) */ a.k, count(*), sum(a.v),
count(b._2)
|FROM (SELECT * FROM cached WHERE k NOT IN (SELECT _1 FROM
excl)) a
|LEFT JOIN small b ON a.k = b._1
|GROUP BY a.k""".stripMargin
def rows(): Seq[(Int, Long, Long, Long)] =
sql(query)
.collect()
.map(r => (r.getInt(0), r.getLong(1), r.getLong(2),
r.getLong(3)))
.toSeq
.sorted
val cometRows = rows()
val sparkRows = withSQLConf(CometConf.COMET_ENABLED.key ->
"false")(rows())
assert(
cometRows.map(_._1).distinct.size == cometRows.size,
"a key is grouped twice")
assert(cometRows == sparkRows)
} finally {
spark.catalog.clearCache()
}
}
}
}
}
}
```
--
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]
