Merge pull request #7547 from Xuanwo/remove-meta-index

refactor: Always list from OpenDAL instead of meta

Author: mergify[bot]

src/meta/types/src/user_stage.rs

@@ -258,7 +258,7 @@ impl UserStageInfo {
     }
 }
 
-#[derive(Default, Clone)]
+#[derive(Default, Debug, Clone)]
 pub struct StageFile {
     pub path: String,
     pub size: u64,

src/query/service/src/interpreters/interpreter_common.rs

@@ -24,10 +24,10 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_types::GrantObject;
 use common_meta_types::StageFile;
-use common_meta_types::StageType;
 use common_meta_types::UserStageInfo;
 use futures::TryStreamExt;
 use regex::Regex;
+use tracing::debug;
 use tracing::warn;
 
 use crate::pipelines::executor::ExecutorSettings;
@@ -138,10 +138,7 @@ pub async fn list_files(
     path: &str,
     pattern: &str,
 ) -> Result<Vec<StageFile>> {
-    match stage.stage_type {
-        StageType::Internal => list_files_from_meta_api(ctx, stage, path, pattern).await,
-        StageType::External => list_files_from_dal(ctx, stage, path, pattern).await,
-    }
+    list_files_from_dal(ctx, stage, path, pattern).await
 }
 
 /// List files from DAL in recursive way.
@@ -155,8 +152,9 @@ pub async fn list_files_from_dal(
     path: &str,
     pattern: &str,
 ) -> Result<Vec<StageFile>> {
-    let rename_me_qry_ctx: Arc<dyn TableContext> = ctx.clone();
-    let op = StageSourceHelper::get_op(&rename_me_qry_ctx, stage).await?;
+    let table_ctx: Arc<dyn TableContext> = ctx.clone();
+    let op = StageSourceHelper::get_op(&table_ctx, stage).await?;
+    let prefix = stage.get_prefix();
     let mut files = Vec::new();
 
     // - If the path itself is a dir, return directly.
@@ -165,7 +163,7 @@ pub async fn list_files_from_dal(
     let dir_path = match op.object(path).metadata().await {
         Ok(meta) if meta.mode().is_dir() => Some(path.to_string()),
         Ok(meta) if !meta.mode().is_dir() => {
-            files.push((path.to_string(), meta));
+            files.push((path.trim_start_matches(&prefix).to_string(), meta));
 
             Some(format!("{path}/"))
         }
@@ -181,7 +179,7 @@ pub async fn list_files_from_dal(
                 let mut ds = op.batch().walk_top_down(&dir)?;
                 while let Some(de) = ds.try_next().await? {
                     if de.mode().is_file() {
-                        let path = de.path().to_string();
+                        let path = de.path().trim_start_matches(&prefix[1..]).to_string();
                         let meta = de.metadata().await?;
                         files.push((path, meta));
                     }
@@ -222,66 +220,7 @@ pub async fn list_files_from_dal(
             creator: None,
         })
         .collect::<Vec<StageFile>>();
-    Ok(matched_files)
-}
-
-pub async fn list_files_from_meta_api(
-    ctx: &Arc<QueryContext>,
-    stage: &UserStageInfo,
-    path: &str,
-    pattern: &str,
-) -> Result<Vec<StageFile>> {
-    let tenant = ctx.get_tenant();
-    let user_mgr = ctx.get_user_manager();
-    let prefix = stage.get_prefix();
-
-    if stage.number_of_files == 0 {
-        // try to sync files from dal
-        if let Ok(files) = list_files_from_dal(ctx, stage, &prefix, "").await {
-            for file in files.iter() {
-                let mut file = file.clone();
-                // In internal stage, files with `/stage/<stage_name>/` prefix will be ignored.
-                // TODO: prefix of internal stage should be as root path.
-                file.path = file
-                    .path
-                    .trim_start_matches(&prefix.trim_start_matches('/'))
-                    .to_string();
-                let _ = user_mgr.add_file(&tenant, &stage.stage_name, file).await;
-            }
-        }
-    }
-
-    let regex = if !pattern.is_empty() {
-        Some(Regex::new(pattern).map_err(|e| {
-            ErrorCode::SyntaxException(format!(
-                "Pattern format invalid, got:{}, error:{:?}",
-                pattern, e
-            ))
-        })?)
-    } else {
-        None
-    };
 
-    let files = user_mgr
-        .list_files(&tenant, &stage.stage_name)
-        .await?
-        .iter()
-        .filter(|file| {
-            let name = format!("{}{}", prefix, file.path);
-            if path.ends_with('/') {
-                name.starts_with(path)
-            } else {
-                name.starts_with(&format!("{path}/")) || name == path
-            }
-        })
-        .filter(|file| {
-            if let Some(regex) = &regex {
-                regex.is_match(&file.path)
-            } else {
-                true
-            }
-        })
-        .cloned()
-        .collect::<Vec<_>>();
-    Ok(files)
+    debug!("listed files: {:?}", matched_files);
+    Ok(matched_files)
 }

src/query/service/src/interpreters/interpreter_user_stage_remove.rs

@@ -15,11 +15,9 @@
 use std::sync::Arc;
 
 use common_exception::Result;
-use common_meta_types::StageType;
 use common_planners::RemoveUserStagePlan;
 use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
-use tracing::debug;
 
 use crate::interpreters::interpreter_common::list_files;
 use crate::interpreters::Interpreter;
@@ -48,23 +46,14 @@ impl Interpreter for RemoveUserStageInterpreter {
     #[tracing::instrument(level = "info", skip(self), fields(ctx.id = self.ctx.get_id().as_str()))]
     async fn execute(&self) -> Result<SendableDataBlockStream> {
         let plan = self.plan.clone();
-        let user_mgr = self.ctx.get_user_manager();
-        let tenant = self.ctx.get_tenant();
+        let prefix = plan.stage.get_prefix();
 
         let files = list_files(&self.ctx, &plan.stage, &plan.path, &plan.pattern).await?;
-        let files = files.iter().map(|f| f.path.clone()).collect::<Vec<_>>();
-        let rename_me: Arc<dyn TableContext> = self.ctx.clone();
-        let op = StageSourceHelper::get_op(&rename_me, &self.plan.stage).await?;
-        if plan.stage.stage_type == StageType::Internal {
-            user_mgr
-                .remove_files(&tenant, &plan.stage.stage_name, files.clone())
-                .await?;
-        }
+        let table_ctx: Arc<dyn TableContext> = self.ctx.clone();
+        let op = StageSourceHelper::get_op(&table_ctx, &self.plan.stage).await?;
 
-        for name in files.into_iter() {
-            let obj = format!("{}/{}", plan.stage.get_prefix(), name);
-            debug!("Removing object: {}", obj);
-            let _ = op.object(&obj).delete().await;
+        for name in files.iter().map(|f| f.path.as_str()) {
+            op.object(&format!("{prefix}{name}")).delete().await?;
         }
 
         Ok(Box::pin(DataBlockStream::create(

src/query/service/src/interpreters/mod.rs

@@ -109,7 +109,6 @@ pub use interpreter_cluster_key_drop::DropTableClusterKeyInterpreter;
 pub use interpreter_common::append2table;
 pub use interpreter_common::commit2table;
 pub use interpreter_common::list_files_from_dal;
-pub use interpreter_common::list_files_from_meta_api;
 pub use interpreter_database_create::CreateDatabaseInterpreter;
 pub use interpreter_database_drop::DropDatabaseInterpreter;
 pub use interpreter_database_rename::RenameDatabaseInterpreter;

src/query/service/src/procedures/systems/mod.rs

@@ -16,12 +16,10 @@ mod clustering_information;
 mod fuse_segment;
 mod fuse_snapshot;
 mod search_tables;
-mod sync_stage;
 mod system;
 
 pub use clustering_information::ClusteringInformationProcedure;
 pub use fuse_segment::FuseSegmentProcedure;
 pub use fuse_snapshot::FuseSnapshotProcedure;
 pub use search_tables::SearchTablesProcedure;
-pub use sync_stage::SyncStageFileProcedure;
 pub use system::SystemProcedure;

src/query/service/src/procedures/systems/sync_stage.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::SyncStageFileProcedure;
 use crate::procedures::systems::ClusteringInformationProcedure;
 use crate::procedures::systems::FuseSegmentProcedure;
 use crate::procedures::systems::FuseSnapshotProcedure;
@@ -39,9 +38,5 @@ impl SystemProcedure {
             "system$search_tables",
             Box::new(SearchTablesProcedure::try_create),
         );
-        factory.register(
-            "system$sync_stage_file",
-            Box::new(SyncStageFileProcedure::try_create),
-        )
     }
 }

src/query/service/src/procedures/systems/system.rs

@@ -14,11 +14,6 @@
 
 use std::sync::Arc;
 
-use chrono::DateTime;
-use chrono::TimeZone;
-use common_datavalues::chrono::Utc;
-use common_meta_types::StageFile;
-use common_meta_types::StageType;
 use poem::error::InternalServerError;
 use poem::error::Result as PoemResult;
 use poem::http::StatusCode;
@@ -85,12 +80,6 @@ pub async fn upload_to_stage(
 
     let prefix = stage.get_prefix();
 
-    let tenant = context.get_tenant();
-    let user = context
-        .get_current_user()
-        .map_err(InternalServerError)?
-        .identity();
-
     let mut files = vec![];
     while let Ok(Some(field)) = multipart.next_field().await {
         let name = match field.file_name() {
@@ -108,26 +97,6 @@ pub async fn upload_to_stage(
             .await
             .map_err(InternalServerError)?;
 
-        if stage.stage_type == StageType::Internal {
-            let meta = op
-                .object(&obj)
-                .metadata()
-                .await
-                .map_err(InternalServerError)?;
-            let file = StageFile {
-                path: file_path,
-                size: meta.content_length(),
-                md5: meta.content_md5().map(str::to_string),
-                last_modified: meta.last_modified().map_or(DateTime::default(), |t| {
-                    Utc.timestamp(t.unix_timestamp(), 0)
-                }),
-                creator: Some(user.clone()),
-            };
-            user_mgr
-                .add_file(&tenant, &stage.stage_name, file)
-                .await
-                .map_err(InternalServerError)?;
-        }
         files.push(name.clone());
     }
 

src/query/service/src/servers/http/v1/stage.rs

@@ -116,6 +116,8 @@ impl StageSourceHelper {
         ))
     }
 
+    /// TODO: we should support construct operator with
+    /// correct root.
     pub async fn get_op(ctx: &Arc<dyn TableContext>, stage: &UserStageInfo) -> Result<Operator> {
         if stage.stage_type == StageType::Internal {
             ctx.get_storage_operator()