crosscluster: disable write buffering

The txnWriteBuffer does not yet support the OriginTimestamp and
OriginID write options used by LDR.  Soon, it will reject requests
with these options set.

This opts LDR out of this setting even in a cluster were the cluster
setting has changed the default.

Epic: none
Release note: None

Author: stevendanna

pkg/crosscluster/logical/lww_kv_processor.go

@@ -224,6 +224,7 @@ func (p *kvRowProcessor) processOneRow(
 	refreshCount int,
 ) error {
 	if err := p.cfg.DB.KV().Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+		txn.SetBufferedWritesEnabled(false)
 		b := makeKVBatch(useLowPriority.Get(&p.cfg.Settings.SV), txn)
 
 		if err := p.addToBatch(ctx, txn, b, dstTableID, row, k, prevValue); err != nil {

pkg/crosscluster/logical/lww_row_processor.go

@@ -388,8 +388,9 @@ func (srp *sqlRowProcessor) GetLastRow() cdcevent.Row {
 }
 
 var (
-	forceGenericPlan = sessiondatapb.PlanCacheModeForceGeneric
-	ieOverrideBase   = sessiondata.InternalExecutorOverride{
+	bufferedWritesEnabled = false
+	forceGenericPlan      = sessiondatapb.PlanCacheModeForceGeneric
+	ieOverrideBase        = sessiondata.InternalExecutorOverride{
 		// The OriginIDForLogicalDataReplication session variable will bind the
 		// origin ID 1 to each per-statement batch request header sent by the
 		// internal executor. This metadata will be plumbed to the MVCCValueHeader
@@ -412,8 +413,9 @@ var (
 		GrowStackSize: true,
 		// We don't get any benefits from generating plan gists for internal
 		// queries, so we disable them.
-		DisablePlanGists: true,
-		QualityOfService: &sessiondatapb.BulkLowQoS,
+		DisablePlanGists:      true,
+		QualityOfService:      &sessiondatapb.BulkLowQoS,
+		BufferedWritesEnabled: &bufferedWritesEnabled,
 	}
 )
 

pkg/sql/internal.go

@@ -962,6 +962,9 @@ func applyOverrides(o sessiondata.InternalExecutorOverride, sd *sessiondata.Sess
 	if o.DisablePlanGists {
 		sd.DisablePlanGists = true
 	}
+	if o.BufferedWritesEnabled != nil {
+		sd.BufferedWritesEnabled = *o.BufferedWritesEnabled
+	}
 
 	if o.MultiOverride != "" {
 		overrides := strings.Split(o.MultiOverride, ",")

pkg/sql/sessiondata/internal.go

@@ -84,6 +84,9 @@ type InternalExecutorOverride struct {
 	GrowStackSize bool
 	// DisablePlanGists, if true, overrides the disable_plan_gists session var.
 	DisablePlanGists bool
+	// BufferedWritesEnabled, if set, controls whether the buffered writes KV transaction
+	// protocol is used for user queries on the current session.
+	BufferedWritesEnabled *bool
 }
 
 // NoSessionDataOverride is the empty InternalExecutorOverride which does not