Merge #75737 #76277 #76354 #76384

75737: cdc: support ALTER CHANGEFEED statement for adding/dropping targets r=sherman-grewal a=sherman-grewal

This PR introduces a new SQL statement called ALTER
CHANGEFEED, which allows users to add/drop targets
for an existing changefeed. Note that the changefeed
job must be paused in order for the alterations to
be applied to the changefeed. The syntax of the
statement is as follows:

ALTER CHANGEFEED <job_id> {{ADD|DROP} \<targets\>}...

Where there can be an arbitrary number of ADD/
DROP commands in any order. Once a user
executes this statement, the targets will be
added/dropped, and the statement will return the
job id and the new job description of the changefeed
job. It is also important to note that a user cannot
drop all targets in a changefeed.

Example:
ALTER CHANGEFEED 123 ADD foo,bar DROP baz;

Release note (enterprise change): Added support
for a new SQL statement called ALTER CHANGEFEED,
which allows users to add/drop targets for an
existing changefeed. The syntax of the statement
is as follows:

ALTER CHANGEFEED <job_id> {{ADD|DROP} \<targets\>}...

Where there can be an abritrary number of ADD/DROP
commands in any order. The following is an example
of this statement:

ALTER CHANGEFEED 123 ADD foo,bar DROP baz;

With this statement, users can avoid going through
the process of altering a changefeed on their own,
and rely on CRDB to carry out this task.

76277: sql: clean up comments when swapping primary keys r=fqazi a=fqazi

Fixes: #71984

Previously, when swapping out primary keys comments
on old indexes would be leftover. This was inadequate because
entries would be leftover in the system.comments table
that could prevent SHOW CREATE  to fail when it attempts
to map comments for dropped indexes. To address this, this patch
will clean up comments on primary key swaps.

Release note (bug fix): Comments are not cleaned up when the table
primary keys are swapped, which can cause SHOW TABLE to fail.

76354: tree: extract new packages to break dependency of execgen on tree r=yuzefovich a=yuzefovich

**execgen: move BinaryOverloadHelper into colexecbase**

This is so that it's easier to break the dependency on `tree` from
`execgen`. Additionally, this commit makes some of the comments
template-time only.

Release note: None

**treecmp: extract ComparisonOperator into a separate package**

This commit extracts things that `execgen` depends on (like some
constants for the comparison operators as well as some other stuff) into
a new package `treecmp`. The goal is to break the dependency of
`execgen` on `tree`.

It was mostly a simple move of some code from `tree/expr.go` into
`treecmp/comparison_operator.go`. I had to export a few things. Also,
I had to make `ComparisonOperatorSymbol.Inverse` take the receiver as an
argument and left the function in `tree` (in order to not mess with
`FoldComparisonExpr`).

Release note: None

**treebin: extract BinaryOperator into a separate package**

This commit extracts things that `execgen` depends on (like some
constants for the binary operators as well as some other stuff) into
a new package `treebin`. The goal is to break the dependency of
`execgen` on `tree`.

It was a simple move of some code from `tree/expr.go` into
`treebin/binary_operator.go`. I only had to export `IsPadded` function.

Release note: None

**treewindow: extract some window constants into a new package out of tree**

This commit extracts some constants about the window functions into
a new package which allows us to finally break the dependency of
`execgen` on `tree`. Now the absence of such dependency is enforced with
`VerifyNoImports` in `colexec` package.

The only notable change that I had to make is making
`WindowFrameExclusion` not implement `tree.NodeFormatter` interface;
instead, the callsites that wanted to call `Format` on that type now
call `String` which provides the same behavior.

Release note: None

76384: spanconfig: mark reconciliation job as idle r=irfansharif a=irfansharif

Fixes #70538.

We have a forever running background AUTO SPAN CONFIG RECONCILIATION job
on tenant pods. To know when it's safe to wind down pods, we use the
number of currently running jobs as an indicator. Given the job is
forever running, we need an indicator to suggest that despite the job's
presence, it's safe to wind down.

In #74747 we added a thin API to the jobs subsystem to do just that,
with the intent of using it for idle changefeed jobs. We just cargo-cult
that same approach here to mark the reconciliation job as always idle.

Release note: None

Co-authored-by: Sherman Grewal <[email protected]>
Co-authored-by: Faizan Qazi <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: irfan sharif <[email protected]>

docs/generated/sql/bnf/BUILD.bazel

@@ -4,6 +4,7 @@ FILES = [
     "abort_stmt",
     "add_column",
     "add_constraint",
+    "alter_changefeed_stmt",
     "alter_column",
     "alter_database_add_region_stmt",
     "alter_database_drop_region",

docs/generated/sql/bnf/alter_changefeed_stmt.bnf

@@ -0,0 +1,2 @@
+alter_changefeed_stmt ::=
+	'ALTER' 'CHANGEFEED' a_expr alter_changefeed_cmds

docs/generated/sql/bnf/alter_ddl_stmt.bnf

@@ -9,3 +9,4 @@ alter_ddl_stmt ::=
 	| alter_schema_stmt
 	| alter_type_stmt
 	| alter_default_privileges_stmt
+	| alter_changefeed_stmt

docs/generated/sql/bnf/stmt_block.bnf

@@ -399,6 +399,7 @@ alter_ddl_stmt ::=
 	| alter_schema_stmt
 	| alter_type_stmt
 	| alter_default_privileges_stmt
+	| alter_changefeed_stmt
 
 alter_role_stmt ::=
 	'ALTER' role_or_group_or_user role_spec opt_role_options
@@ -1385,6 +1386,9 @@ alter_default_privileges_stmt ::=
 	| 'ALTER' 'DEFAULT' 'PRIVILEGES' 'FOR' 'ALL' 'ROLES' opt_in_schemas abbreviated_grant_stmt
 	| 'ALTER' 'DEFAULT' 'PRIVILEGES' 'FOR' 'ALL' 'ROLES' opt_in_schemas abbreviated_revoke_stmt
 
+alter_changefeed_stmt ::=
+	'ALTER' 'CHANGEFEED' a_expr alter_changefeed_cmds
+
 role_or_group_or_user ::=
 	'ROLE'
 	| 'USER'
@@ -1879,6 +1883,9 @@ abbreviated_revoke_stmt ::=
 	'REVOKE' privileges 'ON' alter_default_privileges_target_object 'FROM' role_spec_list opt_drop_behavior
 	| 'REVOKE' 'GRANT' 'OPTION' 'FOR' privileges 'ON' alter_default_privileges_target_object 'FROM' role_spec_list opt_drop_behavior
 
+alter_changefeed_cmds ::=
+	( alter_changefeed_cmd ) ( ( alter_changefeed_cmd ) )*
+
 role_options ::=
 	( role_option ) ( ( role_option ) )*
 
@@ -2358,6 +2365,10 @@ alter_default_privileges_target_object ::=
 	| 'TYPES'
 	| 'SCHEMAS'
 
+alter_changefeed_cmd ::=
+	'ADD' changefeed_targets
+	| 'DROP' changefeed_targets
+
 role_option ::=
 	'CREATEROLE'
 	| 'NOCREATEROLE'

pkg/ccl/changefeedccl/BUILD.bazel

@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "changefeedccl",
     srcs = [
+        "alter_changefeed_stmt.go",
         "avro.go",
         "changefeed.go",
         "changefeed_dist.go",
@@ -65,6 +66,7 @@ go_library(
         "//pkg/sql/execinfra",
         "//pkg/sql/execinfrapb",
         "//pkg/sql/flowinfra",
+        "//pkg/sql/parser",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/pgwire/pgnotice",
@@ -122,6 +124,7 @@ go_test(
     name = "changefeedccl_test",
     size = "enormous",
     srcs = [
+        "alter_changefeed_test.go",
         "avro_test.go",
         "bench_test.go",
         "changefeed_test.go",

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -0,0 +1,202 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Licensed as a CockroachDB Enterprise file under the Cockroach Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+//     https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt
+
+package changefeedccl
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/parser"
+	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/types"
+	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/errors"
+)
+
+func init() {
+	sql.AddPlanHook("alter changefeed", alterChangefeedPlanHook)
+}
+
+type alterChangefeedOpts struct {
+	AddTargets  []tree.TargetList
+	DropTargets []tree.TargetList
+}
+
+// alterChangefeedPlanHook implements sql.PlanHookFn.
+func alterChangefeedPlanHook(
+	ctx context.Context, stmt tree.Statement, p sql.PlanHookState,
+) (sql.PlanHookRowFn, colinfo.ResultColumns, []sql.PlanNode, bool, error) {
+	alterChangefeedStmt, ok := stmt.(*tree.AlterChangefeed)
+	if !ok {
+		return nil, nil, nil, false, nil
+	}
+
+	header := colinfo.ResultColumns{
+		{Name: "job_id", Typ: types.Int},
+		{Name: "job_description", Typ: types.String},
+	}
+	lockForUpdate := false
+
+	fn := func(ctx context.Context, _ []sql.PlanNode, resultsCh chan<- tree.Datums) error {
+		if err := validateSettings(ctx, p); err != nil {
+			return err
+		}
+
+		typedExpr, err := alterChangefeedStmt.Jobs.TypeCheck(ctx, p.SemaCtx(), types.Int)
+		if err != nil {
+			return err
+		}
+		jobID := jobspb.JobID(tree.MustBeDInt(typedExpr))
+
+		job, err := p.ExecCfg().JobRegistry.LoadJobWithTxn(ctx, jobID, p.ExtendedEvalContext().Txn)
+		if err != nil {
+			err = errors.Wrapf(err, `could not load job with job id %d`, jobID)
+			return err
+		}
+
+		details, ok := job.Details().(jobspb.ChangefeedDetails)
+		if !ok {
+			return errors.Errorf(`job %d is not changefeed job`, jobID)
+		}
+
+		if job.Status() != jobs.StatusPaused {
+			return errors.Errorf(`job %d is not paused`, jobID)
+		}
+
+		var opts alterChangefeedOpts
+		for _, cmd := range alterChangefeedStmt.Cmds {
+			switch v := cmd.(type) {
+			case *tree.AlterChangefeedAddTarget:
+				opts.AddTargets = append(opts.AddTargets, v.Targets)
+			case *tree.AlterChangefeedDropTarget:
+				opts.DropTargets = append(opts.DropTargets, v.Targets)
+			}
+		}
+
+		var initialHighWater hlc.Timestamp
+		statementTime := hlc.Timestamp{
+			WallTime: p.ExtendedEvalContext().GetStmtTimestamp().UnixNano(),
+		}
+
+		if opts.AddTargets != nil {
+			var targetDescs []catalog.Descriptor
+
+			for _, targetList := range opts.AddTargets {
+				descs, err := getTableDescriptors(ctx, p, &targetList, statementTime, initialHighWater)
+				if err != nil {
+					return err
+				}
+				targetDescs = append(targetDescs, descs...)
+			}
+
+			newTargets, err := getTargets(ctx, p, targetDescs, details.Opts)
+			if err != nil {
+				return err
+			}
+			// add old targets
+			for id, target := range details.Targets {
+				newTargets[id] = target
+			}
+			details.Targets = newTargets
+		}
+
+		if opts.DropTargets != nil {
+			var targetDescs []catalog.Descriptor
+
+			for _, targetList := range opts.DropTargets {
+				descs, err := getTableDescriptors(ctx, p, &targetList, statementTime, initialHighWater)
+				if err != nil {
+					return err
+				}
+				targetDescs = append(targetDescs, descs...)
+			}
+
+			for _, desc := range targetDescs {
+				if table, isTable := desc.(catalog.TableDescriptor); isTable {
+					if err := p.CheckPrivilege(ctx, desc, privilege.SELECT); err != nil {
+						return err
+					}
+					delete(details.Targets, table.GetID())
+				}
+			}
+		}
+
+		if len(details.Targets) == 0 {
+			return errors.Errorf("cannot drop all targets for changefeed job %d", jobID)
+		}
+
+		if err := validateSink(ctx, p, jobID, details, details.Opts); err != nil {
+			return err
+		}
+
+		oldStmt, err := parser.ParseOne(job.Payload().Description)
+		if err != nil {
+			return err
+		}
+		oldChangefeedStmt, ok := oldStmt.AST.(*tree.CreateChangefeed)
+		if !ok {
+			return errors.Errorf(`could not parse create changefeed statement for job %d`, jobID)
+		}
+
+		var targets tree.TargetList
+		for _, target := range details.Targets {
+			targetName := tree.MakeTableNameFromPrefix(tree.ObjectNamePrefix{}, tree.Name(target.StatementTimeName))
+			targets.Tables = append(targets.Tables, &targetName)
+		}
+
+		oldChangefeedStmt.Targets = targets
+		jobDescription := tree.AsString(oldChangefeedStmt)
+
+		newPayload := job.Payload()
+		newPayload.Description = jobDescription
+		newPayload.Details = jobspb.WrapPayloadDetails(details)
+
+		finalDescs, err := getTableDescriptors(ctx, p, &targets, statementTime, initialHighWater)
+		if err != nil {
+			return err
+		}
+
+		newPayload.DescriptorIDs = func() (sqlDescIDs []descpb.ID) {
+			for _, desc := range finalDescs {
+				sqlDescIDs = append(sqlDescIDs, desc.GetID())
+			}
+			return sqlDescIDs
+		}()
+
+		err = p.ExecCfg().JobRegistry.UpdateJobWithTxn(ctx, jobID, p.ExtendedEvalContext().Txn, lockForUpdate, func(
+			txn *kv.Txn, md jobs.JobMetadata, ju *jobs.JobUpdater,
+		) error {
+			ju.UpdatePayload(&newPayload)
+			return nil
+		})
+
+		if err != nil {
+			return err
+		}
+
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case resultsCh <- tree.Datums{
+			tree.NewDInt(tree.DInt(jobID)),
+			tree.NewDString(jobDescription),
+		}:
+			return nil
+		}
+	}
+
+	return fn, header, nil, false, nil
+}