[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 <weimingdiit@gmail.com>

Author: weimingdiit

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")

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

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,20 +57,39 @@ 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
     }
 
     val partitions = inputPartitions(exec)
     // 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.

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)

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)