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