[Future][Connector-V2]Support the automatic creation of non-primary key table

[hawk9821]

Purpose of this pull request

1. Support the automatic creation of non-primary key table ( append-only table) when Source is the primary key table
2. Fix losing the primary key when loading the Schema in Paimon Source

Does this PR introduce any user-facing change?

How was this patch tested?

UT: PaimonCatalogPrimaryTest
E2E: PaimonSinkCDCIT#testChangelogLookup

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config
• Update the release-note.

[Copilot]

Pull Request Overview

This pull request introduces support for the automatic creation of non-primary key (append-only) tables when the source is a primary key table and fixes the issue of losing primary key information when loading the schema in Paimon Source.

• Adds a new unit test to verify primary key preservation.
• Updates schema conversion and sink configuration to handle the non-primary key scenario.
• Introduces a new error code and configuration option for non-primary key validation.

Reviewed Changes

Copilot reviewed 6 out of 7 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
seatunnel-connectors-v2/connector-paimon/src/test/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalogPrimaryTest.java	Adds tests to verify primary key behavior.
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/utils/SchemaUtil.java	Updates schema conversion to clear primary keys when non-primary key is enabled.
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/exception/PaimonConnectorErrorCode.java	Introduces a new error code for non-primary key check errors.
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonSinkOptions.java	Adds a new NON_PRIMARY_KEY option with a description that appears ambiguous.
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonSinkConfig.java	Implements validation to throw an exception when non-primary key is true but primary keys are provided.
seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalog.java	Adjusts catalog conversion logic and sets primary key and constraint key based on the schema.

Files not reviewed (1)

• seatunnel-e2e/seatunnel-connector-v2-e2e/connector-paimon-e2e/src/test/resources/changelog_paimon_to_paimon.conf: Language not supported

Comments suppressed due to low confidence (2)

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/config/PaimonSinkConfig.java:55

• The exception for the non-primary key check is thrown with an empty error message. Provide a descriptive message to aid debugging.

throw new PaimonConnectorException(PaimonConnectorErrorCode.NON_PRIMARY_KEY_CHECK_ERROR, "");

seatunnel-connectors-v2/connector-paimon/src/main/java/org/apache/seatunnel/connectors/seatunnel/paimon/catalog/PaimonCatalog.java:276

• Typo in variable name 'primaryKyes'. Consider renaming it to 'primaryKeys' for clarity.

List<String> primaryKyes = schema.primaryKeys();

[Copilot]

The description for the NON_PRIMARY_KEY option does not match its functionality (clearing primary keys). Consider updating the description to reflect that setting this option to true disables primary key handling.

[Hisoka-X]

I think if a field is a primary key, it must be unique. So do not put it into constraintKey.

Wrong click