BufferExec and AnalyzeExec should report eager evaluation

[geoffreyclaude]

Describe the bug

EvaluationType::Eager is documented as an operator stream that eagerly generates RecordBatch values in one or more spawned Tokio tasks. BufferExec and AnalyzeExec both appear to match that behavior, but their PlanProperties do not report eager evaluation.

In 5c92390:

• BufferExec::new(...) clones the input properties and only changes SchedulingType to Cooperative, so it keeps the input evaluation type.
• BufferExec::execute(...) wraps the input stream in MemoryBufferedStream::new(...).
• MemoryBufferedStream::new(...) immediately creates a SpawnedTask that polls the input stream into an internal queue.

Similarly:

• AnalyzeExec::compute_properties(...) calls PlanProperties::new(...), leaving evaluation_type at the default Lazy.
• AnalyzeExec::execute(...) uses RecordBatchReceiverStream::builder(...) and calls builder.run_input(...) for each input partition.
• The comments in AnalyzeExec::execute(...) describe those futures as running input partitions in parallel on separate Tokio tasks.

This makes EvaluationType less reliable for physical-plan analysis. DataFusion already exposes need_data_exchange(plan), implemented as:

plan.properties().evaluation_type == EvaluationType::Eager

so stale or incomplete EvaluationType metadata can cause eager child-polling operators to be missed.

To Reproduce

Inspect the physical properties of these operators:

1. Construct a BufferExec over a child whose evaluation_type is Lazy.
2. Check buffer_exec.properties().evaluation_type.
3. Construct an AnalyzeExec with a normal input plan.
4. Check analyze_exec.properties().evaluation_type.

Both report EvaluationType::Lazy in these cases, even though their execute(...) paths drive input polling from spawned tasks.

Expected behavior

If the documented contract for EvaluationType::Eager is intended to identify operators that drive child stream polling from spawned Tokio tasks, then BufferExec and AnalyzeExec should set PlanProperties::with_evaluation_type(EvaluationType::Eager).

BufferExec should probably always report eager evaluation because its buffering stream creates a background task to poll the input stream.

AnalyzeExec should probably report eager evaluation because it runs input partitions through RecordBatchReceiverStream::builder(...).run_input(...).

If this is not the intended meaning of EvaluationType::Eager, then the docs for EvaluationType and/or need_data_exchange(...) should be clarified so callers know which eager child-polling operators are intentionally excluded.

Additional context

This does not appear to be a query-result correctness issue. It is a physical-plan metadata consistency issue for optimizers and integrations that use DataFusion metadata to reason about execution topology.

[gabotechs]

There's one subtle different from EvaluationType::Eager's definition and what BufferExec does:

    /// The stream generated by [`execute`](ExecutionPlan::execute) eagerly generates `RecordBatch`
    /// in one or more spawned Tokio tasks. Eager evaluation is only started the first time
    /// `Stream::poll_next` is called.
    /// Examples of eager operators are repartition, coalesce partitions, and sort preserving merge.
    ///
    /// Eager operators are also known as a data-driven operators.
    Eager,

BufferExec does eagerly generates RecordBatches on a tokio spawned task, but it does not start evaluation after Stream::poll_next is called. It starts evaluation whether Stream::poll_next is called or not, which is the main point and the main reason why it brings performance benefits at the cost of memory usage.

That being said, I agree that it's more accurate if both declare EvaluationType::Eager

[gabotechs]

One thing that I worry about is this piece of code:

#[expect(clippy::needless_pass_by_value)]
pub fn need_data_exchange(plan: Arc<dyn ExecutionPlan>) -> bool {
    plan.properties().evaluation_type == EvaluationType::Eager
}

I think this is wrong, eager evaluation does not necessarily means that data needs to be exchanged. I know datafusion-distributed does not use this, and I also do not see Ballista using this (https://github.com/search?q=repo%3Aapache%2Fdatafusion-ballista+need_data_exchange&type=code), but if we mark BufferExec and AnalyzeExec as Eager, this will start being wrong.

Probably the solution is to change the need_data_exchange function body along with the evaluation type of BufferExec and AnalyzeExec.

[geoffreyclaude]

Thanks @gabotechs, I agree with both points.

For the first point, the EvaluationType::Eager rustdoc was defining the category too tightly by saying evaluation starts on the first Stream::poll_next. I think that startup timing is an implementation detail of eager operators. BufferExec starts its producer from execute, while other eager operators may start producer work on first poll; both are still eager from the caller's perspective because downstream demand can cause the operator to drive child input or produce batches independently of one downstream poll at a time.

For need_data_exchange, I agree that using evaluation_type == Eager makes the helper answer the wrong question once BufferExec and AnalyzeExec are classified accurately. The history is useful here: #4585 proposed need_data_exchange for callers that need to identify physical operators requiring exchange-style handling, and listed non-round-robin RepartitionExec, CoalescePartitionsExec, and SortPreservingMergeExec. #4586 implemented that helper around that exchange/gather meaning; it even moved the logic out of the ExecutionPlan trait into a free helper. Later, #16398 introduced EvaluationType for cooperative scheduling and made need_data_exchange delegate to eager evaluation. That was understandable while the eager set effectively matched the exchange/gather set, but it conflates two different properties.

I updated the PR so EvaluationType is documented as execution/evaluation behavior, while need_data_exchange(...) is restored to the exchange/gather predicate. That lets BufferExec and AnalyzeExec report eager evaluation without being treated as data-exchange operators.

replied through Codex