salimTutuncu opened a new issue, #972: URL: https://github.com/apache/incubator-xtable/issues/972
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/incubator-xtable/issues?q=is%3Aissue) and found no similar issues. ### Please describe the bug 🐞 ### Search before asking - [x] I had searched in the [issues](https://github.com/apache/incubator-xtable/issues?q=is%3Aissue) and found no similar issues. ### Please describe the bug 🐞 With `delta.columnMapping.mode=name` set at table creation, the Delta Kernel conversion source (`useDeltaKernel=true`) writes NULL identity-partition values into the Iceberg target. Delta stores `USA` / `Italy` in its log and reads them correctly; the Iceberg target returns `country = NULL` for every row in Spark 4.1.1 / Iceberg 1.11.0, PyIceberg 0.12.0 and DuckDB 1.5.6, and `WHERE country = 'USA'` returns zero rows instead of two (PyIceberg `plan_files()` selects no files). The sync reports `SUCCESS`. ```text expected (Delta source): (1, USA, John Smith) (2, USA, Mary Johnson) (6, Italy, Luca Verdi) actual (Iceberg target): (1, NULL, John Smith) (2, NULL, Mary Johnson) (6, NULL, Luca Verdi) WHERE country = 'USA': expected 2 rows, actual 0 rows in Spark, PyIceberg and DuckDB ``` Reproduced on unmodified `main` 9d9f9b3 (`0.5.0-SNAPSHOT`) in both `FULL` and `INCREMENTAL` mode (returned modes, same Iceberg table UUID across the incremental sync), and on `0.4.0-incubating` and `0.3.0-incubating`. The pre-rename baseline already shows the NULLs, so a rename is not required to trigger it; the fixture keeps one rename only so it can be compared with #B and #C. ### Why this is not a duplicate of #828 #828 fixed NULL partition values for composite generated-column partitions (date transforms). This is a plain identity partition (`PARTITIONED BY (country)`), and the trigger is column mapping: the add action keys the partition value by the physical column name. #765 / #766 made column-mapped Delta sources convertible to Iceberg, but partition values under column mapping on the Kernel path are not covered by that test. Relevant to #886 prerequisite 2 (Kernel path verification before it becomes the default). ### Root cause Delta PROTOCOL.md, Writer Requirements for Column Mapping: "Track partition values, column level statistics, and clustering column names with the physical name of the column in the transaction log." The add actions in this fixture carry `partitionValues: {"col-<uuid>": "USA"}`. `DeltaKernelPartitionExtractor.getSerializedPartitionValue` (https://github.com/apache/incubator-xtable/blob/9d9f9b3ced22adea03001b4b5e547b51fa7c8adc/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelPartitionExtractor.java) resolves a `VALUE` partition with `values.getOrDefault(partitionField.getSourceField().getName(), null)`, i.e. the logical name `country`. That key does not exist in the map, the value is null, and the Iceberg data file is written with partition `Record[None]`. The partition column is not stored in the Delta Parquet files, so no reader can recover it afterwards. `DeltaPartitionExtractor` (Standalone, https://github.com/apache/incubator-xtable/blob/9d9f9b3ced22adea03001b4b5e547b51fa7c8adc/xtable-core/src/main/java/org/apache/xtable/delta/DeltaPartitionExtractor.java) has the same lookup; I did not run the Standalone source in this test. Exact log keys, Parquet fields and generated Iceberg partition records are in the attached `physical-files-and-par titions.json`. ### To Reproduce Environment: sync host Spark 3.5.5 / JDK 17; XTable `main` 9d9f9b3 built with Java 11 (`./mvnw -pl xtable-spark-runtime -am -DskipTests package`, artifact SHA-256 `f863a05924b31230bf4ba2a655e7e0712b5d96c91dfcdf1171fba5c751759fc6`); classpath: the built `xtable-spark-runtime_2.12-0.5.0-SNAPSHOT.jar`, `delta-kernel-api-4.0.0`, `delta-kernel-defaults-4.0.0`, `delta-storage-3.3.0`, `iceberg-spark-runtime-3.5_2.12-1.9.2`, `avro-1.12.0`, `jol-core-0.16`, `spark.driver.userClassPathFirst=true`. Source table written with Delta 4.2.0 / Spark 4.1.1, deletion vectors disabled. Local path-based tables, no catalog, fresh consumer processes. ```sql CREATE TABLE delta_mapped (id INT, country STRING, name STRING) USING DELTA PARTITIONED BY (country) LOCATION '<path>' TBLPROPERTIES ('delta.enableDeletionVectors'='false', 'delta.columnMapping.mode'='name'); INSERT INTO delta_mapped (id, name, country) VALUES (1, 'John Smith', 'USA'), (2, 'Mary Johnson', 'USA'); -- incremental case: sync here and read the two-row target (country is already NULL) ALTER TABLE delta_mapped RENAME COLUMN name TO new_name; INSERT INTO delta_mapped (id, new_name, country) VALUES (6, 'Luca Verdi', 'Italy'); -- sync: first call returns FULL, second call returns INCREMENTAL ``` ```python targets = spark._jvm.java.util.ArrayList(); targets.add("ICEBERG") spec = (spark._jvm.org.apache.xtable.spark.TableSyncSpec.builder() .key("delta_mapped").basePath(path).sourceFormat("DELTA") .targets(targets).useDeltaKernel(True).build()) r = spark._jvm.org.apache.xtable.spark.XTableSyncService().sync(spec, spark.sparkContext._jsc.hadoopConfiguration()) print(r.get("ICEBERG").getMode(), r.get("ICEBERG").getTableFormatSyncStatus().getStatusCode()) ``` Observed: `FULL SUCCESS` then `INCREMENTAL SUCCESS`; `country` is NULL for all rows in all three readers; `country = 'USA'` returns `[]` in all three; PyIceberg `scan(row_filter=EqualTo("country", "USA")).plan_files()` is empty. Contrast: the same table without `delta.columnMapping.mode` returns the correct partition values in all three readers (pre-mapping baseline of #B). Generated metadata for the cold target (schema and spec are correct, the per-file partition records are not): ```json {"partition-specs":[{"spec-id":0,"fields":[]}, {"spec-id":1,"fields":[{"name":"country","transform":"identity","source-id":2,"field-id":1000}]}], "default-spec-id":1, "schema.name-mapping.default":[{"field-id":1,"names":["id"]},{"field-id":2,"names":["country"]},{"field-id":3,"names":["new_name"]}]} ``` ### Expected behavior Partition values resolved by the physical column name when column mapping is enabled, so the Iceberg manifests carry `USA` / `Italy` and partition filters prune correctly, on both the Kernel and the Standalone source. Failing that, refuse or warn on column-mapped partitioned sources instead of reporting `SUCCESS` with NULL partitions. ### Affected versions `main` 9d9f9b3 (0.5.0-SNAPSHOT, Delta Kernel 4.0.0); `0.4.0-incubating` spark-runtime (Delta Kernel 4.0.0); `0.3.0-incubating` utilities (Delta reader 2.4.0). Standalone source on `main` not run. ### Are you willing to submit PR? - [ ] I am willing to submit a PR! - [ ] I am willing to submit a PR but need help getting started! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
