(iceberg) branch main updated: Flink: Clear globalStatisticsState in init to avoid duplication (#14294)

[pv...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 59344e83df Flink: Clear globalStatisticsState in init to avoid duplication (#14294)
59344e83df is described below

commit 59344e83dfa61ce5bd834d5a6de9a51dc04b9b6b
Author: GuoYu <51...@qq.com>
AuthorDate: Mon Oct 13 17:51:08 2025 +0800

    Flink: Clear globalStatisticsState in init to avoid duplication (#14294)
    
    Co-authored-by: hydrogenlee <14...@qq.com>
---
 .../apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java | 8 ++++++++
 .../iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java    | 6 ++++++
 2 files changed, 14 insertions(+)

diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
index a873136c91..a481794db7 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
@@ -122,6 +122,9 @@ public class DataStatisticsOperator extends AbstractStreamOperator<StatisticsOrR
         this.globalStatistics = restoredStatistics;
       }
 
+      // Perform a cleanup first to ensure that the state is empty.
+      globalStatisticsState.clear();
+
       // Always request for new statistics from coordinator upon task initialization.
       // There are a few scenarios this is needed
       // 1. downstream writer parallelism changed due to rescale.
@@ -266,4 +269,9 @@ public class DataStatisticsOperator extends AbstractStreamOperator<StatisticsOrR
   GlobalStatistics globalStatistics() {
     return globalStatistics;
   }
+
+  @VisibleForTesting
+  ListState<GlobalStatistics> globalStatisticsState() {
+    return globalStatisticsState;
+  }
 }
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
index 09b2b6371e..408309b2b9 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
@@ -242,6 +242,12 @@ public class TestDataStatisticsOperator {
       testHarness2.setup();
       testHarness2.initializeState(snapshot);
 
+      // When we restore from the savepoint, we should ensure that `globalStatisticsState` has been
+      // completely cleaned up
+      Iterable<GlobalStatistics> globalStatisticsIterable =
+          restoredOperator.globalStatisticsState().get();
+      assertThat(globalStatisticsIterable).isEmpty();
+
       GlobalStatistics globalStatistics = restoredOperator.globalStatistics();
       // global statistics is always restored and used initially even if
       // downstream parallelism changed.