perf: remove two redundant object-store calls in datasets metadata fetch (#1760)

get_manifest_list fetched the base stream.json unconditionally, but on
Query/Prism-mode nodes that result was never read since those modes
merge every ingestor's stream.json instead. Move the fetch into the
branch that actually needs it (Mode::All).

get_first_and_latest_event_from_storage awaited its two independent
extract_timestamp_for_date calls sequentially. Run them concurrently
with try_join! instead.

Both are part of the object-store I/O behind POST /api/prism/v1/datasets,
which has been observed taking 50s+ on multi-TB, multi-day deployments.

Author: prabhaks

src/query/mod.rs

@@ -798,14 +798,6 @@ pub async fn get_manifest_list(
     time_range: &TimeRange,
     tenant_id: &Option<String>,
 ) -> Result<Vec<Manifest>, QueryError> {
-    // get object store
-    let object_store_format: ObjectStoreFormat = serde_json::from_slice(
-        &PARSEABLE
-            .metastore
-            .get_stream_json(stream_name, false, tenant_id, false)
-            .await?,
-    )?;
-
     // all the manifests will go here
     let mut merged_snapshot: Snapshot = Snapshot::default();
 
@@ -826,6 +818,14 @@ pub async fn get_manifest_list(
             }
         }
     } else {
+        // Only needed here: Query/Prism mode merges every ingestor's stream.json
+        // above instead and never reads the base file.
+        let object_store_format: ObjectStoreFormat = serde_json::from_slice(
+            &PARSEABLE
+                .metastore
+                .get_stream_json(stream_name, false, tenant_id, false)
+                .await?,
+        )?;
         merged_snapshot = object_store_format.snapshot;
     }
 

src/storage/object_storage.rs

@@ -931,13 +931,11 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         let min_date = &parsed_dates[0].1;
         let max_date = &parsed_dates[parsed_dates.len() - 1].1;
 
-        // Extract timestamps for min and max dates
-        let first_timestamp = self
-            .extract_timestamp_for_date(stream_name, min_date, true, tenant_id)
-            .await?;
-        let latest_timestamp = self
-            .extract_timestamp_for_date(stream_name, max_date, false, tenant_id)
-            .await?;
+        // Extract timestamps for min and max dates concurrently; independent reads, no shared state
+        let (first_timestamp, latest_timestamp) = tokio::try_join!(
+            self.extract_timestamp_for_date(stream_name, min_date, true, tenant_id),
+            self.extract_timestamp_for_date(stream_name, max_date, false, tenant_id)
+        )?;
 
         let first_event_at = first_timestamp.map(|ts| ts.to_rfc3339());
         let latest_event_at = latest_timestamp.map(|ts| ts.to_rfc3339());