You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by "lyne7-sc (via GitHub)" <gi...@apache.org> on 2026/06/10 07:39:08 UTC
[PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
lyne7-sc opened a new pull request, #2322:
URL: https://github.com/apache/auron/pull/2322
# Which issue does this PR close?
Closes #2321
# Rationale for this change
The native Iceberg scan matches data-file columns by name, but Iceberg tracks them by field-id. After a column rename, old files read as all-NULL; after a drop-then-add of the same name, the new column reads the old column's data.
# What changes are included in this PR?
Resolve columns by Iceberg field-id instead of by name:
- proto: add `field_id` to `Field`.
- JVM (`AuronIcebergSourceUtil`, `IcebergScanSupport`, `NativeConverters`): extract top-level `name → field-id` from the scan's `expectedSchema()` and serialize it into the plan.
- native (`auron-planner`, `scan/mod.rs`): stamp the id into Arrow field metadata (`PARQUET:field_id`); `fields_match` matches by id when present, else falls back to case-insensitive name matching (non-Iceberg scans unchanged).
Nested-struct evolution and ORC rename/drop fall back to Spark, additive evolution stays native.
# Are there any user-facing changes?
Yes. Iceberg queries on renamed or drop-then-added columns now return correct results under the native scan. Unsupported cases fall back to Spark. No API change.
# How was this patch tested?
Added cases to `AuronIcebergIntegrationSuite`
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3420768114
##########
native-engine/datafusion-ext-plans/src/scan/mod.rs:
##########
@@ -89,6 +89,16 @@ impl SchemaAdapter for AuronSchemaAdapter {
}
}
+fn fields_match(table_field: &Field, file_field: &Field) -> bool {
+ match table_field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
+ Some(table_field_id) => file_field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .is_some_and(|file_field_id| file_field_id == table_field_id),
+ None => table_field.name().eq_ignore_ascii_case(file_field.name()),
Review Comment:
Updated `fields_match` to use a nested match. When `file_field` lacks a field id, it now falls back to case-insensitive name matching.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -75,6 +76,27 @@ object IcebergScanSupport extends Logging {
partitionSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType)),
"Has unsupported schema type.")
Review Comment:
Split this into two independent inspection steps.
`expectedFieldIds` failure returns `None` because field-id mapping is required for safe native planning.
`detectRenameOrDrop` failure returns `None` because rename/drop safety cannot be determined reliably.
This avoids reporting a misleading nested rename/drop fallback reason when the actual issue is schema-history inspection failure.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -75,6 +76,27 @@ object IcebergScanSupport extends Logging {
partitionSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType)),
"Has unsupported schema type.")
Review Comment:
Added `TreeNodeTag` caching on the `BatchScanExec`. `plan(exec)` now reuses the cached result on the second call
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala:
##########
@@ -16,13 +16,77 @@
*/
package org.apache.iceberg.spark.source
+import scala.collection.JavaConverters._
+
object AuronIcebergSourceUtil {
+ final case class RenameOrDrop(topLevel: Boolean, nested: Boolean)
+
def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = {
classOf[SparkBatchQueryScan]
}
def getClassOfSparkInputPartition(): Class[SparkInputPartition] = {
classOf[SparkInputPartition]
}
+
+ def expectedFieldIds(scan: AnyRef): Map[String, Int] = {
+ val expectedSchema = asBatchQueryScan(scan).expectedSchema()
+ expectedSchema.columns().asScala.map(field => field.name() -> field.fieldId()).toMap
+ }
+
+ def detectRenameOrDrop(scan: AnyRef): RenameOrDrop = {
+ val table = asBatchQueryScan(scan).table()
+ val currentFields = collectFieldIdToName(table.schema())
Review Comment:
Addressed both points:
1. Skipped the current schema.
2. Replaced the local recursive collector with `TypeUtil.indexById(schema.asStruct())`.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -75,6 +76,27 @@ object IcebergScanSupport extends Logging {
partitionSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType)),
"Has unsupported schema type.")
+ val inspected: Option[(AuronIcebergSourceUtil.RenameOrDrop, Map[String, Int])] =
+ try {
+ Some(
+ (
+ AuronIcebergSourceUtil.detectRenameOrDrop(scan.asInstanceOf[AnyRef]),
+ AuronIcebergSourceUtil.expectedFieldIds(scan.asInstanceOf[AnyRef])))
+ } catch {
+ case NonFatal(t) =>
+ logWarning(s"Failed to inspect Iceberg field ids for $scanClassName.", t)
+ None
+ }
+ if (inspected.isEmpty) {
+ return None
+ }
+ val (renameOrDrop, fieldIdsByName) = inspected.get
+ assert(!renameOrDrop.nested, "Nested Iceberg rename or drop is not supported.")
+
+ assert(
+ fileSchema.fields.forall(field => fieldIdsByName.contains(field.name)),
+ "Failed to find field ids for all Iceberg data columns.")
+
val partitions = inputPartitions(exec)
Review Comment:
Updated the assertion to be more explicit.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3464121161
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
+ !(format == FileFormat.ORC && renameOrDrop.topLevel),
Review Comment:
Agreed — the conservative fallback is the right call for this PR, and the new comment captures the why. The real root cause is the field-id-in-Arrow-metadata gap in the native ORC reader; until that's closed, falling back is the only safe option for ORC files written before a top-level rename/drop. The two ORC fallback tests (`t_orc_rename`, `t_orc_drop_add`) pin the behavior down nicely.
The longer-term direction you sketched — carrying Iceberg field IDs into the native ORC reader so it can match by field id like Parquet — sounds like the right end state, with a file-level fallback as the lighter option if that proves too invasive. Maybe worth a tracking issue so it isn't lost.
Nothing further from me — the change looks good. I'll leave the final call to the maintainers.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3442652678
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
Review Comment:
Makes sense, I added a small ORC drop-then-add test for this.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "merrily01 (via GitHub)" <gi...@apache.org>.
merrily01 commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3490811214
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala:
##########
@@ -16,13 +16,72 @@
*/
package org.apache.iceberg.spark.source
+import scala.collection.JavaConverters._
+
+import org.apache.iceberg.types.TypeUtil
+
object AuronIcebergSourceUtil {
+ final case class RenameOrDrop(topLevel: Boolean, nested: Boolean)
+
def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = {
classOf[SparkBatchQueryScan]
}
def getClassOfSparkInputPartition(): Class[SparkInputPartition] = {
classOf[SparkInputPartition]
}
+
+ def expectedFieldIds(scan: AnyRef): Map[String, Int] = {
+ val expectedSchema = asBatchQueryScan(scan).expectedSchema()
+ expectedSchema.columns().asScala.map(field => field.name() -> field.fieldId()).toMap
+ }
+
+ def detectRenameOrDrop(scan: AnyRef): RenameOrDrop = {
+ val table = asBatchQueryScan(scan).table()
+ val currentFields = collectFieldIdToName(table.schema())
+
+ table
+ .schemas()
+ .asScala
+ .filterNot(_._1 == table.schema().schemaId())
+ .values
+ .foldLeft(RenameOrDrop(topLevel = false, nested = false)) { (result, schema) =>
+ collectFieldIdToName(schema).foldLeft(result) {
+ case (currentResult, (fieldId, historicalField)) =>
+ currentFields.get(fieldId) match {
+ case Some(currentField) if currentField.name != historicalField.name =>
+ if (historicalField.topLevel || currentField.topLevel) {
+ currentResult.copy(topLevel = true)
+ } else {
+ currentResult.copy(nested = true)
+ }
+ case None =>
+ if (historicalField.topLevel) {
+ currentResult.copy(topLevel = true)
+ } else {
+ currentResult.copy(nested = true)
+ }
+ case _ =>
+ currentResult
+ }
+ }
+ }
+ }
+
+ final private case class FieldIdentity(name: String, topLevel: Boolean)
+
+ private def collectFieldIdToName(schema: org.apache.iceberg.Schema): Map[Int, FieldIdentity] = {
+ val topLevelFieldIds = schema.columns().asScala.map(_.fieldId()).toSet
+ TypeUtil
+ .indexById(schema.asStruct())
+ .asScala
+ .map { case (fieldId, field) =>
+ fieldId.toInt -> FieldIdentity(field.name(), topLevelFieldIds.contains(fieldId.toInt))
+ }
+ .toMap
+ }
+
+ private def asBatchQueryScan(scan: AnyRef): SparkBatchQueryScan =
+ scan.asInstanceOf[SparkBatchQueryScan]
Review Comment:
This uses `asInstanceOf[SparkBatchQueryScan]` for an unchecked cast. If `scan` isn't actually a `SparkBatchQueryScan` at runtime (e.g. a future scan subclass, or a runtime filtering wrapper around it), wouldn't this throw `ClassCastException`? The outer `try/catch NonFatal` does contain it, but the exception ends up being swallowed as a warning and silently falls back to the Spark scan.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on PR #2322:
URL: https://github.com/apache/auron/pull/2322#issuecomment-4861605570
Thanks everyone for the reviews!
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on PR #2322:
URL: https://github.com/apache/auron/pull/2322#issuecomment-4851114147
@lyne7-sc Thanks for the contribution!
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3442652678
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
Review Comment:
Makes sense, I added a ORC drop-then-add test for this.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3497028680
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala:
##########
@@ -16,13 +16,72 @@
*/
package org.apache.iceberg.spark.source
+import scala.collection.JavaConverters._
+
+import org.apache.iceberg.types.TypeUtil
+
object AuronIcebergSourceUtil {
+ final case class RenameOrDrop(topLevel: Boolean, nested: Boolean)
+
def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = {
classOf[SparkBatchQueryScan]
}
def getClassOfSparkInputPartition(): Class[SparkInputPartition] = {
classOf[SparkInputPartition]
}
+
+ def expectedFieldIds(scan: AnyRef): Map[String, Int] = {
+ val expectedSchema = asBatchQueryScan(scan).expectedSchema()
+ expectedSchema.columns().asScala.map(field => field.name() -> field.fieldId()).toMap
+ }
+
+ def detectRenameOrDrop(scan: AnyRef): RenameOrDrop = {
+ val table = asBatchQueryScan(scan).table()
+ val currentFields = collectFieldIdToName(table.schema())
+
+ table
+ .schemas()
+ .asScala
+ .filterNot(_._1 == table.schema().schemaId())
+ .values
+ .foldLeft(RenameOrDrop(topLevel = false, nested = false)) { (result, schema) =>
+ collectFieldIdToName(schema).foldLeft(result) {
+ case (currentResult, (fieldId, historicalField)) =>
+ currentFields.get(fieldId) match {
+ case Some(currentField) if currentField.name != historicalField.name =>
+ if (historicalField.topLevel || currentField.topLevel) {
+ currentResult.copy(topLevel = true)
+ } else {
+ currentResult.copy(nested = true)
+ }
+ case None =>
+ if (historicalField.topLevel) {
+ currentResult.copy(topLevel = true)
+ } else {
+ currentResult.copy(nested = true)
+ }
+ case _ =>
+ currentResult
+ }
+ }
+ }
+ }
+
+ final private case class FieldIdentity(name: String, topLevel: Boolean)
+
+ private def collectFieldIdToName(schema: org.apache.iceberg.Schema): Map[Int, FieldIdentity] = {
+ val topLevelFieldIds = schema.columns().asScala.map(_.fieldId()).toSet
+ TypeUtil
+ .indexById(schema.asStruct())
+ .asScala
+ .map { case (fieldId, field) =>
+ fieldId.toInt -> FieldIdentity(field.name(), topLevelFieldIds.contains(fieldId.toInt))
+ }
+ .toMap
+ }
+
+ private def asBatchQueryScan(scan: AnyRef): SparkBatchQueryScan =
+ scan.asInstanceOf[SparkBatchQueryScan]
Review Comment:
I think the unchecked cast should be safe here, since the caller already guards it.
In the current call path, `asBatchQueryScan` is only reached from `IcebergScanSupport.planUncached`, after this guard:
https://github.com/apache/auron/blob/236d8e965ca2cbba11e43f983110858c2d1ae555/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala#L89-L91
This guard ensures that:
- A non-`SparkBatchQueryScan` scan(e.g. a future scan implementation or runtime filtering wrapper), returns `None` here, so it will not reach the `expectedFieldIds` / `detectRenameOrDrop`
logic and the query stays on Spark's scan path.
- A subclass of `SparkBatchQueryScan` still passes the `isInstance` check and the cast remains valid.
So by the time execution reaches `asBatchQueryScan`, `scan` should already be a `SparkBatchQueryScan` or one of its subclasses. The changelog scan path is handled separately and does not call these helper methods. Let me know if I'm missing something though.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "merrily01 (via GitHub)" <gi...@apache.org>.
merrily01 commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3503753023
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala:
##########
@@ -16,13 +16,72 @@
*/
package org.apache.iceberg.spark.source
+import scala.collection.JavaConverters._
+
+import org.apache.iceberg.types.TypeUtil
+
object AuronIcebergSourceUtil {
+ final case class RenameOrDrop(topLevel: Boolean, nested: Boolean)
+
def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = {
classOf[SparkBatchQueryScan]
}
def getClassOfSparkInputPartition(): Class[SparkInputPartition] = {
classOf[SparkInputPartition]
}
+
+ def expectedFieldIds(scan: AnyRef): Map[String, Int] = {
+ val expectedSchema = asBatchQueryScan(scan).expectedSchema()
+ expectedSchema.columns().asScala.map(field => field.name() -> field.fieldId()).toMap
+ }
+
+ def detectRenameOrDrop(scan: AnyRef): RenameOrDrop = {
+ val table = asBatchQueryScan(scan).table()
+ val currentFields = collectFieldIdToName(table.schema())
+
+ table
+ .schemas()
+ .asScala
+ .filterNot(_._1 == table.schema().schemaId())
+ .values
+ .foldLeft(RenameOrDrop(topLevel = false, nested = false)) { (result, schema) =>
+ collectFieldIdToName(schema).foldLeft(result) {
+ case (currentResult, (fieldId, historicalField)) =>
+ currentFields.get(fieldId) match {
+ case Some(currentField) if currentField.name != historicalField.name =>
+ if (historicalField.topLevel || currentField.topLevel) {
+ currentResult.copy(topLevel = true)
+ } else {
+ currentResult.copy(nested = true)
+ }
+ case None =>
+ if (historicalField.topLevel) {
+ currentResult.copy(topLevel = true)
+ } else {
+ currentResult.copy(nested = true)
+ }
+ case _ =>
+ currentResult
+ }
+ }
+ }
+ }
+
+ final private case class FieldIdentity(name: String, topLevel: Boolean)
+
+ private def collectFieldIdToName(schema: org.apache.iceberg.Schema): Map[Int, FieldIdentity] = {
+ val topLevelFieldIds = schema.columns().asScala.map(_.fieldId()).toSet
+ TypeUtil
+ .indexById(schema.asStruct())
+ .asScala
+ .map { case (fieldId, field) =>
+ fieldId.toInt -> FieldIdentity(field.name(), topLevelFieldIds.contains(fieldId.toInt))
+ }
+ .toMap
+ }
+
+ private def asBatchQueryScan(scan: AnyRef): SparkBatchQueryScan =
+ scan.asInstanceOf[SparkBatchQueryScan]
Review Comment:
You're right, I missed the guard details. No further concerns from my side.
Approved, Thank you for the contribute.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3452671012
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
+ !(format == FileFormat.ORC && renameOrDrop.topLevel),
Review Comment:
Yes, this ORC guard is conservative. The native ORC path still maps columns by `position/name`, so if the table has a top-level `rename/drop` in its history, older ORC files may be unsafe to read natively.
Longer term, I think the right fix is to make the native ORC reader preserve Iceberg field IDs and then use the same field-id based matching as Parquet. From the current orc-rust code, it looks like ORC type attributes are not carried into Arrow Field metadata yet. If so, we’d need to extend the ORC reader/schema conversion path first before native ORC can do field-id based matching.
Maybe we could first try the ORC reader-side fix. If that turns out to be too invasive, a more precise file-level fallback would also be a reasonable improvement. For this PR, I’d prefer to keep the conservative fallback for correctness and add a comment here. WDYT?
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "lyne7-sc (via GitHub)" <gi...@apache.org>.
lyne7-sc commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3452671012
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
+ !(format == FileFormat.ORC && renameOrDrop.topLevel),
Review Comment:
Yes, this ORC guard is conservative. The native ORC path still maps columns by position/name, so if the table has a top-level rename/drop in its history, older ORC files may be unsafe to read natively.
Longer term, I think the right fix is to make the native ORC reader preserve Iceberg field IDs and then use the same field-id based matching as Parquet. From the current orc-rust code, it looks like ORC type attributes are not carried into Arrow Field metadata yet. If so, we’d need to extend the ORC reader/schema conversion path first before native ORC can do field-id based matching.
Maybe we could first try the ORC reader-side fix. If that turns out to be too invasive, a more precise file-level fallback would also be a reasonable improvement. For this PR, I’d prefer to keep the conservative fallback for correctness and add a short comment near the guard. WDYT?
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "weiqingy (via GitHub)" <gi...@apache.org>.
weiqingy commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3450144903
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
+ !(format == FileFormat.ORC && renameOrDrop.topLevel),
Review Comment:
`detectRenameOrDrop` walks the table's entire schema history (`table.schemas()`), so `renameOrDrop.topLevel` stays `true` for any table that has *ever* had a top-level rename or drop — and this assert then sends every ORC scan on that table back to Spark, even when all the data files being read were written after the rename and would resolve fine by name. For Parquet it's harmless (field-id matching still runs); for ORC a table that did a single early rename seems to lose the native scan permanently. Is that the intended scope, or worth narrowing to the files that actually predate the rename — or at least noting the permanence in a comment so it isn't surprising later?
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "merrily01 (via GitHub)" <gi...@apache.org>.
merrily01 commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3435448730
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -110,6 +150,9 @@ object IcebergScanSupport extends Logging {
assert(
!(format != FileFormat.PARQUET && format != FileFormat.ORC),
"Only support parquet or orc.")
+ assert(
Review Comment:
We already test ORC rename fallback, but will drop-then-add also sets topLevel=true ? Would it help to add a small ORC drop-then-add test so this fallback path is explicitly covered?
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb merged PR #2322:
URL: https://github.com/apache/auron/pull/2322
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3413392428
##########
native-engine/datafusion-ext-plans/src/scan/mod.rs:
##########
@@ -89,6 +89,16 @@ impl SchemaAdapter for AuronSchemaAdapter {
}
}
+fn fields_match(table_field: &Field, file_field: &Field) -> bool {
+ match table_field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
+ Some(table_field_id) => file_field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .is_some_and(|file_field_id| file_field_id == table_field_id),
+ None => table_field.name().eq_ignore_ascii_case(file_field.name()),
+ }
+}
Review Comment:
`fields_match` only falls back to case-insensitive name matching when the *table* field is missing a field-id. If the table has a field-id but the Parquet file schema does not carry `PARQUET:field_id` metadata, this will refuse to match even when names match, which contradicts the intended "match by id when present, else fall back to name" behavior described in the PR and can cause all-NULL reads.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #2321] Support Iceberg column rename and drop-then-add in the native scan [auron]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #2322:
URL: https://github.com/apache/auron/pull/2322#discussion_r3413461018
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -75,6 +76,27 @@ object IcebergScanSupport extends Logging {
partitionSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType)),
"Has unsupported schema type.")
Review Comment:
Minor: `plan()` is called twice per query (once in `isSupported`, once in `convert` — pattern from `IcebergConvertProvider`). This PR adds `detectRenameOrDrop` inside `plan()`, which iterates **all** historical schema versions via `table.schemas().values()`. For long-lived tables this loading + comparison now happens twice per scan node. Consider caching the plan result (e.g., via a `TreeNodeTag` on the `BatchScanExec`) to avoid this.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -75,6 +76,27 @@ object IcebergScanSupport extends Logging {
partitionSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType)),
"Has unsupported schema type.")
Review Comment:
The `inspected` block bundles `detectRenameOrDrop` and `expectedFieldIds` in a single try/catch. If `detectRenameOrDrop` throws (e.g., a catalog timeout in `table.schemas()`), `expectedFieldIds` — which is independent and likely to succeed since `expectedSchema()` is a local field access — is also discarded. The scan falls back to Spark entirely.
Consider separating them:
```scala
val fieldIdsByName = try {
AuronIcebergSourceUtil.expectedFieldIds(scan.asInstanceOf[AnyRef])
} catch { case NonFatal(t) => logWarning(...); return None }
val renameOrDrop = try {
AuronIcebergSourceUtil.detectRenameOrDrop(scan.asInstanceOf[AnyRef])
} catch { case NonFatal(t) =>
logWarning(...)
AuronIcebergSourceUtil.RenameOrDrop(topLevel = true, nested = true) // conservative
}
```
This way a transient schema-history failure can still fall back on the ORC/nested guards while preserving field-id matching for Parquet.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/iceberg/spark/source/AuronIcebergSourceUtil.scala:
##########
@@ -16,13 +16,77 @@
*/
package org.apache.iceberg.spark.source
+import scala.collection.JavaConverters._
+
object AuronIcebergSourceUtil {
+ final case class RenameOrDrop(topLevel: Boolean, nested: Boolean)
+
def getClassOfSparkBatchQueryScan(): Class[SparkBatchQueryScan] = {
classOf[SparkBatchQueryScan]
}
def getClassOfSparkInputPartition(): Class[SparkInputPartition] = {
classOf[SparkInputPartition]
}
+
+ def expectedFieldIds(scan: AnyRef): Map[String, Int] = {
+ val expectedSchema = asBatchQueryScan(scan).expectedSchema()
+ expectedSchema.columns().asScala.map(field => field.name() -> field.fieldId()).toMap
+ }
+
+ def detectRenameOrDrop(scan: AnyRef): RenameOrDrop = {
+ val table = asBatchQueryScan(scan).table()
+ val currentFields = collectFieldIdToName(table.schema())
Review Comment:
Two observations on `detectRenameOrDrop`:
1. `table.schemas().values()` includes the **current** schema. When compared against `currentFields` (built from `table.schema()`), every field matches itself — the entire iteration is a no-op. Consider filtering it out: `table.schemas().asScala.filterNot(_._1 == table.schema().schemaId())`.
2. `collectFieldIdToName` hand-rolls a recursive field-ID collector. Iceberg provides `TypeUtil.indexById(schema.asStruct())` which returns `Map<Integer, NestedField>` covering all nested fields. The only extra piece is the `topLevel` flag, which can be derived trivially from `schema.columns()`. Using the Iceberg utility would reduce maintenance surface and benefit from upstream fixes for new type variants.
##########
thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/auron/iceberg/IcebergScanSupport.scala:
##########
@@ -75,6 +76,27 @@ object IcebergScanSupport extends Logging {
partitionSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType)),
"Has unsupported schema type.")
+ val inspected: Option[(AuronIcebergSourceUtil.RenameOrDrop, Map[String, Int])] =
+ try {
+ Some(
+ (
+ AuronIcebergSourceUtil.detectRenameOrDrop(scan.asInstanceOf[AnyRef]),
+ AuronIcebergSourceUtil.expectedFieldIds(scan.asInstanceOf[AnyRef])))
+ } catch {
+ case NonFatal(t) =>
+ logWarning(s"Failed to inspect Iceberg field ids for $scanClassName.", t)
+ None
+ }
+ if (inspected.isEmpty) {
+ return None
+ }
+ val (renameOrDrop, fieldIdsByName) = inspected.get
+ assert(!renameOrDrop.nested, "Nested Iceberg rename or drop is not supported.")
+
+ assert(
+ fileSchema.fields.forall(field => fieldIdsByName.contains(field.name)),
+ "Failed to find field ids for all Iceberg data columns.")
+
val partitions = inputPartitions(exec)
Review Comment:
Nit: the assertion message `"Failed to find field ids for all Iceberg data columns."` doesn't include **which** columns are missing, making debugging harder. Consider:
```scala
val missing = fileSchema.fields.filterNot(f => fieldIdsByName.contains(f.name)).map(_.name)
assert(missing.isEmpty, s"Missing Iceberg field ids for columns: ${missing.mkString(", ")}")
```
##########
native-engine/datafusion-ext-plans/src/scan/mod.rs:
##########
@@ -89,6 +89,16 @@ impl SchemaAdapter for AuronSchemaAdapter {
}
}
+fn fields_match(table_field: &Field, file_field: &Field) -> bool {
+ match table_field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
+ Some(table_field_id) => file_field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .is_some_and(|file_field_id| file_field_id == table_field_id),
+ None => table_field.name().eq_ignore_ascii_case(file_field.name()),
Review Comment:
When `table_field` has a `PARQUET:field_id` but `file_field` does **not**, `is_some_and` returns `false` and there is no name-based fallback — the column simply doesn't match.
For spec-compliant Iceberg Parquet files this is fine (the Iceberg spec mandates field IDs, and arrow-rs populates them into Arrow metadata). But if an older Parquet writer omitted the `field_id` in the Thrift `SchemaElement`, or if a non-Iceberg Parquet file happens to be served through this path, every column would fail to match and the scan would produce all-NULL rows.
Consider falling back to name matching when `file_field` lacks a field ID:
```rust
fn fields_match(table_field: &Field, file_field: &Field) -> bool {
match table_field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
Some(table_field_id) => match file_field.metadata().get(PARQUET_FIELD_ID_META_KEY) {
Some(file_field_id) => file_field_id == table_field_id,
None => table_field.name().eq_ignore_ascii_case(file_field.name()),
},
None => table_field.name().eq_ignore_ascii_case(file_field.name()),
}
}
```
This preserves field-id matching when both sides have IDs, but degrades gracefully to name matching otherwise.
--
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: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org