Merge #82489 #82768

82489: sql: disallow adding column as primary key r=jasonmchan a=jasonmchan

Previously, the behavior of ALTER TABLE ... ADD COLUMN ... PRIMARY KEY
was undefined. With the legacy schema changer, this statement would
appear to succeed, but it would break the new schema changer. This
commit changes this by returning an explicit error.

In the future, we should support this statement if the table's primary
key was originally the default rowid primary key (see #82735).

Fixes #82489

Release note: None

82768: changefeedccl: Do not inhibit node shutdown r=miretskiy a=miretskiy

See individual commits for details:
  * Ensure running changefeed do not inhibit node shutdown (Informs #82765)
  * Treat node unavailable error as retryable.

Test not being merged as part of this PR -- see #82767

Release Notes (bug fix): Ensure running changefeeds do
not inhibit node shutdown.
Release Notes (bug fix): Treat node unavailable error as a retryable changefeed error.

Co-authored-by: Jason Chan <[email protected]>
Co-authored-by: Yevgeniy Miretskiy <[email protected]>

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -208,17 +208,14 @@ func (ca *changeAggregator) MustBeStreaming() bool {
 
 // Start is part of the RowSource interface.
 func (ca *changeAggregator) Start(ctx context.Context) {
+	// Derive a separate context so that we can shutdown the poller.
+	ctx, ca.cancel = ca.flowCtx.Stopper().WithCancelOnQuiesce(ctx)
+
 	if ca.spec.JobID != 0 {
 		ctx = logtags.AddTag(ctx, "job", ca.spec.JobID)
 	}
 	ctx = ca.StartInternal(ctx, changeAggregatorProcName)
 
-	// Derive a separate context so that we can shutdown the poller. Note that
-	// we need to update both ctx (used throughout this function) and
-	// ProcessorBase.Ctx (used in all other methods) to the new context.
-	ctx, ca.cancel = context.WithCancel(ctx)
-	ca.Ctx = ctx
-
 	spans, err := ca.setupSpansAndFrontier()
 
 	endTime := ca.spec.Feed.EndTime

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -1030,7 +1030,12 @@ func (b *changefeedResumer) resumeWithRetries(
 			}
 		}
 
-		if !changefeedbase.IsRetryableError(err) {
+		// Retry changefeed is error is retryable.  In addition, we want to handle
+		// context cancellation as retryable, but only if the resumer context has not been cancelled.
+		// (resumer context is canceled by the jobs framework -- so we should respect it).
+		isRetryableErr := changefeedbase.IsRetryableError(err) ||
+			(ctx.Err() == nil && errors.Is(err, context.Canceled))
+		if !isRetryableErr {
 			if ctx.Err() != nil {
 				return ctx.Err()
 			}

pkg/ccl/changefeedccl/changefeedbase/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/clusterversion",
         "//pkg/jobs/joberror",
         "//pkg/jobs/jobspb",
+        "//pkg/roachpb",
         "//pkg/settings",
         "//pkg/sql",
         "//pkg/sql/catalog",

pkg/ccl/changefeedccl/changefeedbase/errors.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 
 	"github.com/cockroachdb/cockroach/pkg/jobs/joberror"
+	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/flowinfra"
 	"github.com/cockroachdb/errors"
 )
@@ -117,7 +118,9 @@ func IsRetryableError(err error) bool {
 		return true
 	}
 
-	return (joberror.IsDistSQLRetryableError(err) || flowinfra.IsNoInboundStreamConnectionError(err))
+	return (joberror.IsDistSQLRetryableError(err) ||
+		flowinfra.IsNoInboundStreamConnectionError(err) ||
+		errors.HasType(err, (*roachpb.NodeUnavailableError)(nil)))
 }
 
 // MaybeStripRetryableErrorMarker performs some minimal attempt to clean the

pkg/sql/alter_table.go

@@ -188,6 +188,10 @@ func (n *alterTableNode) startExec(params runParams) error {
 						"UNIQUE WITHOUT INDEX constraint on the column",
 				)
 			}
+			if t.ColumnDef.PrimaryKey.IsPrimaryKey {
+				return pgerror.Newf(pgcode.InvalidColumnDefinition,
+					"multiple primary keys for table %q are not allowed", tn.Object())
+			}
 			var err error
 			params.p.runWithOptions(resolveFlags{contextDatabaseID: n.tableDesc.ParentID}, func() {
 				err = params.p.addColumnImpl(params, n, tn, n.tableDesc, t)

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -2593,3 +2593,14 @@ CREATE INDEX idx_j ON t_add_column_not_null (j);
 
 statement ok
 DROP TABLE t_add_column_not_null
+
+subtest regression_81448
+
+statement ok
+CREATE TABLE t81448 (a INT PRIMARY KEY)
+
+statement error pq: multiple primary keys for table "t81448" are not allowed
+ALTER TABLE t81448 ADD COLUMN b INT PRIMARY KEY
+
+statement ok
+DROP TABLE t81448

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

@@ -78,6 +78,19 @@ func alterTableAddColumn(
 				"UNIQUE WITHOUT INDEX constraint on the column",
 		))
 	}
+	if d.PrimaryKey.IsPrimaryKey {
+		publicTargets := b.QueryByID(tbl.TableID).Filter(
+			func(_ scpb.Status, target scpb.TargetStatus, _ scpb.Element) bool {
+				return target == scpb.ToPublic
+			},
+		)
+		_, _, primaryIdx := scpb.FindPrimaryIndex(publicTargets)
+		// TODO(#82735): support when primary key is implicit
+		if primaryIdx != nil {
+			panic(pgerror.Newf(pgcode.InvalidColumnDefinition,
+				"multiple primary keys for table %q are not allowed", tn.Object()))
+		}
+	}
 	if d.IsComputed() {
 		d.Computed.Expr = schemaexpr.MaybeRewriteComputedColumn(d.Computed.Expr, b.SessionData())
 	}