viirya commented on code in PR #2369:
URL: https://github.com/apache/arrow-rs/pull/2369#discussion_r945109641
##########
arrow/src/ipc/writer.rs:
##########
@@ -1096,29 +1210,56 @@ fn write_array_data(
offset,
data_ref.len(),
data_ref.null_count(),
+ compression_codec,
write_options,
- );
- });
+ )?;
+ }
}
- offset
+ Ok(offset)
}
-/// Write a buffer to a vector of bytes, and add its ipc::Buffer to a vector
+/// Write a buffer into `arrow_data`, a vector of bytes, and adds its
+/// [`ipc::Buffer`] to `buffers`. Returns the new offset in `arrow_data`
+///
+///
+/// From
<https://github.com/apache/arrow/blob/6a936c4ff5007045e86f65f1a6b6c3c955ad5103/format/Message.fbs#L58>
+/// Each constituent buffer is first compressed with the indicated
+/// compressor, and then written with the uncompressed length in the first 8
+/// bytes as a 64-bit little-endian signed integer followed by the compressed
+/// buffer bytes (and then padding as required by the protocol). The
+/// uncompressed length may be set to -1 to indicate that the data that
+/// follows is not compressed, which can be useful for cases where
+/// compression does not yield appreciable savings.
fn write_buffer(
- buffer: &[u8],
- buffers: &mut Vec<ipc::Buffer>,
- arrow_data: &mut Vec<u8>,
- offset: i64,
-) -> i64 {
- let len = buffer.len();
- let pad_len = pad_to_8(len as u32);
- let total_len: i64 = (len + pad_len) as i64;
- // assert_eq!(len % 8, 0, "Buffer width not a multiple of 8 bytes");
- buffers.push(ipc::Buffer::new(offset, total_len));
- arrow_data.extend_from_slice(buffer);
- arrow_data.extend_from_slice(&vec![0u8; pad_len][..]);
- offset + total_len
+ buffer: &[u8], // input
+ buffers: &mut Vec<ipc::Buffer>, // output buffer descriptors
+ arrow_data: &mut Vec<u8>, // output stream
+ offset: i64, // current output stream offset
+ compression_codec: &Option<CompressionCodec>,
+) -> Result<i64> {
+ let len: i64 = match compression_codec {
+ Some(compressor) => compressor.compress_to_vec(buffer, arrow_data)?,
+ None => {
+ arrow_data.extend_from_slice(buffer);
+ buffer.len()
+ }
+ }
+ .try_into()
+ .map_err(|e| {
+ ArrowError::InvalidArgumentError(format!(
+ "Could not convert compressed size to i64: {}",
+ e
+ ))
+ })?;
+
+ // make new indx entry
Review Comment:
```suggestion
// make new index entry
```
--
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]
