update logic for check or load stream

Author: nikhilsinhaparseable

src/parseable/mod.rs

@@ -212,13 +212,14 @@ impl Parseable {
 
     /// Checks for the stream in memory, or loads it from storage when in distributed mode
     pub async fn check_or_load_stream(&self, stream_name: &str) -> bool {
-        !self.streams.contains(stream_name)
-            && (self.options.mode != Mode::Query
-                || self.options.mode != Mode::Prism
-                || !self
-                    .create_stream_and_schema_from_storage(stream_name)
-                    .await
-                    .unwrap_or_default())
+        if self.streams.contains(stream_name) {
+            return true;
+        }
+        (self.options.mode == Mode::Query || self.options.mode == Mode::Prism)
+            && self
+                .create_stream_and_schema_from_storage(stream_name)
+                .await
+                .unwrap_or_default()
     }
 
     // validate the storage, if the proper path for staging directory is provided

src/prism/logstream/mod.rs

@@ -86,7 +86,7 @@ pub async fn get_prism_logstream_info(
 
 async fn get_stream_schema_helper(stream_name: &str) -> Result<Arc<Schema>, StreamError> {
     // Ensure parseable is aware of stream in distributed mode
-    if PARSEABLE.check_or_load_stream(stream_name).await {
+    if !PARSEABLE.check_or_load_stream(stream_name).await {
         return Err(StreamNotFound(stream_name.to_owned()).into());
     }
 
@@ -152,7 +152,7 @@ async fn get_stream_info_helper(stream_name: &str) -> Result<StreamInfo, StreamE
     // For query mode, if the stream not found in memory map,
     //check if it exists in the storage
     //create stream and schema from storage
-    if PARSEABLE.check_or_load_stream(stream_name).await {
+    if !PARSEABLE.check_or_load_stream(stream_name).await {
         return Err(StreamNotFound(stream_name.to_owned()).into());
     }
 
@@ -289,7 +289,7 @@ impl PrismDatasetRequest {
         }
 
         // Skip streams that don't exist
-        if !self.stream_exists(&stream).await {
+        if !PARSEABLE.check_or_load_stream(&stream).await {
             return Ok(None);
         }
 
@@ -311,15 +311,6 @@ impl PrismDatasetRequest {
         }
     }
 
-    async fn stream_exists(&self, stream: &str) -> bool {
-        if PARSEABLE.check_or_load_stream(stream).await {
-            warn!("Stream not found: {stream}");
-            false
-        } else {
-            true
-        }
-    }
-
     async fn build_dataset_response(
         &self,
         stream: String,