This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 809be25122 [flink] Add data evolution delete action (#8716)
809be25122 is described below
commit 809be251226efa0ae0c36cace451035695aa551a
Author: wangwj <[email protected]>
AuthorDate: Thu Jul 23 21:07:38 2026 +0800
[flink] Add data evolution delete action (#8716)
---
docs/docs/flink/action-jars.md | 118 +++++
docs/docs/multimodal-table/data-evolution.mdx | 7 +-
.../append/BaseAppendDeleteFileMaintainer.java | 18 +
.../paimon/flink/action/DataEvolutionDelete.java | 441 ++++++++++++++++
.../apache/paimon/flink/action/DeleteAction.java | 82 ++-
.../paimon/flink/action/DeleteActionFactory.java | 35 +-
.../dataevolution/DataEvolutionDeleteOperator.java | 568 +++++++++++++++++++++
.../action/DeleteActionDataEvolutionITCase.java | 494 ++++++++++++++++++
.../paimon/flink/action/DeleteActionITCase.java | 28 +
.../paimon/flink/procedure/ProcedureTest.java | 3 +-
10 files changed, 1784 insertions(+), 10 deletions(-)
diff --git a/docs/docs/flink/action-jars.md b/docs/docs/flink/action-jars.md
index 419c342008..b44546518e 100644
--- a/docs/docs/flink/action-jars.md
+++ b/docs/docs/flink/action-jars.md
@@ -259,6 +259,124 @@ For more information of 'delete', see
delete --help
```
+## Deleting from a Data Evolution table
+
+For a non-primary-key append table in
+[Data Evolution](../multimodal-table/data-evolution) mode, Paimon supports
+logically deleting matching rows through the same `delete` action used by
+primary-key tables. The action dispatches to the appropriate implementation
+based on the target table type. Regular append-only tables are not supported.
+
+The action evaluates the filter against a fixed snapshot of the
+`$row_tracking` system table and records matched rows in deletion vectors. It
+does not rewrite existing data files or dedicated BLOB files.
+
+:::info
+
+The target table must:
+
+1. have no primary key;
+2. use bucket-unaware mode (`bucket = -1`);
+3. enable `row-tracking.enabled`;
+4. enable `data-evolution.enabled`;
+5. enable `deletion-vectors.enabled`.
+
+:::
+
+Run the following command to submit a `delete` job:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-@@VERSION@@.jar \
+ delete \
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ [--source_sql <sql> [--source_sql <sql> ...]] \
+ --where "<filter_spec>" \
+ [--sink_parallelism <sink-parallelism>] \
+ [--catalog_conf <paimon-catalog-conf> [--catalog_conf
<paimon-catalog-conf> ...]]
+```
+
+`filter_spec` uses Flink SQL expression syntax and is equivalent to the
+predicate in a SQL `WHERE` clause. For example:
+
+```text
+last_access_time < TIMESTAMP '2026-07-01 00:00:00'
+status = 'expired'
+id >= 100 AND id < 200
+```
+
+`source_sql` is repeatable. Each statement is executed in order before the
+target query, so any bounded table supported by Flink SQL and optional views
+can be registered and referenced from a subquery in `filter_spec`. This is not
+limited to a particular external system: JDBC databases, data warehouses, and
+other bounded connectors can all be used as long as the corresponding
+connector is available in the action job's classpath.
+
+The subquery is also where source-side filtering belongs. In the example below,
+the action reads an access-state table, selects only cold URLs, and deletes the
+matching rows from the target Paimon table:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-@@VERSION@@.jar \
+ delete \
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ --source_sql "CREATE TEMPORARY TABLE access_state (
+ url STRING,
+ last_ingest_time TIMESTAMP(3),
+ last_request_time TIMESTAMP(3)
+ ) WITH (
+ 'connector' = 'jdbc',
+ 'url' = '<jdbc-url>',
+ 'table-name' = '<access-state-table>'
+ )" \
+ --where "url IN (
+ SELECT url
+ FROM access_state
+ WHERE last_ingest_time < TIMESTAMP '2026-07-01 00:00:00'
+ AND (last_request_time IS NULL
+ OR last_request_time < TIMESTAMP '2026-07-01 00:00:00')
+ )" \
+ --sink_parallelism 8
+```
+
+Using a subquery avoids copying a large candidate set into a temporary Paimon
+table. The external source must be bounded so the action can finish and commit
+one delete snapshot.
+
+:::warning
+
+- This action performs a logical delete. Physical data and BLOB file
+ reclamation still depends on subsequent Data Evolution compaction with
+ `data-evolution.compaction.rewrite-row-ids=true` and snapshot expiration.
+- Do not run multiple delete actions, or concurrent `APPEND`, `COMPACT`, or
+ `OVERWRITE` operations, against the same table. A conflicting
+ commit causes the action to fail instead of silently overwriting deletion
+ vectors.
+- Row positions are aggregated in parallel per anchor file. Deletion vectors
+ are then written in parallel across independent rewrite groups. One existing
+ deletion-vector index file is an atomic rewrite group and always has a single
+ writer owner, because it can contain deletion vectors for several anchor
+ files.
+- Split large deletes into bounded batches to limit row-tracking planning,
+ deletion-vector memory usage, and external-source scan size.
+- Snapshot retention must preserve the action's fixed base snapshot until the
+ job finishes.
+
+:::
+
+For more information, run:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ /path/to/paimon-flink-action-@@VERSION@@.jar \
+ delete --help
+```
+
## Drop Partition
Run the following command to submit a 'drop_partition' job for the table.
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index e11e9f079b..ba5220a1f6 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -273,8 +273,11 @@ WHEN MATCHED AND s.op = 'update' THEN UPDATE SET t.b = t.b
+ 10
WHEN NOT MATCHED BY SOURCE AND t.id > 10 THEN DELETE;
```
-The `WHEN NOT MATCHED BY SOURCE` clause requires Spark 3.4 or later. Currently,
-only Spark supports writing deletes for Data Evolution tables.
+The `WHEN NOT MATCHED BY SOURCE` clause requires Spark 3.4 or later.
+
+Flink SQL does not currently support `DELETE FROM` for Data Evolution tables,
+but Flink users can submit the
+[`delete` action](../flink/action-jars#deleting-from-a-data-evolution-table).
## Self Updates
diff --git
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
index b61985a177..c23556e2ee 100644
---
a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/BaseAppendDeleteFileMaintainer.java
@@ -38,6 +38,7 @@ import java.util.stream.Collectors;
import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/**
@@ -80,8 +81,25 @@ public interface BaseAppendDeleteFileMaintainer {
indexFileHandler.scan(snapshot,
DELETION_VECTORS_INDEX).stream()
.filter(e -> e.partition().equals(partition))
.collect(Collectors.toList());
+ return forUnawareAppend(indexFileHandler, partition, manifestEntries);
+ }
+
+ /**
+ * Creates an unaware-bucket maintainer which owns only the provided index
files.
+ *
+ * <p>This overload is used by parallel rewrite-group writers. Each old
deletion-vector index
+ * file must be owned by exactly one writer; a writer for new anchors
receives an empty list.
+ */
+ static AppendDeleteFileMaintainer forUnawareAppend(
+ IndexFileHandler indexFileHandler,
+ BinaryRow partition,
+ List<IndexManifestEntry> manifestEntries) {
Map<String, DeletionFile> deletionFiles = new HashMap<>();
for (IndexManifestEntry file : manifestEntries) {
+ checkArgument(
+ file.partition().equals(partition),
+ "Index file %s belongs to a different partition.",
+ file.indexFile().fileName());
LinkedHashMap<String, DeletionVectorMeta> dvMetas =
file.indexFile().dvRanges();
checkNotNull(dvMetas);
for (DeletionVectorMeta dvMeta : dvMetas.values()) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java
new file mode 100644
index 0000000000..03ea04d144
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java
@@ -0,0 +1,441 @@
+/*
+ * 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.paimon.flink.action;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.flink.dataevolution.DataEvolutionDeleteOperator;
+import
org.apache.paimon.flink.dataevolution.DataEvolutionDeleteOperator.DeletionTarget;
+import
org.apache.paimon.flink.dataevolution.DataEvolutionDeleteOperator.DeletionVectorAggregator;
+import
org.apache.paimon.flink.dataevolution.DataEvolutionDeleteOperator.DeletionVectorUpdate;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.sink.CommittableTypeInfo;
+import org.apache.paimon.flink.sink.CommitterOperatorFactory;
+import org.apache.paimon.flink.sink.NoopCommittableStateManager;
+import org.apache.paimon.flink.sink.StoreCommitter;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestCommittable;
+import org.apache.paimon.operation.DataEvolutionSplitRead;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DeletionFile;
+import org.apache.paimon.utils.DataEvolutionUtils;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.SerializationUtils;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.Partitioner;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.types.Row;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Internal implementation which logically deletes rows from a data-evolution
append table.
+ *
+ * <p>The action evaluates a Flink SQL filter against a fixed row-tracking
snapshot, maps every
+ * matched {@code _ROW_ID} to its data-evolution anchor file, and commits
deletion-vector index
+ * files. Optional source SQL statements can register bounded external tables
used by subqueries in
+ * the filter. Existing data and BLOB files are not rewritten by this action.
+ *
+ * <p>Only one instance of this action should run against the same table at a
time. The fixed base
+ * snapshot and strict commit mode detect conflicting commits, including
append, compaction, and
+ * overwrite, instead of silently applying deletion vectors to a stale row-id
mapping.
+ *
+ * <p>The current implementation plans anchor ranges on the coordinator. Row
positions are first
+ * aggregated per anchor and then shuffled by rewrite group. Anchors backed by
the same existing
+ * deletion-vector index file always have one writer owner; anchors without
existing deletion
+ * vectors are split into stable shards. Large deletes should still be split
into bounded batches to
+ * limit coordinator and deletion-vector memory usage.
+ */
+class DataEvolutionDelete implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DataEvolutionDelete.class);
+
+ private final DeleteAction action;
+ private final String filter;
+ private final long baseSnapshotId;
+
+ private int sinkParallelism = 1;
+
+ DataEvolutionDelete(DeleteAction action, String filter) {
+ this.action = action;
+ Preconditions.checkArgument(
+ filter != null && !filter.trim().isEmpty(),
+ "Deletion filter must not be null or blank.");
+ this.filter = filter;
+
+ if (!(action.table instanceof FileStoreTable)) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Only FileStoreTable supports Data Evolution
delete. The table type is '%s'.",
+ action.table.getClass().getName()));
+ }
+
+ FileStoreTable storeTable = (FileStoreTable) action.table;
+ Long latestSnapshotId =
storeTable.snapshotManager().latestSnapshotId();
+ if (latestSnapshotId == null) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete action doesn't support deleting
from an empty table.");
+ }
+ this.baseSnapshotId = latestSnapshotId;
+
+ CoreOptions coreOptions = storeTable.coreOptions();
+ if (!storeTable.schema().primaryKeys().isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete action only supports append tables
without primary keys.");
+ }
+ if (!coreOptions.rowTrackingEnabled()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete action requires
row-tracking.enabled to be true.");
+ }
+ if (!coreOptions.dataEvolutionEnabled()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete action requires
data-evolution.enabled to be true.");
+ }
+ if (!coreOptions.deletionVectorsEnabled()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete action requires
deletion-vectors.enabled to be true.");
+ }
+ if (storeTable.bucketMode() != BucketMode.BUCKET_UNAWARE) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Data-evolution delete action only supports
unaware bucket mode, but table bucket mode is %s.",
+ storeTable.bucketMode()));
+ }
+
+ action.table =
+ action.table.copy(
+ Collections.singletonMap(
+
CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key(),
+ latestSnapshotId.toString()));
+ }
+
+ DataEvolutionDelete withSinkParallelism(int sinkParallelism) {
+ Preconditions.checkArgument(
+ sinkParallelism > 0,
+ "Sink parallelism must be a positive integer, but is %s.",
+ sinkParallelism);
+ this.sinkParallelism = sinkParallelism;
+ return this;
+ }
+
+ /** Builds and executes the Flink batch topology. */
+ TableResult runInternal() {
+ FileStoreTable storeTable = (FileStoreTable) action.table;
+ List<AnchorRange> anchorRanges = planAnchorRanges(storeTable);
+ String commitUser =
+
CoreOptions.createCommitUser(storeTable.coreOptions().toConfiguration());
+
+ String query =
+ String.format(
+ "SELECT `_ROW_ID` FROM `%s`.`%s`.`%s$row_tracking` "
+ + "/*+ OPTIONS('scan.snapshot-id'='%d') */
WHERE %s",
+ action.catalogName,
+ action.identifier.getDatabaseName(),
+ action.identifier.getObjectName(),
+ baseSnapshotId,
+ filter);
+ LOG.info("Data-evolution delete source query: {}", query);
+
+ Table matchedRows = action.batchTEnv.sqlQuery(query);
+ DataStream<Long> rowIds =
+ action.batchTEnv
+ .toDataStream(matchedRows)
+ .map(
+ (MapFunction<Row, Long>) row -> (Long)
row.getField(0),
+ TypeInformation.of(Long.class));
+
+ DataStream<DeletionTarget> targets =
+ rowIds.rebalance()
+ .map(
+ new RowIdToDeletionTarget(anchorRanges),
+ TypeInformation.of(DeletionTarget.class))
+ // Anchor ranges are part of the mapper closure. Bound
the number of copies
+ // by the configured sink parallelism instead of the
source scan
+ // parallelism.
+ .setParallelism(sinkParallelism)
+ .partitionCustom(new StringHashPartitioner(), new
AnchorKeySelector());
+
+ DataStream<DeletionVectorUpdate> deletionVectorUpdates =
+ targets.transform(
+ "AGGREGATE DELETION VECTORS",
+ TypeInformation.of(DeletionVectorUpdate.class),
+ new DeletionVectorAggregator(
+
storeTable.coreOptions().deletionVectorBitmap64()))
+ .setParallelism(sinkParallelism)
+ .partitionCustom(
+ new StringHashPartitioner(), new
RewriteGroupKeySelector());
+
+ DataStream<Committable> written =
+ deletionVectorUpdates
+ .transform(
+ "WRITE DELETION VECTORS",
+ new CommittableTypeInfo(),
+ new DataEvolutionDeleteOperator(storeTable,
baseSnapshotId))
+ .setParallelism(sinkParallelism);
+
+ CommitterOperatorFactory<Committable, ManifestCommittable>
committerOperator =
+ new CommitterOperatorFactory<>(
+ false,
+ true,
+ commitUser,
+ context ->
+ new StoreCommitter(
+ storeTable,
+ storeTable
+
.newCommit(context.commitUser())
+
.withOperation(Snapshot.Operation.DELETE)
+
.rowIdCheckConflict(baseSnapshotId),
+ context),
+ new NoopCommittableStateManager());
+
+ DataStream<Committable> committed =
+ written.transform("COMMIT OPERATOR", new
CommittableTypeInfo(), committerOperator)
+ .setParallelism(1)
+ .setMaxParallelism(1);
+
+ Transformation<?> end =
+ committed
+ .sinkTo(new DiscardingSink<>())
+ .name("END")
+ .setParallelism(1)
+ .getTransformation();
+
+ return action.executeInternal(
+ Collections.singletonList(end),
+ Collections.singletonList(action.identifier.getFullName()));
+ }
+
+ private List<AnchorRange> planAnchorRanges(FileStoreTable storeTable) {
+ List<AnchorRange> anchorRanges = new ArrayList<>();
+ for (DataSplit split :
+
storeTable.newSnapshotReader().withSnapshot(baseSnapshotId).read().dataSplits())
{
+ Map<String, String> oldIndexFileByDataFile = new HashMap<>();
+ if (split.deletionFiles().isPresent()) {
+ List<DeletionFile> deletionFiles = split.deletionFiles().get();
+ Preconditions.checkState(
+ deletionFiles.size() == split.dataFiles().size(),
+ "Deletion files and data files have different sizes in
bucket path %s.",
+ split.bucketPath());
+ for (int i = 0; i < deletionFiles.size(); i++) {
+ DeletionFile deletionFile = deletionFiles.get(i);
+ if (deletionFile != null) {
+ oldIndexFileByDataFile.put(
+ split.dataFiles().get(i).fileName(),
+ new Path(deletionFile.path()).getName());
+ }
+ }
+ }
+
+ for (List<DataFileMeta> group :
+
DataEvolutionSplitRead.mergeRangesAndSort(split.dataFiles())) {
+ DataFileMeta anchor =
DataEvolutionUtils.retrieveAnchorFile(group, file -> file);
+ Range range = anchor.nonNullRowIdRange();
+ String anchorFilePath =
+ anchor.externalPath().isPresent()
+ ? anchor.externalPath().get()
+ : split.bucketPath() + "/" + anchor.fileName();
+ String rewriteGroup =
+ rewriteGroup(
+ split.bucketPath(),
+ oldIndexFileByDataFile.get(anchor.fileName()),
+ anchorFilePath,
+ sinkParallelism);
+ String oldIndexFileName =
oldIndexFileByDataFile.get(anchor.fileName());
+ anchorRanges.add(
+ new AnchorRange(
+ range.from,
+ range.to,
+ rewriteGroup,
+ split.bucketPath(),
+ oldIndexFileName,
+
SerializationUtils.serializeBinaryRow(split.partition()),
+ anchorFilePath));
+ }
+ }
+
+ anchorRanges.sort(Comparator.comparingLong(range -> range.from));
+ Preconditions.checkState(
+ !anchorRanges.isEmpty(),
+ "Cannot find data-evolution anchor files in snapshot %s.",
+ baseSnapshotId);
+ for (int i = 1; i < anchorRanges.size(); i++) {
+ AnchorRange previous = anchorRanges.get(i - 1);
+ AnchorRange current = anchorRanges.get(i);
+ Preconditions.checkState(
+ previous.to < current.from,
+ "Data-evolution anchor ranges overlap: [%s, %s] and [%s,
%s].",
+ previous.from,
+ previous.to,
+ current.from,
+ current.to);
+ }
+ return anchorRanges;
+ }
+
+ /**
+ * Returns the ownership key for rewriting a deletion-vector index file.
+ *
+ * <p>An existing index file is the atomic rewrite unit because it may
contain deletion vectors
+ * for multiple anchors. New anchors have no shared old file and can
therefore be distributed
+ * over stable shards.
+ */
+ @VisibleForTesting
+ static String rewriteGroup(
+ String bucketPath,
+ @Nullable String oldIndexFile,
+ String anchorFilePath,
+ int parallelism) {
+ if (oldIndexFile != null) {
+ return bucketPath + "\u0000old\u0000" + oldIndexFile;
+ }
+ int shard = Math.floorMod(anchorFilePath.hashCode(), parallelism);
+ return bucketPath + "\u0000new\u0000" + shard;
+ }
+
+ /** A data-evolution anchor file and its covered global row-id range. */
+ private static class AnchorRange implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final long from;
+ private final long to;
+ private final String rewriteGroup;
+ private final String bucketPath;
+ @Nullable private final String oldIndexFileName;
+ private final byte[] serializedPartition;
+ private final String dataFilePath;
+
+ private AnchorRange(
+ long from,
+ long to,
+ String rewriteGroup,
+ String bucketPath,
+ @Nullable String oldIndexFileName,
+ byte[] serializedPartition,
+ String dataFilePath) {
+ this.from = from;
+ this.to = to;
+ this.rewriteGroup = rewriteGroup;
+ this.bucketPath = bucketPath;
+ this.oldIndexFileName = oldIndexFileName;
+ this.serializedPartition = serializedPartition;
+ this.dataFilePath = dataFilePath;
+ }
+ }
+
+ /** Maps a global row id to its anchor data file and local deletion-vector
position. */
+ private static class RowIdToDeletionTarget implements MapFunction<Long,
DeletionTarget> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<AnchorRange> anchorRanges;
+
+ private RowIdToDeletionTarget(List<AnchorRange> anchorRanges) {
+ this.anchorRanges = anchorRanges;
+ }
+
+ @Override
+ public DeletionTarget map(Long rowId) {
+ int low = 0;
+ int high = anchorRanges.size() - 1;
+ int candidate = -1;
+
+ while (low <= high) {
+ int mid = (low + high) >>> 1;
+ if (anchorRanges.get(mid).from <= rowId) {
+ candidate = mid;
+ low = mid + 1;
+ } else {
+ high = mid - 1;
+ }
+ }
+
+ if (candidate < 0 || rowId > anchorRanges.get(candidate).to) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot find data-evolution deletion-vector
anchor range for row id %s.",
+ rowId));
+ }
+
+ AnchorRange anchor = anchorRanges.get(candidate);
+ return new DeletionTarget(
+ anchor.rewriteGroup,
+ anchor.bucketPath,
+ anchor.oldIndexFileName,
+ anchor.serializedPartition,
+ anchor.dataFilePath,
+ rowId - anchor.from);
+ }
+ }
+
+ private static class AnchorKeySelector implements
KeySelector<DeletionTarget, String> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String getKey(DeletionTarget value) {
+ return value.getBucketPath() + "\u0000" + value.getDataFilePath();
+ }
+ }
+
+ private static class RewriteGroupKeySelector
+ implements KeySelector<DeletionVectorUpdate, String> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String getKey(DeletionVectorUpdate value) {
+ return value.getRewriteGroup();
+ }
+ }
+
+ private static class StringHashPartitioner implements Partitioner<String> {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public int partition(String key, int numPartitions) {
+ return Math.floorMod(key.hashCode(), numPartitions);
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
index ed9423cf52..41e3e49ae6 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteAction.java
@@ -19,6 +19,8 @@
package org.apache.paimon.flink.action;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.utils.Preconditions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.Table;
@@ -30,18 +32,22 @@ import org.apache.flink.types.RowKind;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE;
-/** Delete from table action for Flink. */
+/** Delete from table action for Flink, dispatching by table type. */
public class DeleteAction extends TableActionBase {
private static final Logger LOG =
LoggerFactory.getLogger(DeleteAction.class);
private final String filter;
+ @Nullable private final DataEvolutionDelete dataEvolutionDelete;
+ @Nullable private String[] sourceSqls;
public DeleteAction(
String databaseName,
@@ -49,11 +55,54 @@ public class DeleteAction extends TableActionBase {
String filter,
Map<String, String> catalogConfig) {
super(databaseName, tableName, catalogConfig);
+ Preconditions.checkArgument(
+ filter != null && !filter.trim().isEmpty(),
+ "Deletion filter must not be null or blank.");
this.filter = filter;
+
+ if (table instanceof FileStoreTable
+ && ((FileStoreTable)
table).coreOptions().dataEvolutionEnabled()) {
+ dataEvolutionDelete = new DataEvolutionDelete(this, filter);
+ } else {
+ dataEvolutionDelete = null;
+ }
+ }
+
+ /** SQL statements used to configure the batch environment and register
external sources. */
+ public DeleteAction withSourceSqls(String... sourceSqls) {
+ this.sourceSqls = sourceSqls;
+ return this;
+ }
+
+ /** Configures deletion-vector aggregation and writing parallelism for
Data Evolution tables. */
+ public DeleteAction withSinkParallelism(int sinkParallelism) {
+ if (dataEvolutionDelete == null) {
+ throw new UnsupportedOperationException(
+ "Sink parallelism is only supported when deleting from a
Data Evolution table.");
+ }
+ dataEvolutionDelete.withSinkParallelism(sinkParallelism);
+ return this;
}
@Override
public void run() throws Exception {
+ handleSqls();
+ if (dataEvolutionDelete != null) {
+ dataEvolutionDelete.runInternal().await();
+ return;
+ }
+
+ runPrimaryKeyDelete();
+ }
+
+ private void runPrimaryKeyDelete() throws Exception {
+ if (!(table instanceof FileStoreTable)
+ || ((FileStoreTable) table).schema().primaryKeys().isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Delete does not support regular append-only tables. "
+ + "Only primary-key tables and Data Evolution
append tables are supported.");
+ }
+
CoreOptions.MergeEngine mergeEngine =
CoreOptions.fromMap(table.options()).mergeEngine();
if (mergeEngine != DEDUPLICATE) {
throw new UnsupportedOperationException(
@@ -67,8 +116,11 @@ public class DeleteAction extends TableActionBase {
Table queriedTable =
batchTEnv.sqlQuery(
String.format(
- "SELECT * FROM %s WHERE %s",
- identifier.getEscapedFullName(), filter));
+ "SELECT * FROM `%s`.`%s`.`%s` WHERE %s",
+ catalogName,
+ identifier.getDatabaseName(),
+ identifier.getObjectName(),
+ filter));
List<DataStructureConverter<Object, Object>> converters =
queriedTable.getResolvedSchema().getColumnDataTypes().stream()
@@ -95,4 +147,28 @@ public class DeleteAction extends TableActionBase {
batchSink(dataStream).await();
}
+
+ private void handleSqls() {
+ // NOTE: a source SQL statement may change the current catalog and
database. Both target
+ // query paths therefore use fully-qualified target identifiers.
+ if (sourceSqls != null) {
+ for (int i = 0; i < sourceSqls.length; i++) {
+ try {
+ batchTEnv.executeSql(sourceSqls[i]).await();
+ } catch (Exception e) {
+ // Source DDL may contain JDBC credentials. Do not include
the SQL or the
+ // original exception (whose message may echo the SQL) in
logs or the wrapper.
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ String message =
+ String.format(
+ "Failed to execute source SQL statement %s
(cause type: %s).",
+ i + 1, e.getClass().getName());
+ LOG.error(message);
+ throw new RuntimeException(message);
+ }
+ }
+ }
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteActionFactory.java
index 774094168a..9508bfbffa 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteActionFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteActionFactory.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action;
+import java.util.Collection;
import java.util.Optional;
/** Factory to create {@link DeleteAction}. */
@@ -26,6 +27,8 @@ public class DeleteActionFactory implements ActionFactory {
public static final String IDENTIFIER = "delete";
private static final String WHERE = "where";
+ private static final String SOURCE_SQL = "source_sql";
+ private static final String SINK_PARALLELISM = "sink_parallelism";
@Override
public String identifier() {
@@ -35,7 +38,7 @@ public class DeleteActionFactory implements ActionFactory {
@Override
public Optional<Action> create(MultipleParameterToolAdapter params) {
String filter = params.get(WHERE);
- if (filter == null) {
+ if (filter == null || filter.trim().isEmpty()) {
throw new IllegalArgumentException(
"Please specify deletion filter. If you want to delete all
records, please use overwrite (see doc).");
}
@@ -47,11 +50,29 @@ public class DeleteActionFactory implements ActionFactory {
filter,
catalogConfigMap(params));
+ if (params.has(SOURCE_SQL)) {
+ Collection<String> sourceSqls =
params.getMultiParameter(SOURCE_SQL);
+ action.withSourceSqls(sourceSqls.toArray(new String[0]));
+ }
+
+ if (params.has(SINK_PARALLELISM)) {
+ int sinkParallelism;
+ try {
+ sinkParallelism =
Integer.parseInt(params.getRequired(SINK_PARALLELISM));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid sink parallelism, must be an integer", e);
+ }
+ action.withSinkParallelism(sinkParallelism);
+ }
+
return Optional.of(action);
}
+ @Override
public void printHelp() {
- System.out.println("Action \"delete\" deletes data from a table.");
+ System.out.println(
+ "Action \"delete\" deletes data from a primary-key or Data
Evolution table.");
System.out.println();
System.out.println("Syntax:");
@@ -60,18 +81,24 @@ public class DeleteActionFactory implements ActionFactory {
+ " --warehouse <warehouse_path> \\\n"
+ "--database <database_name> \\\n"
+ "--table <table_name> \\\n"
- + "--where <filter_spec>");
+ + "[--source_sql <sql> ...] \\\n"
+ + "--where <filter_spec> \\\n"
+ + "[--sink_parallelism <sink_parallelism>]");
System.out.println(" delete --path <table_path> --where
<filter_spec>");
System.out.println();
System.out.println(
"The '--where <filter_spec>' part is equal to the 'WHERE'
clause in SQL DELETE statement. If you want to delete all records, please use
overwrite (see doc).");
+ System.out.println(
+ "For Data Evolution tables, use repeated --source_sql
arguments to register bounded external tables referenced by subqueries in
--where.");
+ System.out.println(
+ "The --sink_parallelism option applies only to Data Evolution
deletion-vector writing.");
System.out.println();
System.out.println("Examples:");
System.out.println(
" delete --path
hdfs:///path/to/warehouse/test_db.db/test_table --where 'id > (SELECT count(*)
FROM employee)'");
System.out.println(
- " It's equal to 'DELETE FROM test_table WHERE id > (SELECT
count(*) FROM employee)");
+ " delete --warehouse hdfs:///path/to/warehouse --database
test_db --table test_table --source_sql \"CREATE TEMPORARY TABLE candidates
(url STRING) WITH (...)\" --where \"url IN (SELECT url FROM candidates)\"
--sink_parallelism 4");
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteOperator.java
new file mode 100644
index 0000000000..5cf1c94dff
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteOperator.java
@@ -0,0 +1,568 @@
+/*
+ * 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.paimon.flink.dataevolution;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.deletionvectors.Bitmap64DeletionVector;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.deletionvectors.append.BaseAppendDeleteFileMaintainer;
+import org.apache.paimon.flink.sink.Committable;
+import org.apache.paimon.flink.utils.BoundedOneInputOperator;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.utils.SerializationUtils;
+
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
+import static org.apache.paimon.table.BucketMode.UNAWARE_BUCKET;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/**
+ * A bounded Flink operator which persists deletion vectors for a
data-evolution table.
+ *
+ * <p>All updates of the same rewrite group must be shuffled to the same
operator subtask before
+ * this operator. A rewrite group contains all anchors backed by the same
existing deletion-vector
+ * index file. Anchors without an existing deletion vector are assigned to
stable shards. This
+ * ownership rule prevents two subtasks from concurrently replacing the same
old index file while
+ * still allowing an unaware-bucket partition to be processed in parallel.
+ */
+public class DataEvolutionDeleteOperator
+ extends BoundedOneInputOperator<
+ DataEvolutionDeleteOperator.DeletionVectorUpdate, Committable>
{
+
+ private static final long serialVersionUID = 1L;
+
+ private final FileStoreTable table;
+ private final long baseSnapshotId;
+
+ private transient Map<String, RewriteGroupDeletionVectors>
deletionVectorsByRewriteGroup;
+
+ public DataEvolutionDeleteOperator(FileStoreTable table, long
baseSnapshotId) {
+ this.table = table;
+ this.baseSnapshotId = baseSnapshotId;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ checkArgument(
+ table.bucketMode() == BucketMode.BUCKET_UNAWARE,
+ "Data-evolution delete only supports unaware bucket mode, but
table bucket mode is %s.",
+ table.bucketMode());
+ checkArgument(
+ table.coreOptions().dataEvolutionEnabled(),
+ "Data-evolution delete requires data-evolution.enabled to be
true.");
+ checkArgument(
+ table.coreOptions().deletionVectorsEnabled(),
+ "Data-evolution delete requires deletion-vectors.enabled to be
true.");
+
+ // Fail early if the snapshot used to map row ids has already expired.
+ table.snapshot(baseSnapshotId);
+
+ deletionVectorsByRewriteGroup = new LinkedHashMap<>();
+ }
+
+ @Override
+ public void processElement(StreamRecord<DeletionVectorUpdate> element) {
+ DeletionVectorUpdate update =
+ checkNotNull(element.getValue(), "Deletion-vector update is
null.");
+ String rewriteGroup = checkNotNull(update.getRewriteGroup(), "Rewrite
group is null.");
+ String bucketPath = checkNotNull(update.getBucketPath(), "Bucket path
is null.");
+ byte[] serializedPartition =
+ checkNotNull(update.getSerializedPartition(), "Serialized
partition is null.");
+ String dataFilePath =
+ checkNotNull(update.getDataFilePath(), "Anchor data file path
is null.");
+ byte[] serializedDeletionVector =
+ checkNotNull(
+ update.getSerializedDeletionVector(),
+ "Serialized deletion vector is null.");
+
+ RewriteGroupDeletionVectors rewriteGroupDeletionVectors =
+ deletionVectorsByRewriteGroup.get(rewriteGroup);
+ if (rewriteGroupDeletionVectors == null) {
+ // BinaryRow created by deserializeBinaryRow points to the input
byte array. Copy both
+ // representations because Flink may reuse the input object after
processElement
+ // returns. Deserialization is needed only for the first update of
a rewrite group.
+ byte[] copiedPartition = Arrays.copyOf(serializedPartition,
serializedPartition.length);
+ BinaryRow partition =
SerializationUtils.deserializeBinaryRow(copiedPartition).copy();
+ rewriteGroupDeletionVectors =
+ new RewriteGroupDeletionVectors(
+ bucketPath, update.getOldIndexFileName(),
partition, copiedPartition);
+ deletionVectorsByRewriteGroup.put(rewriteGroup,
rewriteGroupDeletionVectors);
+ } else {
+ checkArgument(
+ rewriteGroupDeletionVectors.bucketPath.equals(bucketPath),
+ "Rewrite group %s is associated with different bucket
paths.",
+ rewriteGroup);
+ checkArgument(
+ Objects.equals(
+ rewriteGroupDeletionVectors.oldIndexFileName,
+ update.getOldIndexFileName()),
+ "Rewrite group %s is associated with different old index
files.",
+ rewriteGroup);
+ checkArgument(
+ Arrays.equals(
+ rewriteGroupDeletionVectors.serializedPartition,
serializedPartition),
+ "Rewrite group %s is associated with different
partitions.",
+ rewriteGroup);
+ }
+
+ String dataFileName = new Path(dataFilePath).getName();
+ DeletionVector deletionVector =
+ DeletionVector.deserializeFromBytes(serializedDeletionVector);
+ DeletionVector previous =
+
rewriteGroupDeletionVectors.deletionVectorsByDataFile.putIfAbsent(
+ dataFileName, deletionVector);
+ if (previous != null) {
+ // This is not expected after anchor-level shuffling, but merging
keeps the writer
+ // correct if upstream parallelism or partitioning is changed in
the future.
+ previous.merge(deletionVector);
+ }
+ }
+
+ @Override
+ public void endInput() {
+ if (deletionVectorsByRewriteGroup.isEmpty()) {
+ return;
+ }
+
+ Snapshot baseSnapshot = table.snapshot(baseSnapshotId);
+ IndexFileHandler indexFileHandler =
table.store().newIndexFileHandler();
+ Map<BinaryRow, Set<String>> ownedIndexFileNamesByPartition = new
HashMap<>();
+ for (RewriteGroupDeletionVectors rewriteGroupDeletionVectors :
+ deletionVectorsByRewriteGroup.values()) {
+ if (rewriteGroupDeletionVectors.oldIndexFileName != null) {
+ ownedIndexFileNamesByPartition
+ .computeIfAbsent(
+ rewriteGroupDeletionVectors.partition.copy(),
+ ignored -> new HashSet<>())
+ .add(rewriteGroupDeletionVectors.oldIndexFileName);
+ }
+ }
+
+ Map<BinaryRow, Map<String, IndexManifestEntry>> indexFilesByPartition
= new HashMap<>();
+ if (!ownedIndexFileNamesByPartition.isEmpty()) {
+ for (IndexManifestEntry entry :
+ indexFileHandler.scan(
+ baseSnapshot,
+ candidate -> {
+ if (!candidate
+ .indexFile()
+ .indexType()
+ .equals(DELETION_VECTORS_INDEX)) {
+ return false;
+ }
+ Set<String> ownedIndexFileNames =
+
ownedIndexFileNamesByPartition.get(candidate.partition());
+ return ownedIndexFileNames != null
+ && ownedIndexFileNames.contains(
+
candidate.indexFile().fileName());
+ })) {
+ indexFilesByPartition
+ .computeIfAbsent(entry.partition().copy(), ignored ->
new HashMap<>())
+ .put(entry.indexFile().fileName(), entry);
+ }
+ }
+
+ for (RewriteGroupDeletionVectors rewriteGroupDeletionVectors :
+ deletionVectorsByRewriteGroup.values()) {
+ List<IndexManifestEntry> ownedIndexFiles;
+ if (rewriteGroupDeletionVectors.oldIndexFileName == null) {
+ ownedIndexFiles = Collections.emptyList();
+ } else {
+ Map<String, IndexManifestEntry> partitionIndexFiles =
+
indexFilesByPartition.get(rewriteGroupDeletionVectors.partition);
+ IndexManifestEntry ownedIndexFile =
+ partitionIndexFiles == null
+ ? null
+ : partitionIndexFiles.get(
+
rewriteGroupDeletionVectors.oldIndexFileName);
+ ownedIndexFiles =
+ Collections.singletonList(
+ checkNotNull(
+ ownedIndexFile,
+ "Cannot find owned deletion-vector
index file %s in the base snapshot.",
+
rewriteGroupDeletionVectors.oldIndexFileName));
+ }
+ BaseAppendDeleteFileMaintainer maintainer =
+ BaseAppendDeleteFileMaintainer.forUnawareAppend(
+ indexFileHandler,
+ rewriteGroupDeletionVectors.partition,
+ ownedIndexFiles);
+
+ for (Map.Entry<String, DeletionVector> entry :
+
rewriteGroupDeletionVectors.deletionVectorsByDataFile.entrySet()) {
+ // The maintainer merges the new bitmap with an existing
deletion vector, if any.
+ maintainer.notifyNewDeletionVector(entry.getKey(),
entry.getValue());
+ }
+
+ List<IndexFileMeta> addedIndexFiles = new ArrayList<>();
+ List<IndexFileMeta> deletedIndexFiles = new ArrayList<>();
+ for (IndexManifestEntry entry : maintainer.persist()) {
+ if (entry.kind() == FileKind.ADD) {
+ addedIndexFiles.add(entry.indexFile());
+ } else if (entry.kind() == FileKind.DELETE) {
+ deletedIndexFiles.add(entry.indexFile());
+ } else {
+ throw new IllegalStateException(
+ "Unsupported index manifest entry kind: " +
entry.kind());
+ }
+ }
+
+ CommitMessage commitMessage =
+ new CommitMessageImpl(
+ maintainer.getPartition(),
+ UNAWARE_BUCKET,
+ null,
+ new DataIncrement(
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ addedIndexFiles,
+ deletedIndexFiles),
+ CompactIncrement.emptyIncrement());
+ output.collect(
+ new StreamRecord<>(
+ new
Committable(BatchWriteBuilder.COMMIT_IDENTIFIER, commitMessage)));
+ }
+
+ deletionVectorsByRewriteGroup.clear();
+ }
+
+ private static class RewriteGroupDeletionVectors {
+
+ private final String bucketPath;
+ @Nullable private final String oldIndexFileName;
+ private final BinaryRow partition;
+ private final byte[] serializedPartition;
+ private final Map<String, DeletionVector> deletionVectorsByDataFile;
+
+ private RewriteGroupDeletionVectors(
+ String bucketPath,
+ @Nullable String oldIndexFileName,
+ BinaryRow partition,
+ byte[] serializedPartition) {
+ this.bucketPath = bucketPath;
+ this.oldIndexFileName = oldIndexFileName;
+ this.partition = partition;
+ this.serializedPartition = serializedPartition;
+ this.deletionVectorsByDataFile = new LinkedHashMap<>();
+ }
+ }
+
+ /**
+ * First-stage bounded operator which aggregates row positions into one
compressed deletion
+ * vector per anchor file.
+ *
+ * <p>Targets must be shuffled by anchor file before this operator. The
resulting compressed
+ * updates are shuffled again by rewrite group before they reach {@link
+ * DataEvolutionDeleteOperator}. This keeps high-cardinality row-id
traffic distributed even
+ * when several anchors share one old deletion-vector index file.
+ */
+ public static class DeletionVectorAggregator
+ extends BoundedOneInputOperator<DeletionTarget,
DeletionVectorUpdate> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final boolean useBitmap64;
+
+ private transient Map<String, AggregatedDeletionVector>
deletionVectorsByAnchor;
+
+ public DeletionVectorAggregator(boolean useBitmap64) {
+ this.useBitmap64 = useBitmap64;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+ deletionVectorsByAnchor = new LinkedHashMap<>();
+ }
+
+ @Override
+ public void processElement(StreamRecord<DeletionTarget> element) {
+ DeletionTarget target = checkNotNull(element.getValue(), "Deletion
target is null.");
+ String dataFilePath =
+ checkNotNull(target.getDataFilePath(), "Anchor data file
path is null.");
+ checkNotNull(target.getRewriteGroup(), "Rewrite group is null.");
+ checkNotNull(target.getBucketPath(), "Bucket path is null.");
+ checkNotNull(target.getSerializedPartition(), "Serialized
partition is null.");
+ checkArgument(
+ target.getRowIndex() >= 0,
+ "Deletion-vector row index must be non-negative, but is
%s.",
+ target.getRowIndex());
+
+ String anchorKey = target.getBucketPath() + "\u0000" +
dataFilePath;
+ AggregatedDeletionVector aggregated =
deletionVectorsByAnchor.get(anchorKey);
+ if (aggregated == null) {
+ aggregated = new AggregatedDeletionVector(target,
newDeletionVector());
+ deletionVectorsByAnchor.put(anchorKey, aggregated);
+ } else {
+ aggregated.validate(target);
+ }
+ aggregated.deletionVector.checkedDelete(target.getRowIndex());
+ }
+
+ @Override
+ public void endInput() {
+ for (AggregatedDeletionVector aggregated :
deletionVectorsByAnchor.values()) {
+ output.collect(
+ new StreamRecord<>(
+ new DeletionVectorUpdate(
+ aggregated.rewriteGroup,
+ aggregated.bucketPath,
+ aggregated.oldIndexFileName,
+ aggregated.serializedPartition,
+ aggregated.dataFilePath,
+ DeletionVector.serializeToBytes(
+ aggregated.deletionVector))));
+ }
+ deletionVectorsByAnchor.clear();
+ }
+
+ private DeletionVector newDeletionVector() {
+ return useBitmap64 ? new Bitmap64DeletionVector() : new
BitmapDeletionVector();
+ }
+
+ private static class AggregatedDeletionVector {
+
+ private final String rewriteGroup;
+ private final String bucketPath;
+ @Nullable private final String oldIndexFileName;
+ private final byte[] serializedPartition;
+ private final String dataFilePath;
+ private final DeletionVector deletionVector;
+
+ private AggregatedDeletionVector(DeletionTarget target,
DeletionVector deletionVector) {
+ this.rewriteGroup = target.getRewriteGroup();
+ this.bucketPath = target.getBucketPath();
+ this.oldIndexFileName = target.getOldIndexFileName();
+ this.serializedPartition =
+ Arrays.copyOf(
+ target.getSerializedPartition(),
+ target.getSerializedPartition().length);
+ this.dataFilePath = target.getDataFilePath();
+ this.deletionVector = deletionVector;
+ }
+
+ private void validate(DeletionTarget target) {
+ checkArgument(
+ rewriteGroup.equals(target.getRewriteGroup())
+ && bucketPath.equals(target.getBucketPath())
+ && Objects.equals(oldIndexFileName,
target.getOldIndexFileName())
+ && Arrays.equals(
+ serializedPartition,
target.getSerializedPartition()),
+ "Anchor file %s is associated with inconsistent
deletion metadata.",
+ dataFilePath);
+ }
+ }
+ }
+
+ /**
+ * A network-serializable deletion target.
+ *
+ * <p>{@code dataFilePath} is the anchor data-file path and {@code
rowIndex} is the row's local
+ * position relative to the anchor file's row-id range. The partition is
serialized because a
+ * {@link BinaryRow} can point to reused memory.
+ */
+ public static class DeletionTarget implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private String rewriteGroup;
+ private String bucketPath;
+ @Nullable private String oldIndexFileName;
+ private byte[] serializedPartition;
+ private String dataFilePath;
+ private long rowIndex;
+
+ /** Public no-argument constructor for Flink POJO serialization. */
+ public DeletionTarget() {}
+
+ public DeletionTarget(
+ String rewriteGroup,
+ String bucketPath,
+ @Nullable String oldIndexFileName,
+ byte[] serializedPartition,
+ String dataFilePath,
+ long rowIndex) {
+ this.rewriteGroup = rewriteGroup;
+ this.bucketPath = bucketPath;
+ this.oldIndexFileName = oldIndexFileName;
+ this.serializedPartition = serializedPartition;
+ this.dataFilePath = dataFilePath;
+ this.rowIndex = rowIndex;
+ }
+
+ public String getRewriteGroup() {
+ return rewriteGroup;
+ }
+
+ public void setRewriteGroup(String rewriteGroup) {
+ this.rewriteGroup = rewriteGroup;
+ }
+
+ public String getBucketPath() {
+ return bucketPath;
+ }
+
+ public void setBucketPath(String bucketPath) {
+ this.bucketPath = bucketPath;
+ }
+
+ public @Nullable String getOldIndexFileName() {
+ return oldIndexFileName;
+ }
+
+ public void setOldIndexFileName(@Nullable String oldIndexFileName) {
+ this.oldIndexFileName = oldIndexFileName;
+ }
+
+ public byte[] getSerializedPartition() {
+ return serializedPartition;
+ }
+
+ public void setSerializedPartition(byte[] serializedPartition) {
+ this.serializedPartition = serializedPartition;
+ }
+
+ public String getDataFilePath() {
+ return dataFilePath;
+ }
+
+ public void setDataFilePath(String dataFilePath) {
+ this.dataFilePath = dataFilePath;
+ }
+
+ public long getRowIndex() {
+ return rowIndex;
+ }
+
+ public void setRowIndex(long rowIndex) {
+ this.rowIndex = rowIndex;
+ }
+ }
+
+ /** A compressed anchor deletion vector sent across the rewrite-group
shuffle. */
+ public static class DeletionVectorUpdate implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private String rewriteGroup;
+ private String bucketPath;
+ @Nullable private String oldIndexFileName;
+ private byte[] serializedPartition;
+ private String dataFilePath;
+ private byte[] serializedDeletionVector;
+
+ /** Public no-argument constructor for Flink POJO serialization. */
+ public DeletionVectorUpdate() {}
+
+ public DeletionVectorUpdate(
+ String rewriteGroup,
+ String bucketPath,
+ @Nullable String oldIndexFileName,
+ byte[] serializedPartition,
+ String dataFilePath,
+ byte[] serializedDeletionVector) {
+ this.rewriteGroup = rewriteGroup;
+ this.bucketPath = bucketPath;
+ this.oldIndexFileName = oldIndexFileName;
+ this.serializedPartition = serializedPartition;
+ this.dataFilePath = dataFilePath;
+ this.serializedDeletionVector = serializedDeletionVector;
+ }
+
+ public String getRewriteGroup() {
+ return rewriteGroup;
+ }
+
+ public void setRewriteGroup(String rewriteGroup) {
+ this.rewriteGroup = rewriteGroup;
+ }
+
+ public String getBucketPath() {
+ return bucketPath;
+ }
+
+ public void setBucketPath(String bucketPath) {
+ this.bucketPath = bucketPath;
+ }
+
+ public @Nullable String getOldIndexFileName() {
+ return oldIndexFileName;
+ }
+
+ public void setOldIndexFileName(@Nullable String oldIndexFileName) {
+ this.oldIndexFileName = oldIndexFileName;
+ }
+
+ public byte[] getSerializedPartition() {
+ return serializedPartition;
+ }
+
+ public void setSerializedPartition(byte[] serializedPartition) {
+ this.serializedPartition = serializedPartition;
+ }
+
+ public String getDataFilePath() {
+ return dataFilePath;
+ }
+
+ public void setDataFilePath(String dataFilePath) {
+ this.dataFilePath = dataFilePath;
+ }
+
+ public byte[] getSerializedDeletionVector() {
+ return serializedDeletionVector;
+ }
+
+ public void setSerializedDeletionVector(byte[]
serializedDeletionVector) {
+ this.serializedDeletionVector = serializedDeletionVector;
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java
new file mode 100644
index 0000000000..08d3e58b11
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java
@@ -0,0 +1,494 @@
+/*
+ * 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.paimon.flink.action;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.index.DeletionVectorMeta;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static
org.apache.flink.table.planner.factories.TestValuesTableFactory.changelogRow;
+import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.buildDdl;
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.init;
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.insertInto;
+import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.sEnv;
+import static
org.apache.paimon.flink.util.ReadWriteTableTestUtil.testBatchRead;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Data Evolution IT cases for the unified {@link DeleteAction}. */
+public class DeleteActionDataEvolutionITCase extends ActionITCaseBase {
+
+ private static final String TABLE = "T";
+
+ @BeforeEach
+ public void setUp() {
+ init(warehouse);
+ }
+
+ @Test
+ public void testDeleteAcrossPartitionsAndMergeExistingDeletionVectors()
throws Exception {
+ createDataEvolutionTable(TABLE, false, true);
+ insertInto(TABLE, "(1, 'one', 'A')", "(2, 'two', 'A')", "(3, 'three',
'A')");
+ insertInto(TABLE, "(4, 'four', 'B')", "(5, 'five', 'B')", "(6, 'six',
'B')");
+
+ FileStoreTable table = getFileStoreTable(TABLE);
+ List<String> originalFiles = plannedFiles(table);
+
+ action(TABLE, "id IN (2, 4)", 2).run();
+
+ testBatchRead(
+ "SELECT id, name, dt FROM T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, "one", "A"),
+ changelogRow("+I", 3, "three", "A"),
+ changelogRow("+I", 5, "five", "B"),
+ changelogRow("+I", 6, "six", "B")));
+ assertDeleteSnapshotAndFiles(table, originalFiles, 2L);
+
+ // A second action must merge with the existing DV instead of losing
the first deletion.
+ action(TABLE, "id IN (2, 3)", 2).run();
+
+ testBatchRead(
+ "SELECT id, name, dt FROM T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, "one", "A"),
+ changelogRow("+I", 5, "five", "B"),
+ changelogRow("+I", 6, "six", "B")));
+ assertDeleteSnapshotAndFiles(table, originalFiles, 3L);
+ }
+
+ @Test
+ public void testDeleteBlobRowWithoutRewritingDataOrBlobFiles() throws
Exception {
+ createDataEvolutionTable("BLOB_T", true, true);
+ insertInto("BLOB_T", "(1, 'one', 'A', X'48656C6C6F')", "(2, 'two',
'A', X'5041494D4F4E')");
+
+ FileStoreTable table = getFileStoreTable("BLOB_T");
+ List<String> originalFiles = plannedFiles(table);
+
+ action("BLOB_T", "id = 1", 1).run();
+
+ testBatchRead(
+ "SELECT id, name, picture FROM BLOB_T",
+ Collections.singletonList(
+ changelogRow("+I", 2, "two", new byte[] {80, 65, 73,
77, 79, 78})));
+ assertDeleteSnapshotAndFiles(table, originalFiles, 1L);
+ }
+
+ @Test
+ public void testNoMatchedRowsDoesNotCreateSnapshot() throws Exception {
+ createDataEvolutionTable(TABLE, false, true);
+ insertInto(TABLE, "(1, 'one', 'A')");
+
+ FileStoreTable table = getFileStoreTable(TABLE);
+ long snapshotId = table.latestSnapshot().get().id();
+
+ action(TABLE, "id = 999", 1).run();
+
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+ assertThat(deletionVectorCardinality(table)).isZero();
+ }
+
+ @Test
+ public void testDeleteWithBoundedExternalCandidates() throws Exception {
+ createDataEvolutionTable(TABLE, false, true);
+ insertInto(TABLE, "(1, 'one', 'A')", "(2, 'two', 'A')", "(3, 'three',
'A')");
+
+ String dataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ changelogRow("+I", 2),
+ changelogRow("+I", 2),
+ changelogRow("+I", 999)));
+ String createCandidates =
+ String.format(
+ "CREATE TEMPORARY TABLE deletion_candidate_source (id
INT) "
+ + "WITH ('connector' = 'values', 'bounded' =
'true', "
+ + "'data-id' = '%s')",
+ dataId);
+ String createCandidateView =
+ "CREATE TEMPORARY VIEW deletion_candidates AS "
+ + "SELECT id FROM deletion_candidate_source";
+
+ actionWithSourceSqls(
+ TABLE,
+ "id IN (SELECT id FROM deletion_candidates)",
+ 2,
+ createCandidates,
+ createCandidateView)
+ .run();
+
+ testBatchRead(
+ "SELECT id, name, dt FROM T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 1, "one", "A"), changelogRow("+I",
3, "three", "A")));
+
assertThat(deletionVectorCardinality(getFileStoreTable(TABLE))).isEqualTo(1L);
+ }
+
+ @Test
+ public void testDeleteWithFilteredBoundedExternalTable() throws Exception {
+ createUnpartitionedUrlTable("URL_T");
+ insertInto(
+ "URL_T",
+ "('cold-url', 'cold')",
+ "('hot-url', 'hot')",
+ "('old-never-requested-url', 'old-never-requested')",
+ "('fresh-never-requested-url', 'fresh-never-requested')",
+ "('url-without-access-state', 'no-access-state')");
+
+ String dataId =
+ TestValuesTableFactory.registerData(
+ Arrays.asList(
+ changelogRow(
+ "+I",
+ "cold-url",
+
LocalDateTime.parse("2026-06-01T00:00:00"),
+
LocalDateTime.parse("2026-06-15T00:00:00")),
+ changelogRow(
+ "+I",
+ "hot-url",
+
LocalDateTime.parse("2026-06-01T00:00:00"),
+
LocalDateTime.parse("2026-07-15T00:00:00")),
+ changelogRow(
+ "+I",
+ "old-never-requested-url",
+
LocalDateTime.parse("2026-06-01T00:00:00"),
+ null),
+ changelogRow(
+ "+I",
+ "fresh-never-requested-url",
+
LocalDateTime.parse("2026-07-15T00:00:00"),
+ null),
+ changelogRow(
+ "+I",
+ "missing-target-url",
+
LocalDateTime.parse("2026-06-01T00:00:00"),
+ null)));
+ String createAccessState =
+ String.format(
+ "CREATE TEMPORARY TABLE external_access_state ("
+ + "url STRING, last_ingest_time TIMESTAMP(3), "
+ + "last_request_time TIMESTAMP(3)) "
+ + "WITH ('connector' = 'values', 'bounded' =
'true', "
+ + "'data-id' = '%s')",
+ dataId);
+
+ action(
+ "URL_T",
+ "url IN (SELECT url FROM external_access_state "
+ + "WHERE last_ingest_time < TIMESTAMP
'2026-07-01 00:00:00' "
+ + "AND (last_request_time IS NULL "
+ + "OR last_request_time < TIMESTAMP
'2026-07-01 00:00:00'))",
+ 2)
+ .withSourceSqls(createAccessState)
+ .run();
+
+ testBatchRead(
+ "SELECT url, payload FROM URL_T ORDER BY url",
+ Arrays.asList(
+ changelogRow("+I", "fresh-never-requested-url",
"fresh-never-requested"),
+ changelogRow("+I", "hot-url", "hot"),
+ changelogRow("+I", "url-without-access-state",
"no-access-state")));
+
assertThat(deletionVectorCardinality(getFileStoreTable("URL_T"))).isEqualTo(2L);
+ }
+
+ @Test
+ public void testSourceSqlFailureDoesNotLeakSqlOrCreateSnapshot() throws
Exception {
+ createDataEvolutionTable(TABLE, false, true);
+ insertInto(TABLE, "(1, 'one', 'A')");
+
+ FileStoreTable table = getFileStoreTable(TABLE);
+ long snapshotId = table.latestSnapshot().get().id();
+ String sensitiveMarker = "password_should_not_be_logged";
+
+ assertThatThrownBy(
+ () ->
+ action(TABLE, "id = 1", 1)
+ .withSourceSqls("INVALID SQL " +
sensitiveMarker)
+ .run())
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("source SQL statement 1")
+ .hasMessageNotContaining(sensitiveMarker);
+
assertThat(table.snapshotManager().latestSnapshotId()).isEqualTo(snapshotId);
+ assertThat(deletionVectorCardinality(table)).isZero();
+ }
+
+ @Test
+ public void testUnpartitionedTableWithParallelRewriteGroups() throws
Exception {
+ createUnpartitionedDataEvolutionTable(TABLE);
+ for (int id = 1; id <= 12; id++) {
+ insertInto(TABLE, String.format("(%d, 'value_%d')", id, id));
+ }
+
+ FileStoreTable table = getFileStoreTable(TABLE);
+ List<String> originalFiles = plannedFiles(table);
+
+ action(TABLE, "MOD(id, 2) = 1", 4).run();
+
+ testBatchRead(
+ "SELECT id, name FROM T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 2, "value_2"),
+ changelogRow("+I", 4, "value_4"),
+ changelogRow("+I", 6, "value_6"),
+ changelogRow("+I", 8, "value_8"),
+ changelogRow("+I", 10, "value_10"),
+ changelogRow("+I", 12, "value_12")));
+ assertDeleteSnapshotAndFiles(table, originalFiles, 6L);
+
+ // The second action must safely merge updates which may be backed by
several old DV
+ // index files created by the first parallel action.
+ action(TABLE, "id IN (2, 6, 10)", 4).run();
+
+ testBatchRead(
+ "SELECT id, name FROM T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 4, "value_4"),
+ changelogRow("+I", 8, "value_8"),
+ changelogRow("+I", 12, "value_12")));
+ assertDeleteSnapshotAndFiles(table, originalFiles, 9L);
+ }
+
+ @Test
+ public void testRewriteGroupOwnership() {
+ String bucketPath = "dt=20260719/bucket-0";
+ assertThat(DataEvolutionDelete.rewriteGroup(bucketPath,
"old-dv-index", "anchor-a", 4))
+ .isEqualTo(
+ DataEvolutionDelete.rewriteGroup(
+ bucketPath, "old-dv-index", "anchor-b", 4));
+ assertThat(DataEvolutionDelete.rewriteGroup(bucketPath,
"old-dv-index-a", "anchor-a", 4))
+ .isNotEqualTo(
+ DataEvolutionDelete.rewriteGroup(
+ bucketPath, "old-dv-index-b", "anchor-a", 4));
+ assertThat(DataEvolutionDelete.rewriteGroup(bucketPath, null,
"anchor-a", 4))
+ .startsWith(bucketPath + "\u0000new\u0000");
+ }
+
+ @Test
+ public void testActionsFromSameSnapshotConflictInsteadOfBeingFiltered()
throws Exception {
+ createDataEvolutionTable(TABLE, false, true);
+ insertInto(TABLE, "(1, 'one', 'A')", "(2, 'two', 'A')", "(3, 'three',
'A')");
+
+ // Both actions intentionally capture the same base snapshot. The
second action must use a
+ // different commit user so that strict mode detects the first DELETE
snapshot instead of
+ // treating the second action as an already committed retry.
+ DeleteAction first = action(TABLE, "id = 1", 1);
+ DeleteAction stale = action(TABLE, "id = 2", 1);
+
+ first.run();
+
+ assertThatThrownBy(stale::run)
+
.hasStackTraceContaining(CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT.key());
+ testBatchRead(
+ "SELECT id, name, dt FROM T ORDER BY id",
+ Arrays.asList(
+ changelogRow("+I", 2, "two", "A"), changelogRow("+I",
3, "three", "A")));
+ }
+
+ @Test
+ public void testRejectTableWithoutDeletionVectors() throws Exception {
+ createDataEvolutionTable(TABLE, false, false);
+ insertInto(TABLE, "(1, 'one', 'A')");
+
+ assertThatThrownBy(() -> action(TABLE, "id = 1", 1))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("deletion-vectors.enabled");
+ }
+
+ @Test
+ public void testFactoryRequiresWhere() {
+ assertThatThrownBy(
+ () ->
+ createAction(
+ DeleteAction.class,
+ "delete",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ TABLE))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("deletion filter");
+
+ assertThatThrownBy(
+ () ->
+ createAction(
+ DeleteAction.class,
+ "delete",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ TABLE,
+ "--where",
+ " "))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("deletion filter");
+ }
+
+ private void createDataEvolutionTable(
+ String tableName, boolean withBlob, boolean deletionVectors) {
+ List<String> fields =
+ withBlob
+ ? Arrays.asList("id INT", "name STRING", "dt STRING",
"picture BYTES")
+ : Arrays.asList("id INT", "name STRING", "dt STRING");
+
+ Map<String, String> options =
+ new java.util.HashMap<String, String>() {
+ {
+ put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ put(
+ CoreOptions.DELETION_VECTORS_ENABLED.key(),
+ String.valueOf(deletionVectors));
+ put(CoreOptions.BUCKET.key(), "-1");
+ if (withBlob) {
+ put("blob-field", "picture");
+ }
+ }
+ };
+
+ sEnv.executeSql(
+ buildDdl(
+ tableName,
+ fields,
+ Collections.emptyList(),
+ Collections.singletonList("dt"),
+ options));
+ }
+
+ private void createUnpartitionedDataEvolutionTable(String tableName) {
+ Map<String, String> options =
+ new java.util.HashMap<String, String>() {
+ {
+ put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ put(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ };
+ sEnv.executeSql(
+ buildDdl(
+ tableName,
+ Arrays.asList("id INT", "name STRING"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options));
+ }
+
+ private void createUnpartitionedUrlTable(String tableName) {
+ Map<String, String> options =
+ new java.util.HashMap<String, String>() {
+ {
+ put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ put(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ put(CoreOptions.BUCKET.key(), "-1");
+ }
+ };
+ sEnv.executeSql(
+ buildDdl(
+ tableName,
+ Arrays.asList("url STRING", "payload STRING"),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options));
+ }
+
+ private DeleteAction action(String tableName, String filter, int
sinkParallelism) {
+ return createAction(
+ DeleteAction.class,
+ "delete",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--where",
+ filter,
+ "--sink_parallelism",
+ String.valueOf(sinkParallelism));
+ }
+
+ private DeleteAction actionWithSourceSqls(
+ String tableName,
+ String filter,
+ int sinkParallelism,
+ String firstSourceSql,
+ String secondSourceSql) {
+ return createAction(
+ DeleteAction.class,
+ "delete",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--source_sql",
+ firstSourceSql,
+ "--source_sql",
+ secondSourceSql,
+ "--where",
+ filter,
+ "--sink_parallelism",
+ String.valueOf(sinkParallelism));
+ }
+
+ private static List<String> plannedFiles(FileStoreTable table) {
+ return table.store().newScan().plan().files().stream()
+ .map(entry -> entry.file().fileName())
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private static long deletionVectorCardinality(FileStoreTable table) {
+ Snapshot snapshot = table.latestSnapshot().get();
+ return table.store().newIndexFileHandler().scan(snapshot,
DELETION_VECTORS_INDEX).stream()
+ .map(IndexManifestEntry::indexFile)
+ .filter(index -> index.dvRanges() != null)
+ .flatMap(index -> index.dvRanges().values().stream())
+ .mapToLong(DeletionVectorMeta::cardinality)
+ .sum();
+ }
+
+ private static void assertDeleteSnapshotAndFiles(
+ FileStoreTable table, List<String> originalFiles, long
expectedDeletedRows) {
+
assertThat(table.latestSnapshot().get().operation()).isEqualTo(Snapshot.Operation.DELETE);
+ // This list contains both normal data files and dedicated BLOB files.
+
assertThat(plannedFiles(table)).containsExactlyElementsOf(originalFiles);
+
assertThat(deletionVectorCardinality(table)).isEqualTo(expectedDeletedRows);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionITCase.java
index 02ab4fd286..6d302fa4f0 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionITCase.java
@@ -44,6 +44,7 @@ import static
org.apache.paimon.flink.util.ReadWriteTableTestUtil.init;
import static
org.apache.paimon.flink.util.ReadWriteTableTestUtil.testStreamingRead;
import static
org.apache.paimon.flink.util.ReadWriteTableTestUtil.validateStreamingReadResult;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** IT cases for {@link DeleteAction}. */
public class DeleteActionITCase extends ActionITCaseBase {
@@ -95,6 +96,33 @@ public class DeleteActionITCase extends ActionITCaseBase {
iterator.close();
}
+ @Test
+ public void testRejectRegularAppendOnlyTable() throws Exception {
+ createFileStoreTable(
+ ROW_TYPE,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonMap("bucket", "-1"));
+
+ DeleteAction action =
+ createAction(
+ DeleteAction.class,
+ "delete",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--where",
+ "k=1");
+
+ assertThatThrownBy(action::run)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("regular append-only tables");
+ }
+
private void prepareTable() throws Exception {
Map<String, String> options = new HashMap<>();
FileStoreTable table =
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ProcedureTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ProcedureTest.java
index 74e3aeeac5..79ba2d71fd 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ProcedureTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ProcedureTest.java
@@ -38,7 +38,8 @@ public class ProcedureTest {
@Test
public void testProcedureCoverAllActions() {
Set<String> expectedExclusions = new HashSet<>();
- // Can be covered by `DELETE FROM` syntax. No procedure needed.
+ // Delete is exposed through both Flink SQL and the unified delete
action. No procedure is
+ // needed.
expectedExclusions.add("delete");
List<String> actionIdentifiers =
FactoryUtil.discoverIdentifiers(
