executor: check ddl history, timestamp and privilege for flashback

ddl/BUILD.bazel

@@ -13,6 +13,7 @@ go_library(
     srcs = [
         "backfilling.go",
         "callback.go",
+        "cluster.go",
         "column.go",
         "constant.go",
         "ddl.go",

ddl/cluster.go

@@ -0,0 +1,68 @@
+// Copyright 2022 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 (
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/meta"
+	"github.com/pingcap/tidb/parser/model"
+)
+
+func (w *worker) onFlashbackCluster(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error) {
+	var flashbackTS uint64
+	if err := job.DecodeArgs(&flashbackTS); err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	txn, err := d.store.Begin()
+	if err != nil {
+		return ver, errors.Trace(err)
+	}
+
+	fn := func(jobs []*model.Job) (bool, error) {
+		for _, j := range jobs {
+			if j.ID == job.ID {
+				continue
+			}
+			if !j.IsFinished() {
+				return true, errors.Errorf("Can't flashback cluster, other ddl jobs in ddl queue")
+			}
+			if j.Type != model.ActionFlashbackCluster && j.BinlogInfo != nil && j.BinlogInfo.FinishedTS >= flashbackTS {
+				return true, errors.Errorf("Can't flashback cluster because of ddl history")
+			}
+			if j.BinlogInfo != nil && j.BinlogInfo.FinishedTS < flashbackTS {
+				return true, nil
+			}
+		}
+		return false, nil
+	}
+
+	ctx, err := w.sessPool.get()
+	if err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+	defer w.sessPool.put(ctx)
+
+	if err = IterAllDDLJobs(ctx, txn, fn); err != nil {
+		job.State = model.JobStateCancelled
+		return ver, errors.Trace(err)
+	}
+
+	job.State = model.JobStateDone
+
+	return ver, errors.Trace(err)
+}

ddl/ddl.go

@@ -135,6 +135,7 @@ type DDL interface {
 	CreatePlacementPolicy(ctx sessionctx.Context, stmt *ast.CreatePlacementPolicyStmt) error
 	DropPlacementPolicy(ctx sessionctx.Context, stmt *ast.DropPlacementPolicyStmt) error
 	AlterPlacementPolicy(ctx sessionctx.Context, stmt *ast.AlterPlacementPolicyStmt) error
+	FlashbackCluster(ctx sessionctx.Context, flashbackTS uint64) error
 
 	// CreateSchemaWithInfo creates a database (schema) given its database info.
 	//

ddl/ddl_api.go

@@ -2593,6 +2593,17 @@ func (d *ddl) preSplitAndScatter(ctx sessionctx.Context, tbInfo *model.TableInfo
 	}
 }
 
+func (d *ddl) FlashbackCluster(ctx sessionctx.Context, flashbackTS uint64) error {
+	job := &model.Job{
+		Type:       model.ActionFlashbackCluster,
+		BinlogInfo: &model.HistoryInfo{},
+		Args:       []interface{}{flashbackTS},
+	}
+	err := d.DoDDLJob(ctx, job)
+	err = d.callHookOnChanged(job, err)
+	return errors.Trace(err)
+}
+
 func (d *ddl) RecoverTable(ctx sessionctx.Context, recoverInfo *RecoverInfo) (err error) {
 	is := d.GetInfoSchemaWithInterceptor(ctx)
 	schemaID, tbInfo := recoverInfo.SchemaID, recoverInfo.TableInfo

ddl/ddl_worker.go

@@ -1208,6 +1208,8 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 		ver, err = onAlterCacheTable(d, t, job)
 	case model.ActionAlterNoCacheTable:
 		ver, err = onAlterNoCacheTable(d, t, job)
+	case model.ActionFlashbackCluster:
+		ver, err = w.onFlashbackCluster(d, t, job)
 	case model.ActionMultiSchemaChange:
 		ver, err = onMultiSchemaChange(w, d, t, job)
 	default:

ddl/schematracker/checker.go

@@ -268,6 +268,12 @@ func (d Checker) RecoverTable(ctx sessionctx.Context, recoverInfo *ddl.RecoverIn
 	panic("implement me")
 }
 
+// FlashbackCluster implements the DDL interface.
+func (d Checker) FlashbackCluster(ctx sessionctx.Context, flashbackTS uint64) (err error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 // DropView implements the DDL interface.
 func (d Checker) DropView(ctx sessionctx.Context, stmt *ast.DropTableStmt) (err error) {
 	err = d.realDDL.DropView(ctx, stmt)

ddl/schematracker/dm_tracker.go

@@ -306,6 +306,11 @@ func (d SchemaTracker) RecoverTable(ctx sessionctx.Context, recoverInfo *ddl.Rec
 	return nil
 }
 
+// FlashbackCluster implements the DDL interface, which is no-op in DM's case.
+func (d SchemaTracker) FlashbackCluster(ctx sessionctx.Context, flashbackTS uint64) (err error) {
+	return nil
+}
+
 // DropView implements the DDL interface.
 func (d SchemaTracker) DropView(ctx sessionctx.Context, stmt *ast.DropTableStmt) (err error) {
 	notExistTables := make([]string, 0, len(stmt.Tables))

executor/BUILD.bazel

@@ -296,7 +296,7 @@ go_test(
         "pkg_test.go",
         "point_get_test.go",
         "prepared_test.go",
-        "recover_table_test.go",
+        "recover_test.go",
         "resource_tag_test.go",
         "revoke_test.go",
         "rowid_test.go",

executor/ddl.go

@@ -25,19 +25,24 @@ import (
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/parser/ast"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/terror"
 	"github.com/pingcap/tidb/planner/core"
+	"github.com/pingcap/tidb/privilege"
+	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/sessiontxn"
+	"github.com/pingcap/tidb/sessiontxn/staleread"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/table/temptable"
 	"github.com/pingcap/tidb/util/chunk"
 	"github.com/pingcap/tidb/util/dbterror"
 	"github.com/pingcap/tidb/util/gcutil"
 	"github.com/pingcap/tidb/util/logutil"
+	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
 )
 
@@ -59,8 +64,8 @@ func (e *DDLExec) toErr(err error) error {
 	checker := domain.NewSchemaChecker(dom, e.is.SchemaMetaVersion(), nil)
 	txn, err1 := e.ctx.Txn(true)
 	if err1 != nil {
-		logutil.BgLogger().Error("active txn failed", zap.Error(err))
-		return err1
+		logutil.BgLogger().Error("active txn failed", zap.Error(err1))
+		return err
 	}
 	_, schemaInfoErr := checker.Check(txn.StartTS())
 	if schemaInfoErr != nil {
@@ -170,7 +175,7 @@ func (e *DDLExec) Next(ctx context.Context, req *chunk.Chunk) (err error) {
 	case *ast.FlashBackTableStmt:
 		err = e.executeFlashbackTable(x)
 	case *ast.FlashBackClusterStmt:
-		err = e.executeFlashBackCluster(x)
+		err = e.executeFlashBackCluster(ctx, x)
 	case *ast.RenameTableStmt:
 		err = e.executeRenameTable(x)
 	case *ast.TruncateTableStmt:
@@ -516,8 +521,52 @@ func (e *DDLExec) getRecoverTableByTableName(tableName *ast.TableName) (*model.J
 	return jobInfo, tableInfo, nil
 }
 
-func (e *DDLExec) executeFlashBackCluster(s *ast.FlashBackClusterStmt) error {
-	return dbterror.ErrGeneralUnsupportedDDL.GenWithStackByArgs("FLASHBACK CLUSTER")
+// ValidateFlashBackTS validates that readTS does not exceed the current time not strictly.
+func ValidateFlashBackTS(ctx context.Context, sctx sessionctx.Context, flashBackTS uint64) error {
+	currentTS, err := sctx.GetStore().GetOracle().GetStaleTimestamp(ctx, oracle.GlobalTxnScope, 0)
+	// If we fail to calculate currentTS from local time, fallback to get a timestamp from PD
+	if err != nil {
+		metrics.ValidateReadTSFromPDCount.Inc()
+		currentVer, err := sctx.GetStore().CurrentVersion(oracle.GlobalTxnScope)
+		if err != nil {
+			return errors.Errorf("fail to validate flashback timestamp: %v", err)
+		}
+		currentTS = currentVer.Ver
+	}
+	if oracle.GetTimeFromTS(flashBackTS).After(oracle.GetTimeFromTS(currentTS)) {
+		return errors.Errorf("cannot set flashback timestamp to future time")
+	}
+	gcSafePoint, err := gcutil.GetGCSafePoint(sctx)
+	if err != nil {
+		return err
+	}
+
+	return gcutil.ValidateSnapshotWithGCSafePoint(flashBackTS, gcSafePoint)
+}
+
+func (e *DDLExec) executeFlashBackCluster(ctx context.Context, s *ast.FlashBackClusterStmt) error {
+	checker := privilege.GetPrivilegeManager(e.ctx)
+	if !checker.RequestVerification(e.ctx.GetSessionVars().ActiveRoles, "", "", "", mysql.SuperPriv) {
+		return core.ErrSpecificAccessDenied.GenWithStackByArgs("SUPER")
+	}
+
+	tiFlashInfo, err := getTiFlashStores(e.ctx)
+	if err != nil {
+		return err
+	}
+	if len(tiFlashInfo) != 0 {
+		return errors.Errorf("not support flash back cluster with TiFlash stores")
+	}
+
+	flashbackTS, err := staleread.CalculateAsOfTsExpr(e.ctx, &s.AsOf)
+	if err != nil {
+		return err
+	}
+	if err = ValidateFlashBackTS(ctx, e.ctx, flashbackTS); err != nil {
+		return err
+	}
+
+	return domain.GetDomain(e.ctx).DDL().FlashbackCluster(e.ctx, flashbackTS)
 }
 
 func (e *DDLExec) executeFlashbackTable(s *ast.FlashBackTableStmt) error {

executor/recover_table_test.go → executor/recover_test.go

@@ -23,6 +23,9 @@ import (
 	ddlutil "github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/infoschema"
+	"github.com/pingcap/tidb/parser/auth"
+	"github.com/pingcap/tidb/planner/core"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/testkit"
 	"github.com/pingcap/tidb/util/gcutil"
 	"github.com/stretchr/testify/require"
@@ -289,6 +292,51 @@ func TestRecoverTableMeetError(t *testing.T) {
 	tk.MustContainErrMsg("select * from t_recover", "Table 'test_recover.t_recover' doesn't exist")
 }
 
+func TestRecoverClusterMeetError(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	// Get GC safe point error
+	tk.MustContainErrMsg(fmt.Sprintf("flashback cluster as of timestamp '%s'", time.Now().Add(30*time.Second)), "cannot set flashback timestamp to future time")
+	tk.MustContainErrMsg(fmt.Sprintf("flashback cluster as of timestamp '%s'", time.Now().Add(0-30*time.Second)), "can not get 'tikv_gc_safe_point'")
+
+	timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
+	defer resetGC()
+
+	//set GC safe point
+	tk.MustExec(fmt.Sprintf(safePointSQL, timeBeforeDrop))
+
+	// out of GC safe point range
+	tk.MustGetErrCode(fmt.Sprintf("flashback cluster as of timestamp '%s'", time.Now().Add(0-60*60*60*time.Second)), int(variable.ErrSnapshotTooOld.Code()))
+
+	// Flashback with ddl history
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int)")
+	tk.MustContainErrMsg(fmt.Sprintf("flashback cluster as of timestamp '%s'", time.Now().Add(0-30*time.Second)), "Can't flashback cluster because of ddl history")
+
+	// Flashback without super privilege
+	tk.MustExec(`CREATE USER 'testflashback'@'localhost';`)
+	newTk := testkit.NewTestKit(t, store)
+	require.True(t, newTk.Session().Auth(&auth.UserIdentity{Username: "testflashback", Hostname: "localhost"}, nil, nil))
+	newTk.MustGetErrCode(fmt.Sprintf("flashback cluster as of timestamp '%s'", time.Now().Add(0-30*time.Second)), int(core.ErrSpecificAccessDenied.Code()))
+	tk.MustExec("drop user 'testflashback'@'localhost';")
+}
+
+func TestRecoverClusterWithTiFlash(t *testing.T) {
+	store := testkit.CreateMockStore(t, withMockTiFlash(1))
+	tk := testkit.NewTestKit(t, store)
+
+	timeBeforeDrop, _, safePointSQL, resetGC := MockGC(tk)
+	defer resetGC()
+
+	//set GC safe point
+	tk.MustExec(fmt.Sprintf(safePointSQL, timeBeforeDrop))
+
+	tk.MustContainErrMsg(fmt.Sprintf("flashback cluster as of timestamp '%s'", time.Now().Add(0-30*time.Second)),
+		"not support flash back cluster with TiFlash stores")
+}
+
 // MockGC is used to make GC work in the test environment.
 func MockGC(tk *testkit.TestKit) (string, string, string, func()) {
 	originGC := ddlutil.IsEmulatorGCEnable()

parser/model/ddl.go

@@ -97,6 +97,7 @@ const (
 	ActionCreateTables                  ActionType = 60
 	ActionMultiSchemaChange             ActionType = 61
 	ActionSetTiFlashMode                ActionType = 62
+	ActionFlashbackCluster              ActionType = 63
 )
 
 var actionMap = map[ActionType]string{
@@ -158,6 +159,7 @@ var actionMap = map[ActionType]string{
 	ActionAlterTableStatsOptions:        "alter table statistics options",
 	ActionMultiSchemaChange:             "alter table multi-schema change",
 	ActionSetTiFlashMode:                "set tiflash mode",
+	ActionFlashbackCluster:              "flashback cluster",
 
 	// `ActionAlterTableAlterPartition` is removed and will never be used.
 	// Just left a tombstone here for compatibility.
@@ -678,7 +680,7 @@ func (job *Job) IsDependentOn(other *Job) (bool, error) {
 // IsFinished returns whether job is finished or not.
 // If the job state is Done or Cancelled, it is finished.
 func (job *Job) IsFinished() bool {
-	return job.State == JobStateDone || job.State == JobStateRollbackDone || job.State == JobStateCancelled
+	return job.State == JobStateDone || job.State == JobStateRollbackDone || job.State == JobStateCancelled || job.State == JobStateSynced
 }
 
 // IsCancelled returns whether the job is cancelled or not.