workload/changefeed: allow changefeed-cursor

This commit adds a new option, changefeed-cursor, which allows specifying the
timestamp after which the changefeed should start emitting events and trigger a
catch-up scan. If not specified, the changefeed defaults to using the current
cluster logical timestamp. This helps tests specify catch-up scan behavior.

Author: wenyihu6

pkg/workload/changefeeds/changefeeds.go

@@ -27,6 +27,7 @@ func AddChangefeedToQueryLoad(
 	gen workload.ConnFlagser,
 	dbName string,
 	resolvedTarget time.Duration,
+	cursorStr string,
 	urls []string,
 	reg *histogram.Registry,
 	ql *workload.QueryLoad,
@@ -76,9 +77,10 @@ func AddChangefeedToQueryLoad(
 		return err
 	}
 
-	var cursorStr string
-	if err := conn.QueryRow(ctx, "SELECT cluster_logical_timestamp()").Scan(&cursorStr); err != nil {
-		return err
+	if cursorStr == "" {
+		if err := conn.QueryRow(ctx, "SELECT cluster_logical_timestamp()").Scan(&cursorStr); err != nil {
+			return err
+		}
 	}
 
 	tableNames := strings.Builder{}

pkg/workload/cli/run.go

@@ -116,6 +116,8 @@ var (
 		"changefeed-max-rate", 0, "Maximum frequency of changefeed ingestion. If 0, no limit.")
 	changefeedResolvedTarget = runFlags.Duration("changefeed-resolved-target", 5*time.Second,
 		"The target frequency of resolved messages. O to disable resolved reporting and accept server defaults.")
+	changefeedCursor = runFlags.String("changefeed-cursor", "",
+		"The cursor to start the changefeed from. If empty, the changefeed will start from the current cluster logical timestamp.")
 )
 
 func init() {
@@ -507,7 +509,8 @@ func runRun(gen workload.Generator, urls []string, dbName string) error {
 
 			if *changefeed {
 				log.Dev.Infof(ctx, "adding changefeed to query load...")
-				err = changefeeds.AddChangefeedToQueryLoad(ctx, gen.(workload.ConnFlagser), dbName, *changefeedResolvedTarget, urls, reg, &ops)
+				err = changefeeds.AddChangefeedToQueryLoad(ctx, gen.(workload.ConnFlagser), dbName,  *changefeedResolvedTarget, *changefeedCursor, urls, reg, &ops)
+				
 				if err != nil && !*tolerateErrors {
 					return errors.Wrapf(err, "failed to initialize changefeed")
 				}