distsql: fix sql.mem.distsql.current metric

This commit fixes double counting of remote DistSQL flows' memory usage
against `sql.mem.distsql.current` metric. This was the case since we
passed the metric to both the `flow` monitor (created for all flows,
both local and remote) and the `distsql` monitor (which tracks the
memory usage of all remote flows). This is now fixed by only passing the
metric in the former case.

Release note (bug fix): Previously, `sql.mem.distsql.current` metric
would double count the memory usage of remote DistSQL flows and this has
been fixed.

pkg/sql/buffer_util.go

@@ -81,6 +81,8 @@ func (c *rowContainerHelper) initMonitors(
 	ctx context.Context, evalContext *extendedEvalContext, opName redact.RedactableString,
 ) {
 	distSQLCfg := &evalContext.DistSQLPlanner.distSQLSrv.ServerConfig
+	// TODO(yuzefovich): currently the memory usage of c.memMonitor doesn't
+	// count against sql.mem.distsql.current metric. Fix it.
 	c.memMonitor = execinfra.NewLimitedMonitorNoFlowCtx(
 		ctx, evalContext.Planner.Mon(), distSQLCfg, evalContext.SessionData(),
 		redact.Sprintf("%s-limited", opName),

pkg/sql/distsql/server.go

@@ -85,9 +85,12 @@ func NewServer(
 		memMonitor: mon.NewMonitor(
 			"distsql",
 			mon.MemoryResource,
-			cfg.Metrics.CurBytesCount,
-			cfg.Metrics.MaxBytesHist,
-			-1, /* increment: use default block size */
+			// Note that we don't use 'sql.mem.distsql.*' metrics here since
+			// that would double count them with the 'flow' monitor in
+			// setupFlow.
+			nil, /* curCount */
+			nil, /* maxHist */
+			-1,  /* increment: use default block size */
 			noteworthyMemoryUsageBytes,
 			cfg.Settings,
 		),

pkg/sql/sem/eval/deps.go

@@ -195,6 +195,10 @@ type Planner interface {
 	tree.FunctionReferenceResolver
 
 	// Mon returns the Planner's monitor.
+	//
+	// TODO(yuzefovich): memory usage against this monitor doesn't count against
+	// sql.mem.distsql.current metric, audit the callers to see whether this is
+	// undesirable in some places.
 	Mon() *mon.BytesMonitor
 
 	// ExecutorConfig returns *ExecutorConfig

pkg/sql/tablewriter_upsert_opt.go

@@ -105,7 +105,7 @@ func (tu *optTableUpserter) init(
 		return err
 	}
 
-	// rowsNeeded, set upon initialization, indicates pkg/sql/backfill.gowhether or not we want
+	// rowsNeeded, set upon initialization, indicates whether or not we want
 	// rows returned from the operation.
 	if tu.rowsNeeded {
 		tu.resultRow = make(tree.Datums, len(tu.returnCols))