[core] Aggregate merge function should not return empty row with non-null column type exist

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeFunction.java

@@ -36,7 +36,10 @@
  */
 public interface MergeFunction<T> {
 
-    /** Reset the merge function to its default state. */
+    /**
+     * Reset the merge function to its default state, call this before calling {@link
+     * #add(KeyValue)} for the first time or after {@link #getResult}.
+     */
     void reset();
 
     /** Add the given {@link KeyValue} to the merge function. */

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java

@@ -31,14 +31,14 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.RowKind;
+import org.apache.paimon.utils.ArrayUtils;
 import org.apache.paimon.utils.Projection;
 
 import javax.annotation.Nullable;
 
 import java.util.Arrays;
 import java.util.List;
 
-import static org.apache.paimon.CoreOptions.AGGREGATION_REMOVE_RECORD_ON_DELETE;
 import static org.apache.paimon.utils.InternalRowUtils.createFieldGetters;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
@@ -50,30 +50,30 @@ public class AggregateMergeFunction implements MergeFunction<KeyValue> {
 
     private final InternalRow.FieldGetter[] getters;
     private final FieldAggregator[] aggregators;
+    private final boolean[] nullables;
 
     private KeyValue latestKv;
     private GenericRow row;
     private KeyValue reused;
     private boolean currentDeleteRow;
     private final boolean removeRecordOnDelete;
-
-    public AggregateMergeFunction(
-            InternalRow.FieldGetter[] getters, FieldAggregator[] aggregators) {
-        this(getters, aggregators, false);
-    }
+    private boolean notNullColumnFilled;
 
     public AggregateMergeFunction(
             InternalRow.FieldGetter[] getters,
             FieldAggregator[] aggregators,
-            boolean removeRecordOnDelete) {
+            boolean removeRecordOnDelete,
+            boolean[] nullables) {
         this.getters = getters;
         this.aggregators = aggregators;
         this.removeRecordOnDelete = removeRecordOnDelete;
+        this.nullables = nullables;
     }
 
     @Override
     public void reset() {
         this.latestKv = null;
+        this.notNullColumnFilled = false;
         this.row = new GenericRow(getters.length);
         Arrays.stream(aggregators).forEach(FieldAggregator::reset);
         this.currentDeleteRow = false;
@@ -87,6 +87,10 @@ public void add(KeyValue kv) {
 
         currentDeleteRow = removeRecordOnDelete && isRetract;
         if (currentDeleteRow) {
+            if (!notNullColumnFilled) {
+                initRow(row, kv.value());
+                notNullColumnFilled = true;
+            }
             return;
         }
 
@@ -100,6 +104,20 @@ public void add(KeyValue kv) {
                             : fieldAggregator.agg(accumulator, inputField);
             row.setField(i, mergedField);
         }
+        notNullColumnFilled = true;
+    }
+
+    private void initRow(GenericRow row, InternalRow value) {
+        for (int i = 0; i < getters.length; i++) {
+            Object field = getters[i].getFieldOrNull(value);
+            if (!nullables[i]) {
+                if (field != null) {
+                    row.setField(i, field);
+                } else {
+                    throw new IllegalArgumentException("Field " + i + " can not be null");
+                }
+            }
+        }
     }
 
     @Override
@@ -147,7 +165,7 @@ private Factory(
             this.tableNames = tableNames;
             this.tableTypes = tableTypes;
             this.primaryKeys = primaryKeys;
-            this.removeRecordOnDelete = conf.get(AGGREGATION_REMOVE_RECORD_ON_DELETE);
+            this.removeRecordOnDelete = options.aggregationRemoveRecordOnDelete();
         }
 
         @Override
@@ -172,7 +190,11 @@ public MergeFunction<KeyValue> create(@Nullable int[][] projection) {
             }
 
             return new AggregateMergeFunction(
-                    createFieldGetters(fieldTypes), fieldAggregators, removeRecordOnDelete);
+                    createFieldGetters(fieldTypes),
+                    fieldAggregators,
+                    removeRecordOnDelete,
+                    ArrayUtils.toPrimitiveBoolean(
+                            fieldTypes.stream().map(DataType::isNullable).toArray(Boolean[]::new)));
         }
 
         private String getAggFuncName(String fieldName, List<String> sequenceFields) {

paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java

@@ -286,7 +286,9 @@ public void testSum(boolean changelogRowDeduplicate) {
                                                 new FieldAggregator[] {
                                                     new FieldSumAggFactory()
                                                             .create(DataTypes.INT(), null, null)
-                                                })),
+                                                },
+                                                false,
+                                                null)),
                         key -> null,
                         changelogRowDeduplicate ? EQUALISER : null,
                         LookupStrategy.from(false, true, false, false),
@@ -373,7 +375,9 @@ public void testMergeHighLevelOrder() {
                                                 new FieldAggregator[] {
                                                     new FieldLastValueAggFactory()
                                                             .create(DataTypes.INT(), null, null)
-                                                })),
+                                                },
+                                                false,
+                                                null)),
                         highLevel::get,
                         null,
                         LookupStrategy.from(false, true, false, false),

paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java

@@ -1326,6 +1326,44 @@ public void testAggregationRemoveRecordOnDelete() throws Exception {
         commit.close();
     }
 
+    @Test
+    public void testAggregationRemoveRecordOnDeleteWithoutInsertData() throws Exception {
+        RowType rowType =
+                RowType.of(
+                        new DataType[] {
+                            DataTypes.INT(), DataTypes.INT(), DataTypes.INT(), DataTypes.INT()
+                        },
+                        new String[] {"pt", "a", "b", "c"});
+        FileStoreTable table =
+                createFileStoreTable(
+                        options -> {
+                            options.set("merge-engine", "aggregation");
+                            options.set("aggregation.remove-record-on-delete", "true");
+                        },
+                        rowType);
+
+        // delete twice to trigger merge when read
+        try (StreamTableWrite write = table.newWrite("");
+                StreamTableCommit commit = table.newCommit("")) {
+            write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, 2, 2));
+            commit.commit(0, write.prepareCommit(true, 0));
+            write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, 2, 2));
+            commit.commit(1, write.prepareCommit(true, 1));
+        }
+
+        // read auditLog table
+        ReadBuilder builder = new AuditLogTable(table).newReadBuilder();
+        try (RecordReader<InternalRow> reader =
+                builder.newRead().createReader(builder.newScan().plan())) {
+            reader.forEachRemaining(
+                    row -> {
+                        assertThat(row.getString(0).toString()).isEqualTo("-D");
+                        assertThat(row.isNullAt(1)).isFalse();
+                        assertThat(row.getInt(1)).isEqualTo(1);
+                    });
+        }
+    }
+
     @Test
     public void testPartialUpdateRemoveRecordOnDelete() throws Exception {
         RowType rowType =