fix: fix create or replace stage bug

src/query/service/src/interpreters/interpreter_user_stage_create.rs

@@ -20,11 +20,14 @@ use databend_common_exception::Result;
 use databend_common_management::RoleApi;
 use databend_common_meta_app::principal::OwnershipObject;
 use databend_common_meta_app::principal::StageType;
+use databend_common_meta_app::schema::CreateOption;
 use databend_common_meta_types::MatchSeq;
 use databend_common_sql::plans::CreateStagePlan;
+use databend_common_storages_stage::StageTable;
 use databend_common_users::RoleCacheManager;
 use databend_common_users::UserApiProvider;
 use log::debug;
+use log::info;
 
 use crate::interpreters::Interpreter;
 use crate::pipelines::PipelineBuildResult;
@@ -79,18 +82,26 @@ impl Interpreter for CreateUserStageInterpreter {
             )));
         };
 
-        if user_stage.stage_type != StageType::External {
-            let op = self.ctx.get_data_operator()?.operator();
-            op.create_dir(&user_stage.stage_prefix()).await?
-        }
-
         let mut user_stage = user_stage;
         user_stage.creator = Some(self.ctx.get_current_user()?.identity());
         user_stage.created_on = Utc::now();
         let _ = user_mgr
-            .add_stage(&plan.tenant, user_stage, &plan.create_option)
+            .add_stage(&plan.tenant, user_stage.clone(), &plan.create_option)
             .await?;
 
+        if user_stage.stage_type != StageType::External {
+            if let CreateOption::CreateOrReplace = plan.create_option {
+                let op = StageTable::get_op(&user_stage)?;
+                op.remove_all("/").await?;
+                info!(
+                    "create or replace stage {:?} with all objects removed in stage",
+                    user_stage.stage_name
+                );
+            }
+            let op = self.ctx.get_data_operator()?.operator();
+            op.create_dir(&user_stage.stage_prefix()).await?
+        }
+
         // Grant ownership as the current role
         let tenant = self.ctx.get_tenant();
         let role_api = UserApiProvider::instance().get_role_api_client(&tenant)?;

tests/sqllogictests/suites/base/05_ddl/05_0016_ddl_stage.test

@@ -52,9 +52,37 @@ CREATE OR REPLACE STAGE if not exists replace_test_stage_huggingface url='hf://o
 statement ok
 CREATE OR REPLACE STAGE replace_test_stage_huggingface url='hf://opendal/huggingface-testdata/'
 
+query I
+select count(*) > 0 from replace_test_stage_huggingface;
+----
+1
+
 statement ok
 DROP STAGE replace_test_stage_huggingface
 
 statement ok
 SHOW STAGES
 
+statement ok
+create stage s;
+
+statement ok
+create table t(c int);
+
+statement ok
+insert into t values(1);
+
+statement ok
+copy into @s from t;
+
+statement ok
+create or replace stage s;
+
+statement error 1006.*no file found
+select * from @s;
+
+statement ok
+DROP STAGE s;
+
+statement ok
+drop table t;