Re: [DISCUSS] FLIP-107: Reading table columns from different parts of source records

[Timo Walther]

Hi everyone,

I updated the FLIP again and hope that I could address the mentioned 
concerns.

@Leonard: Thanks for the explanation. I wasn't aware that ts_ms and 
source.ts_ms have different semantics. I updated the FLIP and expose the 
most commonly used properties separately. So frequently used properties 
are not hidden in the MAP anymore:

debezium-json.ingestion-timestamp
debezium-json.source.timestamp
debezium-json.source.database
debezium-json.source.schema
debezium-json.source.table

However, since other properties depend on the used connector/vendor, the 
remaining options are stored in:

debezium-json.source.properties

And accessed with:

CAST(SYSTEM_METADATA('debezium-json.source.properties') AS MAP<STRING, 
STRING>)['table']

Otherwise it is not possible to figure out the value and column type 
during validation.

@Jark: You convinced me in relaxing the CAST constraints. I added a 
dedicacated sub-section to the FLIP:

For making the use of SYSTEM_METADATA easier and avoid nested casting we 
allow explicit casting to a target data type:

rowtime AS CAST(SYSTEM_METADATA("timestamp") AS TIMESTAMP(3) WITH LOCAL 
TIME ZONE)

A connector still produces and consumes the data type returned by 
`listMetadata()`. The planner will insert necessary explicit casts.

In any case, the user must provide a CAST such that the computed column 
receives a valid data type when constructing the table schema.

"I don't see a reason why `DecodingFormat#applyReadableMetadata` needs a
DataType argument."

Correct he DeserializationSchema doesn't need TypeInfo, it is always 
executed locally. It is the source that needs TypeInfo for serializing 
the record to the next operator. And that's this is what we provide.

@Danny:

“SYSTEM_METADATA("offset")` returns the NULL type by default”

We can also use some other means to represent an UNKNOWN data type. In 
the Flink type system, we use the NullType for it. The important part is 
that the final data type is known for the entire computed column. As I 
mentioned before, I would avoid the suggested option b) that would be 
similar to your suggestion. The CAST should be enough and allows for 
complex expressions in the computed column. Option b) would need parser 
changes.

Regards,
Timo



On 08.09.20 06:21, Leonard Xu wrote:
Hi, Timo

Thanks for you explanation and update,  I have only one question  for the 
latest FLIP.

About the MAP<STRING, STRING> DataType of key 'debezium-json.source', if user 
want to use the table name metadata, they need to write:
tableName STRING AS CAST(SYSTEM_METADATA('debeuim-json.source') AS MAP<STRING, 
STRING>)['table']

the expression is a little complex for user, Could we only support necessary 
metas with simple DataType as following?
tableName STRING AS CAST(SYSTEM_METADATA('debeuim-json.source.table') AS 
STRING),
transactionTime LONG AS CAST(SYSTEM_METADATA('debeuim-json.source.ts_ms') AS 
BIGINT),

In this way, we can simplify the expression, the mainly used metadata in 
changelog format may include 'database','table','source.ts_ms','ts_ms' from my 
side,
maybe we could only support them at first version.

Both Debezium and Canal have above four metadata, and I‘m willing to take some 
subtasks in next development if necessary.

Debezium:
{
   "before": null,
   "after": {  "id": 101,"name": "scooter"},
   "source": {
     "db": "inventory",                  # 1. database name the changelog 
belongs to.
     "table": "products",                # 2. table name the changelog belongs 
to.
     "ts_ms": 1589355504100,             # 3. timestamp of the change happened 
in database system, i.e.: transaction time in database.
     "connector": "mysql",
     ….
   },
   "ts_ms": 1589355606100,              # 4. timestamp when the debezium 
processed the changelog.
   "op": "c",
   "transaction": null
}

Canal:
{
   "data": [{  "id": "102", "name": "car battery" }],
   "database": "inventory",      # 1. database name the changelog belongs to.
   "table": "products",          # 2. table name the changelog belongs to.
   "es": 1589374013000,          # 3. execution time of the change in database 
system, i.e.: transaction time in database.
   "ts": 1589374013680,          # 4. timestamp when the cannal processed the 
changelog.
   "isDdl": false,
   "mysqlType": {},
   ....
}


Best
Leonard

在 2020年9月8日，11:57，Danny Chan <yuzhao....@gmail.com> 写道：

Thanks Timo ~

The FLIP was already in pretty good shape, I have only 2 questions here:


1. “`CAST(SYSTEM_METADATA("offset") AS INT)` would be a valid read-only 
computed column for Kafka and can be extracted by the planner.”


What is the pros we follow the SQL-SERVER syntax here ? Usually an expression 
return type can be inferred automatically. But I guess SQL-SERVER does not have 
function like SYSTEM_METADATA which actually does not have a specific return 
type.

And why not use the Oracle or MySQL syntax there ?

column_name [datatype] [GENERATED ALWAYS] AS (expression) [VIRTUAL]
Which is more straight-forward.

2. “SYSTEM_METADATA("offset")` returns the NULL type by default”

The default type should not be NULL because only NULL literal does that. 
Usually we use ANY as the type if we do not know the specific type in the SQL 
context. ANY means the physical value can be any java object.

[1] https://oracle-base.com/articles/11g/virtual-columns-11gr1
[2] https://dev.mysql.com/doc/refman/5.7/en/create-table-generated-columns.html

Best,
Danny Chan
在 2020年9月4日 +0800 PM4:48，Timo Walther <twal...@apache.org>，写道：
Hi everyone,

I completely reworked FLIP-107. It now covers the full story how to read
and write metadata from/to connectors and formats. It considers all of
the latest FLIPs, namely FLIP-95, FLIP-132 and FLIP-122. It introduces
the concept of PERSISTED computed columns and leaves out partitioning
for now.

Looking forward to your feedback.

Regards,
Timo


On 04.03.20 09:45, Kurt Young wrote:
Sorry, forgot one question.

4. Can we make the value.fields-include more orthogonal? Like one can
specify it as "EXCEPT_KEY, EXCEPT_TIMESTAMP".
With current EXCEPT_KEY and EXCEPT_KEY_TIMESTAMP, users can not config to
just ignore timestamp but keep key.

Best,
Kurt


On Wed, Mar 4, 2020 at 4:42 PM Kurt Young <ykt...@gmail.com> wrote:

Hi Dawid,

I have a couple of questions around key fields, actually I also have some
other questions but want to be focused on key fields first.

1. I don't fully understand the usage of "key.fields". Is this option only
valid during write operation? Because for
reading, I can't imagine how such options can be applied. I would expect
that there might be a SYSTEM_METADATA("key")
to read and assign the key to a normal field?

2. If "key.fields" is only valid in write operation, I want to propose we
can simplify the options to not introducing key.format.type and
other related options. I think a single "key.field" (not fields) would be
enough, users can use UDF to calculate whatever key they
want before sink.

3. Also I don't want to introduce "value.format.type" and
"value.format.xxx" with the "value" prefix. Not every connector has a
concept
of key and values. The old parameter "format.type" already good enough to
use.

Best,
Kurt


On Tue, Mar 3, 2020 at 10:40 PM Jark Wu <imj...@gmail.com> wrote:

Thanks Dawid,

I have two more questions.

SupportsMetadata
Introducing SupportsMetadata sounds good to me. But I have some questions
regarding to this interface.
1) How do the source know what the expected return type of each metadata?
2) Where to put the metadata fields? Append to the existing physical
fields?
If yes, I would suggest to change the signature to `TableSource
appendMetadataFields(String[] metadataNames, DataType[] metadataTypes)`

SYSTEM_METADATA("partition")
Can SYSTEM_METADATA() function be used nested in a computed column
expression? If yes, how to specify the return type of SYSTEM_METADATA?

Best,
Jark

On Tue, 3 Mar 2020 at 17:06, Dawid Wysakowicz <dwysakow...@apache.org>
wrote:

Hi,

1. I thought a bit more on how the source would emit the columns and I
now see its not exactly the same as regular columns. I see a need to
elaborate a bit more on that in the FLIP as you asked, Jark.

I do agree mostly with Danny on how we should do that. One additional
things I would introduce is an

interface SupportsMetadata {

boolean supportsMetadata(Set<String> metadataFields);

TableSource generateMetadataFields(Set<String> metadataFields);

}

This way the source would have to declare/emit only the requested
metadata fields. In order not to clash with user defined fields. When
emitting the metadata field I would prepend the column name with
__system_{property_name}. Therefore when requested
SYSTEM_METADATA("partition") the source would append a field
__system_partition to the schema. This would be never visible to the
user as it would be used only for the subsequent computed columns. If
that makes sense to you, I will update the FLIP with this description.

2. CAST vs explicit type in computed columns

Here I agree with Danny. It is also the current state of the proposal.

3. Partitioning on computed column vs function

Here I also agree with Danny. I also think those are orthogonal. I would
leave out the STORED computed columns out of the discussion. I don't see
how do they relate to the partitioning. I already put both of those
cases in the document. We can either partition on a computed column or
use a udf in a partioned by clause. I am fine with leaving out the
partitioning by udf in the first version if you still have some
concerns.

As for your question Danny. It depends which partitioning strategy you
use.

For the HASH partitioning strategy I thought it would work as you
explained. It would be N = MOD(expr, num). I am not sure though if we
should introduce the PARTITIONS clause. Usually Flink does not own the
data and the partitions are already an intrinsic property of the
underlying source e.g. for kafka we do not create topics, but we just
describe pre-existing pre-partitioned topic.

4. timestamp vs timestamp.field vs connector.field vs ...

I am fine with changing it to timestamp.field to be consistent with
other value.fields and key.fields. Actually that was also my initial
proposal in a first draft I prepared. I changed it afterwards to shorten
the key.

Best,

Dawid

On 03/03/2020 09:00, Danny Chan wrote:
Thanks Dawid for bringing up this discussion, I think it is a useful
feature ~

About how the metadata outputs from source

I think it is completely orthogonal, computed column push down is
another topic, this should not be a blocker but a promotion, if we do
not
have any filters on the computed column, there is no need to do any
pushings; the source node just emit the complete record with full
metadata
with the declared physical schema, then when generating the virtual
columns, we would extract the metadata info and output as full
columns(with
full schema).

About the type of metadata column

Personally i prefer explicit type instead of CAST, they are symantic
equivalent though, explict type is more straight-forward and we can
declare
the nullable attribute there.

About option A: partitioning based on acomputed column VS option B:
partitioning with just a function

 From the FLIP, it seems that B's partitioning is just a strategy when
writing data, the partiton column is not included in the table schema,
so
it's just useless when reading from that.

- Compared to A, we do not need to generate the partition column when
selecting from the table(but insert into)
- For A we can also mark the column as STORED when we want to persist
that

So in my opition they are orthogonal, we can support both, i saw that
MySQL/Oracle[1][2] would suggest to also define the PARTITIONS num, and
the
partitions are managed under a "tablenamespace", the partition in which
the
record is stored is partition number N, where N = MOD(expr, num), for
your
design, which partiton the record would persist ?

[1] https://dev.mysql.com/doc/refman/5.7/en/partitioning-hash.html
[2]

https://docs.oracle.com/database/121/VLDBG/GUID-F023D3ED-262F-4B19-950A-D3C8F8CDB4F4.htm#VLDBG1270

Best,
Danny Chan
在 2020年3月2日 +0800 PM6:16，Dawid Wysakowicz <dwysakow...@apache.org
，写道：
Hi Jark,
Ad. 2 I added a section to discuss relation to FLIP-63
Ad. 3 Yes, I also tried to somewhat keep hierarchy of properties.
Therefore you have the key.format.type.
I also considered exactly what you are suggesting (prefixing with
connector or kafka). I should've put that into an Option/Rejected
alternatives.
I agree timestamp, key.*, value.* are connector properties. Why I
wanted to suggest not adding that prefix in the first version is that
actually all the properties in the WITH section are connector
properties.
Even format is in the end a connector property as some of the sources
might
not have a format, imo. The benefit of not adding the prefix is that it
makes the keys a bit shorter. Imagine prefixing all the properties with
connector (or if we go with FLINK-12557: elasticsearch):
elasticsearch.key.format.type: csv
elasticsearch.key.format.field: ....
elasticsearch.key.format.delimiter: ....
elasticsearch.key.format.*: ....
I am fine with doing it though if this is a preferred approach in the
community.
Ad in-line comments:
I forgot to update the `value.fields.include` property. It should be
value.fields-include. Which I think you also suggested in the comment,
right?
As for the cast vs declaring output type of computed column. I think
it's better not to use CAST, but declare a type of an expression and
later
on infer the output type of SYSTEM_METADATA. The reason is I think this
way
it will be easier to implement e.g. filter push downs when working with
the
native types of the source, e.g. in case of Kafka's offset, i think it's
better to pushdown long rather than string. This could let us push
expression like e.g. offset > 12345 & offset < 59382. Otherwise we would
have to push down cast(offset, long) > 12345 && cast(offset, long) <
59382.
Moreover I think we need to introduce the type for computed columns
anyway
to support functions that infer output type based on expected return
type.
As for the computed column push down. Yes, SYSTEM_METADATA would have
to be pushed down to the source. If it is not possible the planner
should
fail. As far as I know computed columns push down will be part of source
rework, won't it? ;)
As for the persisted computed column. I think it is completely
orthogonal. In my current proposal you can also partition by a computed
column. The difference between using a udf in partitioned by vs
partitioned
by a computed column is that when you partition by a computed column
this
column must be also computed when reading the table. If you use a udf in
the partitioned by, the expression is computed only when inserting into
the
table.
Hope this answers some of your questions. Looking forward for further
suggestions.
Best,
Dawid


On 02/03/2020 05:18, Jark Wu wrote:
Hi,

Thanks Dawid for starting such a great discussion. Reaing metadata
and
key-part information from source is an important feature for
streaming
users.

In general, I agree with the proposal of the FLIP.
I will leave my thoughts and comments here:

1) +1 to use connector properties instead of introducing HEADER
keyword as
the reason you mentioned in the FLIP.
2) we already introduced PARTITIONED BY in FLIP-63. Maybe we should
add a
section to explain what's the relationship between them.
Do their concepts conflict? Could INSERT PARTITION be used on the
PARTITIONED table in this FLIP?
3) Currently, properties are hierarchical in Flink SQL. Shall we
make
the
new introduced properties more hierarchical?
For example, "timestamp" => "connector.timestamp"? (actually, I
prefer
"kafka.timestamp" which is another improvement for properties
FLINK-12557)
A single "timestamp" in properties may mislead users that the
field
is
a rowtime attribute.

I also left some minor comments in the FLIP.

Thanks,
Jark



On Sun, 1 Mar 2020 at 22:30, Dawid Wysakowicz <
dwysakow...@apache.org>
wrote:

Hi,

I would like to propose an improvement that would enable reading
table
columns from different parts of source records. Besides the main
payload
majority (if not all of the sources) expose additional
information. It
can be simply a read-only metadata such as offset, ingestion time
or a
read and write parts of the record that contain data but
additionally
serve different purposes (partitioning, compaction etc.), e.g. key
or
timestamp in Kafka.

We should make it possible to read and write data from all of those
locations. In this proposal I discuss reading partitioning data,
for
completeness this proposal discusses also the partitioning when
writing
data out.

I am looking forward to your comments.

You can access the FLIP here:



https://cwiki.apache.org/confluence/display/FLINK/FLIP-107%3A+Reading+table+columns+from+different+parts+of+source+records?src=contextnavpagetreemode

Best,

Dawid