andersglindstrom opened a new issue, #490:
URL: https://github.com/apache/arrow-js/issues/490
### Describe the bug, including details regarding any error messages,
version, and platform.
When `RecordBatchReader.from()` reads an async byte source that stays open
(a WebSocket, a network stream) and a chunk ends exactly at the end of a record
batch message, that batch is not yielded until the *next* chunk arrives. On a
live stream the latest batch is always one step behind, and the last batch
never appears until the stream ends.
The cause looks like an off-by-one in `src/io/adapters.ts`,
`fromAsyncIterable` (quoted, lightly simplified, from the published 21.2.0
build):
```ts
do {
({ cmd, size } = yield byteRange()); // serve the request; receive the
next one
} while (size < bufferLength); // BUG: stops when the next
request is exactly
// the bytes still buffered, then
waits for
// the next chunk before serving
them
```
The reader asks for a message body with `size === bufferLength` (exactly the
bytes left in the chunk). The inner loop exits because `size < bufferLength` is
false, and the outer loop awaits the next chunk before serving bytes that are
already buffered.
A likely fix:
```diff
- } while (size < bufferLength);
+ } while (size <= bufferLength);
```
The same `while (size < bufferLength)` appears in `fromIterable`,
`fromDOMStream` and `fromNodeStream`, so browser `ReadableStream` sources are
probably affected too (not tested).
### To reproduce
apache-arrow 21.2.0, Node 24. Three one-batch IPC messages, one per chunk,
one chunk per second, source never ends:
```js
// make the inputs with pyarrow:
// s = pa.schema([("x", pa.int64())]);
open("schema.bin","wb").write(s.serialize().to_pybytes())
// for i in range(3): open(f"b{i}.bin","wb").write(
// pa.record_batch([pa.array([i*10, i*10+1])],
schema=s).serialize().to_pybytes())
const { RecordBatchReader } = require("apache-arrow");
const fs = require("fs");
const frames = ["schema", "b0", "b1", "b2"].map(f => new
Uint8Array(fs.readFileSync(f + ".bin")));
const sleep = ms => new Promise(r => setTimeout(r, ms));
const t0 = Date.now();
async function* source() { // one message per chunk, like a
WebSocket; never ends
for (const f of frames) { yield f; await sleep(1000); }
await sleep(1e9);
}
(async () => {
const reader = await RecordBatchReader.from(source());
await reader.open();
for await (const b of reader)
console.log(`batch ${b.getChildAt(0).get(0)} at ${((Date.now() - t0) /
1000).toFixed(1)} s`);
})();
setTimeout(() => process.exit(0), 5500);
```
Output:
```
batch 0 at 2.0 s <- sent at 1.0 s
batch 10 at 3.0 s <- sent at 2.0 s
<- batch 20, sent at 3.0 s, never yielded
```
Expected: each batch at the time its chunk arrives (1.0 s, 2.0 s, 3.0 s).
Splitting each chunk so that its last byte arrives as a separate chunk makes
every batch arrive on time, which is consistent with the off-by-one above.
---
This report was prepared with the help of an AI coding assistant (Claude),
which traced the cause in the library source and wrote and ran the reproduction.
--
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]
