Merge #156594 #156935 #156943

156594: backup: test adding/dropping database in cluster backup and restore r=jeffswenson a=msbutler

Epic: None

Release note: none

156935: logical: enable crud writer by default r=jeffswenson a=jeffswenson

This change enables the crud writer by default for IMMEDIATE and VALIDATED mode LDR jobs.

Release note: LDR now supports partial indexes and is tolerant of mismatched column ids in the source and destination tables.

Epic: CRDB-51533

156943: workflow: backport stale action should mark PRs stale after 14 days r=RaduBerinde a=rail

Previously, the backport stale action marked PRs as stale after 21 days of inactivity. This change updates the configuration to mark PRs as stale after 14 days of inactivity, ensuring that stale backport PRs are identified and addressed more promptly.

Epic: none
Release note: none

Co-authored-by: Michael Butler <[email protected]>
Co-authored-by: Jeff Swenson <[email protected]>
Co-authored-by: Rail Aliiev <[email protected]>

Author: 4 people

.github/workflows/backport-stale.yml

@@ -2,6 +2,7 @@ name: Mark stale backport requests
 
 on:
   schedule:
+  # Run at 10am UTC daily, except weekends
   - cron: "0 10 * * 1-4"
   workflow_dispatch:
 
@@ -12,17 +13,17 @@ jobs:
       issues: write
       pull-requests: write
     steps:
-    - uses: actions/stale@v3
+    - uses: actions/stale@v10
       with:
         operations-per-run: 1000
         repo-token: ${{ secrets.GITHUB_TOKEN }}
-        stale-issue-message: 'Blah'
-        stale-pr-message: 'Reminder: it has been 3 weeks please merge or close your backport!'
+        stale-issue-message: 'Ignored'
+        stale-pr-message: 'Reminder: it has been 2 weeks please merge or close your backport!'
         stale-issue-label: 'no-backport-issue-activity'
         stale-pr-label: 'no-backport-pr-activity'
         close-issue-label: 'X-stale'
         close-pr-label: 'X-stale'
-        days-before-pr-stale: 21
+        days-before-pr-stale: 14
         # Disable this for issues, by setting a very high bar
         days-before-issue-stale: 99999
         days-before-close: 99999

pkg/backup/backup_test.go

@@ -11198,3 +11198,94 @@ func TestBackupRestoreFunctionDependenciesRevisionHistory(t *testing.T) {
 
 	checkFunctions("test6")
 }
+
+func TestBackupRestoreDatabaseRevisionHistory(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	dataDir, dirCleanupFunc := testutils.TempDir(t)
+	defer dirCleanupFunc()
+
+	_, sqlDB, cleanupFn := backupRestoreTestSetupEmpty(t, singleNode, dataDir, InitManualReplication, base.TestClusterArgs{})
+	defer cleanupFn()
+
+	// Helper function to check if a database exists.
+	checkDatabase := func(dbName string, shouldExist bool) {
+		var count int
+		sqlDB.QueryRow(t, `SELECT count(*) FROM system.namespace WHERE name = $1 AND "parentID" = 0`, dbName).Scan(&count)
+		if shouldExist {
+			if count != 1 {
+				t.Fatalf("expected database %s to exist, but it doesn't", dbName)
+			}
+		} else {
+			if count != 0 {
+				t.Fatalf("expected database %s to not exist, but it does", dbName)
+			}
+		}
+	}
+
+	sqlDB.Exec(t, `CREATE DATABASE test_db`)
+
+	// T1: Full cluster backup with revision history.
+	backupPath := "nodelocal://1/database_revision_backup"
+	sqlDB.Exec(t, `BACKUP INTO $1 WITH revision_history`, backupPath)
+
+	// T2: Capture timestamp after backup (database exists).
+	var t2 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t2)
+
+	sqlDB.Exec(t, `DROP DATABASE test_db`)
+
+	checkDatabase("test_db", false)
+
+	// T3: Capture timestamp after dropping database.
+	var t3 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t3)
+
+	sqlDB.Exec(t, `CREATE DATABASE ephemeral`)
+
+	// T4: Capture timestamp the has the ephemeral database.
+	var t4 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t4)
+
+	sqlDB.Exec(t, `DROP DATABASE ephemeral`)
+
+	// T5: Capture timestamp after dropping ephemeral database.
+	var t5 string
+	sqlDB.QueryRow(t, `SELECT cluster_logical_timestamp()`).Scan(&t5)
+
+	sqlDB.Exec(t, `BACKUP INTO LATEST IN $1 WITH revision_history`, backupPath)
+
+	// Choose one of the 4 restores to run, to reduce test runtime:
+	i := rand.Intn(4)
+
+	switch i {
+	case 0:
+		// Restore AOST t2 -> expect database to exist.
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t2)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("test_db", true)
+
+		sqlDB.Exec(t, `DROP DATABASE test_db`)
+	case 1:
+		// Restore AOST t3 -> expect database to not exist.
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t3)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("test_db", false)
+	case 2:
+		// Restore AOST t4 -> ephemeral db exists.
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t4)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("ephemeral", true)
+
+		sqlDB.Exec(t, `DROP DATABASE ephemeral`)
+	case 3:
+		restoreQuery := fmt.Sprintf(
+			"RESTORE FROM LATEST IN $1 AS OF SYSTEM TIME %s", t5)
+		sqlDB.Exec(t, restoreQuery, backupPath)
+		checkDatabase("ephemeral", false)
+	}
+}

pkg/crosscluster/logical/BUILD.bazel

@@ -150,6 +150,7 @@ go_test(
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/storageccl",
         "//pkg/clusterversion",
+        "//pkg/crosscluster",
         "//pkg/crosscluster/replicationtestutils",
         "//pkg/crosscluster/streamclient",
         "//pkg/crosscluster/streamclient/randclient",

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/cdcevent"
 	"github.com/cockroachdb/cockroach/pkg/clusterversion"
+	"github.com/cockroachdb/cockroach/pkg/crosscluster"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
 	_ "github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient/randclient"
@@ -895,9 +896,6 @@ func TestRandomTables(t *testing.T) {
 	tc, s, runnerA, runnerB := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
 	defer tc.Stopper().Stop(ctx)
 
-	// TODO(#148303): Remove this once the crud writer supports tables with array primary keys.
-	runnerA.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.immediate_mode_writer = 'legacy-kv'")
-
 	sqlA := s.SQLConn(t, serverutils.DBName("a"))
 
 	var tableName, streamStartStmt string
@@ -2897,8 +2895,14 @@ func TestGetWriterType(t *testing.T) {
 
 	t.Run("validated-mode", func(t *testing.T) {
 		st := cluster.MakeTestingClusterSettings()
+
 		wt, err := getWriterType(ctx, jobspb.LogicalReplicationDetails_Validated, st)
 		require.NoError(t, err)
+		require.Equal(t, sqlclustersettings.LDRWriterTypeCRUD, wt)
+
+		crosscluster.LogicalReplicationUDFWriterEnabled.Override(ctx, &st.SV, true)
+		wt, err = getWriterType(ctx, jobspb.LogicalReplicationDetails_Validated, st)
+		require.NoError(t, err)
 		require.Equal(t, sqlclustersettings.LDRWriterTypeSQL, wt)
 	})
 

pkg/crosscluster/logical/logical_replication_writer_processor.go

@@ -741,11 +741,17 @@ func (lrw *logicalReplicationWriterProcessor) setupBatchHandlers(ctx context.Con
 func getWriterType(
 	ctx context.Context, mode jobspb.LogicalReplicationDetails_ApplyMode, settings *cluster.Settings,
 ) (sqlclustersettings.LDRWriterType, error) {
+	// TODO(jeffswenson): delete the kv and legacy sql ldr writers
 	switch mode {
 	case jobspb.LogicalReplicationDetails_Immediate:
 		return sqlclustersettings.LDRWriterType(sqlclustersettings.LDRImmediateModeWriter.Get(&settings.SV)), nil
 	case jobspb.LogicalReplicationDetails_Validated:
-		return sqlclustersettings.LDRWriterTypeSQL, nil
+		if crosscluster.LogicalReplicationUDFWriterEnabled.Get(&settings.SV) {
+			// If the UDF writer is enabled, fall back to the legacy SQL writer for
+			// validated mode.
+			return sqlclustersettings.LDRWriterTypeSQL, nil
+		}
+		return sqlclustersettings.LDRWriterTypeCRUD, nil
 	default:
 		return "", errors.Newf("unknown logical replication writer type: %s", mode)
 	}

pkg/sql/sqlclustersettings/clustersettings.go

@@ -144,8 +144,7 @@ var LDRImmediateModeWriter = settings.RegisterStringSetting(
 	settings.ApplicationLevel,
 	"logical_replication.consumer.immediate_mode_writer",
 	"the writer to use when in immediate mode",
-	// TODO(jeffswenson): make the crud writer the default
-	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeLegacyKV), string(LDRWriterTypeCRUD), string(LDRWriterTypeSQL)),
+	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeCRUD), string(LDRWriterTypeLegacyKV), string(LDRWriterTypeSQL)),
 	settings.WithValidateString(func(sv *settings.Values, val string) error {
 		if val != string(LDRWriterTypeSQL) && val != string(LDRWriterTypeLegacyKV) && val != string(LDRWriterTypeCRUD) {
 			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)