diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java index 30b9e2bc618b..bbd59aa083de 100644 --- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java +++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java @@ -184,6 +184,11 @@ public Class[] javaClasses() { classes[i] = Object.class; } else { Type sourceType = schema.findType(field.sourceId()); + if (null == sourceType) { + // When the source field has been dropped we cannot determine the type + 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 32168d9a0904..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,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.nullsFirst(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..b31d157302b9 100644 --- a/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java +++ b/core/src/main/java/org/apache/iceberg/avro/ValueWriters.java @@ -144,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(); } } 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..56a385ebf2f4 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()); @@ -61,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/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/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 d3d0d9b9103e..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 @@ -601,10 +601,8 @@ 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); } @TestTemplate @@ -629,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(); + } }