[feature](schema) support temporary table like mysql

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -945,6 +945,17 @@ public class Config extends ConfigBase {
     // All frontends will get tablet stat from all backends at each interval
     @ConfField public static int tablet_stat_update_interval_second = 60;  // 1 min
 
+    // update interval of alive session
+    // Only master FE collect this info from all frontends at each interval
+    @ConfField public static int alive_session_update_interval_second = 5;
+
+    @ConfField public static int fe_session_mgr_threads_num = 1;
+
+    @ConfField public static int fe_session_mgr_blocking_queue_size = 1024;
+
+    @ConfField(mutable = true, masterOnly = true)
+    public static int loss_conn_fe_temp_table_keep_second = 60;
+
     /**
      * Max bytes a broker scanner can process in one broker load job.
      * Commonly, each Backends has one broker scanner.

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -177,6 +177,19 @@ supportedCreateStatement
         properties=propertyClause?
         (BROKER extProperties=propertyClause)?
         (AS query)?                                                       #createTable
+    | CREATE (TEMPORARY)? TABLE (IF NOT EXISTS)? name=multipartIdentifier
+        ((ctasCols=identifierList)? | (LEFT_PAREN columnDefs (COMMA indexDefs)? COMMA? RIGHT_PAREN))
+        (ENGINE EQ engine=identifier)?
+        ((AGGREGATE | UNIQUE | DUPLICATE) KEY keys=identifierList
+        (CLUSTER BY clusterKeys=identifierList)?)?
+        (COMMENT STRING_LITERAL)?
+        (partition=partitionTable)?
+        (DISTRIBUTED BY (HASH hashKeys=identifierList | RANDOM)
+        (BUCKETS (INTEGER_VALUE | autoBucket=AUTO))?)?
+        (ROLLUP LEFT_PAREN rollupDefs RIGHT_PAREN)?
+        properties=propertyClause?
+        (BROKER extProperties=propertyClause)?
+        (AS query)?                                                       #createTable
     | CREATE (OR REPLACE)? VIEW (IF NOT EXISTS)? name=multipartIdentifier
         (LEFT_PAREN cols=simpleColumnDefs RIGHT_PAREN)?
         (COMMENT STRING_LITERAL)? AS query                                #createView
@@ -185,6 +198,9 @@ supportedCreateStatement
     | CREATE (EXTERNAL)? TABLE (IF NOT EXISTS)? name=multipartIdentifier
         LIKE existedTable=multipartIdentifier
         (WITH ROLLUP (rollupNames=identifierList)?)?                      #createTableLike
+    | CREATE (TEMPORARY)? TABLE (IF NOT EXISTS)? name=multipartIdentifier
+        LIKE existedTable=multipartIdentifier
+        (WITH ROLLUP (rollupNames=identifierList)?)?                      #createTableLike
     | CREATE ROLE (IF NOT EXISTS)? name=identifier (COMMENT STRING_LITERAL)?    #createRole
     | CREATE WORKLOAD GROUP (IF NOT EXISTS)?
         name=identifierOrText properties=propertyClause?                        #createWorkloadGroup          

fe/fe-core/src/main/java/org/apache/doris/alter/Alter.java

@@ -127,7 +127,7 @@ public void processCreateMaterializedView(CreateMaterializedViewStmt stmt)
         Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(dbName);
         Env.getCurrentInternalCatalog().checkAvailableCapacity(db);
 
-        OlapTable olapTable = (OlapTable) db.getTableOrMetaException(tableName, TableType.OLAP);
+        OlapTable olapTable = (OlapTable) db.getNonTempTableOrMetaException(tableName, TableType.OLAP);
         ((MaterializedViewHandler) materializedViewHandler).processCreateMaterializedView(stmt, db, olapTable);
     }
 
@@ -539,6 +539,9 @@ public void processAlterTable(AlterTableStmt stmt) throws UserException {
         switch (tableIf.getType()) {
             case MATERIALIZED_VIEW:
             case OLAP:
+                if (tableIf.isTemporary()) {
+                    throw new DdlException("Do not support alter temporary table[" + tableName + "]");
+                }
                 OlapTable olapTable = (OlapTable) tableIf;
                 needProcessOutsideTableLock = processAlterOlapTable(stmt, olapTable, alterClauses, (Database) dbIf);
                 break;
@@ -711,6 +714,9 @@ public void processReplaceTable(Database db, OlapTable origTable, String newTblN
         try {
             List<TableType> tableTypes = Lists.newArrayList(TableType.OLAP, TableType.MATERIALIZED_VIEW);
             Table newTbl = db.getTableOrMetaException(newTblName, tableTypes);
+            if (newTbl.isTemporary()) {
+                throw new UserException("Do not support replace with temporary table");
+            }
             OlapTable olapNewTbl = (OlapTable) newTbl;
             List<Table> tableList = Lists.newArrayList(origTable, newTbl);
             tableList.sort((Comparator.comparing(Table::getId)));

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateMTMVStmt.java

@@ -38,8 +38,8 @@ public CreateMTMVStmt(boolean ifNotExists, TableName mvName, List<Column> column
             MTMVRefreshInfo refreshInfo, KeysDesc keyDesc, DistributionDesc distributionDesc,
             Map<String, String> properties, Map<String, String> mvProperties, String querySql, String comment,
             PartitionDesc partitionDesc, MTMVPartitionInfo mvPartitionInfo, MTMVRelation relation) {
-        super(ifNotExists, false, mvName, columns, new ArrayList<Index>(), DEFAULT_ENGINE_NAME, keyDesc, partitionDesc,
-                distributionDesc, properties, null, comment, null, null);
+        super(ifNotExists, false, false, mvName, columns, new ArrayList<Index>(), DEFAULT_ENGINE_NAME, keyDesc,
+                partitionDesc, distributionDesc, properties, null, comment, null, null);
         this.refreshInfo = refreshInfo;
         this.querySql = querySql;
         this.mvProperties = mvProperties;

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableLikeStmt.java

@@ -44,14 +44,21 @@ public class CreateTableLikeStmt extends DdlStmt implements NotFallbackInParser
     private static final Logger LOG = LogManager.getLogger(CreateTableLikeStmt.class);
 
     private final boolean ifNotExists;
+    private final boolean isTemp;
     private final TableName tableName;
     private final TableName existedTableName;
     private final ArrayList<String> rollupNames;
     private final boolean withAllRollup;
 
     public CreateTableLikeStmt(boolean ifNotExists, TableName tableName, TableName existedTableName,
             ArrayList<String> rollupNames, boolean withAllRollup) throws DdlException {
+        this(ifNotExists, false, tableName, existedTableName, rollupNames, withAllRollup);
+    }
+
+    public CreateTableLikeStmt(boolean ifNotExists, boolean isTemp, TableName tableName, TableName existedTableName,
+            ArrayList<String> rollupNames, boolean withAllRollup) throws DdlException {
         this.ifNotExists = ifNotExists;
+        this.isTemp = isTemp;
         this.tableName = tableName;
         this.existedTableName = existedTableName;
         if (!CollectionUtils.isEmpty(rollupNames) && withAllRollup) {
@@ -65,6 +72,10 @@ public boolean isIfNotExists() {
         return ifNotExists;
     }
 
+    public boolean isTemp() {
+        return isTemp;
+    }
+
     public String getDbName() {
         return tableName.getDb();
     }
@@ -115,7 +126,11 @@ public void analyze(Analyzer analyzer) throws UserException {
     @Override
     public String toSql() {
         StringBuilder sb = new StringBuilder();
-        sb.append("CREATE TABLE ").append(tableName.toSql()).append(" LIKE ").append(existedTableName.toSql());
+        sb.append("CREATE ");
+        if (isTemp) {
+            sb.append("TEMPORARY ");
+        }
+        sb.append("TABLE ").append(tableName.toSql()).append(" LIKE ").append(existedTableName.toSql());
         if (withAllRollup && CollectionUtils.isEmpty(rollupNames)) {
             sb.append(" WITH ROLLUP");
         }

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java

@@ -81,6 +81,7 @@ public class CreateTableStmt extends DdlStmt implements NotFallbackInParser {
 
     protected boolean ifNotExists;
     private boolean isExternal;
+    private boolean isTemp;
     protected TableName tableName;
     protected List<ColumnDef> columnDefs;
     private List<IndexDef> indexDefs;
@@ -188,12 +189,14 @@ public CreateTableStmt(boolean ifNotExists, boolean isExternal, TableName tableN
     }
 
     // for Nereids
-    public CreateTableStmt(boolean ifNotExists, boolean isExternal, TableName tableName, List<Column> columns,
-            List<Index> indexes, String engineName, KeysDesc keysDesc, PartitionDesc partitionDesc,
-            DistributionDesc distributionDesc, Map<String, String> properties, Map<String, String> extProperties,
-            String comment, List<AlterClause> rollupAlterClauseList, Void unused) {
+    public CreateTableStmt(boolean ifNotExists, boolean isExternal, boolean isTemp, TableName tableName,
+            List<Column> columns, List<Index> indexes, String engineName, KeysDesc keysDesc,
+            PartitionDesc partitionDesc, DistributionDesc distributionDesc, Map<String, String> properties,
+            Map<String, String> extProperties, String comment,
+            List<AlterClause> rollupAlterClauseList, Void unused) {
         this.ifNotExists = ifNotExists;
         this.isExternal = isExternal;
+        this.isTemp = isTemp;
         this.tableName = tableName;
         this.columns = columns;
         this.indexes = indexes;
@@ -225,6 +228,14 @@ public boolean isExternal() {
         return isExternal;
     }
 
+    public boolean isTemp() {
+        return isTemp;
+    }
+
+    public void setTemp(boolean temp) {
+        isTemp = temp;
+    }
+
     public TableName getDbTbl() {
         return tableName;
     }
@@ -297,6 +308,9 @@ public void analyze(Analyzer analyzer) throws UserException {
         if (Strings.isNullOrEmpty(engineName) || engineName.equalsIgnoreCase(DEFAULT_ENGINE_NAME)) {
             this.properties = maybeRewriteByAutoBucket(distributionDesc, properties);
         }
+        if (isTemp && !engineName.equalsIgnoreCase(DEFAULT_ENGINE_NAME)) {
+            throw new AnalysisException("Temporary table should be OLAP table");
+        }
 
         super.analyze(analyzer);
         tableName.analyze(analyzer);
@@ -693,6 +707,9 @@ public String toSql() {
         StringBuilder sb = new StringBuilder();
 
         sb.append("CREATE ");
+        if (isTemp) {
+            sb.append("TEMPORARY ");
+        }
         if (isExternal) {
             sb.append("EXTERNAL ");
         }

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowColumnStatsStmt.java

@@ -28,6 +28,7 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.Util;
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
@@ -170,7 +171,7 @@ public ShowResultSet constructResultSet(List<Pair<Pair<String, String>, ColumnSt
             List<String> row = Lists.newArrayList();
             // p data structure is Pair<Pair<IndexName, ColumnName>, ColumnStatistic>
             row.add(p.first.second);
-            row.add(p.first.first);
+            row.add(Util.getTempTableDisplayName(p.first.first));
             row.add(String.valueOf(p.second.count));
             row.add(String.valueOf(p.second.ndv));
             row.add(String.valueOf(p.second.numNulls));

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowDataStmt.java

@@ -220,13 +220,22 @@ public int compare(Table t1, Table t2) {
             }
         });
 
+        boolean isAdmin = Env.getCurrentEnv().getAccessManager()
+                .checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN);
         for (Table table : tables) {
             if (!Env.getCurrentEnv().getAccessManager()
                     .checkTblPriv(ConnectContext.get(), InternalCatalog.INTERNAL_CATALOG_NAME, dbName,
                             table.getName(),
                             PrivPredicate.SHOW)) {
                 continue;
             }
+            // admin users can see all temporary tables no matter they are created by which session
+            if (!isAdmin) {
+                // non admin user can only see temporary tables in current session
+                if (table.isTemporary() && !Util.isTempTableInCurrentSession(table.getName())) {
+                    continue;
+                }
+            }
             sortedTables.add(table);
         }
 
@@ -244,14 +253,20 @@ public int compare(Table t1, Table t2) {
             replicaCount = olapTable.getReplicaCount();
             remoteSize = olapTable.getRemoteDataSize();
 
+            boolean useDisplayName = false;
+            if (!isAdmin && olapTable.isTemporary()) {
+                useDisplayName = true;
+            }
+            String tableName = useDisplayName
+                    ? Util.getTempTableDisplayName(olapTable.getName()) : olapTable.getName();
             if (!detailed) {
-                totalRowsObject.add(Arrays.asList(table.getName(), tableSize, replicaCount, remoteSize));
+                totalRowsObject.add(Arrays.asList(tableName, tableSize, replicaCount, remoteSize));
             } else {
                 long localIndexSize = olapTable.getLocalIndexFileSize();
                 long localSegmentSize = olapTable.getLocalSegmentSize();
                 long remoteIndexSize = olapTable.getRemoteIndexFileSize();
                 long remoteSegmentSize = olapTable.getRemoteSegmentSize();
-                totalRowsObject.add(Arrays.asList(table.getName(), tableSize, replicaCount, remoteSize,
+                totalRowsObject.add(Arrays.asList(tableName, tableSize, replicaCount, remoteSize,
                         localIndexSize, localSegmentSize, remoteIndexSize, remoteSegmentSize));
                 totalLocalInvertedSize += localIndexSize;
                 totalLocalSegmentSize += localSegmentSize;

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowQueryStatsStmt.java

@@ -25,6 +25,7 @@
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.Util;
 import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ShowResultSetMetaData;
@@ -137,7 +138,14 @@ public void analyze(Analyzer analyzer) throws UserException {
             stats.forEach((tableName, queryHit) -> {
                 if (Env.getCurrentEnv().getAccessManager()
                         .checkTblPriv(ConnectContext.get(), ctlName, dbName, tableName, PrivPredicate.SHOW)) {
-                    totalRows.add(Arrays.asList(tableName, String.valueOf(queryHit)));
+                    if (Util.isTempTable(tableName)) {
+                        if (Util.isTempTableInCurrentSession(tableName)) {
+                            totalRows.add(Arrays.asList(Util.getTempTableDisplayName(tableName),
+                                    String.valueOf(queryHit)));
+                        }
+                    } else {
+                        totalRows.add(Arrays.asList(tableName, String.valueOf(queryHit)));
+                    }
                 }
             });
         } else {

fe/fe-core/src/main/java/org/apache/doris/backup/BackupHandler.java

@@ -48,6 +48,7 @@
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.util.MasterDaemon;
 import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.common.util.Util;
 import org.apache.doris.fs.FileSystemFactory;
 import org.apache.doris.fs.remote.AzureFileSystem;
 import org.apache.doris.fs.remote.RemoteFileSystem;
@@ -435,14 +436,27 @@ private void backup(Repository repository, Database db, BackupStmt stmt) throws
                 if (Config.ignore_backup_not_support_table_type) {
                     LOG.warn("Table '{}' is a {} table, can not backup and ignore it."
                             + "Only OLAP(Doris)/ODBC/VIEW table can be backed up",
-                            tblName, tbl.getType().toString());
+                            tblName, tbl.isTemporary() ? "temporary" : tbl.getType().toString());
                     tblRefsNotSupport.add(tblRef);
                     continue;
                 } else {
                     ErrorReport.reportDdlException(ErrorCode.ERR_NOT_OLAP_TABLE, tblName);
                 }
             }
 
+            if (tbl.isTemporary()) {
+                if (Config.ignore_backup_not_support_table_type) {
+                    LOG.warn("Table '{}' is a temporary table, can not backup and ignore it."
+                            + "Only OLAP(Doris)/ODBC/VIEW table can be backed up",
+                            Util.getTempTableDisplayName(tblName));
+                    tblRefsNotSupport.add(tblRef);
+                    continue;
+                } else {
+                    ErrorReport.reportDdlException("Table " + Util.getTempTableDisplayName(tblName)
+                            + " is a temporary table, do not support backup");
+                }
+            }
+
             OlapTable olapTbl = (OlapTable) tbl;
             tbl.readLock();
             try {

fe/fe-core/src/main/java/org/apache/doris/binlog/BinlogManager.java

@@ -154,7 +154,11 @@ private void addBinlog(long dbId, List<Long> tableIds, long commitSeq, long time
         if (tableIds != null) {
             for (long tableId : tableIds) {
                 boolean tableBinlogEnable = binlogConfigCache.isEnableTable(dbId, tableId);
-                anyEnable = anyEnable || tableBinlogEnable;
+                if (tableIds.size() > 1) {
+                    anyEnable = anyEnable || tableBinlogEnable;
+                } else {
+                    anyEnable = tableBinlogEnable;
+                }
                 if (anyEnable) {
                     break;
                 }