Hi, Timo Thanks for the update I have a minor suggestion about the debezium metadata key, Could we use the original debezium key rather than import new key?
debezium-json.schema => debezium-json.schema debezium-json.ingestion-timestamp => debezium-json.ts_ms debezium-json.source.database => debezium-json.source.db debezium-json.source.schema => debezium-json.source.schema debezium-json.source.table => debezium-json.source.table debezium-json.source.timestamp => debezium-json.source.ts_ms debezium-json.source.properties => debezium-json.source MAP<STRING, STRING> User who familiar with debezium will understand the key easier, and the key syntax is more json-path like. HDYT? The other part looks really good to me. Regards, Leonard > 在 2020年9月10日，18:26，Aljoscha Krettek <aljos...@apache.org> 写道： > > I've only been watching this from the sidelines but that latest proposal > looks very good to me! > > Aljoscha > > On 10.09.20 12:20, Kurt Young wrote: >> 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 >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> >>>>>>>>>> >>>>>>>> >>>>>>>> >>>>>>> >>>>>> >>>>> >>>> >>> >
