changefeedccl: warn user when creating changefeed

with an older cursor

Older but valid cursors can cause changefeeds to lag behind a lot.
To ensure that customers are not accidentally using a cursor that's too
old, we now provide a warning if the provided cursor is older than 5
hours. This warning would not show if the changefeed option is
initial_scan='only' as the changefeed would never lag behind with
this option.

The warning looks like:
NOTICE: the provided cursor is 7 hours old; older cursors can result in
increased changefeed latency

Resolves #124290
Epic: CRDB-32401

Release note (general): Provide a warning when creating an enterprise
changefeed with a cursor older than 5 hours

Author: KeithCh

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -133,6 +133,49 @@ func changefeedTypeCheck(
 	return true, withSinkHeader, nil
 }
 
+func maybeShowCursorAgeWarning(
+	ctx context.Context, p sql.PlanHookState, opts changefeedbase.StatementOptions,
+) error {
+	st, err := opts.GetInitialScanType()
+	if err != nil {
+		return err
+	}
+
+	if !opts.HasStartCursor() || st == changefeedbase.OnlyInitialScan {
+		return nil
+	}
+	statementTS := p.ExtendedEvalContext().GetStmtTimestamp().UnixNano()
+	cursorTS, err := evalCursor(ctx, p, hlc.Timestamp{WallTime: statementTS}, opts.GetCursor())
+	if err != nil {
+		return err
+	}
+
+	warningAge := int64(5 * time.Hour)
+	cursorAge := func() int64 {
+		knobs, _ := p.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs)
+		if knobs != nil && knobs.OverrideCursorAge != nil {
+			return knobs.OverrideCursorAge()
+		}
+		return statementTS - cursorTS.WallTime
+	}()
+
+	if cursorAge > warningAge {
+		err = p.SendClientNotice(ctx,
+			pgnotice.Newf(
+				`the provided cursor is %d hours old; `+
+					`older cursors can result in increased changefeed latency`,
+				cursorAge/int64(time.Hour),
+			),
+			true,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // changefeedPlanHook implements sql.planHookFn.
 func changefeedPlanHook(
 	ctx context.Context, stmt tree.Statement, p sql.PlanHookState,
@@ -229,6 +272,9 @@ func changefeedPlanHook(
 
 			telemetry.Count(`changefeed.create.core`)
 			logChangefeedCreateTelemetry(ctx, jr, changefeedStmt.Select != nil)
+			if err := maybeShowCursorAgeWarning(ctx, p, opts); err != nil {
+				return err
+			}
 
 			err := coreChangefeed(ctx, p, details, description, progress, resultsCh)
 			// TODO(yevgeniy): This seems wrong -- core changefeeds always terminate
@@ -314,6 +360,10 @@ func changefeedPlanHook(
 			return err
 		}
 
+		if err := maybeShowCursorAgeWarning(ctx, p, opts); err != nil {
+			return err
+		}
+
 		logChangefeedCreateTelemetry(ctx, jr, changefeedStmt.Select != nil)
 
 		select {
@@ -388,6 +438,22 @@ func coreChangefeed(
 	}
 }
 
+func evalCursor(
+	ctx context.Context, p sql.PlanHookState, statementTime hlc.Timestamp, timeString string,
+) (hlc.Timestamp, error) {
+	if knobs, ok := p.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs); ok {
+		if knobs != nil && knobs.OverrideCursor != nil {
+			timeString = knobs.OverrideCursor(&statementTime)
+		}
+	}
+	asOfClause := tree.AsOfClause{Expr: tree.NewStrVal(timeString)}
+	asOf, err := p.EvalAsOfTimestamp(ctx, asOfClause)
+	if err != nil {
+		return hlc.Timestamp{}, err
+	}
+	return asOf.Timestamp, nil
+}
+
 func createChangefeedJobRecord(
 	ctx context.Context,
 	p sql.PlanHookState,
@@ -408,22 +474,10 @@ func createChangefeedJobRecord(
 		WallTime: p.ExtendedEvalContext().GetStmtTimestamp().UnixNano(),
 	}
 	var initialHighWater hlc.Timestamp
-	evalTimestamp := func(s string) (hlc.Timestamp, error) {
-		if knobs, ok := p.ExecCfg().DistSQLSrv.TestingKnobs.Changefeed.(*TestingKnobs); ok {
-			if knobs != nil && knobs.OverrideCursor != nil {
-				s = knobs.OverrideCursor(&statementTime)
-			}
-		}
-		asOfClause := tree.AsOfClause{Expr: tree.NewStrVal(s)}
-		asOf, err := p.EvalAsOfTimestamp(ctx, asOfClause)
-		if err != nil {
-			return hlc.Timestamp{}, err
-		}
-		return asOf.Timestamp, nil
-	}
+
 	if opts.HasStartCursor() {
 		var err error
-		initialHighWater, err = evalTimestamp(opts.GetCursor())
+		initialHighWater, err = evalCursor(ctx, p, statementTime, opts.GetCursor())
 		if err != nil {
 			return nil, err
 		}

pkg/ccl/changefeedccl/changefeed_test.go

@@ -5864,6 +5864,62 @@ func TestChangefeedOutdatedCursor(t *testing.T) {
 	cdcTestWithSystem(t, testFn, feedTestNoTenants)
 }
 
+// TestChangefeedCursorWarning ensures that we show a warning if
+// any of the tables we're creating a changefeed is past
+// the warning threshold.
+func TestChangefeedCursorAgeWarning(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	var cursorAges = []time.Duration{
+		time.Hour,
+		6 * time.Hour,
+	}
+
+	testutils.RunValues(t, "cursor age", cursorAges, func(t *testing.T, cursorAge time.Duration) {
+		s, stopServer := makeServer(t)
+		defer stopServer()
+		knobs := s.TestingKnobs.
+			DistSQL.(*execinfra.TestingKnobs).
+			Changefeed.(*TestingKnobs)
+		knobs.OverrideCursorAge = func() int64 {
+			return int64(cursorAge)
+		}
+
+		warning := fmt.Sprintf(
+			"the provided cursor is %d hours old; older cursors can result in increased changefeed latency",
+			int64(cursorAge/time.Hour))
+		noWarning := "(no notice)"
+
+		expectedWarning := func(initial_scan string) string {
+			if cursorAge == time.Hour || initial_scan == "only" {
+				return noWarning
+			}
+			return warning
+		}
+
+		sqlDB := sqlutils.MakeSQLRunner(s.DB)
+		sqlDB.Exec(t, `CREATE TABLE f (a INT PRIMARY KEY)`)
+		sqlDB.Exec(t, `INSERT INTO f VALUES (1)`)
+		timeNow := strings.Split(s.Server.Clock().Now().AsOfSystemTime(), ".")[0]
+
+		expectNotice(t, s.Server,
+			fmt.Sprintf(
+				`CREATE CHANGEFEED FOR TABLE d.f INTO 'null://' with cursor = '%s', initial_scan='only'`,
+				timeNow), expectedWarning("only"))
+
+		expectNotice(t, s.Server,
+			fmt.Sprintf(
+				`CREATE CHANGEFEED FOR TABLE d.f INTO 'null://' with cursor = '%s', initial_scan='yes'`,
+				timeNow), expectedWarning("yes"))
+
+		expectNotice(t, s.Server,
+			fmt.Sprintf(
+				`CREATE CHANGEFEED FOR TABLE d.f INTO 'null://' with cursor = '%s', initial_scan='no'`,
+				timeNow), expectedWarning("no"))
+	})
+}
+
 // TestChangefeedSchemaTTL ensures that changefeeds fail with an error in the case
 // where the feed has fallen behind the GC TTL of the table's schema.
 func TestChangefeedSchemaTTL(t *testing.T) {

pkg/ccl/changefeedccl/testing_knobs.go

@@ -108,6 +108,9 @@ type TestingKnobs struct {
 
 	// WrapTelemetryLogger is used to wrap the periodic telemetry logger in tests.
 	WrapTelemetryLogger func(logger telemetryLogger) telemetryLogger
+
+	// OverrideCursorAge is used to change how old a cursor is. Returns time in nanoseconds.
+	OverrideCursorAge func() int64
 }
 
 // ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.