[IGNORE] pass to everywhere that use arrow ipc reader the option to share or not share buffers

Cargo.toml

@@ -89,30 +89,51 @@ version = "54.0.0"
 #
 # See for more details: https://github.com/rust-lang/cargo/issues/11329
 apache-avro = { version = "0.21", default-features = false }
-arrow = { version = "59.1.0", features = [
+arrow = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+#    version = "59.1.0",
+    features = [
     "prettyprint",
     "chrono-tz",
 ] }
-arrow-avro = { version = "59.1.0", default-features = false, features = [
+arrow-avro = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+#    version = "59.1.0",
+    default-features = false, features = [
     "deflate",
     "snappy",
     "zstd",
     "bzip2",
     "xz",
 ] }
-arrow-buffer = { version = "59.1.0", default-features = false }
-arrow-data = { version = "59.1.0", default-features = false }
-arrow-flight = { version = "59.1.0", features = [
+arrow-buffer = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+#    version = "59.1.0",
+
+    default-features = false }
+arrow-data = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+#    version = "59.1.0",
+    default-features = false }
+arrow-flight = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+
+#    version = "59.1.0",
+    features = [
     "flight-sql-experimental",
 ] }
 # Both codecs are required here to make sure that code paths like
 # file-spilling have access to all compression codecs.
-arrow-ipc = { version = "59.1.0", default-features = false, features = [
+arrow-ipc = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+
+#    version = "59.1.0",
+    default-features = false, features = [
     "lz4",
     "zstd",
 ] }
-arrow-ord = { version = "59.1.0", default-features = false }
-arrow-schema = { version = "59.1.0", default-features = false }
+arrow-ord = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+
+#    version = "59.1.0",
+    default-features = false }
+arrow-schema = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+
+#    version = "59.1.0",
+    default-features = false }
 async-trait = "0.1.89"
 bigdecimal = "0.4.8"
 bytes = "1.11"
@@ -178,7 +199,10 @@ memchr = "2.8.1"
 num-traits = { version = "0.2" }
 object_store = { version = "0.13.2", default-features = false }
 parking_lot = "0.12"
-parquet = { version = "59.1.0", default-features = false, features = [
+parquet = { git = "https://github.com/rluvaton/arrow-rs", rev = "9075b4bc94f9a0ba3caf0befb7c93c897f4260ba",
+
+#    version = "59.1.0",
+    default-features = false, features = [
     "arrow",
     "async",
     "object_store",

datafusion/common/src/config.rs

@@ -34,9 +34,11 @@ use std::collections::{BTreeMap, HashMap};
 use std::error::Error;
 use std::fmt::{self, Display};
 use std::num::NonZeroUsize;
+use std::ops::Deref;
 use std::str::FromStr;
 #[cfg(feature = "parquet_encryption")]
 use std::sync::Arc;
+use arrow_ipc::reader::BufferAllocationStrategy;
 
 /// A macro that wraps a configuration struct and automatically derives
 /// [`Default`] and [`ConfigField`] for it, allowing it to be used
@@ -583,6 +585,72 @@ impl Display for SpillCompression {
     }
 }
 
+#[derive(Clone, Copy, PartialEq, Eq, Default)]
+pub struct BufferAllocationStrategyWrapper(BufferAllocationStrategy);
+
+impl Deref for BufferAllocationStrategyWrapper {
+    type Target = BufferAllocationStrategy;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl From<BufferAllocationStrategy> for BufferAllocationStrategyWrapper {
+    fn from(value: BufferAllocationStrategy) -> Self {
+        Self(value)
+    }
+}
+
+impl Into<BufferAllocationStrategy> for BufferAllocationStrategyWrapper {
+    fn into(self) -> BufferAllocationStrategy {
+        self.0
+    }
+}
+
+impl fmt::Debug for BufferAllocationStrategyWrapper {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl FromStr for BufferAllocationStrategyWrapper {
+    type Err = DataFusionError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_ascii_lowercase().as_str() {
+            "shared" | "" => Ok(Self(BufferAllocationStrategy::Shared)),
+            "owned_if_free" | "owned-if-free" => Ok(Self(BufferAllocationStrategy::OwnedIfFree)),
+            "owned" => Ok(Self(BufferAllocationStrategy::Owned)),
+            other => Err(DataFusionError::Configuration(format!(
+                "Invalid Arrow IPC buffer allocation strategy type: {other}. Expected one of: shared, owned_if_free, owned"
+            ))),
+        }
+    }
+}
+
+impl ConfigField for BufferAllocationStrategyWrapper {
+    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+        v.some(key, self, description)
+    }
+
+    fn set(&mut self, _: &str, value: &str) -> Result<()> {
+        *self = BufferAllocationStrategyWrapper::from_str(value)?;
+        Ok(())
+    }
+}
+
+impl Display for BufferAllocationStrategyWrapper {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let str = match self.0 {
+            BufferAllocationStrategy::Shared => "shared",
+            BufferAllocationStrategy::OwnedIfFree => "owned_if_free",
+            BufferAllocationStrategy::Owned => "owned",
+        };
+        write!(f, "{str}")
+    }
+}
+
 /// A `usize` configuration value that rejects zero when set from strings.
 ///
 /// Use this for options where zero is never a meaningful runtime value.
@@ -814,6 +882,9 @@ config_namespace! {
         /// higher compression ratios at the cost of slower (de)compression speed.
         pub spill_compression: SpillCompression, default = SpillCompression::Uncompressed
 
+        /// In Arrow IPC reader whether to read back each buffer owned, or shared betwwen
+        pub ipc_buffer_allocation_strategy: BufferAllocationStrategyWrapper, default = BufferAllocationStrategyWrapper::default()
+
         /// Specifies the reserved memory for each spillable sort operation to
         /// facilitate an in-memory merge.
         ///

datafusion/datasource-arrow/src/file_format.rs

@@ -204,8 +204,8 @@ impl FileFormat for ArrowFormat {
         let mut source: Arc<dyn FileSource> =
             match is_object_in_arrow_ipc_file_format(object_store, object_location).await
             {
-                Ok(true) => Arc::new(ArrowSource::new_file_source(table_schema)),
-                Ok(false) => Arc::new(ArrowSource::new_stream_file_source(table_schema)),
+                Ok(true) => Arc::new(ArrowSource::new_file_source(table_schema).with_buffer_allocation_strategy(state.config().buffer_allocation_strategy())),
+                Ok(false) => Arc::new(ArrowSource::new_stream_file_source(table_schema).with_buffer_allocation_strategy(state.config().buffer_allocation_strategy())),
                 Err(e) => Err(e)?,
             };