You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2025/10/13 11:25:29 UTC
(iceberg) branch main updated: Flink: Backport clear globalStatisticsState in init to avoid duplication (#14315)
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 6ca4009297 Flink: Backport clear globalStatisticsState in init to avoid duplication (#14315)
6ca4009297 is described below
commit 6ca4009297483eaf6a66c865bb0498d778000b1a
Author: GuoYu <51...@qq.com>
AuthorDate: Mon Oct 13 19:25:24 2025 +0800
Flink: Backport clear globalStatisticsState in init to avoid duplication (#14315)
Backports #14294
---
.../apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java | 8 ++++++++
.../iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java | 6 ++++++
.../apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java | 8 ++++++++
.../iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java | 6 ++++++
4 files changed, 28 insertions(+)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
index 7995a8a5b1..945deeec27 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
@@ -119,6 +119,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.
@@ -263,4 +266,9 @@ public class DataStatisticsOperator extends AbstractStreamOperator<StatisticsOrR
GlobalStatistics globalStatistics() {
return globalStatistics;
}
+
+ @VisibleForTesting
+ ListState<GlobalStatistics> globalStatisticsState() {
+ return globalStatisticsState;
+ }
}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
index f7a7a147e7..da0a498da9 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
@@ -250,6 +250,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.
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
index a873136c91..a481794db7 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsOperator.java
+++ b/flink/v2.0/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.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
index 09b2b6371e..408309b2b9 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsOperator.java
+++ b/flink/v2.0/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.