[
https://issues.apache.org/jira/browse/FLINK-20105?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17230532#comment-17230532
]
Jark Wu edited comment on FLINK-20105 at 11/12/20, 11:18 AM:
-------------------------------------------------------------
Hi [~BenoitParis], thanks for trying out the new feature. I would like to
highlight some notes:
1. PK is not allowed on an append-only kafka table (it's almost impossible to
guarantee the primary key semantic on such table), e.g with the csv, json, avro
format.
2. PK is allowed on changelog kafka table, e.g. with debezium-json format.
3. The primary key fields must be NOT NULL. Primary key fields will be NOT NULL
by default.
4. If you want to have a primary key on the append-only table, you can add a
[deduplicate
operation|https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/queries.html#deduplication]
on the source table.
5. I would be more nature to temporal join an [upsert kafka
table|https://cwiki.apache.org/confluence/display/FLINK/FLIP-149%3A+Introduce+the+upsert-kafka+Connector]
which has the primary key semantic natively.
was (Author: jark):
Hi [~BenoitParis], thanks for trying out the new feature. I would like to
highlight some notes:
1. PK is not allowed on an append-only kafka table (it's almost impossible to
guarantee the primary key semantic on such table), e.g with the csv, json, avro
format.
2. PK is allowed on changelog kafka table, e.g. with debezium-json format.
3. The primary key fields must be NOT NULL. Primary key fields will be NOT NULL
by default.
4. If you want to have a primary key on the append-only table, you can add a
[deduplicate
operation](https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/queries.html#deduplication)
on the source table.
5. I would be more nature to temporal join an [upsert kafka
table](https://cwiki.apache.org/confluence/display/FLINK/FLIP-149%3A+Introduce+the+upsert-kafka+Connector)
which has the primary key semantic natively.
> Temporary Table does not work when Kafka is used as the versioned side
> (planner PK problem)
> -------------------------------------------------------------------------------------------
>
> Key: FLINK-20105
> URL: https://issues.apache.org/jira/browse/FLINK-20105
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka, Table SQL / Planner
> Affects Versions: 1.12.0
> Reporter: Benoît Paris
> Priority: Major
> Fix For: 1.12.0
>
> Attachments: flink-test-temporal-table.zip
>
>
>
> This is probably an early bug report, that I'm making before 0.12 is out.
> In 1.12-SNAPSHOT, doing this:
>
> {code:java}
> INSERT INTO output_table
> SELECT
> o.f_sequence * r.f_int,
> o.f_random_str
> FROM datagen_1 AS o
> LEFT JOIN input_2 FOR SYSTEM_TIME AS OF o.ts r
> ON o.f_random = r.f_int_pk{code}
>
> works when input_2 is build with datagen, but fails when data comes from
> kafka; yielding the following error that comes from planner code:
>
> {code:java}
> Type INT NOT NULL of table field 'f_int_pk' does not match with the physical
> type INT of the 'f_int_pk' field of the TableSource return type.{code}
>
> Included is code for a complete reproduction, with:
> * docker-compose file for ZooKeeper and Kafka (latest)
> * pom.xml
> * OK_TempTableSQLTestDatagen.java: it works with the datagen
> * KO_TempTableSQLTestKafka.java: fails with Kafka
> * KO_TempTableSQLTestKafkaNoPK.java: I tried to have no PK, it fails
> * KO_TempTableSQLTestKafkaNull.java: I tried with the PK being nullable, it
> fails
> * KO_TempTableSQLTestKafkaNullif.java: I tried with using the PK in a
> NULLIF(pk, '') as [advertised
> here|http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Inserting-nullable-data-into-NOT-NULL-columns-td34198.html],
> but if fails (loses PK powers)
> I just can't think of a workaround. I even tried to GROUP BY on the PK.
> From memory, the Temporal Table Function suffers from a similar problem; My
> usual workaround being to do a
> COALESCE(problematic_temp_table_function_primary_key, null), but it fails
> here as well (interestingly, it does not fail because of losing PK powers,
> but because of the NOT NULL planner difference).
> It seems like the same problem of having transformations of the same field
> being NULL and NOT NULL between planner transformations.
> ----
>
> This isssue is probably related to the ongoing FLIP-132 Temporal Table DDL
> and Temporal Table Join developments.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
