This is an automated email from the ASF dual-hosted git repository.
parthchandra pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 60413eb52c feat: support Iceberg metadata columns _pos, _spec, _file,
and _partition (#4752)
60413eb52c is described below
commit 60413eb52c1f6dda98965e328598ac3b90948b6a
Author: Parth Chandra <[email protected]>
AuthorDate: Fri Jul 31 18:10:00 2026 -0700
feat: support Iceberg metadata columns _pos, _spec, _file, and _partition
(#4752)
* feat: [iceberg] support iceberg metadata columns _pos, _spec, _file,
_partition
---
dev/diffs/iceberg/1.10.0.diff | 14 +
dev/diffs/iceberg/1.11.0.diff | 14 +
dev/diffs/iceberg/1.8.1.diff | 14 +
dev/diffs/iceberg/1.9.1.diff | 14 +
native/Cargo.lock | 4 +-
native/Cargo.toml | 4 +-
.../core/src/execution/operators/iceberg_scan.rs | 1 +
native/core/src/execution/planner.rs | 262 +++++++
native/proto/src/proto/operator.proto | 6 +-
.../apache/comet/iceberg/IcebergReflection.scala | 84 ++
.../org/apache/comet/rules/CometScanRule.scala | 87 ++-
.../serde/operator/CometIcebergNativeScan.scala | 291 ++++---
.../iceberg/metadata_column_partition.sql | 854 +++++++++++++++++++++
.../org/apache/comet/CometIcebergNativeSuite.scala | 670 +++++++++++++++-
14 files changed, 2159 insertions(+), 160 deletions(-)
diff --git a/dev/diffs/iceberg/1.10.0.diff b/dev/diffs/iceberg/1.10.0.diff
index b7c5ce8a84..775f190e99 100644
--- a/dev/diffs/iceberg/1.10.0.diff
+++ b/dev/diffs/iceberg/1.10.0.diff
@@ -1519,6 +1519,20 @@ index 4c1a509591..b22f3f9ad9 100644
.enableHiveSupport()
.getOrCreate();
+diff --git
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+index 893f9931cf..17e5fef217 100644
+---
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
++++
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+@@ -460,7 +460,8 @@ public abstract class SparkRowLevelOperationsTestBase
extends ExtensionsTestBase
+
+ protected void assertAllBatchScansVectorized(SparkPlan plan) {
+ List<SparkPlan> batchScans = SparkPlanUtil.collectBatchScans(plan);
+-
assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar);
++ // When Comet is enabled, its native scan replaces BatchScanExec nodes
entirely
++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar);
+ }
+
+ protected void createTableWithDeleteGranularity(
diff --git
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
index ecf9e6f8a5..23dbcf7af2 100644
---
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
diff --git a/dev/diffs/iceberg/1.11.0.diff b/dev/diffs/iceberg/1.11.0.diff
index ca56ab9b93..a8dc4d3150 100644
--- a/dev/diffs/iceberg/1.11.0.diff
+++ b/dev/diffs/iceberg/1.11.0.diff
@@ -57,6 +57,20 @@ index f766fbb79a..c5e31185a9 100644
.config(TestBase.DISABLE_UI)
.enableHiveSupport()
.getOrCreate();
+diff --git
a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+index b5d6415763..0763a723d3 100644
+---
a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
++++
b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+@@ -448,7 +448,8 @@ public abstract class SparkRowLevelOperationsTestBase
extends ExtensionsTestBase
+
+ protected void assertAllBatchScansVectorized(SparkPlan plan) {
+ List<SparkPlan> batchScans = SparkPlanUtil.collectBatchScans(plan);
+-
assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar);
++ // When Comet is enabled, its native scan replaces BatchScanExec nodes
entirely
++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar);
+ }
+
+ protected void createTableWithDeleteGranularity(
diff --git
a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/DeleteOrphanFilesBenchmark.java
b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/DeleteOrphanFilesBenchmark.java
index 3fd84553f0..350b4a7561 100644
---
a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/DeleteOrphanFilesBenchmark.java
diff --git a/dev/diffs/iceberg/1.8.1.diff b/dev/diffs/iceberg/1.8.1.diff
index 7c951c7f5b..7456bb745e 100644
--- a/dev/diffs/iceberg/1.8.1.diff
+++ b/dev/diffs/iceberg/1.8.1.diff
@@ -78,6 +78,20 @@ index 4f137f5b8d..145ef2d00c 100644
.getOrCreate();
SparkTestBase.catalog =
+diff --git
a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+index 5fa76f9da6..9184aaba83 100644
+---
a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
++++
b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+@@ -452,7 +452,8 @@ public abstract class SparkRowLevelOperationsTestBase
extends SparkExtensionsTes
+
+ protected void assertAllBatchScansVectorized(SparkPlan plan) {
+ List<SparkPlan> batchScans = SparkPlanUtil.collectBatchScans(plan);
+-
assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar);
++ // When Comet is enabled, its native scan replaces BatchScanExec nodes
entirely
++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar);
+ }
+
+ protected void createTableWithDeleteGranularity(
diff --git
a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
index 55a413063e..a1895c2faf 100644
---
a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
diff --git a/dev/diffs/iceberg/1.9.1.diff b/dev/diffs/iceberg/1.9.1.diff
index e086b43a30..32c4e693fc 100644
--- a/dev/diffs/iceberg/1.9.1.diff
+++ b/dev/diffs/iceberg/1.9.1.diff
@@ -1631,6 +1631,20 @@ index 578845e3da..3cc1598035 100644
.enableHiveSupport()
.getOrCreate();
+diff --git
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+index b5d6415763..0763a723d3 100644
+---
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
++++
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java
+@@ -448,7 +448,8 @@ public abstract class SparkRowLevelOperationsTestBase
extends ExtensionsTestBase
+
+ protected void assertAllBatchScansVectorized(SparkPlan plan) {
+ List<SparkPlan> batchScans = SparkPlanUtil.collectBatchScans(plan);
+-
assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar);
++ // When Comet is enabled, its native scan replaces BatchScanExec nodes
entirely
++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar);
+ }
+
+ protected void createTableWithDeleteGranularity(
diff --git
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
index ecf9e6f8a5..23dbcf7af2 100644
---
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java
diff --git a/native/Cargo.lock b/native/Cargo.lock
index a231b50010..a920991922 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -3497,7 +3497,7 @@ dependencies = [
[[package]]
name = "iceberg"
version = "0.10.0"
-source =
"git+https://github.com/apache/iceberg-rust?rev=4532da8#4532da80f3930cdbda1cf2effe05b50165da9875";
+source =
"git+https://github.com/apache/iceberg-rust?rev=3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7#3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7";
dependencies = [
"aes-gcm",
"anyhow",
@@ -3553,7 +3553,7 @@ dependencies = [
[[package]]
name = "iceberg-storage-opendal"
version = "0.10.0"
-source =
"git+https://github.com/apache/iceberg-rust?rev=4532da8#4532da80f3930cdbda1cf2effe05b50165da9875";
+source =
"git+https://github.com/apache/iceberg-rust?rev=3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7#3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7";
dependencies = [
"anyhow",
"async-trait",
diff --git a/native/Cargo.toml b/native/Cargo.toml
index fd7d7d3a66..e41b3a1265 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -58,8 +58,8 @@ 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 = "4532da8" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust";,
rev = "4532da8", features = ["opendal-memory", "opendal-fs", "opendal-s3",
"opendal-gcs", "opendal-oss", "opendal-azdls"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust";, rev =
"3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust";,
rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7", features = ["opendal-memory",
"opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
reqsign-core = "3"
[profile.release]
diff --git a/native/core/src/execution/operators/iceberg_scan.rs
b/native/core/src/execution/operators/iceberg_scan.rs
index d77b208964..e727294fd9 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -660,6 +660,7 @@ mod tests {
partition: None,
partition_spec: None,
name_mapping: None,
+ unified_partition_type: None,
case_sensitive: false,
key_metadata: None,
}
diff --git a/native/core/src/execution/planner.rs
b/native/core/src/execution/planner.rs
index 39e7ccc2e5..5accd31609 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -3835,6 +3835,77 @@ 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, matching Iceberg Java's
+ // Partitioning.buildPartitionProjectionType()/iceberg-rust's own
+ // compute_unified_partition_type(): process specs by spec_id descending
(so the newest
+ // spec's field name wins when two specs share a field id), then sort the
merged fields
+ // ascending by field id. The ascending sort matters beyond cosmetics: it
fixes the
+ // physical Arrow struct field order for `_partition`, which must match
the field order
+ // Spark's analyzer bound `_partition.<field>` accesses to (also computed
via
+ // Partitioning.partitionType(), spec-descending + ascending-sort) -- a
mismatch would
+ // read the wrong value under the right field name instead of erroring.
+ //
+ // partition_type_pool is index-aligned with partition_spec_pool (see the
.proto comment),
+ // so partition_spec_cache[i].spec_id() gives the spec_id for
partition_type_pool[i].
+ //
+ // NOTE: The resolved field types here come from Iceberg Java's
PartitionSpec.partitionType()
+ // (via Scala reflection), which already applied Transform::result_type()
per field. Because
+ // we only have resolved types (not the original Transform), we cannot
replicate one further
+ // nuance of the reference algorithms: preferring a non-void transform's
type over an older
+ // void transform's for the same field id. In practice this only affects a
field whose
+ // partition source column was later dropped from the schema, which Comet
already filters out
+ // upstream (see the "unknown type" filtering in serializePartitionData).
+ let unified_partition_type = {
+ let mut indexed_types = proto_common
+ .partition_type_pool
+ .iter()
+ .enumerate()
+ .map(|(idx, type_json)| {
+ let struct_type =
serde_json::from_str::<iceberg::spec::StructType>(type_json)
+ .map_err(|e| {
+ ExecutionError::GeneralError(format!(
+ "Failed to deserialize partition type JSON from
pool: {e}"
+ ))
+ })?;
+ // Recover spec_id from the index-aligned spec JSON's
"spec-id" field directly,
+ // rather than from partition_spec_cache: a spec that uses a
transform iceberg-rust
+ // doesn't recognize (e.g. forward-compatibility tests) fails
to deserialize into a
+ // PartitionSpec, but its "spec-id" is still present in the
JSON and its partition
+ // type entry has no usable fields anyway. Falling back to idx
keeps ordering
+ // deterministic if the field is somehow absent.
+ let spec_id = proto_common
+ .partition_spec_pool
+ .get(idx)
+ .and_then(|json|
serde_json::from_str::<serde_json::Value>(json).ok())
+ .and_then(|v|
v.get("spec-id").and_then(serde_json::Value::as_i64))
+ .map(|id| id as i32)
+ .unwrap_or(idx as i32);
+ Ok((spec_id, struct_type))
+ })
+ .collect::<Result<Vec<_>, ExecutionError>>()?;
+
+ indexed_types.sort_by_key(|(spec_id, _)| std::cmp::Reverse(*spec_id));
+
+ let mut seen_field_ids = std::collections::HashSet::new();
+ let mut struct_fields: Vec<iceberg::spec::NestedFieldRef> = Vec::new();
+
+ for (_, struct_type) in &indexed_types {
+ for field in struct_type.fields() {
+ if seen_field_ids.insert(field.id) {
+ struct_fields.push(Arc::clone(field));
+ }
+ }
+ }
+
+ struct_fields.sort_by_key(|f| f.id);
+
+ 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| {
@@ -3939,6 +4010,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(),
@@ -3953,6 +4032,7 @@ fn parse_file_scan_tasks_from_common(
partition,
partition_spec,
name_mapping,
+ unified_partition_type: unified_partition_type_for_task,
case_sensitive: false,
// Plaintext StandardKeyMetadata forwarded verbatim from the
JVM; decoded by
// iceberg-rust with no KMS unwrap. None for unencrypted data
files.
@@ -4383,6 +4463,7 @@ mod tests {
use crate::execution::operators::ExecutionError;
use crate::execution::planner::literal_to_array_ref;
+ use crate::execution::planner::parse_file_scan_tasks_from_common;
use crate::parquet::parquet_support::SparkParquetOptions;
use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
use datafusion_comet_proto::spark_expression::expr::ExprStruct;
@@ -5455,4 +5536,185 @@ 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"
+ );
+ }
+
+ #[test]
+ fn test_unified_partition_type_merges_specs_by_descending_spec_id() {
+ use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Type};
+
+ let iceberg_schema = iceberg::spec::Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![
+ NestedField::required(1, "id",
Type::Primitive(PrimitiveType::Int)).into(),
+ NestedField::required(2, "region",
Type::Primitive(PrimitiveType::String)).into(),
+ NestedField::required(3, "category",
Type::Primitive(PrimitiveType::String)).into(),
+ ])
+ .build()
+ .expect("schema");
+ let schema_arc = Arc::new(iceberg_schema);
+
+ // Spec 0 (older): field 1000 named "region_old".
+ let spec0 = PartitionSpec::builder(Arc::clone(&schema_arc))
+ .with_spec_id(0)
+ .add_unbound_field(iceberg::spec::UnboundPartitionField {
+ source_id: 2,
+ field_id: Some(1000),
+ name: "region_old".to_string(),
+ transform: iceberg::spec::Transform::Identity,
+ })
+ .expect("add field")
+ .build()
+ .expect("build spec0");
+
+ // Spec 1 (newer): same field id 1000 renamed to "region_new", plus a
new field 2000.
+ let spec1 = PartitionSpec::builder(Arc::clone(&schema_arc))
+ .with_spec_id(1)
+ .add_unbound_field(iceberg::spec::UnboundPartitionField {
+ source_id: 2,
+ field_id: Some(1000),
+ name: "region_new".to_string(),
+ transform: iceberg::spec::Transform::Identity,
+ })
+ .expect("add field")
+ .add_unbound_field(iceberg::spec::UnboundPartitionField {
+ source_id: 3,
+ field_id: Some(2000),
+ name: "category".to_string(),
+ transform: iceberg::spec::Transform::Identity,
+ })
+ .expect("add field")
+ .build()
+ .expect("build spec1");
+
+ let spec0_type_json =
+
serde_json::to_string(&spec0.partition_type(&schema_arc).expect("partition_type"))
+ .expect("serialize");
+ let spec1_type_json =
+
serde_json::to_string(&spec1.partition_type(&schema_arc).expect("partition_type"))
+ .expect("serialize");
+ let spec0_json = serde_json::to_string(&spec0).expect("serialize
spec0");
+ let spec1_json = serde_json::to_string(&spec1).expect("serialize
spec1");
+
+ let schema_json =
serde_json::to_string(schema_arc.as_ref()).expect("serialize schema");
+
+ // Pool insertion order deliberately does NOT match spec_id order: the
newer spec
+ // (spec_id 1) is inserted first, at index 0, and the older spec
(spec_id 0) second, at
+ // index 1. If the merge relied on pool-insertion order instead of
spec_id, this would
+ // produce the wrong result (region_old kept instead of region_new).
+ let proto_common = spark_operator::IcebergScanCommon {
+ schema_pool: vec![schema_json],
+ partition_type_pool: vec![spec1_type_json, spec0_type_json],
+ partition_spec_pool: vec![spec1_json, spec0_json],
+ project_field_ids_pool: vec![spark_operator::ProjectFieldIdList {
+ field_ids:
vec![iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION],
+ }],
+ ..Default::default()
+ };
+
+ let proto_task = spark_operator::IcebergFileScanTask {
+ data_file_path: "file:///tmp/data.parquet".to_string(),
+ file_size_in_bytes: 100,
+ schema_idx: 0,
+ partition_spec_idx: Some(0),
+ project_field_ids_idx: 0,
+ ..Default::default()
+ };
+
+ let tasks =
+ parse_file_scan_tasks_from_common(&proto_common,
&[proto_task]).expect("parse tasks");
+ assert_eq!(tasks.len(), 1);
+
+ let unified = tasks[0]
+ .unified_partition_type
+ .as_ref()
+ .expect("unified_partition_type must be set when _partition is
projected");
+
+ let fields = unified.fields();
+ assert_eq!(
+ fields.len(),
+ 2,
+ "expected fields 1000 and 2000, got {fields:?}"
+ );
+ // Ascending by field id: 1000 before 2000.
+ assert_eq!(fields[0].id, 1000);
+ assert_eq!(fields[1].id, 2000);
+ // Newest spec (spec_id 1) wins the name for field 1000, despite being
inserted into the
+ // pool before spec_id 0's entry.
+ assert_eq!(fields[0].name, "region_new");
+ assert_eq!(fields[1].name, "category");
+ }
+
+ #[test]
+ fn test_unified_partition_type_tolerates_unparseable_spec() {
+ // Regression for
TestForwardCompatibility.testSparkCanReadUnknownTransform: a spec that
+ // uses a transform iceberg-rust doesn't recognize fails to
deserialize into a
+ // PartitionSpec, but its partition field is filtered out on the Scala
side (unknown type),
+ // so its partition_type_pool entry is an empty struct. Recovering
spec_id must not require
+ // the full spec to parse -- the merge must succeed (with no fields)
rather than erroring.
+ let schema_json = serde_json::to_string(
+ &iceberg::spec::Schema::builder()
+ .with_schema_id(0)
+ .with_fields(vec![iceberg::spec::NestedField::required(
+ 1,
+ "id",
+
iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::Int),
+ )
+ .into()])
+ .build()
+ .expect("schema"),
+ )
+ .expect("serialize schema");
+
+ // A spec JSON whose transform iceberg-rust cannot deserialize, paired
with an empty type
+ // entry (as Scala produces once the unknown-type field is filtered).
+ let unparseable_spec_json =
r#"{"spec-id":7,"fields":[{"source-id":1,"field-id":1000,"name":"x","transform":"totally_unknown[9]"}]}"#;
+ let empty_type_json = r#"{"type":"struct","fields":[]}"#;
+
+ let proto_common = spark_operator::IcebergScanCommon {
+ schema_pool: vec![schema_json],
+ partition_type_pool: vec![empty_type_json.to_string()],
+ partition_spec_pool: vec![unparseable_spec_json.to_string()],
+ project_field_ids_pool: vec![spark_operator::ProjectFieldIdList {
+ field_ids:
vec![iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION],
+ }],
+ ..Default::default()
+ };
+
+ let proto_task = spark_operator::IcebergFileScanTask {
+ data_file_path: "file:///tmp/data.parquet".to_string(),
+ file_size_in_bytes: 100,
+ schema_idx: 0,
+ partition_spec_idx: Some(0),
+ project_field_ids_idx: 0,
+ ..Default::default()
+ };
+
+ let tasks = parse_file_scan_tasks_from_common(&proto_common,
&[proto_task])
+ .expect("parse tasks must not error on an unparseable spec");
+ assert_eq!(tasks.len(), 1);
+ assert!(
+ tasks[0]
+ .unified_partition_type
+ .as_ref()
+ .map(|t| t.fields().is_empty())
+ .unwrap_or(true),
+ "unified partition type should have no fields for an
all-unknown-transform spec"
+ );
+ }
}
diff --git a/native/proto/src/proto/operator.proto
b/native/proto/src/proto/operator.proto
index dae7eed9fb..d9515edc99 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -261,6 +261,11 @@ message IcebergScanCommon {
// Deduplication pools (must contain all entries for cross-partition
deduplication)
repeated string schema_pool = 4;
+ // Index-aligned with partition_spec_pool: entry i is the StructType JSON
for the spec at
+ // partition_spec_pool[i]. Native uses this pairing to recover each entry's
spec_id when
+ // merging partition types across historical specs (see
parse_file_scan_tasks_from_common
+ // in planner.rs), which partition_type_pool cannot express on its own since
it is only
+ // resolved field types, not specs.
repeated string partition_type_pool = 5;
repeated string partition_spec_pool = 6;
repeated string name_mapping_pool = 7;
@@ -332,7 +337,6 @@ message IcebergFileScanTask {
// Indices into IcebergScan deduplication pools
uint32 schema_idx = 15;
- optional uint32 partition_type_idx = 16;
optional uint32 partition_spec_idx = 17;
optional uint32 name_mapping_idx = 18;
uint32 project_field_ids_idx = 19;
diff --git
a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 72e885e834..f867d4bc9b 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -52,6 +52,8 @@ object IcebergReflection extends Logging {
val SPARK_BATCH_QUERY_SCAN =
"org.apache.iceberg.spark.source.SparkBatchQueryScan"
val SPARK_STAGED_SCAN = "org.apache.iceberg.spark.source.SparkStagedScan"
val SPARK_SCHEMA_UTIL = "org.apache.iceberg.spark.SparkSchemaUtil"
+ val TABLE = "org.apache.iceberg.Table"
+ val PARTITIONING = "org.apache.iceberg.Partitioning"
}
/**
@@ -66,6 +68,37 @@ object IcebergReflection extends Logging {
def isIcebergScanClass(name: String): Boolean =
ICEBERG_SCAN_CLASSES.contains(name)
+ // Iceberg FileIO implementations whose backing storage Comet's native
reader can reach.
+ // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache)
are not compatible
+ // because Comet's native reader bypasses Java FileIO entirely.
+ val COMPATIBLE_FILE_IO_CLASSES: 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")
+
+ // Prefix of the EncryptingFileIO family. An encrypted table's io() is not
the bare
+ // EncryptingFileIO but a nested variant chosen from the wrapped delegate's
capabilities
+ // (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 through iceberg-rust's own storage layer, so any
EncryptingFileIO
+ // variant is compatible.
+ private val ENCRYPTING_FILE_IO_PREFIX =
"org.apache.iceberg.encryption.EncryptingFileIO"
+
+ /**
+ * True if `fileIO` is a FileIO whose backing storage Comet's native reader
can reach. Matches
+ * on `fileIO`'s own class hierarchy (via [[classNameInHierarchy]]) rather
than its exact leaf
+ * class, so a subclass that only adds metrics/retry/credential-routing on
top of a known-
+ * compatible FileIO (e.g. a custom S3FileIO subclass) still matches,
instead of silently
+ * falling back to Spark.
+ */
+ def isCompatibleFileIO(fileIO: Any): Boolean =
+ classNameInHierarchy(fileIO.getClass, COMPATIBLE_FILE_IO_CLASSES) ||
+ fileIO.getClass.getName.startsWith(ENCRYPTING_FILE_IO_PREFIX)
+
/**
* Iceberg content types.
*/
@@ -125,6 +158,23 @@ object IcebergReflection extends Logging {
None
}
+ /**
+ * True if `clazz` or any of its superclasses has a name in `names`. Walks
the already-loaded
+ * class object's own hierarchy, so unlike [[loadClass]] it never risks a
+ * `ClassNotFoundException` for a candidate name that isn't on this JVM's
classpath (e.g.
+ * checking for a GCS/Azure FileIO class when only iceberg-aws is bundled).
+ */
+ 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
+ }
+
/**
* Extracts file location from Iceberg ContentFile, handling both location()
and path().
*
@@ -461,6 +511,40 @@ object IcebergReflection extends Logging {
}
}
+ /**
+ * Validates that the table's unified partition type can be computed -- the
merge of every
+ * historical partition spec, which is what the `_partition` metadata column
projects.
+ *
+ * Iceberg Java's `Partitioning.partitionType(table)` runs the same
cross-spec compatibility
+ * check that iceberg-rust does natively: a V1 table does not guarantee
partition field ids are
+ * unique across specs, so two specs can bind the same id to incompatible
source/transform
+ * pairs, which cannot be merged into one struct field. iceberg-rust returns
a DataInvalid error
+ * in that case, but only at scan time -- too late for Comet to fall back.
Calling the Java
+ * check here, at plan time, lets `CometScanRule` fall back to Spark instead
of failing inside
+ * the native reader.
+ *
+ * Returns None when the unified type is computable, or Some(reason) when it
is not -- either
+ * the specs conflict or the reflection call itself failed. Both mean Comet
cannot safely serve
+ * `_partition`, so both map to a fallback.
+ */
+ def validateUnifiedPartitionType(table: Any): Option[String] = {
+ try {
+ val tableClass = loadClass(ClassNames.TABLE)
+ val partitioningClass = loadClass(ClassNames.PARTITIONING)
+ partitioningClass
+ .getMethod("partitionType", tableClass)
+ .invoke(null, table.asInstanceOf[AnyRef])
+ None
+ } catch {
+ // A conflict surfaces as the ValidationException thrown by
partitionType(), wrapped by
+ // reflection in InvocationTargetException; unwrap it for a meaningful
reason.
+ case e: java.lang.reflect.InvocationTargetException =>
+ Some(Option(e.getCause).getOrElse(e).getMessage)
+ case e: Exception =>
+ Some(e.getMessage)
+ }
+ }
+
/**
* Gets the table metadata from an Iceberg table.
*
diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
index 5e451b382a..ca917335da 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -88,12 +88,6 @@ case class CometScanRule(session: SparkSession)
case _ => false
}
- def metadataCols(plan: SparkPlan): Seq[String] = {
- plan.expressions.collect {
- case a: Attribute if a.isMetadataCol => a.name
- }
- }
-
def isIcebergMetadataTable(scanExec: BatchScanExec): Boolean = {
// List of Iceberg metadata tables:
//
https://iceberg.apache.org/docs/latest/spark-queries/#inspecting-tables
@@ -130,11 +124,6 @@ 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 =>
- withFallbackReason(
- scan,
- s"Metadata column(s) ${metadataCols(scan).mkString(", ")} is not
supported")
-
// data source V1
case scanExec: FileSourceScanExec =>
transformV1Scan(fullPlan, scanExec)
@@ -154,6 +143,12 @@ case class CometScanRule(session: SparkSession)
}
private def transformV1Scan(plan: SparkPlan, scanExec: FileSourceScanExec):
SparkPlan = {
+ val metadataColNames = metadataCols(scanExec)
+ if (metadataColNames.nonEmpty) {
+ return withFallbackReason(
+ scanExec,
+ s"Metadata column(s) ${metadataColNames.mkString(", ")} is not
supported")
+ }
// On Spark 3.4, injectQueryStageOptimizerRule is unavailable, so
// CometPlanAdaptiveDynamicPruningFilters cannot run. Fall back this scan
to Spark so that
@@ -302,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)
@@ -361,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 " +
@@ -451,15 +472,17 @@ 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 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.
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 IcebergReflection.isCompatibleFileIO(fileIO) =>
true
+ case Some(fileIO) =>
+ fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not
supported by " +
+ "Comet's native reader (bypasses Iceberg Java FileIO)"
+ false
case None =>
fallbackReasons += "Could not check FileIO compatibility"
false
@@ -603,6 +626,25 @@ case class CometScanRule(session: SparkSession)
false
}
+ // The `_partition` metadata column projects the table's unified
partition type -- the
+ // merge of all historical specs. iceberg-rust computes that merge at
scan time and errors
+ // (DataInvalid) when specs bind one field id to incompatible
source/transform pairs (only
+ // possible in V1 tables, which do not keep partition field ids unique
across specs). That
+ // error would fail the native scan with no chance to fall back, so
when `_partition` is
+ // projected we run Iceberg Java's equivalent check here and fall back
if it cannot merge.
+ 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
+ }
+
// Get filter expressions for complex predicates check
val filterExpressionsOpt =
IcebergReflection.getFilterExpressions(scanExec.scan)
@@ -788,6 +830,7 @@ case class CometScanRule(session: SparkSession)
if (schemaSupported && fileIOCompatible && formatVersionSupported &&
defaultValuesSupported && schemaTypesSupported &&
encryptionKeyLengthSupported &&
taskValidation.allParquet && allSupportedFilesystems &&
partitionTypesSupported &&
+ unifiedPartitionTypeSupported &&
complexTypePredicatesSupported && transformFunctionsSupported &&
deleteFileTypesSupported && dppSubqueriesSupported) {
CometBatchScanExec(
@@ -824,6 +867,12 @@ case class CometScanRule(session: SparkSession)
case _ => false
}
+ private def metadataCols(plan: SparkPlan): Seq[String] = {
+ plan.expressions.collect {
+ case a: Attribute if a.isMetadataCol => a.name
+ }
+ }
+
private def isSchemaSupported(scanExec: FileSourceScanExec, r:
HadoopFsRelation): Boolean = {
val fallbackReasons = new ListBuffer[String]()
val typeChecker = CometScanTypeChecker()
diff --git
a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index f61f1368b2..bb81fd69ca 100644
---
a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++
b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -85,6 +85,20 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
Constants.Operations.GT ->
OperatorOuterClass.IcebergPredicateOperator.GreaterThan,
Constants.Operations.GT_EQ ->
OperatorOuterClass.IcebergPredicateOperator.GreaterThanOrEq)
+ // 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_POS = i32::MAX - 2 (2147483645)
+ // RESERVED_FIELD_ID_SPEC_ID = i32::MAX - 4 (2147483643)
+ // 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),
+ "_pos" -> (Int.MaxValue - 2),
+ "_spec_id" -> (Int.MaxValue - 4),
+ "_partition" -> (Int.MaxValue - 5))
+
/**
* Wraps an Iceberg partition value (a typed primitive) in a PartitionValue.
The value encoding
* is shared with predicate literals via [[icebergLiteralToProto]].
@@ -343,7 +357,6 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
fileScanTaskClass: Class[_],
taskBuilder: OperatorOuterClass.IcebergFileScanTask.Builder,
commonBuilder: OperatorOuterClass.IcebergScanCommon.Builder,
- partitionTypeToPoolIndex: mutable.HashMap[String, Int],
partitionSpecToPoolIndex: mutable.HashMap[String, Int],
partitionDataToPoolIndex: mutable.HashMap[String, Int]): Unit = {
try {
@@ -351,27 +364,96 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
val spec = specMethod.invoke(task)
if (spec != null) {
- // Deduplicate partition spec
- try {
- val partitionSpecParserClass =
-
IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER)
- val toJsonMethod = partitionSpecParserClass.getMethod(
- "toJson",
-
IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC))
- val partitionSpecJson = toJsonMethod
- .invoke(null, spec)
- .asInstanceOf[String]
-
- val specIdx = partitionSpecToPoolIndex.getOrElseUpdate(
- partitionSpecJson, {
- val idx = partitionSpecToPoolIndex.size
- commonBuilder.addPartitionSpecPool(partitionSpecJson)
- idx
- })
- taskBuilder.setPartitionSpecIdx(specIdx)
- } catch {
- case e: Exception =>
- logWarning(s"Failed to serialize partition spec to JSON:
${e.getMessage}")
+ // Get the partition type/schema from the spec. Needed regardless of
whether this task
+ // ends up value-less below.
+ val partitionTypeMethod = spec.getClass.getMethod("partitionType")
+ val partitionType = partitionTypeMethod.invoke(spec)
+ val fieldsMethod = partitionType.getClass.getMethod("fields")
+ val fields = fieldsMethod
+ .invoke(partitionType)
+ .asInstanceOf[java.util.List[_]]
+
+ // Helper to get field type string (shared by both type and data
serialization)
+ def getFieldType(field: Any): String = {
+ val typeMethod = field.getClass.getMethod("type")
+ typeMethod.invoke(field).toString
+ }
+
+ // Filter out fields with unknown types (dropped partition fields).
+ // Unknown type fields represent partition columns that have been
dropped
+ // from the schema. Per the Iceberg spec, unknown type fields are not
+ // stored in data files and iceberg-rust doesn't support deserializing
+ // them. Since these columns are dropped, we don't need to expose their
+ // partition values when reading.
+ val fieldsJson = fields.asScala.flatMap { field =>
+ val fieldTypeStr = getFieldType(field)
+
+ // Skip fields with unknown type (dropped partition columns)
+ if (fieldTypeStr == IcebergReflection.TypeNames.UNKNOWN) {
+ None
+ } else {
+ val fieldIdMethod = field.getClass.getMethod("fieldId")
+ val fieldId = fieldIdMethod.invoke(field).asInstanceOf[Int]
+
+ val nameMethod = field.getClass.getMethod("name")
+ val fieldName = nameMethod.invoke(field).asInstanceOf[String]
+
+ val isOptionalMethod = field.getClass.getMethod("isOptional")
+ val isOptional =
+ isOptionalMethod.invoke(field).asInstanceOf[Boolean]
+ val required = !isOptional
+
+ Some(
+ ("id" -> fieldId) ~
+ ("name" -> fieldName) ~
+ ("required" -> required) ~
+ ("type" -> fieldTypeStr))
+ }
+ }.toList
+
+ // Serializes the file's real partition spec plus its paired
partition-type-pool entry,
+ // then points the task at that pool index. Only invoked for tasks
that actually carry
+ // partition values: a value-less task (partition evolution) takes the
empty-spec path
+ // below instead, so this avoids the toJson reflection call and pool
intern that would
+ // otherwise be computed and then immediately overwritten.
+ //
+ // The spec and type pools are index-aligned: native correlates the
two by index to
+ // recover, per spec, the spec_id needed to merge partition types
across historical specs
+ // in the same order Iceberg Java's
Partitioning.partitionType()/iceberg-rust's
+ // compute_unified_partition_type do (newest spec_id first); see
+ // parse_file_scan_tasks_from_common in planner.rs. Every path that
adds a spec-pool entry
+ // adds its type-pool entry in the same block to keep them aligned.
+ def serializeRealSpec(): Unit = {
+ try {
+ val partitionSpecParserClass =
+
IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER)
+ val toJsonMethod = partitionSpecParserClass.getMethod(
+ "toJson",
+
IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC))
+ val partitionSpecJson = toJsonMethod
+ .invoke(null, spec)
+ .asInstanceOf[String]
+
+ val specIdx = partitionSpecToPoolIndex.getOrElseUpdate(
+ partitionSpecJson, {
+ val idx = partitionSpecToPoolIndex.size
+ commonBuilder.addPartitionSpecPool(partitionSpecJson)
+ // Manually build StructType JSON to match iceberg-rust
expectations.
+ // Using Iceberg's SchemaParser.toJson() would include
schema-level
+ // metadata (e.g., "schema-id") that iceberg-rust's StructType
+ // deserializer rejects. We need pure StructType format:
+ // {"type":"struct","fields":[...]}
+ val partitionTypeJson = compact(
+ render(("type" -> "struct") ~
+ ("fields" -> fieldsJson)))
+ commonBuilder.addPartitionTypePool(partitionTypeJson)
+ idx
+ })
+ taskBuilder.setPartitionSpecIdx(specIdx)
+ } catch {
+ case e: Exception =>
+ logWarning(s"Failed to serialize partition spec to JSON:
${e.getMessage}")
+ }
}
// Get partition data from the task (via file().partition())
@@ -379,84 +461,6 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
val partitionData = partitionMethod.invoke(task)
if (partitionData != null) {
- // Get the partition type/schema from the spec
- val partitionTypeMethod = spec.getClass.getMethod("partitionType")
- val partitionType = partitionTypeMethod.invoke(spec)
-
- // Check if partition type has any fields before serializing
- val fieldsMethod = partitionType.getClass.getMethod("fields")
- val fields = fieldsMethod
- .invoke(partitionType)
- .asInstanceOf[java.util.List[_]]
-
- // Helper to get field type string (shared by both type and data
serialization)
- def getFieldType(field: Any): String = {
- val typeMethod = field.getClass.getMethod("type")
- typeMethod.invoke(field).toString
- }
-
- // Only serialize partition type if there are actual partition fields
- if (!fields.isEmpty) {
- try {
- // Manually build StructType JSON to match iceberg-rust
expectations.
- // Using Iceberg's SchemaParser.toJson() would include
schema-level
- // metadata (e.g., "schema-id") that iceberg-rust's StructType
- // deserializer rejects. We need pure StructType format:
- // {"type":"struct","fields":[...]}
-
- // Filter out fields with unknown types (dropped partition
fields).
- // Unknown type fields represent partition columns that have
been dropped
- // from the schema. Per the Iceberg spec, unknown type fields
are not
- // stored in data files and iceberg-rust doesn't support
deserializing
- // them. Since these columns are dropped, we don't need to
expose their
- // partition values when reading.
- val fieldsJson = fields.asScala.flatMap { field =>
- val fieldTypeStr = getFieldType(field)
-
- // Skip fields with unknown type (dropped partition columns)
- if (fieldTypeStr == IcebergReflection.TypeNames.UNKNOWN) {
- None
- } else {
- val fieldIdMethod = field.getClass.getMethod("fieldId")
- val fieldId = fieldIdMethod.invoke(field).asInstanceOf[Int]
-
- val nameMethod = field.getClass.getMethod("name")
- val fieldName = nameMethod.invoke(field).asInstanceOf[String]
-
- val isOptionalMethod = field.getClass.getMethod("isOptional")
- val isOptional =
- isOptionalMethod.invoke(field).asInstanceOf[Boolean]
- val required = !isOptional
-
- Some(
- ("id" -> fieldId) ~
- ("name" -> fieldName) ~
- ("required" -> required) ~
- ("type" -> fieldTypeStr))
- }
- }.toList
-
- // Only serialize if we have non-unknown fields
- if (fieldsJson.nonEmpty) {
- val partitionTypeJson = compact(
- render(
- ("type" -> "struct") ~
- ("fields" -> fieldsJson)))
-
- val typeIdx = partitionTypeToPoolIndex.getOrElseUpdate(
- partitionTypeJson, {
- val idx = partitionTypeToPoolIndex.size
- commonBuilder.addPartitionTypePool(partitionTypeJson)
- idx
- })
- taskBuilder.setPartitionTypeIdx(typeIdx)
- }
- } catch {
- case e: Exception =>
- logWarning(s"Failed to serialize partition type to JSON:
${e.getMessage}")
- }
- }
-
// Serialize partition data to protobuf for native execution.
// The native execution engine uses partition_data protobuf messages
to
// build a constants_map, which provides partition values to
identity-
@@ -487,25 +491,65 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
}
}.toSeq
- // Only serialize partition data if we have non-unknown fields
- if (partitionValues.nonEmpty) {
- val partitionDataProto = OperatorOuterClass.PartitionData
- .newBuilder()
- .addAllValues(partitionValues.asJava)
- .build()
-
- // Deduplicate by protobuf bytes (use Base64 string as key)
- val partitionDataBytes = partitionDataProto.toByteArray
- val partitionDataKey =
Base64.getEncoder.encodeToString(partitionDataBytes)
-
- val partitionDataIdx = partitionDataToPoolIndex.getOrElseUpdate(
- partitionDataKey, {
- val idx = partitionDataToPoolIndex.size
- commonBuilder.addPartitionDataPool(partitionDataProto)
+ // Native requires a task to carry both a partition spec and
partition data, or neither:
+ // iceberg-rust errors when the unified partition type has fields
but a task is missing
+ // its spec/data. A file written while a partition field was dropped
(partition
+ // evolution) has no partition values, so partitionValues is empty
here.
+ //
+ // For that value-less case we must NOT send the file's real spec:
it may retain a void
+ // field whose id collides with a unified field, which would make
iceberg-rust index the
+ // empty partition data out of range. Instead send an empty-fields
spec that keeps the
+ // real spec id (so _spec_id stays correct) plus empty data, so
native fills every
+ // unified _partition field with null -- matching Spark and the
pre-tightening behaviour.
+ // Pair it with an empty partition-type-pool entry too, keeping the
two pools aligned.
+ //
+ // Only the value-carrying path serializes the real spec
(serializeRealSpec), so the
+ // value-less path never does the reflection/intern work just to
overwrite it.
+ if (partitionValues.isEmpty) {
+ val specId =
spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int]
+ val emptySpecJson =
+ compact(
+ render(("spec-id" -> specId) ~ ("fields" ->
List.empty[org.json4s.JObject])))
+ val emptySpecIdx = partitionSpecToPoolIndex.getOrElseUpdate(
+ emptySpecJson, {
+ val idx = partitionSpecToPoolIndex.size
+ commonBuilder.addPartitionSpecPool(emptySpecJson)
+ val emptyTypeJson = compact(
+ render(("type" -> "struct") ~
+ ("fields" -> List.empty[org.json4s.JObject])))
+ commonBuilder.addPartitionTypePool(emptyTypeJson)
idx
})
- taskBuilder.setPartitionDataIdx(partitionDataIdx)
+ taskBuilder.setPartitionSpecIdx(emptySpecIdx)
+ } else {
+ serializeRealSpec()
}
+
+ // Always send partition data (empty when there are no values) so
native never sees a
+ // spec without data. Native uses it to build the identity-transform
constants_map and,
+ // together with the spec above, the _partition column.
+ val partitionDataProto = OperatorOuterClass.PartitionData
+ .newBuilder()
+ .addAllValues(partitionValues.asJava)
+ .build()
+
+ // Deduplicate by protobuf bytes (use Base64 string as key)
+ val partitionDataBytes = partitionDataProto.toByteArray
+ val partitionDataKey =
Base64.getEncoder.encodeToString(partitionDataBytes)
+
+ val partitionDataIdx = partitionDataToPoolIndex.getOrElseUpdate(
+ partitionDataKey, {
+ val idx = partitionDataToPoolIndex.size
+ commonBuilder.addPartitionDataPool(partitionDataProto)
+ idx
+ })
+ taskBuilder.setPartitionDataIdx(partitionDataIdx)
+ } else {
+ // Defensive: ContentScanTask.partition() returns an empty struct
(never null) for
+ // unpartitioned tables in practice. If it is ever null we cannot
compute values, so
+ // just register the file's real spec (preserving the pre-refactor
behaviour) to keep
+ // _spec_id correct.
+ serializeRealSpec()
}
}
} catch {
@@ -850,7 +894,6 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
// Deduplication structures - map unique values to pool indices
val schemaToPoolIndex = mutable.HashMap[AnyRef, Int]()
- val partitionTypeToPoolIndex = mutable.HashMap[String, Int]()
val partitionSpecToPoolIndex = mutable.HashMap[String, Int]()
val nameMappingToPoolIndex = mutable.HashMap[String, Int]()
val projectFieldIdsToPoolIndex = mutable.HashMap[Seq[Int], Int]()
@@ -1028,14 +1071,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.")
}
}
@@ -1103,7 +1148,6 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
fileScanTaskClass,
taskBuilder,
commonBuilder,
- partitionTypeToPoolIndex,
partitionSpecToPoolIndex,
partitionDataToPoolIndex)
@@ -1145,7 +1189,6 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
// Log deduplication summary
val allPoolSizes = Seq(
schemaToPoolIndex.size,
- partitionTypeToPoolIndex.size,
partitionSpecToPoolIndex.size,
nameMappingToPoolIndex.size,
projectFieldIdsToPoolIndex.size,
diff --git
a/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
new file mode 100644
index 0000000000..9ceda6bb8e
--- /dev/null
+++ b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
@@ -0,0 +1,854 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Native Iceberg scan is unsupported on Spark 4.2: no Iceberg spark-runtime
is published for
+-- 4.2 yet, so the build reuses the 4.0 runtime, which is binary-incompatible
with 4.2 (e.g.
+-- connector.catalog.View became a class). Mirrors the assume(!isSpark42Plus)
guard in
+-- CometIcebergNativeSuite. See
https://github.com/apache/datafusion-comet/issues/4969.
+-- MaxSparkVersion: 4.1
+
+-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog
+-- Config: spark.sql.catalog.test_cat.type=hadoop
+-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test
+-- Config: spark.comet.enabled=true
+-- Config: spark.comet.exec.enabled=true
+-- Config: spark.comet.iceberg.native.enabled=true
+
+-- All `query` assertions below implicitly validate that Comet's partition type
+-- computation (from partition_type_pool JSON) matches Spark's (from Iceberg
Java's
+-- PartitionSpec.partitionType()). If the two diverged,
checkSparkAnswerAndOperator
+-- would detect differing struct values. This covers the "dual computation
path"
+-- cross-validation concern.
+
+-- =============================================================
+-- Setup: Create table with bucket partitioning (format v2 for MoR)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.meta_test
+
+statement
+CREATE TABLE test_cat.db.meta_test (id INT, name STRING, value DOUBLE) USING
iceberg PARTITIONED BY (bucket(4, id)) TBLPROPERTIES ('format-version' = '2')
+
+statement
+INSERT INTO test_cat.db.meta_test VALUES (1, 'alice', 10.0), (2, 'bob', 20.0),
(3, 'charlie', 30.0)
+
+statement
+INSERT INTO test_cat.db.meta_test VALUES (4, 'dave', 40.0), (5, 'eve', 50.0)
+
+-- =============================================================
+-- _file: basic projection (native)
+-- =============================================================
+
+query
+SELECT id, _file FROM test_cat.db.meta_test ORDER BY id
+
+-- Multiple inserts produce multiple files
+query
+SELECT COUNT(DISTINCT _file) > 1 FROM test_cat.db.meta_test
+
+-- _file in a WHERE clause exercises filter pushdown interaction
+query
+SELECT id, name FROM test_cat.db.meta_test WHERE _file LIKE '%.parquet' ORDER
BY id
+
+-- =============================================================
+-- _partition: basic struct projection (native)
+-- =============================================================
+
+query
+SELECT id, _partition FROM test_cat.db.meta_test ORDER BY id
+
+-- Access individual partition fields within the struct
+query
+SELECT id, _partition.id_bucket FROM test_cat.db.meta_test ORDER BY id
+
+-- =============================================================
+-- _spec_id: partition spec ID (scalar per file)
+-- =============================================================
+
+query
+SELECT id, _spec_id FROM test_cat.db.meta_test ORDER BY id
+
+-- All rows should have spec_id = 0 (initial spec)
+query
+SELECT DISTINCT _spec_id FROM test_cat.db.meta_test
+
+-- =============================================================
+-- _pos: row position within each file (0-based)
+-- =============================================================
+
+-- Basic: positions should be 0-based within each file
+query
+SELECT id, _file, _pos FROM test_cat.db.meta_test ORDER BY _file, _pos
+
+-- Positions should be contiguous 0..N-1 per file
+query
+SELECT _file, MIN(_pos) as min_pos, MAX(_pos) as max_pos, COUNT(*) as cnt FROM
test_cat.db.meta_test GROUP BY _file
+
+-- =============================================================
+-- Partition evolution requires IcebergSparkSessionExtensions
+-- (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.
+-- =============================================================
+
+-- =============================================================
+-- Combined: all metadata columns together (native -- all supported)
+-- =============================================================
+
+query
+SELECT id, _file, _pos, _spec_id, _partition FROM test_cat.db.meta_test ORDER
BY _file, _pos
+
+-- Row-level operation simulation: what CoW/MoR would project
+query
+SELECT _file, _pos, _spec_id, _partition, id, name FROM test_cat.db.meta_test
WHERE id > 3 ORDER BY _file, _pos
+
+-- =============================================================
+-- _file + _partition combined (native -- both supported)
+-- =============================================================
+
+query
+SELECT id, _file, _partition FROM test_cat.db.meta_test ORDER BY id
+
+statement
+DROP TABLE test_cat.db.meta_test
+
+-- =============================================================
+-- Identity partition transform
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.identity_part
+
+statement
+CREATE TABLE test_cat.db.identity_part (id INT, name STRING, dept STRING)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.identity_part VALUES (1, 'Alice', 'eng'), (2, 'Bob',
'sales'), (3, 'Charlie', 'eng')
+
+query
+SELECT id, _partition FROM test_cat.db.identity_part ORDER BY id
+
+query
+SELECT id, _partition.dept FROM test_cat.db.identity_part ORDER BY id
+
+query
+SELECT id, _file, _partition FROM test_cat.db.identity_part ORDER BY id
+
+statement
+DROP TABLE test_cat.db.identity_part
+
+-- =============================================================
+-- Multiple partition fields
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.multi_part
+
+statement
+CREATE TABLE test_cat.db.multi_part (id INT, year INT, month INT, data STRING)
USING iceberg PARTITIONED BY (year, month)
+
+statement
+INSERT INTO test_cat.db.multi_part VALUES (1, 2023, 6, 'a'), (2, 2023, 7,
'b'), (3, 2024, 1, 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.multi_part ORDER BY id
+
+statement
+DROP TABLE test_cat.db.multi_part
+
+-- =============================================================
+-- Partition with filter
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.filter_part
+
+statement
+CREATE TABLE test_cat.db.filter_part (id INT, dept STRING, value DOUBLE) USING
iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.filter_part VALUES (1, 'eng', 10.5), (2, 'sales',
20.3), (3, 'eng', 30.7)
+
+query
+SELECT id, _partition FROM test_cat.db.filter_part WHERE dept = 'eng' ORDER BY
id
+
+statement
+DROP TABLE test_cat.db.filter_part
+
+-- =============================================================
+-- Edge case: Unpartitioned table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.unpart
+
+statement
+CREATE TABLE test_cat.db.unpart (id INT, val STRING) USING iceberg
+
+statement
+INSERT INTO test_cat.db.unpart VALUES (1, 'a'), (2, 'b')
+
+-- _partition on unpartitioned table is an empty struct (null), which Comet's
+-- shuffle doesn't support. Verify correctness only, not native operator
coverage.
+query spark_answer_only
+SELECT id, _partition FROM test_cat.db.unpart ORDER BY id
+
+-- Without ORDER BY (no shuffle), the native scan executes end-to-end
+query
+SELECT id, _partition FROM test_cat.db.unpart
+
+query
+SELECT id, _file FROM test_cat.db.unpart ORDER BY id
+
+-- _spec_id on unpartitioned: should be 0
+-- (spark_answer_only because _partition is an empty struct that Comet shuffle
rejects)
+query spark_answer_only
+SELECT id, _spec_id, _partition FROM test_cat.db.unpart ORDER BY id
+
+statement
+DROP TABLE test_cat.db.unpart
+
+-- =============================================================
+-- Edge case: Large batch (tests multi-batch _pos correctness)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.large_pos
+
+statement
+CREATE TABLE test_cat.db.large_pos (id INT) USING iceberg
+
+statement
+INSERT INTO test_cat.db.large_pos SELECT id FROM range(10000)
+
+query
+SELECT id, _pos FROM test_cat.db.large_pos ORDER BY id LIMIT 5
+
+statement
+DROP TABLE test_cat.db.large_pos
+
+-- =============================================================
+-- Edge case: Multiple row groups (forces _pos offset tracking)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.multi_rg
+
+statement
+CREATE TABLE test_cat.db.multi_rg (id INT, data STRING) USING iceberg
TBLPROPERTIES ('write.parquet.row-group-size-bytes' = '1024')
+
+statement
+INSERT INTO test_cat.db.multi_rg SELECT id, REPEAT('x', 100) FROM range(1000)
+
+query
+SELECT _pos, id FROM test_cat.db.multi_rg WHERE _pos IN (0, 1, 998, 999) ORDER
BY _pos
+
+statement
+DROP TABLE test_cat.db.multi_rg
+
+-- =============================================================
+-- DML operations (DELETE, UPDATE, MERGE) internally project metadata columns
+-- (_file, _pos, _spec_id, _partition) to identify rows. All four are now
+-- supported natively. The subsequent SELECT queries verify that reads after
+-- DML return correct results through the native scan path (applying position
+-- deletes, seeing updated data).
+-- =============================================================
+
+-- Copy-on-write DELETE
+statement
+DROP TABLE IF EXISTS test_cat.db.cow_delete
+
+statement
+CREATE TABLE test_cat.db.cow_delete (id INT, name STRING, dept STRING) USING
iceberg PARTITIONED BY (dept) TBLPROPERTIES
('write.delete.mode'='copy-on-write')
+
+statement
+INSERT INTO test_cat.db.cow_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob',
'sales'), (3, 'Charlie', 'eng')
+
+statement
+DELETE FROM test_cat.db.cow_delete WHERE id = 2
+
+query
+SELECT id, _partition FROM test_cat.db.cow_delete ORDER BY id
+
+statement
+DROP TABLE test_cat.db.cow_delete
+
+-- Merge-on-read DELETE (native read applies position deletes)
+statement
+DROP TABLE IF EXISTS test_cat.db.mor_delete
+
+statement
+CREATE TABLE test_cat.db.mor_delete (id INT, name STRING, dept STRING) USING
iceberg PARTITIONED BY (dept) TBLPROPERTIES
('write.delete.mode'='merge-on-read')
+
+statement
+INSERT INTO test_cat.db.mor_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob',
'sales'), (3, 'Charlie', 'eng')
+
+statement
+DELETE FROM test_cat.db.mor_delete WHERE id = 2
+
+query
+SELECT id, _partition FROM test_cat.db.mor_delete ORDER BY id
+
+-- After MoR delete, _pos still references original file positions
+query
+SELECT id, _pos FROM test_cat.db.mor_delete ORDER BY id
+
+statement
+DROP TABLE test_cat.db.mor_delete
+
+-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not
supported in Spark
+-- 4.0, and is not covered anywhere else either: it requires Iceberg's SQL
extensions
+-- (IcebergSparkSessionExtensions), which are registered at SparkSession
construction time and
+-- are not wired into CometTestBase's shared test session (same blocker as
ALTER TABLE ADD
+-- PARTITION FIELD elsewhere in this test suite). Adding coverage needs that
session-level
+-- infra, not just a new test case.
+
+-- =============================================================
+-- Temporal transforms: years() on DATE column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_years_date
+
+statement
+CREATE TABLE test_cat.db.part_years_date (id INT, event_date DATE, data
STRING) USING iceberg PARTITIONED BY (years(event_date))
+
+statement
+INSERT INTO test_cat.db.part_years_date VALUES (1, DATE '2022-03-15', 'a'),
(2, DATE '2023-07-20', 'b'), (3, DATE '2022-11-01', 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.part_years_date ORDER BY id
+
+query
+SELECT id, _partition.event_date_year FROM test_cat.db.part_years_date ORDER
BY id
+
+statement
+DROP TABLE test_cat.db.part_years_date
+
+-- =============================================================
+-- Temporal transforms: months() on DATE column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_months_date
+
+statement
+CREATE TABLE test_cat.db.part_months_date (id INT, event_date DATE, data
STRING) USING iceberg PARTITIONED BY (months(event_date))
+
+statement
+INSERT INTO test_cat.db.part_months_date VALUES (1, DATE '2023-01-10', 'jan'),
(2, DATE '2023-02-15', 'feb'), (3, DATE '2023-01-25', 'jan2')
+
+query
+SELECT id, _partition FROM test_cat.db.part_months_date ORDER BY id
+
+query
+SELECT id, _partition.event_date_month FROM test_cat.db.part_months_date ORDER
BY id
+
+statement
+DROP TABLE test_cat.db.part_months_date
+
+-- =============================================================
+-- Temporal transforms: days() on DATE column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_days_date
+
+statement
+CREATE TABLE test_cat.db.part_days_date (id INT, event_date DATE, data STRING)
USING iceberg PARTITIONED BY (days(event_date))
+
+statement
+INSERT INTO test_cat.db.part_days_date VALUES (1, DATE '2023-06-01', 'x'), (2,
DATE '2023-06-02', 'y'), (3, DATE '2023-06-01', 'z')
+
+query
+SELECT id, _partition FROM test_cat.db.part_days_date ORDER BY id
+
+query
+SELECT id, _partition.event_date_day FROM test_cat.db.part_days_date ORDER BY
id
+
+statement
+DROP TABLE test_cat.db.part_days_date
+
+-- =============================================================
+-- Temporal transforms: hours() on TIMESTAMP column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_hours_ts
+
+statement
+CREATE TABLE test_cat.db.part_hours_ts (id INT, ts TIMESTAMP, data STRING)
USING iceberg PARTITIONED BY (hours(ts))
+
+statement
+INSERT INTO test_cat.db.part_hours_ts VALUES (1, TIMESTAMP '2023-06-15
10:30:00', 'morning'), (2, TIMESTAMP '2023-06-15 14:45:00', 'afternoon'), (3,
TIMESTAMP '2023-06-15 10:55:00', 'morning2')
+
+query
+SELECT id, _partition FROM test_cat.db.part_hours_ts ORDER BY id
+
+query
+SELECT id, _partition.ts_hour FROM test_cat.db.part_hours_ts ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_hours_ts
+
+-- =============================================================
+-- Truncate transform: truncate(N, int_col)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_trunc_int
+
+statement
+CREATE TABLE test_cat.db.part_trunc_int (id INT, value INT, data STRING) USING
iceberg PARTITIONED BY (truncate(10, value))
+
+statement
+INSERT INTO test_cat.db.part_trunc_int VALUES (1, 5, 'a'), (2, 15, 'b'), (3,
7, 'c'), (4, 23, 'd')
+
+query
+SELECT id, _partition FROM test_cat.db.part_trunc_int ORDER BY id
+
+query
+SELECT id, _partition.value_trunc FROM test_cat.db.part_trunc_int ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_trunc_int
+
+-- =============================================================
+-- Truncate transform: truncate(N, string_col)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_trunc_str
+
+statement
+CREATE TABLE test_cat.db.part_trunc_str (id INT, city STRING, data STRING)
USING iceberg PARTITIONED BY (truncate(3, city))
+
+statement
+INSERT INTO test_cat.db.part_trunc_str VALUES (1, 'San Francisco', 'x'), (2,
'Santa Cruz', 'y'), (3, 'Boston', 'z')
+
+query
+SELECT id, _partition FROM test_cat.db.part_trunc_str ORDER BY id
+
+query
+SELECT id, _partition.city_trunc FROM test_cat.db.part_trunc_str ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_trunc_str
+
+-- =============================================================
+-- Multiple non-identity transforms on same table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_multi_transform
+
+statement
+CREATE TABLE test_cat.db.part_multi_transform (id INT, name STRING, ts
TIMESTAMP) USING iceberg PARTITIONED BY (bucket(4, id), truncate(2, name),
hours(ts))
+
+statement
+INSERT INTO test_cat.db.part_multi_transform VALUES (1, 'Alice', TIMESTAMP
'2023-06-15 10:00:00'), (2, 'Bob', TIMESTAMP '2023-06-15 11:00:00'), (3,
'Alice', TIMESTAMP '2023-06-15 10:30:00')
+
+query
+SELECT id, _partition FROM test_cat.db.part_multi_transform ORDER BY id
+
+query
+SELECT id, _partition.id_bucket, _partition.name_trunc, _partition.ts_hour
FROM test_cat.db.part_multi_transform ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_multi_transform
+
+-- =============================================================
+-- NULL partition values
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_nulls
+
+statement
+CREATE TABLE test_cat.db.part_nulls (id INT, dept STRING, value DOUBLE) USING
iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_nulls VALUES (1, 'eng', 10.0), (2, NULL, 20.0),
(3, 'sales', 30.0), (4, NULL, 40.0)
+
+query
+SELECT id, _partition FROM test_cat.db.part_nulls ORDER BY id
+
+query
+SELECT id, _partition.dept FROM test_cat.db.part_nulls ORDER BY id
+
+query
+SELECT id, _partition FROM test_cat.db.part_nulls WHERE _partition.dept IS
NULL ORDER BY id
+
+query
+SELECT id, _partition FROM test_cat.db.part_nulls WHERE _partition.dept IS NOT
NULL ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_nulls
+
+-- =============================================================
+-- _partition in WHERE clause (filter on metadata column itself)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_where
+
+statement
+CREATE TABLE test_cat.db.part_where (id INT, dept STRING, value DOUBLE) USING
iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_where VALUES (1, 'eng', 10.0), (2, 'sales',
20.0), (3, 'eng', 30.0), (4, 'hr', 40.0)
+
+query
+SELECT id, value FROM test_cat.db.part_where WHERE _partition.dept = 'eng'
ORDER BY id
+
+query
+SELECT id, _partition FROM test_cat.db.part_where WHERE _partition.dept IN
('eng', 'hr') ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_where
+
+-- =============================================================
+-- _partition in GROUP BY
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_groupby
+
+statement
+CREATE TABLE test_cat.db.part_groupby (id INT, dept STRING, salary DOUBLE)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_groupby VALUES (1, 'eng', 100.0), (2, 'eng',
120.0), (3, 'sales', 80.0), (4, 'sales', 90.0), (5, 'hr', 70.0)
+
+query
+SELECT _partition.dept, COUNT(*) as cnt, SUM(salary) as total FROM
test_cat.db.part_groupby GROUP BY _partition.dept ORDER BY dept
+
+-- ORDER BY on _partition (StructType) falls back because Comet does not
support Sort on structs
+query spark_answer_only
+SELECT _partition, COUNT(*) as cnt FROM test_cat.db.part_groupby GROUP BY
_partition ORDER BY _partition
+
+statement
+DROP TABLE test_cat.db.part_groupby
+
+-- =============================================================
+-- _partition in JOIN condition
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_join_left
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_join_right
+
+statement
+CREATE TABLE test_cat.db.part_join_left (id INT, dept STRING, value INT) USING
iceberg PARTITIONED BY (dept)
+
+statement
+CREATE TABLE test_cat.db.part_join_right (id INT, dept STRING, budget INT)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_join_left VALUES (1, 'eng', 10), (2, 'sales',
20), (3, 'eng', 30)
+
+statement
+INSERT INTO test_cat.db.part_join_right VALUES (100, 'eng', 500), (200,
'sales', 300)
+
+query
+SELECT l.id, l.value, r.budget FROM test_cat.db.part_join_left l JOIN
test_cat.db.part_join_right r ON l._partition.dept = r._partition.dept ORDER BY
l.id
+
+statement
+DROP TABLE test_cat.db.part_join_left
+
+statement
+DROP TABLE test_cat.db.part_join_right
+
+-- =============================================================
+-- _partition in subquery
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_subq
+
+statement
+CREATE TABLE test_cat.db.part_subq (id INT, dept STRING, value INT) USING
iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_subq VALUES (1, 'eng', 10), (2, 'sales', 20), (3,
'hr', 30), (4, 'eng', 40)
+
+query
+SELECT id, value FROM test_cat.db.part_subq WHERE _partition.dept IN (SELECT
_partition.dept FROM test_cat.db.part_subq WHERE value > 25) ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_subq
+
+-- =============================================================
+-- _partition with HAVING clause
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_having
+
+statement
+CREATE TABLE test_cat.db.part_having (id INT, dept STRING, salary DOUBLE)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_having VALUES (1, 'eng', 100.0), (2, 'eng',
120.0), (3, 'eng', 110.0), (4, 'sales', 80.0), (5, 'hr', 70.0)
+
+query
+SELECT _partition.dept, COUNT(*) as cnt FROM test_cat.db.part_having GROUP BY
_partition.dept HAVING COUNT(*) > 1 ORDER BY dept
+
+statement
+DROP TABLE test_cat.db.part_having
+
+-- =============================================================
+-- _partition with ORDER BY on partition field
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_orderby
+
+statement
+CREATE TABLE test_cat.db.part_orderby (id INT, dept STRING, priority INT)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_orderby VALUES (1, 'eng', 3), (2, 'sales', 1),
(3, 'hr', 2), (4, 'eng', 4)
+
+query
+SELECT id, _partition.dept FROM test_cat.db.part_orderby ORDER BY
_partition.dept, id
+
+statement
+DROP TABLE test_cat.db.part_orderby
+
+-- =============================================================
+-- _partition with DISTINCT
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_distinct
+
+statement
+CREATE TABLE test_cat.db.part_distinct (id INT, region STRING, data STRING)
USING iceberg PARTITIONED BY (region)
+
+statement
+INSERT INTO test_cat.db.part_distinct VALUES (1, 'US', 'a'), (2, 'EU', 'b'),
(3, 'US', 'c'), (4, 'EU', 'd'), (5, 'APAC', 'e')
+
+query spark_answer_only
+SELECT DISTINCT _partition.region FROM test_cat.db.part_distinct ORDER BY
region
+
+query spark_answer_only
+SELECT DISTINCT _partition FROM test_cat.db.part_distinct ORDER BY
_partition.region
+
+statement
+DROP TABLE test_cat.db.part_distinct
+
+-- =============================================================
+-- Wide partition struct (many partition fields)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_wide
+
+statement
+CREATE TABLE test_cat.db.part_wide (id INT, a INT, b INT, c INT, d STRING, e
STRING) USING iceberg PARTITIONED BY (a, b, c, d, e)
+
+statement
+INSERT INTO test_cat.db.part_wide VALUES (1, 10, 20, 30, 'x', 'y'), (2, 11,
21, 31, 'p', 'q'), (3, 10, 20, 30, 'x', 'y')
+
+query
+SELECT id, _partition FROM test_cat.db.part_wide ORDER BY id
+
+query
+SELECT id, _partition.a, _partition.b, _partition.c, _partition.d,
_partition.e FROM test_cat.db.part_wide ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_wide
+
+-- =============================================================
+-- DATE identity partition
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_date_identity
+
+statement
+CREATE TABLE test_cat.db.part_date_identity (id INT, event_date DATE, data
STRING) USING iceberg PARTITIONED BY (event_date)
+
+statement
+INSERT INTO test_cat.db.part_date_identity VALUES (1, DATE '2023-01-15', 'a'),
(2, DATE '2023-02-20', 'b'), (3, DATE '2023-01-15', 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.part_date_identity ORDER BY id
+
+query
+SELECT id, _partition.event_date FROM test_cat.db.part_date_identity ORDER BY
id
+
+query
+SELECT id, data FROM test_cat.db.part_date_identity WHERE
_partition.event_date = DATE '2023-01-15' ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_date_identity
+
+-- =============================================================
+-- String identity with special characters
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_special_chars
+
+statement
+CREATE TABLE test_cat.db.part_special_chars (id INT, tag STRING, data STRING)
USING iceberg PARTITIONED BY (tag)
+
+statement
+INSERT INTO test_cat.db.part_special_chars VALUES (1, 'hello world', 'a'), (2,
'foo/bar', 'b'), (3, '', 'c'), (4, 'a=b&c=d', 'd')
+
+query
+SELECT id, _partition FROM test_cat.db.part_special_chars ORDER BY id
+
+query
+SELECT id, _partition.tag FROM test_cat.db.part_special_chars ORDER BY id
+
+query
+SELECT id FROM test_cat.db.part_special_chars WHERE _partition.tag = 'hello
world' ORDER BY id
+
+query
+SELECT id FROM test_cat.db.part_special_chars WHERE _partition.tag = '' ORDER
BY id
+
+statement
+DROP TABLE test_cat.db.part_special_chars
+
+-- =============================================================
+-- Bucket on STRING column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_bucket_str
+
+statement
+CREATE TABLE test_cat.db.part_bucket_str (id INT, name STRING, value INT)
USING iceberg PARTITIONED BY (bucket(8, name))
+
+statement
+INSERT INTO test_cat.db.part_bucket_str VALUES (1, 'Alice', 10), (2, 'Bob',
20), (3, 'Charlie', 30), (4, 'Alice', 40)
+
+query
+SELECT id, _partition FROM test_cat.db.part_bucket_str ORDER BY id
+
+query
+SELECT id, _partition.name_bucket FROM test_cat.db.part_bucket_str ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_bucket_str
+
+-- =============================================================
+-- years() on TIMESTAMP column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_years_ts
+
+statement
+CREATE TABLE test_cat.db.part_years_ts (id INT, ts TIMESTAMP, data STRING)
USING iceberg PARTITIONED BY (years(ts))
+
+statement
+INSERT INTO test_cat.db.part_years_ts VALUES (1, TIMESTAMP '2021-03-15
08:00:00', 'a'), (2, TIMESTAMP '2022-07-20 12:00:00', 'b'), (3, TIMESTAMP
'2021-11-01 16:00:00', 'c')
+
+query
+SELECT id, _partition FROM test_cat.db.part_years_ts ORDER BY id
+
+query
+SELECT id, _partition.ts_year FROM test_cat.db.part_years_ts ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_years_ts
+
+-- =============================================================
+-- months() on TIMESTAMP column
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_months_ts
+
+statement
+CREATE TABLE test_cat.db.part_months_ts (id INT, ts TIMESTAMP, data STRING)
USING iceberg PARTITIONED BY (months(ts))
+
+statement
+INSERT INTO test_cat.db.part_months_ts VALUES (1, TIMESTAMP '2023-01-10
09:00:00', 'jan'), (2, TIMESTAMP '2023-02-15 10:00:00', 'feb'), (3, TIMESTAMP
'2023-01-25 11:00:00', 'jan2')
+
+query
+SELECT id, _partition FROM test_cat.db.part_months_ts ORDER BY id
+
+query
+SELECT id, _partition.ts_month FROM test_cat.db.part_months_ts ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_months_ts
+
+-- =============================================================
+-- Combined: temporal transform + identity on same table
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_mixed
+
+statement
+CREATE TABLE test_cat.db.part_mixed (id INT, event_date DATE, region STRING,
data STRING) USING iceberg PARTITIONED BY (years(event_date), region)
+
+statement
+INSERT INTO test_cat.db.part_mixed VALUES (1, DATE '2022-03-01', 'US', 'a'),
(2, DATE '2023-07-15', 'EU', 'b'), (3, DATE '2022-09-20', 'US', 'c'), (4, DATE
'2023-01-10', 'US', 'd')
+
+query
+SELECT id, _partition FROM test_cat.db.part_mixed ORDER BY id
+
+query
+SELECT id, _partition.event_date_year, _partition.region FROM
test_cat.db.part_mixed ORDER BY id
+
+query
+SELECT _partition.event_date_year, COUNT(*) as cnt FROM test_cat.db.part_mixed
GROUP BY _partition.event_date_year ORDER BY event_date_year
+
+query
+SELECT id, data FROM test_cat.db.part_mixed WHERE _partition.region = 'US'
ORDER BY id
+
+statement
+DROP TABLE test_cat.db.part_mixed
+
+-- =============================================================
+-- _partition after multiple inserts (different files, same partition)
+-- =============================================================
+
+statement
+DROP TABLE IF EXISTS test_cat.db.part_multi_insert
+
+statement
+CREATE TABLE test_cat.db.part_multi_insert (id INT, dept STRING, value INT)
USING iceberg PARTITIONED BY (dept)
+
+statement
+INSERT INTO test_cat.db.part_multi_insert VALUES (1, 'eng', 10), (2, 'sales',
20)
+
+statement
+INSERT INTO test_cat.db.part_multi_insert VALUES (3, 'eng', 30), (4, 'hr', 40)
+
+statement
+INSERT INTO test_cat.db.part_multi_insert VALUES (5, 'sales', 50)
+
+query
+SELECT id, _partition FROM test_cat.db.part_multi_insert ORDER BY id
+
+query
+SELECT _partition.dept, COUNT(DISTINCT _file) as file_cnt, COUNT(*) as row_cnt
FROM test_cat.db.part_multi_insert GROUP BY _partition.dept ORDER BY dept
+
+statement
+DROP TABLE test_cat.db.part_multi_insert
diff --git
a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index bdad3163d4..c816255924 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -39,7 +39,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{StringType, TimestampType}
import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus,
isSpark40Plus, isSpark42Plus}
-import org.apache.comet.iceberg.RESTCatalogHelper
+import org.apache.comet.iceberg.{IcebergReflection, RESTCatalogHelper}
import org.apache.comet.serde.OperatorOuterClass
import org.apache.comet.testing.{FuzzDataGenerator, SchemaGenOptions}
@@ -4806,8 +4806,9 @@ class CometIcebergNativeSuite
}
}
- test("CometScanRule should report unsupported metadata columns") {
+ test("metadata columns - still-unsupported column falls back to Spark") {
assume(icebergAvailable, "Iceberg not available in classpath")
+
withTempIcebergDir { warehouseDir =>
withSQLConf(
"spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
@@ -4815,24 +4816,286 @@ class CometIcebergNativeSuite
"spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
CometConf.COMET_ENABLED.key -> "true",
CometConf.COMET_EXEC_ENABLED.key -> "true",
- CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true",
- CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true") {
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
- val table = "test_cat.db.test_meta_cols"
+ val table = "test_cat.db.unsupported_meta_col"
try {
spark.sql(s"""
CREATE TABLE $table (id INT, value DOUBLE) USING iceberg
TBLPROPERTIES ('format-version' = '2')
""")
+ spark.sql(s"INSERT INTO $table VALUES (1, 10.5), (2, 20.3)")
+
+ // _deleted is a real, selectable Iceberg metadata column
(SparkTable.metadataColumns())
+ // but is not in CometIcebergNativeScan.MetadataFieldIds, so this
must fall back rather
+ // than silently drop the column or crash. Guards against a future
change to
+ // MetadataFieldIds or the unsupportedMetadataCols filter regressing
this path.
+ checkIcebergNativeScanFallback(
+ s"SELECT id, _deleted FROM $table ORDER BY id",
+ "_deleted is not in CometIcebergNativeScan.MetadataFieldIds")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _pos returns row position within each file") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.pos_test"
+ try {
spark.sql(s"""
- INSERT INTO $table
- VALUES (1, 10.5), (2, 20.3), (3, 30.7)
- """)
+ CREATE TABLE $table (id INT, name STRING) USING iceberg
+ """)
+
+ // coalesce(1) guarantees a single file so _pos is 0,1,2
+ spark
+ .sql("SELECT 1 as id, 'Alice' as name UNION ALL SELECT 2, 'Bob'
UNION ALL SELECT 3, 'Charlie'")
+ .coalesce(1)
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable(table)
+
+ checkIcebergNativeScan(s"SELECT id, _pos FROM $table ORDER BY id")
+
+ // _pos should be 0-indexed within the single file
+ val positions = spark
+ .sql(s"SELECT _pos FROM $table ORDER BY _pos")
+ .collect()
+ .map(_.getLong(0))
+ assert(positions.toSeq == Seq(0L, 1L, 2L), s"Expected [0,1,2], got
${positions.toSeq}")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _pos resets per file with multiple data files") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.pos_multi_file_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, value DOUBLE) USING iceberg
+ """)
+
+ // Two coalesced inserts guarantee exactly two data files
+ spark
+ .range(1, 4)
+ .selectExpr("CAST(id AS INT)", "CAST(id * 10.0 AS DOUBLE) as
value")
+ .coalesce(1)
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable(table)
+ spark
+ .range(4, 6)
+ .selectExpr("CAST(id AS INT)", "CAST(id * 10.0 AS DOUBLE) as
value")
+ .coalesce(1)
+ .write
+ .format("iceberg")
+ .mode("append")
+ .saveAsTable(table)
+
+ // _pos resets to 0 at the start of each file
+ checkIcebergNativeScan(s"SELECT id, _pos, _file FROM $table ORDER BY
_file, _pos")
+
+ // Verify each file starts _pos at 0
+ val rows = spark
+ .sql(s"SELECT _file, _pos FROM $table ORDER BY _file, _pos")
+ .collect()
+ val byFile = rows.groupBy(_.getString(0))
+ assert(byFile.size == 2, s"Expected 2 files, got ${byFile.size}")
+ byFile.foreach { case (file, fileRows) =>
+ val positions = fileRows.map(_.getLong(1))
+ assert(
+ positions.head == 0L,
+ s"File $file should start _pos at 0, got ${positions.head}")
+ }
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _spec_id returns partition spec ID") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.spec_id_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, category STRING, value DOUBLE)
+ USING iceberg PARTITIONED BY (category)
+ """)
+
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, 'A', 10.0), (2, 'B', 20.0), (3, 'A', 30.0)
+ """)
+
+ checkIcebergNativeScan(s"SELECT id, _spec_id FROM $table ORDER BY
id")
+
+ // All rows written under the initial spec should have _spec_id = 0
+ val result = spark.sql(s"SELECT DISTINCT _spec_id FROM
$table").collect()
+ assert(result.length == 1)
+ assert(result(0).getInt(0) == 0)
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _spec_id changes after partition evolution") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ val table = "test_cat.db.spec_id_evolution"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, region STRING, category STRING)
+ USING iceberg PARTITIONED BY (region)
+ """)
+
+ // Data under spec 0
+ spark.sql(s"INSERT INTO $table VALUES (1, 'US', 'A'), (2, 'EU',
'B')")
+
+ // Evolve partition spec: add category field -> spec 1
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val iceTable = sparkCatalog
+ .icebergCatalog()
+ .loadTable(TableIdentifier.of("db", "spec_id_evolution"))
+ iceTable.updateSpec().addField("category").commit()
+
+ // Data under spec 1
+ spark.sql(s"INSERT INTO $table VALUES (3, 'APAC', 'C'), (4, 'US',
'D')")
+
+ checkIcebergNativeScan(s"SELECT id, _spec_id FROM $table ORDER BY
id")
+
+ // Rows 1,2 should have spec_id=0; rows 3,4 should have spec_id=1
+ val result = spark.sql(s"SELECT id, _spec_id FROM $table ORDER BY
id").collect()
+ assert(result(0).getInt(1) == 0, "row 1 should have spec_id=0")
+ assert(result(1).getInt(1) == 0, "row 2 should have spec_id=0")
+ assert(result(2).getInt(1) == 1, "row 3 should have spec_id=1")
+ assert(result(3).getInt(1) == 1, "row 4 should have spec_id=1")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - all metadata columns together") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.all_meta_cols"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, value DOUBLE)
+ USING iceberg PARTITIONED BY (bucket(4, id))
+ TBLPROPERTIES ('format-version' = '2')
+ """)
+
+ spark.sql(s"""
+ INSERT INTO $table VALUES (1, 10.5), (2, 20.3), (3, 30.7)
+ """)
- checkSparkAnswerAndFallbackReason(
- s"SELECT id, value, _spec_id, _pos, _file, _partition FROM $table
WHERE id >= 2 ORDER BY id",
- "Metadata column(s) _spec_id, _partition, _file, _pos is not
supported")
+ // Query all supported metadata columns together
+ checkIcebergNativeScan(
+ s"SELECT id, value, _file, _pos, _spec_id, _partition FROM $table
ORDER BY id")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("metadata columns - _pos with MOR deletes") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.pos_delete_test"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, name STRING) USING iceberg
+ TBLPROPERTIES (
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+ )
+ """)
+
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, 'Alice'), (2, 'Bob'), (3, 'Charlie'), (4, 'Diana'), (5,
'Eve')
+ """)
+
+ // Delete some rows (creates position delete files)
+ spark.sql(s"DELETE FROM $table WHERE id IN (2, 4)")
+
+ // _pos should still reflect original file positions for surviving
rows
+ checkIcebergNativeScan(s"SELECT id, _pos FROM $table ORDER BY id")
} finally {
spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
}
@@ -4870,7 +5133,6 @@ class CometIcebergNativeSuite
test("variant column filter falls back to Spark") {
assume(isSpark40Plus, "VARIANT type requires Spark 4.0+")
assume(icebergAvailable, "Iceberg not available in classpath")
- assume(icebergVersionAtLeast(1, 11), "VARIANT column type requires Iceberg
1.11+")
withTempIcebergDir { warehouseDir =>
withSQLConf(
"spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
@@ -4896,4 +5158,388 @@ class CometIcebergNativeSuite
}
}
}
+
+ test("partition evolution - _partition contains fields from all historical
specs") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ spark.sql("""
+ CREATE TABLE test_cat.db.part_evolution (
+ id INT, region STRING, category STRING, value DOUBLE
+ ) USING iceberg PARTITIONED BY (region)
+ """)
+
+ spark.sql("""
+ INSERT INTO test_cat.db.part_evolution VALUES
+ (1, 'US', 'A', 10.0), (2, 'EU', 'B', 20.0)
+ """)
+
+ // Add a second partition field via Iceberg Java API (partition
evolution -> spec_id 1)
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val table = sparkCatalog
+ .icebergCatalog()
+ .loadTable(TableIdentifier.of("db", "part_evolution"))
+ table.updateSpec().addField("category").commit()
+
+ spark.sql("""
+ INSERT INTO test_cat.db.part_evolution VALUES
+ (3, 'US', 'C', 30.0), (4, 'APAC', 'A', 40.0)
+ """)
+
+ // _partition should be a struct with BOTH region and category fields,
+ // covering all historical specs. Files written under spec 0 will have
+ // category=null in _partition; files under spec 1 have both populated.
+ checkIcebergNativeScan(
+ "SELECT id, _partition FROM test_cat.db.part_evolution ORDER BY id")
+
+ // Verify the struct fields are accessible
+ checkIcebergNativeScan(
+ "SELECT id, _partition.region, _partition.category " +
+ "FROM test_cat.db.part_evolution ORDER BY id")
+
+ // Verify correctness: old rows have null category in _partition
+ val result = spark
+ .sql("SELECT id, _partition.category " +
+ "FROM test_cat.db.part_evolution ORDER BY id")
+ .collect()
+ assert(result(0).isNullAt(1), "row 1 (spec 0) should have null
_partition.category")
+ assert(result(1).isNullAt(1), "row 2 (spec 0) should have null
_partition.category")
+ assert(result(2).getString(1) == "C", "row 3 (spec 1) should have
category=C")
+ assert(result(3).getString(1) == "A", "row 4 (spec 1) should have
category=A")
+
+ spark.sql("DROP TABLE test_cat.db.part_evolution")
+ }
+ }
+ }
+
+ test("partition evolution - _partition struct fields sorted by field_id
ascending") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ // Create table partitioned by region (gets lower field_id in
partition spec)
+ spark.sql("""
+ CREATE TABLE test_cat.db.field_order (
+ id INT, region STRING, category STRING
+ ) USING iceberg PARTITIONED BY (region)
+ """)
+
+ spark.sql("""
+ INSERT INTO test_cat.db.field_order VALUES (1, 'US', 'A'), (2, 'EU',
'B')
+ """)
+
+ // Evolve: add category as second partition field (gets higher
field_id).
+ // Without sort-by-id, spec-descending traversal would emit category
before region.
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val table = sparkCatalog
+ .icebergCatalog()
+ .loadTable(TableIdentifier.of("db", "field_order"))
+ table.updateSpec().addField("category").commit()
+
+ spark.sql("""
+ INSERT INTO test_cat.db.field_order VALUES (3, 'APAC', 'C'), (4,
'US', 'D')
+ """)
+
+ // Verify _partition struct field order matches Spark (sorted by
field_id ascending).
+ // Spark's Java reader uses buildPartitionProjectionType which sorts
by natural key order.
+ checkIcebergNativeScan("SELECT id, _partition FROM
test_cat.db.field_order ORDER BY id")
+
+ // Access fields by name to confirm both are present and correctly
ordered
+ checkIcebergNativeScan(
+ "SELECT id, _partition.region, _partition.category " +
+ "FROM test_cat.db.field_order ORDER BY id")
+
+ spark.sql("DROP TABLE test_cat.db.field_order")
+ }
+ }
+ }
+
+ test("partition evolution - dropped partition field preserved in _partition
struct") {
+ // Exercises the case CTTY raised in iceberg-rust PR #2668:
+ // Java's Partitioning.buildPartitionProjectionType preserves the type of a
+ // partition field that was dropped (Void in newer spec, real transform in
older spec).
+ // This test verifies iceberg-rust matches that behavior through Comet.
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ val table = "test_cat.db.drop_part_field"
+ try {
+ // Spec 0: partitioned by (region, category)
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, region STRING, category STRING, value
DOUBLE)
+ USING iceberg PARTITIONED BY (region, category)
+ """)
+
+ spark.sql(s"INSERT INTO $table VALUES (1, 'US', 'A', 10.0), (2,
'EU', 'B', 20.0)")
+
+ // Evolve: drop 'category' from partition spec -> spec 1 marks
category as Void
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val iceTable = sparkCatalog
+ .icebergCatalog()
+ .loadTable(TableIdentifier.of("db", "drop_part_field"))
+ iceTable.updateSpec().removeField("category").commit()
+
+ // Insert data under spec 1 (only region partitioning)
+ spark.sql(s"INSERT INTO $table VALUES (3, 'APAC', 'C', 30.0), (4,
'US', 'D', 40.0)")
+
+ // _partition struct should still contain BOTH region and category
fields.
+ // Java preserves dropped partition fields in the unified type;
+ // rows written under spec 1 have category=null in _partition.
+ checkIcebergNativeScan(s"SELECT id, _partition FROM $table ORDER BY
id")
+
+ checkIcebergNativeScan(
+ s"SELECT id, _partition.region, _partition.category FROM $table
ORDER BY id")
+
+ // Verify: spec-0 rows have category populated, spec-1 rows have
category=null
+ val result = spark
+ .sql(s"SELECT id, _partition.category FROM $table ORDER BY id")
+ .collect()
+ assert(
+ result(0).getString(1) == "A",
+ "row 1 (spec 0) should have _partition.category=A")
+ assert(
+ result(1).getString(1) == "B",
+ "row 2 (spec 0) should have _partition.category=B")
+ assert(result(2).isNullAt(1), "row 3 (spec 1) should have null
_partition.category")
+ assert(result(3).isNullAt(1), "row 4 (spec 1) should have null
_partition.category")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("partition evolution - re-add dropped partition field") {
+ // Further exercises the void-transform handling: drop a field then re-add
it.
+ // Java's unified type should contain the field once (not duplicated).
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ val table = "test_cat.db.readd_part_field"
+ try {
+ // Spec 0: partitioned by (region)
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, region STRING, value DOUBLE)
+ USING iceberg PARTITIONED BY (region)
+ """)
+
+ spark.sql(s"INSERT INTO $table VALUES (1, 'US', 10.0)")
+
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val iceTable = sparkCatalog
+ .icebergCatalog()
+ .loadTable(TableIdentifier.of("db", "readd_part_field"))
+
+ // Spec 1: drop region
+ iceTable.updateSpec().removeField("region").commit()
+ spark.sql(s"INSERT INTO $table VALUES (2, 'EU', 20.0)")
+
+ // Spec 2: re-add region (gets a new partition field_id but same
source column)
+ iceTable.refresh()
+ iceTable.updateSpec().addField("region").commit()
+ spark.sql(s"INSERT INTO $table VALUES (3, 'APAC', 30.0)")
+
+ // All rows should be queryable with _partition
+ checkIcebergNativeScan(s"SELECT id, _partition FROM $table ORDER BY
id")
+
+ checkIcebergNativeScan(s"SELECT id, _partition.region FROM $table
ORDER BY id")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("partition evolution - conflicting V1 spec field detected before native
scan") {
+ // Iceberg Java's own write path (e.g. updateSpec()) can never produce two
specs that bind
+ // the same partition field id to different source columns:
TableMetadata.Builder always
+ // reassigns fresh, non-conflicting field ids when a spec is added. Only
V1 tables written by
+ // a non-Java writer (or, as here, a hand-edited metadata.json) can have
this defect --
+ // exactly the case
IcebergReflection.validateUnifiedPartitionType/CometScanRule's
+ // unifiedPartitionTypeSupported guards against. Table *loading* does not
validate this
+ // (TableMetadataParser binds all non-default specs via
UnboundPartitionSpec#bindUnchecked),
+ // so the conflict only surfaces when something asks for the unified
partition type.
+ //
+ // That "something" turns out to always be Spark itself, not just Comet:
resolving
+ // _partition's output type for any query that touches metadata columns on
this table goes
+ // through SparkTable.metadataColumns(), which eagerly calls the same
+ // Partitioning.partitionType(table) and throws during analysis, before
CometScanRule (which
+ // runs during physical planning) ever sees the query. So there's no SQL
query that
+ // reaches CometScanRule's fallback for this case -- plain Spark can't
analyze it either. This
+ // test instead exercises IcebergReflection.validateUnifiedPartitionType
directly, which is
+ // the reflection call unifiedPartitionTypeSupported relies on to fall
back safely in the
+ // (currently unreached, but not guaranteed to stay that way across
Iceberg versions) case
+ // where analysis succeeds but the native scan's own merge would otherwise
fail.
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+
+ val table = "test_cat.db.conflicting_spec"
+ try {
+ spark.sql(s"""
+ CREATE TABLE $table (id INT, region STRING, ts TIMESTAMP)
+ USING iceberg PARTITIONED BY (region)
+ TBLPROPERTIES ('format-version' = '1')
+ """)
+ spark.sql(s"INSERT INTO $table VALUES (1, 'US', TIMESTAMP
'2024-01-01 00:00:00')")
+
+ import org.apache.iceberg.catalog.TableIdentifier
+ import org.apache.iceberg.spark.SparkCatalog
+
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("test_cat")
+ .asInstanceOf[SparkCatalog]
+ val iceTable = sparkCatalog
+ .icebergCatalog()
+ .loadTable(TableIdentifier.of("db", "conflicting_spec"))
+
+ // Derive the metadata dir from the table's own reported location
rather than
+ // reconstructing the warehouse layout by hand, since location() may
or may not carry a
+ // "file:" URI scheme depending on the FileIO in use.
+ val tableLocationUri = iceTable.location()
+ val tableDir =
+ if (tableLocationUri.contains(":")) new File(new
java.net.URI(tableLocationUri))
+ else new File(tableLocationUri)
+ val metadataDir = new File(tableDir, "metadata")
+ // Discover the current metadata version by listing files rather
than trusting
+ // version-hint.text: HadoopCatalog (unlike plain HadoopTables) does
not always write
+ // it, and Iceberg's own HadoopTableOperations falls back to a
directory scan for the
+ // same reason.
+ val versionPattern = "^v(\\d+)\\.metadata\\.json$".r
+ val currentVersion = metadataDir
+ .listFiles()
+ .flatMap(f =>
versionPattern.findFirstMatchIn(f.getName).map(_.group(1).toInt))
+ .max
+ val currentMetadataFile = new File(metadataDir,
s"v$currentVersion.metadata.json")
+ val currentMetadataJson =
+ new
String(java.nio.file.Files.readAllBytes(currentMetadataFile.toPath), UTF_8)
+
+ val mapper = new com.fasterxml.jackson.databind.ObjectMapper()
+ val root = mapper
+ .readTree(currentMetadataJson)
+ .asInstanceOf[com.fasterxml.jackson.databind.node.ObjectNode]
+ val specs = root
+ .get("partition-specs")
+ .asInstanceOf[com.fasterxml.jackson.databind.node.ArrayNode]
+ val spec0 = specs.get(0)
+ val regionFieldId =
spec0.get("fields").get(0).get("field-id").asInt()
+ val idFieldId = root
+ .get("schemas")
+ .get(0)
+ .get("fields")
+ .elements()
+ .asScala
+ .find(_.get("name").asText() == "id")
+ .get
+ .get("id")
+ .asInt()
+
+ // New spec (id 1) reuses region's field id for a field bound to a
*different* source
+ // column ("id" instead of "region") -- exactly what
Partitioning.partitionType()/
+ // compute_unified_partition_type's equivalentIgnoringNames rejects.
+ val conflictingSpec = mapper.createObjectNode()
+ conflictingSpec.put("spec-id", 1)
+ val conflictingFields = mapper.createArrayNode()
+ val conflictingField = mapper.createObjectNode()
+ conflictingField.put("source-id", idFieldId)
+ conflictingField.put("field-id", regionFieldId)
+ conflictingField.put("name", "id_as_region")
+ conflictingField.put("transform", "identity")
+ conflictingFields.add(conflictingField)
+ conflictingSpec.set("fields", conflictingFields)
+ specs.add(conflictingSpec)
+
+ // Round-trip through Iceberg's own parser before writing, so a
malformed hand-edit
+ // fails this test with a clear parse error rather than producing an
unloadable table.
+ val newMetadataJson = mapper.writeValueAsString(root)
+ org.apache.iceberg.TableMetadataParser.fromJson(newMetadataJson)
+
+ val newVersion = currentVersion + 1
+ val newMetadataFile = new File(metadataDir,
s"v$newVersion.metadata.json")
+ java.nio.file.Files.write(newMetadataFile.toPath,
newMetadataJson.getBytes(UTF_8))
+ // Best-effort: some catalog configurations track the current
version via this file;
+ // others (see the directory-listing fallback above) don't require
it.
+ java.nio.file.Files.write(
+ new File(metadataDir, "version-hint.text").toPath,
+ newVersion.toString.getBytes(UTF_8))
+
+ // Not testable end-to-end via a SQL query: resolving _partition's
output type for
+ // *any* query touching metadata columns on this table goes through
Spark's own
+ // SparkTable.metadataColumns() -> SparkMetadataColumns.partition()
->
+ // Partitioning.partitionType(table), which throws this exact
ValidationException
+ // during analysis -- before CometScanRule (which runs during
physical planning) ever
+ // sees the query. So this exercises the same reflection call
CometScanRule's
+ // unifiedPartitionTypeSupported guards with, directly, against the
conflicting table.
+ iceTable.refresh()
+ val reason = IcebergReflection.validateUnifiedPartitionType(iceTable)
+ assert(
+ reason.isDefined,
+ "Expected a validation failure for conflicting partition specs")
+ assert(
+ reason.get.contains("Conflicting partition fields"),
+ s"Expected a conflicting-partition-fields reason, got:
${reason.get}")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
