Skip to content

[core] Restart aggregation after a removing delete #10952

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cxzl25 wants to merge 1 commit into
base: master
Choose a base branch
from
+89 −2
Open

[core] Restart aggregation after a removing delete #10952

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions ...e/src/main/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

@JingsongLi JingsongLi Nov 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve the delete barrier through buffer preaggregation

Resetting this accumulator is insufficient when DELETE and the first replacement INSERT are in the same checkpoint. MergeTreeWriter.flushWriteBuffer invokes this function before persisting that buffer, so getResult() emits an ordinary INSERT and loses the fact that older aggregates must be discarded. I reproduced this by adapting the new table test with supported write-only=true: commit (pt=1,a=1,b=10,c=1), then buffer its DELETE and (1,1,3,7) together before the next prepareCommit, then commit (1,1,4,8). The actual table read returns +I[1,1,17,1] instead of +I[1,1,7,7]; putting DELETE in its own commit passes the identical control. Thus the claimed delete/reinsert fix still depends on checkpoint grouping, and old sum/first_value values survive. Please preserve reset/tombstone semantics through buffer reduction and partial compaction, or avoid collapsing this boundary until older versions participate, and add a same-checkpoint persistence regression.

row = new GenericRow(getters.length);
Arrays.stream(aggregators).forEach(FieldAggregator::reset);
}

boolean isRetract = kv.valueKind().isRetract();
for (int i = 0; i < getters.length; i++) {
Expand Down
34 changes: 34 additions & 0 deletions ...c/test/java/org/apache/paimon/mergetree/compact/aggregate/AggregateMergeFunctionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down
46 changes: 46 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down
Loading