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
