comphead commented on code in PR #7316:
URL: https://github.com/apache/datafusion-comet/pull/7316#discussion_r4417471261
##########
native/core/src/execution/planner.rs:
##########
@@ -777,6 +805,7 @@ impl PhysicalPlanner {
}
}
};
+ self.note_scalar_bytes(&scalar_value);
Review Comment:
Every item of an `In` list is planned through this arm, so its bytes count
toward `scalar_bytes`, although list items are only compared and never expanded
to the batch length. `CometInSet` turns a Spark `InSet` into this `In` message,
so `col IN (<25,000 UUID strings>)` is about 900 KB of literals and lowers the
batch size to about 1,165 rows for every operator in the task, with no overflow
risk to justify it. Could the `In` arm plan its items without counting them?
##########
native/core/src/execution/jni_api.rs:
##########
@@ -1555,6 +1573,17 @@ fn execute_plan(
exec_context.scans = scans;
exec_context.shuffle_scans = shuffle_scans;
+ // The task context created below copies the session config, so every
source and the
+ // JVM-fed leaves see the smaller batch size.
+ let batch_size = exec_context.session_ctx.copied_config().batch_size();
+ let rows = scalar_batch_rows(planner.scalar_bytes(), batch_size);
+ if rows < batch_size {
+ let state = exec_context.session_ctx.state_ref();
+ let mut state = state.write();
+ let config = state.config_mut();
+ *config = std::mem::take(config).with_batch_size(rows);
Review Comment:
`FilterExec` keeps its own batch size, 8192 by default in DataFusion 55.2
(`FILTER_EXEC_DEFAULT_BATCH_SIZE`), and coalesces its output up to it through
`LimitedBatchCoalescer` without reading the session value at execute time.
Comet builds it with `FilterExec::try_new`, so a filter under the expanding
projection turns two capped 4096-row batches back into one 8192-row batch. I
expect `SELECT concat((SELECT max(s) FROM wide), CAST(k AS STRING)) FROM ids
WHERE k >= 0` to still hit the offset overflow. #7280 passes the session batch
size into `FilterExecBuilder`, but at planning time, before this cap lowers it,
so it would not cover this either. Could the cap reach filters, for example by
computing the scalar bytes from the protobuf plan before the planner runs, and
could the fixture add a query with a `WHERE`?
##########
native/core/src/execution/operators/scan.rs:
##########
@@ -302,6 +309,7 @@ impl ScanStream<'_> {
baseline_metrics,
cast_options: CastOptions::default(),
cast_time,
+ slicer: BatchSlicer::new(batch_size),
Review Comment:
The description says that with the default configuration nothing is sliced,
but the slicer is installed with the session batch size whether or not the cap
lowered it. `Utils.coalesceBroadcastBatches` deliberately merges a broadcast
into few large batches, so a broadcast build side can arrive here as one batch
of millions of rows and is now cut into 8192-row slices. `HashJoinExec` then
rebuilds the build side with `concat_batches`, which copies every slice, where
a single input batch took arrow's zero-copy path. Could the slicer be installed
only when the cap actually lowered the batch size? I have not confirmed at
runtime that a broadcast reaches `ScanExec` as one batch.
--
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]
