voonhous opened a new issue, #20543:
URL: https://github.com/apache/hudi/issues/20543
### Describe the problem
Initializing the partitioned (non-global) record level index fails with
`StackOverflowError` on tables with a few hundred data partitions or more.
`PartitionedRecordIndexer.buildInitialization` returns one
`DataPartitionAndRecords` per data partition, and
`HoodieBackedTableMetadataWriter.initializeMetadataPartition`
(`HoodieBackedTableMetadataWriter.java:467-472`) unions them one at a time:
```java
HoodieData<HoodieRecord> records = engineContext.emptyHoodieData();
for (DataPartitionAndRecords dataPartitionAndRecords:
initializationPlan.dataPartitionAndRecords()) {
initializeFileGroups(...);
records = records.union(dataPartitionAndRecords.indexRecords());
}
bulkCommit(instantTimeForPartition, relativePartitionPath, records,
initializationPlan.indexParser());
```
`HoodieJavaRDD.union` is `JavaRDD.union`, and Spark does not flatten nested
unions, so the lineage passed to `bulkCommit` is a chain of nested `UnionRDD`s
as deep as the number of data partitions. Spark Java-serializes that lineage
into the task binary recursively, once per level, so the driver overflows in
`DAGScheduler.submitMissingTasks` (`Task serialization failed:
java.lang.StackOverflowError`, surfacing as `HoodieMetadataException: Bootstrap
on record_index partition failed`). When driver serialization succeeds,
executors can overflow while deserializing the task instead, which kills the
executor.
The loop has been there since partitioned RLI was added in #13603 (release
1.1.0); the indexer refactor in #18348 kept it.
Measured on JDK 11 (arm64), Spark 3.5, local mode, one file per partition:
| `-Xss` | passes up to | fails from |
|---|---|---|
| 512k | 250 partitions | 260-300 |
| 1m (Linux x86_64 default) | ~450-520 | ~500-550 |
| 2m (macOS arm64 default) | ~1000 | 1100+ |
So with the usual 1 MB stack, roughly 500 data partitions (about 18 months
of daily partitions) is enough to break initialization.
### Repro
Write a COW table with N partitions (one record each) with RLI off, then
enable `hoodie.metadata.record.level.index.enable=true` so the next metadata
writer initializes the partitioned RLI. With N around 600 on a 1 MB stack the
initialization fails as above. A deterministic check is to register a
`SparkListener` during initialization and compute the longest chain of
`UnionRDD`s from the stage `rddInfos`: it equals N (50 for N = 50).
### Proposed fix
Collect the `indexRecords()` into a list in the loop and union them once
with a flat union (`SparkContext.union(Seq)` yields a single `UnionRDD`). There
is no `HoodieEngineContext.union(List)` today, so this needs a small
engine-context API (with a fold fallback for non-Spark engines), plus a
regression test asserting the union depth stays flat.
--
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]
