parthchandra opened a new pull request, #7280:
URL: https://github.com/apache/datafusion-comet/pull/7280
## Which issue does this PR close?
Closes #7268.
Part of #7267. That epic stays open: this change only helps when
`spark.comet.batchSize` is lowered, and it doesn't remove the 2 GiB limit.
## Rationale for this change
Comet builds the native filter with `FilterExec::try_new` and never sets its
batch size. `FilterExec` merges small output batches into bigger ones, and
without a batch size it uses DataFusion's built-in default of 8192 rows.
DataFusion's own physical planner passes the session batch size here; Comet's
planner skipped that step.
So the filter ignores `spark.comet.batchSize`. The scan honors the setting
and emits small batches, then the filter merges them back up to 8192 rows. With
large string or binary values (for example blobs of a few hundred KB each), one
merged column can pass Arrow's 2 GiB limit for 32-bit offsets. The query then
fails with:
org.apache.comet.CometNativeException: Offset overflow error: <n>
Lowering `spark.comet.batchSize` is the natural workaround, but today it has
no effect on this path, even at `batchSize=1`.
## What changes are included in this PR?
- The planner builds the filter with `FilterExecBuilder` and the session
batch size, the same way DataFusion's planner does.
- Every place that later rebuilds the filter keeps the value. That covers
the projection-pruning rebuild and the `CometFilterExec` wrapper used for join
probe sides. I checked that this match arm is the only place Comet creates a
`FilterExec`.
- Adds a small `CometFilterExec::batch_size()` accessor, so a test can check
the wrapped filter.
Behavior change: the filter now follows `spark.comet.batchSize` in both
directions.
- At the default of 8192, nothing changes at runtime.
- Lowering it now bounds the filter's output. That gives users with large
values a working workaround (for ~1 MB values, about 1024 or less).
- Raising it above 8192 now gives larger filter output batches, as other
operators already do with the session value.
This is still a row-count limit. At 8192 rows, any string/binary column
averaging more than about 256 KiB per value can still overflow. A byte limit in
the coalescer (#7271) is the fix that doesn't depend on row counts. Other
operators and batch sources with the same problem are tracked in #7267.
## How are these changes tested?
Rust tests. Each one fails without the fix and passes with it:
- `filter_output_respects_session_batch_size`: with a session batch size of
2, five 1-row batches that all pass the filter must come out as `[2, 2, 1]`.
Without the fix they merge into one 5-row batch. The batches include a string
column with NULL, empty and multi-byte values, and the test checks every value
comes through in order.
- `projection_prunes_filter_output_and_preserves_metrics` (existing test):
now plans with a session batch size of 5 and asserts the filter keeps it, with
and without projection pruning.
- `probe_filters_keep_session_batch_size` (`planner/join.rs`): prepares a
filter -> projection -> filter plan as a join probe side, and checks both
wrapped `CometFilterExec`s keep the session batch size.
SQL file test `operators/filter_varlen_values.sql`: with
`spark.comet.batchSize=4`, filters over binary, string, `array<struct<binary,
string>>` and `map<string, binary>` columns, including NULL, empty and
non-ASCII values, compared with Spark. One filter keeps about one row in three,
so every batch gets merged. This test checks that merged batches match Spark.
It passes with or without the fix, because SQL can't see batch sizes; the Rust
tests are the regression guard.
--
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]
