feat(csharp): Use RecyclableMemoryStream for LZ4 decompression to reduce memory pressure #1
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…uce memory pressure Changes: - Modified Lz4Utilities.DecompressLz4Async to return MemoryStream instead of (buffer, length) tuple - Stream is now disposed by the caller (Arrow) after reading, not immediately - Uses RecyclableMemoryStream for pooled memory allocation - Decompression uses CopyAsync with 80KB chunks for efficient streaming - Fixed bug where buffer was recycled before Arrow finished reading Benefits: - Reduces GC pressure by reusing memory streams from pool - Fixes "Unexpectedly reached end of stream" error - Memory is only returned to pool after Arrow completes reading - More efficient memory usage for high-volume CloudFetch operations 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <[email protected]>
Owner
Author
|
Closing - will create PR against apache/arrow-adbc instead |
eric-wang-1990
pushed a commit
that referenced
this pull request
Jan 15, 2026
…che#3870) ## Fix Critical Deadlocks and Race Conditions in Snowflake Record Reader This PR addresses multiple critical concurrency issues in the Snowflake driver's `recordReader` that could cause complete application hangs under normal racing conditions. ### Issues Fixed *1. Critical Deadlock: `Release()` Blocking Forever* *Problem*: When `Release()` was called while producer goroutines were blocked on channel sends, a permanent deadlock occurred: * `Release()` cancels context and attempts to drain channels * Producer goroutines blocked on `ch <- rec` cannot see the cancellation * Channels never close because producers never exit * `Release()` blocks forever on `for rec := range ch` *Fix:* Added a `done` channel that signals when all producer goroutines have completed. `Release()` now waits for this signal before attempting to drain channels. *2. Severe Deadlock: Non-Context-Aware Channel Sends* *Problem:* Channel send operations at lines 694 and 732 checked context before the send but not during: ```go for rr.Next() && ctx.Err() == nil { // Context checked here // ... ch <- rec // But send blocks here without checking context } ``` *Fix:* Wrapped all channel sends in `select` statements with context awareness: ```go select { case chs[0] <- rec: // Successfully sent case <-ctx.Done(): rec.Release() return ctx.Err() } ``` *3. Critical Race Condition: Nil Channel Reads* *Problem:* Channels were created asynchronously in goroutines after `newRecordReader` returned. If `Next()` was called quickly after creation, it could read from uninitialized (nil) channels, causing infinite blocking. *Fix:* Initialize all channels upfront before starting any goroutines: ```go chs := make([]chan arrow.RecordBatch, len(batches)) for i := range chs { chs[i] = make(chan arrow.RecordBatch, bufferSize) } ``` *4. Goroutine Leaks on Initialization Errors* *Problem:* Error paths only cleaned up the first channel, potentially leaking goroutines if initialization failed after starting concurrent operations. *Fix:* Moved all error-prone initialization (GetStream, NewReader) before goroutine creation, and added proper cleanup on errors. ---------------------- #### Changes * Added `done` channel to `reader` struct to signal goroutine completion * Initialize all channels upfront to eliminate race conditions * Use context-aware sends with `select` statements for all channel operations * Update `Release()` to wait on `done` channel before draining * Reorganize initialization to handle errors before starting goroutines * Signal completion by closing `done` channel after all producers finish #### Reproduction Scenarios Prevented *Deadlock #1:* 1. bufferSize = 1, producer generates 2 records quickly 2. Channel becomes full after first record 3. Producer blocks on send 4. Consumer calls Release() before Next() 5. Without fix: permanent deadlock 6. With fix: producer responds to cancellation, Release() completes *Race Condition:* 1. Query returns 3 batches 2. First batch processes quickly 3. Next() advances to second channel 4. Without fix: reads from nil channel, blocks forever 5. With fix: channel already initialized, works correctly See apache#3730
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Implements RecyclableMemoryStream for LZ4 decompression to reduce memory pressure and fix stream corruption issues.
Changes
Lz4Utilities.DecompressLz4Asyncto returnMemoryStreaminstead of(buffer, length)tupleRecyclableMemoryStreamManagerfor pooled memory allocationCopyAsyncwith 80KB chunks for efficient streamingProblem Fixed
Previously, the RecyclableMemoryStream was disposed immediately after decompression, returning its buffer to the pool. This caused "Unexpectedly reached end of stream" errors when Arrow tried to read from the buffer later, as it could have been overwritten by other operations.
Solution
The new API returns the stream itself without disposing it. Arrow reads from the stream and then disposes it when done, ensuring the buffer is only returned to the pool after reading completes.
Benefits
Testing
🤖 Generated with Claude Code