viirya opened a new pull request, #56928:
URL: https://github.com/apache/spark/pull/56928

   ### What changes were proposed in this pull request?
   
   This adds a conservative optimizer rule `MarkSingleTaskExecution` that marks 
small single-partition scans, optionally with a shuffle-inducing operator on 
top (sort, aggregate, distinct, window, limit/offset, expand) or an in-memory 
`LocalRelation`, as candidates for single-task execution. Such a scan reports a 
`SinglePartition` output partitioning, allowing `EnsureRequirements` to elide 
the shuffle that would otherwise be inserted before the operator on top.
   
   Details:
   - The rule runs as the last optimizer batch (so it sees the final plan 
shape) and marks eligible `LogicalRelation`/`LocalRelation` nodes with a 
`TreeNodeTag`.
   - `FileSourceStrategy`/`SparkStrategies` propagate the mark to 
`FileSourceScanExec`/`LocalTableScanExec`.
   - `FileSourceScanExec` additionally gates on file count and size thresholds 
using the generic `ScanFileListing`, reports `SinglePartition`, and coalesces 
its input RDD to a single partition as a correctness backstop when the estimate 
does not match the runtime partition count.
   - `LocalTableScanExec` reads its data in a single partition and reports 
`SinglePartition`.
   - `ExpandExec` forwards `SinglePartition` from its child, since Expand only 
replicates rows within a partition and never moves rows across partitions.
   
   The feature is controlled by new internal configs under 
`spark.sql.optimizer.singleTaskExecution.*` and is disabled by default. Join is 
intentionally left out for now and can be added as a follow-up; union is 
already covered by the existing `spark.sql.unionOutputPartitioning`.
   
   This is part of the SPIP umbrella 
[SPARK-56978](https://issues.apache.org/jira/browse/SPARK-56978) (Faster 
queries in local laptop mode), covering the shuffle-free local execution for 
small queries category.
   
   ### Why are the changes needed?
   
   For small, low-latency queries the fixed cost of a shuffle (scheduling, 
serialization, network) dominates the total runtime. When the input is already 
a single small partition, the shuffle inserted before a sort/aggregate/window 
is unnecessary and can be removed to reduce latency, without affecting 
correctness.
   
   ### Does this PR introduce _any_ user-facing change?
   
   No. The optimization is behind internal configs 
(`spark.sql.optimizer.singleTaskExecution.*`) and is disabled by default.
   
   ### How was this patch tested?
   
   New `MarkSingleTaskExecutionSuite` (14 tests) covering:
   - the marking decision for the supported plan shapes;
   - `SinglePartition` output with no shuffle in the final physical plan;
   - empty-scan correctness (a global aggregation over an empty scan still 
returns a single row);
   - disabled-flag negatives (master flag and per-operator sub-flags);
   - ineligibility of unsupported shapes (join) and subquery expressions;
   - the leaf-node parallelism override disabling the local-relation case.
   
   `SQLConfSuite` passes as a config-wiring regression check.
   
   ### Was this patch authored or co-authored using generative AI tooling?
   
   Yes, using Claude Code.
   


-- 
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]

Reply via email to