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

docs/docs/primary-key-table/index.md

@@ -30,6 +30,36 @@ Primary keys consist of a set of columns that contain unique values for each rec
 sorting the primary key within each bucket, allowing users to achieve high performance by applying filtering conditions
 on the primary key. See [CREATE TABLE](../flink/sql-ddl#create-table).
 
+## Nullable Primary Keys
+
+Primary key fields are `NOT NULL` by default. Set `primary-key.nullable` to `true` when a source
+system can produce null key components:
+
+```sql
+CREATE TABLE orders (
+    order_id BIGINT,
+    payload STRING
+) WITH (
+    'primary-key' = 'order_id',
+    'primary-key.nullable' = 'true'
+);
+```
+
+Null key components use null-safe equality. For example, two records whose key is `(1, NULL)` are
+treated as the same key and are merged by the configured merge engine. The option is disabled by
+default and cannot be changed after the table has snapshots.
+
+In Flink, define a nullable Paimon primary key with the `primary-key` table option as shown above.
+The standard SQL `PRIMARY KEY` constraint implies `NOT NULL`, so Paimon does not expose a nullable
+key as a Flink SQL primary-key constraint.
+
+Flink streaming reads that emit updates or deletes require a full changelog producer, for example
+`changelog-producer=input`. The default `changelog-producer=none` produces an upsert changelog,
+which Flink can normalize only when the table exposes a SQL primary-key constraint. Because a
+nullable key cannot be exposed as that constraint, Paimon rejects this streaming-read combination
+instead of producing an invalid Flink plan. Insert-only streaming reads, such as tables using the
+`first-row` merge engine, are not affected.
+
 ## Bucket
 
 Unpartitioned tables, or partitions in partitioned tables, are sub-divided into buckets, to provide extra structure to the data that may be used for more efficient querying.

docs/generated/core_configuration.html

@@ -1344,6 +1344,12 @@
             <td>String</td>
             <td>Define primary key by table options, cannot define primary key on DDL and table options at the same time.</td>
         </tr>
+        <tr>
+            <td><h5>primary-key.nullable</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether primary key fields can contain null values. Null values use null-safe equality when records are merged.</td>
+        </tr>
         <tr>
             <td><h5>query-auth.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1478,6 +1478,14 @@ public String toString() {
                     .withDescription(
                             "Define primary key by table options, cannot define primary key on DDL and table options at the same time.");
 
+    @Immutable
+    public static final ConfigOption<Boolean> PRIMARY_KEY_NULLABLE =
+            key("primary-key.nullable")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether primary key fields can contain null values. Null values use null-safe equality when records are merged.");
+
     @Immutable
     public static final ConfigOption<String> PARTITION =
             key("partition")
@@ -3145,6 +3153,15 @@ public String fieldsDefaultFunc() {
         return options.get(FIELDS_DEFAULT_AGG_FUNC);
     }
 
+    public boolean primaryKeyNullable() {
+        return options.get(PRIMARY_KEY_NULLABLE);
+    }
+
+    public static boolean primaryKeyNullable(Map<String, String> options) {
+        return Options.fromMap(options)
+                .getBoolean(PRIMARY_KEY_NULLABLE.key(), PRIMARY_KEY_NULLABLE.defaultValue());
+    }
+
     public static String createCommitUser(Options options) {
         String commitUserPrefix = options.get(COMMIT_USER_PREFIX);
         return commitUserPrefix == null

paimon-api/src/main/java/org/apache/paimon/schema/Schema.java

@@ -88,7 +88,12 @@ public Schema(
         this.options = new HashMap<>(options);
         this.partitionKeys = normalizePartitionKeys(partitionKeys);
         this.primaryKeys = normalizePrimaryKeys(primaryKeys);
-        this.fields = normalizeFields(fields, this.primaryKeys, this.partitionKeys);
+        this.fields =
+                normalizeFields(
+                        fields,
+                        this.primaryKeys,
+                        this.partitionKeys,
+                        CoreOptions.primaryKeyNullable(this.options));
         this.comment = comment;
     }
 
@@ -126,7 +131,10 @@ public Schema copy(RowType rowType) {
     }
 
     private static List<DataField> normalizeFields(
-            List<DataField> fields, List<String> primaryKeys, List<String> partitionKeys) {
+            List<DataField> fields,
+            List<String> primaryKeys,
+            List<String> partitionKeys,
+            boolean primaryKeyNullable) {
         List<String> fieldNames = fields.stream().map(DataField::name).collect(Collectors.toList());
 
         Set<String> duplicateColumns = duplicateFields(fieldNames);
@@ -165,16 +173,17 @@ private static List<DataField> normalizeFields(
                 fieldNames,
                 primaryKeys);
 
-        // primary key should not nullable
+        // SQL engines may implicitly make primary key fields NOT NULL. Normalize them to the
+        // nullability selected by the table option so all engines expose the same table schema.
         Set<String> pkSet = new HashSet<>(primaryKeys);
         List<DataField> newFields = new ArrayList<>();
         for (DataField field : fields) {
-            if (pkSet.contains(field.name()) && field.type().isNullable()) {
+            if (pkSet.contains(field.name()) && field.type().isNullable() != primaryKeyNullable) {
                 newFields.add(
                         new DataField(
                                 field.id(),
                                 field.name(),
-                                field.type().copy(false),
+                                field.type().copy(primaryKeyNullable),
                                 field.description(),
                                 field.defaultValue()));
             } else {
@@ -345,7 +354,8 @@ public Builder partitionKeys(List<String> columnNames) {
 
         /**
          * Declares a primary key constraint for a set of given columns. Primary key uniquely
-         * identify a row in a table. Neither of columns in a primary can be nullable.
+         * identify a row in a table. By default, primary key columns are not nullable. Set {@link
+         * CoreOptions#PRIMARY_KEY_NULLABLE} to allow null values.
          *
          * @param columnNames columns that form a unique primary key
          */
@@ -355,7 +365,8 @@ public Builder primaryKey(String... columnNames) {
 
         /**
          * Declares a primary key constraint for a set of given columns. Primary key uniquely
-         * identify a row in a table. Neither of columns in a primary can be nullable.
+         * identify a row in a table. By default, primary key columns are not nullable. Set {@link
+         * CoreOptions#PRIMARY_KEY_NULLABLE} to allow null values.
          *
          * @param columnNames columns that form a unique primary key
          */

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -144,6 +144,12 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
 
         validateOnlyContainPrimitiveType(schema.fields(), schema.primaryKeys(), "primary key");
         validateOnlyContainPrimitiveType(schema.fields(), schema.partitionKeys(), "partition");
+        if (options.primaryKeyNullable() && schema.primaryKeys().isEmpty()) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Option '%s' can only be enabled for a table with primary keys.",
+                            CoreOptions.PRIMARY_KEY_NULLABLE.key()));
+        }
 
         validateBucket(schema, options);
 

paimon-core/src/test/java/org/apache/paimon/schema/SchemaBuilderTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.schema;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.types.DataTypes;
 
 import org.junit.jupiter.api.Test;
@@ -65,6 +66,21 @@ public void testDuplicatePartitionKeys() {
                                 "Partition key constraint [id, id] must not contain duplicate columns. Found: [id]"));
     }
 
+    @Test
+    public void testPrimaryKeyNullability() {
+        Schema defaultSchema =
+                Schema.newBuilder().column("id", DataTypes.INT()).primaryKey("id").build();
+        assertThat(defaultSchema.fields().get(0).type().isNullable()).isFalse();
+
+        Schema nullableSchema =
+                Schema.newBuilder()
+                        .column("id", DataTypes.INT().notNull())
+                        .primaryKey("id")
+                        .option(CoreOptions.PRIMARY_KEY_NULLABLE.key(), "true")
+                        .build();
+        assertThat(nullableSchema.fields().get(0).type().isNullable()).isTrue();
+    }
+
     @Test
     public void testHighestFieldId() {
         Schema.Builder builder =

paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java

@@ -38,6 +38,7 @@
 import static java.util.Collections.singletonList;
 import static org.apache.paimon.CoreOptions.BUCKET;
 import static org.apache.paimon.CoreOptions.DATA_EVOLUTION_ENABLED;
+import static org.apache.paimon.CoreOptions.PRIMARY_KEY_NULLABLE;
 import static org.apache.paimon.CoreOptions.SCAN_SNAPSHOT_ID;
 import static org.apache.paimon.CoreOptions.VECTOR_FIELD;
 import static org.apache.paimon.CoreOptions.VECTOR_FILE_FORMAT;
@@ -50,6 +51,26 @@
 
 class SchemaValidationTest {
 
+    @Test
+    void testNullablePrimaryKeyRequiresPrimaryKeyTable() {
+        Map<String, String> options = new HashMap<>();
+        options.put(PRIMARY_KEY_NULLABLE.key(), "true");
+        TableSchema schema =
+                new TableSchema(
+                        1,
+                        singletonList(new DataField(0, "f0", DataTypes.INT())),
+                        10,
+                        emptyList(),
+                        emptyList(),
+                        options,
+                        "");
+
+        assertThatThrownBy(() -> validateTableSchema(schema))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessage(
+                        "Option 'primary-key.nullable' can only be enabled for a table with primary keys.");
+    }
+
     private void validateTableSchemaExec(Map<String, String> options) {
         List<DataField> fields =
                 Arrays.asList(

paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java

@@ -137,6 +137,7 @@
 import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE;
 import static org.apache.paimon.CoreOptions.MergeEngine.FIRST_ROW;
 import static org.apache.paimon.CoreOptions.MergeEngine.PARTIAL_UPDATE;
+import static org.apache.paimon.CoreOptions.PRIMARY_KEY_NULLABLE;
 import static org.apache.paimon.CoreOptions.SNAPSHOT_EXPIRE_LIMIT;
 import static org.apache.paimon.CoreOptions.SOURCE_SPLIT_OPEN_FILE_COST;
 import static org.apache.paimon.CoreOptions.SOURCE_SPLIT_TARGET_SIZE;
@@ -156,6 +157,76 @@
 /** Tests for {@link PrimaryKeyFileStoreTable}. */
 public class PrimaryKeySimpleTableTest extends SimpleTableTestBase {
 
+    @Test
+    public void testNullablePrimaryKey() throws Exception {
+        FileStoreTable table =
+                createFileStoreTable(
+                        options -> {
+                            options.set(BUCKET, 3);
+                            options.set(PRIMARY_KEY_NULLABLE, true);
+                        });
+        assertThat(table.rowType().getTypeAt(0).isNullable()).isTrue();
+        assertThat(table.rowType().getTypeAt(1).isNullable()).isTrue();
+
+        try (StreamTableWrite write = table.newWrite(commitUser);
+                StreamTableCommit commit = table.newCommit(commitUser)) {
+            write.write(rowData(null, 1, 10L));
+            write.write(rowData(1, null, 20L));
+            write.write(rowData(null, null, 30L));
+            commit.commit(0, write.prepareCommit(true, 0));
+
+            write.write(rowData(null, 1, 11L));
+            write.write(rowData(1, null, 21L));
+            write.write(rowData(null, null, 31L));
+            write.write(rowData(1, 2, 22L));
+            commit.commit(1, write.prepareCommit(true, 1));
+        }
+
+        Function<InternalRow, String> toString =
+                row ->
+                        (row.isNullAt(0) ? "null" : String.valueOf(row.getInt(0)))
+                                + "|"
+                                + (row.isNullAt(1) ? "null" : String.valueOf(row.getInt(1)))
+                                + "|"
+                                + row.getLong(2);
+        assertThat(
+                        getResult(
+                                table.newRead(),
+                                toSplits(table.newSnapshotReader().read().dataSplits()),
+                                toString))
+                .containsExactlyInAnyOrder("null|null|31", "null|1|11", "1|null|21", "1|2|22");
+
+        List<DataSplit> splits = table.newSnapshotReader().read().dataSplits();
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            for (DataSplit split : splits) {
+                write.compact(split.partition(), split.bucket(), true);
+            }
+            commit.commit(write.prepareCommit());
+        }
+        assertThat(
+                        getResult(
+                                table.newRead(),
+                                toSplits(table.newSnapshotReader().read().dataSplits()),
+                                toString))
+                .containsExactlyInAnyOrder("null|null|31", "null|1|11", "1|null|21", "1|2|22");
+
+        try (StreamTableWrite write = table.newWrite(commitUser);
+                StreamTableCommit commit = table.newCommit(commitUser)) {
+            write.write(rowDataWithKind(RowKind.DELETE, null, 1, 0L));
+            write.write(rowDataWithKind(RowKind.DELETE, 1, null, 0L));
+            write.write(rowDataWithKind(RowKind.DELETE, null, null, 0L));
+            commit.commit(2, write.prepareCommit(true, 2));
+        }
+        assertThat(
+                        getResult(
+                                table.newRead(),
+                                toSplits(table.newSnapshotReader().read().dataSplits()),
+                                toString))
+                .containsExactly("1|2|22");
+    }
+
     @Test
     public void testPostponeBucketWithManyPartitions() throws Exception {
         FileStoreTable table =

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSchema.java

@@ -156,7 +156,8 @@ public Builder column(String columnName, DataType dataType, @Nullable String des
 
         /**
          * Declares a primary key constraint for a set of given columns. Primary key uniquely
-         * identify a row in a table. Neither of columns in a primary can be nullable.
+         * identify a row in a table. Primary key columns are not nullable unless the target table
+         * enables {@code primary-key.nullable}.
          *
          * @param columnNames columns that form a unique primary key
          */
@@ -166,7 +167,8 @@ public Builder primaryKey(String... columnNames) {
 
         /**
          * Declares a primary key constraint for a set of given columns. Primary key uniquely
-         * identify a row in a table. Neither of columns in a primary can be nullable.
+         * identify a row in a table. Primary key columns are not nullable unless the target table
+         * enables {@code primary-key.nullable}.
          *
          * @param columnNames columns that form a unique primary key
          */

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java

@@ -1017,9 +1017,14 @@ private CatalogBaseTable toCatalogTable(Table table) {
             deserializeWatermarkSpec(newOptions, builder);
         }
 
-        // add primary keys
-        if (!table.primaryKeys().isEmpty()) {
+        // Flink primary-key constraints imply NOT NULL. For a nullable Paimon primary key, expose
+        // the key through the Paimon table option instead so Flink preserves the physical column
+        // nullability while the underlying table remains a primary-key table.
+        boolean nullablePrimaryKey = CoreOptions.primaryKeyNullable(newOptions);
+        if (!table.primaryKeys().isEmpty() && !nullablePrimaryKey) {
             builder.primaryKey(table.primaryKeys());
+        } else if (!table.primaryKeys().isEmpty()) {
+            newOptions.put(CoreOptions.PRIMARY_KEY.key(), String.join(",", table.primaryKeys()));
         }
 
         org.apache.flink.table.api.Schema schema = builder.build();

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java

@@ -144,8 +144,9 @@ public ChangelogMode getChangelogMode() {
         }
 
         Options options = Options.fromMap(table.options());
+        CoreOptions coreOptions = new CoreOptions(options);
 
-        if (new CoreOptions(options).mergeEngine() == FIRST_ROW) {
+        if (coreOptions.mergeEngine() == FIRST_ROW) {
             return ChangelogMode.insertOnly();
         }
 
@@ -157,6 +158,12 @@ public ChangelogMode getChangelogMode() {
             return ChangelogMode.all();
         }
 
+        if (coreOptions.primaryKeyNullable()) {
+            throw new UnsupportedOperationException(
+                    "Flink streaming reads with nullable primary keys require a full changelog. "
+                            + "Configure 'changelog-producer' to a value other than 'none'.");
+        }
+
         return ChangelogMode.upsert();
     }