grantatspothero commented on code in PR #18076:
URL: https://github.com/apache/iceberg/pull/18076#discussion_r4006908585
##########
core/src/main/java/org/apache/iceberg/ManifestMergeManager.java:
##########
@@ -193,34 +200,54 @@ private ManifestFile createManifest(int specId,
List<ManifestFile> bin) {
}
ManifestWriter<F> writer = newManifestWriter(spec(specId));
+ ExecutorService workerPool = workerPoolSupplier.get();
+ Deque<FutureTask<List<ManifestEntry<F>>>> pendingReads = new
ArrayDeque<>();
boolean threw = true;
try {
- for (ManifestFile manifest : bin) {
- boolean isCommitted =
- manifest.snapshotId() != null && snapshotId() !=
manifest.snapshotId();
- try (ManifestReader<F> reader = newManifestReader(manifest,
isCommitted)) {
- for (ManifestEntry<F> entry : reader.entries()) {
- if (entry.status() == Status.DELETED) {
- // suppress deletes from previous snapshots. only files deleted
by this snapshot
- // should be added to the new manifest
- if (entry.snapshotId() == snapshotId()) {
- writer.delete(entry);
- }
- } else if (entry.status() == Status.ADDED && entry.snapshotId() ==
snapshotId()) {
- // adds from this snapshot are still adds, otherwise they should
be existing
- writer.add(entry);
- } else {
- // add all files from the old manifest as existing files
- writer.existing(entry);
- }
+ // reads run on the worker pool ahead of the writer, which consumes them
in bin order. a
Review Comment:
This code is quite dense, could you reuse or extend iceberg's existing
`ParallelIterable` here?
Two main differences I see:
1. Large manifest behavior:
`ParallelIterable` previously had problems with holding open too many
connections, see: https://github.com/apache/iceberg/pull/11781 which solves the
problems by forcing drain of a whole task once it is started. This allows
potentially unbounded memory usage if a single iterable is very large, but
allows for maximum parallelism.
This approach is different, given you know the size of the manifest entries
upfront, only parallelize those manifests which have small bounded size to
avoid unbounded memory usage. Less memory usage but also less parallelism.
2. Ordering behavior:
`ParallelIterable` interleaves entries across manifests but this code
preserves ordering within a manifest. This is not strictly necessary, just
makes testing easier right?
It seems like both differences are not strictly needed and you could reuse
ParallelIterable? I bet (1) is not an issue in practice given
`ParallelIterable` already buffers whole manifests in full in other places like
scan planning.
--
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]
