dataset tags and labels in home api response

Author: nikhilsinhaparseable

src/prism/home/mod.rs

@@ -237,8 +237,16 @@ async fn get_stream_metadata(stream: String, tenant_id: &Option<String>) -> Stre
     let ingested = stream_jsons
         .iter()
         .any(|s| s.stats.current_stats.events > 0);
-    let dataset_tags = stream_jsons[0].dataset_tags.clone();
-    let dataset_labels = stream_jsons[0].dataset_labels.clone();
+    let dataset_tags = stream_jsons
+        .iter()
+        .flat_map(|s| s.dataset_tags.clone())
+        .unique_by(|t| *t)
+        .collect_vec();
+    let dataset_labels = stream_jsons
+        .iter()
+        .flat_map(|s| s.dataset_labels.clone())
+        .unique_by(|l| l.clone())
+        .collect_vec();
 
     Ok(StreamMetadata {
         stream,

src/storage/object_storage.rs

@@ -21,6 +21,7 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use chrono::{DateTime, Utc};
 use datafusion::{datasource::listing::ListingTableUrl, execution::runtime_env::RuntimeEnvBuilder};
+use itertools::Itertools;
 use object_store::ListResult;
 use object_store::ObjectMeta;
 use object_store::buffered::BufReader;
@@ -754,6 +755,18 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
                 stats: FullStats::default(),
                 snapshot: Snapshot::default(),
                 log_source: merged_log_sources,
+                dataset_tags: stream_metadata_obs
+                    .iter()
+                    .filter_map(|bytes| serde_json::from_slice::<ObjectStoreFormat>(bytes).ok())
+                    .flat_map(|meta| meta.dataset_tags)
+                    .unique()
+                    .collect(),
+                dataset_labels: stream_metadata_obs
+                    .iter()
+                    .filter_map(|bytes| serde_json::from_slice::<ObjectStoreFormat>(bytes).ok())
+                    .flat_map(|meta| meta.dataset_labels)
+                    .unique()
+                    .collect(),
                 ..stream_ob_metadata
             };