feat(sink): `auto.create` for automatically creating sink tables/topics

xiangjinwu commented 10 months ago

Is your feature request related to a problem? Please describe.

https://docs.confluent.io/platform/current/schema-registry/schema_registry_onprem_tutorial.html#auto-schema-registration

This is very convenient in development environments, but in production environments we recommend that client applications do not automatically register new schemas. Best practice is to register schemas outside of the client application to control when schemas are registered with Schema Registry and how they evolve.

Even for a simple table create table t (k int, val varchar); The avro definition could be:

{
  "type": "record",
  "name": "t",
  "fields": [
    {"name": "k", "type": ["null", "int"]},
    {"name": "v", "type": ["null", "string"]}
  ]
}

Inputting this manually is cumbersome and error-prone.

Task items

[ ] auto.create for Kafka Protobuf/Avro sink https://github.com/risingwavelabs/risingwave/issues/13139
~~[ ] https://github.com/risingwavelabs/risingwave/issues/14254~~ auto.create for Iceberg sink is unplanned because there are sophisticated options in Iceberg schemas.
[ ] auto.create for Postgres sink
[ ] auto.create for MySQL sink
[ ] auto.create for Clickhouse sink
[ ] auto.create for Doris sink
[x] #15586

Describe the solution you'd like

https://github.com/risingwavelabs/risingwave/blob/main/e2e_test/sink/kafka/avro.slt

create sink sink0 from into_kafka with (
  connector = 'kafka',
  topic = 'test-rw-sink-upsert-avro',
  properties.bootstrap.server = 'message_queue:29092',
  primary_key = 'int32_field,string_field')
format upsert encode avro (
  schema.registry = 'http://message_queue:8081',
  auto.register.schemas = true);

Describe alternatives you've considered

Provide a variant of DESCRIBE (or some other dedicated command) to generate the definition for a relation:

dev=> GENERATE SINK SCHEMA ENCODE AVRO into_kafka;
{
"type": "record",
"name": "into_kafka",
"fields": [
{"name": "bool_field", "type": ["null", "boolean"]},
...
{"name": "time_millis_field", "type": ["null", {"type": "int", "logicalType": "time-millis"}]}
]
}

Leverage existing avro schema editing tools.

Additional context

Limitations of an auto generated definition:

Everything will be considered nullable.
The same struct may be defined multiple times.

neverchanje commented 10 months ago

Once we can show the sink schema via a sql command, we can then implement a python script that auto registers the schema registry. It doesn't have to be implemented in the kernel, unless we see a similar requirements.

hzxa21 commented 10 months ago

Once we can show the sink schema via a sql command, we can then implement a python script that auto registers the schema registry. It doesn't have to be implemented in the kernel, unless we see a similar requirements.

Hmm... Where and who will run the python script? If it is the DB admin, not the SQL user and the script is not run in SQL, I think it is still a development burden.

neverchanje commented 9 months ago

Can be anyone who has the privilege to read table columns and register schema to Schema Registry.

neverchanje commented 9 months ago

As discussed previously with Patrick, we agree that auto-registering schemas is a valid requirement. Almost all commercial Extract&Load systems like Fivetran will create downstream tables automatically. https://github.com/risingwavelabs/risingwave/issues/13718

For the user-facing option name, let's align it to auto.create. We'll support this option for other sinks as well in the future.

risingwavelabs / risingwave