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).

...

The `VIRTUAL` keyword excludes the column from the query-to-sink schema which means that a query cannot write to a this metadata column.

Reading metadata via DynamicTableSource

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`
SchemaBuilder.columnByMetadata(String, DataType)
// for `offset INT METADATA VIRTUAL`
SchemaBuilder.columnByMetadata(String, DataType, boolean isVirtual)
// for `myOffset INT METADATA FROM 'offset' VIRTUAL`
SchemaBuilder.columnByMetadata(String, DataType, String)
SchemaBuilder.columnByMetadata(String, DataType, String, boolean isVirtual)

An example would look like:

.schema(Schema.newBuilder()
   .column("user_id", DataTypes.BIGINT())
   .column("user_name", DataTypes.STRING())
   .columnByMetadata("offset", DataTypes.INT())
   .columnByMetadata("myOffset", DataTypes.INT(), "offset")
   .watermark("ts", $("ts").minus(lit(3).seconds()))
   .primaryKey("user_id")
   .build()
)

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);

CREATE TABLE t (timestamp TIMESTAMP(3) WITH LOCAL TIME ZONE METADATA FROM 'other-system.timestamp')
WITH (...)
LIKE t (
  INCLUDING ALL 
  OVERWRITING OPTIONS
  OVERWRITING METADATA
)

Reading metadata via DynamicTableSource

LetLet's assume the following example:

...

topic

partition

headers

leader-epoch

offset

timestamp

timestamp-type

...

debezium-json.schema

debezium-json.ingestion-timestamp

debezium-json.source.timestamp

debezium-json.source.database

debezium-json.source.schema

debezium-json.source.table

debezium-json.source.properties

...