Merge #121138 #121164 #121212

121138: storage,ingest: enable ingestion optimization using flushableIngest r=aadityasondhi a=aadityasondhi

In cockroachdb/pebble#3398, we introduced an optimization in pebble where we could use flushableIngests with excises for ingesting SSTs with RangeDels and RangeKeyDels.

This patch turns this optimization on using
`sstContainsExciseTombstone`.

Informs cockroachdb/pebble#3335.

Release note: None

121164: Revert "rangefeed: unconditionally use rangefeed scheduler" r=erikgrinaker a=erikgrinaker

This reverts 3/4 commits from #114410, only leaving 1dbdbe9 which enabled the scheduler by default.

There was significant code drift in tests, but I think I've been able to address the conflicts.

This revert does not integrate with the changes from #120347, so the legacy rangefeed processor will not use the improved memory accounting. We still shouldn't leak budget allocations, since the processor implementations are entirely independent, but it's worth having a second look to make sure the registry (which is used by both processors) don't make assumptions about this.

Touches #121054.
Epic: none.
Release note: the following release note from #114410 should be reverted/ignored: "the setting kv.rangefeed.scheduler.enabled has been removed, as the rangefeed scheduler is now unconditionally enabled".

121212: security: wrap error when parsing DN r=rafiss a=rafiss

Epic: CRDB-34126
Release note: None

Co-authored-by: Aaditya Sondhi <[email protected]>
Co-authored-by: Erik Grinaker <[email protected]>
Co-authored-by: Rafi Shamim <[email protected]>

DEPS.bzl

@@ -1693,10 +1693,10 @@ def go_deps():
         patches = [
             "@com_github_cockroachdb_cockroach//build/patches:com_github_cockroachdb_pebble.patch",
         ],
-        sha256 = "f91c1724434bde2eb5d1c5fa5678f485056e963faa23c3974ac6bc9899aa8b18",
-        strip_prefix = "github.com/cockroachdb/[email protected]20240322192100-10ebcdd794ec",
+        sha256 = "e847edf768847f4f2034ebed41c2834728206c9bea79dc1fb69db5a5c0e66114",
+        strip_prefix = "github.com/cockroachdb/[email protected]20240326153331-5babbee7ffe8",
         urls = [
-            "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/pebble/com_github_cockroachdb_pebble-v0.0.0-20240322192100-10ebcdd794ec.zip",
+            "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/pebble/com_github_cockroachdb_pebble-v0.0.0-20240326153331-5babbee7ffe8.zip",
         ],
     )
     go_repository(

build/bazelutil/distdir_files.bzl

@@ -330,7 +330,7 @@ DISTDIR_FILES = {
     "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/gostdlib/com_github_cockroachdb_gostdlib-v1.19.0.zip": "c4d516bcfe8c07b6fc09b8a9a07a95065b36c2855627cb3514e40c98f872b69e",
     "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/logtags/com_github_cockroachdb_logtags-v0.0.0-20230118201751-21c54148d20b.zip": "ca7776f47e5fecb4c495490a679036bfc29d95bd7625290cfdb9abb0baf97476",
     "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/metamorphic/com_github_cockroachdb_metamorphic-v0.0.0-20231108215700-4ba948b56895.zip": "28c8cf42192951b69378cf537be5a9a43f2aeb35542908cc4fe5f689505853ea",
-    "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/pebble/com_github_cockroachdb_pebble-v0.0.0-20240322192100-10ebcdd794ec.zip": "f91c1724434bde2eb5d1c5fa5678f485056e963faa23c3974ac6bc9899aa8b18",
+    "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/pebble/com_github_cockroachdb_pebble-v0.0.0-20240326153331-5babbee7ffe8.zip": "e847edf768847f4f2034ebed41c2834728206c9bea79dc1fb69db5a5c0e66114",
     "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/redact/com_github_cockroachdb_redact-v1.1.5.zip": "11b30528eb0dafc8bc1a5ba39d81277c257cbe6946a7564402f588357c164560",
     "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/returncheck/com_github_cockroachdb_returncheck-v0.0.0-20200612231554-92cdbca611dd.zip": "ce92ba4352deec995b1f2eecf16eba7f5d51f5aa245a1c362dfe24c83d31f82b",
     "https://storage.googleapis.com/cockroach-godeps/gomod/github.com/cockroachdb/stress/com_github_cockroachdb_stress-v0.0.0-20220803192808-1806698b1b7b.zip": "3fda531795c600daf25532a4f98be2a1335cd1e5e182c72789bca79f5f69fcc1",

go.mod

@@ -124,7 +124,7 @@ require (
 	github.com/cockroachdb/go-test-teamcity v0.0.0-20191211140407-cff980ad0a55
 	github.com/cockroachdb/gostdlib v1.19.0
 	github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b
-	github.com/cockroachdb/pebble v0.0.0-20240322192100-10ebcdd794ec
+	github.com/cockroachdb/pebble v0.0.0-20240326153331-5babbee7ffe8
 	github.com/cockroachdb/redact v1.1.5
 	github.com/cockroachdb/returncheck v0.0.0-20200612231554-92cdbca611dd
 	github.com/cockroachdb/stress v0.0.0-20220803192808-1806698b1b7b

go.sum

@@ -507,8 +507,8 @@ github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZe
 github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs=
 github.com/cockroachdb/metamorphic v0.0.0-20231108215700-4ba948b56895 h1:XANOgPYtvELQ/h4IrmPAohXqe2pWA8Bwhejr3VQoZsA=
 github.com/cockroachdb/metamorphic v0.0.0-20231108215700-4ba948b56895/go.mod h1:aPd7gM9ov9M8v32Yy5NJrDyOcD8z642dqs+F0CeNXfA=
-github.com/cockroachdb/pebble v0.0.0-20240322192100-10ebcdd794ec h1:pvg/EIMk3MMADtSNi8i9akNZupSpdLygaHXGIAXyVOw=
-github.com/cockroachdb/pebble v0.0.0-20240322192100-10ebcdd794ec/go.mod h1:4vn8KzcL6D2yW6hZAabweFFHVYSIL6z9BKTAEBvAmS4=
+github.com/cockroachdb/pebble v0.0.0-20240326153331-5babbee7ffe8 h1:OdeQ7vEOrVUQ/okyHNQWF3JJm3H1H0STy0EtbnVxjxw=
+github.com/cockroachdb/pebble v0.0.0-20240326153331-5babbee7ffe8/go.mod h1:4vn8KzcL6D2yW6hZAabweFFHVYSIL6z9BKTAEBvAmS4=
 github.com/cockroachdb/redact v1.1.3/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg=
 github.com/cockroachdb/redact v1.1.5 h1:u1PMllDkdFfPWaNGMyLD1+so+aq3uUItthCFqzwPJ30=
 github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg=

pkg/cmd/roachtest/tests/cdc.go

@@ -496,6 +496,7 @@ func (f *enumFeatureFlag) enabled(r enthropy, choose func(enthropy) string) (str
 // cdcFeatureFlags describes various cdc feature flags.
 // zero value cdcFeatureFlags uses metamorphic settings for features.
 type cdcFeatureFlags struct {
+	RangeFeedScheduler   featureFlag
 	SchemaLockTables     featureFlag
 	DistributionStrategy enumFeatureFlag
 }
@@ -2864,6 +2865,16 @@ func (cfc *changefeedCreator) applySettings() error {
 		return err
 	}
 
+	schedEnabled := cfc.flags.RangeFeedScheduler.enabled(cfc.rng)
+	if schedEnabled != featureUnset {
+		cfc.logger.Printf("Setting kv.rangefeed.scheduler.enabled to %t", schedEnabled == featureEnabled)
+		if _, err := cfc.db.Exec(
+			"SET CLUSTER SETTING kv.rangefeed.scheduler.enabled = $1", schedEnabled == featureEnabled,
+		); err != nil {
+			return err
+		}
+	}
+
 	rangeDistribution, rangeDistributionEnabled := cfc.flags.DistributionStrategy.enabled(cfc.rng,
 		chooseDistributionStrategy)
 	if rangeDistributionEnabled == featureEnabled {

pkg/cmd/roachtest/tests/cdc_bench.go

@@ -39,6 +39,7 @@ import (
 )
 
 type cdcBenchScanType string
+type cdcBenchServer string
 
 const (
 	// cdcBenchInitialScan runs an initial scan across a table, i.e. it scans and
@@ -59,11 +60,16 @@ const (
 	// do so efficiently. Ideally, this wouldn't take any time at all, but in
 	// practice it can.
 	cdcBenchColdCatchupScan cdcBenchScanType = "catchup-cold"
+
+	cdcBenchNoServer        cdcBenchServer = ""
+	cdcBenchProcessorServer cdcBenchServer = "processor" // legacy processor
+	cdcBenchSchedulerServer cdcBenchServer = "scheduler" // new scheduler
 )
 
 var (
 	cdcBenchScanTypes = []cdcBenchScanType{
 		cdcBenchInitialScan, cdcBenchCatchupScan, cdcBenchColdCatchupScan}
+	cdcBenchServers = []cdcBenchServer{cdcBenchProcessorServer, cdcBenchSchedulerServer}
 )
 
 func registerCDCBench(r registry.Registry) {
@@ -119,26 +125,29 @@ func registerCDCBench(r registry.Registry) {
 				RequiresLicense:  true,
 				Timeout:          time.Hour,
 				Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-					runCDCBenchWorkload(ctx, t, c, ranges, readPercent, "")
+					runCDCBenchWorkload(ctx, t, c, ranges, readPercent, "", "")
 				},
 			})
 
 			// Workloads with a concurrent changefeed running.
-			r.Add(registry.TestSpec{
-				Name: fmt.Sprintf(
-					"cdc/workload/kv%d/nodes=%d/cpu=%d/ranges=%s/server=scheduler/protocol=mux/format=%s/sink=null",
-					readPercent, nodes, cpus, formatSI(ranges), format),
-				Owner:            registry.OwnerCDC,
-				Benchmark:        true,
-				Cluster:          r.MakeClusterSpec(nodes+2, spec.CPU(cpus)),
-				CompatibleClouds: registry.AllExceptAWS,
-				Suites:           registry.Suites(registry.Nightly),
-				RequiresLicense:  true,
-				Timeout:          time.Hour,
-				Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
-					runCDCBenchWorkload(ctx, t, c, ranges, readPercent, format)
-				},
-			})
+			for _, server := range cdcBenchServers {
+				server := server // pin loop variable
+				r.Add(registry.TestSpec{
+					Name: fmt.Sprintf(
+						"cdc/workload/kv%d/nodes=%d/cpu=%d/ranges=%s/server=%s/protocol=mux/format=%s/sink=null",
+						readPercent, nodes, cpus, formatSI(ranges), server, format),
+					Owner:            registry.OwnerCDC,
+					Benchmark:        true,
+					Cluster:          r.MakeClusterSpec(nodes+2, spec.CPU(cpus)),
+					CompatibleClouds: registry.AllExceptAWS,
+					Suites:           registry.Suites(registry.Nightly),
+					RequiresLicense:  true,
+					Timeout:          time.Hour,
+					Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
+						runCDCBenchWorkload(ctx, t, c, ranges, readPercent, server, format)
+					},
+				})
+			}
 		}
 	}
 }
@@ -352,6 +361,7 @@ func runCDCBenchWorkload(
 	c cluster.Cluster,
 	numRanges int64,
 	readPercent int,
+	server cdcBenchServer,
 	format string,
 ) {
 	const sink = "null://"
@@ -370,12 +380,28 @@ func runCDCBenchWorkload(
 	if readPercent == 100 {
 		insertCount = 1_000_000 // ingest some data to read
 	}
+	// Either of these will disable changefeeds. Make sure they're all disabled.
+	if server == "" || format == "" {
+		require.Empty(t, server)
+		require.Empty(t, format)
+		cdcEnabled = false
+	}
 
 	// Start data nodes first to place data on them. We'll start the changefeed
 	// coordinator later, since we don't want any data on it.
 	opts, settings := makeCDCBenchOptions(c)
 	settings.ClusterSettings["kv.rangefeed.enabled"] = strconv.FormatBool(cdcEnabled)
 
+	switch server {
+	case cdcBenchProcessorServer:
+		settings.ClusterSettings["kv.rangefeed.scheduler.enabled"] = "false"
+	case cdcBenchSchedulerServer:
+		settings.ClusterSettings["kv.rangefeed.scheduler.enabled"] = "true"
+	case cdcBenchNoServer:
+	default:
+		t.Fatalf("unknown server type %q", server)
+	}
+
 	c.Start(ctx, t.L(), opts, settings, nData)
 	m := c.NewMonitor(ctx, nData.Merge(nCoord))
 

pkg/cmd/roachtest/tests/mixed_version_cdc.go

@@ -371,6 +371,14 @@ func (cmvt *cdcMixedVersionTester) createChangeFeed(
 	}
 
 	var ff cdcFeatureFlags
+	rangefeedSchedulerSupported, err := cmvt.rangefeedSchedulerSupported(r, h)
+	if err != nil {
+		return err
+	}
+	if !rangefeedSchedulerSupported {
+		ff.RangeFeedScheduler.v = &featureUnset
+	}
+
 	distributionStrategySupported, err := cmvt.distributionStrategySupported(r, h)
 	if err != nil {
 		return err
@@ -426,17 +434,14 @@ func (cmvt *cdcMixedVersionTester) muxRangeFeedSupported(
 	)
 }
 
+const v232CV = "23.2"
 const v241CV = "24.1"
 
 func (cmvt *cdcMixedVersionTester) rangefeedSchedulerSupported(
-	h *mixedversion.Helper,
-) (bool, option.NodeListOption, error) {
-	// kv.rangefeed.scheduler.enabled only exists in 23.2. In 24.1, it is enabled
-	// unconditionally.
-	return canMixedVersionUseDeletedClusterSetting(h,
-		clusterupgrade.MustParseVersion("v23.2.0"),
-		clusterupgrade.MustParseVersion("v24.1.0-alpha.00000000"),
-	)
+	r *rand.Rand, h *mixedversion.Helper,
+) (bool, error) {
+	// kv.rangefeed.scheduler.enabled only exists since 23.2.
+	return h.ClusterVersionAtLeast(r, v232CV)
 }
 
 func (cmvt *cdcMixedVersionTester) distributionStrategySupported(
@@ -530,14 +535,15 @@ func runCDCMixedVersions(ctx context.Context, t test.Test, c cluster.Cluster) {
 
 	// Rangefeed scheduler available in 23.2
 	setRangeFeedSchedulerEnabled := func(ctx context.Context, l *logger.Logger, r *rand.Rand, h *mixedversion.Helper) error {
-		supported, gatewayNodes, err := tester.rangefeedSchedulerSupported(h)
+		supported, err := tester.rangefeedSchedulerSupported(r, h)
 		if err != nil {
 			return err
 		}
+		l.Printf("kv.rangefeed.scheduler.enabled=%t", supported)
 		if supported {
 			coin := r.Int()%2 == 0
 			l.PrintfCtx(ctx, "Setting kv.rangefeed.scheduler.enabled=%t", coin)
-			return h.ExecWithGateway(r, gatewayNodes, "SET CLUSTER SETTING kv.rangefeed.scheduler.enabled=$1", coin)
+			return h.Exec(r, "SET CLUSTER SETTING kv.rangefeed.scheduler.enabled=$1", coin)
 		}
 		return nil
 	}