Fix created time in cubestore parquet files

rust/cubestore/cubestore/src/remotefs/cleanup.rs

@@ -4,7 +4,9 @@ use crate::remotefs::RemoteFs;
 use crate::{app_metrics, CubeError};
 use chrono::Utc;
 use datafusion::cube_ext;
+use datafusion::parquet::file::reader::{FileReader, SerializedFileReader};
 use std::collections::HashSet;
+use std::fs::{DirEntry, File};
 use std::path::Path;
 use std::sync::Arc;
 use tokio::time::Duration;
@@ -219,27 +221,11 @@ impl RemoteFsCleanup {
                         if !file_name.ends_with(".parquet") {
                             continue;
                         }
-
-                        let should_deleted = if let Ok(metadata) = entry.metadata() {
-                            match metadata.created() {
-                                Ok(created) => {
-                                    if created
-                                        .elapsed()
-                                        .map_or(true, |e| e < cleanup_local_files_delay)
-                                    {
-                                        false
-                                    } else {
-                                        true
-                                    }
-                                }
-                                Err(e) => {
-                                    log::error!(
-                                        "error while getting created time for file {:?}:{}",
-                                        entry.file_name(),
-                                        e
-                                    );
-                                    false
-                                }
+                        let should_deleted = if let Some(created) = get_timestamp(&entry) {
+                            if created < cleanup_local_files_delay {
+                                false
+                            } else {
+                                true
                             }
                         } else {
                             false
@@ -286,6 +272,38 @@ impl RemoteFsCleanup {
         }
     }
 }
+
+fn get_timestamp(entry: &DirEntry) -> Option<Duration> {
+    let path = entry.path();
+    let file = match File::open(&path) {
+        Ok(file) => file,
+        Err(e) => {
+            log::error!("Error opening file {:?}: {}", path, e);
+            return None;
+        }
+    };
+    let reader = match SerializedFileReader::new(file) {
+        Ok(reader) => reader,
+        Err(e) => {
+            log::error!("Error reading Parquet file {:?}: {}", path, e);
+            return None;
+        }
+    };
+    let metadata = reader.metadata().file_metadata().key_value_metadata();
+    if let Some(key_values) = metadata {
+        let created = key_values
+            .iter()
+            .find(|key_value| key_value.key == "created_at");
+        if let Some(created) = created {
+            if let Some(value) = created.value.as_ref() {
+                if let Ok(timestamp) = value.parse::<u64>() {
+                    return Some(Duration::from_secs(timestamp));
+                }
+            }
+        }
+    }
+    None
+}
 #[cfg(test)]
 mod test {
     use super::*;

rust/cubestore/cubestore/src/table/parquet.rs

@@ -7,12 +7,14 @@
 use datafusion::arrow::datatypes::Schema;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::parquet::arrow::{ArrowReader, ArrowWriter, ParquetFileArrowReader};
+use datafusion::parquet::file::metadata::KeyValue;
 use datafusion::parquet::file::properties::{
     WriterProperties, WriterPropertiesBuilder, WriterVersion,
 };
 use datafusion::physical_plan::parquet::{MetadataCacheFactory, ParquetMetadataCache};
 use std::fs::File;
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 pub trait CubestoreParquetMetadataCache: DIService + Send + Sync {
     fn cache(self: &Self) -> Arc<dyn ParquetMetadataCache>;
@@ -128,14 +130,22 @@
     pub fn arrow_schema(&self) -> Schema {
         arrow_schema(&self.table)
     }
 
     pub async fn writer_props(&self, table: &IdRow<Table>) -> Result<WriterProperties, CubeError> {
+        let now = SystemTime::now().duration_since(UNIX_EPOCH);
+        let created_at_key_value = KeyValue::new(
+            "created_at".to_string(),
+            now?.as_secs().to_string(),
+        );
+        let mut metadata = Vec::new();
+        metadata.push(created_at_key_value);
         self.metadata_cache_factory
             .build_writer_props(
                 table,
                 WriterProperties::builder()
                     .set_max_row_group_size(self.row_group_size)
-                    .set_writer_version(WriterVersion::PARQUET_2_0),
+                    .set_writer_version(WriterVersion::PARQUET_2_0)
+                    .set_key_value_metadata(Some(metadata)),
             )
             .await
             .map_err(CubeError::from)