diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 1d1505a28c05..45500e9fae49 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -20,6 +20,8 @@ import java.io.Closeable; import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Base64; import java.util.Collections; import java.util.List; import java.util.Map; @@ -45,6 +47,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; @@ -70,6 +73,8 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.Literal; import org.apache.iceberg.flink.util.FlinkAlterTableUtil; import org.apache.iceberg.flink.util.FlinkCompatibilityUtil; import org.apache.iceberg.io.CloseableIterable; @@ -80,6 +85,10 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -725,8 +734,139 @@ public void createPartition( @Override public void dropPartition( ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) - throws CatalogException { - throw new UnsupportedOperationException(); + throws PartitionNotExistException, CatalogException { + Table table; + try { + table = loadIcebergTable(tablePath); + } catch (TableNotExistException e) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e); + } + + if (table.spec().isUnpartitioned()) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + org.apache.iceberg.expressions.Expression filter; + try { + filter = toPartitionFilter(table, partitionSpec); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Invalid partition spec %s for table %s", partitionSpec, tablePath), e); + } + + boolean partitionExists; + try (CloseableIterable tasks = table.newScan().filter(filter).planFiles()) { + partitionExists = tasks.iterator().hasNext(); + } catch (IOException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } + + if (!partitionExists) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + try { + table.newDelete().deleteFromRowFilter(filter).commit(); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to drop partition %s from table %s", partitionSpec, tablePath), e); + } + } + + private static org.apache.iceberg.expressions.Expression toPartitionFilter( + Table table, CatalogPartitionSpec partitionSpec) { + Map values = partitionSpec.getPartitionSpec(); + List fields = table.spec().fields(); + + Preconditions.checkArgument( + values.size() == fields.size(), + "Partition spec %s does not match the partition fields of table %s", + partitionSpec, + table.name()); + + org.apache.iceberg.expressions.Expression filter = null; + for (PartitionField field : fields) { + Preconditions.checkArgument( + field.transform().isIdentity(), + "Dropping partitions with transform %s is not supported for field %s", + field.transform(), + field.name()); + + Preconditions.checkArgument( + values.containsKey(field.name()), + "Partition spec %s is missing partition field %s", + partitionSpec, + field.name()); + + Type sourceType = table.schema().findType(field.sourceId()); + Preconditions.checkArgument( + sourceType != null, + "Cannot find source field %s for partition field %s", + field.sourceId(), + field.name()); + + String value = values.get(field.name()); + Object parsedValue = fromPartitionString(sourceType, value); + org.apache.iceberg.expressions.Expression fieldFilter = + parsedValue == null + ? Expressions.isNull(table.schema().findColumnName(field.sourceId())) + : Expressions.equal(table.schema().findColumnName(field.sourceId()), parsedValue); + filter = filter == null ? fieldFilter : Expressions.and(filter, fieldFilter); + } + + return filter; + } + + private static Object fromPartitionString(Type type, String value) { + if (value == null || "__HIVE_DEFAULT_PARTITION__".equals(value)) { + return null; + } + + try { + switch (type.typeId()) { + case BOOLEAN: + Preconditions.checkArgument( + "true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value), + "Invalid boolean partition value: %s", + value); + return Boolean.valueOf(value); + case TIME: + case TIMESTAMP: + case TIMESTAMP_NANO: + Literal literal = Literal.of(value).to(type); + Preconditions.checkArgument( + literal != null, "Cannot convert partition value %s to type %s", value, type); + return literal.value(); + case BINARY: + case FIXED: + byte[] bytes = Base64.getDecoder().decode(value); + if (type instanceof Types.FixedType) { + Preconditions.checkArgument( + bytes.length == ((Types.FixedType) type).length(), + "Invalid fixed partition value length: %s", + bytes.length); + } + return ByteBuffer.wrap(bytes); + default: + return Conversions.fromPartitionString(type, value); + } + } catch (RuntimeException e) { + throw new IllegalArgumentException( + String.format("Cannot convert partition value %s to type %s", value, type), e); + } } @Override @@ -825,7 +965,13 @@ public List listPartitions(ObjectPath tablePath) StructLike structLike = dataFile.partition(); PartitionSpec spec = table.specs().get(dataFile.specId()); for (int i = 0; i < structLike.size(); i++) { - map.put(spec.fields().get(i).name(), String.valueOf(structLike.get(i, Object.class))); + PartitionField field = spec.fields().get(i); + Type sourceType = table.schema().findType(field.sourceId()); + Type partitionType = field.transform().getResultType(sourceType); + Object value = structLike.get(i, partitionType.typeId().javaClass()); + map.put( + field.name(), + value == null ? null : toHumanString(field.transform(), partitionType, value)); } set.add(new CatalogPartitionSpec(map)); } @@ -837,6 +983,11 @@ public List listPartitions(ObjectPath tablePath) return Lists.newArrayList(set); } + @SuppressWarnings({"rawtypes", "unchecked"}) + private static String toHumanString(Transform transform, Type type, Object value) { + return transform.toHumanString(type, value); + } + @Override public List listPartitions( ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException { diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java index 94e7348e4e4f..8a094d0d5d9c 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java @@ -19,20 +19,28 @@ package org.apache.iceberg.flink; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Collections; import java.util.List; import org.apache.flink.table.catalog.CatalogPartitionSpec; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Parameter; import org.apache.iceberg.Parameters; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; @@ -66,8 +74,8 @@ protected static List parameters() { @Override @BeforeEach public void before() { - super.before(); config.put(CatalogProperties.CACHE_ENABLED, String.valueOf(cacheEnabled)); + super.before(); sql("CREATE DATABASE %s", flinkDatabase); sql("USE CATALOG %s", catalogName); sql("USE %s", DATABASE); @@ -87,9 +95,9 @@ public void testListPartitionsWithUnpartitionedTable() { tableName, format.name()); sql("INSERT INTO %s SELECT 1,'a'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - assertThatThrownBy(() -> flinkCatalog.listPartitions(objectPath)) + assertThatThrownBy(() -> flinkCatalog.listPartitions(tablePath)) .isInstanceOf(TableNotPartitionedException.class) .hasMessageStartingWith("Table db.test_table in catalog") .hasMessageEndingWith("is not partitioned."); @@ -105,9 +113,9 @@ public void testListPartitionsWithPartitionedTable() sql("INSERT INTO %s SELECT 1,'a'", tableName); sql("INSERT INTO %s SELECT 2,'b'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - List list = flinkCatalog.listPartitions(objectPath); + List list = flinkCatalog.listPartitions(tablePath); assertThat(list).hasSize(2); List expected = Lists.newArrayList(); CatalogPartitionSpec partitionSpec1 = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); @@ -116,4 +124,209 @@ public void testListPartitionsWithPartitionedTable() expected.add(partitionSpec2); assertThat(list).as("Should produce the expected catalog partition specs.").isEqualTo(expected); } + + @TestTemplate + void dropPartitionThroughSql() throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 2,'b'", tableName); + + sql("ALTER TABLE %s DROP PARTITION (data = 'a')", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "b"))); + } + + @TestTemplate + void dropPartitionWithMultipleFields() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (id, data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 1,'b'", tableName); + sql("INSERT INTO %s SELECT 2,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "a")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactlyInAnyOrder( + new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "b")), + new CatalogPartitionSpec(ImmutableMap.of("id", "2", "data", "a"))); + } + + @TestTemplate + void dropPartitionWithDatePartition() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, dt DATE) PARTITIONED BY (dt) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, DATE '2024-01-01'", tableName); + sql("INSERT INTO %s SELECT 2, DATE '2024-01-02'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-01")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-02"))); + } + + @TestTemplate + void dropPartitionWithNullValue() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + sql("INSERT INTO %s SELECT 2, 'null'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(Collections.singletonMap("data", null)), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "null"))); + } + + @TestTemplate + void listPartitionsPreservesNullValue() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + + assertThat(flinkCatalog().listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(Collections.singletonMap("data", null))); + } + + @TestTemplate + void dropPartitionRejectsNonIdentityTransform() { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + validationCatalog.createTable( + TableIdentifier.of(icebergNamespace, tableName), + schema, + PartitionSpec.builderFor(schema).bucket("id", 2).build()); + + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + objectPath, + new CatalogPartitionSpec(ImmutableMap.of("id_bucket", "0")), + false)) + .isInstanceOf(CatalogException.class) + .hasMessageContaining("Invalid partition spec"); + } + + @TestTemplate + void dropPartitionIfNotExistsDoesNotCreateSnapshot() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + long snapshotId = currentSnapshotId(); + sql("ALTER TABLE %s DROP IF EXISTS PARTITION (data = 'missing')", tableName); + + assertThat(currentSnapshotId()).isEqualTo(snapshotId); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "a"))); + } + + @TestTemplate + void dropPartitionThrowsWhenPartitionDoesNotExist() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = + new CatalogPartitionSpec(ImmutableMap.of("data", "missing")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithUnpartitionedTable() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithIgnoreOnUnpartitionedTable() { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("data", "a")), true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionWithIgnoreOnMissingTable() { + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionThrowsForMissingTable() { + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + private FlinkCatalog flinkCatalog() { + return (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); + } + + private long currentSnapshotId() { + return validationCatalog + .loadTable(TableIdentifier.of(icebergNamespace, tableName)) + .currentSnapshot() + .snapshotId(); + } + + private final ObjectPath objectPath = new ObjectPath(DATABASE, tableName); } diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 1d1505a28c05..45500e9fae49 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -20,6 +20,8 @@ import java.io.Closeable; import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Base64; import java.util.Collections; import java.util.List; import java.util.Map; @@ -45,6 +47,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; @@ -70,6 +73,8 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.Literal; import org.apache.iceberg.flink.util.FlinkAlterTableUtil; import org.apache.iceberg.flink.util.FlinkCompatibilityUtil; import org.apache.iceberg.io.CloseableIterable; @@ -80,6 +85,10 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -725,8 +734,139 @@ public void createPartition( @Override public void dropPartition( ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) - throws CatalogException { - throw new UnsupportedOperationException(); + throws PartitionNotExistException, CatalogException { + Table table; + try { + table = loadIcebergTable(tablePath); + } catch (TableNotExistException e) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e); + } + + if (table.spec().isUnpartitioned()) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + org.apache.iceberg.expressions.Expression filter; + try { + filter = toPartitionFilter(table, partitionSpec); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Invalid partition spec %s for table %s", partitionSpec, tablePath), e); + } + + boolean partitionExists; + try (CloseableIterable tasks = table.newScan().filter(filter).planFiles()) { + partitionExists = tasks.iterator().hasNext(); + } catch (IOException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } + + if (!partitionExists) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + try { + table.newDelete().deleteFromRowFilter(filter).commit(); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to drop partition %s from table %s", partitionSpec, tablePath), e); + } + } + + private static org.apache.iceberg.expressions.Expression toPartitionFilter( + Table table, CatalogPartitionSpec partitionSpec) { + Map values = partitionSpec.getPartitionSpec(); + List fields = table.spec().fields(); + + Preconditions.checkArgument( + values.size() == fields.size(), + "Partition spec %s does not match the partition fields of table %s", + partitionSpec, + table.name()); + + org.apache.iceberg.expressions.Expression filter = null; + for (PartitionField field : fields) { + Preconditions.checkArgument( + field.transform().isIdentity(), + "Dropping partitions with transform %s is not supported for field %s", + field.transform(), + field.name()); + + Preconditions.checkArgument( + values.containsKey(field.name()), + "Partition spec %s is missing partition field %s", + partitionSpec, + field.name()); + + Type sourceType = table.schema().findType(field.sourceId()); + Preconditions.checkArgument( + sourceType != null, + "Cannot find source field %s for partition field %s", + field.sourceId(), + field.name()); + + String value = values.get(field.name()); + Object parsedValue = fromPartitionString(sourceType, value); + org.apache.iceberg.expressions.Expression fieldFilter = + parsedValue == null + ? Expressions.isNull(table.schema().findColumnName(field.sourceId())) + : Expressions.equal(table.schema().findColumnName(field.sourceId()), parsedValue); + filter = filter == null ? fieldFilter : Expressions.and(filter, fieldFilter); + } + + return filter; + } + + private static Object fromPartitionString(Type type, String value) { + if (value == null || "__HIVE_DEFAULT_PARTITION__".equals(value)) { + return null; + } + + try { + switch (type.typeId()) { + case BOOLEAN: + Preconditions.checkArgument( + "true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value), + "Invalid boolean partition value: %s", + value); + return Boolean.valueOf(value); + case TIME: + case TIMESTAMP: + case TIMESTAMP_NANO: + Literal literal = Literal.of(value).to(type); + Preconditions.checkArgument( + literal != null, "Cannot convert partition value %s to type %s", value, type); + return literal.value(); + case BINARY: + case FIXED: + byte[] bytes = Base64.getDecoder().decode(value); + if (type instanceof Types.FixedType) { + Preconditions.checkArgument( + bytes.length == ((Types.FixedType) type).length(), + "Invalid fixed partition value length: %s", + bytes.length); + } + return ByteBuffer.wrap(bytes); + default: + return Conversions.fromPartitionString(type, value); + } + } catch (RuntimeException e) { + throw new IllegalArgumentException( + String.format("Cannot convert partition value %s to type %s", value, type), e); + } } @Override @@ -825,7 +965,13 @@ public List listPartitions(ObjectPath tablePath) StructLike structLike = dataFile.partition(); PartitionSpec spec = table.specs().get(dataFile.specId()); for (int i = 0; i < structLike.size(); i++) { - map.put(spec.fields().get(i).name(), String.valueOf(structLike.get(i, Object.class))); + PartitionField field = spec.fields().get(i); + Type sourceType = table.schema().findType(field.sourceId()); + Type partitionType = field.transform().getResultType(sourceType); + Object value = structLike.get(i, partitionType.typeId().javaClass()); + map.put( + field.name(), + value == null ? null : toHumanString(field.transform(), partitionType, value)); } set.add(new CatalogPartitionSpec(map)); } @@ -837,6 +983,11 @@ public List listPartitions(ObjectPath tablePath) return Lists.newArrayList(set); } + @SuppressWarnings({"rawtypes", "unchecked"}) + private static String toHumanString(Transform transform, Type type, Object value) { + return transform.toHumanString(type, value); + } + @Override public List listPartitions( ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException { diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java index 94e7348e4e4f..8a094d0d5d9c 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java +++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java @@ -19,20 +19,28 @@ package org.apache.iceberg.flink; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Collections; import java.util.List; import org.apache.flink.table.catalog.CatalogPartitionSpec; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Parameter; import org.apache.iceberg.Parameters; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; @@ -66,8 +74,8 @@ protected static List parameters() { @Override @BeforeEach public void before() { - super.before(); config.put(CatalogProperties.CACHE_ENABLED, String.valueOf(cacheEnabled)); + super.before(); sql("CREATE DATABASE %s", flinkDatabase); sql("USE CATALOG %s", catalogName); sql("USE %s", DATABASE); @@ -87,9 +95,9 @@ public void testListPartitionsWithUnpartitionedTable() { tableName, format.name()); sql("INSERT INTO %s SELECT 1,'a'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - assertThatThrownBy(() -> flinkCatalog.listPartitions(objectPath)) + assertThatThrownBy(() -> flinkCatalog.listPartitions(tablePath)) .isInstanceOf(TableNotPartitionedException.class) .hasMessageStartingWith("Table db.test_table in catalog") .hasMessageEndingWith("is not partitioned."); @@ -105,9 +113,9 @@ public void testListPartitionsWithPartitionedTable() sql("INSERT INTO %s SELECT 1,'a'", tableName); sql("INSERT INTO %s SELECT 2,'b'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - List list = flinkCatalog.listPartitions(objectPath); + List list = flinkCatalog.listPartitions(tablePath); assertThat(list).hasSize(2); List expected = Lists.newArrayList(); CatalogPartitionSpec partitionSpec1 = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); @@ -116,4 +124,209 @@ public void testListPartitionsWithPartitionedTable() expected.add(partitionSpec2); assertThat(list).as("Should produce the expected catalog partition specs.").isEqualTo(expected); } + + @TestTemplate + void dropPartitionThroughSql() throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 2,'b'", tableName); + + sql("ALTER TABLE %s DROP PARTITION (data = 'a')", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "b"))); + } + + @TestTemplate + void dropPartitionWithMultipleFields() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (id, data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 1,'b'", tableName); + sql("INSERT INTO %s SELECT 2,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "a")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactlyInAnyOrder( + new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "b")), + new CatalogPartitionSpec(ImmutableMap.of("id", "2", "data", "a"))); + } + + @TestTemplate + void dropPartitionWithDatePartition() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, dt DATE) PARTITIONED BY (dt) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, DATE '2024-01-01'", tableName); + sql("INSERT INTO %s SELECT 2, DATE '2024-01-02'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-01")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-02"))); + } + + @TestTemplate + void dropPartitionWithNullValue() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + sql("INSERT INTO %s SELECT 2, 'null'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(Collections.singletonMap("data", null)), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "null"))); + } + + @TestTemplate + void listPartitionsPreservesNullValue() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + + assertThat(flinkCatalog().listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(Collections.singletonMap("data", null))); + } + + @TestTemplate + void dropPartitionRejectsNonIdentityTransform() { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + validationCatalog.createTable( + TableIdentifier.of(icebergNamespace, tableName), + schema, + PartitionSpec.builderFor(schema).bucket("id", 2).build()); + + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + objectPath, + new CatalogPartitionSpec(ImmutableMap.of("id_bucket", "0")), + false)) + .isInstanceOf(CatalogException.class) + .hasMessageContaining("Invalid partition spec"); + } + + @TestTemplate + void dropPartitionIfNotExistsDoesNotCreateSnapshot() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + long snapshotId = currentSnapshotId(); + sql("ALTER TABLE %s DROP IF EXISTS PARTITION (data = 'missing')", tableName); + + assertThat(currentSnapshotId()).isEqualTo(snapshotId); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "a"))); + } + + @TestTemplate + void dropPartitionThrowsWhenPartitionDoesNotExist() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = + new CatalogPartitionSpec(ImmutableMap.of("data", "missing")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithUnpartitionedTable() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithIgnoreOnUnpartitionedTable() { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("data", "a")), true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionWithIgnoreOnMissingTable() { + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionThrowsForMissingTable() { + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + private FlinkCatalog flinkCatalog() { + return (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); + } + + private long currentSnapshotId() { + return validationCatalog + .loadTable(TableIdentifier.of(icebergNamespace, tableName)) + .currentSnapshot() + .snapshotId(); + } + + private final ObjectPath objectPath = new ObjectPath(DATABASE, tableName); } diff --git a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 1d1505a28c05..45500e9fae49 100644 --- a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -20,6 +20,8 @@ import java.io.Closeable; import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Base64; import java.util.Collections; import java.util.List; import java.util.Map; @@ -45,6 +47,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; @@ -70,6 +73,8 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.Literal; import org.apache.iceberg.flink.util.FlinkAlterTableUtil; import org.apache.iceberg.flink.util.FlinkCompatibilityUtil; import org.apache.iceberg.io.CloseableIterable; @@ -80,6 +85,10 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -725,8 +734,139 @@ public void createPartition( @Override public void dropPartition( ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) - throws CatalogException { - throw new UnsupportedOperationException(); + throws PartitionNotExistException, CatalogException { + Table table; + try { + table = loadIcebergTable(tablePath); + } catch (TableNotExistException e) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e); + } + + if (table.spec().isUnpartitioned()) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + org.apache.iceberg.expressions.Expression filter; + try { + filter = toPartitionFilter(table, partitionSpec); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Invalid partition spec %s for table %s", partitionSpec, tablePath), e); + } + + boolean partitionExists; + try (CloseableIterable tasks = table.newScan().filter(filter).planFiles()) { + partitionExists = tasks.iterator().hasNext(); + } catch (IOException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } + + if (!partitionExists) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + try { + table.newDelete().deleteFromRowFilter(filter).commit(); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to drop partition %s from table %s", partitionSpec, tablePath), e); + } + } + + private static org.apache.iceberg.expressions.Expression toPartitionFilter( + Table table, CatalogPartitionSpec partitionSpec) { + Map values = partitionSpec.getPartitionSpec(); + List fields = table.spec().fields(); + + Preconditions.checkArgument( + values.size() == fields.size(), + "Partition spec %s does not match the partition fields of table %s", + partitionSpec, + table.name()); + + org.apache.iceberg.expressions.Expression filter = null; + for (PartitionField field : fields) { + Preconditions.checkArgument( + field.transform().isIdentity(), + "Dropping partitions with transform %s is not supported for field %s", + field.transform(), + field.name()); + + Preconditions.checkArgument( + values.containsKey(field.name()), + "Partition spec %s is missing partition field %s", + partitionSpec, + field.name()); + + Type sourceType = table.schema().findType(field.sourceId()); + Preconditions.checkArgument( + sourceType != null, + "Cannot find source field %s for partition field %s", + field.sourceId(), + field.name()); + + String value = values.get(field.name()); + Object parsedValue = fromPartitionString(sourceType, value); + org.apache.iceberg.expressions.Expression fieldFilter = + parsedValue == null + ? Expressions.isNull(table.schema().findColumnName(field.sourceId())) + : Expressions.equal(table.schema().findColumnName(field.sourceId()), parsedValue); + filter = filter == null ? fieldFilter : Expressions.and(filter, fieldFilter); + } + + return filter; + } + + private static Object fromPartitionString(Type type, String value) { + if (value == null || "__HIVE_DEFAULT_PARTITION__".equals(value)) { + return null; + } + + try { + switch (type.typeId()) { + case BOOLEAN: + Preconditions.checkArgument( + "true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value), + "Invalid boolean partition value: %s", + value); + return Boolean.valueOf(value); + case TIME: + case TIMESTAMP: + case TIMESTAMP_NANO: + Literal literal = Literal.of(value).to(type); + Preconditions.checkArgument( + literal != null, "Cannot convert partition value %s to type %s", value, type); + return literal.value(); + case BINARY: + case FIXED: + byte[] bytes = Base64.getDecoder().decode(value); + if (type instanceof Types.FixedType) { + Preconditions.checkArgument( + bytes.length == ((Types.FixedType) type).length(), + "Invalid fixed partition value length: %s", + bytes.length); + } + return ByteBuffer.wrap(bytes); + default: + return Conversions.fromPartitionString(type, value); + } + } catch (RuntimeException e) { + throw new IllegalArgumentException( + String.format("Cannot convert partition value %s to type %s", value, type), e); + } } @Override @@ -825,7 +965,13 @@ public List listPartitions(ObjectPath tablePath) StructLike structLike = dataFile.partition(); PartitionSpec spec = table.specs().get(dataFile.specId()); for (int i = 0; i < structLike.size(); i++) { - map.put(spec.fields().get(i).name(), String.valueOf(structLike.get(i, Object.class))); + PartitionField field = spec.fields().get(i); + Type sourceType = table.schema().findType(field.sourceId()); + Type partitionType = field.transform().getResultType(sourceType); + Object value = structLike.get(i, partitionType.typeId().javaClass()); + map.put( + field.name(), + value == null ? null : toHumanString(field.transform(), partitionType, value)); } set.add(new CatalogPartitionSpec(map)); } @@ -837,6 +983,11 @@ public List listPartitions(ObjectPath tablePath) return Lists.newArrayList(set); } + @SuppressWarnings({"rawtypes", "unchecked"}) + private static String toHumanString(Transform transform, Type type, Object value) { + return transform.toHumanString(type, value); + } + @Override public List listPartitions( ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException { diff --git a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java index 94e7348e4e4f..8a094d0d5d9c 100644 --- a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java +++ b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java @@ -19,20 +19,28 @@ package org.apache.iceberg.flink; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Collections; import java.util.List; import org.apache.flink.table.catalog.CatalogPartitionSpec; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Parameter; import org.apache.iceberg.Parameters; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; @@ -66,8 +74,8 @@ protected static List parameters() { @Override @BeforeEach public void before() { - super.before(); config.put(CatalogProperties.CACHE_ENABLED, String.valueOf(cacheEnabled)); + super.before(); sql("CREATE DATABASE %s", flinkDatabase); sql("USE CATALOG %s", catalogName); sql("USE %s", DATABASE); @@ -87,9 +95,9 @@ public void testListPartitionsWithUnpartitionedTable() { tableName, format.name()); sql("INSERT INTO %s SELECT 1,'a'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - assertThatThrownBy(() -> flinkCatalog.listPartitions(objectPath)) + assertThatThrownBy(() -> flinkCatalog.listPartitions(tablePath)) .isInstanceOf(TableNotPartitionedException.class) .hasMessageStartingWith("Table db.test_table in catalog") .hasMessageEndingWith("is not partitioned."); @@ -105,9 +113,9 @@ public void testListPartitionsWithPartitionedTable() sql("INSERT INTO %s SELECT 1,'a'", tableName); sql("INSERT INTO %s SELECT 2,'b'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - List list = flinkCatalog.listPartitions(objectPath); + List list = flinkCatalog.listPartitions(tablePath); assertThat(list).hasSize(2); List expected = Lists.newArrayList(); CatalogPartitionSpec partitionSpec1 = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); @@ -116,4 +124,209 @@ public void testListPartitionsWithPartitionedTable() expected.add(partitionSpec2); assertThat(list).as("Should produce the expected catalog partition specs.").isEqualTo(expected); } + + @TestTemplate + void dropPartitionThroughSql() throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 2,'b'", tableName); + + sql("ALTER TABLE %s DROP PARTITION (data = 'a')", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "b"))); + } + + @TestTemplate + void dropPartitionWithMultipleFields() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (id, data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 1,'b'", tableName); + sql("INSERT INTO %s SELECT 2,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "a")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactlyInAnyOrder( + new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "b")), + new CatalogPartitionSpec(ImmutableMap.of("id", "2", "data", "a"))); + } + + @TestTemplate + void dropPartitionWithDatePartition() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, dt DATE) PARTITIONED BY (dt) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, DATE '2024-01-01'", tableName); + sql("INSERT INTO %s SELECT 2, DATE '2024-01-02'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-01")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-02"))); + } + + @TestTemplate + void dropPartitionWithNullValue() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + sql("INSERT INTO %s SELECT 2, 'null'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(Collections.singletonMap("data", null)), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "null"))); + } + + @TestTemplate + void listPartitionsPreservesNullValue() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + + assertThat(flinkCatalog().listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(Collections.singletonMap("data", null))); + } + + @TestTemplate + void dropPartitionRejectsNonIdentityTransform() { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + validationCatalog.createTable( + TableIdentifier.of(icebergNamespace, tableName), + schema, + PartitionSpec.builderFor(schema).bucket("id", 2).build()); + + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + objectPath, + new CatalogPartitionSpec(ImmutableMap.of("id_bucket", "0")), + false)) + .isInstanceOf(CatalogException.class) + .hasMessageContaining("Invalid partition spec"); + } + + @TestTemplate + void dropPartitionIfNotExistsDoesNotCreateSnapshot() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + long snapshotId = currentSnapshotId(); + sql("ALTER TABLE %s DROP IF EXISTS PARTITION (data = 'missing')", tableName); + + assertThat(currentSnapshotId()).isEqualTo(snapshotId); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "a"))); + } + + @TestTemplate + void dropPartitionThrowsWhenPartitionDoesNotExist() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = + new CatalogPartitionSpec(ImmutableMap.of("data", "missing")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithUnpartitionedTable() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithIgnoreOnUnpartitionedTable() { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("data", "a")), true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionWithIgnoreOnMissingTable() { + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionThrowsForMissingTable() { + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + private FlinkCatalog flinkCatalog() { + return (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); + } + + private long currentSnapshotId() { + return validationCatalog + .loadTable(TableIdentifier.of(icebergNamespace, tableName)) + .currentSnapshot() + .snapshotId(); + } + + private final ObjectPath objectPath = new ObjectPath(DATABASE, tableName); } diff --git a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java index 1d1505a28c05..45500e9fae49 100644 --- a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java +++ b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java @@ -20,6 +20,8 @@ import java.io.Closeable; import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Base64; import java.util.Collections; import java.util.List; import java.util.Map; @@ -45,6 +47,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; @@ -70,6 +73,8 @@ import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.Literal; import org.apache.iceberg.flink.util.FlinkAlterTableUtil; import org.apache.iceberg.flink.util.FlinkCompatibilityUtil; import org.apache.iceberg.io.CloseableIterable; @@ -80,6 +85,10 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.transforms.Transform; +import org.apache.iceberg.types.Conversions; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; /** * A Flink Catalog implementation that wraps an Iceberg {@link Catalog}. @@ -725,8 +734,139 @@ public void createPartition( @Override public void dropPartition( ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) - throws CatalogException { - throw new UnsupportedOperationException(); + throws PartitionNotExistException, CatalogException { + Table table; + try { + table = loadIcebergTable(tablePath); + } catch (TableNotExistException e) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec, e); + } + + if (table.spec().isUnpartitioned()) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + org.apache.iceberg.expressions.Expression filter; + try { + filter = toPartitionFilter(table, partitionSpec); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Invalid partition spec %s for table %s", partitionSpec, tablePath), e); + } + + boolean partitionExists; + try (CloseableIterable tasks = table.newScan().filter(filter).planFiles()) { + partitionExists = tasks.iterator().hasNext(); + } catch (IOException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to check partition %s of table %s", partitionSpec, tablePath), e); + } + + if (!partitionExists) { + if (ignoreIfNotExists) { + return; + } + throw new PartitionNotExistException(getName(), tablePath, partitionSpec); + } + + try { + table.newDelete().deleteFromRowFilter(filter).commit(); + } catch (RuntimeException e) { + throw new CatalogException( + String.format("Failed to drop partition %s from table %s", partitionSpec, tablePath), e); + } + } + + private static org.apache.iceberg.expressions.Expression toPartitionFilter( + Table table, CatalogPartitionSpec partitionSpec) { + Map values = partitionSpec.getPartitionSpec(); + List fields = table.spec().fields(); + + Preconditions.checkArgument( + values.size() == fields.size(), + "Partition spec %s does not match the partition fields of table %s", + partitionSpec, + table.name()); + + org.apache.iceberg.expressions.Expression filter = null; + for (PartitionField field : fields) { + Preconditions.checkArgument( + field.transform().isIdentity(), + "Dropping partitions with transform %s is not supported for field %s", + field.transform(), + field.name()); + + Preconditions.checkArgument( + values.containsKey(field.name()), + "Partition spec %s is missing partition field %s", + partitionSpec, + field.name()); + + Type sourceType = table.schema().findType(field.sourceId()); + Preconditions.checkArgument( + sourceType != null, + "Cannot find source field %s for partition field %s", + field.sourceId(), + field.name()); + + String value = values.get(field.name()); + Object parsedValue = fromPartitionString(sourceType, value); + org.apache.iceberg.expressions.Expression fieldFilter = + parsedValue == null + ? Expressions.isNull(table.schema().findColumnName(field.sourceId())) + : Expressions.equal(table.schema().findColumnName(field.sourceId()), parsedValue); + filter = filter == null ? fieldFilter : Expressions.and(filter, fieldFilter); + } + + return filter; + } + + private static Object fromPartitionString(Type type, String value) { + if (value == null || "__HIVE_DEFAULT_PARTITION__".equals(value)) { + return null; + } + + try { + switch (type.typeId()) { + case BOOLEAN: + Preconditions.checkArgument( + "true".equalsIgnoreCase(value) || "false".equalsIgnoreCase(value), + "Invalid boolean partition value: %s", + value); + return Boolean.valueOf(value); + case TIME: + case TIMESTAMP: + case TIMESTAMP_NANO: + Literal literal = Literal.of(value).to(type); + Preconditions.checkArgument( + literal != null, "Cannot convert partition value %s to type %s", value, type); + return literal.value(); + case BINARY: + case FIXED: + byte[] bytes = Base64.getDecoder().decode(value); + if (type instanceof Types.FixedType) { + Preconditions.checkArgument( + bytes.length == ((Types.FixedType) type).length(), + "Invalid fixed partition value length: %s", + bytes.length); + } + return ByteBuffer.wrap(bytes); + default: + return Conversions.fromPartitionString(type, value); + } + } catch (RuntimeException e) { + throw new IllegalArgumentException( + String.format("Cannot convert partition value %s to type %s", value, type), e); + } } @Override @@ -825,7 +965,13 @@ public List listPartitions(ObjectPath tablePath) StructLike structLike = dataFile.partition(); PartitionSpec spec = table.specs().get(dataFile.specId()); for (int i = 0; i < structLike.size(); i++) { - map.put(spec.fields().get(i).name(), String.valueOf(structLike.get(i, Object.class))); + PartitionField field = spec.fields().get(i); + Type sourceType = table.schema().findType(field.sourceId()); + Type partitionType = field.transform().getResultType(sourceType); + Object value = structLike.get(i, partitionType.typeId().javaClass()); + map.put( + field.name(), + value == null ? null : toHumanString(field.transform(), partitionType, value)); } set.add(new CatalogPartitionSpec(map)); } @@ -837,6 +983,11 @@ public List listPartitions(ObjectPath tablePath) return Lists.newArrayList(set); } + @SuppressWarnings({"rawtypes", "unchecked"}) + private static String toHumanString(Transform transform, Type type, Object value) { + return transform.toHumanString(type, value); + } + @Override public List listPartitions( ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException { diff --git a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java index 94e7348e4e4f..8a094d0d5d9c 100644 --- a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java +++ b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTablePartitions.java @@ -19,20 +19,28 @@ package org.apache.iceberg.flink; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Collections; import java.util.List; import org.apache.flink.table.catalog.CatalogPartitionSpec; import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotExistException; import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Parameter; import org.apache.iceberg.Parameters; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; @@ -66,8 +74,8 @@ protected static List parameters() { @Override @BeforeEach public void before() { - super.before(); config.put(CatalogProperties.CACHE_ENABLED, String.valueOf(cacheEnabled)); + super.before(); sql("CREATE DATABASE %s", flinkDatabase); sql("USE CATALOG %s", catalogName); sql("USE %s", DATABASE); @@ -87,9 +95,9 @@ public void testListPartitionsWithUnpartitionedTable() { tableName, format.name()); sql("INSERT INTO %s SELECT 1,'a'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - assertThatThrownBy(() -> flinkCatalog.listPartitions(objectPath)) + assertThatThrownBy(() -> flinkCatalog.listPartitions(tablePath)) .isInstanceOf(TableNotPartitionedException.class) .hasMessageStartingWith("Table db.test_table in catalog") .hasMessageEndingWith("is not partitioned."); @@ -105,9 +113,9 @@ public void testListPartitionsWithPartitionedTable() sql("INSERT INTO %s SELECT 1,'a'", tableName); sql("INSERT INTO %s SELECT 2,'b'", tableName); - ObjectPath objectPath = new ObjectPath(DATABASE, tableName); + ObjectPath tablePath = new ObjectPath(DATABASE, tableName); FlinkCatalog flinkCatalog = (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); - List list = flinkCatalog.listPartitions(objectPath); + List list = flinkCatalog.listPartitions(tablePath); assertThat(list).hasSize(2); List expected = Lists.newArrayList(); CatalogPartitionSpec partitionSpec1 = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); @@ -116,4 +124,209 @@ public void testListPartitionsWithPartitionedTable() expected.add(partitionSpec2); assertThat(list).as("Should produce the expected catalog partition specs.").isEqualTo(expected); } + + @TestTemplate + void dropPartitionThroughSql() throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 2,'b'", tableName); + + sql("ALTER TABLE %s DROP PARTITION (data = 'a')", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "b"))); + } + + @TestTemplate + void dropPartitionWithMultipleFields() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (id, data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + sql("INSERT INTO %s SELECT 1,'b'", tableName); + sql("INSERT INTO %s SELECT 2,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "a")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactlyInAnyOrder( + new CatalogPartitionSpec(ImmutableMap.of("id", "1", "data", "b")), + new CatalogPartitionSpec(ImmutableMap.of("id", "2", "data", "a"))); + } + + @TestTemplate + void dropPartitionWithDatePartition() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, dt DATE) PARTITIONED BY (dt) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, DATE '2024-01-01'", tableName); + sql("INSERT INTO %s SELECT 2, DATE '2024-01-02'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-01")), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("dt", "2024-01-02"))); + } + + @TestTemplate + void dropPartitionWithNullValue() + throws TableNotExistException, TableNotPartitionedException, PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + sql("INSERT INTO %s SELECT 2, 'null'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + flinkCatalog.dropPartition( + objectPath, new CatalogPartitionSpec(Collections.singletonMap("data", null)), false); + + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "null"))); + } + + @TestTemplate + void listPartitionsPreservesNullValue() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1, CAST(NULL AS VARCHAR)", tableName); + + assertThat(flinkCatalog().listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(Collections.singletonMap("data", null))); + } + + @TestTemplate + void dropPartitionRejectsNonIdentityTransform() { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + validationCatalog.createTable( + TableIdentifier.of(icebergNamespace, tableName), + schema, + PartitionSpec.builderFor(schema).bucket("id", 2).build()); + + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + objectPath, + new CatalogPartitionSpec(ImmutableMap.of("id_bucket", "0")), + false)) + .isInstanceOf(CatalogException.class) + .hasMessageContaining("Invalid partition spec"); + } + + @TestTemplate + void dropPartitionIfNotExistsDoesNotCreateSnapshot() + throws TableNotExistException, TableNotPartitionedException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + FlinkCatalog flinkCatalog = flinkCatalog(); + long snapshotId = currentSnapshotId(); + sql("ALTER TABLE %s DROP IF EXISTS PARTITION (data = 'missing')", tableName); + + assertThat(currentSnapshotId()).isEqualTo(snapshotId); + assertThat(flinkCatalog.listPartitions(objectPath)) + .containsExactly(new CatalogPartitionSpec(ImmutableMap.of("data", "a"))); + } + + @TestTemplate + void dropPartitionThrowsWhenPartitionDoesNotExist() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) PARTITIONED BY (data) " + + "with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = + new CatalogPartitionSpec(ImmutableMap.of("data", "missing")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithUnpartitionedTable() throws PartitionNotExistException { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + sql("INSERT INTO %s SELECT 1,'a'", tableName); + + CatalogPartitionSpec partitionSpec = new CatalogPartitionSpec(ImmutableMap.of("data", "a")); + + assertThatThrownBy(() -> flinkCatalog().dropPartition(objectPath, partitionSpec, false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + @TestTemplate + void dropPartitionWithIgnoreOnUnpartitionedTable() { + sql( + "CREATE TABLE %s (id INT, data VARCHAR) with ('write.format.default'='%s')", + tableName, format.name()); + + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + objectPath, new CatalogPartitionSpec(ImmutableMap.of("data", "a")), true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionWithIgnoreOnMissingTable() { + assertThatCode( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + true)) + .doesNotThrowAnyException(); + } + + @TestTemplate + void dropPartitionThrowsForMissingTable() { + assertThatThrownBy( + () -> + flinkCatalog() + .dropPartition( + new ObjectPath(DATABASE, "missing_table"), + new CatalogPartitionSpec(ImmutableMap.of("data", "a")), + false)) + .isInstanceOf(PartitionNotExistException.class) + .hasMessageContaining("does not exist"); + } + + private FlinkCatalog flinkCatalog() { + return (FlinkCatalog) getTableEnv().getCatalog(catalogName).get(); + } + + private long currentSnapshotId() { + return validationCatalog + .loadTable(TableIdentifier.of(icebergNamespace, tableName)) + .currentSnapshot() + .snapshotId(); + } + + private final ObjectPath objectPath = new ObjectPath(DATABASE, tableName); }