refactor: Always list from OpenDAL instead of meta

Signed-off-by: Xuanwo <github@xuanwo.io>

Author: Xuanwo

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

@@ -24,7 +24,6 @@ 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;
@@ -138,10 +137,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.
@@ -224,64 +220,3 @@ pub async fn list_files_from_dal(
         .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)
-}

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

@@ -15,7 +15,6 @@
 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;
@@ -48,20 +47,12 @@ 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 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?;
-        }
 
-        for name in files.into_iter() {
+        for name in files.iter().map(|f| f.path.as_str()) {
             let obj = format!("{}/{}", plan.stage.get_prefix(), name);
             debug!("Removing object: {}", obj);
             let _ = op.object(&obj).delete().await;

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());
     }