[jira] [Comment Edited] (FLINK-30386) Column constraint lacks primary key not enforced check

["Jane Chan (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-30386?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17646395#comment-17646395 ] 

Jane Chan edited comment on FLINK-30386 at 12/13/22 2:18 AM:
-------------------------------------------------------------

Re [~dwysakowicz] . The `NOT ENFORCED` check only performs on the table constraint but skips the column constraint. As a result, `PRIMARY KEY(a)` will throw an exception, but `a BIGINT NOT NULL PRIMARY KEY` is allowed.

I think we should also throw the exception when encountering `a BIGINT NOT NULL PRIMARY KEY` and inform users to use `a BIGINT NOT NULL PRIMARY KEY NOT ENFORCED` instead.


was (Author: qingyue):
Re [~dwysakowicz] . The `NOT ENFORCED` check only performs on the table constraint but skips the column constraint. As a result, `PRIMARY KEY(a)` will throw exception, but `a BIGINT NOT NULL PRIMARY KEY` is allowed.

> Column constraint lacks primary key not enforced check
> ------------------------------------------------------
>
>                 Key: FLINK-30386
>                 URL: https://issues.apache.org/jira/browse/FLINK-30386
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / API
>    Affects Versions: 1.16.0, 1.15.2, 1.15.3
>            Reporter: Jane Chan
>            Priority: Major
>
> Currently, only table constraint performs the enforced check. Not sure if it is by design or a bug.
> The following case can be reproduced on Flink 1.16.0, 1.15.3, and 1.15.2. I think the earlier version might also reveal it.
> {code:sql}
> Flink SQL> create table T (f0 int not null primary key, f1 string) with ('connector' = 'datagen');
> [INFO] Execute statement succeed.
> Flink SQL> explain select * from T;
> == Abstract Syntax Tree ==
> LogicalProject(f0=[$0], f1=[$1])
> +- LogicalTableScan(table=[[default_catalog, default_database, T]])
> == Optimized Physical Plan ==
> TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1])
> == Optimized Execution Plan ==
> TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1])
> Flink SQL> create table S (f0 int not null, f1 string, primary key(f0)) with ('connector' = 'datagen');
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.table.api.ValidationException: Flink doesn't support ENFORCED mode for PRIMARY KEY constraint. ENFORCED/NOT ENFORCED  controls if the constraint checks are performed on the incoming/outgoing data. Flink does not own the data therefore the only supported mode is the NOT ENFORCED mode
> {code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)