[Bug] Flink Catalog loses nested NOT NULL constraints in complex types

[gyang94]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

Environment

• Flink 1.20.3
• Fluss Flink connector built from the current development branch

Description

When a Fluss table is created through Flink SQL/Fluss Catalog, NOT NULL declared on fields nested inside ROW, ARRAY<ROW<...>>, or similar complex types is lost.

For example:

CREATE TABLE nested_nullability_repro (
  headers ARRAY<ROW<name STRING NOT NULL, header_value BYTES>>
) WITH (
  'bucket.num' = '1'
);

Expected:

headers ARRAY<ROW<name STRING NOT NULL, header_value BYTES>>

Actual schema retrieved from Fluss metadata:

headers ARRAY<ROW<name STRING, header_value BYTES>>

Top-level NOT NULL constraints are retained; the problem affects nested fields.

Minimal reproduction without Fluss

The same loss can be reproduced in Flink 1.20.3 with the default in-memory Catalog, before the schema reaches FlinkCatalog:

TableEnvironment tableEnvironment =
        TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());

tableEnvironment.executeSql(
        "CREATE TABLE nested_nullability_repro ("
                + "headers ARRAY<ROW<name STRING NOT NULL, header_value BYTES>>) "
                + "WITH ('connector' = 'blackhole')");

CatalogBaseTable table =
        tableEnvironment
                .getCatalog(tableEnvironment.getCurrentCatalog())
                .get()
                .getTable(
                        new ObjectPath(
                                tableEnvironment.getCurrentDatabase(),
                                "nested_nullability_repro"));

System.out.println(table.getUnresolvedSchema());

Output:

(
  `headers` ARRAY<ROW<`name` STRING, `header_value` BYTES>>
)

This indicates that nested nullability has already been lost in Flink's SQL/Catalog schema representation before the Fluss connector converts the type.

Are you willing to submit a PR?

• I'm willing to submit a PR!