include null, add query param in prism home api

Author: nikhilsinhaparseable

src/handlers/http/prism_home.rs

@@ -26,7 +26,7 @@ use crate::{
 };
 
 const HOME_SEARCH_QUERY_PARAM: &str = "key";
-
+const HOME_QUERY_PARAM: &str = "includeInternal";
 /// Fetches the data to populate Prism's home
 ///
 ///
@@ -36,8 +36,14 @@ const HOME_SEARCH_QUERY_PARAM: &str = "key";
 pub async fn home_api(req: HttpRequest) -> Result<impl Responder, PrismHomeError> {
     let key = extract_session_key_from_req(&req)
         .map_err(|err| PrismHomeError::Anyhow(anyhow::Error::msg(err.to_string())))?;
+    let query_map = web::Query::<HashMap<String, String>>::from_query(req.query_string())
+        .map_err(|_| PrismHomeError::InvalidQueryParameter(HOME_QUERY_PARAM.to_string()))?;
 
-    let res = generate_home_response(&key).await?;
+    let include_internal = query_map
+        .get(HOME_QUERY_PARAM)
+        .map_or(false, |v| v == "true");
+
+    let res = generate_home_response(&key, include_internal).await?;
 
     Ok(web::Json(res))
 }
@@ -52,11 +58,12 @@ pub async fn home_search(req: HttpRequest) -> Result<impl Responder, PrismHomeEr
         return Ok(web::Json(serde_json::json!({})));
     }
 
-    let query_value = query_map
+    let query_key = query_map
         .get(HOME_SEARCH_QUERY_PARAM)
         .ok_or_else(|| PrismHomeError::InvalidQueryParameter(HOME_SEARCH_QUERY_PARAM.to_string()))?
         .to_lowercase();
-    let res = generate_home_search_response(&key, &query_value).await?;
+
+    let res = generate_home_search_response(&key, &query_key).await?;
     let json_res = serde_json::to_value(res)
         .map_err(|err| PrismHomeError::Anyhow(anyhow::Error::msg(err.to_string())))?;
 

src/parseable/streams.rs

@@ -52,11 +52,8 @@ use crate::{
         format::{LogSource, LogSourceEntry},
         DEFAULT_TIMESTAMP_KEY,
     },
-    handlers::http::{
-        cluster::INTERNAL_STREAM_NAME, ingest::PostError,
-        modal::utils::ingest_utils::flatten_and_push_logs,
-    },
     handlers::http::modal::{ingest_server::INGESTOR_META, query_server::QUERIER_META},
+    handlers::http::{ingest::PostError, modal::utils::ingest_utils::flatten_and_push_logs},
     metadata::{LogStreamMetadata, SchemaVersion},
     metrics,
     option::Mode,
@@ -74,7 +71,7 @@ use super::{
     },
     LogStream, ARROW_FILE_EXTENSION,
 };
-
+const DATASET_STATS_STREAM_NAME: &str = "pstats";
 const MAX_CONCURRENT_FIELD_STATS: usize = 10;
 
 #[derive(Serialize, Debug)]
@@ -512,7 +509,7 @@ impl Stream {
         let (schema, rbs) = self.convert_disk_files_to_parquet(
             time_partition.as_ref(),
             custom_partition.as_ref(),
-                init_signal,
+            init_signal,
             shutdown_signal,
         )?;
         // check if there is already a schema file in staging pertaining to this stream
@@ -732,8 +729,8 @@ impl Stream {
         let mut writer = ArrowWriter::try_new(&mut part_file, schema.clone(), Some(props.clone()))?;
         for ref record in record_reader.merged_iter(schema.clone(), time_partition.cloned()) {
             writer.write(record)?;
-                // Collect record batches for finding statistics later
-                record_batches.push(record.clone());
+            // Collect record batches for finding statistics later
+            record_batches.push(record.clone());
         }
         writer.close()?;
 
@@ -1033,11 +1030,10 @@ impl Stream {
         record_batches: Vec<RecordBatch>,
         schema: Arc<Schema>,
     ) -> Result<(), PostError> {
-        let stats_dataset_name = format!("dataset_{INTERNAL_STREAM_NAME}");
         let log_source_entry = LogSourceEntry::new(LogSource::Json, HashSet::new());
         PARSEABLE
             .create_stream_if_not_exists(
-                &stats_dataset_name,
+                DATASET_STATS_STREAM_NAME,
                 StreamType::Internal,
                 vec![log_source_entry],
             )
@@ -1061,7 +1057,7 @@ impl Stream {
 
         flatten_and_push_logs(
             stats_value,
-            &stats_dataset_name,
+            DATASET_STATS_STREAM_NAME,
             &LogSource::Json,
             &HashMap::new(),
         )
@@ -1189,7 +1185,7 @@ impl Stream {
         field_name: &str,
     ) -> Vec<DistinctStat> {
         let sql = format!(
-        "select count(*) as distinct_count, \"{field_name}\" from \"{stream_name}\" where \"{field_name}\" is not null group by \"{field_name}\" order by distinct_count desc limit {}",
+        "select count(*) as distinct_count, \"{field_name}\" from \"{stream_name}\" group by \"{field_name}\" order by distinct_count desc limit {}",
         PARSEABLE.options.max_field_statistics
     );
         let mut distinct_stats = Vec::new();
@@ -1300,7 +1296,8 @@ impl Streams {
             .map(Arc::clone)
             .collect();
         for stream in streams {
-            joinset.spawn(async move { stream.flush_and_convert(init_signal, shutdown_signal).await });
+            joinset
+                .spawn(async move { stream.flush_and_convert(init_signal, shutdown_signal).await });
         }
     }
 }

src/prism/home/mod.rs

@@ -33,7 +33,7 @@ use crate::{
     handlers::http::{cluster::fetch_daily_stats, logstream::error::StreamError},
     parseable::PARSEABLE,
     rbac::{map::SessionKey, role::Action, Users},
-    storage::{ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY},
+    storage::{ObjectStorageError, ObjectStoreFormat, StreamType, STREAM_ROOT_DIRECTORY},
     users::{dashboards::DASHBOARDS, filters::FILTERS},
 };
 
@@ -88,7 +88,10 @@ pub struct HomeSearchResponse {
     resources: Vec<Resource>,
 }
 
-pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, PrismHomeError> {
+pub async fn generate_home_response(
+    key: &SessionKey,
+    include_internal: bool,
+) -> Result<HomeResponse, PrismHomeError> {
     // Execute these operations concurrently
     let (stream_titles_result, alerts_info_result) =
         tokio::join!(get_stream_titles(key), get_alerts_info());
@@ -120,6 +123,14 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
     for result in stream_metadata_results {
         match result {
             Ok((stream, metadata, dataset_type)) => {
+                // Skip internal streams if the flag is false
+                if !include_internal
+                    && metadata
+                        .iter()
+                        .any(|m| m.stream_type == StreamType::Internal)
+                {
+                    continue;
+                }
                 stream_wise_stream_json.insert(stream.clone(), metadata);
                 datasets.push(DataSet {
                     title: stream,