persist: fetch source parts concurrently

src/persist-client/src/cfg.rs

@@ -312,6 +312,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&crate::cfg::USE_CRITICAL_SINCE_CATALOG)
         .add(&crate::cfg::USE_CRITICAL_SINCE_SOURCE)
         .add(&crate::cfg::USE_CRITICAL_SINCE_SNAPSHOT)
+        .add(&crate::cfg::SOURCE_FETCH_CONCURRENCY)
         .add(&BATCH_BUILDER_MAX_OUTSTANDING_PARTS)
         .add(&COMPACTION_HEURISTIC_MIN_INPUTS)
         .add(&COMPACTION_HEURISTIC_MIN_PARTS)
@@ -489,6 +490,17 @@ pub const USE_CRITICAL_SINCE_SOURCE: Config<bool> = Config::new(
     "Use the critical since (instead of the overall since) in the Persist source.",
 );
 
+/// Maximum number of part fetches the persist source issues concurrently, per
+/// worker. Concurrent fetches amortize the blob-store round-trip, which
+/// dominates when there are many small parts (e.g. a fine-grained hydration
+/// replay). `1` keeps the previous serial behavior.
+pub const SOURCE_FETCH_CONCURRENCY: Config<usize> = Config::new(
+    "persist_source_fetch_concurrency",
+    1,
+    "Maximum number of part fetches the persist source issues concurrently per worker \
+     (1 = serial).",
+);
+
 /// Migrate snapshots to use the critical since when opening a new read handle.
 pub const USE_CRITICAL_SINCE_SNAPSHOT: Config<bool> = Config::new(
     "persist_use_critical_since_snapshot",

src/persist-client/src/fetch.rs

@@ -155,6 +155,33 @@ where
     pub(crate) _phantom: PhantomData<fn() -> (K, V, T, D)>,
 }
 
+// Hand-written (rather than derived) so cloning does not require `K: Clone`
+// etc.: every field is an `Arc` or independently `Clone`. The `schema_cache`
+// clone shares the schema-lookup maps and applier, so clones reuse cached
+// schema fetches and only duplicate a small per-clone migration memo. Used to
+// run several `fetch_leased_part` calls concurrently, each on its own clone.
+impl<K, V, T, D> Clone for BatchFetcher<K, V, T, D>
+where
+    T: Timestamp + Lattice + Codec64,
+    K: Debug + Codec,
+    V: Debug + Codec,
+    D: Monoid + Codec64 + Send + Sync,
+{
+    fn clone(&self) -> Self {
+        Self {
+            cfg: self.cfg.clone(),
+            blob: Arc::clone(&self.blob),
+            metrics: Arc::clone(&self.metrics),
+            shard_metrics: Arc::clone(&self.shard_metrics),
+            shard_id: self.shard_id.clone(),
+            read_schemas: self.read_schemas.clone(),
+            schema_cache: self.schema_cache.clone(),
+            is_transient: self.is_transient,
+            _phantom: PhantomData,
+        }
+    }
+}
+
 impl<K, V, T, D> BatchFetcher<K, V, T, D>
 where
     K: Debug + Codec,