revert)

stream_arrow_query used a synchronous Arrow IPC StreamWriter to an
async mpsc by spawning one tokio task per std::io::Write::write call:

    let handle = TOKIO.spawn(async move { tx.send(bytes).await });
    self.handles.push(handle);

Each StreamWriter::write(&batch) makes ~6 sequential write_all calls
The spawned tasks have no ordering relation; on a multi-thread tokio
runtime they race to send into the receiver, so bytes arrive in
arbitrary order and the resulting Arrow IPC stream gets corrupted.

The fix is to not call sync Write from inside an async future at
all. stream_arrow_query now hands StreamWriter a Vec<u8> and drains
the buffer between batches via std::mem::take(writer.get_mut()), then
yields a single ordered Bytes chunk per batch. Memory cost is bounded
by one record batch; behaviour matches stream_json_query, which has
always used this shape.

ChannelWriter retains its AsyncFileWriter impl for the parquet path
(AsyncArrowWriter awaits each write future before issuing the next,
so ordering there is already safe); the racy std::io::Write impl, the
handles vec, and the cfg(test) reordering shim are all removed.

Refs: #3923 #3792 #4287 #4226 #5814 #4240
Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

PR #5814 introduced pyarrow as an optional extra (feldera[arrow]),
gated behind a lazy import that surfaced a 'pip install feldera[arrow]'

We make this a non-optional import here because this is suppsed
to become the default format anyways going forward.

Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

 revert)

stream_arrow_query used a synchronous Arrow IPC StreamWriter to an
async mpsc by spawning one tokio task per std::io::Write::write call:

    let handle = TOKIO.spawn(async move { tx.send(bytes).await });
    self.handles.push(handle);

Each StreamWriter::write(&batch) makes ~6 sequential write_all calls
The spawned tasks have no ordering relation; on a multi-thread tokio
runtime they race to send into the receiver, so bytes arrive in
arbitrary order and the resulting Arrow IPC stream gets corrupted.

The fix is to not call sync Write from inside an async future at
all. stream_arrow_query now hands StreamWriter a Vec<u8> and drains
the buffer between batches via std::mem::take(writer.get_mut()), then
yields a single ordered Bytes chunk per batch. Memory cost is bounded
by one record batch; behaviour matches stream_json_query, which has
always used this shape.

ChannelWriter retains its AsyncFileWriter impl for the parquet path
(AsyncArrowWriter awaits each write future before issuing the next,
so ordering there is already safe); the racy std::io::Write impl, the
handles vec, and the cfg(test) reordering shim are all removed.

Refs: #3923 #3792 #4287 #4226 #5814 #4240
Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

PR #5814 introduced pyarrow as an optional extra (feldera[arrow]),
gated behind a lazy import that surfaced a 'pip install feldera[arrow]'

We make this a non-optional import here because this is suppsed
to become the default format anyways going forward.

Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

…tically

Adds a unit test that drives StreamWriter through the existing
ChannelWriter (the sync std::io::Write adapter that spawns one tokio
task per write call) and verifies the receiver-side StreamReader can
parse the byte stream back.

With four record batches and a four-worker tokio runtime the test
fails reliably with the same errors report in issues:

    ParseError("Unable to get root as message:
                RangeOutOfBounds { range: 655360..655364, .. }")
    IpcError("Expected schema message, found empty stream.")

Refs: #3923 #3792 #4287 #4226 #5814

 revert)

stream_arrow_query used a synchronous Arrow IPC StreamWriter to an
async mpsc by spawning one tokio task per std::io::Write::write call:

    let handle = TOKIO.spawn(async move { tx.send(bytes).await });
    self.handles.push(handle);

Each StreamWriter::write(&batch) makes ~6 sequential write_all calls
The spawned tasks have no ordering relation; on a multi-thread tokio
runtime they race to send into the receiver, so bytes arrive in
arbitrary order and the resulting Arrow IPC stream gets corrupted.

The fix is to not call sync Write from inside an async future at
all. stream_arrow_query now hands StreamWriter a Vec<u8> and drains
the buffer between batches via std::mem::take(writer.get_mut()), then
yields a single ordered Bytes chunk per batch. Memory cost is bounded
by one record batch; behaviour matches stream_json_query, which has
always used this shape.

ChannelWriter retains its AsyncFileWriter impl for the parquet path
(AsyncArrowWriter awaits each write future before issuing the next,
so ordering there is already safe); the racy std::io::Write impl, the
handles vec, and the cfg(test) reordering shim are all removed.

Refs: #3923 #3792 #4287 #4226 #5814 #4240
Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

PR #5814 introduced pyarrow as an optional extra (feldera[arrow]),
gated behind a lazy import that surfaced a 'pip install feldera[arrow]'

We make this a non-optional import here because this is suppsed
to become the default format anyways going forward.

Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

…tically

Adds a unit test that drives StreamWriter through the existing
ChannelWriter (the sync std::io::Write adapter that spawns one tokio
task per write call) and verifies the receiver-side StreamReader can
parse the byte stream back.

With four record batches and a four-worker tokio runtime the test
fails reliably with the same errors report in issues:

    ParseError("Unable to get root as message:
                RangeOutOfBounds { range: 655360..655364, .. }")
    IpcError("Expected schema message, found empty stream.")

Refs: feldera#3923 feldera#3792 feldera#4287 feldera#4226 feldera#5814

…eldera#4226 feldera#5814 revert)

stream_arrow_query used a synchronous Arrow IPC StreamWriter to an
async mpsc by spawning one tokio task per std::io::Write::write call:

    let handle = TOKIO.spawn(async move { tx.send(bytes).await });
    self.handles.push(handle);

Each StreamWriter::write(&batch) makes ~6 sequential write_all calls
The spawned tasks have no ordering relation; on a multi-thread tokio
runtime they race to send into the receiver, so bytes arrive in
arbitrary order and the resulting Arrow IPC stream gets corrupted.

The fix is to not call sync Write from inside an async future at
all. stream_arrow_query now hands StreamWriter a Vec<u8> and drains
the buffer between batches via std::mem::take(writer.get_mut()), then
yields a single ordered Bytes chunk per batch. Memory cost is bounded
by one record batch; behaviour matches stream_json_query, which has
always used this shape.

ChannelWriter retains its AsyncFileWriter impl for the parquet path
(AsyncArrowWriter awaits each write future before issuing the next,
so ordering there is already safe); the racy std::io::Write impl, the
handles vec, and the cfg(test) reordering shim are all removed.

Refs: feldera#3923 feldera#3792 feldera#4287 feldera#4226 feldera#5814 feldera#4240
Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>

PR feldera#5814 introduced pyarrow as an optional extra (feldera[arrow]),
gated behind a lazy import that surfaced a 'pip install feldera[arrow]'

We make this a non-optional import here because this is suppsed
to become the default format anyways going forward.

Signed-off-by: Gerd Zellweger <mail@gerdzellweger.com>