sql: Add blocking schema changer job IDs to log entries for concurrent schema change

Previously, we had log entries that says a schema change job is waiting
for concurrent schema change jobs without saying what those blocking
schema change jobs are. This is inadequate for troubleshooting and
debugging. This commit fixes it. Note that because we have both
legacy and declarative schema changer, there are two paths of handling
concurrent schema changes:
1. legacy ongoing, another legacy comes in: The second, legacy schema
changer will go through, create a job if needed, and the job will wait
for the previous legacy schema changer job to finish before proceeding.
This commit provides all schema change jobs before the current one that
are blocking.

2. legacy + declarative, declarative + declarative, or declarative + legacy:
in all three cases, the second schema change will return an error back
to its upstream -- conn_executor -- and we wait for all concurrent
schema change jobs to finish in a retry loop in the conn_executor. We
also provides all the concurrent schema change job IDs that the
conn_executor is waiting in this commit.

Release note: None

Author: Xiang-Gu

pkg/sql/schema_change_plan_node.go

@@ -21,6 +21,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/descmetadata"
@@ -203,6 +204,7 @@ func (p *planner) waitForDescriptorSchemaChanges(
 	for r := retry.StartWithCtx(ctx, base.DefaultRetryOptions()); r.Next(); {
 		now := p.ExecCfg().Clock.Now()
 		var isBlocked bool
+		var blockingJobIDs []catpb.JobID
 		if err := p.ExecCfg().InternalDB.DescsTxn(ctx, func(
 			ctx context.Context, txn descs.Txn,
 		) error {
@@ -214,6 +216,7 @@ func (p *planner) waitForDescriptorSchemaChanges(
 				return err
 			}
 			isBlocked = desc.HasConcurrentSchemaChanges()
+			blockingJobIDs = desc.ConcurrentSchemaChangeJobIDs()
 			return nil
 		}); err != nil {
 			return err
@@ -225,8 +228,8 @@ func (p *planner) waitForDescriptorSchemaChanges(
 		}
 		if logEvery.ShouldLog() {
 			log.Infof(ctx,
-				"schema change waiting for concurrent schema changes on descriptor %d,"+
-					" waited %v so far", descID, timeutil.Since(start),
+				"schema change waiting for %v concurrent schema change job(s) %v on descriptor %d,"+
+					" waited %v so far", len(blockingJobIDs), blockingJobIDs, descID, timeutil.Since(start),
 			)
 		}
 	}

pkg/sql/schema_changer.go

@@ -584,14 +584,29 @@ func (sc *SchemaChanger) notFirstInLine(ctx context.Context, desc catalog.Descri
 		// descriptor, it seems possible for a job to be resumed after the mutation
 		// has already been removed. If there's a mutation provided, we should check
 		// whether it actually exists on the table descriptor and exit the job if not.
-		for i, mutation := range tableDesc.AllMutations() {
+		allMutations := tableDesc.AllMutations()
+		for i, mutation := range allMutations {
 			if mutation.MutationID() == sc.mutationID {
 				if i != 0 {
+					blockingJobIDsAsSet := make(map[catpb.JobID]struct{})
+					for j := 0; j < i; j++ {
+						blockingJobID, err := mustGetJobIDWithMutationID(tableDesc, allMutations[j].MutationID())
+						if err != nil {
+							return err
+						}
+						blockingJobIDsAsSet[blockingJobID] = struct{}{}
+					}
+					blockingJobIDs := make([]catpb.JobID, 0, len(blockingJobIDsAsSet))
+					for jobID := range blockingJobIDsAsSet {
+						blockingJobIDs = append(blockingJobIDs, jobID)
+					}
 					log.Infof(ctx,
-						"schema change on %q (v%d): another change is still in progress",
-						desc.GetName(), desc.GetVersion(),
+						"schema change on %q (v%d): another %v schema change job(s) %v is still in progress "+
+							"and it is blocking this job from proceeding",
+						desc.GetName(), desc.GetVersion(), len(blockingJobIDs), blockingJobIDs,
 					)
-					return errSchemaChangeNotFirstInLine
+					return errors.Wrapf(errSchemaChangeNotFirstInLine, "schema change is "+
+						"blocked by %v other schema change job(s) %v", len(blockingJobIDs), blockingJobIDs)
 				}
 				break
 			}
@@ -600,6 +615,21 @@ func (sc *SchemaChanger) notFirstInLine(ctx context.Context, desc catalog.Descri
 	return nil
 }
 
+func mustGetJobIDWithMutationID(
+	tableDesc catalog.TableDescriptor, mutationID descpb.MutationID,
+) (jobID catpb.JobID, err error) {
+	for _, mutationJob := range tableDesc.GetMutationJobs() {
+		if mutationJob.MutationID == mutationID {
+			jobID = mutationJob.JobID
+		}
+	}
+	if jobID == catpb.InvalidJobID {
+		return 0, errors.AssertionFailedf("mutation job with mutation ID %v is not found in table %q",
+			mutationID, tableDesc.GetName())
+	}
+	return jobID, nil
+}
+
 func (sc *SchemaChanger) getTargetDescriptor(ctx context.Context) (catalog.Descriptor, error) {
 	// Retrieve the descriptor that is being changed.
 	var desc catalog.Descriptor

pkg/sql/schema_changer_helpers_test.go

@@ -12,13 +12,18 @@ package sql
 
 import (
 	"context"
+	"regexp"
+	"sort"
+	"strings"
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/backfill"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -88,3 +93,89 @@ func TestCalculateSplitAtShards(t *testing.T) {
 		})
 	}
 }
+
+// TestNotFirstInLine tests that if a schema change's mutation is not first in
+// line, the error message clearly state what the blocking schema change job ID
+// is.
+func TestNotFirstInLine(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	// A helper to extract blocking schema change job IDs from the error message
+	// which is of the form "schema change is blocked by x other schema change
+	// job(s) [yyy zzz]".
+	extractSortedBlockingJobIDsFromNotFirstInLineErr := func(errMsg string) []string {
+		p := regexp.MustCompile(`\[(.*?)\]`)
+		m := p.FindStringSubmatch(errMsg)
+		require.NotNilf(t, m, "did not find any blocking job IDs")
+		blockingJobIDs := strings.Fields(m[1])
+		sort.Slice(blockingJobIDs, func(i, j int) bool {
+			return blockingJobIDs[i] <= blockingJobIDs[j]
+		})
+		return blockingJobIDs
+	}
+
+	ctx := context.Background()
+	desc := descpb.TableDescriptor{
+		Name: "t",
+		ID:   104,
+		Mutations: []descpb.DescriptorMutation{
+			{
+				Descriptor_: &descpb.DescriptorMutation_Index{},
+				MutationID:  1,
+			},
+			{
+				Descriptor_: &descpb.DescriptorMutation_Index{},
+				MutationID:  1,
+			},
+			{
+				Descriptor_: &descpb.DescriptorMutation_Column{},
+				MutationID:  2,
+			},
+			{
+				Descriptor_: &descpb.DescriptorMutation_Column{},
+				MutationID:  3,
+			},
+		},
+		MutationJobs: []descpb.TableDescriptor_MutationJob{
+			{
+				JobID:      11111,
+				MutationID: 1,
+			},
+			{
+				JobID:      22222,
+				MutationID: 2,
+			},
+			{
+				JobID:      33333,
+				MutationID: 3,
+			},
+		},
+	}
+	mut := tabledesc.NewBuilder(&desc).BuildExistingMutableTable()
+	{
+		sc := SchemaChanger{
+			descID:     104,
+			mutationID: 1,
+		}
+		err := sc.notFirstInLine(ctx, mut)
+		require.NoError(t, err)
+	}
+	{
+		sc := SchemaChanger{
+			descID:     104,
+			mutationID: 2,
+		}
+		err := sc.notFirstInLine(ctx, mut)
+		require.True(t, errors.Is(err, errSchemaChangeNotFirstInLine))
+		require.Equal(t, []string{"11111"}, extractSortedBlockingJobIDsFromNotFirstInLineErr(err.Error()))
+	}
+	{
+		sc := SchemaChanger{
+			descID:     104,
+			mutationID: 3,
+		}
+		err := sc.notFirstInLine(ctx, mut)
+		require.True(t, errors.Is(err, errSchemaChangeNotFirstInLine))
+		require.Equal(t, []string{"11111", "22222"}, extractSortedBlockingJobIDsFromNotFirstInLineErr(err.Error()))
+	}
+}

pkg/sql/schema_changer_test.go

@@ -16,6 +16,7 @@ import (
 	"database/sql/driver"
 	"fmt"
 	"math/rand"
+	"regexp"
 	"strconv"
 	"strings"
 	"sync"
@@ -8389,3 +8390,41 @@ SELECT fraction_completed > 0
 		}
 	}
 }
+
+// TestLegacySchemaChangerWaitsForOtherSchemaChanges tests concurrent legacy schema changes
+// wait properly for preceding ones if it's not first in line.
+func TestLegacySchemaChangerWaitsForOtherSchemaChanges(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+	ctx := context.Background()
+	params, _ := tests.CreateTestServerParams()
+	params.Knobs.JobsTestingKnobs = jobs.NewTestingKnobsWithShortIntervals()
+
+	s, sqlDB, _ := serverutils.StartServer(t, params)
+	defer s.Stopper().Stop(ctx)
+	tdb := sqlutils.MakeSQLRunner(sqlDB)
+
+	tdb.Exec(t, `SET use_declarative_schema_changer = off`)
+	tdb.Exec(t, `CREATE TABLE t (i INT PRIMARY KEY);`)
+	tdb.Exec(t, `SET CLUSTER SETTING jobs.debug.pausepoints = 'schemachanger.before.exec';`)
+
+	pattern, err := regexp.Compile(`\d+`)
+	require.NoError(t, err)
+	_, err = sqlDB.Exec(`CREATE INDEX idx ON t (i);`)
+	jobID1 := pattern.FindString(err.Error())
+	require.NotEmpty(t, jobID1)
+	_, err = sqlDB.Exec(`ALTER TABLE t ADD COLUMN j INT DEFAULT 30;`)
+	jobID2 := pattern.FindString(err.Error())
+	require.NotEmpty(t, jobID2)
+
+	tdb.Exec(t, `SET CLUSTER SETTING jobs.debug.pausepoints = '';`)
+	tdb.Exec(t, `RESUME JOB $1`, jobID2)
+	tdb.Exec(t, `RESUME JOB $1`, jobID1)
+	testutils.SucceedsSoon(t, func() error {
+		res := tdb.QueryStr(t, `SELECT status FROM [SHOW JOBS] WHERE job_id in ($1, $2)`, jobID1, jobID2)
+		if len(res) == 2 && res[0][0] == "succeeded" && res[1][0] == "succeeded" {
+			return nil
+		}
+		return errors.New("")
+	})
+}