wombatu-kun commented on code in PR #18807:
URL: https://github.com/apache/iceberg/pull/18807#discussion_r4477253012
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java:
##########
@@ -82,6 +83,40 @@ static Set<Integer> checkAndGetEqualityFieldIds(Table table,
List<String> equali
return equalityFieldIds;
}
+ static long getMaxCommittedCheckpointIdBySink(
Review Comment:
This duplicates DynamicCommitter.getMaxCommittedCheckpointId, and
DynamicCommitter still writes the summary with private copies of the keys that
writtenBySameSink now reads from here. Could that lookup move into SinkUtil for
both committers (IcebergCommitter passing SnapshotUtil.ancestorsOf), with
DynamicCommitter using these constants, so writtenBySameSink does not have to
become public?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergWriteAggregator.java:
##########
@@ -119,6 +121,7 @@ public void prepareSnapshotPreBarrier(long checkpointId)
throws IOException {
new IcebergCommittable(
writeToManifest(results, checkpointId),
getContainingTask().getEnvironment().getJobID().toString(),
+ sinkId,
Review Comment:
Nothing pins that the committable gets the sink's sinkId rather than the
jobId here: TestIcebergCommitter builds its committables by hand and none of
the TestIcebergSink* tests read flink.sink-id. Could a test run the real
IcebergSink and assert that the committed snapshot's flink.sink-id differs from
its flink.job-id, as TestDynamicIcebergSink.testAllTablesShareTheSinkId does?
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
