spanconfigsql{watcher,reconciler}: setup SQLWatcher to watch for pts updates

[adityamaru]

This change sets up a rangefeed on the system.pts_records table,
and adds this to the set of rangefeeds managed by the SQLWatcher. We
switch the unit emitted by the SQLWatcher to a union type called
SQLUpdate which captures either an update to a descriptor or to a
protected timestamp target. The zones and descriptor table rangefeeds
will continue to emit descriptor updates. The pts rangefeed will emit
descriptor updates for records that target schema objects, and pts target
updates for records that target cluster or tenants.

The SQLWatcher continues to dedup the SQLUpdates that it emits, so as to
ensure there is only one SQLUpdate per descriptor ID, and one SQLUpdate
per cluster or tenant target. The reconciler registers a handler that
is invoked everytime a batch of SQLUpdates are emitted by the SQLWatcher.
There is change in semantics in this part of the code, a future PR will teach
the reconciler to parse the pts target SQLUpdates, and in turn instruct the
SQLTranslator to generate the appropriate SystemSpanConfigs.

Note, this file moves the sqlwatcher tests into a ccl package so as to be
able to run backup statements to test the rangefeed on the pts table.

Informs: #73727

Release note: None

[cockroach-teamcity]

This change is 

[adityamaru]

@irfansharif @arulajmani this is an early draft of what I had in mind for how the SQLWatcher would watch for protected timestamp updates. Before I clean up the tests and tidy up the PR I wanted inputs on some of the changes to the SQLWatcher interface.

[adityamaru]

This is RFAL now!

[adityamaru]

@irfansharif wrapped the target in a ProtectedTimestampUpdate - https://github.com/cockroachdb/cockroach/pull/75122/files#diff-879b43c0345e0967f0b1ee30422ea1993314a45a3195cb2cff030a62791ff434R309

Instead of storing it as separate Cluster and Tenant fields, I went with an opaque ptpb.Target field that we sanity check in the constructor. The reason I went with this is it becomes cleaner to dedup these entries in the watcher, based on the ptpb.Target stringer. Let me know what you think.

[adityamaru]

@irfansharif changed to unwrap the Tenants_Target into individual SQLUpdates in the SQLWatcher. Also made the sort stable, since each SQLUpdate only has one tenant ID in it now.

[irfansharif]

LGTM, nice work! Left only nits.

[arulajmani]

I only have a few nits! Good stuff

Reviewed 1 of 13 files at r1, 10 of 12 files at r2, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @adityamaru, @ajwerner, and @irfansharif)

---

pkg/spanconfig/spanconfigsqlwatcher/buffer.go, line 109 at r2 (raw file):

// Less implements the sort.Interface methods.
func (s bufferedEvents) Less(i, j int) bool {

nit: do you think it's worth capturing the sort order here? I know you already have it inlined; your call.

---

pkg/spanconfig/spanconfigsqlwatcher/buffer.go, line 255 at r2 (raw file):

		// target ProtectedTimestampUpdates there is no deduplication possible.
		if prevUpdate.IsClusterUpdate() && curUpdate.IsTenantsUpdate() ||
			prevUpdate.IsTenantsUpdate() && curUpdate.IsClusterUpdate() {

Can we ever have prevUpdate.IsClusterUpdate() and curUpdate.IsTenantUpdate?

---

pkg/spanconfig/spanconfigsqlwatcher/buffer.go, line 266 at r2 (raw file):

		}

		// If the previous buffered event, and the current event are both tenant

nit: move this at the top of the loop considering we sort tenant targets before cluster targets?

---

pkg/spanconfig/spanconfigsqlwatcher/sqlwatcher.go, line 374 at r2 (raw file):

		case *ptpb.Target_SchemaObjects:
			// For PTS records with schema object targets, unwrap the descriptor IDs,
			// and emit them as descriptor SQLUpdates. This allows for the deduplication

Coming back to this after reading your tests for the buffer, hijacking PTS updates on descriptor IDs at this level worked out well 💯

---

pkg/spanconfig/spanconfigsqlwatcher/buffer_test.go, line 109 at r2 (raw file):

	buffer.advance(descriptorsRangefeed, ts(10))
	require.NoError(t, err)
	buffer.advance(protectedTimestampRangefeed, ts(10))

nit: here, and below, instead of moving descriptors/pts rangefeeds in lockstep you could make the test case more interesting by choosing different values and making sure the lowest of the 3 is the combined frontier timestamp.

[adityamaru]

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @adityamaru, @ajwerner, @arulajmani, and @irfansharif)

---

pkg/spanconfig/spanconfigsqlwatcher/buffer.go, line 109 at r2 (raw file):

Previously, arulajmani (Arul Ajmani) wrote…

nit: do you think it's worth capturing the sort order here? I know you already have it inlined; your call.

Heh, I did have it here for a second and then moved it back inline cause it felt more natural to highlight to the reader that this sort.Sort is actually more involved than meets the eye.

---

pkg/spanconfig/spanconfigsqlwatcher/buffer.go, line 255 at r2 (raw file):

Previously, arulajmani (Arul Ajmani) wrote…

Can we ever have prevUpdate.IsClusterUpdate() and curUpdate.IsTenantUpdate?

I refactored this to Irfan's suggestion above. You're right I don't think we could have in the older version of the code.

---

pkg/spanconfig/spanconfigsqlwatcher/buffer.go, line 266 at r2 (raw file):

Previously, arulajmani (Arul Ajmani) wrote…

nit: move this at the top of the loop considering we sort tenant targets before cluster targets?

refactored to irfan's suggestion but also moveed this clause higher.

---

pkg/spanconfig/spanconfigsqlwatcher/buffer_test.go, line 109 at r2 (raw file):

Previously, arulajmani (Arul Ajmani) wrote…

nit: here, and below, instead of moving descriptors/pts rangefeeds in lockstep you could make the test case more interesting by choosing different values and making sure the lowest of the 3 is the combined frontier timestamp.

staggered the timestamps as suggested.

[adityamaru]

Thanks for all the review on this!

bors r=irfansharif,arulajmani

[craig]

Build failed (retrying...):

• GitHub CI (Cockroach)

[otan]

[04:31:38][run] pkg/spanconfig/spanconfigsqlwatcher/sqlwatcher.go:406:21: not enough arguments in call to rangefeed.WithDiff
[04:31:38][run] 	have ()
[04:31:38][run] 	want (bool)

merge skew :(

(yay bors!!!)

[otan]

bors r-

[craig]

Canceled.

[adityamaru]

CI failure is #75960.

bors r+

[craig]

Build succeeded:

• GitHub CI (Cockroach)