ddl: support refresh meta ddl

pkg/ddl/executor.go

@@ -134,6 +134,8 @@ type Executor interface {
 	AlterResourceGroup(ctx sessionctx.Context, stmt *ast.AlterResourceGroupStmt) error
 	DropResourceGroup(ctx sessionctx.Context, stmt *ast.DropResourceGroupStmt) error
 	FlashbackCluster(ctx sessionctx.Context, flashbackTS uint64) error
+	// RefreshMeta can only be called by BR during the log restore phase.
+	RefreshMeta(ctx sessionctx.Context, args *model.RefreshMetaArgs) error
 
 	// CreateSchemaWithInfo creates a database (schema) given its database info.
 	//
@@ -7058,3 +7060,29 @@ func NewDDLReorgMeta(ctx sessionctx.Context) *model.DDLReorgMeta {
 		Version:           model.CurrentReorgMetaVersion,
 	}
 }
+
+// RefreshMeta is a internal DDL job. In some cases, BR log restore will
+// exchange table partition in meta kv directly, and table info in meta kv
+// is not consistent with info schema. So when BR call AlterTableMode for new
+// table will failure. RefreshMeta will reload schema diff to update info schema
+// by schema ID and table ID.
+func (e *executor) RefreshMeta(sctx sessionctx.Context, args *model.RefreshMetaArgs) error {
+	is := e.infoCache.GetLatest()
+	_, ok := is.SchemaByID(args.SchemaID)
+	if !ok {
+		return infoschema.ErrDatabaseNotExists.GenWithStackByArgs(fmt.Sprintf("SchemaID: %v", args.SchemaID))
+	}
+
+	job := &model.Job{
+		Version:        model.JobVersion2,
+		SchemaID:       args.SchemaID,
+		TableID:        args.TableID,
+		Type:           model.ActionRefreshMeta,
+		BinlogInfo:     &model.HistoryInfo{},
+		CDCWriteSource: sctx.GetSessionVars().CDCWriteSource,
+		SQLMode:        sctx.GetSessionVars().SQLMode,
+	}
+	sctx.SetValue(sessionctx.QueryString, "skip")
+	err := e.doDDLJob2(sctx, job, args)
+	return errors.Trace(err)
+}

pkg/ddl/job_worker.go

@@ -1046,6 +1046,8 @@ func (w *worker) runOneJobStep(
 		ver, err = onDropCheckConstraint(jobCtx, job)
 	case model.ActionAlterCheckConstraint:
 		ver, err = w.onAlterCheckConstraint(jobCtx, job)
+	case model.ActionRefreshMeta:
+		ver, err = onRefreshMeta(jobCtx, job)
 	default:
 		// Invalid job, cancel it.
 		job.State = model.JobStateCancelled

pkg/ddl/schematracker/checker.go

@@ -404,6 +404,11 @@ func (d *Checker) AlterTableMode(ctx sessionctx.Context, args *model.AlterTableM
 	return d.realExecutor.AlterTableMode(ctx, args)
 }
 
+// RefreshMeta implements the DDL interface.
+func (d *Checker) RefreshMeta(ctx sessionctx.Context, args *model.RefreshMetaArgs) error {
+	return d.realExecutor.RefreshMeta(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

@@ -1197,3 +1197,8 @@ func (d *SchemaTracker) BatchCreateTableWithInfo(ctx sessionctx.Context, schema
 func (*SchemaTracker) CreatePlacementPolicyWithInfo(_ sessionctx.Context, _ *model.PolicyInfo, _ ddl.OnExist) error {
 	return nil
 }
+
+// RefreshMeta implements the DDL interface, it's no-op in DM's case.
+func (*SchemaTracker) RefreshMeta(_ sessionctx.Context, _ *model.RefreshMetaArgs) error {
+	return nil
+}

pkg/ddl/table.go

@@ -1699,3 +1699,25 @@ func onAlterNoCacheTable(jobCtx *jobContext, job *model.Job) (ver int64, err err
 	}
 	return ver, err
 }
+
+func onRefreshMeta(jobCtx *jobContext, job *model.Job) (ver int64, err error) {
+	_, err = model.GetRefreshMetaArgs(job)
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return 0, errors.Trace(err)
+	}
+	// update schema version
+	ver, err = updateSchemaVersion(jobCtx, job)
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	var tbInfo *model.TableInfo
+	metaMut := jobCtx.metaMut
+	tbInfo, err = GetTableInfoAndCancelFaultJob(metaMut, job, job.SchemaID)
+	if err != nil {
+		return ver, err
+	}
+	job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tbInfo)
+	return ver, nil
+}

pkg/ddl/table_mode.go

@@ -15,6 +15,7 @@
 package ddl
 
 import (
+	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/meta/model"
 )
@@ -47,6 +48,9 @@ func onAlterTableMode(jobCtx *jobContext, job *model.Job) (ver int64, err error)
 		}
 		// update table info and schema version
 		ver, err = updateVersionAndTableInfo(jobCtx, job, tbInfo, true)
+		if err != nil {
+			return ver, errors.Trace(err)
+		}
 		job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tbInfo)
 	default:
 		job.State = model.JobStateCancelled

pkg/ddl/table_mode_test.go

@@ -16,6 +16,7 @@ package ddl_test
 
 import (
 	"context"
+	"strings"
 	"sync"
 	"testing"
 
@@ -24,9 +25,12 @@ import (
 	"github.com/pingcap/tidb/pkg/ddl/testutil"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/errno"
+	"github.com/pingcap/tidb/pkg/kv"
+	"github.com/pingcap/tidb/pkg/meta"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/terror"
+	"github.com/pingcap/tidb/pkg/table"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/stretchr/testify/require"
 )
@@ -300,3 +304,66 @@ func TestTableModeConcurrent(t *testing.T) {
 	require.NotNil(t, failedErr3)
 	checkErrorCode(t, failedErr3, errno.ErrInvalidTableModeSet)
 }
+
+// TestTableModeWithRefreshMeta tests update table meta by txn(exchange partition ID),
+// after RefreshMeta can modify TableMode.
+func TestTableModeWithRefreshMeta(t *testing.T) {
+	store, domain := testkit.CreateMockStoreAndDomain(t)
+	de := domain.DDLExecutor()
+	tk := testkit.NewTestKit(t, store)
+	sctx := testkit.NewTestKit(t, store).Session()
+
+	tk.MustExec("use test")
+	tk.MustExec("create table nt(id int, c1 int)")
+	tk.MustExec("create table pt(id int, c1 int) partition by range (c1) (partition p10 values less than (10))")
+	tk.MustExec("insert into nt values(3, 3), (4, 4), (5, 5)")
+	tk.MustExec("insert into pt values(1, 1), (2, 2)")
+
+	dbInfo, ok := domain.InfoSchema().SchemaByName(ast.NewCIStr("test"))
+	require.True(t, ok)
+	require.NotNil(t, dbInfo)
+	ntInfo, ptInfo := getClonedTableInfoFromDomain(t, "test", "nt", domain), getClonedTableInfoFromDomain(t, "test", "pt", domain)
+	// change non-partition table ID to partition ID
+	partID := ptInfo.Partition.Definitions[0].ID
+	recreateTableWithPartitionID(t, &store, dbInfo.ID, ntInfo, ptInfo, "p10")
+	ntInfo = testutil.GetTableInfoByTxn(t, store, dbInfo.ID, ntInfo.ID)
+	require.Equal(t, partID, ntInfo.ID)
+	// set table mode failure before refresh meta
+	err := testutil.SetTableMode(sctx, t, store, de, dbInfo, ntInfo, model.TableModeImport)
+	require.ErrorContains(t, err, "doesn't exist")
+	testutil.RefreshMeta(sctx, t, de, dbInfo.ID, ntInfo.ID)
+	// set table mode success after refresh meta
+	err = testutil.SetTableMode(sctx, t, store, de, dbInfo, ntInfo, model.TableModeImport)
+	require.NoError(t, err)
+	tk.MustGetErrCode("select * from nt", errno.ErrProtectedTableMode)
+	err = testutil.SetTableMode(sctx, t, store, de, dbInfo, ntInfo, model.TableModeNormal)
+	require.NoError(t, err)
+	tk.MustExec("select * from nt")
+}
+
+// recreateTableWithPartitionID update table ID to partition ID and recreate table.
+func recreateTableWithPartitionID(t *testing.T, store *kv.Storage, dbID int64, ntInfo, ptInfo *model.TableInfo, partName string) {
+	_, partDef, err := getPartitionDef(ptInfo, partName)
+	require.NoError(t, err)
+	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL)
+	err = kv.RunInNewTxn(ctx, *store, true, func(ctx context.Context, txn kv.Transaction) error {
+		m := meta.NewMutator(txn)
+		err := m.DropTableOrView(dbID, ntInfo.ID)
+		require.NoError(t, err)
+		ntInfo.ID = partDef.ID
+		err = m.CreateTableOrView(dbID, ntInfo)
+		require.NoError(t, err)
+		return nil
+	})
+	require.NoError(t, err)
+}
+
+func getPartitionDef(tblInfo *model.TableInfo, partName string) (index int, def *model.PartitionDefinition, _ error) {
+	defs := tblInfo.Partition.Definitions
+	for i := 0; i < len(defs); i++ {
+		if strings.EqualFold(defs[i].Name.L, strings.ToLower(partName)) {
+			return i, &(defs[i]), nil
+		}
+	}
+	return index, nil, table.ErrUnknownPartition.GenWithStackByArgs(partName, tblInfo.Name.O)
+}

pkg/ddl/table_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/ddl"
+	"github.com/pingcap/tidb/pkg/ddl/testutil"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta"
@@ -836,3 +837,32 @@ func TestIssue59238(t *testing.T) {
 	tk.MustExec("alter table t exchange partition p1 with table t1")
 	require.True(t, tk.MustQuery("select distinct create_time from information_schema.partitions where table_name = 't'").Equal(testkit.Rows(rs)))
 }
+
+func TestRefreshMetaBasic(t *testing.T) {
+	store, domain := testkit.CreateMockStoreAndDomain(t)
+	de := domain.DDLExecutor()
+	tk := testkit.NewTestKit(t, store)
+	sctx := testkit.NewTestKit(t, store).Session()
+
+	// get t1 table info
+	tk.MustExec("use test")
+	tk.MustExec("create table t1(id int)")
+	dbInfo, ok := domain.InfoSchema().SchemaByName(ast.NewCIStr("test"))
+	require.True(t, ok)
+	t1TableInfo := getClonedTableInfoFromDomain(t, "test", "t1", domain)
+	// update t1 table name to t2 by txn
+	t1TableInfo.Name = ast.NewCIStr("t2")
+	updateTableMeta(t, store, dbInfo.ID, t1TableInfo)
+	t2TableInfo := testutil.GetTableInfoByTxn(t, store, dbInfo.ID, t1TableInfo.ID)
+	require.Equal(t, t1TableInfo, t2TableInfo)
+	// validate infoschema doesn't conatain t2 table info
+	_, err := domain.InfoSchema().TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t2"))
+	require.ErrorContains(t, err, "Table 'test.t2' doesn't exist")
+	// refresh meta, validate infoschema store table t2 and schema version increase 1
+	oldSchemaVer := getSchemaVer(t, sctx)
+	testutil.RefreshMeta(sctx, t, de, dbInfo.ID, t1TableInfo.ID)
+	newSchemaVer := getSchemaVer(t, sctx)
+	require.Equal(t, oldSchemaVer+1, newSchemaVer)
+	_, err = domain.InfoSchema().TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t2"))
+	require.NoError(t, err)
+}

pkg/ddl/testutil/testutil.go

@@ -187,3 +187,36 @@ func SetTableMode(
 
 	return err
 }
+
+// GetTableInfoByTxn get table info by transaction.
+func GetTableInfoByTxn(t *testing.T, store kv.Storage, dbID int64, tableID int64) *model.TableInfo {
+	var (
+		tableInfo *model.TableInfo
+		err       error
+	)
+	err = kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(_ context.Context, txn kv.Transaction) error {
+		m := meta.NewMutator(txn)
+		_, err = m.GetDatabase(dbID)
+		require.NoError(t, err)
+		tableInfo, err = m.GetTable(dbID, tableID)
+		require.NoError(t, err)
+		require.NotNil(t, tableInfo)
+		return nil
+	})
+	return tableInfo
+}
+
+// RefreshMeta sets the table mode of a table in the store.
+func RefreshMeta(
+	ctx sessionctx.Context,
+	t *testing.T,
+	de ddl.Executor,
+	dbID, tableID int64,
+) {
+	args := &model.RefreshMetaArgs{
+		SchemaID: dbID,
+		TableID:  tableID,
+	}
+	err := de.RefreshMeta(ctx, args)
+	require.NoError(t, err)
+}

pkg/meta/model/bdr.go

@@ -104,6 +104,7 @@ var BDRActionMap = map[DDLBDRType][]ActionType{
 		ActionAddColumnarIndex,
 		ActionModifyEngineAttribute,
 		ActionAlterTableMode,
+		ActionRefreshMeta,
 	},
 	UnmanagementDDL: {
 		ActionCreatePlacementPolicy,

pkg/meta/model/job.go

@@ -114,6 +114,7 @@ const (
 	ActionAddColumnarIndex       ActionType = 73
 	ActionModifyEngineAttribute  ActionType = 74
 	ActionAlterTableMode         ActionType = 75
+	ActionRefreshMeta            ActionType = 76
 )
 
 // ActionMap is the map of DDL ActionType to string.
@@ -188,6 +189,7 @@ var ActionMap = map[ActionType]string{
 	ActionAddColumnarIndex:              "add columnar index",
 	ActionModifyEngineAttribute:         "modify engine attribute",
 	ActionAlterTableMode:                "alter table mode",
+	ActionRefreshMeta:                   "refresh meta",
 
 	// `ActionAlterTableAlterPartition` is removed and will never be used.
 	// Just left a tombstone here for compatibility.

pkg/meta/model/job_args.go

@@ -1785,3 +1785,22 @@ func GetFinishedModifyColumnArgs(job *Job) (*ModifyColumnArgs, error) {
 	}
 	return getOrDecodeArgsV2[*ModifyColumnArgs](job)
 }
+
+// RefreshMetaArgs is the argument for RefreshMeta.
+type RefreshMetaArgs struct {
+	SchemaID int64 `json:"schema_id,omitempty"`
+	TableID  int64 `json:"table_id,omitempty"`
+}
+
+func (a *RefreshMetaArgs) getArgsV1(*Job) []any {
+	return []any{a}
+}
+
+func (a *RefreshMetaArgs) decodeV1(job *Job) error {
+	return errors.Trace(job.decodeArgs(a))
+}
+
+// GetRefreshMetaArgs get the refresh meta argument.
+func GetRefreshMetaArgs(job *Job) (*RefreshMetaArgs, error) {
+	return getOrDecodeArgs[*RefreshMetaArgs](&RefreshMetaArgs{}, job)
+}