feat: [iceberg] support iceberg metadata columns _file, _partition

native/Cargo.toml

@@ -58,10 +58,13 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"]
 url = "2.2"
 aws-config = "1.8.18"
 aws-credential-types = "1.2.13"
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "80a30d3" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "80a30d3", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "77c5d710" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "77c5d710", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
 reqsign-core = "3"
 
+[patch."https://github.com/apache/iceberg-rust"]
+iceberg = { git = "https://github.com/parthchandra/iceberg-rust", branch = "metadata-columns-test" }
+
 [profile.release]
 debug = true
 overflow-checks = false

native/core/src/execution/operators/iceberg_scan.rs

@@ -651,6 +651,7 @@ mod tests {
             partition: None,
             partition_spec: None,
             name_mapping: None,
+            unified_partition_type: None,
             case_sensitive: false,
         }
     }

native/core/src/execution/planner.rs

@@ -3678,6 +3678,41 @@ fn parse_file_scan_tasks_from_common(
         })
         .collect::<Result<Vec<_>, _>>()?;
 
+    // Compute unified partition type from the partition_type_pool.
+    // Each entry is a StructType JSON with the resolved field types for one partition spec.
+    // Merge all specs into a single unified type (dedup by field_id).
+    //
+    // NOTE: The type information here comes from Iceberg Java's PartitionSpec.partitionType()
+    // (via Scala reflection). This is the same type computation as iceberg-rust's
+    // Transform::result_type() -- both implement the Iceberg spec's transform result rules.
+    // When iceberg-rust's own scan planning is used (not Comet's proto path), it computes
+    // this via compute_unified_partition_type(specs, schema) instead.
+    let unified_partition_type = {
+        let mut seen_field_ids = std::collections::HashSet::new();
+        let mut struct_fields: Vec<iceberg::spec::NestedFieldRef> = Vec::new();
+
+        for type_json in &proto_common.partition_type_pool {
+            match serde_json::from_str::<iceberg::spec::StructType>(type_json) {
+                Ok(struct_type) => {
+                    for field in struct_type.fields() {
+                        if seen_field_ids.insert(field.id) {
+                            struct_fields.push(Arc::clone(field));
+                        }
+                    }
+                }
+                Err(e) => {
+                    return Err(ExecutionError::GeneralError(format!(
+                        "Failed to deserialize partition type JSON from pool: {e}"
+                    )));
+                }
+            }
+        }
+
+        iceberg::spec::StructType::new(struct_fields)
+    };
+
+    let unified_partition_type_arc = Arc::new(unified_partition_type);
+
     let results: Result<Vec<_>, _> = proto_tasks
         .iter()
         .map(|proto_task| {
@@ -3778,6 +3813,14 @@ fn parse_file_scan_tasks_from_common(
                 .field_ids
                 .clone();
 
+            let unified_partition_type_for_task = if project_field_ids
+                .contains(&iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION)
+            {
+                Some(Arc::clone(&unified_partition_type_arc))
+            } else {
+                None
+            };
+
             Ok(iceberg::scan::FileScanTask {
                 file_size_in_bytes: proto_task.file_size_in_bytes,
                 data_file_path: proto_task.data_file_path.clone(),
@@ -3792,6 +3835,7 @@ fn parse_file_scan_tasks_from_common(
                 partition,
                 partition_spec,
                 name_mapping,
+                unified_partition_type: unified_partition_type_for_task,
                 case_sensitive: false,
             })
         })
@@ -5408,4 +5452,21 @@ mod tests {
             }
         });
     }
+
+    #[test]
+    fn test_metadata_field_id_constants_match_iceberg_rust() {
+        // These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds)
+        // as Int.MaxValue - 1 and Int.MaxValue - 5. This test ensures the Rust constants
+        // haven't drifted, which would cause a silent mismatch with the Scala side.
+        assert_eq!(
+            iceberg::metadata_columns::RESERVED_FIELD_ID_FILE,
+            i32::MAX - 1,
+            "RESERVED_FIELD_ID_FILE must be i32::MAX - 1 to match Scala MetadataFieldIds"
+        );
+        assert_eq!(
+            iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION,
+            i32::MAX - 5,
+            "RESERVED_FIELD_ID_PARTITION must be i32::MAX - 5 to match Scala MetadataFieldIds"
+        );
+    }
 }

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -125,7 +125,7 @@ case class CometScanRule(session: SparkSession)
       case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) =>
         withFallbackReason(scan, "Comet Scan is not enabled")
 
-      case scan if metadataCols(scan).nonEmpty =>
+      case scan: FileSourceScanExec if metadataCols(scan).nonEmpty =>
         withFallbackReason(
           scan,
           s"Metadata column(s) ${metadataCols(scan).mkString(", ")} is not supported")
@@ -297,6 +297,11 @@ case class CometScanRule(session: SparkSession)
 
     scanExec.scan match {
       case scan: CSVScan if COMET_CSV_V2_NATIVE_ENABLED.get() =>
+        if (scanExec.output.exists(_.isMetadataCol)) {
+          return withFallbackReason(
+            scanExec,
+            "Metadata columns are not supported for CSV V2 scans")
+        }
         val fallbackReasons = new ListBuffer[String]()
         val schemaSupported =
           CometBatchScanExec.isSchemaSupported(scan.readDataSchema, fallbackReasons)
@@ -356,9 +361,30 @@ case class CometScanRule(session: SparkSession)
           return withFallbackReasons(scanExec, fallbackReasons.toSet)
         }
 
+        // Check for unsupported metadata columns in Iceberg scans
+        val unsupportedMetadataCols = scanExec.output.filter(_.isMetadataCol).filterNot { attr =>
+          CometIcebergNativeScan.MetadataFieldIds.keySet.contains(attr.name)
+        }
+        if (unsupportedMetadataCols.nonEmpty) {
+          fallbackReasons += "Unsupported Iceberg metadata columns: " +
+            unsupportedMetadataCols.map(_.name).mkString(", ")
+          return withFallbackReasons(scanExec, fallbackReasons.toSet)
+        }
+
         val typeChecker = CometScanTypeChecker()
+        // Filter out metadata columns from schema check -- their types are handled
+        // by iceberg-rust directly (e.g., _partition can be an empty struct for
+        // unpartitioned tables which the general type checker rejects).
+        val metadataColNames = scanExec.output.filter(_.isMetadataCol).map(_.name).toSet
+        val dataSchema = if (metadataColNames.nonEmpty) {
+          val filtered =
+            scanExec.scan.readSchema().filter(f => !metadataColNames.contains(f.name))
+          new org.apache.spark.sql.types.StructType(filtered.toArray)
+        } else {
+          scanExec.scan.readSchema()
+        }
         val schemaSupported =
-          typeChecker.isSchemaSupported(scanExec.scan.readSchema(), fallbackReasons)
+          typeChecker.isSchemaSupported(dataSchema, fallbackReasons)
 
         if (!schemaSupported) {
           fallbackReasons += "Comet extension is not enabled for " +
@@ -446,15 +472,18 @@ case class CometScanRule(session: SparkSession)
         }
 
         // Now perform all validation using the pre-extracted metadata
-        // Check if table uses a FileIO implementation compatible with iceberg-rust
-
+        // Check if table uses a FileIO implementation compatible with iceberg-rust.
+        // Comet's native reader uses object_store (Rust) for I/O, bypassing Iceberg Java's
+        // FileIO entirely. Only allow known-compatible implementations whose underlying
+        // storage object_store can reach via standard URL schemes.
         val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match {
           case Some(fileIO)
-              if fileIO.getClass.getName == "org.apache.iceberg.inmemory.InMemoryFileIO" =>
-            fallbackReasons += "InMemoryFileIO is not supported by Comet's native reader"
-            false
-          case Some(_) =>
+              if CometScanRule.CompatibleFileIOClasses.contains(fileIO.getClass.getName) =>
             true
+          case Some(fileIO) =>
+            fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not supported by " +
+              "Comet's native reader (object_store bypasses Iceberg Java FileIO)"
+            false
           case None =>
             fallbackReasons += "Could not check FileIO compatibility"
             false
@@ -772,6 +801,18 @@ case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim {
 
 object CometScanRule extends Logging {
 
+  // Iceberg FileIO implementations whose underlying storage object_store can reach.
+  // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible
+  // because Comet's native reader bypasses Java FileIO entirely.
+  private val CompatibleFileIOClasses: Set[String] = Set(
+    "org.apache.iceberg.hadoop.HadoopFileIO",
+    "org.apache.iceberg.aws.s3.S3FileIO",
+    "org.apache.iceberg.gcp.gcs.GCSFileIO",
+    "org.apache.iceberg.io.ResolvingFileIO",
+    "org.apache.iceberg.spark.SparkFileIO",
+    "org.apache.iceberg.azure.adlsv2.ADLSFileIO",
+    "org.apache.iceberg.CachingFileIO")
+
   // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer depends only on the
   // URL scheme, so we cache by scheme and never re-cross the JNI boundary for a repeated scheme.
   private val schemeSupportCache =

spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala

@@ -71,6 +71,14 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     }
   }
 
+  // Iceberg reserved field IDs for metadata columns.
+  // These must match iceberg-rust's constants in crates/iceberg/src/metadata_columns.rs:
+  //   RESERVED_FIELD_ID_FILE      = i32::MAX - 1  (2147483646)
+  //   RESERVED_FIELD_ID_PARTITION = i32::MAX - 5  (2147483642)
+  // Scala's Int.MaxValue == 2^31 - 1 == Rust's i32::MAX.
+  val MetadataFieldIds: Map[String, Int] =
+    Map("_file" -> (Int.MaxValue - 1), "_partition" -> (Int.MaxValue - 5))
+
   /**
    * Converts an Iceberg partition value to protobuf format. Protobuf is less verbose than JSON.
    * The following types are also serialized as integer values instead of as strings - Timestamps,
@@ -866,14 +874,16 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
 
                 val nameToFieldId = IcebergReflection.buildFieldIdMapping(schema)
 
-                val projectFieldIds = output.flatMap { attr =>
+                val projectFieldIds = output.map { attr =>
                   nameToFieldId
                     .get(attr.name)
                     .orElse(metadata.globalFieldIdMapping.get(attr.name))
-                    .orElse {
-                      logWarning(s"Column '${attr.name}' not found in task or scan schema, " +
-                        "skipping projection")
-                      None
+                    .orElse(CometIcebergNativeScan.MetadataFieldIds.get(attr.name))
+                    .getOrElse {
+                      throw new IllegalStateException(
+                        s"Column '${attr.name}' not found in task schema, global schema, " +
+                          "or metadata field IDs. This indicates a bug in CometScanRule " +
+                          "validation -- all output columns should be resolvable.")
                     }
                 }