This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 32bba5ad27 rename Compression struct (#10221)
32bba5ad27 is described below
commit 32bba5ad277e3f12dceec4e21e6dc6f81c90714e
Author: RIchard Baah <[email protected]>
AuthorDate: Wed Jul 1 17:38:17 2026 -0400
rename Compression struct (#10221)
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->
- smaller piece of https://github.com/apache/arrow-rs/pull/10137
- see comment
https://github.com/apache/arrow-rs/pull/10137#pullrequestreview-4581920672
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
Changed Compression context to IPCwritecontext as now its holding onto
the fbb and scratch space for buffers.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
- compression.rs: Renamed CompressionContext → IpcWriteContext. Added
scratch: Vec<u8> field (unused for now, #[allow(dead_code)]). Added
#[deprecated(since = "57.0.0", note = "Use IpcWriteContext instead")]
pub type CompressionContext = IpcWriteContext; for backward
compatibility.
- writer.rs: Updated all internal type references from
CompressionContext -> IpcWriteContext. Renamed all
field/parameter/variable occurrences of compression_context ->
ipc_write_context. Both names are re-exported from writer so existing
code compiles with a deprecation warning rather than a hard error.
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
If this PR claims a performance improvement, please include evidence
such as benchmark results.
-->
n/a
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please call them out.
-->
Deprecate old name, but no breaking changes
---------
Co-authored-by: Andrew Lamb <[email protected]>
---
arrow-flight/src/encode.rs | 12 +-
arrow-flight/src/utils.rs | 6 +-
.../flight_client_scenarios/integration_test.rs | 12 +-
arrow-ipc/src/compression.rs | 22 ++--
arrow-ipc/src/writer.rs | 141 ++++++++++-----------
5 files changed, 96 insertions(+), 97 deletions(-)
diff --git a/arrow-flight/src/encode.rs b/arrow-flight/src/encode.rs
index 6c17669956..c88ab71ce4 100644
--- a/arrow-flight/src/encode.rs
+++ b/arrow-flight/src/encode.rs
@@ -20,7 +20,7 @@ use std::{collections::VecDeque, fmt::Debug, pin::Pin,
sync::Arc, task::Poll};
use crate::{FlightData, FlightDescriptor, SchemaAsIpc, error::Result};
use arrow_array::{Array, ArrayRef, RecordBatch, RecordBatchOptions,
UnionArray};
-use arrow_ipc::writer::{CompressionContext, DictionaryTracker,
IpcDataGenerator, IpcWriteOptions};
+use arrow_ipc::writer::{DictionaryTracker, IpcDataGenerator, IpcWriteContext,
IpcWriteOptions};
use arrow_schema::{DataType, Field, FieldRef, Fields, Schema, SchemaRef,
UnionMode};
use bytes::Bytes;
@@ -701,7 +701,7 @@ struct FlightIpcEncoder {
options: IpcWriteOptions,
data_gen: IpcDataGenerator,
dictionary_tracker: DictionaryTracker,
- compression_context: CompressionContext,
+ ipc_write_context: IpcWriteContext,
}
impl FlightIpcEncoder {
@@ -710,7 +710,7 @@ impl FlightIpcEncoder {
options,
data_gen: IpcDataGenerator::default(),
dictionary_tracker: DictionaryTracker::new(error_on_replacement),
- compression_context: CompressionContext::default(),
+ ipc_write_context: IpcWriteContext::default(),
}
}
@@ -729,7 +729,7 @@ impl FlightIpcEncoder {
batch,
&mut self.dictionary_tracker,
&self.options,
- &mut self.compression_context,
+ &mut self.ipc_write_context,
)?;
let flight_dictionaries = encoded_dictionaries.into_iter().map(|e|
e.into());
@@ -1833,14 +1833,14 @@ mod tests {
) -> (Vec<FlightData>, FlightData) {
let data_gen = IpcDataGenerator::default();
let mut dictionary_tracker = DictionaryTracker::new(false);
- let mut compression_context = CompressionContext::default();
+ let mut ipc_write_context = IpcWriteContext::default();
let (encoded_dictionaries, encoded_batch) = data_gen
.encode(
batch,
&mut dictionary_tracker,
options,
- &mut compression_context,
+ &mut ipc_write_context,
)
.expect("DictionaryTracker configured above to not error on
replacement");
diff --git a/arrow-flight/src/utils.rs b/arrow-flight/src/utils.rs
index 6effb5f86a..07a1801914 100644
--- a/arrow-flight/src/utils.rs
+++ b/arrow-flight/src/utils.rs
@@ -24,7 +24,7 @@ use std::sync::Arc;
use arrow_array::{ArrayRef, RecordBatch};
use arrow_buffer::Buffer;
use arrow_ipc::convert::fb_to_schema;
-use arrow_ipc::writer::CompressionContext;
+use arrow_ipc::writer::IpcWriteContext;
use arrow_ipc::{reader, root_as_message, writer, writer::IpcWriteOptions};
use arrow_schema::{ArrowError, Schema, SchemaRef};
@@ -92,14 +92,14 @@ pub fn batches_to_flight_data(
let data_gen = writer::IpcDataGenerator::default();
let mut dictionary_tracker = writer::DictionaryTracker::new(false);
- let mut compression_context = CompressionContext::default();
+ let mut ipc_write_context = IpcWriteContext::default();
for batch in batches.iter() {
let (encoded_dictionaries, encoded_batch) = data_gen.encode(
batch,
&mut dictionary_tracker,
&options,
- &mut compression_context,
+ &mut ipc_write_context,
)?;
dictionaries.extend(encoded_dictionaries.into_iter().map(Into::into));
diff --git
a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
index 05ca5627ec..950a64562e 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -26,7 +26,7 @@ use arrow::{
datatypes::SchemaRef,
ipc::{
self, reader,
- writer::{self, CompressionContext},
+ writer::{self, IpcWriteContext},
},
record_batch::RecordBatch,
};
@@ -95,7 +95,7 @@ async fn upload_data(
let mut original_data_iter = original_data.iter().enumerate();
- let mut compression_context = CompressionContext::default();
+ let mut ipc_write_context = IpcWriteContext::default();
if let Some((counter, first_batch)) = original_data_iter.next() {
let metadata = counter.to_string().into_bytes();
@@ -106,7 +106,7 @@ async fn upload_data(
first_batch,
&options,
&mut dict_tracker,
- &mut compression_context,
+ &mut ipc_write_context,
)
.await?;
@@ -129,7 +129,7 @@ async fn upload_data(
batch,
&options,
&mut dict_tracker,
- &mut compression_context,
+ &mut ipc_write_context,
)
.await?;
@@ -159,12 +159,12 @@ async fn send_batch(
batch: &RecordBatch,
options: &writer::IpcWriteOptions,
dictionary_tracker: &mut writer::DictionaryTracker,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
) -> Result {
let data_gen = writer::IpcDataGenerator::default();
let (encoded_dictionaries, encoded_batch) = data_gen
- .encode(batch, dictionary_tracker, options, compression_context)
+ .encode(batch, dictionary_tracker, options, ipc_write_context)
.expect("DictionaryTracker configured above to not error on
replacement");
let dictionary_flight_data: Vec<FlightData> =
diff --git a/arrow-ipc/src/compression.rs b/arrow-ipc/src/compression.rs
index fd8469c2ba..79879332d4 100644
--- a/arrow-ipc/src/compression.rs
+++ b/arrow-ipc/src/compression.rs
@@ -30,13 +30,15 @@ const DEFAULT_ZSTD_COMPRESSION_LEVEL: i32 = 3;
/// compression calls to avoid the performance overhead of initialising a new
context for every
/// compression. Also holds a [`FlatBufferBuilder`] that is reused across IPC
writes.
#[derive(Default)]
-pub struct CompressionContext {
+pub struct IpcWriteContext {
+ #[expect(dead_code)]
+ pub(crate) scratch: Vec<u8>,
fbb: FlatBufferBuilder<'static>,
#[cfg(feature = "zstd")]
compressor: Option<zstd::bulk::Compressor<'static>>,
}
-impl CompressionContext {
+impl IpcWriteContext {
/// Get a mutable reference to the [`FlatBufferBuilder`] that is reused
across IPC writes.
pub(crate) fn mut_fbb(&mut self) -> &mut FlatBufferBuilder<'static> {
&mut self.fbb
@@ -50,9 +52,9 @@ impl CompressionContext {
}
}
-impl std::fmt::Debug for CompressionContext {
+impl std::fmt::Debug for IpcWriteContext {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- let mut ds = f.debug_struct("CompressionContext");
+ let mut ds = f.debug_struct("IpcWriteContext");
#[cfg(feature = "zstd")]
ds.field(
@@ -64,6 +66,10 @@ impl std::fmt::Debug for CompressionContext {
}
}
+/// Deprecated alias for [`IpcWriteContext`].
+#[deprecated(since = "59.1.0", note = "Use IpcWriteContext instead")]
+pub type CompressionContext = IpcWriteContext;
+
/// Additional context that may be needed for decompression.
///
/// In the case of zstd, this will contain the zstd decompression context,
which can be reused
@@ -169,7 +175,7 @@ impl CompressionCodec {
&self,
input: &[u8],
output: &mut Vec<u8>,
- context: &mut CompressionContext,
+ context: &mut IpcWriteContext,
) -> Result<usize, ArrowError> {
let uncompressed_data_len = input.len();
let original_output_len = output.len();
@@ -235,7 +241,7 @@ impl CompressionCodec {
&self,
input: &[u8],
output: &mut Vec<u8>,
- context: &mut CompressionContext,
+ context: &mut IpcWriteContext,
) -> Result<(), ArrowError> {
match self {
CompressionCodec::Lz4Frame => compress_lz4(input, output),
@@ -304,7 +310,7 @@ fn decompress_lz4(_input: &[u8], _decompressed_size: usize)
-> Result<Vec<u8>, A
fn compress_zstd(
input: &[u8],
output: &mut Vec<u8>,
- context: &mut CompressionContext,
+ context: &mut IpcWriteContext,
level: i32,
) -> Result<(), ArrowError> {
let result = context.zstd_compressor(level).compress(input)?;
@@ -317,7 +323,7 @@ fn compress_zstd(
fn compress_zstd(
_input: &[u8],
_output: &mut Vec<u8>,
- _context: &mut CompressionContext,
+ _context: &mut IpcWriteContext,
_level: i32,
) -> Result<(), ArrowError> {
Err(ArrowError::InvalidArgumentError(
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index b88abdec7e..618f8b2fc9 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -43,7 +43,8 @@ use arrow_schema::*;
use crate::CONTINUATION_MARKER;
use crate::compression::CompressionCodec;
-pub use crate::compression::CompressionContext;
+#[expect(deprecated)]
+pub use crate::compression::{CompressionContext, IpcWriteContext};
use crate::convert::IpcSchemaEncoder;
/// IPC write options used to control the behaviour of the [`IpcDataGenerator`]
@@ -306,7 +307,7 @@ impl Default for IpcWriteOptions {
/// # use std::sync::Arc;
/// # use arrow_array::UInt64Array;
/// # use arrow_array::RecordBatch;
-/// # use arrow_ipc::writer::{CompressionContext, DictionaryTracker,
IpcDataGenerator, IpcWriteOptions};
+/// # use arrow_ipc::writer::{IpcWriteContext, DictionaryTracker,
IpcDataGenerator, IpcWriteOptions};
///
/// // Create a record batch
/// let batch = RecordBatch::try_from_iter(vec![
@@ -318,13 +319,13 @@ impl Default for IpcWriteOptions {
/// let options = IpcWriteOptions::default();
/// let mut dictionary_tracker = DictionaryTracker::new(error_on_replacement);
///
-/// let mut compression_context = CompressionContext::default();
+/// let mut ipc_write_context = IpcWriteContext::default();
///
/// // encode the batch into zero or more encoded dictionaries
/// // and the data for the actual array.
/// let data_gen = IpcDataGenerator::default();
/// let (encoded_dictionaries, encoded_message) = data_gen
-/// .encode(&batch, &mut dictionary_tracker, &options, &mut
compression_context)
+/// .encode(&batch, &mut dictionary_tracker, &options, &mut
ipc_write_context)
/// .unwrap();
/// # }
/// ```
@@ -372,7 +373,7 @@ impl IpcDataGenerator {
dictionary_tracker: &mut DictionaryTracker,
write_options: &IpcWriteOptions,
dict_id: &mut I,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
) -> Result<(), ArrowError> {
match column.data_type() {
DataType::Struct(fields) => {
@@ -385,7 +386,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
}
@@ -407,7 +408,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::List(field) => {
@@ -419,7 +420,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::LargeList(field) => {
@@ -431,7 +432,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::ListView(field) => {
@@ -443,7 +444,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::LargeListView(field) => {
@@ -455,7 +456,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::FixedSizeList(field, _) => {
@@ -470,7 +471,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::Map(field, _) => {
@@ -489,7 +490,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
// values
@@ -500,7 +501,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
DataType::Union(fields, _) => {
@@ -514,7 +515,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
}
@@ -533,7 +534,7 @@ impl IpcDataGenerator {
dictionary_tracker: &mut DictionaryTracker,
write_options: &IpcWriteOptions,
dict_id_seq: &mut I,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
) -> Result<(), ArrowError> {
match column.data_type() {
DataType::Dictionary(_key_type, _value_type) => {
@@ -548,7 +549,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id_seq,
- compression_context,
+ ipc_write_context,
)?;
// It's important to only take the dict_id at this point,
because the dict ID
@@ -575,7 +576,7 @@ impl IpcDataGenerator {
dict_values,
write_options,
false,
- compression_context,
+ ipc_write_context,
)?);
}
DictionaryUpdate::Delta(data) => {
@@ -584,7 +585,7 @@ impl IpcDataGenerator {
&data,
write_options,
true,
- compression_context,
+ ipc_write_context,
)?);
}
}
@@ -595,7 +596,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
dict_id_seq,
- compression_context,
+ ipc_write_context,
)?,
}
@@ -610,19 +611,15 @@ impl IpcDataGenerator {
batch: &RecordBatch,
dictionary_tracker: &mut DictionaryTracker,
write_options: &IpcWriteOptions,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
) -> Result<(Vec<EncodedData>, EncodedData), ArrowError> {
- let encoded_dictionaries = self.encode_all_dicts(
- batch,
- dictionary_tracker,
- write_options,
- compression_context,
- )?;
+ let encoded_dictionaries =
+ self.encode_all_dicts(batch, dictionary_tracker, write_options,
ipc_write_context)?;
let mut arrow_data = Vec::new();
let (ipc_message, _, tail_pad) = self.record_batch_to_bytes(
batch,
write_options,
- compression_context,
+ ipc_write_context,
&mut IpcBodySink::Write(&mut arrow_data),
)?;
arrow_data.extend_from_slice(&PADDING[..tail_pad]);
@@ -641,7 +638,7 @@ impl IpcDataGenerator {
batch: &RecordBatch,
dictionary_tracker: &mut DictionaryTracker,
write_options: &IpcWriteOptions,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
) -> Result<Vec<EncodedData>, ArrowError> {
let schema = batch.schema();
let mut encoded_dictionaries =
Vec::with_capacity(schema.flattened_fields().len());
@@ -654,7 +651,7 @@ impl IpcDataGenerator {
dictionary_tracker,
write_options,
&mut dict_id,
- compression_context,
+ ipc_write_context,
)?;
}
Ok(encoded_dictionaries)
@@ -668,15 +665,11 @@ impl IpcDataGenerator {
batch: &RecordBatch,
dictionary_tracker: &mut DictionaryTracker,
write_options: &IpcWriteOptions,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
writer: &mut W,
) -> Result<IpcWriteMetadata, ArrowError> {
- let encoded_dictionaries = self.encode_all_dicts(
- batch,
- dictionary_tracker,
- write_options,
- compression_context,
- )?;
+ let encoded_dictionaries =
+ self.encode_all_dicts(batch, dictionary_tracker, write_options,
ipc_write_context)?;
let mut dictionary_block_sizes =
Vec::with_capacity(encoded_dictionaries.len());
for dict in encoded_dictionaries {
@@ -692,7 +685,7 @@ impl IpcDataGenerator {
let (ipc_message, body_len, tail_pad) = self.record_batch_to_bytes(
batch,
write_options,
- compression_context,
+ ipc_write_context,
&mut IpcBodySink::Collect(&mut encoded_buffers),
)?;
@@ -751,13 +744,13 @@ impl IpcDataGenerator {
&self,
batch: &RecordBatch,
write_options: &IpcWriteOptions,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
sink: &mut IpcBodySink<'_>,
) -> Result<(Vec<u8>, usize, usize), ArrowError> {
let batch_compression_type = write_options.batch_compression_type;
let compression = batch_compression_type.map(|batch_compression_type| {
- let fbb = compression_context.mut_fbb();
+ let fbb = ipc_write_context.mut_fbb();
let mut c = crate::BodyCompressionBuilder::new(fbb);
c.add_method(crate::BodyCompressionMethod::BUFFER);
c.add_codec(batch_compression_type);
@@ -787,7 +780,7 @@ impl IpcDataGenerator {
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
append_variadic_buffer_counts(&mut variadic_buffer_counts,
&array_data);
@@ -796,7 +789,7 @@ impl IpcDataGenerator {
let tail_pad = pad_to_alignment(alignment, offset as usize);
let body_len = offset as usize + tail_pad;
- let fbb = compression_context.mut_fbb();
+ let fbb = ipc_write_context.mut_fbb();
let buffers = fbb.create_vector(&meta.buffers);
let nodes = fbb.create_vector(&meta.nodes);
let variadic_buffer = if variadic_buffer_counts.is_empty() {
@@ -839,7 +832,7 @@ impl IpcDataGenerator {
array_data: &ArrayData,
write_options: &IpcWriteOptions,
is_delta: bool,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
) -> Result<EncodedData, ArrowError> {
let mut arrow_data: Vec<u8> = vec![];
@@ -847,7 +840,7 @@ impl IpcDataGenerator {
let batch_compression_type = write_options.batch_compression_type;
let compression = batch_compression_type.map(|batch_compression_type| {
- let fbb = compression_context.mut_fbb();
+ let fbb = ipc_write_context.mut_fbb();
let mut c = crate::BodyCompressionBuilder::new(fbb);
c.add_method(crate::BodyCompressionMethod::BUFFER);
c.add_codec(batch_compression_type);
@@ -873,7 +866,7 @@ impl IpcDataGenerator {
&mut sink,
0,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
@@ -885,7 +878,7 @@ impl IpcDataGenerator {
let body_len = offset as usize + tail_pad;
arrow_data.extend_from_slice(&PADDING[..tail_pad]);
- let fbb = compression_context.mut_fbb();
+ let fbb = ipc_write_context.mut_fbb();
let buffers = fbb.create_vector(&meta.buffers);
let nodes = fbb.create_vector(&meta.nodes);
let variadic_buffer = if variadic_buffer_counts.is_empty() {
@@ -1314,7 +1307,7 @@ pub struct FileWriter<W> {
data_gen: IpcDataGenerator,
- compression_context: CompressionContext,
+ ipc_write_context: IpcWriteContext,
}
impl<W: Write> FileWriter<BufWriter<W>> {
@@ -1376,7 +1369,7 @@ impl<W: Write> FileWriter<W> {
dictionary_tracker,
custom_metadata: HashMap::new(),
data_gen,
- compression_context: CompressionContext::default(),
+ ipc_write_context: IpcWriteContext::default(),
})
}
@@ -1397,7 +1390,7 @@ impl<W: Write> FileWriter<W> {
batch,
&mut self.dictionary_tracker,
&self.write_options,
- &mut self.compression_context,
+ &mut self.ipc_write_context,
&mut self.writer,
)?;
@@ -1605,7 +1598,7 @@ pub struct StreamWriter<W> {
data_gen: IpcDataGenerator,
- compression_context: CompressionContext,
+ ipc_write_context: IpcWriteContext,
}
impl<W: Write> StreamWriter<BufWriter<W>> {
@@ -1656,7 +1649,7 @@ impl<W: Write> StreamWriter<W> {
finished: false,
dictionary_tracker,
data_gen,
- compression_context: CompressionContext::default(),
+ ipc_write_context: IpcWriteContext::default(),
})
}
@@ -1672,7 +1665,7 @@ impl<W: Write> StreamWriter<W> {
batch,
&mut self.dictionary_tracker,
&self.write_options,
- &mut self.compression_context,
+ &mut self.ipc_write_context,
&mut self.writer,
)?;
Ok(())
@@ -2033,7 +2026,7 @@ fn write_array_data(
sink: &mut IpcBodySink<'_>,
offset: i64,
compression_codec: Option<CompressionCodec>,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
write_options: &IpcWriteOptions,
) -> Result<i64, ArrowError> {
let mut offset = offset;
@@ -2067,7 +2060,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
}
@@ -2082,7 +2075,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
}
@@ -2100,7 +2093,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
@@ -2111,7 +2104,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
}
@@ -2124,7 +2117,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
}
@@ -2145,7 +2138,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
} else if matches!(data_type, DataType::Boolean) {
@@ -2161,7 +2154,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
} else if matches!(
@@ -2184,7 +2177,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
offset = write_array_data(
@@ -2193,7 +2186,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
return Ok(offset);
@@ -2216,7 +2209,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
offset = encode_sink_buffer(
@@ -2225,7 +2218,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
@@ -2235,7 +2228,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
return Ok(offset);
@@ -2253,7 +2246,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
return Ok(offset);
@@ -2265,7 +2258,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options.alignment,
)?;
}
@@ -2285,7 +2278,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
}
@@ -2300,7 +2293,7 @@ fn write_array_data(
sink,
offset,
compression_codec,
- compression_context,
+ ipc_write_context,
write_options,
)?;
}
@@ -2317,8 +2310,8 @@ fn write_array_data(
/// - `sink`: destination for the actual encoded bytes; either a contiguous
`Vec<u8>` for
/// in-memory writes, or a list of [`EncodedBuffer`] segments for deferred
zero-copy streaming.
/// - `offset`: running byte offset into the IPC message body, used to compute
the metadata entry.
-/// - `compression_codec` / `compression_context`: if `Some`, the buffer is
compressed before
-/// writing; `compression_context` provides reusable scratch space across
calls.
+/// - `compression_codec` / `ipc_write_context`: if `Some`, the buffer is
compressed before
+/// writing; `ipc_write_context` provides reusable scratch space across
calls.
/// - `alignment`: each buffer is padded to this many bytes so the next buffer
starts aligned.
///
/// Returns the updated `offset` (advanced by the encoded length plus any
alignment padding).
@@ -2328,7 +2321,7 @@ fn encode_sink_buffer(
sink: &mut IpcBodySink<'_>,
offset: i64,
compression_codec: Option<CompressionCodec>,
- compression_context: &mut CompressionContext,
+ ipc_write_context: &mut IpcWriteContext,
alignment: u8,
) -> Result<i64, ArrowError> {
let (encoded, len) = match compression_codec {
@@ -2339,7 +2332,7 @@ fn encode_sink_buffer(
Some(codec) => {
let mut scratch = Vec::new();
let written =
- codec.compress_to_vec(buffer.as_slice(), &mut scratch,
compression_context)?;
+ codec.compress_to_vec(buffer.as_slice(), &mut scratch,
ipc_write_context)?;
let len = i64::try_from(written)
.map_err(|e|
ArrowError::InvalidArgumentError(format!("{e}")))?;
(EncodedBuffer::Compressed(scratch), len)
@@ -4524,7 +4517,7 @@ mod tests {
let data_gen = IpcDataGenerator::default();
let mut dictionary_tracker = DictionaryTracker::new(false);
let writer_options = IpcWriteOptions::default();
- let mut compression_ctx = CompressionContext::default();
+ let mut compression_ctx = IpcWriteContext::default();
let schema = Arc::new(Schema::new(vec