From 79c4c86fcb27f416c95b87661416f34e1b70b2dc Mon Sep 17 00:00:00 2001 From: Andrew Werner Date: Tue, 21 Dec 2021 23:59:25 -0500 Subject: [PATCH] sql: fix InternalExecutor bug Any time we use WithSyntheticDescriptors, it has to be on an unshared internal executor. The move in #71246 to not have an internal executor hanging around for the current session hurts here. Fixes #73788 Release note: None --- pkg/sql/alter_table.go | 15 +++++++++++---- pkg/sql/backfill.go | 31 ++++++++++++++++--------------- pkg/sql/check.go | 4 ++-- pkg/sql/index_backfiller.go | 2 +- pkg/sql/planner.go | 2 -- pkg/sql/schema_changer.go | 14 ++++---------- 6 files changed, 34 insertions(+), 34 deletions(-) diff --git a/pkg/sql/alter_table.go b/pkg/sql/alter_table.go index 77fe68813f3a..db421e7759e5 100644 --- a/pkg/sql/alter_table.go +++ b/pkg/sql/alter_table.go @@ -809,7 +809,8 @@ func (n *alterTableNode) startExec(params runParams) error { "constraint %q in the middle of being added, try again later", t.Constraint) } if err := validateCheckInTxn( - params.ctx, ¶ms.p.semaCtx, params.ExecCfg().InternalExecutor, params.SessionData(), n.tableDesc, params.EvalContext().Txn, ck.Expr, + params.ctx, ¶ms.p.semaCtx, params.ExecCfg().InternalExecutorFactory, + params.SessionData(), n.tableDesc, params.EvalContext().Txn, ck.Expr, ); err != nil { return err } @@ -831,8 +832,12 @@ func (n *alterTableNode) startExec(params runParams) error { "constraint %q in the middle of being added, try again later", t.Constraint) } if err := validateFkInTxn( - params.ctx, params.p.LeaseMgr(), params.ExecCfg().InternalExecutor, - n.tableDesc, params.EvalContext().Txn, name, params.EvalContext().Codec, + params.ctx, params.p.LeaseMgr(), + params.ExecCfg().InternalExecutorFactory, + params.p.SessionData(), + n.tableDesc, + params.EvalContext().Txn, + name, params.EvalContext().Codec, ); err != nil { return err } @@ -855,7 +860,9 @@ func (n *alterTableNode) startExec(params runParams) error { "constraint %q in the middle of being added, try again later", t.Constraint) } if err := validateUniqueWithoutIndexConstraintInTxn( - params.ctx, params.ExecCfg().InternalExecutor, n.tableDesc, params.EvalContext().Txn, name, + params.ctx, params.ExecCfg().InternalExecutorFactory( + params.ctx, params.SessionData(), + ), n.tableDesc, params.EvalContext().Txn, name, ); err != nil { return err } diff --git a/pkg/sql/backfill.go b/pkg/sql/backfill.go index 636d126bf350..9464327de54f 100644 --- a/pkg/sql/backfill.go +++ b/pkg/sql/backfill.go @@ -152,7 +152,7 @@ func (sc *SchemaChanger) makeFixedTimestampRunner(readAsOf hlc.Timestamp) histor ctx context.Context, txn *kv.Txn, descriptors *descs.Collection, ) error { // We need to re-create the evalCtx since the txn may retry. - evalCtx := createSchemaChangeEvalCtx(ctx, sc.execCfg, readAsOf, sc.ieFactory, descriptors) + evalCtx := createSchemaChangeEvalCtx(ctx, sc.execCfg, readAsOf, descriptors) return retryable(ctx, txn, &evalCtx) }) } @@ -168,9 +168,7 @@ func (sc *SchemaChanger) makeFixedTimestampInternalExecRunner( ctx context.Context, txn *kv.Txn, descriptors *descs.Collection, ) error { // We need to re-create the evalCtx since the txn may retry. - ie := createSchemaChangeEvalCtx( - ctx, sc.execCfg, readAsOf, sc.ieFactory, descriptors, - ).SchemaChangeInternalExecutor + ie := sc.ieFactory(ctx, NewFakeSessionData(sc.execCfg.SV())) return retryable(ctx, txn, ie) }) } @@ -713,21 +711,21 @@ func (sc *SchemaChanger) validateConstraints( defer func() { collection.ReleaseAll(ctx) }() if c.IsCheck() { if err := validateCheckInTxn( - ctx, &semaCtx, evalCtx.SchemaChangeInternalExecutor, evalCtx.SessionData(), desc, txn, c.Check().Expr, + ctx, &semaCtx, sc.ieFactory, evalCtx.SessionData(), desc, txn, c.Check().Expr, ); err != nil { return err } } else if c.IsForeignKey() { - if err := validateFkInTxn(ctx, sc.leaseMgr, evalCtx.SchemaChangeInternalExecutor, desc, txn, c.GetName(), evalCtx.Codec); err != nil { + if err := validateFkInTxn(ctx, sc.leaseMgr, sc.ieFactory, evalCtx.SessionData(), desc, txn, c.GetName(), evalCtx.Codec); err != nil { return err } } else if c.IsUniqueWithoutIndex() { - if err := validateUniqueWithoutIndexConstraintInTxn(ctx, evalCtx.SchemaChangeInternalExecutor, desc, txn, c.GetName()); err != nil { + if err := validateUniqueWithoutIndexConstraintInTxn(ctx, sc.ieFactory(ctx, evalCtx.SessionData()), desc, txn, c.GetName()); err != nil { return err } } else if c.IsNotNull() { if err := validateCheckInTxn( - ctx, &semaCtx, evalCtx.SchemaChangeInternalExecutor, evalCtx.SessionData(), desc, txn, c.Check().Expr, + ctx, &semaCtx, sc.ieFactory, evalCtx.SessionData(), desc, txn, c.Check().Expr, ); err != nil { // TODO (lucy): This should distinguish between constraint // validation errors and other types of unexpected errors, and @@ -991,7 +989,7 @@ func (sc *SchemaChanger) distIndexBackfill( if err != nil { return err } - evalCtx = createSchemaChangeEvalCtx(ctx, sc.execCfg, txn.ReadTimestamp(), sc.ieFactory, descriptors) + evalCtx = createSchemaChangeEvalCtx(ctx, sc.execCfg, txn.ReadTimestamp(), descriptors) planCtx = sc.distSQLPlanner.NewPlanningCtx(ctx, &evalCtx, nil /* planner */, txn, true /* distribute */) indexBatchSize := indexBackfillBatchSize.Get(&sc.execCfg.Settings.SV) @@ -1265,7 +1263,7 @@ func (sc *SchemaChanger) distColumnBackfill( return nil } cbw := MetadataCallbackWriter{rowResultWriter: &errOnlyResultWriter{}, fn: metaFn} - evalCtx := createSchemaChangeEvalCtx(ctx, sc.execCfg, txn.ReadTimestamp(), sc.ieFactory, descriptors) + evalCtx := createSchemaChangeEvalCtx(ctx, sc.execCfg, txn.ReadTimestamp(), descriptors) recv := MakeDistSQLReceiver( ctx, &cbw, @@ -2090,7 +2088,8 @@ func runSchemaChangesInTxn( check := &c.ConstraintToUpdateDesc().Check if check.Validity == descpb.ConstraintValidity_Validating { if err := validateCheckInTxn( - ctx, &planner.semaCtx, planner.ExecCfg().InternalExecutor, planner.SessionData(), tableDesc, planner.txn, check.Expr, + ctx, &planner.semaCtx, planner.ExecCfg().InternalExecutorFactory, + planner.SessionData(), tableDesc, planner.txn, check.Expr, ); err != nil { return err } @@ -2187,7 +2186,7 @@ func runSchemaChangesInTxn( func validateCheckInTxn( ctx context.Context, semaCtx *tree.SemaContext, - ie *InternalExecutor, + ief sqlutil.SessionBoundInternalExecutorFactory, sessionData *sessiondata.SessionData, tableDesc *tabledesc.Mutable, txn *kv.Txn, @@ -2197,6 +2196,7 @@ func validateCheckInTxn( if tableDesc.Version > tableDesc.ClusterVersion.Version { syntheticDescs = append(syntheticDescs, tableDesc) } + ie := ief(ctx, sessionData) return ie.WithSyntheticDescriptors(syntheticDescs, func() error { return validateCheckExpr(ctx, semaCtx, sessionData, checkExpr, tableDesc, ie, txn) }) @@ -2217,7 +2217,8 @@ func validateCheckInTxn( func validateFkInTxn( ctx context.Context, leaseMgr *lease.Manager, - ie *InternalExecutor, + ief sqlutil.SessionBoundInternalExecutorFactory, + sd *sessiondata.SessionData, tableDesc *tabledesc.Mutable, txn *kv.Txn, fkName string, @@ -2239,7 +2240,7 @@ func validateFkInTxn( if fk == nil { return errors.AssertionFailedf("foreign key %s does not exist", fkName) } - + ie := ief(ctx, sd) return ie.WithSyntheticDescriptors(syntheticDescs, func() error { return validateForeignKey(ctx, tableDesc, fk, ie, txn, codec) }) @@ -2259,7 +2260,7 @@ func validateFkInTxn( // reuse an existing kv.Txn safely. func validateUniqueWithoutIndexConstraintInTxn( ctx context.Context, - ie *InternalExecutor, + ie sqlutil.InternalExecutor, tableDesc *tabledesc.Mutable, txn *kv.Txn, constraintName string, diff --git a/pkg/sql/check.go b/pkg/sql/check.go index 109afbb655f6..5a884eabcb37 100644 --- a/pkg/sql/check.go +++ b/pkg/sql/check.go @@ -44,7 +44,7 @@ func validateCheckExpr( sessionData *sessiondata.SessionData, exprStr string, tableDesc *tabledesc.Mutable, - ie *InternalExecutor, + ie sqlutil.InternalExecutor, txn *kv.Txn, ) error { expr, err := schemaexpr.FormatExprForDisplay(ctx, tableDesc, exprStr, semaCtx, sessionData, tree.FmtParsable) @@ -237,7 +237,7 @@ func validateForeignKey( ctx context.Context, srcTable *tabledesc.Mutable, fk *descpb.ForeignKeyConstraint, - ie *InternalExecutor, + ie sqlutil.InternalExecutor, txn *kv.Txn, codec keys.SQLCodec, ) error { diff --git a/pkg/sql/index_backfiller.go b/pkg/sql/index_backfiller.go index c11b537dc566..68818b594452 100644 --- a/pkg/sql/index_backfiller.go +++ b/pkg/sql/index_backfiller.go @@ -133,7 +133,7 @@ func (ib *IndexBackfillPlanner) plan( if err := DescsTxn(ctx, ib.execCfg, func( ctx context.Context, txn *kv.Txn, descriptors *descs.Collection, ) error { - evalCtx = createSchemaChangeEvalCtx(ctx, ib.execCfg, nowTimestamp, ib.ieFactory, descriptors) + evalCtx = createSchemaChangeEvalCtx(ctx, ib.execCfg, nowTimestamp, descriptors) planCtx = ib.execCfg.DistSQLPlanner.NewPlanningCtx(ctx, &evalCtx, nil /* planner */, txn, true /* distribute */) // TODO(ajwerner): Adopt util.ConstantWithMetamorphicTestRange for the diff --git a/pkg/sql/planner.go b/pkg/sql/planner.go index df6b30d8fd77..465562008e77 100644 --- a/pkg/sql/planner.go +++ b/pkg/sql/planner.go @@ -101,8 +101,6 @@ type extendedEvalContext struct { indexUsageStats *idxusage.LocalIndexUsageStats SchemaChangerState *SchemaChangerState - - SchemaChangeInternalExecutor *InternalExecutor } // copyFromExecCfg copies relevant fields from an ExecutorConfig. diff --git a/pkg/sql/schema_changer.go b/pkg/sql/schema_changer.go index 16de3ac05b1c..728186e38ded 100644 --- a/pkg/sql/schema_changer.go +++ b/pkg/sql/schema_changer.go @@ -1941,24 +1941,18 @@ func (sc *SchemaChanger) txn( // used in the surrounding SQL session, so session tracing is unable // to capture schema change activity. func createSchemaChangeEvalCtx( - ctx context.Context, - execCfg *ExecutorConfig, - ts hlc.Timestamp, - ieFactory sqlutil.SessionBoundInternalExecutorFactory, - descriptors *descs.Collection, + ctx context.Context, execCfg *ExecutorConfig, ts hlc.Timestamp, descriptors *descs.Collection, ) extendedEvalContext { sd := NewFakeSessionData(execCfg.SV()) - ie := ieFactory(ctx, sd) evalCtx := extendedEvalContext{ // Make a session tracing object on-the-fly. This is OK // because it sets "enabled: false" and thus none of the // other fields are used. - Tracing: &SessionTracing{}, - ExecCfg: execCfg, - Descs: descriptors, - SchemaChangeInternalExecutor: ie.(*InternalExecutor), + Tracing: &SessionTracing{}, + ExecCfg: execCfg, + Descs: descriptors, EvalContext: tree.EvalContext{ SessionDataStack: sessiondata.NewStack(sd), // TODO(andrei): This is wrong (just like on the main code path on