sql: always use cput for ldr primary key writes

jeffswenson · jeffswenson · commit ab8a53b62e55 · 2025-06-05T11:46:01.000-04:00
This change reworks the SQL layer so that it always uses a CPUT with an origin timestamp if the LDR origin timestamp option is set. This change allows the classic and crud SQL writers to correctly implement LWW in the presence of tombstones. Note: the classic SQL writer only depends on the CPUT when inserting or upserting over a tombstone. The crud SQL writer relies on the CPut of inserts, updates, and deletes. Release note: none Fixes: #146117
diff --git a/pkg/crosscluster/logical/batch_handler_test.go b/pkg/crosscluster/logical/batch_handler_test.go
@@ -31,7 +31,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
-	"github.com/cockroachdb/cockroach/pkg/testutils/skip"
 	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
@@ -198,17 +197,13 @@ func TestBatchHandlerExhaustiveSQL(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 146117)
-
 	testBatchHandlerExhaustive(t, newSqlBatchHandler)
 }
 
 func TestBatchHandlerExhaustiveCrud(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)
 
-	skip.WithIssue(t, 146117)
-
 	testBatchHandlerExhaustive(t, newCrudBatchHandler)
 }
 
diff --git a/pkg/crosscluster/logical/lww_kv_processor.go b/pkg/crosscluster/logical/lww_kv_processor.go
@@ -493,7 +493,7 @@ func (p *kvTableWriter) insertRow(ctx context.Context, b *kv.Batch, after cdceve
 	// TODO(dt): support partial indexes.
 	var vh row.VectorIndexUpdateHelper
 	// TODO(mw5h, drewk): support vector indexes.
-	oth := &row.OriginTimestampCPutHelper{
+	oth := row.OriginTimestampCPutHelper{
 		OriginTimestamp: after.MvccTimestamp,
 		// TODO(ssd): We should choose this based by comparing the cluster IDs of the source
 		// and destination clusters.
@@ -516,7 +516,7 @@ func (p *kvTableWriter) updateRow(
 	// TODO(dt): support partial indexes.
 	var vh row.VectorIndexUpdateHelper
 	// TODO(mw5h, drewk): support vector indexes.
-	oth := &row.OriginTimestampCPutHelper{
+	oth := row.OriginTimestampCPutHelper{
 		OriginTimestamp: after.MvccTimestamp,
 		// TODO(ssd): We should choose this based by comparing the cluster IDs of the source
 		// and destination clusters.
@@ -540,7 +540,7 @@ func (p *kvTableWriter) deleteRow(
 	// TODO(dt): support partial indexes.
 	var vh row.VectorIndexUpdateHelper
 	// TODO(mw5h, drewk): support vector indexes.
-	oth := &row.OriginTimestampCPutHelper{
+	oth := row.OriginTimestampCPutHelper{
 		PreviousWasDeleted: before.IsDeleted(),
 		OriginTimestamp:    after.MvccTimestamp,
 		// TODO(ssd): We should choose this based by comparing the cluster IDs of the source
diff --git a/pkg/crosscluster/logical/lww_row_processor.go b/pkg/crosscluster/logical/lww_row_processor.go
@@ -897,7 +897,8 @@ DELETE FROM [%d as t] WHERE %s
    AND ((t.crdb_internal_mvcc_timestamp < $%[3]d
         AND t.crdb_internal_origin_timestamp IS NULL)
     OR (t.crdb_internal_origin_timestamp < $%[3]d
-        AND t.crdb_internal_origin_timestamp IS NOT NULL))`
+        AND t.crdb_internal_origin_timestamp IS NOT NULL))
+RETURNING *`
 	stmt, err := parser.ParseOne(
 		fmt.Sprintf(baseQuery, dstTableDescID, whereClause.String(), originTSIdx))
 	if err != nil {
diff --git a/pkg/crosscluster/logical/replication_statements.go b/pkg/crosscluster/logical/replication_statements.go
@@ -254,8 +254,11 @@ func newDeleteStatement(
 			TableID: int64(table.GetID()),
 			As:      tree.AliasClause{Alias: "replication_target"},
 		},
-		Where:     &tree.Where{Type: tree.AstWhere, Expr: whereClause},
-		Returning: tree.AbsentReturningClause,
+		Where: &tree.Where{Type: tree.AstWhere, Expr: whereClause},
+		// NOTE: we use RETURNING * to ensure that every column in the table is decoded.
+		// This ensures that the Deleter can reconstruct the previous value when generating
+		// the cput to update the primary key.
+		Returning: &tree.ReturningExprs{tree.StarSelectExpr()},
 	}
 
 	return toParsedStatement(delete)
diff --git a/pkg/crosscluster/logical/testdata/ldr_statements b/pkg/crosscluster/logical/testdata/ldr_statements
@@ -18,7 +18,7 @@ UPDATE [104 AS replication_target] SET id = $5::INT8, name = $6::STRING, value =
 
 show-delete table=basic_table
 ----
-DELETE FROM [104 AS replication_target] WHERE (((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (value IS NOT DISTINCT FROM $3::INT8)) AND (data IS NOT DISTINCT FROM $4::BYTES)
+DELETE FROM [104 AS replication_target] WHERE (((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (value IS NOT DISTINCT FROM $3::INT8)) AND (data IS NOT DISTINCT FROM $4::BYTES) RETURNING *
 
 show-select table=basic_table
 ----
@@ -52,7 +52,7 @@ UPDATE [107 AS replication_target] SET id = $7::INT8, title = $8::STRING, descri
 
 show-delete table=tasks
 ----
-DELETE FROM [107 AS replication_target] WHERE (((((id = $1::INT8) AND (title IS NOT DISTINCT FROM $2::STRING)) AND (description IS NOT DISTINCT FROM $3::STRING)) AND (status IS NOT DISTINCT FROM $4::@100105)) AND (priority IS NOT DISTINCT FROM $5::INT8)) AND (created_at IS NOT DISTINCT FROM $6::TIMESTAMP)
+DELETE FROM [107 AS replication_target] WHERE (((((id = $1::INT8) AND (title IS NOT DISTINCT FROM $2::STRING)) AND (description IS NOT DISTINCT FROM $3::STRING)) AND (status IS NOT DISTINCT FROM $4::@100105)) AND (priority IS NOT DISTINCT FROM $5::INT8)) AND (created_at IS NOT DISTINCT FROM $6::TIMESTAMP) RETURNING *
 
 show-select table=tasks
 ----
@@ -86,7 +86,7 @@ UPDATE [108 AS replication_target] SET id = $7::INT8, name = $8::STRING, unit_pr
 # NOTE: total_price and discount_price are not included since they are computed.[
 show-delete table=products
 ----
-DELETE FROM [108 AS replication_target] WHERE ((((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (unit_price IS NOT DISTINCT FROM $3::DECIMAL(10,2))) AND (quantity IS NOT DISTINCT FROM $4::INT8)) AND (last_updated IS NOT DISTINCT FROM $6::TIMESTAMP)
+DELETE FROM [108 AS replication_target] WHERE ((((id = $1::INT8) AND (name IS NOT DISTINCT FROM $2::STRING)) AND (unit_price IS NOT DISTINCT FROM $3::DECIMAL(10,2))) AND (quantity IS NOT DISTINCT FROM $4::INT8)) AND (last_updated IS NOT DISTINCT FROM $6::TIMESTAMP) RETURNING *
 
 # NOTE: total_price is not included because it is a computed column, but
 # discount_price is included because its part of the primary key.
@@ -137,7 +137,7 @@ UPDATE [109 AS replication_target] SET id = $8::INT8, first_name = $9::STRING, l
 
 show-delete table=employees
 ----
-DELETE FROM [109 AS replication_target] WHERE ((((((id = $1::INT8) AND (first_name IS NOT DISTINCT FROM $2::STRING)) AND (last_name IS NOT DISTINCT FROM $3::STRING)) AND (email IS NOT DISTINCT FROM $4::STRING)) AND (salary IS NOT DISTINCT FROM $5::DECIMAL(12,2))) AND (department IS NOT DISTINCT FROM $6::STRING)) AND (hire_date IS NOT DISTINCT FROM $7::DATE)
+DELETE FROM [109 AS replication_target] WHERE ((((((id = $1::INT8) AND (first_name IS NOT DISTINCT FROM $2::STRING)) AND (last_name IS NOT DISTINCT FROM $3::STRING)) AND (email IS NOT DISTINCT FROM $4::STRING)) AND (salary IS NOT DISTINCT FROM $5::DECIMAL(12,2))) AND (department IS NOT DISTINCT FROM $6::STRING)) AND (hire_date IS NOT DISTINCT FROM $7::DATE) RETURNING *
 
 show-select table=employees
 ----
@@ -171,7 +171,7 @@ UPDATE [112 AS replication_target] SET id = $7::UUID, user_id = $8::INT8, event_
 
 show-delete table=user_events
 ----
-DELETE FROM [112 AS replication_target] WHERE (((((id = $1::UUID) AND (user_id IS NOT DISTINCT FROM $2::INT8)) AND (event_type IS NOT DISTINCT FROM $3::STRING)) AND (event_data IS NOT DISTINCT FROM $4::JSONB)) AND (created_at IS NOT DISTINCT FROM $5::TIMESTAMP)) AND (region = $6::@100110)
+DELETE FROM [112 AS replication_target] WHERE (((((id = $1::UUID) AND (user_id IS NOT DISTINCT FROM $2::INT8)) AND (event_type IS NOT DISTINCT FROM $3::STRING)) AND (event_data IS NOT DISTINCT FROM $4::JSONB)) AND (created_at IS NOT DISTINCT FROM $5::TIMESTAMP)) AND (region = $6::@100110) RETURNING *
 
 show-select table=user_events
 ----
diff --git a/pkg/crosscluster/logical/tombstone_updater.go b/pkg/crosscluster/logical/tombstone_updater.go
@@ -144,7 +144,7 @@ func (tu *tombstoneUpdater) addToBatch(
 		afterRow,
 		ph,
 		vh,
-		&row.OriginTimestampCPutHelper{
+		row.OriginTimestampCPutHelper{
 			OriginTimestamp:    mvccTimestamp,
 			PreviousWasDeleted: true,
 		},
diff --git a/pkg/sql/backfill/backfill.go b/pkg/sql/backfill/backfill.go
@@ -404,8 +404,9 @@ func (cb *ColumnBackfiller) RunColumnBackfillChunk(
 		// VectorIndexUpdateHelper in this case.
 		var pm row.PartialIndexUpdateHelper
 		var vh row.VectorIndexUpdateHelper
+		var oth row.OriginTimestampCPutHelper
 		if _, err := ru.UpdateRow(
-			ctx, b, oldValues, updateValues, pm, vh, nil, false /* mustValidateOldPKValues */, traceKV,
+			ctx, b, oldValues, updateValues, pm, vh, oth, false /* mustValidateOldPKValues */, traceKV,
 		); err != nil {
 			return roachpb.Key{}, err
 		}
diff --git a/pkg/sql/colenc/encode_test.go b/pkg/sql/colenc/encode_test.go
@@ -619,8 +619,9 @@ func buildRowKVs(
 	p := &capturePutter{}
 	var pm row.PartialIndexUpdateHelper
 	var vh row.VectorIndexUpdateHelper
+	var oth row.OriginTimestampCPutHelper
 	for _, d := range datums {
-		if err := inserter.InsertRow(context.Background(), p, d, pm, vh, nil, row.CPutOp, true /* traceKV */); err != nil {
+		if err := inserter.InsertRow(context.Background(), p, d, pm, vh, oth, row.CPutOp, true /* traceKV */); err != nil {
 			return kvs{}, err
 		}
 	}
diff --git a/pkg/sql/create_table.go b/pkg/sql/create_table.go
@@ -614,7 +614,8 @@ func (n *createTableNode) startExec(params runParams) error {
 				// indexes, partial, vector, or otherwise, to update.
 				var pm row.PartialIndexUpdateHelper
 				var vh row.VectorIndexUpdateHelper
-				if err := ti.row(params.ctx, rowBuffer, pm, vh, params.extendedEvalCtx.Tracing.KVTracingEnabled()); err != nil {
+				var oth row.OriginTimestampCPutHelper
+				if err := ti.row(params.ctx, rowBuffer, pm, vh, oth, params.extendedEvalCtx.Tracing.KVTracingEnabled()); err != nil {
 					return err
 				}
 			}
diff --git a/pkg/sql/delete.go b/pkg/sql/delete.go
@@ -63,12 +63,19 @@ type deleteRun struct {
 	numPassthrough int
 
 	mustValidateOldPKValues bool
+
+	originTimestampCPutHelper row.OriginTimestampCPutHelper
 }
 
-func (r *deleteRun) initRowContainer(params runParams, columns colinfo.ResultColumns) {
+func (r *deleteRun) init(params runParams, columns colinfo.ResultColumns) {
+	if ots := params.extendedEvalCtx.SessionData().OriginTimestampForLogicalDataReplication; ots.IsSet() {
+		r.originTimestampCPutHelper.OriginTimestamp = ots
+	}
+
 	if !r.rowsNeeded {
 		return
 	}
+
 	r.td.rows = rowcontainer.NewRowContainer(
 		params.p.Mon().MakeBoundAccount(),
 		colinfo.ColTypeInfoFromResCols(columns),
@@ -83,7 +90,7 @@ func (d *deleteNode) startExec(params runParams) error {
 	// cache traceKV during execution, to avoid re-evaluating it for every row.
 	d.run.traceKV = params.p.ExtendedEvalContext().Tracing.KVTracingEnabled()
 
-	d.run.initRowContainer(params, d.columns)
+	d.run.init(params, d.columns)
 
 	return d.run.td.init(params.ctx, params.p.txn, params.EvalContext())
 }
@@ -189,7 +196,7 @@ func (r *deleteRun) processSourceRow(params runParams, sourceVals tree.Datums) e
 
 	// Queue the deletion in the KV batch.
 	if err := r.td.row(
-		params.ctx, deleteVals, pm, vh, r.mustValidateOldPKValues, r.traceKV,
+		params.ctx, deleteVals, pm, vh, r.originTimestampCPutHelper, r.mustValidateOldPKValues, r.traceKV,
 	); err != nil {
 		return err
 	}
diff --git a/pkg/sql/delete_swap.go b/pkg/sql/delete_swap.go
@@ -42,7 +42,7 @@ func (d *deleteSwapNode) startExec(params runParams) error {
 
 	d.run.mustValidateOldPKValues = true
 
-	d.run.initRowContainer(params, d.columns)
+	d.run.init(params, d.columns)
 
 	return d.run.td.init(params.ctx, params.p.txn, params.EvalContext())
 }
diff --git a/pkg/sql/insert.go b/pkg/sql/insert.go
@@ -93,6 +93,8 @@ type insertRun struct {
 	// regionLocalInfo handles erroring out the INSERT when the
 	// enforce_home_region setting is on.
 	regionLocalInfo regionLocalInfoType
+
+	originTimestampCPutHelper row.OriginTimestampCPutHelper
 }
 
 // regionLocalInfoType contains common items needed for determining the home region
@@ -158,7 +160,10 @@ func (r *regionLocalInfoType) checkHomeRegion(row tree.Datums) error {
 	return nil
 }
 
-func (r *insertRun) initRowContainer(params runParams, columns colinfo.ResultColumns) {
+func (r *insertRun) init(params runParams, columns colinfo.ResultColumns) {
+	if ots := params.extendedEvalCtx.SessionData().OriginTimestampForLogicalDataReplication; ots.IsSet() {
+		r.originTimestampCPutHelper.OriginTimestamp = ots
+	}
 	if !r.rowsNeeded {
 		return
 	}
@@ -243,7 +248,7 @@ func (r *insertRun) processSourceRow(params runParams, rowVals tree.Datums) erro
 	}
 
 	// Queue the insert in the KV batch.
-	if err := r.ti.row(params.ctx, insertVals, pm, vh, r.traceKV); err != nil {
+	if err := r.ti.row(params.ctx, insertVals, pm, vh, r.originTimestampCPutHelper, r.traceKV); err != nil {
 		return err
 	}
 
@@ -272,7 +277,7 @@ func (n *insertNode) startExec(params runParams) error {
 	// Cache traceKV during execution, to avoid re-evaluating it for every row.
 	n.run.traceKV = params.p.ExtendedEvalContext().Tracing.KVTracingEnabled()
 
-	n.run.initRowContainer(params, n.columns)
+	n.run.init(params, n.columns)
 
 	return n.run.ti.init(params.ctx, params.p.txn, params.EvalContext())
 }
diff --git a/pkg/sql/insert_fast_path.go b/pkg/sql/insert_fast_path.go
@@ -398,7 +398,7 @@ func (n *insertFastPathNode) startExec(params runParams) error {
 	// Cache traceKV during execution, to avoid re-evaluating it for every row.
 	n.run.traceKV = params.p.ExtendedEvalContext().Tracing.KVTracingEnabled()
 
-	n.run.initRowContainer(params, n.columns)
+	n.run.init(params, n.columns)
 
 	n.run.numInputCols = len(n.input[0])
 	n.run.inputBuf = make(tree.Datums, len(n.input)*n.run.numInputCols)
diff --git a/pkg/sql/row/deleter.go b/pkg/sql/row/deleter.go
@@ -140,7 +140,7 @@ func (rd *Deleter) DeleteRow(
 	values []tree.Datum,
 	pm PartialIndexUpdateHelper,
 	vh VectorIndexUpdateHelper,
-	oth *OriginTimestampCPutHelper,
+	oth OriginTimestampCPutHelper,
 	mustValidateOldPKValues bool,
 	traceKV bool,
 ) error {
diff --git a/pkg/sql/row/inserter.go b/pkg/sql/row/inserter.go
@@ -168,7 +168,7 @@ func (ri *Inserter) InsertRow(
 	values []tree.Datum,
 	pm PartialIndexUpdateHelper,
 	vh VectorIndexUpdateHelper,
-	oth *OriginTimestampCPutHelper,
+	oth OriginTimestampCPutHelper,
 	kvOp KVInsertOp,
 	traceKV bool,
 ) error {
diff --git a/pkg/sql/row/row_converter.go b/pkg/sql/row/row_converter.go
@@ -584,14 +584,15 @@ func (c *DatumRowConverter) Row(ctx context.Context, sourceID int32, rowIndex in
 	// TODO(mw5h, drewk): call into the vector index library to determine the partitions
 	// to update.
 	var vh VectorIndexUpdateHelper
+	var oth OriginTimestampCPutHelper
 
 	if err := c.ri.InsertRow(
 		ctx,
 		c.kvInserter,
 		insertRow,
 		pm,
 		vh,
-		nil, /* OriginTimestampCPutHelper */
+		oth,
 		// Lock acquisition ask doesn't matter for the DatumRowConverter, but
 		// we're being conservative and are choosing a "safer" option of asking
 		// for the lock.
diff --git a/pkg/sql/row/updater.go b/pkg/sql/row/updater.go
@@ -241,7 +241,7 @@ func (ru *Updater) UpdateRow(
 	updateValues []tree.Datum,
 	pm PartialIndexUpdateHelper,
 	vh VectorIndexUpdateHelper,
-	oth *OriginTimestampCPutHelper,
+	oth OriginTimestampCPutHelper,
 	mustValidateOldPKValues bool,
 	traceKV bool,
 ) ([]tree.Datum, error) {
diff --git a/pkg/sql/row/writer.go b/pkg/sql/row/writer.go
@@ -94,7 +94,7 @@ func prepareInsertOrUpdateBatch(
 	kvKey *roachpb.Key,
 	kvValue *roachpb.Value,
 	rawValueBuf []byte,
-	oth *OriginTimestampCPutHelper,
+	oth OriginTimestampCPutHelper,
 	oldValues []tree.Datum,
 	kvOp KVInsertOp,
 	mustValidateOldPKValues bool,
diff --git a/pkg/sql/sqlclustersettings/clustersettings.go b/pkg/sql/sqlclustersettings/clustersettings.go
@@ -129,9 +129,7 @@ var LDRImmediateModeWriter = settings.RegisterStringSetting(
 	settings.ApplicationLevel,
 	"logical_replication.consumer.immediate_mode_writer",
 	"the writer to use when in immediate mode",
-	// TODO(jeffswenson): make the sql writer the default.
-	//metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeSQL), string(LDRWriterTypeLegacyKV), string(LDRWriterTypeCRUD)),
-	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeLegacyKV)),
+	metamorphic.ConstantWithTestChoice("logical_replication.consumer.immediate_mode_writer", string(LDRWriterTypeLegacyKV), string(LDRWriterTypeSQL)),
 	settings.WithValidateString(func(sv *settings.Values, val string) error {
 		if val != string(LDRWriterTypeSQL) && val != string(LDRWriterTypeLegacyKV) && val != string(LDRWriterTypeCRUD) {
 			return errors.Newf("immediate mode writer must be either 'sql', 'legacy-kv', or 'crud', got '%s'", val)
diff --git a/pkg/sql/tablewriter_delete.go b/pkg/sql/tablewriter_delete.go
@@ -55,11 +55,12 @@ func (td *tableDeleter) row(
 	values tree.Datums,
 	pm row.PartialIndexUpdateHelper,
 	vh row.VectorIndexUpdateHelper,
+	oth row.OriginTimestampCPutHelper,
 	mustValidateOldPKValues bool,
 	traceKV bool,
 ) error {
 	td.currentBatchSize++
-	return td.rd.DeleteRow(ctx, td.b, values, pm, vh, nil, mustValidateOldPKValues, traceKV)
+	return td.rd.DeleteRow(ctx, td.b, values, pm, vh, oth, mustValidateOldPKValues, traceKV)
 }
 
 // deleteIndex runs the kv operations necessary to delete all kv entries in the
diff --git a/pkg/sql/tablewriter_insert.go b/pkg/sql/tablewriter_insert.go
@@ -48,10 +48,11 @@ func (ti *tableInserter) row(
 	values tree.Datums,
 	pm row.PartialIndexUpdateHelper,
 	vh row.VectorIndexUpdateHelper,
+	oth row.OriginTimestampCPutHelper,
 	traceKV bool,
 ) error {
 	ti.currentBatchSize++
-	return ti.ri.InsertRow(ctx, &ti.putter, values, pm, vh, nil, row.CPutOp, traceKV)
+	return ti.ri.InsertRow(ctx, &ti.putter, values, pm, vh, oth, row.CPutOp, traceKV)
 }
 
 // tableDesc returns the TableDescriptor for the table that the tableInserter
diff --git a/pkg/sql/tablewriter_update.go b/pkg/sql/tablewriter_update.go
@@ -51,12 +51,13 @@ func (tu *tableUpdater) rowForUpdate(
 	oldValues, updateValues tree.Datums,
 	pm row.PartialIndexUpdateHelper,
 	vh row.VectorIndexUpdateHelper,
+	oth row.OriginTimestampCPutHelper,
 	mustValidateOldPKValues bool,
 	traceKV bool,
 ) (tree.Datums, error) {
 	tu.currentBatchSize++
 	return tu.ru.UpdateRow(
-		ctx, tu.b, oldValues, updateValues, pm, vh, nil, mustValidateOldPKValues, traceKV,
+		ctx, tu.b, oldValues, updateValues, pm, vh, oth, mustValidateOldPKValues, traceKV,
 	)
 }
 
diff --git a/pkg/sql/tablewriter_upsert.go b/pkg/sql/tablewriter_upsert.go
diff --git a/pkg/sql/update.go b/pkg/sql/update.go
diff --git a/pkg/sql/update_swap.go b/pkg/sql/update_swap.go
diff --git a/pkg/sql/upsert.go b/pkg/sql/upsert.go

Original file line number	Diff line number	Diff line change
`@@ -619,8 +619,9 @@ func buildRowKVs(`
`619`	`619`	`p := &capturePutter{}`
`620`	`620`	`var pm row.PartialIndexUpdateHelper`
`621`	`621`	`var vh row.VectorIndexUpdateHelper`
	`622`	`+ var oth row.OriginTimestampCPutHelper`
`622`	`623`	`for _, d := range datums {`
`623`		`- if err := inserter.InsertRow(context.Background(), p, d, pm, vh, nil, row.CPutOp, true /* traceKV */); err != nil {`
	`624`	`+ if err := inserter.InsertRow(context.Background(), p, d, pm, vh, oth, row.CPutOp, true /* traceKV */); err != nil {`
`624`	`625`	`return kvs{}, err`
`625`	`626`	`}`
`626`	`627`	`}`
Original file line number	Diff line number	Diff line change
`@@ -614,7 +614,8 @@ func (n *createTableNode) startExec(params runParams) error {`
`614`	`614`	`// indexes, partial, vector, or otherwise, to update.`
`615`	`615`	`var pm row.PartialIndexUpdateHelper`
`616`	`616`	`var vh row.VectorIndexUpdateHelper`
`617`		`- if err := ti.row(params.ctx, rowBuffer, pm, vh, params.extendedEvalCtx.Tracing.KVTracingEnabled()); err != nil {`
	`617`	`+ var oth row.OriginTimestampCPutHelper`
	`618`	`+ if err := ti.row(params.ctx, rowBuffer, pm, vh, oth, params.extendedEvalCtx.Tracing.KVTracingEnabled()); err != nil {`
`618`	`619`	`return err`
`619`	`620`	`}`
`620`	`621`	`}`