From 918df0ae50568ebc54bd6d89f2d9a49bc9b79b06 Mon Sep 17 00:00:00 2001
From: Lynn <zimu_xia@126.com>
Date: Thu, 14 Dec 2023 11:58:49 +0800
Subject: [PATCH] *: speed up `create table` when the number of tables is
 relatively large (#49371)

close pingcap/tidb#49370
---
 pkg/ddl/table.go          |  2 +-
 pkg/meta/meta.go          | 32 ++++++++++++++++++++++++++++++++
 pkg/meta/meta_test.go     |  8 ++++++++
 pkg/parser/model/model.go |  6 ++++++
 4 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/pkg/ddl/table.go b/pkg/ddl/table.go
index eefb14d55a922..ba387bf20bdd8 100644
--- a/pkg/ddl/table.go
+++ b/pkg/ddl/table.go
@@ -1545,7 +1545,7 @@ func checkTableNotExistsFromInfoSchema(is infoschema.InfoSchema, schemaID int64,
 
 func checkTableNotExistsFromStore(t *meta.Meta, schemaID int64, tableName string) error {
 	// Check this table's database.
-	tbls, err := t.ListTables(schemaID)
+	tbls, err := t.ListSimpleTables(schemaID)
 	if err != nil {
 		if meta.ErrDBNotExists.Equal(err) {
 			return infoschema.ErrDatabaseNotExists.GenWithStackByArgs("")
diff --git a/pkg/meta/meta.go b/pkg/meta/meta.go
index 79ac5085a7630..1a67f53fd70da 100644
--- a/pkg/meta/meta.go
+++ b/pkg/meta/meta.go
@@ -967,6 +967,38 @@ func (m *Meta) ListTables(dbID int64) ([]*model.TableInfo, error) {
 	return tables, nil
 }
 
+// ListSimpleTables shows all simple tables in database.
+func (m *Meta) ListSimpleTables(dbID int64) ([]*model.TableNameInfo, error) {
+	dbKey := m.dbKey(dbID)
+	if err := m.checkDBExists(dbKey); err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	res, err := m.txn.HGetAll(dbKey)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	tables := make([]*model.TableNameInfo, 0, len(res)/2)
+	for _, r := range res {
+		// only handle table meta
+		tableKey := string(r.Field)
+		if !strings.HasPrefix(tableKey, mTablePrefix) {
+			continue
+		}
+
+		tbInfo := &model.TableNameInfo{}
+		err = json.Unmarshal(r.Value, tbInfo)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+
+		tables = append(tables, tbInfo)
+	}
+
+	return tables, nil
+}
+
 // ListDatabases shows all databases.
 func (m *Meta) ListDatabases() ([]*model.DBInfo, error) {
 	res, err := m.txn.HGetAll(mDBs)
diff --git a/pkg/meta/meta_test.go b/pkg/meta/meta_test.go
index a0e310d849d4f..8ada35c0ffdaa 100644
--- a/pkg/meta/meta_test.go
+++ b/pkg/meta/meta_test.go
@@ -269,6 +269,11 @@ func TestMeta(t *testing.T) {
 	err = m.CreateTableOrView(1, tbInfo2)
 	require.NoError(t, err)
 
+	tblName := &model.TableNameInfo{ID: tbInfo.ID, Name: tbInfo.Name}
+	tblName2 := &model.TableNameInfo{ID: tbInfo2.ID, Name: tbInfo2.Name}
+	tableNames, err := m.ListSimpleTables(1)
+	require.NoError(t, err)
+	require.Equal(t, []*model.TableNameInfo{tblName, tblName2}, tableNames)
 	tables, err := m.ListTables(1)
 	require.NoError(t, err)
 	require.Equal(t, []*model.TableInfo{tbInfo, tbInfo2}, tables)
@@ -304,6 +309,9 @@ func TestMeta(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int64(0), n)
 
+	tableNames, err = m.ListSimpleTables(1)
+	require.NoError(t, err)
+	require.Equal(t, []*model.TableNameInfo{tblName}, tableNames)
 	tables, err = m.ListTables(1)
 	require.NoError(t, err)
 	require.Equal(t, []*model.TableInfo{tbInfo}, tables)
diff --git a/pkg/parser/model/model.go b/pkg/parser/model/model.go
index 77e5aec0a9e03..831bb85946b77 100644
--- a/pkg/parser/model/model.go
+++ b/pkg/parser/model/model.go
@@ -543,6 +543,12 @@ type TableInfo struct {
 	TTLInfo *TTLInfo `json:"ttl_info"`
 }
 
+// TableNameInfo provides meta data describing a table name info.
+type TableNameInfo struct {
+	ID   int64 `json:"id"`
+	Name CIStr `json:"name"`
+}
+
 // SepAutoInc decides whether _rowid and auto_increment id use separate allocator.
 func (t *TableInfo) SepAutoInc() bool {
 	return t.Version >= TableInfoVersion5 && t.AutoIdCache == 1