schemachanger: Enable ALTER PRIMARY KEY USING HASH

We enable ALTER PRIMARY KEY USING HASH in the declarative schema
changer. This commit also cleaned up some common logic to be re-used
with CREATE INDEX ... USING HASH.

Note that we still fallback to legacy schema changer if the old primary
key is on the implicit `rowid` column, because we don't support
`ADD COLUMN, DROP COLUMN` in one statement yet (adding a shard column
and dropping the `rowid` column).

Release note: None

pkg/sql/backfill.go

@@ -1574,6 +1574,9 @@ func ValidateConstraint(
 			return txn.WithSyntheticDescriptors(
 				[]catalog.Descriptor{tableDesc},
 				func() error {
+					if skip, err := canSkipCheckValidation(tableDesc, ck); err != nil || skip {
+						return err
+					}
 					violatingRow, formattedCkExpr, err := validateCheckExpr(ctx, &semaCtx, txn, sessionData, ck.GetExpr(),
 						tableDesc.(*tabledesc.Mutable), indexIDForValidation)
 					if err != nil {
@@ -1630,6 +1633,29 @@ func ValidateConstraint(
 	})
 }
 
+// canSkipCheckValidation returns true if
+//  1. ck is from a hash-sharded column (because the shard column's computed
+//     expression is a modulo operation and thus the check constraint is
+//     valid by construction).
+//  2. ck is a NOT-NULL check and the column is a shard column (because shard
+//     column is a computed column and thus not null by construction).
+func canSkipCheckValidation(
+	tableDesc catalog.TableDescriptor, ck catalog.CheckConstraint,
+) (bool, error) {
+	if ck.IsHashShardingConstraint() {
+		return true, nil
+	}
+	if ck.IsNotNullColumnConstraint() {
+		colID := ck.GetReferencedColumnID(0)
+		col, err := catalog.MustFindColumnByID(tableDesc, colID)
+		if err != nil {
+			return false, err
+		}
+		return tableDesc.IsShardColumn(col), nil
+	}
+	return false, nil
+}
+
 // ValidateInvertedIndexes checks that the indexes have entries for
 // all the items of data in rows.
 //

pkg/sql/catalog/tabledesc/table.go

@@ -554,6 +554,16 @@ func RenameColumnInTable(
 		if !shardedDesc.IsSharded {
 			return
 		}
+		// Simpler case: If the shard column is to be renamed, keep the
+		// shard descriptor name in sync.
+		if shardedDesc.Name == string(col.ColName()) {
+			shardedDesc.Name = string(newName)
+			return
+		}
+		// Harder case: If one of the columns that the shard column is based on is
+		// to be renamed, then rename the base column in the descriptor as well as
+		// the shard descriptor name. We also record this fact in `shardColumnsToRename`
+		// so the next recursive call will rename the shard column.
 		oldShardColName := tree.Name(GetShardColumnName(
 			shardedDesc.ColumnNames, shardedDesc.ShardBuckets))
 		var changed bool

pkg/sql/logictest/testdata/logic_test/alter_primary_key

@@ -669,7 +669,7 @@ t  CREATE TABLE public.t (
 statement ok
 CREATE INDEX i ON t (x);
 
-statement error 42P07 constraint with name \"i\" already exists
+statement error pgcode 42P07 constraint with name \"i\" already exists
 ALTER TABLE t DROP CONSTRAINT "my_pk", ADD CONSTRAINT "i" PRIMARY KEY (x);
 
 # Regression for #45362.
@@ -1860,3 +1860,24 @@ INSERT INTO t_97296 VALUES (0.0, 5, 32);
 
 statement ok
 ALTER TABLE t_97296 ALTER PRIMARY KEY USING COLUMNS (b, a);
+
+# This subtest ensures we properly support ALTER PRIMARY KEY USING HASH
+subtest alter_primary_key_using_hash_96730
+
+statement ok
+CREATE TABLE t_96730 (i INT PRIMARY KEY, j INT NOT NULL, k STRING NOT NULL, FAMILY "primary" (i, j, k));
+
+statement ok
+ALTER TABLE t_96730 ALTER PRIMARY KEY USING COLUMNS (j, k) USING HASH
+
+query TT
+SHOW CREATE TABLE t_96730
+----
+t_96730  CREATE TABLE public.t_96730 (
+           i INT8 NOT NULL,
+           j INT8 NOT NULL,
+           k STRING NOT NULL,
+           crdb_internal_j_k_shard_16 INT8 NOT VISIBLE NOT NULL AS (mod(fnv32(crdb_internal.datums_to_bytes(j, k)), 16:::INT8)) VIRTUAL,
+           CONSTRAINT t_96730_pkey PRIMARY KEY (j ASC, k ASC) USING HASH WITH (bucket_count=16),
+           UNIQUE INDEX t_96730_i_key (i ASC)
+         )

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/alter_table.go

@@ -78,6 +78,8 @@ func init() {
 
 // alterTableChecks determines if the entire set of alter table commands
 // are supported.
+// One side-effect is that this function will modify `n` when it hoists
+// add column constraints.
 func alterTableChecks(
 	n *tree.AlterTable,
 	mode sessiondatapb.NewSchemaChangerMode,

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/alter_table_add_column.go

@@ -408,7 +408,7 @@ func handleAddColumnFreshlyAddedPrimaryIndex(
 		Filter(isColumnFilter).
 		Filter(absentTargetFilter).
 		IsEmpty(); haveDroppingColumn {
-		panic(scerrors.NotImplementedErrorf(n, "DROP COLUMN after ADD COLUMN"))
+		panic(scerrors.NotImplementedErrorf(n, "ADD COLUMN after DROP COLUMN"))
 	}
 
 	var tempIndex *scpb.TemporaryIndex