xx789633 commented on code in PR #1555:
URL: https://github.com/apache/fluss/pull/1555#discussion_r2282468657
##########
fluss-lake/fluss-lake-iceberg/src/main/java/com/alibaba/fluss/lake/iceberg/tiering/IcebergLakeCommitter.java:
##########
@@ -88,22 +95,46 @@ public long commit(IcebergCommittable committable,
Map<String, String> snapshotP
try {
// Refresh table to get latest metadata
icebergTable.refresh();
- // Simple append-only case: only data files, no delete files or
compaction
- AppendFiles appendFiles = icebergTable.newAppend();
- for (DataFile dataFile : committable.getDataFiles()) {
- appendFiles.appendFile(dataFile);
- }
- if (!committable.getDeleteFiles().isEmpty()) {
- throw new IllegalStateException(
- "Delete files are not supported in append-only mode. "
- + "Found "
- + committable.getDeleteFiles().size()
- + " delete files.");
- }
- addFlussProperties(appendFiles, snapshotProperties);
+ if (committable.getDeleteFiles().isEmpty()) {
+ // Simple append-only case: only data files, no delete files
or compaction
+ AppendFiles appendFiles = icebergTable.newAppend();
+ for (DataFile dataFile : committable.getDataFiles()) {
+ appendFiles.appendFile(dataFile);
+ }
+ if (!committable.getDeleteFiles().isEmpty()) {
+ throw new IllegalStateException(
+ "Delete files are not supported in append-only
mode. "
+ + "Found "
+ + committable.getDeleteFiles().size()
+ + " delete files.");
+ }
+
+ addFlussProperties(appendFiles, snapshotProperties);
- appendFiles.commit();
+ appendFiles.commit();
+ } else {
+ // 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.
+ RowDelta rowDelta = icebergTable.newRowDelta();
Review Comment:
This doesn't pass java 8.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
