Internal sinks, part 1

[teskje]

This PR is the first step towards enabling internal sinks in Materialize. It includes various changes to the existing persist sinks, to make them work like the envisioned internal sinks.

The PR does not make significant changes to the UI, the CREATE SINK ... INTO PERSIST syntax remains (albeit in a simpler form). Replacing it with the new CREATE RECORDED VIEW syntax will be the subject of a follow-up PR. The intention of this split is to keep the amount of changes that need to be reviewed somewhat manageable.

Changes made in this PR (see the separate commits for more detail):

• SinkDesc receives a type parameter for storage metadata, akin to SourceInstanceDesc.
• The persist location parameters are removed from CREATE SINK ... INTO PERSIST. Instead, the persist location is provided by the storage controller now.
• When creating a persist sink, a source is registered with the storage controller. Right now this is only done to get a storage collection, but later we will use that source to read back sinked data.
• computed's persist_sink is made compatible to storaged's persist_source, so sinked data can be read back.
• The compute controller is extended to report updates of persist sink uppers to the storage controller, so the storage controller can track the frontiers of sink collections.

Usage

Sinking data now looks as follows from SQL:

CREATE TABLE table (a int);
CREATE SINK sink FROM table INTO PERSIST;
INSERT INTO table VALUES (1), (2), (2);
DELETE FROM table WHERE a != 1;

It is not currently possible to read back sinked data from SQL. Instead, you can write a simple tail command using the mz-persist-client crate directly, or clone https://github.com/teskje/persist-test.

When you crate a persist sink, computed prints the shard ID in an INFO trace:

[...] INFO mz_compute::sink::persist_sink: persist_sink shard ID: s00e6f38f-ccfc-4471-bb8c-49718abccc19

You can plug that into your tail command, e.g.:

$ cargo run --bin tail -- 'postgres://jan@%2Ftmp?options=--search_path=consensus' 'file:///Users/jan/devel/materialize/mzdata/persist/blob' 's00e6f38f-ccfc-4471-bb8c-49718abccc19'
[...]
(2), 1656079696243, -2
(1), 1656079696243, 1
(2), 1656079696243, 2

(Here, inserts and deletes appear to happen at the same time because the since of the persist shard has already advanced beyond the times of both events. If you want to see them happen in real-time, attach the tail first, before inserting/deleting from the table.)

Caveats

Persist sinks should behave correctly on replicated compute instances and when compute replicas are killed and restarted. They currently don't survive restarts of environmentd: The sink gets recreated but gets a new persist shard ID, so data is written to a new storage collection instead of the existing one. This happens because the storage controller currently doesn't persist shard ID.

Motivation

• This PR adds a known-desirable feature.

  Part of https://github.com/MaterializeInc/database-issues/issues/3692.

Tips for reviewer

I would suggest looking at the individual commits separately.

Testing

• This PR has adequate test coverage / QA involvement has been duly considered.

Release notes

This PR includes no user-facing behavior changes.

[benesch]

Nice! I love these sorts of incremental steps.

[antiguru]

This looks great, thanks!

[aljoscha]

I'll approve retroactively, since I didn't get to this in time. 👍

[teskje]

@aljoscha Thanks! Sorry, I was too eager with the merge trigger!

[aljoscha]

Not at all, enough people had reviewed it by then! 👍