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

Abierto
  • ¿Algo no está bien?
#27,298 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
5/5
Tiempo estimado
Más de una semana
Aptitud para principiantes
25/100
Tipo de issue
Nueva funcionalidad
Claridad
Bastante claro
Estado de actividad
Activo
Stack tecnológico
rust
Área
backend, performance

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

#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.

Lenguaje dominante
Rust
Estrellas
9.6k
Forks
2.6k
Merge medio
3 d 4 h
PR fusionados (30 d)
345

Preparar el entorno

Abrir en Codespaces

Inicia el contenedor de desarrollo del proyecto en tu navegador, con tu propia cuenta de GitHub.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/datafusion

Todos los issues de apache/datafusion

Issues similares

Más issues de Rust

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.