fix unit test and refine document

platinumhamburg · platinumhamburg · commit c8bb2d420bdf · 2025-12-15T14:54:52.000+08:00
diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
@@ -1433,8 +1433,10 @@ public class ConfigOptions {
                                     + "This configuration is inspired by similar settings in database systems like MySQL's binlog_row_image and PostgreSQL's replica identity. "
                                     + "The supported modes are `FULL` (default) and `WAL`. "
                                     + "The `FULL` mode produces both UPDATE_BEFORE and UPDATE_AFTER records for update operations, capturing complete information about updates and allowing tracking of previous values. "
-                                    + "The `WAL` mode emits only UPDATE_AFTER records for both INSERT and UPDATE operations, without UPDATE_BEFORE records. This is similar to database WAL (Write-Ahead Log) behavior. "
-                                    + "DELETE records are emitted if allowed. This mode reduces storage and transmission costs but loses the ability to distinguish between inserts and updates, and to track previous values. "
+                                    + "The `WAL` mode does not produce UPDATE_BEFORE records. Only INSERT, UPDATE_AFTER (and DELETE if allowed) records are emitted. "
+                                    + "When WAL mode is enabled with default merge engine (no merge engine configured) and full row updates (not partial update), an optimization is applied to skip looking up old values, "
+                                    + "and in this case INSERT operations are converted to UPDATE_AFTER events. "
+                                    + "This mode reduces storage and transmission costs but loses the ability to track previous values. "
                                     + "This option only affects primary key tables.");
 
     // ------------------------------------------------------------------------
diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/ChangelogImage.java b/fluss-common/src/main/java/org/apache/fluss/metadata/ChangelogImage.java
@@ -35,11 +35,12 @@ public enum ChangelogImage {
     FULL,
 
     /**
-     * WAL mode produces only UPDATE_AFTER records for both INSERT and UPDATE operations, without
-     * UPDATE_BEFORE records. This mode is similar to database WAL (Write-Ahead Log) behavior, where
-     * all inserts and updates are represented as UPDATE_AFTER events. DELETE records are emitted if
-     * allowed. This mode reduces storage and transmission costs but loses the ability to
-     * distinguish between inserts and updates, and to track previous values.
+     * WAL mode does not produce UPDATE_BEFORE records. Only INSERT, UPDATE_AFTER (and DELETE if
+     * allowed) records are emitted. When WAL mode is enabled with default merge engine (no merge
+     * engine configured) and full row updates (not partial update), an optimization is applied to
+     * skip looking up old values, and in this case INSERT operations are converted to UPDATE_AFTER
+     * events, similar to database WAL (Write-Ahead Log) behavior. This mode reduces storage and
+     * transmission costs but loses the ability to track previous values.
      */
     WAL;
 
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
@@ -213,17 +213,20 @@ public ChangelogMode getChangelogMode() {
                     ChangelogImage changelogImage =
                             tableConf.get(ConfigOptions.TABLE_CHANGELOG_IMAGE);
                     if (changelogImage == ChangelogImage.WAL) {
-                        // When using WAL mode, only produce UPDATE_AFTER (and DELETE if allowed)
-                        // Note: INSERT operations are also converted to UPDATE_AFTER in WAL mode
+                        // When using WAL mode, produce INSERT and UPDATE_AFTER (and DELETE if
+                        // allowed), without UPDATE_BEFORE. Note: with default merge engine and full
+                        // row updates, an optimization converts INSERT to UPDATE_AFTER.
                         if (deleteBehavior == DeleteBehavior.ALLOW) {
                             // DELETE is still produced when delete behavior is allowed
                             return ChangelogMode.newBuilder()
+                                    .addContainedKind(RowKind.INSERT)
                                     .addContainedKind(RowKind.UPDATE_AFTER)
                                     .addContainedKind(RowKind.DELETE)
                                     .build();
                         } else {
                             // No DELETE when delete operations are ignored or disabled
                             return ChangelogMode.newBuilder()
+                                    .addContainedKind(RowKind.INSERT)
                                     .addContainedKind(RowKind.UPDATE_AFTER)
                                     .build();
                         }
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvTablet.java
@@ -444,10 +444,10 @@ private long processUpsert(
             PaddingRow latestSchemaRow,
             long logOffset)
             throws Exception {
-        // Optimization: when using NO_UPDATE_BEFORE mode and merger is
-        // DefaultRowMerger (full update, not partial update), we can skip
-        // fetching old value for better performance since it always returns
-        // new value.
+        // Optimization: when using WAL mode and merger is DefaultRowMerger (full update, not
+        // partial update), we can skip fetching old value for better performance since it
+        // always returns new value. In this case, both INSERT and UPDATE will produce
+        // UPDATE_AFTER.
         if (changelogImage == ChangelogImage.WAL && currentMerger instanceof DefaultRowMerger) {
             return applyUpdate(key, null, currentValue, walBuilder, latestSchemaRow, logOffset);
         }
@@ -487,10 +487,7 @@ private long applyInsert(
             PaddingRow latestSchemaRow,
             long logOffset)
             throws Exception {
-        // In WAL mode, INSERT operations are also emitted as UPDATE_AFTER
-        ChangeType changeType =
-                changelogImage == ChangelogImage.WAL ? ChangeType.UPDATE_AFTER : ChangeType.INSERT;
-        walBuilder.append(changeType, latestSchemaRow.replaceRow(currentValue.row));
+        walBuilder.append(ChangeType.INSERT, latestSchemaRow.replaceRow(currentValue.row));
         kvPreWriteBuffer.put(key, currentValue.encodeValue(), logOffset);
         return logOffset + 1;
     }
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/KvTabletTest.java
@@ -1045,8 +1045,9 @@ void testAppendDuplicatedKvBatch() throws Exception {
     }
 
     @Test
-    void testChangelogImageNoUpdateBefore() throws Exception {
-        // WAL mode - INSERT and UPDATE both produce UPDATE_AFTER, no UPDATE_BEFORE
+    void testWalModeChangelogImageNoUpdateBefore() throws Exception {
+        // WAL mode - no UPDATE_BEFORE. With default merge engine and full row update,
+        // optimization converts INSERT to UPDATE_AFTER
         Map<String, String> config = new HashMap<>();
         config.put("table.changelog.image", "WAL");
         initLogTabletAndKvTablet(DATA1_SCHEMA_PK, config);
@@ -1061,7 +1062,8 @@ void testChangelogImageNoUpdateBefore() throws Exception {
         kvTablet.putAsLeader(kvRecordBatch1, null);
         long endOffset = logTablet.localLogEndOffset();
 
-        // Verify inserts produce +U (WAL mode converts INSERT to UPDATE_AFTER)
+        // Verify inserts produce +U (optimization in WAL mode with default merge engine and full
+        // row update)
         LogRecords actualLogRecords = readLogRecords();
         MemoryLogRecords expectedLogs =
                 logRecords(
@@ -1110,8 +1112,9 @@ void testChangelogImageNoUpdateBefore() throws Exception {
     }
 
     @Test
-    void testChangelogImageNoUpdateBeforeWithPartialUpdate() throws Exception {
-        // WAL mode with partial update - still need to fetch oldValue
+    void testWalModeChangelogImageNoUpdateBeforeWithPartialUpdate() throws Exception {
+        // WAL mode with partial update - INSERT produces INSERT, UPDATE produces UPDATE_AFTER
+        // only (no optimization applied)
         Map<String, String> config = new HashMap<>();
         config.put("table.changelog.image", "WAL");
         initLogTabletAndKvTablet(DATA2_SCHEMA, config);
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
@@ -85,7 +85,7 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties)
 | table.merge-engine                      | Enum     | (None)                              | Defines the merge engine for the primary key table. By default, primary key table uses the [default merge engine(last_row)](table-design/table-types/pk-table/merge-engines/default.md). It also supports two merge engines are `first_row` and `versioned`. The [first_row merge engine](table-design/table-types/pk-table/merge-engines/first-row.md) will keep the first row of the same primary key. The [versioned merge engine](table-design/table-types/pk-table/merge-engines/versioned.md) will keep the row with the largest version of the same primary key.                                                                                                                                                                                                                                                                                                                                                                |
 | table.merge-engine.versioned.ver-column | String   | (None)                              | The column name of the version column for the `versioned` merge engine. If the merge engine is set to `versioned`, the version column must be set.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 | table.delete.behavior                   | Enum     | ALLOW                               | Controls the behavior of delete operations on primary key tables. Three modes are supported: `ALLOW` (default) - allows normal delete operations; `IGNORE` - silently ignores delete requests without errors; `DISABLE` - rejects delete requests and throws explicit errors. This configuration provides system-level guarantees for some downstream pipelines (e.g., Flink Delta Join) that must not receive any delete events in the changelog of the table. For tables with `first_row` or `versioned` merge engines, this option is automatically set to `IGNORE` and cannot be overridden. Only applicable to primary key tables.                                                                                                                                                                                                                                                                                                |
-| table.changelog.image                   | Enum     | FULL                                | Defines the changelog image mode for primary key tables. This configuration is inspired by similar settings in database systems like MySQL's `binlog_row_image` and PostgreSQL's `replica identity`. Two modes are supported: `FULL` (default) - produces both UPDATE_BEFORE and UPDATE_AFTER records for update operations, capturing complete information about updates and allowing tracking of previous values; `WAL` - emits only UPDATE_AFTER records for both INSERT and UPDATE operations, without UPDATE_BEFORE records. This is similar to database WAL (Write-Ahead Log) behavior. DELETE records are emitted if allowed. This mode reduces storage and transmission costs but loses the ability to distinguish between inserts and updates, and to track previous values. Only applicable to primary key tables.                                                                                                                    |
+| table.changelog.image                   | Enum     | FULL                                | Defines the changelog image mode for primary key tables. This configuration is inspired by similar settings in database systems like MySQL's `binlog_row_image` and PostgreSQL's `replica identity`. Two modes are supported: `FULL` (default) - produces both UPDATE_BEFORE and UPDATE_AFTER records for update operations, capturing complete information about updates and allowing tracking of previous values; `WAL` - does not produce UPDATE_BEFORE records. Only INSERT, UPDATE_AFTER (and DELETE if allowed) records are emitted. When WAL mode is enabled with default merge engine (no merge engine configured) and full row updates (not partial update), an optimization is applied to skip looking up old values, and in this case INSERT operations are converted to UPDATE_AFTER events. This mode reduces storage and transmission costs but loses the ability to track previous values. Only applicable to primary key tables.                                                                                                                    |
 
 
 ## Read Options
