Merge #147117

147117: kvcoord: deterministic error ordering r=jeffswenson a=jeffswenson

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: #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.

Co-authored-by: Jeff Swenson <[email protected]>

Author: craig[bot]

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

@@ -1688,9 +1688,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