Skip to content

fix: Yield buffered bytes before pulling the next chunk in stream adapters - #502

Open
cpruijsen wants to merge 2 commits into
apache:mainfrom
cpruijsen:fix/issue-490
Open

cpruijsen wants to merge 2 commits into
apache:mainfrom
cpruijsen:fix/issue-490

Conversation

@cpruijsen

Copy link
Copy Markdown

What's Changed

Yield buffered bytes before the next source chunk for positive exact-fit requests in src/io/adapters.ts.
Record batches no longer arrive one chunk late when an IPC message ends exactly at a source chunk boundary.
This applies to Iterable, AsyncIterable, DOM ReadableStream, and Node ReadableStream.

The condition keeps the existing < case and adds equality only for positive sizes.
Null and zero requests retain their existing behavior.

test/unit/ipc/reader/stream-reader-tests.ts adds deterministic tests for schema and batch reads through iterable sources without a pull of the next message chunk.
The tests also check batch contents, finite-stream completion, and consecutive null/zero reads through iterable sources.

The latency tests fail on the unchanged baseline with Expected: 1 and Received: 2, and pass with the fix.

Closes #490.

@kou
kou requested a balanced review from Copilot October 14, 2026 00:40
@kou

kou commented Oct 14, 2026

Copy link
Copy Markdown
Member

@andersglindstrom Could you review this?

Copilot AI left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Copilot review overview

3 open findings
What changed in this PR

Fixes stream adapter buffering behavior so exact-fit positive read requests yield remaining buffered bytes before pulling the next source chunk, preventing IPC record batches from arriving one chunk late at chunk boundaries.

Changes:

  • Adjusted adapter generator loop conditions to handle positive size === bufferLength reads without prefetching another chunk.
  • Added unit tests verifying schema/batch reads don’t trigger an extra pull for iterable/async-iterable sources and that null/zero reads advance correctly.
File Description
src/​io/​adapters.ts Updates adapter yield loop conditions for Iterable/AsyncIterable/DOM/Node stream sources to avoid prefetching on exact-fit reads.
test/​unit/​ipc/​reader/​stream-reader-tests.ts Adds deterministic regression tests for chunk-boundary behavior and ByteStream null/zero read semantics.

🧠 Review effort: Lite


💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Comment thread src/io/adapters.ts
Comment on lines 93 to 97
if (done || size <= bufferLength) {
do {
({ cmd, size } = yield byteRange());
} while (size < bufferLength);
} while (size < bufferLength || (size > 0 && size === bufferLength));
}
Comment thread src/io/adapters.ts
Comment on lines 192 to 196
if (done || size <= bufferLength) {
do {
({ cmd, size } = yield byteRange());
} while (size < bufferLength);
} while (size < bufferLength || (size > 0 && size === bufferLength));
}
Comment on lines +80 to +96
async function* chunks() {
for (const message of messages) { pulled++; yield message; }
// the byte stream stays open, like a WebSocket or network stream
await new Promise(() => { });
}

const reader = await (await RecordBatchReader.from(chunks())).open();
const iterator = reader[Symbol.asyncIterator]();

// reading the schema should not pull the next chunk
expect(pulled).toBe(1);
for (let i = 0; i < table.batches.length; i++) {
const { value } = await iterator.next();
expect(value).toEqualRecordBatch(table.batches[i]);
expect(pulled).toBe(i + 2);
}
await iterator.return!(null);
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

RecordBatchReader on an open-ended async stream withholds each record batch until the next chunk arrives

3 participants