You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: schedule_pts_chaining_test.go
    path of file: ./repos/cockroach/pkg/ccl/backupccl
    
    the code of the file until where you have to start completion: // Copyright 2021 The Cockroach Authors.
//
// Licensed as a CockroachDB Enterprise file under the Cockroach Community
// License (the "License"); you may not use this file except in compliance with
// the License. You may obtain a copy of the License at
//
//     https://github.com/cockroachdb/cockroach/blob/master/licenses/CCL.txt

package backupccl

import (
	"context"
	"fmt"
	"regexp"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/cockroachdb/cockroach/pkg/jobs"
	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
	"github.com/cockroachdb/cockroach/pkg/jobs/jobsprotectedts"
	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts"
	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/protectedts/ptpb"
	"github.com/cockroachdb/cockroach/pkg/sql/isql"
	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
	"github.com/cockroachdb/cockroach/pkg/testutils"
	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
	"github.com/cockroachdb/cockroach/pkg/util/hlc"
	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
	"github.com/cockroachdb/cockroach/pkg/util/log"
	"github.com/cockroachdb/cockroach/pkg/util/uuid"
	"github.com/cockroachdb/errors"
	"github.com/stretchr/testify/require"
)

// Create backup schedules for this test.
// Returns schedule IDs for full and incremental schedules, plus a cleanup function.
func (th *testHelper) createSchedules(
	t *testing.T, backupStmt string, opts ...string,
) (jobspb.ScheduleID, jobspb.ScheduleID, func()) {
	backupOpts := ""
	if len(opts) > 0 {
		backupOpts = fmt.Sprintf(" WITH %s", strings.Join(opts, ", "))
	}
	backupQuery := fmt.Sprintf("%s%s", backupStmt, backupOpts)
	schedules, err := th.createBackupSchedule(t,
		fmt.Sprintf("CREATE SCHEDULE FOR %s RECURRING '*/2 * * * *'", backupQuery))
	require.NoError(t, err)

	// We expect full & incremental schedule to be created.
	require.Equal(t, 2, len(schedules))

	// Order schedules so that the full schedule is the first one
	fullID, incID := schedules[0].ScheduleID(), schedules[1].ScheduleID()
	if sched
    
func TestScheduleChainingEdgeCases(t *testing.T) {
	defer leaktest.AfterTest(t)()
	defer log.Scope(t).Close(t)

	th, cleanup := newTestHelper(t)
	defer cleanup()

	th.sqlDB.Exec(t, `
CREATE DATABASE db;
USE db;
CREATE TABLE t(a int);
INSERT INTO t values (1), (10), (100);
`)

	backupAsOfTimes := make([]time.Time, 0)
	th.cfg.TestingKnobs.(*jobs.TestingKnobs).OverrideAsOfClause = func(clause *tree.AsOfClause, _ time.Time) {
		backupAsOfTime := th.cfg.DB.KV().Clock().PhysicalTime()
		expr, err := tree.MakeDTimestampTZ(backupAsOfTime, time.Microsecond)
		require.NoError(t, err)
		clause.Expr = expr
		backupAsOfTimes = append(backupAsOfTimes, backupAsOfTime)
	}

	runSchedule := func(t *testing.T, schedule *jobs.ScheduledJob) {
		th.env.SetTime(schedule.NextRun().Add(time.Second))
		require.NoError(t, th.executeSchedules())
		th.waitForSuccessfulScheduledJob(t, schedule.ScheduleID())
	}

	clusterBackupStmt := `BACKUP INTO 'nodelocal://1/%s'`

	ctx := context.Background()

	t.Run("inc schedule is dropped", func(t *testing.T) {
		fullID, incID, cleanupSchedules := th.createSchedules(t, fmt.Sprintf(clusterBackupStmt, "foo"))
		defer cleanupSchedules()
		defer func() { backupAsOfTimes = backupAsOfTimes[:0] }()

		th.sqlDB.Exec(t, `DROP SCHEDULE $1`, incID)

		fullSchedule := th.loadSchedule(t, fullID)
		// Force full backup to execute (this unpauses incremental, which should be a no-op).
		runSchedule(t, fullSchedule)

		var numRows int
		th.sqlDB.QueryRow(t, `SELECT count(*) FROM system.protected_ts_records`).Scan(&numRows)
		require.Zero(t, numRows)
	})

	t.Run("full schedule is dropped after first run", func(t *testing.T) {
		fullID, incID, cleanupSchedules := th.createSchedules(t, fmt.Sprintf(clusterBackupStmt, "bar"))
		defer cleanupSchedules()
		defer func() { backupAsOfTimes = backupAsOfTimes[:0] }()

		schedules := jobs.ScheduledJobDB(th.internalDB())

		fullSchedule := th.loadSchedule(t, fullID)
		_, fullArgs, err := getScheduledBackupExecutionArgsFromSchedule(
			ctx, th.env, schedules, fullID,
		)
		require.NoError(t, err)
		// Force full backup to execute (this unpauses incremental).
		runSchedule(t, fullSchedule)

		// Check that there is no PTS record on the full schedule.
		require.Nil(t, fullArgs.ProtectedTimestampRecord)

		// Check that there is a PTS record on the incremental schedule.
		incSchedule := th.loadSchedule(t, incID)
		_, incArgs, err := getScheduledBackupExecutionArgsFromSchedule(
			ctx, th.env, schedules, incID,
		)
		require.NoError(t, err)
		ptsOnIncID := incArgs.ProtectedTimestampRecord
		require.NotNil(t, ptsOnIncID)
		checkPTSRecord(ctx, t, th, *ptsOnIncID, incSchedule,
			hlc.Timestamp{WallTime: backupAsOfTimes[0].Round(time.Microsecond).UnixNano()})

		th.sqlDB.Exec(t, `DROP SCHEDULE $1`, fullID)

		_, err = schedules.Load(ctx, th.env, incID)
		require.Error(t, err)

		// Check that the incremental schedule's PTS is dropped
		require.NoError(t, th.cfg.DB.Txn(ctx, func(ctx context.Context, txn isql.Txn) error {
			pts := th.protectedTimestamps().WithTxn(txn)
			_, err := pts.GetRecord(ctx, *ptsOnIncID)
			require.True(t, errors.Is(err, protectedts.ErrNotExists))
			return nil
		}))
	})
}