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 2026/07/22 12:39:44 UTC
(iceberg) branch main updated: Flink: Add plannedGroups counter metric and log for DataFileRewritePlanner (#17316)
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 a8d1e1464b Flink: Add plannedGroups counter metric and log for DataFileRewritePlanner (#17316)
a8d1e1464b is described below
commit a8d1e1464b55a41526bafa0683964fee1b466ec6
Author: GuoYu <51...@qq.com>
AuthorDate: Wed Jul 22 20:39:37 2026 +0800
Flink: Add plannedGroups counter metric and log for DataFileRewritePlanner (#17316)
---
.../operator/DataFileRewritePlanner.java | 8 ++++++-
.../operator/TableMaintenanceMetrics.java | 3 +++
.../maintenance/api/TestRewriteDataFiles.java | 25 ++++++++++++++++++++++
3 files changed, 35 insertions(+), 1 deletion(-)
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
index b78c602c64..d6c3c95d55 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
@@ -63,6 +63,7 @@ public class DataFileRewritePlanner
private final long maxRewriteBytes;
private final Map<String, String> rewriterOptions;
private transient Counter errorCounter;
+ private transient Counter plannedGroupsCounter;
private final String branch;
private final SerializableSupplier<Expression> filterSupplier;
@@ -100,6 +101,9 @@ public class DataFileRewritePlanner
this.errorCounter =
TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
.counter(TableMaintenanceMetrics.ERROR_COUNTER);
+ this.plannedGroupsCounter =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
+ .counter(TableMaintenanceMetrics.PLANNED_GROUPS_COUNTER);
}
@Override
@@ -157,12 +161,14 @@ public class DataFileRewritePlanner
IntMath.divide(groups.size(), partialProgressMaxCommits, RoundingMode.CEILING);
LOG.info(
- DataFileRewritePlanner.MESSAGE_PREFIX + "Rewrite plan created {}",
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewrite plan created, {} groups: {}",
tableName,
taskName,
taskIndex,
ctx.timestamp(),
+ groups.size(),
groups);
+ plannedGroupsCounter.inc(groups.size());
for (RewriteFileGroup group : groups) {
LOG.info(
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
index 897760caaa..a4805356d3 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
@@ -51,6 +51,9 @@ public class TableMaintenanceMetrics {
public static final String REMOVED_DATA_FILE_NUM_METRIC = "removedDataFileNum";
public static final String REMOVED_DATA_FILE_SIZE_METRIC = "removedDataFileSize";
+ // DataFileRewritePlanner metrics
+ public static final String PLANNED_GROUPS_COUNTER = "plannedGroups";
+
static MetricGroup groupFor(
RuntimeContext context, String tableName, String taskName, int taskIndex) {
return groupFor(groupFor(context, tableName), taskName, taskIndex);
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
index c27f6081af..e0ce691304 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
@@ -25,6 +25,7 @@ import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.REWRITE_
import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.ADDED_DATA_FILE_NUM_METRIC;
import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.ADDED_DATA_FILE_SIZE_METRIC;
import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.ERROR_COUNTER;
+import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.PLANNED_GROUPS_COUNTER;
import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.REMOVED_DATA_FILE_NUM_METRIC;
import static org.apache.iceberg.flink.maintenance.operator.TableMaintenanceMetrics.REMOVED_DATA_FILE_SIZE_METRIC;
import static org.assertj.core.api.Assertions.assertThat;
@@ -255,6 +256,14 @@ class TestRewriteDataFiles extends MaintenanceTaskTestBase {
"0",
ERROR_COUNTER),
1L)
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ PLANNED_GROUPS_COUNTER),
+ 1L)
.put(
ImmutableList.of(
REWRITE_TASK_NAME + "[0]",
@@ -371,6 +380,14 @@ class TestRewriteDataFiles extends MaintenanceTaskTestBase {
"0",
ERROR_COUNTER),
0L)
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ PLANNED_GROUPS_COUNTER),
+ 1L)
.put(
ImmutableList.of(
REWRITE_TASK_NAME + "[0]",
@@ -450,6 +467,14 @@ class TestRewriteDataFiles extends MaintenanceTaskTestBase {
"0",
ERROR_COUNTER),
0L)
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ PLANNED_GROUPS_COUNTER),
+ 1L)
.put(
ImmutableList.of(
REWRITE_TASK_NAME + "[0]",