q2788116586-tech opened a new issue, #20344:
URL: https://github.com/apache/hudi/issues/20344
### Describe the problem you faced
## Summary
We observed spill decisions driven by sampled-size extrapolation in Spark
FileGroupReader (FGR), even with the buffered-record changes from #13225
present and in use. The map stores `BufferedRecord<UnsafeRow>`; this is not a
report that the new representation was bypassed.
In one observed map, 13,201 resident rows contained **136,456,568 B of
UnsafeRow data**, but ExternalSpillableMap accounted for **604,500,192 B**,
exceeding its effective budget of **412,316,860 B**. The 4.43x ratio compares
the estimate to row bytes, **not to measured retained heap**.
The large sampled row was genuinely large: getSizeInBytes() and backing
byte-array capacity were both 346,688 B, and the value estimator returned
346,912 B. Our evidence points to extrapolating a real outlier across the
resident population, rather than counting a huge unused backing buffer.
## Where the threshold crossing occurs
In ExternalSpillableMap.put(), a new in-memory entry periodically updates
the estimated average:
```java
estimatedPayloadSize = (long) (estimatedPayloadSize * 0.9
+ (keySizeEstimator.sizeEstimate(key)
+ valueSizeEstimator.sizeEstimate(value)) * 0.1);
currentInMemoryMapSize = inMemoryMap.size() * estimatedPayloadSize;
// The incoming entry is then charged at the new average.
```
Sanitized observations for the example above (all sizes in bytes):
```text
entriesBeforeSample=13200
previousEstimate=12322
sampleKeyBytes=112
sampleValueBytes=346912
sampleRowBytes=346688
sampleBackingArrayBytes=346688
newEstimate=45792
residentEntries=13201
residentRowBytesSum=136456568
residentBackingArrayBytesSum=136456568
estimatedMemory=604500192
effectiveBudget=412316860
unknownRows=0
oversizedBackingArrays=0
```
The calculation matches exactly:
- floor(0.9 * 12322 + 0.1 * (112 + 346912)) = 45792.
- 13201 * 45792 = 604500192.
- The next new key goes to disk.
Thus, the sample does not merely consume its own roughly 347 KB: it changes
the accounting for all resident entries. In this append-only scan, subsequent
disk-bound entries do not enter the in-memory sampling branch, so later smaller
rows cannot correct this average. This is not a claim that memory admission can
never resume after removals.
## Evidence across 64 maps
All 64 maps have complete initialization, first-spill and close diagnostics.
Every resident row at first spill was measured.
| Metric at first spill | Minimum | Median | Maximum |
|---|---:|---:|---:|
| Resident entries | 13,201 | 31,001 | 39,701 |
| Summed UnsafeRow bytes | 136,456,568 | 320,661,452 | 412,558,344 |
| Estimated map bytes | 412,317,432 | 457,816,281.5 | 868,016,268 |
| Per-map estimate / summed row bytes | 1.091x | 1.453x | 4.430x |
- 60/64 first spills immediately followed an average-size update crossing
the budget; the preceding estimate was below it.
- All resident snapshots had backing-array capacities equal to row lengths,
with zero unknown rows or oversized backing arrays.
- In 512 early samples and 64 last-sample observations, estimated value size
minus row length was always 224 B. This is a sample observation, not a
universal object-overhead constant.
- 63/64 row-byte totals were below budget; one was already slightly above
it. This does **not** prove that all 63 complete object graphs fit or that all
spill is avoidable.
The query completed successfully. This report concerns the spill decision,
not an exception or an OOM.
### To Reproduce
This is an instrumented downstream workload report, not yet a public minimal
reproducer or a reproduction on an unmodified Apache release.
Observed workload:
1. A frozen MOR table, version 5: 10,000,000 unique keys, 64 bucket/file
groups, 384 log files and no Parquet base files.
2. Initial full-row batch with 188 business columns, largely strings, plus
map/numeric fields and long-tailed row lengths. No in-memory key replacements
were observed. This diagnostic did not use subsequent partial-update batches.
3. Spark snapshot read with a full-width aggregate consuming the fields,
without join, sort or LIMIT.
4. Diagnostic logging of the existing samples, and one traversal of resident
values immediately before the first disk-map initialization to sum UnsafeRow
lengths and backing-array capacities.
The instrumentation leaves the sampling schedule, estimator calls, 0.9/0.1
formula, budget and spill branches unchanged. No row contents are logged. The
table manifest and timeline were unchanged before/after the run.
The private input cannot be shared. A proposed synthetic reproducer would
use variable-width rows with unique keys and a fixed insertion order, placing
occasional large rows at sampling boundaries. That synthetic reproducer has NOT
yet been executed; a constant-width generator would miss the observed mechanism.
### Expected behavior
Spill accounting should track resident data with a defensible allowance for
object/map overhead, without an occasional large sample revaluing all cached
rows and leaving an unrepresentative average in an append-only scan.
### Possible direction for discussion — not an implemented fix
For the Spark FGR path, could we incrementally account for sealed,
independently owned UnsafeRow storage, plus explicit key/wrapper/map and
ordering-state overhead, rather than multiplying a sampled average by all
resident entries?
This would need consistent insert/replace/remove accounting (possibly
retaining each entry's previous charge) and admission checks against the
projected total. A Spark-specific implementation could retain the existing
generic strategy as a fallback. It should avoid an expensive whole-object-graph
traversal for every row.
Open trade-offs include mutable/shared objects, accounting overhead,
additional per-entry metadata, and transient memory outside this map. We are
not proposing row bytes alone as the complete heap budget, nor claiming this
removes all spill.
Is this a known limitation after #13225? Would this direction fit the
existing design, or is there a better approach / related change we should
evaluate?
### Environment Description
- Hudi: downstream based on the 1.1.1 branch; changes from #13225, #13232
and #13408 are present.
- Spark: downstream 3.4.4; Scala 2.12; vendor JDK 8.
- Deployment/storage: YARN / HDFS.
- Executors: 16 x 4 cores x 8 GiB; dynamic allocation, AQE and speculation
disabled.
- Flink/Hive: not involved in this read diagnostic.
- Exact Hadoop version: not verified.
Effective configuration verified from all 64 map constructor logs:
```text
context=KeyBasedFileGroupRecordBuffer
rawBudget=515396075
effectiveBudget=412316860
diskMapType=BITCASK
compression=true
valueEstimator=org.apache.hudi.common.util.HoodieRecordSizeEstimator
serializer=org.apache.hudi.common.serialization.DefaultSerializer
```
The existing 0.8 sizing factor converts the raw budget to the effective map
budget. Compression acts after the spill decision and does not explain the
threshold crossing.
### Additional context
### Relationship to existing work
- #13225: BufferedRecord is present and used. This report concerns remaining
size-extrapolation behavior, not evidence that the refactoring has no benefit
or caused a regression.
- #13408 / HUDI-9499: the Avro estimator excludes shared schema overhead.
Our observed Spark path uses UnsafeRow and the generic estimator, not that
Avro-specific estimator. This alone does not establish a missing optimization
in Spark.
- #13232: re-estimation is confined to new in-memory entries. This behavior
is present but does not change the sampled-average formula.
- #19084 concerns BitCask/MDT-compaction heap usage. It is related context,
not the same observed failure: ours is a successful Spark snapshot read with
sample-driven threshold crossing, not MDT compaction OOM.
For a scoped comparison against Apache 1.1.0
(4d0280c975865873dec3345418649ba30ae26f25), HoodieRecordSizeEstimator is
unchanged and ExternalSpillableMap differs only by our diagnostic additions.
The downstream BufferedRecord has an additional nullable nested-ordering map;
it is null in every logged sample. These checks do not establish equivalence of
the entire downstream reader to upstream.
### Measurement limitations
Summed row bytes exclude keys, wrappers, HashMap storage, array headers and
alignment. We did not take a heap dump or measure retained heap. JOL reported
unavailable attach instrumentation; independent row/array lengths do not
validate every object-layout assumption.
The first-spill traversal took roughly 9–34 ms per map and adds diagnostic
overhead; this run is not a performance benchmark. We have not quantified how
much disk I/O or latency a corrected accounting strategy would avoid.
We would appreciate pointers to an existing issue covering this specific
sampling/extrapolation behavior and guidance on a suitable upstream reproducer.
### Stacktrace
```shell
No exception: the diagnostic query completed successfully. Sanitized
first-spill observations are included in the problem description.
```
--
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]
