kv, sql/row: don't always convert ConditionFailedError to dupe key error

KV returns a `ConditionFailedError` when an existing row does not match
the expected value in a CPut. Up until now, the SQL layer has mostly
only used CPut with an empty expected value to verify that no row exists
with the key. For this reason, `ConvertBatchError` always converts
`ConditionFailedError` to a duplicate key error. (The one exception is
backfill of a unique index, which uses CPutAllowingIfNotExists with a
non-empty expected value.)

In the new UPDATE and DELETE fast paths, the SQL layer will now use CPut
with a non-empty expected value to verify that a certain row already
exists. If this verification fails, we don't want to return a dupe key
error, but rather undo the write and continue without an error. We need
`ConvertBatchError` to be aware of the difference.

This commit changes `ConvertBatchError` to only convert
`ConditionFailedError` to a dupe key error if the original request was a
CPut with an empty expected value (or the caller is backfill of a unique
index). If the original request had a non-empty expected value, we leave
it as a `ConditionFailedError` for the new fast path nodes to handle.

Epic: None

Release note: None

Author: michae2

pkg/kv/batch.go

@@ -1161,25 +1161,35 @@ func (b *Batch) bulkRequest(
 	b.initResult(numKeys, numKeys, notRaw, nil)
 }
 
-// GetResult retrieves the Result and Result row KeyValue for a particular index.
+// GetResult retrieves the Result, expected value bytes, and Result row KeyValue
+// for a particular index.
 //
 // WARNING: introduce new usages of this function with care. See discussion in
 // https://github.com/cockroachdb/cockroach/pull/112937.
 // TODO(yuzefovich): look into removing this confusing function.
-func (b *Batch) GetResult(idx int) (*Result, KeyValue, error) {
+func (b *Batch) GetResult(idx int) (*Result, []byte, KeyValue, error) {
 	origIdx := idx
 	for i := range b.Results {
 		r := &b.Results[i]
 		if idx < r.calls {
+			var expBytes []byte
+			if origIdx < len(b.reqs) {
+				req := &b.reqs[origIdx]
+				switch t := req.Value.(type) {
+				case *kvpb.RequestUnion_ConditionalPut:
+					expBytes = t.ConditionalPut.ExpBytes
+				}
+			}
 			if idx < len(r.Rows) {
-				return r, r.Rows[idx], nil
+				return r, expBytes, r.Rows[idx], nil
 			} else if idx < len(r.Keys) {
-				return r, KeyValue{Key: r.Keys[idx]}, nil
+				return r, expBytes, KeyValue{Key: r.Keys[idx]}, nil
 			} else {
-				return r, KeyValue{}, nil
+				return r, expBytes, KeyValue{}, nil
 			}
 		}
 		idx -= r.calls
 	}
-	return nil, KeyValue{}, errors.AssertionFailedf("index %d outside of results: %+v", origIdx, b.Results)
+	return nil, nil, KeyValue{},
+		errors.AssertionFailedf("index %d outside of results: %+v", origIdx, b.Results)
 }

pkg/kv/db_test.go

@@ -1144,25 +1144,39 @@ func TestGetResult(t *testing.T) {
 	b.PutBytes(&byteSliceBulkSource[[]byte]{kys2, vals})
 	keyToDel := roachpb.Key("b")
 	b.Del(keyToDel)
+	keyToCPut := roachpb.Key("c")
+	var expVal roachpb.Value
+	expVal.SetBytes(vals[2])
+	b.CPut(keyToCPut, []byte("who"), expVal.TagAndDataBytes())
 	err := txn.CommitInBatch(ctx, b)
 	require.NoError(t, err)
 	for i := 0; i < len(kys1)+len(kys2); i++ {
-		res, row, err := b.GetResult(i)
-		require.Equal(t, res, &b.Results[i/3])
-		require.Equal(t, row, b.Results[i/3].Rows[i%3])
+		res, exp, row, err := b.GetResult(i)
+		require.Equal(t, &b.Results[i/3], res)
+		require.Nil(t, exp)
+		require.Equal(t, b.Results[i/3].Rows[i%3], row)
 		require.NoError(t, err)
 	}
 	// test Del request (it uses Result.Keys rather than Result.Rows)
-	_, kv, err := b.GetResult(len(kys1) + len(kys2))
+	_, exp, kv, err := b.GetResult(len(kys1) + len(kys2))
 	require.NoError(t, err)
-	require.Equal(t, keyToDel, kv.Key)
+	require.Nil(t, exp)
+	require.Equal(t, kv.Key, keyToDel)
 	require.Nil(t, kv.Value)
+
+	// test CPut request (it fills in expBytes)
+	res, exp, row, err := b.GetResult(len(kys1) + len(kys2) + 1)
+	require.Equal(t, &b.Results[3], res)
+	require.Equal(t, expVal.TagAndDataBytes(), exp)
+	require.Equal(t, b.Results[3].Rows[0], row)
+	require.NoError(t, err)
+
 	// test EndTxn result
-	_, _, err = b.GetResult(len(kys1) + len(kys2) + 1)
+	_, _, _, err = b.GetResult(len(kys1) + len(kys2) + 2)
 	require.NoError(t, err)
 
 	// test out of bounds
-	_, _, err = b.GetResult(len(kys1) + len(kys2) + 2)
+	_, _, _, err = b.GetResult(len(kys1) + len(kys2) + 3)
 	require.Error(t, err)
 }
 

pkg/sql/backfill/backfill.go

@@ -448,7 +448,7 @@ func ConvertBackfillError(
 	if err != nil {
 		return err
 	}
-	return row.ConvertBatchError(ctx, desc, b)
+	return row.ConvertBatchError(ctx, desc, b, true /* alwaysConvertCondFailed */)
 }
 
 type muBoundAccount struct {

pkg/sql/colexec/insert.go

@@ -184,7 +184,9 @@ func (v *vectorInserter) Next() coldata.Batch {
 			err = v.flowCtx.Txn.Run(ctx, kvba.Batch)
 		}
 		if err != nil {
-			colexecerror.ExpectedError(row.ConvertBatchError(ctx, v.desc, kvba.Batch))
+			colexecerror.ExpectedError(row.ConvertBatchError(
+				ctx, v.desc, kvba.Batch, false, /* alwaysConvertCondFailed */
+			))
 		}
 		numRows := end - start
 		start = end

pkg/sql/delete_range.go

@@ -121,7 +121,7 @@ func (d *deleteRangeNode) startExec(params runParams) error {
 			d.deleteSpans(params, b, spans)
 			log.VEventf(ctx, 2, "fast delete: processing %d spans", len(spans))
 			if err := params.p.txn.Run(ctx, b); err != nil {
-				return row.ConvertBatchError(ctx, d.desc, b)
+				return row.ConvertBatchError(ctx, d.desc, b, false /* alwaysConvertCondFailed */)
 			}
 
 			spans = spans[:0]
@@ -144,7 +144,7 @@ func (d *deleteRangeNode) startExec(params runParams) error {
 		d.deleteSpans(params, b, spans)
 		log.VEventf(ctx, 2, "fast delete: processing %d spans and committing", len(spans))
 		if err := params.p.txn.CommitInBatch(ctx, b); err != nil {
-			return row.ConvertBatchError(ctx, d.desc, b)
+			return row.ConvertBatchError(ctx, d.desc, b, false /* alwaysConvertCondFailed */)
 		}
 		if resumeSpans, err := d.processResults(b.Results, nil /* resumeSpans */); err != nil {
 			return err

pkg/sql/row/errors.go

@@ -31,7 +31,14 @@ import (
 // ConvertBatchError attempts to map a key-value error generated during a
 // key-value batch operating over the specified table to a user friendly SQL
 // error.
-func ConvertBatchError(ctx context.Context, tableDesc catalog.TableDescriptor, b *kv.Batch) error {
+//
+// If alwaysConvertCondFailed is set to true, ConditionFailedError (from CPut
+// failures) will always be converted to a duplicate key error even if the
+// expValue of the CPut was not empty. This is necessary when backfilling a
+// unique index.
+func ConvertBatchError(
+	ctx context.Context, tableDesc catalog.TableDescriptor, b *kv.Batch, alwaysConvertCondFailed bool,
+) error {
 	origPErr := b.MustPErr()
 	switch v := origPErr.GetDetail().(type) {
 	case *kvpb.MinTimestampBoundUnsatisfiableError:
@@ -50,11 +57,14 @@ func ConvertBatchError(ctx context.Context, tableDesc catalog.TableDescriptor, b
 			break
 		}
 		j := origPErr.Index.Index
-		_, kv, err := b.GetResult(int(j))
+		_, expBytes, kv, err := b.GetResult(int(j))
 		if err != nil {
 			return err
 		}
-		return NewUniquenessConstraintViolationError(ctx, tableDesc, kv.Key, v.ActualValue)
+		if alwaysConvertCondFailed || len(expBytes) == 0 {
+			// If we didn't expect the row to exist, this is a uniqueness violation.
+			return NewUniquenessConstraintViolationError(ctx, tableDesc, kv.Key, v.ActualValue)
+		}
 
 	case *kvpb.WriteIntentError:
 		key := v.Locks[0].Key

pkg/sql/tablewriter.go

@@ -141,7 +141,7 @@ func (tb *tableWriterBase) setRowsWrittenLimit(sd *sessiondata.SessionData) {
 func (tb *tableWriterBase) flushAndStartNewBatch(ctx context.Context) error {
 	log.VEventf(ctx, 2, "writing batch with %d requests", len(tb.b.Requests()))
 	if err := tb.txn.Run(ctx, tb.b); err != nil {
-		return row.ConvertBatchError(ctx, tb.desc, tb.b)
+		return row.ConvertBatchError(ctx, tb.desc, tb.b, false /* alwaysConvertCondFailed */)
 	}
 	if err := tb.tryDoResponseAdmission(ctx); err != nil {
 		return err
@@ -180,7 +180,7 @@ func (tb *tableWriterBase) finalize(ctx context.Context) (err error) {
 	}
 	tb.lastBatchSize = tb.currentBatchSize
 	if err != nil {
-		return row.ConvertBatchError(ctx, tb.desc, tb.b)
+		return row.ConvertBatchError(ctx, tb.desc, tb.b, false /* alwaysConvertCondFailed */)
 	}
 	return tb.tryDoResponseAdmission(ctx)
 }