<a id="ksqldb-schema-inference-with-id"></a>

# Schema Inference With ID in ksqlDB for Confluent Platform

Schema inference with ID lets you specify a `KEY_SCHEMA_ID` or
`VALUE_SCHEMA_ID` in a `CREATE` statement so ksqlDB retrieves and
registers that exact schema from Schema Registry, instead of inferring the latest
one automatically. This means you don’t have to define columns and
data types manually, and your data is serialized using the schema the
ID points to. Before using schema inference with explicit IDs, make
sure that Schema Registry is up and running and ksqlDB is
[configured to use it](installation/avro-schema.md#ksqldb-installation-configure-serialization-formats).

Here’s what you can do with schema inference with IDs in ksqlDB:

- Declare streams and tables on Kafka topics with supported key
  and value formats by using `CREATE STREAM` and `CREATE TABLE`
  statements with `KEY_SCHEMA_ID` or `VALUE_SCHEMA_ID` properties,
  without the need to declare the key and value columns.
- Declare derived views with `CREATE STREAM AS SELECT` and
  `CREATE TABLE AS SELECT` statements with `KEY_SCHEMA_ID` or
  `VALUE_SCHEMA_ID` properties. The schema of the view is registered
  in Schema Registry automatically.
- Serialize output data using the schema referred to by
  `KEY_SCHEMA_ID` or `VALUE_SCHEMA_ID`, instead of the logical data
  source schema stored in ksqlDB.

If you’re declaring a stream or table with a key format that’s different
from its value format, and only one of the two formats supports schema
inference, you can explicitly provide the columns for the format that
doesn’t support schema inference while still having ksqlDB load columns
for the format that does support schema inference from Schema Registry. This
is known as *partial schema inference*. To infer value columns for a
keyless stream, set the key format to
[NONE](../reference/serialization.md#ksqldb-serialization-formats-none).

Tables require a `PRIMARY KEY`, so you must supply one explicitly in
your `CREATE TABLE` statement. `KEY` columns are optional for
streams, so if you don’t supply one, the stream is created without a key
column.

#### NOTE
The following example statements show how to create streams and tables
that have Avro-formatted data. If you want to use Protobuf or
JSON-formatted data, substitute `PROTOBUF` or `JSON_SR` for `AVRO`
in each statement.

## Create a new stream or table

When `KEY_SCHEMA_ID` or `VALUE_SCHEMA_ID` is used in statements to
create a stream or table, the schema fetched from Schema Registry is used to
infer data source’s columns and serialize output data. See
[Schema inference and data serialization](#ksqldb-schema-inference-with-id-and-data-serialization)
for details about how columns are inferred and data are serialized.

#### IMPORTANT
- The schemas referred to by `KEY_SCHEMA_ID` and `VALUE_SCHEMA_ID` must
  be registered in Schema Registry. They can be under any subject but must match the
  formats defined by `KEY_FORMAT` and `VALUE_FORMAT`, respectively.
- You can’t define key or value columns in a statement if a corresponding
  `KEY_SCHEMA_ID` or `VALUE_SCHEMA_ID` is supplied.

### Without a key column

The following statement shows how to create a new `pageviews` stream
by reading from a Kafka topic that has Avro-formatted message
values.

```sql
CREATE STREAM pageviews
  WITH (
    KAFKA_TOPIC='pageviews-avro-topic',
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_ID=1
  );
```

In this example, you don’t need to define any columns in the CREATE
statement. ksqlDB infers this information automatically from Schema Registry
using the provided `VALUE_SCHEMA_ID`.

### With a key column

The following statement shows how to create a new `pageviews` stream
by reading from a Kafka topic that has Avro-formatted key and
message values.

```sql
CREATE STREAM pageviews WITH (
    KAFKA_TOPIC='pageviews-avro-topic',
    KEY_FORMAT='AVRO',
    VALUE_FORMAT='AVRO',
    KEY_SCHEMA_ID=1,
    VALUE_SCHEMA_ID=2
  );
```

In this example, ksqlDB infers the key and value columns automatically
from Schema Registry using the provided `KEY_SCHEMA_ID` and
`VALUE_SCHEMA_ID`.

### With partial schema inference

The following statement shows how to create a new `pageviews` table by
reading from a Kafka topic that has Avro-formatted message values
and a `KAFKA`-formatted `INT` primary key.

```sql
CREATE TABLE pageviews (
    pageId INT PRIMARY KEY
  ) WITH (
    KAFKA_TOPIC='pageviews-avro-topic',
    KEY_FORMAT='KAFKA',
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_ID=1
  );
```

In this example, only the key column is supplied in the CREATE
statement. ksqlDB infers the value columns automatically from
Schema Registry using the provided `VALUE_SCHEMA_ID`.

## Declaring a derived view with schema ID.

The following statement shows how to create a materialized view derived
from an existing source with the `VALUE_SCHEMA_ID` property. The
schema referred to by `VALUE_SCHEMA_ID` is used to check column
compatibility with output columns and serialize output data. For more
information, see
[Schema inference and data serialization](#ksqldb-schema-inference-with-id-and-data-serialization).

```sql
CREATE STREAM pageviews_new
  WITH (
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_ID=1
  ) AS
  SELECT
    pageId,
    ts
  FROM pageviews
```

#### IMPORTANT
The schema referred to by `VALUE_SCHEMA_ID` must be compatible with the
logical schema defined by the `SELECT` clause. For more information, see
[Schema inference and data serialization](#ksqldb-schema-inference-with-id-and-data-serialization).

<a id="ksqldb-schema-inference-with-id-and-data-serialization"></a>

## Schema inference and data serialization

The schema in Schema Registry is a “physical schema”, and the schema in
ksqlDB is a “logical schema”. The physical schema, not the logical
schema, is registered under the subject `<topic-name>-key` or
`<topic-name>-value` if corresponding `KEY_SCHEMA_ID` or
`VALUE_SCHEMA_ID` values are provided.

### Schema inference schema requirements

If `WRAP_SINGLE_VALUE` is set to `true` in the SQL statement, the
physical schema is expected to be a `struct` type, and the field names
are used as data source column names. Field types are inferred from
corresponding column data types.

- In `AVRO`, the `struct` type corresponds with the `record` type.
- In `PROTOBUF` the `struct` type corresponds with the `message` type.
- In `JSON_SR`, the `struct` type corresponds with the `object` type.

#### NOTE
In the following examples, the `AVRO` schema string in Schema Registry is a
single-line raw string without newline characters (`\n`). The strings
are shown as human-readable text for convenience.

For example, the following a physical schema is in `AVRO` format and
is registered with Schema Registry with ID 1:

```json
{
  "schema": {
    "type": "record",
    "name": "PageViewValueSchema",
    "namespace": "io.confluent.ksql.avro_schemas",
    "fields": [
      {
        "name": "page_name",
        "type": "string",
        "default": "abc"
      },
      {
        "name": "ts",
        "type": "int",
        "default": 123
      }
    ]
  }
}
```

The following `CREATE` statement defines a stream on the `pageviews`
topic and specifies the physical schema that has an ID of `1`.

```sql
CREATE STREAM pageviews (pageId INT KEY)
WITH (
   KAFKA_TOPIC='pageviews-avro-topic', KEY_FORMAT='KAFKA', VALUE_FORMAT='AVRO',VALUE_SCHEMA_ID=1,PARTITIONS=1
   );
```

The following output from the `describe pageviews` command shows the
inferred logical schema for the `pageviews` stream:

```sql
DESCRIBE pageviews;

Name                 : PAGEVIEWS
 Field     | Type
------------------------------------
 PAGEID    | INTEGER          (key)
 page_name | VARCHAR(STRING)
 ts        | INTEGER
------------------------------------
```

If `WRAP_SINGLE_VALUE` is `false` in the statement, and if
`KEY_SCHEMA_ID` is set, `ROWKEY` is used as the key’s column name.

If `VALUE_SCHEMA_ID` is set, `ROWVAL` is used as the value’s column
name. The physical schema is used as the column data type.

For example, the following physical schema is `AVRO` and is defined in
Schema Registry with ID `2`:

```json
{"schema": "int"}
```

The following `CREATE` statement defines a table on the
`pageview-count` topic and specifies the physical schema that has ID
`2`.

`sql hl_lines="7" CREATE TABLE pageview_count (     pageId INT PRIMARY KEY   ) WITH (     KAFKA_TOPIC='pageview-count',     KEY_FORMAT='KAFKA',     VALUE_FORMAT='AVRO',     VALUE_SCHEMA_ID=2,     WRAP_SINGLE_VALUE=false,     PARTITIONS=1   );`

The inferred logical schema for the `pageview_count` table is:

```none
Name                 : PAGEVIEW_COUNT
 Field  | Type
-----------------------------------------
 PAGEID | INTEGER          (primary key)
 ROWVAL | INTEGER
-----------------------------------------
```

For more information about `WRAP_SINGLE_VALUE`, see
[Single field unwrapping](../reference/serialization.md#ksqldb-serialization-formats-single-field-unwrapping).

### Schema inference type handling

ksqlDB supports the `null` value for `KEY` and `VALUE` columns. If
a field in the physical schema is a required type, it’s translated to an
optional type in the logical schema. This has subtle implications for
data serialization which are explained in the following section.

#### IMPORTANT
- ksqlDB ignores unsupported types in the physical schema and continues
  translating supported types to the logical schema. You should verify
  that the logical schema is translated as expected.
- During schema translation from a physical schema to a logical schema,
  `struct` type field names are used as column names in the logical
  schema. Field names *are not* translated to uppercase, in contrast with
  schema inference without a schema id, which *does* translate field names
  to uppercase.

### Schema compatibility check for derived view

You can use schema IDs when creating a materialized view, but instead of
inferring the logical schema for the view, the schema is used to check
compatibility against the query’s projection and serialized output data.
For compatibility checks, the inferred logical schema must be a superset
of the query’s projection schema, which means corresponding column
names, types, and order must match. The inferred logical schema may have
extra columns.

The following example creates the `pageviews_new` topic as the result
of a SELECT query:

```sql
CREATE STREAM pageviews_new
  WITH (
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_ID=1
  ) AS
  SELECT
    pageId,
    ts
  FROM pageviews
```

If the `pageviews` value column has the type `ts INT`, the logical
schema of `pageviews_new` is decided by the projection in the query
`SELECT pageId, ts FROM pageviews`. When `VALUE_SCHEMA_ID` is used,
the inferred logical schema is checked against `ts INT` for
compatibility.

The following example shows a compatible physical schema:

`json hl_lines="8-9" {   "schema": {     "type": "record",     "name": "PageViewNewSchema",     "namespace": "io.confluent.ksql.avro_schemas",     "fields": [       {         "name": "ts",         "type": "int",         "default": 123       },       {         "name": "title",         "type": "string",         "default": "title"       }     ]   } }`

In this `AVRO` schema, `title` is an extra field. Because the
physical schema is used for data serialization, the `title` field with
a default value appears in serialized data, even though the inserted
value can never set the `title` field, because it’s not in the logical
schema defined by the `SELECT` clause of the query.

The following example shows an incompatible physical schema, which is
incompatible because of the type mismatch for `pageId`.

`json hl_lines="8-9" {   "schema": {     "type": "record",     "name": "PageViewNewSchema",     "namespace": "io.confluent.ksql.avro_schemas",     "fields": [       {         "name": "pageId",         "type": "string",         "default": "id"       },       {         "name": "ts",         "type": "int",         "default": 123       }     ]   } }`

### Data serialization

When a schema ID is provided, and schema inference is successful, ksqlDB
can create the data source. When writing to the data source, the
physical schema inferred by the schema ID is used to serialize data,
instead of the logical schema that’s used in other cases. Because
ksqlDB’s logical schema accepts `null` values but the physical schema
may not, serialization can fail even if the inserted value is valid for
the logical schema.

The following example shows a physical schema that’s defined in
Schema Registry with ID `1`. No default values are specified for the
`page_name` and `ts` fields.

`json hl_lines="8-9 12-13" {   "schema": {     "type": "record",     "name": "PageViewValueSchema",     "namespace": "io.confluent.ksql.avro_schemas",     "fields": [       {         "name": "page_name",         "type": "string"       },       {         "name": "ts",         "type": "int"       }     ]   } }`

The following example creates a stream with schema ID `1`:

```sql
CREATE STREAM pageviews (
    pageId INT KEY
  ) WITH (
    KAFKA_TOPIC='pageviews-avro-topic',
    KEY_FORMAT='KAFKA',
    VALUE_FORMAT='AVRO',
    VALUE_SCHEMA_ID=1,
    PARTITIONS=1
  );
```

ksqlDB infers the following schema for `pageviews`:

```none
Name                 : PAGEVIEWS
 Field     | Type
------------------------------------
 PAGEID    | INTEGER          (key)
 page_name | VARCHAR(STRING)
 ts        | INTEGER
------------------------------------
```

If you insert values to `pageviews` with `null` values, ksqlDB
returns an error:

```sql
INSERT INTO pageviews VALUES (1, null, null);
```

```none
Failed to insert values into 'PAGEVIEWS'. Could not serialize value: [ null | null ]. Error serializing message to topic: pageviews-avro-topic1. Invalid value: null used for required field: "page_name", schema type: STRING
```

This error occurs because `page_name` and `ts` are required fields
without default values in the specified physical schema.

#### IMPORTANT
ksqlDB doesn’t check that `null` can be serialized in a physical schema
that contains required fields. You must ensure that `null` can be handled
properly, either by making physical schema fields optional or by using the
[IFNULL](../developer-guide/ksqldb-reference/scalar-functions.md#ksqldb-reference-scalar-functions-ifnull)
function to ensure that `null` is never inserted.

#### IMPORTANT
If `key_schema_id` is used in table creation with windowed aggregation,
the serialized key value also contain window information in addition to
original key value.