From 9a71f929fdb8d0f4cf61b6c36afb2dc27b28d3d0 Mon Sep 17 00:00:00 2001 From: sychen Date: Mon, 2 Nov 2026 21:35:35 +0800 Subject: [PATCH] [core] Restart aggregation after a removing delete With aggregation.remove-record-on-delete enabled, a delete followed by new records for the same key kept aggregating on top of the deleted record's values, because the accumulator row was replaced by the delete's own image and the field aggregators were not reset. Reset the accumulator row and every field aggregator when the first record after such a delete arrives. --- .../aggregate/AggregateMergeFunction.java | 11 ++++- .../aggregate/AggregateMergeFunctionTest.java | 34 ++++++++++++++ .../table/PrimaryKeySimpleTableTest.java | 46 +++++++++++++++++++ 3 files changed, 89 insertions(+), 2 deletions(-) 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 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 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 =