Don't create CompressionContext when no compression is selected [IPC]

[EmilyMatt]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

When attempting to run Miri on a program using arrow-ipc, there are always error on foreign calls, even when not using ZSTD, due to the CompressionContext being created all the time, even when compression is None.
I don't think we should be making unnecessary FFI calls unless we intend to use the encoder/decoder.
This is relevant both for the arrow-ipc crate, and the arrow-flight crate, but both are from the same underlying issue.

Describe the solution you'd like

CompressionContext should be an Option, or the zstd encoder/decoder within it should be an Option.
This Option should only be Some when compression is specified.

Describe alternatives you've considered

Tried disabling the feature, but it leads to a cascade of compilation failures so big it's not really worth trying to untangle right now.

Additional context

[mbutrovich]

We hit this in Apache DataFusion Comet when upgrading to arrow-ipc 58.1.0. Our Miri CI started failing on a test that only uses lz4 compression, because StreamWriter::try_new() unconditionally creates a CompressionContext::default() which eagerly calls ZSTD_createCCtx().

Call chain:

StreamWriter::try_new()
  -> CompressionContext::default()
    -> zstd::bulk::Compressor::new()
      -> ZSTD_createCCtx()  // FFI, fails under Miri

The root cause is in arrow-ipc/src/compression.rs:

impl Default for CompressionContext {
    fn default() -> Self {
        CompressionContext {
            #[cfg(feature = "zstd")]
            compressor: zstd::bulk::Compressor::new(zstd::DEFAULT_COMPRESSION_LEVEL)
                .expect("can use default compression level"),
        }
    }
}

Same issue exists for DecompressionContext.

Proposed fix

Wrap the compressor/decompressor in Option and lazy-init on first use:

pub struct CompressionContext {
    #[cfg(feature = "zstd")]
    compressor: Option<zstd::bulk::Compressor<'static>>,
}

impl Default for CompressionContext {
    fn default() -> Self {
        CompressionContext {
            #[cfg(feature = "zstd")]
            compressor: None,
        }
    }
}

impl CompressionContext {
    #[cfg(feature = "zstd")]
    fn zstd_compressor(&mut self) -> &mut zstd::bulk::Compressor<'static> {
        self.compressor.get_or_insert_with(|| {
            zstd::bulk::Compressor::new(zstd::DEFAULT_COMPRESSION_LEVEL)
                .expect("can use default compression level")
        })
    }
}

Then in compress_zstd, use ctx.zstd_compressor() instead of ctx.compressor. Same pattern for DecompressionContext.

This preserves the context reuse optimization from #8405/#9400 while avoiding the unnecessary FFI call when zstd isn't the selected codec.

CC @albertlockett @Dandandan

[albertlockett]

sounds good @mbutrovich ! Thanks for the detailed investigation

[alamb]

label_issue.py automatically added labels {'arrow'} from #9808

[alamb]

label_issue.py automatically added labels {'performance'} from #9808