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
