sql: incorrect behaviour setting NOT NULL on column and dropping it

Fixes: #47719

Previously, if a column was mutated and then dropped in a single
transaction we would still try to validate the dropped column.
Additionally, in general if a drop operation occurred in a
different transaction we ran the danger of doing something
similar. From a behaviour perspective this was bad, since
it can lead to unexpected behaviour for users. To address, this
patch introduces logic to scan later mutations to check if
a drop column occurs. If one does occur then it will skip any
operations made irrelevant by the drop. Additionally, for a
correctness perspective it will wait for the drop to complete
before returning back.

Release note (bug fix): A constraint like not null or check on
a column made irrelevant by a drop in a later concurrent transaction
would lead to incorrect errors / behaviour.

pkg/jobs/registry.go

@@ -299,18 +299,14 @@ func (r *Registry) NotifyToAdoptJobs(ctx context.Context) error {
 	return nil
 }
 
-// Run starts previously unstarted jobs from a list of scheduled
-// jobs. Canceling ctx interrupts the waiting but doesn't cancel the jobs.
-func (r *Registry) Run(
+// WaitForJobs waits for a given list of jobs to reach some sort
+// of terminal state.
+func (r *Registry) WaitForJobs(
 	ctx context.Context, ex sqlutil.InternalExecutor, jobs []jobspb.JobID,
 ) error {
 	if len(jobs) == 0 {
 		return nil
 	}
-	log.Infof(ctx, "scheduled jobs %+v", jobs)
-	if err := r.NotifyToAdoptJobs(ctx); err != nil {
-		return err
-	}
 	buf := bytes.Buffer{}
 	for i, id := range jobs {
 		if i > 0 {
@@ -372,6 +368,25 @@ func (r *Registry) Run(
 	return nil
 }
 
+// Run starts previously unstarted jobs from a list of scheduled
+// jobs. Canceling ctx interrupts the waiting but doesn't cancel the jobs.
+func (r *Registry) Run(
+	ctx context.Context, ex sqlutil.InternalExecutor, jobs []jobspb.JobID,
+) error {
+	if len(jobs) == 0 {
+		return nil
+	}
+	log.Infof(ctx, "scheduled jobs %+v", jobs)
+	if err := r.NotifyToAdoptJobs(ctx); err != nil {
+		return err
+	}
+	err := r.WaitForJobs(ctx, ex, jobs)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 // NewJob creates a new Job.
 func (r *Registry) NewJob(record Record, jobID jobspb.JobID) *Job {
 	job := &Job{

pkg/sql/backfill.go

@@ -206,10 +206,16 @@ func (sc *SchemaChanger) runBackfill(ctx context.Context) error {
 	log.Infof(ctx, "running backfill for %q, v=%d", tableDesc.Name, tableDesc.Version)
 
 	needColumnBackfill := false
-	for _, m := range tableDesc.Mutations {
+	for mutationIdx, m := range tableDesc.Mutations {
 		if m.MutationID != sc.mutationID {
 			break
 		}
+		// If the current mutation is discarded, then
+		// skip over processing.
+		if discarded, _ := isCurrentMutationDiscarded(tableDesc, m, mutationIdx+1); discarded {
+			continue
+		}
+
 		switch m.Direction {
 		case descpb.DescriptorMutation_ADD:
 			switch t := m.Descriptor_.(type) {
@@ -1865,6 +1871,11 @@ func runSchemaChangesInTxn(
 	// mutations that need to be processed.
 	for i := 0; i < len(tableDesc.Mutations); i++ {
 		m := tableDesc.Mutations[i]
+		// Skip mutations that get cancelled by later operations
+		if discarded, _ := isCurrentMutationDiscarded(tableDesc, m, i+1); discarded {
+			continue
+		}
+
 		immutDesc := tabledesc.NewBuilder(tableDesc.TableDesc()).BuildImmutableTable()
 		switch m.Direction {
 		case descpb.DescriptorMutation_ADD:

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -1689,6 +1689,26 @@ SELECT count(descriptor_id)
 ----
 0
 
+# Validation for #47719 where a mutation is cancelled out by a drop.
+statement ok
+create table tColDrop (a int);
+
+statement ok
+begin;
+alter table tColDrop alter column a set not null;
+alter table tColDrop drop column a;
+commit;
+
+statement ok
+drop table tColDrop;
+
+statement ok
+begin;
+create table tColDrop (a int);
+alter table tColDrop alter column a set not null;
+alter table tColDrop drop column a;
+commit;
+
 # Validate that the schema_change_successful metric
 query T
 SELECT feature_name FROM crdb_internal.feature_usage

pkg/sql/schema_changer.go

@@ -1045,10 +1045,13 @@ func (sc *SchemaChanger) done(ctx context.Context) error {
 	// Jobs (for GC, etc.) that need to be started immediately after the table
 	// descriptor updates are published.
 	var didUpdate bool
+	var depMutationJobs []jobspb.JobID
+	var scTable *tabledesc.Mutable
 	modified, err := sc.txnWithModified(ctx, func(
 		ctx context.Context, txn *kv.Txn, descsCol *descs.Collection,
 	) error {
-		scTable, err := descsCol.GetMutableTableVersionByID(ctx, sc.descID, txn)
+		var err error
+		scTable, err = descsCol.GetMutableTableVersionByID(ctx, sc.descID, txn)
 		if err != nil {
 			return err
 		}
@@ -1292,9 +1295,17 @@ func (sc *SchemaChanger) done(ctx context.Context) error {
 			// The table descriptor is unchanged, return without writing anything.
 			return nil
 		}
+		committedMutations := scTable.Mutations[:i]
 		// Trim the executed mutations from the descriptor.
 		scTable.Mutations = scTable.Mutations[i:]
 
+		// Check any jobs that we need to depend on for the current
+		// job to be successful.
+		depMutationJobs, err = sc.getDependentMutationsJobs(ctx, scTable, committedMutations)
+		if err != nil {
+			return err
+		}
+
 		for i, g := range scTable.MutationJobs {
 			if g.MutationID == sc.mutationID {
 				// Trim the executed mutation group from the descriptor.
@@ -1385,6 +1396,14 @@ func (sc *SchemaChanger) done(ctx context.Context) error {
 	if err := sc.jobRegistry.NotifyToAdoptJobs(ctx); err != nil {
 		return err
 	}
+
+	// If any operations was skipped because a mutation was made
+	// redundant due to a column getting dropped later on then we should
+	// wait for those jobs to complete before returning our result back.
+	if err := sc.jobRegistry.WaitForJobs(ctx, sc.execCfg.InternalExecutor, depMutationJobs); err != nil {
+		return errors.Wrap(err, "A dependent transaction failed for this schema change")
+	}
+
 	return nil
 }
 
@@ -1525,6 +1544,7 @@ func (sc *SchemaChanger) maybeReverseMutations(ctx context.Context, causingError
 		// Mutation is already reversed, so we don't need to do any more work.
 		// This can happen if the mutations were already reversed, but before
 		// the rollback completed the job was adopted.
+		// the rollback completed the job was adopted.
 		if alreadyReversed {
 			return nil
 		}
@@ -1564,6 +1584,12 @@ func (sc *SchemaChanger) maybeReverseMutations(ctx context.Context, causingError
 				return errors.AssertionFailedf("mutation already rolled back: %v", mutation)
 			}
 
+			// Ignore mutations that would be skipped, nothing
+			// to reverse here.
+			if discarded, _ := isCurrentMutationDiscarded(scTable, mutation, i+1); discarded {
+				continue
+			}
+
 			log.Warningf(ctx, "reverse schema change mutation: %+v", mutation)
 			scTable.Mutations[i], columns = sc.reverseMutation(mutation, false /*notStarted*/, columns)
 
@@ -2094,7 +2120,6 @@ func (r schemaChangeResumer) Resume(ctx context.Context, execCtx interface{}) er
 			return err
 		}
 	}
-
 	execSchemaChange := func(descID descpb.ID, mutationID descpb.MutationID, droppedDatabaseID descpb.ID) error {
 		sc := SchemaChanger{
 			descID:               descID,
@@ -2519,3 +2544,64 @@ func DeleteTableDescAndZoneConfig(
 		return txn.Run(ctx, b)
 	})
 }
+
+// getDependentMutationJobs gets the dependent jobs that need to complete for
+// the current schema change to be considered successful. For example, if column
+// A will have a unique index, but it later gets dropped. Then for this mutation
+// to be successful the drop column job has to be successful too.
+func (sc *SchemaChanger) getDependentMutationsJobs(
+	ctx context.Context, tableDesc *tabledesc.Mutable, mutations []descpb.DescriptorMutation,
+) ([]jobspb.JobID, error) {
+	dependentJobs := make([]jobspb.JobID, 0, len(tableDesc.MutationJobs))
+	for _, m := range mutations {
+		// Find all the mutations that we depend
+		discarded, dependentID := isCurrentMutationDiscarded(tableDesc, m, 0)
+		if discarded {
+			jobID, err := getJobIDForMutationWithDescriptor(ctx, tableDesc, dependentID)
+			if err != nil {
+				return nil, err
+			}
+			dependentJobs = append(dependentJobs, jobID)
+		}
+	}
+	return dependentJobs, nil
+}
+
+// isCurrentMutationDiscarded returns if the current column mutation is made irrelevant
+// by a later operation. The nextMutationIdx provides the index at which to check for
+// later mutation.
+func isCurrentMutationDiscarded(
+	tableDesc *tabledesc.Mutable, currentMutation descpb.DescriptorMutation, nextMutationIdx int,
+) (bool, descpb.MutationID) {
+	if nextMutationIdx+1 > len(tableDesc.Mutations) {
+		return false, descpb.InvalidMutationID
+	}
+
+	colToCheck := make([]descpb.ColumnID, 0, 1)
+	// Both NOT NULL related updates and check constraint updates
+	// involving this column will get cancelled out by a drop column.
+	if constraint := currentMutation.GetConstraint(); constraint != nil {
+		if constraint.ConstraintType == descpb.ConstraintToUpdate_NOT_NULL {
+			colToCheck = append(colToCheck, constraint.NotNullColumn)
+		} else if constraint.ConstraintType == descpb.ConstraintToUpdate_CHECK {
+			colToCheck = constraint.Check.ColumnIDs
+		}
+	}
+
+	for _, m := range tableDesc.Mutations[nextMutationIdx:] {
+		colDesc := m.GetColumn()
+		if m.Direction == descpb.DescriptorMutation_DROP &&
+			colDesc != nil &&
+			!m.Rollback {
+			// Column was dropped later on, so this operation
+			// should be a no-op.
+			for _, col := range colToCheck {
+				if colDesc.ID == col {
+					return true, m.MutationID
+				}
+			}
+		}
+	}
+	// Not discarded by any later operation.
+	return false, descpb.InvalidMutationID
+}