sql: avoid contention between long mutation and auto stats

Previously, when processing large mutations in batches (see
`BatchedNext` method), we would notify the stats refresher about the
number of rows affected after each batch. If the mutation is long, then
this could have triggered the auto stats collection that would contend
with the mutation that triggered it. There is actually no reason to try
collecting auto stats on those just-mutated rows since in most cases
they haven't been committed yet. This commit fixes this issue by
notifying the stats refresher about the total number of rows affected
when the last batch of the mutation has just been processed. The bug has
been present since about 19.1 version.

Note that we generally don't perform well with long mutations, but it
seems like a worthwhile improvement nonetheless. We can't remove
contention with auto stats triggered by _other_ mutations, but at least
we can do so for self-inflicted ones. (Also probably a more important
retry reason is the closed TS interval.)

Release note (bug fix): Large mutation statements (INSERTs, UPDATEs,
DELETEs, UPSERTs) are now less likely to encounter contention with
automatic table stats collection in some cases. The bug has been present
since 19.1 version.

Author: yuzefovich

pkg/sql/colexec/insert.go

@@ -45,6 +45,11 @@ type vectorInserter struct {
 	mutationQuota int
 	// If auto commit is true we'll commit the last batch.
 	autoCommit bool
+	// rowsWritten tracks the number of rows written by the vectorInserter so
+	// far.
+	rowsWritten int
+	// statsRefresherNotified is set once we notify the stats refresher.
+	statsRefresherNotified bool
 }
 
 var _ colexecop.Operator = &vectorInserter{}
@@ -123,6 +128,15 @@ func (v *vectorInserter) Next() coldata.Batch {
 	ctx := v.Ctx
 	b := v.Input.Next()
 	if b.Length() == 0 {
+		if !v.statsRefresherNotified {
+			// We've just exhausted the input, so let's notify the stats
+			// refresher.
+			// TODO(yuzefovich): when auto-commit enabled, the inserted rows
+			// will be visible sooner than at the end. Is it worth notifying the
+			// stats refresher earlier in that case?
+			v.flowCtx.Cfg.StatsRefresher.NotifyMutation(v.desc, v.rowsWritten)
+			v.statsRefresherNotified = true
+		}
 		return coldata.ZeroBatch
 	}
 
@@ -200,7 +214,7 @@ func (v *vectorInserter) Next() coldata.Batch {
 	v.retBatch.ColVec(0).Int64()[0] = int64(b.Length())
 	v.retBatch.SetLength(1)
 
-	v.flowCtx.Cfg.StatsRefresher.NotifyMutation(v.desc, b.Length())
+	v.rowsWritten += b.Length()
 
 	return v.retBatch
 }

pkg/sql/delete.go

@@ -159,11 +159,10 @@ func (d *deleteNode) BatchedNext(params runParams) (bool, error) {
 		}
 		// Remember we're done for the next call to BatchedNext().
 		d.run.done = true
+		// Possibly initiate a run of CREATE STATISTICS.
+		params.ExecCfg().StatsRefresher.NotifyMutation(d.run.td.tableDesc(), int(d.run.td.rowsWritten))
 	}
 
-	// Possibly initiate a run of CREATE STATISTICS.
-	params.ExecCfg().StatsRefresher.NotifyMutation(d.run.td.tableDesc(), d.run.td.lastBatchSize)
-
 	return d.run.td.lastBatchSize > 0, nil
 }
 

pkg/sql/insert.go

@@ -360,11 +360,10 @@ func (n *insertNode) BatchedNext(params runParams) (bool, error) {
 		}
 		// Remember we're done for the next call to BatchedNext().
 		n.run.done = true
+		// Possibly initiate a run of CREATE STATISTICS.
+		params.ExecCfg().StatsRefresher.NotifyMutation(n.run.ti.tableDesc(), int(n.run.ti.rowsWritten))
 	}
 
-	// Possibly initiate a run of CREATE STATISTICS.
-	params.ExecCfg().StatsRefresher.NotifyMutation(n.run.ti.tableDesc(), n.run.ti.lastBatchSize)
-
 	return n.run.ti.lastBatchSize > 0, nil
 }
 

pkg/sql/update.go

@@ -168,11 +168,10 @@ func (u *updateNode) BatchedNext(params runParams) (bool, error) {
 		}
 		// Remember we're done for the next call to BatchedNext().
 		u.run.done = true
+		// Possibly initiate a run of CREATE STATISTICS.
+		params.ExecCfg().StatsRefresher.NotifyMutation(u.run.tu.tableDesc(), int(u.run.tu.rowsWritten))
 	}
 
-	// Possibly initiate a run of CREATE STATISTICS.
-	params.ExecCfg().StatsRefresher.NotifyMutation(u.run.tu.tableDesc(), u.run.tu.lastBatchSize)
-
 	return u.run.tu.lastBatchSize > 0, nil
 }
 

pkg/sql/upsert.go

@@ -133,11 +133,10 @@ func (n *upsertNode) BatchedNext(params runParams) (bool, error) {
 		}
 		// Remember we're done for the next call to BatchedNext().
 		n.run.done = true
+		// Possibly initiate a run of CREATE STATISTICS.
+		params.ExecCfg().StatsRefresher.NotifyMutation(n.run.tw.tableDesc(), int(n.run.tw.rowsWritten))
 	}
 
-	// Possibly initiate a run of CREATE STATISTICS.
-	params.ExecCfg().StatsRefresher.NotifyMutation(n.run.tw.tableDesc(), n.run.tw.lastBatchSize)
-
 	return n.run.tw.lastBatchSize > 0, nil
 }