kvcoord: deterministic error ordering

DistSender splits requests up in range order and collects errors in the
order requests were dispatched. If there are multiple errors, the
highest priority error is returned and ties are broken by dispatch
order. Now, the merge logic consults the index of the request that
generated the error. This is particularly useful for CPuts as it allows
the caller to observe the first cput in the batch that failed.

The deterministic order allows LDR, `deleteSwap`, and `updateSwap` to
guarantee the cput error is for the primary key. This property is
essential because they are using cput to optimistically guess at the
value of the row in the database. The batches may also contain cput
failures for unique indexes, but those errors should only be raised to
the user if the primary key cput succeeded.

This is still a bit of a hack. In a perfect world, the cput failure
would be treated as a result instead of an error. That would allow the
caller to inspect each cput result independently and would make it clear
that cput failures do not poison any of the other requests in the
batch.

Informs: cockroachdb#146117
Release note: fixes an issue with LDR where the presence of a unique
index may cause spurious DLQ entries if the unique index has a smaller
index id than the primary key index.

Author: jeffswenson

pkg/crosscluster/logical/batch_handler_test.go

@@ -183,8 +183,7 @@ func TestBatchHandlerExhaustive(t *testing.T) {
 	if uniqueConstraint {
 		runner.Exec(t, `ALTER TABLE test_table ADD CONSTRAINT unique_value UNIQUE (value)`)
 	}
-	// TODO(jeffswenson): enable this metamorphic option after fixing #146465.
-	if false && addAndRemoveColumn {
+	if addAndRemoveColumn {
 		runner.Exec(t, `ALTER TABLE test_table ADD COLUMN temp_col INT`)
 		runner.Exec(t, `ALTER TABLE test_table DROP COLUMN temp_col`)
 	}

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -1684,9 +1684,19 @@ func maybeSwapErrorIndex(pErr *kvpb.Error, a, b int) {
 }
 
 // mergeErrors merges the two errors, combining their transaction state and
-// returning the error with the highest priority.
+// returning the error with the highest priority. If errors have the same
+// priority, the error with the lowest request index is preferred. This allows
+// a caller issuing multiple cputs to control which ConditionFailedError is
+// returned. Specifically, it allows returning a primary key cput failure
+// instead of a unique index conflict.
 func mergeErrors(pErr1, pErr2 *kvpb.Error) *kvpb.Error {
 	ret, drop := pErr1, pErr2
+
+	hasIndex := ret.Index != nil && drop.Index != nil
+	if hasIndex && drop.Index.Index < ret.Index.Index {
+		ret, drop = drop, ret
+	}
+
 	if kvpb.ErrPriority(drop.GoError()) > kvpb.ErrPriority(ret.GoError()) {
 		ret, drop = drop, ret
 	}

pkg/kv/kvclient/kvcoord/dist_sender_test.go

@@ -4022,6 +4022,27 @@ func TestMultipleErrorsMerged(t *testing.T) {
 	}
 }
 
+func TestMergeErrorIndex(t *testing.T) {
+	// When merging, the error with lower index should be preferred. Create two
+	// errors with the same priority but different indices.
+	err1 := kvpb.NewError(&kvpb.ConditionFailedError{})
+	err1.Index = &kvpb.ErrPosition{Index: 2}
+	err2 := kvpb.NewError(&kvpb.ConditionFailedError{})
+	err2.Index = &kvpb.ErrPosition{Index: 1}
+
+	require.Equal(t, int32(1), mergeErrors(err1, err2).Index.Index)
+	require.Equal(t, int32(1), mergeErrors(err2, err1).Index.Index)
+
+	// Test that priority still takes precedence over index
+	highPriorityErr := kvpb.NewError(&kvpb.TransactionAbortedError{})
+	highPriorityErr.Index = &kvpb.ErrPosition{Index: 5}
+	lowPriorityErr := kvpb.NewError(&kvpb.ConditionFailedError{})
+	lowPriorityErr.Index = &kvpb.ErrPosition{Index: 3}
+
+	require.Equal(t, int32(5), mergeErrors(highPriorityErr, lowPriorityErr).Index.Index)
+	require.Equal(t, int32(5), mergeErrors(lowPriorityErr, highPriorityErr).Index.Index)
+}
+
 // Regression test for #20067.
 // If a batch is partitioned into multiple partial batches, the
 // kvpb.Error.Index of each batch should correspond to its original index in