Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ddl: get latest old table ID before replace view #53720

Merged
merged 7 commits into from
Jun 3, 2024
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pkg/ddl/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -164,7 +164,6 @@ go_library(
"//pkg/util/sqlexec",
"//pkg/util/sqlkiller",
"//pkg/util/stringutil",
"//pkg/util/syncutil",
"//pkg/util/tiflash",
"//pkg/util/timeutil",
"//pkg/util/topsql",
@@ -340,6 +339,7 @@ go_test(
"@com_github_tikv_client_go_v2//util",
"@io_etcd_go_etcd_client_v3//:client",
"@org_golang_google_grpc//:grpc",
"@org_golang_x_sync//errgroup",
"@org_uber_go_atomic//:atomic",
"@org_uber_go_goleak//:goleak",
"@org_uber_go_zap//:zap",
6 changes: 0 additions & 6 deletions pkg/ddl/ddl.go
Original file line number Diff line number Diff line change
@@ -65,7 +65,6 @@ import (
"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
"github.com/pingcap/tidb/pkg/util/gcutil"
"github.com/pingcap/tidb/pkg/util/generic"
"github.com/pingcap/tidb/pkg/util/syncutil"
"github.com/tikv/client-go/v2/tikvrpc"
clientv3 "go.etcd.io/etcd/client/v3"
"go.etcd.io/etcd/client/v3/concurrency"
@@ -388,11 +387,6 @@ type ddlCtx struct {

// reorgCtx is used for reorganization.
reorgCtx reorgContexts
// backfillCtx is used for backfill workers.
backfillCtx struct {
syncutil.RWMutex
jobCtxMap map[int64]*JobContext
}

jobCtx struct {
sync.RWMutex
41 changes: 41 additions & 0 deletions pkg/ddl/ddl_api_test.go
Original file line number Diff line number Diff line change
@@ -31,6 +31,7 @@ import (
"github.com/pingcap/tidb/pkg/testkit"
"github.com/pingcap/tidb/pkg/util/chunk"
"github.com/stretchr/testify/require"
"golang.org/x/sync/errgroup"
)

func TestGetDDLJobs(t *testing.T) {
@@ -151,6 +152,46 @@ func enQueueDDLJobs(t *testing.T, sess sessiontypes.Session, txn kv.Transaction,
}
}

func TestCreateViewConcurrently(t *testing.T) {
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test")

tk.MustExec("create table t (a int);")
tk.MustExec("create view v as select * from t;")
var (
counterErr error
counter int
)
failpoint.EnableCall("github.com/pingcap/tidb/pkg/ddl/onDDLCreateView", func(job *model.Job) {
counter++
if counter > 1 {
counterErr = fmt.Errorf("create view job should not run concurrently")
return
}
})
failpoint.EnableCall("github.com/pingcap/tidb/pkg/ddl/afterDelivery2Worker", func(job *model.Job) {
if job.Type == model.ActionCreateView {
counter--
}
})
var eg errgroup.Group
for i := 0; i < 5; i++ {
eg.Go(func() error {
newTk := testkit.NewTestKit(t, store)
_, err := newTk.Exec("use test")
if err != nil {
return err
}
_, err = newTk.Exec("create or replace view v as select * from t;")
return err
})
}
err := eg.Wait()
require.NoError(t, err)
require.NoError(t, counterErr)
}

func TestCreateDropCreateTable(t *testing.T) {
store, dom := testkit.CreateMockStoreAndDomain(t)
tk := testkit.NewTestKit(t, store)
1 change: 1 addition & 0 deletions pkg/ddl/job_table.go
Original file line number Diff line number Diff line change
@@ -511,6 +511,7 @@ func (s *jobScheduler) delivery2Worker(wk *worker, pool *workerPool, job *model.
s.wg.Run(func() {
metrics.DDLRunningJobCount.WithLabelValues(pool.tp().String()).Inc()
defer func() {
failpoint.InjectCall("afterDelivery2Worker", job)
s.runningJobs.remove(job)
asyncNotify(s.ddlJobNotifyCh)
metrics.DDLRunningJobCount.WithLabelValues(pool.tp().String()).Dec()
4 changes: 2 additions & 2 deletions pkg/ddl/placement_policy.go
Original file line number Diff line number Diff line change
@@ -119,7 +119,7 @@ func getPlacementPolicyByName(d *ddlCtx, t *meta.Meta, policyName model.CIStr) (
}

is := d.infoCache.GetLatest()
if is.SchemaMetaVersion() == currVer {
if is != nil && is.SchemaMetaVersion() == currVer {
// Use cached policy.
policy, ok := is.PolicyByName(policyName)
if ok {
@@ -346,7 +346,7 @@ func checkPlacementPolicyNotInUse(d *ddlCtx, t *meta.Meta, policy *model.PolicyI
return err
}
is := d.infoCache.GetLatest()
if is.SchemaMetaVersion() == currVer {
if is != nil && is.SchemaMetaVersion() == currVer {
err = CheckPlacementPolicyNotInUseFromInfoSchema(is, policy)
} else {
err = CheckPlacementPolicyNotInUseFromMeta(t, policy)
2 changes: 1 addition & 1 deletion pkg/ddl/schema.go
Original file line number Diff line number Diff line change
@@ -76,7 +76,7 @@ func checkSchemaNotExists(d *ddlCtx, t *meta.Meta, schemaID int64, dbInfo *model
return err
}
is := d.infoCache.GetLatest()
if is.SchemaMetaVersion() == currVer {
if is != nil && is.SchemaMetaVersion() == currVer {
return checkSchemaNotExistsFromInfoSchema(is, schemaID, dbInfo)
}
return checkSchemaNotExistsFromStore(t, schemaID, dbInfo)
63 changes: 55 additions & 8 deletions pkg/ddl/table.go
Original file line number Diff line number Diff line change
@@ -308,14 +308,19 @@ func onCreateView(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
schemaID := job.SchemaID
tbInfo := &model.TableInfo{}
var orReplace bool
var oldTbInfoID int64
if err := job.DecodeArgs(tbInfo, &orReplace, &oldTbInfoID); err != nil {
var _placeholder int64 // oldTblInfoID
if err := job.DecodeArgs(tbInfo, &orReplace, &_placeholder); err != nil {
// Invalid arguments, cancel this job.
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
}
tbInfo.State = model.StateNone
err := checkTableNotExists(d, t, schemaID, tbInfo.Name.L)

oldTableID, err := findTableIDByName(d, t, schemaID, tbInfo.Name.L)
if infoschema.ErrTableNotExists.Equal(err) {
err = nil
}
failpoint.InjectCall("onDDLCreateView", job)
if err != nil {
if infoschema.ErrDatabaseNotExists.Equal(err) {
job.State = model.JobStateCancelled
@@ -337,13 +342,13 @@ func onCreateView(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
// none -> public
tbInfo.State = model.StatePublic
tbInfo.UpdateTS = t.StartTS
if oldTbInfoID > 0 && orReplace {
err = t.DropTableOrView(schemaID, job.SchemaName, oldTbInfoID, tbInfo.Name.L)
if oldTableID > 0 && orReplace {
err = t.DropTableOrView(schemaID, job.SchemaName, oldTableID, tbInfo.Name.L)
if err != nil {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
}
err = t.GetAutoIDAccessors(schemaID, oldTbInfoID).Del()
err = t.GetAutoIDAccessors(schemaID, oldTableID).Del()
if err != nil {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
@@ -1513,7 +1518,7 @@ func checkTableNotExists(d *ddlCtx, t *meta.Meta, schemaID int64, tableName stri
return err
}
is := d.infoCache.GetLatest()
if is.SchemaMetaVersion() == currVer {
if is != nil && is.SchemaMetaVersion() == currVer {
return checkTableNotExistsFromInfoSchema(is, schemaID, tableName)
}

@@ -1527,7 +1532,7 @@ func checkTableNotExistsByName(d *ddlCtx, t *meta.Meta, schemaID int64, schemaNa
return err
}
is := d.infoCache.GetLatest()
if is.SchemaMetaVersion() == currVer {
if is != nil && is.SchemaMetaVersion() == currVer {
return checkTableNotExistsFromInfoSchema(is, schemaID, tableName)
}
return t.CheckTableNameNotExists(t.TableNameKey(schemaName, tableName))
@@ -1601,6 +1606,48 @@ func checkTableNotExistsFromStore(t *meta.Meta, schemaID int64, tableName string
return nil
}

func findTableIDByName(d *ddlCtx, t *meta.Meta, schemaID int64, tableName string) (int64, error) {
// Try to use memory schema info to check first.
currVer, err := t.GetSchemaVersion()
if err != nil {
return 0, err
}
is := d.infoCache.GetLatest()
if is != nil && is.SchemaMetaVersion() == currVer {
return findTableIDFromInfoSchema(is, schemaID, tableName)
}

return findTableIDFromStore(t, schemaID, tableName)
}

func findTableIDFromInfoSchema(is infoschema.InfoSchema, schemaID int64, tableName string) (int64, error) {
schema, ok := is.SchemaByID(schemaID)
if !ok {
return 0, infoschema.ErrDatabaseNotExists.GenWithStackByArgs("")
}
tbl, err := is.TableByName(schema.Name, model.NewCIStr(tableName))
if err != nil {
return 0, err
}
return tbl.Meta().ID, nil
}

func findTableIDFromStore(t *meta.Meta, schemaID int64, tableName string) (int64, error) {
tbls, err := t.ListSimpleTables(schemaID)
if err != nil {
if meta.ErrDBNotExists.Equal(err) {
return 0, infoschema.ErrDatabaseNotExists.GenWithStackByArgs("")
}
return 0, errors.Trace(err)
}
for _, tbl := range tbls {
if tbl.Name.L == tableName {
return tbl.ID, nil
}
}
return 0, infoschema.ErrTableNotExists.FastGenByArgs(tableName)
}

// updateVersionAndTableInfoWithCheck checks table info validate and updates the schema version and the table information
func updateVersionAndTableInfoWithCheck(d *ddlCtx, t *meta.Meta, job *model.Job, tblInfo *model.TableInfo, shouldUpdateVer bool, multiInfos ...schemaIDAndTableInfo) (
ver int64, err error) {
3 changes: 2 additions & 1 deletion pkg/ddl/table_test.go
Original file line number Diff line number Diff line change
@@ -309,7 +309,8 @@ func TestCreateView(t *testing.T) {
}
ctx.SetValue(sessionctx.QueryString, "skip")
err = d.DoDDLJob(ctx, job)
require.Error(t, err)
// The non-existing table id in job args will not be considered anymore.
require.NoError(t, err)
}

func checkTableCacheTest(t *testing.T, store kv.Storage, dbInfo *model.DBInfo, tblInfo *model.TableInfo) {