Hi! I'm facing an issue with the output schema for FileSystemTableSink. In FileSystemTableSink#createWriter (FileSystemTableSink.java#L29 <https://github.com/apache/flink/blob/release-1.16.0/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java#L29>9) the original nullability of the underlying logical data type is not preserved, which introduces unnecessarily union with null in the schema, i.e. for avro, it generates [null, {"type":"record", "fields": ...}] instead of {"type":"record", "fields": ...}. https://issues.apache.org/jira/browse/FLINK-30438 describes the same problem, but not sure if the root cause is the same. We use Flink 1.16.0, but it's relevant for newer versions.
Looking at the source code, the issue exists because DataType::ROW instantiates RowType with isNullable = true by default (constructor here <https://github.com/apache/flink/blob/master/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/logical/RowType.java#L161-L163>). Similar DataType creation is followed by nullability check and calling .notNull() in DataTypeUtils <https://github.com/apache/flink/blob/1.16/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/utils/DataTypeUtils.java#L124-L127> . I wonder whether someone had the same issue and whether there is a workaround. Thank you, Yuliya
