[core][flink][python] Support nullable primary keys

[JingsongLi]

What changed

• add the immutable primary-key.nullable table option, disabled by default
• normalize primary-key field nullability from the option and reject it on non-primary-key tables
• preserve nullable primary-key schemas through Flink Catalog without publishing a Flink SQL primary-key constraint
• support nullable key validation and nulls-first key ordering in PyPaimon
• document null-safe key equality and add core, Flink, and Python regression coverage

Why

Some upstream systems can produce null components in logical merge keys. Paimon previously rewrote every primary-key field to NOT NULL, preventing true primary-key tables from representing and consistently merging these records.

With the option enabled, null components use null-safe equality, so repeated records for keys such as (1, NULL) merge as the same key.

User impact

Existing tables remain unchanged because the option defaults to false. Tables that opt in can write, compact, read, update, and delete records whose primary-key components are null. The option is immutable after the table has snapshots.

For Flink, nullable primary keys are declared with the primary-key table option because Flink SQL primary-key constraints imply NOT NULL.

Validation

• Core: SchemaBuilderTest, SchemaValidationTest#testNullablePrimaryKeyRequiresPrimaryKeyTable, and PrimaryKeySimpleTableTest#testNullablePrimaryKey (7 tests)
• Flink 1.x: FlinkCatalogTest#testNullablePrimaryKeyExposedAsTableOption
• PyPaimon: targeted schema, merge-buffer, and reader tests (25 tests), plus Flake8
• Spotless checks for changed Java and Scala modules
• generated configuration docs via the paimon-docs generate-docs profile

[leaves12138]

I found a blocking Flink compatibility issue with the default streaming read path.

Because a nullable primary key is intentionally omitted from the Flink CatalogTable schema, Flink sees no primary-key constraint. However, BaseDataTableSource#getChangelogMode() still returns an upsert changelog for the default unbounded primary-key source (UPDATE_AFTER/DELETE without UPDATE_BEFORE). Flink therefore rejects even a plain streaming SELECT * FROM T during planning:

Table 'PAIMON.default.T' produces a changelog stream that contains UPDATE_AFTER but no UPDATE_BEFORE. This requires defining a primary key constraint on the table.

I reproduced this on this PR with a table created using 'primary-key'='id' and 'primary-key.nullable'='true'. The same query can be planned with changelog-producer=input, because that mode includes UPDATE_BEFORE, but the default streaming path is currently unusable. Please adapt the changelog/planner integration and add a Flink IT case, or explicitly constrain the feature if nullable keys cannot support default streaming reads.

There is also a cross-language compatibility gap worth addressing or documenting: current paimon-rust and paimon-cpp schema creation normalize primary-key fields to non-nullable unconditionally. Their comparators already handle nulls, but tables created or evolved through those clients cannot currently preserve the nullable-primary-key schema semantics. Please define the compatibility/versioning expectation and add cross-implementation coverage or coordinated follow-ups, especially for Java/PyPaimon/Rust/C++ table interchange.

[JingsongLi]

Thanks for catching this. Addressed in b3e73b725d.

For Flink, I chose to constrain the unsupported combination instead of advertising ChangelogMode.all(): the default Paimon source does not actually produce UPDATE_BEFORE, so claiming a full changelog would make downstream aggregations incorrect. A nullable-primary-key streaming read now fails early with a clear instruction to configure a full changelog producer (for example, changelog-producer=input). Insert-only reads such as merge-engine=first-row remain supported.

I added both a direct changelog-mode test and a Flink planner IT case covering the rejected default path and the supported input path, documented the limitation, and verified the tests with both the Flink 1.x and Flink 2.x profiles.

For Rust and C++, agreed that schema normalization and cross-implementation interchange coverage are still needed. I will coordinate those as follow-up work so this PR can remain focused on the Java/PyPaimon/Flink implementation.

[leaves12138]

Verified the follow-up fix in b3e73b725d.

The early rejection is the correct behavior for the default nullable-key streaming path: it avoids advertising UPDATE_BEFORE records that the source does not produce, while preserving batch reads, insert-only reads, and full-changelog producers. The new planner and changelog-mode tests pass locally with both the Flink 1.x and Flink 2.x profiles.

The Rust/C++ schema-normalization work can be handled in coordinated follow-ups as discussed. LGTM.