You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by sl...@apache.org on 2026/04/12 03:11:10 UTC
(auron) branch master updated: [AURON #2175] Add native support for the _file metadata column (#2184)
This is an automated email from the ASF dual-hosted git repository.
slfan1989 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 193b5cbd [AURON #2175] Add native support for the _file metadata column (#2184)
193b5cbd is described below
commit 193b5cbd540f4e632ad8f694fcf4d617d545606b
Author: Ming Wei <we...@gmail.com>
AuthorDate: Sun Apr 12 11:11:05 2026 +0800
[AURON #2175] Add native support for the _file metadata column (#2184)
# Which issue does this PR close?
Closes #2175
# Rationale for this change
This PR adds native support for Iceberg metadata columns in Auron,
starting with `_file`.
Previously, Iceberg scans fell back whenever metadata columns were
projected. With this change, queries that read `_file` can remain on the
native Iceberg scan path.
Iceberg metadata columns are useful in real workloads for debugging,
lineage, and inspection queries. However, Auron previously treated
metadata columns as unsupported and fell back to Spark.
This PR improves native Iceberg scan coverage by supporting metadata
columns that can be represented as file-level constant values, while
still falling back for unsupported row-level metadata columns.
# What changes are included in this PR?
This PR:
- adds native support for the Iceberg `_file` metadata column
- keeps unsupported metadata columns such as `_pos` on the fallback path
- extends `IcebergScanPlan` to distinguish between:
- file-backed data columns
- metadata columns materialized outside the file payload
- updates `IcebergScanSupport` to stop rejecting all metadata columns
unconditionally
- passes supported metadata values through the native Iceberg scan path
as per-file constant values
- updates `NativeIcebergTableScanExec` to project both normal data
columns and supported metadata columns
- adds integration tests in `AuronIcebergIntegrationSuite`
# Scope of support in this PR
This PR intentionally takes a conservative approach.
Supported in native scan:
- `_file`
Still falls back:
- `_pos`
- other unsupported metadata columns that require row-level metadata
handling
# Why this design?
`_file` is a file-level metadata column: every row coming from the same
file shares the same value. That makes it a good fit for the existing
native file-scan path by treating it as a per-file constant column.
In contrast, `_pos` is row-level metadata and cannot be represented
correctly with the same mechanism, so it remains unsupported in native
execution for now.
# How was this patch tested?
CI.
---------
Signed-off-by: weimingdiit <we...@gmail.com>
---
.../org/apache/auron/AuronFunctionSuite.scala | 16 +++++--
.../apache/spark/sql/auron/NativeConverters.scala | 8 ++++
.../sql/auron/iceberg/IcebergScanSupport.scala | 53 ++++++++++++++++++----
.../auron/plan/NativeIcebergTableScanExec.scala | 36 +++++++++++----
.../iceberg/AuronIcebergIntegrationSuite.scala | 48 ++++++++++++++++++--
5 files changed, 136 insertions(+), 25 deletions(-)
diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala
index 6b6c0adf..8dadd4d5 100644
--- a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala
+++ b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala
@@ -358,7 +358,7 @@ class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite {
val err = intercept[Exception] {
df.collect()
}
- assert(err.getMessage.contains("null map keys"))
+ assert(allCauseMessages(err).toLowerCase.contains("null map keys"))
val plan = stripAQEPlan(df.queryExecution.executedPlan)
plan
.collectFirst { case op if !isNativeOrPassThrough(op) => op }
@@ -406,7 +406,7 @@ class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite {
|plan:
|${plan}""".stripMargin)
}
- assert(err.getMessage.toLowerCase.contains("null map key"))
+ assert(allCauseMessages(err).toLowerCase.contains("null map key"))
}
}
@@ -430,10 +430,20 @@ class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite {
|plan:
|${plan}""".stripMargin)
}
- assert(err.getMessage.toLowerCase.contains("duplicate key"))
+ assert(allCauseMessages(err).toLowerCase.contains("duplicate key"))
}
}
+ private def allCauseMessages(err: Throwable): String = {
+ val messages = scala.collection.mutable.ArrayBuffer.empty[String]
+ var current = err
+ while (current != null) {
+ Option(current.getMessage).foreach(messages += _)
+ current = current.getCause
+ }
+ messages.mkString(" | caused by: ")
+ }
+
test("map_from_entries last win dedup policy") {
withTable("t1") {
sql("create table t1(c1 array<struct<k:string,v:int>>) using parquet")
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala
index f4bf67fd..750aaa52 100644
--- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala
@@ -469,6 +469,9 @@ object NativeConverters extends Logging {
.setReturnNullable(subquery.nullable))
}
+ case expr if isNoOpAnsiCast(expr) =>
+ convertExprWithFallback(expr.children.head, isPruningExpr, fallback)
+
// cast
case cast: Cast =>
val involvesDateOrTimestamp =
@@ -1439,6 +1442,11 @@ object NativeConverters extends Logging {
Cast(expr, dataType)
}
+ private def isNoOpAnsiCast(expr: Expression): Boolean =
+ expr.getClass.getSimpleName == "AnsiCast" &&
+ expr.children.size == 1 &&
+ expr.children.head.dataType == expr.dataType
+
def unpackBinaryTypeCast(expr: Expression): Expression =
expr match {
case expr: Cast if expr.dataType == BinaryType => expr.child
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 39e14f3e..1d9efbc3 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
@@ -31,10 +31,14 @@ import org.apache.spark.sql.types.{BinaryType, DataType, DecimalType, StringType
import org.apache.auron.{protobuf => pb}
+// fileSchema is read from the data files. partitionSchema carries supported metadata columns
+// (for example _file) that are materialized as per-file constant values in the native scan.
final case class IcebergScanPlan(
fileTasks: Seq[FileScanTask],
fileFormat: FileFormat,
readSchema: StructType,
+ fileSchema: StructType,
+ partitionSchema: StructType,
pruningPredicates: Seq[pb.PhysicalExprNode])
object IcebergScanSupport extends Logging {
@@ -53,12 +57,24 @@ object IcebergScanSupport extends Logging {
}
val readSchema = scan.readSchema
- // Native scan does not support Iceberg metadata columns (e.g. _file, _pos).
- if (hasMetadataColumns(readSchema)) {
+ val unsupportedMetadataColumns = collectUnsupportedMetadataColumns(readSchema)
+ // Native scan can project file-level metadata columns such as _file via partition values.
+ // Metadata columns that require per-row materialization (for example _pos) still fallback.
+ if (unsupportedMetadataColumns.nonEmpty) {
return None
}
- if (!readSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType))) {
+ val fileSchema = StructType(readSchema.fields.filterNot(isSupportedMetadataColumn))
+ // Supported metadata columns are materialized via per-file constant values rather than
+ // read from the Iceberg data file payload.
+ val partitionSchema = StructType(readSchema.fields.filter(isSupportedMetadataColumn))
+
+ if (!fileSchema.fields.forall(field => NativeConverters.isTypeSupported(field.dataType))) {
+ return None
+ }
+
+ if (!partitionSchema.fields.forall(field =>
+ NativeConverters.isTypeSupported(field.dataType))) {
return None
}
@@ -66,7 +82,14 @@ object IcebergScanSupport extends Logging {
// Empty scan (e.g. empty table) should still build a plan to return no rows.
if (partitions.isEmpty) {
logWarning(s"Native Iceberg scan planned with empty partitions for $scanClassName.")
- return Some(IcebergScanPlan(Seq.empty, FileFormat.PARQUET, readSchema, Seq.empty))
+ return Some(
+ IcebergScanPlan(
+ Seq.empty,
+ FileFormat.PARQUET,
+ readSchema,
+ fileSchema,
+ partitionSchema,
+ Seq.empty))
}
val icebergPartitions = partitions.flatMap(icebergPartition)
@@ -94,12 +117,26 @@ object IcebergScanSupport extends Logging {
}
val pruningPredicates = collectPruningPredicates(scan.asInstanceOf[AnyRef], readSchema)
-
- Some(IcebergScanPlan(fileTasks, format, readSchema, pruningPredicates))
+ Some(
+ IcebergScanPlan(
+ fileTasks,
+ format,
+ readSchema,
+ fileSchema,
+ partitionSchema,
+ pruningPredicates))
}
- private def hasMetadataColumns(schema: StructType): Boolean =
- schema.fields.exists(field => MetadataColumns.isMetadataColumn(field.name))
+ private def collectUnsupportedMetadataColumns(schema: StructType): Seq[String] =
+ schema.fields.collect {
+ case field
+ if MetadataColumns.isMetadataColumn(field.name) &&
+ !isSupportedMetadataColumn(field) =>
+ field.name
+ }
+
+ private def isSupportedMetadataColumn(field: org.apache.spark.sql.types.StructField): Boolean =
+ field.name == MetadataColumns.FILE_PATH.name()
private def inputPartitions(exec: BatchScanExec): Seq[InputPartition] = {
// Prefer DataSource V2 batch API; if not available, fallback to exec methods via reflection.
diff --git a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala
index 3cf5cf56..38697792 100644
--- a/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala
+++ b/thirdparty/auron-iceberg/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeIcebergTableScanExec.scala
@@ -24,7 +24,7 @@ import java.util.UUID
import scala.collection.JavaConverters._
import org.apache.hadoop.fs.FileSystem
-import org.apache.iceberg.{FileFormat, FileScanTask}
+import org.apache.iceberg.{FileFormat, FileScanTask, MetadataColumns}
import org.apache.spark.Partition
import org.apache.spark.TaskContext
import org.apache.spark.broadcast.Broadcast
@@ -33,13 +33,14 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.auron.{EmptyNativeRDD, NativeConverters, NativeHelper, NativeRDD, NativeSupports, Shims}
import org.apache.spark.sql.auron.iceberg.IcebergScanPlan
import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.Literal
import org.apache.spark.sql.execution.LeafExecNode
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.datasources.{FilePartition, PartitionedFile}
import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{StringType, StructType}
import org.apache.spark.util.SerializableConfiguration
import org.apache.auron.{protobuf => pb}
@@ -57,32 +58,37 @@ case class NativeIcebergTableScanExec(basedScan: BatchScanExec, plan: IcebergSca
override val output = basedScan.output
override val outputPartitioning = basedScan.outputPartitioning
- private lazy val readSchema: StructType = plan.readSchema
+ private lazy val fileSchema: StructType = plan.fileSchema
+ private lazy val partitionSchema: StructType = plan.partitionSchema
+ private lazy val projectableSchema: StructType =
+ StructType(fileSchema.fields ++ partitionSchema.fields)
private lazy val fileTasks: Seq[FileScanTask] = plan.fileTasks
private lazy val pruningPredicates: Seq[pb.PhysicalExprNode] = plan.pruningPredicates
private lazy val partitions: Array[FilePartition] = buildFilePartitions()
private lazy val fileSizes: Map[String, Long] = buildFileSizes()
- private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(readSchema)
+ private lazy val nativeFileSchema: pb.Schema = NativeConverters.convertSchema(fileSchema)
private lazy val nativePartitionSchema: pb.Schema =
- NativeConverters.convertSchema(StructType(Nil))
+ NativeConverters.convertSchema(partitionSchema)
private lazy val caseSensitive: Boolean = SQLConf.get.caseSensitiveAnalysis
private lazy val fieldIndexByName: Map[String, Int] = {
if (caseSensitive) {
- readSchema.fieldNames.zipWithIndex.toMap
+ projectableSchema.fieldNames.zipWithIndex.toMap
} else {
- readSchema.fieldNames.map(_.toLowerCase(Locale.ROOT)).zipWithIndex.toMap
+ projectableSchema.fieldNames.map(_.toLowerCase(Locale.ROOT)).zipWithIndex.toMap
}
}
private def fieldIndexFor(name: String): Int = {
if (caseSensitive) {
- fieldIndexByName.getOrElse(name, readSchema.fieldIndex(name))
+ fieldIndexByName.getOrElse(name, projectableSchema.fieldIndex(name))
} else {
- fieldIndexByName.getOrElse(name.toLowerCase(Locale.ROOT), readSchema.fieldIndex(name))
+ fieldIndexByName.getOrElse(
+ name.toLowerCase(Locale.ROOT),
+ projectableSchema.fieldIndex(name))
}
}
@@ -99,6 +105,7 @@ case class NativeIcebergTableScanExec(basedScan: BatchScanExec, plan: IcebergSca
.setPath(filePath)
.setSize(size)
.setLastModifiedNs(0)
+ .addAllPartitionValues(metadataPartitionValues(filePath).asJava)
.setRange(
pb.FileRange
.newBuilder()
@@ -113,6 +120,17 @@ case class NativeIcebergTableScanExec(basedScan: BatchScanExec, plan: IcebergSca
.build()
}
+ private def metadataPartitionValues(filePath: String): Seq[pb.ScalarValue] =
+ partitionSchema.fields.map { field =>
+ field.name match {
+ case name if name == MetadataColumns.FILE_PATH.name() =>
+ NativeConverters.convertExpr(Literal.create(filePath, StringType)).getLiteral
+ case name =>
+ throw new IllegalStateException(
+ s"unsupported Iceberg metadata column in native scan: $name")
+ }
+ }
+
override def doExecuteNative(): NativeRDD = {
if (partitions.isEmpty) {
return new EmptyNativeRDD(sparkContext)
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 c74fdd26..0c5b752a 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
@@ -207,13 +207,36 @@ class AuronIcebergIntegrationSuite
}
}
- test("iceberg scan falls back when reading metadata columns") {
+ test("iceberg native scan supports _file metadata column") {
withTable("local.db.t4") {
sql("create table local.db.t4 using iceberg as select 1 as id, 'a' as v")
- val df = sql("select _file from local.db.t4")
- df.collect()
- val plan = df.queryExecution.executedPlan.toString()
- assert(!plan.contains("NativeIcebergTableScan"))
+ checkSparkAnswerAndOperator("select _file from local.db.t4")
+ }
+ }
+
+ test("iceberg native scan supports data columns with _file metadata column") {
+ withTable("local.db.t4_mixed") {
+ sql("create table local.db.t4_mixed using iceberg as select 1 as id, 'a' as v")
+ checkSparkAnswerAndOperator("select id, _file from local.db.t4_mixed")
+ }
+ }
+
+ test("iceberg native scan preserves projected order for _file metadata column") {
+ withTable("local.db.t4_metadata_first") {
+ sql("create table local.db.t4_metadata_first using iceberg as select 1 as id, 'a' as v")
+ checkSparkAnswerAndOperator("select _file, id from local.db.t4_metadata_first")
+ }
+ }
+
+ test("iceberg scan falls back when reading unsupported metadata columns") {
+ withTable("local.db.t4_pos") {
+ sql("create table local.db.t4_pos using iceberg as select 1 as id, 'a' as v")
+ withSQLConf("spark.auron.enable" -> "true", "spark.auron.enable.iceberg.scan" -> "true") {
+ val df = sql("select _pos from local.db.t4_pos")
+ df.collect()
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(!plan.contains("NativeIcebergTableScan"))
+ }
}
}
@@ -303,6 +326,21 @@ class AuronIcebergIntegrationSuite
}
}
+ private def checkSparkAnswerAndOperator(sqlText: String): DataFrame = {
+ var expected: Seq[Row] = Nil
+ withSQLConf("spark.auron.enable" -> "false") {
+ expected = sql(sqlText).collect().toSeq
+ }
+
+ var df: DataFrame = null
+ withSQLConf("spark.auron.enable" -> "true", "spark.auron.enable.iceberg.scan" -> "true") {
+ df = sql(sqlText)
+ checkAnswer(df, expected)
+ val plan = df.queryExecution.executedPlan.toString()
+ assert(plan.contains("NativeIcebergTableScan"))
+ }
+ df
+ }
private def icebergScanPlan(df: DataFrame) =
df.queryExecution.sparkPlan.collectFirst { case scan: BatchScanExec =>
IcebergScanSupport.plan(scan)