logical: wrap recoverable errors in savepoints

Internal executor statements failures are not atomic. They may fail and
leave behind live KV operations. Usually this is okay because the error
is returned up the stack and the caller aborts the transaction. There is
a bug LDR because it recovers condition failed errors to detect LWW
losses. The insert fast path writes indexes before it writes to the
primary key, so these fast path failures could corrupt index entries.

Now, all LDR code paths that commit a transaction containing a
recoverable internal executor error wrap the statement in a savepoint.

Informs: #144645
Epic: CRDB-48647
Release note: None

Author: jeffswenson

pkg/crosscluster/logical/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "purgatory.go",
         "range_stats.go",
         "replication_statements.go",
+        "savepoint.go",
         "sql_crud_writer.go",
         "sql_row_reader.go",
         "sql_row_writer.go",
@@ -127,6 +128,7 @@ go_test(
         "purgatory_test.go",
         "range_stats_test.go",
         "replication_statements_test.go",
+        "savepoint_test.go",
         "sql_row_reader_test.go",
         "sql_row_writer_test.go",
         "table_batch_handler_test.go",

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -181,6 +181,61 @@ func TestLogicalStreamIngestionJobNameResolution(t *testing.T) {
 	}
 }
 
+func TestOptimsitcInsertCorruption(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	skip.UnderDeadlock(t)
+	defer log.Scope(t).Close(t)
+
+	// This is a regression test for #144645. Running the optimistic insert code
+	// path could corrupt indexes if the insert is not wrapped in a savepoint.
+
+	ctx := context.Background()
+
+	server, s, dbSource, dbDest := setupLogicalTestServer(t, ctx, testClusterBaseClusterArgs, 1)
+	defer server.Stopper().Stop(ctx)
+
+	dbSource.Exec(t, "SET CLUSTER SETTING logical_replication.consumer.try_optimistic_insert.enabled = true")
+
+	createTableStmt := `CREATE TABLE computed_cols (
+		a INT,
+		b INT,
+		c INT,
+		PRIMARY KEY (a, b)
+	)`
+	dbSource.Exec(t, createTableStmt)
+	dbDest.Exec(t, createTableStmt)
+
+	createIdxStmt := `CREATE INDEX c ON computed_cols (c)`
+	dbSource.Exec(t, createIdxStmt)
+	dbDest.Exec(t, createIdxStmt)
+
+	// Insert initial data into destination that should be overwritten. This
+	// gives more opportunities for corruption.
+	dbDest.Exec(t, "INSERT INTO computed_cols (a, b, c) VALUES (1, 2, 3), (3, 4, 5)")
+	dbSource.Exec(t, "INSERT INTO computed_cols (a, b, c) VALUES (1, 2, 6), (3, 4, 7)")
+
+	// Create logical replication stream from source to destination
+	sourceURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("a"))
+	var jobID jobspb.JobID
+	dbDest.QueryRow(t,
+		"CREATE LOGICAL REPLICATION STREAM FROM TABLE computed_cols ON $1 INTO TABLE computed_cols",
+		sourceURL.String(),
+	).Scan(&jobID)
+
+	WaitUntilReplicatedTime(t, s.Clock().Now(), dbDest, jobID)
+
+	dbSource.CheckQueryResults(t, "SELECT * FROM computed_cols", [][]string{
+		{"1", "2", "6"},
+		{"3", "4", "7"},
+	})
+	dbDest.CheckQueryResults(t, "SELECT * FROM computed_cols", [][]string{
+		{"1", "2", "6"},
+		{"3", "4", "7"},
+	})
+
+	compareReplicatedTables(t, s, "a", "b", "computed_cols", dbSource, dbDest)
+}
+
 type fatalDLQ struct{ *testing.T }
 
 func (fatalDLQ) Create(ctx context.Context) error { return nil }

pkg/crosscluster/logical/lww_row_processor.go

@@ -640,7 +640,11 @@ func (lww *lwwQuerier) InsertRow(
 		if !useLowPriority.Get(&lww.settings.SV) {
 			sess.QualityOfService = nil
 		}
-		if _, err = ie.ExecParsed(ctx, replicatedOptimisticInsertOpName, kvTxn, sess, stmt, datums...); err != nil {
+		err = withSavepoint(ctx, kvTxn, func() error {
+			_, err = ie.ExecParsed(ctx, replicatedOptimisticInsertOpName, kvTxn, sess, stmt, datums...)
+			return err
+		})
+		if err != nil {
 			if isLwwLoser(err) {
 				return batchStats{}, nil
 			}
@@ -667,7 +671,10 @@ func (lww *lwwQuerier) InsertRow(
 		sess.QualityOfService = nil
 	}
 	sess.OriginTimestampForLogicalDataReplication = row.MvccTimestamp
-	_, err = ie.ExecParsed(ctx, replicatedInsertOpName, kvTxn, sess, stmt, datums...)
+	err = withSavepoint(ctx, kvTxn, func() error {
+		_, err = ie.ExecParsed(ctx, replicatedInsertOpName, kvTxn, sess, stmt, datums...)
+		return err
+	})
 	if isLwwLoser(err) {
 		return batchStats{}, nil
 	}

pkg/crosscluster/logical/savepoint.go

@@ -0,0 +1,49 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package logical
+
+import (
+	"context"
+
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/errors"
+)
+
+// withSavepoint is a utility function that runs the provided function within
+// the context of the savepoint.
+//
+// If the caller wishes to recover from an error returned by an internal
+// executor and keep the transaction, then the call to the internal executor
+// must be scoped within a savepoint. Failed internal executor calls are not
+// atomic and may have left behind some KV operations. Usually that is okay
+// because the error is passed up the call stack and the transaction is rolled
+// back.
+func withSavepoint(ctx context.Context, txn *kv.Txn, fn func() error) error {
+	// In the classic sql writer the txn is optional.
+	if txn == nil {
+		return fn()
+	}
+	// TODO(jeffswenson): consider changing the internal executor so all calls
+	// implicitly create and apply/rollback savepoints.
+	savepoint, err := txn.CreateSavepoint(ctx)
+	if err != nil {
+		return err
+	}
+	err = fn()
+	if err != nil {
+		// NOTE: we return the save point error if rollback fails because we do not
+		// want something checking error types to attempt to handle the inner
+		// error.
+		if savePointErr := txn.RollbackToSavepoint(ctx, savepoint); savePointErr != nil {
+			return errors.WithSecondaryError(savePointErr, err)
+		}
+		return err
+	}
+	if err := txn.ReleaseSavepoint(ctx, savepoint); err != nil {
+		return err
+	}
+	return nil
+}

pkg/crosscluster/logical/savepoint_test.go

@@ -0,0 +1,70 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package logical
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWithSavepoint(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	srv, rawDB, _ := serverutils.StartServer(t, base.TestServerArgs{})
+	defer srv.Stopper().Stop(context.Background())
+
+	ctx := context.Background()
+	sqlDB := sqlutils.MakeSQLRunner(rawDB)
+	sqlDB.Exec(t, "CREATE TABLE test (id STRING PRIMARY KEY, value STRING)")
+
+	require.NoError(t, srv.DB().Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+		err := withSavepoint(ctx, txn, func() error {
+			_, err := srv.InternalExecutor().(*sql.InternalExecutor).ExecEx(
+				ctx,
+				"test-insert",
+				txn,
+				sessiondata.NodeUserSessionDataOverride,
+				"INSERT INTO defaultdb.test VALUES ('ok', 'is-persisted')",
+			)
+			return err
+		})
+		require.NoError(t, err)
+
+		err = withSavepoint(ctx, txn, func() error {
+			_, err := srv.InternalExecutor().(*sql.InternalExecutor).ExecEx(
+				ctx,
+				"test-insert",
+				txn,
+				sessiondata.NodeUserSessionDataOverride,
+				"INSERT INTO defaultdb.test VALUES ('fails', 'is-rolled-back')",
+			)
+			require.NoError(t, err)
+			// NOTE: the query above is okay, which means it wrote things to KV,
+			// but we're going to return an error which rolls back the
+			// savepoint.
+			return errors.New("something to rollback")
+		})
+		require.ErrorContains(t, err, "something to rollback")
+
+		return nil
+	}))
+
+	sqlDB.CheckQueryResults(t, "SELECT id, value FROM test", [][]string{
+		{"ok", "is-persisted"},
+	})
+}

pkg/crosscluster/logical/table_batch_handler.go

@@ -170,7 +170,9 @@ func (t *tableHandler) attemptBatch(
 				stats.kvLwwLosers += tombstoneUpdateStats.kvWriteTooOld
 			case event.prevRow == nil:
 				stats.inserts++
-				err := t.sqlWriter.InsertRow(ctx, txn, event.originTimestamp, event.row)
+				err := withSavepoint(ctx, txn.KV(), func() error {
+					return t.sqlWriter.InsertRow(ctx, txn, event.originTimestamp, event.row)
+				})
 				if isLwwLoser(err) {
 					// Insert may observe a LWW failure if it attempts to write over a tombstone.
 					stats.kvLwwLosers++