camelcase stream info, fix stats for historical

Author: nikhilsinhaparseable

src/handlers/http/cluster/mod.rs

@@ -622,6 +622,10 @@ pub async fn send_retention_cleanup_request(
             "failed to perform cleanup on retention: {}\nResponse Returned: {:?}",
             ingestor.domain_name, body
         );
+        return Err(ObjectStorageError::Custom(format!(
+            "failed to perform cleanup on retention: {}\nResponse Returned: {:?}",
+            ingestor.domain_name, body
+        )));
     }
 
     Ok(())

src/handlers/http/logstream.rs

@@ -333,7 +333,7 @@ pub async fn get_stream_info(stream_name: Path<String>) -> Result<impl Responder
         return Err(StreamNotFound(stream_name.clone()).into());
     }
 
-    let storage = PARSEABLE.storage.get_object_store();
+    let storage = PARSEABLE.storage().get_object_store();
 
     // Get first and latest event timestamps from storage
     let (stream_first_event_at, stream_latest_event_at) = match storage
@@ -412,7 +412,7 @@ pub async fn put_stream_hot_tier(
     hot_tier_manager
         .put_hot_tier(&stream_name, &mut hottier)
         .await?;
-    let storage = PARSEABLE.storage.get_object_store();
+    let storage = PARSEABLE.storage().get_object_store();
     let mut stream_metadata = storage.get_object_store_format(&stream_name).await?;
     stream_metadata.hot_tier_enabled = true;
     storage

src/handlers/http/modal/ingest/ingestor_logstream.rs

@@ -38,7 +38,7 @@ pub async fn retention_cleanup(
     Json(date_list): Json<Vec<String>>,
 ) -> Result<impl Responder, StreamError> {
     let stream_name = stream_name.into_inner();
-    let storage = PARSEABLE.storage.get_object_store();
+    let storage = PARSEABLE.storage().get_object_store();
     // if the stream not found in memory map,
     //check if it exists in the storage
     //create stream and schema from storage
@@ -62,7 +62,7 @@ pub async fn retention_cleanup(
         });
     }
 
-    Ok(("Cleanup complete", StatusCode::OK))
+    Ok(actix_web::HttpResponse::NoContent().finish())
 }
 
 pub async fn delete(stream_name: Path<String>) -> Result<impl Responder, StreamError> {

src/migration/mod.rs

@@ -366,7 +366,7 @@ async fn setup_logstream_metadata(
         ..
     } = serde_json::from_value(stream_metadata_value).unwrap_or_default();
 
-    let storage = PARSEABLE.storage.get_object_store();
+    let storage = PARSEABLE.storage().get_object_store();
 
     update_data_type_time_partition(arrow_schema, time_partition.as_ref()).await?;
     storage.put_schema(stream, arrow_schema).await?;

src/prism/logstream/mod.rs

@@ -152,7 +152,7 @@ async fn get_stream_info_helper(stream_name: &str) -> Result<StreamInfo, StreamE
         return Err(StreamNotFound(stream_name.to_owned()).into());
     }
 
-    let storage = PARSEABLE.storage.get_object_store();
+    let storage = PARSEABLE.storage().get_object_store();
 
     // Get first and latest event timestamps from storage
     let (stream_first_event_at, stream_latest_event_at) = match storage

src/storage/mod.rs

@@ -130,13 +130,11 @@ pub struct ObjectStoreFormat {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct StreamInfo {
-    #[serde(rename = "created-at")]
     pub created_at: String,
-    #[serde(rename = "first-event-at")]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub first_event_at: Option<String>,
-    #[serde(rename = "latest-event-at")]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub latest_event_at: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]

src/storage/object_storage.rs

@@ -38,10 +38,8 @@ use std::path::Path;
 use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
-use tokio::task;
 use tokio::task::JoinSet;
 use tracing::info;
-use tracing::trace;
 use tracing::{error, warn};
 use ulid::Ulid;
 
@@ -96,7 +94,6 @@ impl UploadContext {
 
 /// Result of a single file upload operation
 pub(crate) struct UploadResult {
-    stats_calculated: bool,
     file_path: std::path::PathBuf,
     manifest_file: Option<catalog::manifest::File>,
 }
@@ -135,10 +132,9 @@ async fn upload_single_parquet_file(
     let manifest = catalog::create_from_parquet_file(absolute_path, &path)?;
 
     // Calculate field stats if enabled
-    let stats_calculated = calculate_stats_if_enabled(&stream_name, &path, &schema).await;
+    calculate_stats_if_enabled(&stream_name, &path, &schema).await;
 
     Ok(UploadResult {
-        stats_calculated,
         file_path: path,
         manifest_file: Some(manifest),
     })
@@ -172,19 +168,19 @@ async fn calculate_stats_if_enabled(
     stream_name: &str,
     path: &std::path::Path,
     schema: &Arc<Schema>,
-) -> bool {
+) {
     if stream_name != DATASET_STATS_STREAM_NAME && PARSEABLE.options.collect_dataset_stats {
         let max_field_statistics = PARSEABLE.options.max_field_statistics;
-        match calculate_field_stats(stream_name, path, schema, max_field_statistics).await {
-            Ok(stats) if stats => return true,
-            Err(err) => trace!(
+        if let Err(err) =
+            calculate_field_stats(stream_name, path, schema, max_field_statistics).await
+        {
+            tracing::trace!(
                 "Error calculating field stats for stream {}: {}",
-                stream_name, err
-            ),
-            _ => {}
+                stream_name,
+                err
+            );
         }
     }
-    false
 }
 
 pub trait ObjectStorageProvider: StorageMetrics + std::fmt::Debug + Send + Sync {
@@ -978,19 +974,18 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
             {
                 match stats {
                     TypedStatistics::Int(int_stats) => {
-                        let min_ts =
-                            DateTime::from_timestamp_millis(int_stats.min).unwrap_or_default();
-                        min_timestamp = Some(match min_timestamp {
-                            Some(existing) => existing.min(min_ts),
-                            None => min_ts,
-                        });
-
-                        let max_ts =
-                            DateTime::from_timestamp_millis(int_stats.max).unwrap_or_default();
-                        max_timestamp = Some(match max_timestamp {
-                            Some(existing) => existing.max(max_ts),
-                            None => max_ts,
-                        });
+                        if let Some(min_ts) = DateTime::from_timestamp_millis(int_stats.min) {
+                            min_timestamp = Some(match min_timestamp {
+                                Some(existing) => existing.min(min_ts),
+                                None => min_ts,
+                            });
+                        }
+                        if let Some(max_ts) = DateTime::from_timestamp_millis(int_stats.max) {
+                            max_timestamp = Some(match max_timestamp {
+                                Some(existing) => existing.max(max_ts),
+                                None => max_ts,
+                            });
+                        }
                     }
                     TypedStatistics::String(str_stats) => {
                         if let Ok(min_ts) = DateTime::parse_from_rfc3339(&str_stats.min) {
@@ -1070,18 +1065,14 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         let upload_context = UploadContext::new(stream);
 
         // Process parquet files concurrently and collect results
-        let (stats_calculated, manifest_files) =
-            process_parquet_files(&upload_context, stream_name).await?;
+        let manifest_files = process_parquet_files(&upload_context, stream_name).await?;
 
         // Update snapshot with collected manifest files
         update_snapshot_with_manifests(stream_name, manifest_files).await?;
 
         // Process schema files
         process_schema_files(&upload_context, stream_name).await?;
 
-        // Handle stats synchronization if needed
-        handle_stats_sync(stats_calculated).await;
-
         Ok(())
     }
 }
@@ -1090,7 +1081,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
 async fn process_parquet_files(
     upload_context: &UploadContext,
     stream_name: &str,
-) -> Result<(bool, Vec<catalog::manifest::File>), ObjectStorageError> {
+) -> Result<Vec<catalog::manifest::File>, ObjectStorageError> {
     let semaphore = Arc::new(tokio::sync::Semaphore::new(100));
     let mut join_set = JoinSet::new();
     let object_store = PARSEABLE.storage().get_object_store();
@@ -1143,16 +1134,12 @@ async fn spawn_parquet_upload_task(
 /// Collects results from all upload tasks
 async fn collect_upload_results(
     mut join_set: JoinSet<Result<UploadResult, ObjectStorageError>>,
-) -> Result<(bool, Vec<catalog::manifest::File>), ObjectStorageError> {
-    let mut stats_calculated = false;
+) -> Result<Vec<catalog::manifest::File>, ObjectStorageError> {
     let mut uploaded_files = Vec::new();
 
     while let Some(result) = join_set.join_next().await {
         match result {
             Ok(Ok(upload_result)) => {
-                if upload_result.stats_calculated {
-                    stats_calculated = true;
-                }
                 if let Some(manifest_file) = upload_result.manifest_file {
                     uploaded_files.push((upload_result.file_path, manifest_file));
                 } else {
@@ -1183,7 +1170,7 @@ async fn collect_upload_results(
         })
         .collect();
 
-    Ok((stats_calculated, manifest_files))
+    Ok(manifest_files)
 }
 
 /// Updates snapshot with collected manifest files
@@ -1215,20 +1202,6 @@ async fn process_schema_files(
     Ok(())
 }
 
-/// Handles stats synchronization if needed
-async fn handle_stats_sync(stats_calculated: bool) {
-    if stats_calculated {
-        // perform local sync for the `pstats` dataset
-        task::spawn(async move {
-            if let Ok(stats_stream) = PARSEABLE.get_stream(DATASET_STATS_STREAM_NAME)
-                && let Err(err) = stats_stream.flush_and_convert(false, false)
-            {
-                error!("Failed in local sync for dataset stats stream: {err}");
-            }
-        });
-    }
-}
-
 /// Builds the stream relative path for a file
 fn stream_relative_path(
     stream_name: &str,