Thanks for driving this Timo, +1 for voting ~
Best,
Danny Chan
在 2020年9月10日 +0800 PM3:47，Timo Walther <twal...@apache.org>，写道：
Thanks everyone for this healthy discussion. I updated the FLIP with the
outcome. I think the result is very powerful but also very easy to
declare. Thanks for all the contributions.
If there are no objections, I would continue with a voting.
What do you think?
Regards,
Timo
On 09.09.20 16:52, Timo Walther wrote:
"If virtual by default, when a user types "timestamp int" ==>
persisted
column, then adds a "metadata" after that ==> virtual column, then
adds
a "persisted" after that ==> persisted column."
Thanks for this nice mental model explanation, Jark. This makes total
sense to me. Also making the the most common case as short at just
adding `METADATA` is a very good idea. Thanks, Danny!
Let me update the FLIP again with all these ideas.
Regards,
Timo
On 09.09.20 15:03, Jark Wu wrote:
I'm also +1 to Danny's proposal: timestamp INT METADATA [FROM
'my-timestamp-field'] [VIRTUAL]
Especially I like the shortcut: timestamp INT METADATA, this makes
the
most
common case to be supported in the simplest way.
I also think the default should be "PERSISTED", so VIRTUAL is
optional
when
you are accessing a read-only metadata. Because:
1. The "timestamp INT METADATA" should be a normal column, because
"METADATA" is just a modifier to indicate it is from metadata, a
normal
column should be persisted.
If virtual by default, when a user types "timestamp int" ==>
persisted
column, then adds a "metadata" after that ==> virtual column, then
adds a
"persisted" after that ==> persisted column.
I think this looks reversed several times and makes users
confused.
Physical fields are also prefixed with "fieldName TYPE", so
"timestamp
INT
METADATA" is persisted is very straightforward.
2. From the collected user question [1], we can see that "timestamp"
is the
most common use case. "timestamp" is a read-write metadata.
Persisted by
default doesn't break the reading behavior.
Best,
Jark
[1]: https://issues.apache.org/jira/browse/FLINK-15869
On Wed, 9 Sep 2020 at 20:56, Leonard Xu <xbjt...@gmail.com> wrote:
Thanks @Dawid for the nice summary, I think you catch all
opinions of
the
long discussion well.
@Danny
“ timestamp INT METADATA [FROM 'my-timestamp-field'] [VIRTUAL]
Note that the "FROM 'field name'" is only needed when the name
conflict
with the declared table column name, when there are no
conflicts,
we can
simplify it to
timestamp INT METADATA"
I really like the proposal, there is no confusion with computed
column any
more, and it’s concise enough.
@Timo @Dawid
“We use `SYSTEM_TIME` for temporal tables. I think prefixing with
SYSTEM
makes it clearer that it comes magically from the system.”
“As for the issue of shortening the SYSTEM_METADATA to METADATA.
Here I
very much prefer the SYSTEM_ prefix.”
I think `SYSTEM_TIME` is different with `SYSTEM_METADATA ` a lot,
First of all, the word `TIME` has broad meanings but the word
`METADATA `
not, `METADATA ` has specific meaning,
Secondly, `FOR SYSTEM_TIME AS OF` exists in SQL standard but
`SYSTEM_METADATA ` not.
Personally, I like more simplify way，sometimes less is more.
Best,
Leonard
Timo Walther <twal...@apache.org> 于2020年9月9日周三 下午6:41写道：
Hi everyone,
"key" and "value" in the properties are a special case
because they
need
to configure a format. So key and value are more than just
metadata.
Jark's example for setting a timestamp would work but as the
FLIP
discusses, we have way more metadata fields like headers,
epoch-leader,
etc. Having a property for all of this metadata would mess up
the WITH
section entirely. Furthermore, we also want to deal with
metadata from
the formats. Solving this through properties as well would
further
complicate the property design.
Personally, I still like the computed column design more
because it
allows to have full flexibility to compute the final column:
timestamp AS adjustTimestamp(CAST(SYSTEM_METADATA("ts") AS
TIMESTAMP(3)))
Instead of having a helper column and a real column in the
table:
helperTimestamp AS CAST(SYSTEM_METADATA("ts") AS TIMESTAMP(3))
realTimestamp AS adjustTimestamp(helperTimestamp)
But I see that the discussion leans towards:
timestamp INT SYSTEM_METADATA("ts")
Which is fine with me. It is the shortest solution, because
we don't
need additional CAST. We can discuss the syntax, so that
confusion
with
computed columns can be avoided.
timestamp INT USING SYSTEM_METADATA("ts")
timestamp INT FROM SYSTEM_METADATA("ts")
timestamp INT FROM SYSTEM_METADATA("ts") PERSISTED
We use `SYSTEM_TIME` for temporal tables. I think prefixing
with
SYSTEM
makes it clearer that it comes magically from the system.
What do you think?
Regards,
Timo
On 09.09.20 11:41, Jark Wu wrote:
Hi Danny,
This is not Oracle and MySQL computed column syntax,
because there is
no
"AS" after the type.
Hi everyone,
If we want to use "offset INT SYSTEM_METADATA("offset")",
then I
think
we
must further discuss about "PERSISED" or "VIRTUAL" keyword
for
query-sink
schema problem.
Personally, I think we can use a shorter keyword "METADATA"
for
"SYSTEM_METADATA". Because "SYSTEM_METADATA" sounds like a
system
function
and confuse users this looks like a computed column.
Best,
Jark
On Wed, 9 Sep 2020 at 17:23, Danny Chan <
danny0...@apache.org> wrote:
"offset INT SYSTEM_METADATA("offset")"
This is actually Oracle or MySQL style computed column
syntax.
"You are right that one could argue that "timestamp",
"headers" are
something like "key" and "value""
I have the same feeling, both key value and headers
timestamp are
*real*
data
stored in the consumed record, they are not computed or
generated.
"Trying to solve everything via properties sounds rather
like a hack
to
me"
Things are not that hack if we can unify the routines or
the
definitions
(all from the computed column way or all from the table
options), i
also
think that it is a hacky that we mix in 2 kinds of syntax
for
different
kinds of metadata (read-only and read-write). In this
FLIP, we
declare
the
Kafka key fields with table options but SYSTEM_METADATA
for other
metadata,
that is a hacky thing or something in-consistent.
Kurt Young <ykt...@gmail.com> 于2020年9月9日周三 下午4:48写道：
I would vote for `offset INT
SYSTEM_METADATA("offset")`.
I don't think we can stick with the SQL standard in DDL
part
forever,
especially as there are more and more
requirements coming from different connectors and
external systems.
Best,
Kurt
On Wed, Sep 9, 2020 at 4:40 PM Timo Walther <
twal...@apache.org>
wrote:
Hi Jark,
now we are back at the original design proposed by
Dawid :D
Yes, we
should be cautious about adding new syntax. But the
length of this
discussion shows that we are looking for a good
long-term
solution.
In
this case I would rather vote for a deep integration
into the
syntax.
Computed columns are also not SQL standard compliant.
And our
DDL is
neither, so we have some degree of freedom here.
Trying to solve everything via properties sounds
rather like a
hack
to
me. You are right that one could argue that
"timestamp", "headers"
are
something like "key" and "value". However, mixing
`offset AS SYSTEM_METADATA("offset")`
and
`'timestamp.field' = 'ts'`
looks more confusing to users that an explicit
`offset AS CAST(SYSTEM_METADATA("offset") AS INT)`
or
`offset INT SYSTEM_METADATA("offset")`
that is symetric for both source and sink.
What do others think?
Regards,
Timo
On 09.09.20 10:09, Jark Wu wrote:
Hi everyone,
I think we have a conclusion that the writable
metadata shouldn't
be
defined as a computed column, but a normal column.
"timestamp STRING SYSTEM_METADATA('timestamp')" is
one of the
approaches.
However, it is not SQL standard compliant, we need
to be cautious
enough
when adding new syntax.
Besides, we have to introduce the `PERSISTED` or
`VIRTUAL`
keyword
to
resolve the query-sink schema problem if it is
read-only
metadata.
That
adds more stuff to learn for users.
From my point of view, the "timestamp",
"headers" are something
like
"key"
and "value" that stores with the real data. So why
not define the
"timestamp" in the same way with "key" by using a
"timestamp.field"
connector option?
On the other side, the read-only metadata, such as
"offset",
shouldn't
be
defined as a normal column. So why not use the
existing computed
column
syntax for such metadata? Then we don't have the
query-sink
schema
problem.
So here is my proposal:
CREATE TABLE kafka_table (
id BIGINT,
name STRING,
col1 STRING,
col2 STRING,
ts TIMESTAMP(3) WITH LOCAL TIME ZONE, -- ts
is a normal
field,
so
can
be read and written.
offset AS SYSTEM_METADATA("offset")
) WITH (
'connector' = 'kafka',
'topic' = 'test-topic',
'key.fields' = 'id, name',
'key.format' = 'csv',
'value.format' = 'avro',
'timestamp.field' = 'ts' -- define the
mapping of Kafka
timestamp
);
INSERT INTO kafka_table
SELECT id, name, col1, col2, rowtime FROM
another_table;
I think this can solve all the problems without
introducing
any new
syntax.
The only minor disadvantage is that we separate the
definition
way/syntax
of read-only metadata and read-write fields.
However, I don't think this is a big problem.
Best,
Jark
On Wed, 9 Sep 2020 at 15:09, Timo Walther <
twal...@apache.org>
wrote:
Hi Kurt,
thanks for sharing your opinion. I'm totally up
for not reusing
computed
columns. I think Jark was a big supporter of this
syntax, @Jark
are
you
fine with this as well? The non-computed column
approach was
only
a
"slightly rejected alternative".
Furthermore, we would need to think about how
such a new design
influences the LIKE clause though.
However, we should still keep the `PERSISTED`
keyword as it
influences
the query->sink schema. If you look at the list
of metadata for
existing
connectors and formats, we currently offer only
two writable
metadata
fields. Otherwise, one would need to declare two
tables
whenever a
metadata columns is read (one for the source, one
for the sink).
This
can be quite inconvientient e.g. for just reading
the topic.
Regards,
Timo
On 09.09.20 08:52, Kurt Young wrote:
I also share the concern that reusing the
computed column
syntax
but
have
different semantics
would confuse users a lot.
Besides, I think metadata fields are
conceptually not the same
with
computed columns. The metadata
field is a connector specific thing and it only
contains the
information
that where does the field come
from (during source) or where does the field
need to write to
(during
sink). It's more similar with normal
fields, with assumption that all these fields
need going to the
data
part.
Thus I'm more lean to the rejected alternative
that Timo
mentioned.
And I
think we don't need the
PERSISTED keyword, SYSTEM_METADATA should be
enough.
During implementation, the framework only needs
to pass such
<field,
metadata field> information to the
connector, and the logic of handling such
fields inside the
connector
should be straightforward.
Regarding the downside Timo mentioned:
The disadvantage is that users cannot call
UDFs or parse
timestamps.
I think this is fairly simple to solve. Since
the metadata
field
isn't
a
computed column anymore, we can support
referencing such fields in the computed column.
For example:
CREATE TABLE kafka_table (
id BIGINT,
name STRING,
timestamp STRING
SYSTEM_METADATA("timestamp"), //
get the
timestamp
field from metadata
ts AS to_timestamp(timestamp) // normal
computed
column,
parse
the
string to TIMESTAMP type by using the metadata
field
) WITH (
...
)
Best,
Kurt
On Tue, Sep 8, 2020 at 11:57 PM Timo Walther
<twal...@apache.org
wrote:
Hi Leonard,
the only alternative I see is that we
introduce a concept that
is
completely different to computed columns.
This is also
mentioned
in
the
rejected alternative section of the FLIP.
Something like:
CREATE TABLE kafka_table (
id BIGINT,
name STRING,
timestamp INT
SYSTEM_METADATA("timestamp") PERSISTED,
headers MAP<STRING, BYTES>
SYSTEM_METADATA("headers")
PERSISTED
) WITH (
...
)
This way we would avoid confusion at all and
can easily map
columns
to
metadata columns. The disadvantage is that
users cannot call
UDFs
or
parse timestamps. This would need to be done
in a real
computed
column.
I'm happy about better alternatives.
Regards,
Timo
On 08.09.20 15:37, Leonard Xu wrote:
HI, Timo
Thanks for driving this FLIP.
Sorry but I have a concern about Writing
metadata via
DynamicTableSink
section:
CREATE TABLE kafka_table (
id BIGINT,
name STRING,
timestamp AS
CAST(SYSTEM_METADATA("timestamp") AS
BIGINT)
PERSISTED,
headers AS
CAST(SYSTEM_METADATA("headers") AS
MAP<STRING,
BYTES>)
PERSISTED
) WITH (
...
)
An insert statement could look like:
INSERT INTO kafka_table VALUES (
(1, "ABC", 1599133672, MAP('checksum',
computeChecksum(...)))
)
The proposed INERT syntax does not make
sense to me,
because it
contains
computed(generated) column.
Both SQL server and Postgresql do not allow
to insert
value to
computed
columns even they are persisted, this boke
the generated
column
semantics
and may confuse user much.
For SQL server computed column[1]:
column_name AS computed_column_expression
[ PERSISTED [ NOT
NULL ]
]...
NOTE: A computed column cannot be the
target of an INSERT or
UPDATE
statement.
For Postgresql generated column[2]:
height_in numeric GENERATED ALWAYS
AS (height_cm /
2.54)
STORED
NOTE: A generated column cannot be
written to directly. In
INSERT
or
UPDATE commands, a value cannot be specified
for a generated
column,
but
the keyword DEFAULT may be specified.
It shouldn't be allowed to set/update value
for generated
column
after
lookup the SQL 2016:
<insert statement> ::=
INSERT INTO <insertion target> <insert
columns and source>
If <contextually typed table value
constructor> CTTVC is
specified,
then every <contextually typed row
value constructor element> simply
contained in CTTVC whose
positionally
corresponding <column name>
in <insert column list> references a
column of which some
underlying
column is a generated column shall
be a <default specification>.
A <default specification> specifies the
default value of
some
associated item.
[1]
https://docs.microsoft.com/en-US/sql/t-sql/statements/alter-table-computed-column-definition-transact-sql?view=sql-server-ver15
<
https://docs.microsoft.com/en-US/sql/t-sql/statements/alter-table-computed-column-definition-transact-sql?view=sql-server-ver15
[2]
https://www.postgresql.org/docs/12/ddl-generated-columns.html
<
https://www.postgresql.org/docs/12/ddl-generated-columns.html>
在 2020年9月8日，17:31，Timo Walther <
twal...@apache.org>
写道：
Hi Jark,
according to Flink's and Calcite's
casting definition in
[1][2]
TIMESTAMP WITH LOCAL TIME ZONE should be
castable from BIGINT.
If
not,
we
will make it possible ;-)
I'm aware of
DeserializationSchema.getProducedType but I
think
that
this method is actually misplaced. The type
should rather be
passed
to
the
source itself.
For our Kafka SQL source, we will also
not use this method
because
the
Kafka source will add own metadata in
addition to the
DeserializationSchema. So
DeserializationSchema.getProducedType
will
never
be read.
For now I suggest to leave out the
`DataType` from
DecodingFormat.applyReadableMetadata. Also
because the
format's
physical
type is passed later in
`createRuntimeDecoder`. If
necessary, it
can
be
computed manually by consumedType + metadata
types. We will
provide
a
metadata utility class for that.
Regards,
Timo
[1]
https://github.com/apache/flink/blob/master/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/utils/LogicalTypeCasts.java#L200
[2]
https://github.com/apache/calcite/blob/master/core/src/main/java/org/apache/calcite/sql/type/SqlTypeCoercionRule.java#L254
On 08.09.20 10:52, Jark Wu wrote:
Hi Timo,
The updated CAST SYSTEM_METADATA
behavior sounds good to
me.
I
just
noticed
that a BIGINT can't be converted to
"TIMESTAMP(3) WITH
LOCAL
TIME
ZONE".
So maybe we need to support this, or
use "TIMESTAMP(3) WITH
LOCAL
TIME
ZONE" as the defined type of Kafka
timestamp? I think this
makes
sense,
because it represents the milli-seconds
since epoch.
Regarding "DeserializationSchema
doesn't need TypeInfo", I
don't
think
so.
The DeserializationSchema implements
ResultTypeQueryable,
thus
the
implementation needs to return an
output TypeInfo.
Besides, FlinkKafkaConsumer also
calls
DeserializationSchema.getProducedType as the produced
type
of
the
source function [1].
Best,
Jark
[1]:
https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumerBase.java#L1066
On Tue, 8 Sep 2020 at 16:35, Timo
Walther <
twal...@apache.org>
wrote:
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
