Skip to content

[flink] Add regression test and docs for nested NOT NULL constraints #4198

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
leekeiabstraction wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[flink] Add regression test and docs for nested NOT NULL constraints #4198

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eb9b056
[flink] Add regression test and docs for nested NOT NULL constraints
leekeiabstraction Sep 2, 2026
32251c6
[flink] Document nested NOT NULL workaround for ARRAY/MAP and write-t…
leekeiabstraction Sep 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 86 additions & 0 deletions ...nk/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<ROW<name STRING NOT NULL, header_value BYTES>>,"
+ " address ROW<city STRING NOT NULL, zip STRING>,"
+ " scores MAP<STRING, ROW<score INT NOT NULL>>"
+ ")",
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);
}
}
}
86 changes: 86 additions & 0 deletions ...nk/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<ROW<name STRING NOT NULL, header_value BYTES>>,"
+ " address ROW<city STRING NOT NULL, zip STRING>,"
+ " scores MAP<STRING, ROW<score INT NOT NULL>>"
+ ")",
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);
}
}
}
8 changes: 8 additions & 0 deletions website/docs/engine-flink/ddl.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,14 @@ 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<name STRING NOT NULL>` or `ARRAY<ROW<name STRING NOT NULL>>`) 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.

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<city STRING NOT NULL, zip STRING> NOT NULL`. For `ARRAY` and `MAP` values, declare the element `ROW` as `NOT NULL`, for example `headers ARRAY<ROW<name STRING NOT NULL> 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

The supported option in `WITH` parameters when creating a table are listed in [Connector Options](engine-flink/options.md) page.
Expand Down
Loading