Populate envelope/key-format/value-format columns on mz_kafka_source_tables table

[rjobanp]

Motivation

• This PR fixes a recognized bug. Fixes https://github.com/MaterializeInc/database-issues/issues/8613

Now that each individual export (table) of a Kafka source can have it's own envelope and/or encoding specified, these columns don't make much sense on the mz_sources catalog table. Instead, we decided to port them to the mz_kafka_source_tables catalog table -- see discussion in the referenced issue.

Tips for reviewer

The large line diff in src/catalog/src/memory/objects.rs is just moving two methods from the Source type to the DataSourceDesc type for re-use

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).
• If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.

[jkosh44]

Code LGTM but I had two questions/comments

• What's happening with these columns in mz_sources are we keeping them but always filling them in with NULL?
• Can you update the docs for mz_kafka_source_tables.

[rjobanp]

What's happening with these columns in mz_sources are we keeping them but always filling them in with NULL?

Yes for now we will - I've put a task in the epic to remove those columns once we're fully migrated to the new model https://github.com/MaterializeInc/database-issues/issues/8677

Can you update the docs for mz_kafka_source_tables

Yes!

[jkosh44]

What's happening with these columns in mz_sources are we keeping them but always filling them in with NULL?

Yes for now we will - I've put a task in the epic to remove those columns once we're fully migrated to the new model https://github.com/MaterializeInc/database-issues/issues/8677

Ah OK, we may want to hard code that into pack_source_update in builtin_table_updates.rs then instead of passing in an Option.

[jkosh44]

LGTM

[rjobanp]

Ah OK, we may want to hard code that into pack_source_update in builtin_table_updates.rs then instead of passing in an Option.

Ah - to clarify - these columns will still be non-null for existing kafka sources that define the envelope and encoding directly on the main source object. However for 'new model' kafka sources, they will be null and the tables will have these values defined. Once we've migrated all sources to the new model then those columns can be removed

[jkosh44]

Ah OK, we may want to hard code that into pack_source_update in builtin_table_updates.rs then instead of passing in an Option.

Ah - to clarify - these columns will still be non-null for existing kafka sources that define the envelope and encoding directly on the main source object. However for 'new model' kafka sources, they will be null and the tables will have these values defined. Once we've migrated all sources to the new model then those columns can be removed

Ok, I understand now. In that case it's good as is.