add wap branch

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.",

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_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_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")
     }

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

@@ -197,6 +197,10 @@ pub struct BindContext {
     /// It's used to avoid infinite loop.
     pub planning_agg_index: bool,
 
+    /// If true, table refs in this scope ignore the session `wap_branch`.
+    /// Used while replaying persisted definitions against base tables.
+    pub suppress_wap_branch: bool,
+
     pub window_definitions: DashMap<String, WindowSpec>,
 }
 
@@ -276,6 +280,7 @@ impl BindContext {
             expr_context: ExprContext::default(),
             group_by_column_first: false,
             planning_agg_index: false,
+            suppress_wap_branch: false,
             window_definitions: DashMap::new(),
         }
     }
@@ -324,6 +329,7 @@ impl BindContext {
             expr_context: ExprContext::default(),
             group_by_column_first: parent.group_by_column_first,
             planning_agg_index: false,
+            suppress_wap_branch: parent.suppress_wap_branch,
             window_definitions: DashMap::new(),
         })
     }
@@ -335,6 +341,8 @@ impl BindContext {
         bind_context.cte_context = self.cte_context.clone();
         bind_context.udf_cache = self.udf_cache.clone();
         bind_context.binding_views = self.binding_views.clone();
+        // Keep WAP suppression across `replace()` (used by bind_join).
+        bind_context.suppress_wap_branch = self.suppress_wap_branch;
         bind_context.group_by_column_first = self.group_by_column_first;
         bind_context
     }

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

@@ -145,6 +145,13 @@ impl Binder {
             target_table_identifier.table_name_alias(),
         );
 
+        let branch_name = self.resolve_write_branch_with_wap_branch(
+            &catalog_name,
+            &database_name,
+            &table_name,
+            branch_name,
+        )?;
+
         // Add table lock before execution.
         let lock_guard = if strategy != MutationStrategy::NotMatchedOnly {
             self.ctx