rowexec: change lookup join batch size to be specified in bytes

[asubiotto]

The previous batch size of 100 rows was overly defensive in order to protect
against result memory blowup. Since the addition of TargetBytes to the roachpb
API, a result will no longer take more than 10MiB. Having a higher batch size
allows us to amortize the lookup cost.

The concern with a larger batch size is no longer whether the results will
cause a memory blowup, but whether the results need to be spilled to disk.
A variation of TPCH Q10 which has a lookup join with a ratio of 8 result bytes
per lookup byte to determine that 6MiB was a good default batch size. However,
since this ratio changes based on the query and lookup columns, we eventually
will want to make this lookup batch size dynamic.

With the new default of 6MiB, the TPCH Q10 variation shows a 15% improvement
on local lookups (2.8s v 3.3s) and an 88% improvement (11s vs 97s) with an
artificial 150ms latency injected into kvfetcher fetches.

Release note (performance improvement): lookup join performance has been
improved.

Closes #39471

[cockroach-teamcity]

This change is 

[yuzefovich]

Reviewed 3 of 3 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto and @jordanlewis)

---

pkg/sql/rowexec/joinreader.go, line 44 at r1 (raw file):

// TODO(asubiotto): Eventually we might want to adjust this batch size
//  dynamically based on whether the result row container spilled or not.
const joinReaderDefaultLookupBatchSizeBytes = 6 << 20 /* 6 MiB */

This number seems too "overfitted" to the query that we ran the benchmarks on, no? Also I think it should depend on 10MiB constant, so maybe something like TargetBytesSize / 2 or TargetBytesSize / 4 would be better?

---

pkg/sql/rowexec/joinreader.go, line 97 at r1 (raw file):

	// Batch size for fetches. Not a constant so we can lower for testing.
	batchSizeBytes int64
	curBatchSize   int64

nit: s/curBatchSize/curBatchSizeBytes/g for symmetry.

---

pkg/sql/rowexec/utils_test.go, line 62 at r1 (raw file):

	case *joinReader:
		// Reduce batch size to exercise batching logic.
		pt.SetBatchSizeBytes(int64(inputRows[0].Size()))

nit: missing a factor of 2, not sure if it matters though.

[asubiotto]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto, @jordanlewis, and @yuzefovich)

---

pkg/sql/rowexec/joinreader.go, line 44 at r1 (raw file):

Previously, yuzefovich wrote…

This number seems too "overfitted" to the query that we ran the benchmarks on, no? Also I think it should depend on 10MiB constant, so maybe something like TargetBytesSize / 2 or TargetBytesSize / 4 would be better?

It could definitely be. The goal of this batch size is to be as big as possible while avoiding disk spilling (performance tanks due to random accesses of the disk container). I think it's a good starting point since the query in question has a big lookup to result byte ratio. I think that we might want to end up calculating a dynamic batch size by assuming that that ratio will stay more or less the same and tracking how much memory is used by the results.
Note that the 10MiB TargetBytes is a per-fetch limit, but we can't stop fetching until we've got all the results for our current lookup batch (since they might be out of order and we want to preserve the order of the input).

I'm also going to look into whether we can change the way we're storing rows on disk. If we match a result row with a lookup ordinal and sort by that, I think we can change the accesses to be sequential which might change this batch size equation.

[yuzefovich]

since they might be out of order and we want to preserve the order of the input

I might be totally off here, but are there cases when we don't care about preserving the order? Maybe we could have different modes of joinReader then? That would probably break the assumptions, but I thought I'd ask.

[asubiotto]

That's actually a very good observation. I'm going to ask the optimizer folks. I'm also wondering whether it would be more efficient to not order anything and if we rely on ordering plan a sorter after the lookup join. If we can't specify the sort in terms of some pre-existing ordering key on the left side, I think that we probably don't care about ordering at all (since the only "ordering" of the lookup side will be accidental in this case).

[yuzefovich]

I had another thought: since the access pattern of the indexed row container is random, maybe trying disabling its cache when we spilled to disk? The comments say that disabling should only be "used in tests", but that could be ignored. Maybe then spilling to disk won't be that bad, and we can continue reusing the same row container implementation.

[asubiotto]

Possibly. I talked to the optimizer folks and it seems like we already communicate a required ordering when we need it. However, it looks like in the distsql physical planning code we expect a lookup joiner to always satisfy this ordering and only plan a merge ordering stage after it.

I think we'll want to provide two different implementations with as much codeshare as possible for the required ordering vs no ordering case. This will also allow us to have different batching strategies. I think what needs to be done is:

1. For the no ordering required case we try to maximize throughput with the highest possible batch size (since we can return results in a streaming fashion).
2. For the ordering case we need to do what we have here. However:
   • Improve the disk spilling case by trying to either remove the caching and see if it has an effect or try to rethink how we do disk spilling (one possibility is to push this ordering to a sortable disk row container for the results so the accesses are sequential).
   • We should try to dynamically increase the batch size but be defensive about it, to avoid disk spilling. We'll have to reevaluate with the improved strategy the cost of doing so.

Before we make any improvements though, I think we need to work on having benchmarks for a lot of different cases. I'll work on adding these and distilling these thoughts into issues.

[blathers-crl]

❌ The GitHub CI (Cockroach) build has failed on 2b6c92ec.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is otan.}

[blathers-crl]

❌ The GitHub CI (Cockroach) build has failed on 95d334bb.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is otan.}

[blathers-crl]

❌ The GitHub CI (Cockroach) build has failed on 832b3ef6.

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is otan.}

[asubiotto]

This is no longer a WIP and is RFAL. The ordering and no ordering cases where separated in another PR, allowing us to use different batch sizes for these cases. Running TPCH queries, the ideal batch size seems to be 2MiB for the no ordering case. Here is a graph for how the runtime evolves for queries 7,10,11:

And now including query 9:

I am running a microbenchmark diff and will post the results here when that completes.

[yuzefovich]

Reviewed 4 of 4 files at r2.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @asubiotto and @jordanlewis)

---

pkg/sql/rowexec/joinreader.go, line 97 at r1 (raw file):

Previously, yuzefovich wrote…

nit: s/curBatchSize/curBatchSizeBytes/g for symmetry.

Ping.

---

pkg/sql/rowexec/utils_test.go, line 62 at r1 (raw file):

Previously, yuzefovich wrote…

nit: missing a factor of 2, not sure if it matters though.

Ping.

[yuzefovich]

nit: the PR description needs an update.

[jordanlewis]

Reviewable status: complete! 2 of 0 LGTMs obtained (waiting on @asubiotto and @jordanlewis)

[asubiotto]

The current status of this is that there are some performance regressions in microbenchmarks which is curious given the TPCH improvements. I want to take a look at this before I merge the PR.

[asubiotto]

I focused on BenchmarkJoinReader/reqOrdering=false/matchratio=onetothirtytwo/lookuprows=16384-16 (here are the benchmark results, although I think I messed up the formatting when copy pasting somehow https://gist.github.com/asubiotto/70428e177b0865b670b154a3f19c4b34).

I first tried modifying the benchmark to use disk (I already thought it did, but it was only using it for the temporary engine, not the primary store) but it looks like there's still a noticeable difference.

Looking at the profiles, the biggest difference seems to be due to more timestamp cache use because of the bigger read batches. One thing that stood out in the profile was some expensive error checking that I'm fixing in #49324. The other thing is that it seems like we're allocating a lot in AddRange

I'm not familiar enough with the timestamp cache to know whether this is necessary or not (cc @nvanbenschoten, maybe you can provide some insight?)

Given that this regression shows up only in the microbenchmarks, I'm fine with merging this as is. @yuzefovich / @jordanlewis lmk what you think and I'll go ahead.

[yuzefovich]

I'm in favor of merging as is, but we should probably open an issue to investigate the timestamp cache allocations.

[nvanbenschoten]

The timestamp cache allocates large slabs of memory at a time - 32 MB chunk by default. I think the problem here is that we dramatically reduce this allocation size for tests. See TestSklPageSize, which is all the way down to 128KB. We see this size allocation in the pprof output.

This is kind of like how we use an in-memory storage engine for tests. We'll either want to configure the tscache properly in benchmarks with the usual allocation size so as to provide a more realistic environment for benchmarks or just ignore it in general.

For this PR though, you'll probably want to set TestSklPageSize = defaultSklPageSize and re-run the benchmarks to see how much of the regression goes away.

[petermattis]

The timestamp cache allocates large slabs of memory at a time - 32 MB chunk by default. I think the problem here is that we dramatically reduce this allocation size for tests. See TestSklPageSize, which is all the way down to 128KB. We see this size allocation in the pprof output.

I wonder if we should do what Pebble does and double the size of the page on each allocation up to some maximum.

[nvanbenschoten]

I like that idea a lot. Out of curiosity, where does Pebble do this? With its block cache? Its memtables? What are the minimum and maximum sizes that it uses?

[petermattis]

I like that idea a lot. Out of curiosity, where does Pebble do this? With its block cache? Its memtables? What are the minimum and maximum sizes that it uses?

It only does this with the memtable. See initialMemTableSize which is 256KB. See also DB.newMemTable() which is where the size is bumped.

[asubiotto]

Thanks for the info. Increasing this page size did help, but the microbenchmarks are still showing a regression. Will investigate further independently from this PR.

bors r=yuzefovich,jordanlewis

[craig]

Build succeeded

• GitHub CI (Cockroach)