ddl: Implement TableMode feature

pkg/ddl/executor.go

@@ -116,6 +116,7 @@ type Executor interface {
 	RenameTable(ctx sessionctx.Context, stmt *ast.RenameTableStmt) error
 	LockTables(ctx sessionctx.Context, stmt *ast.LockTablesStmt) error
 	UnlockTables(ctx sessionctx.Context, lockedTables []model.TableLockTpInfo) error
+	AlterTableMode(ctx sessionctx.Context, args *model.AlterTableModeArgs) error
 	CleanupTableLock(ctx sessionctx.Context, tables []*ast.TableName) error
 	UpdateTableReplicaInfo(ctx sessionctx.Context, physicalID int64, available bool) error
 	RepairTable(ctx sessionctx.Context, createStmt *ast.CreateTableStmt) error
@@ -1059,6 +1060,15 @@ func (e *executor) createTableWithInfoJob(
 		switch cfg.OnExist {
 		case OnExistIgnore:
 			ctx.GetSessionVars().StmtCtx.AppendNote(err)
+			// If table exists, changing from Normal/Import to Restore is not allowed.
+			if tbInfo.TableMode == model.TableModeRestore {
+				oldTableMode := oldTable.Meta().TableMode
+				if oldTableMode != model.TableModeRestore {
+					return nil, infoschema.ErrTableModeInvalidTransition.GenWithStackByArgs(
+						fmt.Sprintf("invalid transition from '%s' to '%s'", oldTableMode, tbInfo.TableMode),
+					)
+				}
+			}
 			return nil, nil
 		case OnExistReplace:
 			// only CREATE OR REPLACE VIEW is supported at the moment.
@@ -1202,6 +1212,7 @@ func (e *executor) CreateTableWithInfo(
 	if err != nil {
 		// table exists, but if_not_exists flags is true, so we ignore this error.
 		if c.OnExist == OnExistIgnore && infoschema.ErrTableExists.Equal(err) {
+			// TODO(xiaoyuan): I think this branch will never be reached, please see e.createTableWithInfoJob
 			ctx.GetSessionVars().StmtCtx.AppendNote(err)
 			err = nil
 		}
@@ -5646,6 +5657,37 @@ func (e *executor) UnlockTables(ctx sessionctx.Context, unlockTables []model.Tab
 	return errors.Trace(err)
 }
 
+func (e *executor) AlterTableMode(ctx sessionctx.Context, args *model.AlterTableModeArgs) error {
+	is := e.infoCache.GetLatest()
+
+	schema, ok := is.SchemaByID(args.SchemaID)
+	if !ok {
+		return infoschema.ErrDatabaseNotExists.GenWithStackByArgs()
+	}
+
+	t, ok := is.TableByID(e.ctx, args.TableID)
+	if !ok {
+		return infoschema.ErrTableNotExists.GenWithStackByArgs()
+	}
+
+	job := &model.Job{
+		Version:        model.GetJobVerInUse(),
+		SchemaID:       args.SchemaID,
+		TableID:        args.TableID,
+		Type:           model.ActionAlterTableMode,
+		BinlogInfo:     &model.HistoryInfo{},
+		CDCWriteSource: ctx.GetSessionVars().CDCWriteSource,
+		InvolvingSchemaInfo: []model.InvolvingSchemaInfo{
+			{
+				Database: schema.Name.L,
+				Table:    t.Meta().Name.L,
+			},
+		},
+	}
+	err := e.doDDLJob2(ctx, job, args)
+	return errors.Trace(err)
+}
+
 func throwErrIfInMemOrSysDB(ctx sessionctx.Context, dbLowerName string) error {
 	if util.IsMemOrSysDB(dbLowerName) {
 		if ctx.GetSessionVars().User != nil {

pkg/ddl/job_worker.go

@@ -980,6 +980,8 @@ func (w *worker) runOneJobStep(
 		ver, err = onLockTables(jobCtx, job)
 	case model.ActionUnlockTable:
 		ver, err = onUnlockTables(jobCtx, job)
+	case model.ActionAlterTableMode:
+		ver, err = onAlterTableMode(jobCtx, job)
 	case model.ActionSetTiFlashReplica:
 		ver, err = w.onSetTableFlashReplica(jobCtx, job)
 	case model.ActionUpdateTiFlashReplicaStatus:

pkg/ddl/schematracker/checker.go

@@ -398,6 +398,11 @@ func (d *Checker) UnlockTables(ctx sessionctx.Context, lockedTables []model.Tabl
 	return d.realExecutor.UnlockTables(ctx, lockedTables)
 }
 
+// AlterTableMode implements the DDL interface.
+func (d *Checker) AlterTableMode(ctx sessionctx.Context, args *model.AlterTableModeArgs) error {
+	return d.realExecutor.AlterTableMode(ctx, args)
+}
+
 // CleanupTableLock implements the DDL interface.
 func (d *Checker) CleanupTableLock(ctx sessionctx.Context, tables []*ast.TableName) error {
 	return d.realExecutor.CleanupTableLock(ctx, tables)

pkg/ddl/schematracker/dm_tracker.go

@@ -1118,6 +1118,11 @@ func (*SchemaTracker) UnlockTables(_ sessionctx.Context, _ []model.TableLockTpIn
 	return nil
 }
 
+// AlterTableMode implements the DDL interface, it's no-op in DM's case.
+func (*SchemaTracker) AlterTableMode(_ sessionctx.Context, _ *model.AlterTableModeArgs) error {
+	return nil
+}
+
 // CleanupTableLock implements the DDL interface, it's no-op in DM's case.
 func (*SchemaTracker) CleanupTableLock(_ sessionctx.Context, _ []*ast.TableName) error {
 	return nil

pkg/ddl/table_mode.go

@@ -0,0 +1,67 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ddl
+
+import (
+	"fmt"
+	"github.com/pingcap/tidb/pkg/infoschema"
+	"github.com/pingcap/tidb/pkg/meta/model"
+)
+
+// onAlterTableMode should only be called by alterTableMode, will call updateVersionAndTableInfo
+func onAlterTableMode(jobCtx *jobContext, job *model.Job) (ver int64, err error) {
+	args, err := model.GetAlterTableModeArgs(job)
+	var tbInfo *model.TableInfo
+	metaMut := jobCtx.metaMut
+	tbInfo, err = GetTableInfoAndCancelFaultJob(metaMut, job, job.SchemaID)
+	if err != nil {
+		return ver, err
+	}
+
+	switch tbInfo.TableMode {
+	case model.TableModeNormal, model.TableModeImport, model.TableModeRestore:
+		// directly change table mode to target mode
+		err = alterTableMode(tbInfo, args)
+		if err != nil {
+			job.State = model.JobStateCancelled
+		} else {
+			// update table info and schema version
+			ver, err = updateVersionAndTableInfo(jobCtx, job, tbInfo, true)
+			job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tbInfo) // TODO: change of schema state
+		}
+	default:
+		job.State = model.JobStateCancelled
+		err = infoschema.ErrTableModeInvalidTransition.GenWithStackByArgs(
+			fmt.Sprintf("invalid transition from '%s' to '%s'", tbInfo.TableMode, args.TableMode),
+		)
+	}
+
+	return ver, err
+}
+
+// alterTableMode first checks if the change is valid and changes table mode to target mode
+func alterTableMode(tbInfo *model.TableInfo, args *model.AlterTableModeArgs) error {
+	// currently we can assume args.TableMode will not be model.TableModeRestore
+	if args.TableMode == model.TableModeImport {
+		// only transition from ModeNormal to ModeImport is allowed
+		if tbInfo.TableMode != model.TableModeNormal {
+			return infoschema.ErrTableModeInvalidTransition.GenWithStackByArgs(
+				fmt.Sprintf("invalid transition from '%s' to '%s'", tbInfo.TableMode, args.TableMode),
+			)
+		}
+	}
+	tbInfo.TableMode = args.TableMode
+	return nil
+}

pkg/executor/infoschema_reader.go

@@ -674,31 +674,32 @@ func (e *memtableRetriever) setDataFromOneTable(
 		}
 
 		record := types.MakeDatums(
-			infoschema.CatalogVal, // TABLE_CATALOG
-			schema.O,              // TABLE_SCHEMA
-			table.Name.O,          // TABLE_NAME
-			tableType,             // TABLE_TYPE
-			"InnoDB",              // ENGINE
-			uint64(10),            // VERSION
-			"Compact",             // ROW_FORMAT
-			rowCount,              // TABLE_ROWS
-			avgRowLength,          // AVG_ROW_LENGTH
-			dataLength,            // DATA_LENGTH
-			uint64(0),             // MAX_DATA_LENGTH
-			indexLength,           // INDEX_LENGTH
-			uint64(0),             // DATA_FREE
-			autoIncID,             // AUTO_INCREMENT
-			createTime,            // CREATE_TIME
-			nil,                   // UPDATE_TIME
-			nil,                   // CHECK_TIME
-			collation,             // TABLE_COLLATION
-			nil,                   // CHECKSUM
-			createOptions,         // CREATE_OPTIONS
-			table.Comment,         // TABLE_COMMENT
-			table.ID,              // TIDB_TABLE_ID
-			shardingInfo,          // TIDB_ROW_ID_SHARDING_INFO
-			pkType,                // TIDB_PK_TYPE
-			policyName,            // TIDB_PLACEMENT_POLICY_NAME
+			infoschema.CatalogVal,    // TABLE_CATALOG
+			schema.O,                 // TABLE_SCHEMA
+			table.Name.O,             // TABLE_NAME
+			tableType,                // TABLE_TYPE
+			"InnoDB",                 // ENGINE
+			uint64(10),               // VERSION
+			"Compact",                // ROW_FORMAT
+			rowCount,                 // TABLE_ROWS
+			avgRowLength,             // AVG_ROW_LENGTH
+			dataLength,               // DATA_LENGTH
+			uint64(0),                // MAX_DATA_LENGTH
+			indexLength,              // INDEX_LENGTH
+			uint64(0),                // DATA_FREE
+			autoIncID,                // AUTO_INCREMENT
+			createTime,               // CREATE_TIME
+			nil,                      // UPDATE_TIME
+			nil,                      // CHECK_TIME
+			collation,                // TABLE_COLLATION
+			nil,                      // CHECKSUM
+			createOptions,            // CREATE_OPTIONS
+			table.Comment,            // TABLE_COMMENT
+			table.ID,                 // TIDB_TABLE_ID
+			shardingInfo,             // TIDB_ROW_ID_SHARDING_INFO
+			pkType,                   // TIDB_PK_TYPE
+			policyName,               // TIDB_PLACEMENT_POLICY_NAME
+			table.TableMode.String(), // TIDB_TABLE_MODE
 		)
 		rows = append(rows, record)
 	} else {
@@ -728,6 +729,7 @@ func (e *memtableRetriever) setDataFromOneTable(
 			nil,                   // TIDB_ROW_ID_SHARDING_INFO
 			pkType,                // TIDB_PK_TYPE
 			nil,                   // TIDB_PLACEMENT_POLICY_NAME
+			nil,                   // TIDB_TABLE_MODE
 		)
 		rows = append(rows, record)
 	}
@@ -831,6 +833,7 @@ func (e *memtableRetriever) setDataFromTables(ctx context.Context, sctx sessionc
 					nil,                   // TIDB_ROW_ID_SHARDING_INFO
 					nil,                   // TIDB_PK_TYPE
 					nil,                   // TIDB_PLACEMENT_POLICY_NAME
+					nil,                   // TIDB_TABLE_MODE
 				)
 				rows = append(rows, record)
 				return true

pkg/executor/infoschema_reader_test.go

@@ -533,6 +533,10 @@ func TestTablesTable(t *testing.T) {
 		}
 	}
 
+	// test table mode
+	tk.MustQuery(`select tidb_table_mode from information_schema.tables where table_schema = 'db1' and
+		table_name = 't1'`).Check(testkit.Rows("ModeNormal"))
+
 	// Predicates are extracted in CNF, so we separate the test cases by the number of disjunctions in the predicate.
 
 	// predicate covers one disjunction

pkg/infoschema/error.go

@@ -108,4 +108,11 @@ var (
 	ErrResourceGroupSupportDisabled = dbterror.ClassSchema.NewStd(mysql.ErrResourceGroupSupportDisabled)
 	// ErrCheckConstraintDupName returns for duplicate constraint names.
 	ErrCheckConstraintDupName = dbterror.ClassSchema.NewStd(mysql.ErrCheckConstraintDupName)
+	// TODO(xiaoyuan): what is the proper mysql error I should use for TableMode?
+	// ErrTableModeImport returns for accessing table in import mode.
+	ErrTableModeImport = dbterror.ClassSchema.NewStd(mysql.ErrTableLocked)
+	// ErrTableModeRestore returns for accessing table in restore mode.
+	ErrTableModeRestore = dbterror.ClassSchema.NewStd(mysql.ErrTableLocked)
+	// ErrTableModeInvalidTransition returns for invalid TableMode transition.
+	ErrTableModeInvalidTransition = dbterror.ClassSchema.NewStd(mysql.ErrTableLocked)
 )

pkg/infoschema/tables.go

@@ -466,6 +466,7 @@ var tablesCols = []columnInfo{
 	{name: "TIDB_ROW_ID_SHARDING_INFO", tp: mysql.TypeVarchar, size: 255},
 	{name: "TIDB_PK_TYPE", tp: mysql.TypeVarchar, size: 64},
 	{name: "TIDB_PLACEMENT_POLICY_NAME", tp: mysql.TypeVarchar, size: 64},
+	{name: "TIDB_TABLE_MODE", tp: mysql.TypeVarchar, size: 16},
 }
 
 // See: http://dev.mysql.com/doc/refman/5.7/en/information-schema-columns-table.html

pkg/meta/model/job.go

@@ -111,6 +111,7 @@ const (
 	ActionAlterTablePartitioning ActionType = 71
 	ActionRemovePartitioning     ActionType = 72
 	ActionAddVectorIndex         ActionType = 73
+	ActionAlterTableMode         ActionType = 74
 )
 
 // ActionMap is the map of DDL ActionType to string.

pkg/meta/model/job_args.go

@@ -1083,6 +1083,25 @@ func GetLockTablesArgs(job *Job) (*LockTablesArgs, error) {
 	return getOrDecodeArgs[*LockTablesArgs](&LockTablesArgs{}, job)
 }
 
+type AlterTableModeArgs struct {
+	TableMode TableModeState
+	SchemaID  int64
+	TableID   int64
+}
+
+func (a *AlterTableModeArgs) getArgsV1(*Job) []any {
+	return []any{a}
+}
+
+func (a *AlterTableModeArgs) decodeV1(job *Job) error {
+	return errors.Trace(job.decodeArgs(a))
+}
+
+// GetAlterTableModeArgs get the AlterTableModeArgs argument.
+func GetAlterTableModeArgs(job *Job) (*AlterTableModeArgs, error) {
+	return getOrDecodeArgs[*AlterTableModeArgs](&AlterTableModeArgs{}, job)
+}
+
 // RepairTableArgs is the argument for repair table
 type RepairTableArgs struct {
 	TableInfo *TableInfo `json:"table_info"`

pkg/meta/model/table.go

@@ -195,6 +195,8 @@ type TableInfo struct {
 	Revision uint64 `json:"revision"`
 
 	DBID int64 `json:"-"`
+
+	TableMode TableModeState `json:"table_mode"`
 }
 
 // Hash64 implement HashEquals interface.
@@ -647,6 +649,28 @@ const (
 	TableLockStatePublic
 )
 
+type TableModeState byte
+
+const (
+	TableModeNormal TableModeState = iota
+	TableModeImport
+	TableModeRestore
+)
+
+// String implements fmt.Stringer interface.
+func (t TableModeState) String() string {
+	switch t {
+	case TableModeNormal:
+		return "ModeNormal"
+	case TableModeImport:
+		return "ModeImport"
+	case TableModeRestore:
+		return "ModeRestore"
+	default:
+		return ""
+	}
+}
+
 // String implements fmt.Stringer interface.
 func (t TableLockState) String() string {
 	switch t {

pkg/planner/core/optimizer.go

@@ -247,6 +247,25 @@ func CheckTableLock(ctx tablelock.TableLockReadContext, is infoschema.InfoSchema
 	return nil
 }
 
+// CheckTableMode checks if the table is accessible by table mode.
+func CheckTableMode(is infoschema.InfoSchema, vs []visitInfo) error {
+	for i := range vs {
+		tb, err := is.TableByName(context.Background(), ast.NewCIStr(vs[i].db), ast.NewCIStr(vs[i].table))
+		if infoschema.ErrTableNotExists.Equal(err) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		if tb.Meta().TableMode == model.TableModeImport {
+			return infoschema.ErrTableModeImport.GenWithStackByArgs(tb.Meta().Name.O)
+		} else if tb.Meta().TableMode == model.TableModeRestore {
+			return infoschema.ErrTableModeRestore.GenWithStackByArgs(tb.Meta().Name.O)
+		}
+	}
+	return nil
+}
+
 func checkStableResultMode(sctx base.PlanContext) bool {
 	s := sctx.GetSessionVars()
 	st := s.StmtCtx

pkg/planner/optimize.go

@@ -474,6 +474,10 @@ func optimize(ctx context.Context, sctx planctx.PlanContext, node *resolve.NodeW
 		return nil, nil, 0, err
 	}
 
+	if err := core.CheckTableMode(is, builder.GetVisitInfo()); err != nil {
+		return nil, nil, 0, err
+	}
+
 	names := p.OutputNames()
 
 	// Handle the non-logical plan statement.