sqlbase: release memory when popping rows from RowContainer

This wasn't critical before because we were always popping rows during a
draining (followed by a call to `Clear` or `Close`). With the buffering router,
we will be alternating between popping rows and adding more rows, so we need to
release the memory.

Removing a distsql_agg "memory exceeded" test - it inadvertently used tuples and
wasn't run via distsql, and it now times out with local SQL (it no longer
exceeds the budget).

pkg/sql/logictest/testdata/logic_test/distsql_agg

@@ -414,9 +414,6 @@ SELECT COUNT(*) FROM one AS a, one AS b, two AS c
 ----
 1000
 
-statement error memory budget exceeded
-SELECT SUM(d1.a), MAX(d1.a), MIN(d1.a) FROM data as d1, data as d2 GROUP BY (d1.b, d1.c, d1.d, d2.a, d2.b, d2.c, d2.d)
-
 query T
 SELECT "URL" FROM [EXPLAIN (DISTSQL) SELECT SUM(a), SUM(b), SUM(c) FROM data GROUP BY d HAVING SUM(a+b) > 10]
 ----

pkg/sql/mon/mem_usage.go

@@ -541,11 +541,16 @@ func (b *BoundAccount) ResizeItem(ctx context.Context, oldSz, newSz int64) error
 	return b.mon.ResizeItem(ctx, &b.MemoryAccount, oldSz, newSz)
 }
 
-// Grow is an accessor for b.mon.Grow.
+// Grow is an accessor for b.mon.GrowAccount.
 func (b *BoundAccount) Grow(ctx context.Context, x int64) error {
 	return b.mon.GrowAccount(ctx, &b.MemoryAccount, x)
 }
 
+// Shrink is an accessor for b.mon.ShrinkAccount.
+func (b *BoundAccount) Shrink(ctx context.Context, x int64) {
+	b.mon.ShrinkAccount(ctx, &b.MemoryAccount, x)
+}
+
 // reserveMemory declares an allocation to this monitor. An error is
 // returned if the allocation is denied.
 func (mm *MemoryMonitor) reserveMemory(ctx context.Context, x int64) error {

pkg/sql/sqlbase/row_container.go

@@ -305,10 +305,20 @@ func (c *RowContainer) PopFirst() {
 	if c.numCols != 0 {
 		c.deletedRows++
 		if c.deletedRows == c.rowsPerChunk {
-			c.deletedRows = 0
+			// We release the memory for rows in chunks. This includes the
+			// chunk slice (allocated by allocChunks) and the Datums.
+			size := c.chunkMemSize
+			for i, pos := 0, 0; i < c.rowsPerChunk; i, pos = i+1, pos+c.numCols {
+				size += c.rowSize(c.chunks[0][pos : pos+c.numCols])
+			}
 			// Reset the pointer so the slice can be garbage collected.
 			c.chunks[0] = nil
+			c.deletedRows = 0
 			c.chunks = c.chunks[1:]
+
+			// We don't have a context plumbed here, but that's ok: it's not actually
+			// used in the shrink paths.
+			c.memAcc.Shrink(context.TODO(), size)
 		}
 	}
 }