Add integration test

Author: dantengsky

Cargo.lock

@@ -66,6 +66,7 @@ uuid = { workspace = true }
 
 [dev-dependencies]
 databend-common-functions = { workspace = true }
+goldenfile = { workspace = true }
 jsonb = { workspace = true }
 tantivy = { workspace = true }
 walkdir = { workspace = true }

src/query/ee/Cargo.toml

@@ -17,23 +17,84 @@ use std::path::Path;
 use databend_common_base::base::tokio;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::block_debug::pretty_format_blocks;
+use databend_common_expression::DataBlock;
 use databend_enterprise_query::test_kits::context::EESetup;
 use databend_query::sessions::QueryContext;
 use databend_query::sessions::TableContext;
 use databend_query::test_kits::TestFixture;
+use futures_util::TryStreamExt;
+use goldenfile::Mint;
 
 // TODO investigate this
 // NOTE: SHOULD specify flavor = "multi_thread", otherwise query execution might be hanged
 #[tokio::test(flavor = "multi_thread")]
-async fn test_vacuum2_all() -> Result<()> {
+async fn test_table_function_fuse_vacuum2_all() -> Result<()> {
     let ee_setup = EESetup::new();
     let fixture = TestFixture::setup_with_custom(ee_setup).await?;
+
+    setup(&fixture).await?;
+
+
+    // vacuum them all
+    let res = fixture.execute_command("call system$fuse_vacuum2()").await;
+
+    // Check that:
+
+    // 1. non-fuse tables should not stop us
+
+    assert!(res.is_ok());
+
+    //  2. fuse table data should be vacuumed
+
+    let storage_root = fixture.storage_root();
+
+    let ctx = fixture.new_query_ctx().await?;
+    check_files_left(&ctx, storage_root, "db1", "t1").await?;
+    check_files_left(&ctx, storage_root, "default", "t1").await?;
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_vacuum_all_stmt() -> Result<()> {
+    let ee_setup = EESetup::new();
+    let fixture = TestFixture::setup_with_custom(ee_setup).await?;
+
+    setup(&fixture).await?;
+
+    let mut mint = Mint::new("tests/it/testdata");
+    let file = &mut mint.new_goldenfile("vacuum_all_stmt.txt").unwrap();
+
+    // execute the VACUUM TABLE ALL
+    let res_block = fixture
+        .execute_query("vacuum table all")
+        .await?
+        .try_collect::<Vec<DataBlock>>()
+        .await?;
+
+    let block_string = pretty_format_blocks(&res_block).unwrap();
+    use std::io::Write;
+    writeln!(file, "{}", block_string).unwrap();
+
+    //  Check: fuse table data should be vacuumed
+
+    let storage_root = fixture.storage_root();
+
+    let ctx = fixture.new_query_ctx().await?;
+    check_files_left(&ctx, storage_root, "db1", "t1").await?;
+    check_files_left(&ctx, storage_root, "default", "t1").await?;
+
+    Ok(())
+}
+
+async fn setup(fixture: &TestFixture) -> Result<()>{
+
     // Adjust retention period to 0, so that dropped tables will be vacuumed immediately
     let session = fixture.default_session();
     session.get_settings().set_data_retention_time_in_days(0)?;
 
-    let ctx = fixture.new_query_ctx().await?;
-
+    // Prepare test db / tables
     let setup_statements = vec![
         // create non-system db1, create fuse and non-fuse table in it.
         "create database db1",
@@ -53,67 +114,49 @@ async fn test_vacuum2_all() -> Result<()> {
     for stmt in setup_statements {
         fixture.execute_command(stmt).await?;
     }
-
-    // vacuum them all
-    let res = fixture.execute_command("call system$fuse_vacuum2()").await;
-
-    // Check that:
-
-    // 1. non-fuse tables should not stop us
-
-    assert!(res.is_ok());
-
-    //  2. fuse table data should be vacuumed
-
-    let storage_root = fixture.storage_root();
-
-    async fn check_files_left(
-        ctx: &QueryContext,
-        storage_root: &str,
-        db_name: &str,
-        tbl_name: &str,
-    ) -> Result<()> {
-        let tenant = ctx.get_tenant();
-        let table = ctx
-            .get_default_catalog()?
-            .get_table(&tenant, db_name, tbl_name)
-            .await?;
-
-        let db = ctx
-            .get_default_catalog()?
-            .get_database(&tenant, db_name)
-            .await?;
-
-        let path = Path::new(storage_root)
-            .join(db.get_db_info().database_id.db_id.to_string())
-            .join(table.get_id().to_string());
-
-        let walker = walkdir::WalkDir::new(path).into_iter();
-
-        let mut files_left = Vec::new();
-        for entry in walker {
-            let entry = entry.unwrap();
-            if entry.file_type().is_file() {
-                files_left.push(entry);
-            }
+    Ok(())
+}
+async fn check_files_left(
+    ctx: &QueryContext,
+    storage_root: &str,
+    db_name: &str,
+    tbl_name: &str,
+) -> Result<()> {
+    let tenant = ctx.get_tenant();
+    let table = ctx
+        .get_default_catalog()?
+        .get_table(&tenant, db_name, tbl_name)
+        .await?;
+
+    let db = ctx
+        .get_default_catalog()?
+        .get_database(&tenant, db_name)
+        .await?;
+
+    let path = Path::new(storage_root)
+        .join(db.get_db_info().database_id.db_id.to_string())
+        .join(table.get_id().to_string());
+
+    let walker = walkdir::WalkDir::new(path).into_iter();
+
+    let mut files_left = Vec::new();
+    for entry in walker {
+        let entry = entry.unwrap();
+        if entry.file_type().is_file() {
+            files_left.push(entry);
         }
-
-        // There should be one snapshot file and one snapshot hint file left
-        assert_eq!(files_left.len(), 2);
-
-        files_left.sort_by(|a, b| a.file_name().cmp(b.file_name()));
-        // First is the only snapshot left
-        files_left[0].path().to_string_lossy().contains("/_ss/");
-        // Second one is the last snapshot location hint
-        files_left[1]
-            .path()
-            .to_string_lossy()
-            .contains("last_snapshot_location_hint_v2");
-        Ok::<(), ErrorCode>(())
     }
 
-    check_files_left(&ctx, storage_root, "db1", "t1").await?;
-    check_files_left(&ctx, storage_root, "default", "t1").await?;
-
-    Ok(())
+    // There should be one snapshot file and one snapshot hint file left
+    assert_eq!(files_left.len(), 2);
+
+    files_left.sort_by(|a, b| a.file_name().cmp(b.file_name()));
+    // First is the only snapshot left
+    files_left[0].path().to_string_lossy().contains("/_ss/");
+    // Second one is the last snapshot location hint
+    files_left[1]
+        .path()
+        .to_string_lossy()
+        .contains("last_snapshot_location_hint_v2");
+    Ok::<(), ErrorCode>(())
 }

src/query/ee/tests/it/storages/fuse/operations/vacuum2.rs

@@ -0,0 +1,5 @@
++----------+
+| Column 0 |
++----------+
+| 20       |
++----------+

src/query/ee/tests/it/testdata/vacuum_all_stmt.txt

@@ -130,7 +130,7 @@ impl VacuumTablePlan {
     pub fn schema(&self) -> DataSchemaRef {
         if !self.use_legacy_vacuum {
             return Arc::new(DataSchema::new(vec![DataField::new(
-                "num_object_removed",
+                "num_of_objects_removed",
                 DataType::Number(NumberDataType::UInt64),
             )]));
         }