You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "cpruijsen (via GitHub)" <gi...@apache.org> on 2026/10/13 18:35:09 UTC
[PR] fix: Yield buffered bytes before pulling the next chunk in stream adapters [arrow-js]
cpruijsen opened a new pull request, #502:
URL: https://github.com/apache/arrow-js/pull/502
## 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.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: Yield buffered bytes before pulling the next chunk in stream adapters [arrow-js]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on PR #502:
URL: https://github.com/apache/arrow-js/pull/502#issuecomment-6158311874
@andersglindstrom Could you review this?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix: Yield buffered bytes before pulling the next chunk in stream adapters [arrow-js]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #502:
URL: https://github.com/apache/arrow-js/pull/502#discussion_r4265669624
##########
src/io/adapters.ts:
##########
@@ -93,7 +93,7 @@ function* fromIterable<T extends ArrayBufferViewInput>(source: Iterable<T> | T):
if (done || size <= bufferLength) {
do {
({ cmd, size } = yield byteRange());
- } while (size < bufferLength);
+ } while (size < bufferLength || (size > 0 && size === bufferLength));
}
Review Comment:
The same non-trivial loop condition is duplicated in four adapters (Iterable/AsyncIterable/DOM/Node). To reduce drift risk and make the exact-fit rationale clearer, consider extracting this predicate into a small shared helper (or at least a named local function/constant) and add a brief comment explaining why equality is included only for positive sizes.
##########
test/unit/ipc/reader/stream-reader-tests.ts:
##########
@@ -59,3 +67,84 @@ for (const table of generateRandomTables([10, 20, 30])) {
});
});
}
+
+describe(`RecordBatchStreamReader over message-sized chunks`, () => {
+
+ const source = tableFromArrays({ x: [0, 10, 20, 30, 40, 50] });
+ const batch = source.batches[0];
+ const table = new Table(source.schema, [batch.slice(0, 2), batch.slice(2, 4), batch.slice(4, 6)]);
+
+ test(`AsyncIterable yields each RecordBatch as soon as its chunk arrives`, async () => {
+ const messages = [...ipcMessages(await RecordBatchStreamWriter.writeAll(table).toUint8Array())];
+ let pulled = 0;
+ 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);
Review Comment:
The async generator intentionally awaits a never-resolving Promise to model an open stream; if a regression causes the reader to over-pull (e.g., past EOS) this can turn into a hard hang that’s difficult to diagnose. Consider making the wait cancellable (e.g., via an AbortController or a deferred you resolve in a `finally` when the generator is closed) so the test still models an open stream but always terminates cleanly under failure.
##########
src/io/adapters.ts:
##########
@@ -192,7 +192,7 @@ async function* fromDOMStream<T extends ArrayBufferViewInput>(source: ReadableSt
if (done || size <= bufferLength) {
do {
({ cmd, size } = yield byteRange());
- } while (size < bufferLength);
+ } while (size < bufferLength || (size > 0 && size === bufferLength));
}
Review Comment:
This change is applied to the DOM and Node stream adapters as well, but the added tests appear to exercise only Iterable/AsyncIterable sources (plus ByteStream semantics). To prevent adapter-specific regressions, add targeted unit tests that reproduce the exact-fit boundary case through `ReadableStream` and Node `Readable` inputs too (matching the PR description’s scope).
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org