[python legacy] use process pool instead of thread pool for files planning

[puchengy]

Based on analysis, plan_files() spent most of the time on doing copy and file reading which are both CPU bounded operations. However, current implementation is using thread pool instead of process pool which does not speed up the operation at all.

This diff proposes switch to process pool instead of thread pool.

Frame graph

Command
py-spy record -o profile.svg -- python myprogram.py

Code

from iceberg.hive import HiveTables

conf = {"hive.metastore.uris": 'xxxx', 'iceberg.scan.plan-in-worker-pool': True, 'iceberg.worker.num-threads': 4}
tables = HiveTables(conf)
table = tables.load('abc.xyz')
scan = table.new_scan()
files = scan.plan_files()

[samredai]

Looking at the function passed to the pool (get_scans_for_manifest) I can understand how this improves the parsing and evaluator logic. For the avro file reads, I'm not sure but I'd expect this would scale better with multithreading since that's I/O bound. @Fokko is it possible that threading may be better here beyond a certain number of files?

[rdblue]

Sorry, I didn't mean to close this. That was an accident.

[Fokko]

Looks good. If we run into memory issues, we could also replace the current map function:

iceberg/python_legacy/iceberg/core/data_table_scan.py

Lines 65 to 73 in 93111e8

	if self.ops.conf.get(SCAN_THREAD_POOL_ENABLED):
	with Pool(self.ops.conf.get(WORKER_THREAD_POOL_SIZE_PROP,
	cpu_count())) as reader_scan_pool:
	return itertools.chain.from_iterable([scan for scan
	in reader_scan_pool.map(self.get_scans_for_manifest,
	matching_manifests)])
	else:
	return itertools.chain.from_iterable([self.get_scans_for_manifest(manifest)
	for manifest in matching_manifests])

Replace the map with the it's lazy imap brother:
https://docs.python.org/3/library/multiprocessing.html#multiprocessing.pool.Pool.imap

[Fokko]

nit; we could combine the imports:

from multiprocessing import cpu_count, Pool

[Fokko]

@samredai

Looking at the function passed to the pool (get_scans_for_manifest) I can understand how this improves the parsing and evaluator logic. For the avro file reads, I'm not sure but I'd expect this would scale better with multithreading since that's I/O bound. @Fokko is it possible that threading may be better here beyond a certain number of files?

I think it would make sense to parallelize that as well to reduce the runtime of the function. Especially if we need to fetch files from external storage that introduces some latency, I would parallelize that kind of calls as well to improve overall throughput.

[kbendick]

Overall I think this looks good as well.

I don't know if scrutinizing it super closely for corner cases is necessary given that we've seen the performance gains and it is the legacy python project.

I've put out some calls for testing in the community for people who I know do actively use the python_legacy project a lot. If there's any negative feedback, we can always revert if need be. 👍

Good find @puchengy!

[chyzzqo2]

In general in python its hard to deal with libraries that make this choice for you. Parallelism via threads and processes don't mix so the choice is better left to the application rather than the library. Can we make the choice configurable?

[samredai]

@chyzzqo2 this is a great point. We already have the iceberg.scan.plan-in-worker-pool config property that enables this code path. Maybe we can add a config along the lines of iceberg.scan.plan-in-worker-pool.type with values of "threads" or "processes". Also I just noticed that "threads" is in the config property name for pool sizes (here) which can be confusing if processes are actually being used.

[rdblue]

I think we want to follow up and implement @samredai's suggestion to add a pool type that controls this. We can do that in a follow-up, though. In the meantime, I'll merge this.

[rdblue]

Thanks, @puchengy!

[puchengy]

@rdblue Thanks I will have a follow up PR for that.