[KV] Support configurable change log image mode for Primary Key Tables

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

@@ -20,6 +20,7 @@
 import org.apache.fluss.annotation.Internal;
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.compression.ArrowCompressionType;
+import org.apache.fluss.metadata.ChangelogImage;
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.DeleteBehavior;
 import org.apache.fluss.metadata.KvFormat;
@@ -1423,6 +1424,21 @@ public class ConfigOptions {
                                     + "The auto increment column can only be used in primary-key table. The data type of the auto increment column must be INT or BIGINT."
                                     + "Currently a table can have only one auto-increment column.");
 
+    public static final ConfigOption<ChangelogImage> TABLE_CHANGELOG_IMAGE =
+            key("table.changelog.image")
+                    .enumType(ChangelogImage.class)
+                    .defaultValue(ChangelogImage.FULL)
+                    .withDescription(
+                            "Defines the changelog image mode for the primary key table. "
+                                    + "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 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.");
+
     // ------------------------------------------------------------------------
     //  ConfigOptions for Kv
     // ------------------------------------------------------------------------

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

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.compression.ArrowCompressionInfo;
+import org.apache.fluss.metadata.ChangelogImage;
 import org.apache.fluss.metadata.DataLakeFormat;
 import org.apache.fluss.metadata.DeleteBehavior;
 import org.apache.fluss.metadata.KvFormat;
@@ -117,6 +118,14 @@ public Optional<DeleteBehavior> getDeleteBehavior() {
         return config.getOptional(ConfigOptions.TABLE_DELETE_BEHAVIOR);
     }
 
+    /**
+     * Gets the changelog image mode of the table. The changelog image mode defines what information
+     * is included in the changelog for update operations.
+     */
+    public ChangelogImage getChangelogImage() {
+        return config.get(ConfigOptions.TABLE_CHANGELOG_IMAGE);
+    }
+
     /** Gets the Arrow compression type and compression level of the table. */
     public ArrowCompressionInfo getArrowCompressionInfo() {
         return ArrowCompressionInfo.fromConf(config);

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

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.metadata;
+
+/**
+ * The changelog image mode for the primary key table.
+ *
+ * <p>This enum defines what information is included in the changelog for update operations. It is
+ * inspired by similar configurations in database systems like MySQL's binlog_row_image and
+ * PostgreSQL's replica identity.
+ *
+ * @since 0.9
+ */
+public enum ChangelogImage {
+
+    /**
+     * Full changelog with both UPDATE_BEFORE and UPDATE_AFTER records. This is the default behavior
+     * that captures complete information about updates, allowing tracking of previous values.
+     */
+    FULL,
+
+    /**
+     * 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;
+
+    /** Creates a {@link ChangelogImage} from the given string. */
+    public static ChangelogImage fromString(String image) {
+        switch (image.toUpperCase().replace("-", "_")) {
+            case "FULL":
+                return FULL;
+            case "WAL":
+                return WAL;
+
+            default:
+                throw new IllegalArgumentException("Unsupported changelog image: " + image);
+        }
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java

@@ -31,6 +31,7 @@
 import org.apache.fluss.flink.utils.PushdownUtils.FieldEqual;
 import org.apache.fluss.lake.source.LakeSource;
 import org.apache.fluss.lake.source.LakeSplit;
+import org.apache.fluss.metadata.ChangelogImage;
 import org.apache.fluss.metadata.DeleteBehavior;
 import org.apache.fluss.metadata.MergeEngineType;
 import org.apache.fluss.metadata.TablePath;
@@ -206,10 +207,32 @@ public ChangelogMode getChangelogMode() {
                 if (mergeEngineType == MergeEngineType.FIRST_ROW) {
                     return ChangelogMode.insertOnly();
                 } else {
-                    // Check delete behavior configuration
                     Configuration tableConf = Configuration.fromMap(tableOptions);
                     DeleteBehavior deleteBehavior =
                             tableConf.get(ConfigOptions.TABLE_DELETE_BEHAVIOR);
+                    ChangelogImage changelogImage =
+                            tableConf.get(ConfigOptions.TABLE_CHANGELOG_IMAGE);
+                    if (changelogImage == ChangelogImage.WAL) {
+                        // 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();
+                        }
+                    }
+
+                    // Using FULL mode, produce full changelog
                     if (deleteBehavior == DeleteBehavior.ALLOW) {
                         return ChangelogMode.all();
                     } else {

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

@@ -185,7 +185,8 @@ public KvTablet getOrCreateKv(
                                     kvFormat,
                                     merger,
                                     arrowCompressionInfo,
-                                    schemaGetter);
+                                    schemaGetter,
+                                    tableConfig.getChangelogImage());
                     currentKvs.put(tableBucket, tablet);
 
                     LOG.info(
@@ -277,9 +278,8 @@ public KvTablet loadKv(File tabletDir, SchemaGetter schemaGetter) throws Excepti
         TablePath tablePath = physicalTablePath.getTablePath();
         TableInfo tableInfo = getTableInfo(zkClient, tablePath);
 
-        RowMerger rowMerger =
-                RowMerger.create(
-                        tableInfo.getTableConfig(), tableInfo.getTableConfig().getKvFormat());
+        TableConfig tableConfig = tableInfo.getTableConfig();
+        RowMerger rowMerger = RowMerger.create(tableConfig, tableConfig.getKvFormat());
         KvTablet kvTablet =
                 KvTablet.create(
                         physicalTablePath,
@@ -290,10 +290,11 @@ public KvTablet loadKv(File tabletDir, SchemaGetter schemaGetter) throws Excepti
                         serverMetricGroup,
                         arrowBufferAllocator,
                         memorySegmentPool,
-                        tableInfo.getTableConfig().getKvFormat(),
+                        tableConfig.getKvFormat(),
                         rowMerger,
-                        tableInfo.getTableConfig().getArrowCompressionInfo(),
-                        schemaGetter);
+                        tableConfig.getArrowCompressionInfo(),
+                        schemaGetter,
+                        tableConfig.getChangelogImage());
         if (this.currentKvs.containsKey(tableBucket)) {
             throw new IllegalStateException(
                     String.format(