diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
index 3e8aaa1b8d0f..5d1a856f8a09 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
@@ -81,12 +81,19 @@ public void reset() {
     public void add(KeyValue kv) {
         latestKv = kv;
 
-        currentDeleteRow = removeRecordOnDelete && kv.valueKind() == RowKind.DELETE;
-        if (currentDeleteRow) {
+        if (removeRecordOnDelete && kv.valueKind() == RowKind.DELETE) {
+            currentDeleteRow = true;
             row = new GenericRow(getters.length);
             initRow(row, kv.value());
             return;
         }
+        if (currentDeleteRow) {
+            // The record was removed by the previous delete, so later records start a new
+            // aggregation instead of accumulating onto the deleted record's values.
+            currentDeleteRow = false;
+            row = new GenericRow(getters.length);
+            Arrays.stream(aggregators).forEach(FieldAggregator::reset);
+        }
 
         boolean isRetract = kv.valueKind().isRetract();
         for (int i = 0; i < getters.length; i++) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunctionTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunctionTest.java
index f1ae73151176..9d1538875cc6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunctionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunctionTest.java
@@ -171,6 +171,40 @@ void testInitRowWithNullableFieldOnDelete() {
         assertThat(aggregateFunction.getResult().value()).isEqualTo(GenericRow.of(1, 2, 2, null));
     }
 
+    @Test
+    void testInsertAfterRemovingDeleteStartsFresh() {
+        Options options = new Options();
+        options.set(FIELDS_DEFAULT_AGG_FUNC, "sum");
+        options.set("fields.c.aggregate-function", "first_value");
+        options.set("aggregation.remove-record-on-delete", "true");
+        MergeFunction<KeyValue> aggregateFunction =
+                AggregateMergeFunction.factory(
+                                options,
+                                RowType.builder()
+                                        .fields(
+                                                new DataType[] {
+                                                    DataTypes.INT().notNull(),
+                                                    DataTypes.INT(),
+                                                    DataTypes.INT(),
+                                                    DataTypes.INT()
+                                                },
+                                                new String[] {"k", "a", "b", "c"})
+                                        .build(),
+                                Collections.singletonList("k"))
+                        .create();
+        aggregateFunction.reset();
+
+        aggregateFunction.add(value(1, 10, 10, 1));
+        // the delete carries the old image of the record, as a CDC delete does
+        aggregateFunction.add(deleteValue(1, 10, 10, 1));
+        // the record was removed, so the new records aggregate from scratch
+        aggregateFunction.add(value(1, 3, null, 7));
+        aggregateFunction.add(value(1, 4, 5, 8));
+        KeyValue result = aggregateFunction.getResult();
+        assertThat(result.valueKind()).isEqualTo(RowKind.INSERT);
+        assertThat(result.value()).isEqualTo(GenericRow.of(1, 7, 5, 7));
+    }
+
     private KeyValue value(Integer... values) {
         return new KeyValue()
                 .replace(GenericRow.of(values[0]), RowKind.INSERT, GenericRow.of(values));
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index f6bc8fcc423c..cd0b9937b1b9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -1790,6 +1790,52 @@ public void testAuditLog() throws Exception {
         assertThat(result).containsExactlyInAnyOrder("+I[20, 2]", "+I[30, 1]", "+I[10, 1]");
     }
 
+    @Test
+    public void testAggregationRemoveRecordOnDeleteThenInsert() 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");
+                            options.set("fields.b.aggregate-function", "sum");
+                            options.set("fields.c.aggregate-function", "first_value");
+                        },
+                        rowType);
+        Function<InternalRow, String> rowToString = row -> internalRowToString(row, rowType);
+        SnapshotReader snapshotReader = table.newSnapshotReader();
+        TableRead read = table.newRead();
+        StreamTableWrite write = table.newWrite("");
+        StreamTableCommit commit = table.newCommit("");
+
+        write.write(GenericRow.of(1, 1, 10, 1));
+        commit.commit(0, write.prepareCommit(true, 0));
+        assertThat(getResult(read, toSplits(snapshotReader.read().dataSplits()), rowToString))
+                .containsExactly("+I[1, 1, 10, 1]");
+
+        // the delete removes the record, carrying the old image like a CDC delete
+        write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, 10, 1));
+        commit.commit(1, write.prepareCommit(true, 1));
+        assertThat(getResult(read, toSplits(snapshotReader.read().dataSplits()), rowToString))
+                .isEmpty();
+
+        // later inserts start a new aggregation, the deleted values must not survive
+        write.write(GenericRow.of(1, 1, 3, 7));
+        commit.commit(2, write.prepareCommit(true, 2));
+        write.write(GenericRow.of(1, 1, 4, 8));
+        commit.commit(3, write.prepareCommit(true, 3));
+        assertThat(getResult(read, toSplits(snapshotReader.read().dataSplits()), rowToString))
+                .containsExactly("+I[1, 1, 7, 7]");
+
+        write.close();
+        commit.close();
+    }
+
     @Test
     public void testAggregationRemoveRecordOnDelete() throws Exception {
         RowType rowType =