From 43710e57ee83711504eb9ba9741ba4b5ba6d14d8 Mon Sep 17 00:00:00 2001
From: Alfonso Subiotto Marques <alfonso@cockroachlabs.com>
Date: Wed, 19 Aug 2020 12:34:17 +0200
Subject: [PATCH 1/3] colexec: create new message to send metadata in unordered
 synchronizer

This commit fixes a race condition where a metadata message would be
double-freed and therefore the same object returned to two different goroutines
from a sync.Pool.

The root cause of this issue was that input goroutines in the parallel
unordered synchronizer use a single message that is sent repeatedly over a
channel instead of multiple messages to avoid allocations. A scenario could
occur where an input would drain metadata and set its message's metadata field
while its message was still unread in the channel. The message would then be
sent on the channel again, and the synchronizer's DrainMeta method would read
the first message with the metadata field set, followed by the same message a
second time. This results in returning the same metadata message twice to the
distsql receiver, which would release the same metadata twice.

The solution is to instead allocate a new message when draining, which will
leave message already present in the channel untouched.

Release note: None (no release with bug)
---
 pkg/sql/colexec/parallel_unordered_synchronizer.go | 12 +++++++++---
 pkg/sql/rowexec/processors_test.go                 |  1 -
 2 files changed, 9 insertions(+), 4 deletions(-)

diff --git a/pkg/sql/colexec/parallel_unordered_synchronizer.go b/pkg/sql/colexec/parallel_unordered_synchronizer.go
index c2a07b8f3d92..c89405b011c3 100644
--- a/pkg/sql/colexec/parallel_unordered_synchronizer.go
+++ b/pkg/sql/colexec/parallel_unordered_synchronizer.go
@@ -23,9 +23,10 @@ import (
 	"github.com/cockroachdb/errors"
 )
 
-// unorderedSynchronizerMsg is a light wrapper over a coldata.Batch sent over a
-// channel so that the main goroutine can know which input this message
-// originated from.
+// unorderedSynchronizerMsg is a light wrapper over a coldata.Batch or metadata
+// sent over a channel so that the main goroutine can know which input this
+// message originated from.
+// Note that either a batch or metadata must be sent, but not both.
 type unorderedSynchronizerMsg struct {
 	inputIdx int
 	b        coldata.Batch
@@ -237,6 +238,11 @@ func (s *ParallelUnorderedSynchronizer) init(ctx context.Context) {
 					// In case of a zero-length batch, proceed to drain the input.
 					fallthrough
 				case parallelUnorderedSynchronizerStateDraining:
+					// Create a new message for metadata. The previous message cannot be
+					// overwritten since it might still be in the channel.
+					msg = &unorderedSynchronizerMsg{
+						inputIdx: inputIdx,
+					}
 					if input.MetadataSources != nil {
 						msg.meta = input.MetadataSources.DrainMeta(ctx)
 					}
diff --git a/pkg/sql/rowexec/processors_test.go b/pkg/sql/rowexec/processors_test.go
index 153c8741ade9..b33407033a69 100644
--- a/pkg/sql/rowexec/processors_test.go
+++ b/pkg/sql/rowexec/processors_test.go
@@ -852,7 +852,6 @@ func TestUncertaintyErrorIsReturned(t *testing.T) {
 	testutils.RunTrueAndFalse(t, "vectorize", func(t *testing.T, vectorize bool) {
 		vectorizeOpt := "off"
 		if vectorize {
-			skip.WithIssue(t, 52948)
 			vectorizeOpt = "on"
 		}
 		for _, testCase := range testCases {

From f04cc3c7a0b8fe50954e0ce1f35989fb54580b47 Mon Sep 17 00:00:00 2001
From: Yevgeniy Miretskiy <yevgeniy@cockroachlabs.com>
Date: Wed, 19 Aug 2020 08:29:42 -0400
Subject: [PATCH 2/3] bulkio: Make incremental scheduled backup wait for full
 backup.

Add ability to record schedule groups: set of related schedules.
Use this functionality to makean incremental schedule wait
until the full one completes before it begins its execution.

Release Notes: None
---
 pkg/ccl/backupccl/backup.pb.go                | 276 ++++++++++--------
 pkg/ccl/backupccl/backup.proto                |   1 +
 pkg/ccl/backupccl/backup_job.go               |  50 +++-
 pkg/ccl/backupccl/create_scheduled_backup.go  | 122 ++++----
 .../backupccl/create_scheduled_backup_test.go |  51 +++-
 pkg/ccl/backupccl/schedule_exec.go            |  68 ++++-
 pkg/jobs/executor_impl.go                     |   9 +-
 pkg/jobs/job_scheduler_test.go                |   8 +-
 pkg/jobs/scheduled_job.go                     |  47 +++
 pkg/jobs/scheduled_job_executor.go            |  48 +--
 pkg/jobs/scheduled_job_executor_test.go       |   9 +-
 11 files changed, 435 insertions(+), 254 deletions(-)

diff --git a/pkg/ccl/backupccl/backup.pb.go b/pkg/ccl/backupccl/backup.pb.go
index 7945071714f7..5b7cd83a9be1 100644
--- a/pkg/ccl/backupccl/backup.pb.go
+++ b/pkg/ccl/backupccl/backup.pb.go
@@ -54,7 +54,7 @@ func (x MVCCFilter) String() string {
 	return proto.EnumName(MVCCFilter_name, int32(x))
 }
 func (MVCCFilter) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{0}
+	return fileDescriptor_backup_062e272dddc240ce, []int{0}
 }
 
 type EncryptionInfo_Scheme int32
@@ -74,7 +74,7 @@ func (x EncryptionInfo_Scheme) String() string {
 	return proto.EnumName(EncryptionInfo_Scheme_name, int32(x))
 }
 func (EncryptionInfo_Scheme) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{3, 0}
+	return fileDescriptor_backup_062e272dddc240ce, []int{3, 0}
 }
 
 type ScheduledBackupExecutionArgs_BackupType int32
@@ -97,7 +97,7 @@ func (x ScheduledBackupExecutionArgs_BackupType) String() string {
 	return proto.EnumName(ScheduledBackupExecutionArgs_BackupType_name, int32(x))
 }
 func (ScheduledBackupExecutionArgs_BackupType) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{5, 0}
+	return fileDescriptor_backup_062e272dddc240ce, []int{5, 0}
 }
 
 // RowCount tracks the size and row/index entry counts.
@@ -111,7 +111,7 @@ func (m *RowCount) Reset()         { *m = RowCount{} }
 func (m *RowCount) String() string { return proto.CompactTextString(m) }
 func (*RowCount) ProtoMessage()    {}
 func (*RowCount) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{0}
+	return fileDescriptor_backup_062e272dddc240ce, []int{0}
 }
 func (m *RowCount) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -189,7 +189,7 @@ func (m *BackupManifest) Reset()         { *m = BackupManifest{} }
 func (m *BackupManifest) String() string { return proto.CompactTextString(m) }
 func (*BackupManifest) ProtoMessage()    {}
 func (*BackupManifest) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{1}
+	return fileDescriptor_backup_062e272dddc240ce, []int{1}
 }
 func (m *BackupManifest) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -232,7 +232,7 @@ func (m *BackupManifest_File) Reset()         { *m = BackupManifest_File{} }
 func (m *BackupManifest_File) String() string { return proto.CompactTextString(m) }
 func (*BackupManifest_File) ProtoMessage()    {}
 func (*BackupManifest_File) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{1, 0}
+	return fileDescriptor_backup_062e272dddc240ce, []int{1, 0}
 }
 func (m *BackupManifest_File) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -267,7 +267,7 @@ func (m *BackupManifest_DescriptorRevision) Reset()         { *m = BackupManifes
 func (m *BackupManifest_DescriptorRevision) String() string { return proto.CompactTextString(m) }
 func (*BackupManifest_DescriptorRevision) ProtoMessage()    {}
 func (*BackupManifest_DescriptorRevision) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{1, 1}
+	return fileDescriptor_backup_062e272dddc240ce, []int{1, 1}
 }
 func (m *BackupManifest_DescriptorRevision) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -301,7 +301,7 @@ func (m *BackupManifest_Progress) Reset()         { *m = BackupManifest_Progress
 func (m *BackupManifest_Progress) String() string { return proto.CompactTextString(m) }
 func (*BackupManifest_Progress) ProtoMessage()    {}
 func (*BackupManifest_Progress) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{1, 2}
+	return fileDescriptor_backup_062e272dddc240ce, []int{1, 2}
 }
 func (m *BackupManifest_Progress) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -335,7 +335,7 @@ func (m *BackupManifest_Tenant) Reset()         { *m = BackupManifest_Tenant{} }
 func (m *BackupManifest_Tenant) String() string { return proto.CompactTextString(m) }
 func (*BackupManifest_Tenant) ProtoMessage()    {}
 func (*BackupManifest_Tenant) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{1, 3}
+	return fileDescriptor_backup_062e272dddc240ce, []int{1, 3}
 }
 func (m *BackupManifest_Tenant) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -370,7 +370,7 @@ func (m *BackupPartitionDescriptor) Reset()         { *m = BackupPartitionDescri
 func (m *BackupPartitionDescriptor) String() string { return proto.CompactTextString(m) }
 func (*BackupPartitionDescriptor) ProtoMessage()    {}
 func (*BackupPartitionDescriptor) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{2}
+	return fileDescriptor_backup_062e272dddc240ce, []int{2}
 }
 func (m *BackupPartitionDescriptor) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -410,7 +410,7 @@ func (m *EncryptionInfo) Reset()         { *m = EncryptionInfo{} }
 func (m *EncryptionInfo) String() string { return proto.CompactTextString(m) }
 func (*EncryptionInfo) ProtoMessage()    {}
 func (*EncryptionInfo) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{3}
+	return fileDescriptor_backup_062e272dddc240ce, []int{3}
 }
 func (m *EncryptionInfo) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -446,7 +446,7 @@ func (m *StatsTable) Reset()         { *m = StatsTable{} }
 func (m *StatsTable) String() string { return proto.CompactTextString(m) }
 func (*StatsTable) ProtoMessage()    {}
 func (*StatsTable) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{4}
+	return fileDescriptor_backup_062e272dddc240ce, []int{4}
 }
 func (m *StatsTable) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -473,15 +473,16 @@ var xxx_messageInfo_StatsTable proto.InternalMessageInfo
 
 // ScheduledBackupExecutionArgs is the arguments to the scheduled backup executor.
 type ScheduledBackupExecutionArgs struct {
-	BackupType      ScheduledBackupExecutionArgs_BackupType `protobuf:"varint,1,opt,name=backup_type,json=backupType,proto3,enum=cockroach.ccl.backupccl.ScheduledBackupExecutionArgs_BackupType" json:"backup_type,omitempty"`
-	BackupStatement string                                  `protobuf:"bytes,2,opt,name=backup_statement,json=backupStatement,proto3" json:"backup_statement,omitempty"`
+	BackupType       ScheduledBackupExecutionArgs_BackupType `protobuf:"varint,1,opt,name=backup_type,json=backupType,proto3,enum=cockroach.ccl.backupccl.ScheduledBackupExecutionArgs_BackupType" json:"backup_type,omitempty"`
+	BackupStatement  string                                  `protobuf:"bytes,2,opt,name=backup_statement,json=backupStatement,proto3" json:"backup_statement,omitempty"`
+	UnpauseOnSuccess int64                                   `protobuf:"varint,3,opt,name=unpause_on_success,json=unpauseOnSuccess,proto3" json:"unpause_on_success,omitempty"`
 }
 
 func (m *ScheduledBackupExecutionArgs) Reset()         { *m = ScheduledBackupExecutionArgs{} }
 func (m *ScheduledBackupExecutionArgs) String() string { return proto.CompactTextString(m) }
 func (*ScheduledBackupExecutionArgs) ProtoMessage()    {}
 func (*ScheduledBackupExecutionArgs) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{5}
+	return fileDescriptor_backup_062e272dddc240ce, []int{5}
 }
 func (m *ScheduledBackupExecutionArgs) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -518,7 +519,7 @@ func (m *RestoreProgress) Reset()         { *m = RestoreProgress{} }
 func (m *RestoreProgress) String() string { return proto.CompactTextString(m) }
 func (*RestoreProgress) ProtoMessage()    {}
 func (*RestoreProgress) Descriptor() ([]byte, []int) {
-	return fileDescriptor_backup_7f21dfad1af56536, []int{6}
+	return fileDescriptor_backup_062e272dddc240ce, []int{6}
 }
 func (m *RestoreProgress) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -1234,6 +1235,11 @@ func (m *ScheduledBackupExecutionArgs) MarshalTo(dAtA []byte) (int, error) {
 		i = encodeVarintBackup(dAtA, i, uint64(len(m.BackupStatement)))
 		i += copy(dAtA[i:], m.BackupStatement)
 	}
+	if m.UnpauseOnSuccess != 0 {
+		dAtA[i] = 0x18
+		i++
+		i = encodeVarintBackup(dAtA, i, uint64(m.UnpauseOnSuccess))
+	}
 	return i, nil
 }
 
@@ -1566,6 +1572,9 @@ func (m *ScheduledBackupExecutionArgs) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovBackup(uint64(l))
 	}
+	if m.UnpauseOnSuccess != 0 {
+		n += 1 + sovBackup(uint64(m.UnpauseOnSuccess))
+	}
 	return n
 }
 
@@ -3645,6 +3654,25 @@ func (m *ScheduledBackupExecutionArgs) Unmarshal(dAtA []byte) error {
 			}
 			m.BackupStatement = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field UnpauseOnSuccess", wireType)
+			}
+			m.UnpauseOnSuccess = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowBackup
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.UnpauseOnSuccess |= (int64(b) & 0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipBackup(dAtA[iNdEx:])
@@ -3900,111 +3928,113 @@ var (
 	ErrIntOverflowBackup   = fmt.Errorf("proto: integer overflow")
 )
 
-func init() { proto.RegisterFile("ccl/backupccl/backup.proto", fileDescriptor_backup_7f21dfad1af56536) }
-
-var fileDescriptor_backup_7f21dfad1af56536 = []byte{
-	// 1641 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x58, 0x4b, 0x6f, 0xdb, 0xd8,
-	0x15, 0x36, 0x25, 0x5a, 0x8f, 0x23, 0x59, 0x56, 0x6e, 0x9c, 0x84, 0x55, 0xa7, 0x92, 0xa2, 0x60,
-	0x50, 0xf5, 0x01, 0x0a, 0xe3, 0x34, 0x7d, 0x78, 0x11, 0xd4, 0x7a, 0x78, 0x42, 0xbf, 0x90, 0x52,
-	0x4e, 0x16, 0x03, 0x14, 0xc2, 0x15, 0x79, 0x2d, 0x11, 0xa6, 0x48, 0x0e, 0xef, 0x95, 0x26, 0x9a,
-	0x65, 0x7f, 0x41, 0xb7, 0x5d, 0xb5, 0x7f, 0xa1, 0xe8, 0x9f, 0xc8, 0x72, 0x16, 0x2d, 0x30, 0xe8,
-	0x42, 0x6d, 0x95, 0x4d, 0xd7, 0x5d, 0x74, 0x11, 0xcc, 0xa2, 0xb8, 0x97, 0xa4, 0xc8, 0x8c, 0xc7,
-	0xb1, 0x52, 0x2d, 0x0c, 0x1c, 0x1d, 0x9d, 0xf3, 0xdd, 0x7b, 0xcf, 0xe3, 0x3b, 0x47, 0x86, 0x8a,
-	0x61, 0xd8, 0xad, 0x21, 0x36, 0xae, 0xa6, 0x5e, 0x2c, 0xa9, 0x9e, 0xef, 0x32, 0x17, 0x3d, 0x30,
-	0x5c, 0xe3, 0xca, 0x77, 0xb1, 0x31, 0x56, 0x0d, 0xc3, 0x56, 0x57, 0x56, 0x95, 0xf2, 0x70, 0x6a,
-	0xd9, 0x66, 0xcb, 0x72, 0x2e, 0xdd, 0xc0, 0xb4, 0x72, 0x47, 0x98, 0x79, 0xc3, 0x16, 0xf6, 0xac,
-	0x50, 0x85, 0x22, 0x95, 0x89, 0x19, 0x0e, 0x75, 0x35, 0xfa, 0xb9, 0xdd, 0xa2, 0x0c, 0x33, 0xda,
-	0x62, 0x78, 0x68, 0x93, 0x01, 0x97, 0x2d, 0xca, 0x2c, 0x23, 0x34, 0x78, 0xc4, 0x0d, 0x0c, 0xcc,
-	0xb0, 0xed, 0x8e, 0x5a, 0x26, 0xa1, 0x86, 0x37, 0x6c, 0x51, 0xe6, 0x4f, 0x0d, 0x36, 0xf5, 0x89,
-	0x19, 0x1a, 0x29, 0x53, 0x66, 0xd9, 0xad, 0xb1, 0x6d, 0xb4, 0x98, 0x35, 0x21, 0x94, 0xe1, 0x49,
-	0x78, 0xe3, 0xca, 0xde, 0xc8, 0x1d, 0xb9, 0x42, 0x6c, 0x71, 0x29, 0xd0, 0x36, 0x2e, 0x21, 0xa7,
-	0xbb, 0x5f, 0x74, 0xdc, 0xa9, 0xc3, 0xd0, 0xf7, 0x21, 0xcf, 0xef, 0x33, 0xa0, 0xd6, 0x97, 0x44,
-	0x91, 0xea, 0x52, 0x33, 0xad, 0xe7, 0xb8, 0xa2, 0x6f, 0x7d, 0x49, 0x10, 0x02, 0xd9, 0x77, 0xbf,
-	0xa0, 0x4a, 0x4a, 0xe8, 0x85, 0x8c, 0x1e, 0xc1, 0x8e, 0xe5, 0x98, 0xe4, 0xd5, 0x80, 0x38, 0xcc,
-	0xb7, 0x08, 0x55, 0xd2, 0xe2, 0xcb, 0xa2, 0x50, 0xf6, 0x02, 0xdd, 0xb1, 0x9c, 0x93, 0xcb, 0xdb,
-	0x8d, 0xff, 0xde, 0x83, 0x52, 0x5b, 0x04, 0xe9, 0x0c, 0x3b, 0xd6, 0x25, 0xa1, 0x0c, 0xb5, 0x01,
-	0x28, 0xc3, 0x3e, 0x1b, 0xf0, 0x9b, 0x8a, 0xf3, 0x0a, 0xfb, 0x3f, 0x50, 0xe3, 0xb8, 0xf2, 0x97,
-	0xa8, 0x63, 0xdb, 0x50, 0x2f, 0xa2, 0x97, 0xb4, 0xe5, 0xd7, 0x8b, 0xda, 0x96, 0x9e, 0x17, 0x6e,
-	0x5c, 0x8b, 0x9e, 0x42, 0x8e, 0x38, 0x66, 0x80, 0x90, 0x5a, 0x1f, 0x21, 0x4b, 0x1c, 0x53, 0xf8,
-	0x3f, 0x86, 0x6d, 0xea, 0x61, 0x87, 0xdf, 0x3c, 0xdd, 0x2c, 0xec, 0x3f, 0x48, 0x38, 0x87, 0x29,
-	0x52, 0xfb, 0x1e, 0x76, 0x42, 0xb7, 0xc0, 0x16, 0x3d, 0x83, 0xed, 0x4b, 0xcb, 0x26, 0x54, 0x91,
-	0x85, 0xd3, 0x4f, 0xd5, 0x1b, 0x6a, 0x41, 0x7d, 0xf7, 0xc1, 0xea, 0x91, 0x65, 0x93, 0x08, 0x49,
-	0x00, 0x20, 0x0d, 0x0a, 0x3c, 0x91, 0xbe, 0xe5, 0x31, 0xd7, 0xa7, 0xca, 0xb6, 0xc0, 0x7b, 0x98,
-	0xc0, 0xa3, 0x9f, 0xdb, 0xfc, 0x6f, 0x88, 0x29, 0x51, 0xbb, 0x2b, 0xcb, 0x10, 0x24, 0xe9, 0x8b,
-	0x0e, 0x20, 0x6d, 0x5a, 0xbe, 0x92, 0x15, 0x41, 0x68, 0x7c, 0xc7, 0x3b, 0x7a, 0xaf, 0x18, 0xf1,
-	0x1d, 0x6c, 0xf7, 0x99, 0xeb, 0xe3, 0x51, 0x74, 0x11, 0xee, 0x84, 0x3e, 0x86, 0xd2, 0xa5, 0xeb,
-	0x4f, 0x30, 0x1b, 0xcc, 0x88, 0x4f, 0x2d, 0xd7, 0x51, 0x72, 0x75, 0xa9, 0xb9, 0xa3, 0xef, 0x04,
-	0xda, 0x97, 0x81, 0x12, 0x8d, 0x00, 0x0c, 0x7b, 0x4a, 0x19, 0xf1, 0x07, 0x96, 0xa9, 0xe4, 0xeb,
-	0x52, 0xb3, 0xd8, 0x7e, 0xc6, 0x51, 0xfe, 0xbe, 0xa8, 0x3d, 0x1e, 0x59, 0x6c, 0x3c, 0x1d, 0xaa,
-	0x86, 0x3b, 0x69, 0xad, 0xce, 0x36, 0x87, 0xb1, 0xdc, 0xf2, 0xae, 0x46, 0x2d, 0x51, 0x9c, 0xd3,
-	0xa9, 0x65, 0xaa, 0x2f, 0x5e, 0x68, 0xdd, 0xe5, 0xa2, 0x96, 0xef, 0x04, 0x80, 0x5a, 0x57, 0xcf,
-	0x87, 0xd8, 0x9a, 0x89, 0x3e, 0x83, 0xac, 0xe3, 0x9a, 0x84, 0x9f, 0x02, 0x75, 0xa9, 0xb9, 0xdd,
-	0x3e, 0x5c, 0x2e, 0x6a, 0x99, 0x73, 0xd7, 0x24, 0x5a, 0xf7, 0xed, 0xba, 0x67, 0x45, 0xef, 0x0e,
-	0xdc, 0xf4, 0x0c, 0x47, 0xd4, 0x4c, 0x74, 0x00, 0x20, 0x3a, 0x74, 0xc0, 0x3b, 0x54, 0x29, 0x88,
-	0x70, 0xdd, 0x4b, 0x84, 0x4b, 0x7c, 0xa9, 0x6a, 0xce, 0xa5, 0x1b, 0x55, 0x9b, 0xd0, 0x70, 0x05,
-	0x3a, 0x86, 0x22, 0xaf, 0xf4, 0xf9, 0xc0, 0xe0, 0xfd, 0x42, 0x95, 0xa2, 0xf0, 0x7e, 0x78, 0x63,
-	0xfe, 0xa3, 0xce, 0x8a, 0xf2, 0x25, 0x9c, 0x85, 0x86, 0xa2, 0x0b, 0x28, 0x4c, 0x66, 0x86, 0x31,
-	0xb8, 0xb4, 0x6c, 0x46, 0x7c, 0x65, 0xa7, 0x2e, 0x35, 0x4b, 0xfb, 0x8f, 0x6e, 0x84, 0x3a, 0x7b,
-	0xd9, 0xe9, 0x1c, 0x09, 0xd3, 0x76, 0x69, 0xb9, 0xa8, 0x41, 0xfc, 0x59, 0x07, 0x8e, 0x13, 0xc8,
-	0x08, 0x43, 0xd1, 0x70, 0x27, 0x9e, 0x4d, 0x18, 0x19, 0x98, 0x43, 0xaa, 0x94, 0xea, 0xe9, 0xe6,
-	0x4e, 0xfb, 0xe9, 0xdb, 0x45, 0xed, 0x60, 0xad, 0xa0, 0x5d, 0xa7, 0x18, 0x55, 0xeb, 0xea, 0x85,
-	0x08, 0xb3, 0x3b, 0xe4, 0xd5, 0x5f, 0xb6, 0x1c, 0xe6, 0xbb, 0xe6, 0xd4, 0x20, 0xe6, 0x20, 0xe8,
-	0x9e, 0xdd, 0x75, 0xba, 0x67, 0x37, 0x76, 0xeb, 0x8b, 0x3e, 0x72, 0x01, 0xc5, 0x15, 0x3c, 0x30,
-	0xc6, 0xd8, 0x19, 0x11, 0xaa, 0x94, 0x05, 0xd6, 0xc1, 0xba, 0x4d, 0x15, 0x77, 0x85, 0x4e, 0x66,
-	0x16, 0xaf, 0xd3, 0xf0, 0xb8, 0x3b, 0x31, 0x76, 0x27, 0x80, 0x46, 0x7d, 0xb8, 0xeb, 0x87, 0x46,
-	0x83, 0x04, 0xf5, 0xdc, 0x59, 0x9f, 0x38, 0xee, 0x44, 0xfe, 0xfd, 0x15, 0x05, 0xfd, 0x06, 0x52,
-	0x96, 0xa9, 0x20, 0xd1, 0x0d, 0x87, 0x9b, 0x75, 0x43, 0x4a, 0xeb, 0xea, 0x29, 0xcb, 0x44, 0x5d,
-	0xa8, 0x7a, 0xd8, 0x67, 0x16, 0xe3, 0x17, 0x4d, 0x84, 0x88, 0x93, 0x86, 0x83, 0x27, 0x84, 0x2a,
-	0x77, 0xeb, 0xe9, 0x66, 0x5e, 0xff, 0x68, 0x65, 0x15, 0x47, 0xe1, 0x28, 0xb2, 0x41, 0xfb, 0x50,
-	0xb4, 0x5d, 0x03, 0xdb, 0x16, 0x9b, 0x0f, 0xae, 0x66, 0x54, 0xd9, 0xe3, 0x3e, 0xed, 0xdd, 0xe5,
-	0xa2, 0x56, 0x38, 0x0d, 0xf5, 0x27, 0x2f, 0xa9, 0x5e, 0x88, 0x8c, 0x4e, 0x66, 0x14, 0xfd, 0x16,
-	0xee, 0x99, 0xc4, 0xf3, 0x89, 0x81, 0x19, 0x4f, 0x6e, 0x34, 0x81, 0xa8, 0x72, 0x4f, 0x64, 0xa5,
-	0xf9, 0x6d, 0x6a, 0xe2, 0xe3, 0x4a, 0xbd, 0xe0, 0xe3, 0xaa, 0x1f, 0xd9, 0x3e, 0xe7, 0x73, 0x45,
-	0xdf, 0x8b, 0x61, 0x56, 0xdf, 0x50, 0x34, 0x87, 0xbb, 0xc9, 0x8c, 0xbb, 0x33, 0xc2, 0xa9, 0x48,
-	0xb9, 0x2f, 0x9a, 0xfc, 0xd9, 0xdb, 0x45, 0xad, 0xbb, 0x76, 0x95, 0x52, 0x32, 0x69, 0x31, 0x9f,
-	0x24, 0x69, 0xb1, 0x13, 0xe2, 0xe9, 0x89, 0xb2, 0x8a, 0x74, 0xe8, 0x2f, 0x12, 0xec, 0xc5, 0xef,
-	0x49, 0x84, 0xf2, 0x81, 0x78, 0xd9, 0xaf, 0xd7, 0xad, 0xb7, 0xf8, 0x35, 0xab, 0x48, 0xf3, 0x49,
-	0x37, 0x6f, 0x3f, 0xfd, 0xdd, 0x3f, 0x36, 0x6a, 0xb2, 0xbb, 0xf4, 0x3a, 0x32, 0x3a, 0x87, 0x2c,
-	0x23, 0x0e, 0xe6, 0x64, 0xa3, 0x88, 0x7b, 0xaa, 0xeb, 0xde, 0xf3, 0x42, 0xb8, 0x45, 0xf3, 0x2e,
-	0x04, 0xa9, 0xfc, 0x27, 0x05, 0x32, 0x47, 0x47, 0x9f, 0x80, 0xcc, 0x5b, 0x37, 0x1c, 0xbb, 0xb7,
-	0x74, 0xae, 0x30, 0xe5, 0x1b, 0x80, 0x87, 0xd9, 0x58, 0xcc, 0xd9, 0xbc, 0x2e, 0x64, 0x74, 0x1f,
-	0x32, 0x74, 0x8c, 0x9f, 0x7c, 0xb2, 0xaf, 0xc8, 0xbc, 0x01, 0xf4, 0xf0, 0xd3, 0x35, 0xa6, 0xcc,
-	0x6c, 0xc0, 0x94, 0xef, 0xee, 0x09, 0xd9, 0x8d, 0xf7, 0x84, 0xdc, 0xff, 0xb1, 0x27, 0xb4, 0xa0,
-	0x90, 0xe8, 0x25, 0x31, 0xfb, 0xf2, 0x01, 0x11, 0xc7, 0xad, 0xa4, 0x43, 0xdc, 0x49, 0xc7, 0x72,
-	0x2e, 0x5d, 0x96, 0x8f, 0xe5, 0xdc, 0x76, 0x39, 0x53, 0xf9, 0xab, 0x04, 0xe8, 0x3a, 0x4d, 0xa1,
-	0x5f, 0x80, 0xfc, 0xa1, 0x9b, 0x8f, 0x70, 0x40, 0xe7, 0x90, 0xd2, 0xba, 0x22, 0x0d, 0x9b, 0x53,
-	0x7b, 0x4a, 0xeb, 0xa2, 0x27, 0x20, 0x73, 0x85, 0xd8, 0xde, 0xd6, 0x59, 0x3f, 0x74, 0x61, 0x5e,
-	0xf9, 0xa3, 0x04, 0xb9, 0xe7, 0xbe, 0x3b, 0xf2, 0x09, 0x4d, 0xec, 0x44, 0xd2, 0xe6, 0x3b, 0x51,
-	0xc9, 0x27, 0xb3, 0x24, 0x3f, 0x7f, 0xc0, 0x62, 0x57, 0xf4, 0xc9, 0x6c, 0x45, 0xcd, 0x95, 0x9f,
-	0x41, 0x26, 0x68, 0x03, 0x74, 0x5f, 0x90, 0x34, 0x8f, 0xb4, 0xdc, 0xce, 0x24, 0x98, 0x16, 0x81,
-	0x2c, 0xf6, 0x80, 0x94, 0xa8, 0x5e, 0x21, 0x57, 0x8e, 0x40, 0xb9, 0xa9, 0xc9, 0x51, 0x19, 0xd2,
-	0x57, 0x64, 0x2e, 0x80, 0x76, 0x74, 0x2e, 0xa2, 0x3d, 0xd8, 0x9e, 0x61, 0x7b, 0x4a, 0xc2, 0xb6,
-	0x08, 0x3e, 0x1c, 0xa4, 0x7e, 0x29, 0x1d, 0xcb, 0xb9, 0x4c, 0x39, 0xdb, 0xf8, 0x46, 0x82, 0xef,
-	0x05, 0x6f, 0x7e, 0x7e, 0x9d, 0xac, 0xbf, 0x5d, 0x57, 0xd2, 0x6d, 0x75, 0x15, 0xc7, 0x39, 0xb5,
-	0x69, 0x9c, 0x4d, 0xc8, 0x07, 0xd6, 0x7c, 0xcd, 0x4a, 0x8b, 0xf1, 0xf5, 0xe9, 0x66, 0xe3, 0x2b,
-	0x17, 0x9c, 0xa9, 0x75, 0xf5, 0x5c, 0x80, 0xac, 0x99, 0x8d, 0x6f, 0x52, 0x50, 0xea, 0x39, 0x86,
-	0x3f, 0xf7, 0xf8, 0xcb, 0xc5, 0x16, 0x75, 0x04, 0x19, 0x6a, 0x8c, 0x49, 0x58, 0xf9, 0xa5, 0xf7,
-	0x50, 0xda, 0xbb, 0x8e, 0x6a, 0x5f, 0x78, 0xe9, 0xa1, 0x37, 0xcf, 0x1d, 0xc5, 0x36, 0x8b, 0x72,
-	0xc7, 0x65, 0xf4, 0x07, 0x09, 0xea, 0x24, 0xf0, 0x22, 0x66, 0x17, 0x33, 0x7c, 0x42, 0xe6, 0xed,
-	0xf9, 0xc9, 0x59, 0xff, 0x0c, 0xf3, 0xdd, 0xf2, 0x84, 0xcc, 0xb5, 0x6e, 0xb8, 0xeb, 0x9f, 0xad,
-	0x7b, 0x6c, 0xef, 0x16, 0x3c, 0x51, 0x19, 0xfa, 0xad, 0xc7, 0x56, 0xfa, 0xf0, 0xf1, 0x5a, 0x50,
-	0xc9, 0x22, 0xcb, 0x7f, 0x47, 0x91, 0x15, 0x13, 0x45, 0xd6, 0x78, 0x00, 0x99, 0x20, 0x2c, 0x68,
-	0x07, 0xf2, 0x87, 0xbd, 0xfe, 0xfe, 0x93, 0x9f, 0x7f, 0xda, 0x39, 0x2b, 0x6f, 0x1d, 0xc8, 0xff,
-	0xfe, 0x53, 0x4d, 0x6a, 0xbc, 0x04, 0xe0, 0xb5, 0x4c, 0xc5, 0x88, 0x46, 0xcf, 0x04, 0x93, 0x46,
-	0x23, 0x5d, 0xfa, 0xc0, 0x91, 0x9e, 0xf0, 0x6d, 0xfc, 0x4d, 0x82, 0x8f, 0xf8, 0xb9, 0xe6, 0xd4,
-	0x26, 0x66, 0x90, 0xf6, 0xde, 0x2b, 0x62, 0x4c, 0x79, 0xd0, 0x0e, 0xfd, 0x11, 0x45, 0x18, 0x0a,
-	0x61, 0x75, 0xb1, 0xb9, 0x17, 0x65, 0xfa, 0xe6, 0x21, 0xfb, 0x3e, 0xac, 0xb0, 0x94, 0x2f, 0xe6,
-	0x1e, 0xd1, 0x61, 0xb8, 0x92, 0xd1, 0x8f, 0xa0, 0x1c, 0x1e, 0xc1, 0x2f, 0x46, 0x26, 0xc4, 0x61,
-	0x61, 0x13, 0xee, 0x06, 0xfa, 0x7e, 0xa4, 0x6e, 0xfc, 0x10, 0x20, 0x06, 0x41, 0x39, 0x90, 0x8f,
-	0x5e, 0x9c, 0x9e, 0x96, 0xb7, 0xd0, 0x2e, 0x14, 0xb4, 0xf3, 0x8e, 0xde, 0x3b, 0xeb, 0x9d, 0x5f,
-	0x1c, 0x9e, 0x96, 0xa5, 0xc6, 0x9f, 0x25, 0xd8, 0xd5, 0x09, 0x65, 0xae, 0x4f, 0x56, 0xd4, 0x76,
-	0x08, 0x59, 0x3a, 0x9d, 0x4c, 0xb0, 0x3f, 0x0f, 0xa9, 0x7a, 0xed, 0x31, 0x16, 0xf9, 0xa1, 0x3a,
-	0x14, 0xbc, 0x10, 0x4e, 0x33, 0x5f, 0x85, 0xbf, 0xa1, 0x93, 0x2a, 0xf4, 0x2b, 0x08, 0x7e, 0x6a,
-	0xf3, 0x99, 0x9c, 0x5e, 0x67, 0x26, 0xaf, 0xcc, 0x7f, 0xfc, 0x10, 0x12, 0xbf, 0x06, 0x10, 0x40,
-	0xe6, 0x14, 0x33, 0x42, 0x59, 0x79, 0x0b, 0x65, 0x21, 0x7d, 0x68, 0xdb, 0x65, 0xa9, 0xfd, 0x93,
-	0xd7, 0xff, 0xaa, 0x6e, 0xbd, 0x5e, 0x56, 0xa5, 0xaf, 0x96, 0x55, 0xe9, 0xeb, 0x65, 0x55, 0xfa,
-	0xe7, 0xb2, 0x2a, 0xfd, 0xfe, 0x4d, 0x75, 0xeb, 0xab, 0x37, 0xd5, 0xad, 0xaf, 0xdf, 0x54, 0xb7,
-	0x3e, 0xcb, 0xaf, 0x1e, 0x31, 0xcc, 0x88, 0xff, 0x0c, 0x3c, 0xfe, 0x5f, 0x00, 0x00, 0x00, 0xff,
-	0xff, 0xe0, 0xf9, 0xc8, 0x16, 0xff, 0x10, 0x00, 0x00,
+func init() { proto.RegisterFile("ccl/backupccl/backup.proto", fileDescriptor_backup_062e272dddc240ce) }
+
+var fileDescriptor_backup_062e272dddc240ce = []byte{
+	// 1667 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x58, 0xcd, 0x6f, 0xdb, 0xc8,
+	0x15, 0x37, 0x25, 0x5a, 0x96, 0x9e, 0x6c, 0x59, 0x99, 0x38, 0x09, 0xab, 0x6e, 0x25, 0x45, 0xc1,
+	0xa2, 0x6a, 0xbb, 0xa0, 0xb0, 0x4e, 0xd3, 0x0f, 0x1f, 0x82, 0x5a, 0x1f, 0xde, 0xd0, 0x5f, 0x4d,
+	0x29, 0x27, 0x87, 0x05, 0x0a, 0x61, 0x44, 0x8e, 0x25, 0xc2, 0x14, 0xc9, 0xe5, 0x0c, 0xb5, 0xd1,
+	0x1e, 0xdb, 0x7f, 0xa0, 0xd7, 0x9e, 0xda, 0x7f, 0xa1, 0xe8, 0x3f, 0x91, 0xe3, 0x1e, 0x7a, 0x58,
+	0xf4, 0xa0, 0xb6, 0xca, 0xa5, 0xe7, 0x1e, 0x7a, 0x08, 0xf6, 0x50, 0xcc, 0x90, 0x14, 0x99, 0xf5,
+	0x7a, 0xad, 0x54, 0x07, 0x01, 0xc3, 0xc7, 0xf7, 0x7e, 0x33, 0xf3, 0x3e, 0x7e, 0xef, 0x51, 0x50,
+	0x31, 0x0c, 0xbb, 0x35, 0xc4, 0xc6, 0x55, 0xe0, 0x25, 0x2b, 0xd5, 0xf3, 0x5d, 0xe6, 0xa2, 0x07,
+	0x86, 0x6b, 0x5c, 0xf9, 0x2e, 0x36, 0xc6, 0xaa, 0x61, 0xd8, 0xea, 0x52, 0xab, 0x52, 0x1e, 0x06,
+	0x96, 0x6d, 0xb6, 0x2c, 0xe7, 0xd2, 0x0d, 0x55, 0x2b, 0x77, 0x84, 0x9a, 0x37, 0x6c, 0x61, 0xcf,
+	0x8a, 0x44, 0x28, 0x16, 0x99, 0x98, 0xe1, 0x48, 0x56, 0xa3, 0x9f, 0xd9, 0x2d, 0xca, 0x30, 0xa3,
+	0x2d, 0x86, 0x87, 0x36, 0x19, 0xf0, 0xb5, 0x45, 0x99, 0x65, 0x44, 0x0a, 0x8f, 0xb8, 0x82, 0x81,
+	0x19, 0xb6, 0xdd, 0x51, 0xcb, 0x24, 0xd4, 0xf0, 0x86, 0x2d, 0xca, 0xfc, 0xc0, 0x60, 0x81, 0x4f,
+	0xcc, 0x48, 0x49, 0x09, 0x98, 0x65, 0xb7, 0xc6, 0xb6, 0xd1, 0x62, 0xd6, 0x84, 0x50, 0x86, 0x27,
+	0xd1, 0x89, 0x2b, 0x7b, 0x23, 0x77, 0xe4, 0x8a, 0x65, 0x8b, 0xaf, 0x42, 0x69, 0xe3, 0x12, 0xf2,
+	0xba, 0xfb, 0x79, 0xc7, 0x0d, 0x1c, 0x86, 0xbe, 0x0f, 0x05, 0x7e, 0x9e, 0x01, 0xb5, 0xbe, 0x20,
+	0x8a, 0x54, 0x97, 0x9a, 0x59, 0x3d, 0xcf, 0x05, 0x7d, 0xeb, 0x0b, 0x82, 0x10, 0xc8, 0xbe, 0xfb,
+	0x39, 0x55, 0x32, 0x42, 0x2e, 0xd6, 0xe8, 0x11, 0xec, 0x58, 0x8e, 0x49, 0x5e, 0x0d, 0x88, 0xc3,
+	0x7c, 0x8b, 0x50, 0x25, 0x2b, 0x5e, 0x6e, 0x0b, 0x61, 0x2f, 0x94, 0x1d, 0xcb, 0x79, 0xb9, 0xbc,
+	0xd9, 0xf8, 0xef, 0x3d, 0x28, 0xb5, 0x85, 0x93, 0xce, 0xb0, 0x63, 0x5d, 0x12, 0xca, 0x50, 0x1b,
+	0x80, 0x32, 0xec, 0xb3, 0x01, 0x3f, 0xa9, 0xd8, 0xaf, 0xb8, 0xff, 0x03, 0x35, 0xf1, 0x2b, 0xbf,
+	0x89, 0x3a, 0xb6, 0x0d, 0xf5, 0x22, 0xbe, 0x49, 0x5b, 0x7e, 0x3d, 0xaf, 0x6d, 0xe8, 0x05, 0x61,
+	0xc6, 0xa5, 0xe8, 0x29, 0xe4, 0x89, 0x63, 0x86, 0x08, 0x99, 0xd5, 0x11, 0xb6, 0x88, 0x63, 0x0a,
+	0xfb, 0xc7, 0xb0, 0x49, 0x3d, 0xec, 0xf0, 0x93, 0x67, 0x9b, 0xc5, 0xfd, 0x07, 0x29, 0xe3, 0x28,
+	0x44, 0x6a, 0xdf, 0xc3, 0x4e, 0x64, 0x16, 0xea, 0xa2, 0x67, 0xb0, 0x79, 0x69, 0xd9, 0x84, 0x2a,
+	0xb2, 0x30, 0xfa, 0x48, 0xbd, 0x21, 0x17, 0xd4, 0x77, 0x2f, 0xac, 0x1e, 0x59, 0x36, 0x89, 0x91,
+	0x04, 0x00, 0xd2, 0xa0, 0xc8, 0x03, 0xe9, 0x5b, 0x1e, 0x73, 0x7d, 0xaa, 0x6c, 0x0a, 0xbc, 0x87,
+	0x29, 0x3c, 0xfa, 0x99, 0xcd, 0x7f, 0x43, 0x4c, 0x89, 0xda, 0x5d, 0x6a, 0x46, 0x20, 0x69, 0x5b,
+	0x74, 0x00, 0x59, 0xd3, 0xf2, 0x95, 0x2d, 0xe1, 0x84, 0xc6, 0xb7, 0xdc, 0xa3, 0xf7, 0x8a, 0x11,
+	0xdf, 0xc1, 0x76, 0x9f, 0xb9, 0x3e, 0x1e, 0xc5, 0x07, 0xe1, 0x46, 0xe8, 0x43, 0x28, 0x5d, 0xba,
+	0xfe, 0x04, 0xb3, 0xc1, 0x94, 0xf8, 0xd4, 0x72, 0x1d, 0x25, 0x5f, 0x97, 0x9a, 0x3b, 0xfa, 0x4e,
+	0x28, 0x7d, 0x19, 0x0a, 0xd1, 0x08, 0xc0, 0xb0, 0x03, 0xca, 0x88, 0x3f, 0xb0, 0x4c, 0xa5, 0x50,
+	0x97, 0x9a, 0xdb, 0xed, 0x67, 0x1c, 0xe5, 0xef, 0xf3, 0xda, 0xe3, 0x91, 0xc5, 0xc6, 0xc1, 0x50,
+	0x35, 0xdc, 0x49, 0x6b, 0xb9, 0xb7, 0x39, 0x4c, 0xd6, 0x2d, 0xef, 0x6a, 0xd4, 0x12, 0xc9, 0x19,
+	0x04, 0x96, 0xa9, 0xbe, 0x78, 0xa1, 0x75, 0x17, 0xf3, 0x5a, 0xa1, 0x13, 0x02, 0x6a, 0x5d, 0xbd,
+	0x10, 0x61, 0x6b, 0x26, 0xfa, 0x14, 0xb6, 0x1c, 0xd7, 0x24, 0x7c, 0x17, 0xa8, 0x4b, 0xcd, 0xcd,
+	0xf6, 0xe1, 0x62, 0x5e, 0xcb, 0x9d, 0xbb, 0x26, 0xd1, 0xba, 0x6f, 0x57, 0xdd, 0x2b, 0xbe, 0x77,
+	0x68, 0xa6, 0xe7, 0x38, 0xa2, 0x66, 0xa2, 0x03, 0x00, 0x51, 0xa1, 0x03, 0x5e, 0xa1, 0x4a, 0x51,
+	0xb8, 0xeb, 0x5e, 0xca, 0x5d, 0xe2, 0xa5, 0xaa, 0x39, 0x97, 0x6e, 0x9c, 0x6d, 0x42, 0xc2, 0x05,
+	0xe8, 0x18, 0xb6, 0x79, 0xa6, 0xcf, 0x06, 0x06, 0xaf, 0x17, 0xaa, 0x6c, 0x0b, 0xeb, 0x87, 0x37,
+	0xc6, 0x3f, 0xae, 0xac, 0x38, 0x5e, 0xc2, 0x58, 0x48, 0x28, 0xba, 0x80, 0xe2, 0x64, 0x6a, 0x18,
+	0x83, 0x4b, 0xcb, 0x66, 0xc4, 0x57, 0x76, 0xea, 0x52, 0xb3, 0xb4, 0xff, 0xe8, 0x46, 0xa8, 0xb3,
+	0x97, 0x9d, 0xce, 0x91, 0x50, 0x6d, 0x97, 0x16, 0xf3, 0x1a, 0x24, 0xcf, 0x3a, 0x70, 0x9c, 0x70,
+	0x8d, 0x30, 0x6c, 0x1b, 0xee, 0xc4, 0xb3, 0x09, 0x23, 0x03, 0x73, 0x48, 0x95, 0x52, 0x3d, 0xdb,
+	0xdc, 0x69, 0x3f, 0x7d, 0x3b, 0xaf, 0x1d, 0xac, 0xe4, 0xb4, 0xeb, 0x14, 0xa3, 0x6a, 0x5d, 0xbd,
+	0x18, 0x63, 0x76, 0x87, 0x3c, 0xfb, 0xcb, 0x96, 0xc3, 0x7c, 0xd7, 0x0c, 0x0c, 0x62, 0x0e, 0xc2,
+	0xea, 0xd9, 0x5d, 0xa5, 0x7a, 0x76, 0x13, 0xb3, 0xbe, 0xa8, 0x23, 0x17, 0x50, 0x92, 0xc1, 0x03,
+	0x63, 0x8c, 0x9d, 0x11, 0xa1, 0x4a, 0x59, 0x60, 0x1d, 0xac, 0x5a, 0x54, 0x49, 0x55, 0xe8, 0x64,
+	0x6a, 0xf1, 0x3c, 0x8d, 0xb6, 0xbb, 0x93, 0x60, 0x77, 0x42, 0x68, 0xd4, 0x87, 0xbb, 0x7e, 0xa4,
+	0x34, 0x48, 0x51, 0xcf, 0x9d, 0xd5, 0x89, 0xe3, 0x4e, 0x6c, 0xdf, 0x5f, 0x52, 0xd0, 0x6f, 0x20,
+	0x63, 0x99, 0x0a, 0x12, 0xd5, 0x70, 0xb8, 0x5e, 0x35, 0x64, 0xb4, 0xae, 0x9e, 0xb1, 0x4c, 0xd4,
+	0x85, 0xaa, 0x87, 0x7d, 0x66, 0x31, 0x7e, 0xd0, 0x94, 0x8b, 0x38, 0x69, 0x38, 0x78, 0x42, 0xa8,
+	0x72, 0xb7, 0x9e, 0x6d, 0x16, 0xf4, 0x0f, 0x96, 0x5a, 0x89, 0x17, 0x8e, 0x62, 0x1d, 0xb4, 0x0f,
+	0xdb, 0xb6, 0x6b, 0x60, 0xdb, 0x62, 0xb3, 0xc1, 0xd5, 0x94, 0x2a, 0x7b, 0xdc, 0xa6, 0xbd, 0xbb,
+	0x98, 0xd7, 0x8a, 0xa7, 0x91, 0xfc, 0xe4, 0x25, 0xd5, 0x8b, 0xb1, 0xd2, 0xc9, 0x94, 0xa2, 0xdf,
+	0xc2, 0x3d, 0x93, 0x78, 0x3e, 0x31, 0x30, 0xe3, 0xc1, 0x8d, 0x3b, 0x10, 0x55, 0xee, 0x89, 0xa8,
+	0x34, 0xbf, 0x49, 0x4d, 0xbc, 0x5d, 0xa9, 0x17, 0xbc, 0x5d, 0xf5, 0x63, 0xdd, 0xe7, 0xbc, 0xaf,
+	0xe8, 0x7b, 0x09, 0xcc, 0xf2, 0x0d, 0x45, 0x33, 0xb8, 0x9b, 0x8e, 0xb8, 0x3b, 0x25, 0x9c, 0x8a,
+	0x94, 0xfb, 0xa2, 0xc8, 0x9f, 0xbd, 0x9d, 0xd7, 0xba, 0x2b, 0x67, 0x29, 0x25, 0x93, 0x16, 0xf3,
+	0x49, 0x9a, 0x16, 0x3b, 0x11, 0x9e, 0x9e, 0x4a, 0xab, 0x58, 0x86, 0xfe, 0x2a, 0xc1, 0x5e, 0x72,
+	0x9f, 0x94, 0x2b, 0x1f, 0x88, 0x9b, 0xfd, 0x6a, 0xd5, 0x7c, 0x4b, 0x6e, 0xb3, 0xf4, 0x34, 0xef,
+	0x74, 0xb3, 0xf6, 0xd3, 0xdf, 0xfd, 0x63, 0xad, 0x22, 0xbb, 0x4b, 0xaf, 0x23, 0xa3, 0x73, 0xd8,
+	0x62, 0xc4, 0xc1, 0x9c, 0x6c, 0x14, 0x71, 0x4e, 0x75, 0xd5, 0x73, 0x5e, 0x08, 0xb3, 0xb8, 0xdf,
+	0x45, 0x20, 0x95, 0xff, 0x64, 0x40, 0xe6, 0xe8, 0xe8, 0x63, 0x90, 0x79, 0xe9, 0x46, 0x6d, 0xf7,
+	0x96, 0xca, 0x15, 0xaa, 0x7c, 0x02, 0xf0, 0x30, 0x1b, 0x8b, 0x3e, 0x5b, 0xd0, 0xc5, 0x1a, 0xdd,
+	0x87, 0x1c, 0x1d, 0xe3, 0x27, 0x1f, 0xef, 0x2b, 0x32, 0x2f, 0x00, 0x3d, 0x7a, 0xba, 0xc6, 0x94,
+	0xb9, 0x35, 0x98, 0xf2, 0xdd, 0x39, 0x61, 0x6b, 0xed, 0x39, 0x21, 0xff, 0x7f, 0xcc, 0x09, 0x2d,
+	0x28, 0xa6, 0x6a, 0x49, 0xf4, 0xbe, 0x42, 0x48, 0xc4, 0x49, 0x29, 0xe9, 0x90, 0x54, 0xd2, 0xb1,
+	0x9c, 0xcf, 0x96, 0xe5, 0x63, 0x39, 0xbf, 0x59, 0xce, 0x55, 0xfe, 0x26, 0x01, 0xba, 0x4e, 0x53,
+	0xe8, 0xe7, 0x20, 0xbf, 0xef, 0xe4, 0x23, 0x0c, 0xd0, 0x39, 0x64, 0xb4, 0xae, 0x08, 0xc3, 0xfa,
+	0xd4, 0x9e, 0xd1, 0xba, 0xe8, 0x09, 0xc8, 0x5c, 0x20, 0xa6, 0xb7, 0x55, 0xc6, 0x0f, 0x5d, 0xa8,
+	0x57, 0xfe, 0x24, 0x41, 0xfe, 0xb9, 0xef, 0x8e, 0x7c, 0x42, 0x53, 0x33, 0x91, 0xb4, 0xfe, 0x4c,
+	0x54, 0xf2, 0xc9, 0x34, 0xcd, 0xcf, 0xef, 0x31, 0xd8, 0x6d, 0xfb, 0x64, 0xba, 0xa4, 0xe6, 0xca,
+	0x4f, 0x21, 0x17, 0x96, 0x01, 0xba, 0x2f, 0x48, 0x9a, 0x7b, 0x5a, 0x6e, 0xe7, 0x52, 0x4c, 0x8b,
+	0x40, 0x16, 0x73, 0x40, 0x46, 0x64, 0xaf, 0x58, 0x57, 0x8e, 0x40, 0xb9, 0xa9, 0xc8, 0x51, 0x19,
+	0xb2, 0x57, 0x64, 0x26, 0x80, 0x76, 0x74, 0xbe, 0x44, 0x7b, 0xb0, 0x39, 0xc5, 0x76, 0x40, 0xa2,
+	0xb2, 0x08, 0x1f, 0x0e, 0x32, 0xbf, 0x90, 0x8e, 0xe5, 0x7c, 0xae, 0xbc, 0xd5, 0xf8, 0x5a, 0x82,
+	0xef, 0x85, 0x77, 0x7e, 0x7e, 0x9d, 0xac, 0xbf, 0x99, 0x57, 0xd2, 0x6d, 0x79, 0x95, 0xf8, 0x39,
+	0xb3, 0xae, 0x9f, 0x4d, 0x28, 0x84, 0xda, 0x7c, 0xcc, 0xca, 0x8a, 0xf6, 0xf5, 0xc9, 0x7a, 0xed,
+	0x2b, 0x1f, 0xee, 0xa9, 0x75, 0xf5, 0x7c, 0x88, 0xac, 0x99, 0x8d, 0xaf, 0x33, 0x50, 0xea, 0x39,
+	0x86, 0x3f, 0xf3, 0xf8, 0xcd, 0xc5, 0x14, 0x75, 0x04, 0x39, 0x6a, 0x8c, 0x49, 0x94, 0xf9, 0xa5,
+	0xef, 0xa0, 0xb4, 0x77, 0x0d, 0xd5, 0xbe, 0xb0, 0xd2, 0x23, 0x6b, 0x1e, 0x3b, 0x8a, 0x6d, 0x16,
+	0xc7, 0x8e, 0xaf, 0xd1, 0x1f, 0x25, 0xa8, 0x93, 0xd0, 0x8a, 0x98, 0x5d, 0xcc, 0xf0, 0x09, 0x99,
+	0xb5, 0x67, 0x27, 0x67, 0xfd, 0x33, 0xcc, 0x67, 0xcb, 0x13, 0x32, 0xd3, 0xba, 0xd1, 0xac, 0x7f,
+	0xb6, 0xea, 0xb6, 0xbd, 0x5b, 0xf0, 0x44, 0x66, 0xe8, 0xb7, 0x6e, 0x5b, 0xe9, 0xc3, 0x87, 0x2b,
+	0x41, 0xa5, 0x93, 0xac, 0xf0, 0x2d, 0x49, 0xb6, 0x9d, 0x4a, 0xb2, 0xc6, 0x03, 0xc8, 0x85, 0x6e,
+	0x41, 0x3b, 0x50, 0x38, 0xec, 0xf5, 0xf7, 0x9f, 0xfc, 0xec, 0x93, 0xce, 0x59, 0x79, 0xe3, 0x40,
+	0xfe, 0xf7, 0x9f, 0x6b, 0x52, 0xe3, 0x25, 0x00, 0xcf, 0x65, 0x2a, 0x5a, 0x34, 0x7a, 0x26, 0x98,
+	0x34, 0x6e, 0xe9, 0xd2, 0x7b, 0xb6, 0xf4, 0x94, 0x6d, 0xe3, 0xf7, 0x19, 0xf8, 0x80, 0xef, 0x6b,
+	0x06, 0x36, 0x31, 0xc3, 0xb0, 0xf7, 0x5e, 0x11, 0x23, 0xe0, 0x4e, 0x3b, 0xf4, 0x47, 0x14, 0x61,
+	0x28, 0x46, 0xd9, 0xc5, 0x66, 0x5e, 0x1c, 0xe9, 0x9b, 0x9b, 0xec, 0x77, 0x61, 0x45, 0xa9, 0x7c,
+	0x31, 0xf3, 0x88, 0x0e, 0xc3, 0xe5, 0x1a, 0xfd, 0x08, 0xca, 0xd1, 0x16, 0xfc, 0x60, 0x64, 0x42,
+	0x1c, 0x16, 0x15, 0xe1, 0x6e, 0x28, 0xef, 0xc7, 0x62, 0xf4, 0x11, 0xa0, 0xc0, 0xf1, 0x70, 0x40,
+	0xc9, 0x80, 0x8f, 0x7e, 0x81, 0x61, 0x10, 0x1a, 0x7f, 0xad, 0x96, 0xa3, 0x37, 0xbf, 0x76, 0xfa,
+	0xa1, 0xbc, 0xf1, 0x43, 0x80, 0x64, 0x4b, 0x94, 0x07, 0xf9, 0xe8, 0xc5, 0xe9, 0x69, 0x79, 0x03,
+	0xed, 0x42, 0x51, 0x3b, 0xef, 0xe8, 0xbd, 0xb3, 0xde, 0xf9, 0xc5, 0xe1, 0x69, 0x59, 0x6a, 0xfc,
+	0x45, 0x82, 0x5d, 0x9d, 0x50, 0xe6, 0xfa, 0x64, 0x49, 0x84, 0x87, 0xb0, 0x45, 0x83, 0xc9, 0x04,
+	0xfb, 0xb3, 0x88, 0xd8, 0x57, 0x6e, 0x7a, 0xb1, 0x1d, 0xaa, 0x43, 0xd1, 0x8b, 0xe0, 0x34, 0xf3,
+	0x55, 0xf4, 0xc5, 0x9d, 0x16, 0xa1, 0x5f, 0x42, 0xf8, 0x61, 0xce, 0x3b, 0x78, 0x76, 0x95, 0x0e,
+	0xbe, 0x54, 0xff, 0xf1, 0x43, 0x48, 0x7d, 0x3b, 0x20, 0x80, 0xdc, 0x29, 0x66, 0x84, 0xb2, 0xf2,
+	0x06, 0xda, 0x82, 0xec, 0xa1, 0x6d, 0x97, 0xa5, 0xf6, 0x4f, 0x5e, 0xff, 0xab, 0xba, 0xf1, 0x7a,
+	0x51, 0x95, 0xbe, 0x5c, 0x54, 0xa5, 0xaf, 0x16, 0x55, 0xe9, 0x9f, 0x8b, 0xaa, 0xf4, 0x87, 0x37,
+	0xd5, 0x8d, 0x2f, 0xdf, 0x54, 0x37, 0xbe, 0x7a, 0x53, 0xdd, 0xf8, 0xb4, 0xb0, 0xbc, 0xc4, 0x30,
+	0x27, 0xfe, 0x47, 0x78, 0xfc, 0xbf, 0x00, 0x00, 0x00, 0xff, 0xff, 0x9c, 0xa0, 0x40, 0xc7, 0x2d,
+	0x11, 0x00, 0x00,
 }
diff --git a/pkg/ccl/backupccl/backup.proto b/pkg/ccl/backupccl/backup.proto
index bcfffc139ad8..fb6767a1b86b 100644
--- a/pkg/ccl/backupccl/backup.proto
+++ b/pkg/ccl/backupccl/backup.proto
@@ -173,6 +173,7 @@ message ScheduledBackupExecutionArgs {
   }
   BackupType backup_type = 1;
   string backup_statement = 2;
+  int64 unpause_on_success = 3;
 }
 
 // RestoreProgress is the information that the RestoreData processor sends back
diff --git a/pkg/ccl/backupccl/backup_job.go b/pkg/ccl/backupccl/backup_job.go
index 4371ac13d250..17004b138e48 100644
--- a/pkg/ccl/backupccl/backup_job.go
+++ b/pkg/ccl/backupccl/backup_job.go
@@ -23,6 +23,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
+	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
@@ -32,7 +34,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/storage/cloud"
 	"github.com/cockroachdb/cockroach/pkg/storage/cloudimpl"
@@ -571,24 +572,45 @@ func (b *backupResumer) Resume(
 		}
 	}
 
-	b.maybeNotifyScheduledJobCompletion(ctx, jobs.StatusSucceeded, p.ExecCfg().InternalExecutor)
-
+	b.maybeNotifyScheduledJobCompletion(ctx, jobs.StatusSucceeded, p.ExecCfg())
 	return nil
 }
 
 func (b *backupResumer) maybeNotifyScheduledJobCompletion(
-	ctx context.Context, jobStatus jobs.Status, ex sqlutil.InternalExecutor,
+	ctx context.Context, jobStatus jobs.Status, exec *sql.ExecutorConfig,
 ) {
-	if b.job.CreatedBy() == nil || b.job.CreatedBy().Name != jobs.CreatedByScheduledJobs {
-		return
-	}
-	info := b.job.CreatedBy()
+	env := scheduledjobs.ProdJobSchedulerEnv
+
+	if err := exec.DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+		// Do not rely on b.job containing created_by_id.  Query it directly.
+		datums, err := exec.InternalExecutor.QueryRowEx(
+			ctx,
+			"lookup-schedule-info",
+			txn,
+			sqlbase.InternalExecutorSessionDataOverride{User: security.NodeUser},
+			fmt.Sprintf(
+				"SELECT created_by_id FROM %s WHERE id=$1 AND created_by_type=$2",
+				env.SystemJobsTableName()),
+			*b.job.ID(), jobs.CreatedByScheduledJobs)
+
+		if err != nil {
+			return errors.Wrap(err, "schedule info lookup")
+		}
+		if datums == nil {
+			// Not a scheduled backup.
+			return nil
+		}
 
-	if err := jobs.NotifyJobTermination(
-		ctx, nil /* env */, *b.job.ID(), jobStatus, info.ID, ex, nil); err != nil {
-		log.Warningf(ctx,
-			"failed to notify schedule %d of completion of job %d; err=%s",
-			info.ID, *b.job.ID(), err)
+		scheduleID := int64(tree.MustBeDInt(datums[0]))
+		if err := jobs.NotifyJobTermination(
+			ctx, env, *b.job.ID(), jobStatus, scheduleID, exec.InternalExecutor, txn); err != nil {
+			log.Warningf(ctx,
+				"failed to notify schedule %d of completion of job %d; err=%s",
+				scheduleID, *b.job.ID(), err)
+		}
+		return nil
+	}); err != nil {
+		log.Errorf(ctx, "maybeNotifySchedule error: %v", err)
 	}
 }
 
@@ -615,7 +637,7 @@ func (b *backupResumer) OnFailOrCancel(ctx context.Context, phs interface{}) err
 	defer b.maybeNotifyScheduledJobCompletion(
 		ctx,
 		jobs.StatusFailed,
-		phs.(sql.PlanHookState).ExecCfg().InternalExecutor,
+		phs.(sql.PlanHookState).ExecCfg(),
 	)
 
 	telemetry.Count("backup.total.failed")
diff --git a/pkg/ccl/backupccl/create_scheduled_backup.go b/pkg/ccl/backupccl/create_scheduled_backup.go
index 711de2d926f8..6b7f29a06c8b 100644
--- a/pkg/ccl/backupccl/create_scheduled_backup.go
+++ b/pkg/ccl/backupccl/create_scheduled_backup.go
@@ -24,7 +24,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/storage/cloudimpl"
 	"github.com/cockroachdb/errors"
@@ -152,19 +151,6 @@ func computeScheduleRecurrence(
 	return &scheduleRecurrence{cron, frequency}, nil
 }
 
-var humanDurations = map[time.Duration]string{
-	time.Hour:          "hour",
-	24 * time.Hour:     "day",
-	7 * 24 * time.Hour: "week",
-}
-
-func (r *scheduleRecurrence) Humanize() string {
-	if d, ok := humanDurations[r.frequency]; ok {
-		return "every " + d
-	}
-	return "every " + r.frequency.String()
-}
-
 var forceFullBackup *scheduleRecurrence
 
 func pickFullRecurrenceFromIncremental(inc *scheduleRecurrence) *scheduleRecurrence {
@@ -319,61 +305,74 @@ func doCreateBackupSchedules(
 
 	ex := p.ExecCfg().InternalExecutor
 	return p.ExecCfg().DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
-		// Create FULL backup schedule.
-		fullFirstRun := firstRun
-		if eval.isEnterpriseUser && fullFirstRun == nil && fullRecurrencePicked {
-			// The enterprise user did not indicate preference when to run full backups,
-			// and we picked the schedule ourselves.
-			// Run full backup immediately so that we do not wind up waiting for a long
-			// time before the first full backup runs.  Without full backup, we can't
-			// execute incrementals.
-			now := env.Now()
-			fullFirstRun = &now
-		}
-
-		if err := createBackupSchedule(
-			ctx, env, p.User(), fullScheduleName, fullRecurrence,
-			fullFirstRun, details, backupNode, resultsCh, ex, txn,
-		); err != nil {
-			return err
-		}
+		unpauseOnSuccessID := jobs.InvalidScheduleID
 
 		// If needed, create incremental.
 		if incRecurrence != nil {
 			backupNode.AppendToLatest = true
+			inc, err := makeBackupSchedule(
+				env, p.User(), fullScheduleName+": INCREMENTAL",
+				incRecurrence, details, unpauseOnSuccessID, backupNode)
 
-			if err := createBackupSchedule(
-				ctx, env, p.User(), fullScheduleName+": INCREMENTAL", incRecurrence,
-				firstRun, details, backupNode, resultsCh, ex, txn,
-			); err != nil {
+			if err != nil {
 				return err
 			}
+			// Incremental is paused until FULL completes.
+			inc.Pause("Waiting for initial backup to complete")
+
+			if err := inc.Create(ctx, ex, txn); err != nil {
+				return err
+			}
+			if err := emitSchedule(inc, tree.AsString(backupNode), resultsCh); err != nil {
+				return err
+			}
+			unpauseOnSuccessID = inc.ScheduleID()
 		}
 
-		return nil
-	})
+		// Create FULL backup schedule.
+		backupNode.AppendToLatest = false
+		fullBackupStmt := tree.AsString(backupNode)
+		full, err := makeBackupSchedule(
+			env, p.User(), fullScheduleName,
+			fullRecurrence, details, unpauseOnSuccessID, backupNode)
+		if err != nil {
+			return err
+		}
 
+		if firstRun != nil {
+			full.SetNextRun(*firstRun)
+		} else if eval.isEnterpriseUser && fullRecurrencePicked {
+			// The enterprise user did not indicate preference when to run full backups,
+			// and we picked the schedule ourselves.
+			// Run full backup immediately so that we do not wind up waiting for a long
+			// time before the first full backup runs.  Without full backup, we can't
+			// execute incremental.
+			full.SetNextRun(env.Now())
+		}
+
+		// Create the schedule (we need its ID to create incremental below).
+		if err := full.Create(ctx, ex, txn); err != nil {
+			return err
+		}
+		return emitSchedule(full, fullBackupStmt, resultsCh)
+	})
 }
 
-func createBackupSchedule(
-	ctx context.Context,
+func makeBackupSchedule(
 	env scheduledjobs.JobSchedulerEnv,
 	owner string,
 	name string,
 	recurrence *scheduleRecurrence,
-	firstRun *time.Time,
 	details jobspb.ScheduleDetails,
+	unpauseOnSuccess int64,
 	backupNode *tree.Backup,
-	resultsCh chan<- tree.Datums,
-	ex sqlutil.InternalExecutor,
-	txn *kv.Txn,
-) error {
+) (*jobs.ScheduledJob, error) {
 	sj := jobs.NewScheduledJob(env)
 	sj.SetScheduleName(name)
 	sj.SetOwner(owner)
 
 	// Prepare arguments for scheduled backup execution.
-	args := &ScheduledBackupExecutionArgs{}
+	args := &ScheduledBackupExecutionArgs{UnpauseOnSuccess: unpauseOnSuccess}
 	if backupNode.AppendToLatest {
 		args.BackupType = ScheduledBackupExecutionArgs_INCREMENTAL
 	} else {
@@ -381,13 +380,10 @@ func createBackupSchedule(
 	}
 
 	if err := sj.SetSchedule(recurrence.cron); err != nil {
-		return err
+		return nil, err
 	}
 
 	sj.SetScheduleDetails(details)
-	if firstRun != nil {
-		sj.SetNextRun(*firstRun)
-	}
 
 	// TODO(yevgeniy): Validate backup schedule:
 	//  * Verify targets exist.  Provide a way for user to override this via option.
@@ -398,21 +394,25 @@ func createBackupSchedule(
 	args.BackupStatement = tree.AsString(backupNode)
 	any, err := pbtypes.MarshalAny(args)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	sj.SetExecutionDetails(
 		tree.ScheduledBackupExecutor.InternalName(),
 		jobspb.ExecutionArguments{Args: any},
 	)
 
-	// Create the schedule.
-	if err := sj.Create(ctx, ex, txn); err != nil {
-		return err
-	}
+	return sj, nil
+}
 
+func emitSchedule(sj *jobs.ScheduledJob, backupStmt string, resultsCh chan<- tree.Datums) error {
 	var nextRun tree.Datum
+	status := "ACTIVE"
 	if sj.IsPaused() {
 		nextRun = tree.DNull
+		status = "PAUSED"
+		if reason := sj.LastChangeReason(); reason != "" {
+			status += ": " + reason
+		}
 	} else {
 		next, err := tree.MakeDTimestampTZ(sj.NextRun(), time.Microsecond)
 		if err != nil {
@@ -423,10 +423,11 @@ func createBackupSchedule(
 
 	resultsCh <- tree.Datums{
 		tree.NewDInt(tree.DInt(sj.ScheduleID())),
-		tree.NewDString(name),
+		tree.NewDString(sj.ScheduleName()),
+		tree.NewDString(status),
 		nextRun,
-		tree.NewDString(recurrence.Humanize()),
-		tree.NewDString(tree.AsString(backupNode)),
+		tree.NewDString(sj.ScheduleExpr()),
+		tree.NewDString(backupStmt),
 	}
 	return nil
 }
@@ -507,8 +508,9 @@ func makeScheduledBackupEval(
 var scheduledBackupHeader = sqlbase.ResultColumns{
 	{Name: "schedule_id", Typ: types.Int},
 	{Name: "name", Typ: types.String},
-	{Name: "next_run", Typ: types.TimestampTZ},
-	{Name: "frequency", Typ: types.String},
+	{Name: "status", Typ: types.String},
+	{Name: "first_run", Typ: types.TimestampTZ},
+	{Name: "schedule", Typ: types.String},
 	{Name: "backup_stmt", Typ: types.String},
 }
 
diff --git a/pkg/ccl/backupccl/create_scheduled_backup_test.go b/pkg/ccl/backupccl/create_scheduled_backup_test.go
index f28136952f90..64bc4582520f 100644
--- a/pkg/ccl/backupccl/create_scheduled_backup_test.go
+++ b/pkg/ccl/backupccl/create_scheduled_backup_test.go
@@ -130,7 +130,7 @@ func (h *testHelper) createBackupSchedule(
 	var schedules []*jobs.ScheduledJob
 	for rows.Next() {
 		var id int64
-		require.NoError(t, rows.Scan(&id, &unusedStr, &unusedTS, &unusedStr, &unusedStr))
+		require.NoError(t, rows.Scan(&id, &unusedStr, &unusedStr, &unusedTS, &unusedStr, &unusedStr))
 		// Query system.scheduled_job table and load those schedules.
 		datums, cols, err := h.cfg.InternalExecutor.QueryWithCols(
 			context.Background(), "sched-load", nil,
@@ -181,6 +181,7 @@ func TestSerializesScheduledBackupExecutionArgs(t *testing.T) {
 		period     time.Duration
 		runsNow    bool
 		shownStmt  string
+		paused     bool
 	}
 
 	testCases := []struct {
@@ -237,6 +238,7 @@ func TestSerializesScheduledBackupExecutionArgs(t *testing.T) {
 					nameRe:     "BACKUP .*: INCREMENTAL",
 					backupStmt: "BACKUP INTO LATEST IN 'somewhere' WITH detached",
 					period:     time.Hour,
+					paused:     true,
 				},
 				{
 					nameRe:     "BACKUP .+",
@@ -255,6 +257,7 @@ func TestSerializesScheduledBackupExecutionArgs(t *testing.T) {
 					nameRe:     "my-backup: INCREMENTAL",
 					backupStmt: "BACKUP INTO LATEST IN 'somewhere' WITH detached",
 					period:     time.Hour,
+					paused:     true,
 				},
 				{
 					nameRe:     "my-backup",
@@ -304,6 +307,7 @@ func TestSerializesScheduledBackupExecutionArgs(t *testing.T) {
 					nameRe:     "my_backup_name: INCREMENTAL",
 					backupStmt: "BACKUP INTO LATEST IN 'somewhere' WITH revision_history, detached",
 					period:     time.Hour,
+					paused:     true,
 				},
 				{
 					nameRe:     "my_backup_name",
@@ -411,6 +415,7 @@ func TestSerializesScheduledBackupExecutionArgs(t *testing.T) {
 				require.NoError(t, err)
 				require.EqualValues(t, expectedSchedule.period, frequency, expectedSchedule)
 
+				require.Equal(t, expectedSchedule.paused, s.IsPaused())
 				if expectedSchedule.runsNow {
 					require.EqualValues(t, th.env.Now().Round(time.Microsecond), s.ScheduledRunTime())
 				}
@@ -475,7 +480,7 @@ INSERT INTO t1 values (-1), (10), (-100);
 	}{
 		{
 			name:     "cluster-backup",
-			schedule: "CREATE SCHEDULE FOR BACKUP INTO $1 RECURRING '@hourly' FULL BACKUP ALWAYS",
+			schedule: "CREATE SCHEDULE FOR BACKUP INTO $1 RECURRING '@hourly'",
 			verifyTables: expectBackupTables(
 				dbTables{"db", []string{"t1", "t2", "t3"}},
 				dbTables{"other_db", []string{"t1"}},
@@ -502,18 +507,52 @@ INSERT INTO t1 values (-1), (10), (-100);
 			destination := "nodelocal://0/backup/" + tc.name
 			schedules, err := th.createBackupSchedule(t, tc.schedule, destination)
 			require.NoError(t, err)
-			require.EqualValues(t, 1, len(schedules))
-			defer th.sqlDB.Exec(t, "DROP SCHEDULE $1", schedules[0].ScheduleID())
+			require.LessOrEqual(t, 1, len(schedules))
+
+			// Either 1 or two schedules will be created.
+			// One of them (incremental) must be paused.
+			var full, inc *jobs.ScheduledJob
+			if len(schedules) == 1 {
+				full = schedules[0]
+			} else {
+				require.Equal(t, 2, len(schedules))
+				full, inc = schedules[0], schedules[1]
+				if full.IsPaused() {
+					full, inc = inc, full // Swap: inc should be paused.
+				}
+				require.True(t, inc.IsPaused())
+				require.False(t, full.IsPaused())
+
+				// The full should list incremental as a schedule to unpause.
+				args := &ScheduledBackupExecutionArgs{}
+				require.NoError(t, pbtypes.UnmarshalAny(full.ExecutionArgs().Args, args))
+				require.EqualValues(t, inc.ScheduleID(), args.UnpauseOnSuccess)
+			}
+
+			defer func() {
+				th.sqlDB.Exec(t, "DROP SCHEDULE $1", full.ScheduleID())
+				if inc != nil {
+					th.sqlDB.Exec(t, "DROP SCHEDULE $1", inc.ScheduleID())
+				}
+			}()
 
 			// Force the schedule to execute.
-			th.env.SetTime(schedules[0].NextRun().Add(time.Second))
+			th.env.SetTime(full.NextRun().Add(time.Second))
 			require.NoError(t,
 				th.cfg.DB.Txn(context.Background(), func(ctx context.Context, txn *kv.Txn) error {
 					return th.executeSchedules(ctx, allSchedules, txn)
 				}))
 
 			// Wait for the backup complete.
-			th.waitForSuccessfulScheduledJob(t, schedules[0].ScheduleID())
+			th.waitForSuccessfulScheduledJob(t, full.ScheduleID())
+
+			if inc != nil {
+				// Once the full backup completes, the incremental one should no longer be paused.
+				loadedInc, err := jobs.LoadScheduledJob(
+					context.Background(), th.env, inc.ScheduleID(), th.cfg.InternalExecutor, nil)
+				require.NoError(t, err)
+				require.False(t, loadedInc.IsPaused())
+			}
 
 			// Verify backup.
 			latest, err := ioutil.ReadFile(path.Join(th.iodir, "backup", tc.name, latestFileName))
diff --git a/pkg/ccl/backupccl/schedule_exec.go b/pkg/ccl/backupccl/schedule_exec.go
index a8ec567c851a..49718f0530de 100644
--- a/pkg/ccl/backupccl/schedule_exec.go
+++ b/pkg/ccl/backupccl/schedule_exec.go
@@ -14,11 +14,13 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
+	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
@@ -122,18 +124,68 @@ func (e *scheduledBackupExecutor) executeBackup(
 
 // NotifyJobTermination implements jobs.ScheduledJobExecutor interface.
 func (e *scheduledBackupExecutor) NotifyJobTermination(
-	ctx context.Context, jobID int64, jobStatus jobs.Status, schedule *jobs.ScheduledJob, _ *kv.Txn,
+	ctx context.Context,
+	jobID int64,
+	jobStatus jobs.Status,
+	env scheduledjobs.JobSchedulerEnv,
+	schedule *jobs.ScheduledJob,
+	ex sqlutil.InternalExecutor,
+	txn *kv.Txn,
 ) error {
 	if jobStatus == jobs.StatusSucceeded {
 		e.metrics.NumSucceeded.Inc(1)
-	} else {
-		err := errors.Errorf(
-			"backup job %d scheduled by %d failed with status %s",
-			jobID, schedule.ScheduleID(), jobStatus)
-		log.Errorf(ctx, "backup error: %v	", err)
-		e.metrics.NumFailed.Inc(1)
-		jobs.DefaultHandleFailedRun(schedule, jobID, err)
+		log.Infof(ctx, "backup job %d scheduled by %d succeeded", jobID, schedule.ScheduleID())
+		return e.backupSucceeded(ctx, schedule, env, ex, txn)
+	}
+
+	e.metrics.NumFailed.Inc(1)
+	err := errors.Errorf(
+		"backup job %d scheduled by %d failed with status %s",
+		jobID, schedule.ScheduleID(), jobStatus)
+	log.Errorf(ctx, "backup error: %v	", err)
+	jobs.DefaultHandleFailedRun(schedule, jobID, err)
+
+	return nil
+}
+
+func (e *scheduledBackupExecutor) backupSucceeded(
+	ctx context.Context,
+	schedule *jobs.ScheduledJob,
+	env scheduledjobs.JobSchedulerEnv,
+	ex sqlutil.InternalExecutor,
+	txn *kv.Txn,
+) error {
+	args := &ScheduledBackupExecutionArgs{}
+	if err := pbtypes.UnmarshalAny(schedule.ExecutionArgs().Args, args); err != nil {
+		return errors.Wrap(err, "un-marshaling args")
 	}
+
+	if args.UnpauseOnSuccess == jobs.InvalidScheduleID {
+		return nil
+	}
+
+	s, err := jobs.LoadScheduledJob(ctx, env, args.UnpauseOnSuccess, ex, txn)
+	if err != nil {
+		return err
+	}
+	if err := s.Unpause("FULL backup completed"); err != nil {
+		return err
+	}
+	if err := s.Update(ctx, ex, txn); err != nil {
+		return err
+	}
+
+	// Clear UnpauseOnSuccess; caller updates schedule.
+	args.UnpauseOnSuccess = jobs.InvalidScheduleID
+	any, err := pbtypes.MarshalAny(args)
+	if err != nil {
+		return errors.Wrap(err, "marshaling args")
+	}
+	schedule.SetExecutionDetails(
+		schedule.ExecutorType(),
+		jobspb.ExecutionArguments{Args: any},
+	)
+
 	return nil
 }
 
diff --git a/pkg/jobs/executor_impl.go b/pkg/jobs/executor_impl.go
index ec9d2ca105d7..9bfab2b282c9 100644
--- a/pkg/jobs/executor_impl.go
+++ b/pkg/jobs/executor_impl.go
@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
 	"github.com/cockroachdb/cockroach/pkg/security"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/errors"
 	"github.com/gogo/protobuf/types"
@@ -70,7 +71,13 @@ func (e *inlineScheduledJobExecutor) ExecuteJob(
 
 // NotifyJobTermination implements ScheduledJobExecutor interface.
 func (e *inlineScheduledJobExecutor) NotifyJobTermination(
-	ctx context.Context, jobID int64, jobStatus Status, schedule *ScheduledJob, _ *kv.Txn,
+	ctx context.Context,
+	jobID int64,
+	jobStatus Status,
+	env scheduledjobs.JobSchedulerEnv,
+	schedule *ScheduledJob,
+	ex sqlutil.InternalExecutor,
+	txn *kv.Txn,
 ) error {
 	// For now, only interested in failed status.
 	if jobStatus == StatusFailed {
diff --git a/pkg/jobs/job_scheduler_test.go b/pkg/jobs/job_scheduler_test.go
index b1810ccec617..509d1b4b821c 100644
--- a/pkg/jobs/job_scheduler_test.go
+++ b/pkg/jobs/job_scheduler_test.go
@@ -232,7 +232,13 @@ func (n *recordScheduleExecutor) ExecuteJob(
 }
 
 func (n *recordScheduleExecutor) NotifyJobTermination(
-	ctx context.Context, jobID int64, jobStatus Status, schedule *ScheduledJob, txn *kv.Txn,
+	ctx context.Context,
+	jobID int64,
+	jobStatus Status,
+	env scheduledjobs.JobSchedulerEnv,
+	schedule *ScheduledJob,
+	ex sqlutil.InternalExecutor,
+	txn *kv.Txn,
 ) error {
 	return nil
 }
diff --git a/pkg/jobs/scheduled_job.go b/pkg/jobs/scheduled_job.go
index d27caea035b3..b2324c5be424 100644
--- a/pkg/jobs/scheduled_job.go
+++ b/pkg/jobs/scheduled_job.go
@@ -46,6 +46,9 @@ type scheduledJobRecord struct {
 	ScheduleChanges jobspb.ScheduleChangeInfo `col:"schedule_changes"`
 }
 
+// InvalidScheduleID is a constant indicating the schedule ID is not valid.
+const InvalidScheduleID int64 = 0
+
 // ScheduledJob  is a representation of the scheduled job.
 // This struct can marshal/unmarshal changes made to the underlying system.scheduled_job table.
 type ScheduledJob struct {
@@ -72,6 +75,36 @@ func NewScheduledJob(env scheduledjobs.JobSchedulerEnv) *ScheduledJob {
 	}
 }
 
+// LoadScheduledJob loads scheduled job record from the database.
+func LoadScheduledJob(
+	ctx context.Context,
+	env scheduledjobs.JobSchedulerEnv,
+	id int64,
+	ex sqlutil.InternalExecutor,
+	txn *kv.Txn,
+) (*ScheduledJob, error) {
+	rows, cols, err := ex.QueryWithCols(ctx, "lookup-schedule", txn,
+		sqlbase.InternalExecutorSessionDataOverride{User: security.RootUser},
+		fmt.Sprintf("SELECT * FROM %s WHERE schedule_id = %d",
+			env.ScheduledJobsTableName(), id))
+
+	if err != nil {
+		return nil, err
+	}
+
+	if len(rows) != 1 {
+		return nil, errors.Newf(
+			"expected to find 1 schedule, found %d with schedule_id=%d",
+			len(rows), id)
+	}
+
+	j := NewScheduledJob(env)
+	if err := j.InitFromDatums(rows[0], cols); err != nil {
+		return nil, err
+	}
+	return j, nil
+}
+
 // ScheduleID returns schedule ID.
 func (j *ScheduledJob) ScheduleID() int64 {
 	return j.rec.ScheduleID
@@ -203,6 +236,20 @@ func (j *ScheduledJob) AddScheduleChangeReason(reasonFmt string, args ...interfa
 	j.markDirty("schedule_changes")
 }
 
+// LastChangeReason returns the last schedule change reason.
+func (j *ScheduledJob) LastChangeReason() string {
+	l := len(j.rec.ScheduleChanges.Changes)
+	if l > 0 {
+		return j.rec.ScheduleChanges.Changes[l-1].Reason
+	}
+	return ""
+}
+
+// ScheduleExpr returns the schedule expression for this schedule.
+func (j *ScheduledJob) ScheduleExpr() string {
+	return j.rec.ScheduleExpr
+}
+
 // Pause pauses this schedule.
 func (j *ScheduledJob) Pause(reason string) {
 	j.rec.NextRun = time.Time{}
diff --git a/pkg/jobs/scheduled_job_executor.go b/pkg/jobs/scheduled_job_executor.go
index 00dc90465dbd..4fd75a3a6264 100644
--- a/pkg/jobs/scheduled_job_executor.go
+++ b/pkg/jobs/scheduled_job_executor.go
@@ -17,8 +17,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
-	"github.com/cockroachdb/cockroach/pkg/security"
-	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/errors"
@@ -43,7 +41,9 @@ type ScheduledJobExecutor interface {
 		ctx context.Context,
 		jobID int64,
 		jobStatus Status,
+		env scheduledjobs.JobSchedulerEnv,
 		schedule *ScheduledJob,
+		ex sqlutil.InternalExecutor,
 		txn *kv.Txn,
 	) error
 
@@ -108,15 +108,17 @@ func NotifyJobTermination(
 		env = scheduledjobs.ProdJobSchedulerEnv
 	}
 
-	// Get the executor for this schedule.
-	schedule, executor, err := lookupScheduleAndExecutor(
-		ctx, env, ex, scheduleID, txn)
+	schedule, err := LoadScheduledJob(ctx, env, scheduleID, ex, txn)
+	if err != nil {
+		return err
+	}
+	executor, err := NewScheduledJobExecutor(schedule.ExecutorType())
 	if err != nil {
 		return err
 	}
 
 	// Delegate handling of the job termination to the executor.
-	err = executor.NotifyJobTermination(ctx, jobID, jobStatus, schedule, txn)
+	err = executor.NotifyJobTermination(ctx, jobID, jobStatus, env, schedule, ex, txn)
 	if err != nil {
 		return err
 	}
@@ -124,37 +126,3 @@ func NotifyJobTermination(
 	// Update this schedule in case executor made changes to it.
 	return schedule.Update(ctx, ex, txn)
 }
-
-func lookupScheduleAndExecutor(
-	ctx context.Context,
-	env scheduledjobs.JobSchedulerEnv,
-	ex sqlutil.InternalExecutor,
-	scheduleID int64,
-	txn *kv.Txn,
-) (*ScheduledJob, ScheduledJobExecutor, error) {
-	rows, cols, err := ex.QueryWithCols(ctx, "lookup-schedule", txn,
-		sqlbase.InternalExecutorSessionDataOverride{User: security.RootUser},
-		fmt.Sprintf(
-			"SELECT schedule_id, schedule_details, executor_type FROM %s WHERE schedule_id = %d",
-			env.ScheduledJobsTableName(), scheduleID))
-
-	if err != nil {
-		return nil, nil, err
-	}
-
-	if len(rows) != 1 {
-		return nil, nil, errors.Newf(
-			"expected to find 1 schedule, found %d with schedule_id=%d",
-			len(rows), scheduleID)
-	}
-
-	j := NewScheduledJob(env)
-	if err := j.InitFromDatums(rows[0], cols); err != nil {
-		return nil, nil, err
-	}
-	executor, err := NewScheduledJobExecutor(j.ExecutorType())
-	if err == nil {
-		return j, executor, nil
-	}
-	return nil, nil, err
-}
diff --git a/pkg/jobs/scheduled_job_executor_test.go b/pkg/jobs/scheduled_job_executor_test.go
index cf9abda4e3db..279d35bf9837 100644
--- a/pkg/jobs/scheduled_job_executor_test.go
+++ b/pkg/jobs/scheduled_job_executor_test.go
@@ -16,6 +16,7 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlutil"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
@@ -39,7 +40,13 @@ func (s *statusTrackingExecutor) ExecuteJob(
 }
 
 func (s *statusTrackingExecutor) NotifyJobTermination(
-	ctx context.Context, jobID int64, jobStatus Status, schedule *ScheduledJob, txn *kv.Txn,
+	ctx context.Context,
+	jobID int64,
+	jobStatus Status,
+	env scheduledjobs.JobSchedulerEnv,
+	schedule *ScheduledJob,
+	ex sqlutil.InternalExecutor,
+	txn *kv.Txn,
 ) error {
 	s.counts[jobStatus]++
 	return nil

From 71288b9bc49342d873b9440e728735fea4115aec Mon Sep 17 00:00:00 2001
From: Spas Bojanov <spas@cockroachlabs.com>
Date: Wed, 19 Aug 2020 11:57:22 +0300
Subject: [PATCH 3/3] jobs: unskip TestRegistryLifecycle/rollback

The test flakiness was introduced by #52697and fixed by #52710.

Fixes #52767.

Release note: none.
---
 pkg/jobs/jobs_test.go | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/pkg/jobs/jobs_test.go b/pkg/jobs/jobs_test.go
index 196605e38383..7992ccf5db36 100644
--- a/pkg/jobs/jobs_test.go
+++ b/pkg/jobs/jobs_test.go
@@ -34,7 +34,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/tests"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
-	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
@@ -545,8 +544,6 @@ func TestRegistryLifecycle(t *testing.T) {
 
 	// Verify that pause and cancel in a rollback do nothing.
 	t.Run("rollback", func(t *testing.T) {
-		skip.WithIssue(t, 52767, "flaky")
-
 		rts := registryTestSuite{}
 		rts.setUp(t)
 		defer rts.tearDown()