Merge #149115 #150169 #150418 #150435

149115: opt: add tests for tpc-ds r=yuzefovich a=DrewKimball

#### workload/tpcds: remove remaining queries that can't be parsed

This commit removes some queries from `QueriesByNumber` that can't
be parsed because of the `rollup()` function.

Epic: None

Release note: None

#### opt: add tests for tpc-ds

This commit adds a set of optimizer tests for the TPC-DS benchmark.
Currently unsupported queries are left commented out. There is a version
with stats (collected with scale factor = 10) and one without stats.

Epic: None

Release note: None

150169: backup: fix race condition in starting compaction job r=msbutler a=kev-cao

In #145930, scheduled compactions are blocked from running if another compaction job is running for the schedule. However, it is currently possible for there to be a race condition which results in a compaction job being unable to find an incremental backup. Take the following circumstance:

1. Compaction job A starts.
2. A scheduled backup B completes and begins considering whether a compaction job should run. It fetches the current chain to its end time and finds that it should run a compaction.
3. Compaction job A completes.
4. B starts a transaction to create the compaction job. Because A has completed, it does not block the job from being created.
5. B creates a compaction job C that has a start time that is now skipped due to A's completion.
6. When C is picked up by the job system, it resolves the backup chain again, which now no longer has its start time and it fails.

This is resolved by opening a transaction before fetching the backup chain to check for an already running compaction job ID.

Fixes: #149867, #147264

Release note: None

150418: teamcity: add `s390x` unit tests build configuration r=rishabh7m a=rickystewart

Epic: CRDB-21133
Release note: None

150435: ccl/cdc: change when drop column is detected in schema feed r=fqazi a=fqazi

Previously, before automatic schema_locked toggle was added in the declarative schema changer, we had a guarantee that schema_locked would be unset before any column mutations are added. Unfortunately, the automatic toggle logic will disable schema_locked at the same time as the column is removed. This doesn't play nice with the existing schema feed logic. To address this, this patch detects the mutation once the new primary index is backfilled, so that is separate from the schema_locked being disabled. This is still valid behavior because the schema change is still in flight at this point, and doesn't have a usable new primary index.

Informs: #150003

Release note (bug fix): Addressed a bug on schema_locked tables when a column is dropped, and schema_locked is toggled for the user.

Co-authored-by: Drew Kimball <drewk@cockroachlabs.com>
Co-authored-by: Kevin Cao <kevin.cao@cockroachlabs.com>
Co-authored-by: Ricky Stewart <ricky@cockroachlabs.com>
Co-authored-by: Faizan Qazi <faizan@cockroachlabs.com>

Author: 5 people

build/teamcity/cockroach/ci/tests-ibm-cloud-linux-s390x/unit_tests.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+
+# Copyright 2022 The Cockroach Authors.
+#
+# Use of this software is governed by the CockroachDB Software License
+# included in the /LICENSE file.
+
+
+set -euo pipefail
+
+dir="$(dirname $(dirname $(dirname $(dirname $(dirname "${0}")))))"
+
+source "$dir/teamcity-support.sh"  # For $root
+source "$dir/teamcity-bazel-support.sh"  # For run_bazel
+
+tc_start_block "Run unit tests"
+BAZEL_SUPPORT_EXTRA_DOCKER_ARGS="-e TC_BUILD_BRANCH -e GITHUB_API_TOKEN -e BUILD_VCS_NUMBER -e TC_BUILD_ID -e TC_SERVER_URL -e TC_BUILDTYPE_ID -e GITHUB_REPO" run_bazel build/teamcity/cockroach/ci/tests-ibm-cloud-linux-s390x/unit_tests_impl.sh
+tc_end_block "Run unit tests"

build/teamcity/cockroach/ci/tests-ibm-cloud-linux-s390x/unit_tests_impl.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+
+# Copyright 2024 The Cockroach Authors.
+#
+# Use of this software is governed by the CockroachDB Software License
+# included in the /LICENSE file.
+
+
+set -euo pipefail
+
+dir="$(dirname $(dirname $(dirname $(dirname $(dirname "${0}")))))"
+
+source "$dir/teamcity-support.sh"  # for 'tc_release_branch'
+
+bazel build //pkg/cmd/bazci
+
+# Omit the ui_test as it depends on Javascript stuff; we don't have nodejs stuff
+# for s390x and it's expensive to pull in anyway.
+TESTS=$(bazel query 'kind(go_test, pkg/...) except attr("tags", "[\[ ]integration[,\]]", kind(go_test, pkg/...))' | grep -v ui_test)
+
+set -x
+
+$(bazel info bazel-bin)/pkg/cmd/bazci/bazci_/bazci -- test --config=ci --config=dev \
+    $TESTS \
+    --profile=/artifacts/profile.gz

pkg/backup/compaction_job.go

@@ -99,6 +99,28 @@ func maybeStartCompactionJob(
 		user,
 	)
 
+	// A race condition can occur where a compaction job ends after we fetch the
+	// backup chain but before we open a transaction to write the record. As a
+	// result, the written record is based on a chain that did not include the
+	// newly completed compaction job. In this scenario, it is possible that the
+	// chosen times for this compaction job actually no longer exist in the chain
+	// because it was compacted away. To avoid this, we need to check for the lock
+	// before fetching the backup chain.
+	//
+	// Note: _Technically_, this isn't entirely alleviated as a compaction job
+	// could start and finish in between the time we grab the backup chain and
+	// before we write the job record. However, this would require the schedule to
+	// have `on_previous_running=start` and realistically speaking, no compaction
+	// job would complete that quickly.
+	if err := execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
+		_, err := getScheduledExecutionArgsAndCheckCompactionLock(
+			ctx, txn, env, triggerJob.ScheduleID,
+		)
+		return err
+	}); err != nil {
+		return 0, err
+	}
+
 	chain, _, _, _, err := getBackupChain(
 		ctx, execCfg, user, triggerJob.Destination, triggerJob.EncryptionOptions,
 		triggerJob.EndTime, &kmsEnv,
@@ -115,23 +137,14 @@ func maybeStartCompactionJob(
 		return 0, err
 	}
 	startTS, endTS := chain[start].StartTime, chain[end-1].EndTime
-	log.Infof(ctx, "compacting backups from %s to %s", startTS, endTS)
 
 	var jobID jobspb.JobID
 	err = execCfg.InternalDB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
-		_, args, err := getScheduledBackupExecutionArgsFromSchedule(
-			ctx, env, jobs.ScheduledJobTxn(txn), triggerJob.ScheduleID,
+		args, err := getScheduledExecutionArgsAndCheckCompactionLock(
+			ctx, txn, env, triggerJob.ScheduleID,
 		)
 		if err != nil {
-			return errors.Wrapf(
-				err, "failed to get scheduled backup execution args for schedule %d", triggerJob.ScheduleID,
-			)
-		}
-		if args.CompactionJobID != 0 {
-			return errors.Newf(
-				"compaction job %d already running for schedule %d",
-				args.CompactionJobID, triggerJob.ScheduleID,
-			)
+			return err
 		}
 		datums, err := txn.QueryRowEx(
 			ctx,
@@ -171,6 +184,9 @@ func maybeStartCompactionJob(
 		)
 		return scheduledJob.Update(ctx, backupSchedule)
 	})
+	if err == nil {
+		log.Infof(ctx, "compacting backups from %s to %s", startTS, endTS)
+	}
 	return jobID, err
 }
 
@@ -931,6 +947,33 @@ func maybeWriteBackupCheckpoint(
 	return true, nil
 }
 
+// getScheduledExecutionArgsAndCheckCompactionLock retrieves the scheduled
+// backup execution args and also checks if a compaction jobs is already
+// running. If we fail to fetch the args or if a compaction job is already
+// running for this schedule, an error is returned.
+func getScheduledExecutionArgsAndCheckCompactionLock(
+	ctx context.Context,
+	txn isql.Txn,
+	env scheduledjobs.JobSchedulerEnv,
+	scheduleID jobspb.ScheduleID,
+) (*backuppb.ScheduledBackupExecutionArgs, error) {
+	_, args, err := getScheduledBackupExecutionArgsFromSchedule(
+		ctx, env, jobs.ScheduledJobTxn(txn), scheduleID,
+	)
+	if err != nil {
+		return nil, errors.Wrapf(
+			err, "failed to get scheduled backup execution args for schedule %d", scheduleID,
+		)
+	}
+	if args.CompactionJobID != 0 {
+		return nil, errors.Newf(
+			"compaction job %d already running for schedule %d",
+			args.CompactionJobID, scheduleID,
+		)
+	}
+	return args, nil
+}
+
 func init() {
 	builtins.StartCompactionJob = StartCompactionJob
 }

pkg/ccl/changefeedccl/changefeed_test.go

@@ -2180,6 +2180,54 @@ func TestChangefeedColumnDropsOnTheSameTableWithMultipleFamilies(t *testing.T) {
 	})
 }
 
+func TestChangefeedColumnDropsOnTheSameTableWithMultipleFamiliesWithManualSchemaLocked(
+	t *testing.T,
+) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	require.NoError(t, log.SetVModule("kv_feed=2,changefeed_processors=2"))
+
+	testFn := func(t *testing.T, s TestServer, f cdctest.TestFeedFactory) {
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+
+		sqlDB.Exec(t, `CREATE TABLE hasfams (id int primary key, a string, b string, c string, FAMILY id_a (id, a), FAMILY b_and_c (b, c))`)
+		sqlDB.Exec(t, `INSERT INTO hasfams values (0, 'a', 'b', 'c')`)
+
+		var args []any
+		if _, ok := f.(*webhookFeedFactory); ok {
+			args = append(args, optOutOfMetamorphicEnrichedEnvelope{reason: "metamorphic enriched envelope does not support column families for webhook sinks"})
+		}
+		// Open up the changefeed.
+		cf := feed(t, f, `CREATE CHANGEFEED FOR TABLE hasfams FAMILY id_a, TABLE hasfams FAMILY b_and_c`, args...)
+		defer closeFeed(t, cf)
+		assertPayloads(t, cf, []string{
+			`hasfams.b_and_c: [0]->{"after": {"b": "b", "c": "c"}}`,
+			`hasfams.id_a: [0]->{"after": {"a": "a", "id": 0}}`,
+		})
+
+		// Check that dropping a watched column will backfill the changefeed.
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=false)`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams DROP COLUMN a`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=true)`)
+		assertPayloads(t, cf, []string{
+			`hasfams.id_a: [0]->{"after": {"id": 0}}`,
+		})
+
+		// Check that dropping a watched column will backfill the changefeed.
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=false)`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams DROP COLUMN b`)
+		sqlDB.Exec(t, `ALTER TABLE hasfams SET (schema_locked=true)`)
+		assertPayloads(t, cf, []string{
+			`hasfams.b_and_c: [0]->{"after": {"c": "c"}}`,
+		})
+	}
+
+	runWithAndWithoutRegression141453(t, testFn, func(t *testing.T, testFn cdcTestFn) {
+		cdcTest(t, testFn)
+	})
+}
+
 func TestNoStopAfterNonTargetAddColumnWithBackfill(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
@@ -3203,7 +3251,7 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 				`drop_column: [2]->{"after": {"a": 2, "b": "2"}}`,
 			})
 			sqlDB.Exec(t, `ALTER TABLE drop_column DROP COLUMN b`)
-			ts := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `drop_column`, 2)
+			ts := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `drop_column`, 6)
 
 			// Backfill for DROP COLUMN b.
 			assertPayloads(t, dropColumn, []string{
@@ -3256,7 +3304,7 @@ func TestChangefeedSchemaChangeAllowBackfill(t *testing.T) {
 			// the 7th step (version 15). Finally, when adding column d, it goes from 17->25 ith the schema change
 			// being visible at the 7th step (version 23).
 			// TODO(#142936): Investigate if this descriptor version hardcoding is sound.
-			dropTS := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 2)
+			dropTS := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 6)
 			addTS := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 15)
 			addTS2 := schematestutils.FetchDescVersionModificationTime(t, s.Server, `d`, `public`, `multiple_alters`, 23)
 

pkg/ccl/changefeedccl/schemafeed/table_event_filter.go

@@ -188,6 +188,26 @@ func shouldFilterAddColumnEvent(e TableEvent, targets changefeedbase.Targets) (b
 	return !watched, nil
 }
 
+// notDeclarativeOrHasMergedIndex returns true if the descriptor has a declarative
+// schema changer with a merged index.
+func notDeclarativeOrHasMergedIndex(desc catalog.TableDescriptor) bool {
+	// If there are not declarative schema changes then this will always be
+	// true.
+	if desc.GetDeclarativeSchemaChangerState() == nil {
+		return true
+	}
+	// For declarative schema changes detect when a new primary index becomes
+	// WRITE_ONLY (i.e. backfill has been completed).
+	for idx, target := range desc.GetDeclarativeSchemaChangerState().Targets {
+		if target.GetPrimaryIndex() != nil &&
+			target.TargetStatus == scpb.Status_PUBLIC &&
+			desc.GetDeclarativeSchemaChangerState().CurrentStatuses[idx] == scpb.Status_WRITE_ONLY {
+			return true
+		}
+	}
+	return false
+}
+
 // Returns true if the changefeed targets a column which has a drop mutation inside the table event.
 func droppedColumnIsWatched(e TableEvent, targets changefeedbase.Targets) (bool, error) {
 	// If no column families are specified, then all columns are targeted.
@@ -212,7 +232,13 @@ func droppedColumnIsWatched(e TableEvent, targets changefeedbase.Targets) (bool,
 		if m.AsColumn() == nil || m.AsColumn().IsHidden() {
 			continue
 		}
-		if m.Dropped() && m.WriteAndDeleteOnly() && watchedColumnIDs.Contains(int(m.AsColumn().GetID())) {
+		// For dropped columns wait for WriteAndDeleteOnly to be hit. When using
+		// the declarative schema changer we will wait a bit later in the plan to
+		// publish the dropped column, since schema_locked and the column being
+		// write and delete only happen at the same stage. Since the schema change
+		// is still in progress, there is a gray area in terms of when the change
+		// should be visible.
+		if m.Dropped() && m.WriteAndDeleteOnly() && notDeclarativeOrHasMergedIndex(e.After) && watchedColumnIDs.Contains(int(m.AsColumn().GetID())) {
 			return true, nil
 		}
 	}
@@ -273,7 +299,13 @@ func dropVisibleColumnMutationExists(desc catalog.TableDescriptor) bool {
 		if m.AsColumn() == nil || m.AsColumn().IsHidden() {
 			continue
 		}
-		if m.Dropped() && m.WriteAndDeleteOnly() {
+		// For dropped columns wait for WriteAndDeleteOnly to be hit. When using
+		// the declarative schema changer we will wait a bit later in the plan to
+		// publish the dropped column, since schema_locked and the column being
+		// write and delete only happen at the same stage. Since the schema change
+		// is still in progress, there is a gray area in terms of when the change
+		// should be visible.
+		if m.Dropped() && m.WriteAndDeleteOnly() && notDeclarativeOrHasMergedIndex(desc) {
 			return true
 		}
 	}

pkg/ccl/changefeedccl/schemafeed/testdata/drop_column

@@ -12,11 +12,11 @@ ALTER TABLE t DROP COLUMN j;
 
 pop f=1
 ----
-t 1->2: DropColumn
+t 1->2: Unknown
 t 2->3: Unknown
 t 3->4: Unknown
 t 4->5: Unknown
-t 5->6: Unknown
+t 5->6: DropColumn
 t 6->7: PrimaryKeyChange (no column changes)
 t 7->8: Unknown
 t 8->9: AddHiddenColumn