importccl: blazing fast IMPORT

[danhhz]

IMPORT is one of the first things our customers use in a POC and we use it extensively in our internal testing. Anything we can do to speed it up is pretty high-leverage.

We've laid a great foundation with direct ingestion of SSTables into RocksDB, but there is currently a big distributed sort that is limiting scalability and throughput. Luckily, the schema change team has recently switched index backfills to be based on a technique that uses direct SSTable ingestion, but without the distributed sort! There are a couple details to work out, but no fundamental reason we can't do the same thing for IMPORT, so now is a great time to revisit where we're at and lay down a roadmap for the medium-term future.

Specifically, I have the following goals:

• fixtures import should be as fast or faster than fixtures load
• fixtures import should scale to 100s of terabytes

fixtures import works by running IMPORT over CSVs that are synthesized from thin air, while fixtures load is a thin wrapper around RESTORE, so this goal is more-or-less that IMPORTon data already on a cluster should be faster thanRESTOREing data from google storage to a cluster on google compute. IMPORThas to do more work thanRESTORE` so if this is possible, it's because of network bandwidth constraints.

At a high-level, IMPORT currently reads a CSV line by line, converting it to our internal mvcc key/values. It currently works by putting these kvs in non-overlapping SSTables, which it then directly ingests. Because we don't have control of the order of rows in the CSVs (plus secondary indexes are basically never sorted in the same order), this requires that we first compute key splits such that the kv data between them is approximately the same size. Then we use these splits as buckets in a big distributed sort.

This big distributed sort means we do the CSV -> kv conversion twice. A distsql flow is set up to sample the splits, converting every CSV -> kvs in the process. Then a followup flow reruns the CSV -> kv conversion, except this time it routes everything according to the splits. This feeds to a second set of processors which buffers every kv in a bucket, sorts them, and runs AddSSTable. These second processors are also responsible for detecting unique index violations (in either the primary index or a secondary index) by finding duplicate keys when it constructs the SSTable for ingestion.

There are a bunch of inefficiencies in this process, but running the CSV->kv conversion twice and doing the distributed sort, along with all the buffering involved, dominates IMPORT's overall throughput. This is all because when we first built IMPORT, it was easier for AddSSTable to only work on non-overlapping buckets (and this was all it needed for RESTORE) The recent index backfill stuff has paved the way for doing overlapping AddSSTables, which means we can rip the double sample and the distributed sort out. @dt has even prototyped this in #34751! (Update: This merged in #34751!)

After that happens, it's worth looking into our new bottlenecks. As always, performance work should be benchmark driven, but here's some preliminary guesses for low hanging fruit.

• It's silly to be using CSV as an intermediate format for importing workload data. The magic IMPORT thing works by taking a set of rows as query parameters and outputting the data as a CSV, which IMPORT then has to parse into rows of Datums. All of this is allocation and cpu heavy. The sql team is adding efficient columnar formats and they'll have to be able to serialize them to use them in distributed flows. If they end up using a standard format, we could add it as an alternate non-CSV format for IMPORT, reducing basically all the allocations between the data generation and the kv generation. (Update: dt had a better idea and it merged in importccl: skip detour though CSV when importing from workload #34892!)

• In AddSSTable, we compute range mvccstats by making an iterator that merges existing rocksdb data and sstable data and handing that iterator to our common mvccstats code. This is because there could be existing data in the range and (because of unavoidable internal retries) we could end up applying the same sstable twice. Instead, we could compute the mvccstats for the sstable as we construct it and blindly add it to the existing stats, setting the estimate flag. This will not always be correct, but it's probably a good enough guess for the duration of the IMPORT. Then, after the IMPORT is finished, we go back and compute the exact mvcc stats and clear the estimate flag. This means that we only iterate the keyspace being imported into once, not once per sstable over a given span.

• SSTables are ingested into RocksDB via the IngestExternalFile call. Historically, this meant we had to write them to disk twice, because RocksDB might need to modify a small part of the file, but raft might later need back the exact original payload. At some point, we added an optimization to try the IngestExternalFile disallowing any modifications to the file, and if that failed, a copy was made and IngestExternalFile was tried again allowing modifications. As of some recent RocksDB version, it's now possible to run IngestExternalFile in a mode that never modifies the file, guaranteeing that it's only written once. (Update: This merged in storage/engine,libroach: 1x write amplification on ingest sst #34886!)

• Profile and speed up tpcc generator's impls of initial data generation: workload/tpcc: initial data load perf improvements #35322, and probably more

• Switch workload.InitialData.Batch to be columnar, which will allow for drastically reduced allocations during initial data generation of frequently used workloads (tpcc, bank), while keeping the easy interface for other workloads: workload: allow columnar data generation (alloc -73%, MB/s +60%) #35349

• Profile and speed up IMPORT: sqlbase: avoid allocations due to ErrNameString, speed up IMPORT by 10% #35317, and certainly more

TODO(dan): continue fleshing this all out and starting running perf experiments

Epic CRDB-2340

Jira issue: CRDB-4624

[danhhz]

In an attempt to start some standard measurements I added a benchmark that does a fixtures import tpcc --warehouses=1 on a single node. I also ran a fixtures import tpcc --warehouses=100 on a single node n1-standard-16 cluster. Multi-node is more representative of real world performance, but I'd first like to get a sense of (and improve) our speed of light.

tpcc-1 old is default new is with --experimental-direct-ingest

name                 old time/op    new time/op    delta
ImportFixtureTPCC-8     7.74s ± 3%     4.09s ± 6%  -47.18%  (p=0.008 n=5+5)

name                 old speed      new speed      delta
ImportFixtureTPCC-8  11.4MB/s ± 3%  21.7MB/s ± 6%  +89.46%  (p=0.008 n=5+5)

name                 old alloc/op   new alloc/op   delta
ImportFixtureTPCC-8    7.18GB ± 0%    4.12GB ± 0%  -42.62%  (p=0.008 n=5+5)

name                 old allocs/op  new allocs/op  delta
ImportFixtureTPCC-8     89.0M ± 0%     39.5M ± 0%  -55.56%  (p=0.008 n=5+5)

tpcc-100

default: 8m49s
direct ingest: 5m45s
delta: -34.8%

So we're not quite getting our linear speedup to 100, but we already know of some issues that could cause that. The biggest is probably that the old import with non-overlapping sstables shouldn't need any compactions, but the new overlapping ones will. I'd bet that write amplification goes up from tpcc-1 to tpcc-100. We might also now be filling up L1 and increasing read amplification for system traffic (jobs updates, liveness, etc). cc @dt

[dt]

Another thing I was thinking might be a factor is the fixed vs variable costs of running the sampling phase flow -- and in smaller import, those fixed costs don't shrink, meaning sampling is a bigger % of the overall import... thus skipping sampling entirely might see a bigger % impact on smaller imports.

[danhhz]

Perhaps, but sampling sets up a fairly simple flow, which I'd guess would be small in comparison to the 4s that tpcc-1 takes end-to-end.

My wild guesses are:

• Non-linear behavior in IngestExternalFile after some point
• Non-linear behavior in something else because the CSV "files" are really big
• GC. This thing generates a TON of garbage, so we're probably missing some of the cost of GC on smaller tests
• Tail latencies

I'm currently trying a single node tpcc-1000.

[danhhz]

Note to self: run our new benchmarks to measure the impact of #34886

[dt]

FWIW, in that PR I mentioned results on a single node MBP we measurable but not very large -- about 5% on 1.7gb tpcc import. It did cut the the direct ingestion write traffic almost half, but our direct writes were a fraction of the compactor's. The total disk write used (as measured by the OS) for importing 1.7gb went from 7.5gb to 6gb. #34886 (comment)

[danhhz]

Yup! I saw. I actually left it as a note to self instead of doing it right now because I want to see if we make any progress on the direct ingestion compactions. That bit, and the stability issues it comes with, is really preventing us from measuring our progress well right now. I'm mulling about it in the background.

[dt]

Wrote up some thoughts on #35219 re: how much we're re-compacting.

[danhhz]

cc @nvanbenschoten

[dt]

A few recently merged changes related to this from me and @adityamaru27 :

• ⬆Direct ingest is on by default.
• ⬇ Import SST ingestion scans existing data and prohibits key shadowing (required for correctness with unique / pk indexes) added ~1% overhead when ingesting non-overlapping SSTs, higher with overlaps.
• ⬆Pebble-based SSTable creation (~10% faster in RESTORE).
• ⬆Caller-provided MVCCStats.
• ⬇ Removal of pre-buffering by table/index ID (fixed OOM crashes). Thanks to range splits at index bounds this didn't change overlapping SSTs but did mean we flushed small index SSTs when the overall buffer filled but...
• ⬆ Splitting KV buffering into pk-vs-index buffers to mitigate removal of per-index buffers above (still uses much less mem, in part because Adder avoids expensive roachpb.Value).

A couple more random ideas from the last time I profiled an ingestion:

a) computing MVCCStats incrementally during SST creation iteration (instead of with a second iteration) should be pretty straightforward given the significant constraints on what we're putting in them (all unique user keys).

b) pass ~1mb kvBuf as the kvbatches sent from the frontends, which could also potentially maintain a boolean sorted to skip re-sorting sorted input (this would probably require plumbing separate batches or batch channels for pk-vs-index data).

c) figure out how we can do better at splitting and scattering.

[dt]

Direct-ingest is now on by default.

The direct-ingest tpcc1k roachtest is now sitting at in teens of minutes down from 40+ and manual testing of tpcc2k looks like 25-40m down from 50-60min in 19.1 (also using direct-ingest, or well over an hour with sorted).

At this point I think the big pieces of this that bulk-io had on its roadmap are done, so I think bulk-io will be removing this issue form its roadmap now. I'll leave it up to @danhhz to decide if we should actually close it though.

[awoods187]

this is so exciting! i'm looking forward to testing it out. Will this be used for nightlies? can we expect some savings and speedups there?

[dt]

it is on in nightlies (that use workload import). several of the changes that affect this all merged late last week, so that change in the tpcc1k roachtest is observed since 8/16.

[danhhz]

huge effort from bulk io on direct ingest, thanks @dt!

i've got two last ideas i've been meaning to write up. 1) speed up workload colbatch -> kv conversion (currently roundtrips through datums but doesn't need to) and 2) fewer copies in the addsstable path

[dt]

2a13ce6#diff-b010dc66fcc28059908e35277ad85915R157 is another likely big win during heavy ingestion loads -- we see the hangover of our containsEstiamtes flag coming up in how expensive splits become, particularly since we need to split a lot during ingestion.

If @adityamaru27 doesn't get to it before he leaves us (😢) we should try to make sure that actually gets done.

[danhhz]

Ah, good to know!

[dt]

I did a little more profiling and one of the big places where we're still spending time is sort.Sort on buffers. We're producing sorted data so I got to thinking it'd be cheaper to keep track of if it is sorted (just one comparison during each append to the buffer while we still think it is sorted) than to sort it from scratch (so O(n) instead of O(nlogn) ) but when I went to actually try this, I realized that our parallel creation of individually sorted batches feeding into one buffer ends up making an unsorted buffer. We could potentially sort batches to solve this, but it seems like that might be somewhat workload specific, so I got to thinking maybe the workload frontend should do that internally, and emit a sorted stream of sorted batches on kvCh. That would also eliminate the potential "jagged edge" problem that still pushes some of workload's SSTs to L0, when a flush occurs while one producer worker is ahead of another, causing the whole SST to overlap just because of one of the boundary batches.

I'm thinking it might make sense to add one more worker that just collects batches from the kvCh that currently all the other workers are pushing on to and does a little bit of buffering (e.g. numWorkers*2) in an attempt to emit batches sequentially.

That, combined with switch to a coldata.Bytes or bulk.kvBuf as the row.KVBatch internal format and bumping it up to 1mb should make the pipeline to the sst_batcher a fair bit cheaper.

I don't know if I'll actually get around to playing with this right away though so I stopped here to write it down.

[nvanbenschoten]

@dt is there more to do here or can this be closed?

[github-actions]

We have marked this issue as stale because it has been inactive for
18 months. If this issue is still relevant, removing the stale label
or adding a comment will keep it active. Otherwise, we'll close it in
10 days to keep the issue queue tidy. Thank you for your contribution
to CockroachDB!