Page properties | |||||
---|---|---|---|---|---|
|
Status
Current state: "Under discussion"
Discussion thread: http://apache-flink-mailing-list-archive.1008284.n3.nabble.com/DISCUSS-FLIP-107-Reading-table-columns-from-different-parts-of-source-records-td38277.html
...
|
...
|
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
CREATE TABLE kinesis_table ( |
...
INSERT INTO kafka_table VALUES ( |
Or with no persisted metadata:
...
Let's assume the following example:
CREATE TABLE t (i INT, s STRING, timestamp TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA, d DOUBLE) |
The `timestamp` column is declared as a `METADATA` column. By default, the column name is used to map to a corresponding metadata key ("timestamp" in this case).
The data type of the column is used to perform an explicit cast of the original metadata data type. For example, a Kafka metadata timestamp is defined as `BIGINT`. A user can declare the column as `TIMESTAMP(3)` though and thus force an explicit cast from sources (BIGINT to TIMESTAMP) and to sinks (TIMESTAMP to BIGINT).
CREATE TABLE t (i INT, s STRING, myOffset INT METADATA FROM 'offset', d DOUBLE) |
In order to use a different column name, it is possible to use `FROM` to reference the metadata key ("offset" in this case).
CREATE TABLE t (i INT, s STRING, offset INT METADATA VIRTUAL, d DOUBLE) |
`offset INT METADATA` would be a valid read-only column for Kafka and can be extracted by the planner. However, since declared tables can be used for sources and sinks, we need to be able to exclude read-only metadata from writing.
The `VIRTUAL` keyword excludes the column from the query-to-sink schema which means that a query cannot write to a this metadata column from the query-to-sink schema which means that a query cannot write to a this metadata column..
Implications to other components
Because we are extending the DDL, this has implication to other components.
Schema
We propose to extend the `Schema` class from FLIP-129 by:
// for `offset INT METADATA` |
An example would look like:
.schema(Schema.newBuilder() |
LIKE clause
Currently, the LIKE clause offers the following table features:
- CONSTRAINTS - constraints such as primary and unique keys
- GENERATED - computed columns
- OPTIONS - connector options that describe connector and format properties
- PARTITIONS - partition of the tables
- WATERMARKS - watermark declarations
We propose to extend the LIKE clause to add `METADATA` as another table feature.
Metadata columns are not generated. They represent data that is present in the external system. Thus, it should not be categorized into the GENERATED feature.
Furthermore, METADATA is connector dependent. It is safer to have a fine-grained control over the table feature. The user should control whether metadata can be inherited or not.
For example, this is important when switching from filesystem to Kafka or vice versa.
METADATA supports OVERWRITING by column name.
The following example shows how to overwrite a metadata column with another metadata column:
CREATE TABLE t (i INT, s STRING, timestamp TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA FROM 'timestamp', d DOUBLE); |
Reading metadata via DynamicTableSource
Let's assume the following example:
CREATE TABLE t (i INT, s STRING, offset INT METADATA VIRTUAL, d DOUBLE) |
We suggest the following interfaces for integrating reading metadata into FLIP-95 interfaces:
...
Key | Data type | Read/Write | Notes |
---|---|---|---|
topic | STRING | r | We don't allow writing to different topics for now. Maybe we will allow that in the future via a property. |
partition | INT | r | We don't allow writing to different partitions for now. Maybe we will allow that in the future via a property. |
headers | MAP<STRING, BYTES> | r/w | |
leader-epoch | INT | r | |
offset | BIGINT | r | |
timestampBIGINT | TIMESTAMP(3) WITH LOCAL TIME ZONE | r/w | Directly forward the underlying type. |
timestamp-type | STRING | r | ['NoTimestampType', 'CreateTime', 'LogAppendTime'] |
...
Key | Data type | Read/Write | Notes |
---|---|---|---|
debezium-json.schema | STRING | r | Pure JSON string, can be handled with Flink's built-in JSON SQL functions |
debezium-json.ingestion-timestampBIGINT | TIMESTAMP(3) WITH LOCAL TIME ZONE | r | Directly forward the underlying type. Do not try to abstract it into a TIMESTAMP. |
debezium-json.source.timestampBIGINT | TIMESTAMP(3) WITH LOCAL TIME ZONE | r | Shortcut for debezium-json.source[ts_ms]. |
debezium-json.source.database | STRING | r | Unified shortcut for `debezium-json.source[db]` across SQL vendors. |
debezium-json.source.schema | STRING | r | Unified shortcut for `debezium-json.source[schema]` across SQL vendors. |
debezium-json.source.table | STRING | r | Unified shortcut for `debezium-json.source[table/collection]` across SQL vendors. |
debezium-json.source.properties | MAP<STRING, STRING> | r | All properties that Debezium provides under `source`. Frequently used properties have a a dedicated key above. It seems that this is always a property list. So we can make it available as a map for easier access. |
...
Declare everything via properties:
CREATE TABLE kafka_table ( |
Pros:
- "timestamp", "headers" are something like "key" and "value" that are stored with the real data. So why not define the "timestamp" in the same way with "key" by using a "timestamp.field" connector option?
...
Option 2
Use computed columns:
CREATE TABLE kafka_table ( |
Pros:
- Allows to have full flexibility to compute the final column and avoid helper columns:
timestamp AS adjustTimestamp(CAST(SYSTEM_METADATA("ts") AS TIMESTAMP(3)))
...