raunaqmorarka commented on code in PR #18076:
URL: https://github.com/apache/iceberg/pull/18076#discussion_r4007880119
##########
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:
Extracted the read-ahead into a `BinReader` so `createManifest` is just a
loop over the bin.
`ParallelIterable` deadlocks in this position. Bins already run on the
worker pool through `Tasks.range(...).executeWith(workerPool)`, so a
`ParallelIterable` consumed inside a bin blocks in `hasNext()` while its own
read tasks sit behind the bin tasks in the same queue. With a pool of 4 I saw 1
to 3 bins finish fine, while 4 and 5 bins hung in 2 of 3 runs. Scan planning is
safe because its consumer is the planner thread, not a pool thread.
Ordering is required here, not only for tests. `mergeGroup` already
documents that manifest and entry order is preserved so data files are not
deleted at random when they age off. Entries with a null `first_row_id` also
get one assigned positionally at read time, and a bin can hold more than one
uncommitted manifest, so a non-deterministic order could hand out different row
ids across commit retries.
The buffering in scan planning is not comparable either. `ManifestGroup`
copies entries through `ContentFileUtil.copy(..., shouldKeepStats,
columnsToKeepStats)` and usually drops column stats, while a merge has to write
full stats back out. The default queue of 30000 in `ParallelIterable` assumes
about 500 bytes per file, and a full entry with stats on a wide table is far
larger. It also queues entries without copying them, and
`ManifestReader.entries()` reuses one container, so a copy would be needed
anyway.
--
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]
