Merge pull request #9318 from dantengsky/feat-purge-before

feat: purge at given point

Author: mergify[bot]

src/query/ast/src/ast/format/ast_format.rs

@@ -1377,11 +1377,9 @@ impl<'ast> Visitor<'ast> for AstFormatVisitor {
         let mut children = Vec::new();
         self.visit_table_ref(&stmt.catalog, &stmt.database, &stmt.table);
         children.push(self.children.pop().unwrap());
-        if let Some(action) = &stmt.action {
-            let action_name = format!("Action {}", action);
-            let action_format_ctx = AstFormatContext::new(action_name);
-            children.push(FormatTreeNode::new(action_format_ctx));
-        }
+        let action_name = format!("Action {}", stmt.action);
+        let action_format_ctx = AstFormatContext::new(action_name);
+        children.push(FormatTreeNode::new(action_format_ctx));
 
         let name = "OptimizeTable".to_string();
         let format_ctx = AstFormatContext::with_children(name, children.len());

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

@@ -364,7 +364,7 @@ impl Display for RenameTableStmt<'_> {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct TruncateTableStmt<'a> {
     pub catalog: Option<Identifier<'a>>,
     pub database: Option<Identifier<'a>>,
@@ -395,7 +395,7 @@ pub struct OptimizeTableStmt<'a> {
     pub catalog: Option<Identifier<'a>>,
     pub database: Option<Identifier<'a>>,
     pub table: Identifier<'a>,
-    pub action: Option<OptimizeTableAction<'a>>,
+    pub action: OptimizeTableAction<'a>,
 }
 
 impl Display for OptimizeTableStmt<'_> {
@@ -408,9 +408,7 @@ impl Display for OptimizeTableStmt<'_> {
                 .chain(&self.database)
                 .chain(Some(&self.table)),
         )?;
-        if let Some(action) = &self.action {
-            write!(f, " {action}")?;
-        }
+        write!(f, " {}", &self.action)?;
 
         Ok(())
     }
@@ -488,7 +486,9 @@ pub enum CompactTarget {
 #[derive(Debug, Clone, PartialEq)]
 pub enum OptimizeTableAction<'a> {
     All,
-    Purge,
+    Purge {
+        before: Option<TimeTravelPoint<'a>>,
+    },
     Compact {
         target: CompactTarget,
         limit: Option<Expr<'a>>,
@@ -499,7 +499,13 @@ impl<'a> Display for OptimizeTableAction<'a> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             OptimizeTableAction::All => write!(f, "ALL"),
-            OptimizeTableAction::Purge => write!(f, "PURGE"),
+            OptimizeTableAction::Purge { before } => {
+                write!(f, "PURGE")?;
+                if let Some(point) = before {
+                    write!(f, " BEFORE {}", point)?;
+                }
+                Ok(())
+            }
             OptimizeTableAction::Compact { target, limit } => {
                 match target {
                     CompactTarget::Block => {

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

@@ -516,7 +516,7 @@ pub fn statement(i: Input) -> IResult<StatementMsg> {
     );
     let optimize_table = map(
         rule! {
-            OPTIMIZE ~ TABLE ~ #peroid_separated_idents_1_to_3 ~ #optimize_table_action?
+            OPTIMIZE ~ TABLE ~ #peroid_separated_idents_1_to_3 ~ #optimize_table_action
         },
         |(_, _, (catalog, database, table), action)| {
             Statement::OptimizeTable(OptimizeTableStmt {
@@ -1483,7 +1483,12 @@ pub fn alter_table_action(i: Input) -> IResult<AlterTableAction> {
 pub fn optimize_table_action(i: Input) -> IResult<OptimizeTableAction> {
     alt((
         value(OptimizeTableAction::All, rule! { ALL }),
-        value(OptimizeTableAction::Purge, rule! { PURGE }),
+        map(
+            rule! { PURGE ~ (BEFORE ~ #travel_point)?},
+            |(_, opt_travel_point)| OptimizeTableAction::Purge {
+                before: opt_travel_point.map(|(_, p)| p),
+            },
+        ),
         map(
             rule! { COMPACT ~ (SEGMENT)? ~ ( LIMIT ~ ^#expr )?},
             |(_, opt_segment, opt_limit)| OptimizeTableAction::Compact {

src/query/ast/src/parser/token.rs

@@ -269,6 +269,8 @@ pub enum TokenKind {
     AWS_SECRET_KEY,
     #[token("ANTI", ignore(ascii_case))]
     ANTI,
+    #[token("BEFORE", ignore(ascii_case))]
+    BEFORE,
     #[token("BETWEEN", ignore(ascii_case))]
     BETWEEN,
     #[token("BIGINT", ignore(ascii_case))]

src/query/catalog/src/table.rs

@@ -320,7 +320,7 @@ pub trait TableExt: Table {
 
 impl<T: ?Sized> TableExt for T where T: Table {}
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Eq, PartialEq)]
 pub enum NavigationPoint {
     SnapshotID(String),
     TimePoint(DateTime<Utc>),

src/query/service/src/interpreters/interpreter_table_optimize.rs

@@ -55,7 +55,7 @@ impl Interpreter for OptimizeTableInterpreter {
         let action = &plan.action;
         let do_purge = matches!(
             action,
-            OptimizeTableAction::Purge | OptimizeTableAction::All
+            OptimizeTableAction::Purge(_) | OptimizeTableAction::All
         );
         let do_compact_blocks = matches!(
             action,
@@ -114,7 +114,13 @@ impl Interpreter for OptimizeTableInterpreter {
         }
 
         if do_purge {
-            table.purge(self.ctx.clone(), true).await?;
+            let table = if let OptimizeTableAction::Purge(Some(point)) = action {
+                table.navigate_to(point).await?
+            } else {
+                table
+            };
+            let keep_latest = true;
+            table.purge(self.ctx.clone(), keep_latest).await?;
         }
 
         Ok(PipelineBuildResult::create())

src/query/service/tests/it/storages/fuse/operations/navigate.rs

@@ -12,21 +12,17 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
-use std::ops::Add;
 use std::ops::Sub;
 use std::time::Duration;
 
 use common_base::base::tokio;
-use common_catalog::table::AppendMode;
 use common_datablocks::DataBlock;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_storages_fuse::io::SnapshotHistoryReader;
-use databend_query::pipelines::Pipeline;
 use databend_query::storages::fuse::io::MetaReaders;
 use databend_query::storages::fuse::io::TableMetaLocationGenerator;
 use databend_query::storages::fuse::FuseTable;
-use databend_query::storages::Table;
 use futures::TryStreamExt;
 
 use crate::storages::fuse::table_test_fixture::execute_query;
@@ -121,73 +117,3 @@ async fn test_fuse_navigate() -> Result<()> {
     };
     Ok(())
 }
-
-#[tokio::test]
-async fn test_fuse_historical_table_is_read_only() -> Result<()> {
-    // 1. Setup
-    let fixture = TestFixture::new().await;
-    let db = fixture.default_db_name();
-    let tbl = fixture.default_table_name();
-    let ctx = fixture.ctx();
-    fixture.create_default_table().await?;
-
-    let qry = format!("insert into {}.{} values (1, (2, 3))", db, tbl);
-    execute_query(ctx.clone(), qry.as_str())
-        .await?
-        .try_collect::<Vec<DataBlock>>()
-        .await?;
-
-    // 2. grab the history
-    let table = fixture.latest_default_table().await?;
-    let fuse_table = FuseTable::try_from_table(table.as_ref())?;
-    let loc = fuse_table.snapshot_loc().await?.unwrap();
-    let reader = MetaReaders::table_snapshot_reader(fuse_table.get_operator());
-    let version = TableMetaLocationGenerator::snapshot_version(loc.as_str());
-    let snapshots: Vec<_> = reader
-        .snapshot_history(loc, version, fuse_table.meta_location_generator().clone())
-        .try_collect()
-        .await?;
-
-    let snapshot = &snapshots[0];
-    let instant = snapshot
-        .timestamp
-        .unwrap()
-        .add(chrono::Duration::milliseconds(1));
-    let tbl = fuse_table.navigate_to_time_point(instant).await?;
-
-    // check append2
-    let res = tbl.append_data(
-        ctx.clone(),
-        &mut Pipeline::create(),
-        AppendMode::Normal,
-        false,
-    );
-    assert_not_writable(res, "append2");
-
-    // check append_data
-    let res = fixture
-        .append_commit_blocks(tbl.clone(), vec![], false, true)
-        .await;
-    assert_not_writable(res, "append2");
-
-    // check truncate
-    let res = tbl.truncate(ctx.clone(), false).await;
-    assert_not_writable(res, "truncate");
-
-    Ok(())
-}
-
-fn assert_not_writable<T>(res: Result<T>, case_name: &str) {
-    match res {
-        Ok(_) => panic!(
-            "historical table should NOT be writable, case {}",
-            case_name
-        ),
-        Err(e) => assert_eq!(
-            e.code(),
-            ErrorCode::TABLE_NOT_WRITABLE,
-            " case {}",
-            case_name
-        ),
-    }
-}

src/query/service/tests/it/storages/fuse/operations/optimize.rs

@@ -23,25 +23,10 @@ use crate::storages::fuse::table_test_fixture::TestFixture;
 use crate::storages::fuse::utils::do_purge_test;
 use crate::storages::fuse::utils::TestTableOperation;
 
-#[tokio::test]
-async fn test_fuse_snapshot_optimize() -> Result<()> {
-    do_purge_test(
-        "implicit pure",
-        TestTableOperation::Optimize("".to_string()),
-        1,
-        0,
-        1,
-        1,
-        1,
-        None,
-    )
-    .await
-}
-
 #[tokio::test]
 async fn test_fuse_snapshot_optimize_purge() -> Result<()> {
     do_purge_test(
-        "explicit pure",
+        "explicit purge",
         TestTableOperation::Optimize("purge".to_string()),
         1,
         0,
@@ -56,7 +41,7 @@ async fn test_fuse_snapshot_optimize_purge() -> Result<()> {
 #[tokio::test]
 async fn test_fuse_snapshot_optimize_all() -> Result<()> {
     do_purge_test(
-        "explicit pure",
+        "explicit purge",
         TestTableOperation::Optimize("all".to_string()),
         1,
         0,

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

@@ -160,7 +160,7 @@ impl<'a> Binder {
             Statement::AlterTable(stmt) => self.bind_alter_table(bind_context, stmt).await?,
             Statement::RenameTable(stmt) => self.bind_rename_table(stmt).await?,
             Statement::TruncateTable(stmt) => self.bind_truncate_table(stmt).await?,
-            Statement::OptimizeTable(stmt) => self.bind_optimize_table(stmt).await?,
+            Statement::OptimizeTable(stmt) => self.bind_optimize_table(bind_context, stmt).await?,
             Statement::AnalyzeTable(stmt) => self.bind_analyze_table(stmt).await?,
             Statement::ExistsTable(stmt) => self.bind_exists_table(stmt).await?,
 

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

@@ -739,13 +739,14 @@ impl<'a> Binder {
 
     pub(in crate::planner::binder) async fn bind_optimize_table(
         &mut self,
+        bind_context: &BindContext,
         stmt: &OptimizeTableStmt<'a>,
     ) -> Result<Plan> {
         let OptimizeTableStmt {
             catalog,
             database,
             table,
-            action,
+            action: ast_action,
         } = stmt;
 
         let catalog = catalog
@@ -757,29 +758,33 @@ impl<'a> Binder {
             .map(|ident| normalize_identifier(ident, &self.name_resolution_ctx).name)
             .unwrap_or_else(|| self.ctx.get_current_database());
         let table = normalize_identifier(table, &self.name_resolution_ctx).name;
-        let action = if let Some(ast_action) = action {
-            match ast_action {
-                AstOptimizeTableAction::All => OptimizeTableAction::All,
-                AstOptimizeTableAction::Purge => OptimizeTableAction::Purge,
-                AstOptimizeTableAction::Compact { target, limit } => {
-                    let limit_cnt = match limit {
-                        Some(Expr::Literal {
-                            lit: Literal::Integer(uint),
-                            ..
-                        }) => Some(*uint as usize),
-                        Some(_) => {
-                            return Err(ErrorCode::IllegalDataType("Unsupported limit type"));
-                        }
-                        _ => None,
-                    };
-                    match target {
-                        CompactTarget::Block => OptimizeTableAction::CompactBlocks(limit_cnt),
-                        CompactTarget::Segment => OptimizeTableAction::CompactSegments(limit_cnt),
+        let action = match ast_action {
+            AstOptimizeTableAction::All => OptimizeTableAction::All,
+            AstOptimizeTableAction::Purge { before } => {
+                let p = if let Some(point) = before {
+                    let point = self.resolve_data_travel_point(bind_context, point).await?;
+                    Some(point)
+                } else {
+                    None
+                };
+                OptimizeTableAction::Purge(p)
+            }
+            AstOptimizeTableAction::Compact { target, limit } => {
+                let limit_cnt = match limit {
+                    Some(Expr::Literal {
+                        lit: Literal::Integer(uint),
+                        ..
+                    }) => Some(*uint as usize),
+                    Some(_) => {
+                        return Err(ErrorCode::IllegalDataType("Unsupported limit type"));
                     }
+                    _ => None,
+                };
+                match target {
+                    CompactTarget::Block => OptimizeTableAction::CompactBlocks(limit_cnt),
+                    CompactTarget::Segment => OptimizeTableAction::CompactSegments(limit_cnt),
                 }
             }
-        } else {
-            OptimizeTableAction::Purge
         };
 
         Ok(Plan::OptimizeTable(Box::new(OptimizeTablePlan {

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

@@ -16,6 +16,7 @@ use std::collections::BTreeMap;
 use std::sync::Arc;
 
 use common_ast::ast::Engine;
+use common_catalog::table::NavigationPoint;
 use common_datavalues::DataField;
 use common_datavalues::DataSchema;
 use common_datavalues::DataSchemaRef;
@@ -117,11 +118,10 @@ impl OptimizeTablePlan {
     }
 }
 
-#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub enum OptimizeTableAction {
     All,
-    Purge,
-    Statistic,
+    Purge(Option<NavigationPoint>),
     CompactBlocks(Option<usize>),
     CompactSegments(Option<usize>),
 }