[IcebergIO] Fix stale openWriters count in RecordWriterManager - #39157
Conversation
RecordWriterManager uses a Guava Cache with expireAfterAccess(1 min) for writer eviction. However, Guava Cache eviction is lazy — expired entries are only removed on subsequent access or explicit cleanUp(). The write() method checks `openWriters >= maxNumWriters` without first calling cleanUp(), so expired writers remain in the cache and openWriters stays stale. With DEFAULT_MAX_WRITERS_PER_BUNDLE = 20 and more than 20 partitions, this causes 100% false spill — every record for the 21st+ partition is rejected as if the writer pool is full. Add writers.cleanUp() before the capacity check to evict expired entries and update openWriters accurately. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request addresses a bug in the IcebergIO RecordWriterManager where stale writer counts caused premature spilling. By forcing a cache cleanup before evaluating the maximum writer capacity, the system now correctly identifies available slots, preventing unnecessary disk usage when handling tables with many partitions. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request modifies the RecordWriterManager to call writers.cleanUp() when the open writers limit is reached, attempting to free up space before rejecting a write. The review feedback highlights that calling cleanUp() on every write attempt under limit pressure can cause significant CPU overhead and lock contention, and suggests rate-limiting these cleanup calls.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if (writer == null && openWriters >= maxNumWriters) { | ||
| return false; | ||
| writers.cleanUp(); | ||
| if (openWriters >= maxNumWriters) { | ||
| return false; | ||
| } | ||
| } |
There was a problem hiding this comment.
Calling writers.cleanUp() on every single write attempt that results in a spill can introduce significant CPU overhead and lock contention, especially when processing a large volume of records for partitions that exceed maxNumWriters. Since Guava's cleanUp() acquires locks on all cache segments, calling it repeatedly in a tight loop for every spilled record is inefficient.\n\nTo optimize this, consider rate-limiting the cleanUp() calls (e.g., at most once every second) by tracking the last cleanup time using a timestamp field (e.g., lastCleanupNanos) in DestinationState.\n\nNote: You will need to define private long lastCleanupNanos = 0L; as a field in the DestinationState class for this suggestion to compile.
if (writer == null && openWriters >= maxNumWriters) {\n long now = System.nanoTime();\n if (now - lastCleanupNanos > 1000000000L) {\n writers.cleanUp();\n lastCleanupNanos = now;\n }\n if (openWriters >= maxNumWriters) {\n return false;\n }\n }
|
@ahmedabu98 — Would you mind reviewing this? These are a set of IcebergIO fixes validated on production-scale benchmarks (39M users, 400 partitions, 99-column schema). The four PRs are independent and can be reviewed/merged in any order. |
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
Failing test is an unrelated flake (#19480). Merging |
2 participants
Summary
writers.cleanUp()before theopenWriters >= maxNumWriterscapacity check inRecordWriterManager.DestinationState.write()cleanUp(), not proactivelyopenWritersstays stale after writers expire, causing 100% false spill for tables with more partitions thanmaxNumWriters(default 20)Motivation
With
DEFAULT_MAX_WRITERS_PER_BUNDLE = 20and more than 20 partitions, every record for the 21st+ partition is rejected as if the writer pool is full, triggering the spill path. Observed on a 400-partition table: 100% of records spilled, causing disk exhaustion.After the fix, expired writers are properly evicted before the capacity check, and spill only occurs when the pool is genuinely full.
Test plan
RecordWriterManagerTestpasses🤖 Generated with Claude Code