From 7dbf54538b72a329ebe9c1572d85a2a11da4e703 Mon Sep 17 00:00:00 2001
From: Andrew Werner <awerner32@gmail.com>
Date: Wed, 24 Aug 2022 11:35:29 -0400
Subject: [PATCH] sql/schemachanger: version gate element creation

We cannot create elements the old version of the code does not know about.

Release justification: fixed mixed version incompatibility
Release note: None
---
 pkg/sql/schemachanger/scbuild/build.go      |  6 +++++
 pkg/sql/schemachanger/screl/BUILD.bazel     |  1 +
 pkg/sql/schemachanger/screl/scalars.go      | 27 +++++++++++++++++++++
 pkg/sql/schemachanger/screl/scalars_test.go | 19 ++++++++++++---
 4 files changed, 50 insertions(+), 3 deletions(-)

diff --git a/pkg/sql/schemachanger/scbuild/build.go b/pkg/sql/schemachanger/scbuild/build.go
index 6fbf279b6dd6..2c76fc3c9da6 100644
--- a/pkg/sql/schemachanger/scbuild/build.go
+++ b/pkg/sql/schemachanger/scbuild/build.go
@@ -89,12 +89,18 @@ func Build(
 		Authorization: els.authorization,
 	}
 	current := make([]scpb.Status, 0, len(bs.output))
+	version := dependencies.ClusterSettings().Version.ActiveVersion(ctx)
 	for _, e := range bs.output {
 		if e.metadata.Size() == 0 {
 			// Exclude targets which weren't explicitly set.
 			// Explicitly-set targets have non-zero values in the target metadata.
 			continue
 		}
+		// Exclude targets which are not yet usable in the currently active
+		// cluster version.
+		if !version.IsActive(screl.MinVersion(e.element)) {
+			continue
+		}
 		ts.Targets = append(ts.Targets, scpb.MakeTarget(e.target, e.element, &e.metadata))
 		current = append(current, e.current)
 	}
diff --git a/pkg/sql/schemachanger/screl/BUILD.bazel b/pkg/sql/schemachanger/screl/BUILD.bazel
index 56f031f84c65..0e2b08ddd320 100644
--- a/pkg/sql/schemachanger/screl/BUILD.bazel
+++ b/pkg/sql/schemachanger/screl/BUILD.bazel
@@ -17,6 +17,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/schemachanger/screl",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/clusterversion",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/catpb",
         "//pkg/sql/catalog/descpb",
diff --git a/pkg/sql/schemachanger/screl/scalars.go b/pkg/sql/schemachanger/screl/scalars.go
index 673c1c46b5e7..b8d8d24b581f 100644
--- a/pkg/sql/schemachanger/screl/scalars.go
+++ b/pkg/sql/schemachanger/screl/scalars.go
@@ -11,6 +11,7 @@
 package screl
 
 import (
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
@@ -91,3 +92,29 @@ func ContainsDescID(haystack scpb.Element, needle catid.DescID) (contains bool)
 	})
 	return contains
 }
+
+// MinVersion returns the minimum cluster version at which an element may
+// be used.
+func MinVersion(el scpb.Element) clusterversion.Key {
+	switch el.(type) {
+	case *scpb.Database, *scpb.Schema, *scpb.View, *scpb.Sequence, *scpb.Table,
+		*scpb.AliasType, *scpb.ColumnFamily, *scpb.Column, *scpb.PrimaryIndex,
+		*scpb.SecondaryIndex, *scpb.TemporaryIndex, *scpb.EnumType,
+		*scpb.UniqueWithoutIndexConstraint, *scpb.CheckConstraint,
+		*scpb.ForeignKeyConstraint, *scpb.TableComment, *scpb.RowLevelTTL,
+		*scpb.TableLocalityGlobal, *scpb.TableLocalityPrimaryRegion,
+		*scpb.TableLocalitySecondaryRegion, *scpb.TableLocalityRegionalByRow,
+		*scpb.ColumnName, *scpb.ColumnType, *scpb.ColumnDefaultExpression,
+		*scpb.ColumnOnUpdateExpression, *scpb.SequenceOwner, *scpb.ColumnComment,
+		*scpb.IndexName, *scpb.IndexPartitioning, *scpb.SecondaryIndexPartial,
+		*scpb.IndexComment, *scpb.ConstraintName, *scpb.ConstraintComment,
+		*scpb.Namespace, *scpb.Owner, *scpb.UserPrivileges,
+		*scpb.DatabaseRegionConfig, *scpb.DatabaseRoleSetting, *scpb.DatabaseComment,
+		*scpb.SchemaParent, *scpb.SchemaComment, *scpb.ObjectParent:
+		return clusterversion.V22_1
+	case *scpb.IndexColumn, *scpb.EnumTypeValue, *scpb.TableZoneConfig:
+		return clusterversion.UseDelRangeInGCJob
+	default:
+		panic(errors.AssertionFailedf("unknown element %T", el))
+	}
+}
diff --git a/pkg/sql/schemachanger/screl/scalars_test.go b/pkg/sql/schemachanger/screl/scalars_test.go
index 1686692a89bb..1c19326114b3 100644
--- a/pkg/sql/schemachanger/screl/scalars_test.go
+++ b/pkg/sql/schemachanger/screl/scalars_test.go
@@ -24,11 +24,24 @@ import (
 
 // TestAllElementsHaveDescID ensures that all element types have a DescID.
 func TestAllElementsHaveDescID(t *testing.T) {
+	forEachElementType(func(elem scpb.Element) {
+		require.Equalf(t, descpb.ID(0), GetDescID(elem), "elem %T", elem)
+	})
+}
+
+func TestAllElementsHaveMinVersion(t *testing.T) {
+	forEachElementType(func(elem scpb.Element) {
+		// If `elem` does not have a min version, the following function call will panic.
+		MinVersion(elem)
+	})
+}
+
+func forEachElementType(f func(element scpb.Element)) {
 	typ := reflect.TypeOf((*scpb.ElementProto)(nil)).Elem()
 	for i := 0; i < typ.NumField(); i++ {
-		f := typ.Field(i)
-		elem := reflect.New(f.Type.Elem()).Interface().(scpb.Element)
-		require.Equal(t, descpb.ID(0), GetDescID(elem))
+		field := typ.Field(i)
+		elem := reflect.New(field.Type.Elem()).Interface().(scpb.Element)
+		f(elem)
 	}
 }