From eb9b05628fbbb461dc1750f7aa2942437fc21599 Mon Sep 17 00:00:00 2001 From: Keith Lee Date: Wed, 2 Sep 2026 08:13:06 +0100 Subject: [PATCH 1/2] [flink] Add regression test and docs for nested NOT NULL constraints Flink before 2.2 drops nested NOT NULL constraints during DDL resolution (FLINK-20539), so this adds a Flink 2.2/2.3 catalog ITCase for the Fluss round trip and documents the limitation for older Flink versions. --- .../flink/catalog/Flink22CatalogITCase.java | 86 +++++++++++++++++++ .../flink/catalog/Flink23CatalogITCase.java | 86 +++++++++++++++++++ website/docs/engine-flink/ddl.md | 6 ++ 3 files changed, 178 insertions(+) diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java index 5c2cd15843a..3a5bea7a1a7 100644 --- a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java +++ b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java @@ -17,7 +17,12 @@ package org.apache.fluss.flink.catalog; +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; import org.apache.fluss.testutils.common.MultiVersionTest; +import org.apache.fluss.types.RowType; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Schema; @@ -175,4 +180,85 @@ protected void addDefaultIndexKey(Schema.Builder schemaBuilder) { Schema currentSchema = schemaBuilder.build(); currentSchema.getPrimaryKey().ifPresent(pk -> schemaBuilder.index(pk.getColumnNames())); } + + @Test + @MultiVersionTest + void testCreateTableWithNestedNotNull() throws Exception { + // Flink before 2.2 drops nested NOT NULL constraints during DDL resolution (FLINK-20539). + String tableName = "nested_not_null_table"; + tEnv.executeSql( + String.format( + "create table %s (" + + " headers ARRAY>," + + " address ROW," + + " scores MAP>" + + ")", + tableName)); + + Schema expectedSchema = + Schema.newBuilder() + .column( + "headers", + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "name", DataTypes.STRING().notNull()), + DataTypes.FIELD( + "header_value", DataTypes.BYTES())))) + .column( + "address", + DataTypes.ROW( + DataTypes.FIELD("city", DataTypes.STRING().notNull()), + DataTypes.FIELD("zip", DataTypes.STRING()))) + .column( + "scores", + // map keys are always non-nullable in Fluss + DataTypes.MAP( + DataTypes.STRING().notNull(), + DataTypes.ROW( + DataTypes.FIELD( + "score", DataTypes.INT().notNull())))) + .build(); + CatalogTable table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName)); + assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema); + + RowType expectedRowType = + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "headers", + org.apache.fluss.types.DataTypes.ARRAY( + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "name", + org.apache.fluss.types.DataTypes.STRING() + .copy(false)), + org.apache.fluss.types.DataTypes.FIELD( + "header_value", + org.apache.fluss.types.DataTypes + .BYTES())))), + org.apache.fluss.types.DataTypes.FIELD( + "address", + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "city", + org.apache.fluss.types.DataTypes.STRING() + .copy(false)), + org.apache.fluss.types.DataTypes.FIELD( + "zip", org.apache.fluss.types.DataTypes.STRING()))), + org.apache.fluss.types.DataTypes.FIELD( + "scores", + org.apache.fluss.types.DataTypes.MAP( + org.apache.fluss.types.DataTypes.STRING().copy(false), + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "score", + org.apache.fluss.types.DataTypes.INT() + .copy(false)))))); + try (Connection conn = + ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig())) { + TableInfo tableInfo = + conn.getAdmin().getTableInfo(TablePath.of(DEFAULT_DB, tableName)).get(); + assertThat(tableInfo.getRowType()).isEqualTo(expectedRowType); + } + } } diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java index f7e9878d324..dd255c3af5a 100644 --- a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java +++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java @@ -17,7 +17,12 @@ package org.apache.fluss.flink.catalog; +import org.apache.fluss.client.Connection; +import org.apache.fluss.client.ConnectionFactory; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; import org.apache.fluss.testutils.common.MultiVersionTest; +import org.apache.fluss.types.RowType; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.api.Schema; @@ -175,4 +180,85 @@ protected void addDefaultIndexKey(Schema.Builder schemaBuilder) { Schema currentSchema = schemaBuilder.build(); currentSchema.getPrimaryKey().ifPresent(pk -> schemaBuilder.index(pk.getColumnNames())); } + + @Test + @MultiVersionTest + void testCreateTableWithNestedNotNull() throws Exception { + // Flink before 2.2 drops nested NOT NULL constraints during DDL resolution (FLINK-20539). + String tableName = "nested_not_null_table"; + tEnv.executeSql( + String.format( + "create table %s (" + + " headers ARRAY>," + + " address ROW," + + " scores MAP>" + + ")", + tableName)); + + Schema expectedSchema = + Schema.newBuilder() + .column( + "headers", + DataTypes.ARRAY( + DataTypes.ROW( + DataTypes.FIELD( + "name", DataTypes.STRING().notNull()), + DataTypes.FIELD( + "header_value", DataTypes.BYTES())))) + .column( + "address", + DataTypes.ROW( + DataTypes.FIELD("city", DataTypes.STRING().notNull()), + DataTypes.FIELD("zip", DataTypes.STRING()))) + .column( + "scores", + // map keys are always non-nullable in Fluss + DataTypes.MAP( + DataTypes.STRING().notNull(), + DataTypes.ROW( + DataTypes.FIELD( + "score", DataTypes.INT().notNull())))) + .build(); + CatalogTable table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName)); + assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema); + + RowType expectedRowType = + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "headers", + org.apache.fluss.types.DataTypes.ARRAY( + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "name", + org.apache.fluss.types.DataTypes.STRING() + .copy(false)), + org.apache.fluss.types.DataTypes.FIELD( + "header_value", + org.apache.fluss.types.DataTypes + .BYTES())))), + org.apache.fluss.types.DataTypes.FIELD( + "address", + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "city", + org.apache.fluss.types.DataTypes.STRING() + .copy(false)), + org.apache.fluss.types.DataTypes.FIELD( + "zip", org.apache.fluss.types.DataTypes.STRING()))), + org.apache.fluss.types.DataTypes.FIELD( + "scores", + org.apache.fluss.types.DataTypes.MAP( + org.apache.fluss.types.DataTypes.STRING().copy(false), + org.apache.fluss.types.DataTypes.ROW( + org.apache.fluss.types.DataTypes.FIELD( + "score", + org.apache.fluss.types.DataTypes.INT() + .copy(false)))))); + try (Connection conn = + ConnectionFactory.createConnection(FLUSS_CLUSTER_EXTENSION.getClientConfig())) { + TableInfo tableInfo = + conn.getAdmin().getTableInfo(TablePath.of(DEFAULT_DB, tableName)).get(); + assertThat(tableInfo.getRowType()).isEqualTo(expectedRowType); + } + } } diff --git a/website/docs/engine-flink/ddl.md b/website/docs/engine-flink/ddl.md index 37ca0090a9a..9970a840f2c 100644 --- a/website/docs/engine-flink/ddl.md +++ b/website/docs/engine-flink/ddl.md @@ -181,6 +181,12 @@ CREATE TABLE my_auto_part_log_table ( For more details about Auto Partitioned (Primary Key/Log) Table, refer to [Auto Partitioning](table-design/data-distribution/partitioning.md#auto-partitioning). +### Nested NOT NULL Constraints + +`NOT NULL` constraints on top-level columns are always preserved. `NOT NULL` constraints on fields nested inside a nullable `ROW` (for example `ROW` or `ARRAY>`) are only preserved on Flink 2.2 and later. Flink 2.1 and earlier drop nested `NOT NULL` constraints while resolving the DDL statement (see [FLINK-20539](https://issues.apache.org/jira/browse/FLINK-20539)), so the table is silently created with those fields nullable. + +If you need nested `NOT NULL` constraints on Flink 2.1 and earlier, declare the enclosing `ROW` column itself as `NOT NULL`, for example `address ROW NOT NULL`. On Flink 2.2 and later, keep `table.legacy-nested-row-nullability` at its default value of `false`, otherwise nested constraints are dropped again. + ### Options The supported option in `WITH` parameters when creating a table are listed in [Connector Options](engine-flink/options.md) page. From 32251c614da0ecbf37d88ab60cc0e11ad10824ae Mon Sep 17 00:00:00 2001 From: Keith Lee Date: Wed, 2 Sep 2026 09:09:24 +0100 Subject: [PATCH 2/2] [flink] Document nested NOT NULL workaround for ARRAY/MAP and write-time enforcement --- website/docs/engine-flink/ddl.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/website/docs/engine-flink/ddl.md b/website/docs/engine-flink/ddl.md index 9970a840f2c..3b58cc7ed90 100644 --- a/website/docs/engine-flink/ddl.md +++ b/website/docs/engine-flink/ddl.md @@ -185,7 +185,9 @@ For more details about Auto Partitioned (Primary Key/Log) Table, refer to [Auto `NOT NULL` constraints on top-level columns are always preserved. `NOT NULL` constraints on fields nested inside a nullable `ROW` (for example `ROW` or `ARRAY>`) are only preserved on Flink 2.2 and later. Flink 2.1 and earlier drop nested `NOT NULL` constraints while resolving the DDL statement (see [FLINK-20539](https://issues.apache.org/jira/browse/FLINK-20539)), so the table is silently created with those fields nullable. -If you need nested `NOT NULL` constraints on Flink 2.1 and earlier, declare the enclosing `ROW` column itself as `NOT NULL`, for example `address ROW NOT NULL`. On Flink 2.2 and later, keep `table.legacy-nested-row-nullability` at its default value of `false`, otherwise nested constraints are dropped again. +On Flink 2.1 and earlier, a nested `NOT NULL` constraint survives only if every enclosing `ROW` on its path is itself `NOT NULL`. Declare the enclosing `ROW` column as `NOT NULL`, for example `address ROW NOT NULL`. For `ARRAY` and `MAP` values, declare the element `ROW` as `NOT NULL`, for example `headers ARRAY NOT NULL>`. + +On Flink 2.2 and later, keep `table.legacy-nested-row-nullability` at its default value of `false`, otherwise nested constraints are dropped again. Fluss does not enforce nested `NOT NULL` constraints on write, and Flink's `table.exec.sink.nested-constraint-enforcer` defaults to `IGNORE`, so a null written into a nested `NOT NULL` field is accepted and later fails every read of that record. Set `table.exec.sink.nested-constraint-enforcer` to `ROWS_AND_COLLECTIONS` to reject such writes. ### Options