From 08d8060f543d280669c80083d1a6a182988ec6c1 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Mon, 15 Jun 2026 13:20:11 -0700 Subject: [PATCH 1/5] feat: [iceberg] support iceberg metadata columns _file, _partition --- native/Cargo.toml | 3 + .../src/execution/operators/iceberg_scan.rs | 1 + native/core/src/execution/planner.rs | 61 ++ .../apache/comet/rules/CometScanRule.scala | 57 +- .../operator/CometIcebergNativeScan.scala | 20 +- .../iceberg/metadata_column_partition.sql | 850 ++++++++++++++++++ .../comet/CometIcebergNativeSuite.scala | 122 ++- 7 files changed, 1100 insertions(+), 14 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql diff --git a/native/Cargo.toml b/native/Cargo.toml index e2277596aa..4431ccc8f6 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -62,6 +62,9 @@ iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "ff678ca" } iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "ff678ca", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } reqsign-core = "3" +[patch."https://github.com/apache/iceberg-rust"] +iceberg = { git = "https://github.com/parthchandra/iceberg-rust", branch = "metadata-columns-unified" } + [profile.release] debug = true overflow-checks = false diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index d89a25c879..0fed54bc60 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -659,6 +659,7 @@ mod tests { partition: None, partition_spec: None, name_mapping: None, + unified_partition_type: None, case_sensitive: false, } } diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 4f5d3ceaab..d0cf8df505 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3694,6 +3694,41 @@ fn parse_file_scan_tasks_from_common( }) .collect::, _>>()?; + // Compute unified partition type from the partition_type_pool. + // Each entry is a StructType JSON with the resolved field types for one partition spec. + // Merge all specs into a single unified type (dedup by field_id). + // + // NOTE: The type information here comes from Iceberg Java's PartitionSpec.partitionType() + // (via Scala reflection). This is the same type computation as iceberg-rust's + // Transform::result_type() -- both implement the Iceberg spec's transform result rules. + // When iceberg-rust's own scan planning is used (not Comet's proto path), it computes + // this via compute_unified_partition_type(specs, schema) instead. + let unified_partition_type = { + let mut seen_field_ids = std::collections::HashSet::new(); + let mut struct_fields: Vec = Vec::new(); + + for type_json in &proto_common.partition_type_pool { + match serde_json::from_str::(type_json) { + Ok(struct_type) => { + for field in struct_type.fields() { + if seen_field_ids.insert(field.id) { + struct_fields.push(Arc::clone(field)); + } + } + } + Err(e) => { + return Err(ExecutionError::GeneralError(format!( + "Failed to deserialize partition type JSON from pool: {e}" + ))); + } + } + } + + iceberg::spec::StructType::new(struct_fields) + }; + + let unified_partition_type_arc = Arc::new(unified_partition_type); + let results: Result, _> = proto_tasks .iter() .map(|proto_task| { @@ -3797,6 +3832,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(), @@ -3811,6 +3854,7 @@ fn parse_file_scan_tasks_from_common( partition, partition_spec, name_mapping, + unified_partition_type: unified_partition_type_for_task, case_sensitive: false, }) }) @@ -5427,4 +5471,21 @@ mod tests { } }); } + + #[test] + fn test_metadata_field_id_constants_match_iceberg_rust() { + // These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds) + // as Int.MaxValue - 1 and Int.MaxValue - 5. This test ensures the Rust constants + // haven't drifted, which would cause a silent mismatch with the Scala side. + assert_eq!( + iceberg::metadata_columns::RESERVED_FIELD_ID_FILE, + i32::MAX - 1, + "RESERVED_FIELD_ID_FILE must be i32::MAX - 1 to match Scala MetadataFieldIds" + ); + assert_eq!( + iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION, + i32::MAX - 5, + "RESERVED_FIELD_ID_PARTITION must be i32::MAX - 5 to match Scala MetadataFieldIds" + ); + } } 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 59331129d5..2bff179732 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -125,7 +125,7 @@ case class CometScanRule(session: SparkSession) case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) => withFallbackReason(scan, "Comet Scan is not enabled") - case scan if metadataCols(scan).nonEmpty => + case scan: FileSourceScanExec if metadataCols(scan).nonEmpty => withFallbackReason( scan, s"Metadata column(s) ${metadataCols(scan).mkString(", ")} is not supported") @@ -297,6 +297,11 @@ case class CometScanRule(session: SparkSession) scanExec.scan match { case scan: CSVScan if COMET_CSV_V2_NATIVE_ENABLED.get() => + if (scanExec.output.exists(_.isMetadataCol)) { + return withFallbackReason( + scanExec, + "Metadata columns are not supported for CSV V2 scans") + } val fallbackReasons = new ListBuffer[String]() val schemaSupported = CometBatchScanExec.isSchemaSupported(scan.readDataSchema, fallbackReasons) @@ -356,9 +361,30 @@ case class CometScanRule(session: SparkSession) return withFallbackReasons(scanExec, fallbackReasons.toSet) } + // Check for unsupported metadata columns in Iceberg scans + val unsupportedMetadataCols = scanExec.output.filter(_.isMetadataCol).filterNot { attr => + CometIcebergNativeScan.MetadataFieldIds.keySet.contains(attr.name) + } + if (unsupportedMetadataCols.nonEmpty) { + fallbackReasons += "Unsupported Iceberg metadata columns: " + + unsupportedMetadataCols.map(_.name).mkString(", ") + return withFallbackReasons(scanExec, fallbackReasons.toSet) + } + val typeChecker = CometScanTypeChecker() + // Filter out metadata columns from schema check -- their types are handled + // by iceberg-rust directly (e.g., _partition can be an empty struct for + // unpartitioned tables which the general type checker rejects). + val metadataColNames = scanExec.output.filter(_.isMetadataCol).map(_.name).toSet + val dataSchema = if (metadataColNames.nonEmpty) { + val filtered = + scanExec.scan.readSchema().filter(f => !metadataColNames.contains(f.name)) + new org.apache.spark.sql.types.StructType(filtered.toArray) + } else { + scanExec.scan.readSchema() + } val schemaSupported = - typeChecker.isSchemaSupported(scanExec.scan.readSchema(), fallbackReasons) + typeChecker.isSchemaSupported(dataSchema, fallbackReasons) if (!schemaSupported) { fallbackReasons += "Comet extension is not enabled for " + @@ -446,15 +472,18 @@ case class CometScanRule(session: SparkSession) } // Now perform all validation using the pre-extracted metadata - // Check if table uses a FileIO implementation compatible with iceberg-rust - + // Check if table uses a FileIO implementation compatible with iceberg-rust. + // Comet's native reader uses object_store (Rust) for I/O, bypassing Iceberg Java's + // FileIO entirely. Only allow known-compatible implementations whose underlying + // storage object_store can reach via standard URL schemes. val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match { case Some(fileIO) - if fileIO.getClass.getName == "org.apache.iceberg.inmemory.InMemoryFileIO" => - fallbackReasons += "InMemoryFileIO is not supported by Comet's native reader" - false - case Some(_) => + if CometScanRule.CompatibleFileIOClasses.contains(fileIO.getClass.getName) => true + case Some(fileIO) => + fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not supported by " + + "Comet's native reader (object_store bypasses Iceberg Java FileIO)" + false case None => fallbackReasons += "Could not check FileIO compatibility" false @@ -808,6 +837,18 @@ case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim { object CometScanRule extends Logging { + // Iceberg FileIO implementations whose underlying storage object_store can reach. + // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible + // because Comet's native reader bypasses Java FileIO entirely. + private val CompatibleFileIOClasses: Set[String] = Set( + "org.apache.iceberg.hadoop.HadoopFileIO", + "org.apache.iceberg.aws.s3.S3FileIO", + "org.apache.iceberg.gcp.gcs.GCSFileIO", + "org.apache.iceberg.io.ResolvingFileIO", + "org.apache.iceberg.spark.SparkFileIO", + "org.apache.iceberg.azure.adlsv2.ADLSFileIO", + "org.apache.iceberg.CachingFileIO") + // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer depends only on the // URL scheme, so we cache by scheme and never re-cross the JNI boundary for a repeated scheme. private val schemeSupportCache = 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 879762818f..3d94130c4d 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 @@ -71,6 +71,14 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } } + // Iceberg reserved field IDs for metadata columns. + // These must match iceberg-rust's constants in crates/iceberg/src/metadata_columns.rs: + // RESERVED_FIELD_ID_FILE = i32::MAX - 1 (2147483646) + // RESERVED_FIELD_ID_PARTITION = i32::MAX - 5 (2147483642) + // Scala's Int.MaxValue == 2^31 - 1 == Rust's i32::MAX. + val MetadataFieldIds: Map[String, Int] = + Map("_file" -> (Int.MaxValue - 1), "_partition" -> (Int.MaxValue - 5)) + /** * Converts an Iceberg partition value to protobuf format. Protobuf is less verbose than JSON. * The following types are also serialized as integer values instead of as strings - Timestamps, @@ -883,14 +891,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.") } } 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..0efde9888a --- /dev/null +++ b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql @@ -0,0 +1,850 @@ +-- 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. + +-- 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: falls back until implemented. +-- When _spec_id is implemented, change these to plain `query`. +-- ============================================================= + +query expect_fallback(Unsupported Iceberg metadata columns) +SELECT id, _spec_id FROM test_cat.db.meta_test ORDER BY id + +-- All rows should have spec_id = 0 (initial spec) +query expect_fallback(Unsupported Iceberg metadata columns) +SELECT DISTINCT _spec_id FROM test_cat.db.meta_test + +-- ============================================================= +-- _pos: falls back until implemented. +-- When _pos is implemented, change these to plain `query`. +-- ============================================================= + +-- Basic: positions should be 0-based within each file +query expect_fallback(Unsupported Iceberg metadata columns) +SELECT id, _file, _pos FROM test_cat.db.meta_test ORDER BY _file, _pos + +-- Positions should be contiguous 0..N-1 per file +query expect_fallback(Unsupported Iceberg metadata columns) +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 (falls back due to _pos/_spec_id) +-- ============================================================= + +query expect_fallback(Unsupported Iceberg metadata columns) +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 expect_fallback(Unsupported Iceberg metadata columns) +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 +query expect_fallback(Unsupported Iceberg metadata columns) +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) +-- Falls back due to _pos, but verifies correctness via Spark parity. +-- ============================================================= + +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 expect_fallback(Unsupported Iceberg metadata columns) +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) +-- Falls back due to _pos, but verifies correctness via Spark parity. +-- ============================================================= + +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 expect_fallback(Unsupported Iceberg metadata columns) +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. Since we don't yet +-- support all four, the DML scan falls back to Spark's JVM reader. The +-- subsequent SELECT queries verify that reads after DML return correct results +-- through the native scan path (applying position deletes, seeing updated data). +-- Once all metadata columns are implemented, Spark will use the native scan for +-- the DML read side too (no code change needed here -- the fallback guard lifts +-- automatically when supportedIcebergMetadataCols covers all projected columns). +-- ============================================================= + +-- 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 (falls back) +query expect_fallback(Unsupported Iceberg metadata columns) +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. Covered in CometIcebergNativeSuite instead. + +-- ============================================================= +-- 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 9656a13c82..3ba7937825 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -4335,7 +4335,7 @@ class CometIcebergNativeSuite 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") + "Unsupported Iceberg metadata columns: _spec_id, _pos") } finally { spark.sql(s"DROP TABLE IF EXISTS $table PURGE") } @@ -4398,4 +4398,124 @@ 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") + } + } + } } From d0efe99161ebab19c564a74209cce4768893f30f Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 8 Jul 2026 16:44:31 -0700 Subject: [PATCH 2/5] add _pos, _spec_id --- native/Cargo.lock | 2 +- .../operator/CometIcebergNativeScan.scala | 8 +- .../iceberg/metadata_column_partition.sql | 44 ++-- .../comet/CometIcebergNativeSuite.scala | 223 +++++++++++++++++- 4 files changed, 240 insertions(+), 37 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index c54812c818..d2b620b9d8 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -3602,7 +3602,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=ff678ca#ff678ca6d1123d9d82902b34379ccba2caf172b6" +source = "git+https://github.com/parthchandra/iceberg-rust?branch=metadata-columns-unified#7fb6a0e5555b83545fd6102e7f6c802b82026bd9" dependencies = [ "aes-gcm", "anyhow", 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 3d94130c4d..98bc2e2491 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 @@ -74,10 +74,16 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit // Iceberg reserved field IDs for metadata columns. // These must match iceberg-rust's constants in crates/iceberg/src/metadata_columns.rs: // RESERVED_FIELD_ID_FILE = i32::MAX - 1 (2147483646) + // RESERVED_FIELD_ID_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), "_partition" -> (Int.MaxValue - 5)) + Map( + "_file" -> (Int.MaxValue - 1), + "_pos" -> (Int.MaxValue - 2), + "_spec_id" -> (Int.MaxValue - 4), + "_partition" -> (Int.MaxValue - 5)) /** * Converts an Iceberg partition value to protobuf format. Protobuf is less verbose than JSON. 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 index 0efde9888a..423faeb418 100644 --- a/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql +++ b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql @@ -71,28 +71,26 @@ query SELECT id, _partition.id_bucket FROM test_cat.db.meta_test ORDER BY id -- ============================================================= --- _spec_id: falls back until implemented. --- When _spec_id is implemented, change these to plain `query`. +-- _spec_id: partition spec ID (scalar per file) -- ============================================================= -query expect_fallback(Unsupported Iceberg metadata columns) +query SELECT id, _spec_id FROM test_cat.db.meta_test ORDER BY id -- All rows should have spec_id = 0 (initial spec) -query expect_fallback(Unsupported Iceberg metadata columns) +query SELECT DISTINCT _spec_id FROM test_cat.db.meta_test -- ============================================================= --- _pos: falls back until implemented. --- When _pos is implemented, change these to plain `query`. +-- _pos: row position within each file (0-based) -- ============================================================= -- Basic: positions should be 0-based within each file -query expect_fallback(Unsupported Iceberg metadata columns) +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 expect_fallback(Unsupported Iceberg metadata columns) +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 -- ============================================================= @@ -104,14 +102,14 @@ SELECT _file, MIN(_pos) as min_pos, MAX(_pos) as max_pos, COUNT(*) as cnt FROM t -- ============================================================= -- ============================================================= --- Combined: all metadata columns together (falls back due to _pos/_spec_id) +-- Combined: all metadata columns together (native -- all supported) -- ============================================================= -query expect_fallback(Unsupported Iceberg metadata columns) +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 expect_fallback(Unsupported Iceberg metadata columns) +query SELECT _file, _pos, _spec_id, _partition, id, name FROM test_cat.db.meta_test WHERE id > 3 ORDER BY _file, _pos -- ============================================================= @@ -213,7 +211,8 @@ query SELECT id, _file FROM test_cat.db.unpart ORDER BY id -- _spec_id on unpartitioned: should be 0 -query expect_fallback(Unsupported Iceberg metadata columns) +-- (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 @@ -221,7 +220,6 @@ DROP TABLE test_cat.db.unpart -- ============================================================= -- Edge case: Large batch (tests multi-batch _pos correctness) --- Falls back due to _pos, but verifies correctness via Spark parity. -- ============================================================= statement @@ -233,7 +231,7 @@ 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 expect_fallback(Unsupported Iceberg metadata columns) +query SELECT id, _pos FROM test_cat.db.large_pos ORDER BY id LIMIT 5 statement @@ -241,7 +239,6 @@ DROP TABLE test_cat.db.large_pos -- ============================================================= -- Edge case: Multiple row groups (forces _pos offset tracking) --- Falls back due to _pos, but verifies correctness via Spark parity. -- ============================================================= statement @@ -253,7 +250,7 @@ CREATE TABLE test_cat.db.multi_rg (id INT, data STRING) USING iceberg TBLPROPERT statement INSERT INTO test_cat.db.multi_rg SELECT id, REPEAT('x', 100) FROM range(1000) -query expect_fallback(Unsupported Iceberg metadata columns) +query SELECT _pos, id FROM test_cat.db.multi_rg WHERE _pos IN (0, 1, 998, 999) ORDER BY _pos statement @@ -261,13 +258,10 @@ DROP TABLE test_cat.db.multi_rg -- ============================================================= -- DML operations (DELETE, UPDATE, MERGE) internally project metadata columns --- (_file, _pos, _spec_id, _partition) to identify rows. Since we don't yet --- support all four, the DML scan falls back to Spark's JVM reader. The --- subsequent SELECT queries verify that reads after DML return correct results --- through the native scan path (applying position deletes, seeing updated data). --- Once all metadata columns are implemented, Spark will use the native scan for --- the DML read side too (no code change needed here -- the fallback guard lifts --- automatically when supportedIcebergMetadataCols covers all projected 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 @@ -305,8 +299,8 @@ 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 (falls back) -query expect_fallback(Unsupported Iceberg metadata columns) +-- After MoR delete, _pos still references original file positions +query SELECT id, _pos FROM test_cat.db.mor_delete ORDER BY id statement diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 3ba7937825..ed9d70563b 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -4310,7 +4310,9 @@ class CometIcebergNativeSuite } } - test("CometScanRule should report unsupported metadata columns") { + 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", @@ -4318,24 +4320,225 @@ 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.pos_test" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, name STRING) USING iceberg + """) + + spark.sql(s""" + INSERT INTO $table VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie') + """) + + checkIcebergNativeScan(s"SELECT id, _pos FROM $table ORDER BY id") + + // _pos should be 0-indexed per file + val result = spark.sql(s"SELECT id, _pos FROM $table ORDER BY id").collect() + assert(result(0).getLong(1) == 0L) + assert(result(1).getLong(1) == 1L) + assert(result(2).getLong(1) == 2L) + } 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 separate inserts create two data files + spark.sql(s"INSERT INTO $table VALUES (1, 10.0), (2, 20.0), (3, 30.0)") + spark.sql(s"INSERT INTO $table VALUES (4, 40.0), (5, 50.0)") + + // _pos resets to 0 at the start of each file + checkIcebergNativeScan(s"SELECT id, _pos, _file FROM $table ORDER BY _file, _pos") + + // Verify _pos starts at 0 in each file + val result = spark.sql(s"SELECT _pos FROM $table ORDER BY _file, _pos").collect() + val positions = result.map(_.getLong(0)) + // First file has 0,1,2; second file has 0,1 (or vice versa) + assert(positions.contains(0L)) + assert(positions.count(_ == 0L) == 2, "Each file should start _pos at 0") + } 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) - """) + INSERT INTO $table VALUES (1, 10.5), (2, 20.3), (3, 30.7) + """) + + // 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)") - checkSparkAnswerAndFallbackReason( - s"SELECT id, value, _spec_id, _pos, _file, _partition FROM $table WHERE id >= 2 ORDER BY id", - "Unsupported Iceberg metadata columns: _spec_id, _pos") + // _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") } From 82d7c4325e7ef7b8ea60930dbfe9750c08372467 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 9 Jul 2026 09:54:26 -0700 Subject: [PATCH 3/5] update cargo.lock --- native/Cargo.lock | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index d2b620b9d8..efbe51fc91 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -2973,7 +2973,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -3863,7 +3863,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5454,7 +5454,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5954,7 +5954,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -6012,7 +6012,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -6593,7 +6593,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7274,7 +7274,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] From 70bbf95262e07c51f39c3ed9ad2a3592db63ee1b Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 9 Jul 2026 14:16:58 -0700 Subject: [PATCH 4/5] fix tests --- native/Cargo.lock | 14 ++--- .../comet/CometIcebergNativeSuite.scala | 60 +++++++++++++------ 2 files changed, 50 insertions(+), 24 deletions(-) diff --git a/native/Cargo.lock b/native/Cargo.lock index efbe51fc91..d2b620b9d8 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -2973,7 +2973,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3863,7 +3863,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5454,7 +5454,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5954,7 +5954,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6012,7 +6012,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6593,7 +6593,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7274,7 +7274,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index ed9d70563b..b03d4aad7e 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -4328,17 +4328,23 @@ class CometIcebergNativeSuite CREATE TABLE $table (id INT, name STRING) USING iceberg """) - spark.sql(s""" - INSERT INTO $table VALUES (1, 'Alice'), (2, 'Bob'), (3, 'Charlie') - """) + // 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 per file - val result = spark.sql(s"SELECT id, _pos FROM $table ORDER BY id").collect() - assert(result(0).getLong(1) == 0L) - assert(result(1).getLong(1) == 1L) - assert(result(2).getLong(1) == 2L) + // _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") } @@ -4364,19 +4370,39 @@ class CometIcebergNativeSuite CREATE TABLE $table (id INT, value DOUBLE) USING iceberg """) - // Two separate inserts create two data files - spark.sql(s"INSERT INTO $table VALUES (1, 10.0), (2, 20.0), (3, 30.0)") - spark.sql(s"INSERT INTO $table VALUES (4, 40.0), (5, 50.0)") + // 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 _pos starts at 0 in each file - val result = spark.sql(s"SELECT _pos FROM $table ORDER BY _file, _pos").collect() - val positions = result.map(_.getLong(0)) - // First file has 0,1,2; second file has 0,1 (or vice versa) - assert(positions.contains(0L)) - assert(positions.count(_ == 0L) == 2, "Each file should start _pos at 0") + // 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") } From cfa8149cb6a6bda702c2f21f41593c938bb4007e Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 9 Jul 2026 17:54:38 -0700 Subject: [PATCH 5/5] update diff --- dev/diffs/iceberg/1.11.0.diff | 82 ++++++++++++++++++++--------------- 1 file changed, 48 insertions(+), 34 deletions(-) diff --git a/dev/diffs/iceberg/1.11.0.diff b/dev/diffs/iceberg/1.11.0.diff index a404ded20a..aa02e07fb6 100644 --- a/dev/diffs/iceberg/1.11.0.diff +++ b/dev/diffs/iceberg/1.11.0.diff @@ -1,5 +1,5 @@ diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml -index db659dc06b..78f5e3440f 100644 +index db659dc..78f5e34 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -40,6 +40,7 @@ bouncycastle = "1.84" @@ -11,7 +11,7 @@ index db659dc06b..78f5e3440f 100644 delta-standalone = "3.3.2" delta-spark = "3.3.2" diff --git a/spark/v4.1/build.gradle b/spark/v4.1/build.gradle -index e6455fa34f..08b685f61a 100644 +index e6455fa..08b685f 100644 --- a/spark/v4.1/build.gradle +++ b/spark/v4.1/build.gradle @@ -109,6 +109,7 @@ project(":iceberg-spark:iceberg-spark-${sparkMajorVersion}_${scalaVersion}") { @@ -39,7 +39,7 @@ index e6455fa34f..08b685f61a 100644 // runtime dependencies for running Hive Catalog based integration test integrationRuntimeOnly project(':iceberg-hive-metastore') diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java -index f766fbb79a..c5e31185a9 100644 +index f766fbb..c5e3118 100644 --- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java +++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/ExtensionsTestBase.java @@ -71,6 +71,14 @@ public abstract class ExtensionsTestBase extends CatalogTestBase { @@ -57,8 +57,22 @@ 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 b5d6415..0763a72 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 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 +index 3fd8455..350b4a7 100644 --- 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 @@ -180,6 +180,14 @@ public class DeleteOrphanFilesBenchmark { @@ -77,7 +91,7 @@ index 3fd84553f0..350b4a7561 100644 .master("local"); spark = builder.getOrCreate(); diff --git a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java -index 683f6bb46d..402561a241 100644 +index 683f6bb..402561a 100644 --- a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java +++ b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/IcebergSortCompactionBenchmark.java @@ -395,6 +395,14 @@ public class IcebergSortCompactionBenchmark { @@ -96,7 +110,7 @@ index 683f6bb46d..402561a241 100644 .master("local[*]"); spark = builder.getOrCreate(); diff --git a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVReaderBenchmark.java b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVReaderBenchmark.java -index 3f242ce228..004db59f1f 100644 +index 3f242ce..004db59 100644 --- a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVReaderBenchmark.java +++ b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVReaderBenchmark.java @@ -240,6 +240,14 @@ public class DVReaderBenchmark { @@ -115,7 +129,7 @@ index 3f242ce228..004db59f1f 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java -index db57897240..b3fde306c8 100644 +index db57897..b3fde30 100644 --- a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java +++ b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/DVWriterBenchmark.java @@ -224,6 +224,14 @@ public class DVWriterBenchmark { @@ -134,7 +148,7 @@ index db57897240..b3fde306c8 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java -index debe37866f..62daec8291 100644 +index debe378..62daec8 100644 --- a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java +++ b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java @@ -95,7 +95,17 @@ public abstract class IcebergSourceBenchmark { @@ -157,7 +171,7 @@ index debe37866f..62daec8291 100644 builder .config("parquet.dictionary.page.size", "1") diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/SparkDistributedDataScanTestBase.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/SparkDistributedDataScanTestBase.java -index d1c724425c..325dcb95d1 100644 +index d1c7244..325dcb9 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/SparkDistributedDataScanTestBase.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/SparkDistributedDataScanTestBase.java @@ -90,6 +90,14 @@ public abstract class SparkDistributedDataScanTestBase @@ -176,7 +190,7 @@ index d1c724425c..325dcb95d1 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanDeletes.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanDeletes.java -index a21c6a08ec..1292883f55 100644 +index a21c6a0..1292883 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanDeletes.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanDeletes.java @@ -73,6 +73,14 @@ public class TestSparkDistributedDataScanDeletes @@ -195,7 +209,7 @@ index a21c6a08ec..1292883f55 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanFilterFiles.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanFilterFiles.java -index 5edf482822..92166f7b48 100644 +index 5edf482..92166f7 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanFilterFiles.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanFilterFiles.java @@ -62,6 +62,14 @@ public class TestSparkDistributedDataScanFilterFiles @@ -214,7 +228,7 @@ index 5edf482822..92166f7b48 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanReporting.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanReporting.java -index e6f3c75475..0a9a958b2c 100644 +index e6f3c75..0a9a958 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanReporting.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/TestSparkDistributedDataScanReporting.java @@ -63,6 +63,14 @@ public class TestSparkDistributedDataScanReporting @@ -233,7 +247,7 @@ index e6f3c75475..0a9a958b2c 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestBase.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestBase.java -index 507d7b313b..3b73dcc014 100644 +index 507d7b3..3b73dcc 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestBase.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestBase.java @@ -86,6 +86,14 @@ public abstract class TestBase extends SparkTestHelperBase { @@ -252,7 +266,7 @@ index 507d7b313b..3b73dcc014 100644 .config(DISABLE_UI) .enableHiveSupport() diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java -index b61ecfa2f4..d696e85139 100644 +index b61ecfa..d696e85 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java @@ -66,6 +66,14 @@ public class TestParquetDictionaryEncodedVectorizedReads extends TestParquetVect @@ -271,7 +285,7 @@ index b61ecfa2f4..d696e85139 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/ScanTestBase.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/ScanTestBase.java -index da9cd63921..fae1357a40 100644 +index da9cd63..fae1357 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/ScanTestBase.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/ScanTestBase.java @@ -62,6 +62,14 @@ public abstract class ScanTestBase extends AvroDataTestBase { @@ -290,7 +304,7 @@ index da9cd63921..fae1357a40 100644 .config(TestBase.DISABLE_UI) .getOrCreate(); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCompressionSettings.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCompressionSettings.java -index d381822483..043d6596c0 100644 +index d381822..043d659 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCompressionSettings.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestCompressionSettings.java @@ -144,7 +144,18 @@ public class TestCompressionSettings extends CatalogTestBase { @@ -314,7 +328,7 @@ index d381822483..043d6596c0 100644 @BeforeEach diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java -index e67ec5fd62..ed47a307c3 100644 +index e67ec5f..ed47a30 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestDataSourceOptions.java @@ -79,7 +79,18 @@ public class TestDataSourceOptions extends TestBaseWithCatalog { @@ -338,7 +352,7 @@ index e67ec5fd62..ed47a307c3 100644 @AfterAll diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java -index 24fecf4eb2..abab7798bc 100644 +index 24fecf4..abab779 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestFilteredScan.java @@ -117,6 +117,14 @@ public class TestFilteredScan { @@ -357,7 +371,7 @@ index 24fecf4eb2..abab7798bc 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestForwardCompatibility.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestForwardCompatibility.java -index d0103ff46e..62a7bdc04d 100644 +index d0103ff..62a7bdc 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestForwardCompatibility.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestForwardCompatibility.java @@ -99,6 +99,14 @@ public class TestForwardCompatibility { @@ -376,7 +390,7 @@ index d0103ff46e..62a7bdc04d 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java -index a637b975fe..26304e8894 100644 +index a637b97..26304e8 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSpark.java @@ -52,6 +52,14 @@ public class TestIcebergSpark { @@ -395,7 +409,7 @@ index a637b975fe..26304e8894 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionPruning.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionPruning.java -index 8098db81f9..0e83b02f45 100644 +index 8098db8..0e83b02 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionPruning.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionPruning.java @@ -120,6 +120,14 @@ public class TestPartitionPruning { @@ -414,7 +428,7 @@ index 8098db81f9..0e83b02f45 100644 .getOrCreate(); TestPartitionPruning.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext()); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionValues.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionValues.java -index 9b5b22a73f..5d41bbeb5e 100644 +index 9b5b22a..5d41bbe 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionValues.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestPartitionValues.java @@ -113,6 +113,14 @@ public class TestPartitionValues { @@ -433,7 +447,7 @@ index 9b5b22a73f..5d41bbeb5e 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java -index 3004e8fa5c..38b6a1f6a4 100644 +index 3004e8f..38b6a1f 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSnapshotSelection.java @@ -98,6 +98,14 @@ public class TestSnapshotSelection { @@ -452,7 +466,7 @@ index 3004e8fa5c..38b6a1f6a4 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataFile.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataFile.java -index d719ca6751..eba566120e 100644 +index d719ca6..eba5661 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataFile.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataFile.java @@ -127,6 +127,14 @@ public class TestSparkDataFile { @@ -471,7 +485,7 @@ index d719ca6751..eba566120e 100644 .getOrCreate(); TestSparkDataFile.sparkContext = JavaSparkContext.fromSparkContext(spark.sparkContext()); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java -index c2f5afef0e..0dac36653c 100644 +index c2f5afe..0dac366 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkDataWrite.java @@ -104,6 +104,14 @@ public class TestSparkDataWrite { @@ -562,7 +576,7 @@ index c2f5afef0e..0dac36653c 100644 .hasSize(records.size() + records2.size()) .containsExactlyInAnyOrder( diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadProjection.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadProjection.java -index de6a5e5902..e220f0dd31 100644 +index de6a5e5..e220f0d 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadProjection.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadProjection.java @@ -89,6 +89,14 @@ public class TestSparkReadProjection extends TestReadProjection { @@ -581,7 +595,7 @@ index de6a5e5902..e220f0dd31 100644 .getOrCreate(); ImmutableMap config = diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java -index 0d61930571..108d31b965 100644 +index 0d61930..108d31b 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderDeletes.java @@ -138,6 +138,14 @@ public class TestSparkReaderDeletes extends DeleteReadTests { @@ -610,7 +624,7 @@ index 0d61930571..108d31b965 100644 @Override diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java -index cb2f866fab..a161d541cf 100644 +index cb2f866..a161d54 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java @@ -183,6 +183,14 @@ public class TestSparkReaderWithBloomFilter { @@ -629,7 +643,7 @@ index cb2f866fab..a161d541cf 100644 .enableHiveSupport() .getOrCreate(); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java -index 5e900ea0ba..c9299d1ebf 100644 +index 5e900ea..c9299d1 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestStructuredStreaming.java @@ -73,6 +73,14 @@ public class TestStructuredStreaming { @@ -648,7 +662,7 @@ index 5e900ea0ba..c9299d1ebf 100644 .getOrCreate(); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTimestampWithoutZone.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTimestampWithoutZone.java -index 79781a8fc3..748b0a8327 100644 +index 79781a8..748b0a8 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTimestampWithoutZone.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestTimestampWithoutZone.java @@ -75,7 +75,18 @@ public class TestTimestampWithoutZone extends TestBase { @@ -672,7 +686,7 @@ index 79781a8fc3..748b0a8327 100644 @AfterAll diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java -index ab2479d610..9b2864bb33 100644 +index ab2479d..9b2864b 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java @@ -86,6 +86,14 @@ public class TestWriteMetricsConfig { @@ -691,7 +705,7 @@ index ab2479d610..9b2864bb33 100644 .getOrCreate(); TestWriteMetricsConfig.sc = JavaSparkContext.fromSparkContext(spark.sparkContext()); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java -index 1669301d2d..f48b4f674a 100644 +index 1669301..f48b4f6 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestAggregatePushDown.java @@ -63,6 +63,14 @@ public class TestAggregatePushDown extends CatalogTestBase { @@ -710,7 +724,7 @@ index 1669301d2d..f48b4f674a 100644 .enableHiveSupport() .getOrCreate(); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestFilterPushDown.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestFilterPushDown.java -index e5a9d63b68..220d445851 100644 +index e5a9d63..220d445 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestFilterPushDown.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestFilterPushDown.java @@ -667,9 +667,11 @@ public class TestFilterPushDown extends TestBaseWithCatalog { @@ -727,7 +741,7 @@ index e5a9d63b68..220d445851 100644 assertThat(planAsString).as("Should be no post scan filter").doesNotContain("Filter ("); } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java -index 8cdcf22e58..820f9fc03b 100644 +index 8cdcf22..820f9fc 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java @@ -100,6 +100,14 @@ public class TestVariantShredding extends CatalogTestBase {