Merge #149094 #149260 #149300 #149341

149094: changefeedccl: add test-only assertion for checkpoint fields invariant r=DarrylWong,asg0451 a=andyyang890

Fixes #149100

---

**roachtest: surface changefeed failures in mixed-version CDC tests** 

Previously, if the changefeed failed in a mixed-version CDC test,
the test wouldn't fail until it hit the timeout. We now periodically
check for changefeed failures and fail the test if one is found.

Release note: None

---

**mixedversion: add RunE method to Test** 

This patch adds a `RunE` method to `Test`. Unlike the existing `Run` function,
it will not immediately fatal the test if an error is encountered and will
instead return the error along with the test plan so that callers may decide
how to handle the error.

Release note: None

---

**changefeedccl: add test-only assertion for checkpoint fields invariant** 

This patch adds a test-only assertion that we won't ever see both
checkpoint fields set on a changefeed job progress struct. It is
test-only because a bug that existed on earlier versions of 25.2
could cause both checkpoint fields to be set and so the production
code continues to discard the legacy checkpoint when that happens.

Release note: None

149260: sql: fix check external connection evaluation r=jeffswenson a=jeffswenson

Previously, `CHECK EXTERNAL CONNECTION '' WITH CONCURRENCY = (SELECT 1)` would panic because its not able to evaluate an expression containing a sub query.

Now, passing a sub query will fail with a user error.

```
CHECK EXTERNAL CONNECTION NULLIF WITH CONCURRENTLY = EXISTS ( ( TABLE error ) );
ERROR: subqueries are not allowed in check_external_connection
SQLSTATE: 0A000
```

Release note: none
Informs: #147876
Informs: #147877

149300: logictest: fix up recent change r=yuzefovich a=yuzefovich

Just merged 55108a1 made one query flaky: namely, we removed ORDER BY from `array_agg` so now the order of two items can be arbitrary. This commit changes the test to assert that the query succeeds (which this test is really about) without hitting an error.

Epic: None
Release note: None

149341: upgrades: fix order of AddHotRangeLoggerJob migration r=rafiss a=rafiss

The V25_3_AddHotRangeLoggerJob cluster version comes _after_ the V25_3_AddEstimatedLastLoginTime version. Yet, the corresponding upgrade for adding the hot ranges logger was incorrectly being done _before_ the last login time column.

informs: #148981
informs: #148998

Release note: None

Co-authored-by: Andy Yang <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

Author: 5 people

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -277,11 +277,15 @@ func startDistChangefeed(
 			return errors.AssertionFailedf("both legacy and current checkpoint set on " +
 				"changefeed job progress and legacy checkpoint has later timestamp")
 		}
-		// This should be an assertion failure but unfortunately due to a bug
+		// This should always be an assertion failure but unfortunately due to a bug
 		// that was included in earlier versions of 25.2 (#148620), we may fail
 		// to clear the legacy checkpoint when we start writing the new one.
 		// We instead discard the legacy checkpoint here and it will eventually be
 		// cleared once the cluster is running a newer patch release with the fix.
+		if buildutil.CrdbTestBuild {
+			return errors.AssertionFailedf("both legacy and current checkpoint set on " +
+				"changefeed job progress")
+		}
 		log.Warningf(ctx, "both legacy and current checkpoint set on changefeed job progress; "+
 			"discarding legacy checkpoint")
 		legacyCheckpoint = nil

pkg/ccl/changefeedccl/changefeed_test.go

@@ -11874,13 +11874,9 @@ func TestChangefeedResumeWithBothLegacyAndCurrentCheckpoint(t *testing.T) {
 		require.NoError(t, err)
 
 		sqlDB.Exec(t, `RESUME JOB $1`, jobFeed.JobID())
-		waitForJobState(sqlDB, t, jobFeed.JobID(), jobs.StateRunning)
-
-		// Wait for highwater to advance past the current time.
-		var tsStr string
-		sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&tsStr)
-		ts := parseTimeToHLC(t, tsStr)
-		require.NoError(t, jobFeed.WaitForHighWaterMark(ts))
+		waitForJobState(sqlDB, t, jobFeed.JobID(), jobs.StateFailed)
+		require.ErrorContains(t, jobFeed.FetchTerminalJobErr(),
+			"both legacy and current checkpoint set on changefeed job progress")
 	}
 
 	cdcTest(t, testFn, feedTestEnterpriseSinks)

pkg/ccl/logictestccl/testdata/logic_test/refcursor

@@ -693,10 +693,8 @@ SELECT lag(y) OVER w, lead(y) OVER w, first_value(y) OVER w, last_value(y) OVER
 
 # Array functions are allowed (but ordering on REFCURSOR and REFCURSOR[] columns
 # is not).
-query TT
+statement ok
 SELECT array_agg(x), array_cat_agg(y) FROM t;
-----
-{foo,baz}  {bar}
 
 query TT rowsort
 SELECT array_append(y, x), array_prepend(x, y) FROM t;

pkg/cmd/roachtest/roachtestutil/mixedversion/mixedversion.go

@@ -834,21 +834,29 @@ func (t *Test) Workload(
 	return t.BackgroundCommand(fmt.Sprintf("%s workload", name), node, runCmd)
 }
 
-// Run runs the mixed-version test. It should be called once all
+// Run is like RunE, except it fatals the test if any error occurs.
+func (t *Test) Run() {
+	_, err := t.RunE()
+	if err != nil {
+		t.rt.Fatal(err)
+	}
+}
+
+// RunE runs the mixed-version test. It should be called once all
 // startup, mixed-version, and after-test hooks have been declared. A
 // test plan will be generated (and logged), and the test will be
-// carried out.
-func (t *Test) Run() {
+// carried out. A non-nil plan will be returned unless planning fails.
+func (t *Test) RunE() (*TestPlan, error) {
 	plan, err := t.plan()
 	if err != nil {
-		t.rt.Fatal(err)
+		return nil, err
 	}
 
 	t.logger.Printf("mixed-version test:\n%s", plan.PrettyPrint())
 
 	if override := os.Getenv(dryRunEnv); override != "" {
 		t.logger.Printf("skipping test run in dry-run mode")
-		return
+		return plan, nil
 	}
 
 	// Mark the deployment mode and versions, so they show up in the github issue. This makes
@@ -857,8 +865,10 @@ func (t *Test) Run() {
 	t.rt.AddParam("mvtVersions", formatVersions(plan.Versions()))
 
 	if err := t.run(plan); err != nil {
-		t.rt.Fatal(err)
+		return plan, err
 	}
+
+	return plan, nil
 }
 
 func (t *Test) run(plan *TestPlan) error {

pkg/cmd/roachtest/tests/mixed_version_cdc.go

@@ -140,6 +140,8 @@ type cdcMixedVersionTester struct {
 
 	validator *cdctest.CountValidator
 	fprintV   *cdctest.FingerprintValidator
+
+	jobID int
 }
 
 func newCDCMixedVersionTester(ctx context.Context, c cluster.Cluster) cdcMixedVersionTester {
@@ -186,6 +188,25 @@ func (cmvt *cdcMixedVersionTester) StartKafka(t test.Test, c cluster.Cluster) (c
 func (cmvt *cdcMixedVersionTester) waitAndValidate(
 	ctx context.Context, l *logger.Logger, r *rand.Rand, h *mixedversion.Helper,
 ) error {
+	cancel := h.GoWithCancel(func(ctx context.Context, l *logger.Logger) error {
+		for {
+			select {
+			case <-time.After(5 * time.Second):
+				_, db := h.RandomDB(r)
+				info, err := getChangefeedInfo(db, cmvt.jobID)
+				if err != nil {
+					return err
+				}
+				if info.GetStatus() == "failed" {
+					return errors.Newf("changefeed failed: %s", info.GetError())
+				}
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+	})
+	defer cancel()
+
 	l.Printf("waiting for %d resolved timestamps", resolvedTimestampsPerState)
 	// create a new channel for the resolved timestamps, allowing any
 	// new resolved timestamps to be captured and account for in the
@@ -403,6 +424,7 @@ func (cmvt *cdcMixedVersionTester) createChangeFeed(
 		return err
 	}
 	l.Printf("created changefeed job %d", jobID)
+	cmvt.jobID = jobID
 	return nil
 }
 
@@ -650,5 +672,40 @@ func runCDCMixedVersionCheckpointing(ctx context.Context, t test.Test, c cluster
 	mvt.InMixedVersion("wait and validate", tester.waitAndValidate)
 	mvt.AfterUpgradeFinalized("wait and validate", tester.waitAndValidate)
 
-	mvt.Run()
+	// Run the test.
+	plan, err := mvt.RunE()
+	if err != nil {
+		isAffectedBy148620 := func(plan *mixedversion.TestPlan) bool {
+			if plan == nil {
+				return false
+			}
+			for _, v := range []string{"v25.2.0", "v25.2.1", "v25.2.2"} {
+				version := clusterupgrade.MustParseVersion(v)
+				for _, planVersion := range plan.Versions() {
+					if planVersion.Equal(version) {
+						return true
+					}
+				}
+			}
+			return false
+		}
+
+		isExpectedErrorDueTo148620 := func(err error) bool {
+			for _, s := range []string{
+				"both legacy and current checkpoint set on change aggregator spec",
+				"both legacy and current checkpoint set on changefeed job progress",
+			} {
+				if strings.Contains(err.Error(), s) {
+					return true
+				}
+			}
+			return false
+		}
+
+		if isAffectedBy148620(plan) && isExpectedErrorDueTo148620(err) {
+			t.Skipf("expected error due to #148620: %s", err)
+		}
+
+		t.Fatal(err)
+	}
 }

pkg/sql/check_external_connection.go

@@ -150,6 +150,7 @@ func (n *checkExternalConnectionNode) Close(_ context.Context) {
 }
 
 func (n *checkExternalConnectionNode) parseParams(params runParams) error {
+	params.p.SemaCtx().Properties.Require("check_external_connection", tree.RejectSubqueries)
 	exprEval := params.p.ExprEvaluator("CHECK EXTERNAL CONNECTION")
 	loc, err := exprEval.String(params.ctx, n.node.URI)
 	if err != nil {

pkg/sql/logictest/testdata/logic_test/show_external_connections

@@ -68,3 +68,7 @@ SHOW EXTERNAL CONNECTIONS
 connection_name connection_uri    connection_type
 foo_conn        nodelocal://1/foo STORAGE
 baz_conn        nodelocal://1/baz STORAGE
+
+# Regression test for #147877
+statement error subqueries are not allowed in check_external_connection
+CHECK EXTERNAL CONNECTION NULLIF WITH CONCURRENTLY = EXISTS ( ( TABLE error ) );

pkg/upgrade/upgrades/upgrades.go

@@ -96,13 +96,6 @@ var upgrades = []upgradebase.Upgrade{
 		eventLogTableMigration,
 		upgrade.RestoreActionNotRequired("cluster restore does not restore the new column or index"),
 	),
-	upgrade.NewTenantUpgrade(
-		"add new hot range logger job",
-		clusterversion.V25_3_AddHotRangeLoggerJob.Version(),
-		upgrade.NoPrecondition,
-		addHotRangeLoggerJob,
-		upgrade.RestoreActionNotRequired("cluster restore does not restore this job"),
-	),
 
 	upgrade.NewTenantUpgrade(
 		"add 'estimated_last_login_time' column to system.users table",
@@ -112,6 +105,14 @@ var upgrades = []upgradebase.Upgrade{
 		upgrade.RestoreActionNotRequired("cluster restore does not restore the new column"),
 	),
 
+	upgrade.NewTenantUpgrade(
+		"add new hot range logger job",
+		clusterversion.V25_3_AddHotRangeLoggerJob.Version(),
+		upgrade.NoPrecondition,
+		addHotRangeLoggerJob,
+		upgrade.RestoreActionNotRequired("cluster restore does not restore this job"),
+	),
+
 	// Note: when starting a new release version, the first upgrade (for
 	// Vxy_zStart) must be a newFirstUpgrade. Keep this comment at the bottom.
 }