Stable 25.8 Backport of #109551 - Fix crash in Iceberg due to type confusion

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp

@@ -856,8 +856,12 @@ void IcebergMetadata::addDeleteTransformers(
             {
                 NameAndTypePair name_and_type
                     = persistent_components.schema_processor->getFieldCharacteristics(delete_file.schema_id, col_id);
-                block_for_set.insert(ColumnWithTypeAndName(name_and_type.type, name_and_type.name));
-                equality_indexes_delete_file.push_back(delete_file_header.getPositionByName(name_and_type.name));
+                size_t position_in_delete_file = delete_file_header.getPositionByName(name_and_type.name);
+                /// Take the type from the delete file header rather than from the table schema:
+                /// the nullability of a column in the delete file may differ from its nullability
+                /// in the table schema, and the columns read below have exactly the header's types.
+                block_for_set.insert(ColumnWithTypeAndName(delete_file_header.getByPosition(position_in_delete_file).type, name_and_type.name));
+                equality_indexes_delete_file.push_back(position_in_delete_file);
             }
             /// Then we read the content of the delete file.
             auto mutable_columns_for_set = block_for_set.cloneEmptyColumns();

tests/queries/0_stateless/03595_equality_deletes_simple.reference

@@ -1,2 +1,5 @@
 448755
 448755	881
+--- equality deletes with mismatched nullability ---
+bar
+world

tests/queries/0_stateless/03595_equality_deletes_simple.sql

@@ -2,3 +2,12 @@
 
 SELECT sum(id) FROM icebergS3(s3_conn, filename = 'deletes_db/eq_deletes_table');
 SELECT sum(id), count(name) FROM icebergS3(s3_conn, filename = 'deletes_db/eq_deletes_table');
+
+-- Equality deletes on the `data` column, which is required in the table schema
+-- but optional in the delete file schema (their nullability may legally differ).
+SELECT '--- equality deletes with mismatched nullability ---';
+-- The table's metadata declares `location` as `/out/warehouse/default/eq_deletes_required_table`,
+-- so the queried path is kept as a component-aligned suffix of it. This branch resolves a data path
+-- with `getProperFilePathFromMetadataInfo`, whose fast path needs `table_location` to end with the
+-- queried path; `IcebergPathResolver`, which does not, is not in 25.8.
+SELECT data FROM icebergS3(s3_conn, filename = 'default/eq_deletes_required_table') ORDER BY data;

tests/queries/0_stateless/data_minio/default/eq_deletes_required_table/metadata/v1.metadata.json

@@ -0,0 +1,46 @@
+{
+  "format-version" : 2,
+  "table-uuid" : "35a5afa6-4947-4e9a-ab1a-0ae79844b538",
+  "location" : "/out/warehouse/default/eq_deletes_required_table",
+  "last-sequence-number" : 0,
+  "last-updated-ms" : 1783354452257,
+  "last-column-id" : 2,
+  "current-schema-id" : 0,
+  "schemas" : [ {
+    "type" : "struct",
+    "schema-id" : 0,
+    "fields" : [ {
+      "id" : 1,
+      "name" : "id",
+      "required" : true,
+      "type" : "int"
+    }, {
+      "id" : 2,
+      "name" : "data",
+      "required" : true,
+      "type" : "string"
+    } ]
+  } ],
+  "default-spec-id" : 0,
+  "partition-specs" : [ {
+    "spec-id" : 0,
+    "fields" : [ ]
+  } ],
+  "last-partition-id" : 999,
+  "default-sort-order-id" : 0,
+  "sort-orders" : [ {
+    "order-id" : 0,
+    "fields" : [ ]
+  } ],
+  "properties" : {
+    "owner" : "root",
+    "write.parquet.compression-codec" : "zstd"
+  },
+  "current-snapshot-id" : -1,
+  "refs" : { },
+  "snapshots" : [ ],
+  "statistics" : [ ],
+  "partition-statistics" : [ ],
+  "snapshot-log" : [ ],
+  "metadata-log" : [ ]
+}

tests/queries/0_stateless/data_minio/default/eq_deletes_required_table/metadata/v2.metadata.json

@@ -0,0 +1,81 @@
+{
+  "format-version" : 2,
+  "table-uuid" : "35a5afa6-4947-4e9a-ab1a-0ae79844b538",
+  "location" : "/out/warehouse/default/eq_deletes_required_table",
+  "last-sequence-number" : 1,
+  "last-updated-ms" : 1783354456558,
+  "last-column-id" : 2,
+  "current-schema-id" : 0,
+  "schemas" : [ {
+    "type" : "struct",
+    "schema-id" : 0,
+    "fields" : [ {
+      "id" : 1,
+      "name" : "id",
+      "required" : true,
+      "type" : "int"
+    }, {
+      "id" : 2,
+      "name" : "data",
+      "required" : true,
+      "type" : "string"
+    } ]
+  } ],
+  "default-spec-id" : 0,
+  "partition-specs" : [ {
+    "spec-id" : 0,
+    "fields" : [ ]
+  } ],
+  "last-partition-id" : 999,
+  "default-sort-order-id" : 0,
+  "sort-orders" : [ {
+    "order-id" : 0,
+    "fields" : [ ]
+  } ],
+  "properties" : {
+    "owner" : "root",
+    "write.parquet.compression-codec" : "zstd"
+  },
+  "current-snapshot-id" : 597133758536090507,
+  "refs" : {
+    "main" : {
+      "snapshot-id" : 597133758536090507,
+      "type" : "branch"
+    }
+  },
+  "snapshots" : [ {
+    "sequence-number" : 1,
+    "snapshot-id" : 597133758536090507,
+    "timestamp-ms" : 1783354456558,
+    "summary" : {
+      "operation" : "append",
+      "spark.app.id" : "local-1783354446783",
+      "added-data-files" : "1",
+      "added-records" : "4",
+      "added-files-size" : "684",
+      "changed-partition-count" : "1",
+      "total-records" : "4",
+      "total-files-size" : "684",
+      "total-data-files" : "1",
+      "total-delete-files" : "0",
+      "total-position-deletes" : "0",
+      "total-equality-deletes" : "0",
+      "engine-version" : "3.5.5",
+      "app-id" : "local-1783354446783",
+      "engine-name" : "spark",
+      "iceberg-version" : "Apache Iceberg 1.8.1 (commit 9ce0fcf0af7becf25ad9fc996c3bad2afdcfd33d)"
+    },
+    "manifest-list" : "/out/warehouse/default/eq_deletes_required_table/metadata/snap-597133758536090507-1-183cb62a-4751-4276-93b1-989e617b5d04.avro",
+    "schema-id" : 0
+  } ],
+  "statistics" : [ ],
+  "partition-statistics" : [ ],
+  "snapshot-log" : [ {
+    "timestamp-ms" : 1783354456558,
+    "snapshot-id" : 597133758536090507
+  } ],
+  "metadata-log" : [ {
+    "timestamp-ms" : 1783354452257,
+    "metadata-file" : "/out/warehouse/default/eq_deletes_required_table/metadata/v1.metadata.json"
+  } ]
+}

tests/queries/0_stateless/data_minio/default/eq_deletes_required_table/metadata/v3.metadata.json

@@ -0,0 +1,112 @@
+{
+  "format-version" : 2,
+  "table-uuid" : "35a5afa6-4947-4e9a-ab1a-0ae79844b538",
+  "location" : "/out/warehouse/default/eq_deletes_required_table",
+  "last-sequence-number" : 2,
+  "last-updated-ms" : 1783354456969,
+  "last-column-id" : 2,
+  "current-schema-id" : 0,
+  "schemas" : [ {
+    "type" : "struct",
+    "schema-id" : 0,
+    "fields" : [ {
+      "id" : 1,
+      "name" : "id",
+      "required" : true,
+      "type" : "int"
+    }, {
+      "id" : 2,
+      "name" : "data",
+      "required" : true,
+      "type" : "string"
+    } ]
+  } ],
+  "default-spec-id" : 0,
+  "partition-specs" : [ {
+    "spec-id" : 0,
+    "fields" : [ ]
+  } ],
+  "last-partition-id" : 999,
+  "default-sort-order-id" : 0,
+  "sort-orders" : [ {
+    "order-id" : 0,
+    "fields" : [ ]
+  } ],
+  "properties" : {
+    "owner" : "root",
+    "write.parquet.compression-codec" : "zstd"
+  },
+  "current-snapshot-id" : 667782107838695408,
+  "refs" : {
+    "main" : {
+      "snapshot-id" : 667782107838695408,
+      "type" : "branch"
+    }
+  },
+  "snapshots" : [ {
+    "sequence-number" : 1,
+    "snapshot-id" : 597133758536090507,
+    "timestamp-ms" : 1783354456558,
+    "summary" : {
+      "operation" : "append",
+      "spark.app.id" : "local-1783354446783",
+      "added-data-files" : "1",
+      "added-records" : "4",
+      "added-files-size" : "684",
+      "changed-partition-count" : "1",
+      "total-records" : "4",
+      "total-files-size" : "684",
+      "total-data-files" : "1",
+      "total-delete-files" : "0",
+      "total-position-deletes" : "0",
+      "total-equality-deletes" : "0",
+      "engine-version" : "3.5.5",
+      "app-id" : "local-1783354446783",
+      "engine-name" : "spark",
+      "iceberg-version" : "Apache Iceberg 1.8.1 (commit 9ce0fcf0af7becf25ad9fc996c3bad2afdcfd33d)"
+    },
+    "manifest-list" : "/out/warehouse/default/eq_deletes_required_table/metadata/snap-597133758536090507-1-183cb62a-4751-4276-93b1-989e617b5d04.avro",
+    "schema-id" : 0
+  }, {
+    "sequence-number" : 2,
+    "snapshot-id" : 667782107838695408,
+    "parent-snapshot-id" : 597133758536090507,
+    "timestamp-ms" : 1783354456969,
+    "summary" : {
+      "operation" : "delete",
+      "added-equality-delete-files" : "1",
+      "added-delete-files" : "1",
+      "added-files-size" : "489",
+      "added-equality-deletes" : "2",
+      "changed-partition-count" : "1",
+      "total-records" : "4",
+      "total-files-size" : "1173",
+      "total-data-files" : "1",
+      "total-delete-files" : "1",
+      "total-position-deletes" : "0",
+      "total-equality-deletes" : "2",
+      "engine-version" : "3.5.5",
+      "app-id" : "local-1783354446783",
+      "engine-name" : "spark",
+      "iceberg-version" : "Apache Iceberg 1.8.1 (commit 9ce0fcf0af7becf25ad9fc996c3bad2afdcfd33d)"
+    },
+    "manifest-list" : "/out/warehouse/default/eq_deletes_required_table/metadata/snap-667782107838695408-1-373fb659-dd36-47eb-81a0-32bbe12c0e09.avro",
+    "schema-id" : 0
+  } ],
+  "statistics" : [ ],
+  "partition-statistics" : [ ],
+  "snapshot-log" : [ {
+    "timestamp-ms" : 1783354456558,
+    "snapshot-id" : 597133758536090507
+  }, {
+    "timestamp-ms" : 1783354456969,
+    "snapshot-id" : 667782107838695408
+  } ],
+  "metadata-log" : [ {
+    "timestamp-ms" : 1783354452257,
+    "metadata-file" : "/out/warehouse/default/eq_deletes_required_table/metadata/v1.metadata.json"
+  }, {
+    "timestamp-ms" : 1783354456558,
+    "metadata-file" : "/out/warehouse/default/eq_deletes_required_table/metadata/v2.metadata.json"
+  } ]
+}

tests/queries/0_stateless/data_minio/default/eq_deletes_required_table/metadata/version-hint.text

@@ -0,0 +1 @@
+3

tests/queries/0_stateless/data_minio/deletes_db/README.md

@@ -217,3 +217,76 @@ public class IcebergInsert {
 }
 
 ```
+
+## How is `eq_deletes_required_table` generated?
+
+The table has a `data STRING NOT NULL` column (required in the Iceberg schema) and an
+equality delete file in which the same column is optional, so the nullability of the
+column differs between the delete file and the table schema. The delete file removes
+`hello` and `foo`, leaving 2 of 4 rows.
+
+Generated with the Iceberg Java API through the PySpark JVM gateway (Spark SQL alone
+cannot write equality deletes), inside the `clickhouse/integration-tests-runner` image:
+
+``` python
+import pyspark
+
+spark = (
+    pyspark.sql.SparkSession.builder.appName("gen")
+    .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog")
+    .config("spark.sql.catalog.spark_catalog.type", "hadoop")
+    .config("spark.sql.catalog.spark_catalog.warehouse", "/out/warehouse")
+    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
+    .master("local")
+    .getOrCreate()
+)
+
+TABLE = "eq_deletes_required_table"
+spark.sql(
+    f"CREATE TABLE {TABLE} (id INT NOT NULL, data STRING NOT NULL) USING iceberg"
+    " TBLPROPERTIES ('format-version'='2')"
+)
+spark.sql(f"INSERT INTO {TABLE} VALUES (1, 'hello'), (2, 'world'), (3, 'foo'), (4, 'bar')")
+
+# Write an equality delete file whose `data` column is optional in the file schema,
+# although it is required in the table schema.
+jvm = spark.sparkContext._jvm
+gateway = spark.sparkContext._gateway
+hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()
+
+table = jvm.org.apache.iceberg.hadoop.HadoopTables(hadoop_conf).load(f"/out/warehouse/default/{TABLE}")
+schema = table.schema()
+field_id = schema.findField("data").fieldId()
+types = jvm.org.apache.iceberg.types.Types
+delete_field = types.NestedField.optional(field_id, "data", types.StringType.get())
+delete_schema = jvm.org.apache.iceberg.Schema([delete_field])
+
+equality_ids = gateway.new_array(jvm.int, 1)
+equality_ids[0] = field_id
+
+appender_factory = jvm.org.apache.iceberg.data.GenericAppenderFactory(
+    schema, table.spec(), equality_ids, delete_schema, None
+)
+out = table.io().newOutputFile(table.locationProvider().newDataLocation("eq-delete-required-optional.parquet"))
+writer = appender_factory.newEqDeleteWriter(
+    jvm.org.apache.iceberg.encryption.EncryptedFiles.plainAsEncryptedOutput(out),
+    jvm.org.apache.iceberg.FileFormat.PARQUET,
+    None,
+)
+for value in ["hello", "foo"]:
+    record = jvm.org.apache.iceberg.data.GenericRecord.create(delete_schema)
+    record.setField("data", value)
+    writer.write(record)
+writer.close()
+
+table.newRowDelta().addDeletes(writer.toDeleteFile()).commit()
+spark.stop()
+```
+
+The resulting `/out/warehouse/default/eq_deletes_required_table` directory is copied here
+as is, except for the `.*.crc` files.
+
+On this branch `eq_deletes_required_table` is kept at `data_minio/default/eq_deletes_required_table`
+instead of here, so that the path the test queries is a component-aligned suffix of the `location`
+recorded in its metadata (`/out/warehouse/default/eq_deletes_required_table`). 25.8 resolves data
+paths with `getProperFilePathFromMetadataInfo`, which needs that alignment.