feat: enable time partition

create stream with custom time partition
use header -
X-P-Time-Partition: <field-name> - this has to be a timestamp field
X-P-Time-Partition-Limit: max historical range default to 30d

server validates if all events in the batch has this time partition field
if false, it rejects the whole batch

Author: nikhilsinhaparseable

src/handlers/http/modal/utils/ingest_utils.rs

@@ -108,25 +108,14 @@ pub async fn push_logs(
     let schema_version = stream.get_schema_version();
     let p_timestamp = Utc::now();
 
-    let data = if time_partition.is_some() || custom_partition.is_some() {
-        convert_array_to_object(
-            json,
-            time_partition.as_ref(),
-            time_partition_limit,
-            custom_partition.as_ref(),
-            schema_version,
-            log_source,
-        )?
-    } else {
-        vec![convert_to_array(convert_array_to_object(
-            json,
-            None,
-            None,
-            None,
-            schema_version,
-            log_source,
-        )?)?]
-    };
+    let data = convert_array_to_object(
+        json,
+        time_partition.as_ref(),
+        time_partition_limit,
+        custom_partition.as_ref(),
+        schema_version,
+        log_source,
+    )?;
 
     for json in data {
         let origin_size = serde_json::to_vec(&json).unwrap().len() as u64; // string length need not be the same as byte length

src/parseable/mod.rs

@@ -525,17 +525,23 @@ impl Parseable {
                 .await;
         }
 
-        if !time_partition.is_empty() || !time_partition_limit.is_empty() {
-            return Err(StreamError::Custom {
-                msg: "Creating stream with time partition is not supported anymore".to_string(),
-                status: StatusCode::BAD_REQUEST,
-            });
-        }
+        let time_partition_in_days = if !time_partition_limit.is_empty() {
+            Some(validate_time_partition_limit(&time_partition_limit)?)
+        } else {
+            None
+        };
 
         if let Some(custom_partition) = &custom_partition {
             validate_custom_partition(custom_partition)?;
         }
 
+        if !time_partition.is_empty() && custom_partition.is_some() {
+            return Err(StreamError::Custom {
+                msg: "Cannot set both time partition and custom partition".to_string(),
+                status: StatusCode::BAD_REQUEST,
+            });
+        }
+
         let schema = validate_static_schema(
             body,
             stream_name,
@@ -547,7 +553,7 @@ impl Parseable {
         self.create_stream(
             stream_name.to_string(),
             &time_partition,
-            None,
+            time_partition_in_days,
             custom_partition.as_ref(),
             static_schema_flag,
             schema,

src/parseable/staging/writer.rs

@@ -31,7 +31,8 @@ use arrow_schema::Schema;
 use arrow_select::concat::concat_batches;
 use chrono::Utc;
 use itertools::Itertools;
-use tracing::{error, warn};
+use rand::distributions::{Alphanumeric, DistString};
+use tracing::error;
 
 use crate::{
     parseable::{ARROW_FILE_EXTENSION, PART_FILE_EXTENSION},
@@ -92,8 +93,15 @@ impl Drop for DiskWriter {
         let mut arrow_path = self.path.to_owned();
         arrow_path.set_extension(ARROW_FILE_EXTENSION);
 
+        // If file exists, append a random string before .date to avoid overwriting
         if arrow_path.exists() {
-            warn!("File {arrow_path:?} exists and will be overwritten");
+            let file_name = arrow_path.file_name().unwrap().to_string_lossy();
+            let date_pos = file_name
+                .find(".date")
+                .expect("File name should contain .date");
+            let random_suffix = Alphanumeric.sample_string(&mut rand::thread_rng(), 8);
+            let new_name = format!("{}{}", random_suffix, &file_name[date_pos..]);
+            arrow_path.set_file_name(new_name);
         }
 
         if let Err(err) = std::fs::rename(&self.path, &arrow_path) {

src/parseable/streams.rs

@@ -50,7 +50,6 @@ use crate::{
         DEFAULT_TIMESTAMP_KEY,
         format::{LogSource, LogSourceEntry},
     },
-    handlers::http::modal::{ingest_server::INGESTOR_META, query_server::QUERIER_META},
     metadata::{LogStreamMetadata, SchemaVersion},
     metrics,
     option::Mode,
@@ -274,7 +273,7 @@ impl Stream {
         init_signal: bool,
         shutdown_signal: bool,
     ) -> HashMap<PathBuf, Vec<PathBuf>> {
-        let random_string = self.get_node_id_string();
+        let random_string = ulid::Ulid::new().to_string();
         let inprocess_dir = Self::inprocess_folder(&self.data_path, group_minute);
 
         let arrow_files = self.fetch_arrow_files_for_conversion(exclude, shutdown_signal);
@@ -322,21 +321,6 @@ impl Stream {
         grouped
     }
 
-    /// Returns the node id string for file naming.
-    fn get_node_id_string(&self) -> String {
-        match self.options.mode {
-            Mode::Query => QUERIER_META
-                .get()
-                .map(|querier_metadata| querier_metadata.get_node_id())
-                .expect("Querier metadata should be set"),
-            Mode::Ingest => INGESTOR_META
-                .get()
-                .map(|ingestor_metadata| ingestor_metadata.get_node_id())
-                .expect("Ingestor metadata should be set"),
-            _ => "000000000000000".to_string(),
-        }
-    }
-
     /// Returns a mapping for inprocess arrow files (init_signal=true).
     fn group_inprocess_arrow_files(&self, random_string: &str) -> HashMap<PathBuf, Vec<PathBuf>> {
         let mut grouped: HashMap<PathBuf, Vec<PathBuf>> = HashMap::new();