fix review comments

Author: zhyass

src/meta/api/src/api_impl/ref_api.rs

@@ -14,7 +14,6 @@
 
 use std::collections::HashMap;
 use std::collections::HashSet;
-use std::sync::Arc;
 
 use chrono::DateTime;
 use chrono::Utc;
@@ -25,7 +24,6 @@ use databend_common_meta_app::app_error::TableSnapshotExpired;
 use databend_common_meta_app::app_error::TableVersionMismatched;
 use databend_common_meta_app::app_error::UndropTableRetentionGuard;
 use databend_common_meta_app::app_error::UnknownReference;
-use databend_common_meta_app::app_error::UnknownTable;
 use databend_common_meta_app::app_error::UnknownTableId;
 use databend_common_meta_app::data_mask::MaskPolicyIdTableId;
 use databend_common_meta_app::data_mask::MaskPolicyTableId;
@@ -41,7 +39,6 @@ use databend_common_meta_app::schema::AutoIncrementStorageValue;
 use databend_common_meta_app::schema::BranchIdToName;
 use databend_common_meta_app::schema::CreateTableBranchReq;
 use databend_common_meta_app::schema::CreateTableTagReq;
-use databend_common_meta_app::schema::DBIdTableName;
 use databend_common_meta_app::schema::DatabaseType;
 use databend_common_meta_app::schema::DropTableBranchReq;
 use databend_common_meta_app::schema::DropTableTagReq;
@@ -85,9 +82,7 @@ use log::debug;
 use log::warn;
 
 use crate::api_impl::schema_api::restore_policy_references_on_undrop;
-use crate::database_util::get_db_or_err;
 use crate::fetch_id;
-use crate::get_u64_value;
 use crate::kv_app_error::KVAppError;
 use crate::kv_pb_api::KVPbApi;
 use crate::txn_backoff::txn_backoff;
@@ -310,13 +305,13 @@ where
         debug!(req :? =(&req); "RefApi: {}", func_name!());
 
         let branch_name = &req.branch_name;
-        let source_table_id = req.from_branch_id.unwrap_or(req.base_table_id);
+        // `base_table_id` identifies the base table namespace where the new branch name lives.
+        // `source_table_id` identifies the object being forked from: the base table itself, or
+        // a source branch table when `source_branch_id` is set.
+        let source_table_id = req.source_branch_id.unwrap_or(req.base_table_id);
         let key_source_table_id = TableId {
             table_id: source_table_id,
         };
-        let key_table_id = TableId {
-            table_id: req.base_table_id,
-        };
         let key_branch = TableIdBranchName::new(req.base_table_id, branch_name);
 
         let mut maybe_branch_id = None;
@@ -351,9 +346,12 @@ where
                 txn_cond_seq(&key_branch, Eq, 0),
             ];
 
-            if req.from_branch_id.is_some() {
+            if req.source_branch_id.is_some() {
                 // The source may be a branch table, but the new branch is still created under the
                 // base table namespace, so the base table must still exist and be active.
+                let key_table_id = TableId {
+                    table_id: req.base_table_id,
+                };
                 let Some(seq_table_meta) = self.get_pb(&key_table_id).await? else {
                     return Err(KVAppError::AppError(AppError::UnknownTableId(
                         UnknownTableId::new(req.base_table_id, "create_table_branch: base table"),
@@ -594,36 +592,11 @@ where
     /// Resolve a visible branch name to its current branch table metadata.
     #[logcall::logcall]
     #[fastrace::trace]
-    async fn get_table_branch(&self, req: GetTableBranchReq) -> Result<Arc<TableInfo>, KVAppError> {
+    async fn get_table_branch(&self, req: GetTableBranchReq) -> Result<TableInfo, KVAppError> {
         debug!(req :? =(&req); "RefApi: {}", func_name!());
 
-        let tenant_dbname = req.name_ident.db_name_ident();
-        let (seq_db_id, _db_meta) = get_db_or_err(
-            self,
-            &tenant_dbname,
-            format!("{}: {}", "get_table_branch", tenant_dbname.display()),
-        )
-        .await?;
-        let db_id = *seq_db_id.data;
-
-        let dbid_tbname = DBIdTableName {
-            db_id,
-            table_name: req.name_ident.table_name.clone(),
-        };
-        // Branch names are resolved under the *currently visible* base table id of `db.table`.
-        // This is intentional: branches are treated as refs of the current base table namespace,
-        // not as independently addressable objects. If the base table is dropped or replaced and
-        // `db.table` points to a new table id, previously created branches under the old base
-        // table id become unreachable by design.
-        let (tb_id_seq, table_id) = get_u64_value(self, &dbid_tbname).await?;
-        if tb_id_seq == 0 {
-            return Err(KVAppError::AppError(AppError::UnknownTable(
-                UnknownTable::new(&req.name_ident.table_name, "get_table_branch"),
-            )));
-        }
-
         let branch_name = &req.branch_name;
-        let key_branch = TableIdBranchName::new(table_id, branch_name);
+        let key_branch = TableIdBranchName::new(req.table_id, branch_name);
         let seq_branch = self.get_pb(&key_branch).await?;
         let Some(seq_branch) = seq_branch else {
             return Err(KVAppError::AppError(AppError::from(UnknownReference::new(
@@ -658,19 +631,20 @@ where
                 table_id: branch_id,
                 seq: seq_table_meta.seq,
             },
-            desc: format!(
-                "'{}'.'{}'/'{}'",
-                req.name_ident.db_name, req.name_ident.table_name, branch_name,
-            ),
-            name: req.name_ident.table_name.clone(),
+            // Meta lookup only resolves the branch object by `table_id + branch_name`.
+            // The caller rewrites this into the user-facing form `db.table/branch`.
+            desc: branch_name.clone(),
+            // `name` should be the base table name in the final TableInfo used by query code.
+            // It is filled by the caller after resolving the branch under a specific table name.
+            name: String::new(),
             meta: seq_table_meta.data,
             db_type: DatabaseType::NormalDB,
             // This meta API is currently reached only through the default catalog's branch lookup
             // path, so the returned TableInfo stays aligned with that default-catalog-only contract.
             catalog_info: Default::default(),
         };
 
-        Ok(Arc::new(tb_info))
+        Ok(tb_info)
     }
 
     /// List table branches.

src/meta/api/src/api_impl/table_api.rs

@@ -1734,6 +1734,26 @@ where
         Ok(table_name)
     }
 
+    #[logcall::logcall]
+    #[fastrace::trace]
+    async fn resolve_table_id(&self, name_ident: &TableNameIdent) -> Result<u64, KVAppError> {
+        let tenant_dbname = name_ident.db_name_ident();
+        let (seq_db_id, _db_meta) = get_db_or_err(
+            self,
+            &tenant_dbname,
+            format!("resolve_table_id: {}", tenant_dbname.display()),
+        )
+        .await?;
+        let dbid_tbname = DBIdTableName::new(*seq_db_id.data, name_ident.table_name.clone());
+        let (tb_id_seq, table_id) = get_u64_value(self, &dbid_tbname).await?;
+        if tb_id_seq == 0 {
+            return Err(KVAppError::AppError(AppError::UnknownTable(
+                UnknownTable::new(&name_ident.table_name, "resolve_table_id"),
+            )));
+        }
+        Ok(table_id)
+    }
+
     #[logcall::logcall]
     #[fastrace::trace]
     async fn get_table_copied_file_info(

src/meta/app/src/schema/table/refs.rs

@@ -24,7 +24,6 @@ use databend_meta_client::types::SeqV;
 use super::TableId;
 use super::TableLvtCheck;
 use super::TableMeta;
-use super::TableNameIdent;
 use crate::tenant::Tenant;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -145,7 +144,7 @@ pub struct CreateTableBranchReq {
     /// Base table id that owns the branch namespace.
     pub base_table_id: u64,
     /// Fork from this branch. `None` means fork from the base table itself.
-    pub from_branch_id: Option<u64>,
+    pub source_branch_id: Option<u64>,
     pub branch_name: String,
     /// Optimistic seq check for the fork source.
     pub seq: MatchSeq,
@@ -165,7 +164,7 @@ pub struct DropTableBranchReq {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct GetTableBranchReq {
-    pub name_ident: TableNameIdent,
+    pub table_id: u64,
     pub branch_name: String,
     pub include_expired: bool,
 }

src/meta/schema-api-test-suite/src/schema_api_test_suite.rs

@@ -7579,7 +7579,7 @@ impl SchemaApiTestSuite {
             .create_table_branch(CreateTableBranchReq {
                 tenant: tenant.clone(),
                 base_table_id: table_info.ident.table_id,
-                from_branch_id: None,
+                source_branch_id: None,
                 seq: MatchSeq::Exact(table_info.ident.seq),
                 branch_name: branch_name.to_string(),
                 new_table_meta: table_info.meta.clone(),

src/query/ee/src/table_ref/handler.rs

@@ -183,7 +183,7 @@ impl TableRefHandler for RealTableRefHandler {
             .create_table_branch(CreateTableBranchReq {
                 tenant: ctx.get_tenant(),
                 base_table_id,
-                from_branch_id: (source_table_id != base_table_id).then_some(source_table_id),
+                source_branch_id: (source_table_id != base_table_id).then_some(source_table_id),
                 branch_name: plan.branch_name.clone(),
                 seq: MatchSeq::Exact(seq),
                 new_table_meta: branch_table_meta,

src/query/service/src/catalogs/default/mutable_catalog.rs

@@ -671,13 +671,18 @@ impl Catalog for MutableCatalog {
         branch_name: &str,
         include_expired: bool,
     ) -> Result<Arc<dyn Table>> {
+        let name_ident = TableNameIdent::new(tenant, db_name, table_name);
+        let table_id = self.ctx.meta.resolve_table_id(&name_ident).await?;
         let req = GetTableBranchReq {
-            name_ident: TableNameIdent::new(tenant, db_name, table_name),
+            table_id,
             branch_name: branch_name.to_string(),
             include_expired,
         };
         let info = self.ctx.meta.get_table_branch(req).await?;
-        self.get_table_by_info(info.as_ref())
+        let mut info = info;
+        info.name = table_name.to_string();
+        info.desc = format!("'{}'.'{}'/'{}'", db_name, table_name, branch_name);
+        self.get_table_by_info(&info)
     }
 
     #[async_backtrace::framed]