schemachanger: Implement ADD UNIQUE WITHOUT INDEX

pkg/sql/backfill.go

@@ -1505,8 +1505,6 @@ func (e InvalidIndexesError) Error() string {
 
 // ValidateConstraint validates the constraint against all rows
 // in `tbl`.
-//
-// TODO (xiang): Support validating UNIQUE_WITHOUT_INDEX constraint in this function.
 func ValidateConstraint(
 	ctx context.Context,
 	tableDesc catalog.TableDescriptor,
@@ -1558,6 +1556,15 @@ func ValidateConstraint(
 						indexIDForValidation, txn, ie)
 				},
 			)
+		case catconstants.ConstraintTypeUniqueWithoutIndex:
+			uwi := constraint.AsUniqueWithoutIndex()
+			return ie.WithSyntheticDescriptors(
+				[]catalog.Descriptor{tableDesc},
+				func() error {
+					return validateUniqueConstraint(ctx, tableDesc, uwi.GetName(), uwi.CollectKeyColumnIDs().Ordered(),
+						uwi.GetPredicate(), indexIDForValidation, ie, txn, sessionData.User(), false)
+				},
+			)
 		default:
 			return errors.AssertionFailedf("validation of unsupported constraint type")
 		}
@@ -2049,6 +2056,7 @@ func countIndexRowsAndMaybeCheckUniqueness(
 					idx.GetName(),
 					idx.IndexDesc().KeyColumnIDs[idx.ImplicitPartitioningColumnCount():],
 					idx.GetPredicate(),
+					0, /* indexIDForValidation */
 					ie,
 					txn,
 					username.NodeUserName(),
@@ -2737,6 +2745,7 @@ func validateUniqueWithoutIndexConstraintInTxn(
 				uc.Name,
 				uc.ColumnIDs,
 				uc.Predicate,
+				0, /* indexIDForValidation */
 				ie,
 				txn,
 				user,

pkg/sql/catalog/rewrite/rewrite.go

@@ -173,6 +173,8 @@ func TableDescs(
 			}
 		}
 
+		// Rewrite unique_without_index in both `UniqueWithoutIndexConstraints`
+		// and `Mutations` slice.
 		origUniqueWithoutIndexConstraints := table.UniqueWithoutIndexConstraints
 		table.UniqueWithoutIndexConstraints = nil
 		for _, unique := range origUniqueWithoutIndexConstraints {
@@ -186,6 +188,18 @@ func TableDescs(
 					"UniqueWithoutIndexConstraint %d was not found", table.Name, unique.TableID)
 			}
 		}
+		for idx := range table.Mutations {
+			if c := table.Mutations[idx].GetConstraint(); c != nil &&
+				c.ConstraintType == descpb.ConstraintToUpdate_UNIQUE_WITHOUT_INDEX {
+				uwi := &c.UniqueWithoutIndexConstraint
+				if rewrite, ok := descriptorRewrites[uwi.TableID]; ok {
+					uwi.TableID = rewrite.ID
+				} else {
+					return errors.AssertionFailedf("cannot restore %q because referenced table ID in "+
+						"UniqueWithoutIndexConstraint %d was not found", table.Name, uwi.TableID)
+				}
+			}
+		}
 
 		if table.IsSequence() && table.SequenceOpts.HasOwner() {
 			if ownerRewrite, ok := descriptorRewrites[table.SequenceOpts.SequenceOwner.OwnerTableID]; ok {

pkg/sql/check.go

@@ -368,8 +368,15 @@ func validateForeignKey(
 // The pred argument is a partial unique constraint predicate, which filters the
 // subset of rows that are guaranteed unique by the constraint. If the unique
 // constraint is not partial, pred should be empty.
+//
+// `indexIDForValidation`, if non-zero, will be used to force the sql query to
+// use this particular index by hinting the query.
 func duplicateRowQuery(
-	srcTbl catalog.TableDescriptor, columnIDs []descpb.ColumnID, pred string, limitResults bool,
+	srcTbl catalog.TableDescriptor,
+	columnIDs []descpb.ColumnID,
+	pred string,
+	indexIDForValidation descpb.IndexID,
+	limitResults bool,
 ) (sql string, colNames []string, _ error) {
 	colNames, err := srcTbl.NamesForColumnIDs(columnIDs)
 	if err != nil {
@@ -397,13 +404,24 @@ func duplicateRowQuery(
 	if limitResults {
 		limit = " LIMIT 1"
 	}
-	return fmt.Sprintf(
+	query := fmt.Sprintf(
 		`SELECT %[1]s FROM [%[2]d AS tbl] WHERE %[3]s GROUP BY %[1]s HAVING count(*) > 1 %[4]s`,
 		strings.Join(srcCols, ", "),     // 1
 		srcTbl.GetID(),                  // 2
 		strings.Join(srcWhere, " AND "), // 3
 		limit,                           // 4
-	), colNames, nil
+	)
+	if indexIDForValidation != 0 {
+		query = fmt.Sprintf(
+			`SELECT %[1]s FROM [%[2]d AS tbl]@[%[3]d] WHERE %[4]s GROUP BY %[1]s HAVING count(*) > 1 %[5]s`,
+			strings.Join(srcCols, ", "),     // 1
+			srcTbl.GetID(),                  // 2
+			indexIDForValidation,            // 3
+			strings.Join(srcWhere, " AND "), // 4
+			limit,                           // 5
+		)
+	}
+	return query, colNames, nil
 }
 
 // RevalidateUniqueConstraintsInCurrentDB verifies that all unique constraints
@@ -473,6 +491,7 @@ func (p *planner) RevalidateUniqueConstraint(
 					index.GetName(),
 					index.IndexDesc().KeyColumnIDs[index.ImplicitPartitioningColumnCount():],
 					index.GetPredicate(),
+					0, /* indexIDForValidation */
 					p.ExecCfg().InternalExecutor,
 					p.Txn(),
 					p.User(),
@@ -493,6 +512,7 @@ func (p *planner) RevalidateUniqueConstraint(
 				uc.GetName(),
 				uc.CollectKeyColumnIDs().Ordered(),
 				uc.GetPredicate(),
+				0, /* indexIDForValidation */
 				p.ExecCfg().InternalExecutor,
 				p.Txn(),
 				p.User(),
@@ -557,6 +577,7 @@ func RevalidateUniqueConstraintsInTable(
 				index.GetName(),
 				index.IndexDesc().KeyColumnIDs[index.ImplicitPartitioningColumnCount():],
 				index.GetPredicate(),
+				0, /* indexIDForValidation */
 				ie,
 				txn,
 				user,
@@ -577,6 +598,7 @@ func RevalidateUniqueConstraintsInTable(
 				uc.GetName(),
 				uc.CollectKeyColumnIDs().Ordered(),
 				uc.GetPredicate(),
+				0, /* indexIDForValidation */
 				ie,
 				txn,
 				user,
@@ -595,6 +617,11 @@ func RevalidateUniqueConstraintsInTable(
 // validateUniqueConstraint verifies that all the rows in the srcTable
 // have unique values for the given columns.
 //
+// `indexIDForValidation`, if non-zero, will be used to force validation
+// against this particular index. This is used to facilitate the declarative
+// schema changer when the validation should be against a yet non-public
+// primary index.
+//
 // It operates entirely on the current goroutine and is thus able to
 // reuse an existing kv.Txn safely.
 //
@@ -606,13 +633,14 @@ func validateUniqueConstraint(
 	constraintName string,
 	columnIDs []descpb.ColumnID,
 	pred string,
+	indexIDForValidation descpb.IndexID,
 	ie sqlutil.InternalExecutor,
 	txn *kv.Txn,
 	user username.SQLUsername,
 	preExisting bool,
 ) error {
 	query, colNames, err := duplicateRowQuery(
-		srcTable, columnIDs, pred, true, /* limitResults */
+		srcTable, columnIDs, pred, indexIDForValidation, true, /* limitResults */
 	)
 	if err != nil {
 		return err

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -1560,7 +1560,9 @@ statement ok
 INSERT INTO unique_without_index (e, f) VALUES (1, 1), (1, 2)
 
 # But trying to add a unique constraint now fails.
-statement error pgcode 23505 pq: could not create unique constraint "my_unique_e"\nDETAIL: Key \(e\)=\(1\) is duplicated\.
+# Note that we omit the constraint name in the expected error message because if the declarative schema changer is used,
+# the constraint name, at the time of validation failure, is still a place-holder name.
+statement error pgcode 23505 pq: could not create unique constraint ".*"\nDETAIL: Key \(e\)=\(1\) is duplicated\.
 ALTER TABLE unique_without_index ADD CONSTRAINT my_unique_e UNIQUE WITHOUT INDEX (e)
 
 # We can create not-valid constraints, however.
@@ -1603,11 +1605,14 @@ SELECT usage_count
  WHERE feature_name = 'sql.schema_changer.errors.constraint_violation';
 
 # Trying to add a partial unique constraint fails when there are duplicates.
-statement error pgcode 23505 pq: could not create unique constraint "uniq_a_1"\nDETAIL: Key \(a\)=\(1\) is duplicated\.
+statement error pgcode 23505 pq: could not create unique constraint ".*"\nDETAIL: Key \(a\)=\(1\) is duplicated\.
 ALTER TABLE unique_without_index_partial ADD CONSTRAINT uniq_a_1 UNIQUE WITHOUT INDEX (a) WHERE b > 0 OR c > 0
 
 # Sanity: Check the number of user errors and
 # database errors in the test.
+# We only do this check for the legacy schema changer because the declarative schema changer does not increment
+# exactly the same counters.
+onlyif config local-legacy-schema-changer
 query B
 SELECT usage_count > $constraint_violations_before
   FROM crdb_internal.feature_usage

pkg/sql/schema_changer_test.go

@@ -5627,6 +5627,7 @@ func TestTableValidityWhileAddingUniqueConstraint(t *testing.T) {
 	if _, err := sqlDB.Exec(`
 CREATE DATABASE t;
 CREATE TABLE t.tab (a INT PRIMARY KEY, b INT, c INT);
+SET use_declarative_schema_changer = off;
 `); err != nil {
 		t.Fatal(err)
 	}

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

@@ -54,6 +54,8 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 		// Support ALTER TABLE ... ADD PRIMARY KEY
 		if d, ok := t.ConstraintDef.(*tree.UniqueConstraintTableDef); ok && d.PrimaryKey && t.ValidationBehavior == tree.ValidationDefault {
 			return true
+		} else if ok && d.WithoutIndex && t.ValidationBehavior == tree.ValidationDefault {
+			return true
 		}
 
 		// Support ALTER TABLE ... ADD CONSTRAINT CHECK
@@ -75,9 +77,10 @@ var supportedAlterTableStatements = map[reflect.Type]supportedAlterTableCommand{
 // They key is constructed as "ADD" + constraint type + validation behavior, joined with "_".
 // E.g. "ADD_PRIMARY_KEY_DEFAULT", "ADD_CHECK_SKIP", "ADD_FOREIGN_KEY_DEFAULT", etc.
 var alterTableAddConstraintMinSupportedClusterVersion = map[string]clusterversion.Key{
-	"ADD_PRIMARY_KEY_DEFAULT": clusterversion.V22_2Start,
-	"ADD_CHECK_DEFAULT":       clusterversion.V23_1Start,
-	"ADD_FOREIGN_KEY_DEFAULT": clusterversion.V23_1Start,
+	"ADD_PRIMARY_KEY_DEFAULT":          clusterversion.V22_2Start,
+	"ADD_CHECK_DEFAULT":                clusterversion.V23_1Start,
+	"ADD_FOREIGN_KEY_DEFAULT":          clusterversion.V23_1Start,
+	"ADD_UNIQUE_WITHOUT_INDEX_DEFAULT": clusterversion.V23_1Start,
 }
 
 func init() {
@@ -166,6 +169,8 @@ func alterTableAddConstraintSupportedInCurrentClusterVersion(
 	case *tree.UniqueConstraintTableDef:
 		if d.PrimaryKey {
 			cmdKey = "ADD_PRIMARY_KEY"
+		} else if d.WithoutIndex {
+			cmdKey = "ADD_UNIQUE_WITHOUT_INDEX"
 		}
 	case *tree.CheckConstraintTableDef:
 		cmdKey = "ADD_CHECK"

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

@@ -41,6 +41,8 @@ func alterTableAddConstraint(
 	case *tree.UniqueConstraintTableDef:
 		if d.PrimaryKey && t.ValidationBehavior == tree.ValidationDefault {
 			alterTableAddPrimaryKey(b, tn, tbl, t)
+		} else if d.WithoutIndex && t.ValidationBehavior == tree.ValidationDefault {
+			alterTableAddUniqueWithoutIndex(b, tn, tbl, t)
 		}
 	case *tree.CheckConstraintTableDef:
 		if t.ValidationBehavior == tree.ValidationDefault {
@@ -373,9 +375,110 @@ func alterTableAddForeignKey(
 	})
 }
 
-// getFullyResolvedColNames returns fully resolved column names.
-// For each name in `colNames`, its fully resolved name will be "db.sc.tbl.col".
-// The order of column names in the return is in syc with the input `colNames`.
+// alterTableAddUniqueWithoutIndex contains logic for building ALTER TABLE ... ADD CONSTRAINT ... UNIQUE WITHOUT INDEX.
+// It assumes `t` is such a command.
+func alterTableAddUniqueWithoutIndex(
+	b BuildCtx, tn *tree.TableName, tbl *scpb.Table, t *tree.AlterTableAddConstraint,
+) {
+	d := t.ConstraintDef.(*tree.UniqueConstraintTableDef)
+
+	// 1. A bunch of checks.
+	if !b.SessionData().EnableUniqueWithoutIndexConstraints {
+		panic(pgerror.New(pgcode.FeatureNotSupported,
+			"unique constraints without an index are not yet supported",
+		))
+	}
+	if len(d.Storing) > 0 {
+		panic(pgerror.New(pgcode.FeatureNotSupported,
+			"unique constraints without an index cannot store columns",
+		))
+	}
+	if d.PartitionByIndex.ContainsPartitions() {
+		panic(pgerror.New(pgcode.FeatureNotSupported,
+			"partitioned unique constraints without an index are not supported",
+		))
+	}
+	if d.NotVisible {
+		// Theoretically, this should never happen because this is not supported by
+		// the parser. This is just a safe check.
+		panic(pgerror.Newf(pgcode.FeatureNotSupported,
+			"creating a unique constraint using UNIQUE WITH NOT VISIBLE INDEX is not supported",
+		))
+	}
+
+	// 2. Check that columns that we want to have uniqueness should have no duplicate.
+	var colSet catalog.TableColSet
+	var colIDs []catid.ColumnID
+	var colNames []string
+	for _, col := range d.Columns {
+		colID := getColumnIDFromColumnName(b, tbl.TableID, col.Column)
+		if colSet.Contains(colID) {
+			panic(pgerror.Newf(pgcode.DuplicateColumn,
+				"column %q appears twice in unique constraint", col.Column))
+		}
+		colSet.Add(colID)
+		colIDs = append(colIDs, colID)
+		colNames = append(colNames, string(col.Column))
+	}
+
+	// 3. If a name is provided, check that this name is not used; Otherwise, generate
+	// a unique name for it.
+	if skip, err := validateConstraintNameIsNotUsed(b, tn, tbl, t); err != nil {
+		panic(err)
+	} else if skip {
+		return
+	}
+	if d.Name == "" {
+		d.Name = tree.Name(tabledesc.GenerateUniqueName(
+			fmt.Sprintf("unique_%s", strings.Join(colNames, "_")),
+			func(name string) bool {
+				return constraintNameInUse(b, tbl.TableID, name)
+			},
+		))
+	}
+
+	// 4. If there is a predicate, validate it.
+	if d.Predicate != nil {
+		predicate, _, _, err := schemaexpr.DequalifyAndValidateExprImpl(b, d.Predicate, types.Bool, "unique without index predicate", b.SemaCtx(), volatility.Immutable, tn,
+			func() colinfo.ResultColumns {
+				return getNonDropResultColumns(b, tbl.TableID)
+			},
+			func(columnName tree.Name) (exists bool, accessible bool, id catid.ColumnID, typ *types.T) {
+				return columnLookupFn(b, tbl.TableID, columnName)
+			},
+		)
+		if err != nil {
+			panic(err)
+		}
+		typedPredicate, err := parser.ParseExpr(predicate)
+		if err != nil {
+			panic(err)
+		}
+		d.Predicate = typedPredicate
+	}
+
+	// 5. (Finally!) Add a UniqueWithoutIndex, ConstraintName element to builder state.
+	constraintID := b.NextTableConstraintID(tbl.TableID)
+	uwi := &scpb.UniqueWithoutIndexConstraint{
+		TableID:              tbl.TableID,
+		ConstraintID:         constraintID,
+		ColumnIDs:            colIDs,
+		IndexIDForValidation: getIndexIDForValidationForConstraint(b, tbl.TableID),
+	}
+	if d.Predicate != nil {
+		uwi.Predicate = b.WrapExpression(tbl.TableID, d.Predicate)
+	}
+	b.Add(uwi)
+	b.Add(&scpb.ConstraintWithoutIndexName{
+		TableID:      tbl.TableID,
+		ConstraintID: constraintID,
+		Name:         string(d.Name),
+	})
+}
+
+// getFullyResolvedColNames returns fully resolved column names for `colNames`.
+// For each column name in `colNames`, its fully resolved name will be "db.sc.tbl.col".
+// The order of column names in the return is in syc with that in the input `colNames`.
 func getFullyResolvedColNames(
 	b BuildCtx, tableID catid.DescID, colNames tree.NameList,
 ) (ret tree.NameList) {
@@ -433,7 +536,8 @@ func validateConstraintNameIsNotUsed(
 		name = d.Name
 		ifNotExists = d.IfNotExists
 	case *tree.UniqueConstraintTableDef:
-		panic(scerrors.NotImplementedErrorf(t, "UNIQUE constraint %v not yet implemented", d.Name))
+		name = d.Name
+		ifNotExists = d.IfNotExists
 	default:
 		return false, errors.AssertionFailedf(
 			"unsupported constraint: %T", t.ConstraintDef)