[Bug] Aggregation SUM retains deleted values after delete/reinsert #10963

Search before asking

• I searched in the issues and found no duplicate of this reproduction.

Paimon version

2.1.0 (org.apache.paimon:paimon-bundle:2.1.0).

Compute Engine

Java API, JDK 11.0.26, macOS arm64. The standalone reproduction below invokes Paimon's AggregateMergeFunction directly; no Fluss, Flink, or Spark runtime is involved.

Minimal reproduce step

Configure an aggregation primary-key table with:

merge-engine=aggregation
ignore-delete=false
aggregation.remove-record-on-delete=true
fields.amount.aggregate-function=sum

For the same primary key, merge these records in order, without a flush/reset between them:

INSERT(id=1, amount=10)
DELETE(id=1, amount=10)
INSERT(id=1, amount=3)

The DELETE contains the old row's value. With whole-row deletion enabled, the final amount should be 3, but the merge result is 13.

The following standalone program also checks a DELETE-only case and a DELETE with a null amount as controls.

Save this pom.xml in an empty directory:

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>repro</groupId>
    <artifactId>paimon-delete-reinsert</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.paimon</groupId>
            <artifactId>paimon-bundle</artifactId>
            <version>2.1.0</version>
        </dependency>
    </dependencies>
</project>

Save this class as src/main/java/PaimonAggregateDeleteReinsertRepro.java:

import org.apache.paimon.KeyValue;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.mergetree.compact.MergeFunction;
import org.apache.paimon.mergetree.compact.aggregate.AggregateMergeFunction;
import org.apache.paimon.options.Options;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;

import java.util.Collections;

public class PaimonAggregateDeleteReinsertRepro {
    public static void main(String[] args) {
        run("delete_only", 10, false);
        run("delete_old_value_reinsert", 10, true);
        run("delete_null_reinsert", null, true);
    }

    private static void run(String name, Integer deleteAmount, boolean reinsert) {
        Options options = new Options();
        options.setString("merge-engine", "aggregation");
        options.setString("ignore-delete", "false");
        options.setString("aggregation.remove-record-on-delete", "true");
        options.setString("fields.amount.aggregate-function", "sum");
        RowType rowType = RowType.of(
                new DataType[] {DataTypes.INT().notNull(), DataTypes.INT()},
                new String[] {"id", "amount"});
        MergeFunction<KeyValue> merge = AggregateMergeFunction
                .factory(options, rowType, Collections.singletonList("id"))
                .create();
        merge.reset();
        merge.add(value(0L, RowKind.INSERT, 10));
        merge.add(value(1L, RowKind.DELETE, deleteAmount));
        if (reinsert) {
            merge.add(value(2L, RowKind.INSERT, 3));
        }
        KeyValue result = merge.getResult();
        System.out.println(name + " => " + result.valueKind().name()
                + " amount=" + result.value().getInt(1));
    }

    private static KeyValue value(long sequence, RowKind kind, Integer amount) {
        return new KeyValue().replace(
                GenericRow.of(1), sequence, kind, GenericRow.of(1, amount));
    }
}

Run:

mvn -q clean compile org.codehaus.mojo:exec-maven-plugin:3.5.0:java \
  -Dexec.mainClass=PaimonAggregateDeleteReinsertRepro

What doesn't meet your expectations?

Actual output:

delete_only => DELETE amount=10
delete_old_value_reinsert => INSERT amount=13
delete_null_reinsert => INSERT amount=3

Expected: delete_old_value_reinsert should produce INSERT amount=3. A whole-row DELETE should discard the previous aggregate before a new INSERT starts aggregating.

The DELETE-only control correctly returns a delete marker. Its payload is not the problem by itself; the incorrect result appears when a subsequent INSERT reuses the deleted value. The null-payload control uses a nullable amount field and helps isolate the issue.

Anything else?

In AggregateMergeFunction.add, a whole-row DELETE creates a new accumulator row and initializes it from the DELETE payload. The next INSERT clears the delete flag but aggregates into that same initialized row. For SUM, this computes 10 + 3 = 13.

A possible fix is to ensure that a whole-row DELETE leaves a fresh aggregation state for a subsequent INSERT, while still preserving a valid DELETE output row for serialization and non-nullable fields. Regression coverage should include delete/reinsert in the same merge/flush.

This was originally observed through the Paimon BatchTableWrite API while investigating apache/fluss#4956. The standalone reproduction above isolates the issue inside Paimon without relying on Fluss.

Related investigations:

• [core] Fix initRow in AggregateMergeFunction and PartialUpdateMergeFunction to set all fields including nullable ones #7645 changed initRow to copy nullable fields as well.
• [Bug] Agg with DELETE has strange behavior #2561 concerns older contribution-retraction behavior; [Bug] Flink lookup join keeps returning rows deleted from an aggregation or partial-update table #10923 concerns lookup-cache invalidation. Neither covers this whole-row delete/reinsert accumulator reproduction.

Are you willing to submit a PR?

• I'm willing to submit a PR!