This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 061a992946 [core][flink][python] Support nullable primary keys (#9094)
061a992946 is described below
commit 061a9929469b1bb9a045088ccc17998201d48c74
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Aug 7 17:58:00 2026 +0800
[core][flink][python] Support nullable primary keys (#9094)
---
docs/docs/primary-key-table/index.md | 30 ++
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 17 ++
.../main/java/org/apache/paimon/schema/Schema.java | 25 +-
.../org/apache/paimon/schema/SchemaValidation.java | 6 +
.../apache/paimon/schema/SchemaBuilderTest.java | 16 +
.../apache/paimon/schema/SchemaValidationTest.java | 21 ++
.../paimon/table/PrimaryKeySimpleTableTest.java | 71 +++++
.../apache/paimon/flink/sink/cdc/CdcSchema.java | 6 +-
.../java/org/apache/paimon/flink/FlinkCatalog.java | 9 +-
.../paimon/flink/source/BaseDataTableSource.java | 9 +-
.../org/apache/paimon/flink/ChangelogModeTest.java | 24 +-
.../org/apache/paimon/flink/FlinkCatalogTest.java | 18 ++
.../paimon/flink/NullablePrimaryKeyITCase.java | 325 +++++++++++++++++++++
.../pypaimon/common/options/core_options.py | 18 ++
paimon-python/pypaimon/schema/schema.py | 18 +-
.../pypaimon/tests/reader_primary_key_test.py | 38 +++
paimon-python/pypaimon/tests/table_schema_test.py | 28 ++
.../pypaimon/tests/test_write_merge_buffer.py | 29 ++
paimon-python/pypaimon/write/writer/data_writer.py | 3 +-
.../pypaimon/write/writer/key_value_data_writer.py | 5 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 2 +-
22 files changed, 699 insertions(+), 25 deletions(-)
diff --git a/docs/docs/primary-key-table/index.md
b/docs/docs/primary-key-table/index.md
index ec48c1277e..dba23325cd 100644
--- a/docs/docs/primary-key-table/index.md
+++ b/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.
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 0e399478c8..434ad80139 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1344,6 +1344,12 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<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>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 0ad4beb001..bc1a2102bc 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -1478,6 +1478,14 @@ public class CoreOptions implements Serializable {
.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 class CoreOptions implements Serializable {
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
diff --git a/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java
b/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java
index 9842b5db9f..5b8be18e45 100644
--- a/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java
+++ b/paimon-api/src/main/java/org/apache/paimon/schema/Schema.java
@@ -88,7 +88,12 @@ public class 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 class Schema {
}
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 @@ public class Schema {
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 class Schema {
/**
* 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 class Schema {
/**
* 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
*/
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index fa28403e82..1fd428a851 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -144,6 +144,12 @@ public class SchemaValidation {
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);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaBuilderTest.java
index 50552db619..a21c491f2e 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaBuilderTest.java
+++ b/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 class SchemaBuilderTest {
"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 =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index e0f141680a..417a00b93b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -38,6 +38,7 @@ import static java.util.Collections.emptyList;
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 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
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(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
index 3ea3628fa5..0fa383dd4d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java
@@ -137,6 +137,7 @@ import static
org.apache.paimon.CoreOptions.MergeEngine.AGGREGATE;
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 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** 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 =
diff --git
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSchema.java
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSchema.java
index 229dcc69bb..31abc780f3 100644
---
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSchema.java
+++
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSchema.java
@@ -156,7 +156,8 @@ public class CdcSchema implements Serializable {
/**
* 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 class CdcSchema implements Serializable {
/**
* 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
*/
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
index 12b18e5d70..fb3cd0bcab 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -1017,9 +1017,14 @@ public class FlinkCatalog extends AbstractCatalog {
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();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
index 827b7b04b5..40a8fe1125 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java
@@ -144,8 +144,9 @@ public abstract class BaseDataTableSource extends
FlinkTableSource
}
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 abstract class BaseDataTableSource extends
FlinkTableSource
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();
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
index 6a068b2ee3..eb7536ae12 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
@@ -43,7 +43,9 @@ import java.util.Collections;
import static org.apache.paimon.CoreOptions.CHANGELOG_PRODUCER;
import static org.apache.paimon.CoreOptions.ChangelogProducer.INPUT;
import static org.apache.paimon.CoreOptions.ChangelogProducer.LOOKUP;
+import static org.apache.paimon.CoreOptions.PRIMARY_KEY_NULLABLE;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Test for changelog mode with flink source and sink. */
public class ChangelogModeTest {
@@ -59,8 +61,7 @@ public class ChangelogModeTest {
path = new Path(temp.toUri().toString());
}
- private void test(Options options, ChangelogMode expectSource,
ChangelogMode expectSink)
- throws Exception {
+ private FileStoreTable createTable(Options options) throws Exception {
new SchemaManager(LocalFileIO.create(), path)
.createTable(
new Schema(
@@ -69,7 +70,12 @@ public class ChangelogModeTest {
Collections.singletonList("f0"),
options.toMap(),
""));
- FileStoreTable table =
FileStoreTableFactory.create(LocalFileIO.create(), path);
+ return FileStoreTableFactory.create(LocalFileIO.create(), path);
+ }
+
+ private void test(Options options, ChangelogMode expectSource,
ChangelogMode expectSink)
+ throws Exception {
+ FileStoreTable table = createTable(options);
DataTableSource source = new DataTableSource(identifier, table, true,
null);
assertThat(source.getChangelogMode()).isEqualTo(expectSource);
@@ -78,6 +84,18 @@ public class ChangelogModeTest {
assertThat(sink.getChangelogMode(ChangelogMode.all())).isEqualTo(expectSink);
}
+ @Test
+ public void testNullablePrimaryKey() throws Exception {
+ Options options = new Options();
+ options.set(PRIMARY_KEY_NULLABLE, true);
+ FileStoreTable table = createTable(options);
+
+ DataTableSource source = new DataTableSource(identifier, table, true,
null);
+ assertThatThrownBy(source::getChangelogMode)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("nullable primary keys require a full
changelog");
+ }
+
@Test
public void testDefault() throws Exception {
test(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
index 5322d81eb4..2bb9c580e5 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkCatalogTest.java
@@ -742,6 +742,24 @@ public class FlinkCatalogTest extends FlinkCatalogTestBase
{
checkCreateTable(path1, catalogTable, (CatalogTable)
catalog.getTable(path1));
}
+ @Test
+ void testNullablePrimaryKeyExposedAsTableOption() throws Exception {
+ catalog.createDatabase(path1.getDatabaseName(), null, false);
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PRIMARY_KEY.key(), "first");
+ options.put(CoreOptions.PRIMARY_KEY_NULLABLE.key(), "true");
+
+ catalog.createTable(path1, createTable(options), false);
+ DataCatalogTable table = (DataCatalogTable) catalog.getTable(path1);
+
+ assertThat(table.table().primaryKeys()).containsExactly("first");
+ assertThat(table.table().rowType().getTypeAt(0).isNullable()).isTrue();
+ assertThat(table.getUnresolvedSchema().getPrimaryKey()).isEmpty();
+ assertThat(table.getOptions())
+ .containsEntry(CoreOptions.PRIMARY_KEY.key(), "first")
+ .containsEntry(CoreOptions.PRIMARY_KEY_NULLABLE.key(), "true");
+ }
+
@Test
void testBuildPaimonTableWithCustomScheme() throws Exception {
catalog.createDatabase(path1.getDatabaseName(), null, false);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NullablePrimaryKeyITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NullablePrimaryKeyITCase.java
new file mode 100644
index 0000000000..9c312816a6
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/NullablePrimaryKeyITCase.java
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for nullable primary-key tables. */
+@Timeout(180)
+public class NullablePrimaryKeyITCase extends CatalogITCaseBase {
+
+ private void createTable(
+ String tableName, String fields, String primaryKey, String...
additionalOptions) {
+ String options =
+ Arrays.stream(additionalOptions)
+ .map(option -> ", " + option)
+ .collect(Collectors.joining());
+ sql(
+ "CREATE TABLE %s (%s) WITH ("
+ + "'primary-key' = '%s', "
+ + "'primary-key.nullable' = 'true', "
+ + "'bucket' = '1'%s)",
+ tableName, fields, primaryKey, options);
+ }
+
+ @Test
+ public void testCatalogKeepsNullablePrimaryKeyAsTableOption() throws
Exception {
+ createTable("T", "id INT, v STRING", "id");
+
+ DataCatalogTable catalogTable = (DataCatalogTable) table("T");
+
assertThat(catalogTable.getUnresolvedSchema().getPrimaryKey()).isEmpty();
+ assertThat(catalogTable.getOptions())
+ .containsEntry(CoreOptions.PRIMARY_KEY.key(), "id")
+ .containsEntry(CoreOptions.PRIMARY_KEY_NULLABLE.key(), "true");
+ assertThat(catalogTable.table().primaryKeys()).containsExactly("id");
+
assertThat(catalogTable.table().rowType().getTypeAt(0).isNullable()).isTrue();
+ }
+
+ @Test
+ public void testSingleNullablePrimaryKeyDeduplicateAcrossCommits() {
+ createTable("T", "id INT, v STRING", "id");
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v1'), (1,
'one-v1')");
+ sql("INSERT INTO T VALUES " + "(CAST(NULL AS INT), 'null-v2'), (1,
'one-v2'), (2, 'two')");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(
+ Row.of(null, "null-v2"), Row.of(1, "one-v2"),
Row.of(2, "two"));
+ assertThat(sql("SELECT v FROM T WHERE id IS
NULL")).containsExactly(Row.of("null-v2"));
+ }
+
+ @Test
+ public void testDuplicateNullablePrimaryKeyWithinSingleCommit() {
+ createTable("T", "id INT, v STRING", "id");
+
+ sql(
+ "INSERT INTO T VALUES "
+ + "(CAST(NULL AS INT), 'null-v1'), "
+ + "(CAST(NULL AS INT), 'null-v2'), "
+ + "(1, 'one')");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(null, "null-v2"), Row.of(1,
"one"));
+ }
+
+ @Test
+ public void testCompositeNullablePrimaryKeyDeduplicateAcrossCommits() {
+ createTable("T", "k1 INT, k2 INT, v STRING", "k1,k2");
+
+ sql(
+ "INSERT INTO T VALUES "
+ + "(CAST(NULL AS INT), CAST(NULL AS INT), 'nn-v1'), "
+ + "(CAST(NULL AS INT), 1, 'n1-v1'), "
+ + "(1, CAST(NULL AS INT), '1n-v1'), "
+ + "(1, 1, '11-v1')");
+ sql(
+ "INSERT INTO T VALUES "
+ + "(CAST(NULL AS INT), CAST(NULL AS INT), 'nn-v2'), "
+ + "(CAST(NULL AS INT), 1, 'n1-v2'), "
+ + "(1, CAST(NULL AS INT), '1n-v2'), "
+ + "(1, 1, '11-v2')");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(
+ Row.of(null, null, "nn-v2"),
+ Row.of(null, 1, "n1-v2"),
+ Row.of(1, null, "1n-v2"),
+ Row.of(1, 1, "11-v2"));
+ }
+
+ @Test
+ public void testPartitionedNullablePrimaryKey() {
+ createTable("T", "dt STRING, id INT, v STRING", "dt,id", "'partition'
= 'dt'");
+
+ sql(
+ "INSERT INTO T VALUES "
+ + "(CAST(NULL AS STRING), 1, 'null-partition-v1'), "
+ + "('A', CAST(NULL AS INT), 'a-null-v1')");
+ sql(
+ "INSERT INTO T VALUES "
+ + "(CAST(NULL AS STRING), 1, 'null-partition-v2'), "
+ + "('A', CAST(NULL AS INT), 'a-null-v2'), "
+ + "('B', 2, 'b-two')");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(
+ Row.of(null, 1, "null-partition-v2"),
+ Row.of("A", null, "a-null-v2"),
+ Row.of("B", 2, "b-two"));
+ assertThat(sql("SELECT id, v FROM T WHERE dt IS NULL"))
+ .containsExactly(Row.of(1, "null-partition-v2"));
+ }
+
+ @Test
+ public void testSequenceFieldWithNullablePrimaryKey() {
+ createTable("T", "id INT, v STRING, seq BIGINT", "id",
"'sequence.field' = 'seq'");
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'newer', 2)");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'older', 1), (1, 'one',
1)");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(null, "newer", 2L),
Row.of(1, "one", 1L));
+ }
+
+ @Test
+ public void testPartialUpdateWithNullablePrimaryKey() {
+ createTable("T", "id INT, a INT, b STRING", "id", "'merge-engine' =
'partial-update'");
+
+ sql(
+ "INSERT INTO T VALUES "
+ + "(CAST(NULL AS INT), 1, CAST(NULL AS STRING)), "
+ + "(1, 10, 'one')");
+ sql("INSERT INTO T VALUES " + "(CAST(NULL AS INT), CAST(NULL AS INT),
'merged')");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(null, 1, "merged"),
Row.of(1, 10, "one"));
+ }
+
+ @Test
+ public void testAggregationWithNullablePrimaryKey() {
+ createTable(
+ "T",
+ "id INT, total BIGINT",
+ "id",
+ "'merge-engine' = 'aggregation'",
+ "'fields.total.aggregate-function' = 'sum'");
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 1), (1, 10)");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 2), (1, 20)");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(null, 3L), Row.of(1, 30L));
+ }
+
+ @Test
+ public void testFirstRowWithNullablePrimaryKey() {
+ createTable("T", "id INT, v STRING", "id", "'merge-engine' =
'first-row'");
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'first'), (1,
'one-first')");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'ignored'), (1,
'one-ignored')");
+
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(null, "first"), Row.of(1,
"one-first"));
+ }
+
+ @Test
+ public void testSchemaEvolutionPreservesNullablePrimaryKey() throws
Exception {
+ createTable("T", "id INT, v STRING", "id");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'before')");
+
+ sql("ALTER TABLE T ADD extra INT");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'after', 2)");
+
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(null,
"after", 2));
+ assertThat(((DataCatalogTable)
table("T")).table().rowType().getTypeAt(0).isNullable())
+ .isTrue();
+ }
+
+ @Test
+ public void testUpdateAndDeleteNullablePrimaryKey() {
+ createTable("T", "id INT, v STRING", "id");
+ sql("INSERT INTO T VALUES " + "(CAST(NULL AS INT), 'null-v1'), (1,
'one'), (2, 'two')");
+
+ sql("UPDATE T SET v = 'null-v2' WHERE id IS NULL");
+ assertThat(sql("SELECT * FROM T WHERE id IS NULL"))
+ .containsExactly(Row.of(null, "null-v2"));
+
+ sql("DELETE FROM T WHERE v = 'null-v2' OR id = 1");
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(2, "two"));
+ }
+
+ @Test
+ public void testCompactionPreservesNullablePrimaryKey() {
+ createTable(
+ "T",
+ "id INT, v STRING",
+ "id",
+ "'write-only' = 'true'",
+ "'num-sorted-run.compaction-trigger' = '10'");
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v1'), (1,
'one-v1')");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v2'), (1,
'one-v2')");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v3'), (2, 'two')");
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql("CALL sys.compact(`table` => 'default.T', compact_strategy =>
'full')");
+
+
assertThat(findLatestSnapshot("T").commitKind()).isEqualTo(Snapshot.CommitKind.COMPACT);
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(
+ Row.of(null, "null-v3"), Row.of(1, "one-v2"),
Row.of(2, "two"));
+ }
+
+ @Test
+ public void testNullablePrimaryKeyOptionIsImmutableAfterWrite() {
+ createTable("T", "id INT, v STRING", "id");
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-value')");
+
+ assertThatThrownBy(() -> sql("ALTER TABLE T SET (" +
"'primary-key.nullable' = 'false')"))
+ .hasStackTraceContaining("Change 'primary-key.nullable' is not
supported yet.");
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(null,
"null-value"));
+ }
+
+ @Test
+ public void testStreamingReadRequiresFullChangelog() {
+ createTable("T_DEFAULT", "id INT, v STRING", "id");
+ createTable("T_INPUT", "id INT, v STRING", "id", "'changelog-producer'
= 'input'");
+
+ assertThatThrownBy(() -> sEnv.explainSql("SELECT * FROM T_DEFAULT"))
+ .hasStackTraceContaining("nullable primary keys require a full
changelog");
+ assertThatCode(() -> sEnv.explainSql("SELECT * FROM
T_INPUT")).doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testLookupChangelogStreamingReadWithNullablePrimaryKey()
throws Exception {
+ createTable(
+ "T",
+ "id INT, v STRING",
+ "id",
+ "'changelog-producer' = 'lookup'",
+ "'continuous.discovery-interval' = '100ms'");
+
+ try (BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT
* FROM T")) {
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v1'), (1,
'one')");
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(Row.of(null, "null-v1"),
Row.of(1, "one"));
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v2')");
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, null, "null-v1"),
+ Row.ofKind(RowKind.UPDATE_AFTER, null, "null-v2"));
+ }
+ }
+
+ @Test
+ public void testInputChangelogStreamingReadWithNullablePrimaryKey() throws
Exception {
+ createTable(
+ "T",
+ "id INT, v STRING",
+ "id",
+ "'changelog-producer' = 'input'",
+ "'continuous.discovery-interval' = '100ms'");
+
+ try (BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT
* FROM T")) {
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v1')");
+ assertThat(iterator.collect(1)).containsExactly(Row.of(null,
"null-v1"));
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v2')");
+ assertThat(iterator.collect(1)).containsExactly(Row.of(null,
"null-v2"));
+ }
+ }
+
+ @Test
+ public void testFullCompactionStreamingReadWithNullablePrimaryKey() throws
Exception {
+ createTable(
+ "T",
+ "id INT, v STRING",
+ "id",
+ "'changelog-producer' = 'full-compaction'",
+ "'changelog-producer.compaction-interval' = '1s'",
+ "'continuous.discovery-interval' = '100ms'");
+
+ try (BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT
* FROM T")) {
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v1'), (1,
'one')");
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(Row.of(null, "null-v1"),
Row.of(1, "one"));
+
+ sql("INSERT INTO T VALUES (CAST(NULL AS INT), 'null-v2')");
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, null, "null-v1"),
+ Row.ofKind(RowKind.UPDATE_AFTER, null, "null-v2"));
+ }
+ }
+}
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index c51225baec..e2cc0ddef2 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -124,6 +124,7 @@ class CoreOptions:
"partial-update.remove-record-on-sequence-group",
"rowkind.field",
"primary-key",
+ "primary-key.nullable",
"partition",
"dynamic-bucket.initial-buckets",
"force-lookup",
@@ -208,6 +209,16 @@ class CoreOptions:
)
)
+ PRIMARY_KEY_NULLABLE: ConfigOption[bool] = (
+ ConfigOptions.key("primary-key.nullable")
+ .boolean_type()
+ .default_value(False)
+ .with_description(
+ "Whether primary key fields can contain null values. Null values "
+ "use null-safe equality when records are merged."
+ )
+ )
+
DYNAMIC_BUCKET_TARGET_ROW_NUM: ConfigOption[int] = (
ConfigOptions.key("dynamic-bucket.target-row-num")
.int_type()
@@ -1082,6 +1093,10 @@ class CoreOptions:
def from_dict(options: dict) -> 'CoreOptions':
return CoreOptions(Options(options))
+ @staticmethod
+ def primary_key_nullable_from_dict(options: dict) -> bool:
+ return Options(options).get(CoreOptions.PRIMARY_KEY_NULLABLE)
+
def path(self, default=None):
return self.options.get(CoreOptions.PATH, default)
@@ -1100,6 +1115,9 @@ class CoreOptions:
def bucket_key(self, default=None):
return self.options.get(CoreOptions.BUCKET_KEY, default)
+ def primary_key_nullable(self, default=None):
+ return self.options.get(CoreOptions.PRIMARY_KEY_NULLABLE, default)
+
def dynamic_bucket_target_row_num(self, default=None):
return self.options.get(CoreOptions.DYNAMIC_BUCKET_TARGET_ROW_NUM,
default)
diff --git a/paimon-python/pypaimon/schema/schema.py
b/paimon-python/pypaimon/schema/schema.py
index 3354afebc8..1fd231168d 100644
--- a/paimon-python/pypaimon/schema/schema.py
+++ b/paimon-python/pypaimon/schema/schema.py
@@ -48,6 +48,17 @@ class Schema:
self.options = options if options is not None else {}
self.comment = comment
+ primary_key_nullable =
CoreOptions.primary_key_nullable_from_dict(self.options)
+ if primary_key_nullable and not self.primary_keys:
+ raise ValueError(
+ "Option 'primary-key.nullable' can only be enabled for a table
"
+ "with primary keys."
+ )
+ pk_set = set(self.primary_keys)
+ for field in self.fields:
+ if field.name in pk_set:
+ field.type.nullable = primary_key_nullable
+
changelog_producer =
self.options.get(CoreOptions.CHANGELOG_PRODUCER.key(), 'none')
if changelog_producer != 'none' and not self.primary_keys:
raise ValueError(
@@ -62,13 +73,6 @@ class Schema:
# Convert PyArrow schema to Paimon fields
fields = PyarrowFieldParser.to_paimon_schema(pa_schema)
- # Primary key fields must be NOT NULL
- pk_set = set(primary_keys) if primary_keys else set()
- if pk_set:
- for field in fields:
- if field.name in pk_set:
- field.type.nullable = False
-
# Check if Vector type with dedicated file format
vector_names = [
field.name for field in fields
diff --git a/paimon-python/pypaimon/tests/reader_primary_key_test.py
b/paimon-python/pypaimon/tests/reader_primary_key_test.py
index 7cae0a77c7..1ed2148099 100644
--- a/paimon-python/pypaimon/tests/reader_primary_key_test.py
+++ b/paimon-python/pypaimon/tests/reader_primary_key_test.py
@@ -236,6 +236,44 @@ class PkReaderTest(unittest.TestCase):
}, schema=self.pa_schema)
self.assertEqual(actual, expected)
+ def test_nullable_primary_key(self):
+ nullable_schema = pa.schema([
+ pa.field('id', pa.int64()),
+ pa.field('value', pa.string()),
+ ])
+ schema = Schema.from_pyarrow_schema(
+ nullable_schema,
+ primary_keys=['id'],
+ options={
+ 'bucket': '3',
+ CoreOptions.PRIMARY_KEY_NULLABLE.key(): 'true',
+ },
+ )
+ self.catalog.create_table('default.test_nullable_pk', schema, False)
+ table = self.catalog.get_table('default.test_nullable_pk')
+
+ for rows in [
+ [{'id': None, 'value': 'old'}],
+ [{'id': None, 'value': 'new'},
+ {'id': 1, 'value': 'one'},
+ {'id': 2, 'value': 'two'}]]:
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ writer.write_arrow(pa.Table.from_pylist(rows,
schema=nullable_schema))
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ actual = self._read_test_table(table.new_read_builder()).to_pylist()
+ actual.sort(key=lambda row: (-1 if row['id'] is None else row['id']))
+ self.assertEqual(
+ actual,
+ [{'id': None, 'value': 'new'},
+ {'id': 1, 'value': 'one'},
+ {'id': 2, 'value': 'two'}],
+ )
+
def test_pk_reader_with_filter(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'],
diff --git a/paimon-python/pypaimon/tests/table_schema_test.py
b/paimon-python/pypaimon/tests/table_schema_test.py
index 19854f7af6..71f426e7ca 100644
--- a/paimon-python/pypaimon/tests/table_schema_test.py
+++ b/paimon-python/pypaimon/tests/table_schema_test.py
@@ -17,7 +17,11 @@
import unittest
+import pyarrow as pa
+
+from pypaimon.common.options.core_options import CoreOptions
from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.schema.schema import Schema
from pypaimon.schema.table_schema import TableSchema
@@ -88,5 +92,29 @@ class TableSchemaBucketKeysTest(unittest.TestCase):
self.assertEqual(schema.bucket_keys, ['id'])
+class SchemaPrimaryKeyNullabilityTest(unittest.TestCase):
+
+ def test_primary_key_is_not_nullable_by_default(self):
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([pa.field('id', pa.int64())]), primary_keys=['id'])
+ self.assertFalse(schema.fields[0].type.nullable)
+
+ def test_primary_key_nullable_option_overrides_not_null_input(self):
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([pa.field('id', pa.int64(), nullable=False)]),
+ primary_keys=['id'],
+ options={CoreOptions.PRIMARY_KEY_NULLABLE.key(): 'true'},
+ )
+ self.assertTrue(schema.fields[0].type.nullable)
+
+ def test_primary_key_nullable_requires_primary_key_table(self):
+ with self.assertRaisesRegex(
+ ValueError, "can only be enabled for a table with primary
keys"):
+ Schema.from_pyarrow_schema(
+ pa.schema([pa.field('id', pa.int64())]),
+ options={CoreOptions.PRIMARY_KEY_NULLABLE.key(): 'true'},
+ )
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/tests/test_write_merge_buffer.py
b/paimon-python/pypaimon/tests/test_write_merge_buffer.py
index 0b86b59138..031d1a3a02 100644
--- a/paimon-python/pypaimon/tests/test_write_merge_buffer.py
+++ b/paimon-python/pypaimon/tests/test_write_merge_buffer.py
@@ -51,6 +51,15 @@ _SCHEMA = pa.schema([
pa.field('b', pa.string()),
])
+_NULLABLE_PK_SCHEMA = pa.schema([
+ pa.field('_KEY_id', pa.int64()),
+ pa.field('_SEQUENCE_NUMBER', pa.int64(), nullable=False),
+ pa.field('_VALUE_KIND', pa.int8(), nullable=False),
+ pa.field('id', pa.int64()),
+ pa.field('a', pa.string()),
+ pa.field('b', pa.string()),
+])
+
def _row(pk, seq, a, b):
return {
@@ -122,6 +131,26 @@ class WriteMergeBufferTest(unittest.TestCase):
_row(3, 3, 'C', None)],
)
+ def test_dedupe_nullable_pk_uses_null_safe_equality_and_nulls_first(self):
+ writer = _Harness(DeduplicateMergeFunction())
+ writer.pending_data = pa.Table.from_pylist(
+ [_row(2, 1, 'two', None),
+ _row(None, 2, 'null-old', None),
+ _row(1, 3, 'one', None),
+ _row(None, 4, 'null-new', None)],
+ schema=_NULLABLE_PK_SCHEMA,
+ )
+
+ writer._flush_all()
+
+ self.assertEqual(len(writer.written_chunks), 1)
+ self.assertEqual(
+ writer.written_chunks[0].to_pylist(),
+ [_row(None, 4, 'null-new', None),
+ _row(1, 3, 'one', None),
+ _row(2, 1, 'two', None)],
+ )
+
# -- partial-update ---------------------------------------------------
def _partial_update(self):
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 3a191bc3f7..e34127ddaf 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -283,7 +283,8 @@ class DataWriter(ABC):
}
key_fields = self.trimmed_primary_keys_fields
key_stats = self._collect_value_stats(data, key_fields, column_stats)
- if not all(count == 0 for count in key_stats.null_counts):
+ if not self.options.primary_key_nullable() and not all(
+ count == 0 for count in key_stats.null_counts):
raise RuntimeError("Primary key should not be null")
value_fields = stats_fields if value_stats_enabled else []
diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index 5200dd2078..86b6ad365a 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -278,5 +278,8 @@ class KeyValueDataWriter(DataWriter):
if '_SEQUENCE_NUMBER' in data.schema.names:
sort_keys.append(('_SEQUENCE_NUMBER', 'ascending'))
- sorted_indices = pc.sort_indices(data, sort_keys=sort_keys)
+ # Java MergeTree comparators order null keys first. Keep
Python-written files in the same
+ # order so their key ranges and sorted-run invariants are
interoperable with Java readers.
+ sorted_indices = pc.sort_indices(
+ data, sort_keys=sort_keys, null_placement='at_start')
return data.take(sorted_indices)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 1f0b1b75b0..7b376c2f79 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -157,7 +157,7 @@ case class PaimonScan(
.map(Expressions.identity)
.map {
sortExpr =>
- // Primary key can not be null, the null ordering is no matter.
+ // Paimon MergeTree comparators and Spark ascending expressions both
order nulls first.
Expressions.sort(sortExpr, SortDirection.ASCENDING)
}
.toArray
