feat: support Iceberg metadata columns _pos, _spec, _file, and _partition by parthchandra · Pull Request #4752 · apache/datafusion-comet

parthchandra · 2026-06-29T16:37:40Z

Which issue does this PR close?

Adds support for iceberg metadata columns

Rationale for this change

What changes are included in this PR?

How are these changes tested?

## Which issue does this PR close?  - Partially closes #2607 ## What changes are included in this PR?  - If a projection includes `_spec_id`, which is a constant like the `_file` metadata column for all rows, add it to `RecordBatchTransformerBuilder` ## Are these changes tested?  - Yes, unit tests. - Iceberg Spark: apache/datafusion-comet#4752 Sample plan ```shell == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- CometSort [id#1247L, data#1248, _spec_id#1249], [id#1247L ASC NULLS FIRST] +- CometExchange rangepartitioning(id#1247L ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=3533] +- CometFilter [id#1247L, data#1248, _spec_id#1249], (id#1247L >= 10) +- CometIcebergNativeScan [id#1247L, data#1248, _spec_id#1249], /var/folders/d2/b93h6k7174ddqxltrxgb51040000gn/T/iceberg_warehouse16657344237870771906/iceberg_data/default/table/metadata/00004-70b616c5-1bbe-4e1d-bdd6-e9ac4910674a.metadata.json, testrest.default.table (branch=null) [filters=id IS NOT NULL, id >= 10, groupedBy=], 1 ```

## Which issue does this PR close?  - Partially closes apache#2607 ## What changes are included in this PR?  - If a projection includes `_spec_id`, which is a constant like the `_file` metadata column for all rows, add it to `RecordBatchTransformerBuilder` ## Are these changes tested?  - Yes, unit tests. - Iceberg Spark: apache/datafusion-comet#4752 Sample plan ```shell == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- CometSort [id#1247L, data#1248, _spec_id#1249], [id#1247L ASC NULLS FIRST] +- CometExchange rangepartitioning(id#1247L ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=3533] +- CometFilter [id#1247L, data#1248, _spec_id#1249], (id#1247L >= 10) +- CometIcebergNativeScan [id#1247L, data#1248, _spec_id#1249], /var/folders/d2/b93h6k7174ddqxltrxgb51040000gn/T/iceberg_warehouse16657344237870771906/iceberg_data/default/table/metadata/00004-70b616c5-1bbe-4e1d-bdd6-e9ac4910674a.metadata.json, testrest.default.table (branch=null) [filters=id IS NOT NULL, id >= 10, groupedBy=], 1 ```

## Which issue does this PR close?  - Partially closes #2607 ## What changes are included in this PR?  - Support `_pos` metadata column by delegating it to the Parquet reader's `RowNumber` virtual column, inspired by apache/datafusion#22026. - Refactor the `compare_schemas` function b/c `_pos` field id exists in both `source_schema` and `target_schema` and the original implementation might result in `SchemaComparison::NameChangesOnly`, which produces a `RecordBatch` with correct content but incorrect column order. ## Are these changes tested?  - Yes, unit tests. - Iceberg Spark: apache/datafusion-comet#4752 Sample plan ```shell == Physical Plan == AdaptiveSparkPlan isFinalPlan=false +- CometSort [id#863L, data#864, _pos#868L], [id#863L ASC NULLS FIRST] +- CometExchange rangepartitioning(id#863L ASC NULLS FIRST, 200), ENSURE_REQUIREMENTS, CometNativeShuffle, [plan_id=2634] +- CometFilter [id#863L, data#864, _pos#868L], (id#863L >= 10) +- CometIcebergNativeScan [id#863L, data#864, _pos#868L], file:/var/folders/d2/b93h6k7174ddqxltrxgb51040000gn/T/hive2622216829356851627/table/metadata/00004-95932e2b-bbe6-444e-81b8-069c5e748a20.metadata.json, spark_catalog.default.table (branch=null) [filters=id IS NOT NULL, id >= 10, groupedBy=], 1 ```

hsiang-c · 2026-07-21T05:01:54Z

Spark 4.2 fails probably b/c of #4969

parthchandra · 2026-07-21T14:26:16Z

Spark 4.2 fails probably b/c of #4969

Thanks for the pointer @hsiang-c !

hsiang-c · 2026-07-30T05:13:49Z

@@ -1,5 +1,5 @@
 diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
-index db659dc06b..78f5e3440f 100644


Nit, we need to use 10 digits hash: git config core.abbrev 10

fixed (thanks for reminding me!)

hsiang-c · 2026-07-30T05:19:38Z

+  // (e.g. EncryptingFileIO$WithSupportsPrefixOperations when the delegate is HadoopFileIO), so
+  // an exact class-name match misses it. Comet forwards each file's key_metadata to iceberg-rust
+  // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible.
+  private val EncryptingFileIOPrefix = "org.apache.iceberg.encryption.EncryptingFileIO"


I had to do this because encryption tests failed.

hsiang-c · 2026-07-30T05:24:34Z

+-- (ALTER TABLE ADD PARTITION FIELD / CALL system.add_partition_field).
+-- The SQL file test framework cannot register session extensions via
+-- Config directives. Partition evolution is covered in
+-- CometIcebergNativeSuite which has full session control.


hsiang-c · 2026-07-30T05:31:55Z

        withFallbackReason(scan, "Comet Scan is not enabled")

-      case scan if metadataCols(scan).nonEmpty =>
+      case scan: FileSourceScanExec if metadataCols(scan).nonEmpty =>


(nit) Maybe we can move this check into L139 b/c this is V1 only check.

Good point. Done

hsiang-c

LGTM

…_partition

hsiang-c · 2026-07-30T20:55:35Z

+            (1, 'US', 'A', 10.0), (2, 'EU', 'B', 20.0)
+        """)
+
+        // Add a second partition field via Iceberg Java API (partition evolution -> spec_id 1)


(nit) Not a must, we can also use ALTER TABLE here.

Oh. Actually I had to revert the change for the previous case as well. ALTER TABLE ... ADD PARTITION FIELD is only recognized by Iceberg's IcebergSparkSessionExtensions SQL parser, and that's registered at SparkSession construction time via spark.sql.extensions.
This suite is based on CometTestBase which does not add the iceberg extension. So leaving this as is.

mbutrovich

First pass, thanks @parthchandra! Awesome to see all of the upstream work that landed in iceberg-rust to enable this.

mbutrovich · 2026-07-30T21:55:39Z

+    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}"
+                    )));
+                }
+            }
+        }


This builds the unified _partition type by scanning partition_type_pool in pool-insertion order and keeping the first-seen field for each field id. Pool-insertion order is whatever order the Scala side happens to encounter distinct per-spec partition types while iterating DataSourceRDD partitions (CometIcebergNativeScan.scala, serializePartitionData), not a defined order.

Compare with iceberg-rust's own compute_unified_partition_type (crates/iceberg/src/partitioning.rs in iceberg-rust, pinned rev 3d84c81), which this code's own comment says it's meant to match: it sorts specs by spec_id descending before merging, and explicitly sorts the final field id list ascending before building the struct. Comet's version does neither.

For a simple case (fields only ever added, never dropped and re-added), ascending field ids happen to fall out of the pool naturally, since each spec's own field list is already ascending and ids only increase across specs. But the "re-add dropped partition field" scenario breaks that: a field dropped in one spec (voided, so it's filtered out of that spec's serialized partition type JSON, see CometIcebergNativeScan.scala around the fieldsJson construction) and re-added in a later spec gets a new, higher field id. If the pool happens to serialize that later spec's JSON before the id of the original, lower-numbered field is (re-)encountered from an earlier spec's JSON, struct_fields ends up with the higher id before the lower one, i.e. not ascending.

CometIcebergNativeSuite.scala has a test for exactly this shape ("partition evolution - re-add dropped partition field"), but it only calls checkIcebergNativeScan without asserting on struct field values directly, so I can't tell from reading the diff alone whether the ordering actually diverges in that test's specific data, or whether it happens to work out. Given build_partition_constant in iceberg-rust (record_batch_transformer.rs) builds the Arrow struct by iterating unified_partition_type.fields() in whatever order it's given, an order mismatch against the schema Spark's analyzer fixed for _partition would show up as wrong values in the wrong named fields, not a crash.

Could you confirm whether the field order this computation produces is actually guaranteed to be ascending by field id in all cases, or add the same sort iceberg-rust's compute_unified_partition_type does?

You're right. This was wrong.
Fixed and this should match compute_unified_partition_type/Partitioning.buildPartitionProjectionType. To get spec_id on the native side, I made partition_type_pool index-aligned with partition_spec_pool (documented in the .proto) and read partition_spec_cache[i].spec_id().
One caveat: because Comet ships resolved types (not the original Transform), the void-vs-non-void type preference isn't replicated — but that only affects a dropped-source-column field, which Comet already filters upstream

mbutrovich · 2026-07-30T21:56:27Z

+        // 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(_) =>
+          case Some(fileIO) if CometScanRule.isCompatibleFileIO(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


This PR flips the FileIO compatibility check from a deny-list (block only InMemoryFileIO, allow everything else) to an allow-list of seven exact class names plus an EncryptingFileIO prefix match:

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

Failing closed on an unrecognized FileIO is the right default here: Comet's native reader bypasses Iceberg Java's FileIO entirely and goes straight to object_store, so a deny-list would silently accelerate any new FileIO a future Iceberg release ships, with no way to know whether its I/O behavior (credential vending, request signing, etc.) is actually replicated on the native side. That's the same fail-closed pattern already used for formatVersionSupported (falls back on v > 3 rather than assuming forward compatibility), so the allow-list approach itself fits the codebase.

The actual gap is the matching mechanism: isCompatibleFileIO matches on exact class name (CompatibleFileIOClasses.contains(className)), not on the class hierarchy. A FileIO that subclasses one of these seven with no change to actual I/O behavior (a common pattern for adding metrics, retry logic, or multi-tenant credential routing on top of S3FileIO/HadoopFileIO) will no longer match, and will now silently fall back to Spark where it previously ran natively.

Matching should walk the already-loaded fileIO.getClass ancestry rather than loading each candidate name via IcebergReflection.loadClass. ClassLoaders.loadClass (spark/src/main/java/org/apache/comet/util/ClassLoaders.java:32) declares throws ClassNotFoundException uncaught, and isCompatibleFileIO is only reached from fileIOCompatible at plan time (CometScanRule.scala:480), where the codebase's established invariant is that a reflection failure becomes a fallback reason, never an uncaught exception (see every other check in transformV2Scan, e.g. defaultValuesSupported, schemaTypesSupported, taskValidation). A deployment that only bundles iceberg-aws (no iceberg-gcp/iceberg-azure on the classpath) would hit ClassNotFoundException trying to load GCSFileIO/ADLSFileIO just to test isInstance, crashing planning instead of falling back — likely why the PR used a string-name comparison against the live instance in the first place.

IcebergReflection.scala already has both pieces of idiom this needs, just not combined: findMethodInHierarchy (IcebergReflection.scala:112-126) walks a Class[_]'s superclass chain with a plain while loop, and isIcebergScanClass/ICEBERG_SCAN_CLASSES.contains(name) matches a live object's getClass.getName against a Set[String], the same pattern this PR's isCompatibleFileIO already uses at the leaf class only. Combining them (and placing the result in IcebergReflection, alongside the other "is this class one of these known names" helpers, rather than in CometScanRule's companion object) avoids ever loading a class that might not exist in this JVM:

// in IcebergReflection def classNameInHierarchy(clazz: Class[_], names: Set[String]): Boolean = { var current: Class[_] = clazz while (current != null) { if (names.contains(current.getName)) return true current = current.getSuperclass } false }

// in CometScanRule def isCompatibleFileIO(fileIO: AnyRef): Boolean = IcebergReflection.classNameInHierarchy(fileIO.getClass, CompatibleFileIOClasses) || fileIO.getClass.getName.startsWith(EncryptingFileIOPrefix)

This also means the call site at CometScanRule.scala:480 needs to change from passing fileIO.getClass.getName to passing fileIO itself, since matching subclasses requires the live Class[_], not just its name.

Right again. Fixed exactly as you suggested.

mbutrovich · 2026-07-30T21:57:02Z

+-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not
+-- supported in Spark 4.0. Covered in CometIcebergNativeSuite instead.


I searched CometIcebergNativeSuite.scala (including this PR's additions) for any UPDATE statement and found none, with or without metadata columns. There's no MoR UPDATE test anywhere in the suite, so this comment is inaccurate and MoR UPDATE's interaction with _file/_pos/_spec_id/_partition has no coverage at all right now. Could you either add that coverage (maybe gated on a newer Spark version where UPDATE is supported, similar to other version-gated tests in this suite) or fix the comment so it doesn't claim coverage that doesn't exist?

mbutrovich · 2026-07-30T21:57:22Z

+        val unifiedPartitionTypeSupported =
+          if (scanExec.output.exists(a => a.isMetadataCol && a.name == "_partition")) {
+            IcebergReflection.validateUnifiedPartitionType(metadata.table) match {
+              case Some(reason) =>
+                fallbackReasons += "Iceberg table has partition specs whose unified partition " +
+                  s"type cannot be computed for the _partition metadata column: $reason"
+                false
+              case None => true
+            }
+          } else {
+            true
+          }


This is new fallback logic for the case that motivated it: a V1 table where two specs bind the same partition field id to incompatible source/transform pairs. I didn't find a test that constructs such a table and confirms Comet actually falls back (with the right reason) instead of trying the native path. Since this is the one case this whole check exists to catch, it seems worth a dedicated test, even if constructing the conflicting-spec table requires going through Iceberg's Java API directly (as the other partition-evolution tests in this PR already do via updateSpec()).

Added a test. This couldn't be tested end to end via SQL because Spark's analyzer catches it and throws a validation error. The test hand-builds the conflicting V1 metadata.json (round-tripped through TableMetadataParser to catch malformed edits) and asserts directly on IcebergReflection.validateUnifiedPartitionType

mbutrovich · 2026-07-30T21:58:11Z

+        // 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)
+        }


The old test that verified an unsupported metadata column triggers fallback ("CometScanRule should report unsupported metadata columns") was replaced with tests for the four now-supported columns, but I don't see a replacement test that queries one of the still-unsupported Iceberg metadata columns (e.g. _deleted, _row_id, _change_type) and checks it still falls back correctly. Worth keeping a test for that path so a future change to MetadataFieldIds or this filter doesn't silently regress it.

Test added.

mbutrovich · 2026-07-30T21:58:34Z

+          if (partitionValues.isEmpty) {
+            val specId = spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int]
+            val emptySpecJson = s"""{"spec-id":$specId,"fields":[]}"""
+            val emptySpecIdx = partitionSpecToPoolIndex.getOrElseUpdate(
+              emptySpecJson, {
+                val idx = partitionSpecToPoolIndex.size
+                commonBuilder.addPartitionSpecPool(emptySpecJson)
                idx
              })
-            taskBuilder.setPartitionDataIdx(partitionDataIdx)
+            // Override the real spec registered above.
+            taskBuilder.setPartitionSpecIdx(emptySpecIdx)
          }


Two things here:

Everywhere else in this file, spec/type JSON is produced through PartitionSpecParser.toJson() (real spec) or the json4s DSL (compact(render(...)), for the partition type). This is the one spot that hand-interpolates a JSON string. It's safe today since specId is an Int, but it's inconsistent with the rest of the file's approach, and any later change that adds a string-typed field to this literal would need to remember to escape it. Might be worth building it the same way (json4s DSL) as the partition type JSON just above it.

By the time this branch runs, the "real" spec for this task has already been resolved, JSON-serialized via PartitionSpecParser.toJson(), and interned into partitionSpecToPoolIndex earlier in the same method (the specIdx/taskBuilder.setPartitionSpecIdx(specIdx) call above this block). This code then unconditionally overwrites that with the empty-spec index. For every value-less task, that's a reflection call plus a JSON serialization done and then thrown away. It's probably not hot enough to matter (value-less tasks only occur for partition-evolution edge cases), but if it's easy to restructure so the empty-spec check happens before the real spec is computed, it'd avoid the redundant work entirely.

Fixed the JSON consistency: the empty-spec literal now uses the json4s DSL like the rest of the file

parthchandra

Thanks for the amazing review @mbutrovich

parthchandra · 2026-07-31T01:42:35Z

+    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}"
+                    )));
+                }
+            }
+        }


You're right. This was wrong.
Fixed and this should match compute_unified_partition_type/Partitioning.buildPartitionProjectionType. To get spec_id on the native side, I made partition_type_pool index-aligned with partition_spec_pool (documented in the .proto) and read partition_spec_cache[i].spec_id().
One caveat: because Comet ships resolved types (not the original Transform), the void-vs-non-void type preference isn't replicated — but that only affects a dropped-source-column field, which Comet already filters upstream

parthchandra · 2026-07-31T01:43:29Z

+        // 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)
+        }


Test added.

parthchandra · 2026-07-31T01:44:35Z

+        // 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(_) =>
+          case Some(fileIO) if CometScanRule.isCompatibleFileIO(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


Right again. Fixed exactly as you suggested.

parthchandra · 2026-07-31T01:46:59Z

+        val unifiedPartitionTypeSupported =
+          if (scanExec.output.exists(a => a.isMetadataCol && a.name == "_partition")) {
+            IcebergReflection.validateUnifiedPartitionType(metadata.table) match {
+              case Some(reason) =>
+                fallbackReasons += "Iceberg table has partition specs whose unified partition " +
+                  s"type cannot be computed for the _partition metadata column: $reason"
+                false
+              case None => true
+            }
+          } else {
+            true
+          }


Added a test. This couldn't be tested end to end via SQL because Spark's analyzer catches it and throws a validation error. The test hand-builds the conflicting V1 metadata.json (round-tripped through TableMetadataParser to catch malformed edits) and asserts directly on IcebergReflection.validateUnifiedPartitionType

parthchandra · 2026-07-31T01:48:35Z

+          if (partitionValues.isEmpty) {
+            val specId = spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int]
+            val emptySpecJson = s"""{"spec-id":$specId,"fields":[]}"""
+            val emptySpecIdx = partitionSpecToPoolIndex.getOrElseUpdate(
+              emptySpecJson, {
+                val idx = partitionSpecToPoolIndex.size
+                commonBuilder.addPartitionSpecPool(emptySpecJson)
                idx
              })
-            taskBuilder.setPartitionDataIdx(partitionDataIdx)
+            // Override the real spec registered above.
+            taskBuilder.setPartitionSpecIdx(emptySpecIdx)
          }


Fixed the JSON consistency: the empty-spec literal now uses the json4s DSL like the rest of the file

mbutrovich · 2026-07-31T14:56:29Z


      if (spec != null) {
-        // Deduplicate partition spec
+        // Get the partition type/schema from the spec. Needed regardless of whether this task


Raising this again from my first pass, since we're already asking for another round of changes.

spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala, serializePartitionData (currently lines 354-533):

Lines 419-444 compute the real spec's JSON via reflection (PartitionSpecParser.toJson), dedup it into partitionSpecToPoolIndex, add the paired partitionTypeJson to the pool, and call taskBuilder.setPartitionSpecIdx(specIdx). This runs unconditionally, before we know whether the task has partition values.

Lines 496-513 then check partitionValues.isEmpty, and if it's true, compute specId via another reflection call, build a separate emptySpecJson, dedup that into the same map under a different key, and overwrite taskBuilder.setPartitionSpecIdx with the empty-spec index. The real spec entry added at line 429-443 is still sitting in partitionSpecToPoolIndex and commonBuilder's pools; it's just not referenced by this task anymore.

So every value-less task (partition evolution: a file written after a partition field was dropped) still does the toJson reflection call, the JSON string dedup lookup, and interns an unused pool entry, only to throw the result away a few lines later. The fix from my earlier comment stands: check partitionValues.isEmpty (or rather, whether partitionData has any non-unknown fields) before running the lines 419-444 block, so the value-less path only ever computes emptySpecJson, not both.

The fieldsJson computation right above this (lines 367-412) is fine to leave unconditional since its own comment says it's needed regardless of the value-less path -- it feeds partitionValues either way. It's specifically the spec-serialization-and-pool-insert block at 419-444 that should move after the value-less check, not before it.

Thank you for correcting. Done.

mbutrovich

This PR's new comments say the Iceberg native reader does its I/O through object_store. That's not accurate. The Iceberg scan path goes through iceberg-storage-opendal / opendal (native/core/src/execution/operators/iceberg_scan.rs imports iceberg_storage_opendal::OpenDalStorageFactory and iceberg::io::FileIO, and native/Cargo.toml pulls in iceberg-storage-opendal with the opendal-* feature flags). object_store is the crate Comet's non-Iceberg native Parquet/CSV scan uses (native/core/src/parquet/, native/core/src/execution/operators/csv_scan.rs) -- a different code path entirely. My own earlier review comment used "object_store" loosely when describing the Iceberg case, and that imprecision carried into the code comments below. Sorry for the bad steer.

Rather than swap object_store for opendal in these comments, I'd rather we not name either backend at all: iceberg-rust's storage layer could eventually replace opendal to object_store upstream, so a comment naming either one will go stale on the next iceberg dependency bump. Describe the behavior instead of the backend: Comet's native reader bypasses Iceberg Java's FileIO and reads through iceberg-rust's own storage layer.

Five spots to fix:

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:71

  // Iceberg FileIO implementations whose underlying storage object_store can reach.

to something like:

  // Iceberg FileIO implementations whose backing storage Comet's native reader can reach.

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:87

  // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible.

to something like:

  // and reads the ciphertext through iceberg-rust's own storage layer, so any EncryptingFileIO
  // variant is compatible.

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:476-478

        // 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.

to something like:

        // Comet's native reader bypasses Iceberg Java's FileIO entirely and reads through
        // iceberg-rust's own storage layer instead. Only allow known-compatible implementations
        // whose backing storage that layer can reach via standard URL schemes.

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

              "Comet's native reader (object_store bypasses Iceberg Java FileIO)"

to something like:

              "Comet's native reader (bypasses Iceberg Java FileIO)"

Not asking for a search-and-replace to opendal here, just to drop the backend name from these five comments so they describe the behavior rather than an implementation detail that's expected to change.

serializePartitionData computed the file's real partition spec (toJson reflection + pool intern) unconditionally, then overwrote it with the empty-spec index for value-less tasks (partition-evolution files with a dropped field). Move the real-spec serialization into a helper that only the value-carrying path calls, so value-less tasks compute just the empty spec. Spec and partition-type pools stay index-aligned: each path interns both entries together.

…kend The FileIO-compatibility comments named object_store as the backend, but the Iceberg scan path reads through iceberg-rust's own storage layer, not object_store. Since iceberg-rust's storage backend may change on a dependency bump, describe the behaviour (bypasses Iceberg Java FileIO, reads through iceberg-rust's storage layer) rather than naming a backend. Non-Iceberg V1 Parquet/CSV comments that genuinely use object_store are left unchanged.

parthchandra · 2026-07-31T17:19:28Z

This PR's new comments say the Iceberg native reader does its I/O through object_store. That's not accurate. The Iceberg scan path goes through iceberg-storage-opendal / opendal (native/core/src/execution/operators/iceberg_scan.rs imports iceberg_storage_opendal::OpenDalStorageFactory and iceberg::io::FileIO, and native/Cargo.toml pulls in iceberg-storage-opendal with the opendal-* feature flags). object_store is the crate Comet's non-Iceberg native Parquet/CSV scan uses (native/core/src/parquet/, native/core/src/execution/operators/csv_scan.rs) -- a different code path entirely. My own earlier review comment used "object_store" loosely when describing the Iceberg case, and that imprecision carried into the code comments below. Sorry for the bad steer.

Rather than swap object_store for opendal in these comments, I'd rather we not name either backend at all: iceberg-rust's storage layer could eventually replace opendal to object_store upstream, so a comment naming either one will go stale on the next iceberg dependency bump. Describe the behavior instead of the backend: Comet's native reader bypasses Iceberg Java's FileIO and reads through iceberg-rust's own storage layer.

Five spots to fix:

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:71
  // Iceberg FileIO implementations whose underlying storage object_store can reach.
to something like:
  // Iceberg FileIO implementations whose backing storage Comet's native reader can reach.
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:87
  // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible.
to something like:
  // and reads the ciphertext through iceberg-rust's own storage layer, so any EncryptingFileIO
  // variant is compatible.
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:476-478
        // 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.
to something like:
        // Comet's native reader bypasses Iceberg Java's FileIO entirely and reads through
        // iceberg-rust's own storage layer instead. Only allow known-compatible implementations
        // whose backing storage that layer can reach via standard URL schemes.
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:484
              "Comet's native reader (object_store bypasses Iceberg Java FileIO)"
to something like:
              "Comet's native reader (bypasses Iceberg Java FileIO)"
Not asking for a search-and-replace to opendal here, just to drop the backend name from these five comments so they describe the behavior rather than an implementation detail that's expected to change.

Reworded all five spots. Left the non-Iceberg V1 Parquet/CSV comments

parthchandra · 2026-07-31T17:23:23Z

Looking into the iceberg 1.8.1 ci failure

mbutrovich · 2026-07-31T17:31:10Z

+        if (scanExec.output.exists(_.isMetadataCol)) {
+          return withFallbackReason(
+            scanExec,
+            "Metadata columns are not supported for CSV V2 scans")
+        }


Not asking for a change yet, but is this reachable? There's no test for it, and it's not clear to me if Spark can request metadata columns from CSV scans.

I think it might be dead code.

mbutrovich · 2026-07-31T20:18:37Z

I found a bug that this PR exposes where self-joins show that we're not properly disambiguating two scans on the same table with different projections. It's the cause of the 1.8.1 failure on Spark 3.4.

mbutrovich · 2026-07-31T21:04:13Z

I found a bug that this PR exposes where self-joins show that we're not properly disambiguating two scans on the same table with different projections. It's the cause of the 1.8.1 failure on Spark 3.4.

Opened a PR against main to resolve the 1.8.1 issue. Tested the fix on this PR branch and it seemed to resolve the issue. #5180

mbutrovich

Approved pending CI, thanks @parthchandra and also @hsiang-c for some of the upstream work to enable this!

parthchandra · 2026-08-01T01:10:06Z

Merged. Thank you @mbutrovich @hsiang-c !

parthchandra force-pushed the iceberg-metadata-columns branch 2 times, most recently from d23bf63 to ec21656 Compare June 29, 2026 20:56

parthchandra changed the title ~~feat: support iceberg metadata columns _file, _partition~~ feat: [iceberg] support iceberg metadata columns _file, _partition Jun 29, 2026

parthchandra mentioned this pull request Jun 30, 2026

feat: add support for _partition metadata column apache/iceberg-rust#2668

Merged

parthchandra force-pushed the iceberg-metadata-columns branch 2 times, most recently from 50fb6b3 to 0e34cf1 Compare July 6, 2026 18:23

parthchandra force-pushed the iceberg-metadata-columns branch from cf7b59f to 82d7c43 Compare July 9, 2026 16:54

hsiang-c reviewed Jul 9, 2026

View reviewed changes

Comment thread spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala

hsiang-c reviewed Jul 10, 2026

View reviewed changes

Comment thread dev/diffs/iceberg/1.11.0.diff

This was referenced Jul 10, 2026

feat: add support for _spec_id metadata column apache/iceberg-rust#2695

Merged

feat: add support for _pos metadata column apache/iceberg-rust#2746

Merged

parthchandra force-pushed the iceberg-metadata-columns branch from e56d51d to 35f615b Compare July 20, 2026 23:43

parthchandra force-pushed the iceberg-metadata-columns branch 3 times, most recently from 7174036 to d5265be Compare July 27, 2026 00:02

hsiang-c reviewed Jul 30, 2026

View reviewed changes

hsiang-c approved these changes Jul 30, 2026

View reviewed changes

feat: [iceberg] support iceberg metadata columns _pos, _spec, _file, …

d062dd1

…_partition

parthchandra force-pushed the iceberg-metadata-columns branch from a6a8d3b to d062dd1 Compare July 30, 2026 20:28

parthchandra changed the title ~~feat: [iceberg] support iceberg metadata columns _file, _partition~~ feat: [iceberg] support iceberg metadata columns _pos, _spec, _file, and _partition Jul 30, 2026

parthchandra marked this pull request as ready for review July 30, 2026 20:37

parthchandra requested a review from andygrove July 30, 2026 20:37

parthchandra requested review from hsiang-c and mbutrovich July 30, 2026 20:37

mbutrovich changed the title ~~feat: [iceberg] support iceberg metadata columns _pos, _spec, _file, and _partition~~ feat: support Iceberg metadata columns _pos, _spec, _file, and _partition Jul 30, 2026

hsiang-c reviewed Jul 30, 2026

View reviewed changes

parthchandra added the run-iceberg-tests label Jul 30, 2026

mbutrovich reviewed Jul 30, 2026

View reviewed changes

address review comments

e80b37d

parthchandra commented Jul 31, 2026

View reviewed changes

parthchandra added 2 commits July 30, 2026 19:53

fix test

53487f4

fix

b2aaae7

mbutrovich reviewed Jul 31, 2026

View reviewed changes

mbutrovich requested changes Jul 31, 2026

View reviewed changes

parthchandra added 2 commits July 31, 2026 09:44

mbutrovich reviewed Jul 31, 2026

View reviewed changes

mbutrovich mentioned this pull request Jul 31, 2026

fix: disambiguate Iceberg scans that share a metadata_location #5180

Merged

mbutrovich added this to the 1.0.0 milestone Jul 31, 2026

Merge branch 'main' into iceberg-metadata-columns

7219617

mbutrovich approved these changes Jul 31, 2026

View reviewed changes

parthchandra merged commit 60413eb into apache:main Aug 1, 2026
79 checks passed

		@@ -1,5 +1,5 @@
		diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
		index db659dc06b..78f5e3440f 100644

		-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not
		-- supported in Spark 4.0. Covered in CometIcebergNativeSuite instead.

Uh oh!

Conversation

parthchandra commented Jun 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

How are these changes tested?

Uh oh!

Uh oh!

Uh oh!

hsiang-c commented Jul 21, 2026

Uh oh!

parthchandra commented Jul 21, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

parthchandra Jul 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hsiang-c left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mbutrovich left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

parthchandra left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

parthchandra commented Jun 29, 2026 •

edited

Loading

parthchandra Jul 30, 2026 •

edited

Loading