feat: support temporary table (#16250)

* create temporary table

* create as select

* fix create as select

* rename table

* get_table_meta_by_id

* get_table_name_by_id

* not support stream on temp table

* fix get_table_name_by_id

* add comment

* check access

* forbid grant and revoke

* get table

* not support inverted index

* not support virtual column

* not support agg index

* not support data mask

* ,

* drop table by id

* refactor

* update table meta

* fix get table meta by id

* get_table

* fix missing header

* add create table test

* rename folder

* commit before refactor

* refactor

* make lint

* regenerate golden file

* fix catalog and add drop table test

* fix catalog

* add alter temp table test

* fix show create table

* add rename temp table test

* add truncate test

* add copy into test

* fix copied file

* add log

* add some check

* add drop_all_temp_tables()

* refine error message

* forbid alter temp prefix

* remove useless println

* adjust slt

* add privilege test

* fix

* add assert

* change print to log

* add temp table storage prefix

* make lint

* fix prefix

Author: SkyFan2002

Cargo.lock

@@ -52,7 +52,7 @@ members = [
     "src/query/storages/common/io",
     "src/query/storages/common/pruner",
     "src/query/storages/common/stage",
-    "src/query/storages/common/txn",
+    "src/query/storages/common/session",
     "src/query/storages/common/table_meta",
     "src/query/storages/delta",
     "src/query/storages/factory",
@@ -197,9 +197,9 @@ databend-storages-common-cache = { path = "src/query/storages/common/cache" }
 databend-storages-common-index = { path = "src/query/storages/common/index" }
 databend-storages-common-io = { path = "src/query/storages/common/io" }
 databend-storages-common-pruner = { path = "src/query/storages/common/pruner" }
+databend-storages-common-session = { path = "src/query/storages/common/session" }
 databend-storages-common-stage = { path = "src/query/storages/common/stage" }
 databend-storages-common-table-meta = { path = "src/query/storages/common/table_meta" }
-databend-storages-common-txn = { path = "src/query/storages/common/txn" }
 
 # Crates.io dependencies
 anyerror = { version = "=0.1.10" }

Cargo.toml

@@ -19,4 +19,5 @@ fi
 echo "Run suites using argument: $RUN_DIR"
 
 echo "Starting databend-sqllogic tests"
-target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,explain_native,ee --enable_sandbox --parallel 8
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers "mysql" --run_dir temp_table --enable_sandbox --parallel 8
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,explain_native,ee,temp_table --enable_sandbox --parallel 8

scripts/ci/ci-run-sqllogic-tests.sh

@@ -3016,6 +3016,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
             copied_files,
             update_stream_metas,
             deduplicated_labels,
+            update_temp_tables: _,
         } = req;
 
         let mut tbl_seqs = HashMap::new();

src/meta/api/src/schema_api_impl.rs

@@ -136,6 +136,7 @@ pub use table::UpdateMultiTableMetaResult;
 pub use table::UpdateStreamMetaReq;
 pub use table::UpdateTableMetaReply;
 pub use table::UpdateTableMetaReq;
+pub use table::UpdateTempTableReq;
 pub use table::UpsertTableCopiedFileReply;
 pub use table::UpsertTableCopiedFileReq;
 pub use table::UpsertTableOptionReply;

src/meta/app/src/schema/mod.rs

@@ -72,7 +72,7 @@ impl Display for TableIdent {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, Hash)]
 pub struct TableNameIdent {
     pub tenant: Tenant,
     pub db_name: String,
@@ -188,6 +188,9 @@ impl Display for DatabaseType {
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, Eq, PartialEq, Default)]
 pub struct TableInfo {
+    /// For a temp table,
+    /// `ident.seq` is always 0.
+    /// `id.table_id` is set as value of `TempTblId`.
     pub ident: TableIdent,
 
     /// For a table it is `db_name.table_name`.
@@ -581,7 +584,7 @@ impl Display for DropTableByIdReq {
     }
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Default)]
 pub struct DropTableReply {
     // db id, share spec vector
     pub spec_vec: Option<(u64, Vec<ShareSpec>)>,
@@ -720,12 +723,31 @@ pub struct UpdateTableMetaReq {
     pub new_table_meta: TableMeta,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct UpdateTempTableReq {
+    pub table_id: u64,
+    pub desc: String,
+    pub new_table_meta: TableMeta,
+    pub copied_files: BTreeMap<String, TableCopiedFileInfo>,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Default)]
 pub struct UpdateMultiTableMetaReq {
     pub update_table_metas: Vec<(UpdateTableMetaReq, TableInfo)>,
     pub copied_files: Vec<(u64, UpsertTableCopiedFileReq)>,
     pub update_stream_metas: Vec<UpdateStreamMetaReq>,
     pub deduplicated_labels: Vec<String>,
+    pub update_temp_tables: Vec<UpdateTempTableReq>,
+}
+
+impl UpdateMultiTableMetaReq {
+    pub fn is_empty(&self) -> bool {
+        self.update_table_metas.is_empty()
+            && self.copied_files.is_empty()
+            && self.update_stream_metas.is_empty()
+            && self.deduplicated_labels.is_empty()
+            && self.update_temp_tables.is_empty()
+    }
 }
 
 /// The result of updating multiple table meta

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

@@ -31,6 +31,7 @@ use crate::ast::CreateStreamStmt;
 use crate::ast::CreateTableSource;
 use crate::ast::CreateTableStmt;
 use crate::ast::CreateViewStmt;
+use crate::ast::TableType;
 use crate::ast::TimeTravelPoint;
 
 pub(crate) fn pretty_create_table(stmt: CreateTableStmt) -> RcDoc<'static> {
@@ -40,10 +41,10 @@ pub(crate) fn pretty_create_table(stmt: CreateTableStmt) -> RcDoc<'static> {
         } else {
             RcDoc::nil()
         })
-        .append(if stmt.transient {
-            RcDoc::space().append(RcDoc::text("TRANSIENT"))
-        } else {
-            RcDoc::nil()
+        .append(match stmt.table_type {
+            TableType::Transient => RcDoc::space().append(RcDoc::text("TRANSIENT")),
+            TableType::Temporary => RcDoc::space().append(RcDoc::text("TEMPORARY")),
+            TableType::Normal => RcDoc::nil(),
         })
         .append(RcDoc::space().append(RcDoc::text("TABLE")))
         .append(match stmt.create_option {

src/query/ast/src/ast/format/syntax/ddl.rs

@@ -140,7 +140,14 @@ pub struct CreateTableStmt {
     pub cluster_by: Vec<Expr>,
     pub table_options: BTreeMap<String, String>,
     pub as_query: Option<Box<Query>>,
-    pub transient: bool,
+    pub table_type: TableType,
+}
+
+#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
+pub enum TableType {
+    Normal,
+    Transient,
+    Temporary,
 }
 
 impl Display for CreateTableStmt {
@@ -149,9 +156,11 @@ impl Display for CreateTableStmt {
         if let CreateOption::CreateOrReplace = self.create_option {
             write!(f, "OR REPLACE ")?;
         }
-        if self.transient {
-            write!(f, "TRANSIENT ")?;
-        }
+        match self.table_type {
+            TableType::Normal => {}
+            TableType::Transient => write!(f, "TRANSIENT ")?,
+            TableType::Temporary => write!(f, "TEMPORARY ")?,
+        };
         write!(f, "TABLE ")?;
         if let CreateOption::CreateIfNotExists = self.create_option {
             write!(f, "IF NOT EXISTS ")?;

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

@@ -697,7 +697,7 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
     );
     let create_table = map_res(
         rule! {
-            CREATE ~ ( OR ~ ^REPLACE )? ~ TRANSIENT? ~ TABLE ~ ( IF ~ ^NOT ~ ^EXISTS )?
+            CREATE ~ ( OR ~ ^REPLACE )? ~ (TEMP| TEMPORARY|TRANSIENT)? ~ TABLE ~ ( IF ~ ^NOT ~ ^EXISTS )?
             ~ #dot_separated_idents_1_to_3
             ~ #create_table_source?
             ~ ( #engine )?
@@ -709,7 +709,7 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
         |(
             _,
             opt_or_replace,
-            opt_transient,
+            opt_type,
             _,
             opt_if_not_exists,
             (catalog, database, table),
@@ -722,6 +722,12 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
         )| {
             let create_option =
                 parse_create_option(opt_or_replace.is_some(), opt_if_not_exists.is_some())?;
+            let table_type = match opt_type.map(|t| t.kind) {
+                None => TableType::Normal,
+                Some(TRANSIENT) => TableType::Transient,
+                Some(TEMP) | Some(TEMPORARY) => TableType::Temporary,
+                _ => unreachable!(),
+            };
             Ok(Statement::CreateTable(CreateTableStmt {
                 create_option,
                 catalog,
@@ -735,7 +741,7 @@ pub fn statement_body(i: Input) -> IResult<Statement> {
                     .unwrap_or_default(),
                 table_options: opt_table_options.unwrap_or_default(),
                 as_query: opt_as_query.map(|(_, query)| Box::new(query)),
-                transient: opt_transient.is_some(),
+                table_type,
             }))
         },
     );

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

@@ -1305,6 +1305,8 @@ pub enum TokenKind {
     ROLLBACK,
     #[token("TEMPORARY", ignore(ascii_case))]
     TEMPORARY,
+    #[token("TEMP", ignore(ascii_case))]
+    TEMP,
     #[token("SECONDS", ignore(ascii_case))]
     SECONDS,
     #[token("DAYS", ignore(ascii_case))]