Skip to content

Flink: Support DROP PARTITION in catalog #18040

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kaiwangleo wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Flink: Support DROP PARTITION in catalog #18040

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9faf76f
Flink: Support DROP PARTITION in catalog
Sep 10, 2026
a8cbe64
Flink: Fix DROP PARTITION checkstyle violations
Sep 10, 2026
60f9b36
Flink: Format DROP PARTITION catalog changes
Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
157 changes: 154 additions & 3 deletions flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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}.
Expand Down Expand Up @@ -119,8 +128,8 @@
? CachingCatalog.wrap(originalCatalog, cacheExpirationIntervalMs)
: originalCatalog;
asNamespaceCatalog =
originalCatalog instanceof SupportsNamespaces ? (SupportsNamespaces) originalCatalog : null;

Check warning on line 131 in flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

View workflow job for this annotation

GitHub Actions / build-checks (17, pull_request) 

[PatternMatchingInstanceof] This code can be simplified to use a pattern-matching instanceof.
closeable = originalCatalog instanceof Closeable ? (Closeable) originalCatalog : null;

Check warning on line 132 in flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

View workflow job for this annotation

GitHub Actions / build-checks (17, pull_request) 

[PatternMatchingInstanceof] This code can be simplified to use a pattern-matching instanceof.

FlinkEnvironmentContext.init();
}
Expand Down Expand Up @@ -592,7 +601,7 @@
List<TableChange> propertyChanges = Lists.newArrayList();
List<TableChange> schemaChanges = Lists.newArrayList();
for (TableChange change : tableChanges) {
if (change instanceof TableChange.SetOption) {

Check warning on line 604 in flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

View workflow job for this annotation

GitHub Actions / build-checks (17, pull_request) 

[PatternMatchingInstanceof] This code can be simplified to use a pattern-matching instanceof.
TableChange.SetOption set = (TableChange.SetOption) change;

if (FlinkCreateTableOptions.LOCATION_KEY.equalsIgnoreCase(set.getKey())) {
Expand Down Expand Up @@ -725,8 +734,139 @@
@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<FileScanTask> 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<String, String> values = partitionSpec.getPartitionSpec();
List<PartitionField> 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()) {

Check warning on line 839 in flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

View workflow job for this annotation

GitHub Actions / build-checks (17, pull_request) 

[StatementSwitchToExpressionSwitch] This statement switch can be converted to a new-style arrow switch
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) {

Check warning on line 856 in flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

View workflow job for this annotation

GitHub Actions / build-checks (17, pull_request) 

[PatternMatchingInstanceof] This code can be simplified to use a pattern-matching instanceof.
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
Expand Down Expand Up @@ -825,7 +965,13 @@
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));
}
Expand All @@ -837,6 +983,11 @@
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<CatalogPartitionSpec> listPartitions(
ObjectPath tablePath, CatalogPartitionSpec partitionSpec) throws CatalogException {
Expand Down
Loading
Loading