catalog: Add method ConcurrentSchemaChangeJobIDs to Descriptor interface

This commits adds a method ConcurrentSchemaChangeJobIDs that retrieves
all in-progress schema change job IDs on a particular descriptor, legacy
or declarative.

Release note: None

Author: Xiang-Gu

pkg/sql/catalog/dbdesc/database_desc.go

@@ -466,6 +466,15 @@ func (desc *immutable) HasConcurrentSchemaChanges() bool {
 		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID
 }
 
+// ConcurrentSchemaChangeJobIDs implements catalog.Descriptor.
+func (desc *immutable) ConcurrentSchemaChangeJobIDs() (ret []catpb.JobID) {
+	if desc.DeclarativeSchemaChangerState != nil &&
+		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID {
+		ret = append(ret, desc.DeclarativeSchemaChangerState.JobID)
+	}
+	return ret
+}
+
 // GetDefaultPrivilegeDescriptor returns a DefaultPrivilegeDescriptor.
 func (desc *immutable) GetDefaultPrivilegeDescriptor() catalog.DefaultPrivilegeDescriptor {
 	defaultPrivilegeDescriptor := desc.GetDefaultPrivileges()

pkg/sql/catalog/descriptor.go

@@ -246,6 +246,10 @@ type Descriptor interface {
 	// in progress, either legacy or declarative.
 	HasConcurrentSchemaChanges() bool
 
+	// ConcurrentSchemaChangeJobIDs returns all in-progress schema change
+	// jobs, either legacy or declarative.
+	ConcurrentSchemaChangeJobIDs() []catpb.JobID
+
 	// SkipNamespace is true when a descriptor should not have a namespace record.
 	SkipNamespace() bool
 

pkg/sql/catalog/funcdesc/func_desc.go

@@ -394,6 +394,15 @@ func (desc *immutable) HasConcurrentSchemaChanges() bool {
 		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID
 }
 
+// ConcurrentSchemaChangeJobIDs implements the catalog.Descriptor interface.
+func (desc *immutable) ConcurrentSchemaChangeJobIDs() (ret []catpb.JobID) {
+	if desc.DeclarativeSchemaChangerState != nil &&
+		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID {
+		ret = append(ret, desc.DeclarativeSchemaChangerState.JobID)
+	}
+	return ret
+}
+
 // SkipNamespace implements the catalog.Descriptor interface.
 func (desc *immutable) SkipNamespace() bool {
 	return true

pkg/sql/catalog/schemadesc/schema_desc.go

@@ -336,6 +336,15 @@ func (desc *immutable) HasConcurrentSchemaChanges() bool {
 		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID
 }
 
+// ConcurrentSchemaChangeJobIDs implements catalog.Descriptor.
+func (desc *immutable) ConcurrentSchemaChangeJobIDs() (ret []catpb.JobID) {
+	if desc.DeclarativeSchemaChangerState != nil &&
+		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID {
+		ret = append(ret, desc.DeclarativeSchemaChangerState.JobID)
+	}
+	return ret
+}
+
 // MaybeIncrementVersion implements the MutableDescriptor interface.
 func (desc *Mutable) MaybeIncrementVersion() {
 	// Already incremented, no-op.

pkg/sql/catalog/schemadesc/synthetic_schema_desc.go

@@ -111,6 +111,11 @@ func (p synthetic) HasConcurrentSchemaChanges() bool {
 	return false
 }
 
+// ConcurrentSchemaChangeJobIDs implements catalog.Descriptor.
+func (p synthetic) ConcurrentSchemaChangeJobIDs() []catpb.JobID {
+	return nil
+}
+
 // SkipNamespace implements the descriptor interface.
 // We never store synthetic descriptors.
 func (p synthetic) SkipNamespace() bool {

pkg/sql/catalog/tabledesc/table_desc.go

@@ -73,6 +73,20 @@ func (desc *wrapper) HasConcurrentSchemaChanges() bool {
 		len(desc.MutationJobs) > 0
 }
 
+// ConcurrentSchemaChangeJobIDs implements catalog.Descriptor.
+func (desc *wrapper) ConcurrentSchemaChangeJobIDs() (ret []catpb.JobID) {
+	if desc.DeclarativeSchemaChangerState != nil &&
+		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID {
+		ret = append(ret, desc.DeclarativeSchemaChangerState.JobID)
+	}
+	if len(desc.MutationJobs) > 0 {
+		for _, mutationJob := range desc.MutationJobs {
+			ret = append(ret, mutationJob.JobID)
+		}
+	}
+	return ret
+}
+
 // SkipNamespace implements the descriptor interface.
 func (desc *wrapper) SkipNamespace() bool {
 	// Virtual tables are hard-coded and don't have entries in the

pkg/sql/catalog/typedesc/table_implicit_record_type.go

@@ -281,6 +281,11 @@ func (v *tableImplicitRecordType) HasConcurrentSchemaChanges() bool {
 	return false
 }
 
+// ConcurrentSchemaChangeJobIDs implements catalog.Descriptor.
+func (v *tableImplicitRecordType) ConcurrentSchemaChangeJobIDs() []catpb.JobID {
+	return nil
+}
+
 // SkipNamespace implements catalog.Descriptor. We never store table implicit
 // record type which is always constructed in memory.
 func (v *tableImplicitRecordType) SkipNamespace() bool {

pkg/sql/catalog/typedesc/type_desc.go

@@ -872,6 +872,15 @@ func (desc *immutable) HasConcurrentSchemaChanges() bool {
 	return false
 }
 
+// ConcurrentSchemaChangeJobIDs implements catalog.Descriptor.
+func (desc *immutable) ConcurrentSchemaChangeJobIDs() (ret []catpb.JobID) {
+	if desc.DeclarativeSchemaChangerState != nil &&
+		desc.DeclarativeSchemaChangerState.JobID != catpb.InvalidJobID {
+		ret = append(ret, desc.DeclarativeSchemaChangerState.JobID)
+	}
+	return ret
+}
+
 // SkipNamespace implements the descriptor interface.
 func (desc *immutable) SkipNamespace() bool {
 	return false