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 5790aa6350 [flink] Support SQL DELETE for data evolution tables (#8988)
5790aa6350 is described below
commit 5790aa6350126e913871f2a9d2b2783f2da9770f
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Aug 3 18:10:49 2026 +0800
[flink] Support SQL DELETE for data evolution tables (#8988)
---
docs/docs/flink/sql-write.mdx | 6 +-
docs/docs/multimodal-table/data-evolution.mdx | 11 +-
.../connector/RowLevelModificationScanContext.java | 22 ++
.../SupportsRowLevelModificationScan.java | 37 +++
.../flink/DataEvolutionDeleteSql117ITCase.java | 43 +++
.../paimon/flink/AbstractFlinkTableFactory.java | 4 +
.../paimon/flink/action/DataEvolutionDelete.java | 334 +--------------------
.../DataEvolutionDeleteSink.java} | 215 ++++---------
...taEvolutionRowLevelModificationScanContext.java | 69 +++++
.../SupportsRowLevelOperationFlinkTableSink.java | 77 ++++-
.../paimon/flink/source/BaseDataTableSource.java | 19 +-
.../flink/source/DataEvolutionDataTableSource.java | 86 ++++++
.../paimon/flink/source/DataTableSource.java | 146 ++++++++-
.../paimon/flink/DataEvolutionDeleteSqlITCase.java | 222 ++++++++++++++
14 files changed, 806 insertions(+), 485 deletions(-)
diff --git a/docs/docs/flink/sql-write.mdx b/docs/docs/flink/sql-write.mdx
index 33114d00a1..8f1d8e3cfb 100644
--- a/docs/docs/flink/sql-write.mdx
+++ b/docs/docs/flink/sql-write.mdx
@@ -214,11 +214,11 @@ UPDATE my_table SET b = 1, c = 2 WHERE a = 'myTable';
:::info
Important table properties setting:
-1. Only primary key tables support this feature.
-2. If the table has primary keys, the following
[MergeEngine](../primary-key-table/merge-engine/) support this feature:
+1. Primary key tables support this feature. The following
[MergeEngine](../primary-key-table/merge-engine/) are supported:
* [deduplicate](../primary-key-table/merge-engine/#deduplicate).
* [partial-update](../primary-key-table/merge-engine/partial-update) with
option 'partial-update.remove-record-on-delete' enabled.
-3. Do not support deleting from table in streaming mode.
+2. With Flink 1.17 or later, append tables in [Data
Evolution](../multimodal-table/data-evolution) mode also support this feature
when row tracking and deletion vectors are enabled and bucket is `-1`.
+3. Deleting from a table is not supported in streaming mode.
:::
diff --git a/docs/docs/multimodal-table/data-evolution.mdx
b/docs/docs/multimodal-table/data-evolution.mdx
index ba5220a1f6..f630c7dc3e 100644
--- a/docs/docs/multimodal-table/data-evolution.mdx
+++ b/docs/docs/multimodal-table/data-evolution.mdx
@@ -275,8 +275,15 @@ WHEN NOT MATCHED BY SOURCE AND t.id > 10 THEN DELETE;
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
+With Flink 1.17 or later, SQL supports `DELETE FROM` for Data Evolution tables
+in batch mode. For row-level predicates, matching rows are recorded in deletion
+vectors, so data files are not rewritten:
+
+```sql
+DELETE FROM target_table WHERE id = 1;
+```
+
+Flink users can also submit the
[`delete` action](../flink/action-jars#deleting-from-a-data-evolution-table).
## Self Updates
diff --git
a/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/connector/RowLevelModificationScanContext.java
b/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/connector/RowLevelModificationScanContext.java
new file mode 100644
index 0000000000..f8e98e515b
--- /dev/null
+++
b/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/connector/RowLevelModificationScanContext.java
@@ -0,0 +1,22 @@
+/*
+ * 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.flink.table.connector;
+
+/** Dummy placeholder for RowLevelModificationScanContext, which was
introduced in Flink 1.17. */
+public interface RowLevelModificationScanContext {}
diff --git
a/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsRowLevelModificationScan.java
b/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsRowLevelModificationScan.java
new file mode 100644
index 0000000000..ac60f5c078
--- /dev/null
+++
b/paimon-flink/paimon-flink-1.16/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsRowLevelModificationScan.java
@@ -0,0 +1,37 @@
+/*
+ * 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.flink.table.connector.source.abilities;
+
+import org.apache.flink.table.connector.RowLevelModificationScanContext;
+
+import javax.annotation.Nullable;
+
+/** Dummy placeholder for SupportsRowLevelModificationScan, which was
introduced in Flink 1.17. */
+public interface SupportsRowLevelModificationScan {
+
+ RowLevelModificationScanContext applyRowLevelModificationScan(
+ RowLevelModificationType rowLevelModificationType,
+ @Nullable RowLevelModificationScanContext previousContext);
+
+ /** Dummy row-level modification type placeholder. */
+ enum RowLevelModificationType {
+ UPDATE,
+ DELETE
+ }
+}
diff --git
a/paimon-flink/paimon-flink-1.17/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSql117ITCase.java
b/paimon-flink/paimon-flink-1.17/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSql117ITCase.java
new file mode 100644
index 0000000000..185e690cc2
--- /dev/null
+++
b/paimon-flink/paimon-flink-1.17/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSql117ITCase.java
@@ -0,0 +1,43 @@
+/*
+ * 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;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests Data Evolution SQL DELETE compatibility with Flink 1.17. */
+public class DataEvolutionDeleteSql117ITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testDelete() {
+ sql(
+ "CREATE TABLE T (id INT, name STRING) WITH ("
+ + "'bucket' = '-1', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'deletion-vectors.enabled' = 'true')");
+ sql("INSERT INTO T VALUES (1, 'one'), (2, 'two')");
+
+ sql("DELETE FROM T WHERE id = 2");
+
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "one"));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
index c3e55247c3..137a7d7534 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
@@ -26,6 +26,7 @@ import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.sink.FlinkFormatTableSink;
import org.apache.paimon.flink.sink.FlinkTableSink;
+import org.apache.paimon.flink.source.DataEvolutionDataTableSource;
import org.apache.paimon.flink.source.DataTableSource;
import org.apache.paimon.flink.source.SystemTableSource;
import org.apache.paimon.options.Options;
@@ -98,6 +99,9 @@ public abstract class AbstractFlinkTableFactory
}
if (origin instanceof SystemCatalogTable) {
return new SystemTableSource(table, unbounded,
context.getObjectIdentifier());
+ } else if
(CoreOptions.fromMap(table.options()).dataEvolutionEnabled()) {
+ return new DataEvolutionDataTableSource(
+ context.getObjectIdentifier(), table, unbounded, context);
} else {
return new DataTableSource(context.getObjectIdentifier(), table,
unbounded, context);
}
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
index fc0bd93697..0f04337333 100644
---
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
@@ -19,37 +19,15 @@
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.flink.dataevolution.DataEvolutionDeleteSink;
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.streaming.api.datastream.DataStreamSink;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.types.Row;
@@ -59,31 +37,9 @@ 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.
- */
+/** Internal implementation which logically deletes rows from a Data Evolution
append table. */
class DataEvolutionDelete implements Serializable {
private static final long serialVersionUID = 1L;
@@ -111,42 +67,13 @@ class DataEvolutionDelete implements Serializable {
}
FileStoreTable storeTable = (FileStoreTable) action.table;
+ DataEvolutionDeleteSink.validateTable(storeTable);
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) {
@@ -161,10 +88,6 @@ class DataEvolutionDelete implements Serializable {
/** 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` "
@@ -185,258 +108,21 @@ class DataEvolutionDelete implements Serializable {
(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();
-
+ DataStreamSink<?> end =
+ new DataEvolutionDeleteSink(storeTable, baseSnapshotId,
sinkParallelism)
+ .sinkFrom(rowIds);
return action.executeInternal(
- Collections.singletonList(end),
+ Collections.singletonList(end.getTransformation()),
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);
- }
+ return DataEvolutionDeleteSink.rewriteGroup(
+ bucketPath, oldIndexFile, anchorFilePath, parallelism);
}
}
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/dataevolution/DataEvolutionDeleteSink.java
similarity index 63%
copy from
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java
copy to
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionDeleteSink.java
index fc0bd93697..55d6814c73 100644
---
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/dataevolution/DataEvolutionDeleteSink.java
@@ -16,12 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.flink.action;
+package org.apache.paimon.flink.dataevolution;
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;
@@ -46,15 +45,10 @@ 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.datastream.DataStreamSink;
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;
@@ -66,133 +60,45 @@ 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 {
+/** Writes deletion vectors for row ids of a Data Evolution table. */
+public class DataEvolutionDeleteSink 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 FileStoreTable table;
private final long baseSnapshotId;
+ private final int sinkParallelism;
- 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) {
+ public DataEvolutionDeleteSink(FileStoreTable table, long baseSnapshotId,
int sinkParallelism) {
+ validateTable(table);
Preconditions.checkArgument(
sinkParallelism > 0,
"Sink parallelism must be a positive integer, but is %s.",
sinkParallelism);
+ this.table =
+ baseSnapshotId ==
DataEvolutionRowLevelModificationScanContext.EMPTY_TABLE_SNAPSHOT
+ ? table
+ : (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.COMMIT_STRICT_MODE_LAST_SAFE_SNAPSHOT
+ .key(),
+
String.valueOf(baseSnapshotId)));
+ this.baseSnapshotId = baseSnapshotId;
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',
'%s'='full') */ WHERE %s",
- action.catalogName,
- action.identifier.getDatabaseName(),
- action.identifier.getObjectName(),
- baseSnapshotId,
- CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(),
- 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));
+ public DataStreamSink<?> sinkFrom(DataStream<Long> rowIds) {
+ if (baseSnapshotId ==
DataEvolutionRowLevelModificationScanContext.EMPTY_TABLE_SNAPSHOT) {
+ return rowIds.sinkTo(new
DiscardingSink<>()).name("END").setParallelism(1);
+ }
+ List<AnchorRange> anchorRanges = planAnchorRanges();
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());
@@ -201,7 +107,7 @@ class DataEvolutionDelete implements Serializable {
"AGGREGATE DELETION VECTORS",
TypeInformation.of(DeletionVectorUpdate.class),
new DeletionVectorAggregator(
-
storeTable.coreOptions().deletionVectorBitmap64()))
+
table.coreOptions().deletionVectorBitmap64()))
.setParallelism(sinkParallelism)
.partitionCustom(
new StringHashPartitioner(), new
RewriteGroupKeySelector());
@@ -211,9 +117,10 @@ class DataEvolutionDelete implements Serializable {
.transform(
"WRITE DELETION VECTORS",
new CommittableTypeInfo(),
- new DataEvolutionDeleteOperator(storeTable,
baseSnapshotId))
+ new DataEvolutionDeleteOperator(table,
baseSnapshotId))
.setParallelism(sinkParallelism);
+ String commitUser =
CoreOptions.createCommitUser(table.coreOptions().toConfiguration());
CommitterOperatorFactory<Committable, ManifestCommittable>
committerOperator =
new CommitterOperatorFactory<>(
false,
@@ -221,9 +128,8 @@ class DataEvolutionDelete implements Serializable {
commitUser,
context ->
new StoreCommitter(
- storeTable,
- storeTable
-
.newCommit(context.commitUser())
+ table,
+ table.newCommit(context.commitUser())
.withOperation(Snapshot.Operation.DELETE)
.rowIdCheckConflict(baseSnapshotId),
context),
@@ -234,22 +140,42 @@ class DataEvolutionDelete implements Serializable {
.setParallelism(1)
.setMaxParallelism(1);
- Transformation<?> end =
- committed
- .sinkTo(new DiscardingSink<>())
- .name("END")
- .setParallelism(1)
- .getTransformation();
+ DataStreamSink<Committable> end =
+ committed.sinkTo(new
DiscardingSink<>()).name("END").setParallelism(1);
+ end.getTransformation().setMaxParallelism(1);
+ return end;
+ }
- return action.executeInternal(
- Collections.singletonList(end),
- Collections.singletonList(action.identifier.getFullName()));
+ public static void validateTable(FileStoreTable table) {
+ CoreOptions coreOptions = table.coreOptions();
+ if (!table.schema().primaryKeys().isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete only supports append tables without
primary keys.");
+ }
+ if (!coreOptions.rowTrackingEnabled()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete requires row-tracking.enabled to be
true.");
+ }
+ if (!coreOptions.dataEvolutionEnabled()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete requires data-evolution.enabled to
be true.");
+ }
+ if (!coreOptions.deletionVectorsEnabled()) {
+ throw new UnsupportedOperationException(
+ "Data-evolution delete requires deletion-vectors.enabled
to be true.");
+ }
+ if (table.bucketMode() != BucketMode.BUCKET_UNAWARE) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Data-evolution delete only supports unaware
bucket mode, but table bucket mode is %s.",
+ table.bucketMode()));
+ }
}
- private List<AnchorRange> planAnchorRanges(FileStoreTable storeTable) {
+ private List<AnchorRange> planAnchorRanges() {
List<AnchorRange> anchorRanges = new ArrayList<>();
for (DataSplit split :
-
storeTable.newSnapshotReader().withSnapshot(baseSnapshotId).read().dataSplits())
{
+
table.newSnapshotReader().withSnapshot(baseSnapshotId).read().dataSplits()) {
Map<String, String> oldIndexFileByDataFile = new HashMap<>();
if (split.deletionFiles().isPresent()) {
List<DeletionFile> deletionFiles = split.deletionFiles().get();
@@ -275,18 +201,16 @@ class DataEvolutionDelete implements Serializable {
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,
+ rewriteGroup(
+ split.bucketPath(),
+ oldIndexFileName,
+ anchorFilePath,
+ sinkParallelism),
split.bucketPath(),
oldIndexFileName,
SerializationUtils.serializeBinaryRow(split.partition()),
@@ -313,15 +237,8 @@ class DataEvolutionDelete implements Serializable {
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(
+ public static String rewriteGroup(
String bucketPath,
@Nullable String oldIndexFile,
String anchorFilePath,
@@ -333,7 +250,6 @@ class DataEvolutionDelete implements Serializable {
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;
@@ -364,7 +280,6 @@ class DataEvolutionDelete implements Serializable {
}
}
- /** 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;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionRowLevelModificationScanContext.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionRowLevelModificationScanContext.java
new file mode 100644
index 0000000000..659f83d51d
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/dataevolution/DataEvolutionRowLevelModificationScanContext.java
@@ -0,0 +1,69 @@
+/*
+ * 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.utils.Pair;
+
+import org.apache.flink.table.connector.RowLevelModificationScanContext;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** Snapshots used by Paimon sources in a row-level modification statement. */
+public class DataEvolutionRowLevelModificationScanContext
+ implements RowLevelModificationScanContext {
+
+ public static final long EMPTY_TABLE_SNAPSHOT = -1L;
+
+ private final Map<Pair<String, String>, Long> snapshotIds;
+
+ private DataEvolutionRowLevelModificationScanContext(
+ Map<Pair<String, String>, Long> snapshotIds) {
+ this.snapshotIds = Collections.unmodifiableMap(snapshotIds);
+ }
+
+ public static DataEvolutionRowLevelModificationScanContext addSnapshot(
+ @Nullable RowLevelModificationScanContext previous,
+ String tableLocation,
+ String branch,
+ long snapshotId) {
+ Map<Pair<String, String>, Long> snapshotIds = new HashMap<>();
+ if (previous instanceof DataEvolutionRowLevelModificationScanContext) {
+ snapshotIds.putAll(
+ ((DataEvolutionRowLevelModificationScanContext)
previous).snapshotIds);
+ }
+ snapshotIds.put(Pair.of(tableLocation, branch), snapshotId);
+ return new DataEvolutionRowLevelModificationScanContext(snapshotIds);
+ }
+
+ @Nullable
+ public static Long snapshotId(
+ @Nullable RowLevelModificationScanContext context,
+ String tableLocation,
+ String branch) {
+ if (!(context instanceof
DataEvolutionRowLevelModificationScanContext)) {
+ return null;
+ }
+ return ((DataEvolutionRowLevelModificationScanContext) context)
+ .snapshotIds.get(Pair.of(tableLocation, branch));
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
index 7ed8e9f176..8e80038ecd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java
@@ -20,15 +20,25 @@ package org.apache.paimon.flink.sink;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.CoreOptions.MergeEngine;
+import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.LogicalTypeConversion;
+import org.apache.paimon.flink.PaimonDataStreamSinkProvider;
import org.apache.paimon.flink.PredicateConverter;
+import org.apache.paimon.flink.dataevolution.DataEvolutionDeleteSink;
+import
org.apache.paimon.flink.dataevolution.DataEvolutionRowLevelModificationScanContext;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.OnlyPartitionKeyEqualVisitor;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.connector.RowLevelModificationScanContext;
@@ -36,6 +46,7 @@ import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.abilities.SupportsDeletePushDown;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate;
+import org.apache.flink.table.data.RowData;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.types.logical.RowType;
@@ -63,6 +74,7 @@ public abstract class SupportsRowLevelOperationFlinkTableSink
extends FlinkTable
implements SupportsRowLevelUpdate, SupportsRowLevelDelete,
SupportsDeletePushDown {
@Nullable protected Predicate deletePredicate;
+ @Nullable protected Long dataEvolutionDeleteSnapshotId;
public SupportsRowLevelOperationFlinkTableSink(
ObjectIdentifier tableIdentifier, Table table,
DynamicTableFactory.Context context) {
@@ -75,6 +87,7 @@ public abstract class SupportsRowLevelOperationFlinkTableSink
extends FlinkTable
copied.staticPartitions = new HashMap<>(staticPartitions);
copied.overwrite = overwrite;
copied.deletePredicate = deletePredicate;
+ copied.dataEvolutionDeleteSnapshotId = dataEvolutionDeleteSnapshotId;
return copied;
}
@@ -127,6 +140,32 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
@Override
public RowLevelDeleteInfo applyRowLevelDelete(
@Nullable RowLevelModificationScanContext
rowLevelModificationScanContext) {
+ if (isDataEvolutionTable()) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ DataEvolutionDeleteSink.validateTable(fileStoreTable);
+ Long snapshotId =
+ DataEvolutionRowLevelModificationScanContext.snapshotId(
+ rowLevelModificationScanContext,
+ fileStoreTable.location().toString(),
+ fileStoreTable.snapshotManager().branch());
+ if (snapshotId == null) {
+ throw new IllegalStateException(
+ "Data Evolution DELETE requires a snapshot from the
Paimon table source.");
+ }
+ dataEvolutionDeleteSnapshotId = snapshotId;
+ return new RowLevelDeleteInfo() {
+ @Override
+ public Optional<List<Column>> requiredColumns() {
+ return Optional.of(
+ Collections.singletonList(
+ Column.metadata(
+ SpecialFields.ROW_ID.name(),
+ DataTypes.BIGINT().notNull(),
+ SpecialFields.ROW_ID.name(),
+ true)));
+ }
+ };
+ }
validatePKUpsertDeletable(table);
return new RowLevelDeleteInfo() {};
}
@@ -135,7 +174,9 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
@Override
public boolean applyDeleteFilters(List<ResolvedExpression> list) {
- validatePKUpsertDeletable(table);
+ if (!isDataEvolutionTable()) {
+ validatePKUpsertDeletable(table);
+ }
List<Predicate> predicates = new ArrayList<>();
RowType rowType = LogicalTypeConversion.toLogicalType(table.rowType());
for (ResolvedExpression filter : list) {
@@ -151,6 +192,35 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
return canPushDownDeleteFilter();
}
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ if (dataEvolutionDeleteSnapshotId == null) {
+ return super.getSinkRuntimeProvider(context);
+ }
+ if (!context.isBounded()) {
+ throw new UnsupportedOperationException(
+ "Data Evolution DELETE only supports batch mode.");
+ }
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ int sinkParallelism =
+ Options.fromMap(table.options())
+ .getOptional(FlinkConnectorOptions.SINK_PARALLELISM)
+ .orElse(1);
+ return new PaimonDataStreamSinkProvider(
+ dataStream -> {
+ DataStream<Long> rowIds =
+ dataStream.map(
+ (MapFunction<RowData, Long>) row ->
row.getLong(0),
+ TypeInformation.of(Long.class));
+ return new DataEvolutionDeleteSink(
+ fileStoreTable,
dataEvolutionDeleteSnapshotId, sinkParallelism)
+ .sinkFrom(rowIds);
+ },
+ tableIdentifier.asSummaryString(),
+ table);
+ }
+
@Override
public Optional<Long> executeDeletion() {
try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
@@ -188,4 +258,9 @@ public abstract class
SupportsRowLevelOperationFlinkTableSink extends FlinkTable
deletePredicate.visit(visitor);
return visitor.partitions();
}
+
+ private boolean isDataEvolutionTable() {
+ return table instanceof FileStoreTable
+ && ((FileStoreTable)
table).coreOptions().dataEvolutionEnabled();
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
index 58e5f38d4a..827b7b04b5 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
@@ -166,8 +166,10 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
return createPushedAggregateScan();
}
+ Table scanTable = tableForScan();
+
WatermarkStrategy<RowData> watermarkStrategy = this.watermarkStrategy;
- Options options = Options.fromMap(table.options());
+ Options options = Options.fromMap(scanTable.options());
if (watermarkStrategy != null) {
WatermarkEmitStrategy emitStrategy =
options.get(SCAN_WATERMARK_EMIT_STRATEGY);
if (emitStrategy == WatermarkEmitStrategy.ON_EVENT) {
@@ -189,10 +191,10 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
}
FlinkSourceBuilder sourceBuilder =
- new FlinkSourceBuilder(table)
+ new FlinkSourceBuilder(scanTable)
.sourceName(tableIdentifier.asSummaryString())
.sourceBounded(!unbounded)
- .projection(projectFields)
+ .projection(projectFieldsForScan())
.predicate(predicate)
.partitionPredicate(partitionPredicate)
.limit(limit)
@@ -201,7 +203,7 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
return new PaimonDataStreamScanProvider(
!unbounded,
env ->
- PostponeMergeOnRead.usesCustomSource(table)
+ PostponeMergeOnRead.usesCustomSource(scanTable)
? sourceBuilder.env(env).build()
: sourceBuilder
.sourceParallelism(inferSourceParallelism(env))
@@ -211,6 +213,15 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
table);
}
+ protected Table tableForScan() {
+ return table;
+ }
+
+ @Nullable
+ protected int[][] projectFieldsForScan() {
+ return projectFields;
+ }
+
private ScanRuntimeProvider createPushedAggregateScan() {
checkNotNull(pushedAggregateResult);
StaticRowDataSource source =
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionDataTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionDataTableSource.java
new file mode 100644
index 0000000000..a35d35c5bb
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataEvolutionDataTableSource.java
@@ -0,0 +1,86 @@
+/*
+ * 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.source;
+
+import org.apache.paimon.flink.source.aggregate.PushedAggregateResult;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.Table;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
+import
org.apache.flink.table.connector.source.abilities.SupportsRowLevelModificationScan;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DynamicTableFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** A {@link DataTableSource} exposing row-level modification metadata for
Data Evolution tables. */
+public class DataEvolutionDataTableSource extends DataTableSource
+ implements SupportsReadingMetadata, SupportsRowLevelModificationScan {
+
+ public DataEvolutionDataTableSource(
+ ObjectIdentifier tableIdentifier,
+ Table table,
+ boolean unbounded,
+ DynamicTableFactory.Context context) {
+ super(tableIdentifier, table, unbounded, context);
+ }
+
+ private DataEvolutionDataTableSource(
+ ObjectIdentifier tableIdentifier,
+ Table table,
+ boolean unbounded,
+ DynamicTableFactory.Context context,
+ @Nullable Predicate predicate,
+ @Nullable int[][] projectFields,
+ @Nullable Long limit,
+ @Nullable WatermarkStrategy<RowData> watermarkStrategy,
+ @Nullable List<String> dynamicPartitionFilteringFields,
+ @Nullable PushedAggregateResult pushedAggregateResult) {
+ super(
+ tableIdentifier,
+ table,
+ unbounded,
+ context,
+ predicate,
+ projectFields,
+ limit,
+ watermarkStrategy,
+ dynamicPartitionFilteringFields,
+ pushedAggregateResult);
+ }
+
+ @Override
+ protected DataTableSource newSource() {
+ return new DataEvolutionDataTableSource(
+ tableIdentifier,
+ table,
+ unbounded,
+ context,
+ predicate,
+ projectFields,
+ limit,
+ watermarkStrategy,
+ dynamicPartitionFilteringFields,
+ pushedAggregateResult);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
index b5d5636406..c496872993 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
@@ -18,25 +18,43 @@
package org.apache.paimon.flink.source;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.flink.PaimonDataStreamScanProvider;
+import org.apache.paimon.flink.Projection;
+import
org.apache.paimon.flink.dataevolution.DataEvolutionRowLevelModificationScanContext;
import org.apache.paimon.flink.source.aggregate.PushedAggregateResult;
+import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.stats.ColStats;
import org.apache.paimon.stats.Statistics;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
+import org.apache.paimon.table.system.RowTrackingTable;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.RowLevelModificationScanContext;
+import org.apache.flink.table.connector.source.ScanTableSource.ScanContext;
+import
org.apache.flink.table.connector.source.ScanTableSource.ScanRuntimeProvider;
import
org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import
org.apache.flink.table.connector.source.abilities.SupportsRowLevelModificationScan.RowLevelModificationType;
import
org.apache.flink.table.connector.source.abilities.SupportsStatisticReport;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.plan.stats.ColumnStats;
import org.apache.flink.table.plan.stats.TableStats;
+import org.apache.flink.table.types.DataType;
import javax.annotation.Nullable;
import java.util.AbstractMap;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -51,7 +69,9 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
public class DataTableSource extends BaseDataTableSource
implements SupportsStatisticReport, SupportsDynamicFiltering {
- @Nullable private List<String> dynamicPartitionFilteringFields;
+ @Nullable protected List<String> dynamicPartitionFilteringFields;
+ @Nullable private Long rowLevelModificationSnapshotId;
+ private List<String> metadataKeys = Collections.emptyList();
public DataTableSource(
ObjectIdentifier tableIdentifier,
@@ -110,6 +130,13 @@ public class DataTableSource extends BaseDataTableSource
@Override
public DataTableSource copy() {
+ DataTableSource copied = newSource();
+ copied.rowLevelModificationSnapshotId = rowLevelModificationSnapshotId;
+ copied.metadataKeys = metadataKeys;
+ return copied;
+ }
+
+ protected DataTableSource newSource() {
return new DataTableSource(
tableIdentifier,
table,
@@ -123,6 +150,123 @@ public class DataTableSource extends BaseDataTableSource
pushedAggregateResult);
}
+ public RowLevelModificationScanContext applyRowLevelModificationScan(
+ RowLevelModificationType rowLevelModificationType,
+ @Nullable RowLevelModificationScanContext previousContext) {
+ if (rowLevelModificationType != RowLevelModificationType.DELETE
+ || !isDataEvolutionTable()
+ ||
TimeTravelUtil.hasTimeTravelOptions(Options.fromMap(table.options()))) {
+ return previousContext;
+ }
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ Long snapshotId = fileStoreTable.snapshotManager().latestSnapshotId();
+ rowLevelModificationSnapshotId =
+ snapshotId == null
+ ?
DataEvolutionRowLevelModificationScanContext.EMPTY_TABLE_SNAPSHOT
+ : snapshotId;
+ return DataEvolutionRowLevelModificationScanContext.addSnapshot(
+ previousContext,
+ fileStoreTable.location().toString(),
+ fileStoreTable.snapshotManager().branch(),
+ rowLevelModificationSnapshotId);
+ }
+
+ public Map<String, DataType> listReadableMetadata() {
+ // Flink calls this after applyRowLevelModificationScan for row-level
operations.
+ if (rowLevelModificationSnapshotId == null || !isDataEvolutionTable())
{
+ return Collections.emptyMap();
+ }
+ Map<String, DataType> metadata = new LinkedHashMap<>();
+ metadata.put(SpecialFields.ROW_ID.name(),
DataTypes.BIGINT().notNull());
+ return metadata;
+ }
+
+ public void applyReadableMetadata(List<String> metadataKeys, DataType
producedDataType) {
+ for (String metadataKey : metadataKeys) {
+ if (!SpecialFields.ROW_ID.name().equals(metadataKey)) {
+ throw new UnsupportedOperationException(
+ "Unsupported Paimon metadata column: " + metadataKey);
+ }
+ }
+ this.metadataKeys = metadataKeys;
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext)
{
+ if (rowLevelModificationSnapshotId == null
+ || rowLevelModificationSnapshotId
+ !=
DataEvolutionRowLevelModificationScanContext.EMPTY_TABLE_SNAPSHOT) {
+ return super.getScanRuntimeProvider(scanContext);
+ }
+
+ Table scanTable = tableForScan();
+ org.apache.paimon.types.RowType rowType = scanTable.rowType();
+ int[][] projection = projectFieldsForScan();
+ if (projection != null) {
+ rowType = Projection.of(projection).project(rowType);
+ }
+ StaticRowDataSource source = new
StaticRowDataSource(Collections.emptyList(), rowType);
+ return new PaimonDataStreamScanProvider(
+ true,
+ env ->
+ env.fromSource(
+ source,
+ WatermarkStrategy.noWatermarks(),
+ tableIdentifier.asSummaryString())
+ .setParallelism(1),
+ tableIdentifier.asSummaryString(),
+ table);
+ }
+
+ @Override
+ protected Table tableForScan() {
+ if (rowLevelModificationSnapshotId == null) {
+ return table;
+ }
+
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ if (rowLevelModificationSnapshotId
+ !=
DataEvolutionRowLevelModificationScanContext.EMPTY_TABLE_SNAPSHOT) {
+ Map<String, String> options = new HashMap<>();
+ options.put(
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ String.valueOf(rowLevelModificationSnapshotId));
+ options.put(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full");
+ fileStoreTable = (FileStoreTable) fileStoreTable.copy(options);
+ }
+
+ return metadataKeys.isEmpty() ? fileStoreTable : new
RowTrackingTable(fileStoreTable);
+ }
+
+ @Override
+ protected int[][] projectFieldsForScan() {
+ if (metadataKeys.isEmpty()) {
+ return projectFields;
+ }
+
+ int physicalFieldCount = table.rowType().getFieldCount();
+ int[][] physicalProjection = projectFields;
+ if (physicalProjection == null) {
+ physicalProjection = new int[physicalFieldCount][];
+ for (int i = 0; i < physicalFieldCount; i++) {
+ physicalProjection[i] = new int[] {i};
+ }
+ }
+
+ int[][] projection =
+ Arrays.copyOf(physicalProjection, physicalProjection.length +
metadataKeys.size());
+ for (int i = 0; i < metadataKeys.size(); i++) {
+ projection[physicalProjection.length + i] = new int[]
{physicalFieldCount};
+ }
+ return projection;
+ }
+
+ private boolean isDataEvolutionTable() {
+ return table instanceof FileStoreTable
+ && ((FileStoreTable)
table).coreOptions().dataEvolutionEnabled();
+ }
+
@Override
public TableStats reportStatistics() {
if (unbounded) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
new file mode 100644
index 0000000000..2120729098
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/DataEvolutionDeleteSqlITCase.java
@@ -0,0 +1,222 @@
+/*
+ * 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;
+
+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.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static
org.apache.paimon.deletionvectors.DeletionVectorsIndexFile.DELETION_VECTORS_INDEX;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for Flink SQL DELETE on Data Evolution tables. */
+public class DataEvolutionDeleteSqlITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testDeleteRowsWithoutRewritingDataFiles() throws Exception {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'one', 'A'), (2, 'two', 'A'), (3,
'three', 'A')");
+ sql("INSERT INTO T VALUES (4, 'four', 'B'), (5, 'five', 'B'), (6,
'six', 'B')");
+
+ FileStoreTable table = paimonTable("T");
+ List<String> originalFiles = plannedFiles(table);
+
+ sql("DELETE FROM T WHERE id IN (2, 4)");
+
+ assertThat(sql("SELECT id, name, dt FROM T ORDER BY id"))
+ .containsExactly(
+ Row.of(1, "one", "A"),
+ Row.of(3, "three", "A"),
+ Row.of(5, "five", "B"),
+ Row.of(6, "six", "B"));
+ assertDeleteSnapshot(table, originalFiles, 2L);
+
+ sql("DELETE FROM T WHERE id IN (2, 3)");
+
+ assertThat(sql("SELECT id, name, dt FROM T ORDER BY id"))
+ .containsExactly(
+ Row.of(1, "one", "A"), Row.of(5, "five", "B"),
Row.of(6, "six", "B"));
+ assertDeleteSnapshot(table, originalFiles, 3L);
+ }
+
+ @Test
+ public void testDeleteFromEmptyTable() throws Exception {
+ createTable();
+
+ sql("DELETE FROM T WHERE id = 1");
+
+ assertThat(sql("SELECT * FROM T")).isEmpty();
+ assertThat(paimonTable("T").latestSnapshot()).isEmpty();
+ }
+
+ @Test
+ public void testDeleteRequiresDeletionVectors() {
+ createTable();
+ sql("ALTER TABLE T RESET ('deletion-vectors.enabled')");
+
+ assertThatThrownBy(() -> sql("DELETE FROM T WHERE id = 1"))
+ .hasRootCauseMessage(
+ "Data-evolution delete requires
deletion-vectors.enabled to be true.");
+ }
+
+ @Test
+ public void testDeleteWithSubquery() {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'one', 'A'), (2, 'two', 'A'), (3,
'three', 'A')");
+ sql("CREATE TABLE S (id INT)");
+ sql("INSERT INTO S VALUES (2), (3)");
+
+ sql("DELETE FROM T WHERE id IN (SELECT id FROM S)");
+
+ assertThat(sql("SELECT id, name, dt FROM
T")).containsExactly(Row.of(1, "one", "A"));
+ }
+
+ @Test
+ public void testDeleteWithSameTableBranchSubquery() {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'one', 'A'), (2, 'two', 'A')");
+ sql("CALL sys.create_tag('default.T', 'tag1')");
+ sql("CALL sys.create_branch('default.T', 'test', 'tag1')");
+ sql("INSERT INTO `T$branch_test` VALUES (3, 'three', 'A')");
+
+ assertThat(sql("SELECT id FROM `T$branch_test` WHERE id =
2")).containsExactly(Row.of(2));
+
+ sql("DELETE FROM T WHERE id IN (SELECT id FROM `T$branch_test` WHERE
id = 2)");
+
+ assertThat(sql("SELECT id FROM T ORDER BY
id")).containsExactly(Row.of(1));
+ assertThat(sql("SELECT id FROM `T$branch_test` ORDER BY id"))
+ .containsExactly(Row.of(1), Row.of(2), Row.of(3));
+ }
+
+ @Test
+ public void testDeleteWithSameTableTagSubquery() {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'one', 'A'), (2, 'two', 'A')");
+ sql("CALL sys.create_tag('default.T', 'tag1')");
+ sql("INSERT INTO T VALUES (3, 'three', 'A')");
+
+ sql(
+ "DELETE FROM T WHERE id IN (SELECT id FROM T "
+ + "/*+ OPTIONS('scan.tag-name'='tag1') */ WHERE id =
2)");
+
+ assertThat(sql("SELECT id FROM T ORDER BY
id")).containsExactly(Row.of(1), Row.of(3));
+ }
+
+ @Test
+ public void testDeletePartitionRemovesGlobalIndex() throws Exception {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'one', 'A'), (2, 'two', 'B')");
+ createGlobalIndex();
+
+ FileStoreTable table = paimonTable("T");
+
assertThat(globalIndexPartitions(table)).containsExactlyInAnyOrder("A", "B");
+
+ sql("DELETE FROM T WHERE dt = 'A'");
+
+ assertThat(sql("SELECT id, name, dt FROM
T")).containsExactly(Row.of(2, "two", "B"));
+
assertThat(table.latestSnapshot().get().operation()).isEqualTo(Snapshot.Operation.TRUNCATE);
+ assertThat(globalIndexPartitions(table)).containsExactly("B");
+ }
+
+ @Test
+ public void testDeleteWholeTableRemovesGlobalIndex() throws Exception {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'one', 'A'), (2, 'two', 'B')");
+ createGlobalIndex();
+
+ FileStoreTable table = paimonTable("T");
+
assertThat(globalIndexPartitions(table)).containsExactlyInAnyOrder("A", "B");
+
+ sql("DELETE FROM T");
+
+ assertThat(sql("SELECT * FROM T")).isEmpty();
+
assertThat(table.latestSnapshot().get().operation()).isEqualTo(Snapshot.Operation.TRUNCATE);
+ assertThat(plannedFiles(table)).isEmpty();
+
assertThat(table.store().newIndexFileHandler().scanEntries()).isEmpty();
+ }
+
+ @Test
+ public void testDeleteUsesFullScalarIndexSearch() {
+ createTable();
+ sql("INSERT INTO T VALUES (1, 'old', 'A')");
+ createGlobalIndex();
+ sql("INSERT INTO T VALUES (2, 'new', 'A')");
+ sql("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')");
+
+ sql("DELETE FROM T WHERE name = 'new'");
+
+ assertThat(sql("SELECT id, name, dt FROM
T")).containsExactly(Row.of(1, "old", "A"));
+ }
+
+ private void createGlobalIndex() {
+ sql(
+ "CALL sys.create_global_index(`table` => 'default.T', "
+ + "index_column => 'name', index_type => 'btree')");
+ }
+
+ private void createTable() {
+ sql(
+ "CREATE TABLE T (id INT, name STRING, dt STRING) PARTITIONED
BY (dt) WITH ("
+ + "'bucket' = '-1', "
+ + "'row-tracking.enabled' = 'true', "
+ + "'data-evolution.enabled' = 'true', "
+ + "'deletion-vectors.enabled' = 'true', "
+ + "'sink.parallelism' = '2')");
+ }
+
+ 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 List<String> globalIndexPartitions(FileStoreTable table) {
+ return table.store().newIndexFileHandler().scan("btree").stream()
+ .map(entry -> entry.partition().getString(0).toString())
+ .distinct()
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private static void assertDeleteSnapshot(
+ FileStoreTable table, List<String> originalFiles, long
deletedRows) {
+
assertThat(table.latestSnapshot().get().operation()).isEqualTo(Snapshot.Operation.DELETE);
+
assertThat(plannedFiles(table)).containsExactlyElementsOf(originalFiles);
+ assertThat(deletionVectorCardinality(table)).isEqualTo(deletedRows);
+ }
+}
