design: deprecate concept of SUBSOURCE

[sploiselle]

Motivation

Proposes a design for MaterializeInc/database-issues#6051

[sploiselle]

@bkirwi sagely asks, "Why TABLE instead of some other noun in MZ?" My answer is just that TABLEs are a well understood concept, and these tables are vaguely like other tables in MZ (except they can't be written to). Maybe MATERIALIZED VIEW is the best analog (persisted, cannot be written to), but the idea that you'd type CREATE MATERIALIZED VIEW foo FROM SOURCE (referenced_object) to add these new object feels clunky to me.

[sjwiesman]

Fwiw, I like table as an abstraction and hope someday we even allow writing directly to this tables along side sources.

Wanted to jot down some thoughts from IRL conversations of ways this is useful beyond user understanding. You mention being able to create a second table from a Kafka source to enable schema migrations. This would also unlock a form of truncation for append only sources. With just the features in this design doc we would rely on Kafka compaction to delete rows, but one could imagine extending the syntax to support specifying the starting offset behavior.

CREATE TABLE truncated FROM SOURCE  kafka_src START OFFSET (0, 10, 100);

CREATE TABLE truncated FROM SOURCE  kafka_src START TIMESTAMP -10000;

We can extend this logic to webhook sources, assuming the endpoint is tied to the source object. Dropping and recreating webhook sources is difficult today because events can't easily be replayed. Running two webhook sources in parallel is tricky because the upstream system has to be configured to point at both or a proxy put in place. But with a source writing to two tables, users have the ability to seed the new shard for until it contains enough data to satisfy downstream temporal filters and then seamlessly blue/green consuming views.

[sjwiesman]

It just occurred to me this could also open the door to solving MaterializeInc/database-issues#6771 for sources

Writing down errors could be a configuration on the table. Or you could have a table that only contains errrors.

CREATE TABLE t FROM kafka_srcs IGNORE ERRORS;

CREATE TABLE dlq FROM kafka_srcs ERRORS ONLY;

[sploiselle]

Seth's idea seems like it could propose a novel approach to MaterializeInc/database-issues#6771 for sources.

[benesch]

It just occurred to me this could also open the door to solving https://github.com/MaterializeInc/database-issues/issues/6771 for sources

Whether we emit or suppress decoding errors seems orthogonal to me to whether we call these things subsources or tables! We've had https://github.com/MaterializeInc/database-issues/issues/4040 for a while, which proposes adding various types of WITH (IGNORE <BLAH> ERRORS = TRUE) options to sources.

[sploiselle]

Whether we emit or suppress decoding errors seems orthogonal to me to whether we call these things subsources or tables!

The design doc included in this PR is not just a discussion of whether ingestions export data to a noun called subsources or a tables. It deals substantially in a design that move progress collection data in "the source itself" and Kafka sources growing ingestion exports (irrespective of whether they're called subsources or tables).

I think there's a graceful implementation of Seth's idea that requires Kafka sources to have multiple ingestion exports, so it seems germane to mention here.

edit: the idea that I'm supportive of here is having the ability to split Ok and Err into separate collections, which seems like it's greatly simplified if you have can have multiple outputs from all sources. That's not currently possible with Kafka, which only gives you one output. However, after leaving this comment, I realize you could hack together the same effect by creating a kind of bizarro derivative subsource on a Kafka source.

[benesch]

Suggested change

	subsources," which is a concept that users often find confusion.
	subsources," which is a concept that users often find confusing.

[benesch]

I think this is implied, but might be worth explicitly stating that we're also going to remove subsource from any system catalog table names or rows. (Since these aren't quite subsource keywords.)

[benesch]

I think it depends the interface! I'd suggest we use "table" for anything programmatic—i.e., these tables should be in mz_tables, be labeled as objects in mz_objects, that sort of thing. But then the commands for humans (SHOW TABLES,SHOW OBJECTS) probably want to very prominently display read-only vs not. And the console of course can use a prominent badge or tooltip or whatever makes sense.

[morsapaes]

+1

[benesch]

Do we have any users depending on these sources today? If not ... maybe we can just rip it out?

This seems like a piece of the migration we could do incrementally. E.g., just ship a change that prevents users from depending on sources directly.

[sploiselle]

I don't know if any users depend on them today but this seems like the simplest and most fool-proof approach. I'm not sure what exactly you mean by preventing users from depending on sources directly, though––can you elaborate?

[benesch]

I don't follow this, but it doesn't seem quite right to me! That or I'm not understanding correctly. We need to make sure that after shipping the breaking change to Materialize, running terraform apply against an existing project with no changes does not cause sources to be recreated, and similarly with the dbt adapter. In the worst case we can write a migration guide detailing changes that must be apply to the code in Terraform and dbt projects, but that's a pretty disruptive thing to force on users, since they won't have any control over when the breaking change to Materialize goes out.

[morsapaes]

I don't think the dbt workflow is going to be as clean as we'd hope, but...captured my current understanding of how this comes together in this gist. I haven't actually tried to use model versioning, so this is all off the top of my head. Roping in @jtcohen6 for a sanity check! 🫠

[benesch]

Suggested change

	- ID, persist hard becomes `SOURCE`
	- ID, persist shard becomes `SOURCE`

[benesch]

This seems clean from the perspective of downstream views, but will be quite disruptive to the Terraform and dbt adapters. Left a comment along these lines below. We should see if there's a way to have the Terraform and dbt adapters automatically adapt to the new setup.

[benesch]

This is really exciting, Sean! Thanks for writing this up.

[benesch]

The idea that I'm supportive of here is having the ability to split Ok and Err into separate collections, which seems like it's greatly simplified if you have can have multiple outputs from all sources. That's not currently possible with Kafka, which only gives you one output. However, after leaving this comment, I realize you could hack together the same effect by creating a kind of bizarro derivative subsource on a Kafka source.

That's just a performance optimization though, right? persist_source already splits the shard into Ok and Err streams, right?

I might be misunderstanding, but I really feel like we already have all the pieces that we need to support https://github.com/MaterializeInc/database-issues/issues/6771, and we don't need to expand the scope of this migration by considering it in the design!

For example, one proposal for MaterializeInc/database-issues#6771 is to add a special $errors syntax, so that e.g. foo$errors returns the errors stream. AFAICT, that design would be totally orthogonal to the subsources-as-tables design!

[sploiselle]

That's just a performance optimization though, right? persist_source already splits the shard into Ok and Err streams, right?

Yeah; maybe the opportunity for a semantic optimization moreso than performance. My point is that if we had multiple outputs from a Kafka source, we could easily split Ok and Err into separate collections. This would give users the option to never encounter definite errors when querying a topic/table/ingestion export, while still tracking what all of those errors were elsewhere (maybe a user cares about this for monitoring).

However, it sounds like you have a clear and distinct idea for MaterializeInc/database-issues#6771, though, that is not this. Not a big deal, just wanted to defend that the common thread Seth and I saw was not orthogonal to this design.

[morsapaes]

Suggested change

	`kafka_src` is again the progress relation for the source. The data is in a
	table named `my_t`. You would never be able to specify more than one table in
	the original `CREATE SOURCE` statement because you would have to two outputs
	from the same source with the same schema, which we should disallow.
	`kafka_src` is again the progress relation for the source. The data is in a
	table named `my_t`. You would never be able to specify more than one table in
	the original `CREATE SOURCE` statement because you would have two outputs
	from the same source with the same schema, which we should disallow.

[morsapaes]

I don't think the dbt workflow is going to be as clean as we'd hope, but...captured my current understanding of how this comes together in this gist. I haven't actually tried to use model versioning, so this is all off the top of my head. Roping in @jtcohen6 for a sanity check! 🫠

[morsapaes]

+1

[morsapaes]

From a chat with @benesch: does this imply that there would be a FOR NO TABLES (or similar) option that only creates the source, but suppresses automatic table creation? This would be useful so we can decouple source management and table creation in dbt.

[benesch]

Also curious if that feature already exists today as FOR TABLES ().

[sploiselle]

Today, you'd get errors for not having created any objects in CREATE SOURCE. However, we could introduce support for FOR TABLES (). I dont know exactly what would happen w/r/t frontier management w/ PG and MySQL sources, but it's something we could figure out.

[benesch]

That's just a performance optimization though, right? persist_source already splits the shard into Ok and Err streams, right?

Yeah; maybe the opportunity for a semantic optimization moreso than performance. My point is that if we had multiple outputs from a Kafka source, we could easily split Ok and Err into separate collections. This would give users the option to never encounter definite errors when querying a topic/table/ingestion export, while still tracking what all of those errors were elsewhere (maybe a user cares about this for monitoring).

However, it sounds like you have a clear and distinct idea for MaterializeInc/database-issues#6771, though, that is not this. Not a big deal, just wanted to defend that the common thread Seth and I saw was not orthogonal to this design.

Ah, yeah, sorry, we're just thinking about this differently! I wouldn't want to do something special for subsource-tables, because views and matviews have exactly the same underlying oks and errs collections.

Here's how it shakes out in my head:

         source             view       matview   
        /       \           /   \      /   \  
      table_a   table_b    oks errs   oks errs
      /   \      /  \    
     oks errs  oks errs  

Vs what I think you and Seth were describing for multi-output sources:

            source ----------------+
           /   |    \               \
          /    |     \               \
         /     |      \               \
        /      |       \               \
       /       |        \               \
table_a_oks table_a_errs  table_b_oks  table_b_errs

Which makes total sense when considering multi-output sources in isolation! But I don't immediately see how we'd generalize this design to views or matviews.

[sploiselle]

@benesch Yeah, totally! I was not suggesting that this is the way but just that it was something we could do. Agreed that the lack of generalization here isn't as ideal as what you've envisioned.

[benesch]

Small note: purification is also (correctly, IMO) used for work that should only be performed once, when the user types the SQL into Materialize, vs on every boot, when the SQL is loaded from the catalog. That sort of work is not always asynchronous.

[sploiselle]

Ah, sure––that's a good call. The issue I have is moreso where the code lives vs. when it is called. Will clarify.

[sploiselle]

What if we split up purification into two stages to make this more explicit? async, and then normalization (or whatever other one-off work exists).

The annoyance here is that there is one-off work which we currently do in purification which very closely resembles other work we have to do post-purification which depends on knowing the catalog state when we rejoin the main coord thread (e.g. we have to know all of a source's subsources and cannot assume that the set available to us in purification is the set that exists when we rejoin the main coord thread).

I guess the reason that I'm harping on this which I haven't clearly articulated is that there is a world in which creating a source and altering a source can use the exact same code paths w/r/t managing the subsources that it offers. I'll amend the design doc to split these things out more neatly and clearly.

[sploiselle]

Here's what I've had in the back of my mind. In case it moves it's in the section evocatively titled "Mind Palace."

[benesch]

What if we split up purification into two stages to make this more explicit? async, and then normalization (or whatever other one-off work exists).

That makes sense to me!

[rjobanp]

Can you describe what this would look like for CREATE TABLE? Would it just be

CREATE TABLE my_new_table FROM SOURCE postgres_src (upstream_schema.upstream_table) WITH (TEXT COLUMNS (upstream_table.column_1), IGNORE COLUMNS (upstream_table.column_2))

?

[sploiselle]

Yes. They will be the same as the ADD SUBSOURCE options but as options on CREATE TABLE. This is what I meant by move.

[rjobanp]

Why? I didn't see in the above doc why this is necessarily different from kafka sources today, where we also need to know their schema.

[sploiselle]

CreateSubsourceStatement requires knowing the schema of the object.

[rjobanp]

Sure but isn't that also true today? I'm perhaps misunderstanding your phrasing here -- how does that constraint block the implementation of #20208?

[sploiselle]

maybe this phrasing is clearer?

we have to know the schema of the Kafka topic when creating the subsource/table that it will write to.

[rjobanp]

so the issue here is that we have to do similar things in both purification and in sequencing based on whether we are creating a new source vs adding a subsource/table to an existing source?

[sploiselle]

Yes

[rjobanp]

what about the 'genericizing' of subsource planning? Is that part of the success criteria or just a side-effect of this work?

[sploiselle]

To generate the CREATE SUBSOURCE statements after planning the source, I intend to genericize subsource planning.

[rjobanp]

this would make things so much clearer!

[rjobanp]

Suggested change

	`CreateSourceStatement`s using the name, reference mapping; we'll get the
	`CreateSubsourceStatement`s using the name, reference mapping; we'll get the

?

[sploiselle]

Yes, ty!

[rjobanp]

😏

[rjobanp]

I like this idea a lot - will definitely simplify the work in adding new source types going forward and understanding how they work

[sploiselle]

Mentioned elsewhere but I added a "Mind Palace" section that we might be able to just write the initial impls of a source in terms of this, and then every new source supports ALTER SOURCE...ADD SUBSOURCE by default.

[rjobanp]

yes that context was helpful, thanks!

[rjobanp]

How do we validate the options like TEXT COLUMNS on the top-level source plan before we've requested the schema from the relevant subsources? Or will we start allowing invalid references in those options in the top-level source statement and throw an error when we plan all the subsources?

[sploiselle]

TEXT COLUMNS and IGNORE COLUMNS only need to validate that they refer to valid columns in the source's "subsource catalog." That should just be done as part of normalizing them––i.e. when we figure out which object they refer to, we should write down that fully qualified name, and if we can't find it, we should error.

[sploiselle]

idk if this makes things any clearer, but I added a section around some of the unification I'm hoping to achieve between creating and modifying sources. This might just end up being a conceptual, rather than concrete, framework for me. idk how gnarly the typing will get.

[rjobanp]

I also agree the gist seems sound but it could get tricky in the implementation! I would vote for a first pass time-boxed attempt at this to see if you can get it to work

[rjobanp]

responses / updates LGTM on the subsource planning doc!

[morsapaes]

This will have a pretty significant impact in the console, so it'd be good to get @RobinClowers' 👀. More broadly, could you expand on how this affects existing core system catalog tables like mz_sources, mz_source_status_history, or mz_source_statistics?

[RobinClowers]

Thanks for looping me in Marta, I have the same questions about status and statistics, but the change outlined here is pretty straightforward for the console.

[guswynn]

Signing off from my end!

One side thing I want to note is that I plan at some point to split mz_source_statistics into mz_source_statistics and mz_subsource_statistics, as sources and subsources produce different sets of statistics. I suppose after this change, the latter would be something like mz_ingested_table_statistics`?

[sploiselle]

At @bosconi 's request I've moved these to Notion.

• Deprecate subsources
• Generate subsources after planning sources