From 66adb4f238c8a1b0b91069bafd451df5b7133618 Mon Sep 17 00:00:00 2001
From: chenjian2664 <chenjian2664@gmail.com>
Date: Wed, 4 Nov 2026 20:27:53 +0800
Subject: [PATCH] [core] Keep first_non_null_value when an older sequence-group
 record is null

---
 .../aggregate/FieldFirstNonNullValueAgg.java  | 10 ++++++
 .../PartialUpdateMergeFunctionTest.java       | 32 +++++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
index 273af1a957f6..4d76038a7d47 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldFirstNonNullValueAgg.java
@@ -41,6 +41,16 @@ public Object agg(Object accumulator, Object inputField) {
         }
     }
 
+    @Override
+    public Object aggReversed(Object accumulator, Object inputField) {
+        // the input precedes the accumulator, so a non-null input is the first one
+        if (inputField != null) {
+            initialized = true;
+            return inputField;
+        }
+        return accumulator;
+    }
+
     @Override
     public void reset() {
         this.initialized = false;
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
index e00449838ec6..873a2cc72d74 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
@@ -1030,6 +1030,38 @@ public void testFirstValue() {
         validate(func, 1, 2, 3, 2);
     }
 
+    @Test
+    public void testFirstNonNullValue() {
+        Options options = new Options();
+        options.set("fields.f1.sequence-group", "f2,f3");
+        options.set("fields.f2.aggregate-function", "first_non_null_value");
+        RowType rowType =
+                RowType.of(DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.INT());
+        MergeFunction<KeyValue> func =
+                PartialUpdateMergeFunction.factory(options, rowType, ImmutableList.of("f0"))
+                        .create();
+
+        func.reset();
+
+        add(func, 1, 2, 2, 2);
+        validate(func, 1, 2, 2, 2);
+        // an older null does not replace the first non-null value
+        add(func, 1, 1, null, 1);
+        validate(func, 1, 2, 2, 2);
+        // an older non-null value comes first
+        add(func, 1, 0, 3, 3);
+        validate(func, 1, 2, 3, 2);
+
+        func.reset();
+        add(func, 1, 2, null, 2);
+        validate(func, 1, 2, null, 2);
+        add(func, 1, 1, 3, 1);
+        validate(func, 1, 2, 3, 2);
+        // a newer value does not replace the first non-null value from an older record
+        add(func, 1, 3, 4, 3);
+        validate(func, 1, 3, 3, 3);
+    }
+
     @Test
     public void testMultiSequenceFieldsFirstValue() {
         Options options = new Options();