[AURON #2238] Add support for auron.never.convert.reason in Iceberg scan scenarios #2237
Open
guixiaowen wants to merge 1 commit intoapache:masterfrom
Open
[AURON #2238] Add support for auron.never.convert.reason in Iceberg scan scenarios #2237guixiaowen wants to merge 1 commit intoapache:masterfrom
guixiaowen wants to merge 1 commit intoapache:masterfrom
Conversation
guixiaowen
changed the title
…berg scan scenarios
guixiaowen
force-pushed
the
foriceberg_addnever_convert_reason
branch
from
844591e to
7f35ead
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
This PR aims to ensure Iceberg fallback scenarios populate the Spark plan tag auron.never.convert.reason, making fallback reasons visible (e.g., in Spark UI) similarly to other scan conversions.
Changes:
- Updated
AuronConvertProviderto makeisEnableddepend on the currentSparkPlan, and adjusted Iceberg/Hudi/Paimon providers accordingly. - Added exception-based tagging in
AuronConverters.convertSparkPlanto setauron.never.convert.reasonwhen conversion is rejected via assertions/exceptions. - Added Iceberg integration tests asserting
auron.never.convert.reasonis present for disabled Iceberg scan and unsupported metadata-column scenarios.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| thirdparty/auron-paimon/src/main/scala/org/apache/spark/sql/hive/auron/paimon/PaimonConvertProvider.scala | Updates provider enablement to be plan-type aware. |
| thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala | Adds integration tests validating auron.never.convert.reason for Iceberg fallback cases. |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala | Converts multiple “return None” fallbacks into assert-based failures with messages intended for tagging. |
| thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergConvertProvider.scala | Updates isEnabled signature and uses assertions to drive fallback reason tagging. |
| thirdparty/auron-iceberg/pom.xml | Adds scala-library dependency (provided scope). |
| thirdparty/auron-hudi/src/main/scala/org/apache/spark/sql/auron/hudi/HudiConvertProvider.scala | Updates provider enablement to be plan-type aware. |
| spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConvertProvider.scala | Changes isEnabled API to accept SparkPlan. |
| spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala | Uses isEnabled(exec) and adds try/catch-based never-convert reason tagging in generic conversion path. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+31
to
+40
| override def isEnabled(exec: SparkPlan): Boolean = { | ||
| exec match { | ||
| case _: BatchScanExec => | ||
| val enabled = SparkAuronConfiguration.ENABLE_ICEBERG_SCAN.get() | ||
| assert(enabled, "Conversion disabled: auron.enable.iceberg.scan=false.") | ||
| assert( | ||
| sparkCompatible, | ||
| s"Supported Spark versions: 3.4 to 4.0 (Iceberg ${icebergVersionOrUnknown}).") | ||
| return false | ||
| enabled | ||
| case _ => false |
Comment on lines
46
to
56
| def plan(exec: BatchScanExec): Option[IcebergScanPlan] = { | ||
| val scan = exec.scan | ||
| val scanClassName = scan.getClass.getName | ||
| // Only handle Iceberg scans; other sources must stay on Spark's path. | ||
| if (!scanClassName.startsWith("org.apache.iceberg.spark.source.")) { | ||
| return None | ||
| } | ||
| assert(scanClassName.startsWith("org.apache.iceberg.spark.source."), "Not iceberg scans.") | ||
|
|
||
| // Changelog scan carries row-level changes; not supported by native COW-only path. | ||
| if (scanClassName == "org.apache.iceberg.spark.source.SparkChangelogScan") { | ||
| return None | ||
| } | ||
| assert( | ||
| !(scanClassName == "org.apache.iceberg.spark.source.SparkChangelogScan"), | ||
| "Not iceberg cow table.") | ||
|
|
Comment on lines
+61
to
+63
| assert( | ||
| !(unsupportedMetadataColumns.nonEmpty), | ||
| "Has per-row materialization (for example _pos).") |
Comment on lines
+242
to
+260
| try { | ||
| extConvertProviders.find(h => h.isEnabled(exec) && h.isSupported(exec)) match { | ||
| case Some(provider) => tryConvert(exec, provider.convert) | ||
| case None => | ||
| Shims.get.convertMoreSparkPlan(exec) match { | ||
| case Some(exec) => | ||
| exec.setTagValue(convertibleTag, true) | ||
| exec.setTagValue(convertStrategyTag, AlwaysConvert) | ||
| exec | ||
| } else { | ||
| addNeverConvertReasonTag(exec) | ||
| } | ||
| } | ||
| case None => | ||
| if (Shims.get.isNative(exec)) { // for QueryStageInput and CustomShuffleReader | ||
| exec.setTagValue(convertibleTag, true) | ||
| exec.setTagValue(convertStrategyTag, AlwaysConvert) | ||
| exec | ||
| } else { | ||
| addNeverConvertReasonTag(exec) | ||
| } | ||
| } | ||
| } |
Comment on lines
+263
to
+268
| exec.setTagValue(convertToNonNativeTag, true) | ||
| exec.setTagValue(convertibleTag, false) | ||
| exec.setTagValue(convertStrategyTag, NeverConvert) | ||
| exec.setTagValue( | ||
| neverConvertReasonTag, | ||
| s"${e.getMessage.replaceFirst("^assertion failed: ?", "")}") |
Comment on lines
+61
to
+63
| assert( | ||
| !(unsupportedMetadataColumns.nonEmpty), | ||
| "Has per-row materialization (for example _pos).") |
Comment on lines
+43
to
+46
| val neverConvertReasonTag: TreeNodeTag[String] = TreeNodeTag("auron.never.convert.reason") | ||
| assert(collectFirst(df.queryExecution.executedPlan) { case batchScanExec: BatchScanExec => | ||
| batchScanExec.getTagValue(neverConvertReasonTag) | ||
| }.get.get.equals("Conversion disabled: auron.enable.iceberg.scan=false.")) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #2238
Rationale for this change
What changes are included in this PR?
Are there any user-facing changes?
How was this patch tested?