This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new 3d24724fe [AURON #2375] Fix Iceberg changelog scan field-id projection
(#2376)
3d24724fe is described below
commit 3d24724feefe542667649eb46f9007a1149eeb3c
Author: linfeng <[email protected]>
AuthorDate: Thu Jul 9 11:03:26 2026 +0800
[AURON #2375] Fix Iceberg changelog scan field-id projection (#2376)
# Which issue does this PR close?
Closes #2375
# Rationale for this change
The regular native Iceberg scan path already passes Iceberg field IDs to
the native reader, which makes top-level schema evolution such as column
rename and drop-then-add safe for Parquet files.
The newer insert-only Iceberg changelog scan path also reads the
underlying Parquet data files through the native reader, but it does not
pass the same field-id mapping into the native scan plan yet. As a
result, native Parquet schema matching falls back to column names on the
changelog path.
This can return wrong results after Iceberg schema evolution. For
example, after `RENAME COLUMN`, pre-rename files may read as null; after
`DROP` + `ADD` of the same name, the newly added column may read data
from the old dropped column.
# What changes are included in this PR?
- Extract field IDs from `SparkChangelogScan`'s expected Iceberg schema.
- Reuse the existing Iceberg rename/drop detection for changelog scans.
- Pass changelog field IDs into `IcebergScanPlan` instead of
`Map.empty`.
- Keep nested rename/drop unsupported and make ORC changelog scans fall
back after top-level rename/drop, consistent with the regular Iceberg
scan path.
- Add changelog scan integration tests for:
- renamed columns resolved by field-id;
- drop-then-add columns with the same name not reusing the dropped
field-id.
# Are there any user-facing changes?
Yes. Insert-only Iceberg changelog scans on renamed or drop-then-added
Parquet columns now return correct results under the native scan.
Unsupported cases continue to fall back to Spark. No API change.
# How was this patch tested?
Added cases to `AuronIcebergIntegrationSuite`.
---
.../spark/source/AuronIcebergSourceUtil.scala | 20 +++++
.../sql/auron/iceberg/IcebergScanSupport.scala | 90 +++++++++++++++-------
.../iceberg/AuronIcebergIntegrationSuite.scala | 59 ++++++++++++++
3 files changed, 142 insertions(+), 27 deletions(-)
diff --git
a/thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala
b/thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala
index c1d0a58d6..cebda6cfa 100644
---
a/thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala
+++
b/thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala
@@ -18,6 +18,8 @@ package org.apache.iceberg.spark.source
import scala.collection.JavaConverters._
+import org.apache.commons.lang3.reflect.FieldUtils
+import org.apache.iceberg.Table
import org.apache.iceberg.types.TypeUtil
object AuronIcebergSourceUtil {
@@ -37,8 +39,26 @@ object AuronIcebergSourceUtil {
expectedSchema.columns().asScala.map(field => field.name() ->
field.fieldId()).toMap
}
+ def expectedFieldIdsForChangelogScan(scan: AnyRef): Map[String, Int] = {
+ // SparkChangelogScan does not expose Iceberg expectedSchema/table
accessors.
+ // Keep the internal field-name assumptions localized here; callers
fallback if they change.
+ val expectedSchema =
+ FieldUtils.readField(scan, "expectedSchema",
true).asInstanceOf[org.apache.iceberg.Schema]
+ expectedSchema.columns().asScala.map(field => field.name() ->
field.fieldId()).toMap
+ }
+
def detectRenameOrDrop(scan: AnyRef): RenameOrDrop = {
val table = asBatchQueryScan(scan).table()
+ detectRenameOrDrop(table)
+ }
+
+ def detectRenameOrDropForChangelogScan(scan: AnyRef): RenameOrDrop = {
+ // SparkChangelogScan does not expose its Iceberg table.
+ val table = FieldUtils.readField(scan, "table", true).asInstanceOf[Table]
+ detectRenameOrDrop(table)
+ }
+
+ private def detectRenameOrDrop(table: Table): RenameOrDrop = {
val currentFields = collectFieldIdToName(table.schema())
table
diff --git
a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
index 3aa85b2de..192f1455b 100644
---
a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
+++
b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala
@@ -118,31 +118,16 @@ object IcebergScanSupport extends Logging {
}
val (fileSchema, partitionSchema) = schemas.get
- val fieldIdsByName =
- try {
- AuronIcebergSourceUtil.expectedFieldIds(scan.asInstanceOf[AnyRef])
- } catch {
- case NonFatal(t) =>
- logWarning(s"Failed to inspect Iceberg field ids for
$scanClassName.", t)
- return None
+ val (fieldIdsByName, renameOrDrop) =
+ inspectFieldIdSupport(
+ fileSchema,
+ scan.asInstanceOf[AnyRef],
+ AuronIcebergSourceUtil.expectedFieldIds,
+ AuronIcebergSourceUtil.detectRenameOrDrop) match {
+ case Some(fieldIdSupport) => fieldIdSupport
+ case None => return None
}
- val renameOrDrop =
- try {
- AuronIcebergSourceUtil.detectRenameOrDrop(scan.asInstanceOf[AnyRef])
- } catch {
- case NonFatal(t) =>
- logWarning(s"Failed to inspect Iceberg schema history for
$scanClassName.", t)
- return None
- }
- assert(!renameOrDrop.nested, "Nested Iceberg rename or drop is not
supported.")
-
- val missingFieldIds =
- fileSchema.fields.filterNot(field =>
fieldIdsByName.contains(field.name)).map(_.name)
- assert(
- missingFieldIds.isEmpty,
- s"Missing Iceberg field ids for columns: ${missingFieldIds.mkString(",
")}")
-
val partitions = inputPartitions(exec)
// Empty scan (e.g. empty table) should still build a plan to return no
rows.
if (partitions.isEmpty) {
@@ -185,7 +170,8 @@ object IcebergScanSupport extends Logging {
// ORC cannot match Iceberg columns by field-id yet, so any historical
top-level
// rename/drop may make older ORC files unsafe for native name/position
matching.
val supportedFormat =
- format == FileFormat.PARQUET || (format == FileFormat.ORC &&
!renameOrDrop.topLevel)
+ format == FileFormat.PARQUET ||
+ (format == FileFormat.ORC && !renameOrDrop.topLevel)
if (!supportedFormat) {
return None
}
@@ -211,6 +197,16 @@ object IcebergScanSupport extends Logging {
}
val (fileSchema, partitionSchema) = schemas.get
+ val (fieldIdsByName, renameOrDrop) =
+ inspectFieldIdSupport(
+ fileSchema,
+ scan.asInstanceOf[AnyRef],
+ AuronIcebergSourceUtil.expectedFieldIdsForChangelogScan,
+ AuronIcebergSourceUtil.detectRenameOrDropForChangelogScan) match {
+ case Some(fieldIdSupport) => fieldIdSupport
+ case None => return None
+ }
+
val partitions = inputPartitions(exec)
if (partitions.isEmpty) {
return Some(
@@ -221,7 +217,7 @@ object IcebergScanSupport extends Logging {
fileSchema,
partitionSchema,
Seq.empty,
- Map.empty))
+ fieldIdsByName))
}
val icebergPartitions = partitions.flatMap(icebergPartition)
@@ -256,7 +252,12 @@ object IcebergScanSupport extends Logging {
}
val format = formats.headOption.getOrElse(FileFormat.PARQUET)
- if (format != FileFormat.PARQUET && format != FileFormat.ORC) {
+ // ORC cannot match Iceberg columns by field-id yet, so any historical
top-level
+ // rename/drop may make older ORC files unsafe for native name/position
matching.
+ val supportedFormat =
+ format == FileFormat.PARQUET ||
+ (format == FileFormat.ORC && !renameOrDrop.topLevel)
+ if (!supportedFormat) {
return None
}
@@ -270,7 +271,42 @@ object IcebergScanSupport extends Logging {
fileSchema,
partitionSchema,
pruningPredicates,
- Map.empty))
+ fieldIdsByName))
+ }
+
+ private def inspectFieldIdSupport(
+ fileSchema: StructType,
+ scan: AnyRef,
+ expectedFieldIds: AnyRef => Map[String, Int],
+ detectRenameOrDrop: AnyRef => AuronIcebergSourceUtil.RenameOrDrop)
+ : Option[(Map[String, Int], AuronIcebergSourceUtil.RenameOrDrop)] = {
+ val scanClassName = scan.getClass.getName
+ val fieldIdsByName =
+ try {
+ expectedFieldIds(scan)
+ } catch {
+ case NonFatal(t) =>
+ logWarning(s"Failed to inspect Iceberg field ids for
$scanClassName.", t)
+ return None
+ }
+
+ val renameOrDrop =
+ try {
+ detectRenameOrDrop(scan)
+ } catch {
+ case NonFatal(t) =>
+ logWarning(s"Failed to inspect Iceberg schema history for
$scanClassName.", t)
+ return None
+ }
+ assert(!renameOrDrop.nested, "Nested Iceberg rename or drop is not
supported.")
+
+ val missingFieldIds =
+ fileSchema.fields.filterNot(field =>
fieldIdsByName.contains(field.name)).map(_.name)
+ assert(
+ missingFieldIds.isEmpty,
+ s"Missing Iceberg field ids for columns: ${missingFieldIds.mkString(",
")}")
+
+ Some((fieldIdsByName, renameOrDrop))
}
private def supportedSchemas(
diff --git
a/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
b/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
index 1142b6e03..e6f75e3c7 100644
---
a/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
+++
b/thirdparty/auron-iceberg/src/test/scala/org/apache/auron/iceberg/AuronIcebergIntegrationSuite.scala
@@ -475,6 +475,65 @@ class AuronIcebergIntegrationSuite
}
}
+ test("iceberg changelog scan reads renamed columns by field id") {
+ withTable("local.db.t_changelog_rename") {
+ withTempView("t_changelog_rename_changes") {
+ sql("""
+ |create table local.db.t_changelog_rename (id int, old_name
string)
+ |using iceberg
+ |tblproperties ('format-version' = '2')
+ |""".stripMargin)
+ sql("insert into local.db.t_changelog_rename values (0, 'initial')")
+ val startSnapshotId = currentSnapshotId("local.db.t_changelog_rename")
+ sql("insert into local.db.t_changelog_rename values (1, 'before')")
+ sql("alter table local.db.t_changelog_rename rename column old_name to
new_name")
+ sql("insert into local.db.t_changelog_rename values (2, 'after')")
+ val endSnapshotId = currentSnapshotId("local.db.t_changelog_rename")
+ createChangelogView(
+ "local.db.t_changelog_rename",
+ "t_changelog_rename_changes",
+ startSnapshotId,
+ endSnapshotId)
+
+ checkSparkAnswerAndOperator("""
+ |select id, new_name, _change_type, _change_ordinal,
_commit_snapshot_id
+ |from t_changelog_rename_changes
+ |order by id
+ |""".stripMargin)
+ }
+ }
+ }
+
+ test("iceberg changelog scan does not reuse dropped field id for an added
column") {
+ withTable("local.db.t_changelog_drop_add") {
+ withTempView("t_changelog_drop_add_changes") {
+ sql("""
+ |create table local.db.t_changelog_drop_add (id int, value
string)
+ |using iceberg
+ |tblproperties ('format-version' = '2')
+ |""".stripMargin)
+ sql("insert into local.db.t_changelog_drop_add values (0, 'initial')")
+ val startSnapshotId =
currentSnapshotId("local.db.t_changelog_drop_add")
+ sql("insert into local.db.t_changelog_drop_add values (1, 'old')")
+ sql("alter table local.db.t_changelog_drop_add drop column value")
+ sql("alter table local.db.t_changelog_drop_add add column value
string")
+ sql("insert into local.db.t_changelog_drop_add values (2, 'new')")
+ val endSnapshotId = currentSnapshotId("local.db.t_changelog_drop_add")
+ createChangelogView(
+ "local.db.t_changelog_drop_add",
+ "t_changelog_drop_add_changes",
+ startSnapshotId,
+ endSnapshotId)
+
+ checkSparkAnswerAndOperator("""
+ |select id, value, _change_type, _change_ordinal,
_commit_snapshot_id
+ |from t_changelog_drop_add_changes
+ |order by id
+ |""".stripMargin)
+ }
+ }
+ }
+
test("iceberg changelog scan falls back when delete changes exist") {
withTable("local.db.t_changelog_delete") {
withTempView("t_changelog_delete_changes") {
