Rich-T-kid commented on code in PR #10277:
URL: https://github.com/apache/arrow-rs/pull/10277#discussion_r3567918456
##########
arrow-ipc/src/writer.rs:
##########
@@ -2522,6 +2773,89 @@ mod tests {
stream_reader.next().unwrap().unwrap()
}
+ fn encode_stream(
+ schema: &Schema,
+ batches: &[RecordBatch],
+ options: IpcWriteOptions,
+ ) -> Vec<u8> {
+ let mut encoder = StreamEncoder::try_new_with_options(schema,
options).unwrap();
+ let mut bytes = Vec::new();
+ for batch in batches {
+ for buffer in encoder.encode(batch).unwrap() {
+ bytes.extend_from_slice(buffer.as_slice());
+ }
+ }
+ for buffer in encoder.finish().unwrap() {
+ bytes.extend_from_slice(buffer.as_slice());
+ }
+ bytes
+ }
+
+ fn write_stream(schema: &Schema, batches: &[RecordBatch], options:
IpcWriteOptions) -> Vec<u8> {
+ let mut bytes = Vec::new();
+ let mut writer = StreamWriter::try_new_with_options(&mut bytes,
schema, options).unwrap();
+ for batch in batches {
+ writer.write(batch).unwrap();
+ }
+ writer.finish().unwrap();
+ bytes
+ }
+
+ // StreamEncoder and StreamWriter currently use separate encoding paths,
so these tests
+ // verify the new sans-IO API preserves the existing IPC stream byte
layout.
+ #[test]
Review Comment:
Not strictly necessary, but since the [originating
issue](https://github.com/apache/arrow-rs/issues/7812) for this PR wanted to
use StreamWriter<W> in an async context, could you add a test that mirrors the
flow of an async writer (tokio is already a dev-dependency here) and then
assert the final bytes against a sync StreamWriter?
##########
arrow-ipc/src/writer.rs:
##########
@@ -1563,6 +1624,139 @@ impl<W: Write> RecordBatchWriter for FileWriter<W> {
}
}
+/// Arrow IPC stream encoder.
+///
+/// Encodes Arrow [`RecordBatch`]es to byte buffers using the [IPC Streaming
Format],
+/// without performing any IO.
+///
+/// The returned [`Buffer`]s are ordered and should be written to the
destination
+/// stream in order. Uncompressed record batch body buffers can share the
original
+/// Arrow buffers instead of being copied into an intermediate contiguous
buffer.
+///
+/// # Example
+/// ```
+/// # use arrow_array::record_batch;
+/// # use arrow_ipc::writer::StreamEncoder;
+/// # use arrow_schema::ArrowError;
+/// # fn main() -> Result<(), ArrowError> {
+/// let batch = record_batch!(("a", Int32, [1, 2, 3]))?;
+///
+/// let mut encoder = StreamEncoder::try_new(&batch.schema())?;
+/// let mut stream = vec![];
+/// for buffer in encoder.encode(&batch)? {
+/// stream.extend_from_slice(buffer.as_slice());
+/// }
+/// for buffer in encoder.finish()? {
+/// stream.extend_from_slice(buffer.as_slice());
+/// }
+/// # Ok(())
+/// # }
+/// ```
+///
+/// [IPC Streaming Format]:
https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
Review Comment:
we can remove the comment/link to the IPC format
--
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]
