Compute safe reusable expressions before broadcast materialization

[sunchao]

Problem, background and scope

Comet can evaluate the same build-side expression once in every probe task, even when the input comes from a shared broadcast. Moving a native projection below a native hash join does not solve that repetition if the entire native plan is instantiated separately by each Spark task. Computing an eligible expression before Spark broadcast materialization can replace repeated evaluation with one producer computation and a reusable column.

The optimization is general: identify pure, supported, side-local expressions and prove that moving their evaluation preserves results and errors. Do not match TPC-DS query IDs, table names, column names, or particular literals.

This is a focused P1 planning improvement. The first implementation should be a small rule for proven expression classes, with an explicit no-change path for unsupported cases.

Potential technical approach

Start at Comet's Spark extension/preparation wiring. Add a guarded logical/preparation transformation before the producer is materialized; select the precise supported Spark hook and test its ordering rather than injecting unresolved expressions late in physical planning. The broadcast collection boundary must receive the producer projection, not leave it in each consumer.

flowchart TD
    A["Build relation and consumer expression"] --> B{"Semantics and placement proven safe?"}
    B -->|No| C["Keep original plan"]
    B -->|Yes| D["Estimate producer work and added column bytes"]
    D --> E{"Reuse benefit exceeds added work and transport?"}
    E -->|No or unknown| C
    E -->|Yes| F["Compute alias before broadcast materialization"]
    F --> G["Broadcast compatible schema once"]
    G --> H["Consumers read alias instead of recomputing"]

Loading

1. Initially support side-local expressions under Inner joins using an explicit safety contract for implemented builtins. Deterministic metadata is insufficient: moving a throwing expression earlier can fail on rows the original join discarded. Some Spark expressions also report nonthrowing metadata despite ANSI evaluation errors. Exclude opaque UDFs and unproved expression classes.
2. Do not extend this mechanically across outer null extension. For example, a post-join coalesce(build.x, 0) can produce0 on an unmatched row; a precomputed alias can instead be null-padded. Any later outer-join support needs its own proof and tests.
3. Preserve expression IDs, types, nullability, collation/time-zone behavior, join conditions and consumer output schemas. Repeated optimizer application must be idempotent. Do not mutate an already materialized broadcast or reuse an incompatible schema for another consumer.
4. Recompute truthful projection statistics. Added bytes can cross broadcast/AQE thresholds or increase decode/spill enough to erase CPU savings. Preserve hints according to their contract; never keep stale statistics or force a broadcast to hide the added cost.
5. Preserve exact producer ownership, exchange/subquery reuse and dynamic-pruning compatibility through AQE and non-AQE planning. If multiple consumers need different expressions, bound materialization and avoid widening every consumer without evidence of benefit.

DataFusion's join projection pushdown converts projections to column expressions before proceeding. Extending that native helper alone would still leave computation inside each Spark task; it is not a substitute for this producer-placement change.

Success criteria

• Operator/type-based tests show eligible expressions below the actual broadcast producer and absent from the repeated consumer position, with the original plan preserved for unsafe or unprofitable cases.
• Spark/native parity covers NULLs, Unicode/collations, decimals, time zones, ANSI errors, duplicates, empty inputs and supported expression composition. Include outer-join and UDF negative controls, Comet-disabled mode and streaming exclusions.
• Replanning is resolved and idempotent; AQE, reused exchanges/subqueries, pruning and multiple consumers retain compatible schemas and ownership. No extra fallback is silently accepted as proof of native execution.
• Near-threshold tests verify honest statistics, changed strategy when legitimately required, and bounded additional payload/memory. Evaluate low-reuse and wide-output controls where the rule should decline to apply.
• Instrument expression-input rows, producer executions, broadcast rows/bytes, decode work and peak memory. Repeated matched end-to-end runs demonstrate that eliminated consumer computation exceeds added producer/transport work, using identical input snapshots, resources and output validation.
• Report regressions and combined-patch interactions. No speedup is credited from a plan change, reduced expression CPU or a synthetic model alone.

This does not remove meaningful Inner-join duplicates, add arbitrary expression pushdown, or share native mutable join state. Producer-side existence-key normalization/deduplication needs a separate semantic proof; it must not be assumed equivalent to moving scalar expressions.