The new syntax looks good to me. Best, Kurt
On Thu, Sep 10, 2020 at 5:57 PM Jark Wu <imj...@gmail.com> wrote: > Hi Timo, > > I have one minor suggestion. > Maybe the default data type of `timestamp` can be `TIMESTAMP(3) WITH > LOCAL TIME ZONE`, because this is the type that users want to use, this can > avoid unnecessary casting. > Besides, currently, the bigint is casted to timestamp in seconds, so the > implicit cast may not work... > > I don't have other objections. But maybe we should wait for the > opinion from @Kurt for the new syntax. > > Best, > Jark > > > On Thu, 10 Sep 2020 at 16:21, Danny Chan <yuzhao....@gmail.com> wrote: > >> 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 >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > > >> > > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > >> > > > > > > > >> > > > > > > >> > > > > > > >> > > > > >> > > > > >> > > > >> > > >> > >> >
