fix: ignore invalid/corrupt staging files (#990)

* fix: ignore invalid/corrupt staging files

issue: if one arrow file is present in staging directory of an ingestor,
the server panics in the sync flow and does not sync the other files to the backend store

this is a two fold problem -
1. find the reason behind corruption of an arrow file
2. discard the invalid/corrupt arrow file and let server sync the other files by notifying the name or the path of the corrupt file for further analysis

This PR solves problem 2 where server notifies the path of the corrupt file name
and ignores the file so that other files do not get stuck in the staging directory
---------

Co-authored-by: Devdutt Shenoi <[email protected]>

Author: nikhilsinhaparseable

server/src/metadata.rs

@@ -410,9 +410,12 @@ impl StreamInfo {
 
 fn update_schema_from_staging(stream_name: &str, current_schema: Schema) -> Schema {
     let staging_files = StorageDir::new(stream_name).arrow_files();
-    let schema = MergedRecordReader::try_new(&staging_files)
-        .unwrap()
-        .merged_schema();
+    let record_reader = MergedRecordReader::try_new(&staging_files).unwrap();
+    if record_reader.readers.is_empty() {
+        return current_schema;
+    }
+
+    let schema = record_reader.merged_schema();
 
     Schema::try_merge(vec![schema, current_schema]).unwrap()
 }

server/src/storage/staging.rs

@@ -260,6 +260,9 @@ pub fn convert_disk_files_to_parquet(
         }
 
         let record_reader = MergedReverseRecordReader::try_new(&files).unwrap();
+        if record_reader.readers.is_empty() {
+            continue;
+        }
         let merged_schema = record_reader.merged_schema();
         let mut index_time_partition: usize = 0;
         if let Some(time_partition) = time_partition.as_ref() {

server/src/utils/arrow/merged_reader.rs

@@ -49,8 +49,13 @@ impl MergedRecordReader {
                 log::error!("Invalid file detected, removing it: {:?}", file);
                 fs::remove_file(file).unwrap();
             } else {
-                let reader = StreamReader::try_new(BufReader::new(File::open(file).unwrap()), None)
-                    .map_err(|_| ())?;
+                let Ok(reader) =
+                    StreamReader::try_new(BufReader::new(File::open(file).unwrap()), None)
+                else {
+                    log::error!("Invalid file detected, ignoring it: {:?}", file);
+                    continue;
+                };
+
                 readers.push(reader);
             }
         }
@@ -77,9 +82,13 @@ impl MergedReverseRecordReader {
     pub fn try_new(files: &[PathBuf]) -> Result<Self, ()> {
         let mut readers = Vec::with_capacity(files.len());
         for file in files {
-            let reader =
+            let Ok(reader) =
                 utils::arrow::reverse_reader::get_reverse_reader(File::open(file).unwrap())
-                    .map_err(|_| ())?;
+            else {
+                log::error!("Invalid file detected, ignoring it: {:?}", file);
+                continue;
+            };
+
             readers.push(reader);
         }