logical: fix bug in tombstone updater

This fixes a bug in the tombstone updater. The tombstone decoder was
using the columns decoded by the kv writer when it needs to expect the
columns decoded by the sql/crud writers. There is one difference between
these two sets of columns that impacts computed columns:

1. The sql/crud writer only decode computed columns that are in the
	 primary key.
2. The kv writer decodes virtual computed columns in the primary key and
	 all stored computed columns in the row.

This mismatch can cause panics if the stored column is in an index
(which is what the conflict workload noticed) and it can cause the
tombstone updater to generate incorrect keys if there is a stored
computed column with a lower column id than the primary key column. If
that happened, and we hit the tombstone update race condition, we would
issue a cput for the wrong row. That would end up creating a tombstone
with an origin time stamp at that wrong key value. This would not cause
data corruption, since we are using cput, so the write only succeeds if
there was nothing there. Additionally we would be creating a tombstone,
so it does not have side effects from the SQL layer. But it could LDR to
fail to converge, since the tombstone updater is not working.

A secondary discovery here is that the tombstone updater is formatting
deletes for indexes other than the primary index. Now, the deleter will
skip generating values for secondary indexes if the cput helper says its
updating a deleted row.

Fixes: #159306
Release note: None

Author: jeffswenson

pkg/crosscluster/logical/BUILD.bazel

@@ -154,6 +154,7 @@ go_test(
         "//pkg/ccl/changefeedccl/changefeedbase",
         "//pkg/ccl/storageccl",
         "//pkg/crosscluster",
+        "//pkg/crosscluster/ldrrandgen",
         "//pkg/crosscluster/replicationtestutils",
         "//pkg/crosscluster/streamclient",
         "//pkg/crosscluster/streamclient/randclient",

pkg/crosscluster/logical/tombstone_updater.go

@@ -164,9 +164,11 @@ func (tu *tombstoneUpdater) getDeleter(ctx context.Context, txn *kv.Txn) (row.De
 			return row.Deleter{}, err
 		}
 
-		cols, err := writeableColunms(ctx, tu.leased.descriptor.Underlying().(catalog.TableDescriptor))
-		if err != nil {
-			return row.Deleter{}, err
+		table := tu.leased.descriptor.Underlying().(catalog.TableDescriptor)
+		schema := getColumnSchema(table)
+		cols := make([]catalog.Column, len(schema))
+		for i, cs := range schema {
+			cols[i] = cs.column
 		}
 
 		tu.leased.deleter = row.MakeDeleter(tu.codec, tu.leased.descriptor.Underlying().(catalog.TableDescriptor), nil /* lockedIndexes */, cols, tu.sd, &tu.settings.SV, nil /* metrics */)

pkg/crosscluster/logical/tombstone_updater_test.go

@@ -7,17 +7,23 @@ package logical
 
 import (
 	"context"
+	"math/rand"
+	"strings"
 	"testing"
 
+	"github.com/cockroachdb/cockroach/pkg/crosscluster/ldrrandgen"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/desctestutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/lease"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
+	"github.com/cockroachdb/cockroach/pkg/sql/randgen"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -82,3 +88,81 @@ func TestTombstoneUpdaterSetsOriginID(t *testing.T) {
 		{"1", "42"},
 	})
 }
+
+func TestTombstoneUpdaterRandomTables(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	ctx := context.Background()
+	server, s, runners, dbNames := setupServerWithNumDBs(t, ctx, testClusterBaseClusterArgs, 1, 1)
+	defer server.Stopper().Stop(ctx)
+
+	runner := runners[0]
+	dbName := dbNames[0]
+
+	rng, _ := randutil.NewPseudoRand()
+	tableName := "rand_table"
+
+	stmt := tree.SerializeForDisplay(ldrrandgen.GenerateLDRTable(ctx, rng, tableName, true))
+	t.Logf("Creating table with schema: %s", stmt)
+	runner.Exec(t, stmt)
+
+	desc := desctestutils.TestingGetMutableExistingTableDescriptor(
+		s.DB(), s.Codec(), dbName, tableName)
+
+	sd := sql.NewInternalSessionData(ctx, s.ClusterSettings(), "" /* opName */)
+	tu := newTombstoneUpdater(s.Codec(), s.DB(), s.LeaseManager().(*lease.Manager), desc.GetID(), sd, s.ClusterSettings())
+	defer tu.ReleaseLeases(ctx)
+
+	columnSchemas := getColumnSchema(desc)
+	cols := make([]catalog.Column, len(columnSchemas))
+	for i, cs := range columnSchemas {
+		cols[i] = cs.column
+	}
+
+	config := s.ExecutorConfig().(sql.ExecutorConfig)
+
+	session := newInternalSession(t, s)
+	defer session.Close(ctx)
+
+	writer, err := newSQLRowWriter(ctx, desc, session)
+	require.NoError(t, err)
+
+	for i := 0; i < 10; i++ {
+		row := generateRandomRow(rng, cols)
+		before := s.Clock().Now()
+		after := s.Clock().Now()
+
+		err := sql.DescsTxn(ctx, &config, func(
+			ctx context.Context, txn isql.Txn, descriptors *descs.Collection,
+		) error {
+			_, err := tu.updateTombstone(ctx, txn, after, row)
+			return err
+		})
+		require.NoError(t, err)
+
+		stats, err := tu.updateTombstone(ctx, nil, before, row)
+		require.NoError(t, err)
+		require.Equal(t, int64(1), stats.kvWriteTooOld, "expected 1 kv put for tombstone update")
+
+		// Use the table writer to try and insert the previous row using the
+		// `before` timestamp. This should fail with LWW error because the
+		// tombstone was written at a later timestamp. It may fail with integer out
+		// of range error if the table has computed columns and the random datums
+		// add to produces something out of range.
+		err = writer.InsertRow(ctx, before, row)
+		require.Error(t, err)
+		if !(strings.Contains(err.Error(), "integer out of range") || isLwwLoser(err)) {
+			t.Fatalf("expected LWW or integer out of range error, got: %v", err)
+		}
+	}
+}
+
+func generateRandomRow(rng *rand.Rand, cols []catalog.Column) []tree.Datum {
+	row := make([]tree.Datum, 0, len(cols))
+	for _, col := range cols {
+		datum := randgen.RandDatum(rng, col.GetType(), col.IsNullable())
+		row = append(row, datum)
+	}
+	return row
+}

pkg/sql/row/deleter.go

@@ -212,6 +212,12 @@ func (rd *Deleter) DeleteRow(
 		return err
 	}
 
+	if oth.PreviousWasDeleted {
+		// If we are updating a tombstone, then there are no secondary index entries
+		// that need to be deleted.
+		return nil
+	}
+
 	// Delete the row from any secondary indices.
 	for i, index := range rd.Helper.Indexes {
 		// If the index ID exists in the set of indexes to ignore, do not