[core] Keep first_non_null_value when an older sequence-group record is null

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;

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();