remove unsused method that leaked in during merge

Author: Devdutt Shenoi

src/storage/object_storage.rs

@@ -549,83 +549,6 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         Ok(Bytes::new())
     }
 
-    async fn sync(&self, shutdown_signal: bool) -> Result<(), ObjectStorageError> {
-        if !Path::new(&PARSEABLE.staging_dir()).exists() {
-            return Ok(());
-        }
-
-        for stream_name in PARSEABLE.streams.list().iter() {
-            let stream = PARSEABLE.get_or_create_stream(stream_name);
-            let time_partition = stream.get_time_partition();
-            let custom_partition = stream.get_custom_partition();
-            if let Some(schema) = stream
-                .convert_disk_files_to_parquet(
-                    time_partition.as_ref(),
-                    custom_partition.as_ref(),
-                    shutdown_signal,
-                )
-                .map_err(|err| ObjectStorageError::UnhandledError(Box::new(err)))?
-            {
-                let static_schema_flag = stream.get_static_schema_flag();
-                if !static_schema_flag {
-                    commit_schema_to_storage(stream_name, schema).await?;
-                }
-            }
-
-            for file in stream.parquet_files() {
-                let filename = file
-                    .file_name()
-                    .expect("only parquet files are returned by iterator")
-                    .to_str()
-                    .expect("filename is valid string");
-
-                let mut file_date_part = filename.split('.').collect::<Vec<&str>>()[0];
-                file_date_part = file_date_part.split('=').collect::<Vec<&str>>()[1];
-                let compressed_size = file.metadata().map_or(0, |meta| meta.len());
-                STORAGE_SIZE
-                    .with_label_values(&["data", stream_name, "parquet"])
-                    .add(compressed_size as i64);
-                EVENTS_STORAGE_SIZE_DATE
-                    .with_label_values(&["data", stream_name, "parquet", file_date_part])
-                    .add(compressed_size as i64);
-                LIFETIME_EVENTS_STORAGE_SIZE
-                    .with_label_values(&["data", stream_name, "parquet"])
-                    .add(compressed_size as i64);
-                let mut file_suffix = str::replacen(filename, ".", "/", 3);
-
-                let custom_partition_clone = custom_partition.clone();
-                if custom_partition_clone.is_some() {
-                    let custom_partition_fields = custom_partition_clone.unwrap();
-                    let custom_partition_list =
-                        custom_partition_fields.split(',').collect::<Vec<&str>>();
-                    file_suffix =
-                        str::replacen(filename, ".", "/", 3 + custom_partition_list.len());
-                }
-
-                let stream_relative_path = format!("{stream_name}/{file_suffix}");
-
-                // Try uploading the file, handle potential errors without breaking the loop
-                if let Err(e) = self.upload_file(&stream_relative_path, &file).await {
-                    error!("Failed to upload file {}: {:?}", filename, e);
-                    continue; // Skip to the next file
-                }
-
-                let absolute_path = self
-                    .absolute_url(RelativePath::from_path(&stream_relative_path).unwrap())
-                    .to_string();
-                let store = PARSEABLE.storage().get_object_store();
-                let manifest =
-                    catalog::manifest::create_from_parquet_file(absolute_path.clone(), &file)
-                        .unwrap();
-                catalog::update_snapshot(store, stream_name, manifest).await?;
-
-                let _ = fs::remove_file(file);
-            }
-        }
-
-        Ok(())
-    }
-
     async fn get_stream_meta_from_storage(
         &self,
         stream_name: &str,