[flink] Optimize failover recovery for idempotent aggregations

fluss-common/src/main/java/org/apache/fluss/metadata/AggFunctionType.java

@@ -38,28 +38,44 @@
 @PublicEvolving
 public enum AggFunctionType {
     // Numeric aggregation
-    SUM,
-    PRODUCT,
-    MAX,
-    MIN,
+    SUM(false),
+    PRODUCT(false),
+    MAX(true),
+    MIN(true),
 
     // Value selection
-    LAST_VALUE,
-    LAST_VALUE_IGNORE_NULLS,
-    FIRST_VALUE,
-    FIRST_VALUE_IGNORE_NULLS,
+    LAST_VALUE(true),
+    LAST_VALUE_IGNORE_NULLS(true),
+    FIRST_VALUE(true),
+    FIRST_VALUE_IGNORE_NULLS(true),
 
     // String aggregation
-    LISTAGG,
-    STRING_AGG, // Alias for LISTAGG - maps to same factory
+    LISTAGG(false),
+    STRING_AGG(false), // Alias for LISTAGG - maps to same factory
 
     // Boolean aggregation
-    BOOL_AND,
-    BOOL_OR,
+    BOOL_AND(true),
+    BOOL_OR(true),
 
     // Roaring bitmap aggregation
-    RBM32,
-    RBM64;
+    RBM32(true),
+    RBM64(true);
+
+    private final boolean idempotent;
+
+    AggFunctionType(boolean idempotent) {
+        this.idempotent = idempotent;
+    }
+
+    /**
+     * Returns whether applying the same aggregation input again leaves the final materialized
+     * aggregate unchanged.
+     *
+     * @return true when duplicate aggregation input is idempotent
+     */
+    public boolean isIdempotent() {
+        return idempotent;
+    }
 
     // ------------------------------------------------------------------------------------------
 

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java

@@ -30,6 +30,7 @@
 import org.apache.fluss.flink.source.FlinkTableSource;
 import org.apache.fluss.flink.source.reader.LeaseContext;
 import org.apache.fluss.flink.utils.FlinkConnectorOptionsUtils;
+import org.apache.fluss.flink.utils.FlinkConversions;
 import org.apache.fluss.metadata.MergeEngineType;
 import org.apache.fluss.metadata.TablePath;
 
@@ -185,6 +186,8 @@ public DynamicTableSink createDynamicTableSink(Context context) {
         List<String> partitionKeys = resolvedCatalogTable.getPartitionKeys();
 
         RowType rowType = (RowType) context.getPhysicalRowDataType().getLogicalType();
+        org.apache.fluss.metadata.Schema tableSchema =
+                FlinkConversions.toFlussTable(resolvedCatalogTable).getSchema();
 
         MergeEngineType mergeEngineType =
                 tableOptions.get(toFlinkOption(ConfigOptions.TABLE_MERGE_ENGINE));
@@ -203,6 +206,7 @@ public DynamicTableSink createDynamicTableSink(Context context) {
                 toFlussClientConfig(
                         context.getCatalogTable().getOptions(), context.getConfiguration()),
                 rowType,
+                tableSchema,
                 primaryKeyIndexes,
                 partitionKeys,
                 isStreamingMode,

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/AggregationRecoveryDecider.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.sink;
+
+import org.apache.fluss.flink.sink.undo.RecoveryAction;
+import org.apache.fluss.metadata.AggFunctionType;
+import org.apache.fluss.metadata.DeleteBehavior;
+import org.apache.fluss.metadata.Schema;
+
+import javax.annotation.Nullable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Decides the failover correction action for an aggregation sink. */
+final class AggregationRecoveryDecider {
+
+    private AggregationRecoveryDecider() {}
+
+    static RecoveryAction decide(
+            Schema tableSchema,
+            @Nullable int[] targetColumnIndexes,
+            DeleteBehavior deleteBehavior,
+            boolean ignoreDelete,
+            boolean inputKnownUpsertOnly) {
+        checkNotNull(deleteBehavior, "deleteBehavior must not be null.");
+        checkNotNull(tableSchema, "tableSchema must not be null.");
+
+        List<Schema.Column> columns = tableSchema.getColumns();
+        int[] effectiveTargets =
+                targetColumnIndexes == null
+                        ? IntStream.range(0, columns.size()).toArray()
+                        : targetColumnIndexes;
+        for (int targetIndex : effectiveTargets) {
+            checkArgument(
+                    targetIndex >= 0 && targetIndex < columns.size(),
+                    "Invalid target column index %s for schema with %s columns.",
+                    targetIndex,
+                    columns.size());
+        }
+
+        if (!ignoreDelete && deleteBehavior == DeleteBehavior.ALLOW && !inputKnownUpsertOnly) {
+            return RecoveryAction.UNDO;
+        }
+
+        Set<Integer> primaryKeyIndexes = new HashSet<>();
+        for (int primaryKeyIndex : tableSchema.getPrimaryKeyIndexes()) {
+            primaryKeyIndexes.add(primaryKeyIndex);
+        }
+        for (int targetIndex : effectiveTargets) {
+            if (primaryKeyIndexes.contains(targetIndex)) {
+                continue;
+            }
+
+            AggFunctionType functionType =
+                    columns.get(targetIndex)
+                            .getAggFunction()
+                            .map(function -> function.getType())
+                            .orElse(AggFunctionType.LAST_VALUE_IGNORE_NULLS);
+            if (!functionType.isIdempotent()) {
+                return RecoveryAction.UNDO;
+            }
+        }
+        return RecoveryAction.NO_OP;
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/sink/FlinkSink.java

@@ -27,6 +27,7 @@
 import org.apache.fluss.flink.sink.shuffle.StatisticsOrRecordChannelComputer;
 import org.apache.fluss.flink.sink.shuffle.StatisticsOrRecordTypeInformation;
 import org.apache.fluss.flink.sink.undo.ProducerOffsetReporter;
+import org.apache.fluss.flink.sink.undo.RecoveryAction;
 import org.apache.fluss.flink.sink.undo.UndoRecoveryOperatorFactory;
 import org.apache.fluss.flink.sink.writer.AppendSinkWriter;
 import org.apache.fluss.flink.sink.writer.FlinkSinkWriter;
@@ -250,14 +251,15 @@ static class UpsertSinkWriterBuilder<InputT>
         private final DistributionMode distributionMode;
         private final FlussSerializationSchema<InputT> flussSerializationSchema;
         private final boolean enableUndoRecovery;
+        private final RecoveryAction recoveryAction;
         @Nullable private final String producerId;
 
         /**
          * Optional context for reporting offsets to the upstream UndoRecoveryOperator.
          *
-         * <p>This is set internally by {@link #addPreWriteTopology} when UndoRecoveryOperator is
-         * added to the pipeline. The context is then passed to the UpsertSinkWriter during
-         * creation.
+         * <p>This is set internally by {@link #addPreWriteTopology} only when the recovery action
+         * is {@link RecoveryAction#UNDO}. The NO_OP compatibility transform remains in the topology
+         * without installing a writer callback.
          *
          * <p>Note: This field is NOT transient because the ProducerOffsetReporterHolder is
          * serializable and needs to survive job serialization to be passed to the TaskManager.
@@ -276,6 +278,7 @@ static class UpsertSinkWriterBuilder<InputT>
                 DistributionMode distributionMode,
                 FlussSerializationSchema<InputT> flussSerializationSchema,
                 boolean enableUndoRecovery,
+                RecoveryAction recoveryAction,
                 @Nullable String producerId) {
             this.tablePath = tablePath;
             this.flussConfig = flussConfig;
@@ -288,6 +291,7 @@ static class UpsertSinkWriterBuilder<InputT>
             this.distributionMode = distributionMode;
             this.flussSerializationSchema = flussSerializationSchema;
             this.enableUndoRecovery = enableUndoRecovery;
+            this.recoveryAction = recoveryAction;
             this.producerId = producerId;
         }
 
@@ -341,7 +345,8 @@ public DataStream<InputT> addPreWriteTopology(DataStream<InputT> input) {
                                 targetColumnIndexes,
                                 numBucket,
                                 !partitionKeys.isEmpty(),
-                                producerId);
+                                producerId,
+                                recoveryAction);
 
                 stream =
                         stream.transform(
@@ -350,7 +355,9 @@ public DataStream<InputT> addPreWriteTopology(DataStream<InputT> input) {
                                         operatorFactory)
                                 .setParallelism(stream.getParallelism());
 
-                offsetReporter = operatorFactory.getProducerOffsetReporter();
+                if (operatorFactory.getRecoveryAction() == RecoveryAction.UNDO) {
+                    offsetReporter = operatorFactory.getProducerOffsetReporter();
+                }
             }
 
             return stream;