comphead commented on PR #4587:
URL:
https://github.com/apache/datafusion-comet/pull/4587#issuecomment-5723010195
## Review: native ExistenceJoin
I traced the change end to end (Spark plan -> serde -> proto -> `planner.rs`
-> DataFusion 55.1.0) and checked the Spark reference behaviour in
`RewritePredicateSubquery` and `ResolveHints`.
### What checks out
The core mapping is sound, so the findings below are about coverage and
packaging rather than the join semantics:
- `BuildRight` is the only build side Spark produces for `ExistenceJoin`
(`canBuildBroadcastLeft` / `canBuildShuffledHashJoinLeft` both exclude it), so
the hash-join path always takes the `swap_inputs` branch. `JoinType::swap()`
turns `LeftMark` into `RightMark`, `swap_inputs` skips the reverting projection
for mark joins, and `build_join_schema` lays `RightMark` out as right fields
followed by `mark`. After the swap `right` is Spark's left, so the native
schema is `[left.output..., mark]`, which matches Spark's `left.output :+
exists`.
- `Field::new("mark", Boolean, false)` matches `AttributeReference("exists",
BooleanType, nullable = false)`.
- DF 55.1.0 `SortMergeJoinExec` routes `LeftMark` through
`BitwiseSortMergeJoinStream`, streams the left side, and reports
`maintains_input_order = [true, false]`, which matches Spark's
`SortMergeJoinExec.outputOrdering` for `LeftExistence(_)`.
- The join dynamic filter is gated on `join.join_type() == Inner`, so mark
joins are unaffected.
- `outputPartitioning` already handles the new type via the existing
`LeftExistence(_)` case.
### Blocking: the SQL fixture only ever exercises BroadcastHashJoin
`existence_join.sql` states that the hints run each query against BHJ, SHJ
and SMJ. They do not. Spark's `ResolveHints.applyJoinStrategyHint` descends
with `mapChildren`, which visits plan children only and never subquery
expressions, and it stops at `SubqueryAlias`. For
```sql
SELECT /*+ BROADCAST(ex_right) */ * FROM ex_left l
WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right r WHERE r.k = l.k)
```
the traversal reaches `SubqueryAlias(l)` and stops, `ex_right` is never
matched, and the hint is dropped with a `hintRelationsNotFound` warning.
Putting the hint inside the subquery does not help either, since
`EliminateResolvedHint` runs in the first optimizer batch (`FinishAnalysis`)
while `RewritePredicateSubquery` runs in the much later `RewriteSubquery`
batch, so there is no `Join` for the hint to attach to.
Consequences:
- `CometTestBase` sets `autoBroadcastJoinThreshold=1g`, so all ten queries
plan as BHJ. `CometSortMergeJoinExec` and `CometHashJoinExec` get no SQL-level
coverage at all.
- Queries 1, 2 and 3 are the same query three times, and the two empty-build
queries are the same query twice.
Strategy selection has to come from config. The fixture format already
supports this:
```
-- Config: spark.sql.adaptive.autoBroadcastJoinThreshold=-1
-- Config: spark.sql.join.forceApplyShuffledHashJoin=true
-- ConfigMatrix: spark.sql.autoBroadcastJoinThreshold=10485760,-1
-- ConfigMatrix: spark.sql.join.preferSortMergeJoin=true,false
```
That yields four runs covering BHJ, SMJ and SHJ, and lets the three
duplicated query groups collapse to one each.
### Missing coverage for the reachable non-equi paths
`RewritePredicateSubquery` produces `ExistenceJoin` from three shapes, and
the fixture covers only one of them.
1. **Join filter.** `Exists(sub, _, _, conditions, _)` puts the whole
correlated predicate in the join condition. A correlation such as `r.k = l.k
AND r.v > l.v` splits into equi keys plus `condition = Some(r.v > l.v)`, which
serialises and reaches `HashJoinExec` with a `JoinFilter` that `swap_inputs`
then rewrites via `JoinFilter::swap`, and reaches `BitwiseSortMergeJoinStream`
on the SMJ side. This is the least-travelled DataFusion path in the change and
nothing in the PR exercises it. Note the SMJ variant is additionally gated by
`spark.comet.exec.sortMergeJoinWithJoinFilter.enabled`, which defaults to true.
2. **`IN` combined with `OR`.** `InSubquery` lowers to `ExistenceJoin` with
plain equi keys, so it goes straight through the new code and is untested.
3. **`NOT IN` combined with `OR`.** This builds the condition `(l.k = r.k)
OR isnull(l.k = r.k)`, from which `ExtractEquiJoinKeys` extracts nothing, so it
plans as a `BroadcastNestedLoopJoin` and falls back. That is a reasonable scope
boundary, but the config name suggests otherwise. Worth an `expect_fallback`
case plus a sentence in the config doc.
Also missing: a fallback case proving the join falls back when the config is
off (the default), and any non-integer key type. Join keys are the one place
`NormalizeFloatingNumbers` and string/decimal/date comparison differences show
up, so a double key with `NaN` and `-0.0` and a string key would be worth
having.
### Benchmark compares different plans on the two arms
`runExpressionBenchmark` applies `extraCometConfigs` to the Comet arm only,
and the queries in `CometExistenceJoinBenchmark` carry no hints. The Spark
baseline therefore runs with the session defaults and plans a BroadcastHashJoin
in all three cases, so the "ShuffledHashJoin" and "SortMergeJoin" rows measure
Comet SHJ and Comet SMJ against Spark BHJ. (`CometHashJoinBenchmark` gets away
with the same config map because its queries carry `SHUFFLE_HASH` hints that
apply to both arms, and as above, hints are not available here.)
Fix is to wrap each `runBenchmark` in `withSQLConf(strategyConfigs: _*)` so
both arms see the same strategy, leaving only
`COMET_EXEC_EXISTENCE_JOIN_ENABLED` in `extraCometConfigs`.
### Reuse and duplication
- **Benchmark file.** `CometExistenceJoinBenchmark` is a fourth copy of a
`getSparkSession` that already exists verbatim in `CometHashJoinBenchmark`,
`CometBroadcastHashJoinBenchmark` and `CometSortMergeJoinBenchmark`, and its
SHJ config map is identical to `CometHashJoinBenchmark.cometConfigs`. An
`exists OR predicate` case added to each of those three files would be a few
lines each instead of 123, and each file already forces its own strategy.
- **`joinType` match.** The same eight-line match now exists three times in
`operators.scala` (`CometHashJoin.doConvert`, `CometSortMergeJoinExec.convert`,
`CometBroadcastNestedLoopJoinExec.convert`), and the PR edits two of the three.
A single `toProtoJoinType(join, allowExistence)` helper would leave one place
to touch when the next join type lands.
- **`producedAttributes`.** Three byte-identical overrides. Since `output`
differs from `inputSet` only for `ExistenceJoin`, `AttributeSet(output) --
inputSet` is a one-liner that needs no join-type match, or put the existing
version on a small shared trait alongside the already-duplicated
`outputPartitioning`.
- **`CometJoinSuite`.** The three new tests differ only in their config map
and expected exec class. One table-driven test over `Seq((label, configs,
expectedClass))` covers the same ground in about a third of the lines. The `/*+
BROADCAST(b) */` inside the first test's subquery is dead for the reason
described above, and the test already forces broadcast by threshold, so it
should go.
### Smaller points
- When the config is off, the fallback reason is `Unsupported join type
ExistenceJoin(exists#N)`, which reads like a permanent limitation rather than a
flag being off. `CometSortMergeJoinExec` already models the better message for
the `sortMergeJoinWithJoinFilter` gate.
- `createExecEnabledConfig("existenceJoin", ...)` renders as "Whether to
enable existenceJoin by default", but this is not an exec, it is a join-type
gate spanning three execs.
`spark.comet.exec.sortMergeJoinWithJoinFilter.enabled` is the established
precedent for a join sub-feature and uses a plain `conf(...)` with a written
doc string.
- `planner.rs` has join-type coverage in its unit tests around the
`build_side` cases. One case for `JoinType::Existence` asserting the swapped
plan's schema ends in `mark` would pin the mapping without a Spark round trip.
--
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]
