Add optional manifest_url parameter

Author: parmesant

src/catalog/mod.rs

@@ -307,6 +307,7 @@ async fn handle_existing_partition(
                 stream_name,
                 manifests[pos].time_lower_bound,
                 manifests[pos].time_upper_bound,
+                Some(manifests[pos].manifest_path.clone()),
             )
             .await
             .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?

src/enterprise/utils.rs

@@ -100,6 +100,7 @@ pub async fn fetch_parquet_file_paths(
                     stream,
                     manifest_item.time_lower_bound,
                     manifest_item.time_upper_bound,
+                    Some(manifest_item.manifest_path),
                 )
                 .await
                 .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?

src/metastore/metastore_traits.rs

@@ -101,6 +101,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
         stream_name: &str,
         lower_bound: DateTime<Utc>,
         upper_bound: DateTime<Utc>,
+        manifest_url: Option<String>,
     ) -> Result<Option<Manifest>, MetastoreError>;
     async fn put_manifest(
         &self,

src/metastore/metastores/object_store_metastore.rs

@@ -414,9 +414,15 @@ impl Metastore for ObjectStoreMetastore {
         stream_name: &str,
         lower_bound: DateTime<Utc>,
         upper_bound: DateTime<Utc>,
+        manifest_url: Option<String>,
     ) -> Result<Option<Manifest>, MetastoreError> {
-        let path = partition_path(stream_name, lower_bound, upper_bound);
-        let path = manifest_path(path.as_str());
+        let path = match manifest_url {
+            Some(url) => RelativePathBuf::from(url),
+            None => {
+                let path = partition_path(stream_name, lower_bound, upper_bound);
+                manifest_path(path.as_str())
+            }
+        };
         match self.storage.get_object(&path).await {
             Ok(bytes) => {
                 let manifest = serde_json::from_slice(&bytes)?;
@@ -425,6 +431,18 @@ impl Metastore for ObjectStoreMetastore {
             Err(ObjectStorageError::NoSuchKey(_)) => Ok(None),
             Err(err) => Err(MetastoreError::ObjectStorageError(err)),
         }
+        // let path = partition_path(stream_name, lower_bound, upper_bound);
+        // // // need a 'ends with `manifest.json` condition here'
+        // // let obs = self
+        // //     .storage
+        // //     .get_objects(
+        // //         path,
+        // //         Box::new(|file_name| file_name.ends_with("manifest.json")),
+        // //     )
+        // //     .await?;
+        // warn!(partition_path=?path);
+        // let path = manifest_path(path.as_str());
+        // warn!(manifest_path=?path);
     }
 
     /// Get the path for a specific `Manifest` file

src/metastore/mod.rs

@@ -148,25 +148,13 @@ impl MetastoreError {
 
     pub fn status_code(&self) -> StatusCode {
         match self {
-            MetastoreError::ObjectStorageError(_object_storage_error) => {
-                StatusCode::INTERNAL_SERVER_ERROR
-            }
-            MetastoreError::JsonParseError(_error) => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::JsonSchemaError { message: _ } => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::InvalidJsonStructure {
-                expected: _,
-                found: _,
-            } => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::MissingJsonField { field: _ } => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::InvalidJsonValue {
-                field: _,
-                reason: _,
-            } => StatusCode::INTERNAL_SERVER_ERROR,
-            MetastoreError::Error {
-                status_code,
-                message: _,
-                flow: _,
-            } => *status_code,
+            MetastoreError::ObjectStorageError(..) => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::JsonParseError(..) => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::JsonSchemaError { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::InvalidJsonStructure { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::MissingJsonField { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::InvalidJsonValue { .. } => StatusCode::INTERNAL_SERVER_ERROR,
+            MetastoreError::Error { status_code, .. } => *status_code,
         }
     }
 }

src/query/mod.rs

@@ -43,6 +43,7 @@ use std::ops::Bound;
 use std::sync::Arc;
 use sysinfo::System;
 use tokio::runtime::Runtime;
+use tracing::warn;
 
 use self::error::ExecuteError;
 use self::stream_schema_provider::GlobalSchemaProvider;
@@ -546,15 +547,20 @@ pub async fn get_manifest_list(
         PartialTimeFilter::High(Bound::Included(time_range.end.naive_utc())),
     ];
 
+    warn!(merged_snapshot=?merged_snapshot);
+    warn!(time_filter=?time_filter);
+
     let mut all_manifest_files = Vec::new();
     for manifest_item in merged_snapshot.manifests(&time_filter) {
+        warn!(manifest_item=?manifest_item);
         all_manifest_files.push(
             PARSEABLE
                 .metastore
                 .get_manifest(
                     stream_name,
                     manifest_item.time_lower_bound,
                     manifest_item.time_upper_bound,
+                    Some(manifest_item.manifest_path),
                 )
                 .await?
                 .expect("Data is invalid for Manifest"),

src/query/stream_schema_provider.rs

@@ -421,6 +421,7 @@ async fn collect_from_snapshot(
                     stream_name,
                     manifest_item.time_lower_bound,
                     manifest_item.time_upper_bound,
+                    Some(manifest_item.manifest_path),
                 )
                 .await
                 .map_err(|e| DataFusionError::Plan(e.to_string()))?