crosscluster/logical: notify sql stats of ldr immediate writes

Release note (enterprise change): Rows replicated by LDR in immediate mode are now factored into the determination of when to recompute SQL table statistics.

Epic: CRDB-40872.

pkg/ccl/crosscluster/logical/BUILD.bazel

@@ -68,6 +68,7 @@ go_library(
         "//pkg/sql/sem/tree",
         "//pkg/sql/sessiondata",
         "//pkg/sql/sessiondatapb",
+        "//pkg/sql/stats",
         "//pkg/sql/syntheticprivilege",
         "//pkg/sql/types",
         "//pkg/util/admission/admissionpb",

pkg/ccl/crosscluster/logical/logical_replication_writer_processor.go

@@ -31,6 +31,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/rowexec"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
+	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -552,6 +553,10 @@ func (lrw *logicalReplicationWriterProcessor) checkpoint(
 		// have exited based on an error.
 		return nil
 	}
+
+	for _, p := range lrw.bh {
+		p.ReportMutations(lrw.FlowCtx.Cfg.StatsRefresher)
+	}
 	lrw.metrics.CheckpointEvents.Inc(1)
 	lrw.debug.RecordCheckpoint(lrw.frontier.Frontier().GoTime())
 	return nil
@@ -957,6 +962,7 @@ type BatchHandler interface {
 	HandleBatch(context.Context, []streampb.StreamEvent_KV) (batchStats, error)
 	GetLastRow() cdcevent.Row
 	SetSyntheticFailurePercent(uint32)
+	ReportMutations(*stats.Refresher)
 	Close(context.Context)
 }
 
@@ -968,6 +974,7 @@ type RowProcessor interface {
 	ProcessRow(context.Context, isql.Txn, roachpb.KeyValue, roachpb.Value) (batchStats, error)
 	GetLastRow() cdcevent.Row
 	SetSyntheticFailurePercent(uint32)
+	ReportMutations(*stats.Refresher)
 	Close(context.Context)
 }
 
@@ -1029,6 +1036,10 @@ func (t *txnBatch) SetSyntheticFailurePercent(rate uint32) {
 	t.rp.SetSyntheticFailurePercent(rate)
 }
 
+func (t *txnBatch) ReportMutations(s *stats.Refresher) {
+	t.rp.ReportMutations(s)
+}
+
 func (t *txnBatch) Close(ctx context.Context) {
 	t.rp.Close(ctx)
 }

pkg/ccl/crosscluster/logical/lww_kv_processor.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/row"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/util/admission/admissionpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/errors"
@@ -114,6 +115,17 @@ func (p *kvRowProcessor) ProcessRow(
 
 }
 
+func (p *kvRowProcessor) ReportMutations(refresher *stats.Refresher) {
+	for _, w := range p.writers {
+		if w.unreportedMutations > 0 && w.leased != nil {
+			if desc := w.leased.Underlying(); desc != nil {
+				refresher.NotifyMutation(desc.(catalog.TableDescriptor), w.unreportedMutations)
+				w.unreportedMutations = 0
+			}
+		}
+	}
+}
+
 // maxRefreshCount is the maximum number of times we will retry a KV batch that has failed with a
 // ConditionFailedError with HadNewerOriginTimetamp=true.
 const maxRefreshCount = 10
@@ -253,6 +265,9 @@ func (p *kvRowProcessor) addToBatch(
 		}
 	}
 
+	// Note that we should report this mutation to sql stats refresh later.
+	w.unreportedMutations++
+
 	return nil
 }
 
@@ -317,6 +332,10 @@ type kvTableWriter struct {
 	ru               row.Updater
 	ri               row.Inserter
 	rd               row.Deleter
+
+	// Mutations to the table this writer wraps that should be reported to sql
+	// stats at some point.
+	unreportedMutations int
 }
 
 func newKVTableWriter(

pkg/ccl/crosscluster/logical/lww_row_processor.go

@@ -32,6 +32,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondatapb"
+	"github.com/cockroachdb/cockroach/pkg/sql/stats"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/metamorphic"
@@ -232,6 +233,11 @@ func makeSQLProcessorFromQuerier(
 	}, nil
 }
 
+// ReportMutations implements the RowProcessor interface, but is a no-op for
+// sqlRowProcessor because its mutations are already reported by the queries it
+// runs when they are run.
+func (sqlRowProcessor) ReportMutations(_ *stats.Refresher) {}
+
 func (*sqlRowProcessor) Close(ctx context.Context) {}
 
 var errInjected = errors.New("injected synthetic error")