rewrite_table_path: manifest list and version file rewriting is always single-threaded when called from PySpark / SQL procedure
- Dominant language
- Java
- Stars
- 9.2k
- Forks
- 3.5k
- Avg merge
- 2d 11h
- Merged PRs (30d)
- 132
Description
### Feature Request / Improvement
`rewrite_table_path` rewrites manifest list files and version files sequentially by default. The Java API exposes `executeWith(ExecutorService)` to parallelize these stages, but this method is inaccessible from PySpark or SQL because:
1. The `RewriteTablePathProcedure` has no parameter for thread count — it only accepts `table`, `source_prefix`, `target_prefix`, `start_version`, `end_version`, `staging_location`, and `create_file_list`.
2. PySpark has no mechanism to construct and pass a Java `ExecutorService` object to a Spark procedure.
As a result, users running `rewrite_table_path` from PySpark on tables with large snapshot histories are forced into sequential manifest list rewriting with no way to tune parallelism.
**Current Behavior**
In `RewriteTablePathSparkAction.rebuildMetadata()`, the manifest list rewriting loop is:
```java
Tasks.foreach(validSnapshots)
.noRetry()
.throwFailureWhenFinished()
.executeWith(executorService) // null when no ExecutorService was provided
.run(snapshot -> manifestListResults.add(
rewriteManifestList(snapshot, endMetadata, manifestsToRewrite)));
```
When `executorService` is `null`, `Tasks.foreach` falls back to sequential single-threaded execution. For a table with hundreds or thousands of snapshots, this is a significant bottleneck.
The `executeWith(ExecutorService)` method on the `RewriteTablePath` interface documents this explicitly:
> *"If this method is not called, these operations will be performed sequentially."*
However, the `RewriteTablePathProcedure` never calls `executeWith`, and there is no procedure parameter to configure it.
**Expected Behavior**
PySpark users should be able to control the degree of parallelism for manifest list and version file rewriting via a procedure option, consistent with how other Iceberg Spark procedures expose tuning knobs (e.g., `rewrite_data_files` accepts an `options => map(...)` parameter).
**Proposed Solution**
Add a `thread-pool-size` option to `RewriteTablePath` (as a string constant with a documented default, e.g. `4`) and wire it through `RewriteTablePathSparkAction` and `RewriteTablePathProcedure`.
**In `RewriteTablePath` (the interface):**
```java
String THREAD_POOL_SIZE = "thread-pool-size";
int THREAD_POOL_SIZE_DEFAULT = 4;
```
**In `RewriteTablePathSparkAction.rebuildMetadata()`:**
```java
int poolSize = PropertyUtil.propertyAsInt(
options(), RewriteTablePath.THREAD_POOL_SIZE, RewriteTablePath.THREAD_POOL_SIZE_DEFAULT);
ExecutorService pool = ThreadPools.newFixedThreadPool("rewrite-table-path", poolSize);
try {
Tasks.foreach(validSnapshots)
.noRetry()
.throwFailureWhenFinished()
.executeWith(pool)
.run(snapshot -> ...);
} finally {
pool.shutdown();
}
```
**In `RewriteTablePathProcedure`:** add an optional `options` parameter of type `MapType(StringType, StringType)` and forward it to the action via `action.options(optionsMap)`, following the same pattern as `RewriteDataFilesProcedure`.
This would allow PySpark users to tune parallelism without any Java interop:
```python
spark.sql("""
CALL system.rewrite_table_path(
table => 'db.my_table',
source_prefix => 'hdfs://old-cluster/',
target_prefix => 'hdfs://new-cluster/',
options => map('thread-pool-size', '16')
)
""")
```
**Affected Files**
- `api/src/main/java/org/apache/iceberg/actions/RewriteTablePath.java`
- `spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java`
- `spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/RewriteTablePathProcedure.java`
- Same for `spark/v4.0` and `spark/v4.1`
- `docs/docs/spark-procedures.md`
**References**
- `RewriteTablePath.executeWith` — [1](#33-0)
- `rebuildMetadata` sequential loop — [2](#33-1)
- `RewriteTablePathProcedure` — no `options` parameter — [3](#33-2)
### Query engine
Spark
### Willingness to contribute
- [ ] I can contribute this improvement/feature independently
- [ ] I would be willing to contribute this improvement/feature with guidance from the Iceberg community
- [x] I cannot contribute this improvement/feature at this time
Contributor guide
Research direction
Start with RewriteTablePath.java, RewriteTablePathSparkAction.java, and RewriteTablePathProcedure.java in the listed Spark versions; compare RewriteDataFilesProcedure for the options parameter pattern. Trace rebuildMetadata and the existing executeWith behavior, then update docs/docs/spark-procedures.md. Done means the procedure accepts and forwards thread-pool-size, metadata rewriting uses the configured pool, and the option is documented consistently.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, distributed-systems
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100