Skip to content

[flink] Support batch delete and update in flink sink #3863

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
loserwang1024 merged 3 commits into from
Aug 10, 2026
Merged

[flink] Support batch delete and update in flink sink #3863

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4d46148
[flink] Support batch delete in flink sink
Jul 31, 2026
b3ba2b1
[flink] Support batch update in flink sink
Aug 5, 2026
85da8c3
modified based on yangyang's advice
Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 18 additions & 20 deletions fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,23 +123,19 @@ public 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
Expand Down Expand Up @@ -282,7 +278,7 @@ public void applyStaticPartition(Map<String, String> partition) {

@Override
public boolean applyDeleteFilters(List<ResolvedExpression> filters) {
validateUpdatableAndDeletable();
validateDeletable();
if (filters.size() != primaryKeyIndexes.length) {
// only supports delete on primary key
return false;
Expand Down Expand Up @@ -333,15 +329,15 @@ public Optional<Long> executeDeletion() {
@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 -> {
Expand Down Expand Up @@ -373,7 +369,7 @@ public RowLevelUpdateMode getRowLevelUpdateMode() {
};
}

private void validateUpdatableAndDeletable() {
private void validateUpdatable() {
if (primaryKeyIndexes.length == 0) {
throw new UnsupportedOperationException(
String.format(
Expand All @@ -386,8 +382,10 @@ private void validateUpdatableAndDeletable() {
"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(
Expand Down
11 changes: 0 additions & 11 deletions ...link/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -431,12 +428,6 @@ public boolean isBounded() {

@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) {
Expand Down Expand Up @@ -533,7 +524,6 @@ public DynamicTableSource copy() {
source.producedDataType = producedDataType;
source.projectedFields = projectedFields;
source.singleRowFilter = singleRowFilter;
source.modificationScanType = modificationScanType;
source.partitionFilters = partitionFilters;
source.lakeSource = lakeSource;
source.logRecordBatchFilter = logRecordBatchFilter;
Expand Down Expand Up @@ -821,7 +811,6 @@ private Set<String> computeAvailableStatsColumns(RowType flussRowType) {
public RowLevelModificationScanContext applyRowLevelModificationScan(
RowLevelModificationType rowLevelModificationType,
@Nullable RowLevelModificationScanContext rowLevelModificationScanContext) {
modificationScanType = rowLevelModificationType;
return null;
}

Expand Down
183 changes: 154 additions & 29 deletions ...nk/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -914,6 +914,80 @@ void testDeleteAndUpdateStmtOnPkTable() throws Exception {
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";
Expand Down Expand Up @@ -1039,7 +1113,7 @@ void testUnsupportedDeleteAndUpdateStmtOnLogTable(boolean isPartitionedTable) {
}

@Test
void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
void testBatchDeleteAndUpdateStmtOnPartialPK() throws Exception {
// test primary-key table
String t1 = "t1";
tBatchEnv.executeSql(
Expand All @@ -1051,11 +1125,15 @@ void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
+ " 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
Expand All @@ -1065,15 +1143,17 @@ void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
.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";
Expand All @@ -1088,11 +1168,30 @@ void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
+ " 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
Expand All @@ -1102,14 +1201,31 @@ void testUnsupportedDeleteAndUpdateStmtOnPartialPK() {
.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();
Comment thread
loserwang1024 marked this conversation as resolved.
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
Expand Down Expand Up @@ -1507,7 +1623,7 @@ public InsertAndExpectValues(List<String> insertValues, List<String> expectedRow
}

@Test
void testDeleteBehaviorDisabledForDeleteStmt() {
void testDeleteBehaviorDisabledForDeleteStmt() throws Exception {
String tableName = "delete_behavior_disable_table";
tBatchEnv.executeSql(
String.format(
Expand All @@ -1530,6 +1646,15 @@ void testDeleteBehaviorDisabledForDeleteStmt() {
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
Expand Down
Loading
Loading