minor migration update

if mode is Prism, it will read all stream jsons once

Author: parmesant

src/enterprise/utils.rs

@@ -99,7 +99,7 @@ pub async fn fetch_parquet_file_paths(
 
     let obs = PARSEABLE
         .metastore
-        .get_all_stream_jsons(stream, None, tenant_id)
+        .get_all_stream_jsons(stream, None, tenant_id, false)
         .await;
     if let Ok(obs) = obs {
         for ob in obs {

src/handlers/http/cluster/mod.rs

@@ -872,7 +872,7 @@ pub async fn fetch_stats_from_ingestors(
 ) -> Result<Vec<utils::QueriedStats>, StreamError> {
     let obs = PARSEABLE
         .metastore
-        .get_all_stream_jsons(stream_name, Some(Mode::Ingest), tenant_id)
+        .get_all_stream_jsons(stream_name, Some(Mode::Ingest), tenant_id, false)
         .await?;
 
     let mut ingestion_size = 0u64;

src/handlers/http/modal/query/querier_logstream.rs

@@ -197,7 +197,7 @@ pub async fn get_stats(
         if !date_value.is_empty() {
             let obs = PARSEABLE
                 .metastore
-                .get_all_stream_jsons(&stream_name, None, &tenant_id)
+                .get_all_stream_jsons(&stream_name, None, &tenant_id, false)
                 .await?;
 
             let mut stream_jsons = Vec::new();

src/lib.rs

@@ -32,7 +32,7 @@ pub mod handlers;
 pub mod hottier;
 pub mod interactive;
 mod livetail;
-mod metadata;
+pub mod metadata;
 pub mod metastore;
 pub mod metrics;
 pub mod migration;

src/metastore/metastore_traits.rs

@@ -247,6 +247,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         stream_name: &str,
         mode: Option<Mode>,
         tenant_id: &Option<String>,
+        is_migration: bool
     ) -> Result<Vec<Bytes>, MetastoreError>;
 
     /// Fetch manifests only for the explicitly requested date keys

src/metastore/metastores/object_store_metastore.rs

@@ -892,6 +892,7 @@ impl Metastore for ObjectStoreMetastore {
         stream_name: &str,
         mode: Option<Mode>,
         tenant_id: &Option<String>,
+        _is_migration: bool
     ) -> Result<Vec<Bytes>, MetastoreError> {
         let root = tenant_id.as_deref().unwrap_or("");
         let path = RelativePathBuf::from_iter([root, stream_name, STREAM_ROOT_DIRECTORY]);

src/migration/mod.rs

@@ -272,6 +272,14 @@ async fn migration_stream(
     let schema = storage
         .create_schema_from_metastore(stream, tenant_id)
         .await?;
+
+    if PARSEABLE.options.mode == Mode::Prism {
+        // read stream jsons once
+        PARSEABLE
+            .metastore
+            .get_all_stream_jsons(stream, None, tenant_id, true)
+            .await?;
+    }
     let stream_metadata = fetch_or_create_stream_metadata(stream, storage, tenant_id).await?;
 
     let mut stream_meta_found = true;
@@ -327,7 +335,7 @@ async fn fetch_or_create_stream_metadata(
     }
 }
 
-async fn migrate_stream_metadata(
+pub async fn migrate_stream_metadata(
     mut stream_metadata_value: Value,
     stream: &str,
     schema: &Bytes,
@@ -432,7 +440,7 @@ async fn migrate_stream_metadata(
     Ok(stream_metadata_value)
 }
 
-async fn setup_logstream_metadata(
+pub async fn setup_logstream_metadata(
     stream: &str,
     arrow_schema: &mut Schema,
     stream_metadata_value: Value,

src/prism/home/mod.rs

@@ -212,7 +212,7 @@ pub async fn generate_home_response(
 async fn get_stream_metadata(stream: String, tenant_id: &Option<String>) -> StreamMetadataResponse {
     let obs = PARSEABLE
         .metastore
-        .get_all_stream_jsons(&stream, None, tenant_id)
+        .get_all_stream_jsons(&stream, None, tenant_id, false)
         .await?;
     let mut stream_jsons = Vec::new();
     for ob in obs {

src/query/mod.rs

@@ -769,7 +769,7 @@ pub async fn get_manifest_list(
     if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism {
         let obs = PARSEABLE
             .metastore
-            .get_all_stream_jsons(stream_name, None, tenant_id)
+            .get_all_stream_jsons(stream_name, None, tenant_id, false)
             .await;
         if let Ok(obs) = obs {
             for ob in obs {

src/query/stream_schema_provider.rs

@@ -566,7 +566,7 @@ impl TableProvider for StandardTableProvider {
         if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism {
             let obs = PARSEABLE
                 .metastore
-                .get_all_stream_jsons(&self.stream, None, &self.tenant_id)
+                .get_all_stream_jsons(&self.stream, None, &self.tenant_id, false)
                 .await;
             if let Ok(obs) = obs {
                 for ob in obs {