[FLINK-38841][table] Enforce expected nullability in FamilyArgumentStrategy

Author: dylanhz

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/FamilyArgumentTypeStrategy.java

@@ -86,13 +86,16 @@ public Optional<DataType> inferArgumentType(
         if (Objects.equals(expectedNullability, Boolean.FALSE) && actualType.isNullable()) {
             return callContext.fail(
                     throwOnFailure,
-                    "Unsupported argument type. Expected nullable type of family '%s' but actual type was '%s'.",
+                    "Unsupported argument type. Expected NOT NULL type of family '%s' but actual type was '%s'.",
                     expectedFamily,
                     actualType);
         }
 
         // type is part of the family
         if (actualType.getTypeRoot().getFamilies().contains(expectedFamily)) {
+            if (Objects.equals(expectedNullability, Boolean.TRUE) && !actualType.isNullable()) {
+                return Optional.of(actualDataType.nullable());
+            }
             return Optional.of(actualDataType);
         }
 

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/InputTypeStrategiesTest.java

@@ -549,23 +549,33 @@ ANY, explicit(DataTypes.INT())
                                 sequence(
                                         logical(LogicalTypeFamily.CHARACTER_STRING, true),
                                         logical(LogicalTypeFamily.EXACT_NUMERIC),
+                                        logical(LogicalTypeFamily.EXACT_NUMERIC, true),
                                         logical(LogicalTypeFamily.APPROXIMATE_NUMERIC),
                                         logical(LogicalTypeFamily.APPROXIMATE_NUMERIC),
                                         logical(LogicalTypeFamily.APPROXIMATE_NUMERIC, false)))
                         .calledWithArgumentTypes(
                                 DataTypes.NULL(),
                                 DataTypes.TINYINT(),
+                                DataTypes.SMALLINT().notNull(),
                                 DataTypes.INT(),
                                 DataTypes.BIGINT().notNull(),
                                 DataTypes.DECIMAL(10, 2).notNull())
                         .expectSignature(
-                                "f(<CHARACTER_STRING NULL>, <EXACT_NUMERIC>, <APPROXIMATE_NUMERIC>, <APPROXIMATE_NUMERIC>, <APPROXIMATE_NUMERIC NOT NULL>)")
+                                "f(<CHARACTER_STRING NULL>, <EXACT_NUMERIC>, <EXACT_NUMERIC NULL>, <APPROXIMATE_NUMERIC>, <APPROXIMATE_NUMERIC>, <APPROXIMATE_NUMERIC NOT NULL>)")
                         .expectArgumentTypes(
                                 DataTypes.VARCHAR(1),
                                 DataTypes.TINYINT(),
+                                DataTypes.SMALLINT(),
                                 DataTypes.DOUBLE(), // widening with preserved nullability
                                 DataTypes.DOUBLE().notNull(), // widening with preserved nullability
                                 DataTypes.DOUBLE().notNull()),
+                TestSpec.forStrategy(
+                                "Logical type family with invalid nullability",
+                                sequence(logical(LogicalTypeFamily.EXACT_NUMERIC, false)))
+                        .calledWithArgumentTypes(DataTypes.INT())
+                        .expectSignature("f(<EXACT_NUMERIC NOT NULL>)")
+                        .expectErrorMessage(
+                                "Unsupported argument type. Expected NOT NULL type of family 'EXACT_NUMERIC' but actual type was 'INT'."),
                 TestSpec.forStrategy(
                                 "Logical type family with invalid type",
                                 sequence(logical(LogicalTypeFamily.EXACT_NUMERIC)))