fix for converting arrows to parquet

Author: nikhilsinhaparseable

src/event/mod.rs

@@ -47,18 +47,7 @@ pub struct Event {
 // Events holds the schema related to a each event for a single log stream
 impl Event {
     pub async 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.custom_partition_values.is_empty() {
-            for (k, v) in self.custom_partition_values.iter().sorted_by_key(|v| v.0) {
-                key.push_str(&format!("&{k}={v}"));
-            }
-        }
-
+        let key = get_schema_key(&self.rb.schema().fields);
         if self.is_first_event {
             commit_schema(&self.stream_name, self.rb.schema())?;
         }

src/staging/streams.rs

@@ -28,7 +28,7 @@ use std::{
 use arrow_array::RecordBatch;
 use arrow_ipc::writer::StreamWriter;
 use arrow_schema::Schema;
-use chrono::{NaiveDateTime, Timelike, Utc};
+use chrono::{DateTime, Datelike, NaiveDateTime, Timelike, Utc};
 use derive_more::{Deref, DerefMut};
 use itertools::Itertools;
 use parquet::{
@@ -136,8 +136,7 @@ impl<'a> Stream<'a> {
             hostname.push_str(&INGESTOR_META.get_ingestor_id());
         }
         let filename = format!(
-            "{}{stream_hash}.date={}.hour={:02}.minute={}.{}{hostname}.{ARROW_FILE_EXTENSION}",
-            Utc::now().format("%Y%m%dT%H%M"),
+            "{stream_hash}.date={}.hour={:02}.minute={}.{}{hostname}.{ARROW_FILE_EXTENSION}",
             parsed_timestamp.date(),
             parsed_timestamp.hour(),
             minute_to_slot(parsed_timestamp.minute(), OBJECT_STORE_DATA_GRANULARITY).unwrap(),
@@ -155,13 +154,13 @@ impl<'a> Stream<'a> {
             return vec![];
         };
 
-        let paths = dir
+        let paths: Vec<PathBuf> = dir
             .flatten()
             .map(|file| file.path())
             .filter(|file| file.extension().is_some_and(|ext| ext.eq("arrows")))
-            .sorted_by_key(|f| f.metadata().unwrap().modified().unwrap())
+            .sorted_by_key(|f| f.metadata().unwrap().created().unwrap())
             .collect();
-
+        
         paths
     }
 
@@ -172,24 +171,36 @@ impl<'a> Stream<'a> {
     /// Only includes ones starting from the previous minute
     pub fn arrow_files_grouped_exclude_time(
         &self,
-        exclude: NaiveDateTime,
         shutdown_signal: bool,
     ) -> HashMap<PathBuf, Vec<PathBuf>> {
+        let now = Utc::now();
+    
+        // Extract date and time components of current time
+        let now_date = (now.year(), now.month(), now.day());
+        let now_time = (now.hour(), now.minute());
+
         let mut grouped_arrow_file: HashMap<PathBuf, Vec<PathBuf>> = HashMap::new();
         let mut arrow_files = self.arrow_files();
-
-        // if the shutdown signal is false i.e. normal condition
-        // don't keep the ones for the current minute
-        if !shutdown_signal {
-            arrow_files.retain(|path| {
-                !path
-                    .file_name()
-                    .unwrap()
-                    .to_str()
-                    .unwrap()
-                    .starts_with(&exclude.format("%Y%m%dT%H%M").to_string())
-            });
-        }
+        arrow_files = arrow_files
+            .into_iter()
+            .filter(|path| {
+                let created_at = path.metadata().unwrap().created().unwrap();
+                let created_at: DateTime<Utc> = created_at.into();
+                let created_date = (created_at.year(), created_at.month(), created_at.day());
+                let created_time = (created_at.hour(), created_at.minute());
+        
+                let same_date = now_date == created_date;
+                let same_time = now_time == created_time;
+        
+                // if the shutdown signal is false i.e. normal condition
+                // don't keep the ones for the current minute
+                if !shutdown_signal {
+                    !same_date || !same_time
+                } else {
+                    true
+                }
+            })
+            .collect();
 
         let random_string =
             rand::distributions::Alphanumeric.sample_string(&mut rand::thread_rng(), 15);
@@ -306,8 +317,7 @@ impl<'a> Stream<'a> {
     ) -> Result<Option<Schema>, StagingError> {
         let mut schemas = Vec::new();
 
-        let time = chrono::Utc::now().naive_utc();
-        let staging_files = self.arrow_files_grouped_exclude_time(time, shutdown_signal);
+        let staging_files = self.arrow_files_grouped_exclude_time(shutdown_signal);
         if staging_files.is_empty() {
             metrics::STAGING_FILES
                 .with_label_values(&[&self.stream_name])