colexec: emit batches with large rows early

cockroach-teamcity commented 1 year ago

Fix the vectorized engine to emit batches eagerly in case they have wide rows:

merge joiner
cross joiner
vectorized concat_agg when used as window function.

Some reproductions

```sql CREATE TABLE l (l_k INT, l_v STRING, INDEX (l_k) STORING (l_v)); CREATE TABLE r (r_k INT, r_v STRING, INDEX (r_k) STORING (r_v)); INSERT INTO l SELECT i // 100, repeat('l', 60000) FROM generate_series(0, 899) AS g(i); INSERT INTO r SELECT i // 10, repeat('r', 60000) FROM generate_series(0, 99) AS g(i); -- sizes in the next two queries can be tweaked for any target memory usage INSERT INTO l SELECT 100000, repeat('l', 1000000) FROM generate_series(0, 99) AS g(i); INSERT INTO r SELECT 100000, repeat('l', 1000000) FROM generate_series(0, 19) AS g(i); ANALYZE l; ANALYZE r; EXPLAIN ANALYZE (VERBOSE) SELECT length(l_v), length(r_v) FROM l INNER MERGE JOIN r ON l_k = r_k; ``` ```sql CREATE TABLE t (s STRING); INSERT INTO t SELECT repeat('a', 50) FROM generate_series(1, 1000); ANALYZE t; EXPLAIN ANALYZE SELECT concat_agg(t1.s ORDER BY t1.s DESC) FROM t AS t1, t AS t2; ```

Jira issue: CRDB-30864

yuzefovich commented 1 year ago

Node 1 OOMed:


fatal error: runtime: out of memory

goroutine 4294 [running]:
runtime.systemstack_switch()
    GOROOT/src/runtime/asm_amd64.s:459 fp=0xc004d271a8 sp=0xc004d271a0 pc=0x4cb9a0
runtime.(*mheap).alloc(0x57633e000?, 0x2bb19f?, 0x4?)
    GOROOT/src/runtime/mheap.go:912 +0x65 fp=0xc004d271f0 sp=0xc004d271a8 pc=0x488ee5
runtime.(*mcache).allocLarge(0xc001a9cd00?, 0x57633d400, 0x1)
    GOROOT/src/runtime/mcache.go:233 +0x85 fp=0xc004d27240 sp=0xc004d271f0 pc=0x4762c5
runtime.mallocgc(0x57633d400, 0x5663460, 0x1)
    GOROOT/src/runtime/malloc.go:1029 +0x57e fp=0xc004d272b8 sp=0xc004d27240 pc=0x46c51e
runtime.makeslice(0x469401?, 0xc00bd43600?, 0xc004d27340?)
    GOROOT/src/runtime/slice.go:103 +0x52 fp=0xc004d272e0 sp=0xc004d272b8 pc=0x4b1092
github.com/cockroachdb/cockroach/pkg/col/coldata.(*element).setNonInlined(0x60023e0?, {0xc06d06a000, 0x15d8cf5, 0x1?}, 0xc002e845c0)
    github.com/cockroachdb/cockroach/pkg/col/coldata/bytes.go:152 +0x10c fp=0xc004d27350 sp=0xc004d272e0 pc=0x10082ac
github.com/cockroachdb/cockroach/pkg/col/coldata.(*element).copy(0xc0074c5a40?, {{0x0, 0x15d8cf5, 0x15d8cf5}, {0x0, 0x0, 0x0, 0x0, 0x0, 0x0}, ...}, ...)
    github.com/cockroachdb/cockroach/pkg/col/coldata/bytes.go:125 +0x6e fp=0xc004d27388 sp=0xc004d27350 pc=0x100812e
github.com/cockroachdb/cockroach/pkg/col/coldata.(*Bytes).Copy(...)
    github.com/cockroachdb/cockroach/pkg/col/coldata/bytes.go:236
github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecjoin.(*crossJoinerBase).buildFromLeftInput.func1()
    github.com/cockroachdb/cockroach/bazel-out/k8-opt/bin/pkg/sql/colexec/colexecjoin/crossjoiner.eg.go:850 +0x1bad fp=0xc004d27b08 sp=0xc004d27388 pc=0x2a4b26d
github.com/cockroachdb/cockroach/pkg/sql/colmem.(*Allocator).PerformOperation(0xc004d27b78?, {0xc008519500, 0x50, 0xc004cb1cc0?}, 0xc004d27b98?)
    github.com/cockroachdb/cockroach/pkg/sql/colmem/allocator.go:460 +0x49 fp=0xc004d27b60 sp=0xc004d27b08 pc=0x23d3229
github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecjoin.(*crossJoinerBase).buildFromLeftInput(0xc006d5f1e0, {0xc002f24000?, 0x4fc56a?}, 0x0)
    github.com/cockroachdb/cockroach/bazel-out/k8-opt/bin/pkg/sql/colexec/colexecjoin/crossjoiner.eg.go:47 +0xc5 fp=0xc004d27bd0 sp=0xc004d27b60 pc=0x2a49665
github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecjoin.(*crossJoiner).Next(0xc000c1c480)
    github.com/cockroachdb/cockroach/pkg/sql/colexec/colexecjoin/crossjoiner.go:118 +0x1c9 fp=0xc004d27c20 sp=0xc004d27bd0 pc=0x2a41ae9
...