Merge #150270 #150281 #150283

150270: bazelbuilder: add `s390x` support r=rail a=rickystewart

The `s390x` version of the image has the build utilities installed,
but NOT some unnecessary bits like the cloud CLI's. 


Epic: CRDB-21133
Part of: DEVINF-1495

150281: kvcoord: disallow pipelining while write buffering is enabled r=miraradeva,arulajmani a=stevendanna

The interaction between pipelining and write buffering has revealed a couple of subtle bugs. Further, these two optimisations are largely overlapping. In the happy case, when write buffering is enabled, we expect almost not write pipelining.

Here, we disable pipelining when write buffering is enabled.

The upside of this is:

1. It avoids other possible bugs in the interaction between these features.

2. It makes it a bit easier to reason about the behaviour of write-buffered transactions.

The downsides include:

1. This results in a negative performance impact for users who turn on write buffering but whose transactions often result in buffering later being disabled. (Mitigated by the second commit).

2. Write pipelining has been the default for many releases and it is not clear that running with write pipelining disabled is as well tested as the enabled code path.

Fixes #149911
Epic: none
Release note: None

150283: changefeedccl: fix ALTER CHANGEFEED max PTS age bug r=KeithCh,asg0451 a=andyyang890

This patch fixes a bug where modifying a changefeed with ALTER CHANGEFEED
that either unset or left the `gc_protect_expires_after` option unset would
cause the changefeed's max PTS age to become unbounded instead of being set to
the default value configured by the `changefeed.protect_timestamp.max_age`
cluster setting.

This bug was occurring because the ALTER CHANGEFEED code was manually querying
the option instead of relying on the CREATE CHANGEFEED code path that also
consulted the cluster setting. This inconsistency has now been corrected.

Fixes #148029

Release note (bug fix): A bug where modifying a changefeed with ALTER CHANGEFEED
that either unset or left the `gc_protect_expires_after` option unset would
cause the changefeed's max PTS age to become unbounded instead of being set to
the default value configured by the `changefeed.protect_timestamp.max_age`
cluster setting.


Co-authored-by: Ricky Stewart <[email protected]>
Co-authored-by: Steven Danna <[email protected]>
Co-authored-by: Andy Yang <[email protected]>

Author: 4 people

build/bazelbuilder/Dockerfile

@@ -3,6 +3,8 @@
 FROM us-east1-docker.pkg.dev/crl-docker-sync/docker-io/library/ubuntu:focal
 ARG TARGETPLATFORM
 
+SHELL ["/usr/bin/bash", "-c"]
+
 RUN apt-get update \
  && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
     apt-transport-https \
@@ -39,14 +41,27 @@ RUN apt-get update \
 # c-deps/*-rebuild to force recreating the makefiles. This prevents
 # strange build errors caused by those makefiles depending on the
 # installed version of cmake.
-RUN case ${TARGETPLATFORM} in \
+RUN if [[ ${TARGETPLATFORM} == "linux/s390x" ]] ; then \
+   curl -fsSL "https://github.com/Kitware/CMake/archive/refs/tags/v3.20.3.tar.gz" -o cmake.tar.gz \
+   && echo "aa059c7f89b56215301f1baac8f88a70a67a334495c9ab6a728b97e1defab763 cmake.tar.gz" | sha256sum -c - \
+   && tar -xzf cmake.tar.gz \
+   && cd CMake-3.20.3 \
+   && ./bootstrap -- -DCMAKE_USE_OPENSSL=OFF \
+   && make \
+   && make install \
+   && cd .. \
+   && rm -rf CMake-3.20.3 \
+   && rm cmake.tar.gz ; \
+ else \
+  case ${TARGETPLATFORM} in \
     "linux/amd64") ARCH=x86_64; SHASUM=97bf730372f9900b2dfb9206fccbcf92f5c7f3b502148b832e77451aa0f9e0e6 ;; \
     "linux/arm64") ARCH=aarch64; SHASUM=77620f99e9d5f39cf4a49294c6a68c89a978ecef144894618974b9958efe3c2a ;; \
   esac \
- && curl -fsSL "https://github.com/Kitware/CMake/releases/download/v3.20.3/cmake-3.20.3-linux-$ARCH.tar.gz" -o cmake.tar.gz \
- && echo "$SHASUM cmake.tar.gz" | sha256sum -c - \
- && tar --strip-components=1 -C /usr -xzf cmake.tar.gz \
- && rm cmake.tar.gz
+   && curl -fsSL "https://github.com/Kitware/CMake/releases/download/v3.20.3/cmake-3.20.3-linux-$ARCH.tar.gz" -o cmake.tar.gz \
+   && echo "$SHASUM cmake.tar.gz" | sha256sum -c - \
+   && tar --strip-components=1 -C /usr -xzf cmake.tar.gz \
+   && rm cmake.tar.gz ; \
+ fi
 
 # git - Upgrade to a more modern version
 RUN apt-get update && \
@@ -62,16 +77,39 @@ RUN apt-get update && \
     cd .. && \
     rm -rf git-2.29.2.zip git-2.29.2
 
-RUN curl -fsSL https://packages.cloud.google.com/apt/doc/apt-key.gpg | apt-key add - \
- && echo 'deb https://packages.cloud.google.com/apt cloud-sdk main' | tee /etc/apt/sources.list.d/gcloud.list \
- && curl -fsLS https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor | apt-key add - \
+# NB: Don't install the azure CLI on s390x which doesn't support it.
+RUN if [[ ${TARGETPLATFORM} != "linux/s390x" ]]; then \
+ curl -fsLS https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor | apt-key add - \
  && echo "deb https://packages.microsoft.com/repos/azure-cli/ focal main" > /etc/apt/sources.list.d/azure-cli.list \
  && apt-get update \
- && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
-    azure-cli \
-    google-cloud-sdk \
-    google-cloud-cli-gke-gcloud-auth-plugin \
-    && apt-get clean
+ && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends azure-cli \
+ && apt-get clean ; \
+ fi
+
+# NB: As above, this is not available on `s390x`.
+RUN if [[ ${TARGETPLATFORM} != "linux/s390x" ]]; then \
+ case ${TARGETPLATFORM} in \
+     "linux/amd64") ARCH=x86_64; SHASUM= ;; \
+     "linux/arm64") ARCH=arm; SHASUM=e6153461e3154ebce61d35b73005bdd14a0ecacd42e5008f66e25b4ad231e5c9 ;; \
+ esac  \
+ && curl -fsSL "https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/google-cloud-cli-linux-$ARCH.tar.gz" -o gcloud.tar.gz \
+ && tar -xzf gcloud.tar.gz \
+ && rm gcloud.tar.gz ; \
+ fi
+
+ENV PATH="$PATH:/google-cloud-sdk/bin"
+
+# NB: We're going to run `dev` builds inside the Docker image on `s390x`,
+# as we can't cross-compile them (there are no cross-toolchains for `s390x`
+# hosts. This means we need these extra dependencies installed specifically
+# on that platform. Don't install them on other platforms to avoid taking
+# unintended dependencies on them.
+RUN if [[ ${TARGETPLATFORM} == "linux/s390x" ]]; then \
+  DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+    keyutils \
+    libresolv-wrapper \
+  && apt-get clean ; \
+ fi
 
 RUN apt-get purge -y \
     apt-transport-https \
@@ -81,19 +119,23 @@ RUN apt-get purge -y \
 
 # awscli - roachtests
 # NB: we don't use apt-get because we need an up to date version of awscli
-RUN case ${TARGETPLATFORM} in \
+# NB: Don't install these SDK's that are unavailable for s390x.
+RUN if [[ ${TARGETPLATFORM} != "linux/s390x" ]]; then \
+ case ${TARGETPLATFORM} in \
     "linux/amd64") ARCH=x86_64; SHASUM=e679933eec90b0e5a75d485be6c2fae0f89a3f9ccdcb1748be69f8f456e9a85f ;; \
     "linux/arm64") ARCH=aarch64; SHASUM=7d6460f795712ebdac7e3c60d4800dde682d136d909810402aac164f2789b860 ;; \
-  esac && \
- curl -fsSL "https://awscli.amazonaws.com/awscli-exe-linux-$ARCH-2.13.9.zip" -o "awscliv2.zip" && \
- echo "$SHASUM awscliv2.zip" | sha256sum -c - && \
- unzip awscliv2.zip && \
- ./aws/install && \
- rm -rf aws awscliv2.zip
+  esac \
+ && curl -fsSL "https://awscli.amazonaws.com/awscli-exe-linux-$ARCH-2.13.9.zip" -o "awscliv2.zip" \
+ && echo "$SHASUM awscliv2.zip" | sha256sum -c - \
+ && unzip awscliv2.zip \
+ && ./aws/install \
+ && rm -rf aws awscliv2.zip ; \
+ fi
 
 RUN case ${TARGETPLATFORM} in \
     "linux/amd64") ARCH=x86_64; SHASUM=a3fb9c1de3512bc91f27cc47297d6d6cf208adee9b64ed719130da59ac13e26b ;; \
     "linux/arm64") ARCH=aarch64; SHASUM=e5165eb592a317e1f6da0ac7fcbccf60d7fb8e5ac1f0d7336a9be51c23308b06 ;; \
+    "linux/s390x") ARCH=s390x; SHASUM=4969ae702488cb79afd14bf91c10b496996852b82a364907c1ebfa5f1667a139 ;; \
   esac && \
  curl -fsSL "https://github.com/NixOS/patchelf/releases/download/0.17.2/patchelf-0.17.2-$ARCH.tar.gz" -o "patchelf.tar.gz" && \
  echo "$SHASUM patchelf.tar.gz" | sha256sum -c - && \
@@ -105,10 +147,11 @@ RUN case ${TARGETPLATFORM} in \
 # build/bootstrap/bootstrap-debian.sh -- if an update is necessary here, it's probably
 # necessary in the agent as well.
 RUN case ${TARGETPLATFORM} in \
-    "linux/amd64") ARCH=amd64; SHASUM=4cb534c52cdd47a6223d4596d530e7c9c785438ab3b0a49ff347e991c210b2cd ;; \
-    "linux/arm64") ARCH=arm64; SHASUM=c1de6860dd4f8d5e2ec270097bd46d6a211b971a0b8b38559784bd051ea950a1 ;; \
+    "linux/amd64") ARCH=amd64; SHASUM=84916c44c8d81cb64f6c9a9f8fd8fa059342e872bfc1ce185f5dcbf70c6aadea ;; \
+    "linux/arm64") ARCH=arm64; SHASUM=7937c941e5140a6a22f6b84919e561b9b77ec49e307852ed0b3cc2a45beace9e ;; \
+    "linux/s390x") ARCH=s390x; SHASUM=df7a5cfe632da022bb2cdc51824e8b04634d86d3ad4a24610c4da758c2e5708f ;; \
   esac \
- && curl -fsSL "https://github.com/bazelbuild/bazelisk/releases/download/v1.10.1/bazelisk-linux-$ARCH" > /tmp/bazelisk \
+ && curl -fsSL "https://github.com/cockroachdb/bazelisk/releases/download/2025-07-14/bazelisk-linux-$ARCH" > /tmp/bazelisk \
  && echo "$SHASUM /tmp/bazelisk" | sha256sum -c - \
  && chmod +x /tmp/bazelisk \
  && mv /tmp/bazelisk /usr/bin/bazel
@@ -119,14 +162,26 @@ RUN ln -sf /usr/bin/llvm-nm /usr/bin/nm
 
 RUN rm -rf /tmp/* /var/lib/apt/lists/*
 
-RUN case ${TARGETPLATFORM} in \
+RUN if [[ ${TARGETPLATFORM} == "linux/s390x" ]]; then \
+  curl -fsSL "https://github.com/benesch/autouseradd/archive/refs/tags/1.3.0.tar.gz" -o autouseradd.tar.gz \
+  && echo "da70cbb00878ab395276b0f6191815a763bc8aa2fc120fb36580f6313de4c41f autouseradd.tar.gz" | sha256sum -c - \
+  && tar -xzf autouseradd.tar.gz \
+  && cd autouseradd-1.3.0 \
+  && make \
+  && make install \
+  && cd .. \
+  && rm -rf autouseradd-1.3.0 \
+  && rm autouseradd.tar.gz ; \
+ else \
+  case ${TARGETPLATFORM} in \
     "linux/amd64") ARCH=amd64; SHASUM=442dae58b727a79f81368127fac141d7f95501ffa05f8c48943d27c4e807deb7 ;; \
     "linux/arm64") ARCH=arm64; SHASUM=b216bebfbe30c3c156144cff07233654e23025e26ab5827058c9b284e130599e ;; \
    esac \
   && curl -fsSL "https://github.com/benesch/autouseradd/releases/download/1.3.0/autouseradd-1.3.0-$ARCH.tar.gz" -o autouseradd.tar.gz \
   && echo "$SHASUM autouseradd.tar.gz" | sha256sum -c - \
   && tar xzf autouseradd.tar.gz --strip-components 1 \
-  && rm autouseradd.tar.gz
+  && rm autouseradd.tar.gz ; \
+  fi
 
 ENTRYPOINT ["autouseradd", "--user", "roach", "--no-create-home"]
 CMD ["/usr/bin/bash"]

build/teamcity/internal/cockroach/build/ci/build-and-push-bazel-builder-image.sh

@@ -15,7 +15,7 @@ docker_login_gcr "$gar_repository" "$IMAGE_BUILDER_GOOGLE_CREDENTIALS"
 
 TAG=$(date +%Y%m%d-%H%M%S)
 docker buildx create --name "builder-$TAG" --use
-docker buildx build --push --platform linux/amd64,linux/arm64 -t "$gar_repository:$TAG" -t "$gar_repository:latest-do-not-use" build/bazelbuilder
+docker buildx build --push --platform linux/amd64,linux/arm64,linux/s390x -t "$gar_repository:$TAG" -t "$gar_repository:latest-do-not-use" build/bazelbuilder
 
 if [[ "$open_pr_on_success" == "true" ]]; then
     # Trigger "Open New Bazel Builder Image PR".

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -227,11 +227,11 @@ func alterChangefeedPlanHook(
 		newPayload.Details = jobspb.WrapPayloadDetails(newDetails)
 		newPayload.Description = jobRecord.Description
 		newPayload.DescriptorIDs = jobRecord.DescriptorIDs
-		newExpiration, err := newOptions.GetPTSExpiration()
-		if err != nil {
-			return err
-		}
-		newPayload.MaximumPTSAge = newExpiration
+
+		// The maximum PTS age on jobRecord will be set correctly (based on either
+		// the option or cluster setting) by createChangefeedJobRecord.
+		newPayload.MaximumPTSAge = jobRecord.MaximumPTSAge
+
 		j, err := p.ExecCfg().JobRegistry.LoadJobWithTxn(ctx, jobID, p.InternalSQLTxn())
 		if err != nil {
 			return err

pkg/ccl/changefeedccl/protected_timestamps_test.go

@@ -424,27 +424,114 @@ func TestChangefeedCanceledWhenPTSIsOld(t *testing.T) {
 		// single row with multiple versions.
 		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b INT)`)
 
-		feed, err := f.Feed("CREATE CHANGEFEED FOR TABLE foo WITH protect_data_from_gc_on_pause, gc_protect_expires_after='24h'")
-		require.NoError(t, err)
-		defer func() {
-			closeFeed(t, feed)
-		}()
+		t.Run("canceled due to gc_protect_expires_after option", func(t *testing.T) {
+			testutils.RunValues(t, "initially-protected-with", []string{"none", "option", "setting"},
+				func(t *testing.T, initialProtect string) {
+					defer func() {
+						sqlDB.Exec(t, `RESET CLUSTER SETTING changefeed.protect_timestamp.max_age`)
+					}()
+
+					if initialProtect == "option" {
+						// We set the cluster setting to something small to make sure that
+						// the option alone is able to protect the PTS record.
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '1us'`)
+					} else {
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '24h'`)
+					}
 
-		jobFeed := feed.(cdctest.EnterpriseTestFeed)
-		require.NoError(t, jobFeed.Pause())
+					feedStmt := `CREATE CHANGEFEED FOR TABLE foo`
+					switch initialProtect {
+					case "none":
+						feedStmt += ` WITH gc_protect_expires_after='1us'`
+					case "option":
+						feedStmt += ` WITH gc_protect_expires_after='24h'`
+					}
+
+					feed, err := f.Feed(feedStmt)
+					require.NoError(t, err)
+					defer func() {
+						closeFeed(t, feed)
+					}()
 
-		// While the job is paused, take opportunity to test that alter changefeed
-		// works when setting gc_protect_expires_after option.
+					jobFeed := feed.(cdctest.EnterpriseTestFeed)
 
-		// Verify we can set it to 0 -- i.e. disable.
-		sqlDB.Exec(t, fmt.Sprintf("ALTER CHANGEFEED %d SET gc_protect_expires_after = '0s'", jobFeed.JobID()))
-		// Now, set it to something very small.
-		sqlDB.Exec(t, fmt.Sprintf("ALTER CHANGEFEED %d SET gc_protect_expires_after = '250ms'", jobFeed.JobID()))
+					if initialProtect != "none" {
+						require.NoError(t, jobFeed.Pause())
+
+						// Wait a little bit and make sure the job ISN'T canceled.
+						require.ErrorContains(t, jobFeed.WaitDurationForState(10*time.Second, func(s jobs.State) bool {
+							return s == jobs.StateCanceled
+						}), `still waiting for job status; current status is "paused"`)
+
+						if initialProtect == "option" {
+							// Set the cluster setting back to something high to make sure the
+							// option alone can cause the changefeed to be canceled.
+							sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '24h'`)
+						}
+
+						// Set option to something small so that job will be canceled.
+						sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET gc_protect_expires_after = '1us'`, jobFeed.JobID()))
+					}
 
-		// Stale PTS record should trigger job cancellation.
-		require.NoError(t, jobFeed.WaitForState(func(s jobs.State) bool {
-			return s == jobs.StateCanceled
-		}))
+					// Stale PTS record should trigger job cancellation.
+					require.NoError(t, jobFeed.WaitForState(func(s jobs.State) bool {
+						return s == jobs.StateCanceled
+					}))
+				})
+		})
+
+		t.Run("canceled due to changefeed.protect_timestamp.max_age setting", func(t *testing.T) {
+			testutils.RunValues(t, "initially-protected-with", []string{"none", "option", "setting"},
+				func(t *testing.T, initialProtect string) {
+					defer func() {
+						sqlDB.Exec(t, `RESET CLUSTER SETTING changefeed.protect_timestamp.max_age`)
+					}()
+
+					if initialProtect == "setting" {
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '24h'`)
+					} else {
+						sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '1us'`)
+					}
+
+					// Set the max age cluster setting to something small.
+					feedStmt := `CREATE CHANGEFEED FOR TABLE foo`
+					if initialProtect == "option" {
+						feedStmt += ` WITH gc_protect_expires_after='24h'`
+					}
+					feed, err := f.Feed(feedStmt)
+					require.NoError(t, err)
+					defer func() {
+						closeFeed(t, feed)
+					}()
+
+					jobFeed := feed.(cdctest.EnterpriseTestFeed)
+
+					if initialProtect != "none" {
+						require.NoError(t, jobFeed.Pause())
+
+						// Wait a little bit and make sure the job ISN'T canceled.
+						require.ErrorContains(t, jobFeed.WaitDurationForState(10*time.Second, func(s jobs.State) bool {
+							return s == jobs.StateCanceled
+						}), `still waiting for job status; current status is "paused"`)
+
+						switch initialProtect {
+						case "option":
+							// Reset the option so that it defaults to the cluster setting.
+							sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET gc_protect_expires_after = '0s'`, jobFeed.JobID()))
+						case "setting":
+							// Modify the cluster setting and do an ALTER CHANGEFEED so that
+							// the new value is picked up.
+							sqlDB.Exec(t, `SET CLUSTER SETTING changefeed.protect_timestamp.max_age = '1us'`)
+							sqlDB.Exec(t, fmt.Sprintf(`ALTER CHANGEFEED %d SET diff`, jobFeed.JobID()))
+						}
+					}
+
+					// Stale PTS record should trigger job cancellation.
+					require.NoError(t, jobFeed.WaitForState(func(s jobs.State) bool {
+						return s == jobs.StateCanceled
+					}))
+				})
+		})
 	}
 
 	cdcTestWithSystem(t, testFn, feedTestEnterpriseSinks)

pkg/ccl/changefeedccl/testfeed_test.go

@@ -491,7 +491,7 @@ func (f *jobFeed) WaitDurationForState(
 		if statusPred(jobs.State(status)) {
 			return nil
 		}
-		return errors.Newf("still waiting for job status; current %s", status)
+		return errors.Newf("still waiting for job status; current status is %q", status)
 	}, dur)
 }
 

pkg/kv/kvclient/kvcoord/BUILD.bazel

@@ -159,6 +159,7 @@ go_test(
         "txn_interceptor_pipeliner_test.go",
         "txn_interceptor_seq_num_allocator_test.go",
         "txn_interceptor_span_refresher_test.go",
+        "txn_interceptor_write_buffer_client_test.go",
         "txn_interceptor_write_buffer_test.go",
         "txn_test.go",
         ":bufferedwrite_interval_btree_test.go",  # keep

pkg/kv/kvclient/kvcoord/txn_coord_sender.go

@@ -281,6 +281,10 @@ func newRootTxnCoordSender(
 		timeSource: timeutil.DefaultTimeSource{},
 		txn:        &tcs.mu.txn,
 	}
+	tcs.interceptorAlloc.txnWriteBuffer.init(
+		&tcs.interceptorAlloc.txnPipeliner,
+	)
+
 	tcs.initCommonInterceptors(tcf, txn, kv.RootTxn)
 
 	// Once the interceptors are initialized, piece them all together in the
@@ -446,6 +450,7 @@ func (tc *TxnCoordSender) DisablePipelining() error {
 	if tc.mu.active {
 		return errors.Errorf("cannot disable pipelining on a running transaction")
 	}
+	tc.interceptorAlloc.txnPipeliner.disabledExplicitly = true
 	tc.interceptorAlloc.txnPipeliner.disabled = true
 	return nil
 }
@@ -1192,6 +1197,9 @@ func (tc *TxnCoordSender) SetBufferedWritesEnabled(enabled bool) {
 		panic("cannot enable buffered writes on a running transaction")
 	}
 	tc.interceptorAlloc.txnWriteBuffer.setEnabled(enabled)
+	if enabled {
+		tc.interceptorAlloc.txnPipeliner.disabled = true
+	}
 }
 
 // BufferedWritesEnabled is part of the kv.TxnSender interface.

pkg/kv/kvclient/kvcoord/txn_interceptor_pipeliner.go

@@ -226,9 +226,14 @@ var rejectTxnMaxCount = settings.RegisterIntSetting(
 // attached to any end transaction request that is passed through the pipeliner
 // to ensure that they the locks within them are released.
 type txnPipeliner struct {
-	st                       *cluster.Settings
-	riGen                    rangeIteratorFactory // used to condense lock spans, if provided
-	wrapped                  lockedSender
+	st      *cluster.Settings
+	riGen   rangeIteratorFactory // used to condense lock spans, if provided
+	wrapped lockedSender
+	// disabledExplicitly tracks whether the user called txn.DisablePipelining().
+	// This is separate from disabled so that the txnWriteBuffer can enable
+	// pipelining if it has flushed its buffer iff pipelining wasn't previously
+	// explicitly disabled.
+	disabledExplicitly       bool
 	disabled                 bool
 	txnMetrics               *TxnMetrics
 	condensedIntentsEveryN   *log.EveryN
@@ -587,6 +592,12 @@ func (tp *txnPipeliner) canUseAsyncConsensus(ctx context.Context, ba *kvpb.Batch
 	return true
 }
 
+// enableImplicitPipelining enables pipelining unless pipelining was explicitly
+// disabled previously.
+func (tp *txnPipeliner) enableImplicitPipelining() {
+	tp.disabled = tp.disabledExplicitly
+}
+
 // chainToInFlightWrites ensures that we "chain" on to any in-flight writes that
 // overlap the keys we're trying to read/write. We do this by prepending
 // QueryIntent requests with the ErrorIfMissing option before each request that