You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ho...@apache.org on 2026/08/10 07:56:26 UTC
(fluss) branch main updated: [flink] Support batch delete and update in flink sink (#3863)
This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 8bcf22e5c [flink] Support batch delete and update in flink sink (#3863)
8bcf22e5c is described below
commit 8bcf22e5cef30141e09d70e040546fca4110f2be
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Mon Aug 10 15:56:21 2026 +0800
[flink] Support batch delete and update in flink sink (#3863)
---
.../apache/fluss/flink/sink/FlinkTableSink.java | 38 ++---
.../fluss/flink/source/FlinkTableSource.java | 11 --
.../fluss/flink/sink/FlinkTableSinkITCase.java | 183 +++++++++++++++++----
.../enumerator/FlinkSourceEnumeratorTest.java | 58 +++++++
4 files changed, 230 insertions(+), 60 deletions(-)
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java
index 8035028de..a0d4aacc9 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java
@@ -123,23 +123,19 @@ public class FlinkTableSink
@Override
public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
- if (!streaming) {
- return ChangelogMode.insertOnly();
- } else {
- if (primaryKeyIndexes.length > 0 || sinkIgnoreDelete) {
- // primary-key table or ignore_delete mode can accept RowKind.DELETE
- ChangelogMode.Builder builder = ChangelogMode.newBuilder();
- for (RowKind kind : requestedMode.getContainedKinds()) {
- // optimize out the update_before messages
- if (kind != RowKind.UPDATE_BEFORE) {
- builder.addContainedKind(kind);
- }
+ if (primaryKeyIndexes.length > 0 || (streaming && sinkIgnoreDelete)) {
+ // Primary-key tables can accept row-level changes in batch mode. In streaming mode,
+ // ignore-delete sinks can also accept and drop DELETE messages.
+ ChangelogMode.Builder builder = ChangelogMode.newBuilder();
+ for (RowKind kind : requestedMode.getContainedKinds()) {
+ // optimize out the update_before messages
+ if (kind != RowKind.UPDATE_BEFORE) {
+ builder.addContainedKind(kind);
}
- return builder.build();
- } else {
- return ChangelogMode.insertOnly();
}
+ return builder.build();
}
+ return ChangelogMode.insertOnly();
}
@Override
@@ -282,7 +278,7 @@ public class FlinkTableSink
@Override
public boolean applyDeleteFilters(List<ResolvedExpression> filters) {
- validateUpdatableAndDeletable();
+ validateDeletable();
if (filters.size() != primaryKeyIndexes.length) {
// only supports delete on primary key
return false;
@@ -333,15 +329,15 @@ public class FlinkTableSink
@Override
public RowLevelDeleteInfo applyRowLevelDelete(
@Nullable RowLevelModificationScanContext rowLevelModificationScanContext) {
- throw new UnsupportedOperationException(
- "Currently, Fluss table only supports DELETE statement with conditions on primary key.");
+ validateDeletable();
+ return new RowLevelDeleteInfo() {};
}
@Override
public RowLevelUpdateInfo applyRowLevelUpdate(
List<Column> updatedColumns,
@Nullable RowLevelModificationScanContext rowLevelModificationScanContext) {
- validateUpdatableAndDeletable();
+ validateUpdatable();
Set<String> primaryKeys = getPrimaryKeyNames();
updatedColumns.forEach(
column -> {
@@ -373,7 +369,7 @@ public class FlinkTableSink
};
}
- private void validateUpdatableAndDeletable() {
+ private void validateUpdatable() {
if (primaryKeyIndexes.length == 0) {
throw new UnsupportedOperationException(
String.format(
@@ -386,8 +382,10 @@ public class FlinkTableSink
"Table %s uses the '%s' merge engine which does not support DELETE or UPDATE statements.",
tablePath, mergeEngineType));
}
+ }
- // Check table-level delete behavior configuration
+ private void validateDeletable() {
+ validateUpdatable();
if (tableDeleteBehavior == DeleteBehavior.DISABLE) {
throw new UnsupportedOperationException(
String.format(
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
index 782c8c49f..1a48d03c2 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
@@ -160,9 +160,6 @@ public class FlinkTableSource
@Nullable private GenericRowData singleRowFilter;
- // whether the scan is for row-level modification
- @Nullable private RowLevelModificationType modificationScanType;
-
// count(*) push down
private boolean selectRowCount = false;
@@ -431,12 +428,6 @@ public class FlinkTableSource
@Override
public Source<RowData, ?, ?> createSource() {
- if (modificationScanType != null) {
- throw new UnsupportedOperationException(
- "Currently, Fluss table only supports "
- + modificationScanType
- + " statement with conditions on primary key.");
- }
if (hasPrimaryKey()
&& startupOptions.startupMode
!= FlinkConnectorOptions.ScanStartupMode.FULL) {
@@ -533,7 +524,6 @@ public class FlinkTableSource
source.producedDataType = producedDataType;
source.projectedFields = projectedFields;
source.singleRowFilter = singleRowFilter;
- source.modificationScanType = modificationScanType;
source.partitionFilters = partitionFilters;
source.lakeSource = lakeSource;
source.logRecordBatchFilter = logRecordBatchFilter;
@@ -821,7 +811,6 @@ public class FlinkTableSource
public RowLevelModificationScanContext applyRowLevelModificationScan(
RowLevelModificationType rowLevelModificationType,
@Nullable RowLevelModificationScanContext rowLevelModificationScanContext) {
- modificationScanType = rowLevelModificationType;
return null;
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
index 02a8af15f..ce46947aa 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
@@ -914,6 +914,80 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
assertResultsIgnoreOrder(row5, expected, true);
}
+ @Test
+ void testBatchDeleteAndUpdateStmtOnPkTable() throws Exception {
+ String tableName = "pk_table_batch_delete_update_test";
+ tBatchEnv.executeSql(
+ String.format(
+ "create table %s ("
+ + " a int not null,"
+ + " b bigint, "
+ + " c string,"
+ + " primary key (a) not enforced"
+ + ")",
+ tableName));
+ List<String> insertValues =
+ Arrays.asList(
+ "(1, 3501, 'Beijing')",
+ "(2, 3502, 'Shanghai')",
+ "(3, 3503, 'Berlin')",
+ "(4, 3504, 'Seattle')",
+ "(5, 3505, 'Boston')",
+ "(6, 3506, 'London')");
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "INSERT INTO %s(a,b,c) VALUES %s",
+ tableName, String.join(", ", insertValues)))
+ .await();
+
+ tBatchEnv.executeSql("UPDATE " + tableName + " SET c = 'China' WHERE b <= 3503").await();
+
+ CloseableIterator<Row> rowIter =
+ tBatchEnv.executeSql(String.format("select * from %s", tableName)).collect();
+ List<String> expectedRows =
+ Arrays.asList(
+ "+I[1, 3501, China]",
+ "+I[2, 3502, China]",
+ "+I[3, 3503, China]",
+ "+I[4, 3504, Seattle]",
+ "+I[5, 3505, Boston]",
+ "+I[6, 3506, London]");
+ assertResultsIgnoreOrder(rowIter, expectedRows, true);
+
+ tBatchEnv.executeSql("DELETE FROM " + tableName + " WHERE b >= 3504").await();
+
+ rowIter = tBatchEnv.executeSql(String.format("select * from %s", tableName)).collect();
+ expectedRows =
+ Arrays.asList("+I[1, 3501, China]", "+I[2, 3502, China]", "+I[3, 3503, China]");
+ assertResultsIgnoreOrder(rowIter, expectedRows, true);
+
+ CloseableIterator<Row> changelogIter =
+ tEnv.executeSql(
+ String.format(
+ "select * from %s /*+ OPTIONS('scan.startup.mode' = 'earliest') */",
+ tableName))
+ .collect();
+ expectedRows =
+ Arrays.asList(
+ "+I[1, 3501, Beijing]",
+ "+I[2, 3502, Shanghai]",
+ "+I[3, 3503, Berlin]",
+ "+I[4, 3504, Seattle]",
+ "+I[5, 3505, Boston]",
+ "+I[6, 3506, London]",
+ "-U[1, 3501, Beijing]",
+ "+U[1, 3501, China]",
+ "-U[2, 3502, Shanghai]",
+ "+U[2, 3502, China]",
+ "-U[3, 3503, Berlin]",
+ "+U[3, 3503, China]",
+ "-D[4, 3504, Seattle]",
+ "-D[5, 3505, Boston]",
+ "-D[6, 3506, London]");
+ assertResultsIgnoreOrder(changelogIter, expectedRows, true);
+ }
+
@Test
void testDeleteAndUpdateStmtOnPartitionedPkTable() throws Exception {
String tableName = "partitioned_pk_table_delete_test";
@@ -1039,7 +1113,7 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
}
@Test
- void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
+ void testBatchDeleteAndUpdateStmtOnPartialPK() throws Exception {
// test primary-key table
String t1 = "t1";
tBatchEnv.executeSql(
@@ -1051,11 +1125,15 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
+ " primary key (a, b) not enforced"
+ ")",
t1));
- assertThatThrownBy(() -> tBatchEnv.executeSql("DELETE FROM " + t1 + " WHERE a = 1").await())
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Currently, Fluss table only supports DELETE statement with conditions on primary key.");
-
+ tBatchEnv
+ .executeSql(
+ "INSERT INTO "
+ + t1
+ + "(a, b, c) VALUES"
+ + "(1, 1001, 'Beijing'),"
+ + "(1, 1002, 'Shanghai'),"
+ + "(2, 2001, 'Berlin')")
+ .await();
assertThatThrownBy(
() ->
tBatchEnv
@@ -1065,15 +1143,17 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
.hasMessageContaining(
"Updates to primary keys are not supported, primaryKeys ([a, b]), updatedColumns ([b])");
- assertThatThrownBy(
- () ->
- tBatchEnv
- .executeSql(
- "UPDATE " + t1 + " SET c = 'New York' WHERE a = 1")
- .await())
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Currently, Fluss table only supports UPDATE statement with conditions on primary key.");
+ tBatchEnv.executeSql("UPDATE " + t1 + " SET c = 'China' WHERE a = 1").await();
+ CloseableIterator<Row> rowIter =
+ tBatchEnv.executeSql(String.format("select * from %s", t1)).collect();
+ assertResultsIgnoreOrder(
+ rowIter,
+ Arrays.asList("+I[1, 1001, China]", "+I[1, 1002, China]", "+I[2, 2001, Berlin]"),
+ true);
+
+ tBatchEnv.executeSql("DELETE FROM " + t1 + " WHERE a = 1").await();
+ rowIter = tBatchEnv.executeSql(String.format("select * from %s", t1)).collect();
+ assertResultsIgnoreOrder(rowIter, Collections.singletonList("+I[2, 2001, Berlin]"), true);
// test partitioned primary-key table
String t2 = "t2";
@@ -1088,11 +1168,30 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
+ " with ('table.auto-partition.enabled' = 'true',"
+ " 'table.auto-partition.time-unit' = 'year')",
t2));
- assertThatThrownBy(() -> tBatchEnv.executeSql("DELETE FROM " + t2 + " WHERE a = 1").await())
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Currently, Fluss table only supports DELETE statement with conditions on primary key.");
-
+ Collection<String> partitions =
+ waitUntilPartitions(
+ FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(),
+ TablePath.of(DEFAULT_DB, t2))
+ .values();
+ String partition1 = partitions.iterator().next();
+ String partition2 = partition1.equals("2030") ? "2031" : "2030";
+ tBatchEnv.executeSql(
+ String.format("alter table %s add partition (c = '%s')", t2, partition2));
+ tBatchEnv
+ .executeSql(
+ "INSERT INTO "
+ + t2
+ + "(a, b, c) VALUES"
+ + "(1, 1001, '"
+ + partition1
+ + "'),"
+ + "(1, 1002, '"
+ + partition2
+ + "'),"
+ + "(2, 2001, '"
+ + partition2
+ + "')")
+ .await();
assertThatThrownBy(
() ->
tBatchEnv
@@ -1102,14 +1201,31 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
.hasMessageContaining(
"Updates to primary keys are not supported, primaryKeys ([a, c]), updatedColumns ([c])");
- assertThatThrownBy(
- () ->
- tBatchEnv
- .executeSql("UPDATE " + t2 + " SET b = 4004 WHERE a = 1")
- .await())
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Currently, Fluss table only supports UPDATE statement with conditions on primary key.");
+ tBatchEnv.executeSql("UPDATE " + t2 + " SET b = 4004 WHERE a = 1").await();
+ rowIter = tBatchEnv.executeSql(String.format("select * from %s", t2)).collect();
+ assertResultsIgnoreOrder(
+ rowIter,
+ Arrays.asList(
+ "+I[1, 4004, " + partition1 + "]",
+ "+I[1, 4004, " + partition2 + "]",
+ "+I[2, 2001, " + partition2 + "]"),
+ true);
+
+ tBatchEnv.executeSql("DELETE FROM " + t2 + " WHERE a = 1").await();
+ rowIter = tBatchEnv.executeSql(String.format("select * from %s", t2)).collect();
+ assertResultsIgnoreOrder(
+ rowIter, Collections.singletonList("+I[2, 2001, " + partition2 + "]"), true);
+
+ tBatchEnv
+ .executeSql(
+ "INSERT INTO " + t2 + "(a, b, c) VALUES (3, 3001, '" + partition1 + "')")
+ .await();
+
+ // test delete rows by partition-only filter.
+ tBatchEnv.executeSql("DELETE FROM " + t2 + " WHERE c = '" + partition2 + "'").await();
+ rowIter = tBatchEnv.executeSql(String.format("select * from %s", t2)).collect();
+ assertResultsIgnoreOrder(
+ rowIter, Collections.singletonList("+I[3, 3001, " + partition1 + "]"), true);
}
@Test
@@ -1507,7 +1623,7 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
}
@Test
- void testDeleteBehaviorDisabledForDeleteStmt() {
+ void testDeleteBehaviorDisabledForDeleteStmt() throws Exception {
String tableName = "delete_behavior_disable_table";
tBatchEnv.executeSql(
String.format(
@@ -1530,6 +1646,15 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
String.format(
"Table %s has delete behavior set to 'disable' which does not support DELETE statements.",
tablePath));
+
+ tBatchEnv
+ .executeSql(String.format("INSERT INTO %s VALUES (1, 1001, 'Beijing')", tableName))
+ .await();
+ tBatchEnv.executeSql("UPDATE " + tableName + " SET c = 'China' WHERE a = 1").await();
+
+ CloseableIterator<Row> rowIter =
+ tBatchEnv.executeSql(String.format("select * from %s", tableName)).collect();
+ assertResultsIgnoreOrder(rowIter, Collections.singletonList("+I[1, 1001, China]"), true);
}
@ParameterizedTest
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
index 0c73afc3c..a49dbebc7 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java
@@ -46,6 +46,8 @@ import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.predicate.Predicate;
+import org.apache.fluss.predicate.PredicateBuilder;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
import org.apache.fluss.server.zk.ZooKeeperClient;
@@ -53,6 +55,7 @@ import org.apache.fluss.server.zk.data.lake.LakeTableHelper;
import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot;
import org.apache.fluss.shaded.guava32.com.google.common.collect.ImmutableMap;
import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
import org.apache.flink.api.connector.source.ReaderInfo;
import org.apache.flink.api.connector.source.SourceEvent;
@@ -435,6 +438,61 @@ class FlinkSourceEnumeratorTest extends FlinkTestBase {
}
}
+ @Test
+ void testBatchPartitionFilterOnlyGeneratesMatchingPartitionSplits() throws Throwable {
+ createTable(DEFAULT_TABLE_PATH, DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR);
+ ZooKeeperClient zooKeeperClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient();
+ Map.Entry<Long, String> targetPartition =
+ waitUntilPartitions(zooKeeperClient, DEFAULT_TABLE_PATH)
+ .entrySet()
+ .iterator()
+ .next();
+ RowType partitionRowType = RowType.builder().field("name", DataTypes.STRING()).build();
+ Object partitionValue =
+ PredicateBuilder.convertJavaObject(
+ partitionRowType.getTypeAt(0), targetPartition.getValue());
+ Predicate partitionFilter = new PredicateBuilder(partitionRowType).equal(0, partitionValue);
+ int numSubtasks = 3;
+
+ try (MockSplitEnumeratorContext<SourceSplitBase> context =
+ new MockSplitEnumeratorContext<>(numSubtasks);
+ FlinkSourceEnumerator enumerator =
+ new FlinkSourceEnumerator(
+ DEFAULT_TABLE_PATH,
+ flussConf,
+ false,
+ true,
+ context,
+ OffsetsInitializer.earliest(),
+ DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS,
+ false,
+ partitionFilter,
+ null,
+ LeaseContext.DEFAULT,
+ false)) {
+ enumerator.start();
+ for (int i = 0; i < numSubtasks; i++) {
+ registerReader(context, enumerator, i);
+ }
+ context.runNextOneTimeCallable();
+
+ List<SourceSplitBase> assignedSplits =
+ getReadersAssignments(context).values().stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList());
+ assertThat(assignedSplits).hasSize(DEFAULT_BUCKET_NUM);
+ assertThat(assignedSplits)
+ .allSatisfy(
+ split -> {
+ assertThat(split).isInstanceOf(LogSplit.class);
+ assertThat(split.getTableBucket().getPartitionId())
+ .isEqualTo(targetPartition.getKey());
+ assertThat(split.getPartitionName())
+ .isEqualTo(targetPartition.getValue());
+ });
+ }
+ }
+
@Test
void testBatchPkTableRejectsNonFullStartupMode() throws Throwable {
createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR);