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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
