[server] Support update_if_changed merge engine for primary-key tables

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

@@ -36,6 +36,7 @@
 import org.apache.fluss.config.MemorySize;
 import org.apache.fluss.fs.FsPath;
 import org.apache.fluss.fs.TestFileSystem;
+import org.apache.fluss.metadata.ChangelogImage;
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.KvFormat;
 import org.apache.fluss.metadata.LogFormat;
@@ -63,6 +64,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import javax.annotation.Nullable;
@@ -1561,6 +1563,149 @@ void testFirstRowMergeEngine(boolean doProjection) throws Exception {
         }
     }
 
+    @ParameterizedTest
+    @EnumSource(ChangelogImage.class)
+    void testUpdateIfChangedMergeEngine(ChangelogImage changelogImage) throws Exception {
+        TableDescriptor tableDescriptor =
+                TableDescriptor.builder()
+                        .schema(DATA1_SCHEMA_PK)
+                        .property(
+                                ConfigOptions.TABLE_MERGE_ENGINE, MergeEngineType.UPDATE_IF_CHANGED)
+                        .property(ConfigOptions.TABLE_CHANGELOG_IMAGE, changelogImage)
+                        .build();
+        RowType rowType = DATA1_SCHEMA_PK.getRowType();
+        TablePath tablePath =
+                TablePath.of(
+                        "test_db_1",
+                        "test_update_if_changed_merge_engine_" + changelogImage.name());
+        createTable(tablePath, tableDescriptor, false);
+
+        try (Table table = conn.getTable(tablePath)) {
+            UpsertWriter upsertWriter = table.newUpsert().createWriter();
+            // insert a row
+            upsertWriter.upsert(row(0, "v0"));
+            // value-identical upserts: should be no-ops that emit no changelog
+            upsertWriter.upsert(row(0, "v0"));
+            upsertWriter.upsert(row(0, "v0"));
+            // a field differs: should emit an update changelog
+            upsertWriter.upsert(row(0, "v1"));
+            // delete the row: should emit a delete changelog
+            upsertWriter.delete(row(0, "v1"));
+            upsertWriter.flush();
+
+            // No records should be emitted for the value-identical upserts.
+            List<ScanRecord> expected = new ArrayList<>();
+            expected.add(new ScanRecord(-1, -1, ChangeType.INSERT, row(0, "v0")));
+            if (changelogImage == ChangelogImage.FULL) {
+                expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v0")));
+            }
+            expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, "v1")));
+            expected.add(new ScanRecord(-1, -1, ChangeType.DELETE, row(0, "v1")));
+
+            LogScanner logScanner = table.newScan().createLogScanner();
+            logScanner.subscribeFromBeginning(0);
+            List<ScanRecord> actualLogRecords = new ArrayList<>(expected.size());
+            while (actualLogRecords.size() < expected.size()) {
+                ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
+                scanRecords.forEach(actualLogRecords::add);
+            }
+            assertThat(logScanner.poll(Duration.ofSeconds(1))).isEmpty();
+            logScanner.close();
+
+            assertThat(actualLogRecords).hasSize(expected.size());
+            for (int i = 0; i < actualLogRecords.size(); i++) {
+                ScanRecord actual = actualLogRecords.get(i);
+                assertThat(actual.getChangeType()).isEqualTo(expected.get(i).getChangeType());
+                assertThatRow(actual.getRow())
+                        .withSchema(rowType)
+                        .isEqualTo(expected.get(i).getRow());
+            }
+        }
+    }
+
+    @ParameterizedTest
+    @EnumSource(ChangelogImage.class)
+    void testUpdateIfChangedMergeEngineWithPartialUpdate(ChangelogImage changelogImage)
+            throws Exception {
+        Schema schema =
+                Schema.newBuilder()
+                        .column("id", DataTypes.INT())
+                        .column("name", DataTypes.STRING())
+                        .column("data", DataTypes.STRING())
+                        .primaryKey("id")
+                        .build();
+        TableDescriptor tableDescriptor =
+                TableDescriptor.builder()
+                        .schema(schema)
+                        .distributedBy(1, "id")
+                        .property(
+                                ConfigOptions.TABLE_MERGE_ENGINE, MergeEngineType.UPDATE_IF_CHANGED)
+                        .property(ConfigOptions.TABLE_CHANGELOG_IMAGE, changelogImage)
+                        .build();
+        RowType rowType = schema.getRowType();
+        TablePath tablePath =
+                TablePath.of(
+                        "test_db_1",
+                        "test_update_if_changed_partial_update_" + changelogImage.name());
+        createTable(tablePath, tableDescriptor, false);
+
+        try (Table table = conn.getTable(tablePath)) {
+            UpsertWriter fullWriter = table.newUpsert().createWriter();
+            fullWriter.upsert(row(0, "v0", "kept")).get();
+
+            UpsertWriter partialWriter =
+                    table.newUpsert().partialUpdate("id", "name").createWriter();
+            // unchanged partial upsert: no changelog
+            partialWriter.upsert(row(0, "v0", null)).get();
+            // changed partial upsert: normal update changelog
+            partialWriter.upsert(row(0, "v1", null)).get();
+            // unchanged partial upsert: no changelog
+            partialWriter.upsert(row(0, "v1", null)).get();
+            // changed partial delete: clears name and emits an update changelog
+            partialWriter.delete(row(0, "v1", null)).get();
+            // unchanged partial delete: no changelog
+            partialWriter.delete(row(0, null, null)).get();
+            partialWriter.flush();
+
+            List<ScanRecord> expected = new ArrayList<>();
+            expected.add(new ScanRecord(-1, -1, ChangeType.INSERT, row(0, "v0", "kept")));
+            if (changelogImage == ChangelogImage.FULL) {
+                expected.add(
+                        new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v0", "kept")));
+            }
+            expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, "v1", "kept")));
+            if (changelogImage == ChangelogImage.FULL) {
+                expected.add(
+                        new ScanRecord(-1, -1, ChangeType.UPDATE_BEFORE, row(0, "v1", "kept")));
+            }
+            expected.add(new ScanRecord(-1, -1, ChangeType.UPDATE_AFTER, row(0, null, "kept")));
+
+            LogScanner logScanner = table.newScan().createLogScanner();
+            logScanner.subscribeFromBeginning(0);
+            List<ScanRecord> actualLogRecords = new ArrayList<>(expected.size());
+            while (actualLogRecords.size() < expected.size()) {
+                ScanRecords scanRecords = logScanner.poll(Duration.ofSeconds(1));
+                scanRecords.forEach(actualLogRecords::add);
+            }
+            assertThat(logScanner.poll(Duration.ofSeconds(1))).isEmpty();
+            logScanner.close();
+
+            assertThat(actualLogRecords).hasSize(expected.size());
+            for (int i = 0; i < actualLogRecords.size(); i++) {
+                ScanRecord actual = actualLogRecords.get(i);
+                assertThat(actual.getChangeType()).isEqualTo(expected.get(i).getChangeType());
+                assertThatRow(actual.getRow())
+                        .withSchema(rowType)
+                        .isEqualTo(expected.get(i).getRow());
+            }
+
+            Lookuper lookuper = table.newLookup().createLookuper();
+            assertThatRow(lookuper.lookup(row(0)).get().getSingletonRow())
+                    .withSchema(rowType)
+                    .isEqualTo(row(0, null, "kept"));
+        }
+    }
+
     @ParameterizedTest
     @CsvSource({"none,3", "lz4_frame,3", "zstd,3", "zstd,9"})
     void testArrowCompressionAndProject(String compression, String level) throws Exception {

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

@@ -2010,10 +2010,12 @@ public class ConfigOptions {
                     .noDefaultValue()
                     .withDescription(
                             "Defines the merge engine for the primary key table. By default, primary key table doesn't have merge engine. "
-                                    + "The supported merge engines are `first_row`, `versioned`, and `aggregation`. "
+                                    + "The supported merge engines are `first_row`, `versioned`, `aggregation`, and `update_if_changed`. "
                                     + "The `first_row` merge engine will keep the first row of the same primary key. "
                                     + "The `versioned` merge engine will keep the row with the largest version of the same primary key. "
-                                    + "The `aggregation` merge engine will aggregate rows with the same primary key using field-level aggregate functions.");
+                                    + "The `aggregation` merge engine will aggregate rows with the same primary key using field-level aggregate functions. "
+                                    + "The `update_if_changed` merge engine keeps last-row upsert semantics but suppresses value-identical writes: "
+                                    + "when an incoming row is logically equal to the stored row, the write is a no-op and no changelog is emitted.");
 
     public static final ConfigOption<String> TABLE_MERGE_ENGINE_VERSION_COLUMN =
             // we may need to introduce "del-column" in the future to support delete operation

fluss-common/src/main/java/org/apache/fluss/metadata/MergeEngineType.java

@@ -22,10 +22,10 @@
  *
  * <p>A primary key table with a merge engine is a special kind of table, called "merge table".
  * Fluss provides 3 kinds of table: "primary key table", "log table", and "merge table". Merge table
- * is a primary key table that has a primary key definition but doesn't directly UPDATE and DELETE
- * rows in the table, and instead, it merges the append rows into a new data set according to the
- * defined {@link MergeEngineType}. Therefore, it doesn't support direct UPDATE (also
- * partial-update) and DELETE operations and only supports INSERT or APPEND operations.
+ * is a primary key table that merges incoming rows into a new data set according to the defined
+ * {@link MergeEngineType}. Most merge engines accept only INSERT or APPEND operations, rather than
+ * direct UPDATE (including partial update) and DELETE operations. {@link #UPDATE_IF_CHANGED} is an
+ * exception that retains normal primary-key table update and delete semantics.
  *
  * <p>Note: A primary key table doesn't have a merge engine by default.
  *
@@ -61,7 +61,20 @@ public enum MergeEngineType {
      *
      * @since 0.9
      */
-    AGGREGATION;
+    AGGREGATION,
+
+    /**
+     * A merge engine that keeps last-row upsert semantics but suppresses value-identical writes.
+     * When an incoming row is logically equal to the currently stored row (compared by logical
+     * field values, not raw bytes), the write is a no-op and no changelog is emitted. When at least
+     * one field differs, the incoming row replaces the stored row and a normal update changelog is
+     * emitted. Unlike {@link #FIRST_ROW}, legitimate updates are still applied; unlike {@link
+     * #VERSIONED}, no version column is required. Partial updates and delete operations are
+     * supported.
+     *
+     * @since 1.1
+     */
+    UPDATE_IF_CHANGED;
 
     /** Creates a {@link MergeEngineType} from the given string. */
     public static MergeEngineType fromString(String type) {
@@ -72,6 +85,8 @@ public static MergeEngineType fromString(String type) {
                 return VERSIONED;
             case "AGGREGATION":
                 return AGGREGATION;
+            case "UPDATE_IF_CHANGED":
+                return UPDATE_IF_CHANGED;
             default:
                 throw new IllegalArgumentException("Unsupported merge engine type: " + type);
         }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkTableSink.java

@@ -156,7 +156,9 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
                             "Fluss table sink does not support partial updates for table without primary key. Please make sure the "
                                     + "number of specified columns in INSERT INTO matches columns of the Fluss table.");
                 }
-                if (mergeEngineType != null && mergeEngineType != MergeEngineType.AGGREGATION) {
+                if (mergeEngineType != null
+                        && mergeEngineType != MergeEngineType.AGGREGATION
+                        && mergeEngineType != MergeEngineType.UPDATE_IF_CHANGED) {
                     throw new ValidationException(
                             String.format(
                                     "Table %s uses the '%s' merge engine which does not support partial updates. Please make sure the "
@@ -376,7 +378,7 @@ private void validateUpdatable() {
                             "Table %s is a Log Table. Log Table doesn't support DELETE and UPDATE statements.",
                             tablePath));
         }
-        if (mergeEngineType != null) {
+        if (mergeEngineType != null && mergeEngineType != MergeEngineType.UPDATE_IF_CHANGED) {
             throw new UnsupportedOperationException(
                     String.format(
                             "Table %s uses the '%s' merge engine which does not support DELETE or UPDATE statements.",

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

@@ -1319,6 +1319,40 @@ void testUnsupportedStmtOnVersionMergeEngine() {
                         tablePath);
     }
 
+    @Test
+    void testUpdateIfChangedMergeEngineSupportsMutations() throws Exception {
+        String tableName = "updateIfChangedMergeEngineTable";
+        tBatchEnv.executeSql(
+                String.format(
+                        "create table %s ("
+                                + " a int not null,"
+                                + " b bigint null, "
+                                + " c string null, "
+                                + " primary key (a) not enforced"
+                                + ") with ('table.merge-engine' = 'update_if_changed')",
+                        tableName));
+
+        tBatchEnv
+                .executeSql(
+                        String.format(
+                                "INSERT INTO %s VALUES (1, 10, 'initial'), (2, 20, 'delete')",
+                                tableName))
+                .await();
+
+        // Verify that the Flink sink accepts partial updates for this merge engine.
+        tBatchEnv
+                .executeSql(String.format("INSERT INTO %s (a, c) VALUES (1, 'partial')", tableName))
+                .await();
+
+        // Verify that row-level UPDATE and DELETE statements are accepted as well.
+        tBatchEnv.executeSql(String.format("UPDATE %s SET b = 11 WHERE a = 1", tableName)).await();
+        tBatchEnv.executeSql(String.format("DELETE FROM %s WHERE a = 2", tableName)).await();
+
+        CloseableIterator<Row> rowIter =
+                tBatchEnv.executeSql(String.format("SELECT * FROM %s", tableName)).collect();
+        assertResultsIgnoreOrder(rowIter, Collections.singletonList("+I[1, 11, partial]"), true);
+    }
+
     @Test
     void testVersionMergeEngineWithTypeBigint() throws Exception {
         tEnv.executeSql(

fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java

@@ -374,6 +374,11 @@ private long processDeletion(
 
         BinaryValue newValue = currentMerger.delete(oldValue);
 
+        if (newValue == oldValue) {
+            // no actual change, skip this record
+            return logOffset;
+        }
+
         // if newValue is null, it means the row should be deleted
         if (newValue == null) {
             return applyDelete(

fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/RowMerger.java

@@ -50,7 +50,8 @@ public interface RowMerger {
      * DeleteBehavior#ALLOW}.
      *
      * @param oldRow the old row.
-     * @return the merged row, or null if the row is deleted.
+     * @return the merged row, or null if the row is deleted. Returning the same instance as {@code
+     *     oldRow} means that nothing happens to the row.
      */
     @Nullable
     BinaryValue delete(BinaryValue oldRow);
@@ -100,6 +101,8 @@ static RowMerger create(TableConfig tableConf, KvFormat kvFormat, SchemaGetter s
                     return new VersionedRowMerger(versionColumn.get(), deleteBehavior);
                 case AGGREGATION:
                     return new AggregateRowMerger(tableConf, kvFormat, schemaGetter);
+                case UPDATE_IF_CHANGED:
+                    return new UpdateIfChangedRowMerger(kvFormat, schemaGetter, deleteBehavior);
                 default:
                     throw new IllegalArgumentException(
                             "Unsupported merge engine type: " + mergeEngineType.get());