This is an automated email from the ASF dual-hosted git repository.
raminqaf pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6e194b0167e [FLINK-41362][table] Reject VARIANT in primary key columns
(#29801)
6e194b0167e is described below
commit 6e194b0167e9a70c94b776cf43c4c6459633ec4b
Author: Ramin Gharib <[email protected]>
AuthorDate: Thu Nov 5 21:58:36 2026 +0100
[FLINK-41362][table] Reject VARIANT in primary key columns (#29801)
Changelog normalization and upsert sinks key their state by the primary key
and match keys by their binary encoding. One VARIANT value has many encodings,
for example {"a":1,"b":2} and {"b":2,"a":1}, so an upsert source with a VARIANT
primary key kept two rows for one key and dropped deletes.
DefaultSchemaResolver now rejects a primary key column whose type is not
comparable, the rule that GROUP BY and joins use since FLINK-40910.
Generated-by: Claude Code 2.1.318 (Claude Opus 5.5)
---
docs/content.zh/docs/sql/reference/data-types.md | 3 +++
docs/content.zh/docs/sql/reference/ddl/create.md | 1 +
docs/content/docs/sql/reference/data-types.md | 3 +++
docs/content/docs/sql/reference/ddl/create.md | 1 +
.../org/apache/flink/table/catalog/DefaultSchemaResolver.java | 11 +++++++++++
.../org/apache/flink/table/catalog/SchemaResolutionTest.java | 4 ++++
6 files changed, 23 insertions(+)
diff --git a/docs/content.zh/docs/sql/reference/data-types.md
b/docs/content.zh/docs/sql/reference/data-types.md
index 2d0ec548413..cb310a235fc 100644
--- a/docs/content.zh/docs/sql/reference/data-types.md
+++ b/docs/content.zh/docs/sql/reference/data-types.md
@@ -1785,6 +1785,9 @@ SELECT * FROM t WHERE v['a'] = PARSE_JSON('1')
-- fa
SELECT * FROM t WHERE CAST(v['a'] AS INT) = 1 --
compares the number
```
+A `PRIMARY KEY` column cannot be a `VARIANT` or a type that contains one
either. Declare the key
+column with a concrete type.
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
diff --git a/docs/content.zh/docs/sql/reference/ddl/create.md
b/docs/content.zh/docs/sql/reference/ddl/create.md
index 28e6411e0de..0f4c57409ed 100644
--- a/docs/content.zh/docs/sql/reference/ddl/create.md
+++ b/docs/content.zh/docs/sql/reference/ddl/create.md
@@ -410,6 +410,7 @@ CREATE TABLE Orders (
主键用作 Flink 优化的一种提示信息。主键限制表明一张表或视图的某个（些）列是唯一的并且不包含 Null 值。
主键声明的列都是非 nullable 的。因此主键可以被用作表行级别的唯一标识。
+A primary key column also cannot be a [`VARIANT`]({{< ref
"docs/sql/reference/data-types" >}}#variant) or a type that contains one,
because these types have no equality.
主键可以和列的定义一起声明，也可以独立声明为表的限制属性，不管是哪种方式，主键都不可以重复定义，否则 Flink 会报错。
diff --git a/docs/content/docs/sql/reference/data-types.md
b/docs/content/docs/sql/reference/data-types.md
index 02c3352262d..b0d6b95fd82 100644
--- a/docs/content/docs/sql/reference/data-types.md
+++ b/docs/content/docs/sql/reference/data-types.md
@@ -1793,6 +1793,9 @@ SELECT * FROM t WHERE v['a'] = PARSE_JSON('1')
-- fa
SELECT * FROM t WHERE CAST(v['a'] AS INT) = 1 --
compares the number
```
+A `PRIMARY KEY` column cannot be a `VARIANT` or a type that contains one
either. Declare the key
+column with a concrete type.
+
**Declaration**
{{< tabs "25c30432-8460-441d-a036-9416d8202882" >}}
diff --git a/docs/content/docs/sql/reference/ddl/create.md
b/docs/content/docs/sql/reference/ddl/create.md
index 0a1c711f2a6..b77795d59d6 100644
--- a/docs/content/docs/sql/reference/ddl/create.md
+++ b/docs/content/docs/sql/reference/ddl/create.md
@@ -405,6 +405,7 @@ CREATE TABLE Orders (
Primary key constraint is a hint for Flink to leverage for optimizations. It
tells that a column or a set of columns of a table or a view are unique and
they **do not** contain null.
Neither of columns in a primary can be nullable. Primary key therefore
uniquely identify a row in a table.
+A primary key column also cannot be a [`VARIANT`]({{< ref
"docs/sql/reference/data-types" >}}#variant) or a type that contains one,
because these types have no equality.
Primary key constraint can be either declared along with a column definition
(a column constraint) or as a single line (a table constraint).
For both cases, it should only be declared as a singleton. If you define
multiple primary key constraints at the same time, an exception would be thrown.
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/DefaultSchemaResolver.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/DefaultSchemaResolver.java
index 4ce5177832d..000ab84a849 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/DefaultSchemaResolver.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/DefaultSchemaResolver.java
@@ -59,6 +59,7 @@ import java.util.stream.Stream;
import static org.apache.flink.table.expressions.ApiExpressionUtils.localRef;
import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.canBeTimeAttributeType;
import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.getPrecision;
+import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isComparableType;
import static
org.apache.flink.table.types.logical.utils.LogicalTypeChecks.isProctimeAttribute;
import static
org.apache.flink.table.types.utils.DataTypeUtils.replaceLogicalType;
@@ -385,6 +386,16 @@ class DefaultSchemaResolver implements SchemaResolver {
}
final LogicalType columnType =
column.getDataType().getLogicalType();
+ // changelog normalization and upsert sinks key their state by the
primary key
+ if (!isComparableType(columnType)) {
+ throw new ValidationException(
+ String.format(
+ "Invalid primary key '%s'. Column '%s' of type
%s has no equality.",
+ primaryKey.getName(),
+ columnName,
+ columnType.copy(true).asSummaryString()));
+ }
+
if (columnType.isNullable()) {
throw new ValidationException(
String.format(
diff --git
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaResolutionTest.java
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaResolutionTest.java
index 3519ade2907..1e86f5a134e 100644
---
a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaResolutionTest.java
+++
b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/catalog/SchemaResolutionTest.java
@@ -319,6 +319,10 @@ class SchemaResolutionTest {
Schema.newBuilder().column("id",
DataTypes.INT()).primaryKey("id", "id").build(),
"Invalid primary key 'PK_id_id'. A primary key must not
contain duplicate columns. Found: [id]");
+ testError(
+ Schema.newBuilder().column("v",
DataTypes.VARIANT()).primaryKey("v").build(),
+ "Invalid primary key 'PK_v'. Column 'v' of type VARIANT has no
equality.");
+
// indexes
testError(
