sql: change UPSERT to not require pre-fetching everything upfront

[knz]

Commit pull away from #23373 to ease reviewing.
Fixes #23660.

tl;dr: lifts limitation, slight perf boost.
Change required by a separate patch (#23373) that introduces
semantically correct behavior for mutation statements.

Prior to this patch, UPSERT with secondary indexes and INSERT ON
UPDATE DO SET ... would proceed as follows:

• it would load all the inserted rows into memory.
• it would pre-determine in one scan which of these
  already have data in KV (a conflict).
• after this initialization, it would then proceed to resolve the
  upsert, by either inserting new rows (for each source row for which
  no conflict was seen) or updating an existing row (for each source
  row where there was a conflict).

This situation caused three unrelated problems.

1. The first problem is that if there are many rows, the KV batch could
   blow up and fail / make everything slow. In comparison, we like the
   DELETE logic that was recently modified to operate using multiple
   short(er) batches. The UPSERT code should be able to evolve to do the
   same.

2. The second problem is that this code is unable to process two
   subsequent upserts to the same PK. For example, suppose that a table
   has no data to start with, and the following statement is issued:

   INSERT INTO kv(k) VALUES (1), (1) ON CONFLICT(k) DO UPDATE SET v = 2
   

   The initial processing described above would determine that there is
   no conflict for neither values being inserted. Then after this
   initialization, processing would start. The first iteration would see
   there was no conflict, and insert a fresh row. The second iteration
   would see there was no conflict (again, the conflict information was
   computed initially), and (try to) insert a fresh row and fail with an
   error (row already exists).

   Because the conflict detection is done initially, it becomes impossible
   to properly upsert multiple rows with the same PK.

   Even worse, if the conflicting index is a secondary index, it is
   possible for UPSERT to complete without error but without processing
   the upsert properly, which is a plain correctness bug.

   To avoid both this latter bug and the conflict error, the previous
   code was equipped with a detection mechanism to reject multiple
   updates to the same PK. This check was necessary for correctness, but
   it was unfortunate because it rejects upserts that are valid in
   PostgreSQL.

3. for consistency with the check required by problem 2, the "fast
   path" of UPSERT was equipped with a check for the same
   condition. This is also rather unfortunate because the fast path
   uses different code that is not otherwise subject to the
   correctness problem above and could already support multiple
   updates to the same PK outright. The check was there thus only for
   UX consistency, so the user sees the same error regardless of which
   code path is used.

This patch aims to resolve problem 1, and ends up solving problems 2
and 3 as a side effect. It changes the main processing to be like
this:

• like before, it loads the rows and pre-loads the existing
  row information during the initialization.
• however, during the main loop, any insert or update feeds back
  the new row into the existing row information, so that
  subsequent iterations can access the updated data.

Because this new processing enables incremental upsert resolution,
this resolves problem 1, because now the processing can operate
in smaller batches. (This change will be performed separately in #23373)

Also it does so without the risk for correctness problems by feeding
earlier updates into later updates. So the check for duplicates can be
removed and the 2nd problem above is resolved as well.

And then since the slow/general code doesn't need this check, the fast
path code doesn't need it either any more (it was only implemented on
both sides for consistency). So this patch removes the check on the
fast path too and problem 3 is resolved as well.

Very small performance boost at large row counts:

name                                old time/op    new time/op    delta
SQL/Cockroach/Upsert/count=1-16        833µs ± 9%     805µs ± 4%    ~     (p=0.518 n=5+9)
SQL/Cockroach/Upsert/count=10-16      1.09ms ± 1%    1.09ms ± 2%    ~     (p=1.000 n=5+10)
SQL/Cockroach/Upsert/count=100-16     3.99ms ± 3%    3.77ms ± 9%    ~     (p=0.099 n=5+10)
SQL/Cockroach/Upsert/count=1000-16    28.0ms ±12%    25.4ms ± 4%  -9.47%  (p=0.003 n=5+8)

name                                old alloc/op   new alloc/op   delta
SQL/Cockroach/Upsert/count=1-16        113kB ± 1%     112kB ± 1%    ~     (p=0.206 n=5+10)
SQL/Cockroach/Upsert/count=10-16       155kB ± 1%     155kB ± 1%    ~     (p=0.768 n=5+10)
SQL/Cockroach/Upsert/count=100-16      716kB ± 2%     716kB ± 2%    ~     (p=0.859 n=5+10)
SQL/Cockroach/Upsert/count=1000-16    6.31MB ± 1%    6.29MB ± 1%    ~     (p=0.594 n=5+10)

name                                old allocs/op  new allocs/op  delta
SQL/Cockroach/Upsert/count=1-16        1.02k ± 1%     0.99k ± 1%  -2.32%  (p=0.001 n=5+10)
SQL/Cockroach/Upsert/count=10-16       1.39k ± 3%     1.35k ± 5%    ~     (p=0.218 n=5+10)
SQL/Cockroach/Upsert/count=100-16      5.39k ± 4%     5.23k ±15%    ~     (p=0.129 n=5+10)
SQL/Cockroach/Upsert/count=1000-16     45.7k ± 6%     43.5k ±11%    ~     (p=0.129 n=5+10)

Release note (sql change): lifted a limitation that prevented UPSERTs
or INSERT ON CONFLICT DO UPDATE over multiple values with the same
primary key.

[cockroach-teamcity]

This change is 

[knz]

Note this currently implements "only 1 result returned by RETURNING" as discussed on the mailing list (option 1) and thus still precludes batching.

The code would be further simplified and batchability reached if I could change it to "return the rows as they are processed".

[knz]

I have added a second commit that implements option 2 and explains in the commit message the semantics of RETURNING. RFAL

[knz]

I think once the dust settles on option 1 vs option 2 (I'm assuming option 2 will prevail), I can squash the two commits. Thoughts?

[RaduBerinde]

Yeah I would squash.

I have a few minor comments. Looks good overall but I'm pretty unfamiliar with this code so I would get another review.

---

Review status: 0 of 3 files reviewed at latest revision, 7 unresolved discussions, all commit checks successful.

---

pkg/sql/tablewriter.go, line 294 at r2 (raw file):

	// rowTemplate is used to prepare rows to add to rowsUpserted.
	rowTemplate tree.Datums
	// rowIdxToRetIdx maps the indexes in the inserted rows

[nit] indices

---

pkg/sql/tablewriter.go, line 408 at r2 (raw file):

// finalize commits to tu.txn any rows batched up in tu.insertRows.
func (tu *tableUpserter) finalize(

This 200+ function is very hard to follow, can you think of a way to break things up a little?

---

pkg/sql/tablewriter.go, line 528 at r2 (raw file):

				// There is also no RETURNING result.
				// See https://www.postgresql.org/docs/10/static/sql-insert.html and the
				// quoted exceirpt above.

excerpt

---

pkg/sql/tablewriter.go, line 600 at r2 (raw file):

				// Now add the new one.
				newExistingRowIdx = len(existingRows)
				pkToRowIdx[string(newUpsertRowPK)] = newExistingRowIdx

[nit] Using the newExistingRowIdx variable is confusing in this block, maybe just set to len(existingRows) directly

---

pkg/sql/tablewriter.go, line 603 at r2 (raw file):

				existingRows = append(existingRows, updatedRow)
			} else {
				// If the PK has changed, we'll have a new rowIdx. In that case,

[nit] usually having the ok case first is easier to read

---

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

				// remove the previous mapping.
				if newExistingRowIdx != existingRowIdx {
					delete(pkToRowIdx, string(upsertRowPK))

This code is pretty hard to understand. We are (to some extent) using the map as a proxy for comparing upsertRowPK with newUpsertRowPK. Could we just compare them directly?

---

pkg/sql/tablewriter.go, line 761 at r2 (raw file):

func (tu *tableUpserter) fetchExisting(
	ctx context.Context, traceKV bool,
) ([]tree.Datums, map[string]int, map[int]roachpb.Key, error) {

[nit] it would help to give these return values some names

---

Comments from Reviewable

[nvanbenschoten]

The commit message says:

this check was necessary for correctness, but it was unfortunate because it rejects upserts that are valid in PostgreSQL.

What are these upserts that were valid in PG? It was my understanding after your recent email thread that Postgres disallows this too. The following corroborates that understanding:

nathan=# INSERT INTO kv(k) VALUES (1), (1) ON CONFLICT(k) DO UPDATE SET v = 2
nathan-# ;
ERROR:  ON CONFLICT DO UPDATE command cannot affect row a second time
HINT:  Ensure that no rows proposed for insertion within the same command have duplicate constrained values.

We're not planning on cherry-picking this into 2.0, are we?

---

Reviewed 1 of 3 files at r1.
Review status: 1 of 3 files reviewed at latest revision, 11 unresolved discussions, all commit checks successful.

---

pkg/sql/tablewriter.go, line 656 at r1 (raw file):

}

// upsertRowPKs returns the primary keys of every row in tu.insertRows

Not all your words, but we're pluralizing primary key in a few places here, which makes this confusing to read. Please singularize, like returns the primary key of every row.

---

pkg/sql/tablewriter.go, line 659 at r1 (raw file):

// for which we might get some value from KV.
//
// - if the conflicting index is the PK, the primary key for every

I'd revert the move from rows with potential upsert conflicts to [rows] which we might get some value from KV. The former is a lot clearer as it references intent instead of an implementation detail.

---

pkg/sql/tablewriter.go, line 691 at r1 (raw file):

				return nil, nil, err
			}
			if _, ok := uniquePKs[string(upsertRowPK)]; !ok {

Add a comment here.

---

pkg/sql/tablewriter.go, line 749 at r1 (raw file):

return value returned

This is redundant. Also, you should name these return values and then refer to them by name in this comment.

EDIT: just noticed Radu's comment below.

---

Comments from Reviewable

[knz]

What are these upserts that were valid in PG? It was my understanding after your recent email thread that Postgres disallows this too.

Indeed. I wrote the commit message before I understood that pg also rejects this. Thanks for spotting it.

I will respond to your other comments and Radu's after I simplify the code as requested.

---

Review status: 1 of 3 files reviewed at latest revision, 11 unresolved discussions, all commit checks successful.

---

Comments from Reviewable

[knz]

I have squashed the code and clarified the code further by:

• splitting it into multiple smaller, bite-sized functions with proper explanatory comments,
• making the naming of the variables clearer and more consistent.

RFAL.

To the other question:

We're not planning on cherry-picking this into 2.0, are we?

This discussion has been punted until after this PR merges (if/when it does).

---

Review status: 1 of 3 files reviewed at latest revision, 11 unresolved discussions.

---

pkg/sql/tablewriter.go, line 656 at r1 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

Not all your words, but we're pluralizing primary key in a few places here, which makes this confusing to read. Please singularize, like returns the primary key of every row.

Done.

---

pkg/sql/tablewriter.go, line 659 at r1 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

I'd revert the move from rows with potential upsert conflicts to [rows] which we might get some value from KV. The former is a lot clearer as it references intent instead of an implementation detail.

Done.

---

pkg/sql/tablewriter.go, line 691 at r1 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

Add a comment here.

Done.

---

pkg/sql/tablewriter.go, line 749 at r1 (raw file):

Previously, nvanbenschoten (Nathan VanBenschoten) wrote…

return value returned

This is redundant. Also, you should name these return values and then refer to them by name in this comment.

EDIT: just noticed Radu's comment below.

Done.

---

pkg/sql/tablewriter.go, line 294 at r2 (raw file):

Previously, RaduBerinde wrote…

[nit] indices

Done.

---

pkg/sql/tablewriter.go, line 408 at r2 (raw file):

Previously, RaduBerinde wrote…

This 200+ function is very hard to follow, can you think of a way to break things up a little?

Absolutely! Done, also made the naming of variables more consistent.

---

pkg/sql/tablewriter.go, line 528 at r2 (raw file):

Previously, RaduBerinde wrote…

excerpt

Done.

---

pkg/sql/tablewriter.go, line 600 at r2 (raw file):

Previously, RaduBerinde wrote…

[nit] Using the newExistingRowIdx variable is confusing in this block, maybe just set to len(existingRows) directly

Done.

---

pkg/sql/tablewriter.go, line 603 at r2 (raw file):

Previously, RaduBerinde wrote…

[nit] usually having the ok case first is easier to read

Done.

---

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, RaduBerinde wrote…

This code is pretty hard to understand. We are (to some extent) using the map as a proxy for comparing upsertRowPK with newUpsertRowPK. Could we just compare them directly?

I guess we could, but there's no simple operator (that I know of) to compare two byte slices in Go. Want to make a suggestion?

---

pkg/sql/tablewriter.go, line 761 at r2 (raw file):

Previously, RaduBerinde wrote…

[nit] it would help to give these return values some names

Done.

---

Comments from Reviewable

[knz]

Friendly request to look at this this week, so that I can enjoy my vacation without fearing some merge conflicts during a rebase after I come back.

[RaduBerinde]

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, knz (kena) wrote…

I guess we could, but there's no simple operator (that I know of) to compare two byte slices in Go. Want to make a suggestion?

bytes.Compare?

---

Comments from Reviewable

[RaduBerinde]

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, RaduBerinde wrote…

bytes.Compare?

even better bytes.Equal

---

Comments from Reviewable

[knz]

Review status: 0 of 3 files reviewed at latest revision, 5 unresolved discussions, all commit checks successful.

---

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, RaduBerinde wrote…

even better bytes.Equal

TIL. Thanks! I didn't know about the bytes package.

Now, that said, let's reconsider in context. The pkToRowIdx map must be checked first anyway to determine whether we've seen a row with that new pk before or not.

Once we've checked the map, we have an index. We also have the index from the row we're inserting. I agree that these two indexes are a "proxy" for the PK, but since they are there anyway, isn't it preferable for efficiency to just compare that? These two vars have at most 4-8 bytes, whereas a PK can be arbitrarily long.

---

Comments from Reviewable

[RaduBerinde]

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, knz (kena) wrote…

TIL. Thanks! I didn't know about the bytes package.

Now, that said, let's reconsider in context. The pkToRowIdx map must be checked first anyway to determine whether we've seen a row with that new pk before or not.

Once we've checked the map, we have an index. We also have the index from the row we're inserting. I agree that these two indexes are a "proxy" for the PK, but since they are there anyway, isn't it preferable for efficiency to just compare that? These two vars have at most 4-8 bytes, whereas a PK can be arbitrarily long.

Sure. Still, I would separate the deletion of the old PK from the two cases of updating existingRows. For example, have a pkChanged bool and set that in each case, and then consolidate the delete call into a if pkChanged {} block afterwards. Or move the assignment out of the if and have a separate if !(ok && updatedConflictingRowIdx != conflictingRowIdx) { delete() }. The if can have a comment that logically we are actually checking if updatedConflictingRowPK differs from conflictingRowPK.

---

Comments from Reviewable

[RaduBerinde]

Thanks for the update! Just some nits; still need to make a more in-depth pass.

---

Review status: 0 of 3 files reviewed at latest revision, 5 unresolved discussions, all commit checks successful.

---

pkg/sql/tablewriter.go, line 443 at r3 (raw file):

		// At this point, conflictingRowPK is either:
		// - nil if a secondary index was used and determined there is no conflict already;

we determined or it was determined

---

pkg/sql/tablewriter.go, line 447 at r3 (raw file):

		//   we must consult pkToRowIdx to determine whether we already
		//   have data (i.e. a conflict) in existingRows.
		//

nit: blank line

---

pkg/sql/tablewriter.go, line 470 at r3 (raw file):

				return nil, err
			}
		} else if len(tu.updateCols) == 0 {

[nit] This combo of elses and continue is confusing. I would separate this out and check for this first, just to get it out of the way (even if the condition becomes if len() == 0 && conflictingRowIdx != -1. Or at least have a single else { } block and move this if inside that.

---

pkg/sql/tablewriter.go, line 484 at r3 (raw file):

			// of SELECT. Only rows that were successfully inserted or
			// updated will be returned.
			// ""

[nit] """? you could also just indent the quote two spaces

---

Comments from Reviewable

[knz]

Review status: 0 of 3 files reviewed at latest revision, 9 unresolved discussions.

---

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, RaduBerinde wrote…

Sure. Still, I would separate the deletion of the old PK from the two cases of updating existingRows. For example, have a pkChanged bool and set that in each case, and then consolidate the delete call into a if pkChanged {} block afterwards. Or move the assignment out of the if and have a separate if !(ok && updatedConflictingRowIdx != conflictingRowIdx) { delete() }. The if can have a comment that logically we are actually checking if updatedConflictingRowPK differs from conflictingRowPK.

Oooh now I understand what you mean. Thank you!
Done.

---

pkg/sql/tablewriter.go, line 443 at r3 (raw file):

Previously, RaduBerinde wrote…

we determined or it was determined

Done.

---

pkg/sql/tablewriter.go, line 447 at r3 (raw file):

Previously, RaduBerinde wrote…

nit: blank line

Done.

---

pkg/sql/tablewriter.go, line 470 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] This combo of elses and continue is confusing. I would separate this out and check for this first, just to get it out of the way (even if the condition becomes if len() == 0 && conflictingRowIdx != -1. Or at least have a single else { } block and move this if inside that.

Done.

---

pkg/sql/tablewriter.go, line 484 at r3 (raw file):

Previously, RaduBerinde wrote…

[nit] """? you could also just indent the quote two spaces

Done.

---

Comments from Reviewable

[RaduBerinde]

Thanks for the updates!

---

Review status: 0 of 3 files reviewed at latest revision, 5 unresolved discussions, all commit checks successful.

---

pkg/sql/tablewriter.go, line 606 at r2 (raw file):

Previously, knz (kena) wrote…

Oooh now I understand what you mean. Thank you!
Done.

Thanks, this is much easier to understand!

---

pkg/sql/tablewriter.go, line 419 at r4 (raw file):

	// At this point existingRows contains data for the conflicting
	// rows, and pkToRowIdx maps each PK in tu.insertRows to an entry in

"each PK" is not right is it? It's just those PKs that have conflicts

---

pkg/sql/tablewriter.go, line 660 at r4 (raw file):

		//    pkChanged = !bytes.Equal(updatedConflictingRowPK, conflictingRowPK)
		//
		// However we already know that the row indice in existingRows are different

indices

---

Comments from Reviewable

[knz]

Review status: 0 of 3 files reviewed at latest revision, 7 unresolved discussions, all commit checks successful.

---

pkg/sql/tablewriter.go, line 419 at r4 (raw file):

Previously, RaduBerinde wrote…

"each PK" is not right is it? It's just those PKs that have conflicts

Yes correct, fixed.

---

pkg/sql/tablewriter.go, line 660 at r4 (raw file):

Previously, RaduBerinde wrote…

indices

Done.

---

Comments from Reviewable

[knz]

Okay this is ready! Thank you for your insightful and astute remarks, recommendations and requests. I have learned a bunch here.