add wap branch

fix

fix

fix

fix

fix

fix

fix

fix

fix

fix

fix

fix

fix

Author: zhyass

src/query/catalog/src/table.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 use chrono::DateTime;
 use chrono::Utc;
 use databend_common_ast::ast::Expr;
+use databend_common_ast::ast::quote::QuotedIdent;
 use databend_common_ast::parser::Dialect;
 use databend_common_ast::parser::parse_comma_separated_exprs;
 use databend_common_ast::parser::tokenize_sql;
@@ -61,6 +62,22 @@ use crate::statistics::BasicColumnStatistics;
 use crate::table_args::TableArgs;
 use crate::table_context::TableContext;
 
+pub fn quoted_table_reference(
+    database_name: &str,
+    table_name: &str,
+    branch_name: Option<&str>,
+    quote: char,
+) -> String {
+    let database = QuotedIdent(database_name, quote);
+    let table = QuotedIdent(table_name, quote);
+    if let Some(branch_name) = branch_name {
+        let branch = QuotedIdent(branch_name, quote);
+        format!("{database}.{table}/{branch}")
+    } else {
+        format!("{database}.{table}")
+    }
+}
+
 #[async_trait::async_trait]
 pub trait Table: Sync + Send {
     fn name(&self) -> &str {
@@ -366,9 +383,10 @@ pub trait Table: Sync + Send {
         ctx: Arc<dyn TableContext>,
         database_name: &str,
         table_name: &str,
+        branch_name: Option<&str>,
         with_options: &str,
     ) -> Result<String> {
-        let (_, _, _, _) = (ctx, database_name, table_name, with_options);
+        let (_, _, _, _, _) = (ctx, database_name, table_name, branch_name, with_options);
 
         Err(ErrorCode::Unimplemented(format!(
             "Change tracking operation is not supported for the table '{}', which uses the '{}' engine.",
@@ -551,6 +569,17 @@ pub enum TimeNavigation {
     },
 }
 
+impl TimeNavigation {
+    pub fn contains_table_tag(&self) -> bool {
+        match self {
+            TimeNavigation::TimeTravel(point) => point.is_table_tag(),
+            TimeNavigation::Changes { at, end, .. } => {
+                at.is_table_tag() || end.as_ref().is_some_and(NavigationPoint::is_table_tag)
+            }
+        }
+    }
+}
+
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum NavigationPoint {
     SnapshotID(String),
@@ -559,6 +588,12 @@ pub enum NavigationPoint {
     TableTag(String),
 }
 
+impl NavigationPoint {
+    pub fn is_table_tag(&self) -> bool {
+        matches!(self, NavigationPoint::TableTag(_))
+    }
+}
+
 #[derive(Debug, Copy, Clone, Default, serde::Serialize, serde::Deserialize)]
 pub struct TableStatistics {
     pub num_rows: Option<u64>,

src/query/service/src/interpreters/access/privilege_access.rs

@@ -1799,13 +1799,17 @@ impl AccessChecker for PrivilegeAccess {
                 }
             }
             Plan::CreateView(plan) => {
-                let mut planner = Planner::new(self.ctx.clone());
+                // Check stored view SQL against base tables.
+                let mut planner =
+                    Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
                 let (plan, _) = planner.plan_sql(&plan.subquery).await?;
                 self.check(ctx, &plan).await?
             }
             Plan::AlterView(plan) => {
                 self.validate_db_access(&plan.catalog, &plan.database, UserPrivilegeType::Alter, false).await?;
-                let mut planner = Planner::new(self.ctx.clone());
+                // Check stored view SQL against base tables.
+                let mut planner =
+                    Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
                 let (plan, _) = planner.plan_sql(&plan.subquery).await?;
                 self.check(ctx, &plan).await?
             }

src/query/service/src/interpreters/interpreter_table_add_column.rs

@@ -242,7 +242,7 @@ impl Interpreter for AddTableColumnInterpreter {
                 field.name(),
                 field.default_expr().unwrap()
             );
-            let mut planner = Planner::new(self.ctx.clone());
+            let mut planner = Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
             let (plan, _) = planner.plan_sql(&query).await?;
             if let Plan::DataMutation { s_expr, schema, .. } = plan {
                 let mutation: Mutation = s_expr.plan().clone().try_into()?;

src/query/service/src/interpreters/interpreter_table_describe.rs

@@ -65,7 +65,8 @@ impl Interpreter for DescribeTableInterpreter {
 
         let schema = if tbl_info.engine() == VIEW_ENGINE {
             if let Some(query) = tbl_info.options().get(QUERY) {
-                let mut planner = Planner::new(self.ctx.clone());
+                // Replay stored view SQL against base tables.
+                let mut planner = Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
                 let (plan, _) = planner.plan_sql(query).await?;
                 infer_table_schema(&plan.schema())
             } else {

src/query/service/src/interpreters/interpreter_table_modify_column.rs

@@ -871,7 +871,7 @@ pub(crate) async fn build_select_insert_plan(
     table_meta_timestamps: TableMetaTimestamps,
 ) -> Result<PipelineBuildResult> {
     // 1. build plan by sql
-    let mut planner = Planner::new(ctx.clone());
+    let mut planner = Planner::new(ctx.clone()).with_suppress_wap_branch(true);
     let (plan, _extras) = planner.plan_sql(&sql).await?;
     let select_schema = plan.schema();
 

src/query/service/src/interpreters/interpreter_view_alter.rs

@@ -57,7 +57,8 @@ impl Interpreter for AlterViewInterpreter {
             .get_table(&self.plan.tenant, &self.plan.database, &self.plan.view_name)
             .await
         {
-            let mut planner = Planner::new(self.ctx.clone());
+            // Replay stored view SQL against base tables.
+            let mut planner = Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
             let (plan, _) = planner.plan_sql(&self.plan.subquery.clone()).await?;
 
             // Detect circular dependency: ALTER VIEW can introduce cycles

src/query/service/src/interpreters/interpreter_view_create.rs

@@ -59,7 +59,8 @@ impl Interpreter for CreateViewInterpreter {
         let tenant = self.ctx.get_tenant();
         let table_function = catalog.list_table_functions();
         let mut options = BTreeMap::new();
-        let mut planner = Planner::new(self.ctx.clone());
+        // Replay stored view SQL against base tables.
+        let mut planner = Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
         let (plan, _) = planner.plan_sql(&self.plan.subquery.clone()).await?;
         match plan.clone() {
             Plan::Query { metadata, .. } => {

src/query/service/src/interpreters/interpreter_view_describe.rs

@@ -63,7 +63,8 @@ impl Interpreter for DescribeViewInterpreter {
         let engine = table.get_table_info().engine();
         let schema = if engine == VIEW_ENGINE {
             if let Some(query) = tbl_info.options().get(QUERY) {
-                let mut planner = Planner::new(self.ctx.clone());
+                // Replay stored view SQL against base tables.
+                let mut planner = Planner::new(self.ctx.clone()).with_suppress_wap_branch(true);
                 let (plan, _) = planner.plan_sql(query).await?;
                 infer_table_schema(&plan.schema())
             } else {

src/query/settings/src/settings_default.rs

@@ -1675,6 +1675,13 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
+                ("wap_branch", DefaultSettingValue {
+                    value: UserSettingValue::String("".to_string()),
+                    desc: "WAP branch for table reads and writes. Empty string disables.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Session,
+                    range: None,
+                }),
                 ("force_aggregate_shuffle_mode", DefaultSettingValue {
                     value: UserSettingValue::String(String::from("auto")),
                     desc: "For testing only. Shuffle mode for aggregation. Options are 'auto', 'row', 'bucket'. Default is 'auto'.",

src/query/settings/src/settings_getter_setter.rs

@@ -1224,6 +1224,15 @@ impl Settings {
         Ok(self.try_get_u64("enable_experimental_table_ref")? != 0)
     }
 
+    pub fn get_wap_branch(&self) -> Result<Option<String>> {
+        let branch = self.try_get_string("wap_branch")?;
+        if branch.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(branch))
+        }
+    }
+
     pub fn get_force_aggregate_shuffle_mode(&self) -> Result<String> {
         self.try_get_string("force_aggregate_shuffle_mode")
     }