Merge cockroachdb#107199 cockroachdb#108211 cockroachdb#108538 cockroachdb#108550 cockroachdb#108558

107199: cli: simplify output assertion in TestUnavailableZip r=ericharmeling a=ericharmeling

This commit removes the `datadriven.RunTest` call from `TestUnavailableZip`, as this unit test is meant to test the ability of debug zip to output while nodes are unavailable, and not meant to assert specific, non-deterministic output.

Fixes cockroachdb#106961.

Release note: None

108211: plpgsql: add support for ELSIF branches r=DrewKimball a=DrewKimball

This patch adds support for executing PLpgSQL `IF` statements with `ELSIF` branches (else if). `IF` statements were already executed as CASE statements under the hood, so this change only requires building the `ELSIF` branches and appending them to the `whens` list.

Informs cockroachdb#105254

Release note (sql change): Added support for specifying PLpgSQL `IF` statements with `ELSIF` branches.

108538: internal/sqlsmith: don't generate UDFs with collated strings r=yuzefovich a=yuzefovich

Collated strings are not a valid return type for UDFs in Postgres, so they are disallowed. We already had the check in most places, and the only exception was if we created a mutation with a single RETURNING column of collated string type. This omission is now fixed.

Fixes: cockroachdb#108006.

Release note: None

108550: kv: add error return from Txn.PrepareForRetry r=nvanbenschoten a=nvanbenschoten

Instead of using `log.Fatal`, return assertion errors.

Epic: None
Release note: None

108558: kvstreamer: fix up some comments r=yuzefovich a=yuzefovich

As of d564a0c we no longer use FIFO order in the OutOfOrder requests provider, so this commit adjusts the comments accordingly.

Epic: None

Release note: None

Co-authored-by: Eric Harmeling <[email protected]>
Co-authored-by: Drew Kimball <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: Nathan VanBenschoten <[email protected]>

Author: 5 people

pkg/cli/testdata/zip/unavailable

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/keys"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/liveness"
@@ -299,7 +300,6 @@ create table defaultdb."../system"(x int);
 func TestUnavailableZip(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 
-	skip.WithIssue(t, 107738, "flaky")
 	skip.UnderShort(t)
 	// Race builds make the servers so slow that they report spurious
 	// unavailability.
@@ -317,35 +317,51 @@ func TestUnavailableZip(t *testing.T) {
 	close(closedCh)
 	unavailableCh.Store(closedCh)
 	knobs := &kvserver.StoreTestingKnobs{
-		TestingRequestFilter: func(ctx context.Context, _ *kvpb.BatchRequest) *kvpb.Error {
-			select {
-			case <-unavailableCh.Load().(chan struct{}):
-			case <-ctx.Done():
+		TestingRequestFilter: func(ctx context.Context,
+			br *kvpb.BatchRequest) *kvpb.Error {
+			if br.Header.GatewayNodeID == 2 {
+				// For node 2 connections, block all replica requests.
+				select {
+				case <-unavailableCh.Load().(chan struct{}):
+				case <-ctx.Done():
+				}
+			} else if br.Header.GatewayNodeID == 1 {
+				// For node 1 connections, only block requests to table data ranges.
+				if br.Requests[0].GetInner().Header().Key.Compare(keys.
+					TableDataMin) >= 0 {
+					select {
+					case <-unavailableCh.Load().(chan struct{}):
+					case <-ctx.Done():
+					}
+				}
 			}
 			return nil
 		},
 	}
 
-	// Make a 2-node cluster, with an option to make the first node unavailable.
-	tc := testcluster.StartTestCluster(t, 2, base.TestClusterArgs{
-		ServerArgsPerNode: map[int]base.TestServerArgs{
-			0: {Insecure: true, Knobs: base.TestingKnobs{Store: knobs}},
-			1: {Insecure: true},
-		},
-	})
+	// Make a 3-node cluster, with an option to block replica requests.
+	tc := testcluster.StartTestCluster(t, 3,
+		base.TestClusterArgs{ServerArgs: base.TestServerArgs{Insecure: true,
+			Knobs: base.TestingKnobs{Store: knobs}},
+		})
 	defer tc.Stopper().Stop(context.Background())
 
-	// Sanity test: check that a simple operation works.
+	// Sanity test: check that a simple SQL operation works against node 1.
 	if _, err := tc.ServerConn(0).Exec("SELECT * FROM system.users"); err != nil {
 		t.Fatal(err)
 	}
 
-	// Make the first two nodes unavailable.
+	// Block querying table data from node 1.
+	// Block all replica requests from node 2.
 	ch := make(chan struct{})
 	unavailableCh.Store(ch)
 	defer close(ch)
 
-	// Zip it. We fake a CLI test context for this.
+	// Run debug zip against node 1.
+	debugZipCommand :=
+		"debug zip --concurrency=1 --cpu-profile-duration=0 " + os.
+			DevNull + " --timeout=.25s"
+
 	c := TestCLI{
 		t:        t,
 		Server:   tc.Server(0),
@@ -354,20 +370,57 @@ func TestUnavailableZip(t *testing.T) {
 	defer func(prevStderr *os.File) { stderr = prevStderr }(stderr)
 	stderr = os.Stdout
 
-	// Keep the timeout short so that the test doesn't take forever.
-	out, err := c.RunWithCapture("debug zip --concurrency=1 --cpu-profile-duration=0 " + os.DevNull + " --timeout=.5s")
+	out, err := c.RunWithCapture(debugZipCommand)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// Strip any non-deterministic messages.
-	out = eraseNonDeterministicZipOutput(out)
-	out = eraseNonDeterministicErrors(out)
+	// Assert debug zip output for cluster, node 1, node 2, node 3.
+	assert.NotEmpty(t, out)
+	assert.Contains(t, out, "[cluster] requesting nodes... received response...")
+	assert.Contains(t, out, "[cluster] requesting liveness... received response...")
+	for i := 1; i < tc.NumServers()+1; i++ {
+		assert.Contains(t, out, fmt.Sprintf("[node %d] using SQL connection URL",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] retrieving SQL data",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] requesting stacks... received response...",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] requesting stacks with labels... received response...",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] requesting heap file list... received response...",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] requesting goroutine dump list... received response...",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] requesting log files list... received response...",
+			i))
+		assert.Contains(t, out, fmt.Sprintf("[node %d] requesting ranges... received response...",
+			i))
+	}
+
+	// Run debug zip against node 2.
+	c = TestCLI{
+		t:        t,
+		Server:   tc.Server(1),
+		Insecure: true,
+	}
+
+	out, err = c.RunWithCapture(debugZipCommand)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Assert debug zip output for cluster, node 2.
+	assert.NotEmpty(t, out)
+	assert.Contains(t, out, "[cluster] requesting nodes... received response...")
+	assert.Contains(t, out, "[cluster] requesting liveness... received response...")
+	assert.Contains(t, out, "[node 2] using SQL connection URL")
+	assert.Contains(t, out, "[node 2] retrieving SQL data")
+	assert.Contains(t, out, "[node 2] requesting ranges... received response...")
+	assert.Contains(t, out, "[node 2] writing ranges...")
+	assert.NotContains(t, out, "[node 1]")
+	assert.NotContains(t, out, "[node 3]")
 
-	datadriven.RunTest(t, datapathutils.TestDataPath(t, "zip", "unavailable"),
-		func(t *testing.T, td *datadriven.TestData) string {
-			return out
-		})
 }
 
 func eraseNonDeterministicZipOutput(out string) string {
@@ -407,21 +460,6 @@ func eraseNonDeterministicZipOutput(out string) string {
 	return out
 }
 
-func eraseNonDeterministicErrors(out string) string {
-	// In order to avoid non-determinism here, we erase the output of
-	// the range retrieval.
-	re := regexp.MustCompile(`(?m)^(requesting ranges.*found|writing: debug/nodes/\d+/ranges).*\n`)
-	out = re.ReplaceAllString(out, ``)
-
-	re = regexp.MustCompile(`(?m)^\[cluster\] requesting data for debug\/settings.*\n`)
-	out = re.ReplaceAllString(out, ``)
-
-	// In order to avoid non-determinism here, we truncate error messages.
-	re = regexp.MustCompile(`(?m)last request failed: .*$`)
-	out = re.ReplaceAllString(out, `last request failed: ...`)
-	return out
-}
-
 // This tests the operation of zip over partial clusters.
 //
 // We cannot combine this test with TestZip above because TestPartialZip

pkg/cli/zip_test.go

@@ -1043,7 +1043,7 @@ func (s *Smither) makeCreateFunc() (cf *tree.CreateRoutine, ok bool) {
 				// If the rtype isn't a RECORD, change it to rrefs or RECORD depending
 				// how many columns there are to avoid return type mismatch errors.
 				if rtyp != types.AnyTuple {
-					if len(rrefs) == 1 && s.coin() {
+					if len(rrefs) == 1 && s.coin() && rrefs[0].typ.Family() != types.CollatedStringFamily {
 						rtyp = rrefs[0].typ
 					} else {
 						rtyp = types.AnyTuple

pkg/internal/sqlsmith/relational.go

@@ -872,7 +872,7 @@ func TestPreservingSteppingOnSenderReplacement(t *testing.T) {
 		require.NotEqual(t, pErr.TxnID, pErr.Transaction.ID)
 
 		// Reset the handle in order to get a new sender.
-		txn.PrepareForRetry(ctx)
+		require.NoError(t, txn.PrepareForRetry(ctx))
 
 		// Make sure we have a new txn ID.
 		require.NotEqual(t, pErr.TxnID, txn.ID())

pkg/kv/db_test.go

@@ -127,7 +127,9 @@ func TestSavepoints(t *testing.T) {
 
 			case "reset":
 				prevID := txn.ID()
-				txn.PrepareForRetry(ctx)
+				if err := txn.PrepareForRetry(ctx); err != nil {
+					t.Fatal(err)
+				}
 				changed := "changed"
 				if prevID == txn.ID() {
 					changed = "not changed"

pkg/kv/kvclient/kvcoord/txn_coord_sender_savepoints_test.go

@@ -912,7 +912,9 @@ func TestTxnCoordSenderTxnUpdatedOnError(t *testing.T) {
 			stopper.Stop(ctx)
 
 			if test.callPrepareForRetry {
-				txn.PrepareForRetry(ctx)
+				if err := txn.PrepareForRetry(ctx); err != nil {
+					t.Fatal(err)
+				}
 			}
 			if test.name != "nil" && err == nil {
 				t.Fatalf("expected an error")

pkg/kv/kvclient/kvcoord/txn_coord_sender_test.go

@@ -244,8 +244,7 @@ func (b *requestsProviderBase) close() {
 }
 
 // outOfOrderRequestsProvider is a requestProvider that returns requests in an
-// arbitrary order (namely in the same order as the requests are enqueued and
-// added).
+// arbitrary order (namely in the LIFO order of requests being enqueued).
 type outOfOrderRequestsProvider struct {
 	*requestsProviderBase
 }
@@ -283,6 +282,8 @@ func (p *outOfOrderRequestsProvider) nextLocked() singleRangeBatch {
 	if len(p.requests) == 0 {
 		panic(errors.AssertionFailedf("nextLocked called when requestsProvider is empty"))
 	}
+	// Use the last request so that we could reuse its slot if resume request is
+	// added.
 	return p.requests[len(p.requests)-1]
 }
 
@@ -291,6 +292,8 @@ func (p *outOfOrderRequestsProvider) removeNextLocked() {
 	if len(p.requests) == 0 {
 		panic(errors.AssertionFailedf("removeNextLocked called when requestsProvider is empty"))
 	}
+	// Use the last request so that we could reuse its slot if resume request is
+	// added.
 	p.requests = p.requests[:len(p.requests)-1]
 }
 

pkg/kv/kvclient/kvstreamer/requests_provider.go

@@ -994,15 +994,17 @@ func (txn *Txn) exec(ctx context.Context, fn func(context.Context, *Txn) error)
 			break
 		}
 
-		txn.PrepareForRetry(ctx)
+		if err := txn.PrepareForRetry(ctx); err != nil {
+			return err
+		}
 	}
 
 	return err
 }
 
 // PrepareForRetry needs to be called before a retry to perform some
 // book-keeping and clear errors when possible.
-func (txn *Txn) PrepareForRetry(ctx context.Context) {
+func (txn *Txn) PrepareForRetry(ctx context.Context) error {
 	// Reset commit triggers. These must be reconfigured by the client during the
 	// next retry.
 	txn.commitTriggers = nil
@@ -1012,11 +1014,11 @@ func (txn *Txn) PrepareForRetry(ctx context.Context) {
 
 	retryErr := txn.mu.sender.GetTxnRetryableErr(ctx)
 	if retryErr == nil {
-		return
+		return nil
 	}
 	if txn.typ != RootTxn {
-		panic(errors.WithContextTags(errors.NewAssertionErrorWithWrappedErrf(
-			retryErr, "PrepareForRetry() called on leaf txn"), ctx))
+		return errors.WithContextTags(errors.NewAssertionErrorWithWrappedErrf(
+			retryErr, "PrepareForRetry() called on leaf txn"), ctx)
 	}
 	log.VEventf(ctx, 2, "retrying transaction: %s because of a retryable error: %s",
 		txn.debugNameLocked(), retryErr)
@@ -1029,23 +1031,24 @@ func (txn *Txn) PrepareForRetry(ctx context.Context) {
 		// txn IDs. However, at no point can both the old and new incarnation of a
 		// transaction be active at the same time -- this would constitute a
 		// programming error.
-		log.Fatalf(
-			ctx,
-			"unexpected retryable error for old incarnation of the transaction %s; current incarnation %s",
-			retryErr.TxnID,
-			txn.mu.ID,
-		)
+		return errors.WithContextTags(
+			errors.NewAssertionErrorWithWrappedErrf(
+				retryErr,
+				"unexpected retryable error for old incarnation of the transaction %s; current incarnation %s",
+				retryErr.TxnID,
+				txn.mu.ID,
+			), ctx)
 	}
 
 	if !retryErr.PrevTxnAborted() {
 		// If the retryable error doesn't correspond to an aborted transaction,
 		// there's no need to switch out the transaction. We simply clear the
 		// retryable error and proceed.
 		txn.mu.sender.ClearTxnRetryableErr(ctx)
-		return
+		return nil
 	}
 
-	txn.handleTransactionAbortedErrorLocked(ctx, retryErr)
+	return txn.handleTransactionAbortedErrorLocked(ctx, retryErr)
 }
 
 // Send runs the specified calls synchronously in a single batch and
@@ -1354,10 +1357,11 @@ func (txn *Txn) UpdateStateOnRemoteRetryableErr(ctx context.Context, pErr *kvpb.
 // the current one with it.
 func (txn *Txn) handleTransactionAbortedErrorLocked(
 	ctx context.Context, retryErr *kvpb.TransactionRetryWithProtoRefreshError,
-) {
+) error {
 	if !retryErr.PrevTxnAborted() {
 		// Sanity check we're dealing with a TransactionAbortedError.
-		log.Fatalf(ctx, "cannot replace root sender if txn not aborted: %v", retryErr)
+		return errors.WithContextTags(errors.NewAssertionErrorWithWrappedErrf(
+			retryErr, "cannot replace root sender if txn not aborted"), ctx)
 	}
 
 	// The transaction we had been using thus far has been aborted. The proto
@@ -1368,6 +1372,7 @@ func (txn *Txn) handleTransactionAbortedErrorLocked(
 	prevSteppingMode := txn.mu.sender.GetSteppingMode(ctx)
 	txn.mu.sender = txn.db.factory.RootTransactionalSender(newTxn, txn.mu.userPriority)
 	txn.mu.sender.ConfigureStepping(ctx, prevSteppingMode)
+	return nil
 }
 
 // SetFixedTimestamp makes the transaction run in an unusual way, at a "fixed

pkg/kv/txn.go

@@ -552,11 +552,14 @@ func cleanupAndFinishOnError(args fsm.Args) error {
 
 func prepareTxnForRetry(args fsm.Args) error {
 	ts := args.Extended.(*txnState)
-	func() {
+	err := func() error {
 		ts.mu.Lock()
 		defer ts.mu.Unlock()
-		ts.mu.txn.PrepareForRetry(ts.Ctx)
+		return ts.mu.txn.PrepareForRetry(ts.Ctx)
 	}()
+	if err != nil {
+		return err
+	}
 	ts.setAdvanceInfo(
 		advanceOne,
 		noRewind,
@@ -587,13 +590,16 @@ func moveToCommitWaitAfterInternalCommit(args fsm.Args) error {
 func prepareTxnForRetryWithRewind(args fsm.Args) error {
 	pl := args.Payload.(eventRetriableErrPayload)
 	ts := args.Extended.(*txnState)
-	func() {
+	err := func() error {
 		ts.mu.Lock()
 		defer ts.mu.Unlock()
-		ts.mu.txn.PrepareForRetry(ts.Ctx)
 		ts.mu.autoRetryReason = pl.err
 		ts.mu.autoRetryCounter++
+		return ts.mu.txn.PrepareForRetry(ts.Ctx)
 	}()
+	if err != nil {
+		return err
+	}
 	// The caller will call rewCap.rewindAndUnlock().
 	ts.setAdvanceInfo(
 		rewind,