Fix test

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

Author: Xuanwo

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

@@ -27,6 +27,7 @@ use common_meta_types::StageFile;
 use common_meta_types::UserStageInfo;
 use futures::TryStreamExt;
 use regex::Regex;
+use tracing::debug;
 use tracing::warn;
 
 use crate::pipelines::executor::ExecutorSettings;
@@ -151,8 +152,8 @@ 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();
 
@@ -162,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}/"))
         }
@@ -219,5 +220,7 @@ pub async fn list_files_from_dal(
             creator: None,
         })
         .collect::<Vec<StageFile>>();
+
+    debug!("listed files: {:?}", matched_files);
     Ok(matched_files)
 }

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

@@ -18,7 +18,6 @@ use common_exception::Result;
 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;
@@ -47,15 +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 prefix = plan.stage.get_prefix();
 
         let files = list_files(&self.ctx, &plan.stage, &plan.path, &plan.pattern).await?;
-        let rename_me: Arc<dyn TableContext> = self.ctx.clone();
-        let op = StageSourceHelper::get_op(&rename_me, &self.plan.stage).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.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;
+            op.object(&format!("{prefix}{name}")).delete().await?;
         }
 
         Ok(Box::pin(DataBlockStream::create(

src/query/service/src/storages/stage/stage_source.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()