parthchandra opened a new issue, #7268:
URL: https://github.com/apache/datafusion-comet/issues/7268
Part of #7267.
## Problem
Comet builds DataFusion's `FilterExec` with `FilterExec::try_new(predicate,
child)`
in `native/core/src/execution/planner.rs` (the `OpStruct::Filter` arm) and
never
sets its batch size. `FilterExec` merges its output with a
`LimitedBatchCoalescer`
whose target is `FilterExec::batch_size`. That field defaults to
`FILTER_EXEC_DEFAULT_BATCH_SIZE = 8192` (DataFusion 55.2.0, `filter.rs:80`).
DataFusion's own physical planner sets it from the session config
(`physical_planner.rs`,
`FilterExecBuilder::with_batch_size(session_state.config().batch_size())`).
Comet skips that step, so:
- `spark.comet.batchSize` has no effect on filter output. The scan honors
it, then
the filter merges batches back up to 8192 rows.
- With large string/binary values, 8192 rows can exceed 2 GiB and the merge
fails
with `Offset overflow error` inside `concat_bytes` ->
`GenericByteBuilder::append_array`.
This match arm is the only place Comet creates a `FilterExec`
(`planner.rs:1694` on main). Every later rebuild keeps the field:
- The filter/projection fusion in
`native/core/src/execution/operators/projection.rs:68-79`
rebuilds it with `FilterExecBuilder::from`, which copies `batch_size`
(DataFusion `filter.rs:225-232`).
- `CometFilterExec` (now `MetricOwningExec<FilterExec>`,
`native/operators/src/filter.rs:28`,
built only in `planner/join.rs:373-381`) wraps a clone. Its
`replace_input`,
`with_fetch` and `with_execution_input` go through
`FilterExec::replace_children`
and `with_fetch`, which keep the value (`filter.rs:555-575, 822-832`).
So one planner change covers all of them.
## Proposed fix
```rust
let filter = FilterExecBuilder::new(predicate,
Arc::clone(&child.native_plan))
.with_batch_size(self.session_ctx.copied_config().batch_size())
.build()?;
```
## Tests
- Rust planner test: plan a Filter with a session batch size of N and assert
the
native `FilterExec::batch_size() == N`. Also cover the filter+projection
fusion path.
- Batch-size test: with `spark.comet.batchSize=10`, filter a table where
most rows
pass, and assert no native output batch has more than 10 rows. This fails
before
the fix (batches of up to 8192 rows). It can't be a SQL-file test because
SQL
can't see batch sizes.
- Result check: a SQL-file test with a binary column of values a few hundred
KB in
size, under a filter, at a small batch size, compared with Spark.
## Scope and limits
- `spark.comet.batchSize` defaults to 8192, the same as DataFusion's filter
default,
so at default settings this change does nothing at runtime. It makes
lowering
`spark.comet.batchSize` work as a workaround.
- Still row-based. At 8192 rows, any string/binary column averaging more
than ~256
KiB per value still overflows. Users must lower `spark.comet.batchSize`
until
sub-issue 4 lands.
- `LimitedBatchCoalescer` passes an input batch larger than half the target
through
unmerged when nothing is buffered (DataFusion `coalesce/mod.rs:63-64`), so
filter
output batches can be up to the scan's batch size, which is also the
session value.
- Users who raised `spark.comet.batchSize` above 8192 now get bigger filter
batches,
matching what other operators already do with the session value.
- Touches the planner, so run `dev/local-ci.sh spark sql_core-1` or apply the
`run-spark-4.1-tests` label before merging.
--
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]
