(paimon) branch master updated: [flink] Keep NOT NULL of VARIANT columns created through Flink (#10617)

[lzljs3620320]

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 0b715f6f64 [flink] Keep NOT NULL of VARIANT columns created through 
Flink (#10617)
0b715f6f64 is described below

commit 0b715f6f64f3c7650b099abbe2598c78bd8d96b6
Author: YangJie <yangji...@baidu.com>
AuthorDate: Sun Oct 18 04:48:58 2026 -0400

    [flink] Keep NOT NULL of VARIANT columns created through Flink (#10617)
---
 .../org/apache/paimon/flink/LogicalTypeToDataType.java     |  2 +-
 .../java/org/apache/paimon/flink/CatalogTableITCase.java   | 14 ++++++++++++++
 2 files changed, 15 insertions(+), 1 deletion(-)

diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeToDataType.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeToDataType.java
index 3a54c975d1..3e3f2f7810 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeToDataType.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/LogicalTypeToDataType.java
@@ -179,7 +179,7 @@ public class LogicalTypeToDataType extends 
LogicalTypeDefaultVisitor<DataType> {
     }
 
     public DataType visit(VariantType variantType) {
-        return new org.apache.paimon.types.VariantType();
+        return new 
org.apache.paimon.types.VariantType(variantType.isNullable());
     }
 
     @Override
diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index 92cf7e5385..a7f5c3e5e2 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -1346,6 +1346,20 @@ public class CatalogTableITCase extends 
CatalogITCaseBase {
         assertThat(rows).containsExactlyInAnyOrder(expectedRow);
     }
 
+    @Test
+    @EnabledIf("isFlink2_1OrAbove")
+    void testVariantNotNull() throws Exception {
+        sql("CREATE TABLE t (id INT, v VARIANT NOT NULL)");
+
+        assertThatThrownBy(() -> sql("INSERT INTO t VALUES (1, CAST(NULL AS 
VARIANT))"))
+                .hasRootCauseMessage(
+                        "Column 'v' is NOT NULL, however, a null value is 
being written into it. "
+                                + "You can set job configuration 
'table.exec.sink.not-null-enforcer'='DROP' "
+                                + "to suppress this exception and drop such 
records silently.");
+        assertThat(sql("SELECT * FROM t")).isEmpty();
+        
assertThat(paimonTable("t").rowType().getTypeAt(1).isNullable()).isFalse();
+    }
+
     private void innerTestReadOptimizedTableAndCheckData(String 
insertTableName) {
         // full compaction will always be performed at the end of batch jobs, 
as long as
         // full-compaction.delta-commits is set, regardless of its value