[improve] Cdc 2.4.0 has supported no primary key table, and we can optimize the relevant code logic.

[2kyyds]

Search before asking

• [X] I had searched in the issues and found no similar feature requirement.

Description

Improve the code

Are you willing to submit a PR?

• [X] Yes I am willing to submit a PR!

Code of Conduct

• [X] I agree to follow this project's Code of Conduct

[github-actions[bot]]

Thanks for coming to the community to ask questions

[gtk96]

Creating a table without primary key in Flink SQL can be done in the following ways:

1. Do not specify the primary key

The primary key is not specified when the table is created directly:

CREATE TABLE mytable (
  id INT, 
  name STRING
) WITH (
  'connector' = 'kafka',
  ...
);

This will create a table without primary keys.

2. Specify the primary key as PROCTIME()

CREATE TABLE mytable (
  id INT,
  name STRING,
  PRIMARY KEY(PROCTIME()) 
) WITH (
  'connector' = 'kafka',
  ...
);

Using processing time as the primary key is equivalent to no primary key.

3. Set primary key as virtual column

CREATE TABLE mytable (
  id INT,
  name STRING, 
  PRIMARY KEY(`_col0`) 
) WITH (
  'connector' = 'kafka',
  ...
);

Use a virtual column as the primary key.

[gtk96]

Failed to create a statement without a primary key table