More general types of Kafka source

[umanwizard]

Discussed this offline with @frankmcsherry and @quodlibetor ...

Currently we have a hard-coded assumption that our Kafka sources will be in Debezium format; i.e., a log of database modifications with before and after representing row deletions and additions.

It may also be useful to allow other models; for example, reading a topic that is an append-only event log might be broadly useful.

[rjnn]

This issue is now a higher priority than previously. Note that adding support for formats can still be somewhat opinionated - you can require that the CREATE SOURCE command enumerates the list of fields that will be populated, and that if a field is not present in a message it will be NULL (as long as that field is indicated as NULLABLE at source creation time, and error out otherwise), etc.

This doesn't mean we need full support for arbitrary gnarly nested JSON - but we should be able to ingest JSON messages that just have the bare columns as named fields, ingesting them as an append-only list of rows/

Finally, we want the ability to optionally designate one of them as the timestamp column.

[cuongdo]

@rjnn is this different from #207?

[rjnn]

@cuongdo Yes. One could support #207 by parsing and ingesting JSON but still require that that JSON field have the prescribed before: {}, after: {}, ts_ms: 1234 schema. To complete this issue we want to suppose people who have data in either avro or JSON but without that prescriptive before/after schema.

[frankmcsherry]

Related to #777