From 1d3a99627253ba61f9c5d06535c6657960cd1bbd Mon Sep 17 00:00:00 2001
From: DeviousCardi <115358213+DeviousCardi@users.noreply.github.com>
Date: Tue, 8 Sep 2026 09:59:30 +0530
Subject: [PATCH] reduce the multipart part size on the opendal backends

OpenDAL buffers a whole part before flushing it, so the target part size is also
the peak memory a single upload holds. At the shared default of 5 GiB a large
split is sent as one part and resides in memory in full.

The S3 implementation streams each part straight from the payload, so a larger
part there only buys fewer billed PUT requests and costs no memory. Lower the
part size for the OpenDAL backends alone rather than the shared default, which
would raise S3 request costs for no benefit.
---
 .../src/opendal_storage/base.rs               | 67 ++++++++++++++++++-
 1 file changed, 65 insertions(+), 2 deletions(-)

diff --git a/quickwit/quickwit-storage/src/opendal_storage/base.rs b/quickwit/quickwit-storage/src/opendal_storage/base.rs
index 731da433747..5be01ffa800 100644
--- a/quickwit/quickwit-storage/src/opendal_storage/base.rs
+++ b/quickwit/quickwit-storage/src/opendal_storage/base.rs
@@ -19,6 +19,7 @@ use std::time::Duration;
 use std::{fmt, io};
 
 use async_trait::async_trait;
+use bytesize::ByteSize;
 use futures::AsyncWriteExt as FuturesAsyncWriteExt;
 use opendal::layers::{ConcurrentLimitLayer, RetryLayer};
 use opendal::{DeleteInput, IntoDeleteInput, Operator};
@@ -37,6 +38,23 @@ use crate::{
     StorageErrorKind, StorageResolverError, StorageResult,
 };
 
+/// Upload limits for the OpenDAL backends.
+///
+/// These match [`MultiPartPolicy::default`] except for the target part size. OpenDAL
+/// buffers a whole part before flushing it, so the part size is also the peak amount of
+/// memory an upload holds. The S3 implementation streams each part straight from the
+/// payload instead, which is why it can afford the larger default: there, a bigger part
+/// only means fewer billed PUT requests, at no cost in memory.
+///
+/// A gibibyte still keeps the request count low while bounding what a single upload can
+/// pin in RAM.
+fn opendal_multipart_policy() -> MultiPartPolicy {
+    MultiPartPolicy {
+        target_part_num_bytes: ByteSize::gib(1),
+        ..MultiPartPolicy::default()
+    }
+}
+
 /// OpenDAL based storage implementation.
 /// # TODO
 ///
@@ -74,8 +92,7 @@ impl OpendalStorage {
         Self {
             uri,
             op,
-            // limits are the same as on S3
-            multipart_policy: MultiPartPolicy::default(),
+            multipart_policy: opendal_multipart_policy(),
         }
     }
 
@@ -342,6 +359,52 @@ mod tests {
 
     use super::*;
 
+    /// A part is buffered whole before it is flushed, so the part size chosen for an
+    /// upload is what that upload can pin in memory.
+    #[test]
+    fn opendal_policy_bounds_the_part_size() {
+        let split_len = ByteSize::gib(4).as_u64();
+
+        // Under the shared default a split this size is sent as one part, so the whole
+        // thing is held at once.
+        assert_eq!(
+            MultiPartPolicy::default().part_num_bytes(split_len),
+            ByteSize::gib(5).as_u64()
+        );
+        assert_eq!(
+            opendal_multipart_policy().part_num_bytes(split_len),
+            ByteSize::gib(1).as_u64()
+        );
+    }
+
+    /// Only the part size differs. The rest still describes what the backend accepts.
+    #[test]
+    fn opendal_policy_keeps_the_other_limits() {
+        let policy = opendal_multipart_policy();
+        let default = MultiPartPolicy::default();
+
+        assert_eq!(
+            policy.multipart_threshold_num_bytes,
+            default.multipart_threshold_num_bytes
+        );
+        assert_eq!(policy.max_num_parts, default.max_num_parts);
+        assert_eq!(policy.max_object_num_bytes, default.max_object_num_bytes);
+        assert_eq!(
+            policy.max_concurrent_uploads(),
+            default.max_concurrent_uploads()
+        );
+    }
+
+    /// Splits below the multipart threshold are unaffected.
+    #[test]
+    fn opendal_policy_leaves_small_uploads_single_part() {
+        let small_len = ByteSize::mib(64).as_u64();
+        assert_eq!(
+            opendal_multipart_policy().part_num_bytes(small_len),
+            small_len
+        );
+    }
+
     /// `AsyncRead` that returns a predefined sequence of chunks, one chunk per
     /// `poll_read` call. This lets us simulate a source that returns small
     /// payloads at each read.