Materialize: kafka sources: support inserting the kafka key as a data field in dataflows

[quodlibetor]

Folks put legitimate information in Kafka's Key and Published Timestamp fields, and currently materialized has no way to extract it into the system. The current workaround is to use Kafka Streams to insert key/timestamp data into the value part, but that is unfortunate.

Other than the obvious question of what syntax should we use for this, we also need to answer:

• What happens on column name collision (keys and values can both be arbitrary data)
• Is it better to create a single relation that has all the data, or to create two that can be joined?
• Probably whatever we do should support including the cross-product of all (key, value, timestamp) combinations in any given table

[cirego]

• Probably whatever we do should support including the cross-product of all (key, value, timestamp) combinations in any given table

I'm under the impression that (timestamp, key, value) is a tuple that represents a single event. If we split these into different tables, wouldn't the join key be event_id or message_id? I'm not sure how one would generate a cross-product of (key, value, timestamp)?

[quodlibetor]

Ah, I meant you should be able to create a relation that includes any of the combinations, so it should be possible to express any one of the following combinations:

key, value, timestamp
key, value
key, timestamp
value, timestamp
key
value
timestamp

[umanwizard]

I'm imagining (post source orthogonality) that bare Kafka sources will emit "(key, value, timestamp, offset)" records, and then the current avro/debezium/upsert/etc. sources we create will do their own extraction of the correct fields. But if someone needs access to the key they will be able to get it (unlike now, where only Upsert sources look at the key)

[umanwizard]

Ah, I meant you should be able to create a relation that includes any of the combinations

You can do this regardless of whether the metadata lives in a separate relation or not. Even if everything is in the same relation, you can do it through filters.

[bbigras]

Does the Kafka sink support sending a key and a timestamp? The key seems supported https://materialize.com/docs/sql/create-sink/#kafka-connector

[krishmanoh2]

We had a prospect ask about this today.

[umanwizard]

@quodlibetor and I definitely want to do this, but there isn't a concrete timeline yet. I'll keep you posted.

[umanwizard]

Design doc: https://github.com/MaterializeInc/materialize/pull/6661

[quodlibetor]

I'm currently working on this, I've renamed the issue and split off the timestamp &c fields into their own issue since I'm not planning on tackling that as part of the current work.