diff --git a/ddl/db_integration_test.go b/ddl/db_integration_test.go
index 39fee9b9d7432..289254f3b2c34 100644
--- a/ddl/db_integration_test.go
+++ b/ddl/db_integration_test.go
@@ -16,6 +16,7 @@ package ddl_test
 import (
 	"context"
 	"fmt"
+	"math"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -24,6 +25,7 @@ import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/ast"
+	"github.com/pingcap/parser/auth"
 	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
@@ -35,6 +37,7 @@ import (
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
+	"github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
@@ -2621,6 +2624,95 @@ func (s *testSerialDBSuite1) TestAutoIncrementTableOption(c *C) {
 	tk.MustQuery("select * from t;").Check(testkit.Rows("12345678901234567890"))
 }
 
+func (s *testIntegrationSuite3) TestAutoIncrementForce(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("drop database if exists auto_inc_force;")
+	tk.MustExec("create database auto_inc_force;")
+	tk.MustExec("use auto_inc_force;")
+	getNextGlobalID := func() uint64 {
+		gidStr := tk.MustQuery("show table t next_row_id").Rows()[0][3]
+		gid, err := strconv.ParseUint(gidStr.(string), 10, 64)
+		c.Assert(err, IsNil)
+		return gid
+	}
+	// Rebase _tidb_row_id.
+	tk.MustExec("create table t (a int);")
+	tk.MustExec("insert into t values (1),(2);")
+	tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("1 1", "2 2"))
+	// Cannot set next global ID to 0.
+	tk.MustGetErrCode("alter table t force auto_increment = 0;", errno.ErrAutoincReadFailed)
+	tk.MustExec("alter table t force auto_increment = 1;")
+	c.Assert(getNextGlobalID(), Equals, uint64(1))
+	// inserting new rows can overwrite the existing data.
+	tk.MustExec("insert into t values (3);")
+	tk.MustExec("insert into t values (3);")
+	tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("3 1", "3 2"))
+
+	// Rebase auto_increment.
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t (a int primary key auto_increment, b int);")
+	tk.MustExec("insert into t values (1, 1);")
+	tk.MustExec("insert into t values (100000000, 1);")
+	tk.MustExec("delete from t where a = 100000000;")
+	c.Assert(getNextGlobalID(), Greater, uint64(100000000))
+	// Cannot set next global ID to 0.
+	tk.MustGetErrCode("alter table t /*T![force_inc] force */ auto_increment = 0;", errno.ErrAutoincReadFailed)
+	tk.MustExec("alter table t /*T![force_inc] force */ auto_increment = 2;")
+	c.Assert(getNextGlobalID(), Equals, uint64(2))
+	tk.MustExec("insert into t(b) values (2);")
+	tk.MustQuery("select a, b from t;").Check(testkit.Rows("1 1", "2 2"))
+
+	// Rebase auto_random.
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t (a bigint primary key auto_random(5));")
+	tk.MustExec("insert into t values ();")
+	tk.MustExec("set @@allow_auto_random_explicit_insert = true")
+	tk.MustExec("insert into t values (100000000);")
+	tk.MustExec("delete from t where a = 100000000;")
+	c.Assert(getNextGlobalID(), Greater, uint64(100000000))
+	// Cannot set next global ID to 0.
+	tk.MustGetErrCode("alter table t force auto_random_base = 0;", errno.ErrAutoincReadFailed)
+	tk.MustExec("alter table t force auto_random_base = 2;")
+	c.Assert(getNextGlobalID(), Equals, uint64(2))
+	tk.MustExec("insert into t values ();")
+	tk.MustQuery("select (a & 3) from t order by 1;").Check(testkit.Rows("1", "2"))
+
+	// Change next global ID.
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t (a bigint primary key auto_increment);")
+	tk.MustExec("insert into t values (1);")
+	bases := []uint64{1, 65535, 10, math.MaxUint64, math.MaxInt64 + 1, 1, math.MaxUint64, math.MaxInt64, 2}
+	lastBase := fmt.Sprintf("%d", bases[len(bases)-1])
+	for _, b := range bases {
+		tk.MustExec(fmt.Sprintf("alter table t force auto_increment = %d;", b))
+		c.Assert(getNextGlobalID(), Equals, b)
+	}
+	tk.MustExec("insert into t values ();")
+	tk.MustQuery("select a from t;").Check(testkit.Rows("1", lastBase))
+	// Force alter unsigned int auto_increment column.
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t (a bigint unsigned primary key auto_increment);")
+	for _, b := range bases {
+		tk.MustExec(fmt.Sprintf("alter table t force auto_increment = %d;", b))
+		c.Assert(getNextGlobalID(), Equals, b)
+		tk.MustExec("insert into t values ();")
+		tk.MustQuery("select a from t;").Check(testkit.Rows(fmt.Sprintf("%d", b)))
+		tk.MustExec("delete from t;")
+	}
+
+	// Force alter with @@auto_increment_increment and @@auto_increment_offset.
+	tk.MustExec("drop table if exists t;")
+	tk.MustExec("create table t(a int key auto_increment);")
+	tk.MustExec("set @@auto_increment_offset=2;")
+	tk.MustExec("set @@auto_increment_increment = 11;")
+	tk.MustExec("insert into t values (500);")
+	tk.MustExec("alter table t force auto_increment=100;")
+	tk.MustExec("insert into t values (), ();")
+	tk.MustQuery("select * from t;").Check(testkit.Rows("101", "112", "500"))
+	tk.MustQuery("select * from t order by a;").Check(testkit.Rows("101", "112", "500"))
+	tk.MustExec("drop table if exists t;")
+}
+
 func (s *testIntegrationSuite3) TestIssue20490(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test;")
@@ -2849,3 +2941,79 @@ func (s *testIntegrationSuite3) TestCreateTemporaryTable(c *C) {
 	tk.MustExec("set @@tidb_snapshot = '2016-01-01 15:04:05.999999'")
 	tk.MustExec("select * from overlap")
 }
+
+func (s *testIntegrationSuite3) TestAvoidCreateViewOnLocalTemporaryTable(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+
+	tk.Se.Auth(&auth.UserIdentity{Username: "root", Hostname: "localhost", CurrentUser: true, AuthUsername: "root", AuthHostname: "%"}, nil, []byte("012345678901234567890"))
+	tk.MustExec("drop table if exists tt0")
+	tk.MustExec("drop table if exists tt1")
+	tk.MustExec("drop table if exists tt2")
+
+	tk.MustExec("set @@tidb_enable_noop_functions=1")
+	tk.MustExec("create table tt0 (a int, b int)")
+	tk.MustExec("create view v0 as select * from tt0")
+	tk.MustExec("create temporary table tt1 (a int, b int)")
+	tk.MustExec("create temporary table tt2 (c int, d int)")
+
+	checkCreateView := func() {
+		_, err := tk.Exec("create view v1 as select * from tt1")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v1")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v1' doesn't exist")
+
+		_, err = tk.Exec("create view v1 as select * from (select * from tt1) as tt")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v1")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v1' doesn't exist")
+
+		_, err = tk.Exec("create view v2 as select * from tt0 union select * from tt1")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v2")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v2' doesn't exist")
+
+		_, err = tk.Exec("create view v3 as select * from tt0, tt1 where tt0.a = tt1.a")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v3")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v3' doesn't exist")
+
+		_, err = tk.Exec("create view v4 as select a, (select count(1) from tt1 where tt1.a = tt0.a) as tt1a from tt0")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v4")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v4' doesn't exist")
+
+		_, err = tk.Exec("create view v5 as select a, (select count(1) from tt1 where tt1.a = 1) as tt1a from tt0")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v5")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v5' doesn't exist")
+
+		_, err = tk.Exec("create view v6 as select * from tt0 where tt0.a=(select max(tt1.b) from tt1)")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v6")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v6' doesn't exist")
+
+		_, err = tk.Exec("create view v7 as select * from tt0 where tt0.b=(select max(tt1.b) from tt1 where tt0.a=tt1.a)")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+		_, err = tk.Exec("select * from v7")
+		c.Assert(err, NotNil)
+		c.Assert(err.Error(), Equals, "[schema:1146]Table 'test.v7' doesn't exist")
+
+		_, err = tk.Exec("create or replace view v0 as select * from tt1")
+		c.Assert(core.ErrViewSelectTemporaryTable.Equal(err), IsTrue)
+	}
+
+	checkCreateView()
+	tk.MustExec("create temporary table tt0 (a int, b int)")
+	tk.MustExec("create table tt1 (a int, b int)")
+	tk.MustExec("create table tt2 (c int, d int)")
+	tk.MustExec("create view vv as select * from v0")
+	checkCreateView()
+}
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
index af6cc2fd13ae7..91fb1a215e337 100644
--- a/ddl/ddl_api.go
+++ b/ddl/ddl_api.go
@@ -2532,7 +2532,7 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 					}
 					err = d.ShardRowID(ctx, ident, opt.UintValue)
 				case ast.TableOptionAutoIncrement:
-					err = d.RebaseAutoID(ctx, ident, int64(opt.UintValue), autoid.RowIDAllocType)
+					err = d.RebaseAutoID(ctx, ident, int64(opt.UintValue), autoid.RowIDAllocType, opt.BoolValue)
 				case ast.TableOptionAutoIdCache:
 					if opt.UintValue > uint64(math.MaxInt64) {
 						// TODO: Refine this error.
@@ -2540,7 +2540,7 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 					}
 					err = d.AlterTableAutoIDCache(ctx, ident, int64(opt.UintValue))
 				case ast.TableOptionAutoRandomBase:
-					err = d.RebaseAutoID(ctx, ident, int64(opt.UintValue), autoid.AutoRandomType)
+					err = d.RebaseAutoID(ctx, ident, int64(opt.UintValue), autoid.AutoRandomType, opt.BoolValue)
 				case ast.TableOptionComment:
 					spec.Comment = opt.StrValue
 					err = d.AlterTableComment(ctx, ident, spec)
@@ -2596,7 +2596,7 @@ func (d *ddl) AlterTable(ctx sessionctx.Context, ident ast.Ident, specs []*ast.A
 	return nil
 }
 
-func (d *ddl) RebaseAutoID(ctx sessionctx.Context, ident ast.Ident, newBase int64, tp autoid.AllocatorType) error {
+func (d *ddl) RebaseAutoID(ctx sessionctx.Context, ident ast.Ident, newBase int64, tp autoid.AllocatorType, force bool) error {
 	schema, t, err := d.getSchemaAndTableByIdent(ctx, ident)
 	if err != nil {
 		return errors.Trace(err)
@@ -2625,17 +2625,11 @@ func (d *ddl) RebaseAutoID(ctx sessionctx.Context, ident ast.Ident, newBase int6
 		actionType = model.ActionRebaseAutoID
 	}
 
-	if alloc := t.Allocators(ctx).Get(tp); alloc != nil {
-		autoID, err := alloc.NextGlobalAutoID(t.Meta().ID)
+	if !force {
+		newBase, err = adjustNewBaseToNextGlobalID(ctx, t, tp, newBase)
 		if err != nil {
-			return errors.Trace(err)
+			return err
 		}
-		// If newBase < autoID, we need to do a rebase before returning.
-		// Assume there are 2 TiDB servers: TiDB-A with allocator range of 0 ~ 30000; TiDB-B with allocator range of 30001 ~ 60000.
-		// If the user sends SQL `alter table t1 auto_increment = 100` to TiDB-B,
-		// and TiDB-B finds 100 < 30001 but returns without any handling,
-		// then TiDB-A may still allocate 99 for auto_increment column. This doesn't make sense for the user.
-		newBase = int64(mathutil.MaxUint64(uint64(newBase), uint64(autoID)))
 	}
 	job := &model.Job{
 		SchemaID:   schema.ID,
@@ -2643,13 +2637,30 @@ func (d *ddl) RebaseAutoID(ctx sessionctx.Context, ident ast.Ident, newBase int6
 		SchemaName: schema.Name.L,
 		Type:       actionType,
 		BinlogInfo: &model.HistoryInfo{},
-		Args:       []interface{}{newBase},
+		Args:       []interface{}{newBase, force},
 	}
 	err = d.doDDLJob(ctx, job)
 	err = d.callHookOnChanged(err)
 	return errors.Trace(err)
 }
 
+func adjustNewBaseToNextGlobalID(ctx sessionctx.Context, t table.Table, tp autoid.AllocatorType, newBase int64) (int64, error) {
+	alloc := t.Allocators(ctx).Get(tp)
+	if alloc == nil {
+		return newBase, nil
+	}
+	autoID, err := alloc.NextGlobalAutoID(t.Meta().ID)
+	if err != nil {
+		return newBase, errors.Trace(err)
+	}
+	// If newBase < autoID, we need to do a rebase before returning.
+	// Assume there are 2 TiDB servers: TiDB-A with allocator range of 0 ~ 30000; TiDB-B with allocator range of 30001 ~ 60000.
+	// If the user sends SQL `alter table t1 auto_increment = 100` to TiDB-B,
+	// and TiDB-B finds 100 < 30001 but returns without any handling,
+	// then TiDB-A may still allocate 99 for auto_increment column. This doesn't make sense for the user.
+	return int64(mathutil.MaxUint64(uint64(newBase), uint64(autoID))), nil
+}
+
 // ShardRowID shards the implicit row ID by adding shard value to the row ID's first few bits.
 func (d *ddl) ShardRowID(ctx sessionctx.Context, tableIdent ast.Ident, uVal uint64) error {
 	schema, t, err := d.getSchemaAndTableByIdent(ctx, tableIdent)
diff --git a/ddl/table.go b/ddl/table.go
index a3962621e5c63..336471c04bd62 100644
--- a/ddl/table.go
+++ b/ddl/table.go
@@ -555,8 +555,11 @@ func onRebaseAutoRandomType(store kv.Storage, t *meta.Meta, job *model.Job) (ver
 
 func onRebaseAutoID(store kv.Storage, t *meta.Meta, job *model.Job, tp autoid.AllocatorType) (ver int64, _ error) {
 	schemaID := job.SchemaID
-	var newBase int64
-	err := job.DecodeArgs(&newBase)
+	var (
+		newBase int64
+		force   bool
+	)
+	err := job.DecodeArgs(&newBase, &force)
 	if err != nil {
 		job.State = model.JobStateCancelled
 		return ver, errors.Trace(err)
@@ -581,8 +584,13 @@ func onRebaseAutoID(store kv.Storage, t *meta.Meta, job *model.Job, tp autoid.Al
 	if alloc := tbl.Allocators(nil).Get(tp); alloc != nil {
 		// The next value to allocate is `newBase`.
 		newEnd := newBase - 1
-		err = alloc.Rebase(tblInfo.ID, newEnd, false)
+		if force {
+			err = alloc.ForceRebase(tblInfo.ID, newEnd)
+		} else {
+			err = alloc.Rebase(tblInfo.ID, newEnd, false)
+		}
 		if err != nil {
+			job.State = model.JobStateCancelled
 			return ver, errors.Trace(err)
 		}
 	}
diff --git a/distsql/distsql_test.go b/distsql/distsql_test.go
index 41e20258152c4..d361f9ddf1ad0 100644
--- a/distsql/distsql_test.go
+++ b/distsql/distsql_test.go
@@ -167,7 +167,7 @@ func (s *testSuite) TestSelectResultRuntimeStats(c *C) {
 		rpcStat:          tikv.NewRegionRequestRuntimeStats(),
 	}
 	s2 := *s1
-	stmtStats := execdetails.NewRuntimeStatsColl()
+	stmtStats := execdetails.NewRuntimeStatsColl(nil)
 	stmtStats.RegisterStats(1, basic)
 	stmtStats.RegisterStats(1, s1)
 	stmtStats.RegisterStats(1, &s2)
diff --git a/distsql/select_result_test.go b/distsql/select_result_test.go
index 2046f53c51695..cecf231d8f2a1 100644
--- a/distsql/select_result_test.go
+++ b/distsql/select_result_test.go
@@ -33,7 +33,7 @@ func (s *testSuite) TestUpdateCopRuntimeStats(c *C) {
 	sr.rootPlanID = 1234
 	sr.updateCopRuntimeStats(context.Background(), &copr.CopRuntimeStats{ExecDetails: execdetails.ExecDetails{CalleeAddress: "a"}}, 0)
 
-	ctx.GetSessionVars().StmtCtx.RuntimeStatsColl = execdetails.NewRuntimeStatsColl()
+	ctx.GetSessionVars().StmtCtx.RuntimeStatsColl = execdetails.NewRuntimeStatsColl(nil)
 	t := uint64(1)
 	sr.selectResp = &tipb.SelectResponse{
 		ExecutionSummaries: []*tipb.ExecutorExecutionSummary{
diff --git a/errors.toml b/errors.toml
index c7a7f2f4bbc19..9bd98a0a76001 100644
--- a/errors.toml
+++ b/errors.toml
@@ -536,6 +536,11 @@ error = '''
 '%-.192s.%-.192s' is not %s
 '''
 
+["executor:1356"]
+error = '''
+View '%-.192s.%-.192s' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
+'''
+
 ["executor:1390"]
 error = '''
 Prepared statement contains too many placeholders
@@ -976,6 +981,11 @@ error = '''
 EXPLAIN/SHOW can not be issued; lacking privileges for underlying table
 '''
 
+["planner:1352"]
+error = '''
+View's SELECT refers to a temporary table '%-.192s'
+'''
+
 ["planner:1356"]
 error = '''
 View '%-.192s.%-.192s' references invalid table(s) or column(s) or function(s) or definer/invoker of view lack rights to use them
diff --git a/executor/builder.go b/executor/builder.go
index b8d820d2284c2..43555a67c854f 100644
--- a/executor/builder.go
+++ b/executor/builder.go
@@ -970,7 +970,7 @@ func (b *executorBuilder) buildExplain(v *plannercore.Explain) Executor {
 	}
 	if v.Analyze {
 		if b.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl == nil {
-			b.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl = execdetails.NewRuntimeStatsColl()
+			b.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl = execdetails.NewRuntimeStatsColl(nil)
 		}
 		explainExec.analyzeExec = b.build(v.TargetPlan)
 	}
diff --git a/executor/ddl.go b/executor/ddl.go
index ab41a16cc8416..68505134cbace 100644
--- a/executor/ddl.go
+++ b/executor/ddl.go
@@ -280,8 +280,16 @@ func (e *DDLExec) createSessionTemporaryTable(s *ast.CreateTableStmt) error {
 }
 
 func (e *DDLExec) executeCreateView(s *ast.CreateViewStmt) error {
-	err := domain.GetDomain(e.ctx).DDL().CreateView(e.ctx, s)
-	return err
+	ret := &core.PreprocessorReturn{}
+	err := core.Preprocess(e.ctx, s.Select, core.WithPreprocessorReturn(ret))
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if ret.IsStaleness {
+		return ErrViewInvalid.GenWithStackByArgs(s.ViewName.Schema.L, s.ViewName.Name.L)
+	}
+
+	return domain.GetDomain(e.ctx).DDL().CreateView(e.ctx, s)
 }
 
 func (e *DDLExec) executeCreateIndex(s *ast.CreateIndexStmt) error {
diff --git a/executor/ddl_test.go b/executor/ddl_test.go
index fb7cecb63e56e..b763a84673375 100644
--- a/executor/ddl_test.go
+++ b/executor/ddl_test.go
@@ -31,6 +31,7 @@ import (
 	ddlutil "github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/errno"
+	"github.com/pingcap/tidb/executor"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
@@ -290,6 +291,10 @@ func (s *testSuite6) TestCreateView(c *C) {
 	c.Assert(terror.ErrorEqual(err, plannercore.ErrNoSuchTable), IsTrue)
 	tk.MustExec("drop table test_v_nested")
 	tk.MustExec("drop view v_nested, v_nested2")
+
+	// Refer https://github.com/pingcap/tidb/issues/25876
+	err = tk.ExecToErr("create view v_stale as select * from source_table as of timestamp current_timestamp(3)")
+	c.Assert(terror.ErrorEqual(err, executor.ErrViewInvalid), IsTrue, Commentf("err %s", err))
 }
 
 func (s *testSuite6) TestViewRecursion(c *C) {
diff --git a/executor/errors.go b/executor/errors.go
index a4043a7d41790..ca19ecba51373 100644
--- a/executor/errors.go
+++ b/executor/errors.go
@@ -51,6 +51,7 @@ var (
 	ErrDynamicPrivilegeNotRegistered = dbterror.ClassExecutor.NewStd(mysql.ErrDynamicPrivilegeNotRegistered)
 	ErrIllegalPrivilegeLevel         = dbterror.ClassExecutor.NewStd(mysql.ErrIllegalPrivilegeLevel)
 	ErrInvalidSplitRegionRanges      = dbterror.ClassExecutor.NewStd(mysql.ErrInvalidSplitRegionRanges)
+	ErrViewInvalid                   = dbterror.ClassExecutor.NewStd(mysql.ErrViewInvalid)
 
 	ErrBRIEBackupFailed              = dbterror.ClassExecutor.NewStd(mysql.ErrBRIEBackupFailed)
 	ErrBRIERestoreFailed             = dbterror.ClassExecutor.NewStd(mysql.ErrBRIERestoreFailed)
diff --git a/executor/executor.go b/executor/executor.go
index 7569830f3f870..297b6a8e66dc2 100644
--- a/executor/executor.go
+++ b/executor/executor.go
@@ -1808,7 +1808,13 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) {
 		sc.PrevAffectedRows = -1
 	}
 	if globalConfig.EnableCollectExecutionInfo {
-		sc.RuntimeStatsColl = execdetails.NewRuntimeStatsColl()
+		// In ExplainFor case, RuntimeStatsColl should not be reset for reuse,
+		// because ExplainFor need to display the last statement information.
+		reuseObj := vars.StmtCtx.RuntimeStatsColl
+		if _, ok := s.(*ast.ExplainForStmt); ok {
+			reuseObj = nil
+		}
+		sc.RuntimeStatsColl = execdetails.NewRuntimeStatsColl(reuseObj)
 	}
 
 	sc.TblInfo2UnionScan = make(map[*model.TableInfo]bool)
diff --git a/executor/insert.go b/executor/insert.go
index c6195ccef34c9..83a367866b747 100644
--- a/executor/insert.go
+++ b/executor/insert.go
@@ -376,12 +376,13 @@ func (e *InsertExec) doDupRowUpdate(ctx context.Context, handle kv.Handle, oldRo
 	// Update old row when the key is duplicated.
 	e.evalBuffer4Dup.SetDatums(e.row4Update...)
 	for _, col := range cols {
-		val, err1 := col.Expr.Eval(e.evalBuffer4Dup.ToRow())
-		if err1 != nil {
+		row := e.evalBuffer4Dup.ToRow()
+		val, err1 := col.Expr.Eval(row)
+		if err1 = e.handleErr(e.Table.WritableCols()[col.Col.Index], &val, row.Idx(), err1); err1 != nil {
 			return err1
 		}
 		e.row4Update[col.Col.Index], err1 = table.CastValue(e.ctx, val, col.Col.ToInfo(), false, false)
-		if err1 != nil {
+		if err1 = e.handleErr(e.Table.WritableCols()[col.Col.Index], &val, row.Idx(), err1); err1 != nil {
 			return err1
 		}
 		e.evalBuffer4Dup.SetDatum(col.Col.Index, e.row4Update[col.Col.Index])
diff --git a/executor/insert_test.go b/executor/insert_test.go
index 166972dd268cb..88fb65bbfcd38 100644
--- a/executor/insert_test.go
+++ b/executor/insert_test.go
@@ -1071,7 +1071,7 @@ func (s *testSuite3) TestInsertFloatOverflow(c *C) {
 	tk.MustExec("drop table if exists t,t1")
 }
 
-// There is a potential issue in MySQL: when the value of auto_increment_offset is greater
+// TestAutoIDIncrementAndOffset There is a potential issue in MySQL: when the value of auto_increment_offset is greater
 // than that of auto_increment_increment, the value of auto_increment_offset is ignored
 // (https://dev.mysql.com/doc/refman/8.0/en/replication-options-master.html#sysvar_auto_increment_increment),
 // This issue is a flaw of the implementation of MySQL and it doesn't exist in TiDB.
@@ -1537,6 +1537,18 @@ func (s *testSerialSuite) TestIssue20768(c *C) {
 	tk.MustQuery("select /*+ merge_join(t1) */ * from t1 join t2 on t1.a = t2.a").Check(testkit.Rows("0 0"))
 }
 
+func (s *testSerialSuite) TestIssue24044(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1")
+	tk.MustExec("CREATE TABLE `t1` (`id` int NOT NULL, `name` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL, PRIMARY KEY (`id`) USING BTREE) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin ROW_FORMAT=COMPACT;")
+	tk.MustExec("insert into t1 values(1,\"a\")")
+	tk.MustExec("insert into t1 values(2,\"b\")")
+
+	err := tk.ExecToErr("insert into t1 select * from t1 on duplicate key update id = \"\"")
+	c.Assert(err.Error(), Equals, "[table:1366]Incorrect int value: '' for column 'id' at row 1")
+}
+
 func (s *testSuite9) TestIssue10402(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")
@@ -1571,7 +1583,7 @@ func combination(items []string) func() []string {
 	}
 }
 
-// See https://github.com/pingcap/tidb/issues/24582
+// TestDuplicatedEntryErr https://github.com/pingcap/tidb/issues/24582
 func (s *testSuite10) TestDuplicatedEntryErr(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")
diff --git a/executor/stale_txn_test.go b/executor/stale_txn_test.go
index ab5e61f4a04be..fa395246967bd 100644
--- a/executor/stale_txn_test.go
+++ b/executor/stale_txn_test.go
@@ -118,7 +118,7 @@ func (s *testStaleTxnSerialSuite) TestSelectAsOf(c *C) {
 		sql              string
 		expectPhysicalTS int64
 		preSec           int64
-		// IsStaleness is auto cleanup in select stmt.
+		// errorStr is auto cleanup in select stmt.
 		errorStr string
 	}{
 		{
@@ -1020,7 +1020,7 @@ func (s *testStaleTxnSuite) TestStmtCtxStaleFlag(c *C) {
 		},
 		// assert select statement
 		{
-			sql:          fmt.Sprintf("select * from t"),
+			sql:          "select * from t",
 			hasStaleFlag: false,
 		},
 		// assert select statement in stale transaction
@@ -1029,7 +1029,7 @@ func (s *testStaleTxnSuite) TestStmtCtxStaleFlag(c *C) {
 			hasStaleFlag: false,
 		},
 		{
-			sql:          fmt.Sprintf("select * from t"),
+			sql:          "select * from t",
 			hasStaleFlag: true,
 		},
 		{
@@ -1042,12 +1042,12 @@ func (s *testStaleTxnSuite) TestStmtCtxStaleFlag(c *C) {
 			hasStaleFlag: false,
 		},
 		{
-			sql:          fmt.Sprintf("select * from t"),
+			sql:          "select * from t",
 			hasStaleFlag: true,
 		},
 		// assert select statement after consumed set transaction
 		{
-			sql:          fmt.Sprintf("select * from t"),
+			sql:          "select * from t",
 			hasStaleFlag: false,
 		},
 		// assert prepare statement with select as of statement
diff --git a/meta/autoid/autoid.go b/meta/autoid/autoid.go
index c4447c54f8844..cf61beaff9781 100644
--- a/meta/autoid/autoid.go
+++ b/meta/autoid/autoid.go
@@ -135,6 +135,9 @@ type Allocator interface {
 	// If allocIDs is false, it will not allocate IDs.
 	Rebase(tableID, newBase int64, allocIDs bool) error
 
+	// ForceRebase set the next global auto ID to newBase.
+	ForceRebase(tableID, newBase int64) error
+
 	// RebaseSeq rebases the sequence value in number axis with tableID and the new base value.
 	RebaseSeq(table, newBase int64) (int64, bool, error)
 
@@ -370,16 +373,49 @@ func (alloc *allocator) Rebase(tableID, requiredBase int64, allocIDs bool) error
 	if tableID == 0 {
 		return errInvalidTableID.GenWithStack("Invalid tableID")
 	}
-
 	alloc.mu.Lock()
 	defer alloc.mu.Unlock()
-
 	if alloc.isUnsigned {
 		return alloc.rebase4Unsigned(tableID, uint64(requiredBase), allocIDs)
 	}
 	return alloc.rebase4Signed(tableID, requiredBase, allocIDs)
 }
 
+// ForceRebase implements autoid.Allocator ForceRebase interface.
+func (alloc *allocator) ForceRebase(tableID, requiredBase int64) error {
+	if tableID <= 0 {
+		return errInvalidTableID.GenWithStack("Invalid tableID")
+	}
+	if requiredBase == -1 {
+		return ErrAutoincReadFailed.GenWithStack("Cannot force rebase the next global ID to '0'")
+	}
+	alloc.mu.Lock()
+	defer alloc.mu.Unlock()
+	startTime := time.Now()
+	err := kv.RunInNewTxn(context.Background(), alloc.store, true, func(ctx context.Context, txn kv.Transaction) error {
+		m := meta.NewMeta(txn)
+		currentEnd, err1 := GetAutoID(m, alloc.dbID, tableID, alloc.allocType)
+		if err1 != nil {
+			return err1
+		}
+		var step int64
+		if !alloc.isUnsigned {
+			step = requiredBase - currentEnd
+		} else {
+			uRequiredBase, uCurrentEnd := uint64(requiredBase), uint64(currentEnd)
+			step = int64(uRequiredBase - uCurrentEnd)
+		}
+		_, err1 = GenerateAutoID(m, alloc.dbID, tableID, step, alloc.allocType)
+		return err1
+	})
+	metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDRebase, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
+	if err != nil {
+		return err
+	}
+	alloc.base, alloc.end = requiredBase, requiredBase
+	return nil
+}
+
 // Rebase implements autoid.Allocator RebaseSeq interface.
 // The return value is quite same as expression function, bool means whether it should be NULL,
 // here it will be used in setval expression function (true meaning the set value has been satisfied, return NULL).
diff --git a/meta/autoid/memid.go b/meta/autoid/memid.go
index d9a6be9f6058a..39db68e3bc33d 100644
--- a/meta/autoid/memid.go
+++ b/meta/autoid/memid.go
@@ -98,6 +98,12 @@ func (alloc *inMemoryAllocator) Rebase(tableID, requiredBase int64, allocIDs boo
 	return nil
 }
 
+// ForceRebase implements autoid.Allocator ForceRebase interface.
+func (alloc *inMemoryAllocator) ForceRebase(tableID, requiredBase int64) error {
+	alloc.base = requiredBase
+	return nil
+}
+
 func (alloc *inMemoryAllocator) alloc4Signed(n uint64, increment, offset int64) (int64, int64, error) {
 	// Check offset rebase if necessary.
 	if offset-1 > alloc.base {
diff --git a/planner/core/errors.go b/planner/core/errors.go
index e501af7676c5f..de8ff8cb09fcd 100644
--- a/planner/core/errors.go
+++ b/planner/core/errors.go
@@ -101,5 +101,6 @@ var (
 	ErrOptOnTemporaryTable       = dbterror.ClassOptimizer.NewStd(mysql.ErrOptOnTemporaryTable)
 	ErrDropTableOnTemporaryTable = dbterror.ClassOptimizer.NewStd(mysql.ErrDropTableOnTemporaryTable)
 	// ErrPartitionNoTemporary returns when partition at temporary mode
-	ErrPartitionNoTemporary = dbterror.ClassOptimizer.NewStd(mysql.ErrPartitionNoTemporary)
+	ErrPartitionNoTemporary     = dbterror.ClassOptimizer.NewStd(mysql.ErrPartitionNoTemporary)
+	ErrViewSelectTemporaryTable = dbterror.ClassOptimizer.NewStd(mysql.ErrViewSelectTmptable)
 )
diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index 09703b286de85..09b34216c8cbc 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -3819,6 +3819,10 @@ func (b *PlanBuilder) buildDataSource(ctx context.Context, tn *ast.TableName, as
 	}
 
 	tableInfo := tbl.Meta()
+	if b.isCreateView && tableInfo.TempTableType == model.TempTableLocal {
+		return nil, ErrViewSelectTemporaryTable.GenWithStackByArgs(tn.Name)
+	}
+
 	var authErr error
 	if sessionVars.User != nil {
 		authErr = ErrTableaccessDenied.FastGenByArgs("SELECT", sessionVars.User.AuthUsername, sessionVars.User.AuthHostname, tableInfo.Name.L)
diff --git a/planner/core/planbuilder.go b/planner/core/planbuilder.go
index b1de3857541a0..2dd9f0df6e785 100644
--- a/planner/core/planbuilder.go
+++ b/planner/core/planbuilder.go
@@ -487,6 +487,8 @@ type PlanBuilder struct {
 	buildingViewStack set.StringSet
 	// renamingViewName is the name of the view which is being renamed.
 	renamingViewName string
+	// isCreateView indicates whether the query is create view.
+	isCreateView bool
 
 	// evalDefaultExpr needs this information to find the corresponding column.
 	// It stores the OutputNames before buildProjection.
@@ -3608,11 +3610,13 @@ func (b *PlanBuilder) buildDDL(ctx context.Context, node ast.DDLNode) (Plan, err
 				v.ReferTable.Name.L, "", authErr)
 		}
 	case *ast.CreateViewStmt:
+		b.isCreateView = true
 		b.capFlag |= canExpandAST | renameView
 		b.renamingViewName = v.ViewName.Schema.L + "." + v.ViewName.Name.L
 		defer func() {
 			b.capFlag &= ^canExpandAST
 			b.capFlag &= ^renameView
+			b.isCreateView = false
 		}()
 		plan, err := b.Build(ctx, v.Select)
 		if err != nil {
diff --git a/sessionctx/binloginfo/binloginfo_test.go b/sessionctx/binloginfo/binloginfo_test.go
index 6a586615bd8a0..4e52d2b100da6 100644
--- a/sessionctx/binloginfo/binloginfo_test.go
+++ b/sessionctx/binloginfo/binloginfo_test.go
@@ -666,6 +666,14 @@ func (s *testBinlogSuite) TestAddSpecialComment(c *C) {
 			"create table t1 (id int, a varchar(255) key clustered);",
 			"create table t1 (id int, a varchar(255) key /*T![clustered_index] clustered */ );",
 		},
+		{
+			"alter table t force auto_increment = 12;",
+			"alter table t /*T![force_inc] force */  auto_increment = 12;",
+		},
+		{
+			"alter table t force, auto_increment = 12;",
+			"alter table t force, auto_increment = 12;",
+		},
 	}
 	for _, ca := range testCase {
 		re := binloginfo.AddSpecialComment(ca.input)
diff --git a/store/copr/coprocessor_test.go b/store/copr/coprocessor_test.go
index afd99cdbb4c50..41e96c7772503 100644
--- a/store/copr/coprocessor_test.go
+++ b/store/copr/coprocessor_test.go
@@ -156,6 +156,7 @@ func (s *testCoprocessorSuite) TestSplitRegionRanges(c *C) {
 	// nil --- 'g' --- 'n' --- 't' --- nil
 	// <-  0  -> <- 1 -> <- 2 -> <- 3 ->
 	_, cluster, pdClient, err := testutils.NewMockTiKV("", nil)
+	c.Assert(err, IsNil)
 	testutils.BootstrapWithMultiRegions(cluster, []byte("g"), []byte("n"), []byte("t"))
 	pdCli := &tikv.CodecPDClient{Client: pdClient}
 	cache := NewRegionCache(tikv.NewRegionCache(pdCli))
diff --git a/types/parser_driver/special_cmt_ctrl.go b/types/parser_driver/special_cmt_ctrl.go
index 6fa7ebf67d6cd..31505ebc40f71 100644
--- a/types/parser_driver/special_cmt_ctrl.go
+++ b/types/parser_driver/special_cmt_ctrl.go
@@ -35,6 +35,7 @@ func init() {
 	parser.SpecialCommentsController.Register(string(FeatureIDAutoIDCache))
 	parser.SpecialCommentsController.Register(string(FeatureIDAutoRandomBase))
 	parser.SpecialCommentsController.Register(string(FeatureClusteredIndex))
+	parser.SpecialCommentsController.Register(string(FeatureForceAutoInc))
 }
 
 // SpecialCommentVersionPrefix is the prefix of TiDB executable comments.
@@ -61,6 +62,8 @@ const (
 	FeatureIDAutoRandomBase featureID = "auto_rand_base"
 	// FeatureClusteredIndex is the `clustered_index` feature.
 	FeatureClusteredIndex featureID = "clustered_index"
+	// FeatureForceAutoInc is the `force auto_increment` feature.
+	FeatureForceAutoInc featureID = "force_inc"
 )
 
 // FeatureIDPatterns is used to record special comments patterns.
@@ -69,4 +72,5 @@ var FeatureIDPatterns = map[featureID]*regexp.Regexp{
 	FeatureIDAutoIDCache:    regexp.MustCompile(`(?P<REPLACE>(?i)AUTO_ID_CACHE\s*=?\s*\d+\s*)`),
 	FeatureIDAutoRandomBase: regexp.MustCompile(`(?P<REPLACE>(?i)AUTO_RANDOM_BASE\s*=?\s*\d+\s*)`),
 	FeatureClusteredIndex:   regexp.MustCompile(`(?i)(PRIMARY)?\s+KEY(\s*\(.*\))?\s+(?P<REPLACE>(NON)?CLUSTERED\b)`),
+	FeatureForceAutoInc:     regexp.MustCompile(`(?P<REPLACE>(?i)FORCE)\b\s*AUTO_INCREMENT\s*`),
 }
diff --git a/util/execdetails/execdetails.go b/util/execdetails/execdetails.go
index 68d1d9656caef..42f5c270f5030 100644
--- a/util/execdetails/execdetails.go
+++ b/util/execdetails/execdetails.go
@@ -555,9 +555,23 @@ type RuntimeStatsColl struct {
 }
 
 // NewRuntimeStatsColl creates new executor collector.
-func NewRuntimeStatsColl() *RuntimeStatsColl {
-	return &RuntimeStatsColl{rootStats: make(map[int]*RootRuntimeStats),
-		copStats: make(map[int]*CopRuntimeStats)}
+// Reuse the object to reduce allocation when *RuntimeStatsColl is not nil.
+func NewRuntimeStatsColl(reuse *RuntimeStatsColl) *RuntimeStatsColl {
+	if reuse != nil {
+		// Reuse map is cheaper than create a new map object.
+		// Go compiler optimize this cleanup code pattern to a clearmap() function.
+		for k := range reuse.rootStats {
+			delete(reuse.rootStats, k)
+		}
+		for k := range reuse.copStats {
+			delete(reuse.copStats, k)
+		}
+		return reuse
+	}
+	return &RuntimeStatsColl{
+		rootStats: make(map[int]*RootRuntimeStats),
+		copStats:  make(map[int]*CopRuntimeStats),
+	}
 }
 
 // RegisterStats register execStat for a executor.
diff --git a/util/execdetails/execdetails_test.go b/util/execdetails/execdetails_test.go
index 6bfcfbeca703e..9ad39abe390d1 100644
--- a/util/execdetails/execdetails_test.go
+++ b/util/execdetails/execdetails_test.go
@@ -93,7 +93,7 @@ func mockExecutorExecutionSummaryForTiFlash(TimeProcessedNs, NumProducedRows, Nu
 }
 
 func TestCopRuntimeStats(t *testing.T) {
-	stats := NewRuntimeStatsColl()
+	stats := NewRuntimeStatsColl(nil)
 	tableScanID := 1
 	aggID := 2
 	tableReaderID := 3
@@ -156,7 +156,7 @@ func TestCopRuntimeStats(t *testing.T) {
 }
 
 func TestCopRuntimeStatsForTiFlash(t *testing.T) {
-	stats := NewRuntimeStatsColl()
+	stats := NewRuntimeStatsColl(nil)
 	tableScanID := 1
 	aggID := 2
 	tableReaderID := 3
@@ -261,7 +261,7 @@ func TestRootRuntimeStats(t *testing.T) {
 	basic1.Record(time.Second, 20)
 	basic2.Record(time.Second*2, 30)
 	pid := 1
-	stmtStats := NewRuntimeStatsColl()
+	stmtStats := NewRuntimeStatsColl(nil)
 	stmtStats.RegisterStats(pid, basic1)
 	stmtStats.RegisterStats(pid, basic2)
 	concurrency := &RuntimeStatsWithConcurrencyInfo{}
diff --git a/util/memory/tracker.go b/util/memory/tracker.go
index 3b9276c3a0461..eb9cef5d8e13a 100644
--- a/util/memory/tracker.go
+++ b/util/memory/tracker.go
@@ -53,7 +53,7 @@ import (
 type Tracker struct {
 	mu struct {
 		sync.Mutex
-		// The children memory trackers. If the Tracker is the Global Tracker, like executor.GlobalDiskUsageTracker,
+		// children memory trackers. If the Tracker is the Global Tracker, like executor.GlobalDiskUsageTracker,
 		// we wouldn't maintain its children in order to avoiding mutex contention.
 		children map[int][]*Tracker
 	}
@@ -95,7 +95,6 @@ func InitTracker(t *Tracker, label int, bytesLimit int64, action ActionOnExceed)
 	t.bytesSoftLimit = int64(float64(bytesLimit) * softScale)
 	t.maxConsumed = 0
 	t.isGlobal = false
-	return
 }
 
 // NewTracker creates a memory tracker.