Core: Support parallel execution when scanning entries in ManifestGroup

[dramaticlly]

Currently, executorService is only used in ManifestGroup::plan

iceberg/core/src/main/java/org/apache/iceberg/ManifestGroup.java

Lines 215 to 219 in a97b4ec

	if (executorService != null) {
	return new ParallelIterable<>(tasks, executorService);
	} else {
	return CloseableIterable.concat(tasks);
	}

But it can also benefit when scanning entries in multiple manifests.

Also apply defensive copy on entries as ManifestReader reuseContainers across iteration

@stevenzwu @RussellSpitzer

[RussellSpitzer]

Could you elaborate more on the previous behavior, and how you are changing that in this PR?

[dramaticlly]

Could you elaborate more on the previous behavior, and how you are changing that in this PR?

Happy to! Before this change when we scan the manifest entries using ManifestGroup, we will scan manifest sequentially using CloseableIterable.concat even when executorService is provided.

After this change, we now can benefit from using threadpool to scan the manifests if ManifestGroup is created with executorService provided in planWith(). One caveat is that we also need to make defensive copy of entries in parallel scanning as entries are reused across all iterations within a manifest given the reuseContainers in ManifestReader.

[deniskuzZ]

Should we consider migrating to Executors.newVirtualThreadPerTaskExecutor() (jdk21)?

[dramaticlly]

I think we still build against java17, so we can migrate this later once we move to jdk21 as minimal support.

[RussellSpitzer]

I'm a little worried about introducing a sort of hidden "materialize everything" setting.

We have one internal use where this could come up and we are effectively double copying with this change

FindFiles

iceberg/core/src/main/java/org/apache/iceberg/FindFiles.java

Lines 201 to 223 in 8417225

	public CloseableIterable<DataFile> collect() {
	Snapshot snapshot =
	snapshotId != null ? ops.current().snapshot(snapshotId) : ops.current().currentSnapshot();
	// snapshot could be null when the table just gets created
	if (snapshot == null) {
	return CloseableIterable.empty();
	}
	// when snapshot is not null
	CloseableIterable<ManifestEntry<DataFile>> entries =
	new ManifestGroup(ops.io(), snapshot.dataManifests(ops.io()))
	.specsById(ops.current().specsById())
	.filterData(rowFilter)
	.filterFiles(fileFilter)
	.filterPartitions(partitionFilter)
	.ignoreDeleted()
	.caseSensitive(caseSensitive)
	.planWith(executorService)
	.entries();
	return CloseableIterable.transform(entries, entry -> entry.file().copy(includeColumnStats));
	}

So not even counting external users of this library we are already introducing a regression. I'm not saying we shouldn't do this but we should be very careful here to avoid accidentally doubling memory consumption (or greatly increasing consumption) of a downstream consumer.

[dramaticlly]

Thanks Russell, I think you raised a good point as existing consumer of this API might already apply the defensive copy so we dont want to double the memory. I took the stab to follow what's existing

iceberg/core/src/main/java/org/apache/iceberg/ManifestGroup.java

Line 180 in 9534c9b

public <T extends ScanTask> CloseableIterable<T> plan(CreateTasksFunction<T> createTasksFunc) {

is doing and add a new method which takes a function for transform.

So testManifestGroupEntriesWithParallelExecution() kind of illustrate that if we only need filePath, we can skip defensive copy like how we collect DataFiles in FindFiles::collect