wangyum opened a new pull request, #3927:
URL: https://github.com/apache/parquet-java/pull/3927
### Rationale for this change
`MemoryManager#addWriter`/`removeWriter` are `synchronized` on a JVM-global
singleton, and on **every** writer open/close they ran `updateAllocation()`,
which:
- iterated **all** registered writers to sum allocations (O(N)),
- recomputed and **pushed** a new threshold into every writer
(`setRowGroupSizeThreshold`),
- eagerly `String.format`-ed a debug line for **every writer** even when
debug logging was disabled.
With many concurrently opening/closing writers — Spark's dynamic partition
writes being the canonical case (`df.write.partitionBy(...)` producing
thousands of files) — the O(N) critical section turns the global monitor
into a convoy: as profiled in #2860 / SPARK-44003, worker threads were
blocked on this lock for a total of ~20.5 hours across a 10-minute JFR
window, roughly doubling one production job's wall-clock (2h with
partitioned output vs 45min without).
### What changes are included in this PR?
- **Push → pull**: writers no longer store a pushed threshold.
`InternalParquetRecordWriter#getRowGroupSizeThreshold()` computes
`floor(requestedSize * scale)` on demand from the manager's
`volatile double scale`. The per-record write path is unchanged (it
still uses the cached `nextRowGroupSize`, refreshed only at row-group
boundaries).
- **O(1) bookkeeping**: `addWriter`/`removeWriter` maintain an incremental
`totalAllocations` and a size multiset (`TreeMap<Long, Integer>`) so
`recomputeScale()` never visits `writerList`. The `minMemoryAllocation`
guard reduces to checking the smallest live allocation: `floor` is
monotone, so the smallest allocation is always the first to breach the
floor. Per-writer debug logging now sits behind `isDebugEnabled`, out of
the critical section.
- **Atomicity preserved**: both methods remain `synchronized`, so
`recomputeScale()` still executes atomically with respect to
`writerList` — the concern raised on the previous attempt #1240
(ConcurrentHashMap). This change shrinks the critical section instead of
removing the lock.
- **Failure-path hardening**:
- `addWriter` now **rolls back its registration** when `recomputeScale()`
rejects the writer (e.g. its allocation would scale below
`minMemoryAllocation`). Previously the rejected writer leaked its
allocation in the global manager forever — the writer never escaped
its constructor, so nothing could ever close it — keeping every later
writer on the same JVM over-allocated and re-tripping the same
exception.
- `removeWriter` resets `scale` to 1.0 once the last writer is removed,
so a stale shrunken scale cannot be observed while the manager sits
idle.
Lock hold time per add/remove drops from O(N) field writes plus O(N)
string formatting to O(1) arithmetic plus an O(log D) multiset update
(D = distinct row-group sizes, typically 1).
### Are these changes tested?
- New `testConcurrentAddRemoveWriters`: 20 threads × 200 open/close cycles
while the pool is sustained well past its limit — no bookkeeping
corruption, no spurious duplicate-writer/minimum-allocation exceptions,
all writers deregistered, and the degraded (scale < 1.0) path is actually
exercised. This is the concurrent add/remove unit test requested in the
#1240 review.
- New `testMinimumAllocationGuard`: pins both directions of the
`TreeMap#firstKey`-based guard (a writer alone never trips it; enough
concurrent writers deterministically do) and the rollback behavior (a
rejected writer leaves no residue, the manager stays usable, scale
returns to 1.0).
- Full `parquet-hadoop` suite passes locally: 910 tests, 0 failures.
### Are there any user-facing changes?
No public API changes. The removed `setRowGroupSizeThreshold` and the new
`setMemoryManager`/`getScale` usages are package-private. The scaling
heuristic itself is unchanged: identical `floor(allocation * scale)`
values and identical callback firing points (callbacks run exactly when
the pool is exceeded at add/remove time, as before). The only observable
behavioral differences are the failure-path fixes described above.
Closes #2860
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
