fix partitioning

Author: nikhilsinhaparseable

src/catalog/mod.rs

@@ -178,7 +178,7 @@ fn create_partition_bounds(lower_bound: DateTime<Utc>) -> (DateTime<Utc>, DateTi
         .date_naive()
         .and_time(
             NaiveTime::from_num_seconds_from_midnight_opt(23 * 3600 + 59 * 60 + 59, 999_999_999)
-                .unwrap_or(NaiveTime::from_hms_opt(23, 59, 59).unwrap()),
+                .unwrap_or_else(|| NaiveTime::from_hms_opt(23, 59, 59).unwrap()),
         )
         .and_utc();
     (partition_lower, partition_upper)

src/handlers/http/ingest.rs

@@ -143,7 +143,9 @@ pub async fn ingest(
         .await?;
 
     if stream.get_time_partition().is_some() {
-        return Err(PostError::IngestionNotAllowedWithTimePartition);
+        return Err(PostError::CustomError(
+            "Ingestion is not allowed to stream with time partition".to_string(),
+        ));
     }
 
     flatten_and_push_logs(json, &stream_name, &log_source, &p_custom_fields, None).await?;
@@ -566,8 +568,6 @@ pub enum PostError {
     InvalidQueryParameter,
     #[error("Missing query parameter")]
     MissingQueryParameter,
-    #[error("Ingestion is not allowed to stream with time partition")]
-    IngestionNotAllowedWithTimePartition,
 }
 
 impl actix_web::ResponseError for PostError {
@@ -599,7 +599,6 @@ impl actix_web::ResponseError for PostError {
             PostError::FieldsCountLimitExceeded(_, _, _) => StatusCode::BAD_REQUEST,
             PostError::InvalidQueryParameter => StatusCode::BAD_REQUEST,
             PostError::MissingQueryParameter => StatusCode::BAD_REQUEST,
-            PostError::IngestionNotAllowedWithTimePartition => StatusCode::BAD_REQUEST,
         }
     }
 

src/storage/object_storage.rs

@@ -1077,10 +1077,10 @@ 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) {
-                if let Err(err) = stats_stream.flush_and_convert(false, false) {
-                    error!("Failed in local sync for dataset stats stream: {err}");
-                }
+            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}");
             }
         });
     }

src/utils/json/mod.rs

@@ -179,16 +179,17 @@ fn process_partitioned_non_array(
     schema_version: SchemaVersion,
     log_source: &LogSource,
 ) -> Result<Vec<Value>, anyhow::Error> {
-    let data = flatten_json_body(
-        body,
+    // convert to an array for processing
+    let arr = vec![body];
+    let processed_elements = process_partitioned_array(
+        arr,
         time_partition,
         time_partition_limit,
         custom_partition,
         schema_version,
-        true,
         log_source,
     )?;
-    Ok(vec![data])
+    Ok(processed_elements)
 }
 
 /// Processes data when no partitioning is configured (original logic)