Skip to content

Flink: add append capability to dynamic iceberg sink (#14526) #14559

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pvary merged 5 commits into from
Nov 18, 2025
Merged

Flink: add append capability to dynamic iceberg sink (#14526) #14559

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6030782
Flink: add append capability to dynamic iceberg sink
Nov 11, 2025
0621c1b
revert changes to earlier versions
Nov 11, 2025
3f6c923
address PR comments
Nov 11, 2025
4a852ad
test append snapshot operation
Nov 11, 2025
c1064d6
revert change - this is now addressed in #14581
Nov 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 50 additions & 21 deletions flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ReplacePartitions;
import org.apache.iceberg.RowDelta;
Expand Down Expand Up @@ -302,30 +303,58 @@ private void commitDeltaTxn(
CommitSummary summary,
String newFlinkJobId,
String operatorId) {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
List<WriteResult> writeResults = e.getValue();

RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
for (WriteResult result : writeResults) {
// Row delta validations are not needed for streaming changes that write equality deletes.
// Equality deletes are applied to data in all previous sequence numbers, so retries may
// push deletes further in the future, but do not affect correctness. Position deletes
// committed to the table in this path are used only to delete rows from data files that are
// being added in this commit. There is no way for data files added along with the delete
// files to be concurrently removed, so there is no need to validate the files referenced by
// the position delete files that are being committed.
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
if (summary.deleteFilesCount() == 0) {
// To be compatible with iceberg format V1.
Copy link
Contributor

@pvary pvary Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we change this comment to describe correctly why we do this?

AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
for (List<WriteResult> resultList : pendingResults.values()) {
for (WriteResult result : resultList) {
Preconditions.checkState(
result.referencedDataFiles().length == 0,
"Should have no referenced data files for append.");
Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
}
}
String description = "append";
Copy link
Contributor

@pvary pvary Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: newline after the block. See: https://iceberg.apache.org/contribute/#block-spacing


// Every Flink checkpoint contains a set of independent changes which can be committed
// together. While it is technically feasible to combine append-only data across checkpoints,
// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
// intervals or when concurrent checkpointing is enabled.
// fail all commits as really its only one
commitOperation(
table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
table,
branch,
appendFiles,
summary,
description,
Copy link
Contributor

@pvary pvary Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we use the string value here? We don't reuse the description variable anywhere else.

newFlinkJobId,
operatorId,
pendingResults.lastKey());
} else {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
List<WriteResult> writeResults = e.getValue();

RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
for (WriteResult result : writeResults) {
// Row delta validations are not needed for streaming changes that write equality deletes.
// Equality deletes are applied to data in all previous sequence numbers, so retries may
// push deletes further in the future, but do not affect correctness. Position deletes
// committed to the table in this path are used only to delete rows from data files that
// are
// being added in this commit. There is no way for data files added along with the delete
// files to be concurrently removed, so there is no need to validate the files referenced
// by
// the position delete files that are being committed.
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
}

// Every Flink checkpoint contains a set of independent changes which can be committed
// together. While it is technically feasible to combine append-only data across
// checkpoints,
// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
// intervals or when concurrent checkpointing is enabled.
commitOperation(
table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
}
}
}

Expand Down
71 changes: 50 additions & 21 deletions flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ReplacePartitions;
import org.apache.iceberg.RowDelta;
Expand Down Expand Up @@ -302,30 +303,58 @@ private void commitDeltaTxn(
CommitSummary summary,
String newFlinkJobId,
String operatorId) {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
List<WriteResult> writeResults = e.getValue();

RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
for (WriteResult result : writeResults) {
// Row delta validations are not needed for streaming changes that write equality deletes.
// Equality deletes are applied to data in all previous sequence numbers, so retries may
// push deletes further in the future, but do not affect correctness. Position deletes
// committed to the table in this path are used only to delete rows from data files that are
// being added in this commit. There is no way for data files added along with the delete
// files to be concurrently removed, so there is no need to validate the files referenced by
// the position delete files that are being committed.
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
if (summary.deleteFilesCount() == 0) {
// To be compatible with iceberg format V1.
AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
for (List<WriteResult> resultList : pendingResults.values()) {
for (WriteResult result : resultList) {
Preconditions.checkState(
result.referencedDataFiles().length == 0,
"Should have no referenced data files for append.");
Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
}
}
String description = "append";

// Every Flink checkpoint contains a set of independent changes which can be committed
// together. While it is technically feasible to combine append-only data across checkpoints,
// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
// intervals or when concurrent checkpointing is enabled.
// fail all commits as really its only one
commitOperation(
table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
table,
branch,
appendFiles,
summary,
description,
newFlinkJobId,
operatorId,
pendingResults.lastKey());
} else {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
List<WriteResult> writeResults = e.getValue();

RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
for (WriteResult result : writeResults) {
// Row delta validations are not needed for streaming changes that write equality deletes.
// Equality deletes are applied to data in all previous sequence numbers, so retries may
// push deletes further in the future, but do not affect correctness. Position deletes
// committed to the table in this path are used only to delete rows from data files that
// are
// being added in this commit. There is no way for data files added along with the delete
// files to be concurrently removed, so there is no need to validate the files referenced
// by
// the position delete files that are being committed.
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
}

// Every Flink checkpoint contains a set of independent changes which can be committed
// together. While it is technically feasible to combine append-only data across
// checkpoints,
// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
// intervals or when concurrent checkpointing is enabled.
commitOperation(
table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
}
}
}

Expand Down
71 changes: 50 additions & 21 deletions flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ReplacePartitions;
import org.apache.iceberg.RowDelta;
Expand Down Expand Up @@ -302,30 +303,58 @@ private void commitDeltaTxn(
CommitSummary summary,
String newFlinkJobId,
String operatorId) {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
List<WriteResult> writeResults = e.getValue();
Copy link
Contributor

@mxm mxm Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we keep the loop structure? We will need it for both types of snapshots. This should work:

  for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
    long checkpointId = e.getKey();
    List<WriteResult> writeResults = e.getValue();

    boolean appendOnly = true;
    for (WriteResult writeResult : writeResults) {
      if (writeResult.deleteFiles().length > 0) {
        appendOnly = false;
        break;
      }
    }

    final SnapshotUpdate snapshotUpdate;
    if (appendOnly) {
        AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
        for (WriteResult result : writeResults) {
          Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
        }
        snapshotUpdate = appendFiles
    } else {
        RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
        for (WriteResult result : writeResults) {
          Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
          Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
        }
       snapshotUpdate = rowDelta;
    }
   
    commitOperation(
          table,
          branch,
          snapshotUpdate,
          summary,
          appendOnly ? "append" : "rowDelta",
          newFlinkJobId,
          operatorId,
          checkpointId);
  }

Copy link
Contributor Author

@bezdomniy bezdomniy Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this looks good - only thing left is the checkState for result.referencedDataFiles().length == 0 (which exists in IcebergSink) which I will add to the loop checking it is appendOnly


RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
for (WriteResult result : writeResults) {
// Row delta validations are not needed for streaming changes that write equality deletes.
// Equality deletes are applied to data in all previous sequence numbers, so retries may
// push deletes further in the future, but do not affect correctness. Position deletes
// committed to the table in this path are used only to delete rows from data files that are
// being added in this commit. There is no way for data files added along with the delete
// files to be concurrently removed, so there is no need to validate the files referenced by
// the position delete files that are being committed.
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
if (summary.deleteFilesCount() == 0) {
Copy link
Contributor

@mxm mxm Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure about the granularity of this value, as every pending result could contain or not contain deletes. Probably best to check the WriteResults directly.

Copy link
Contributor

@pvary pvary Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The deleteFilesCount should be correct.

This is how it is calculated:

iceberg/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicCommitter.java

Line 223 in 059310e

summary.addAll(pendingResults);

And internally:

iceberg/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/CommitSummary.java

Lines 46 to 66 in 059310e

public void addAll(NavigableMap<Long, List<WriteResult>> pendingResults) {
pendingResults.values().forEach(writeResults -> writeResults.forEach(this::addWriteResult));
}
private void addWriteResult(WriteResult writeResult) {
dataFilesCount.addAndGet(writeResult.dataFiles().length);
Arrays.stream(writeResult.dataFiles())
.forEach(
dataFile -> {
dataFilesRecordCount.addAndGet(dataFile.recordCount());
dataFilesByteCount.addAndGet(dataFile.fileSizeInBytes());
});
deleteFilesCount.addAndGet(writeResult.deleteFiles().length);
Arrays.stream(writeResult.deleteFiles())
.forEach(
deleteFile -> {
deleteFilesRecordCount.addAndGet(deleteFile.recordCount());
long deleteBytes = ScanTaskUtil.contentSizeInBytes(deleteFile);
deleteFilesByteCount.addAndGet(deleteBytes);
});
}

Copy link
Contributor

@mxm mxm Nov 11, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The value is correct but we commit at checkpoint level and the delete file count is done across checkpoints. Strictly speaking, there could be both append-only checkpoints and overwrite checkpoints as part of pendingResults.

Copy link
Contributor

@pvary pvary Nov 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting decision.
In IcebergSink we commit multiple checkpoints together in a single commit, if we happen to accumulate multiple of them.

Copy link
Contributor

@mxm mxm Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When we fixed #14182, we decided not to do that. In fact, you proposed not to do that: #14182 (comment) 🤓

IMHO this kind of optimization is a bit premature. In practice it is rare to even have multiple pending checkpoints.

Copy link
Contributor

@pvary pvary Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, as I have mentioned in the comment, this could cause issues if "replacePartition" is used

Copy link
Contributor

@pvary pvary Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So we should commit them 1-by-1

// To be compatible with iceberg format V1.
AppendFiles appendFiles = table.newAppend().scanManifestsWith(workerPool);
for (List<WriteResult> resultList : pendingResults.values()) {
Copy link
Contributor

@mxm mxm Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we move this loop up one level, as in https://github.com/apache/iceberg/pull/14559/files#r2513719871? This avoids repeating it.

for (WriteResult result : resultList) {
Preconditions.checkState(
result.referencedDataFiles().length == 0,
"Should have no referenced data files for append.");
Arrays.stream(result.dataFiles()).forEach(appendFiles::appendFile);
}
}
String description = "append";

// Every Flink checkpoint contains a set of independent changes which can be committed
// together. While it is technically feasible to combine append-only data across checkpoints,
// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
// intervals or when concurrent checkpointing is enabled.
// fail all commits as really its only one
commitOperation(
table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
table,
branch,
appendFiles,
summary,
description,
newFlinkJobId,
operatorId,
pendingResults.lastKey());
} else {
for (Map.Entry<Long, List<WriteResult>> e : pendingResults.entrySet()) {
long checkpointId = e.getKey();
List<WriteResult> writeResults = e.getValue();

RowDelta rowDelta = table.newRowDelta().scanManifestsWith(workerPool);
for (WriteResult result : writeResults) {
// Row delta validations are not needed for streaming changes that write equality deletes.
// Equality deletes are applied to data in all previous sequence numbers, so retries may
// push deletes further in the future, but do not affect correctness. Position deletes
// committed to the table in this path are used only to delete rows from data files that
// are
// being added in this commit. There is no way for data files added along with the delete
// files to be concurrently removed, so there is no need to validate the files referenced
// by
// the position delete files that are being committed.
Arrays.stream(result.dataFiles()).forEach(rowDelta::addRows);
Arrays.stream(result.deleteFiles()).forEach(rowDelta::addDeletes);
}

// Every Flink checkpoint contains a set of independent changes which can be committed
// together. While it is technically feasible to combine append-only data across
// checkpoints,
// for the sake of simplicity, we do not implement this (premature) optimization. Multiple
// pending checkpoints here are very rare to occur, i.e. only with very short checkpoint
// intervals or when concurrent checkpointing is enabled.
commitOperation(
table, branch, rowDelta, summary, "rowDelta", newFlinkJobId, operatorId, checkpointId);
}
}
}

Expand Down