-
Notifications
You must be signed in to change notification settings - Fork 1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
fix: support partial schemas (#4625)
* fix: support partial schemas fixes: #4566 With this change users can now supply just the key schema and use schema inference to get the value columns. For example, if the key is an `INT` serialized using Kafka's `IntegerSerializer` and the value is an Avro record with the schema stored in the Scheme Registry, then such a stream can be registered in ksqlDB with a statement such as: ```sql -- note: only the key columns are provided between the first set of brackets -- the value columns will be inferred from the Avro schema in the Schema Registry CREATE STREAM users (ROWKEY INT KET) WITH (kafka_topic='users', value_format='avro'); ```
- Loading branch information
1 parent
f6a5469
commit 4f1ce8a
Showing
16 changed files
with
879 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
144 changes: 144 additions & 0 deletions
144
..._plans/elements_-_validate_without_value_elements_OK_-_AVRO/6.0.0_1582711670275/plan.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,144 @@ | ||
| { | ||
| "plan" : [ { | ||
| "@type" : "ksqlPlanV1", | ||
| "statementText" : "CREATE STREAM INPUT (ROWKEY INTEGER KEY, C1 INTEGER) WITH (AVRO_SCHEMA_ID=1, KAFKA_TOPIC='input', VALUE_FORMAT='AvRo');", | ||
| "ddlCommand" : { | ||
| "@type" : "createStreamV1", | ||
| "sourceName" : "INPUT", | ||
| "schema" : "`ROWKEY` INTEGER KEY, `C1` INTEGER", | ||
| "keyField" : null, | ||
| "timestampColumn" : null, | ||
| "topicName" : "input", | ||
| "formats" : { | ||
| "keyFormat" : { | ||
| "format" : "KAFKA", | ||
| "properties" : { } | ||
| }, | ||
| "valueFormat" : { | ||
| "format" : "AVRO", | ||
| "properties" : { } | ||
| }, | ||
| "options" : [ ] | ||
| }, | ||
| "windowInfo" : null | ||
| }, | ||
| "queryPlan" : null | ||
| }, { | ||
| "@type" : "ksqlPlanV1", | ||
| "statementText" : "CREATE STREAM OUTPUT AS SELECT *\nFROM INPUT INPUT\nEMIT CHANGES", | ||
| "ddlCommand" : { | ||
| "@type" : "createStreamV1", | ||
| "sourceName" : "OUTPUT", | ||
| "schema" : "`ROWKEY` INTEGER KEY, `C1` INTEGER", | ||
| "keyField" : null, | ||
| "timestampColumn" : null, | ||
| "topicName" : "OUTPUT", | ||
| "formats" : { | ||
| "keyFormat" : { | ||
| "format" : "KAFKA", | ||
| "properties" : { } | ||
| }, | ||
| "valueFormat" : { | ||
| "format" : "AVRO", | ||
| "properties" : { } | ||
| }, | ||
| "options" : [ ] | ||
| }, | ||
| "windowInfo" : null | ||
| }, | ||
| "queryPlan" : { | ||
| "sources" : [ "INPUT" ], | ||
| "sink" : "OUTPUT", | ||
| "physicalPlan" : { | ||
| "@type" : "streamSinkV1", | ||
| "properties" : { | ||
| "queryContext" : "OUTPUT" | ||
| }, | ||
| "source" : { | ||
| "@type" : "streamSelectV1", | ||
| "properties" : { | ||
| "queryContext" : "Project" | ||
| }, | ||
| "source" : { | ||
| "@type" : "streamSourceV1", | ||
| "properties" : { | ||
| "queryContext" : "KsqlTopic/Source" | ||
| }, | ||
| "topicName" : "input", | ||
| "formats" : { | ||
| "keyFormat" : { | ||
| "format" : "KAFKA", | ||
| "properties" : { } | ||
| }, | ||
| "valueFormat" : { | ||
| "format" : "AVRO", | ||
| "properties" : { } | ||
| }, | ||
| "options" : [ ] | ||
| }, | ||
| "timestampColumn" : null, | ||
| "sourceSchema" : "`ROWKEY` INTEGER KEY, `C1` INTEGER" | ||
| }, | ||
| "selectExpressions" : [ "C1 AS C1" ] | ||
| }, | ||
| "formats" : { | ||
| "keyFormat" : { | ||
| "format" : "KAFKA", | ||
| "properties" : { } | ||
| }, | ||
| "valueFormat" : { | ||
| "format" : "AVRO", | ||
| "properties" : { } | ||
| }, | ||
| "options" : [ ] | ||
| }, | ||
| "topicName" : "OUTPUT" | ||
| }, | ||
| "queryId" : "CSAS_OUTPUT_0" | ||
| } | ||
| } ], | ||
| "configs" : { | ||
| "ksql.extension.dir" : "ext", | ||
| "ksql.streams.cache.max.bytes.buffering" : "0", | ||
| "ksql.security.extension.class" : null, | ||
| "ksql.transient.prefix" : "transient_", | ||
| "ksql.persistence.wrap.single.values" : "true", | ||
| "ksql.authorization.cache.expiry.time.secs" : "30", | ||
| "ksql.schema.registry.url" : "", | ||
| "ksql.streams.default.deserialization.exception.handler" : "io.confluent.ksql.errors.LogMetricAndContinueExceptionHandler", | ||
| "ksql.output.topic.name.prefix" : "", | ||
| "ksql.streams.auto.offset.reset" : "earliest", | ||
| "ksql.query.pull.enable.standby.reads" : "false", | ||
| "ksql.connect.url" : "http://localhost:8083", | ||
| "ksql.service.id" : "some.ksql.service.id", | ||
| "ksql.internal.topic.min.insync.replicas" : "1", | ||
| "ksql.streams.shutdown.timeout.ms" : "300000", | ||
| "ksql.new.api.enabled" : "false", | ||
| "ksql.streams.state.dir" : "/var/folders/2d/3pt97ylj3zngd51bwl91bl3r0000gp/T/confluent3048982612619405901", | ||
| "ksql.internal.topic.replicas" : "1", | ||
| "ksql.insert.into.values.enabled" : "true", | ||
| "ksql.query.pull.max.allowed.offset.lag" : "9223372036854775807", | ||
| "ksql.streams.default.production.exception.handler" : "io.confluent.ksql.errors.ProductionExceptionHandlerUtil$LogAndFailProductionExceptionHandler", | ||
| "ksql.access.validator.enable" : "auto", | ||
| "ksql.streams.bootstrap.servers" : "localhost:0", | ||
| "ksql.streams.commit.interval.ms" : "2000", | ||
| "ksql.metric.reporters" : "", | ||
| "ksql.streams.auto.commit.interval.ms" : "0", | ||
| "ksql.metrics.extension" : null, | ||
| "ksql.streams.topology.optimization" : "all", | ||
| "ksql.query.pull.streamsstore.rebalancing.timeout.ms" : "10000", | ||
| "ksql.hidden.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses", | ||
| "ksql.streams.num.stream.threads" : "4", | ||
| "ksql.authorization.cache.max.entries" : "10000", | ||
| "ksql.metrics.tags.custom" : "", | ||
| "ksql.pull.queries.enable" : "true", | ||
| "ksql.udfs.enabled" : "true", | ||
| "ksql.udf.enable.security.manager" : "true", | ||
| "ksql.connect.worker.config" : "", | ||
| "ksql.sink.window.change.log.additional.retention" : "1000000", | ||
| "ksql.readonly.topics" : "_confluent.*,__confluent.*,_schemas,__consumer_offsets,__transaction_state,connect-configs,connect-offsets,connect-status,connect-statuses", | ||
| "ksql.udf.collect.metrics" : "false", | ||
| "ksql.persistent.prefix" : "query_", | ||
| "ksql.query.persistent.active.limit" : "2147483647" | ||
| } | ||
| } |
22 changes: 22 additions & 0 deletions
22
..._plans/elements_-_validate_without_value_elements_OK_-_AVRO/6.0.0_1582711670275/spec.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| { | ||
| "version" : "6.0.0", | ||
| "timestamp" : 1582711670275, | ||
| "schemas" : { | ||
| "CSAS_OUTPUT_0.KsqlTopic.Source" : "STRUCT<C1 INT> NOT NULL", | ||
| "CSAS_OUTPUT_0.OUTPUT" : "STRUCT<C1 INT> NOT NULL" | ||
| }, | ||
| "inputs" : [ { | ||
| "topic" : "input", | ||
| "key" : 42, | ||
| "value" : { | ||
| "c1" : 4 | ||
| } | ||
| } ], | ||
| "outputs" : [ { | ||
| "topic" : "OUTPUT", | ||
| "key" : 42, | ||
| "value" : { | ||
| "C1" : 4 | ||
| } | ||
| } ] | ||
| } |
13 changes: 13 additions & 0 deletions
13
...l_plans/elements_-_validate_without_value_elements_OK_-_AVRO/6.0.0_1582711670275/topology
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| Topologies: | ||
| Sub-topology: 0 | ||
| Source: KSTREAM-SOURCE-0000000000 (topics: [input]) | ||
| --> KSTREAM-TRANSFORMVALUES-0000000001 | ||
| Processor: KSTREAM-TRANSFORMVALUES-0000000001 (stores: []) | ||
| --> Project | ||
| <-- KSTREAM-SOURCE-0000000000 | ||
| Processor: Project (stores: []) | ||
| --> KSTREAM-SINK-0000000003 | ||
| <-- KSTREAM-TRANSFORMVALUES-0000000001 | ||
| Sink: KSTREAM-SINK-0000000003 (topic: OUTPUT) | ||
| <-- Project | ||
|
|
Oops, something went wrong.