Merge #147879 #148236 #148302

147879: buildutil.bzl: fix disallowed_imports_test logic and refactor BUILD.bazel r=rickystewart a=suvaidkhan

Changes - 
- Changed `allowList` logic to check for empty list
- Set  `allowList` `disallowed_list` & `disallowed_prefixes` defaults to None and added None handling 
- If no list is passed in `disallowed_imports_test` an error is thrown
- Refactored `disallowed_imports_test`  in `pkg/server/diagnostics/BUILD.bazel`and `pkg/server/diagnostics/BUILD.bazel`

Fixes #139892

148236: roachtest: add buffered-writes variants of TPCC bench r=stevendanna a=miraradeva

This commit adds the following variants of TPCC bench:
- write buffering enabled, pipelining enabled (by default),
- write buffering enabled, pipelining disabled.

The former will help us compare the performance of buffered writes to the default setup, and the latter will inform our decision to replace pipelining with buffering altogether.

Part of: #148235
Release note: None

148302: changefeedccl: fix table drop shutdown for enriched envelope r=andyyang890 a=asg0451

Fixes a bug where feeds with enriched envelopes
and a `source` field could treat table drops as
transient errors.

Fixes: #148216
Release note: None


Co-authored-by: suvaidkhan <[email protected]>
Co-authored-by: Mira Radeva <[email protected]>
Co-authored-by: Miles Frankel <[email protected]>

Author: 4 people

pkg/ccl/changefeedccl/enriched_source_provider.go

@@ -540,6 +540,9 @@ func getDescriptors(
 		return nil
 	}
 	if err := execCfg.InternalDB.DescsTxn(ctx, f, isql.WithPriority(admissionpb.NormalPri)); err != nil {
+		if errors.Is(err, catalog.ErrDescriptorDropped) {
+			err = changefeedbase.WithTerminalError(err)
+		}
 		return nil, nil, nil, err
 	}
 	return tableDescriptor, dbDescriptor, schemaDescriptor, nil

pkg/cmd/roachtest/registry/test_spec.go

@@ -131,6 +131,12 @@ type TestSpec struct {
 	// to epoch leases.
 	Leases LeaseType
 
+	// WriteOptimization specifies what kind of write optimization to use.
+	// Defaults to pipelining. This is currently used only in benchmark tests.
+	// If used in a non-benchmark test, this setting will be overwritten to enable
+	// pipelining and additionally, with 50% probability, buffering.
+	WriteOptimization WriteOptimizationType
+
 	// SkipPostValidations is a bit-set of post-validations that should be skipped
 	// after the test completes. This is useful for tests that are known to be
 	// incompatible with some validations. By default, tests will run all
@@ -284,6 +290,35 @@ const (
 // The list does not contain aliases like "default" and "metamorphic".
 var LeaseTypes = []LeaseType{EpochLeases, ExpirationLeases, LeaderLeases}
 
+// WriteOptimizationType specifies the type of write optimization to use.
+type WriteOptimizationType int
+
+func (w WriteOptimizationType) String() string {
+	switch w {
+	case DefaultWriteOptimization:
+		return "default"
+	case Pipelining:
+		return "pipelining"
+	case Buffering:
+		return "buffering"
+	case PipeliningBuffering:
+		return "pipelining-buffering"
+	default:
+		return fmt.Sprintf("writeoptimization-%d", w)
+	}
+}
+
+const (
+	// DefaultWriteOptimization uses the default cluster settings.
+	DefaultWriteOptimization = WriteOptimizationType(iota)
+	// Pipelining uses write pipelining.
+	Pipelining
+	// Buffering uses client-side write buffering.
+	Buffering
+	// PipeliningBuffering uses both buffering and pipelining.
+	PipeliningBuffering
+)
+
 // CloudSet represents a set of clouds.
 //
 // Instances of CloudSet are immutable. The uninitialized (zero) value is not

pkg/cmd/roachtest/test_runner.go

@@ -933,6 +933,22 @@ func (r *testRunner) runWorker(
 					t.Fatalf("unknown lease type %s", leases)
 				}
 
+				// Choose which write optimization to use. These are currently used only
+				// in benchmark tests. For non-benchmark tests, write buffering will be
+				// enabled metamorphically below.
+				switch testSpec.WriteOptimization {
+				case registry.DefaultWriteOptimization:
+				case registry.Pipelining:
+					c.clusterSettings["kv.transaction.write_pipelining.enabled"] = "true"
+					c.clusterSettings["kv.transaction.write_buffering.enabled"] = "false"
+				case registry.Buffering:
+					c.clusterSettings["kv.transaction.write_buffering.enabled"] = "true"
+					c.clusterSettings["kv.transaction.write_pipelining.enabled"] = "false"
+				case registry.PipeliningBuffering:
+					c.clusterSettings["kv.transaction.write_pipelining.enabled"] = "true"
+					c.clusterSettings["kv.transaction.write_buffering.enabled"] = "true"
+				}
+
 				// Apply metamorphic settings not explicitly defined by the test.
 				// These settings should only be applied to non-benchmark tests.
 				if !testSpec.Benchmark {

pkg/cmd/roachtest/tests/tpcc.go

@@ -1619,6 +1619,50 @@ func registerTPCC(r registry.Registry) {
 		Clouds: registry.OnlyGCE,
 		Suites: registry.Suites(registry.Weekly),
 	})
+
+	// Buffered writes benchmarks. These are duplicates of variants above.
+
+	// A variant with only buffering enabled (pipelining disabled).
+	registerTPCCBenchSpec(
+		r, tpccBenchSpec{
+			Nodes: 3,
+			CPUs:  16,
+
+			LoadWarehousesGCE:   3500,
+			LoadWarehousesAWS:   3900,
+			LoadWarehousesAzure: 3900,
+			LoadWarehousesIBM:   3900,
+			EstimatedMaxGCE:     3100,
+			EstimatedMaxAWS:     3600,
+			EstimatedMaxAzure:   3600,
+			EstimatedMaxIBM:     3600,
+			WriteOptimization:   registry.Buffering,
+
+			Clouds: registry.AllClouds,
+			Suites: registry.Suites(registry.Nightly),
+		},
+	)
+
+	// A variant with pipelining and buffering both enabled.
+	registerTPCCBenchSpec(
+		r, tpccBenchSpec{
+			Nodes: 3,
+			CPUs:  16,
+
+			LoadWarehousesGCE:   3500,
+			LoadWarehousesAWS:   3900,
+			LoadWarehousesAzure: 3900,
+			LoadWarehousesIBM:   3900,
+			EstimatedMaxGCE:     3100,
+			EstimatedMaxAWS:     3600,
+			EstimatedMaxAzure:   3600,
+			EstimatedMaxIBM:     3600,
+			WriteOptimization:   registry.PipeliningBuffering,
+
+			Clouds: registry.AllClouds,
+			Suites: registry.Suites(registry.Nightly),
+		},
+	)
 }
 
 func valueForCloud(cloud spec.Cloud, gce, aws, azure, ibm int) int {
@@ -1730,6 +1774,9 @@ type tpccBenchSpec struct {
 	// SharedProcessMT, if true, indicates that the cluster should run in
 	// shared-process mode of multi-tenancy.
 	SharedProcessMT bool
+	// WriteOptimization specifies the write optimization to use (e.g. pipelining,
+	// buffering, or both).
+	WriteOptimization registry.WriteOptimizationType
 }
 
 func (s tpccBenchSpec) EstimatedMax(cloud spec.Cloud) int {
@@ -1829,6 +1876,10 @@ func registerTPCCBenchSpec(r registry.Registry, b tpccBenchSpec) {
 		nameParts = append(nameParts, "mt-shared-process")
 	}
 
+	if b.WriteOptimization != registry.DefaultWriteOptimization {
+		nameParts = append(nameParts, fmt.Sprintf("write-optimization=%s", b.WriteOptimization.String()))
+	}
+
 	name := strings.Join(nameParts, "/")
 
 	numNodes := b.Nodes + b.LoadConfig.numLoadNodes(b.Distribution)
@@ -1844,6 +1895,7 @@ func registerTPCCBenchSpec(r registry.Registry, b tpccBenchSpec) {
 		Suites:                 b.Suites,
 		EncryptionSupport:      encryptionSupport,
 		Leases:                 leases,
+		WriteOptimization:      b.WriteOptimization,
 		PostProcessPerfMetrics: getMaxWarehousesAboveEfficiency,
 		Run: func(ctx context.Context, t test.Test, c cluster.Cluster) {
 			runTPCCBench(ctx, t, c, b)

pkg/roachpb/BUILD.bazel

@@ -163,4 +163,5 @@ stringer(
 disallowed_imports_test(
     "roachpb",
     disallow_cdeps = True,
+    disallowed_list = [],
 )

pkg/server/diagnostics/BUILD.bazel

@@ -87,4 +87,5 @@ disallowed_imports_test(
     "diagnostics",
     # TODO(#81378): This should be flipped to "true".
     disallow_cdeps = False,
+    disallowed_list = [],
 )

pkg/testutils/buildutil/buildutil.bzl

@@ -49,6 +49,8 @@ _deps_aspect = aspect(
 )
 
 def _find_deps_with_disallowed_prefixes(current_pkg, dep_pkgs, prefixes):
+    if prefixes == None:
+        return []
     return [dp for dp_list in [
         [(d, p) for p in prefixes if d.startswith(p) and d != current_pkg]
         for d in dep_pkgs
@@ -61,11 +63,13 @@ def _deps_rule_impl(ctx):
         prefixes = ctx.attr.disallowed_prefixes,
     )
     deps = {k: None for k in ctx.attr.src[_DepsInfo].deps.to_list()}
-    if ctx.attr.allowlist:
+    if ctx.attr.has_allowlist:
         failed = [p for p in deps if p not in ctx.attr.allowlist and
                                      p.label != ctx.attr.src.label]
-    else:
+    elif ctx.attr.has_disallowed_list:
         failed = [p for p in ctx.attr.disallowed_list if p in deps]
+    else:
+        failed = []
     failures = []
     if failed_prefixes:
         failures.extend([
@@ -108,10 +112,14 @@ _deps_rule = rule(
         "disallow_cdeps": attr.bool(mandatory = False, default = False),
         "disallowed_list": attr.label_list(providers = [GoInfo]),
         "disallowed_prefixes": attr.string_list(mandatory = False, allow_empty = True),
+        "has_allowlist": attr.bool(default = False),
+        "has_disallowed_list": attr.bool(default = False),
     },
 )
 
 def _validate_disallowed_prefixes(prefixes):
+    if prefixes == None:
+        return []
     validated = []
     repo_prefix = "github.com/cockroachdb/cockroach/"
     short_prefix = "pkg/"
@@ -130,11 +138,15 @@ def _validate_disallowed_prefixes(prefixes):
 
 def disallowed_imports_test(
         src,
-        disallowed_list = [],
-        disallowed_prefixes = [],
+        disallowed_list = None,
+        disallowed_prefixes = None,
         disallow_cdeps = False,
-        allowlist = []):
-    if (disallowed_list and allowlist) or (disallowed_prefixes and allowlist):
+        allowlist = None):
+
+    if allowlist == None and disallowed_list == None and disallowed_prefixes == None:
+        fail("Either allowlist, disallowed_list or disallowed_prefixes should be passed, to block " +
+           "all imports you must explicitly pass allowlist = []")
+    if (allowlist != None and disallowed_list != None) or (allowlist != None and disallowed_prefixes != None):
         fail("allowlist or (disallowed_list or disallowed_prefixes) can be " +
              "provided, but not both")
     disallowed_prefixes = _validate_disallowed_prefixes(disallowed_prefixes)
@@ -147,6 +159,8 @@ def disallowed_imports_test(
         disallowed_list = disallowed_list,
         disallowed_prefixes = disallowed_prefixes,
         disallow_cdeps = disallow_cdeps,
+        has_allowlist = allowlist != None,
+        has_disallowed_list = disallowed_list != None,
     )
     native.sh_test(
         name = src.strip(":") + "_disallowed_imports_test",