parthchandra opened a new issue, #7273:
URL: https://github.com/apache/datafusion-comet/issues/7273
Part of #7267.
## Problem
Both joins merge the whole build side into a single batch so they can index
into it
directly:
- `HashJoinExec`: `concat_batches` over all build batches (DataFusion 55.2.0,
`joins/hash_join/exec.rs:2365`). The array-map path for a single integer
key does
the same (`exec.rs:185`, `try_create_array_map`).
- `NestedLoopJoinExec`: same (`joins/nested_loop_join.rs:968`). There is a
second
concat in the memory-limited path (`:1789`); unsure whether Comet reaches
it.
If the build side carries more than 2 GiB in any string/binary column,
including
payload columns not used in the join key, the concat fails with
`OffsetOverflowError`. Batch size has no effect.
A second, lower-risk path hands an oversized batch downstream. When the
build side
must emit its unmatched rows at the end (Left, LeftSemi, LeftAnti, LeftMark
and Full
with left as build, `need_produce_result_in_final`, `joins/utils.rs:1181`),
`HashJoinExec` takes them from the build batch as one batch
(`joins/hash_join/stream.rs:1038-1054`) and pushes it into its output
coalescer. The
coalescer passes a batch over half the target through unsplit when nothing is
buffered (`coalesce/mod.rs:64`, arrow `coalesce.rs:451-470`). This can't
overflow
by itself, because the build batch was already concatenated, but downstream
operators get a batch as large as the build side. Comet swaps inputs so the
build
side is on the left, so this only hits Spark joins whose build side is the
outer or
semi side (full outer, or BuildLeft left-outer/semi/anti). The common
BuildRight
left-outer broadcast join doesn't hit it.
In Comet this affects broadcast and shuffled hash joins with a large build
side
holding big values. Comet still uses DataFusion's `HashJoinExec`
(`planner/join.rs:140`, `Partitioned`, or `CollectLeft` for null-aware anti
join),
sometimes wrapped in `DynamicFilterJoinExec` for inner equi-joins, which
still runs
`HashJoinExec`. Nested loop join is built at `join.rs:309`.
## Options
1. **Upstream:** keep the build side as several batches and index with
(batch, row) pairs, or split it into chunks under 2 GiB per column.
2. **Upstream or Comet:** cast large string/binary build-side columns to
`LargeUtf8`/
`LargeBinary` or view types before the concat, and cast back on output.
3. **Comet:** prefer sort-merge join (no build-side concat) when the build
side has
large var-len columns and the size estimate is near the limit.
## Tests
- DataFusion test of the chosen fix with a build side split into chunks.
- Comet SQL-file test joining a small key table to a table with a large
binary
payload column on the build side, compared with Spark.
## Notes
- `SortMergeJoinExec` output uses `interleave` limited by row count only.
Lower
risk, but it would benefit from the same retry approach.
--
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]
