backupccl,spanconfig,kvserver: ExportRequest noops on ranges excluded…

… from backup

This change is the first of two changes that gets us to the goal of backup
ignoring certain table row data, and not holding up GC on these ranges.

This change does a few things:

- It sets up the transport of the exclude_data_from_backup bit set on a
table descriptor, to the span configuration applied in KV.

- It teaches ExportRequest on a range marked as excluded to return
an empty ExportResponse. In this way, a backup processor will receive no row
data to backup up for an ephemeral table.

- A follow up change will also teach the SQLTranslator
to not populate the protected timestamp field on the SpanConfig for such
tables. This way, a long running backup will not hold up GC on such high-churn
tables. With no protection on such ranges, it is possible that an
ExportRequest targetting the range has a StartTime
below the range's GCThreshold. To avoid the returned BatchTimestampBeforeGCError
from failing the backup we decorate the the error with information about the
range being excluded from backup and handle the error in the backup processor.

Informs: #73536

Release note (sql change): BACKUP of a table marked with `exclude_data_from_backup`
via `ALTER TABLE ... SET (exclude_data_from_backup = true)` will no longer backup
that table's row data. The backup will continue to backup the table's descriptor
and related metadata, and so on restore we will end up with an empty version of
the backed up table.

pkg/ccl/backupccl/BUILD.bazel

@@ -75,6 +75,7 @@ go_library(
         "//pkg/sql/catalog/descidgen",
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/descs",
+        "//pkg/sql/catalog/desctestutils",
         "//pkg/sql/catalog/multiregion",
         "//pkg/sql/catalog/resolver",
         "//pkg/sql/catalog/schemadesc",
@@ -103,6 +104,9 @@ go_library(
         "//pkg/sql/types",
         "//pkg/storage",
         "//pkg/testutils",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
         "//pkg/util",
         "//pkg/util/admission",
         "//pkg/util/contextutil",
@@ -123,6 +127,8 @@ go_library(
         "//pkg/util/timeutil",
         "//pkg/util/tracing",
         "//pkg/util/uuid",
+        "//pkg/workload/bank",
+        "//pkg/workload/workloadsql",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_gogo_protobuf//jsonpb",
@@ -148,7 +154,6 @@ go_test(
         "bench_test.go",
         "create_scheduled_backup_test.go",
         "full_cluster_backup_restore_test.go",
-        "helpers_test.go",
         "insert_missing_public_schema_namespace_entry_restore_test.go",
         "key_rewriter_test.go",
         "main_test.go",
@@ -247,7 +252,6 @@ go_test(
         "//pkg/util/timeutil",
         "//pkg/util/uuid",
         "//pkg/workload/bank",
-        "//pkg/workload/workloadsql",
         "@com_github_aws_aws_sdk_go//aws/credentials",
         "@com_github_cockroachdb_cockroach_go_v2//crdb",
         "@com_github_cockroachdb_datadriven//:datadriven",

pkg/ccl/backupccl/backup_cloud_test.go

@@ -62,7 +62,7 @@ func TestCloudBackupRestoreS3(t *testing.T) {
 	const numAccounts = 1000
 
 	ctx := context.Background()
-	tc, _, _, cleanupFn := BackupRestoreTestSetup(t, 1, numAccounts, InitManualReplication)
+	tc, _, _, cleanupFn := backupRestoreTestSetup(t, 1, numAccounts, InitManualReplication)
 	defer cleanupFn()
 	prefix := fmt.Sprintf("TestBackupRestoreS3-%d", timeutil.Now().UnixNano())
 	uri := url.URL{Scheme: "s3", Host: bucket, Path: prefix}
@@ -89,7 +89,7 @@ func TestCloudBackupRestoreGoogleCloudStorage(t *testing.T) {
 	const numAccounts = 1000
 
 	ctx := context.Background()
-	tc, _, _, cleanupFn := BackupRestoreTestSetup(t, 1, numAccounts, InitManualReplication)
+	tc, _, _, cleanupFn := backupRestoreTestSetup(t, 1, numAccounts, InitManualReplication)
 	defer cleanupFn()
 	prefix := fmt.Sprintf("TestBackupRestoreGoogleCloudStorage-%d", timeutil.Now().UnixNano())
 	uri := url.URL{Scheme: "gs", Host: bucket, Path: prefix}
@@ -120,7 +120,7 @@ func TestCloudBackupRestoreAzure(t *testing.T) {
 	const numAccounts = 1000
 
 	ctx := context.Background()
-	tc, _, _, cleanupFn := BackupRestoreTestSetup(t, 1, numAccounts, InitManualReplication)
+	tc, _, _, cleanupFn := backupRestoreTestSetup(t, 1, numAccounts, InitManualReplication)
 	defer cleanupFn()
 	prefix := fmt.Sprintf("TestBackupRestoreAzure-%d", timeutil.Now().UnixNano())
 	uri := url.URL{Scheme: "azure", Host: bucket, Path: prefix}

pkg/ccl/backupccl/backup_processor.go

@@ -451,6 +451,17 @@ func runBackupProcessor(
 						if errors.HasType(exportRequestErr, (*contextutil.TimeoutError)(nil)) {
 							return errors.Wrap(exportRequestErr, "export request timeout")
 						}
+						// BatchTimestampBeforeGCError is returned if the ExportRequest
+						// attempts to read below the range's GC threshold.
+						if batchTimestampBeforeGCError, ok := pErr.GetDetail().(*roachpb.BatchTimestampBeforeGCError); ok {
+							// If the range we are exporting is marked to be excluded from
+							// backup, it is safe to ignore the error. It is likely that the
+							// table has been configured with a low GC TTL, and so the data
+							// the backup is targeting has already been gc'ed.
+							if batchTimestampBeforeGCError.DataExcludedFromBackup {
+								continue
+							}
+						}
 						return errors.Wrapf(exportRequestErr, "exporting %s", span.span)
 					}
 
@@ -633,7 +644,6 @@ func makeSSTSink(
 	// prior to the error.
 	incrementSize := int64(32 << 20)
 	maxSize := smallFileBuffer.Get(s.conf.settings)
-	log.Infof(ctx, "this is the max size we are using %d", (maxSize / (1024 * 1024)))
 	for {
 		if s.queueCap >= maxSize {
 			break

pkg/ccl/backupccl/backup_rand_test.go

@@ -85,8 +85,8 @@ database_name = 'rand' AND schema_name = 'public'`)
 	// Now that we've created our random tables, backup and restore the whole DB
 	// and compare all table descriptors for equality.
 
-	dbBackup := LocalFoo + "wholedb"
-	tablesBackup := LocalFoo + "alltables"
+	dbBackup := localFoo + "wholedb"
+	tablesBackup := localFoo + "alltables"
 	dbBackups := []string{dbBackup, tablesBackup}
 
 	if err := verifyBackupRestoreStatementResult(
@@ -136,7 +136,7 @@ database_name = 'rand' AND schema_name = 'public'`)
 
 	for i, combo := range tableNameCombos {
 		sqlDB.Exec(t, "DROP DATABASE IF EXISTS restoredb; CREATE DATABASE restoredb")
-		backupTarget := fmt.Sprintf("%s%d", LocalFoo, i)
+		backupTarget := fmt.Sprintf("%s%d", localFoo, i)
 		if len(combo) == 0 {
 			continue
 		}