feat: support vacuum leaked table data (#17022)-Revert (#17123)

* Revert "chore: add more log (#17110)"

This reverts commit 0e05b43.

* Revert "feat: support vacuum leaked table data (#17022)"

This reverts commit 3a9f404.

Author: dantengsky

src/query/ast/src/ast/statements/table.rs

@@ -789,7 +789,6 @@ pub struct VacuumDropTableOption {
     // Some(true) means dry run with summary option
     pub dry_run: Option<bool>,
     pub limit: Option<usize>,
-    pub force: bool,
 }
 
 impl Display for VacuumDropTableOption {
@@ -803,9 +802,6 @@ impl Display for VacuumDropTableOption {
         if let Some(limit) = self.limit {
             write!(f, " LIMIT {}", limit)?;
         }
-        if self.force {
-            write!(f, " FORCE")?;
-        }
         Ok(())
     }
 }

src/query/ast/src/parser/statement.rs

@@ -3777,12 +3777,11 @@ pub fn literal_duration(i: Input) -> IResult<Duration> {
 pub fn vacuum_drop_table_option(i: Input) -> IResult<VacuumDropTableOption> {
     alt((map(
         rule! {
-            (DRY ~ ^RUN ~ SUMMARY?)? ~ (LIMIT ~ #literal_u64)? ~ FORCE?
+            (DRY ~ ^RUN ~ SUMMARY?)? ~ (LIMIT ~ #literal_u64)?
         },
-        |(opt_dry_run, opt_limit, opt_force)| VacuumDropTableOption {
+        |(opt_dry_run, opt_limit)| VacuumDropTableOption {
             dry_run: opt_dry_run.map(|dry_run| dry_run.2.is_some()),
             limit: opt_limit.map(|(_, limit)| limit as usize),
-            force: opt_force.is_some(),
         },
     ),))(i)
 }

src/query/ast/tests/it/testdata/stmt.txt

@@ -13016,7 +13016,6 @@ VacuumDropTable(
         option: VacuumDropTableOption {
             dry_run: None,
             limit: None,
-            force: false,
         },
     },
 )
@@ -13036,7 +13035,6 @@ VacuumDropTable(
                 false,
             ),
             limit: None,
-            force: false,
         },
     },
 )
@@ -13056,7 +13054,6 @@ VacuumDropTable(
                 true,
             ),
             limit: None,
-            force: false,
         },
     },
 )
@@ -13083,7 +13080,6 @@ VacuumDropTable(
         option: VacuumDropTableOption {
             dry_run: None,
             limit: None,
-            force: false,
         },
     },
 )
@@ -13112,7 +13108,6 @@ VacuumDropTable(
             limit: Some(
                 10,
             ),
-            force: false,
         },
     },
 )

src/query/service/src/interpreters/interpreter_vacuum_drop_tables.rs

@@ -31,10 +31,8 @@ use databend_common_meta_app::schema::DroppedId;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::ListDroppedTableReq;
 use databend_common_sql::plans::VacuumDropTablePlan;
-use databend_common_storage::DataOperator;
 use databend_common_storages_view::view_table::VIEW_ENGINE;
 use databend_enterprise_vacuum_handler::get_vacuum_handler;
-use futures_util::TryStreamExt;
 use log::info;
 
 use crate::interpreters::Interpreter;
@@ -118,11 +116,6 @@ impl Interpreter for VacuumDropTablesInterpreter {
         LicenseManagerSwitch::instance()
             .check_enterprise_enabled(self.ctx.get_license_key(), Vacuum)?;
 
-        if self.plan.option.force {
-            self.vacuum_drop_tables_force().await?;
-            return Ok(PipelineBuildResult::create());
-        }
-
         let ctx = self.ctx.clone();
         let duration = Duration::days(ctx.get_settings().get_data_retention_time_in_days()? as i64);
 
@@ -140,7 +133,6 @@ impl Interpreter for VacuumDropTablesInterpreter {
         };
 
         let tenant = self.ctx.get_tenant();
-
         let (tables, drop_ids) = catalog
             .get_drop_table_infos(ListDroppedTableReq::new4(
                 &tenant,
@@ -283,62 +275,3 @@ impl Interpreter for VacuumDropTablesInterpreter {
         }
     }
 }
-
-impl VacuumDropTablesInterpreter {
-    async fn vacuum_drop_tables_force(&self) -> Result<()> {
-        let catalog = self.ctx.get_catalog(self.plan.catalog.as_str()).await?;
-        let op = DataOperator::instance().operator();
-        let databases = match self.plan.database.is_empty() {
-            true => catalog.list_databases(&self.ctx.get_tenant()).await?,
-            false => {
-                let database = catalog
-                    .get_database(&self.ctx.get_tenant(), &self.plan.database)
-                    .await?;
-                vec![database]
-            }
-        };
-
-        for database in databases {
-            if database.name() == "system" || database.name() == "information_schema" {
-                continue;
-            }
-            let db_id = database.get_db_info().database_id.db_id;
-            info!(
-                "vacuum drop table force from db name: {}, id: {}",
-                database.name(),
-                db_id
-            );
-            let mut lister = op.lister_with(&db_id.to_string()).recursive(true).await?;
-            let mut paths = vec![];
-            let mut orphan_paths = vec![];
-            while let Some(entry) = lister.try_next().await? {
-                paths.push(entry.path().to_string());
-            }
-            let tables_in_meta = database.list_tables_history().await?;
-            let table_ids_in_meta = tables_in_meta
-                .iter()
-                .map(|t| t.get_id())
-                .collect::<HashSet<_>>();
-            info!("table_ids_in_meta: {:?}", table_ids_in_meta);
-            for path in paths {
-                let Some(table_id) = path.split('/').nth(1) else {
-                    info!("can not parse table id from path: {}", path);
-                    continue;
-                };
-                info!("split table id:{} from path: {}", table_id, path);
-                let Some(table_id) = table_id.parse::<u64>().ok() else {
-                    info!("can not parse table id from path: {}", path);
-                    continue;
-                };
-                info!("parse table id:{} from path: {}", table_id, path);
-                if !table_ids_in_meta.contains(&table_id) {
-                    orphan_paths.push(path);
-                }
-            }
-            info!("orphan_paths summary: {:?}", orphan_paths);
-            op.remove(orphan_paths).await?;
-        }
-
-        Ok(())
-    }
-}

src/query/sql/src/planner/binder/ddl/table.rs

@@ -1287,7 +1287,6 @@ impl Binder {
             VacuumDropTableOption {
                 dry_run: option.dry_run,
                 limit: option.limit,
-                force: option.force,
             }
         };
         Ok(Plan::VacuumDropTable(Box::new(VacuumDropTablePlan {

src/query/sql/src/planner/plans/ddl/table.rs

@@ -188,7 +188,6 @@ pub struct VacuumDropTableOption {
     // Some(true) means dry run with summary option
     pub dry_run: Option<bool>,
     pub limit: Option<usize>,
-    pub force: bool,
 }
 
 #[derive(Debug, Clone)]