branch-4.1: [feature](paimon) Support row-level UPDATE, DELETE, and MERGE

be/src/exec/sink/writer/paimon/jni_paimon_write_backend.cpp

@@ -79,6 +79,16 @@ static constexpr const char* PAIMON_JNI_WRITER_CLASS = "org/apache/doris/paimon/
 static constexpr const char* SCANNER_LOADER_CLASS =
         "org/apache/doris/common/classloader/ScannerLoader";
 
+const char* const PAIMON_JNI_WRITER_OPEN_SIGNATURE =
+        "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZZLjava/lang/"
+        "String;Ljava/lang/String;JJ)V";
+
+PaimonJniWriterOpenMode PaimonJniWriterOpenMode::from_write_mode(
+        TPaimonWriteMode::type write_mode) {
+    return {static_cast<jboolean>(write_mode == TPaimonWriteMode::OVERWRITE),
+            static_cast<jboolean>(write_mode == TPaimonWriteMode::CHANGELOG)};
+}
+
 /// Attach the current native thread to the JVM if not already attached,
 /// and return a valid JNIEnv pointer.
 static Status _get_jni_env(JNIEnv** env) {
@@ -252,10 +262,7 @@ Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s
     RETURN_IF_ERROR(PaimonJniMemoryManager::register_natives(env, _jni_writer_cls));
 
     // Step 2: Cache JNI method IDs for write, prepareCommit, abort, close.
-    jmethodID open_id = env->GetMethodID(
-            _jni_writer_cls, "open",
-            "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZLjava/lang/"
-            "String;Ljava/lang/String;JJ)V");
+    jmethodID open_id = env->GetMethodID(_jni_writer_cls, "open", PAIMON_JNI_WRITER_OPEN_SIGNATURE);
     _write_id = env->GetMethodID(_jni_writer_cls, "write", "(Ljava/nio/ByteBuffer;)V");
     _prepare_commit_id = env->GetMethodID(_jni_writer_cls, "prepareCommit", "()[[B");
     _abort_id = env->GetMethodID(_jni_writer_cls, "abort", "()V");
@@ -293,10 +300,10 @@ Status JniPaimonWriteBackend::open(const TPaimonTableSink& sink, RuntimeState* s
         env->DeleteLocalRef(str);
     }
 
+    PaimonJniWriterOpenMode open_mode = PaimonJniWriterOpenMode::from_write_mode(sink.write_mode);
     env->CallVoidMethod(_jni_writer_obj, open_id, j_serialized_table, j_hadoop_config, j_cols,
-                        static_cast<jlong>(sink.transaction_id), j_commit_user,
-                        static_cast<jboolean>(sink.write_mode == TPaimonWriteMode::OVERWRITE),
-                        j_time_zone, j_spill_directories,
+                        static_cast<jlong>(sink.transaction_id), j_commit_user, open_mode.overwrite,
+                        open_mode.changelog, j_time_zone, j_spill_directories,
                         static_cast<jlong>(_memory_manager->memory_limit()),
                         reinterpret_cast<jlong>(_memory_manager.get()));
     Status st = _check_jni_exception(env, "open");

be/src/exec/sink/writer/paimon/jni_paimon_write_backend.h

@@ -33,6 +33,15 @@ namespace doris {
 
 class RuntimeState;
 
+extern const char* const PAIMON_JNI_WRITER_OPEN_SIGNATURE;
+
+struct PaimonJniWriterOpenMode {
+    jboolean overwrite;
+    jboolean changelog;
+
+    static PaimonJniWriterOpenMode from_write_mode(TPaimonWriteMode::type write_mode);
+};
+
 /// JNI backend that owns the Java PaimonJniWriter object and its JNI method
 /// handles. Creates lightweight JniPaimonWriter adapters that share this
 /// backend's JVM connection.

be/test/exec/sink/writer/paimon/paimon_write_backend_test.cpp

@@ -19,6 +19,8 @@
 
 #include <gtest/gtest.h>
 
+#include "exec/sink/writer/paimon/jni_paimon_write_backend.h"
+
 namespace doris {
 
 TEST(PaimonWriteBackendFactoryTest, SelectBackendType) {
@@ -29,4 +31,23 @@ TEST(PaimonWriteBackendFactoryTest, SelectBackendType) {
     EXPECT_EQ(PaimonBackendType::FFI, PaimonWriteBackendFactory::select_backend_type(sink));
 }
 
+TEST(JniPaimonWriteBackendTest, OpenAbiAndWriteModes) {
+    EXPECT_STREQ(
+            "(Ljava/lang/String;Ljava/util/Map;[Ljava/lang/String;JLjava/lang/String;ZZLjava/lang/"
+            "String;Ljava/lang/String;JJ)V",
+            PAIMON_JNI_WRITER_OPEN_SIGNATURE);
+
+    auto append = PaimonJniWriterOpenMode::from_write_mode(TPaimonWriteMode::APPEND);
+    EXPECT_FALSE(append.overwrite);
+    EXPECT_FALSE(append.changelog);
+
+    auto overwrite = PaimonJniWriterOpenMode::from_write_mode(TPaimonWriteMode::OVERWRITE);
+    EXPECT_TRUE(overwrite.overwrite);
+    EXPECT_FALSE(overwrite.changelog);
+
+    auto changelog = PaimonJniWriterOpenMode::from_write_mode(TPaimonWriteMode::CHANGELOG);
+    EXPECT_FALSE(changelog.overwrite);
+    EXPECT_TRUE(changelog.changelog);
+}
+
 } // namespace doris

fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonJniWriter.java

@@ -139,14 +139,15 @@ public PaimonJniWriter() {
      * @param transactionId  Doris external transaction identifier
      * @param commitUser     Paimon commit user shared with the FE committer
      * @param overwrite      whether this is an overwrite write
+     * @param changelogWrite whether the first input column contains a row change operation
      * @param timeZone       normalized Doris session timezone used for Paimon LTZ values
      * @param spillDirectories Doris storage-root scoped directories for Paimon write-buffer spill
      * @param memoryPoolLimitBytes maximum Doris-managed Paimon write-buffer memory
      * @param nativeMemoryManager opaque BE manager used to allocate tracked native pages
      */
     public void open(String serializedTable, Map<String, String> hadoopConfig,
                      String[] columnNames, long transactionId, String commitUser,
-                     boolean overwrite, String timeZone, String spillDirectories,
+                     boolean overwrite, boolean changelogWrite, String timeZone, String spillDirectories,
                      long memoryPoolLimitBytes, long nativeMemoryManager) throws Exception {
         try (ThreadClassLoaderContext ignored = new ThreadClassLoaderContext(classLoader)) {
             if (memoryPoolLimitBytes <= 0) {
@@ -170,8 +171,10 @@ public void open(String serializedTable, Map<String, String> hadoopConfig,
                     this.bucketMode = table.bucketMode();
 
                     CoreOptions coreOptions = CoreOptions.fromMap(table.options());
-                    this.writeSchema = PaimonWriteSchema.create(table.rowType(), columnNames);
-                    validateWriteColumnsForMergeEngine(columnNames.length, coreOptions);
+                    this.writeSchema = PaimonWriteSchema.create(
+                            table.rowType(), columnNames, changelogWrite);
+                    validateWriteColumnsForMergeEngine(
+                            columnNames.length - (changelogWrite ? 1 : 0), coreOptions);
                     this.fullCompactionChangelog =
                             !coreOptions.writeOnly()
                                     && coreOptions.changelogProducer()

fe/be-java-extensions/paimon-connector/src/main/java/org/apache/doris/paimon/PaimonWriteSchema.java

@@ -20,7 +20,9 @@
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TinyIntType;
 import org.apache.paimon.utils.DefaultValueUtils;
 
 import java.util.Arrays;
@@ -33,6 +35,11 @@
  * table-schema layout expected by the Paimon writer.
  */
 final class PaimonWriteSchema {
+    static final String ROW_KIND_COLUMN = "__DORIS_PAIMON_ROW_KIND__";
+    static final byte INSERT_OPERATION = 0;
+    static final byte UPDATE_OPERATION = 1;
+    static final byte DELETE_OPERATION = 2;
+
     private final DataType[] targetTypes;
     /** Maps Doris input-column position → Paimon table-schema position. */
     private final int[] tableFieldIndexes;
@@ -60,6 +67,11 @@ private PaimonWriteSchema(DataType[] targetTypes, int[] tableFieldIndexes,
      * @throws IllegalArgumentException if any column name is not found in the table schema
      */
     static PaimonWriteSchema create(RowType tableType, String[] columnNames) {
+        return create(tableType, columnNames, false);
+    }
+
+    static PaimonWriteSchema create(
+            RowType tableType, String[] columnNames, boolean changelogWrite) {
         if (columnNames == null || columnNames.length == 0) {
             throw new IllegalArgumentException(
                     "PaimonJniWriter requires explicit column names");
@@ -69,6 +81,15 @@ static PaimonWriteSchema create(RowType tableType, String[] columnNames) {
         int[] tableFieldIndexes = new int[columnNames.length];
         boolean[] specifiedFields = new boolean[tableType.getFieldCount()];
         for (int i = 0; i < columnNames.length; i++) {
+            if (changelogWrite && i == 0) {
+                if (!ROW_KIND_COLUMN.equals(columnNames[i])) {
+                    throw new IllegalArgumentException(
+                            "Paimon changelog write requires row kind as the first column");
+                }
+                targetTypes[i] = new TinyIntType(false);
+                tableFieldIndexes[i] = -1;
+                continue;
+            }
             int tableIndex = tableType.getFieldIndex(columnNames[i]);
             if (tableIndex < 0) {
                 throw new IllegalArgumentException(
@@ -122,10 +143,31 @@ GenericRow tableRow(Object[] columnValues) {
             row.setField(omittedDefaultFieldIndexes[i], omittedDefaultValues[i]);
         }
         for (int i = 0; i < tableFieldIndexes.length; i++) {
+            if (tableFieldIndexes[i] < 0) {
+                row.setRowKind(toRowKind(columnValues[i]));
+                continue;
+            }
             // Actual Doris input is applied last so an explicit NULL remains distinct
             // from an omitted field and retains Paimon's writer-side semantics.
             row.setField(tableFieldIndexes[i], columnValues[i]);
         }
         return row;
     }
+
+    private static RowKind toRowKind(Object operation) {
+        if (!(operation instanceof Byte)) {
+            throw new IllegalArgumentException("Paimon row change operation must be a TINYINT");
+        }
+        switch ((Byte) operation) {
+            case INSERT_OPERATION:
+                return RowKind.INSERT;
+            case UPDATE_OPERATION:
+                return RowKind.UPDATE_AFTER;
+            case DELETE_OPERATION:
+                return RowKind.DELETE;
+            default:
+                throw new IllegalArgumentException(
+                        "Unknown Paimon row change operation: " + operation);
+        }
+    }
 }

fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonJniWriterTest.java

@@ -45,7 +45,7 @@ public void testOpenFailureRestoresContextClassLoader() throws Exception {
         try {
             Assertions.assertThrows(Exception.class, () -> writer.open(
                     "not-a-serialized-table", Collections.emptyMap(), new String[0],
-                    1L, "test-user", false, "UTC", System.getProperty("java.io.tmpdir"),
+                    1L, "test-user", false, false, "UTC", System.getProperty("java.io.tmpdir"),
                     64L * 1024 * 1024, 1L));
             Assertions.assertSame(testClassLoader, thread.getContextClassLoader());
         } finally {

fe/be-java-extensions/paimon-connector/src/test/java/org/apache/doris/paimon/PaimonWriteSchemaTest.java

@@ -25,6 +25,7 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.DoubleType;
 import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowKind;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.types.VarCharType;
 import org.junit.jupiter.api.Assertions;
@@ -36,6 +37,32 @@
 
 public class PaimonWriteSchemaTest {
 
+    @Test
+    public void testChangelogOperationsSetPaimonRowKind() {
+        PaimonWriteSchema schema = PaimonWriteSchema.create(
+                tableType(),
+                new String[] {PaimonWriteSchema.ROW_KIND_COLUMN, "id", "name", "score", "region"},
+                true);
+
+        Assertions.assertEquals(RowKind.INSERT,
+                changelogRow(schema, PaimonWriteSchema.INSERT_OPERATION).getRowKind());
+        Assertions.assertEquals(RowKind.UPDATE_AFTER,
+                changelogRow(schema, PaimonWriteSchema.UPDATE_OPERATION).getRowKind());
+        Assertions.assertEquals(RowKind.DELETE,
+                changelogRow(schema, PaimonWriteSchema.DELETE_OPERATION).getRowKind());
+    }
+
+    @Test
+    public void testUnknownChangelogOperationRejected() {
+        PaimonWriteSchema schema = PaimonWriteSchema.create(
+                tableType(),
+                new String[] {PaimonWriteSchema.ROW_KIND_COLUMN, "id"},
+                true);
+
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> schema.tableRow(new Object[] {(byte) 9, 1}));
+    }
+
     @Test
     public void testReorderedInputProducesTableSchemaRow() {
         PaimonWriteSchema schema = PaimonWriteSchema.create(tableType(),
@@ -197,4 +224,14 @@ private static RowType tableType() {
                 new DataField(2, "score", new DoubleType()),
                 new DataField(3, "region", new VarCharType())));
     }
+
+    private static InternalRow changelogRow(PaimonWriteSchema schema, byte operation) {
+        return schema.tableRow(new Object[] {
+                operation,
+                11,
+                BinaryString.fromString("value"),
+                1.0D,
+                BinaryString.fromString("east")
+        });
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonRowChangeOperation.java

@@ -0,0 +1,29 @@
+// 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.doris.datasource.paimon;
+
+/** Stable Doris wire values for Paimon row-level changes. */
+public final class PaimonRowChangeOperation {
+    public static final String OPERATION_COLUMN = "__DORIS_PAIMON_ROW_KIND__";
+    public static final byte INSERT = 0;
+    public static final byte UPDATE = 1;
+    public static final byte DELETE = 2;
+
+    private PaimonRowChangeOperation() {
+    }
+}