diff --git a/Cargo.lock b/Cargo.lock
index 7f94aa24..4404eacc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1523,21 +1523,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "crc"
-version = "3.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5eb8a2a1cd12ab0d987a5d5e825195d372001a4094a0376319d5a0ad71c1ba0d"
-dependencies = [
- "crc-catalog",
-]
-
-[[package]]
-name = "crc-catalog"
-version = "2.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5"
-
 [[package]]
 name = "crc32fast"
 version = "1.5.0"
@@ -3906,16 +3891,6 @@ dependencies = [
  "twox-hash",
 ]
 
-[[package]]
-name = "lzma-rust2"
-version = "0.13.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c60a23ffb90d527e23192f1246b14746e2f7f071cb84476dd879071696c18a4a"
-dependencies = [
- "crc",
- "sha2",
-]
-
 [[package]]
 name = "mac_address"
 version = "1.1.8"
@@ -6917,9 +6892,9 @@ dependencies = [
 
 [[package]]
 name = "zip"
-version = "6.0.0"
+version = "4.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eb2a05c7c36fde6c09b08576c9f7fb4cda705990f73b58fe011abf7dfb24168b"
+checksum = "caa8cd6af31c3b31c6631b8f483848b91589021b28fffe50adada48d4f4d2ed1"
 dependencies = [
  "aes",
  "arbitrary",
@@ -6931,7 +6906,7 @@ dependencies = [
  "getrandom 0.3.4",
  "hmac",
  "indexmap",
- "lzma-rust2",
+ "liblzma",
  "memchr",
  "pbkdf2",
  "ppmd-rust",
diff --git a/Cargo.toml b/Cargo.toml
index 70f46e58..b0866063 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -51,7 +51,7 @@ arrow = { version = "57.1.0", optional = true }
 hyper-util = { version = "0.1.16", optional = true }
 pretty_assertions = { version = "1.4", optional = true }
 reqwest = { version = "0.12", optional = true }
-zip = { version = "6.0", optional = true }
+zip = { version = "4.0", optional = true }
 tonic-prost = "0.14.2"
 
 [features]
@@ -83,5 +83,5 @@ tokio-stream = "0.1.17"
 hyper-util = "0.1.16"
 pretty_assertions = "1.4"
 reqwest = "0.12"
-zip = "6.0"
+zip = "4.0"
 test-case = "3.3.1"
diff --git a/src/distributed_ext.rs b/src/distributed_ext.rs
index 750c87dc..e06a85e5 100644
--- a/src/distributed_ext.rs
+++ b/src/distributed_ext.rs
@@ -469,6 +469,24 @@ pub trait DistributedExt: Sized {
         &mut self,
         compression: Option<CompressionType>,
     ) -> Result<(), DataFusionError>;
+
+    /// How many rows to collect in each record batch before sending it over the wire in a
+    /// shuffle operation. This value defaults to the same as `datafusion.execution.batch_size`.
+    ///
+    /// Setting it to something smaller than `datafusion.execution.batch_size` has no effect.
+    ///
+    /// It's preferable to set `datafusion.execution.batch_size` directly instead of this
+    /// parameter if the specific use case allows it.
+    fn with_distributed_shuffle_batch_size(
+        self,
+        batch_size: usize,
+    ) -> Result<Self, DataFusionError>;
+
+    /// Same as [DistributedExt::with_distributed_shuffle_batch_size] but with an in-place mutation.
+    fn set_distributed_shuffle_batch_size(
+        &mut self,
+        batch_size: usize,
+    ) -> Result<(), DataFusionError>;
 }
 
 impl DistributedExt for SessionConfig {
@@ -565,6 +583,15 @@ impl DistributedExt for SessionConfig {
         Ok(())
     }
 
+    fn set_distributed_shuffle_batch_size(
+        &mut self,
+        batch_size: usize,
+    ) -> Result<(), DataFusionError> {
+        let d_cfg = DistributedConfig::from_config_options_mut(self.options_mut())?;
+        d_cfg.shuffle_batch_size = batch_size;
+        Ok(())
+    }
+
     delegate! {
         to self {
             #[call(set_distributed_option_extension)]
@@ -618,6 +645,10 @@ impl DistributedExt for SessionConfig {
             #[call(set_distributed_compression)]
             #[expr($?;Ok(self))]
             fn with_distributed_compression(mut self, compression: Option<CompressionType>) -> Result<Self, DataFusionError>;
+
+            #[call(set_distributed_shuffle_batch_size)]
+            #[expr($?;Ok(self))]
+            fn with_distributed_shuffle_batch_size(mut self, batch_size: usize) -> Result<Self, DataFusionError>;
         }
     }
 }
@@ -689,6 +720,11 @@ impl DistributedExt for SessionStateBuilder {
             #[call(set_distributed_compression)]
             #[expr($?;Ok(self))]
             fn with_distributed_compression(mut self, compression: Option<CompressionType>) -> Result<Self, DataFusionError>;
+
+            fn set_distributed_shuffle_batch_size(&mut self, batch_size: usize) -> Result<(), DataFusionError>;
+            #[call(set_distributed_shuffle_batch_size)]
+            #[expr($?;Ok(self))]
+            fn with_distributed_shuffle_batch_size(mut self, batch_size: usize) -> Result<Self, DataFusionError>;
         }
     }
 }
@@ -760,6 +796,11 @@ impl DistributedExt for SessionState {
             #[call(set_distributed_compression)]
             #[expr($?;Ok(self))]
             fn with_distributed_compression(mut self, compression: Option<CompressionType>) -> Result<Self, DataFusionError>;
+
+            fn set_distributed_shuffle_batch_size(&mut self, batch_size: usize) -> Result<(), DataFusionError>;
+            #[call(set_distributed_shuffle_batch_size)]
+            #[expr($?;Ok(self))]
+            fn with_distributed_shuffle_batch_size(mut self, batch_size: usize) -> Result<Self, DataFusionError>;
         }
     }
 }
@@ -831,6 +872,11 @@ impl DistributedExt for SessionContext {
             #[call(set_distributed_compression)]
             #[expr($?;Ok(self))]
             fn with_distributed_compression(self, compression: Option<CompressionType>) -> Result<Self, DataFusionError>;
+
+            fn set_distributed_shuffle_batch_size(&mut self, batch_size: usize) -> Result<(), DataFusionError>;
+            #[call(set_distributed_shuffle_batch_size)]
+            #[expr($?;Ok(self))]
+            fn with_distributed_shuffle_batch_size(self, batch_size: usize) -> Result<Self, DataFusionError>;
         }
     }
 }