Make it explicit when operators start work before their stream is polled

Aperta
  • Qualcosa non va?
#27,298 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
5/5
Tempo stimato
Più di una settimana
Idoneità per principianti
25/100
Tipo di issue
Funzionalità
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
rust
Ambito
backend, performance

Direzione di ricerca

Read the discussion in this issue and the linked #27126 change, especially the cited comment, to understand how operators start work and how that affects query concurrency. The proposal asks for documentation, an option to defer eager starts until the first poll, and plan visibility; done means those behaviors and settings are agreed on and made explicit, with defaults preserving current behavior.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

#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). 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 can behave differently: the former usually has a CoalescePartitionsExec on the build side, the latter a RepartitionExec.
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
  • 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.

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.

Lingua principale
Rust
Stelle
9.6k
Fork
2.6k
Merge medio
3g 4h
PR unite (30g)
345

Preparare l'ambiente

Apri in Codespaces

Avvia il container di sviluppo del progetto nel browser, con il tuo account GitHub.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/datafusion

Tutte le issue di apache/datafusion

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.