[GLUTEN-12273][CORE] Support staged scans for Iceberg data-file rewrites - #12962
[GLUTEN-12273][CORE] Support staged scans for Iceberg data-file rewrites#12962infvg wants to merge 1 commit into
Conversation
|
Run Gluten Clickhouse CI on x86 |
infvg
changed the title
6b123d9 to
c00a321
Compare
|
Run Gluten Clickhouse CI on x86 |
c00a321 to
d7e40b6
Compare
|
Run Gluten Clickhouse CI on x86 |
|
|
||
| def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = { | ||
| classOf[SparkBatchQueryScan] | ||
| def supportsScan(sparkScan: Scan): Boolean = sparkScan match { |
There was a problem hiding this comment.
maybe a clear func name: isSupportedScan
| case _: SparkBatchQueryScan => true | ||
| case scan: SparkStagedScan => | ||
| val tasks = getScanTasks(scan) | ||
| tasks.nonEmpty && |
There was a problem hiding this comment.
do we need to exclude empty scan here?
| case scan: SparkStagedScan => | ||
| val tasks = getScanTasks(scan) | ||
| tasks.nonEmpty && | ||
| (tasks.forall(_.isFileScanTask) || tasks.forall(_.isInstanceOf[CombinedScanTask])) |
There was a problem hiding this comment.
here it also excluded some of scan tasks, is this necessary?
| def supportsScan(sparkScan: Scan): Boolean = sparkScan match { | ||
| case _: SparkBatchQueryScan => true | ||
| case scan: SparkStagedScan => | ||
| val tasks = getScanTasks(scan) |
There was a problem hiding this comment.
this will bring big perf overhead here, can we avoid this?
There was a problem hiding this comment.
changed to just check the scan object, and materializes them later when needed
| import org.apache.iceberg.spark.SparkSchemaUtil | ||
|
|
||
| import java.lang.{Class, Long => JLong} | ||
| import java.lang.{Long => JLong} |
There was a problem hiding this comment.
this one is required for scala 2
| // which will not appear in readFields, they also cannot be filtered. | ||
| val tableFields = spec.schema().columns().asScala.map(_.name()).toSet | ||
| val voidTransformFields = getTable(sparkScan) | ||
| .spec() |
There was a problem hiding this comment.
we already use foreach getScanTasks(sparkScan) to get task.spec, can we skip the .spec() here?
There was a problem hiding this comment.
task.spec() only gives the partition spec used to write the file, we still need table.spec() for the fields dropped from the current spec using void transforms
There was a problem hiding this comment.
🟡 Changes recommended
The new staged-scan test relies on brittle class-name string matching, and the staged scan support check currently materializes full task lists (avoidable overhead) in the planning path.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This PR extends Gluten’s Iceberg V2 scan offload to support Iceberg’s SparkStagedScan (used by CALL ... rewrite_data_files), allowing rewrite reads to use the native Iceberg scan transformer path instead of falling back to vanilla Spark execution.
Changes:
- Add scan-support detection for
SparkStagedScanand extract staged scan tasks fromtaskGroups(). - Reuse existing Iceberg table/schema/partition/file-format extraction logic for both batch and staged scans.
- Add a regression test asserting
rewrite_data_filesusesIcebergScanTransformerwith a staged scan.
File summaries
| File | Description |
|---|---|
| gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala | Adds a regression test covering rewrite_data_files and staged-scan offload behavior. |
| gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/GlutenIcebergSourceUtil.scala | Introduces unified scan support + staged-task extraction to drive existing schema/partition/format logic. |
| gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala | Switches scan support gating to the new unified supportsScan logic. |
Review details
- Files reviewed: 3/3 changed files
- Comments generated: 2
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| def supportsScan(sparkScan: Scan): Boolean = sparkScan match { | ||
| case _: SparkBatchQueryScan => true | ||
| case scan: SparkStagedScan => | ||
| val tasks = getScanTasks(scan) | ||
| tasks.nonEmpty && | ||
| (tasks.forall(_.isFileScanTask) || tasks.forall(_.isInstanceOf[CombinedScanTask])) | ||
| case _ => false | ||
| } |
| qe.executedPlan.exists { | ||
| case scan: IcebergScanTransformer => | ||
| scan.scan.getClass.getSimpleName == "SparkStagedScan" | ||
| case _ => false | ||
| } |
d7e40b6 to
f4706f2
Compare
|
Run Gluten Clickhouse CI on x86 |
What changed
SparkStagedScan.Why
Iceberg
RewriteDataFilesuses a staged scan to read selected data files.Gluten previously supported only
SparkBatchQueryScan. Gluten rejected the staged scan, and Spark used the standard Spark execution path.This change lets Gluten use the native Iceberg scan path for staged scans.
Resolves GLUTEN-12273