feat: add stats for each field

src/cli.rs

@@ -388,6 +388,41 @@ pub struct Options {
         help = "Maximum level of flattening allowed for events"
     )]
     pub event_flatten_level: usize,
+
+    // maximum limit to store the statistics for a field
+    #[arg(
+        long,
+        env = "P_MAX_FIELD_STATISTICS",
+        default_value = "50",
+        help = "Maximum number of field statistics to store"
+    )]
+    pub max_field_statistics: usize,
+
+    // collect dataset stats
+    #[arg(
+        long,
+        env = "P_COLLECT_DATASET_STATS",
+        default_value = "false",
+        help = "Enable/Disable collecting dataset stats"
+    )]
+    pub collect_dataset_stats: bool,
+
+    // the duration during which local sync should be completed
+    #[arg(
+        long,
+        env = "P_LOCAL_SYNC_THRESHOLD",
+        default_value = "30",
+        help = "Local sync threshold in seconds"
+    )]
+    pub local_sync_threshold: u64,
+    // the duration during which object store sync should be completed
+    #[arg(
+        long,
+        env = "P_OBJECT_STORE_SYNC_THRESHOLD",
+        default_value = "60",
+        help = "Object store sync threshold in seconds"
+    )]
+    pub object_store_sync_threshold: u64,
 }
 
 #[derive(Parser, Debug)]

src/connectors/kafka/processor.rs

@@ -55,6 +55,7 @@ impl ParseableSinkProcessor {
             .create_stream_if_not_exists(
                 stream_name,
                 StreamType::UserDefined,
+                None,
                 vec![log_source_entry],
             )
             .await?;

src/handlers/http/health_check.rs

@@ -56,12 +56,35 @@ pub async fn check_shutdown_middleware(
 
 // This function is called when the server is shutting down
 pub async fn shutdown() {
-    // Set the shutdown flag to true
-    let mut shutdown_flag = SIGNAL_RECEIVED.lock().await;
-    *shutdown_flag = true;
+    // Set shutdown flag to true
+    set_shutdown_flag().await;
 
     //sleep for 5 secs to allow any ongoing requests to finish
     tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+
+    // Perform sync operations
+    perform_sync_operations().await;
+
+    // If collect_dataset_stats is enabled, perform sync operations
+    // This is to ensure that all stats data is synced before the server shuts down
+    if PARSEABLE.options.collect_dataset_stats {
+        perform_sync_operations().await;
+    }
+}
+
+async fn set_shutdown_flag() {
+    let mut shutdown_flag = SIGNAL_RECEIVED.lock().await;
+    *shutdown_flag = true;
+}
+
+async fn perform_sync_operations() {
+    // Perform local sync
+    perform_local_sync().await;
+    // Perform object store sync
+    perform_object_store_sync().await;
+}
+
+async fn perform_local_sync() {
     let mut local_sync_joinset = JoinSet::new();
 
     // Sync staging
@@ -76,7 +99,9 @@ pub async fn shutdown() {
             Err(err) => error!("Failed to join async task: {err}"),
         }
     }
+}
 
+async fn perform_object_store_sync() {
     // Sync object store
     let mut object_store_joinset = JoinSet::new();
     sync_all_streams(&mut object_store_joinset);

src/handlers/http/ingest.rs

@@ -100,6 +100,7 @@ pub async fn ingest(
         .create_stream_if_not_exists(
             &stream_name,
             StreamType::UserDefined,
+            None,
             vec![log_source_entry.clone()],
         )
         .await?;
@@ -183,6 +184,7 @@ pub async fn handle_otel_logs_ingestion(
         .create_stream_if_not_exists(
             &stream_name,
             StreamType::UserDefined,
+            None,
             vec![log_source_entry.clone()],
         )
         .await?;
@@ -248,6 +250,7 @@ pub async fn handle_otel_metrics_ingestion(
         .create_stream_if_not_exists(
             &stream_name,
             StreamType::UserDefined,
+            None,
             vec![log_source_entry.clone()],
         )
         .await?;
@@ -313,6 +316,7 @@ pub async fn handle_otel_traces_ingestion(
         .create_stream_if_not_exists(
             &stream_name,
             StreamType::UserDefined,
+            None,
             vec![log_source_entry.clone()],
         )
         .await?;

src/handlers/http/prism_home.rs

@@ -26,7 +26,7 @@ use crate::{
 };
 
 const HOME_SEARCH_QUERY_PARAM: &str = "key";
-
+pub const HOME_QUERY_PARAM: &str = "includeInternal";
 /// Fetches the data to populate Prism's home
 ///
 ///
@@ -36,8 +36,12 @@ 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).is_some_and(|v| v == "true");
+
+    let res = generate_home_response(&key, include_internal).await?;
 
     Ok(web::Json(res))
 }
@@ -52,11 +56,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/mod.rs

@@ -330,6 +330,7 @@ impl Parseable {
             .create_stream_if_not_exists(
                 INTERNAL_STREAM_NAME,
                 StreamType::Internal,
+                None,
                 vec![log_source_entry],
             )
             .await
@@ -354,12 +355,18 @@ impl Parseable {
         &self,
         stream_name: &str,
         stream_type: StreamType,
+        custom_partition: Option<&String>,
         log_source: Vec<LogSourceEntry>,
     ) -> Result<bool, PostError> {
         if self.streams.contains(stream_name) {
             return Ok(true);
         }
 
+        // validate custom partition if provided
+        if let Some(partition) = custom_partition {
+            validate_custom_partition(partition)?;
+        }
+
         // For distributed deployments, if the stream not found in memory map,
         //check if it exists in the storage
         //create stream and schema from storage
@@ -375,7 +382,7 @@ impl Parseable {
             stream_name.to_string(),
             "",
             None,
-            None,
+            custom_partition,
             false,
             Arc::new(Schema::empty()),
             stream_type,

src/parseable/streams.rs

@@ -29,7 +29,7 @@ use std::{
 use arrow_array::RecordBatch;
 use arrow_schema::{Field, Fields, Schema};
 use chrono::{NaiveDateTime, Timelike, Utc};
-use derive_more::{Deref, DerefMut};
+use derive_more::derive::{Deref, DerefMut};
 use itertools::Itertools;
 use parquet::{
     arrow::ArrowWriter,
@@ -463,7 +463,7 @@ impl Stream {
     }
 
     /// Converts arrow files in staging into parquet files, does so only for past minutes when run with `!shutdown_signal`
-    pub fn prepare_parquet(
+    pub async fn prepare_parquet(
         &self,
         init_signal: bool,
         shutdown_signal: bool,
@@ -478,15 +478,12 @@ impl Stream {
 
         // read arrow files on disk
         // convert them to parquet
-        let schema = self
-            .convert_disk_files_to_parquet(
-                time_partition.as_ref(),
-                custom_partition.as_ref(),
-                init_signal,
-                shutdown_signal,
-            )
-            .inspect_err(|err| warn!("Error while converting arrow to parquet- {err:?}"))?;
-
+        let schema = self.convert_disk_files_to_parquet(
+            time_partition.as_ref(),
+            custom_partition.as_ref(),
+            init_signal,
+            shutdown_signal,
+        )?;
         // check if there is already a schema file in staging pertaining to this stream
         // if yes, then merge them and save
 
@@ -641,6 +638,7 @@ impl Stream {
 
         self.update_staging_metrics(&staging_files);
 
+        let mut record_batches = Vec::new();
         for (parquet_path, arrow_files) in staging_files {
             let record_reader = MergedReverseRecordReader::try_new(&arrow_files);
             if record_reader.readers.is_empty() {
@@ -658,6 +656,7 @@ impl Stream {
                 &schema,
                 &props,
                 time_partition,
+                &mut record_batches,
             )? {
                 continue;
             }
@@ -683,6 +682,7 @@ impl Stream {
         schema: &Arc<Schema>,
         props: &WriterProperties,
         time_partition: Option<&String>,
+        record_batches: &mut Vec<RecordBatch>,
     ) -> Result<bool, StagingError> {
         let mut part_file = OpenOptions::new()
             .create(true)
@@ -692,6 +692,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());
         }
         writer.close()?;
 
@@ -958,7 +960,7 @@ impl Stream {
     }
 
     /// First flushes arrows onto disk and then converts the arrow into parquet
-    pub fn flush_and_convert(
+    pub async fn flush_and_convert(
         &self,
         init_signal: bool,
         shutdown_signal: bool,
@@ -972,7 +974,8 @@ impl Stream {
         );
 
         let start_convert = Instant::now();
-        self.prepare_parquet(init_signal, shutdown_signal)?;
+
+        self.prepare_parquet(init_signal, shutdown_signal).await?;
         trace!(
             "Converting arrows to parquet on stream ({}) took: {}s",
             self.stream_name,
@@ -1067,7 +1070,8 @@ impl Streams {
             .map(Arc::clone)
             .collect();
         for stream in streams {
-            joinset.spawn(async move { stream.flush_and_convert(init_signal, shutdown_signal) });
+            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()
+                        .all(|m| m.stream_type == StreamType::Internal)
+                {
+                    continue;
+                }
                 stream_wise_stream_json.insert(stream.clone(), metadata);
                 datasets.push(DataSet {
                     title: stream,