Merge cockroachdb#154341

154341: changefeedccl/resolvedspan: benchmark frontier per-table tracking r=KeithCh,asg0451 a=andyyang890

Fixes cockroachdb#152111

---

This patch adds a new benchmark for frontier per-table tracking.

Results when I run the benchmark on my laptop:
```
goos: darwin
goarch: arm64
cpu: Apple M1 Pro
BenchmarkFrontierPerTableTracking/system/aggregator/per-table-tracking=false
BenchmarkFrontierPerTableTracking/system/aggregator/per-table-tracking=false         	11734448	        93.39 ns/op	       0 B/op	       0 allocs/op
BenchmarkFrontierPerTableTracking/system/aggregator/per-table-tracking=true
BenchmarkFrontierPerTableTracking/system/aggregator/per-table-tracking=true          	 6423613	       171.1 ns/op	      16 B/op	       2 allocs/op
BenchmarkFrontierPerTableTracking/system/coordinator/per-table-tracking=false
BenchmarkFrontierPerTableTracking/system/coordinator/per-table-tracking=false        	14056830	        87.45 ns/op	       0 B/op	       0 allocs/op
BenchmarkFrontierPerTableTracking/system/coordinator/per-table-tracking=true
BenchmarkFrontierPerTableTracking/system/coordinator/per-table-tracking=true         	 7310874	       157.4 ns/op	      16 B/op	       2 allocs/op
BenchmarkFrontierPerTableTracking/tenant/aggregator/per-table-tracking=false
BenchmarkFrontierPerTableTracking/tenant/aggregator/per-table-tracking=false         	14520753	        82.53 ns/op	       0 B/op	       0 allocs/op
BenchmarkFrontierPerTableTracking/tenant/aggregator/per-table-tracking=true
BenchmarkFrontierPerTableTracking/tenant/aggregator/per-table-tracking=true          	 7783419	       160.1 ns/op	      16 B/op	       2 allocs/op
BenchmarkFrontierPerTableTracking/tenant/coordinator/per-table-tracking=false
BenchmarkFrontierPerTableTracking/tenant/coordinator/per-table-tracking=false        	14832232	        83.03 ns/op	       0 B/op	       0 allocs/op
BenchmarkFrontierPerTableTracking/tenant/coordinator/per-table-tracking=true
BenchmarkFrontierPerTableTracking/tenant/coordinator/per-table-tracking=true         	 8280913	       164.1 ns/op	      16 B/op	       2 allocs/op
```

From profiling, the allocs and some of the slowdown when per-table
tracking is on comes from constructing the full table spans.
We could consider caching them in the future if we see a need.

Release note: None

Co-authored-by: Andy Yang <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/resolvedspan/BUILD.bazel

@@ -24,19 +24,31 @@ go_library(
 
 go_test(
     name = "resolvedspan_test",
-    srcs = ["frontier_test.go"],
+    srcs = [
+        "frontier_test.go",
+        "main_test.go",
+    ],
     deps = [
         ":resolvedspan",
+        "//pkg/base",
+        "//pkg/ccl",
         "//pkg/jobs/jobspb",
         "//pkg/keys",
         "//pkg/roachpb",
+        "//pkg/security/securityassets",
+        "//pkg/security/securitytest",
+        "//pkg/server",
         "//pkg/sql/catalog/descpb",
         "//pkg/testutils",
+        "//pkg/testutils/serverutils",
+        "//pkg/testutils/sqlutils",
+        "//pkg/testutils/testcluster",
         "//pkg/util/hlc",
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/randutil",
         "//pkg/util/span",
+        "//pkg/util/timeutil",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],

pkg/ccl/changefeedccl/resolvedspan/frontier_test.go

@@ -6,20 +6,27 @@
 package resolvedspan_test
 
 import (
+	"context"
+	"fmt"
 	"iter"
 	"testing"
+	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/base"
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/resolvedspan"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"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/randutil"
 	"github.com/cockroachdb/cockroach/pkg/util/span"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/require"
 )
@@ -172,6 +179,7 @@ func TestCoordinatorFrontier(t *testing.T) {
 }
 
 type frontier interface {
+	AddSpansAt(startAt hlc.Timestamp, spans ...roachpb.Span) error
 	Frontier() hlc.Timestamp
 	ForwardResolvedSpan(jobspb.ResolvedSpan) (bool, error)
 	InBackfill(jobspb.ResolvedSpan) bool
@@ -530,3 +538,120 @@ func TestFrontierForwardFullTableSpan(t *testing.T) {
 			require.Equal(t, targetTimestamp, f.Frontier())
 		})
 }
+
+func BenchmarkFrontierPerTableTracking(b *testing.B) {
+	defer leaktest.AfterTest(b)()
+	defer log.Scope(b).Close(b)
+
+	ctx := context.Background()
+	rng, _ := randutil.NewTestRand()
+
+	// Generate a random span iteration order that's the same for all
+	// sub-benchmarks.
+	const numSpans = 100
+	order := make([]int, numSpans)
+	for i := range numSpans {
+		order[i] = i
+	}
+	rng.Shuffle(numSpans, func(i, j int) {
+		order[i], order[j] = order[j], order[i]
+	})
+
+	for _, tenantType := range []string{"system", "tenant"} {
+		for _, frontierType := range []string{"aggregator", "coordinator"} {
+			for _, perTableTracking := range []bool{false, true} {
+				b.Run(
+					fmt.Sprintf("%s/%s/per-table-tracking=%t", tenantType, frontierType, perTableTracking),
+					func(b *testing.B) {
+						// Start the server.
+						srv, db, _ := serverutils.StartServer(b, base.TestServerArgs{
+							DefaultTestTenant: base.TestControlsTenantsExplicitly,
+						})
+						defer srv.Stopper().Stop(ctx)
+
+						// Get a SQL connection/codec for the tenant type.
+						sqlDB, codec := func() (*sqlutils.SQLRunner, keys.SQLCodec) {
+							switch tenantType {
+							case "system":
+								return sqlutils.MakeSQLRunner(db), keys.SystemSQLCodec
+							case "tenant":
+								tenantID := roachpb.MinTenantID
+								_, tenantDB := serverutils.StartTenant(b, srv, base.TestTenantArgs{
+									TenantID: tenantID,
+								})
+								return sqlutils.MakeSQLRunner(tenantDB), keys.MakeSQLCodec(tenantID)
+							default:
+								panic("unreachable")
+							}
+						}()
+
+						// Create a table and split it into multiple spans.
+						sqlDB.Exec(b, `CREATE TABLE foo (id INT PRIMARY KEY)`)
+						sqlDB.Exec(b, fmt.Sprintf(
+							`ALTER TABLE foo SPLIT AT SELECT generate_series(10, %d, 10)`, (numSpans-1)*10))
+
+						var fooTableID uint32
+						sqlDB.QueryRow(b, `SELECT 'foo'::regclass::oid::int`).Scan(&fooTableID)
+						fooSpan := codec.TableSpan(fooTableID)
+
+						// Collect all the spans.
+						var spans roachpb.Spans
+						rows := sqlDB.Query(b, `SELECT raw_start_key, raw_end_key
+FROM [SHOW RANGES FROM TABLE foo WITH KEYS]`)
+						for rows.Next() {
+							var startKey, endKey roachpb.Key
+							err := rows.Scan(&startKey, &endKey)
+							require.NoError(b, err)
+							sp := roachpb.Span{Key: startKey, EndKey: endKey}
+							spans = append(spans, fooSpan.Intersect(sp))
+						}
+						require.Len(b, spans, numSpans)
+
+						now := makeTS(timeutil.Now().Unix())
+
+						// Create the frontier and add all the spans.
+						f, err := func() (frontier, error) {
+							switch frontierType {
+							case "aggregator":
+								return resolvedspan.NewAggregatorFrontier(
+									now,
+									now,
+									codec,
+									perTableTracking,
+								)
+							case "coordinator":
+								return resolvedspan.NewCoordinatorFrontier(
+									now,
+									now,
+									codec,
+									perTableTracking,
+								)
+							default:
+								panic("unreachable")
+							}
+						}()
+						require.NoError(b, err)
+						require.NoError(b, f.AddSpansAt(now, spans...))
+
+						// Main benchmark loop: forward (shuffled) spans in a loop.
+						b.ResetTimer()
+						for n := range b.N {
+							ts := now.AddDuration(time.Second)
+							i := n % len(spans)
+							_, err := f.ForwardResolvedSpan(jobspb.ResolvedSpan{
+								Span:      spans[order[i]],
+								Timestamp: ts,
+							})
+							if err != nil {
+								b.Fatalf("error forwarding: %v", err)
+							}
+						}
+						b.StopTimer()
+
+						// Make sure the compiler doesn't optimize away the forwards.
+						require.True(b, now.LessEq(f.Frontier()))
+					})
+			}
+		}
+	}
+}

pkg/ccl/changefeedccl/resolvedspan/main_test.go

@@ -0,0 +1,28 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package resolvedspan_test
+
+import (
+	"os"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/ccl"
+	"github.com/cockroachdb/cockroach/pkg/security/securityassets"
+	"github.com/cockroachdb/cockroach/pkg/security/securitytest"
+	"github.com/cockroachdb/cockroach/pkg/server"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/testcluster"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
+)
+
+func TestMain(m *testing.M) {
+	defer ccl.TestingEnableEnterprise()()
+	securityassets.SetLoader(securitytest.EmbeddedAssets)
+	randutil.SeedForTests()
+	serverutils.InitTestServerFactory(server.TestServerFactory)
+	serverutils.InitTestClusterFactory(testcluster.TestClusterFactory)
+	os.Exit(m.Run())
+}