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/05/07 05:52:21 UTC
(iceberg) branch main updated (a04c532650 -> 27e73fea23)
This is an automated email from the ASF dual-hosted git repository.
stevenwu pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
from a04c532650 Core: Enable HTTP proxy support for the client used by REST Catalog (#12406)
new 4bb60aa0bc Flink: Backport Maintenance - RewriteDataFiles to Flink 1.19
new 27e73fea23 Flink: Backport Maintenance - RewriteDataFiles to Flink 2.0
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../flink/maintenance/api/ExpireSnapshots.java | 2 +-
.../maintenance/api/MaintenanceTaskBuilder.java | 4 +-
.../flink/maintenance/api/RewriteDataFiles.java | 0
.../flink/maintenance/api/TableMaintenance.java | 7 +-
.../operator/DataFileRewriteCommitter.java | 0
.../operator/DataFileRewritePlanner.java | 0
.../operator/DataFileRewriteRunner.java | 0
.../maintenance/operator/DeleteFilesProcessor.java | 29 ++----
.../operator/ExpireSnapshotsProcessor.java | 2 +-
.../flink/maintenance/operator/LockRemover.java | 28 ++---
.../flink/maintenance/operator/LogUtil.java | 0
.../operator/TableMaintenanceMetrics.java | 30 ++++++
.../maintenance/operator/TaskResultAggregator.java | 0
.../flink/maintenance/operator/TriggerManager.java | 29 ++----
.../api/MaintenanceTaskInfraExtension.java | 8 +-
.../maintenance/api/MaintenanceTaskTestBase.java | 39 ++++++-
.../flink/maintenance/api/TestExpireSnapshots.java | 43 ++------
.../flink/maintenance/api/TestMaintenanceE2E.java | 13 +++
.../maintenance/api/TestRewriteDataFiles.java | 0
.../maintenance/operator/OperatorTestBase.java | 113 +++++++++++++++++++++
.../flink/maintenance/operator/RewriteUtil.java | 0
.../operator/TestDataFileRewriteCommitter.java | 0
.../operator/TestDataFileRewritePlanner.java | 0
.../operator/TestDataFileRewriteRunner.java | 0
.../operator/TestDeleteFilesProcessor.java | 3 +-
.../flink/maintenance/api/ExpireSnapshots.java | 2 +-
.../maintenance/api/MaintenanceTaskBuilder.java | 4 +-
.../flink/maintenance/api/RewriteDataFiles.java | 0
.../flink/maintenance/api/TableMaintenance.java | 7 +-
.../operator/DataFileRewriteCommitter.java | 0
.../operator/DataFileRewritePlanner.java | 4 +-
.../operator/DataFileRewriteRunner.java | 4 +-
.../maintenance/operator/DeleteFilesProcessor.java | 29 ++----
.../operator/ExpireSnapshotsProcessor.java | 2 +-
.../flink/maintenance/operator/LockRemover.java | 28 ++---
.../flink/maintenance/operator/LogUtil.java | 0
.../operator/TableMaintenanceMetrics.java | 30 ++++++
.../maintenance/operator/TaskResultAggregator.java | 0
.../flink/maintenance/operator/TriggerManager.java | 29 ++----
.../api/MaintenanceTaskInfraExtension.java | 8 +-
.../maintenance/api/MaintenanceTaskTestBase.java | 39 ++++++-
.../flink/maintenance/api/TestExpireSnapshots.java | 43 ++------
.../flink/maintenance/api/TestMaintenanceE2E.java | 13 +++
.../maintenance/api/TestRewriteDataFiles.java | 0
.../maintenance/operator/OperatorTestBase.java | 113 +++++++++++++++++++++
.../flink/maintenance/operator/RewriteUtil.java | 0
.../operator/TestDataFileRewriteCommitter.java | 0
.../operator/TestDataFileRewritePlanner.java | 0
.../operator/TestDataFileRewriteRunner.java | 0
.../operator/TestDeleteFilesProcessor.java | 3 +-
50 files changed, 488 insertions(+), 220 deletions(-)
copy flink/{v1.20 => v1.19}/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java (100%)
copy flink/{v1.20 => v1.19}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java (98%)
copy flink/{v1.20 => v2.0}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java (98%)
copy flink/{v1.20 => v2.0}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java (100%)
copy flink/{v1.20 => v2.0}/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java (100%)
(iceberg) 01/02: Flink: Backport Maintenance - RewriteDataFiles to Flink 1.19
Posted by st...@apache.org.
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
commit 4bb60aa0bc873c15e4446a0960d381d24fcbe8be
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Tue May 6 14:34:00 2025 +0200
Flink: Backport Maintenance - RewriteDataFiles to Flink 1.19
Backports #11497
---
.../flink/maintenance/api/ExpireSnapshots.java | 2 +-
.../maintenance/api/MaintenanceTaskBuilder.java | 4 +-
.../flink/maintenance/api/RewriteDataFiles.java | 234 ++++++++++++
.../flink/maintenance/api/TableMaintenance.java | 7 +-
.../operator/DataFileRewriteCommitter.java | 199 ++++++++++
.../operator/DataFileRewritePlanner.java | 206 ++++++++++
.../operator/DataFileRewriteRunner.java | 253 +++++++++++++
.../maintenance/operator/DeleteFilesProcessor.java | 29 +-
.../operator/ExpireSnapshotsProcessor.java | 2 +-
.../flink/maintenance/operator/LockRemover.java | 28 +-
.../flink/maintenance/operator/LogUtil.java | 26 ++
.../operator/TableMaintenanceMetrics.java | 30 ++
.../maintenance/operator/TaskResultAggregator.java | 101 +++++
.../flink/maintenance/operator/TriggerManager.java | 29 +-
.../api/MaintenanceTaskInfraExtension.java | 8 +-
.../maintenance/api/MaintenanceTaskTestBase.java | 39 +-
.../flink/maintenance/api/TestExpireSnapshots.java | 43 +--
.../flink/maintenance/api/TestMaintenanceE2E.java | 13 +
.../maintenance/api/TestRewriteDataFiles.java | 417 +++++++++++++++++++++
.../maintenance/operator/OperatorTestBase.java | 113 ++++++
.../flink/maintenance/operator/RewriteUtil.java | 83 ++++
.../operator/TestDataFileRewriteCommitter.java | 278 ++++++++++++++
.../operator/TestDataFileRewritePlanner.java | 193 ++++++++++
.../operator/TestDataFileRewriteRunner.java | 355 ++++++++++++++++++
.../operator/TestDeleteFilesProcessor.java | 3 +-
25 files changed, 2587 insertions(+), 108 deletions(-)
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java
index 9cde5cb173..2bac5163ca 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java
@@ -113,7 +113,7 @@ public class ExpireSnapshots {
operatorName(DELETE_FILES_OPERATOR_NAME),
TypeInformation.of(Void.class),
new DeleteFilesProcessor(
- index(), taskName(), tableLoader().loadTable(), deleteBatchSize))
+ tableLoader().loadTable(), taskName(), index(), deleteBatchSize))
.uid(DELETE_FILES_OPERATOR_NAME + uidSuffix())
.slotSharingGroup(slotSharingGroup())
.setParallelism(parallelism());
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java
index 3fc431d025..c884f13b3e 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java
@@ -195,9 +195,9 @@ public abstract class MaintenanceTaskBuilder<T extends MaintenanceTaskBuilder<?>
DataStream<TaskResult> append(
DataStream<Trigger> sourceStream,
- int taskIndex,
- String newTaskName,
String newTableName,
+ String newTaskName,
+ int taskIndex,
TableLoader newTableLoader,
String defaultUidSuffix,
String defaultSlotSharingGroup,
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
new file mode 100644
index 0000000000..b96635e6ab
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
@@ -0,0 +1,234 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import java.util.Map;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewriteCommitter;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewritePlanner;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewriteRunner;
+import org.apache.iceberg.flink.maintenance.operator.TaskResultAggregator;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * Creates the data file rewriter data stream. Which runs a single iteration of the task for every
+ * {@link Trigger} event.
+ */
+public class RewriteDataFiles {
+ static final String PLANNER_TASK_NAME = "RDF Planner";
+ static final String REWRITE_TASK_NAME = "Rewrite";
+ static final String COMMIT_TASK_NAME = "Rewrite commit";
+ static final String AGGREGATOR_TASK_NAME = "Rewrite aggregator";
+
+ private RewriteDataFiles() {}
+
+ /** Creates the builder for a stream which rewrites data files for the table. */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder extends MaintenanceTaskBuilder<RewriteDataFiles.Builder> {
+ private boolean partialProgressEnabled = false;
+ private int partialProgressMaxCommits = 10;
+ private final Map<String, String> rewriteOptions = Maps.newHashMapWithExpectedSize(6);
+ private long maxRewriteBytes = Long.MAX_VALUE;
+
+ /**
+ * Allows committing compacted data files in batches. See {@link
+ * org.apache.iceberg.actions.RewriteDataFiles#PARTIAL_PROGRESS_ENABLED} for more details.
+ *
+ * @param newPartialProgressEnabled to enable partial commits
+ */
+ public Builder partialProgressEnabled(boolean newPartialProgressEnabled) {
+ this.partialProgressEnabled = newPartialProgressEnabled;
+ return this;
+ }
+
+ /**
+ * Configures the size of batches if {@link #partialProgressEnabled}. See {@link
+ * org.apache.iceberg.actions.RewriteDataFiles#PARTIAL_PROGRESS_MAX_COMMITS} for more details.
+ *
+ * @param newPartialProgressMaxCommits to target number of the commits per run
+ */
+ public Builder partialProgressMaxCommits(int newPartialProgressMaxCommits) {
+ this.partialProgressMaxCommits = newPartialProgressMaxCommits;
+ return this;
+ }
+
+ /**
+ * Configures the maximum byte size of the rewrites for one scheduled compaction. This could be
+ * used to limit the resources used by the compaction.
+ *
+ * @param newMaxRewriteBytes to limit the size of the rewrites
+ */
+ public Builder maxRewriteBytes(long newMaxRewriteBytes) {
+ this.maxRewriteBytes = newMaxRewriteBytes;
+ return this;
+ }
+
+ /**
+ * Configures the target file size. See {@link
+ * org.apache.iceberg.actions.RewriteDataFiles#TARGET_FILE_SIZE_BYTES} for more details.
+ *
+ * @param targetFileSizeBytes target file size
+ */
+ public Builder targetFileSizeBytes(long targetFileSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.TARGET_FILE_SIZE_BYTES, String.valueOf(targetFileSizeBytes));
+ return this;
+ }
+
+ /**
+ * Configures the min file size considered for rewriting. See {@link
+ * SizeBasedFileRewritePlanner#MIN_FILE_SIZE_BYTES} for more details.
+ *
+ * @param minFileSizeBytes min file size
+ */
+ public Builder minFileSizeBytes(long minFileSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MIN_FILE_SIZE_BYTES, String.valueOf(minFileSizeBytes));
+ return this;
+ }
+
+ /**
+ * Configures the max file size considered for rewriting. See {@link
+ * SizeBasedFileRewritePlanner#MAX_FILE_SIZE_BYTES} for more details.
+ *
+ * @param maxFileSizeBytes max file size
+ */
+ public Builder maxFileSizeBytes(long maxFileSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MAX_FILE_SIZE_BYTES, String.valueOf(maxFileSizeBytes));
+ return this;
+ }
+
+ /**
+ * Configures the minimum file number after a rewrite is always initiated. See description see
+ * {@link SizeBasedFileRewritePlanner#MIN_INPUT_FILES} for more details.
+ *
+ * @param minInputFiles min file number
+ */
+ public Builder minInputFiles(int minInputFiles) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MIN_INPUT_FILES, String.valueOf(minInputFiles));
+ return this;
+ }
+
+ /**
+ * Configures the minimum delete file number for a file after a rewrite is always initiated. See
+ * {@link BinPackRewriteFilePlanner#DELETE_FILE_THRESHOLD} for more details.
+ *
+ * @param deleteFileThreshold min delete file number
+ */
+ public Builder deleteFileThreshold(int deleteFileThreshold) {
+ this.rewriteOptions.put(
+ BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD, String.valueOf(deleteFileThreshold));
+ return this;
+ }
+
+ /**
+ * Overrides other options and forces rewriting of all provided files.
+ *
+ * @param rewriteAll enables a full rewrite
+ */
+ public Builder rewriteAll(boolean rewriteAll) {
+ this.rewriteOptions.put(SizeBasedFileRewritePlanner.REWRITE_ALL, String.valueOf(rewriteAll));
+ return this;
+ }
+
+ /**
+ * Configures the group size for rewriting. See {@link
+ * SizeBasedFileRewritePlanner#MAX_FILE_GROUP_SIZE_BYTES} for more details.
+ *
+ * @param maxFileGroupSizeBytes file group size for rewrite
+ */
+ public Builder maxFileGroupSizeBytes(long maxFileGroupSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ String.valueOf(maxFileGroupSizeBytes));
+ return this;
+ }
+
+ /**
+ * The input is a {@link DataStream} with {@link Trigger} events and every event should be
+ * immediately followed by a {@link Watermark} with the same timestamp as the event.
+ *
+ * <p>The output is a {@link DataStream} with the {@link TaskResult} of the run followed by the
+ * {@link Watermark}.
+ */
+ @Override
+ DataStream<TaskResult> append(DataStream<Trigger> trigger) {
+ SingleOutputStreamOperator<DataFileRewritePlanner.PlannedGroup> planned =
+ trigger
+ .process(
+ new DataFileRewritePlanner(
+ tableName(),
+ taskName(),
+ index(),
+ tableLoader(),
+ partialProgressEnabled ? partialProgressMaxCommits : 1,
+ maxRewriteBytes,
+ rewriteOptions))
+ .name(operatorName(PLANNER_TASK_NAME))
+ .uid(PLANNER_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .forceNonParallel();
+
+ SingleOutputStreamOperator<DataFileRewriteRunner.ExecutedGroup> rewritten =
+ planned
+ .rebalance()
+ .process(new DataFileRewriteRunner(tableName(), taskName(), index()))
+ .name(operatorName(REWRITE_TASK_NAME))
+ .uid(REWRITE_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .setParallelism(parallelism());
+
+ SingleOutputStreamOperator<Trigger> updated =
+ rewritten
+ .transform(
+ operatorName(COMMIT_TASK_NAME),
+ TypeInformation.of(Trigger.class),
+ new DataFileRewriteCommitter(tableName(), taskName(), index(), tableLoader()))
+ .uid(COMMIT_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .forceNonParallel();
+
+ return trigger
+ .union(updated)
+ .connect(
+ planned
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .union(
+ rewritten.getSideOutput(TaskResultAggregator.ERROR_STREAM),
+ updated.getSideOutput(TaskResultAggregator.ERROR_STREAM)))
+ .transform(
+ operatorName(AGGREGATOR_TASK_NAME),
+ TypeInformation.of(TaskResult.class),
+ new TaskResultAggregator(tableName(), taskName(), index()))
+ .uid(AGGREGATOR_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .forceNonParallel();
+ }
+ }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java
index 25d0325abb..12230891ac 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.flink.maintenance.api;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
+import java.util.Locale;
import java.util.UUID;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
@@ -257,9 +258,9 @@ public class TableMaintenance {
DataStream<TaskResult> result =
builder.append(
filtered,
- taskIndex,
- taskNames.get(taskIndex),
tableName,
+ taskNames.get(taskIndex),
+ taskIndex,
loader,
uidSuffix,
slotSharingGroup,
@@ -301,7 +302,7 @@ public class TableMaintenance {
private static String nameFor(MaintenanceTaskBuilder<?> streamBuilder, int taskIndex) {
return String.format(
- "%s [%s]", streamBuilder.getClass().getSimpleName(), String.valueOf(taskIndex));
+ Locale.ROOT, "%s [%d]", streamBuilder.getClass().getSimpleName(), taskIndex);
}
}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java
new file mode 100644
index 0000000000..135d3d9b42
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java
@@ -0,0 +1,199 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import java.io.IOException;
+import java.util.Set;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewriteDataFilesCommitManager;
+import org.apache.iceberg.actions.RewriteDataFilesCommitManager.CommitService;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Commits the rewrite changes using {@link RewriteDataFilesCommitManager}. The input is a {@link
+ * DataFileRewriteRunner.ExecutedGroup}. Only {@link Watermark} is emitted which is chained to
+ * {@link TaskResultAggregator} input 1.
+ */
+@Internal
+public class DataFileRewriteCommitter extends AbstractStreamOperator<Trigger>
+ implements OneInputStreamOperator<DataFileRewriteRunner.ExecutedGroup, Trigger> {
+ private static final Logger LOG = LoggerFactory.getLogger(DataFileRewriteCommitter.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+ private final TableLoader tableLoader;
+
+ private transient Table table;
+ private transient CommitService commitService;
+ private transient Counter errorCounter;
+ private transient Counter addedDataFileNumCounter;
+ private transient Counter addedDataFileSizeCounter;
+ private transient Counter removedDataFileNumCounter;
+ private transient Counter removedDataFileSizeCounter;
+
+ public DataFileRewriteCommitter(
+ String tableName, String taskName, int taskIndex, TableLoader tableLoader) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+ Preconditions.checkNotNull(tableLoader, "Table loader should no be null");
+
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ this.tableLoader = tableLoader;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ tableLoader.open();
+ this.table = tableLoader.loadTable();
+
+ MetricGroup taskMetricGroup =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex);
+ this.errorCounter = taskMetricGroup.counter(TableMaintenanceMetrics.ERROR_COUNTER);
+ this.addedDataFileNumCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.ADDED_DATA_FILE_NUM_METRIC);
+ this.addedDataFileSizeCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.ADDED_DATA_FILE_SIZE_METRIC);
+ this.removedDataFileNumCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.REMOVED_DATA_FILE_NUM_METRIC);
+ this.removedDataFileSizeCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.REMOVED_DATA_FILE_SIZE_METRIC);
+ }
+
+ @Override
+ public void processElement(StreamRecord<DataFileRewriteRunner.ExecutedGroup> streamRecord) {
+ DataFileRewriteRunner.ExecutedGroup executedGroup = streamRecord.getValue();
+ try {
+ if (commitService == null) {
+ // Refresh the table to get the latest snapshot for the committer
+ table.refresh();
+
+ FlinkRewriteDataFilesCommitManager commitManager =
+ new FlinkRewriteDataFilesCommitManager(
+ table, executedGroup.snapshotId(), streamRecord.getTimestamp());
+ this.commitService = commitManager.service(executedGroup.groupsPerCommit());
+ commitService.start();
+ }
+
+ commitService.offer(executedGroup.group());
+ } catch (Exception e) {
+ LOG.warn(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception processing {}",
+ tableName,
+ taskName,
+ taskIndex,
+ streamRecord.getTimestamp(),
+ executedGroup,
+ e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ errorCounter.inc();
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ try {
+ if (commitService != null) {
+ commitService.close();
+ }
+
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Successfully completed data file compaction",
+ tableName,
+ taskName,
+ taskIndex,
+ mark.getTimestamp());
+ } catch (Exception e) {
+ LOG.warn(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception closing commit service",
+ tableName,
+ taskName,
+ taskIndex,
+ mark.getTimestamp(),
+ e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ errorCounter.inc();
+ }
+
+ // Cleanup
+ this.commitService = null;
+
+ super.processWatermark(mark);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (commitService != null) {
+ commitService.close();
+ }
+ }
+
+ private class FlinkRewriteDataFilesCommitManager extends RewriteDataFilesCommitManager {
+ private final long timestamp;
+
+ FlinkRewriteDataFilesCommitManager(Table table, long startingSnapshotId, long timestamp) {
+ super(table, startingSnapshotId);
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public void commitFileGroups(Set<RewriteFileGroup> fileGroups) {
+ super.commitFileGroups(fileGroups);
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Committed {}",
+ tableName,
+ taskName,
+ taskIndex,
+ timestamp,
+ fileGroups);
+ updateMetrics(fileGroups);
+ }
+
+ private void updateMetrics(Set<RewriteFileGroup> fileGroups) {
+ for (RewriteFileGroup fileGroup : fileGroups) {
+ for (DataFile added : fileGroup.addedFiles()) {
+ addedDataFileNumCounter.inc();
+ addedDataFileSizeCounter.inc(added.fileSizeInBytes());
+ }
+
+ for (DataFile rewritten : fileGroup.rewrittenFiles()) {
+ removedDataFileNumCounter.inc();
+ removedDataFileSizeCounter.inc(rewritten.fileSizeInBytes());
+ }
+ }
+ }
+ }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
new file mode 100644
index 0000000000..e36c9c6610
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
@@ -0,0 +1,206 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import java.math.RoundingMode;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.SerializableTable;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.actions.FileRewritePlan;
+import org.apache.iceberg.actions.RewriteDataFiles;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.math.IntMath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Plans the rewrite groups using the {@link BinPackRewriteFilePlanner}. The input is the {@link
+ * Trigger}, the output is zero, or some {@link PlannedGroup}s.
+ */
+@Internal
+public class DataFileRewritePlanner
+ extends ProcessFunction<Trigger, DataFileRewritePlanner.PlannedGroup> {
+ static final String MESSAGE_PREFIX = "[For table {} with {}[{}] at {}]: ";
+ private static final Logger LOG = LoggerFactory.getLogger(DataFileRewritePlanner.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+ private final TableLoader tableLoader;
+ private final int partialProgressMaxCommits;
+ private final long maxRewriteBytes;
+ private final Map<String, String> rewriterOptions;
+ private transient Counter errorCounter;
+
+ public DataFileRewritePlanner(
+ String tableName,
+ String taskName,
+ int taskIndex,
+ TableLoader tableLoader,
+ int newPartialProgressMaxCommits,
+ long maxRewriteBytes,
+ Map<String, String> rewriterOptions) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+ Preconditions.checkNotNull(tableLoader, "Table loader should no be null");
+ Preconditions.checkNotNull(rewriterOptions, "Options map should no be null");
+
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ this.tableLoader = tableLoader;
+ this.partialProgressMaxCommits = newPartialProgressMaxCommits;
+ this.maxRewriteBytes = maxRewriteBytes;
+ this.rewriterOptions = rewriterOptions;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ tableLoader.open();
+ this.errorCounter =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
+ .counter(TableMaintenanceMetrics.ERROR_COUNTER);
+ }
+
+ @Override
+ public void processElement(Trigger value, Context ctx, Collector<PlannedGroup> out)
+ throws Exception {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Creating rewrite plan",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp());
+ try {
+ SerializableTable table =
+ (SerializableTable) SerializableTable.copyOf(tableLoader.loadTable());
+ if (table.currentSnapshot() == null) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Nothing to plan for in an empty table",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp());
+ return;
+ }
+
+ BinPackRewriteFilePlanner planner = new BinPackRewriteFilePlanner(table);
+ planner.init(rewriterOptions);
+
+ FileRewritePlan<RewriteDataFiles.FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup>
+ plan = planner.plan();
+
+ long rewriteBytes = 0;
+ List<RewriteFileGroup> groups = Lists.newArrayList();
+ for (CloseableIterator<RewriteFileGroup> groupIterator = plan.groups().iterator();
+ groupIterator.hasNext(); ) {
+ RewriteFileGroup group = groupIterator.next();
+ if (rewriteBytes + group.inputFilesSizeInBytes() > maxRewriteBytes) {
+ // Keep going, maybe some other group might fit in
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Skipping group as max rewrite size reached {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ group);
+ } else {
+ rewriteBytes += group.inputFilesSizeInBytes();
+ groups.add(group);
+ }
+ }
+
+ int groupsPerCommit =
+ IntMath.divide(groups.size(), partialProgressMaxCommits, RoundingMode.CEILING);
+
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewrite plan created {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ groups);
+
+ for (RewriteFileGroup group : groups) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Emitting {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ group);
+ out.collect(new PlannedGroup(table, groupsPerCommit, group));
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Failed to plan data file rewrite groups",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ e);
+ ctx.output(TaskResultAggregator.ERROR_STREAM, e);
+ errorCounter.inc();
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ tableLoader.close();
+ }
+
+ public static class PlannedGroup {
+ private final SerializableTable table;
+ private final int groupsPerCommit;
+ private final RewriteFileGroup group;
+
+ private PlannedGroup(SerializableTable table, int groupsPerCommit, RewriteFileGroup group) {
+ this.table = table;
+ this.groupsPerCommit = groupsPerCommit;
+ this.group = group;
+ }
+
+ SerializableTable table() {
+ return table;
+ }
+
+ int groupsPerCommit() {
+ return groupsPerCommit;
+ }
+
+ RewriteFileGroup group() {
+ return group;
+ }
+ }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java
new file mode 100644
index 0000000000..c03b5cc1c8
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java
@@ -0,0 +1,253 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING;
+
+import java.util.Collections;
+import java.util.Set;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+import org.apache.iceberg.BaseCombinedScanTask;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewritePlanner.PlannedGroup;
+import org.apache.iceberg.flink.sink.RowDataTaskWriterFactory;
+import org.apache.iceberg.flink.sink.TaskWriterFactory;
+import org.apache.iceberg.flink.source.DataIterator;
+import org.apache.iceberg.flink.source.FileScanTaskReader;
+import org.apache.iceberg.flink.source.RowDataFileScanTaskReader;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Executes a rewrite for a single {@link PlannedGroup}. Reads the files with the standard {@link
+ * FileScanTaskReader}, so the delete files are considered, and writes using the {@link
+ * TaskWriterFactory}. The output is an {@link ExecutedGroup}.
+ */
+@Internal
+public class DataFileRewriteRunner
+ extends ProcessFunction<PlannedGroup, DataFileRewriteRunner.ExecutedGroup> {
+ private static final Logger LOG = LoggerFactory.getLogger(DataFileRewriteRunner.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+
+ private transient int subTaskId;
+ private transient int attemptId;
+ private transient Counter errorCounter;
+
+ public DataFileRewriteRunner(String tableName, String taskName, int taskIndex) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ }
+
+ @Override
+ public void open(Configuration parameters) {
+ this.errorCounter =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
+ .counter(TableMaintenanceMetrics.ERROR_COUNTER);
+
+ this.subTaskId = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+ this.attemptId = getRuntimeContext().getTaskInfo().getAttemptNumber();
+ }
+
+ @Override
+ public void processElement(PlannedGroup value, Context ctx, Collector<ExecutedGroup> out)
+ throws Exception {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewriting files for group {} with files: {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().info(),
+ value.group().rewrittenFiles());
+ } else {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Rewriting files for group {} with {} number of files",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().info(),
+ value.group().rewrittenFiles().size());
+ }
+
+ try (TaskWriter<RowData> writer = writerFor(value)) {
+ try (DataIterator<RowData> iterator = readerFor(value)) {
+ while (iterator.hasNext()) {
+ writer.write(iterator.next());
+ }
+
+ Set<DataFile> dataFiles = Sets.newHashSet(writer.dataFiles());
+ value.group().setOutputFiles(dataFiles);
+ out.collect(
+ new ExecutedGroup(
+ value.table().currentSnapshot().snapshotId(),
+ value.groupsPerCommit(),
+ value.group()));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewritten files {} from {} to {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().info(),
+ value.group().rewrittenFiles(),
+ value.group().addedFiles());
+ } else {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewritten {} files to {} files",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().rewrittenFiles().size(),
+ value.group().addedFiles().size());
+ }
+ } catch (Exception ex) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception rewriting datafile group {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group(),
+ ex);
+ ctx.output(TaskResultAggregator.ERROR_STREAM, ex);
+ errorCounter.inc();
+ abort(writer, ctx.timestamp());
+ }
+ } catch (Exception ex) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Exception creating compaction writer for group {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group(),
+ ex);
+ ctx.output(TaskResultAggregator.ERROR_STREAM, ex);
+ errorCounter.inc();
+ }
+ }
+
+ private TaskWriter<RowData> writerFor(PlannedGroup value) {
+ String formatString =
+ PropertyUtil.propertyAsString(
+ value.table().properties(),
+ TableProperties.DEFAULT_FILE_FORMAT,
+ TableProperties.DEFAULT_FILE_FORMAT_DEFAULT);
+ RowDataTaskWriterFactory factory =
+ new RowDataTaskWriterFactory(
+ value.table(),
+ FlinkSchemaUtil.convert(value.table().schema()),
+ value.group().inputSplitSize(),
+ FileFormat.fromString(formatString),
+ value.table().properties(),
+ null,
+ false);
+ factory.initialize(subTaskId, attemptId);
+ return factory.create();
+ }
+
+ private DataIterator<RowData> readerFor(PlannedGroup value) {
+ RowDataFileScanTaskReader reader =
+ new RowDataFileScanTaskReader(
+ value.table().schema(),
+ value.table().schema(),
+ PropertyUtil.propertyAsString(value.table().properties(), DEFAULT_NAME_MAPPING, null),
+ false,
+ Collections.emptyList());
+ return new DataIterator<>(
+ reader,
+ new BaseCombinedScanTask(value.group().fileScanTasks()),
+ value.table().io(),
+ value.table().encryption());
+ }
+
+ private void abort(TaskWriter<RowData> writer, long timestamp) {
+ try {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Aborting rewrite for (subTaskId {}, attemptId {})",
+ tableName,
+ taskName,
+ taskIndex,
+ timestamp,
+ subTaskId,
+ attemptId);
+ writer.abort();
+ } catch (Exception e) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception in abort",
+ tableName,
+ taskName,
+ taskIndex,
+ timestamp,
+ e);
+ }
+ }
+
+ public static class ExecutedGroup {
+ private final long snapshotId;
+ private final int groupsPerCommit;
+ private final RewriteFileGroup group;
+
+ @VisibleForTesting
+ ExecutedGroup(long snapshotId, int groupsPerCommit, RewriteFileGroup group) {
+ this.snapshotId = snapshotId;
+ this.groupsPerCommit = groupsPerCommit;
+ this.group = group;
+ }
+
+ long snapshotId() {
+ return snapshotId;
+ }
+
+ int groupsPerCommit() {
+ return groupsPerCommit;
+ }
+
+ RewriteFileGroup group() {
+ return group;
+ }
+ }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java
index dc7846c4c4..9189f5f018 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.flink.maintenance.operator;
import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -40,17 +41,17 @@ public class DeleteFilesProcessor extends AbstractStreamOperator<Void>
implements OneInputStreamOperator<String, Void> {
private static final Logger LOG = LoggerFactory.getLogger(DeleteFilesProcessor.class);
- private final String taskIndex;
+ private final String tableName;
private final String taskName;
+ private final int taskIndex;
private final SupportsBulkOperations io;
- private final String tableName;
private final Set<String> filesToDelete = Sets.newHashSet();
private final int batchSize;
private transient Counter failedCounter;
private transient Counter succeededCounter;
- public DeleteFilesProcessor(int taskIndex, String taskName, Table table, int batchSize) {
+ public DeleteFilesProcessor(Table table, String taskName, int taskIndex, int batchSize) {
Preconditions.checkNotNull(taskName, "Task name should no be null");
Preconditions.checkNotNull(table, "Table should no be null");
@@ -60,31 +61,21 @@ public class DeleteFilesProcessor extends AbstractStreamOperator<Void>
"%s doesn't support bulk delete",
fileIO.getClass().getSimpleName());
- this.taskIndex = String.valueOf(taskIndex);
+ this.tableName = table.name();
this.taskName = taskName;
+ this.taskIndex = taskIndex;
this.io = (SupportsBulkOperations) fileIO;
- this.tableName = table.name();
this.batchSize = batchSize;
}
@Override
public void open() throws Exception {
+ MetricGroup taskMetricGroup =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex);
this.failedCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, taskName)
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, taskIndex)
- .counter(TableMaintenanceMetrics.DELETE_FILE_FAILED_COUNTER);
+ taskMetricGroup.counter(TableMaintenanceMetrics.DELETE_FILE_FAILED_COUNTER);
this.succeededCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, taskName)
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, taskIndex)
- .counter(TableMaintenanceMetrics.DELETE_FILE_SUCCEEDED_COUNTER);
+ taskMetricGroup.counter(TableMaintenanceMetrics.DELETE_FILE_SUCCEEDED_COUNTER);
}
@Override
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java
index a09d0244e9..098d32e3b6 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java
@@ -74,7 +74,7 @@ public class ExpireSnapshotsProcessor extends ProcessFunction<Trigger, TaskResul
this.table = tableLoader.loadTable();
this.plannerPool =
plannerPoolSize != null
- ? ThreadPools.newWorkerPool(table.name() + "-table--planner", plannerPoolSize)
+ ? ThreadPools.newFixedThreadPool(table.name() + "-table--planner", plannerPoolSize)
: ThreadPools.getWorkerPool();
}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
index 49fa69b487..2066ca8e01 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -98,31 +99,16 @@ public class LockRemover extends AbstractStreamOperator<Void>
this.failedTaskResultCounters = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
this.taskLastRunDurationMs = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
for (int taskIndex = 0; taskIndex < maintenanceTaskNames.size(); ++taskIndex) {
+ MetricGroup taskMetricGroup =
+ TableMaintenanceMetrics.groupFor(
+ getRuntimeContext(), tableName, maintenanceTaskNames.get(taskIndex), taskIndex);
succeededTaskResultCounters.add(
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
- .counter(TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER));
+ taskMetricGroup.counter(TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER));
failedTaskResultCounters.add(
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
- .counter(TableMaintenanceMetrics.FAILED_TASK_COUNTER));
+ taskMetricGroup.counter(TableMaintenanceMetrics.FAILED_TASK_COUNTER));
AtomicLong duration = new AtomicLong(0);
taskLastRunDurationMs.add(duration);
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
- .gauge(TableMaintenanceMetrics.LAST_RUN_DURATION_MS, duration::get);
+ taskMetricGroup.gauge(TableMaintenanceMetrics.LAST_RUN_DURATION_MS, duration::get);
}
lockFactory.open();
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java
new file mode 100644
index 0000000000..8bdcd7ba2b
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java
@@ -0,0 +1,26 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+class LogUtil {
+ static final String MESSAGE_PREFIX = "[For table {} with {}[{}] at {}]: ";
+ static final String MESSAGE_FORMAT_PREFIX = "[For table %s with {%s}[{%d}] at {%d}]: ";
+
+ private LogUtil() {}
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
index 6147c3a5fd..897760caaa 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
@@ -18,12 +18,18 @@
*/
package org.apache.iceberg.flink.maintenance.operator;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.metrics.MetricGroup;
+
public class TableMaintenanceMetrics {
public static final String GROUP_KEY = "maintenance";
public static final String TASK_NAME_KEY = "taskName";
public static final String TASK_INDEX_KEY = "taskIndex";
public static final String TABLE_NAME_KEY = "tableName";
+ // Operator error counter
+ public static final String ERROR_COUNTER = "error";
+
// TriggerManager metrics
public static final String RATE_LIMITER_TRIGGERED = "rateLimiterTriggered";
public static final String CONCURRENT_RUN_THROTTLED = "concurrentRunThrottled";
@@ -39,6 +45,30 @@ public class TableMaintenanceMetrics {
public static final String DELETE_FILE_FAILED_COUNTER = "deleteFailed";
public static final String DELETE_FILE_SUCCEEDED_COUNTER = "deleteSucceeded";
+ // DataFileUpdater metrics
+ public static final String ADDED_DATA_FILE_NUM_METRIC = "addedDataFileNum";
+ public static final String ADDED_DATA_FILE_SIZE_METRIC = "addedDataFileSize";
+ public static final String REMOVED_DATA_FILE_NUM_METRIC = "removedDataFileNum";
+ public static final String REMOVED_DATA_FILE_SIZE_METRIC = "removedDataFileSize";
+
+ static MetricGroup groupFor(
+ RuntimeContext context, String tableName, String taskName, int taskIndex) {
+ return groupFor(groupFor(context, tableName), taskName, taskIndex);
+ }
+
+ static MetricGroup groupFor(RuntimeContext context, String tableName) {
+ return context
+ .getMetricGroup()
+ .addGroup(TableMaintenanceMetrics.GROUP_KEY)
+ .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName);
+ }
+
+ static MetricGroup groupFor(MetricGroup mainGroup, String taskName, int taskIndex) {
+ return mainGroup
+ .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, taskName)
+ .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex));
+ }
+
private TableMaintenanceMetrics() {
// do not instantiate
}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java
new file mode 100644
index 0000000000..cceb043a26
--- /dev/null
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java
@@ -0,0 +1,101 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import java.util.List;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.iceberg.flink.maintenance.api.TaskResult;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Aggregates results of the operators for a given maintenance task.
+ *
+ * <ul>
+ * <li>Input 1 is used:
+ * <ul>
+ * <li>To provide the {@link TaskResult#startEpoch()} - should be chained to the task input
+ * <li>To mark that the task is finished - should be chained at the end of the task, so an
+ * incoming watermark will signal that the task is finished
+ * </ul>
+ * <li>Input 2 expects an {@link Exception} which caused the failure - should be chained to the
+ * {@link #ERROR_STREAM} of the operators
+ * </ul>
+ *
+ * The operator emits a {@link TaskResult} with the overall result on {@link Watermark}.
+ */
+@Internal
+public class TaskResultAggregator extends AbstractStreamOperator<TaskResult>
+ implements TwoInputStreamOperator<Trigger, Exception, TaskResult> {
+ public static final OutputTag<Exception> ERROR_STREAM =
+ new OutputTag<>("error-stream", TypeInformation.of(Exception.class));
+
+ private static final Logger LOG = LoggerFactory.getLogger(TaskResultAggregator.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+ private final List<Exception> exceptions;
+ private transient Long startTime;
+
+ public TaskResultAggregator(String tableName, String taskName, int taskIndex) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ this.exceptions = Lists.newArrayList();
+ this.startTime = 0L;
+ }
+
+ @Override
+ public void processElement1(StreamRecord<Trigger> streamRecord) {
+ startTime = streamRecord.getValue().timestamp();
+ }
+
+ @Override
+ public void processElement2(StreamRecord<Exception> streamRecord) {
+ Preconditions.checkNotNull(streamRecord.getValue(), "Exception could not be `null`.");
+ exceptions.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {
+ TaskResult response = new TaskResult(taskIndex, startTime, exceptions.isEmpty(), exceptions);
+ output.collect(new StreamRecord<>(response));
+ LOG.info(
+ "Aggregated result for table {}, task {}[{}] is {}",
+ tableName,
+ taskName,
+ taskIndex,
+ response);
+ exceptions.clear();
+ startTime = 0L;
+ }
+}
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
index d6d1ebe397..bd8424d726 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.TimerService;
@@ -121,33 +122,17 @@ public class TriggerManager extends KeyedProcessFunction<Boolean, TableChange, T
@Override
public void open(Configuration parameters) throws Exception {
+ MetricGroup mainGroup = TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName);
this.rateLimiterTriggeredCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .counter(TableMaintenanceMetrics.RATE_LIMITER_TRIGGERED);
+ mainGroup.counter(TableMaintenanceMetrics.RATE_LIMITER_TRIGGERED);
this.concurrentRunThrottledCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .counter(TableMaintenanceMetrics.CONCURRENT_RUN_THROTTLED);
- this.nothingToTriggerCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .counter(TableMaintenanceMetrics.NOTHING_TO_TRIGGER);
+ mainGroup.counter(TableMaintenanceMetrics.CONCURRENT_RUN_THROTTLED);
+ this.nothingToTriggerCounter = mainGroup.counter(TableMaintenanceMetrics.NOTHING_TO_TRIGGER);
this.triggerCounters = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
for (int taskIndex = 0; taskIndex < maintenanceTaskNames.size(); ++taskIndex) {
triggerCounters.add(
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
+ TableMaintenanceMetrics.groupFor(
+ mainGroup, maintenanceTaskNames.get(taskIndex), taskIndex)
.counter(TableMaintenanceMetrics.TRIGGERED));
}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java
index 1b69be1fa3..10efb9120c 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java
@@ -48,16 +48,16 @@ class MaintenanceTaskInfraExtension implements BeforeEachCallback {
@Override
public void beforeEach(ExtensionContext context) {
- env = StreamExecutionEnvironment.getExecutionEnvironment();
- source = new ManualSource<>(env, TypeInformation.of(Trigger.class));
+ this.env = StreamExecutionEnvironment.getExecutionEnvironment();
+ this.source = new ManualSource<>(env, TypeInformation.of(Trigger.class));
// Adds the watermark to mimic the behaviour expected for the input of the maintenance tasks
- triggerStream =
+ this.triggerStream =
source
.dataStream()
.assignTimestampsAndWatermarks(new TableMaintenance.PunctuatedWatermarkStrategy())
.name(IGNORED_OPERATOR_NAME)
.forceNonParallel();
- sink = new CollectingSink<>();
+ this.sink = new CollectingSink<>();
}
StreamExecutionEnvironment env() {
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java
index 36041d9c38..6d3a10c4b1 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java
@@ -37,16 +37,42 @@ class MaintenanceTaskTestBase extends OperatorTestBase {
@RegisterExtension MaintenanceTaskInfraExtension infra = new MaintenanceTaskInfraExtension();
void runAndWaitForSuccess(
+ StreamExecutionEnvironment env,
+ ManualSource<Trigger> triggerSource,
+ CollectingSink<TaskResult> collectingSink)
+ throws Exception {
+ runAndWaitForResult(env, triggerSource, collectingSink, false, () -> true);
+ }
+
+ void runAndWaitForSuccess(
+ StreamExecutionEnvironment env,
+ ManualSource<Trigger> triggerSource,
+ CollectingSink<TaskResult> collectingSink,
+ Supplier<Boolean> waitForCondition)
+ throws Exception {
+ runAndWaitForResult(env, triggerSource, collectingSink, false, waitForCondition);
+ }
+
+ void runAndWaitForFailure(
+ StreamExecutionEnvironment env,
+ ManualSource<Trigger> triggerSource,
+ CollectingSink<TaskResult> collectingSink)
+ throws Exception {
+ runAndWaitForResult(env, triggerSource, collectingSink, true, () -> true);
+ }
+
+ void runAndWaitForResult(
StreamExecutionEnvironment env,
ManualSource<Trigger> triggerSource,
CollectingSink<TaskResult> collectingSink,
+ boolean generateFailure,
Supplier<Boolean> waitForCondition)
throws Exception {
JobClient jobClient = null;
try {
jobClient = env.executeAsync();
- // Do a single task run
+ // Do a single successful task run
long time = System.currentTimeMillis();
triggerSource.sendRecord(Trigger.create(time, TESTING_TASK_ID), time);
@@ -56,6 +82,17 @@ class MaintenanceTaskTestBase extends OperatorTestBase {
assertThat(result.success()).isTrue();
assertThat(result.taskIndex()).isEqualTo(TESTING_TASK_ID);
+ if (generateFailure) {
+ dropTable();
+ time = System.currentTimeMillis();
+ triggerSource.sendRecord(Trigger.create(time, TESTING_TASK_ID), time);
+ result = collectingSink.poll(POLL_DURATION);
+
+ assertThat(result.startEpoch()).isEqualTo(time);
+ assertThat(result.success()).isFalse();
+ assertThat(result.taskIndex()).isEqualTo(TESTING_TASK_ID);
+ }
+
Awaitility.await().until(waitForCondition::get);
} finally {
closeJobClient(jobClient);
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java
index f80129f966..b8aa259e2f 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java
@@ -27,7 +27,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.List;
import java.util.Set;
-import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
@@ -69,9 +68,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
.uidSuffix(UID_SUFFIX)
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
"OTHER",
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
@@ -101,36 +100,16 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
ExpireSnapshots.builder()
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
1)
.sinkTo(infra.sink());
- JobClient jobClient = null;
- try {
- jobClient = infra.env().executeAsync();
-
- // Do a single task run
- long time = System.currentTimeMillis();
- infra.source().sendRecord(Trigger.create(time, 1), time);
-
- // First successful run (ensure that the operators are loaded/opened etc.)
- assertThat(infra.sink().poll(Duration.ofSeconds(5)).success()).isTrue();
-
- // Drop the table, so it will cause an exception
- dropTable();
-
- // Failed run
- infra.source().sendRecord(Trigger.create(time + 1, 1), time + 1);
-
- assertThat(infra.sink().poll(Duration.ofSeconds(5)).success()).isFalse();
- } finally {
- closeJobClient(jobClient);
- }
+ runAndWaitForFailure(infra.env(), infra.source(), infra.sink());
// Check the metrics. There are no expired snapshots or data files because ExpireSnapshots has
// no max age of number of snapshots set, so no files are removed.
@@ -162,9 +141,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
.uidSuffix(UID_SUFFIX)
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
@@ -180,9 +159,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
ExpireSnapshots.builder()
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
@@ -204,9 +183,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
.parallelism(1)
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java
index 467ad2d8ce..0a860fec47 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java
@@ -52,6 +52,19 @@ class TestMaintenanceE2E extends OperatorTestBase {
.retainLast(5)
.deleteBatchSize(5)
.parallelism(8))
+ .add(
+ RewriteDataFiles.builder()
+ .scheduleOnDataFileCount(10)
+ .partialProgressEnabled(true)
+ .partialProgressMaxCommits(10)
+ .maxRewriteBytes(1000L)
+ .targetFileSizeBytes(1000L)
+ .minFileSizeBytes(1000L)
+ .maxFileSizeBytes(1000L)
+ .minInputFiles(10)
+ .deleteFileThreshold(10)
+ .rewriteAll(false)
+ .maxFileGroupSizeBytes(1000L))
.append();
JobClient jobClient = null;
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
new file mode 100644
index 0000000000..618647259b
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
@@ -0,0 +1,417 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
+import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.COMMIT_TASK_NAME;
+import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.PLANNER_TASK_NAME;
+import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.REWRITE_TASK_NAME;
+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.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;
+
+import java.util.List;
+import java.util.stream.StreamSupport;
+import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.SimpleDataUtil;
+import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Test;
+
+class TestRewriteDataFiles extends MaintenanceTaskTestBase {
+ @Test
+ void testRewriteUnpartitioned() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+ insert(table, 3, "c");
+ insert(table, 4, "d");
+
+ assertFileNum(table, 4, 0);
+
+ appendRewriteDataFiles(
+ RewriteDataFiles.builder()
+ .parallelism(2)
+ .deleteFileThreshold(10)
+ .targetFileSizeBytes(1_000_000L)
+ .maxFileGroupSizeBytes(10_000_000L)
+ .maxFileSizeBytes(2_000_000L)
+ .minFileSizeBytes(500_000L)
+ .minInputFiles(2)
+ .partialProgressEnabled(true)
+ .partialProgressMaxCommits(1)
+ .maxRewriteBytes(100_000L)
+ .rewriteAll(false));
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ assertFileNum(table, 1, 0);
+
+ SimpleDataUtil.assertTableRecords(
+ table,
+ ImmutableList.of(
+ createRecord(1, "a"),
+ createRecord(2, "b"),
+ createRecord(3, "c"),
+ createRecord(4, "d")));
+ }
+
+ @Test
+ void testRewritePartitioned() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+
+ assertFileNum(table, 4, 0);
+
+ appendRewriteDataFiles();
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ assertFileNum(table, 2, 0);
+
+ SimpleDataUtil.assertTableRecords(
+ table,
+ ImmutableList.of(
+ createRecord(1, "p1"),
+ createRecord(2, "p1"),
+ createRecord(3, "p2"),
+ createRecord(4, "p2")));
+ }
+
+ @Test
+ void testPlannerFailure() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ assertFileNum(table, 2, 0);
+
+ appendRewriteDataFiles();
+
+ runAndWaitForFailure(infra.env(), infra.source(), infra.sink());
+
+ // Check the metrics. The first task should be successful, but the second one should fail. This
+ // should be represented in the counters.
+ MetricsReporterFactoryForTests.assertCounters(
+ new ImmutableMap.Builder<List<String>, Long>()
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 1L)
+ .put(
+ ImmutableList.of(
+ REWRITE_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_NUM_METRIC),
+ 1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_NUM_METRIC),
+ 2L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .build());
+ }
+
+ @Test
+ void testUidAndSlotSharingGroup() {
+ createTable();
+
+ RewriteDataFiles.builder()
+ .slotSharingGroup(SLOT_SHARING_GROUP)
+ .uidSuffix(UID_SUFFIX)
+ .append(
+ infra.triggerStream(),
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
+ tableLoader(),
+ "OTHER",
+ "OTHER",
+ 1)
+ .sinkTo(infra.sink());
+
+ checkUidsAreSet(infra.env(), UID_SUFFIX);
+ checkSlotSharingGroupsAreSet(infra.env(), SLOT_SHARING_GROUP);
+ }
+
+ @Test
+ void testUidAndSlotSharingGroupUnset() {
+ createTable();
+
+ RewriteDataFiles.builder()
+ .append(
+ infra.triggerStream(),
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
+ tableLoader(),
+ UID_SUFFIX,
+ StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
+ 1)
+ .sinkTo(infra.sink());
+
+ checkUidsAreSet(infra.env(), null);
+ checkSlotSharingGroupsAreSet(infra.env(), null);
+ }
+
+ @Test
+ void testMetrics() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ assertFileNum(table, 2, 0);
+
+ appendRewriteDataFiles();
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ // Check the metrics
+ MetricsReporterFactoryForTests.assertCounters(
+ new ImmutableMap.Builder<List<String>, Long>()
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ REWRITE_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_NUM_METRIC),
+ 1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_NUM_METRIC),
+ 2L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .build());
+ }
+
+ @Test
+ void testV2Table() throws Exception {
+ Table table = createTableWithDelete();
+ update(table, 1, null, "a", "b");
+ update(table, 1, "b", "c");
+
+ assertFileNum(table, 2, 3);
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of(createRecord(1, "c")));
+
+ appendRewriteDataFiles();
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ // After #11131 we don't remove the delete files
+ assertFileNum(table, 1, 3);
+
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of(createRecord(1, "c")));
+
+ // Check the metrics
+ MetricsReporterFactoryForTests.assertCounters(
+ new ImmutableMap.Builder<List<String>, Long>()
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ REWRITE_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_NUM_METRIC),
+ 1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_NUM_METRIC),
+ 2L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .build());
+ }
+
+ private void appendRewriteDataFiles() {
+ appendRewriteDataFiles(RewriteDataFiles.builder().rewriteAll(true));
+ }
+
+ private void appendRewriteDataFiles(RewriteDataFiles.Builder builder) {
+ builder
+ .append(
+ infra.triggerStream(),
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
+ tableLoader(),
+ UID_SUFFIX,
+ StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
+ 1)
+ .sinkTo(infra.sink());
+ }
+
+ private static void assertFileNum(
+ Table table, int expectedDataFileNum, int expectedDeleteFileNum) {
+ table.refresh();
+ assertThat(
+ table.currentSnapshot().dataManifests(table.io()).stream()
+ .flatMap(
+ m ->
+ StreamSupport.stream(
+ ManifestFiles.read(m, table.io(), table.specs()).spliterator(), false))
+ .count())
+ .isEqualTo(expectedDataFileNum);
+ assertThat(
+ table.currentSnapshot().deleteManifests(table.io()).stream()
+ .flatMap(
+ m ->
+ StreamSupport.stream(
+ ManifestFiles.readDeleteManifest(m, table.io(), table.specs())
+ .spliterator(),
+ false))
+ .count())
+ .isEqualTo(expectedDeleteFileNum);
+ }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
index a577e40b2c..4a4bceffb1 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
@@ -33,16 +33,26 @@ import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.flink.streaming.api.transformations.SinkTransformation;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TestHelpers;
+import org.apache.iceberg.data.FileHelpers;
import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.flink.HadoopCatalogExtension;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
import org.apache.iceberg.flink.maintenance.api.TriggerLockFactory;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -126,12 +136,85 @@ public class OperatorTestBase {
ImmutableMap.of("format-version", "2", "write.upsert.enabled", "true"));
}
+ protected static Table createPartitionedTable() {
+ return CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TestFixtures.TABLE_IDENTIFIER,
+ SimpleDataUtil.SCHEMA,
+ PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build(),
+ null,
+ ImmutableMap.of("flink.max-continuous-empty-commits", "100000"));
+ }
+
protected void insert(Table table, Integer id, String data) throws IOException {
new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
.appendToTable(Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
table.refresh();
}
+ /**
+ * For the same identifier column id this methods simulate the following row operations: <tr>
+ * <li>add an equality delete on oldData
+ * <li>insert newData </tr>
+ *
+ * @param table to modify
+ * @param id the identifier column id
+ * @param oldData the old data to be deleted
+ * @param newData the new data to be inserted
+ */
+ protected void update(Table table, Integer id, String oldData, String newData)
+ throws IOException {
+ DataFile dataFile =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .writeFile(Lists.newArrayList(SimpleDataUtil.createRecord(id, newData)));
+ DeleteFile eqDelete = writeEqualityDelete(table, id, oldData);
+
+ table.newRowDelta().addRows(dataFile).addDeletes(eqDelete).commit();
+ }
+
+ /**
+ * For the same identifier column id this methods simulate the following row operations: <tr>
+ * <li>add an equality delete on oldData
+ * <li>insert tempData
+ * <li>add a position delete on tempData
+ * <li>insert newData </tr>
+ *
+ * @param table to modify
+ * @param id the identifier column id
+ * @param oldData the old data to be deleted
+ * @param tempData the temp data to be inserted and deleted with a position delete
+ * @param newData the new data to be inserted
+ */
+ protected void update(Table table, Integer id, String oldData, String tempData, String newData)
+ throws IOException {
+ DataFile dataFile =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .writeFile(
+ Lists.newArrayList(
+ SimpleDataUtil.createRecord(id, tempData),
+ SimpleDataUtil.createRecord(id, newData)));
+ DeleteFile eqDelete = writeEqualityDelete(table, id, oldData);
+ DeleteFile posDelete = writePosDelete(table, dataFile.path(), 0, id, tempData);
+
+ table.newRowDelta().addRows(dataFile).addDeletes(eqDelete).addDeletes(posDelete).commit();
+ }
+
+ protected void insertPartitioned(Table table, Integer id, String data) throws IOException {
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .appendToTable(
+ TestHelpers.Row.of(data), Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
+ table.refresh();
+ }
+
+ protected void insertFullPartitioned(Table table, Integer id, String data) throws IOException {
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .appendToTable(
+ TestHelpers.Row.of(data, id),
+ Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
+ table.refresh();
+ }
+
protected void dropTable() {
CATALOG_EXTENSION.catalogLoader().loadCatalog().dropTable(TestFixtures.TABLE_IDENTIFIER);
}
@@ -214,6 +297,36 @@ public class OperatorTestBase {
return config;
}
+ private DeleteFile writeEqualityDelete(Table table, Integer id, String oldData)
+ throws IOException {
+ File file = File.createTempFile("junit", null, warehouseDir.toFile());
+ assertThat(file.delete()).isTrue();
+ return FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(file),
+ new PartitionData(PartitionSpec.unpartitioned().partitionType()),
+ Lists.newArrayList(SimpleDataUtil.createRecord(id, oldData)),
+ SCHEMA_WITH_PRIMARY_KEY);
+ }
+
+ private DeleteFile writePosDelete(
+ Table table, CharSequence path, Integer pos, Integer id, String oldData) throws IOException {
+ File file = File.createTempFile("junit", null, warehouseDir.toFile());
+ assertThat(file.delete()).isTrue();
+ PositionDelete<GenericRecord> posDelete = PositionDelete.create();
+ GenericRecord nested = GenericRecord.create(table.schema());
+ nested.set(0, id);
+ nested.set(1, oldData);
+ posDelete.set(path, pos, nested);
+ return FileHelpers.writePosDeleteFile(
+ table, Files.localOutput(file), null, Lists.newArrayList(posDelete));
+ }
+
+ static void trigger(OneInputStreamOperatorTestHarness<Trigger, ?> harness) throws Exception {
+ long time = System.currentTimeMillis();
+ harness.processElement(Trigger.create(time, 0), time);
+ }
+
private static class MemoryLock implements TriggerLockFactory.Lock {
volatile boolean locked = false;
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java
new file mode 100644
index 0000000000..6e43009e08
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java
@@ -0,0 +1,83 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_FILES;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Set;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+
+class RewriteUtil {
+ private RewriteUtil() {}
+
+ static List<DataFileRewritePlanner.PlannedGroup> planDataFileRewrite(TableLoader tableLoader)
+ throws Exception {
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader,
+ 11,
+ 10_000_000L,
+ ImmutableMap.of(MIN_INPUT_FILES, "2")))) {
+ testHarness.open();
+
+ OperatorTestBase.trigger(testHarness);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ return testHarness.extractOutputValues();
+ }
+ }
+
+ static List<DataFileRewriteRunner.ExecutedGroup> executeRewrite(
+ List<DataFileRewritePlanner.PlannedGroup> elements) throws Exception {
+ try (OneInputStreamOperatorTestHarness<
+ DataFileRewritePlanner.PlannedGroup, DataFileRewriteRunner.ExecutedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewriteRunner(
+ OperatorTestBase.DUMMY_TABLE_NAME, OperatorTestBase.DUMMY_TABLE_NAME, 0))) {
+ testHarness.open();
+
+ for (DataFileRewritePlanner.PlannedGroup element : elements) {
+ testHarness.processElement(element, System.currentTimeMillis());
+ }
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ return testHarness.extractOutputValues();
+ }
+ }
+
+ static Set<DataFile> newDataFiles(Table table) {
+ table.refresh();
+ return Sets.newHashSet(table.currentSnapshot().addedDataFiles(table.io()));
+ }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java
new file mode 100644
index 0000000000..9e8f2ec921
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java
@@ -0,0 +1,278 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.executeRewrite;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
+import static org.apache.iceberg.metrics.CommitMetricsResult.TOTAL_DATA_FILES;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.Test;
+
+class TestDataFileRewriteCommitter extends OperatorTestBase {
+ @Test
+ void testUnpartitioned() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "p1");
+ insert(table, 2, "p2");
+ insert(table, 3, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(1);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(rewritten.get(0), EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+
+ assertDataFiles(
+ table, rewritten.get(0).group().addedFiles(), rewritten.get(0).group().rewrittenFiles(), 1);
+ }
+
+ @Test
+ void testPartitioned() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(2);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(2);
+ assertThat(rewritten.get(0).groupsPerCommit()).isEqualTo(1);
+ assertThat(rewritten.get(1).groupsPerCommit()).isEqualTo(1);
+ ensureDifferentGroups(rewritten);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(rewritten.get(0), EVENT_TIME);
+ assertDataFiles(
+ table,
+ rewritten.get(0).group().addedFiles(),
+ rewritten.get(0).group().rewrittenFiles(),
+ 3);
+
+ testHarness.processElement(rewritten.get(1), EVENT_TIME);
+ assertDataFiles(
+ table,
+ rewritten.get(1).group().addedFiles(),
+ rewritten.get(1).group().rewrittenFiles(),
+ 2);
+
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+ }
+
+ @Test
+ void testNewTable() throws Exception {
+ Table table = createTable();
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten;
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ insert(table, 1, "p1");
+ insert(table, 2, "p2");
+ insert(table, 3, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(1);
+
+ testHarness.processElement(rewritten.get(0), EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+
+ assertDataFiles(
+ table, rewritten.get(0).group().addedFiles(), rewritten.get(0).group().rewrittenFiles(), 1);
+ }
+
+ @Test
+ void testBatchSize() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+ insertPartitioned(table, 5, "p3");
+ insertPartitioned(table, 6, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(3);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(3);
+ ensureDifferentGroups(rewritten);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(0)), EVENT_TIME);
+ assertNoChange(table);
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(1)), EVENT_TIME);
+
+ Set<DataFile> added = Sets.newHashSet(rewritten.get(0).group().addedFiles());
+ added.addAll(rewritten.get(1).group().addedFiles());
+ Set<DataFile> removed = Sets.newHashSet(rewritten.get(0).group().rewrittenFiles());
+ removed.addAll(rewritten.get(1).group().rewrittenFiles());
+ assertDataFiles(table, added, removed, 4);
+
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(2)), EVENT_TIME);
+ assertNoChange(table);
+
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+
+ // This should be committed on close
+ assertDataFiles(
+ table, rewritten.get(2).group().addedFiles(), rewritten.get(2).group().rewrittenFiles(), 3);
+ }
+
+ @Test
+ void testError() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+ insertPartitioned(table, 5, "p3");
+ insertPartitioned(table, 6, "p3");
+ insertPartitioned(table, 7, "p4");
+ insertPartitioned(table, 8, "p4");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(4);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(4);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(0)), EVENT_TIME);
+ assertNoChange(table);
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+
+ DataFileRewriteRunner.ExecutedGroup group = spy(setBatchSizeToTwo(rewritten.get(1)));
+ when(group.group()).thenThrow(new RuntimeException("Testing error"));
+ testHarness.processElement(group, EVENT_TIME);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ assertThat(
+ testHarness
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .poll()
+ .getValue()
+ .getMessage())
+ .contains("Testing error");
+ }
+ }
+
+ private OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger> harness()
+ throws Exception {
+ return new OneInputStreamOperatorTestHarness<>(
+ new DataFileRewriteCommitter(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader()));
+ }
+
+ private static DataFileRewriteRunner.ExecutedGroup setBatchSizeToTwo(
+ DataFileRewriteRunner.ExecutedGroup from) {
+ return new DataFileRewriteRunner.ExecutedGroup(from.snapshotId(), 2, from.group());
+ }
+
+ // Ensure that the groups are different, so the tests are not accidentally passing
+ private static void ensureDifferentGroups(List<DataFileRewriteRunner.ExecutedGroup> rewritten) {
+ List<String> resultFiles =
+ rewritten.stream()
+ .flatMap(task -> task.group().addedFiles().stream().map(ContentFile::location))
+ .collect(Collectors.toList());
+ assertThat(resultFiles).hasSize(Set.copyOf(resultFiles).size());
+ }
+
+ /**
+ * Assert that the number of the data files in the table is as expected. Additionally, tests that
+ * the last commit contains the expected added and removed files.
+ *
+ * @param table the table to check
+ * @param expectedAdded the expected added data files
+ * @param expectedRemoved the expected removed data files
+ * @param expectedCurrent the expected current data files count
+ */
+ private static void assertDataFiles(
+ Table table,
+ Set<DataFile> expectedAdded,
+ Set<DataFile> expectedRemoved,
+ long expectedCurrent) {
+ table.refresh();
+
+ assertThat(table.currentSnapshot().summary().get(TOTAL_DATA_FILES))
+ .isEqualTo(String.valueOf(expectedCurrent));
+ Set<DataFile> actualAdded = Sets.newHashSet(table.currentSnapshot().addedDataFiles(table.io()));
+ Set<DataFile> actualRemoved =
+ Sets.newHashSet(table.currentSnapshot().removedDataFiles(table.io()));
+ assertThat(actualAdded.stream().map(DataFile::location).collect(Collectors.toSet()))
+ .isEqualTo(expectedAdded.stream().map(DataFile::location).collect(Collectors.toSet()));
+ assertThat(actualRemoved.stream().map(DataFile::location).collect(Collectors.toSet()))
+ .isEqualTo(expectedRemoved.stream().map(DataFile::location).collect(Collectors.toSet()));
+ }
+
+ private static void assertNoChange(Table table) {
+ long original = table.currentSnapshot().snapshotId();
+ table.refresh();
+
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(original);
+ }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
new file mode 100644
index 0000000000..1d7e29813a
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
@@ -0,0 +1,193 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_FILES;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.newDataFiles;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.Test;
+
+class TestDataFileRewritePlanner extends OperatorTestBase {
+ @Test
+ void testUnpartitioned() throws Exception {
+ Set<DataFile> expected = Sets.newHashSetWithExpectedSize(3);
+ Table table = createTable();
+ insert(table, 1, "a");
+ expected.addAll(newDataFiles(table));
+ insert(table, 2, "b");
+ expected.addAll(newDataFiles(table));
+ insert(table, 3, "c");
+ expected.addAll(newDataFiles(table));
+
+ List<DataFileRewritePlanner.PlannedGroup> actual = planDataFileRewrite(tableLoader());
+
+ assertThat(actual).hasSize(1);
+ assertRewriteFileGroup(actual.get(0), table, expected);
+ }
+
+ @Test
+ void testPartitioned() throws Exception {
+ Set<DataFile> expectedP1 = Sets.newHashSetWithExpectedSize(2);
+ Set<DataFile> expectedP2 = Sets.newHashSetWithExpectedSize(2);
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ expectedP1.addAll(newDataFiles(table));
+ insertPartitioned(table, 2, "p1");
+ expectedP1.addAll(newDataFiles(table));
+
+ insertPartitioned(table, 3, "p2");
+ expectedP2.addAll(newDataFiles(table));
+ insertPartitioned(table, 4, "p2");
+ expectedP2.addAll(newDataFiles(table));
+
+ // This should not participate in compaction, as there is no more files in the partition
+ insertPartitioned(table, 5, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> actual = planDataFileRewrite(tableLoader());
+
+ assertThat(actual).hasSize(2);
+ if (actual.get(0).group().info().partition().get(0, String.class).equals("p1")) {
+ assertRewriteFileGroup(actual.get(0), table, expectedP1);
+ assertRewriteFileGroup(actual.get(1), table, expectedP2);
+ } else {
+ assertRewriteFileGroup(actual.get(0), table, expectedP2);
+ assertRewriteFileGroup(actual.get(1), table, expectedP1);
+ }
+ }
+
+ @Test
+ void testError() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader(),
+ 11,
+ 1L,
+ ImmutableMap.of(MIN_INPUT_FILES, "2")))) {
+ testHarness.open();
+
+ // Cause an exception
+ dropTable();
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ trigger(testHarness);
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ assertThat(
+ testHarness
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .poll()
+ .getValue()
+ .getMessage())
+ .contains("Table does not exist: ");
+ }
+ }
+
+ @Test
+ void testV2Table() throws Exception {
+ Table table = createTableWithDelete();
+ update(table, 1, null, "a", "b");
+ update(table, 1, "b", "c");
+
+ List<DataFileRewritePlanner.PlannedGroup> actual = planDataFileRewrite(tableLoader());
+
+ assertThat(actual).hasSize(1);
+ List<FileScanTask> tasks = actual.get(0).group().fileScanTasks();
+ assertThat(tasks).hasSize(2);
+ // Find the task with the deletes
+ FileScanTask withDelete = tasks.get(0).deletes().isEmpty() ? tasks.get(1) : tasks.get(0);
+ assertThat(withDelete.deletes()).hasSize(2);
+ assertThat(withDelete.deletes().stream().map(ContentFile::content).collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(FileContent.POSITION_DELETES, FileContent.EQUALITY_DELETES);
+ }
+
+ @Test
+ void testMaxRewriteBytes() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+
+ // First run with high maxRewriteBytes
+ List<DataFileRewritePlanner.PlannedGroup> planWithNoMaxRewriteBytes =
+ planDataFileRewrite(tableLoader());
+ assertThat(planWithNoMaxRewriteBytes).hasSize(2);
+
+ // Second run with low maxRewriteBytes, the 2nd group should be removed from the plan
+ long maxRewriteBytes =
+ planWithNoMaxRewriteBytes.get(0).group().fileScanTasks().get(0).sizeBytes()
+ + planWithNoMaxRewriteBytes.get(1).group().fileScanTasks().get(0).sizeBytes()
+ + 1;
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader(),
+ 11,
+ maxRewriteBytes,
+ ImmutableMap.of(MIN_INPUT_FILES, "2")))) {
+ testHarness.open();
+
+ OperatorTestBase.trigger(testHarness);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ // Only a single group is planned
+ assertThat(testHarness.extractOutputValues()).hasSize(1);
+ }
+ }
+
+ void assertRewriteFileGroup(
+ DataFileRewritePlanner.PlannedGroup plannedGroup, Table table, Set<DataFile> files) {
+ assertThat(plannedGroup.table().currentSnapshot().snapshotId())
+ .isEqualTo(table.currentSnapshot().snapshotId());
+ assertThat(plannedGroup.groupsPerCommit()).isEqualTo(1);
+ assertThat(
+ plannedGroup.group().fileScanTasks().stream()
+ .map(s -> s.file().location())
+ .collect(Collectors.toSet()))
+ .containsExactlyInAnyOrderElementsOf(
+ files.stream().map(ContentFile::location).collect(Collectors.toList()));
+ }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
new file mode 100644
index 0000000000..6dc8fb3c02
--- /dev/null
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
@@ -0,0 +1,355 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.actions.RewriteDataFiles.TARGET_FILE_SIZE_BYTES;
+import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_FILES;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.executeRewrite;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionData;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestDataFileRewriteRunner extends OperatorTestBase {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testExecute(boolean partitioned) throws Exception {
+ Table table;
+ PartitionData partition;
+ if (partitioned) {
+ table = createPartitionedTable();
+ partition = new PartitionData(table.spec().partitionType());
+ partition.set(0, "p1");
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p1");
+ } else {
+ table = createTable();
+ partition = new PartitionData(PartitionSpec.unpartitioned().partitionType());
+ insert(table, 1, "p1");
+ insert(table, 2, "p1");
+ insert(table, 3, "p1");
+ }
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ List<DataFileRewriteRunner.ExecutedGroup> actual = executeRewrite(planned);
+ assertThat(actual).hasSize(1);
+
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(
+ ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"), ImmutableList.of(3, "p1"))),
+ 1,
+ ImmutableSet.of(partition));
+ }
+
+ @Test
+ void testPartitionSpecChange() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ PartitionData oldPartition = new PartitionData(table.spec().partitionType());
+ oldPartition.set(0, "p1");
+
+ try (OneInputStreamOperatorTestHarness<
+ DataFileRewritePlanner.PlannedGroup, DataFileRewriteRunner.ExecutedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewriteRunner(
+ OperatorTestBase.DUMMY_TABLE_NAME, OperatorTestBase.DUMMY_TABLE_NAME, 0))) {
+ testHarness.open();
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+
+ testHarness.processElement(planned.get(0), System.currentTimeMillis());
+ List<DataFileRewriteRunner.ExecutedGroup> actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(1);
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"))),
+ 1,
+ ImmutableSet.of(oldPartition));
+
+ insertPartitioned(table, 3, "p1");
+
+ planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+
+ testHarness.processElement(planned.get(0), System.currentTimeMillis());
+ actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(2);
+ assertRewriteFileGroup(
+ actual.get(1),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(
+ ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"), ImmutableList.of(3, "p1"))),
+ 1,
+ ImmutableSet.of(oldPartition));
+
+ // Alter the table schema
+ table.updateSpec().addField("id").commit();
+ // Insert some now data
+ insertFullPartitioned(table, 4, "p1");
+ insertFullPartitioned(table, 4, "p1");
+ PartitionData newPartition = new PartitionData(table.spec().partitionType());
+ newPartition.set(0, "p1");
+ newPartition.set(1, 4);
+ table.refresh();
+
+ planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(2);
+ DataFileRewritePlanner.PlannedGroup oldCompact = planned.get(0);
+ DataFileRewritePlanner.PlannedGroup newCompact = planned.get(1);
+ if (oldCompact.group().inputFileNum() == 2) {
+ newCompact = planned.get(0);
+ oldCompact = planned.get(1);
+ }
+
+ testHarness.processElement(newCompact, System.currentTimeMillis());
+ actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(3);
+ assertRewriteFileGroup(
+ actual.get(2),
+ table,
+ records(
+ table.schema(),
+ ImmutableList.of(ImmutableList.of(4, "p1"), ImmutableList.of(4, "p1"))),
+ 1,
+ ImmutableSet.of(newPartition));
+
+ testHarness.processElement(oldCompact, System.currentTimeMillis());
+ actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(4);
+ PartitionData[] transformedPartitions = {
+ newPartition.copy(), newPartition.copy(), newPartition.copy()
+ };
+ transformedPartitions[0].set(1, 1);
+ transformedPartitions[1].set(1, 2);
+ transformedPartitions[2].set(1, 3);
+ assertRewriteFileGroup(
+ actual.get(3),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(
+ ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"), ImmutableList.of(3, "p1"))),
+ 3,
+ Sets.newHashSet(transformedPartitions));
+ }
+ }
+
+ @Test
+ void testError() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ try (OneInputStreamOperatorTestHarness<
+ DataFileRewritePlanner.PlannedGroup, DataFileRewriteRunner.ExecutedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewriteRunner(
+ OperatorTestBase.DUMMY_TABLE_NAME, OperatorTestBase.DUMMY_TABLE_NAME, 0))) {
+ testHarness.open();
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ // Cause an exception
+ dropTable();
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ testHarness.processElement(planned.get(0), System.currentTimeMillis());
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ assertThat(
+ testHarness
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .poll()
+ .getValue()
+ .getMessage())
+ .contains("File does not exist: ");
+ }
+ }
+
+ @Test
+ void testV2Table() throws Exception {
+ Table table = createTableWithDelete();
+ update(table, 1, null, "a", "b");
+ update(table, 1, "b", "c");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+
+ List<DataFileRewriteRunner.ExecutedGroup> actual = executeRewrite(planned);
+ assertThat(actual).hasSize(1);
+
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ records(table.schema(), ImmutableSet.of(ImmutableList.of(1, "c"))),
+ 1,
+ ImmutableSet.of(new PartitionData(PartitionSpec.unpartitioned().partitionType())));
+ }
+
+ @Test
+ void testSplitSize() throws Exception {
+ Table table = createTable();
+
+ File dataDir = new File(new Path(table.location(), "data").toUri().getPath());
+ dataDir.mkdir();
+ GenericAppenderHelper dataAppender =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, dataDir.toPath());
+ List<Record> expected = Lists.newArrayListWithExpectedSize(4000);
+ for (int i = 0; i < 4; ++i) {
+ List<Record> batch = RandomGenericData.generate(table.schema(), 1000, 10 + i);
+ dataAppender.appendToTable(batch);
+ expected.addAll(batch);
+ }
+
+ // First run with high target file size
+ List<DataFileRewritePlanner.PlannedGroup> planWithNoTargetFileSize =
+ planDataFileRewrite(tableLoader());
+ assertThat(planWithNoTargetFileSize).hasSize(1);
+
+ // Second run with low target file size
+ long targetFileSize =
+ planWithNoTargetFileSize.get(0).group().fileScanTasks().get(0).sizeBytes()
+ + planWithNoTargetFileSize.get(0).group().fileScanTasks().get(1).sizeBytes();
+ List<DataFileRewritePlanner.PlannedGroup> planned;
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader(),
+ 11,
+ 10_000_000,
+ ImmutableMap.of(
+ MIN_INPUT_FILES,
+ "2",
+ TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize))))) {
+ testHarness.open();
+
+ OperatorTestBase.trigger(testHarness);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ planned = testHarness.extractOutputValues();
+ assertThat(planned).hasSize(1);
+ }
+
+ List<DataFileRewriteRunner.ExecutedGroup> actual = executeRewrite(planned);
+ assertThat(actual).hasSize(1);
+
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ expected,
+ 2,
+ ImmutableSet.of(new PartitionData(PartitionSpec.unpartitioned().partitionType())));
+ }
+
+ void assertRewriteFileGroup(
+ DataFileRewriteRunner.ExecutedGroup actual,
+ Table table,
+ Collection<Record> expectedRecords,
+ int expectedFileNum,
+ Set<StructLike> expectedPartitions)
+ throws IOException {
+ assertThat(actual.snapshotId()).isEqualTo(table.currentSnapshot().snapshotId());
+ assertThat(actual.groupsPerCommit()).isEqualTo(1);
+ assertThat(actual.group().addedFiles()).hasSize(expectedFileNum);
+ Collection<Record> writtenRecords = Lists.newArrayListWithExpectedSize(expectedRecords.size());
+ Set<StructLike> writtenPartitions = Sets.newHashSetWithExpectedSize(expectedPartitions.size());
+ for (DataFile newDataFile : actual.group().addedFiles()) {
+ assertThat(newDataFile.format()).isEqualTo(FileFormat.PARQUET);
+ assertThat(newDataFile.content()).isEqualTo(FileContent.DATA);
+ assertThat(newDataFile.keyMetadata()).isNull();
+ writtenPartitions.add(newDataFile.partition());
+
+ try (CloseableIterable<Record> reader =
+ Parquet.read(table.io().newInputFile(newDataFile.location()))
+ .project(table.schema())
+ .createReaderFunc(
+ fileSchema -> GenericParquetReaders.buildReader(table.schema(), fileSchema))
+ .build()) {
+ List<Record> newRecords = Lists.newArrayList(reader);
+ assertThat(newRecords).hasSize((int) newDataFile.recordCount());
+ writtenRecords.addAll(newRecords);
+ }
+ }
+
+ assertThat(writtenRecords).containsExactlyInAnyOrderElementsOf(expectedRecords);
+ assertThat(writtenPartitions).isEqualTo(expectedPartitions);
+ }
+
+ private List<Record> records(Schema schema, Collection<List<Object>> data) {
+ GenericRecord record = GenericRecord.create(schema);
+
+ ImmutableList.Builder<Record> builder = ImmutableList.builder();
+ data.forEach(
+ recordData ->
+ builder.add(
+ record.copy(ImmutableMap.of("id", recordData.get(0), "data", recordData.get(1)))));
+
+ return builder.build();
+ }
+}
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java
index d70c4aafd5..7511e1029b 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java
@@ -90,8 +90,7 @@ class TestDeleteFilesProcessor extends OperatorTestBase {
tableLoader().open();
try (OneInputStreamOperatorTestHarness<String, Void> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new DeleteFilesProcessor(0, DUMMY_TASK_NAME, tableLoader.loadTable(), 10),
- StringSerializer.INSTANCE)) {
+ new DeleteFilesProcessor(table, DUMMY_TASK_NAME, 0, 10), StringSerializer.INSTANCE)) {
testHarness.open();
testHarness.processElement(fileName, System.currentTimeMillis());
testHarness.processWatermark(EVENT_TIME);
(iceberg) 02/02: Flink: Backport Maintenance - RewriteDataFiles to Flink 2.0
Posted by st...@apache.org.
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
commit 27e73fea23dcf39185da0dba2e3b9e3ef5f3254b
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Tue May 6 15:32:11 2025 +0200
Flink: Backport Maintenance - RewriteDataFiles to Flink 2.0
Backports #11497
---
.../flink/maintenance/api/ExpireSnapshots.java | 2 +-
.../maintenance/api/MaintenanceTaskBuilder.java | 4 +-
.../flink/maintenance/api/RewriteDataFiles.java | 234 ++++++++++++
.../flink/maintenance/api/TableMaintenance.java | 7 +-
.../operator/DataFileRewriteCommitter.java | 199 ++++++++++
.../operator/DataFileRewritePlanner.java | 206 ++++++++++
.../operator/DataFileRewriteRunner.java | 253 +++++++++++++
.../maintenance/operator/DeleteFilesProcessor.java | 29 +-
.../operator/ExpireSnapshotsProcessor.java | 2 +-
.../flink/maintenance/operator/LockRemover.java | 28 +-
.../flink/maintenance/operator/LogUtil.java | 26 ++
.../operator/TableMaintenanceMetrics.java | 30 ++
.../maintenance/operator/TaskResultAggregator.java | 101 +++++
.../flink/maintenance/operator/TriggerManager.java | 29 +-
.../api/MaintenanceTaskInfraExtension.java | 8 +-
.../maintenance/api/MaintenanceTaskTestBase.java | 39 +-
.../flink/maintenance/api/TestExpireSnapshots.java | 43 +--
.../flink/maintenance/api/TestMaintenanceE2E.java | 13 +
.../maintenance/api/TestRewriteDataFiles.java | 417 +++++++++++++++++++++
.../maintenance/operator/OperatorTestBase.java | 113 ++++++
.../flink/maintenance/operator/RewriteUtil.java | 83 ++++
.../operator/TestDataFileRewriteCommitter.java | 278 ++++++++++++++
.../operator/TestDataFileRewritePlanner.java | 193 ++++++++++
.../operator/TestDataFileRewriteRunner.java | 355 ++++++++++++++++++
.../operator/TestDeleteFilesProcessor.java | 3 +-
25 files changed, 2587 insertions(+), 108 deletions(-)
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java
index 9cde5cb173..2bac5163ca 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ExpireSnapshots.java
@@ -113,7 +113,7 @@ public class ExpireSnapshots {
operatorName(DELETE_FILES_OPERATOR_NAME),
TypeInformation.of(Void.class),
new DeleteFilesProcessor(
- index(), taskName(), tableLoader().loadTable(), deleteBatchSize))
+ tableLoader().loadTable(), taskName(), index(), deleteBatchSize))
.uid(DELETE_FILES_OPERATOR_NAME + uidSuffix())
.slotSharingGroup(slotSharingGroup())
.setParallelism(parallelism());
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java
index 3fc431d025..c884f13b3e 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskBuilder.java
@@ -195,9 +195,9 @@ public abstract class MaintenanceTaskBuilder<T extends MaintenanceTaskBuilder<?>
DataStream<TaskResult> append(
DataStream<Trigger> sourceStream,
- int taskIndex,
- String newTaskName,
String newTableName,
+ String newTaskName,
+ int taskIndex,
TableLoader newTableLoader,
String defaultUidSuffix,
String defaultSlotSharingGroup,
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
new file mode 100644
index 0000000000..b96635e6ab
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
@@ -0,0 +1,234 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import java.util.Map;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewriteCommitter;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewritePlanner;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewriteRunner;
+import org.apache.iceberg.flink.maintenance.operator.TaskResultAggregator;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * Creates the data file rewriter data stream. Which runs a single iteration of the task for every
+ * {@link Trigger} event.
+ */
+public class RewriteDataFiles {
+ static final String PLANNER_TASK_NAME = "RDF Planner";
+ static final String REWRITE_TASK_NAME = "Rewrite";
+ static final String COMMIT_TASK_NAME = "Rewrite commit";
+ static final String AGGREGATOR_TASK_NAME = "Rewrite aggregator";
+
+ private RewriteDataFiles() {}
+
+ /** Creates the builder for a stream which rewrites data files for the table. */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public static class Builder extends MaintenanceTaskBuilder<RewriteDataFiles.Builder> {
+ private boolean partialProgressEnabled = false;
+ private int partialProgressMaxCommits = 10;
+ private final Map<String, String> rewriteOptions = Maps.newHashMapWithExpectedSize(6);
+ private long maxRewriteBytes = Long.MAX_VALUE;
+
+ /**
+ * Allows committing compacted data files in batches. See {@link
+ * org.apache.iceberg.actions.RewriteDataFiles#PARTIAL_PROGRESS_ENABLED} for more details.
+ *
+ * @param newPartialProgressEnabled to enable partial commits
+ */
+ public Builder partialProgressEnabled(boolean newPartialProgressEnabled) {
+ this.partialProgressEnabled = newPartialProgressEnabled;
+ return this;
+ }
+
+ /**
+ * Configures the size of batches if {@link #partialProgressEnabled}. See {@link
+ * org.apache.iceberg.actions.RewriteDataFiles#PARTIAL_PROGRESS_MAX_COMMITS} for more details.
+ *
+ * @param newPartialProgressMaxCommits to target number of the commits per run
+ */
+ public Builder partialProgressMaxCommits(int newPartialProgressMaxCommits) {
+ this.partialProgressMaxCommits = newPartialProgressMaxCommits;
+ return this;
+ }
+
+ /**
+ * Configures the maximum byte size of the rewrites for one scheduled compaction. This could be
+ * used to limit the resources used by the compaction.
+ *
+ * @param newMaxRewriteBytes to limit the size of the rewrites
+ */
+ public Builder maxRewriteBytes(long newMaxRewriteBytes) {
+ this.maxRewriteBytes = newMaxRewriteBytes;
+ return this;
+ }
+
+ /**
+ * Configures the target file size. See {@link
+ * org.apache.iceberg.actions.RewriteDataFiles#TARGET_FILE_SIZE_BYTES} for more details.
+ *
+ * @param targetFileSizeBytes target file size
+ */
+ public Builder targetFileSizeBytes(long targetFileSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.TARGET_FILE_SIZE_BYTES, String.valueOf(targetFileSizeBytes));
+ return this;
+ }
+
+ /**
+ * Configures the min file size considered for rewriting. See {@link
+ * SizeBasedFileRewritePlanner#MIN_FILE_SIZE_BYTES} for more details.
+ *
+ * @param minFileSizeBytes min file size
+ */
+ public Builder minFileSizeBytes(long minFileSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MIN_FILE_SIZE_BYTES, String.valueOf(minFileSizeBytes));
+ return this;
+ }
+
+ /**
+ * Configures the max file size considered for rewriting. See {@link
+ * SizeBasedFileRewritePlanner#MAX_FILE_SIZE_BYTES} for more details.
+ *
+ * @param maxFileSizeBytes max file size
+ */
+ public Builder maxFileSizeBytes(long maxFileSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MAX_FILE_SIZE_BYTES, String.valueOf(maxFileSizeBytes));
+ return this;
+ }
+
+ /**
+ * Configures the minimum file number after a rewrite is always initiated. See description see
+ * {@link SizeBasedFileRewritePlanner#MIN_INPUT_FILES} for more details.
+ *
+ * @param minInputFiles min file number
+ */
+ public Builder minInputFiles(int minInputFiles) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MIN_INPUT_FILES, String.valueOf(minInputFiles));
+ return this;
+ }
+
+ /**
+ * Configures the minimum delete file number for a file after a rewrite is always initiated. See
+ * {@link BinPackRewriteFilePlanner#DELETE_FILE_THRESHOLD} for more details.
+ *
+ * @param deleteFileThreshold min delete file number
+ */
+ public Builder deleteFileThreshold(int deleteFileThreshold) {
+ this.rewriteOptions.put(
+ BinPackRewriteFilePlanner.DELETE_FILE_THRESHOLD, String.valueOf(deleteFileThreshold));
+ return this;
+ }
+
+ /**
+ * Overrides other options and forces rewriting of all provided files.
+ *
+ * @param rewriteAll enables a full rewrite
+ */
+ public Builder rewriteAll(boolean rewriteAll) {
+ this.rewriteOptions.put(SizeBasedFileRewritePlanner.REWRITE_ALL, String.valueOf(rewriteAll));
+ return this;
+ }
+
+ /**
+ * Configures the group size for rewriting. See {@link
+ * SizeBasedFileRewritePlanner#MAX_FILE_GROUP_SIZE_BYTES} for more details.
+ *
+ * @param maxFileGroupSizeBytes file group size for rewrite
+ */
+ public Builder maxFileGroupSizeBytes(long maxFileGroupSizeBytes) {
+ this.rewriteOptions.put(
+ SizeBasedFileRewritePlanner.MAX_FILE_GROUP_SIZE_BYTES,
+ String.valueOf(maxFileGroupSizeBytes));
+ return this;
+ }
+
+ /**
+ * The input is a {@link DataStream} with {@link Trigger} events and every event should be
+ * immediately followed by a {@link Watermark} with the same timestamp as the event.
+ *
+ * <p>The output is a {@link DataStream} with the {@link TaskResult} of the run followed by the
+ * {@link Watermark}.
+ */
+ @Override
+ DataStream<TaskResult> append(DataStream<Trigger> trigger) {
+ SingleOutputStreamOperator<DataFileRewritePlanner.PlannedGroup> planned =
+ trigger
+ .process(
+ new DataFileRewritePlanner(
+ tableName(),
+ taskName(),
+ index(),
+ tableLoader(),
+ partialProgressEnabled ? partialProgressMaxCommits : 1,
+ maxRewriteBytes,
+ rewriteOptions))
+ .name(operatorName(PLANNER_TASK_NAME))
+ .uid(PLANNER_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .forceNonParallel();
+
+ SingleOutputStreamOperator<DataFileRewriteRunner.ExecutedGroup> rewritten =
+ planned
+ .rebalance()
+ .process(new DataFileRewriteRunner(tableName(), taskName(), index()))
+ .name(operatorName(REWRITE_TASK_NAME))
+ .uid(REWRITE_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .setParallelism(parallelism());
+
+ SingleOutputStreamOperator<Trigger> updated =
+ rewritten
+ .transform(
+ operatorName(COMMIT_TASK_NAME),
+ TypeInformation.of(Trigger.class),
+ new DataFileRewriteCommitter(tableName(), taskName(), index(), tableLoader()))
+ .uid(COMMIT_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .forceNonParallel();
+
+ return trigger
+ .union(updated)
+ .connect(
+ planned
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .union(
+ rewritten.getSideOutput(TaskResultAggregator.ERROR_STREAM),
+ updated.getSideOutput(TaskResultAggregator.ERROR_STREAM)))
+ .transform(
+ operatorName(AGGREGATOR_TASK_NAME),
+ TypeInformation.of(TaskResult.class),
+ new TaskResultAggregator(tableName(), taskName(), index()))
+ .uid(AGGREGATOR_TASK_NAME + uidSuffix())
+ .slotSharingGroup(slotSharingGroup())
+ .forceNonParallel();
+ }
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java
index 25d0325abb..12230891ac 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/TableMaintenance.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.flink.maintenance.api;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
+import java.util.Locale;
import java.util.UUID;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
@@ -257,9 +258,9 @@ public class TableMaintenance {
DataStream<TaskResult> result =
builder.append(
filtered,
- taskIndex,
- taskNames.get(taskIndex),
tableName,
+ taskNames.get(taskIndex),
+ taskIndex,
loader,
uidSuffix,
slotSharingGroup,
@@ -301,7 +302,7 @@ public class TableMaintenance {
private static String nameFor(MaintenanceTaskBuilder<?> streamBuilder, int taskIndex) {
return String.format(
- "%s [%s]", streamBuilder.getClass().getSimpleName(), String.valueOf(taskIndex));
+ Locale.ROOT, "%s [%d]", streamBuilder.getClass().getSimpleName(), taskIndex);
}
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java
new file mode 100644
index 0000000000..135d3d9b42
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteCommitter.java
@@ -0,0 +1,199 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import java.io.IOException;
+import java.util.Set;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.actions.RewriteDataFilesCommitManager;
+import org.apache.iceberg.actions.RewriteDataFilesCommitManager.CommitService;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Commits the rewrite changes using {@link RewriteDataFilesCommitManager}. The input is a {@link
+ * DataFileRewriteRunner.ExecutedGroup}. Only {@link Watermark} is emitted which is chained to
+ * {@link TaskResultAggregator} input 1.
+ */
+@Internal
+public class DataFileRewriteCommitter extends AbstractStreamOperator<Trigger>
+ implements OneInputStreamOperator<DataFileRewriteRunner.ExecutedGroup, Trigger> {
+ private static final Logger LOG = LoggerFactory.getLogger(DataFileRewriteCommitter.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+ private final TableLoader tableLoader;
+
+ private transient Table table;
+ private transient CommitService commitService;
+ private transient Counter errorCounter;
+ private transient Counter addedDataFileNumCounter;
+ private transient Counter addedDataFileSizeCounter;
+ private transient Counter removedDataFileNumCounter;
+ private transient Counter removedDataFileSizeCounter;
+
+ public DataFileRewriteCommitter(
+ String tableName, String taskName, int taskIndex, TableLoader tableLoader) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+ Preconditions.checkNotNull(tableLoader, "Table loader should no be null");
+
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ this.tableLoader = tableLoader;
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ tableLoader.open();
+ this.table = tableLoader.loadTable();
+
+ MetricGroup taskMetricGroup =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex);
+ this.errorCounter = taskMetricGroup.counter(TableMaintenanceMetrics.ERROR_COUNTER);
+ this.addedDataFileNumCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.ADDED_DATA_FILE_NUM_METRIC);
+ this.addedDataFileSizeCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.ADDED_DATA_FILE_SIZE_METRIC);
+ this.removedDataFileNumCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.REMOVED_DATA_FILE_NUM_METRIC);
+ this.removedDataFileSizeCounter =
+ taskMetricGroup.counter(TableMaintenanceMetrics.REMOVED_DATA_FILE_SIZE_METRIC);
+ }
+
+ @Override
+ public void processElement(StreamRecord<DataFileRewriteRunner.ExecutedGroup> streamRecord) {
+ DataFileRewriteRunner.ExecutedGroup executedGroup = streamRecord.getValue();
+ try {
+ if (commitService == null) {
+ // Refresh the table to get the latest snapshot for the committer
+ table.refresh();
+
+ FlinkRewriteDataFilesCommitManager commitManager =
+ new FlinkRewriteDataFilesCommitManager(
+ table, executedGroup.snapshotId(), streamRecord.getTimestamp());
+ this.commitService = commitManager.service(executedGroup.groupsPerCommit());
+ commitService.start();
+ }
+
+ commitService.offer(executedGroup.group());
+ } catch (Exception e) {
+ LOG.warn(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception processing {}",
+ tableName,
+ taskName,
+ taskIndex,
+ streamRecord.getTimestamp(),
+ executedGroup,
+ e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ errorCounter.inc();
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ try {
+ if (commitService != null) {
+ commitService.close();
+ }
+
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Successfully completed data file compaction",
+ tableName,
+ taskName,
+ taskIndex,
+ mark.getTimestamp());
+ } catch (Exception e) {
+ LOG.warn(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception closing commit service",
+ tableName,
+ taskName,
+ taskIndex,
+ mark.getTimestamp(),
+ e);
+ output.collect(TaskResultAggregator.ERROR_STREAM, new StreamRecord<>(e));
+ errorCounter.inc();
+ }
+
+ // Cleanup
+ this.commitService = null;
+
+ super.processWatermark(mark);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (commitService != null) {
+ commitService.close();
+ }
+ }
+
+ private class FlinkRewriteDataFilesCommitManager extends RewriteDataFilesCommitManager {
+ private final long timestamp;
+
+ FlinkRewriteDataFilesCommitManager(Table table, long startingSnapshotId, long timestamp) {
+ super(table, startingSnapshotId);
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public void commitFileGroups(Set<RewriteFileGroup> fileGroups) {
+ super.commitFileGroups(fileGroups);
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Committed {}",
+ tableName,
+ taskName,
+ taskIndex,
+ timestamp,
+ fileGroups);
+ updateMetrics(fileGroups);
+ }
+
+ private void updateMetrics(Set<RewriteFileGroup> fileGroups) {
+ for (RewriteFileGroup fileGroup : fileGroups) {
+ for (DataFile added : fileGroup.addedFiles()) {
+ addedDataFileNumCounter.inc();
+ addedDataFileSizeCounter.inc(added.fileSizeInBytes());
+ }
+
+ for (DataFile rewritten : fileGroup.rewrittenFiles()) {
+ removedDataFileNumCounter.inc();
+ removedDataFileSizeCounter.inc(rewritten.fileSizeInBytes());
+ }
+ }
+ }
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
new file mode 100644
index 0000000000..8f67ce151a
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
@@ -0,0 +1,206 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import java.math.RoundingMode;
+import java.util.List;
+import java.util.Map;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.SerializableTable;
+import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
+import org.apache.iceberg.actions.FileRewritePlan;
+import org.apache.iceberg.actions.RewriteDataFiles;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.math.IntMath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Plans the rewrite groups using the {@link BinPackRewriteFilePlanner}. The input is the {@link
+ * Trigger}, the output is zero, or some {@link PlannedGroup}s.
+ */
+@Internal
+public class DataFileRewritePlanner
+ extends ProcessFunction<Trigger, DataFileRewritePlanner.PlannedGroup> {
+ static final String MESSAGE_PREFIX = "[For table {} with {}[{}] at {}]: ";
+ private static final Logger LOG = LoggerFactory.getLogger(DataFileRewritePlanner.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+ private final TableLoader tableLoader;
+ private final int partialProgressMaxCommits;
+ private final long maxRewriteBytes;
+ private final Map<String, String> rewriterOptions;
+ private transient Counter errorCounter;
+
+ public DataFileRewritePlanner(
+ String tableName,
+ String taskName,
+ int taskIndex,
+ TableLoader tableLoader,
+ int newPartialProgressMaxCommits,
+ long maxRewriteBytes,
+ Map<String, String> rewriterOptions) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+ Preconditions.checkNotNull(tableLoader, "Table loader should no be null");
+ Preconditions.checkNotNull(rewriterOptions, "Options map should no be null");
+
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ this.tableLoader = tableLoader;
+ this.partialProgressMaxCommits = newPartialProgressMaxCommits;
+ this.maxRewriteBytes = maxRewriteBytes;
+ this.rewriterOptions = rewriterOptions;
+ }
+
+ @Override
+ public void open(OpenContext context) throws Exception {
+ tableLoader.open();
+ this.errorCounter =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
+ .counter(TableMaintenanceMetrics.ERROR_COUNTER);
+ }
+
+ @Override
+ public void processElement(Trigger value, Context ctx, Collector<PlannedGroup> out)
+ throws Exception {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Creating rewrite plan",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp());
+ try {
+ SerializableTable table =
+ (SerializableTable) SerializableTable.copyOf(tableLoader.loadTable());
+ if (table.currentSnapshot() == null) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Nothing to plan for in an empty table",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp());
+ return;
+ }
+
+ BinPackRewriteFilePlanner planner = new BinPackRewriteFilePlanner(table);
+ planner.init(rewriterOptions);
+
+ FileRewritePlan<RewriteDataFiles.FileGroupInfo, FileScanTask, DataFile, RewriteFileGroup>
+ plan = planner.plan();
+
+ long rewriteBytes = 0;
+ List<RewriteFileGroup> groups = Lists.newArrayList();
+ for (CloseableIterator<RewriteFileGroup> groupIterator = plan.groups().iterator();
+ groupIterator.hasNext(); ) {
+ RewriteFileGroup group = groupIterator.next();
+ if (rewriteBytes + group.inputFilesSizeInBytes() > maxRewriteBytes) {
+ // Keep going, maybe some other group might fit in
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Skipping group as max rewrite size reached {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ group);
+ } else {
+ rewriteBytes += group.inputFilesSizeInBytes();
+ groups.add(group);
+ }
+ }
+
+ int groupsPerCommit =
+ IntMath.divide(groups.size(), partialProgressMaxCommits, RoundingMode.CEILING);
+
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewrite plan created {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ groups);
+
+ for (RewriteFileGroup group : groups) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Emitting {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ group);
+ out.collect(new PlannedGroup(table, groupsPerCommit, group));
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Failed to plan data file rewrite groups",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ e);
+ ctx.output(TaskResultAggregator.ERROR_STREAM, e);
+ errorCounter.inc();
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ tableLoader.close();
+ }
+
+ public static class PlannedGroup {
+ private final SerializableTable table;
+ private final int groupsPerCommit;
+ private final RewriteFileGroup group;
+
+ private PlannedGroup(SerializableTable table, int groupsPerCommit, RewriteFileGroup group) {
+ this.table = table;
+ this.groupsPerCommit = groupsPerCommit;
+ this.group = group;
+ }
+
+ SerializableTable table() {
+ return table;
+ }
+
+ int groupsPerCommit() {
+ return groupsPerCommit;
+ }
+
+ RewriteFileGroup group() {
+ return group;
+ }
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java
new file mode 100644
index 0000000000..ad3b045400
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewriteRunner.java
@@ -0,0 +1,253 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING;
+
+import java.util.Collections;
+import java.util.Set;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.streaming.api.functions.ProcessFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.Collector;
+import org.apache.iceberg.BaseCombinedScanTask;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.actions.RewriteFileGroup;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.maintenance.operator.DataFileRewritePlanner.PlannedGroup;
+import org.apache.iceberg.flink.sink.RowDataTaskWriterFactory;
+import org.apache.iceberg.flink.sink.TaskWriterFactory;
+import org.apache.iceberg.flink.source.DataIterator;
+import org.apache.iceberg.flink.source.FileScanTaskReader;
+import org.apache.iceberg.flink.source.RowDataFileScanTaskReader;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Executes a rewrite for a single {@link PlannedGroup}. Reads the files with the standard {@link
+ * FileScanTaskReader}, so the delete files are considered, and writes using the {@link
+ * TaskWriterFactory}. The output is an {@link ExecutedGroup}.
+ */
+@Internal
+public class DataFileRewriteRunner
+ extends ProcessFunction<PlannedGroup, DataFileRewriteRunner.ExecutedGroup> {
+ private static final Logger LOG = LoggerFactory.getLogger(DataFileRewriteRunner.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+
+ private transient int subTaskId;
+ private transient int attemptId;
+ private transient Counter errorCounter;
+
+ public DataFileRewriteRunner(String tableName, String taskName, int taskIndex) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ }
+
+ @Override
+ public void open(OpenContext context) {
+ this.errorCounter =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
+ .counter(TableMaintenanceMetrics.ERROR_COUNTER);
+
+ this.subTaskId = getRuntimeContext().getTaskInfo().getIndexOfThisSubtask();
+ this.attemptId = getRuntimeContext().getTaskInfo().getAttemptNumber();
+ }
+
+ @Override
+ public void processElement(PlannedGroup value, Context ctx, Collector<ExecutedGroup> out)
+ throws Exception {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewriting files for group {} with files: {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().info(),
+ value.group().rewrittenFiles());
+ } else {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Rewriting files for group {} with {} number of files",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().info(),
+ value.group().rewrittenFiles().size());
+ }
+
+ try (TaskWriter<RowData> writer = writerFor(value)) {
+ try (DataIterator<RowData> iterator = readerFor(value)) {
+ while (iterator.hasNext()) {
+ writer.write(iterator.next());
+ }
+
+ Set<DataFile> dataFiles = Sets.newHashSet(writer.dataFiles());
+ value.group().setOutputFiles(dataFiles);
+ out.collect(
+ new ExecutedGroup(
+ value.table().currentSnapshot().snapshotId(),
+ value.groupsPerCommit(),
+ value.group()));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewritten files {} from {} to {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().info(),
+ value.group().rewrittenFiles(),
+ value.group().addedFiles());
+ } else {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Rewritten {} files to {} files",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group().rewrittenFiles().size(),
+ value.group().addedFiles().size());
+ }
+ } catch (Exception ex) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception rewriting datafile group {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group(),
+ ex);
+ ctx.output(TaskResultAggregator.ERROR_STREAM, ex);
+ errorCounter.inc();
+ abort(writer, ctx.timestamp());
+ }
+ } catch (Exception ex) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Exception creating compaction writer for group {}",
+ tableName,
+ taskName,
+ taskIndex,
+ ctx.timestamp(),
+ value.group(),
+ ex);
+ ctx.output(TaskResultAggregator.ERROR_STREAM, ex);
+ errorCounter.inc();
+ }
+ }
+
+ private TaskWriter<RowData> writerFor(PlannedGroup value) {
+ String formatString =
+ PropertyUtil.propertyAsString(
+ value.table().properties(),
+ TableProperties.DEFAULT_FILE_FORMAT,
+ TableProperties.DEFAULT_FILE_FORMAT_DEFAULT);
+ RowDataTaskWriterFactory factory =
+ new RowDataTaskWriterFactory(
+ value.table(),
+ FlinkSchemaUtil.convert(value.table().schema()),
+ value.group().inputSplitSize(),
+ FileFormat.fromString(formatString),
+ value.table().properties(),
+ null,
+ false);
+ factory.initialize(subTaskId, attemptId);
+ return factory.create();
+ }
+
+ private DataIterator<RowData> readerFor(PlannedGroup value) {
+ RowDataFileScanTaskReader reader =
+ new RowDataFileScanTaskReader(
+ value.table().schema(),
+ value.table().schema(),
+ PropertyUtil.propertyAsString(value.table().properties(), DEFAULT_NAME_MAPPING, null),
+ false,
+ Collections.emptyList());
+ return new DataIterator<>(
+ reader,
+ new BaseCombinedScanTask(value.group().fileScanTasks()),
+ value.table().io(),
+ value.table().encryption());
+ }
+
+ private void abort(TaskWriter<RowData> writer, long timestamp) {
+ try {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX
+ + "Aborting rewrite for (subTaskId {}, attemptId {})",
+ tableName,
+ taskName,
+ taskIndex,
+ timestamp,
+ subTaskId,
+ attemptId);
+ writer.abort();
+ } catch (Exception e) {
+ LOG.info(
+ DataFileRewritePlanner.MESSAGE_PREFIX + "Exception in abort",
+ tableName,
+ taskName,
+ taskIndex,
+ timestamp,
+ e);
+ }
+ }
+
+ public static class ExecutedGroup {
+ private final long snapshotId;
+ private final int groupsPerCommit;
+ private final RewriteFileGroup group;
+
+ @VisibleForTesting
+ ExecutedGroup(long snapshotId, int groupsPerCommit, RewriteFileGroup group) {
+ this.snapshotId = snapshotId;
+ this.groupsPerCommit = groupsPerCommit;
+ this.group = group;
+ }
+
+ long snapshotId() {
+ return snapshotId;
+ }
+
+ int groupsPerCommit() {
+ return groupsPerCommit;
+ }
+
+ RewriteFileGroup group() {
+ return group;
+ }
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java
index dc7846c4c4..9189f5f018 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DeleteFilesProcessor.java
@@ -21,6 +21,7 @@ package org.apache.iceberg.flink.maintenance.operator;
import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -40,17 +41,17 @@ public class DeleteFilesProcessor extends AbstractStreamOperator<Void>
implements OneInputStreamOperator<String, Void> {
private static final Logger LOG = LoggerFactory.getLogger(DeleteFilesProcessor.class);
- private final String taskIndex;
+ private final String tableName;
private final String taskName;
+ private final int taskIndex;
private final SupportsBulkOperations io;
- private final String tableName;
private final Set<String> filesToDelete = Sets.newHashSet();
private final int batchSize;
private transient Counter failedCounter;
private transient Counter succeededCounter;
- public DeleteFilesProcessor(int taskIndex, String taskName, Table table, int batchSize) {
+ public DeleteFilesProcessor(Table table, String taskName, int taskIndex, int batchSize) {
Preconditions.checkNotNull(taskName, "Task name should no be null");
Preconditions.checkNotNull(table, "Table should no be null");
@@ -60,31 +61,21 @@ public class DeleteFilesProcessor extends AbstractStreamOperator<Void>
"%s doesn't support bulk delete",
fileIO.getClass().getSimpleName());
- this.taskIndex = String.valueOf(taskIndex);
+ this.tableName = table.name();
this.taskName = taskName;
+ this.taskIndex = taskIndex;
this.io = (SupportsBulkOperations) fileIO;
- this.tableName = table.name();
this.batchSize = batchSize;
}
@Override
public void open() throws Exception {
+ MetricGroup taskMetricGroup =
+ TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex);
this.failedCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, taskName)
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, taskIndex)
- .counter(TableMaintenanceMetrics.DELETE_FILE_FAILED_COUNTER);
+ taskMetricGroup.counter(TableMaintenanceMetrics.DELETE_FILE_FAILED_COUNTER);
this.succeededCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, taskName)
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, taskIndex)
- .counter(TableMaintenanceMetrics.DELETE_FILE_SUCCEEDED_COUNTER);
+ taskMetricGroup.counter(TableMaintenanceMetrics.DELETE_FILE_SUCCEEDED_COUNTER);
}
@Override
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java
index a7e54297ba..a4f8dc2021 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/ExpireSnapshotsProcessor.java
@@ -74,7 +74,7 @@ public class ExpireSnapshotsProcessor extends ProcessFunction<Trigger, TaskResul
this.table = tableLoader.loadTable();
this.plannerPool =
plannerPoolSize != null
- ? ThreadPools.newWorkerPool(table.name() + "-table--planner", plannerPoolSize)
+ ? ThreadPools.newFixedThreadPool(table.name() + "-table--planner", plannerPoolSize)
: ThreadPools.getWorkerPool();
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
index 49fa69b487..2066ca8e01 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LockRemover.java
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.flink.annotation.Internal;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
@@ -98,31 +99,16 @@ public class LockRemover extends AbstractStreamOperator<Void>
this.failedTaskResultCounters = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
this.taskLastRunDurationMs = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
for (int taskIndex = 0; taskIndex < maintenanceTaskNames.size(); ++taskIndex) {
+ MetricGroup taskMetricGroup =
+ TableMaintenanceMetrics.groupFor(
+ getRuntimeContext(), tableName, maintenanceTaskNames.get(taskIndex), taskIndex);
succeededTaskResultCounters.add(
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
- .counter(TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER));
+ taskMetricGroup.counter(TableMaintenanceMetrics.SUCCEEDED_TASK_COUNTER));
failedTaskResultCounters.add(
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
- .counter(TableMaintenanceMetrics.FAILED_TASK_COUNTER));
+ taskMetricGroup.counter(TableMaintenanceMetrics.FAILED_TASK_COUNTER));
AtomicLong duration = new AtomicLong(0);
taskLastRunDurationMs.add(duration);
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
- .gauge(TableMaintenanceMetrics.LAST_RUN_DURATION_MS, duration::get);
+ taskMetricGroup.gauge(TableMaintenanceMetrics.LAST_RUN_DURATION_MS, duration::get);
}
lockFactory.open();
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java
new file mode 100644
index 0000000000..8bdcd7ba2b
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/LogUtil.java
@@ -0,0 +1,26 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+class LogUtil {
+ static final String MESSAGE_PREFIX = "[For table {} with {}[{}] at {}]: ";
+ static final String MESSAGE_FORMAT_PREFIX = "[For table %s with {%s}[{%d}] at {%d}]: ";
+
+ private LogUtil() {}
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
index 6147c3a5fd..897760caaa 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableMaintenanceMetrics.java
@@ -18,12 +18,18 @@
*/
package org.apache.iceberg.flink.maintenance.operator;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.metrics.MetricGroup;
+
public class TableMaintenanceMetrics {
public static final String GROUP_KEY = "maintenance";
public static final String TASK_NAME_KEY = "taskName";
public static final String TASK_INDEX_KEY = "taskIndex";
public static final String TABLE_NAME_KEY = "tableName";
+ // Operator error counter
+ public static final String ERROR_COUNTER = "error";
+
// TriggerManager metrics
public static final String RATE_LIMITER_TRIGGERED = "rateLimiterTriggered";
public static final String CONCURRENT_RUN_THROTTLED = "concurrentRunThrottled";
@@ -39,6 +45,30 @@ public class TableMaintenanceMetrics {
public static final String DELETE_FILE_FAILED_COUNTER = "deleteFailed";
public static final String DELETE_FILE_SUCCEEDED_COUNTER = "deleteSucceeded";
+ // DataFileUpdater metrics
+ public static final String ADDED_DATA_FILE_NUM_METRIC = "addedDataFileNum";
+ public static final String ADDED_DATA_FILE_SIZE_METRIC = "addedDataFileSize";
+ public static final String REMOVED_DATA_FILE_NUM_METRIC = "removedDataFileNum";
+ public static final String REMOVED_DATA_FILE_SIZE_METRIC = "removedDataFileSize";
+
+ static MetricGroup groupFor(
+ RuntimeContext context, String tableName, String taskName, int taskIndex) {
+ return groupFor(groupFor(context, tableName), taskName, taskIndex);
+ }
+
+ static MetricGroup groupFor(RuntimeContext context, String tableName) {
+ return context
+ .getMetricGroup()
+ .addGroup(TableMaintenanceMetrics.GROUP_KEY)
+ .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName);
+ }
+
+ static MetricGroup groupFor(MetricGroup mainGroup, String taskName, int taskIndex) {
+ return mainGroup
+ .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, taskName)
+ .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex));
+ }
+
private TableMaintenanceMetrics() {
// do not instantiate
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java
new file mode 100644
index 0000000000..cceb043a26
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TaskResultAggregator.java
@@ -0,0 +1,101 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import java.util.List;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.OutputTag;
+import org.apache.iceberg.flink.maintenance.api.TaskResult;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Aggregates results of the operators for a given maintenance task.
+ *
+ * <ul>
+ * <li>Input 1 is used:
+ * <ul>
+ * <li>To provide the {@link TaskResult#startEpoch()} - should be chained to the task input
+ * <li>To mark that the task is finished - should be chained at the end of the task, so an
+ * incoming watermark will signal that the task is finished
+ * </ul>
+ * <li>Input 2 expects an {@link Exception} which caused the failure - should be chained to the
+ * {@link #ERROR_STREAM} of the operators
+ * </ul>
+ *
+ * The operator emits a {@link TaskResult} with the overall result on {@link Watermark}.
+ */
+@Internal
+public class TaskResultAggregator extends AbstractStreamOperator<TaskResult>
+ implements TwoInputStreamOperator<Trigger, Exception, TaskResult> {
+ public static final OutputTag<Exception> ERROR_STREAM =
+ new OutputTag<>("error-stream", TypeInformation.of(Exception.class));
+
+ private static final Logger LOG = LoggerFactory.getLogger(TaskResultAggregator.class);
+
+ private final String tableName;
+ private final String taskName;
+ private final int taskIndex;
+ private final List<Exception> exceptions;
+ private transient Long startTime;
+
+ public TaskResultAggregator(String tableName, String taskName, int taskIndex) {
+ Preconditions.checkNotNull(tableName, "Table name should no be null");
+ Preconditions.checkNotNull(taskName, "Task name should no be null");
+
+ this.tableName = tableName;
+ this.taskName = taskName;
+ this.taskIndex = taskIndex;
+ this.exceptions = Lists.newArrayList();
+ this.startTime = 0L;
+ }
+
+ @Override
+ public void processElement1(StreamRecord<Trigger> streamRecord) {
+ startTime = streamRecord.getValue().timestamp();
+ }
+
+ @Override
+ public void processElement2(StreamRecord<Exception> streamRecord) {
+ Preconditions.checkNotNull(streamRecord.getValue(), "Exception could not be `null`.");
+ exceptions.add(streamRecord.getValue());
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {
+ TaskResult response = new TaskResult(taskIndex, startTime, exceptions.isEmpty(), exceptions);
+ output.collect(new StreamRecord<>(response));
+ LOG.info(
+ "Aggregated result for table {}, task {}[{}] is {}",
+ tableName,
+ taskName,
+ taskIndex,
+ response);
+ exceptions.clear();
+ startTime = 0L;
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
index 9dcd5286da..f1f2b51c09 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerManager.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.TimerService;
@@ -121,33 +122,17 @@ public class TriggerManager extends KeyedProcessFunction<Boolean, TableChange, T
@Override
public void open(OpenContext parameters) throws Exception {
+ MetricGroup mainGroup = TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName);
this.rateLimiterTriggeredCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .counter(TableMaintenanceMetrics.RATE_LIMITER_TRIGGERED);
+ mainGroup.counter(TableMaintenanceMetrics.RATE_LIMITER_TRIGGERED);
this.concurrentRunThrottledCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .counter(TableMaintenanceMetrics.CONCURRENT_RUN_THROTTLED);
- this.nothingToTriggerCounter =
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .counter(TableMaintenanceMetrics.NOTHING_TO_TRIGGER);
+ mainGroup.counter(TableMaintenanceMetrics.CONCURRENT_RUN_THROTTLED);
+ this.nothingToTriggerCounter = mainGroup.counter(TableMaintenanceMetrics.NOTHING_TO_TRIGGER);
this.triggerCounters = Lists.newArrayListWithExpectedSize(maintenanceTaskNames.size());
for (int taskIndex = 0; taskIndex < maintenanceTaskNames.size(); ++taskIndex) {
triggerCounters.add(
- getRuntimeContext()
- .getMetricGroup()
- .addGroup(TableMaintenanceMetrics.GROUP_KEY)
- .addGroup(TableMaintenanceMetrics.TABLE_NAME_KEY, tableName)
- .addGroup(TableMaintenanceMetrics.TASK_NAME_KEY, maintenanceTaskNames.get(taskIndex))
- .addGroup(TableMaintenanceMetrics.TASK_INDEX_KEY, String.valueOf(taskIndex))
+ TableMaintenanceMetrics.groupFor(
+ mainGroup, maintenanceTaskNames.get(taskIndex), taskIndex)
.counter(TableMaintenanceMetrics.TRIGGERED));
}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java
index 1b69be1fa3..10efb9120c 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskInfraExtension.java
@@ -48,16 +48,16 @@ class MaintenanceTaskInfraExtension implements BeforeEachCallback {
@Override
public void beforeEach(ExtensionContext context) {
- env = StreamExecutionEnvironment.getExecutionEnvironment();
- source = new ManualSource<>(env, TypeInformation.of(Trigger.class));
+ this.env = StreamExecutionEnvironment.getExecutionEnvironment();
+ this.source = new ManualSource<>(env, TypeInformation.of(Trigger.class));
// Adds the watermark to mimic the behaviour expected for the input of the maintenance tasks
- triggerStream =
+ this.triggerStream =
source
.dataStream()
.assignTimestampsAndWatermarks(new TableMaintenance.PunctuatedWatermarkStrategy())
.name(IGNORED_OPERATOR_NAME)
.forceNonParallel();
- sink = new CollectingSink<>();
+ this.sink = new CollectingSink<>();
}
StreamExecutionEnvironment env() {
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java
index 36041d9c38..6d3a10c4b1 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/MaintenanceTaskTestBase.java
@@ -37,16 +37,42 @@ class MaintenanceTaskTestBase extends OperatorTestBase {
@RegisterExtension MaintenanceTaskInfraExtension infra = new MaintenanceTaskInfraExtension();
void runAndWaitForSuccess(
+ StreamExecutionEnvironment env,
+ ManualSource<Trigger> triggerSource,
+ CollectingSink<TaskResult> collectingSink)
+ throws Exception {
+ runAndWaitForResult(env, triggerSource, collectingSink, false, () -> true);
+ }
+
+ void runAndWaitForSuccess(
+ StreamExecutionEnvironment env,
+ ManualSource<Trigger> triggerSource,
+ CollectingSink<TaskResult> collectingSink,
+ Supplier<Boolean> waitForCondition)
+ throws Exception {
+ runAndWaitForResult(env, triggerSource, collectingSink, false, waitForCondition);
+ }
+
+ void runAndWaitForFailure(
+ StreamExecutionEnvironment env,
+ ManualSource<Trigger> triggerSource,
+ CollectingSink<TaskResult> collectingSink)
+ throws Exception {
+ runAndWaitForResult(env, triggerSource, collectingSink, true, () -> true);
+ }
+
+ void runAndWaitForResult(
StreamExecutionEnvironment env,
ManualSource<Trigger> triggerSource,
CollectingSink<TaskResult> collectingSink,
+ boolean generateFailure,
Supplier<Boolean> waitForCondition)
throws Exception {
JobClient jobClient = null;
try {
jobClient = env.executeAsync();
- // Do a single task run
+ // Do a single successful task run
long time = System.currentTimeMillis();
triggerSource.sendRecord(Trigger.create(time, TESTING_TASK_ID), time);
@@ -56,6 +82,17 @@ class MaintenanceTaskTestBase extends OperatorTestBase {
assertThat(result.success()).isTrue();
assertThat(result.taskIndex()).isEqualTo(TESTING_TASK_ID);
+ if (generateFailure) {
+ dropTable();
+ time = System.currentTimeMillis();
+ triggerSource.sendRecord(Trigger.create(time, TESTING_TASK_ID), time);
+ result = collectingSink.poll(POLL_DURATION);
+
+ assertThat(result.startEpoch()).isEqualTo(time);
+ assertThat(result.success()).isFalse();
+ assertThat(result.taskIndex()).isEqualTo(TESTING_TASK_ID);
+ }
+
Awaitility.await().until(waitForCondition::get);
} finally {
closeJobClient(jobClient);
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java
index f80129f966..b8aa259e2f 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestExpireSnapshots.java
@@ -27,7 +27,6 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.List;
import java.util.Set;
-import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
@@ -69,9 +68,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
.uidSuffix(UID_SUFFIX)
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
"OTHER",
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
@@ -101,36 +100,16 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
ExpireSnapshots.builder()
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
1)
.sinkTo(infra.sink());
- JobClient jobClient = null;
- try {
- jobClient = infra.env().executeAsync();
-
- // Do a single task run
- long time = System.currentTimeMillis();
- infra.source().sendRecord(Trigger.create(time, 1), time);
-
- // First successful run (ensure that the operators are loaded/opened etc.)
- assertThat(infra.sink().poll(Duration.ofSeconds(5)).success()).isTrue();
-
- // Drop the table, so it will cause an exception
- dropTable();
-
- // Failed run
- infra.source().sendRecord(Trigger.create(time + 1, 1), time + 1);
-
- assertThat(infra.sink().poll(Duration.ofSeconds(5)).success()).isFalse();
- } finally {
- closeJobClient(jobClient);
- }
+ runAndWaitForFailure(infra.env(), infra.source(), infra.sink());
// Check the metrics. There are no expired snapshots or data files because ExpireSnapshots has
// no max age of number of snapshots set, so no files are removed.
@@ -162,9 +141,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
.uidSuffix(UID_SUFFIX)
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
@@ -180,9 +159,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
ExpireSnapshots.builder()
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
@@ -204,9 +183,9 @@ class TestExpireSnapshots extends MaintenanceTaskTestBase {
.parallelism(1)
.append(
infra.triggerStream(),
- 0,
- DUMMY_TASK_NAME,
DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
tableLoader(),
UID_SUFFIX,
StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java
index 467ad2d8ce..0a860fec47 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestMaintenanceE2E.java
@@ -52,6 +52,19 @@ class TestMaintenanceE2E extends OperatorTestBase {
.retainLast(5)
.deleteBatchSize(5)
.parallelism(8))
+ .add(
+ RewriteDataFiles.builder()
+ .scheduleOnDataFileCount(10)
+ .partialProgressEnabled(true)
+ .partialProgressMaxCommits(10)
+ .maxRewriteBytes(1000L)
+ .targetFileSizeBytes(1000L)
+ .minFileSizeBytes(1000L)
+ .maxFileSizeBytes(1000L)
+ .minInputFiles(10)
+ .deleteFileThreshold(10)
+ .rewriteAll(false)
+ .maxFileGroupSizeBytes(1000L))
.append();
JobClient jobClient = null;
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
new file mode 100644
index 0000000000..618647259b
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
@@ -0,0 +1,417 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import static org.apache.iceberg.flink.SimpleDataUtil.createRecord;
+import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.COMMIT_TASK_NAME;
+import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.PLANNER_TASK_NAME;
+import static org.apache.iceberg.flink.maintenance.api.RewriteDataFiles.REWRITE_TASK_NAME;
+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.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;
+
+import java.util.List;
+import java.util.stream.StreamSupport;
+import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.SimpleDataUtil;
+import org.apache.iceberg.flink.maintenance.operator.MetricsReporterFactoryForTests;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Test;
+
+class TestRewriteDataFiles extends MaintenanceTaskTestBase {
+ @Test
+ void testRewriteUnpartitioned() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+ insert(table, 3, "c");
+ insert(table, 4, "d");
+
+ assertFileNum(table, 4, 0);
+
+ appendRewriteDataFiles(
+ RewriteDataFiles.builder()
+ .parallelism(2)
+ .deleteFileThreshold(10)
+ .targetFileSizeBytes(1_000_000L)
+ .maxFileGroupSizeBytes(10_000_000L)
+ .maxFileSizeBytes(2_000_000L)
+ .minFileSizeBytes(500_000L)
+ .minInputFiles(2)
+ .partialProgressEnabled(true)
+ .partialProgressMaxCommits(1)
+ .maxRewriteBytes(100_000L)
+ .rewriteAll(false));
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ assertFileNum(table, 1, 0);
+
+ SimpleDataUtil.assertTableRecords(
+ table,
+ ImmutableList.of(
+ createRecord(1, "a"),
+ createRecord(2, "b"),
+ createRecord(3, "c"),
+ createRecord(4, "d")));
+ }
+
+ @Test
+ void testRewritePartitioned() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+
+ assertFileNum(table, 4, 0);
+
+ appendRewriteDataFiles();
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ assertFileNum(table, 2, 0);
+
+ SimpleDataUtil.assertTableRecords(
+ table,
+ ImmutableList.of(
+ createRecord(1, "p1"),
+ createRecord(2, "p1"),
+ createRecord(3, "p2"),
+ createRecord(4, "p2")));
+ }
+
+ @Test
+ void testPlannerFailure() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ assertFileNum(table, 2, 0);
+
+ appendRewriteDataFiles();
+
+ runAndWaitForFailure(infra.env(), infra.source(), infra.sink());
+
+ // Check the metrics. The first task should be successful, but the second one should fail. This
+ // should be represented in the counters.
+ MetricsReporterFactoryForTests.assertCounters(
+ new ImmutableMap.Builder<List<String>, Long>()
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 1L)
+ .put(
+ ImmutableList.of(
+ REWRITE_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_NUM_METRIC),
+ 1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_NUM_METRIC),
+ 2L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .build());
+ }
+
+ @Test
+ void testUidAndSlotSharingGroup() {
+ createTable();
+
+ RewriteDataFiles.builder()
+ .slotSharingGroup(SLOT_SHARING_GROUP)
+ .uidSuffix(UID_SUFFIX)
+ .append(
+ infra.triggerStream(),
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
+ tableLoader(),
+ "OTHER",
+ "OTHER",
+ 1)
+ .sinkTo(infra.sink());
+
+ checkUidsAreSet(infra.env(), UID_SUFFIX);
+ checkSlotSharingGroupsAreSet(infra.env(), SLOT_SHARING_GROUP);
+ }
+
+ @Test
+ void testUidAndSlotSharingGroupUnset() {
+ createTable();
+
+ RewriteDataFiles.builder()
+ .append(
+ infra.triggerStream(),
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
+ tableLoader(),
+ UID_SUFFIX,
+ StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
+ 1)
+ .sinkTo(infra.sink());
+
+ checkUidsAreSet(infra.env(), null);
+ checkSlotSharingGroupsAreSet(infra.env(), null);
+ }
+
+ @Test
+ void testMetrics() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ assertFileNum(table, 2, 0);
+
+ appendRewriteDataFiles();
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ // Check the metrics
+ MetricsReporterFactoryForTests.assertCounters(
+ new ImmutableMap.Builder<List<String>, Long>()
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ REWRITE_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_NUM_METRIC),
+ 1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_NUM_METRIC),
+ 2L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .build());
+ }
+
+ @Test
+ void testV2Table() throws Exception {
+ Table table = createTableWithDelete();
+ update(table, 1, null, "a", "b");
+ update(table, 1, "b", "c");
+
+ assertFileNum(table, 2, 3);
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of(createRecord(1, "c")));
+
+ appendRewriteDataFiles();
+
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ // After #11131 we don't remove the delete files
+ assertFileNum(table, 1, 3);
+
+ SimpleDataUtil.assertTableRecords(table, ImmutableList.of(createRecord(1, "c")));
+
+ // Check the metrics
+ MetricsReporterFactoryForTests.assertCounters(
+ new ImmutableMap.Builder<List<String>, Long>()
+ .put(
+ ImmutableList.of(
+ PLANNER_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ REWRITE_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ERROR_COUNTER),
+ 0L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_NUM_METRIC),
+ 1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ ADDED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_NUM_METRIC),
+ 2L)
+ .put(
+ ImmutableList.of(
+ COMMIT_TASK_NAME + "[0]",
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ "0",
+ REMOVED_DATA_FILE_SIZE_METRIC),
+ -1L)
+ .build());
+ }
+
+ private void appendRewriteDataFiles() {
+ appendRewriteDataFiles(RewriteDataFiles.builder().rewriteAll(true));
+ }
+
+ private void appendRewriteDataFiles(RewriteDataFiles.Builder builder) {
+ builder
+ .append(
+ infra.triggerStream(),
+ DUMMY_TABLE_NAME,
+ DUMMY_TASK_NAME,
+ 0,
+ tableLoader(),
+ UID_SUFFIX,
+ StreamGraphGenerator.DEFAULT_SLOT_SHARING_GROUP,
+ 1)
+ .sinkTo(infra.sink());
+ }
+
+ private static void assertFileNum(
+ Table table, int expectedDataFileNum, int expectedDeleteFileNum) {
+ table.refresh();
+ assertThat(
+ table.currentSnapshot().dataManifests(table.io()).stream()
+ .flatMap(
+ m ->
+ StreamSupport.stream(
+ ManifestFiles.read(m, table.io(), table.specs()).spliterator(), false))
+ .count())
+ .isEqualTo(expectedDataFileNum);
+ assertThat(
+ table.currentSnapshot().deleteManifests(table.io()).stream()
+ .flatMap(
+ m ->
+ StreamSupport.stream(
+ ManifestFiles.readDeleteManifest(m, table.io(), table.specs())
+ .spliterator(),
+ false))
+ .count())
+ .isEqualTo(expectedDeleteFileNum);
+ }
+}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
index 36727c8862..78f34f6c16 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
@@ -33,16 +33,26 @@ import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.flink.streaming.api.transformations.SinkTransformation;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TestHelpers;
+import org.apache.iceberg.data.FileHelpers;
import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.flink.HadoopCatalogExtension;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.flink.TableLoader;
import org.apache.iceberg.flink.TestFixtures;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
import org.apache.iceberg.flink.maintenance.api.TriggerLockFactory;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -126,12 +136,85 @@ public class OperatorTestBase {
ImmutableMap.of("format-version", "2", "write.upsert.enabled", "true"));
}
+ protected static Table createPartitionedTable() {
+ return CATALOG_EXTENSION
+ .catalog()
+ .createTable(
+ TestFixtures.TABLE_IDENTIFIER,
+ SimpleDataUtil.SCHEMA,
+ PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build(),
+ null,
+ ImmutableMap.of("flink.max-continuous-empty-commits", "100000"));
+ }
+
protected void insert(Table table, Integer id, String data) throws IOException {
new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
.appendToTable(Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
table.refresh();
}
+ /**
+ * For the same identifier column id this methods simulate the following row operations: <tr>
+ * <li>add an equality delete on oldData
+ * <li>insert newData </tr>
+ *
+ * @param table to modify
+ * @param id the identifier column id
+ * @param oldData the old data to be deleted
+ * @param newData the new data to be inserted
+ */
+ protected void update(Table table, Integer id, String oldData, String newData)
+ throws IOException {
+ DataFile dataFile =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .writeFile(Lists.newArrayList(SimpleDataUtil.createRecord(id, newData)));
+ DeleteFile eqDelete = writeEqualityDelete(table, id, oldData);
+
+ table.newRowDelta().addRows(dataFile).addDeletes(eqDelete).commit();
+ }
+
+ /**
+ * For the same identifier column id this methods simulate the following row operations: <tr>
+ * <li>add an equality delete on oldData
+ * <li>insert tempData
+ * <li>add a position delete on tempData
+ * <li>insert newData </tr>
+ *
+ * @param table to modify
+ * @param id the identifier column id
+ * @param oldData the old data to be deleted
+ * @param tempData the temp data to be inserted and deleted with a position delete
+ * @param newData the new data to be inserted
+ */
+ protected void update(Table table, Integer id, String oldData, String tempData, String newData)
+ throws IOException {
+ DataFile dataFile =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .writeFile(
+ Lists.newArrayList(
+ SimpleDataUtil.createRecord(id, tempData),
+ SimpleDataUtil.createRecord(id, newData)));
+ DeleteFile eqDelete = writeEqualityDelete(table, id, oldData);
+ DeleteFile posDelete = writePosDelete(table, dataFile.path(), 0, id, tempData);
+
+ table.newRowDelta().addRows(dataFile).addDeletes(eqDelete).addDeletes(posDelete).commit();
+ }
+
+ protected void insertPartitioned(Table table, Integer id, String data) throws IOException {
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .appendToTable(
+ TestHelpers.Row.of(data), Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
+ table.refresh();
+ }
+
+ protected void insertFullPartitioned(Table table, Integer id, String data) throws IOException {
+ new GenericAppenderHelper(table, FileFormat.PARQUET, warehouseDir)
+ .appendToTable(
+ TestHelpers.Row.of(data, id),
+ Lists.newArrayList(SimpleDataUtil.createRecord(id, data)));
+ table.refresh();
+ }
+
protected void dropTable() {
CATALOG_EXTENSION.catalogLoader().loadCatalog().dropTable(TestFixtures.TABLE_IDENTIFIER);
}
@@ -211,6 +294,36 @@ public class OperatorTestBase {
return config;
}
+ private DeleteFile writeEqualityDelete(Table table, Integer id, String oldData)
+ throws IOException {
+ File file = File.createTempFile("junit", null, warehouseDir.toFile());
+ assertThat(file.delete()).isTrue();
+ return FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(file),
+ new PartitionData(PartitionSpec.unpartitioned().partitionType()),
+ Lists.newArrayList(SimpleDataUtil.createRecord(id, oldData)),
+ SCHEMA_WITH_PRIMARY_KEY);
+ }
+
+ private DeleteFile writePosDelete(
+ Table table, CharSequence path, Integer pos, Integer id, String oldData) throws IOException {
+ File file = File.createTempFile("junit", null, warehouseDir.toFile());
+ assertThat(file.delete()).isTrue();
+ PositionDelete<GenericRecord> posDelete = PositionDelete.create();
+ GenericRecord nested = GenericRecord.create(table.schema());
+ nested.set(0, id);
+ nested.set(1, oldData);
+ posDelete.set(path, pos, nested);
+ return FileHelpers.writePosDeleteFile(
+ table, Files.localOutput(file), null, Lists.newArrayList(posDelete));
+ }
+
+ static void trigger(OneInputStreamOperatorTestHarness<Trigger, ?> harness) throws Exception {
+ long time = System.currentTimeMillis();
+ harness.processElement(Trigger.create(time, 0), time);
+ }
+
private static class MemoryLock implements TriggerLockFactory.Lock {
volatile boolean locked = false;
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java
new file mode 100644
index 0000000000..6e43009e08
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/RewriteUtil.java
@@ -0,0 +1,83 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_FILES;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Set;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+
+class RewriteUtil {
+ private RewriteUtil() {}
+
+ static List<DataFileRewritePlanner.PlannedGroup> planDataFileRewrite(TableLoader tableLoader)
+ throws Exception {
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader,
+ 11,
+ 10_000_000L,
+ ImmutableMap.of(MIN_INPUT_FILES, "2")))) {
+ testHarness.open();
+
+ OperatorTestBase.trigger(testHarness);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ return testHarness.extractOutputValues();
+ }
+ }
+
+ static List<DataFileRewriteRunner.ExecutedGroup> executeRewrite(
+ List<DataFileRewritePlanner.PlannedGroup> elements) throws Exception {
+ try (OneInputStreamOperatorTestHarness<
+ DataFileRewritePlanner.PlannedGroup, DataFileRewriteRunner.ExecutedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewriteRunner(
+ OperatorTestBase.DUMMY_TABLE_NAME, OperatorTestBase.DUMMY_TABLE_NAME, 0))) {
+ testHarness.open();
+
+ for (DataFileRewritePlanner.PlannedGroup element : elements) {
+ testHarness.processElement(element, System.currentTimeMillis());
+ }
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ return testHarness.extractOutputValues();
+ }
+ }
+
+ static Set<DataFile> newDataFiles(Table table) {
+ table.refresh();
+ return Sets.newHashSet(table.currentSnapshot().addedDataFiles(table.io()));
+ }
+}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java
new file mode 100644
index 0000000000..9e8f2ec921
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteCommitter.java
@@ -0,0 +1,278 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.executeRewrite;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
+import static org.apache.iceberg.metrics.CommitMetricsResult.TOTAL_DATA_FILES;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.Test;
+
+class TestDataFileRewriteCommitter extends OperatorTestBase {
+ @Test
+ void testUnpartitioned() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "p1");
+ insert(table, 2, "p2");
+ insert(table, 3, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(1);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(rewritten.get(0), EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+
+ assertDataFiles(
+ table, rewritten.get(0).group().addedFiles(), rewritten.get(0).group().rewrittenFiles(), 1);
+ }
+
+ @Test
+ void testPartitioned() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(2);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(2);
+ assertThat(rewritten.get(0).groupsPerCommit()).isEqualTo(1);
+ assertThat(rewritten.get(1).groupsPerCommit()).isEqualTo(1);
+ ensureDifferentGroups(rewritten);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(rewritten.get(0), EVENT_TIME);
+ assertDataFiles(
+ table,
+ rewritten.get(0).group().addedFiles(),
+ rewritten.get(0).group().rewrittenFiles(),
+ 3);
+
+ testHarness.processElement(rewritten.get(1), EVENT_TIME);
+ assertDataFiles(
+ table,
+ rewritten.get(1).group().addedFiles(),
+ rewritten.get(1).group().rewrittenFiles(),
+ 2);
+
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+ }
+
+ @Test
+ void testNewTable() throws Exception {
+ Table table = createTable();
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten;
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ insert(table, 1, "p1");
+ insert(table, 2, "p2");
+ insert(table, 3, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(1);
+
+ testHarness.processElement(rewritten.get(0), EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+
+ assertDataFiles(
+ table, rewritten.get(0).group().addedFiles(), rewritten.get(0).group().rewrittenFiles(), 1);
+ }
+
+ @Test
+ void testBatchSize() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+ insertPartitioned(table, 5, "p3");
+ insertPartitioned(table, 6, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(3);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(3);
+ ensureDifferentGroups(rewritten);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(0)), EVENT_TIME);
+ assertNoChange(table);
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(1)), EVENT_TIME);
+
+ Set<DataFile> added = Sets.newHashSet(rewritten.get(0).group().addedFiles());
+ added.addAll(rewritten.get(1).group().addedFiles());
+ Set<DataFile> removed = Sets.newHashSet(rewritten.get(0).group().rewrittenFiles());
+ removed.addAll(rewritten.get(1).group().rewrittenFiles());
+ assertDataFiles(table, added, removed, 4);
+
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(2)), EVENT_TIME);
+ assertNoChange(table);
+
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+
+ testHarness.processWatermark(EVENT_TIME);
+ assertThat(testHarness.extractOutputValues()).isEmpty();
+ }
+
+ // This should be committed on close
+ assertDataFiles(
+ table, rewritten.get(2).group().addedFiles(), rewritten.get(2).group().rewrittenFiles(), 3);
+ }
+
+ @Test
+ void testError() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+ insertPartitioned(table, 5, "p3");
+ insertPartitioned(table, 6, "p3");
+ insertPartitioned(table, 7, "p4");
+ insertPartitioned(table, 8, "p4");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(4);
+ List<DataFileRewriteRunner.ExecutedGroup> rewritten = executeRewrite(planned);
+ assertThat(rewritten).hasSize(4);
+
+ try (OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger>
+ testHarness = harness()) {
+ testHarness.open();
+
+ testHarness.processElement(setBatchSizeToTwo(rewritten.get(0)), EVENT_TIME);
+ assertNoChange(table);
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+
+ DataFileRewriteRunner.ExecutedGroup group = spy(setBatchSizeToTwo(rewritten.get(1)));
+ when(group.group()).thenThrow(new RuntimeException("Testing error"));
+ testHarness.processElement(group, EVENT_TIME);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ assertThat(
+ testHarness
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .poll()
+ .getValue()
+ .getMessage())
+ .contains("Testing error");
+ }
+ }
+
+ private OneInputStreamOperatorTestHarness<DataFileRewriteRunner.ExecutedGroup, Trigger> harness()
+ throws Exception {
+ return new OneInputStreamOperatorTestHarness<>(
+ new DataFileRewriteCommitter(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader()));
+ }
+
+ private static DataFileRewriteRunner.ExecutedGroup setBatchSizeToTwo(
+ DataFileRewriteRunner.ExecutedGroup from) {
+ return new DataFileRewriteRunner.ExecutedGroup(from.snapshotId(), 2, from.group());
+ }
+
+ // Ensure that the groups are different, so the tests are not accidentally passing
+ private static void ensureDifferentGroups(List<DataFileRewriteRunner.ExecutedGroup> rewritten) {
+ List<String> resultFiles =
+ rewritten.stream()
+ .flatMap(task -> task.group().addedFiles().stream().map(ContentFile::location))
+ .collect(Collectors.toList());
+ assertThat(resultFiles).hasSize(Set.copyOf(resultFiles).size());
+ }
+
+ /**
+ * Assert that the number of the data files in the table is as expected. Additionally, tests that
+ * the last commit contains the expected added and removed files.
+ *
+ * @param table the table to check
+ * @param expectedAdded the expected added data files
+ * @param expectedRemoved the expected removed data files
+ * @param expectedCurrent the expected current data files count
+ */
+ private static void assertDataFiles(
+ Table table,
+ Set<DataFile> expectedAdded,
+ Set<DataFile> expectedRemoved,
+ long expectedCurrent) {
+ table.refresh();
+
+ assertThat(table.currentSnapshot().summary().get(TOTAL_DATA_FILES))
+ .isEqualTo(String.valueOf(expectedCurrent));
+ Set<DataFile> actualAdded = Sets.newHashSet(table.currentSnapshot().addedDataFiles(table.io()));
+ Set<DataFile> actualRemoved =
+ Sets.newHashSet(table.currentSnapshot().removedDataFiles(table.io()));
+ assertThat(actualAdded.stream().map(DataFile::location).collect(Collectors.toSet()))
+ .isEqualTo(expectedAdded.stream().map(DataFile::location).collect(Collectors.toSet()));
+ assertThat(actualRemoved.stream().map(DataFile::location).collect(Collectors.toSet()))
+ .isEqualTo(expectedRemoved.stream().map(DataFile::location).collect(Collectors.toSet()));
+ }
+
+ private static void assertNoChange(Table table) {
+ long original = table.currentSnapshot().snapshotId();
+ table.refresh();
+
+ assertThat(table.currentSnapshot().snapshotId()).isEqualTo(original);
+ }
+}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
new file mode 100644
index 0000000000..1d7e29813a
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
@@ -0,0 +1,193 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_FILES;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.newDataFiles;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.Test;
+
+class TestDataFileRewritePlanner extends OperatorTestBase {
+ @Test
+ void testUnpartitioned() throws Exception {
+ Set<DataFile> expected = Sets.newHashSetWithExpectedSize(3);
+ Table table = createTable();
+ insert(table, 1, "a");
+ expected.addAll(newDataFiles(table));
+ insert(table, 2, "b");
+ expected.addAll(newDataFiles(table));
+ insert(table, 3, "c");
+ expected.addAll(newDataFiles(table));
+
+ List<DataFileRewritePlanner.PlannedGroup> actual = planDataFileRewrite(tableLoader());
+
+ assertThat(actual).hasSize(1);
+ assertRewriteFileGroup(actual.get(0), table, expected);
+ }
+
+ @Test
+ void testPartitioned() throws Exception {
+ Set<DataFile> expectedP1 = Sets.newHashSetWithExpectedSize(2);
+ Set<DataFile> expectedP2 = Sets.newHashSetWithExpectedSize(2);
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ expectedP1.addAll(newDataFiles(table));
+ insertPartitioned(table, 2, "p1");
+ expectedP1.addAll(newDataFiles(table));
+
+ insertPartitioned(table, 3, "p2");
+ expectedP2.addAll(newDataFiles(table));
+ insertPartitioned(table, 4, "p2");
+ expectedP2.addAll(newDataFiles(table));
+
+ // This should not participate in compaction, as there is no more files in the partition
+ insertPartitioned(table, 5, "p3");
+
+ List<DataFileRewritePlanner.PlannedGroup> actual = planDataFileRewrite(tableLoader());
+
+ assertThat(actual).hasSize(2);
+ if (actual.get(0).group().info().partition().get(0, String.class).equals("p1")) {
+ assertRewriteFileGroup(actual.get(0), table, expectedP1);
+ assertRewriteFileGroup(actual.get(1), table, expectedP2);
+ } else {
+ assertRewriteFileGroup(actual.get(0), table, expectedP2);
+ assertRewriteFileGroup(actual.get(1), table, expectedP1);
+ }
+ }
+
+ @Test
+ void testError() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader(),
+ 11,
+ 1L,
+ ImmutableMap.of(MIN_INPUT_FILES, "2")))) {
+ testHarness.open();
+
+ // Cause an exception
+ dropTable();
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ trigger(testHarness);
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ assertThat(
+ testHarness
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .poll()
+ .getValue()
+ .getMessage())
+ .contains("Table does not exist: ");
+ }
+ }
+
+ @Test
+ void testV2Table() throws Exception {
+ Table table = createTableWithDelete();
+ update(table, 1, null, "a", "b");
+ update(table, 1, "b", "c");
+
+ List<DataFileRewritePlanner.PlannedGroup> actual = planDataFileRewrite(tableLoader());
+
+ assertThat(actual).hasSize(1);
+ List<FileScanTask> tasks = actual.get(0).group().fileScanTasks();
+ assertThat(tasks).hasSize(2);
+ // Find the task with the deletes
+ FileScanTask withDelete = tasks.get(0).deletes().isEmpty() ? tasks.get(1) : tasks.get(0);
+ assertThat(withDelete.deletes()).hasSize(2);
+ assertThat(withDelete.deletes().stream().map(ContentFile::content).collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(FileContent.POSITION_DELETES, FileContent.EQUALITY_DELETES);
+ }
+
+ @Test
+ void testMaxRewriteBytes() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p2");
+ insertPartitioned(table, 4, "p2");
+
+ // First run with high maxRewriteBytes
+ List<DataFileRewritePlanner.PlannedGroup> planWithNoMaxRewriteBytes =
+ planDataFileRewrite(tableLoader());
+ assertThat(planWithNoMaxRewriteBytes).hasSize(2);
+
+ // Second run with low maxRewriteBytes, the 2nd group should be removed from the plan
+ long maxRewriteBytes =
+ planWithNoMaxRewriteBytes.get(0).group().fileScanTasks().get(0).sizeBytes()
+ + planWithNoMaxRewriteBytes.get(1).group().fileScanTasks().get(0).sizeBytes()
+ + 1;
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader(),
+ 11,
+ maxRewriteBytes,
+ ImmutableMap.of(MIN_INPUT_FILES, "2")))) {
+ testHarness.open();
+
+ OperatorTestBase.trigger(testHarness);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ // Only a single group is planned
+ assertThat(testHarness.extractOutputValues()).hasSize(1);
+ }
+ }
+
+ void assertRewriteFileGroup(
+ DataFileRewritePlanner.PlannedGroup plannedGroup, Table table, Set<DataFile> files) {
+ assertThat(plannedGroup.table().currentSnapshot().snapshotId())
+ .isEqualTo(table.currentSnapshot().snapshotId());
+ assertThat(plannedGroup.groupsPerCommit()).isEqualTo(1);
+ assertThat(
+ plannedGroup.group().fileScanTasks().stream()
+ .map(s -> s.file().location())
+ .collect(Collectors.toSet()))
+ .containsExactlyInAnyOrderElementsOf(
+ files.stream().map(ContentFile::location).collect(Collectors.toList()));
+ }
+}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
new file mode 100644
index 0000000000..6dc8fb3c02
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewriteRunner.java
@@ -0,0 +1,355 @@
+/*
+ * 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.iceberg.flink.maintenance.operator;
+
+import static org.apache.iceberg.actions.RewriteDataFiles.TARGET_FILE_SIZE_BYTES;
+import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_FILES;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.executeRewrite;
+import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileContent;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionData;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.flink.maintenance.api.Trigger;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class TestDataFileRewriteRunner extends OperatorTestBase {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testExecute(boolean partitioned) throws Exception {
+ Table table;
+ PartitionData partition;
+ if (partitioned) {
+ table = createPartitionedTable();
+ partition = new PartitionData(table.spec().partitionType());
+ partition.set(0, "p1");
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ insertPartitioned(table, 3, "p1");
+ } else {
+ table = createTable();
+ partition = new PartitionData(PartitionSpec.unpartitioned().partitionType());
+ insert(table, 1, "p1");
+ insert(table, 2, "p1");
+ insert(table, 3, "p1");
+ }
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ List<DataFileRewriteRunner.ExecutedGroup> actual = executeRewrite(planned);
+ assertThat(actual).hasSize(1);
+
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(
+ ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"), ImmutableList.of(3, "p1"))),
+ 1,
+ ImmutableSet.of(partition));
+ }
+
+ @Test
+ void testPartitionSpecChange() throws Exception {
+ Table table = createPartitionedTable();
+ insertPartitioned(table, 1, "p1");
+ insertPartitioned(table, 2, "p1");
+ PartitionData oldPartition = new PartitionData(table.spec().partitionType());
+ oldPartition.set(0, "p1");
+
+ try (OneInputStreamOperatorTestHarness<
+ DataFileRewritePlanner.PlannedGroup, DataFileRewriteRunner.ExecutedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewriteRunner(
+ OperatorTestBase.DUMMY_TABLE_NAME, OperatorTestBase.DUMMY_TABLE_NAME, 0))) {
+ testHarness.open();
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+
+ testHarness.processElement(planned.get(0), System.currentTimeMillis());
+ List<DataFileRewriteRunner.ExecutedGroup> actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(1);
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"))),
+ 1,
+ ImmutableSet.of(oldPartition));
+
+ insertPartitioned(table, 3, "p1");
+
+ planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+
+ testHarness.processElement(planned.get(0), System.currentTimeMillis());
+ actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(2);
+ assertRewriteFileGroup(
+ actual.get(1),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(
+ ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"), ImmutableList.of(3, "p1"))),
+ 1,
+ ImmutableSet.of(oldPartition));
+
+ // Alter the table schema
+ table.updateSpec().addField("id").commit();
+ // Insert some now data
+ insertFullPartitioned(table, 4, "p1");
+ insertFullPartitioned(table, 4, "p1");
+ PartitionData newPartition = new PartitionData(table.spec().partitionType());
+ newPartition.set(0, "p1");
+ newPartition.set(1, 4);
+ table.refresh();
+
+ planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(2);
+ DataFileRewritePlanner.PlannedGroup oldCompact = planned.get(0);
+ DataFileRewritePlanner.PlannedGroup newCompact = planned.get(1);
+ if (oldCompact.group().inputFileNum() == 2) {
+ newCompact = planned.get(0);
+ oldCompact = planned.get(1);
+ }
+
+ testHarness.processElement(newCompact, System.currentTimeMillis());
+ actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(3);
+ assertRewriteFileGroup(
+ actual.get(2),
+ table,
+ records(
+ table.schema(),
+ ImmutableList.of(ImmutableList.of(4, "p1"), ImmutableList.of(4, "p1"))),
+ 1,
+ ImmutableSet.of(newPartition));
+
+ testHarness.processElement(oldCompact, System.currentTimeMillis());
+ actual = testHarness.extractOutputValues();
+ assertThat(actual).hasSize(4);
+ PartitionData[] transformedPartitions = {
+ newPartition.copy(), newPartition.copy(), newPartition.copy()
+ };
+ transformedPartitions[0].set(1, 1);
+ transformedPartitions[1].set(1, 2);
+ transformedPartitions[2].set(1, 3);
+ assertRewriteFileGroup(
+ actual.get(3),
+ table,
+ records(
+ table.schema(),
+ ImmutableSet.of(
+ ImmutableList.of(1, "p1"), ImmutableList.of(2, "p1"), ImmutableList.of(3, "p1"))),
+ 3,
+ Sets.newHashSet(transformedPartitions));
+ }
+ }
+
+ @Test
+ void testError() throws Exception {
+ Table table = createTable();
+ insert(table, 1, "a");
+ insert(table, 2, "b");
+
+ try (OneInputStreamOperatorTestHarness<
+ DataFileRewritePlanner.PlannedGroup, DataFileRewriteRunner.ExecutedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewriteRunner(
+ OperatorTestBase.DUMMY_TABLE_NAME, OperatorTestBase.DUMMY_TABLE_NAME, 0))) {
+ testHarness.open();
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+ // Cause an exception
+ dropTable();
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ testHarness.processElement(planned.get(0), System.currentTimeMillis());
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).hasSize(1);
+ assertThat(
+ testHarness
+ .getSideOutput(TaskResultAggregator.ERROR_STREAM)
+ .poll()
+ .getValue()
+ .getMessage())
+ .contains("File does not exist: ");
+ }
+ }
+
+ @Test
+ void testV2Table() throws Exception {
+ Table table = createTableWithDelete();
+ update(table, 1, null, "a", "b");
+ update(table, 1, "b", "c");
+
+ List<DataFileRewritePlanner.PlannedGroup> planned = planDataFileRewrite(tableLoader());
+ assertThat(planned).hasSize(1);
+
+ List<DataFileRewriteRunner.ExecutedGroup> actual = executeRewrite(planned);
+ assertThat(actual).hasSize(1);
+
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ records(table.schema(), ImmutableSet.of(ImmutableList.of(1, "c"))),
+ 1,
+ ImmutableSet.of(new PartitionData(PartitionSpec.unpartitioned().partitionType())));
+ }
+
+ @Test
+ void testSplitSize() throws Exception {
+ Table table = createTable();
+
+ File dataDir = new File(new Path(table.location(), "data").toUri().getPath());
+ dataDir.mkdir();
+ GenericAppenderHelper dataAppender =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, dataDir.toPath());
+ List<Record> expected = Lists.newArrayListWithExpectedSize(4000);
+ for (int i = 0; i < 4; ++i) {
+ List<Record> batch = RandomGenericData.generate(table.schema(), 1000, 10 + i);
+ dataAppender.appendToTable(batch);
+ expected.addAll(batch);
+ }
+
+ // First run with high target file size
+ List<DataFileRewritePlanner.PlannedGroup> planWithNoTargetFileSize =
+ planDataFileRewrite(tableLoader());
+ assertThat(planWithNoTargetFileSize).hasSize(1);
+
+ // Second run with low target file size
+ long targetFileSize =
+ planWithNoTargetFileSize.get(0).group().fileScanTasks().get(0).sizeBytes()
+ + planWithNoTargetFileSize.get(0).group().fileScanTasks().get(1).sizeBytes();
+ List<DataFileRewritePlanner.PlannedGroup> planned;
+ try (OneInputStreamOperatorTestHarness<Trigger, DataFileRewritePlanner.PlannedGroup>
+ testHarness =
+ ProcessFunctionTestHarnesses.forProcessFunction(
+ new DataFileRewritePlanner(
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ OperatorTestBase.DUMMY_TABLE_NAME,
+ 0,
+ tableLoader(),
+ 11,
+ 10_000_000,
+ ImmutableMap.of(
+ MIN_INPUT_FILES,
+ "2",
+ TARGET_FILE_SIZE_BYTES,
+ String.valueOf(targetFileSize))))) {
+ testHarness.open();
+
+ OperatorTestBase.trigger(testHarness);
+
+ assertThat(testHarness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNull();
+ planned = testHarness.extractOutputValues();
+ assertThat(planned).hasSize(1);
+ }
+
+ List<DataFileRewriteRunner.ExecutedGroup> actual = executeRewrite(planned);
+ assertThat(actual).hasSize(1);
+
+ assertRewriteFileGroup(
+ actual.get(0),
+ table,
+ expected,
+ 2,
+ ImmutableSet.of(new PartitionData(PartitionSpec.unpartitioned().partitionType())));
+ }
+
+ void assertRewriteFileGroup(
+ DataFileRewriteRunner.ExecutedGroup actual,
+ Table table,
+ Collection<Record> expectedRecords,
+ int expectedFileNum,
+ Set<StructLike> expectedPartitions)
+ throws IOException {
+ assertThat(actual.snapshotId()).isEqualTo(table.currentSnapshot().snapshotId());
+ assertThat(actual.groupsPerCommit()).isEqualTo(1);
+ assertThat(actual.group().addedFiles()).hasSize(expectedFileNum);
+ Collection<Record> writtenRecords = Lists.newArrayListWithExpectedSize(expectedRecords.size());
+ Set<StructLike> writtenPartitions = Sets.newHashSetWithExpectedSize(expectedPartitions.size());
+ for (DataFile newDataFile : actual.group().addedFiles()) {
+ assertThat(newDataFile.format()).isEqualTo(FileFormat.PARQUET);
+ assertThat(newDataFile.content()).isEqualTo(FileContent.DATA);
+ assertThat(newDataFile.keyMetadata()).isNull();
+ writtenPartitions.add(newDataFile.partition());
+
+ try (CloseableIterable<Record> reader =
+ Parquet.read(table.io().newInputFile(newDataFile.location()))
+ .project(table.schema())
+ .createReaderFunc(
+ fileSchema -> GenericParquetReaders.buildReader(table.schema(), fileSchema))
+ .build()) {
+ List<Record> newRecords = Lists.newArrayList(reader);
+ assertThat(newRecords).hasSize((int) newDataFile.recordCount());
+ writtenRecords.addAll(newRecords);
+ }
+ }
+
+ assertThat(writtenRecords).containsExactlyInAnyOrderElementsOf(expectedRecords);
+ assertThat(writtenPartitions).isEqualTo(expectedPartitions);
+ }
+
+ private List<Record> records(Schema schema, Collection<List<Object>> data) {
+ GenericRecord record = GenericRecord.create(schema);
+
+ ImmutableList.Builder<Record> builder = ImmutableList.builder();
+ data.forEach(
+ recordData ->
+ builder.add(
+ record.copy(ImmutableMap.of("id", recordData.get(0), "data", recordData.get(1)))));
+
+ return builder.build();
+ }
+}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java
index d70c4aafd5..7511e1029b 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDeleteFilesProcessor.java
@@ -90,8 +90,7 @@ class TestDeleteFilesProcessor extends OperatorTestBase {
tableLoader().open();
try (OneInputStreamOperatorTestHarness<String, Void> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new DeleteFilesProcessor(0, DUMMY_TASK_NAME, tableLoader.loadTable(), 10),
- StringSerializer.INSTANCE)) {
+ new DeleteFilesProcessor(table, DUMMY_TASK_NAME, 0, 10), StringSerializer.INSTANCE)) {
testHarness.open();
testHarness.processElement(fileName, System.currentTimeMillis());
testHarness.processWatermark(EVENT_TIME);