HyukjinKwon opened a new pull request, #59943: URL: https://github.com/apache/spark/pull/59943
### What changes were proposed in this pull request? The micro-batch and continuous executions now push the top-level columns that a streaming query needs into the scan builders of Data Source V2 streaming sources that support column pruning (`SupportsPushDownRequiredColumns`), instead of all the columns of their output, as `V2ScanRelationPushDown` does for batch scans. - `StreamingV2ColumnPruning` (new) computes the attributes of the sources that the query needs from its analyzed plan: those in its output, those that its operators reference, and every column of the children of the operators that need all of them (distinct, unions and other set operations, CTE definitions, and nodes that map the columns of a child by position). The scan of a streaming source is built once when the query starts, so the pruning uses that plan; the optimizer prunes each micro-batch further as before, so stateful operators keep getting the same columns. - `MicroBatchExecution` and `ContinuousExecution` call `pruneColumns` on the scan builder with those columns, and use the `readSchema` that the scan returns, so a source that reads more columns than required still works. - `ColumnarScanBuilder` (the adapter of native data sources) keeps the required columns and lets the batch scan prune them through its planner, while its streaming scans keep reading the full schema for now. - The behavior is behind the internal configuration `spark.sql.streaming.v2ColumnPruning.enabled` (true by default). This targets 4.4.0 (master and branch-4.x). ### Why are the changes needed? A streaming scan of a Data Source V2 source reads every column of the source, even when the query needs only some of them, while a batch scan of the same source reads only those. Native data sources (SPARK-60723, #59931) will benefit once their stream readers get the columns to read. ### Does this PR introduce _any_ user-facing change? Yes. Streaming queries over Data Source V2 sources that support column pruning read only the columns that they need. The results do not change. The previous behavior can be restored with `spark.sql.streaming.v2ColumnPruning.enabled=false`. ### How was this patch tested? New `StreamingV2ColumnPruningSuite`, with an in-memory table that records the columns that its streaming scans read: - a projection, a streaming aggregation and a streaming deduplication read only the columns that they need, also after a restart from the checkpoint; - metadata columns that the query needs are read; - the children of a union read all their columns; - a scan that reads more columns than required still returns the right result; - all the columns are read when the configuration is disabled. The existing `DataSourceV2Suite`, `MicroBatchExecutionSuite`, `ColumnarDataSourceSuite` and `NativeDataSourceSuite` pass. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) This pull request and its description were written by Isaac. -- 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]
