tschwarzinger opened a new issue, #27298: URL: https://github.com/apache/datafusion/issues/27298
#27126 changes `CoalescePartitionsExec` to start its inputs on the first poll instead of in `execute()`. I was surprised that `CoalescePartitionsExec` polls its children before its own stream is polled. With dynamic filters this can cost performance: the children start before the filters are ready and can’t use them (TPC-H q7/q21 with `pushdown_filters`). At the same time, other queries rely on this behavior. With simulated I/O latency, TPC-DS q88 is 6.6× slower with the change in#27126, because the parts of the query no longer run concurrently. My impression is that this behavior grew organically (see also [this comment](https://github.com/apache/datafusion/pull/27126#issuecomment-6690789008)). This issue is meant to start a discussion on making it explicit when DataFusion works on several parts of a plan concurrently. I think that `CoalescePartitionsExec` deciding whether the inner streams should be polled instantly grew organically (see also https://github.com/apache/datafusion/pull/27126#issuecomment-6690789008). This issue should start a discussion on whether we should make it more explicit when DataFusion works on multiple parts of the query plan concurrently. A Note: operators that do this spawn tasks that decouple producing batches from the consumer polling the stream. If they spawn these tasks in `execute()` rather than on the first poll, I call this starting eagerly in this issue. Some thoughts and findings: - *Suprising*: Whether an operator starts eagerly differs between operators, and even between partition counts of the same operator. See table below. As a result, `CollectLeft` and `Partitioned` hash joins behave differently: the former usually has a `CoalescePartitionsExec` on the build side, the latter a `RepartitionExec`. - *Not Configurable*: `CoalescePartitionsExec` can’t be used without starting eagerly with multiple partitions. That limits reuse in downstream systems that rely heavily on dynamic filters. - *Implicit*: Telling from a plan which parts run concurrently requires knowing each operator’s implementation. `EvaluationType::Eager` only exists in code, isn’t shown in `EXPLAIN`, and covers both cases: “decouples input from output” (`RepartitionExec`) and “decouples input from output and starts in execute()” (`CoalescePartitionsExec`). - *No CPU budget*: This might be a different part of the conversation but I guess many users assume that the number of target partitions is approximately the maximum number of CPU cores pinned by a single query. Working on multiple concurrent parts of a query in parallel can use more cores. The fact that is implicit may makes it more difficult to estimate the CPU usage. | **Operator** | **Starts eagerly?** | |--------|--------| | `CoalescePartitionsExec` | Yes with several input partitions; no with one | | `SortPreservingMergeExec` | Yes with several input partitions (spawn_buffered); no with one | | `BufferExec` | Yes | | `RepartitionExec` | No: it decouples input from output, but spawns its tasks on the first poll | Whether starting eagerly pays off depends on available resources, filter selectivity, and storage latency and throughput. Downstream users are often in a better position to decide than DataFusion. The code change is usually small: spawn the tasks on the first poll instead of in `execute()`. **Proposal** - *State it*. Every operator that starts eagerly documents this, and` EXPLAIN` shows it. - *Make it configurable*. Operators that start eagerly get a code-level option to start on the first poll instead, defaulting to today’s behavior. #27126 then becomes opt-in, and nothing regresses by default. After that, we can think of a better way of managing concurrency (e.g., let joins decide). Both go together: once the setting exists, showing it in plans covers the first point. A code-level option may be enough for now; a session config can follow if needed. I would also only expose the difference between starting tasks in `execute()` and the first poll in the plan properties once we have a use case for an optimization that relies on this information. **Out of scope** When each operator should start eagerly, for example pipeline breakers deciding based on pending dynamic filters, belongs in a separate issue. I hope to open a separate issue on that later. -- 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]
