sql: disable write pipelining for implicit txns used by COPY

[yuzefovich]

This commit makes it so that we disable write pipelining on implicit txns used by COPY. The rationale for that is that these are throughput-oriented txns which build large batches and shouldn’t be getting much benefit from write pipelining. It also introduces a session variable to be able to go back to the old behavior should we need to.

Note that there are two places where we deal with txns in COPY:

• the initial txn is created by the connExecutor state machine. It also might be explicit. In this case disabling the write pipelining is done only for the implicit txn.
• for implicit txns with non-atomic COPY, we create a fresh txn after each batch.

Epic: None

Release note: None

[cockroach-teamcity]

This change is 

[nvanbenschoten]

Reviewed 8 of 8 files at r1, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @rafiss, @rharding6373, and @yuzefovich)

---

-- commits line 13 at r1:
I feel like we shouldn't bother messing with explicit transactions. Doing so introduces confusing user-visible behavior to subsequent statements in a transaction which runs a COPY.

[yuzefovich]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @nvanbenschoten, @rafiss, and @rharding6373)

---

-- commits line 13 at r1:

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

I feel like we shouldn't bother messing with explicit transactions. Doing so introduces confusing user-visible behavior to subsequent statements in a transaction which runs a COPY.

Ack, switched to disable pipelining only for implicit txns.

[nvanbenschoten]

This feels difficult to test, so I'll defer to you on whether we can add any unit tests for the change. If not though, have you manually tested with tracing to verify that this has the desired effect?

Reviewed 2 of 2 files at r2, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @rafiss and @rharding6373)

[yuzefovich]

On top of #115674, I added the following diff:

diff --git a/pkg/kv/kvclient/kvcoord/txn_coord_sender.go b/pkg/kv/kvclient/kvcoord/txn_coord_sender.go
index 1797935d6dd..2bf49f00a5f 100644
--- a/pkg/kv/kvclient/kvcoord/txn_coord_sender.go
+++ b/pkg/kv/kvclient/kvcoord/txn_coord_sender.go
@@ -422,6 +422,12 @@ func (tc *TxnCoordSender) DisablePipelining() error {
        return nil
 }
 
+func (tc *TxnCoordSender) PipeliningDisabled() bool {
+       tc.mu.Lock()
+       defer tc.mu.Unlock()
+       return tc.interceptorAlloc.txnPipeliner.disabled
+}
+
 func generateTxnDeadlineExceededErr(txn *roachpb.Transaction, deadline hlc.Timestamp) *kvpb.Error {
        exceededBy := txn.WriteTimestamp.GoTime().Sub(deadline.GoTime())
        extraMsg := redact.Sprintf(
diff --git a/pkg/kv/mock_transactional_sender.go b/pkg/kv/mock_transactional_sender.go
index 2f83e4a5079..3c2b467781b 100644
--- a/pkg/kv/mock_transactional_sender.go
+++ b/pkg/kv/mock_transactional_sender.go
@@ -235,6 +235,8 @@ func (m *MockTransactionalSender) Active() bool {
 // DisablePipelining is part of the kv.TxnSender interface.
 func (m *MockTransactionalSender) DisablePipelining() error { return nil }
 
+func (m *MockTransactionalSender) PipeliningDisabled() bool { return false }
+
 // Step is part of the TxnSender interface.
 func (m *MockTransactionalSender) Step(context.Context, bool) error {
        // At least one test (e.g sql/TestPortalsDestroyedOnTxnFinish) requires
diff --git a/pkg/kv/sender.go b/pkg/kv/sender.go
index 6d26e62e02e..3067a2fd3d8 100644
--- a/pkg/kv/sender.go
+++ b/pkg/kv/sender.go
@@ -232,6 +232,8 @@ type TxnSender interface {
        // merges ranges together.
        DisablePipelining() error
 
+       PipeliningDisabled() bool
+
        // ReadTimestamp returns the transaction's current read timestamp.
        // Note a transaction can be internally pushed forward in time
        // before committing so this is not guaranteed to be the commit
diff --git a/pkg/kv/txn.go b/pkg/kv/txn.go
index 4120111d365..bc1d916b1aa 100644
--- a/pkg/kv/txn.go
+++ b/pkg/kv/txn.go
@@ -1726,6 +1726,10 @@ func (txn *Txn) AdmissionHeader() kvpb.AdmissionHeader {
        return h
 }
 
+func (txn *Txn) PipeliningDisabled() bool {
+       return txn.mu.sender.PipeliningDisabled()
+}
+
 // OnePCNotAllowedError signifies that a request had the Require1PC flag set,
 // but 1PC evaluation was not possible for one reason or another.
 type OnePCNotAllowedError struct{}
diff --git a/pkg/sql/copy/copy_in_test.go b/pkg/sql/copy/copy_in_test.go
index b4f8d016af1..fb744a86aa2 100644
--- a/pkg/sql/copy/copy_in_test.go
+++ b/pkg/sql/copy/copy_in_test.go
@@ -514,6 +514,9 @@ func TestCopyFromRetries(t *testing.T) {
                                                                txn.AdmissionHeader().Priority, admissionpb.UserLowPri,
                                                        )
                                                }
+                                               if !txn.PipeliningDisabled() {
+                                                       t.Errorf("pipelining should be disabled")
+                                               }
                                        }
                                        attemptNumber++
                                        return tc.hook(attemptNumber)

which asserts that whenever COPY is executed outside of explicit txns (and within different scenarios), it has the pipelining disabled (and the test passes). Do you think it's worth including such an extension of kv.Txn API for this test case? To me it doesn't seem justified. Or did you have something else in mind in terms of testing rather than simply asserting that pipelining is disabled on all implicit txns?

[yuzefovich]

I think the test I temporarily added is sufficient confirmation that the change works as expected and don't seem worthy of checking it in, so I'll omit it from the PR.

TFTR!

bors r+

[yuzefovich]

Actually, I see a minor nit, one sec.

bors r-

[craig]

Canceled.

[yuzefovich]

bors r+

[craig]

Build succeeded:

• Bazel Essential CI (Cockroach)