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. Behavior under large manifests with many manifest entries:
`ParallelIterable` previously had problems with holding open too
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?
--
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]
