Skip to content

Commit b25ba46

Browse files
litiliulitiliu
authored
[server] Skip unchanged upserts for primary-key tables (#4439)
* [server] Support update_if_changed merge engine for primary-key tables * [common] Use locale-independent merge engine parsing * [server] Add opt-in skip-unchanged upserts Replace the update_if_changed merge engine with table.upsert.skip-unchanged. Compare complete candidate rows in the KV write path using schema-aware logical equality and reject incompatible WAL, TTL, merge engine, and overwrite modes. Cover equality and configuration edge cases, update the Flink sink and documentation, and keep the implementation focused on the write path. * [server] Refine skip-unchanged upsert comparison Compare complete DataType definitions before treating rows as unchanged. Clarify the existing FULL-image lookup cost and describe skipped state writes without implying that upserts are rejected. * [docs] Clarify skip-unchanged upsert behavior * [server] Compare unchanged upserts by serialized bytes Compare the candidate BinaryValue schema ID and serialized row bytes with the stored value without materializing row fields. Cover schema changes, representation differences, batch duplicates, and overwrite rejection. Move the option guide into the primary key table documentation and link the option reference to its section.
1 parent 5af9dda commit b25ba46

14 files changed

Lines changed: 748 additions & 5 deletions

File tree

‎fluss-client/src/test/java/org/apache/fluss/client/table/FlussTableITCase.java‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1561,6 +1561,140 @@ void testFirstRowMergeEngine(boolean doProjection) throws Exception {
15611561
}
15621562
}
15631563

1564+
@ParameterizedTest
1565+
@ValueSource(booleans = {false, true})
1566+
void testSkipUnchangedUpsertsIsOptIn(boolean skipUnchanged) throws Exception {
1567+
TableDescriptor tableDescriptor =
1568+
TableDescriptor.builder()
1569+
.schema(DATA1_SCHEMA_PK)
1570+
.property(ConfigOptions.TABLE_UPSERT_SKIP_UNCHANGED, skipUnchanged)
1571+
.build();
1572+
RowType rowType = DATA1_SCHEMA_PK.getRowType();
1573+
TablePath tablePath =
1574+
TablePath.of("test_db_1", "test_skip_unchanged_upserts_" + skipUnchanged);
1575+
createTable(tablePath, tableDescriptor, false);
1576+
1577+
try (Table table = conn.getTable(tablePath)) {
1578+
UpsertWriter upsertWriter = table.newUpsert().createWriter();
1579+
// insert a row
1580+
upsertWriter.upsert(row(0, "v0"));
1581+
// value-identical upserts: should be no-ops that emit no changelog
1582+
upsertWriter.upsert(row(0, "v0"));
1583+
upsertWriter.upsert(row(0, "v0"));
1584+
// a field differs: should emit an update changelog
1585+
upsertWriter.upsert(row(0, "v1"));
1586+
// delete the row: should emit a delete changelog
1587+
upsertWriter.delete(row(0, "v1"));
1588+
upsertWriter.flush();
1589+
1590+
// Identical writes emit changelog only when suppression is disabled.
1591+
List<ScanRecord> expected = new ArrayList<>();
1592+
expected.add(new ScanRecord(-1, -1, ChangeType.INSERT, row(0, "v0")));
1593+
if (!skipUnchanged) {
1594+
for (int i = 0; i < 2; i++) {
1595+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v0")));
1596+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, "v0")));
1597+
}
1598+
}
1599+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v0")));
1600+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, "v1")));
1601+
expected.add(new ScanRecord(-1, -1, ChangeType.DELETE, row(0, "v1")));
1602+
1603+
LogScanner logScanner = table.newScan().createLogScanner();
1604+
logScanner.subscribeFromBeginning(0);
1605+
List<ScanRecord> actualLogRecords = new ArrayList<>(expected.size());
1606+
while (actualLogRecords.size() < expected.size()) {
1607+
ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
1608+
scanRecords.forEach(actualLogRecords::add);
1609+
}
1610+
assertThat(logScanner.poll(Duration.ofSeconds(1))).isEmpty();
1611+
logScanner.close();
1612+
1613+
assertThat(actualLogRecords).hasSize(expected.size());
1614+
for (int i = 0; i < actualLogRecords.size(); i++) {
1615+
ScanRecord actual = actualLogRecords.get(i);
1616+
assertThat(actual.getChangeType()).isEqualTo(expected.get(i).getChangeType());
1617+
assertThatRow(actual.getRow())
1618+
.withSchema(rowType)
1619+
.isEqualTo(expected.get(i).getRow());
1620+
}
1621+
}
1622+
}
1623+
1624+
@Test
1625+
void testSkipUnchangedPartialUpdatesAndDeletes() throws Exception {
1626+
Schema schema =
1627+
Schema.newBuilder()
1628+
.column("id", DataTypes.INT())
1629+
.column("name", DataTypes.STRING())
1630+
.column("data", DataTypes.STRING())
1631+
.primaryKey("id")
1632+
.build();
1633+
TableDescriptor tableDescriptor =
1634+
TableDescriptor.builder()
1635+
.schema(schema)
1636+
.distributedBy(1, "id")
1637+
.property(ConfigOptions.TABLE_UPSERT_SKIP_UNCHANGED, true)
1638+
.build();
1639+
RowType rowType = schema.getRowType();
1640+
TablePath tablePath = TablePath.of("test_db_1", "test_skip_unchanged_partial_update");
1641+
createTable(tablePath, tableDescriptor, false);
1642+
1643+
try (Table table = conn.getTable(tablePath)) {
1644+
UpsertWriter fullWriter = table.newUpsert().createWriter();
1645+
fullWriter.upsert(row(0, "v0", "kept")).get();
1646+
1647+
UpsertWriter keyOnlyWriter = table.newUpsert().partialUpdate("id").createWriter();
1648+
// A primary-key-only partial upsert leaves the complete row unchanged.
1649+
keyOnlyWriter.upsert(row(0, null, null)).get();
1650+
1651+
UpsertWriter partialWriter =
1652+
table.newUpsert().partialUpdate("id", "name").createWriter();
1653+
// unchanged partial upsert: no changelog
1654+
partialWriter.upsert(row(0, "v0", null)).get();
1655+
// changed partial upsert: normal update changelog
1656+
partialWriter.upsert(row(0, "v1", null)).get();
1657+
// unchanged partial upsert: no changelog
1658+
partialWriter.upsert(row(0, "v1", null)).get();
1659+
// changed partial delete: clears name and emits an update changelog
1660+
partialWriter.delete(row(0, "v1", null)).get();
1661+
// unchanged partial delete: no changelog
1662+
partialWriter.delete(row(0, null, null)).get();
1663+
partialWriter.flush();
1664+
1665+
List<ScanRecord> expected = new ArrayList<>();
1666+
expected.add(new ScanRecord(-1, -1, ChangeType.INSERT, row(0, "v0", "kept")));
1667+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v0", "kept")));
1668+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, "v1", "kept")));
1669+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v1", "kept")));
1670+
expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, null, "kept")));
1671+
1672+
LogScanner logScanner = table.newScan().createLogScanner();
1673+
logScanner.subscribeFromBeginning(0);
1674+
List<ScanRecord> actualLogRecords = new ArrayList<>(expected.size());
1675+
while (actualLogRecords.size() < expected.size()) {
1676+
ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
1677+
scanRecords.forEach(actualLogRecords::add);
1678+
}
1679+
assertThat(logScanner.poll(Duration.ofSeconds(1))).isEmpty();
1680+
logScanner.close();
1681+
1682+
assertThat(actualLogRecords).hasSize(expected.size());
1683+
for (int i = 0; i < actualLogRecords.size(); i++) {
1684+
ScanRecord actual = actualLogRecords.get(i);
1685+
assertThat(actual.getChangeType()).isEqualTo(expected.get(i).getChangeType());
1686+
assertThatRow(actual.getRow())
1687+
.withSchema(rowType)
1688+
.isEqualTo(expected.get(i).getRow());
1689+
}
1690+
1691+
Lookuper lookuper = table.newLookup().createLookuper();
1692+
assertThatRow(lookuper.lookup(row(0)).get().getSingletonRow())
1693+
.withSchema(rowType)
1694+
.isEqualTo(row(0, null, "kept"));
1695+
}
1696+
}
1697+
15641698
@ParameterizedTest
15651699
@CsvSource({"none,3", "lz4_frame,3", "zstd,3", "zstd,9"})
15661700
void testArrowCompressionAndProject(String compression, String level) throws Exception {

‎fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2256,6 +2256,16 @@ public class ConfigOptions {
22562256
+ "The `versioned` merge engine will keep the row with the largest version of the same primary key. "
22572257
+ "The `aggregation` merge engine will aggregate rows with the same primary key using field-level aggregate functions.");
22582258

2259+
public static final ConfigOption<Boolean> TABLE_UPSERT_SKIP_UNCHANGED =
2260+
key("table.upsert.skip-unchanged")
2261+
.booleanType()
2262+
.defaultValue(false)
2263+
.withDescription(
2264+
"For a primary key table using the default merge behavior, skip the state write and changelog when the complete candidate row has the same schema ID and serialized row bytes as the stored row. "
2265+
+ "This also applies to partial updates and partial deletes. "
2266+
+ "It requires FULL changelog image mode and cannot be combined with row-level KV TTL or a merge engine. As with other tables without a merge engine, OVERWRITE writes are unsupported. "
2267+
+ "FULL changelog image mode already reads the previous value for every upsert. Comparing it with the complete candidate row adds comparison costs.");
2268+
22592269
public static final ConfigOption<String> TABLE_MERGE_ENGINE_VERSION_COLUMN =
22602270
// we may need to introduce "del-column" in the future to support delete operation
22612271
key("table.merge-engine.versioned.ver-column")

‎fluss-common/src/main/java/org/apache/fluss/config/TableConfig.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,11 @@ public Optional<MergeEngineType> getMergeEngineType() {
171171
return config.getOptional(ConfigOptions.TABLE_MERGE_ENGINE);
172172
}
173173

174+
/** Whether an unchanged upsert should skip the state write and changelog. */
175+
public boolean isUpsertSkipUnchanged() {
176+
return config.get(ConfigOptions.TABLE_UPSERT_SKIP_UNCHANGED);
177+
}
178+
174179
/**
175180
* Gets the optional {@link MergeEngineType#VERSIONED} merge engine version column of the table.
176181
*/

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.fluss.client.table.scanner.log.LogScanner;
2525
import org.apache.fluss.client.table.scanner.log.ScanRecords;
2626
import org.apache.fluss.config.Configuration;
27+
import org.apache.fluss.exception.InvalidConfigException;
2728
import org.apache.fluss.flink.sink.shuffle.DistributionMode;
2829
import org.apache.fluss.metadata.TableBucket;
2930
import org.apache.fluss.metadata.TablePath;
@@ -1342,6 +1343,73 @@ void testUnsupportedStmtOnVersionMergeEngine() {
13421343
tablePath);
13431344
}
13441345

1346+
@Test
1347+
@MultiVersionTest
1348+
void testSkipUnchangedUpsertsSupportsMutations() throws Exception {
1349+
String tableName = "skipUnchangedUpsertsTable";
1350+
tBatchEnv.executeSql(
1351+
String.format(
1352+
"create table %s ("
1353+
+ " a int not null,"
1354+
+ " b bigint null, "
1355+
+ " c string null, "
1356+
+ " primary key (a) not enforced"
1357+
+ ") with ('table.upsert.skip-unchanged' = 'true')",
1358+
tableName));
1359+
1360+
tBatchEnv
1361+
.executeSql(
1362+
String.format(
1363+
"INSERT INTO %s VALUES (1, 10, 'initial'), (2, 20, 'delete')",
1364+
tableName))
1365+
.await();
1366+
1367+
// Verify that the Flink sink accepts partial updates with unchanged writes skipped.
1368+
tBatchEnv
1369+
.executeSql(String.format("INSERT INTO %s (a, c) VALUES (1, 'partial')", tableName))
1370+
.await();
1371+
1372+
// Verify that row-level UPDATE and DELETE statements are accepted as well.
1373+
tBatchEnv.executeSql(String.format("UPDATE %s SET b = 11 WHERE a = 1", tableName)).await();
1374+
tBatchEnv.executeSql(String.format("DELETE FROM %s WHERE a = 2", tableName)).await();
1375+
1376+
CloseableIterator<Row> rowIter =
1377+
tBatchEnv.executeSql(String.format("SELECT * FROM %s", tableName)).collect();
1378+
assertResultsIgnoreOrder(rowIter, Collections.singletonList("+I[1, 11, partial]"), true);
1379+
}
1380+
1381+
@Test
1382+
@MultiVersionTest
1383+
void testSkipUnchangedUpsertsRejectsWalImage() {
1384+
assertThatThrownBy(
1385+
() ->
1386+
tBatchEnv.executeSql(
1387+
"create table skip_unchanged_wal ("
1388+
+ "a int not null, b string, primary key (a) not enforced"
1389+
+ ") with ('table.upsert.skip-unchanged' = 'true', "
1390+
+ "'table.changelog.image' = 'wal')"))
1391+
.rootCause()
1392+
.isInstanceOf(InvalidConfigException.class)
1393+
.hasMessageContaining("table.upsert.skip-unchanged")
1394+
.hasMessageContaining("table.changelog.image");
1395+
}
1396+
1397+
@Test
1398+
@MultiVersionTest
1399+
void testSkipUnchangedUpsertsRejectsKvTtl() {
1400+
assertThatThrownBy(
1401+
() ->
1402+
tBatchEnv.executeSql(
1403+
"create table skip_unchanged_ttl ("
1404+
+ "a int not null, b string, primary key (a) not enforced"
1405+
+ ") with ('table.upsert.skip-unchanged' = 'true', "
1406+
+ "'table.kv.ttl' = '1 h')"))
1407+
.rootCause()
1408+
.isInstanceOf(InvalidConfigException.class)
1409+
.hasMessageContaining("table.upsert.skip-unchanged")
1410+
.hasMessageContaining("table.kv.ttl");
1411+
}
1412+
13451413
@Test
13461414
void testVersionMergeEngineWithTypeBigint() throws Exception {
13471415
tEnv.executeSql(

‎fluss-server/src/main/java/org/apache/fluss/server/kv/BytewiseRowEqualizer.java‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.fluss.server.kv;
19+
20+
import org.apache.fluss.record.BinaryValue;
21+
import org.apache.fluss.row.BinaryRow;
22+
import org.apache.fluss.row.BinarySegmentUtils;
23+
24+
/** Compares the schema ID and serialized row bytes of two KV values. */
25+
final class BytewiseRowEqualizer {
26+
27+
private BytewiseRowEqualizer() {}
28+
29+
static boolean equals(BinaryValue left, BinaryValue right) {
30+
if (left == right) {
31+
return true;
32+
}
33+
if (left.schemaId != right.schemaId) {
34+
return false;
35+
}
36+
37+
BinaryRow leftRow = left.row;
38+
BinaryRow rightRow = right.row;
39+
int size = leftRow.getSizeInBytes();
40+
return size == rightRow.getSizeInBytes()
41+
&& BinarySegmentUtils.equals(
42+
leftRow.getSegments(),
43+
leftRow.getOffset(),
44+
rightRow.getSegments(),
45+
rightRow.getOffset(),
46+
size);
47+
}
48+
}

‎fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -218,7 +218,8 @@ private KvTablet(
218218
AutoIncrementManager autoIncrementManager,
219219
@Nullable RowTtlTimestampProvider rowTtlTimestampProvider,
220220
Clock clock,
221-
boolean rowTtlEnabled) {
221+
boolean rowTtlEnabled,
222+
boolean skipUnchangedUpsert) {
222223
this.physicalPath = physicalPath;
223224
this.tableBucket = tableBucket;
224225
this.historicalPartition =
@@ -256,7 +257,8 @@ private KvTablet(
256257
stateValueEncoder,
257258
valueDecoder,
258259
rowTtlTimestampProvider,
259-
clock);
260+
clock,
261+
skipUnchangedUpsert);
260262
this.rocksDBStatistics = rocksDBStatistics;
261263
this.autoIncrementManager = autoIncrementManager;
262264
this.flushCompleteListener = flushCompleteListener;
@@ -518,7 +520,8 @@ private static KvTablet create(
518520
autoIncrementManager,
519521
rowTtlTimestampProvider,
520522
clock,
521-
rowTtl.isPresent());
523+
rowTtl.isPresent(),
524+
tableConfig.isUpsertSkipUnchanged());
522525
}
523526

524527
public static KvTablet create(

0 commit comments

Comments
 (0)