From c7a86fbd941c6f6ddccf9030434bed4050c30b74 Mon Sep 17 00:00:00 2001 From: prabhaks Date: Tue, 18 Aug 2026 20:50:54 -0700 Subject: [PATCH] perf: cut two redundant object-store round trips in stream metadata fetch 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. --- src/query/mod.rs | 16 ++++++++-------- src/storage/object_storage.rs | 12 +++++------- 2 files changed, 13 insertions(+), 15 deletions(-) diff --git a/src/query/mod.rs b/src/query/mod.rs index 728e3cdb6..9d1765ef6 100644 --- a/src/query/mod.rs +++ b/src/query/mod.rs @@ -763,14 +763,6 @@ pub async fn get_manifest_list( time_range: &TimeRange, tenant_id: &Option, ) -> Result, 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(); @@ -791,6 +783,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; } diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs index 38b2809be..42d9e120a 100644 --- a/src/storage/object_storage.rs +++ b/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());