You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by st...@apache.org on 2025/08/16 03:45:15 UTC
(iceberg) branch main updated: Flink: Fix hash code comparison for requesting global statistics in DataStatisticsCoordinator (#13827)
This is an automated email from the ASF dual-hosted git repository.
stevenwu 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 3685b5558f Flink: Fix hash code comparison for requesting global statistics in DataStatisticsCoordinator (#13827)
3685b5558f is described below
commit 3685b5558fa9e1c3d9b96aebb5937462f4cd2d07
Author: GuoYu <51...@qq.com>
AuthorDate: Sat Aug 16 11:45:09 2025 +0800
Flink: Fix hash code comparison for requesting global statistics in DataStatisticsCoordinator (#13827)
---
.../sink/shuffle/DataStatisticsCoordinator.java | 4 +-
.../shuffle/TestDataStatisticsCoordinator.java | 53 ++++++++++++++++++++++
2 files changed, 55 insertions(+), 2 deletions(-)
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
index e23ad943fd..773d0fe6c6 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java
@@ -277,9 +277,9 @@ class DataStatisticsCoordinator implements OperatorCoordinator {
if (globalStatistics != null) {
runInCoordinatorThread(
() -> {
- if (event.signature() != null && event.signature() != globalStatistics.hashCode()) {
+ if (event.signature() != null && event.signature() == globalStatistics.hashCode()) {
LOG.debug(
- "Skip responding to statistics request from subtask {}, as hashCode matches or not included in the request",
+ "Skip responding to statistics request from subtask {}, as the operator task already holds the same global statistics",
subtask);
} else {
LOG.info(
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java
index acecc5b60a..0a6caf2aaa 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/shuffle/TestDataStatisticsCoordinator.java
@@ -279,6 +279,59 @@ public class TestDataStatisticsCoordinator {
}
}
+ @Test
+ public void testMultipleRequestGlobalStatisticsEvents() throws Exception {
+ try (DataStatisticsCoordinator dataStatisticsCoordinator =
+ createCoordinator(StatisticsType.Map)) {
+ dataStatisticsCoordinator.start();
+ tasksReady(dataStatisticsCoordinator);
+
+ StatisticsEvent checkpoint1Subtask0DataStatisticEvent =
+ Fixtures.createStatisticsEvent(
+ StatisticsType.Map, Fixtures.TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("a"));
+ StatisticsEvent checkpoint1Subtask1DataStatisticEvent =
+ Fixtures.createStatisticsEvent(
+ StatisticsType.Map, Fixtures.TASK_STATISTICS_SERIALIZER, 1L, CHAR_KEYS.get("b"));
+
+ dataStatisticsCoordinator.handleEventFromOperator(
+ 0, 0, checkpoint1Subtask0DataStatisticEvent);
+ dataStatisticsCoordinator.handleEventFromOperator(
+ 1, 0, checkpoint1Subtask1DataStatisticEvent);
+
+ waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
+
+ // signature is null
+ dataStatisticsCoordinator.handleEventFromOperator(0, 0, new RequestGlobalStatisticsEvent());
+
+ // Checkpoint StatisticEvent + RequestGlobalStatisticsEvent
+ Awaitility.await("wait for first statistics event")
+ .pollInterval(Duration.ofMillis(10))
+ .atMost(Duration.ofSeconds(10))
+ .until(() -> receivingTasks.getSentEventsForSubtask(0).size() == 2);
+
+ // Simulate the scenario where a subtask send global statistics request with the same hash
+ // code. The coordinator would skip the response after comparing the request contained hash
+ // code with latest global statistics hash code.
+ int correctSignature = dataStatisticsCoordinator.globalStatistics().hashCode();
+ dataStatisticsCoordinator.handleEventFromOperator(
+ 0, 0, new RequestGlobalStatisticsEvent(correctSignature));
+
+ waitForCoordinatorToProcessActions(dataStatisticsCoordinator);
+ // Checkpoint StatisticEvent + RequestGlobalStatisticsEvent
+ assertThat(receivingTasks.getSentEventsForSubtask(0).size()).isEqualTo(2);
+
+ // signature is different
+ dataStatisticsCoordinator.handleEventFromOperator(
+ 0, 0, new RequestGlobalStatisticsEvent(correctSignature + 1));
+
+ // Checkpoint StatisticEvent + RequestGlobalStatisticsEvent + RequestGlobalStatisticsEvent
+ Awaitility.await("wait for second statistics event")
+ .pollInterval(Duration.ofMillis(10))
+ .atMost(Duration.ofSeconds(10))
+ .until(() -> receivingTasks.getSentEventsForSubtask(0).size() == 3);
+ }
+ }
+
static void setAllTasksReady(
int subtasks,
DataStatisticsCoordinator dataStatisticsCoordinator,