fix: capture time created from metadata not filename

Author: Devdutt Shenoi

src/parseable/streams.rs

@@ -24,12 +24,13 @@ use std::{
     path::{Path, PathBuf},
     process,
     sync::{Arc, Mutex, RwLock},
+    time::UNIX_EPOCH,
 };
 
 use arrow_array::RecordBatch;
 use arrow_ipc::writer::StreamWriter;
 use arrow_schema::{Field, Fields, Schema};
-use chrono::{NaiveDateTime, Timelike, Utc};
+use chrono::{DateTime, NaiveDateTime, Timelike, Utc};
 use derive_more::{Deref, DerefMut};
 use itertools::Itertools;
 use parquet::{
@@ -193,7 +194,7 @@ impl Stream {
     /// Only includes ones starting from the previous minute
     pub fn arrow_files_grouped_exclude_time(
         &self,
-        exclude: NaiveDateTime,
+        exclude: DateTime<Utc>,
         shutdown_signal: bool,
     ) -> HashMap<PathBuf, Vec<PathBuf>> {
         let mut grouped_arrow_file: HashMap<PathBuf, Vec<PathBuf>> = HashMap::new();
@@ -203,12 +204,14 @@ impl Stream {
         // 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())
+                path.metadata()
+                    .expect("Arrow file should exist on disk")
+                    .created()
+                    .expect("Creation time should be accessible")
+                    .duration_since(UNIX_EPOCH)
+                    .expect("Unix Timestamp Duration")
+                    .as_millis()
+                    < exclude.timestamp_millis() as u128
             });
         }
 
@@ -430,8 +433,7 @@ impl Stream {
     ) -> 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(Utc::now(), shutdown_signal);
         if staging_files.is_empty() {
             metrics::STAGING_FILES
                 .with_label_values(&[&self.stream_name])