refactor

Author: platinumhamburg

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,15 +1424,17 @@ 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<Boolean> TABLE_CHANGELOG_IGNORE_UPDATE_BEFORE =
-            key("table.changelog.ignore-update-before")
-                    .booleanType()
-                    .defaultValue(false)
-                    .withDescription(
-                            "Whether to ignore UPDATE_BEFORE records in changelog for the primary key table. "
-                                    + "When disabled (default), update operations produce both UPDATE_BEFORE and UPDATE_AFTER records. "
-                                    + "When enabled, update operations only produce UPDATE_AFTER records, "
-                                    + "which reduces storage and transmission costs but loses the ability to track previous values. "
+    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 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. "
                                     + "This option only affects primary key tables.");
 
     // ------------------------------------------------------------------------

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;
@@ -118,12 +119,11 @@ public Optional<DeleteBehavior> getDeleteBehavior() {
     }
 
     /**
-     * Whether to ignore UPDATE_BEFORE records in changelog for the primary key table. When false
-     * (default), update operations produce both UPDATE_BEFORE and UPDATE_AFTER records. When true,
-     * update operations only produce UPDATE_AFTER records.
+     * Gets the changelog image mode of the table. The changelog image mode defines what information
+     * is included in the changelog for update operations.
      */
-    public boolean isChangelogIgnoreUpdateBefore() {
-        return config.get(ConfigOptions.TABLE_CHANGELOG_IGNORE_UPDATE_BEFORE);
+    public ChangelogImage getChangelogImage() {
+        return config.get(ConfigOptions.TABLE_CHANGELOG_IMAGE);
     }
 
     /** Gets the Arrow compression type and compression level of the table. */

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

@@ -0,0 +1,58 @@
+/*
+ * 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 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;
+
+    /** 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;
@@ -209,28 +210,26 @@ public ChangelogMode getChangelogMode() {
                     Configuration tableConf = Configuration.fromMap(tableOptions);
                     DeleteBehavior deleteBehavior =
                             tableConf.get(ConfigOptions.TABLE_DELETE_BEHAVIOR);
-                    boolean ignoreUpdateBefore =
-                            tableConf.get(ConfigOptions.TABLE_CHANGELOG_IGNORE_UPDATE_BEFORE);
-                    if (ignoreUpdateBefore) {
-                        // When ignoring UPDATE_BEFORE, only produce INSERT, UPDATE_AFTER (and
-                        // DELETE if allowed)
+                    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
                         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();
                         }
                     }
 
-                    // Not ignoring UPDATE_BEFORE, produce full changelog
+                    // 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

@@ -186,7 +186,7 @@ public KvTablet getOrCreateKv(
                                     merger,
                                     arrowCompressionInfo,
                                     schemaGetter,
-                                    tableConfig.isChangelogIgnoreUpdateBefore());
+                                    tableConfig.getChangelogImage());
                     currentKvs.put(tableBucket, tablet);
 
                     LOG.info(
@@ -294,7 +294,7 @@ public KvTablet loadKv(File tabletDir, SchemaGetter schemaGetter) throws Excepti
                         rowMerger,
                         tableConfig.getArrowCompressionInfo(),
                         schemaGetter,
-                        tableConfig.isChangelogIgnoreUpdateBefore());
+                        tableConfig.getChangelogImage());
         if (this.currentKvs.containsKey(tableBucket)) {
             throw new IllegalStateException(
                     String.format(