This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new e8c614878d Flink: Backport PR #10526 to v1.18 and v1.20 (#11018)
e8c614878d is described below
commit e8c614878dc8f5fd24ec1bbdbd325a7a5d9bfd3f
Author: Qishang Zhong <[email protected]>
AuthorDate: Sat Aug 31 02:53:47 2024 +0800
Flink: Backport PR #10526 to v1.18 and v1.20 (#11018)
---
.../org/apache/iceberg/flink/sink/FlinkSink.java | 11 +-
.../iceberg/flink/sink/FlinkWriteResult.java | 40 ++++
.../iceberg/flink/sink/IcebergFilesCommitter.java | 48 +++--
.../iceberg/flink/sink/IcebergStreamWriter.java | 13 +-
.../flink/sink/TestCompressionSettings.java | 11 +-
.../flink/sink/TestIcebergFilesCommitter.java | 240 +++++++++++++++------
.../flink/sink/TestIcebergStreamWriter.java | 61 ++++--
.../org/apache/iceberg/flink/sink/FlinkSink.java | 11 +-
.../iceberg/flink/sink/FlinkWriteResult.java | 40 ++++
.../iceberg/flink/sink/IcebergFilesCommitter.java | 48 +++--
.../iceberg/flink/sink/IcebergStreamWriter.java | 13 +-
.../flink/sink/TestCompressionSettings.java | 11 +-
.../flink/sink/TestIcebergFilesCommitter.java | 240 +++++++++++++++------
.../flink/sink/TestIcebergStreamWriter.java | 61 ++++--
14 files changed, 598 insertions(+), 250 deletions(-)
diff --git
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index 2256d1e874..5cd43a46de 100644
---
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -67,7 +67,6 @@ import org.apache.iceberg.flink.sink.shuffle.RangePartitioner;
import org.apache.iceberg.flink.sink.shuffle.StatisticsOrRecord;
import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
-import org.apache.iceberg.io.WriteResult;
import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -420,7 +419,7 @@ public class FlinkSink {
distributeDataStream(rowDataInput, equalityFieldIds, flinkRowType,
writerParallelism);
// Add parallel writers that append rows to files
- SingleOutputStreamOperator<WriteResult> writerStream =
+ SingleOutputStreamOperator<FlinkWriteResult> writerStream =
appendWriter(distributeStream, flinkRowType, equalityFieldIds,
writerParallelism);
// Add single-parallelism committer that commits files
@@ -487,7 +486,7 @@ public class FlinkSink {
}
private SingleOutputStreamOperator<Void> appendCommitter(
- SingleOutputStreamOperator<WriteResult> writerStream) {
+ SingleOutputStreamOperator<FlinkWriteResult> writerStream) {
IcebergFilesCommitter filesCommitter =
new IcebergFilesCommitter(
tableLoader,
@@ -507,7 +506,7 @@ public class FlinkSink {
return committerStream;
}
- private SingleOutputStreamOperator<WriteResult> appendWriter(
+ private SingleOutputStreamOperator<FlinkWriteResult> appendWriter(
DataStream<RowData> input,
RowType flinkRowType,
List<Integer> equalityFieldIds,
@@ -545,11 +544,11 @@ public class FlinkSink {
IcebergStreamWriter<RowData> streamWriter =
createStreamWriter(tableSupplier, flinkWriteConf, flinkRowType,
equalityFieldIds);
- SingleOutputStreamOperator<WriteResult> writerStream =
+ SingleOutputStreamOperator<FlinkWriteResult> writerStream =
input
.transform(
operatorName(ICEBERG_STREAM_WRITER_NAME),
- TypeInformation.of(WriteResult.class),
+ TypeInformation.of(FlinkWriteResult.class),
streamWriter)
.setParallelism(writerParallelism);
if (uidPrefix != null) {
diff --git
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
new file mode 100644
index 0000000000..317fb169ae
--- /dev/null
+++
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
@@ -0,0 +1,40 @@
+/*
+ * 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.sink;
+
+import java.io.Serializable;
+import org.apache.iceberg.io.WriteResult;
+
+public class FlinkWriteResult implements Serializable {
+ private final long checkpointId;
+ private final WriteResult writeResult;
+
+ public FlinkWriteResult(long checkpointId, WriteResult writeResult) {
+ this.checkpointId = checkpointId;
+ this.writeResult = writeResult;
+ }
+
+ public long checkpointId() {
+ return checkpointId;
+ }
+
+ public WriteResult writeResult() {
+ return writeResult;
+ }
+}
diff --git
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
index b9bceaa931..7108c20083 100644
---
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
+++
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
@@ -63,7 +63,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
class IcebergFilesCommitter extends AbstractStreamOperator<Void>
- implements OneInputStreamOperator<WriteResult, Void>, BoundedOneInput {
+ implements OneInputStreamOperator<FlinkWriteResult, Void>, BoundedOneInput
{
private static final long serialVersionUID = 1L;
private static final long INITIAL_CHECKPOINT_ID = -1L;
@@ -96,7 +96,7 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
// The completed files cache for current checkpoint. Once the snapshot
barrier received, it will
// be flushed to the 'dataFilesPerCheckpoint'.
- private final List<WriteResult> writeResultsOfCurrentCkpt =
Lists.newArrayList();
+ private final Map<Long, List<WriteResult>> writeResultsSinceLastSnapshot =
Maps.newHashMap();
private final String branch;
// It will have an unique identifier for one job.
@@ -212,7 +212,8 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
// Update the checkpoint state.
long startNano = System.nanoTime();
- dataFilesPerCheckpoint.put(checkpointId, writeToManifest(checkpointId));
+ writeToManifestUptoLatestCheckpoint(checkpointId);
+
// Reset the snapshot state to the latest state.
checkpointsState.clear();
checkpointsState.add(dataFilesPerCheckpoint);
@@ -220,8 +221,6 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
jobIdState.clear();
jobIdState.add(flinkJobId);
- // Clear the local buffer for current checkpoint.
- writeResultsOfCurrentCkpt.clear();
committerMetrics.checkpointDuration(
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
}
@@ -426,30 +425,45 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
}
@Override
- public void processElement(StreamRecord<WriteResult> element) {
- this.writeResultsOfCurrentCkpt.add(element.getValue());
+ public void processElement(StreamRecord<FlinkWriteResult> element) {
+ FlinkWriteResult flinkWriteResult = element.getValue();
+ List<WriteResult> writeResults =
+ writeResultsSinceLastSnapshot.computeIfAbsent(
+ flinkWriteResult.checkpointId(), k -> Lists.newArrayList());
+ writeResults.add(flinkWriteResult.writeResult());
}
@Override
public void endInput() throws IOException {
// Flush the buffered data files into 'dataFilesPerCheckpoint' firstly.
- long currentCheckpointId = Long.MAX_VALUE;
- dataFilesPerCheckpoint.put(currentCheckpointId,
writeToManifest(currentCheckpointId));
- writeResultsOfCurrentCkpt.clear();
-
+ long currentCheckpointId = IcebergStreamWriter.END_INPUT_CHECKPOINT_ID;
+ writeToManifestUptoLatestCheckpoint(currentCheckpointId);
commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, operatorUniqueId,
currentCheckpointId);
}
+ private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws
IOException {
+ if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+ dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
+ writeResultsSinceLastSnapshot.entrySet()) {
+ dataFilesPerCheckpoint.put(
+ writeResultsOfCheckpoint.getKey(),
+ writeToManifest(writeResultsOfCheckpoint.getKey(),
writeResultsOfCheckpoint.getValue()));
+ }
+
+ // Clear the local buffer for current checkpoint.
+ writeResultsSinceLastSnapshot.clear();
+ }
+
/**
* Write all the complete data files to a newly created manifest file and
return the manifest's
* avro serialized bytes.
*/
- private byte[] writeToManifest(long checkpointId) throws IOException {
- if (writeResultsOfCurrentCkpt.isEmpty()) {
- return EMPTY_MANIFEST_DATA;
- }
-
- WriteResult result =
WriteResult.builder().addAll(writeResultsOfCurrentCkpt).build();
+ private byte[] writeToManifest(long checkpointId, List<WriteResult>
writeResults)
+ throws IOException {
+ WriteResult result = WriteResult.builder().addAll(writeResults).build();
DeltaManifests deltaManifests =
FlinkManifestUtil.writeCompletedFiles(
result, () -> manifestOutputFileFactory.create(checkpointId),
spec);
diff --git
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
index 9ea0349fb0..bb5efe982e 100644
---
a/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
+++
b/flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
@@ -29,10 +29,11 @@ import org.apache.iceberg.io.TaskWriter;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-class IcebergStreamWriter<T> extends AbstractStreamOperator<WriteResult>
- implements OneInputStreamOperator<T, WriteResult>, BoundedOneInput {
+class IcebergStreamWriter<T> extends AbstractStreamOperator<FlinkWriteResult>
+ implements OneInputStreamOperator<T, FlinkWriteResult>, BoundedOneInput {
private static final long serialVersionUID = 1L;
+ static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
private final String fullTableName;
private final TaskWriterFactory<T> taskWriterFactory;
@@ -63,7 +64,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
- flush();
+ flush(checkpointId);
this.writer = taskWriterFactory.create();
}
@@ -89,7 +90,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
// Note that if the task is not closed after calling endInput, checkpoint
may be triggered again
// causing files to be sent repeatedly, the writer is marked as null after
the last file is sent
// to guard against duplicated writes.
- flush();
+ flush(END_INPUT_CHECKPOINT_ID);
}
@Override
@@ -102,7 +103,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
}
/** close all open files and emit files to downstream committer operator */
- private void flush() throws IOException {
+ private void flush(long checkpointId) throws IOException {
if (writer == null) {
return;
}
@@ -110,7 +111,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
long startNano = System.nanoTime();
WriteResult result = writer.complete();
writerMetrics.updateFlushResult(result);
- output.collect(new StreamRecord<>(result));
+ output.collect(new StreamRecord<>(new FlinkWriteResult(checkpointId,
result)));
writerMetrics.flushDuration(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNano));
// Set writer to null to prevent duplicate flushes in the corner case of
diff --git
a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
index 8faae1b05a..3299e7a977 100644
---
a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
+++
b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
@@ -40,7 +40,6 @@ import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.io.BaseTaskWriter;
import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
@@ -210,8 +209,10 @@ public class TestCompressionSettings {
.containsEntry(TableProperties.ORC_COMPRESSION_STRATEGY, "speed");
}
- private static OneInputStreamOperatorTestHarness<RowData, WriteResult>
createIcebergStreamWriter(
- Table icebergTable, TableSchema flinkSchema, Map<String, String>
override) throws Exception {
+ private static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
+ createIcebergStreamWriter(
+ Table icebergTable, TableSchema flinkSchema, Map<String, String>
override)
+ throws Exception {
RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(),
flinkSchema);
FlinkWriteConf flinkWriteConfig =
new FlinkWriteConf(
@@ -219,7 +220,7 @@ public class TestCompressionSettings {
IcebergStreamWriter<RowData> streamWriter =
FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig,
flinkRowType, null);
- OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+ OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
harness.setup();
@@ -230,7 +231,7 @@ public class TestCompressionSettings {
private static Map<String, String> appenderProperties(
Table table, TableSchema schema, Map<String, String> override) throws
Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter(table, schema, override)) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
diff --git
a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
index 948c7b3143..ac5babe119 100644
---
a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
+++
b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
@@ -129,7 +129,8 @@ public class TestIcebergFilesCommitter extends TestBase {
long timestamp = 0;
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -161,7 +162,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
long checkpointId = 0;
long timestamp = 0;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
@@ -176,8 +178,8 @@ public class TestIcebergFilesCommitter extends TestBase {
}
}
- private WriteResult of(DataFile dataFile) {
- return WriteResult.builder().addDataFiles(dataFile).build();
+ private FlinkWriteResult of(long checkpointId, DataFile dataFile) {
+ return new FlinkWriteResult(checkpointId,
WriteResult.builder().addDataFiles(dataFile).build());
}
@TestTemplate
@@ -193,7 +195,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobID = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobID)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobID)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -204,7 +207,7 @@ public class TestIcebergFilesCommitter extends TestBase {
for (int i = 1; i <= 3; i++) {
RowData rowData = SimpleDataUtil.createRowData(i, "hello" + i);
DataFile dataFile = writeDataFile("data-" + i,
ImmutableList.of(rowData));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(i, dataFile), ++timestamp);
rows.add(rowData);
harness.snapshot(i, ++timestamp);
@@ -233,7 +236,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -243,21 +247,21 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createRowData(1, "hello");
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ long firstCheckpointId = 1;
+ harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
- long firstCheckpointId = 1;
harness.snapshot(firstCheckpointId, ++timestamp);
assertFlinkManifests(1);
RowData row2 = SimpleDataUtil.createRowData(2, "world");
DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
- harness.processElement(of(dataFile2), ++timestamp);
+ long secondCheckpointId = 2;
+ harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 2. snapshotState for checkpoint#2
- long secondCheckpointId = 2;
harness.snapshot(secondCheckpointId, ++timestamp);
assertFlinkManifests(2);
@@ -286,7 +290,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -296,21 +301,21 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createRowData(1, "hello");
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ long firstCheckpointId = 1;
+ harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
- long firstCheckpointId = 1;
harness.snapshot(firstCheckpointId, ++timestamp);
assertFlinkManifests(1);
RowData row2 = SimpleDataUtil.createRowData(2, "world");
DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
- harness.processElement(of(dataFile2), ++timestamp);
+ long secondCheckpointId = 2;
+ harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 2. snapshotState for checkpoint#2
- long secondCheckpointId = 2;
harness.snapshot(secondCheckpointId, ++timestamp);
assertFlinkManifests(2);
@@ -337,7 +342,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -349,8 +355,8 @@ public class TestIcebergFilesCommitter extends TestBase {
expectedRows.add(row);
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row));
- harness.processElement(of(dataFile1), ++timestamp);
- snapshot = harness.snapshot(++checkpointId, ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile1), ++timestamp);
+ snapshot = harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -362,7 +368,8 @@ public class TestIcebergFilesCommitter extends TestBase {
}
// Restore from the given snapshot
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.initializeState(snapshot);
@@ -375,9 +382,9 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(2, "world");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -400,7 +407,8 @@ public class TestIcebergFilesCommitter extends TestBase {
List<RowData> expectedRows = Lists.newArrayList();
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -411,15 +419,16 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(1, "hello");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-1", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- snapshot = harness.snapshot(++checkpointId, ++timestamp);
+ snapshot = harness.snapshot(checkpointId, ++timestamp);
SimpleDataUtil.assertTableRows(table, ImmutableList.of(), branch);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
assertFlinkManifests(1);
}
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.initializeState(snapshot);
@@ -446,15 +455,15 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(2, "world");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- snapshot = harness.snapshot(++checkpointId, ++timestamp);
+ snapshot = harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
}
// Redeploying flink job from external checkpoint.
JobID newJobId = new JobID();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
createStreamSink(newJobId)) {
harness.setup();
harness.initializeState(snapshot);
@@ -473,9 +482,9 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(3, "foo");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-3", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -496,7 +505,7 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID oldJobId = new JobID();
OperatorID oldOperatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
createStreamSink(oldJobId)) {
harness.setup();
harness.open();
@@ -510,8 +519,8 @@ public class TestIcebergFilesCommitter extends TestBase {
tableRows.addAll(rows);
DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
- harness.processElement(of(dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -528,7 +537,7 @@ public class TestIcebergFilesCommitter extends TestBase {
timestamp = 0;
JobID newJobId = new JobID();
OperatorID newOperatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
createStreamSink(newJobId)) {
harness.setup();
harness.open();
@@ -542,8 +551,8 @@ public class TestIcebergFilesCommitter extends TestBase {
tableRows.addAll(rows);
DataFile dataFile = writeDataFile("data-new-1", rows);
- harness.processElement(of(dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -567,7 +576,8 @@ public class TestIcebergFilesCommitter extends TestBase {
int checkpointId = i / 3;
JobID jobId = jobs[jobIndex];
OperatorID operatorId = operatorIds[jobIndex];
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.open();
@@ -579,7 +589,7 @@ public class TestIcebergFilesCommitter extends TestBase {
tableRows.addAll(rows);
DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId + 1, dataFile), ++timestamp);
harness.snapshot(checkpointId + 1, ++timestamp);
assertFlinkManifests(1);
@@ -603,8 +613,10 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId1 = new OperatorID();
OperatorID operatorId2 = new OperatorID();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
- OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+ createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+ createStreamSink(jobId)) {
harness1.getStreamConfig().setOperatorID(operatorId1);
harness1.setup();
harness1.open();
@@ -620,14 +632,14 @@ public class TestIcebergFilesCommitter extends TestBase {
expectedRows.add(row1);
DataFile dataFile1 = writeDataFile("data-1-1", ImmutableList.of(row1));
- harness1.processElement(of(dataFile1), ++timestamp);
- snapshot1 = harness1.snapshot(++checkpointId, ++timestamp);
+ harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+ snapshot1 = harness1.snapshot(checkpointId, ++timestamp);
RowData row2 = SimpleDataUtil.createRowData(1, "hello2");
expectedRows.add(row2);
DataFile dataFile2 = writeDataFile("data-1-2", ImmutableList.of(row2));
- harness2.processElement(of(dataFile2), ++timestamp);
+ harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
snapshot2 = harness2.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(2);
@@ -643,8 +655,10 @@ public class TestIcebergFilesCommitter extends TestBase {
}
// Restore from the given snapshot
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
- OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+ createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+ createStreamSink(jobId)) {
harness1.getStreamConfig().setOperatorID(operatorId1);
harness1.setup();
harness1.initializeState(snapshot1);
@@ -668,13 +682,13 @@ public class TestIcebergFilesCommitter extends TestBase {
expectedRows.add(row1);
DataFile dataFile1 = writeDataFile("data-2-1", ImmutableList.of(row1));
- harness1.processElement(of(dataFile1), ++timestamp);
- harness1.snapshot(++checkpointId, ++timestamp);
+ harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+ harness1.snapshot(checkpointId, ++timestamp);
RowData row2 = SimpleDataUtil.createRowData(2, "world2");
expectedRows.add(row2);
DataFile dataFile2 = writeDataFile("data-2-2", ImmutableList.of(row2));
- harness2.processElement(of(dataFile2), ++timestamp);
+ harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
harness2.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(2);
@@ -694,7 +708,8 @@ public class TestIcebergFilesCommitter extends TestBase {
public void testBoundedStream() throws Exception {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -706,13 +721,14 @@ public class TestIcebergFilesCommitter extends TestBase {
List<RowData> tableRows =
Lists.newArrayList(SimpleDataUtil.createRowData(1, "word-1"));
DataFile dataFile = writeDataFile("data-1", tableRows);
- harness.processElement(of(dataFile), 1);
+ harness.processElement(of(IcebergStreamWriter.END_INPUT_CHECKPOINT_ID,
dataFile), 1);
((BoundedOneInput) harness.getOneInputOperator()).endInput();
assertFlinkManifests(0);
SimpleDataUtil.assertTableRows(table, tableRows, branch);
assertSnapshotSize(1);
- assertMaxCommittedCheckpointId(jobId, operatorId, Long.MAX_VALUE);
+ assertMaxCommittedCheckpointId(
+ jobId, operatorId, IcebergStreamWriter.END_INPUT_CHECKPOINT_ID);
assertThat(SimpleDataUtil.latestSnapshot(table, branch).summary())
.containsEntry("flink.test",
TestIcebergFilesCommitter.class.getName());
}
@@ -725,7 +741,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -735,7 +752,7 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createRowData(1, "hello");
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ harness.processElement(of(checkpoint, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
@@ -775,7 +792,8 @@ public class TestIcebergFilesCommitter extends TestBase {
OperatorID operatorId;
FileAppenderFactory<RowData> appenderFactory =
createDeletableAppenderFactory();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -784,7 +802,7 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createInsert(1, "aaa");
DataFile dataFile1 = writeDataFile("data-file-1",
ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ harness.processElement(of(checkpoint, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
@@ -816,13 +834,15 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData delete1 = SimpleDataUtil.createDelete(1, "aaa");
DeleteFile deleteFile1 =
writeEqDeleteFile(appenderFactory, "delete-file-1",
ImmutableList.of(delete1));
+ assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
harness.processElement(
-
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build(),
+ new FlinkWriteResult(
+ ++checkpoint,
+
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build()),
++timestamp);
- assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
// 5. snapshotState for checkpoint#2
- harness.snapshot(++checkpoint, ++timestamp);
+ harness.snapshot(checkpoint, ++timestamp);
assertFlinkManifests(2);
// 6. notifyCheckpointComplete for checkpoint#2
@@ -846,7 +866,8 @@ public class TestIcebergFilesCommitter extends TestBase {
OperatorID operatorId;
FileAppenderFactory<RowData> appenderFactory =
createDeletableAppenderFactory();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -860,7 +881,9 @@ public class TestIcebergFilesCommitter extends TestBase {
DeleteFile deleteFile1 =
writeEqDeleteFile(appenderFactory, "delete-file-1",
ImmutableList.of(delete3));
harness.processElement(
-
WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build(),
+ new FlinkWriteResult(
+ checkpoint,
+
WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build()),
++timestamp);
// The 1th snapshotState.
@@ -872,11 +895,13 @@ public class TestIcebergFilesCommitter extends TestBase {
DeleteFile deleteFile2 =
writeEqDeleteFile(appenderFactory, "delete-file-2",
ImmutableList.of(delete2));
harness.processElement(
-
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build(),
+ new FlinkWriteResult(
+ ++checkpoint,
+
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build()),
++timestamp);
// The 2nd snapshotState.
- harness.snapshot(++checkpoint, ++timestamp);
+ harness.snapshot(checkpoint, ++timestamp);
// Notify the 2nd snapshot to complete.
harness.notifyOfCompletedCheckpoint(checkpoint);
@@ -887,6 +912,79 @@ public class TestIcebergFilesCommitter extends TestBase {
}
}
+ /**
+ * The testcase is to simulate upserting to an Iceberg V2 table, and facing
the following
+ * scenario:
+ *
+ * <ul>
+ * <li>A specific row is updated
+ * <li>The prepareSnapshotPreBarrier triggered
+ * <li>Checkpoint failed for reasons outside of the Iceberg connector
+ * <li>The specific row is updated again in the second checkpoint as well
+ * <li>Second snapshot is triggered, and finished
+ * </ul>
+ *
+ * <p>Previously the files from the 2 snapshots were committed in a single
Iceberg commit, as a
+ * results duplicate rows were created in the table.
+ *
+ * @throws Exception Exception
+ */
+ @TestTemplate
+ public void testCommitMultipleCheckpointsForV2Table() throws Exception {
+ assumeThat(formatVersion)
+ .as("Only support equality-delete in format v2 or later.")
+ .isGreaterThan(1);
+
+ long timestamp = 0;
+ long checkpoint = 10;
+
+ JobID jobId = new JobID();
+ OperatorID operatorId;
+
+ FileAppenderFactory<RowData> appenderFactory =
+ new FlinkAppenderFactory(
+ table,
+ table.schema(),
+ FlinkSchemaUtil.convert(table.schema()),
+ table.properties(),
+ table.spec(),
+ new int[] {table.schema().findField("id").fieldId()},
+ table.schema(),
+ null);
+
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
+ harness.setup();
+ harness.open();
+ operatorId = harness.getOperator().getOperatorID();
+
+ assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
+
+ RowData insert1 = null;
+ RowData insert2 = null;
+ for (int i = 1; i <= 3; i++) {
+ insert1 = SimpleDataUtil.createInsert(1, "aaa" + i);
+ insert2 = SimpleDataUtil.createInsert(2, "bbb" + i);
+ DataFile dataFile = writeDataFile("data-file-" + i,
ImmutableList.of(insert1, insert2));
+ DeleteFile deleteFile =
+ writeEqDeleteFile(
+ appenderFactory, "delete-file-" + i, ImmutableList.of(insert1,
insert2));
+ harness.processElement(
+ new FlinkWriteResult(
+ ++checkpoint,
+
WriteResult.builder().addDataFiles(dataFile).addDeleteFiles(deleteFile).build()),
+ ++timestamp);
+ }
+
+ harness.snapshot(checkpoint, ++timestamp);
+ harness.notifyOfCompletedCheckpoint(checkpoint);
+ SimpleDataUtil.assertTableRows(table, ImmutableList.of(insert1,
insert2), branch);
+ assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
+ assertFlinkManifests(0);
+ assertThat(table.snapshots()).hasSize(3);
+ }
+ }
+
@TestTemplate
public void testSpecEvolution() throws Exception {
long timestamp = 0;
@@ -899,7 +997,8 @@ public class TestIcebergFilesCommitter extends TestBase {
DataFile dataFile;
int specId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -910,7 +1009,7 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData rowData = SimpleDataUtil.createRowData(checkpointId, "hello" +
checkpointId);
// table unpartitioned
dataFile = writeDataFile("data-" + checkpointId,
ImmutableList.of(rowData));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId, dataFile), ++timestamp);
rows.add(rowData);
harness.snapshot(checkpointId, ++timestamp);
@@ -929,7 +1028,7 @@ public class TestIcebergFilesCommitter extends TestBase {
rowData = SimpleDataUtil.createRowData(checkpointId, "hello" +
checkpointId);
// write data with old partition spec
dataFile = writeDataFile("data-" + checkpointId,
ImmutableList.of(rowData), oldSpec, null);
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId, dataFile), ++timestamp);
rows.add(rowData);
snapshot = harness.snapshot(checkpointId, ++timestamp);
@@ -947,7 +1046,8 @@ public class TestIcebergFilesCommitter extends TestBase {
}
// Restore from the given snapshot
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.initializeState(snapshot);
@@ -963,7 +1063,7 @@ public class TestIcebergFilesCommitter extends TestBase {
partition.set(0, checkpointId);
dataFile =
writeDataFile("data-" + checkpointId, ImmutableList.of(row),
table.spec(), partition);
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId, dataFile), ++timestamp);
rows.add(row);
harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
@@ -1089,7 +1189,7 @@ public class TestIcebergFilesCommitter extends TestBase {
assertThat(table.snapshots()).hasSize(expectedSnapshotSize);
}
- private OneInputStreamOperatorTestHarness<WriteResult, Void>
createStreamSink(JobID jobID)
+ private OneInputStreamOperatorTestHarness<FlinkWriteResult, Void>
createStreamSink(JobID jobID)
throws Exception {
TestOperatorFactory factory = TestOperatorFactory.of(table.location(),
branch, table.spec());
return new OneInputStreamOperatorTestHarness<>(factory,
createEnvironment(jobID));
@@ -1109,7 +1209,7 @@ public class TestIcebergFilesCommitter extends TestBase {
}
private static class TestOperatorFactory extends
AbstractStreamOperatorFactory<Void>
- implements OneInputStreamOperatorFactory<WriteResult, Void> {
+ implements OneInputStreamOperatorFactory<FlinkWriteResult, Void> {
private final String tablePath;
private final String branch;
private final PartitionSpec spec;
diff --git
a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
index 50283f7ad2..e13721a9f1 100644
---
a/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
+++
b/flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.api.DataTypes;
@@ -102,7 +103,7 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testWritingTable() throws Exception {
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
// The first checkpoint
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
@@ -111,7 +112,8 @@ public class TestIcebergStreamWriter {
testHarness.prepareSnapshotPreBarrier(checkpointId);
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -123,7 +125,8 @@ public class TestIcebergStreamWriter {
testHarness.prepareSnapshotPreBarrier(checkpointId);
expectedDataFiles = partitioned ? 4 : 2;
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -148,14 +151,15 @@ public class TestIcebergStreamWriter {
public void testSnapshotTwice() throws Exception {
long checkpointId = 1;
long timestamp = 1;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"),
timestamp++);
testHarness.processElement(SimpleDataUtil.createRowData(2, "world"),
timestamp);
testHarness.prepareSnapshotPreBarrier(checkpointId++);
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -163,7 +167,10 @@ public class TestIcebergStreamWriter {
for (int i = 0; i < 5; i++) {
testHarness.prepareSnapshotPreBarrier(checkpointId++);
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+ WriteResult.builder()
+ .addAll(getWriteResults(testHarness.extractOutputValues()))
+ .build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
}
@@ -172,14 +179,14 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testTableWithoutSnapshot() throws Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
assertThat(testHarness.extractOutputValues()).isEmpty();
}
// Even if we closed the iceberg stream writer, there's no orphan data
file.
assertThat(scanDataFiles()).isEmpty();
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
// Still not emit the data file yet, because there is no checkpoint.
@@ -212,7 +219,7 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -221,13 +228,15 @@ public class TestIcebergStreamWriter {
((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
// Datafiles should not be sent again
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -236,7 +245,7 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testBoundedStreamTriggeredEndInputBeforeTriggeringCheckpoint()
throws Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -244,13 +253,15 @@ public class TestIcebergStreamWriter {
testHarness.endInput();
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
testHarness.prepareSnapshotPreBarrier(1L);
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
// It should be ensured that after endInput is triggered, when
prepareSnapshotPreBarrier
// is triggered, write should only send WriteResult once
@@ -275,7 +286,7 @@ public class TestIcebergStreamWriter {
}
}
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
for (RowData row : rows) {
testHarness.processElement(row, 1);
@@ -283,7 +294,8 @@ public class TestIcebergStreamWriter {
// snapshot the operator.
testHarness.prepareSnapshotPreBarrier(1);
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(8);
@@ -346,13 +358,14 @@ public class TestIcebergStreamWriter {
record.copy(ImmutableMap.of("tinyint", 2, "smallint", 0, "int",
102)),
record.copy(ImmutableMap.of("tinyint", 3, "smallint", 32767,
"int", 103)));
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter(icebergTable, flinkSchema)) {
for (RowData row : rows) {
testHarness.processElement(row, 1);
}
testHarness.prepareSnapshotPreBarrier(1);
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(partitioned ? 3 : 1);
@@ -365,12 +378,18 @@ public class TestIcebergStreamWriter {
SimpleDataUtil.assertTableRecords(location, expected);
}
- private OneInputStreamOperatorTestHarness<RowData, WriteResult>
createIcebergStreamWriter()
+ private static List<WriteResult> getWriteResults(List<FlinkWriteResult>
flinkWriteResults) {
+ return flinkWriteResults.stream()
+ .map(FlinkWriteResult::writeResult)
+ .collect(Collectors.toList());
+ }
+
+ private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
createIcebergStreamWriter()
throws Exception {
return createIcebergStreamWriter(table, SimpleDataUtil.FLINK_SCHEMA);
}
- private OneInputStreamOperatorTestHarness<RowData, WriteResult>
createIcebergStreamWriter(
+ private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
createIcebergStreamWriter(
Table icebergTable, TableSchema flinkSchema) throws Exception {
RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(),
flinkSchema);
FlinkWriteConf flinkWriteConfig =
@@ -379,7 +398,7 @@ public class TestIcebergStreamWriter {
IcebergStreamWriter<RowData> streamWriter =
FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig,
flinkRowType, null);
- OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+ OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
harness.setup();
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index be2a8db030..c534314909 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -67,7 +67,6 @@ import org.apache.iceberg.flink.sink.shuffle.RangePartitioner;
import org.apache.iceberg.flink.sink.shuffle.StatisticsOrRecord;
import org.apache.iceberg.flink.sink.shuffle.StatisticsType;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
-import org.apache.iceberg.io.WriteResult;
import
org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -421,7 +420,7 @@ public class FlinkSink {
distributeDataStream(rowDataInput, equalityFieldIds, flinkRowType,
writerParallelism);
// Add parallel writers that append rows to files
- SingleOutputStreamOperator<WriteResult> writerStream =
+ SingleOutputStreamOperator<FlinkWriteResult> writerStream =
appendWriter(distributeStream, flinkRowType, equalityFieldIds,
writerParallelism);
// Add single-parallelism committer that commits files
@@ -488,7 +487,7 @@ public class FlinkSink {
}
private SingleOutputStreamOperator<Void> appendCommitter(
- SingleOutputStreamOperator<WriteResult> writerStream) {
+ SingleOutputStreamOperator<FlinkWriteResult> writerStream) {
IcebergFilesCommitter filesCommitter =
new IcebergFilesCommitter(
tableLoader,
@@ -508,7 +507,7 @@ public class FlinkSink {
return committerStream;
}
- private SingleOutputStreamOperator<WriteResult> appendWriter(
+ private SingleOutputStreamOperator<FlinkWriteResult> appendWriter(
DataStream<RowData> input,
RowType flinkRowType,
List<Integer> equalityFieldIds,
@@ -546,11 +545,11 @@ public class FlinkSink {
IcebergStreamWriter<RowData> streamWriter =
createStreamWriter(tableSupplier, flinkWriteConf, flinkRowType,
equalityFieldIds);
- SingleOutputStreamOperator<WriteResult> writerStream =
+ SingleOutputStreamOperator<FlinkWriteResult> writerStream =
input
.transform(
operatorName(ICEBERG_STREAM_WRITER_NAME),
- TypeInformation.of(WriteResult.class),
+ TypeInformation.of(FlinkWriteResult.class),
streamWriter)
.setParallelism(writerParallelism);
if (uidPrefix != null) {
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
new file mode 100644
index 0000000000..317fb169ae
--- /dev/null
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkWriteResult.java
@@ -0,0 +1,40 @@
+/*
+ * 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.sink;
+
+import java.io.Serializable;
+import org.apache.iceberg.io.WriteResult;
+
+public class FlinkWriteResult implements Serializable {
+ private final long checkpointId;
+ private final WriteResult writeResult;
+
+ public FlinkWriteResult(long checkpointId, WriteResult writeResult) {
+ this.checkpointId = checkpointId;
+ this.writeResult = writeResult;
+ }
+
+ public long checkpointId() {
+ return checkpointId;
+ }
+
+ public WriteResult writeResult() {
+ return writeResult;
+ }
+}
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
index 622daa8088..609deb621f 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergFilesCommitter.java
@@ -61,7 +61,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
class IcebergFilesCommitter extends AbstractStreamOperator<Void>
- implements OneInputStreamOperator<WriteResult, Void>, BoundedOneInput {
+ implements OneInputStreamOperator<FlinkWriteResult, Void>, BoundedOneInput
{
private static final long serialVersionUID = 1L;
private static final long INITIAL_CHECKPOINT_ID = -1L;
@@ -94,7 +94,7 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
// The completed files cache for current checkpoint. Once the snapshot
barrier received, it will
// be flushed to the 'dataFilesPerCheckpoint'.
- private final List<WriteResult> writeResultsOfCurrentCkpt =
Lists.newArrayList();
+ private final Map<Long, List<WriteResult>> writeResultsSinceLastSnapshot =
Maps.newHashMap();
private final String branch;
// It will have an unique identifier for one job.
@@ -210,7 +210,8 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
// Update the checkpoint state.
long startNano = System.nanoTime();
- dataFilesPerCheckpoint.put(checkpointId, writeToManifest(checkpointId));
+ writeToManifestUptoLatestCheckpoint(checkpointId);
+
// Reset the snapshot state to the latest state.
checkpointsState.clear();
checkpointsState.add(dataFilesPerCheckpoint);
@@ -218,8 +219,6 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
jobIdState.clear();
jobIdState.add(flinkJobId);
- // Clear the local buffer for current checkpoint.
- writeResultsOfCurrentCkpt.clear();
committerMetrics.checkpointDuration(
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano));
}
@@ -403,30 +402,45 @@ class IcebergFilesCommitter extends
AbstractStreamOperator<Void>
}
@Override
- public void processElement(StreamRecord<WriteResult> element) {
- this.writeResultsOfCurrentCkpt.add(element.getValue());
+ public void processElement(StreamRecord<FlinkWriteResult> element) {
+ FlinkWriteResult flinkWriteResult = element.getValue();
+ List<WriteResult> writeResults =
+ writeResultsSinceLastSnapshot.computeIfAbsent(
+ flinkWriteResult.checkpointId(), k -> Lists.newArrayList());
+ writeResults.add(flinkWriteResult.writeResult());
}
@Override
public void endInput() throws IOException {
// Flush the buffered data files into 'dataFilesPerCheckpoint' firstly.
- long currentCheckpointId = Long.MAX_VALUE;
- dataFilesPerCheckpoint.put(currentCheckpointId,
writeToManifest(currentCheckpointId));
- writeResultsOfCurrentCkpt.clear();
-
+ long currentCheckpointId = IcebergStreamWriter.END_INPUT_CHECKPOINT_ID;
+ writeToManifestUptoLatestCheckpoint(currentCheckpointId);
commitUpToCheckpoint(dataFilesPerCheckpoint, flinkJobId, operatorUniqueId,
currentCheckpointId);
}
+ private void writeToManifestUptoLatestCheckpoint(long checkpointId) throws
IOException {
+ if (!writeResultsSinceLastSnapshot.containsKey(checkpointId)) {
+ dataFilesPerCheckpoint.put(checkpointId, EMPTY_MANIFEST_DATA);
+ }
+
+ for (Map.Entry<Long, List<WriteResult>> writeResultsOfCheckpoint :
+ writeResultsSinceLastSnapshot.entrySet()) {
+ dataFilesPerCheckpoint.put(
+ writeResultsOfCheckpoint.getKey(),
+ writeToManifest(writeResultsOfCheckpoint.getKey(),
writeResultsOfCheckpoint.getValue()));
+ }
+
+ // Clear the local buffer for current checkpoint.
+ writeResultsSinceLastSnapshot.clear();
+ }
+
/**
* Write all the complete data files to a newly created manifest file and
return the manifest's
* avro serialized bytes.
*/
- private byte[] writeToManifest(long checkpointId) throws IOException {
- if (writeResultsOfCurrentCkpt.isEmpty()) {
- return EMPTY_MANIFEST_DATA;
- }
-
- WriteResult result =
WriteResult.builder().addAll(writeResultsOfCurrentCkpt).build();
+ private byte[] writeToManifest(long checkpointId, List<WriteResult>
writeResults)
+ throws IOException {
+ WriteResult result = WriteResult.builder().addAll(writeResults).build();
DeltaManifests deltaManifests =
FlinkManifestUtil.writeCompletedFiles(
result, () -> manifestOutputFileFactory.create(checkpointId),
spec);
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
index 7d86baa14f..412d6c7081 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergStreamWriter.java
@@ -29,10 +29,11 @@ import org.apache.iceberg.io.TaskWriter;
import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
-class IcebergStreamWriter<T> extends AbstractStreamOperator<WriteResult>
- implements OneInputStreamOperator<T, WriteResult>, BoundedOneInput {
+class IcebergStreamWriter<T> extends AbstractStreamOperator<FlinkWriteResult>
+ implements OneInputStreamOperator<T, FlinkWriteResult>, BoundedOneInput {
private static final long serialVersionUID = 1L;
+ static final long END_INPUT_CHECKPOINT_ID = Long.MAX_VALUE;
private final String fullTableName;
private final TaskWriterFactory<T> taskWriterFactory;
@@ -63,7 +64,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
- flush();
+ flush(checkpointId);
this.writer = taskWriterFactory.create();
}
@@ -89,7 +90,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
// Note that if the task is not closed after calling endInput, checkpoint
may be triggered again
// causing files to be sent repeatedly, the writer is marked as null after
the last file is sent
// to guard against duplicated writes.
- flush();
+ flush(END_INPUT_CHECKPOINT_ID);
}
@Override
@@ -102,7 +103,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
}
/** close all open files and emit files to downstream committer operator */
- private void flush() throws IOException {
+ private void flush(long checkpointId) throws IOException {
if (writer == null) {
return;
}
@@ -110,7 +111,7 @@ class IcebergStreamWriter<T> extends
AbstractStreamOperator<WriteResult>
long startNano = System.nanoTime();
WriteResult result = writer.complete();
writerMetrics.updateFlushResult(result);
- output.collect(new StreamRecord<>(result));
+ output.collect(new StreamRecord<>(new FlinkWriteResult(checkpointId,
result)));
writerMetrics.flushDuration(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNano));
// Set writer to null to prevent duplicate flushes in the corner case of
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
index 8faae1b05a..3299e7a977 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestCompressionSettings.java
@@ -40,7 +40,6 @@ import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.io.BaseTaskWriter;
import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
@@ -210,8 +209,10 @@ public class TestCompressionSettings {
.containsEntry(TableProperties.ORC_COMPRESSION_STRATEGY, "speed");
}
- private static OneInputStreamOperatorTestHarness<RowData, WriteResult>
createIcebergStreamWriter(
- Table icebergTable, TableSchema flinkSchema, Map<String, String>
override) throws Exception {
+ private static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
+ createIcebergStreamWriter(
+ Table icebergTable, TableSchema flinkSchema, Map<String, String>
override)
+ throws Exception {
RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(),
flinkSchema);
FlinkWriteConf flinkWriteConfig =
new FlinkWriteConf(
@@ -219,7 +220,7 @@ public class TestCompressionSettings {
IcebergStreamWriter<RowData> streamWriter =
FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig,
flinkRowType, null);
- OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+ OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
harness.setup();
@@ -230,7 +231,7 @@ public class TestCompressionSettings {
private static Map<String, String> appenderProperties(
Table table, TableSchema schema, Map<String, String> override) throws
Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter(table, schema, override)) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
index 67ca9d08b2..7808771d98 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergFilesCommitter.java
@@ -129,7 +129,8 @@ public class TestIcebergFilesCommitter extends TestBase {
long timestamp = 0;
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -160,7 +161,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
long checkpointId = 0;
long timestamp = 0;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
@@ -175,8 +177,8 @@ public class TestIcebergFilesCommitter extends TestBase {
}
}
- private WriteResult of(DataFile dataFile) {
- return WriteResult.builder().addDataFiles(dataFile).build();
+ private FlinkWriteResult of(long checkpointId, DataFile dataFile) {
+ return new FlinkWriteResult(checkpointId,
WriteResult.builder().addDataFiles(dataFile).build());
}
@TestTemplate
@@ -192,7 +194,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobID = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobID)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobID)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -203,7 +206,7 @@ public class TestIcebergFilesCommitter extends TestBase {
for (int i = 1; i <= 3; i++) {
RowData rowData = SimpleDataUtil.createRowData(i, "hello" + i);
DataFile dataFile = writeDataFile("data-" + i,
ImmutableList.of(rowData));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(i, dataFile), ++timestamp);
rows.add(rowData);
harness.snapshot(i, ++timestamp);
@@ -232,7 +235,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -242,21 +246,21 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createRowData(1, "hello");
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ long firstCheckpointId = 1;
+ harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
- long firstCheckpointId = 1;
harness.snapshot(firstCheckpointId, ++timestamp);
assertFlinkManifests(1);
RowData row2 = SimpleDataUtil.createRowData(2, "world");
DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
- harness.processElement(of(dataFile2), ++timestamp);
+ long secondCheckpointId = 2;
+ harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 2. snapshotState for checkpoint#2
- long secondCheckpointId = 2;
harness.snapshot(secondCheckpointId, ++timestamp);
assertFlinkManifests(2);
@@ -285,7 +289,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -295,21 +300,21 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createRowData(1, "hello");
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ long firstCheckpointId = 1;
+ harness.processElement(of(firstCheckpointId, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
- long firstCheckpointId = 1;
harness.snapshot(firstCheckpointId, ++timestamp);
assertFlinkManifests(1);
RowData row2 = SimpleDataUtil.createRowData(2, "world");
DataFile dataFile2 = writeDataFile("data-2", ImmutableList.of(row2));
- harness.processElement(of(dataFile2), ++timestamp);
+ long secondCheckpointId = 2;
+ harness.processElement(of(secondCheckpointId, dataFile2), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 2. snapshotState for checkpoint#2
- long secondCheckpointId = 2;
harness.snapshot(secondCheckpointId, ++timestamp);
assertFlinkManifests(2);
@@ -336,7 +341,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -348,8 +354,8 @@ public class TestIcebergFilesCommitter extends TestBase {
expectedRows.add(row);
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row));
- harness.processElement(of(dataFile1), ++timestamp);
- snapshot = harness.snapshot(++checkpointId, ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile1), ++timestamp);
+ snapshot = harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -361,7 +367,8 @@ public class TestIcebergFilesCommitter extends TestBase {
}
// Restore from the given snapshot
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.initializeState(snapshot);
@@ -374,9 +381,9 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(2, "world");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -399,7 +406,8 @@ public class TestIcebergFilesCommitter extends TestBase {
List<RowData> expectedRows = Lists.newArrayList();
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -410,15 +418,16 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(1, "hello");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-1", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- snapshot = harness.snapshot(++checkpointId, ++timestamp);
+ snapshot = harness.snapshot(checkpointId, ++timestamp);
SimpleDataUtil.assertTableRows(table, ImmutableList.of(), branch);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
assertFlinkManifests(1);
}
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.initializeState(snapshot);
@@ -445,15 +454,15 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(2, "world");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-2", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- snapshot = harness.snapshot(++checkpointId, ++timestamp);
+ snapshot = harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
}
// Redeploying flink job from external checkpoint.
JobID newJobId = new JobID();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
createStreamSink(newJobId)) {
harness.setup();
harness.initializeState(snapshot);
@@ -472,9 +481,9 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row = SimpleDataUtil.createRowData(3, "foo");
expectedRows.add(row);
DataFile dataFile = writeDataFile("data-3", ImmutableList.of(row));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -495,7 +504,7 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID oldJobId = new JobID();
OperatorID oldOperatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
createStreamSink(oldJobId)) {
harness.setup();
harness.open();
@@ -509,8 +518,8 @@ public class TestIcebergFilesCommitter extends TestBase {
tableRows.addAll(rows);
DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
- harness.processElement(of(dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -527,7 +536,7 @@ public class TestIcebergFilesCommitter extends TestBase {
timestamp = 0;
JobID newJobId = new JobID();
OperatorID newOperatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
createStreamSink(newJobId)) {
harness.setup();
harness.open();
@@ -541,8 +550,8 @@ public class TestIcebergFilesCommitter extends TestBase {
tableRows.addAll(rows);
DataFile dataFile = writeDataFile("data-new-1", rows);
- harness.processElement(of(dataFile), ++timestamp);
- harness.snapshot(++checkpointId, ++timestamp);
+ harness.processElement(of(++checkpointId, dataFile), ++timestamp);
+ harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
harness.notifyOfCompletedCheckpoint(checkpointId);
@@ -566,7 +575,8 @@ public class TestIcebergFilesCommitter extends TestBase {
int checkpointId = i / 3;
JobID jobId = jobs[jobIndex];
OperatorID operatorId = operatorIds[jobIndex];
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.open();
@@ -578,7 +588,7 @@ public class TestIcebergFilesCommitter extends TestBase {
tableRows.addAll(rows);
DataFile dataFile = writeDataFile(String.format("data-%d", i), rows);
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId + 1, dataFile), ++timestamp);
harness.snapshot(checkpointId + 1, ++timestamp);
assertFlinkManifests(1);
@@ -602,8 +612,10 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId1 = new OperatorID();
OperatorID operatorId2 = new OperatorID();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
- OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+ createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+ createStreamSink(jobId)) {
harness1.getStreamConfig().setOperatorID(operatorId1);
harness1.setup();
harness1.open();
@@ -619,14 +631,14 @@ public class TestIcebergFilesCommitter extends TestBase {
expectedRows.add(row1);
DataFile dataFile1 = writeDataFile("data-1-1", ImmutableList.of(row1));
- harness1.processElement(of(dataFile1), ++timestamp);
- snapshot1 = harness1.snapshot(++checkpointId, ++timestamp);
+ harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+ snapshot1 = harness1.snapshot(checkpointId, ++timestamp);
RowData row2 = SimpleDataUtil.createRowData(1, "hello2");
expectedRows.add(row2);
DataFile dataFile2 = writeDataFile("data-1-2", ImmutableList.of(row2));
- harness2.processElement(of(dataFile2), ++timestamp);
+ harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
snapshot2 = harness2.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(2);
@@ -642,8 +654,10 @@ public class TestIcebergFilesCommitter extends TestBase {
}
// Restore from the given snapshot
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness1 =
createStreamSink(jobId);
- OneInputStreamOperatorTestHarness<WriteResult, Void> harness2 =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness1 =
+ createStreamSink(jobId);
+ OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness2 =
+ createStreamSink(jobId)) {
harness1.getStreamConfig().setOperatorID(operatorId1);
harness1.setup();
harness1.initializeState(snapshot1);
@@ -667,13 +681,13 @@ public class TestIcebergFilesCommitter extends TestBase {
expectedRows.add(row1);
DataFile dataFile1 = writeDataFile("data-2-1", ImmutableList.of(row1));
- harness1.processElement(of(dataFile1), ++timestamp);
- harness1.snapshot(++checkpointId, ++timestamp);
+ harness1.processElement(of(++checkpointId, dataFile1), ++timestamp);
+ harness1.snapshot(checkpointId, ++timestamp);
RowData row2 = SimpleDataUtil.createRowData(2, "world2");
expectedRows.add(row2);
DataFile dataFile2 = writeDataFile("data-2-2", ImmutableList.of(row2));
- harness2.processElement(of(dataFile2), ++timestamp);
+ harness2.processElement(of(checkpointId, dataFile2), ++timestamp);
harness2.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(2);
@@ -693,7 +707,8 @@ public class TestIcebergFilesCommitter extends TestBase {
public void testBoundedStream() throws Exception {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -705,13 +720,14 @@ public class TestIcebergFilesCommitter extends TestBase {
List<RowData> tableRows =
Lists.newArrayList(SimpleDataUtil.createRowData(1, "word-1"));
DataFile dataFile = writeDataFile("data-1", tableRows);
- harness.processElement(of(dataFile), 1);
+ harness.processElement(of(IcebergStreamWriter.END_INPUT_CHECKPOINT_ID,
dataFile), 1);
((BoundedOneInput) harness.getOneInputOperator()).endInput();
assertFlinkManifests(0);
SimpleDataUtil.assertTableRows(table, tableRows, branch);
assertSnapshotSize(1);
- assertMaxCommittedCheckpointId(jobId, operatorId, Long.MAX_VALUE);
+ assertMaxCommittedCheckpointId(
+ jobId, operatorId, IcebergStreamWriter.END_INPUT_CHECKPOINT_ID);
assertThat(SimpleDataUtil.latestSnapshot(table, branch).summary())
.containsEntry("flink.test",
TestIcebergFilesCommitter.class.getName());
}
@@ -724,7 +740,8 @@ public class TestIcebergFilesCommitter extends TestBase {
JobID jobId = new JobID();
OperatorID operatorId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -734,7 +751,7 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createRowData(1, "hello");
DataFile dataFile1 = writeDataFile("data-1", ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ harness.processElement(of(checkpoint, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
@@ -774,7 +791,8 @@ public class TestIcebergFilesCommitter extends TestBase {
OperatorID operatorId;
FileAppenderFactory<RowData> appenderFactory =
createDeletableAppenderFactory();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -783,7 +801,7 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData row1 = SimpleDataUtil.createInsert(1, "aaa");
DataFile dataFile1 = writeDataFile("data-file-1",
ImmutableList.of(row1));
- harness.processElement(of(dataFile1), ++timestamp);
+ harness.processElement(of(checkpoint, dataFile1), ++timestamp);
assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
// 1. snapshotState for checkpoint#1
@@ -815,13 +833,15 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData delete1 = SimpleDataUtil.createDelete(1, "aaa");
DeleteFile deleteFile1 =
writeEqDeleteFile(appenderFactory, "delete-file-1",
ImmutableList.of(delete1));
+ assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
harness.processElement(
-
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build(),
+ new FlinkWriteResult(
+ ++checkpoint,
+
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile1).build()),
++timestamp);
- assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
// 5. snapshotState for checkpoint#2
- harness.snapshot(++checkpoint, ++timestamp);
+ harness.snapshot(checkpoint, ++timestamp);
assertFlinkManifests(2);
// 6. notifyCheckpointComplete for checkpoint#2
@@ -845,7 +865,8 @@ public class TestIcebergFilesCommitter extends TestBase {
OperatorID operatorId;
FileAppenderFactory<RowData> appenderFactory =
createDeletableAppenderFactory();
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -859,7 +880,9 @@ public class TestIcebergFilesCommitter extends TestBase {
DeleteFile deleteFile1 =
writeEqDeleteFile(appenderFactory, "delete-file-1",
ImmutableList.of(delete3));
harness.processElement(
-
WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build(),
+ new FlinkWriteResult(
+ checkpoint,
+
WriteResult.builder().addDataFiles(dataFile1).addDeleteFiles(deleteFile1).build()),
++timestamp);
// The 1th snapshotState.
@@ -871,11 +894,13 @@ public class TestIcebergFilesCommitter extends TestBase {
DeleteFile deleteFile2 =
writeEqDeleteFile(appenderFactory, "delete-file-2",
ImmutableList.of(delete2));
harness.processElement(
-
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build(),
+ new FlinkWriteResult(
+ ++checkpoint,
+
WriteResult.builder().addDataFiles(dataFile2).addDeleteFiles(deleteFile2).build()),
++timestamp);
// The 2nd snapshotState.
- harness.snapshot(++checkpoint, ++timestamp);
+ harness.snapshot(checkpoint, ++timestamp);
// Notify the 2nd snapshot to complete.
harness.notifyOfCompletedCheckpoint(checkpoint);
@@ -886,6 +911,79 @@ public class TestIcebergFilesCommitter extends TestBase {
}
}
+ /**
+ * The testcase is to simulate upserting to an Iceberg V2 table, and facing
the following
+ * scenario:
+ *
+ * <ul>
+ * <li>A specific row is updated
+ * <li>The prepareSnapshotPreBarrier triggered
+ * <li>Checkpoint failed for reasons outside of the Iceberg connector
+ * <li>The specific row is updated again in the second checkpoint as well
+ * <li>Second snapshot is triggered, and finished
+ * </ul>
+ *
+ * <p>Previously the files from the 2 snapshots were committed in a single
Iceberg commit, as a
+ * results duplicate rows were created in the table.
+ *
+ * @throws Exception Exception
+ */
+ @TestTemplate
+ public void testCommitMultipleCheckpointsForV2Table() throws Exception {
+ assumeThat(formatVersion)
+ .as("Only support equality-delete in format v2 or later.")
+ .isGreaterThan(1);
+
+ long timestamp = 0;
+ long checkpoint = 10;
+
+ JobID jobId = new JobID();
+ OperatorID operatorId;
+
+ FileAppenderFactory<RowData> appenderFactory =
+ new FlinkAppenderFactory(
+ table,
+ table.schema(),
+ FlinkSchemaUtil.convert(table.schema()),
+ table.properties(),
+ table.spec(),
+ new int[] {table.schema().findField("id").fieldId()},
+ table.schema(),
+ null);
+
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
+ harness.setup();
+ harness.open();
+ operatorId = harness.getOperator().getOperatorID();
+
+ assertMaxCommittedCheckpointId(jobId, operatorId, -1L);
+
+ RowData insert1 = null;
+ RowData insert2 = null;
+ for (int i = 1; i <= 3; i++) {
+ insert1 = SimpleDataUtil.createInsert(1, "aaa" + i);
+ insert2 = SimpleDataUtil.createInsert(2, "bbb" + i);
+ DataFile dataFile = writeDataFile("data-file-" + i,
ImmutableList.of(insert1, insert2));
+ DeleteFile deleteFile =
+ writeEqDeleteFile(
+ appenderFactory, "delete-file-" + i, ImmutableList.of(insert1,
insert2));
+ harness.processElement(
+ new FlinkWriteResult(
+ ++checkpoint,
+
WriteResult.builder().addDataFiles(dataFile).addDeleteFiles(deleteFile).build()),
+ ++timestamp);
+ }
+
+ harness.snapshot(checkpoint, ++timestamp);
+ harness.notifyOfCompletedCheckpoint(checkpoint);
+ SimpleDataUtil.assertTableRows(table, ImmutableList.of(insert1,
insert2), branch);
+ assertMaxCommittedCheckpointId(jobId, operatorId, checkpoint);
+ assertFlinkManifests(0);
+ assertThat(table.snapshots()).hasSize(3);
+ }
+ }
+
@TestTemplate
public void testSpecEvolution() throws Exception {
long timestamp = 0;
@@ -898,7 +996,8 @@ public class TestIcebergFilesCommitter extends TestBase {
DataFile dataFile;
int specId;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.setup();
harness.open();
operatorId = harness.getOperator().getOperatorID();
@@ -909,7 +1008,7 @@ public class TestIcebergFilesCommitter extends TestBase {
RowData rowData = SimpleDataUtil.createRowData(checkpointId, "hello" +
checkpointId);
// table unpartitioned
dataFile = writeDataFile("data-" + checkpointId,
ImmutableList.of(rowData));
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId, dataFile), ++timestamp);
rows.add(rowData);
harness.snapshot(checkpointId, ++timestamp);
@@ -928,7 +1027,7 @@ public class TestIcebergFilesCommitter extends TestBase {
rowData = SimpleDataUtil.createRowData(checkpointId, "hello" +
checkpointId);
// write data with old partition spec
dataFile = writeDataFile("data-" + checkpointId,
ImmutableList.of(rowData), oldSpec, null);
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId, dataFile), ++timestamp);
rows.add(rowData);
snapshot = harness.snapshot(checkpointId, ++timestamp);
@@ -946,7 +1045,8 @@ public class TestIcebergFilesCommitter extends TestBase {
}
// Restore from the given snapshot
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
createStreamSink(jobId)) {
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
+ createStreamSink(jobId)) {
harness.getStreamConfig().setOperatorID(operatorId);
harness.setup();
harness.initializeState(snapshot);
@@ -962,7 +1062,7 @@ public class TestIcebergFilesCommitter extends TestBase {
partition.set(0, checkpointId);
dataFile =
writeDataFile("data-" + checkpointId, ImmutableList.of(row),
table.spec(), partition);
- harness.processElement(of(dataFile), ++timestamp);
+ harness.processElement(of(checkpointId, dataFile), ++timestamp);
rows.add(row);
harness.snapshot(checkpointId, ++timestamp);
assertFlinkManifests(1);
@@ -1088,7 +1188,7 @@ public class TestIcebergFilesCommitter extends TestBase {
assertThat(table.snapshots()).hasSize(expectedSnapshotSize);
}
- private OneInputStreamOperatorTestHarness<WriteResult, Void>
createStreamSink(JobID jobID)
+ private OneInputStreamOperatorTestHarness<FlinkWriteResult, Void>
createStreamSink(JobID jobID)
throws Exception {
TestOperatorFactory factory = TestOperatorFactory.of(table.location(),
branch, table.spec());
return new OneInputStreamOperatorTestHarness<>(factory,
createEnvironment(jobID));
@@ -1108,7 +1208,7 @@ public class TestIcebergFilesCommitter extends TestBase {
}
private static class TestOperatorFactory extends
AbstractStreamOperatorFactory<Void>
- implements OneInputStreamOperatorFactory<WriteResult, Void> {
+ implements OneInputStreamOperatorFactory<FlinkWriteResult, Void> {
private final String tablePath;
private final String branch;
private final PartitionSpec spec;
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
index 50283f7ad2..e13721a9f1 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java
@@ -28,6 +28,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.api.DataTypes;
@@ -102,7 +103,7 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testWritingTable() throws Exception {
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
// The first checkpoint
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
@@ -111,7 +112,8 @@ public class TestIcebergStreamWriter {
testHarness.prepareSnapshotPreBarrier(checkpointId);
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -123,7 +125,8 @@ public class TestIcebergStreamWriter {
testHarness.prepareSnapshotPreBarrier(checkpointId);
expectedDataFiles = partitioned ? 4 : 2;
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -148,14 +151,15 @@ public class TestIcebergStreamWriter {
public void testSnapshotTwice() throws Exception {
long checkpointId = 1;
long timestamp = 1;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"),
timestamp++);
testHarness.processElement(SimpleDataUtil.createRowData(2, "world"),
timestamp);
testHarness.prepareSnapshotPreBarrier(checkpointId++);
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -163,7 +167,10 @@ public class TestIcebergStreamWriter {
for (int i = 0; i < 5; i++) {
testHarness.prepareSnapshotPreBarrier(checkpointId++);
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+ WriteResult.builder()
+ .addAll(getWriteResults(testHarness.extractOutputValues()))
+ .build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
}
@@ -172,14 +179,14 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testTableWithoutSnapshot() throws Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
assertThat(testHarness.extractOutputValues()).isEmpty();
}
// Even if we closed the iceberg stream writer, there's no orphan data
file.
assertThat(scanDataFiles()).isEmpty();
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
// Still not emit the data file yet, because there is no checkpoint.
@@ -212,7 +219,7 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testBoundedStreamCloseWithEmittingDataFiles() throws Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -221,13 +228,15 @@ public class TestIcebergStreamWriter {
((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
((BoundedOneInput) testHarness.getOneInputOperator()).endInput();
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
// Datafiles should not be sent again
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
@@ -236,7 +245,7 @@ public class TestIcebergStreamWriter {
@TestTemplate
public void testBoundedStreamTriggeredEndInputBeforeTriggeringCheckpoint()
throws Exception {
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
testHarness.processElement(SimpleDataUtil.createRowData(1, "hello"), 1);
testHarness.processElement(SimpleDataUtil.createRowData(2, "world"), 2);
@@ -244,13 +253,15 @@ public class TestIcebergStreamWriter {
testHarness.endInput();
int expectedDataFiles = partitioned ? 2 : 1;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(expectedDataFiles);
testHarness.prepareSnapshotPreBarrier(1L);
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
// It should be ensured that after endInput is triggered, when
prepareSnapshotPreBarrier
// is triggered, write should only send WriteResult once
@@ -275,7 +286,7 @@ public class TestIcebergStreamWriter {
}
}
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter()) {
for (RowData row : rows) {
testHarness.processElement(row, 1);
@@ -283,7 +294,8 @@ public class TestIcebergStreamWriter {
// snapshot the operator.
testHarness.prepareSnapshotPreBarrier(1);
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(8);
@@ -346,13 +358,14 @@ public class TestIcebergStreamWriter {
record.copy(ImmutableMap.of("tinyint", 2, "smallint", 0, "int",
102)),
record.copy(ImmutableMap.of("tinyint", 3, "smallint", 32767,
"int", 103)));
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createIcebergStreamWriter(icebergTable, flinkSchema)) {
for (RowData row : rows) {
testHarness.processElement(row, 1);
}
testHarness.prepareSnapshotPreBarrier(1);
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+
WriteResult.builder().addAll(getWriteResults(testHarness.extractOutputValues())).build();
assertThat(result.deleteFiles()).isEmpty();
assertThat(result.dataFiles()).hasSize(partitioned ? 3 : 1);
@@ -365,12 +378,18 @@ public class TestIcebergStreamWriter {
SimpleDataUtil.assertTableRecords(location, expected);
}
- private OneInputStreamOperatorTestHarness<RowData, WriteResult>
createIcebergStreamWriter()
+ private static List<WriteResult> getWriteResults(List<FlinkWriteResult>
flinkWriteResults) {
+ return flinkWriteResults.stream()
+ .map(FlinkWriteResult::writeResult)
+ .collect(Collectors.toList());
+ }
+
+ private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
createIcebergStreamWriter()
throws Exception {
return createIcebergStreamWriter(table, SimpleDataUtil.FLINK_SCHEMA);
}
- private OneInputStreamOperatorTestHarness<RowData, WriteResult>
createIcebergStreamWriter(
+ private OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
createIcebergStreamWriter(
Table icebergTable, TableSchema flinkSchema) throws Exception {
RowType flinkRowType = FlinkSink.toFlinkRowType(icebergTable.schema(),
flinkSchema);
FlinkWriteConf flinkWriteConfig =
@@ -379,7 +398,7 @@ public class TestIcebergStreamWriter {
IcebergStreamWriter<RowData> streamWriter =
FlinkSink.createStreamWriter(() -> icebergTable, flinkWriteConfig,
flinkRowType, null);
- OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+ OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
new OneInputStreamOperatorTestHarness<>(streamWriter, 1, 1, 0);
harness.setup();
