feat: re-enable time-partitioning

Author: Devdutt Shenoi

src/event/format/json.rs

@@ -179,7 +179,7 @@ impl EventFormat for Event {
             origin_size,
             is_first_event,
             parsed_timestamp,
-            time_partition: None,
+            time_partitioned: time_partition.is_some(),
             custom_partition_values,
             stream_type,
         })

src/event/mod.rs

@@ -31,7 +31,7 @@ use crate::{
     storage::StreamType,
     LOCK_EXPECT,
 };
-use chrono::NaiveDateTime;
+use chrono::{NaiveDateTime, Utc};
 use std::collections::HashMap;
 
 pub const DEFAULT_TIMESTAMP_KEY: &str = "p_timestamp";
@@ -47,7 +47,7 @@ pub struct Event {
     pub origin_size: u64,
     pub is_first_event: bool,
     pub parsed_timestamp: NaiveDateTime,
-    pub time_partition: Option<String>,
+    pub time_partitioned: bool,
     pub custom_partition_values: HashMap<String, String>,
     pub stream_type: StreamType,
 }
@@ -56,12 +56,14 @@ pub struct Event {
 impl Event {
     pub fn process(self) -> Result<(), EventError> {
         let mut key = get_schema_key(&self.rb.schema().fields);
-        if self.time_partition.is_some() {
-            let parsed_timestamp_to_min = self.parsed_timestamp.format("%Y%m%dT%H%M").to_string();
-            key.push_str(&parsed_timestamp_to_min);
+        if self.time_partitioned {
+            // For time partitioned streams, concatenate timestamp to filename, ensuring we don't write to a finished arrows file
+            let curr_timestamp = Utc::now().format("%Y%m%dT%H%M").to_string();
+            key.push_str(&curr_timestamp);
         }
 
         if !self.custom_partition_values.is_empty() {
+            // For custom partitioned streams, concatenate values to filename, ensuring we write to different arrows files
             for (k, v) in self.custom_partition_values.iter().sorted_by_key(|v| v.0) {
                 key.push_str(&format!("&{k}={v}"));
             }

src/handlers/http/ingest.rs

@@ -296,7 +296,7 @@ pub async fn push_logs_unchecked(
         origin_format: "json",
         origin_size: 0,
         parsed_timestamp: Utc::now().naive_utc(),
-        time_partition: None,
+        time_partitioned: false,
         is_first_event: true,                    // NOTE: Maybe should be false
         custom_partition_values: HashMap::new(), // should be an empty map for unchecked push
         stream_type: StreamType::UserDefined,