From f2fae8822975712a3256f8d8d69c0768cd08fd6c Mon Sep 17 00:00:00 2001 From: Eduard Tudenhoefner Date: Thu, 16 Oct 2025 09:54:15 +0200 Subject: [PATCH 1/3] Core, Spark: Handle unknown type during deletes --- .../org/apache/iceberg/PartitionSpec.java | 9 +++- .../org/apache/iceberg/types/Comparators.java | 1 + .../org/apache/iceberg/types/Conversions.java | 6 +++ .../org/apache/iceberg/avro/ValueWriters.java | 10 ++++- .../apache/iceberg/util/ManifestFileUtil.java | 6 +++ .../org/apache/iceberg/TestPartitioning.java | 43 +++++++++++++++++++ .../TestAlterTablePartitionFields.java | 9 ++-- 7 files changed, 77 insertions(+), 7 deletions(-) diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 30b9e2bc618b..2e9632e89044 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -184,8 +184,13 @@ public Class[] javaClasses() { classes[i] = Object.class; } else { Type sourceType = schema.findType(field.sourceId()); - Type result = field.transform().getResultType(sourceType); - classes[i] = result.typeId().javaClass(); + if (null == sourceType) { + // When the source field has been dropped we cannot determine the type + classes[i] = Types.UnknownType.get().typeId().javaClass(); + } else { + Type result = field.transform().getResultType(sourceType); + classes[i] = result.typeId().javaClass(); + } } } diff --git a/api/src/main/java/org/apache/iceberg/types/Comparators.java b/api/src/main/java/org/apache/iceberg/types/Comparators.java index 32168d9a0904..937307bcf33f 100644 --- a/api/src/main/java/org/apache/iceberg/types/Comparators.java +++ b/api/src/main/java/org/apache/iceberg/types/Comparators.java @@ -48,6 +48,7 @@ private Comparators() {} .put(Types.StringType.get(), Comparators.charSequences()) .put(Types.UUIDType.get(), Comparator.naturalOrder()) .put(Types.BinaryType.get(), Comparators.unsignedBytes()) + .put(Types.UnknownType.get(), Comparator.naturalOrder()) .buildOrThrow(); public static Comparator forType(Types.StructType struct) { diff --git a/api/src/main/java/org/apache/iceberg/types/Conversions.java b/api/src/main/java/org/apache/iceberg/types/Conversions.java index 3b840ba6dfc8..5966c894be5f 100644 --- a/api/src/main/java/org/apache/iceberg/types/Conversions.java +++ b/api/src/main/java/org/apache/iceberg/types/Conversions.java @@ -131,6 +131,9 @@ public static ByteBuffer toByteBuffer(Type.TypeID typeId, Object value) { variantMetadata.writeTo(variantBuffer, 0); variantValue.writeTo(variantBuffer, variantMetadata.sizeInBytes()); return variantBuffer; + case UNKNOWN: + // underlying type not known + return null; default: throw new UnsupportedOperationException("Cannot serialize type: " + typeId); } @@ -193,6 +196,9 @@ private static Object internalFromByteBuffer(Type type, ByteBuffer buffer) { return new BigDecimal(new BigInteger(unscaledBytes), decimal.scale()); case VARIANT: return Variant.from(tmp); + case UNKNOWN: + // underlying type not known + return null; default: throw new UnsupportedOperationException("Cannot deserialize type: " + type); } diff --git a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java index 580175c5f839..e387be5ee9f6 100644 --- a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java +++ b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java @@ -580,7 +580,15 @@ public ValueWriter writer(int pos) { @Override public void write(S row, Encoder encoder) throws IOException { for (int i = 0; i < writers.length; i += 1) { - writers[i].write(get(row, i), encoder); + Object datum = get(row, i); + ValueWriter writer = writers[i]; + + if (NullWriter.INSTANCE.getClass().equals(writer.getClass()) && null != datum) { + // this is an UnknownType that has a value + writer.write(null, encoder); + } else { + writer.write(datum, encoder); + } } } } diff --git a/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java b/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java index a73a00d0e6fd..37d623a6b3b8 100644 --- a/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java +++ b/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java @@ -41,9 +41,11 @@ private static class FieldSummary { private final T upperBound; private final boolean containsNull; private final boolean containsNaN; + private final Type.PrimitiveType type; @SuppressWarnings("unchecked") FieldSummary(Type.PrimitiveType primitive, ManifestFile.PartitionFieldSummary summary) { + this.type = primitive; this.comparator = Comparators.forType(primitive); this.javaClass = (Class) primitive.typeId().javaClass(); this.lowerBound = Conversions.fromByteBuffer(primitive, summary.lowerBound()); @@ -53,6 +55,10 @@ private static class FieldSummary { } boolean canContain(Object value) { + if (Types.UnknownType.get().equals(type)) { + return true; + } + if (value == null) { return containsNull; } diff --git a/core/src/test/java/org/apache/iceberg/TestPartitioning.java b/core/src/test/java/org/apache/iceberg/TestPartitioning.java index eb77a693c799..dc362d33c335 100644 --- a/core/src/test/java/org/apache/iceberg/TestPartitioning.java +++ b/core/src/test/java/org/apache/iceberg/TestPartitioning.java @@ -459,4 +459,47 @@ public void testDeletingPartitionField() { assertThat(table.spec()).isEqualTo(spec); } + + @Test + public void deleteFileAfterDeletingAllPartitionFields() { + TestTables.TestTable table = + TestTables.create(tableDir, "test", SCHEMA, BY_DATA_SPEC, V2_FORMAT_VERSION); + + DataFile dataFile = + DataFiles.builder(BY_DATA_SPEC) + .withPath("/path/to/data-a.parquet") + .withFileSizeInBytes(10) + .withPartitionPath("data=1") + .withRecordCount(1) + .build(); + + table.newAppend().appendFile(dataFile).commit(); + assertThat(table.currentSnapshot().summary()).containsEntry("added-data-files", "1"); + table.updateSpec().removeField("data").commit(); + table.updateSchema().deleteColumn("data").commit(); + table.newDelete().deleteFile(dataFile).commit(); + assertThat(table.currentSnapshot().summary()).containsEntry("deleted-data-files", "1"); + } + + @Test + public void deleteFileAfterDeletingOnePartitionField() { + TestTables.TestTable table = + TestTables.create(tableDir, "test", SCHEMA, BY_CATEGORY_DATA_SPEC, V2_FORMAT_VERSION); + + // drop one out of 2 partition fields + DataFile dataFile = + DataFiles.builder(BY_CATEGORY_DATA_SPEC) + .withPath("/path/to/data-b.parquet") + .withFileSizeInBytes(10) + .withPartitionPath("category=2/data=2") + .withRecordCount(1) + .build(); + + table.newAppend().appendFile(dataFile).commit(); + assertThat(table.currentSnapshot().summary()).containsEntry("added-data-files", "1"); + table.updateSpec().removeField("data").commit(); + table.updateSchema().deleteColumn("data").commit(); + table.newDelete().deleteFile(dataFile).commit(); + assertThat(table.currentSnapshot().summary()).containsEntry("deleted-data-files", "1"); + } } diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java index d3d0d9b9103e..8c0eb108c1ca 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java +++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java @@ -601,10 +601,11 @@ private void runCreateAndDropPartitionField( sql("INSERT INTO %s VALUES (3000, CAST('2024-05-01 19:25:00' as TIMESTAMP), 2300)", tableName); sql("ALTER TABLE %s DROP COLUMN %s", tableName, column); - assertEquals( - "Should return correct data", - expected, - sql("SELECT * FROM %s WHERE %s ORDER BY col_int", tableName, predicate)); + assertThat(sql("SELECT * FROM %s WHERE %s ORDER BY col_int", tableName, predicate)) + .containsExactlyElementsOf(expected); + + sql("DELETE FROM %s WHERE %s", tableName, predicate); + assertThat(sql("SELECT * FROM %s WHERE %s", tableName, predicate)).isEmpty(); } @TestTemplate From d2441c539a44aa0378f40f4a1ff377734ca74e6e Mon Sep 17 00:00:00 2001 From: Eduard Tudenhoefner Date: Thu, 6 Nov 2025 08:46:29 +0100 Subject: [PATCH 2/3] review feedback --- .../org/apache/iceberg/PartitionSpec.java | 8 +- .../org/apache/iceberg/types/Comparators.java | 2 +- .../org/apache/iceberg/avro/ValueWriters.java | 6 + .../apache/iceberg/util/ManifestFileUtil.java | 8 +- .../iceberg/util/TestManifestFileUtil.java | 127 ++++++++++++++++++ .../TestAlterTablePartitionFields.java | 23 +++- 6 files changed, 162 insertions(+), 12 deletions(-) create mode 100644 core/src/test/java/org/apache/iceberg/util/TestManifestFileUtil.java diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 2e9632e89044..bbd59aa083de 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -186,11 +186,11 @@ public Class[] javaClasses() { Type sourceType = schema.findType(field.sourceId()); if (null == sourceType) { // When the source field has been dropped we cannot determine the type - classes[i] = Types.UnknownType.get().typeId().javaClass(); - } else { - Type result = field.transform().getResultType(sourceType); - classes[i] = result.typeId().javaClass(); + sourceType = Types.UnknownType.get(); } + + Type result = field.transform().getResultType(sourceType); + classes[i] = result.typeId().javaClass(); } } diff --git a/api/src/main/java/org/apache/iceberg/types/Comparators.java b/api/src/main/java/org/apache/iceberg/types/Comparators.java index 937307bcf33f..ab59c895686d 100644 --- a/api/src/main/java/org/apache/iceberg/types/Comparators.java +++ b/api/src/main/java/org/apache/iceberg/types/Comparators.java @@ -48,7 +48,7 @@ private Comparators() {} .put(Types.StringType.get(), Comparators.charSequences()) .put(Types.UUIDType.get(), Comparator.naturalOrder()) .put(Types.BinaryType.get(), Comparators.unsignedBytes()) - .put(Types.UnknownType.get(), Comparator.naturalOrder()) + .put(Types.UnknownType.get(), Comparator.nullsFirst(Comparator.naturalOrder())) .buildOrThrow(); public static Comparator forType(Types.StructType struct) { diff --git a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java index e387be5ee9f6..50cc797dccd6 100644 --- a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java +++ b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java @@ -46,6 +46,10 @@ public class ValueWriters { private ValueWriters() {} + /** + * @deprecated since 1.11.0, return type will be changed to {@code ValueWriter} in 1.12.0 + */ + @Deprecated public static ValueWriter nulls() { return NullWriter.INSTANCE; } @@ -583,6 +587,8 @@ public void write(S row, Encoder encoder) throws IOException { Object datum = get(row, i); ValueWriter writer = writers[i]; + // TODO: remove this workaround once the return type of ValueWriters.nulls() has been + // changed from ValueWriter to ValueWriter if (NullWriter.INSTANCE.getClass().equals(writer.getClass()) && null != datum) { // this is an UnknownType that has a value writer.write(null, encoder); diff --git a/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java b/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java index 37d623a6b3b8..56a385ebf2f4 100644 --- a/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java +++ b/core/src/main/java/org/apache/iceberg/util/ManifestFileUtil.java @@ -55,10 +55,6 @@ private static class FieldSummary { } boolean canContain(Object value) { - if (Types.UnknownType.get().equals(type)) { - return true; - } - if (value == null) { return containsNull; } @@ -67,6 +63,10 @@ boolean canContain(Object value) { return containsNaN; } + if (Types.UnknownType.get().equals(type)) { + return true; + } + // if lower bound is null, then there are no non-null values if (lowerBound == null) { // the value is non-null, so it cannot match diff --git a/core/src/test/java/org/apache/iceberg/util/TestManifestFileUtil.java b/core/src/test/java/org/apache/iceberg/util/TestManifestFileUtil.java new file mode 100644 index 000000000000..8d2416032058 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/util/TestManifestFileUtil.java @@ -0,0 +1,127 @@ +/* + * 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. + */ +package org.apache.iceberg.util; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.file.Path; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.Files; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestWriter; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +public class TestManifestFileUtil { + private static final Schema SCHEMA = + new Schema( + optional(1, "id", Types.IntegerType.get()), + optional(2, "unknown", Types.UnknownType.get()), + optional(3, "floats", Types.FloatType.get())); + + @TempDir private Path temp; + + @Test + public void canContainWithUnknownTypeOnly() throws IOException { + PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("unknown").build(); + PartitionData partition = new PartitionData(spec.partitionType()); + partition.set(0, "someValue"); + ManifestFile manifestFile = writeManifestWithDataFile(spec, partition); + + assertThat( + ManifestFileUtil.canContainAny( + manifestFile, + ImmutableList.of(Pair.of(spec.specId(), partition)), + ImmutableMap.of(spec.specId(), spec))) + .isTrue(); + } + + @Test + public void canContainWithNaNValueOnly() throws IOException { + PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("floats").build(); + PartitionData partition = new PartitionData(spec.partitionType()); + partition.set(0, Float.NaN); + ManifestFile manifestFile = writeManifestWithDataFile(spec, partition); + + assertThat( + ManifestFileUtil.canContainAny( + manifestFile, + ImmutableList.of(Pair.of(spec.specId(), partition)), + ImmutableMap.of(spec.specId(), spec))) + .isTrue(); + } + + @Test + public void canContainWithNullValueOnly() throws IOException { + PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("floats").build(); + PartitionData partition = new PartitionData(spec.partitionType()); + partition.set(0, null); + ManifestFile manifestFile = writeManifestWithDataFile(spec, partition); + + assertThat( + ManifestFileUtil.canContainAny( + manifestFile, + ImmutableList.of(Pair.of(spec.specId(), partition)), + ImmutableMap.of(spec.specId(), spec))) + .isTrue(); + } + + @Test + public void canContainWithUnknownType() throws IOException { + PartitionSpec spec = + PartitionSpec.builderFor(SCHEMA).identity("floats").identity("unknown").build(); + PartitionData partition = new PartitionData(spec.partitionType()); + partition.set(0, 1.0f); + partition.set(1, "someValue"); + ManifestFile manifestFile = writeManifestWithDataFile(spec, partition); + + assertThat( + ManifestFileUtil.canContainAny( + manifestFile, + ImmutableList.of(Pair.of(spec.specId(), partition)), + ImmutableMap.of(spec.specId(), spec))) + .isTrue(); + } + + private ManifestFile writeManifestWithDataFile(PartitionSpec spec, PartitionData partition) + throws IOException { + ManifestWriter writer = ManifestFiles.write(spec, Files.localOutput(temp.toFile())); + try (writer) { + writer.add( + DataFiles.builder(spec) + .withPath("/path/to/data-a.parquet") + .withFileSizeInBytes(10) + .withPartition(partition) + .withRecordCount(10) + .build()); + } + + return writer.toManifestFile(); + } +} diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java index 8c0eb108c1ca..296564e20d4a 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java +++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java @@ -603,9 +603,6 @@ private void runCreateAndDropPartitionField( assertThat(sql("SELECT * FROM %s WHERE %s ORDER BY col_int", tableName, predicate)) .containsExactlyElementsOf(expected); - - sql("DELETE FROM %s WHERE %s", tableName, predicate); - assertThat(sql("SELECT * FROM %s WHERE %s", tableName, predicate)).isEmpty(); } @TestTemplate @@ -630,4 +627,24 @@ public void testDropPartitionAndSourceColumnTimestamp() { runCreateAndDropPartitionField("col_long", "truncate(2, col_long)", expected, predicate); runCreateAndDropPartitionField("col_long", "bucket(16, col_long)", expected, predicate); } + + @TestTemplate + public void deleteAfterDroppingPartitionAndSourceColumn() { + sql("DROP TABLE IF EXISTS %s", tableName); + sql( + "CREATE TABLE %s (id INTEGER, data STRING) USING ICEBERG TBLPROPERTIES ('format-version' = %d)", + tableName, formatVersion); + sql("INSERT INTO %s VALUES (1, 'data1')", tableName); + sql("ALTER TABLE %s ADD PARTITION FIELD data", tableName); + sql("INSERT INTO %s VALUES (2, 'data2')", tableName); + sql("ALTER TABLE %s DROP PARTITION FIELD data", tableName); + sql("INSERT INTO %s VALUES (3, 'data3')", tableName); + sql("ALTER TABLE %s DROP COLUMN data", tableName); + + assertThat(sql("SELECT * FROM %s WHERE id >= 1 ORDER BY id", tableName)) + .containsExactly(row(1), row(2), row(3)); + + sql("DELETE FROM %s WHERE id >= 1", tableName); + assertThat(sql("SELECT * FROM %s WHERE id >= 1", tableName)).isEmpty(); + } } From d0121ab122ac6bbdc6c6d1014b43343abc7fcef7 Mon Sep 17 00:00:00 2001 From: Eduard Tudenhoefner Date: Mon, 10 Nov 2025 08:35:17 +0100 Subject: [PATCH 3/3] review feedback --- .../org/apache/iceberg/avro/ValueWriters.java | 23 ++++--------------- 1 file changed, 5 insertions(+), 18 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java index 50cc797dccd6..b31d157302b9 100644 --- a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java +++ b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java @@ -46,10 +46,6 @@ public class ValueWriters { private ValueWriters() {} - /** - * @deprecated since 1.11.0, return type will be changed to {@code ValueWriter} in 1.12.0 - */ - @Deprecated public static ValueWriter nulls() { return NullWriter.INSTANCE; } @@ -148,13 +144,14 @@ public static ValueWriter struct(List> writers) { return new StructLikeWriter(writers); } - private static class NullWriter implements ValueWriter { - private static final NullWriter INSTANCE = new NullWriter(); + private static class NullWriter implements ValueWriter { + @SuppressWarnings({"unchecked", "rawtypes"}) + private static final ValueWriter INSTANCE = (ValueWriter) new NullWriter(); private NullWriter() {} @Override - public void write(Void ignored, Encoder encoder) throws IOException { + public void write(Object ignored, Encoder encoder) throws IOException { encoder.writeNull(); } } @@ -584,17 +581,7 @@ public ValueWriter writer(int pos) { @Override public void write(S row, Encoder encoder) throws IOException { for (int i = 0; i < writers.length; i += 1) { - Object datum = get(row, i); - ValueWriter writer = writers[i]; - - // TODO: remove this workaround once the return type of ValueWriters.nulls() has been - // changed from ValueWriter to ValueWriter - if (NullWriter.INSTANCE.getClass().equals(writer.getClass()) && null != datum) { - // this is an UnknownType that has a value - writer.write(null, encoder); - } else { - writer.write(datum, encoder); - } + writers[i].write(get(row, i), encoder); } } }