[FLINK-33255] [table] Validate argument count during type inference

- Previously, TypeInferenceOperandInference was not validating argument counts
  which led to bugs like FLINK-33248
- This commit adds a check to validate argument count and throws an exception
  if the argument count doesn't match the expected number of arguments

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

@@ -209,6 +209,48 @@ public static TableException createUnexpectedException(
                 cause);
     }
 
+    /**
+     * Validates argument counts.
+     *
+     * @param argumentCount expected argument count
+     * @param actualCount actual argument count
+     * @param throwOnFailure if true, the function throws a {@link ValidationException} if the
+     *     actual value does not meet the expected argument count
+     * @return a boolean indicating if expected argument counts match the actual counts
+     */
+    public static boolean validateArgumentCount(
+            ArgumentCount argumentCount, int actualCount, boolean throwOnFailure) {
+        final int minCount = argumentCount.getMinCount().orElse(0);
+        if (actualCount < minCount) {
+            if (throwOnFailure) {
+                throw new ValidationException(
+                        String.format(
+                                "Invalid number of arguments. At least %d arguments expected but %d passed.",
+                                minCount, actualCount));
+            }
+            return false;
+        }
+        final int maxCount = argumentCount.getMaxCount().orElse(Integer.MAX_VALUE);
+        if (actualCount > maxCount) {
+            if (throwOnFailure) {
+                throw new ValidationException(
+                        String.format(
+                                "Invalid number of arguments. At most %d arguments expected but %d passed.",
+                                maxCount, actualCount));
+            }
+            return false;
+        }
+        if (!argumentCount.isValidCount(actualCount)) {
+            if (throwOnFailure) {
+                throw new ValidationException(
+                        String.format(
+                                "Invalid number of arguments. %d arguments passed.", actualCount));
+            }
+            return false;
+        }
+        return true;
+    }
+
     /**
      * Information what the outer world (i.e. an outer wrapping call) expects from the current
      * function call. This can be helpful for an {@link InputTypeStrategy}.
@@ -385,39 +427,6 @@ private static String formatArgument(Signature.Argument arg) {
         return stringBuilder.toString();
     }
 
-    private static boolean validateArgumentCount(
-            ArgumentCount argumentCount, int actualCount, boolean throwOnFailure) {
-        final int minCount = argumentCount.getMinCount().orElse(0);
-        if (actualCount < minCount) {
-            if (throwOnFailure) {
-                throw new ValidationException(
-                        String.format(
-                                "Invalid number of arguments. At least %d arguments expected but %d passed.",
-                                minCount, actualCount));
-            }
-            return false;
-        }
-        final int maxCount = argumentCount.getMaxCount().orElse(Integer.MAX_VALUE);
-        if (actualCount > maxCount) {
-            if (throwOnFailure) {
-                throw new ValidationException(
-                        String.format(
-                                "Invalid number of arguments. At most %d arguments expected but %d passed.",
-                                maxCount, actualCount));
-            }
-            return false;
-        }
-        if (!argumentCount.isValidCount(actualCount)) {
-            if (throwOnFailure) {
-                throw new ValidationException(
-                        String.format(
-                                "Invalid number of arguments. %d arguments passed.", actualCount));
-            }
-            return false;
-        }
-        return true;
-    }
-
     private static AdaptedCallContext inferInputTypes(
             TypeInference typeInference,
             CallContext callContext,

flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/ExtraCalciteResource.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.calcite.runtime.Resources;
+
+/**
+ * Compiler-checked resources similar to CalciteResource in Calcite project. These are extra
+ * exceptions we want to extend Calcite with. Ref:
+ * https://issues.apache.org/jira/browse/CALCITE-6069
+ */
+public interface ExtraCalciteResource {
+
+    @Resources.BaseMessage(
+            "No match found for function signature {0}.\nSupported signatures are:\n{1}")
+    Resources.ExInst<SqlValidatorException> validatorNoFunctionMatch(
+            String invocation, String allowedSignatures);
+}

flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

@@ -158,10 +158,13 @@
  * Default implementation of {@link SqlValidator}, the class was copied over because of
  * CALCITE-4554.
  *
- * <p>Lines 5079 ~ 5092, Flink enables TIMESTAMP and TIMESTAMP_LTZ for system time period
+ * <p>Lines 1954 ~ 1977, Flink improves error message for functions without appropriate arguments in
+ * handleUnresolvedFunction at {@link SqlValidatorImpl#handleUnresolvedFunction}.
+ *
+ * <p>Lines 5101 ~ 5114, Flink enables TIMESTAMP and TIMESTAMP_LTZ for system time period
  * specification type at {@link org.apache.calcite.sql.validate.SqlValidatorImpl#validateSnapshot}.
  *
- * <p>Lines 5436 ~ 5442, Flink enables TIMESTAMP and TIMESTAMP_LTZ for first orderBy column in
+ * <p>Lines 5458 ~ 5464, Flink enables TIMESTAMP and TIMESTAMP_LTZ for first orderBy column in
  * matchRecognize at {@link SqlValidatorImpl#validateMatchRecognize}.
  */
 public class SqlValidatorImpl implements SqlValidatorWithHints {
@@ -181,6 +184,9 @@ public class SqlValidatorImpl implements SqlValidatorWithHints {
     /** Alias prefix generated for source columns when rewriting UPDATE to MERGE. */
     public static final String UPDATE_ANON_PREFIX = "SYS$ANON";
 
+    private static final ExtraCalciteResource EXTRA_RESOURCE =
+            Resources.create(ExtraCalciteResource.class);
+
     // ~ Instance fields --------------------------------------------------------
 
     private final SqlOperatorTable opTab;
@@ -1946,11 +1952,27 @@ public CalciteException handleUnresolvedFunction(
 
         final String signature;
         if (unresolvedFunction instanceof SqlFunction) {
+            // ----- FLINK MODIFICATION BEGIN -----
             final SqlOperandTypeChecker typeChecking =
                     new AssignableOperandTypeChecker(argTypes, argNames);
-            signature =
+            final String invocation =
                     typeChecking.getAllowedSignatures(
                             unresolvedFunction, unresolvedFunction.getName());
+            if (unresolvedFunction.getOperandTypeChecker() != null) {
+                final String allowedSignatures =
+                        unresolvedFunction
+                                .getOperandTypeChecker()
+                                .getAllowedSignatures(
+                                        unresolvedFunction, unresolvedFunction.getName());
+                throw newValidationError(
+                        call,
+                        EXTRA_RESOURCE.validatorNoFunctionMatch(invocation, allowedSignatures));
+            } else {
+                signature =
+                        typeChecking.getAllowedSignatures(
+                                unresolvedFunction, unresolvedFunction.getName());
+            }
+            // ----- FLINK MODIFICATION END -----
         } else {
             signature = unresolvedFunction.getName();
         }

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/inference/TypeInferenceOperandInference.java

@@ -68,7 +68,12 @@ public void inferOperandTypes(
         final CallContext callContext =
                 new CallBindingCallContext(dataTypeFactory, definition, callBinding, returnType);
         try {
-            inferOperandTypesOrError(unwrapTypeFactory(callBinding), callContext, operandTypes);
+            if (TypeInferenceUtil.validateArgumentCount(
+                    typeInference.getInputTypeStrategy().getArgumentCount(),
+                    callContext.getArgumentDataTypes().size(),
+                    false)) {
+                inferOperandTypesOrError(unwrapTypeFactory(callBinding), callContext, operandTypes);
+            }
         } catch (ValidationException | CalciteContextException e) {
             // let operand checker fail
         } catch (Throwable t) {

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/CollectionFunctionsITCase.java

@@ -975,11 +975,16 @@ private Stream<TestSetSpec> arrayJoinTestCases() {
                                 DataTypes.STRING().nullable())
                         .testSqlValidationError(
                                 "ARRAY_JOIN(f0)",
-                                "No match found for function "
-                                        + "signature ARRAY_JOIN(<VARCHAR(2147483647) ARRAY>)")
+                                "No match found for function signature ARRAY_JOIN(<VARCHAR(2147483647) ARRAY>).\n"
+                                        + "Supported signatures are:\n"
+                                        + "ARRAY_JOIN(ARRAY<STRING>, <CHARACTER_STRING>)\n"
+                                        + "ARRAY_JOIN(ARRAY<STRING>, <CHARACTER_STRING>, <CHARACTER_STRING>)")
                         .testSqlValidationError(
                                 "ARRAY_JOIN()",
-                                "No match found for function signature ARRAY_JOIN()")
+                                "No match found for function signature ARRAY_JOIN().\n"
+                                        + "Supported signatures are:\n"
+                                        + "ARRAY_JOIN(ARRAY<STRING>, <CHARACTER_STRING>)\n"
+                                        + "ARRAY_JOIN(ARRAY<STRING>, <CHARACTER_STRING>, <CHARACTER_STRING>)")
                         .testSqlValidationError(
                                 "ARRAY_JOIN(f5, '+')",
                                 "Invalid input arguments. Expected signatures are:\n"
@@ -1190,7 +1195,16 @@ private Stream<TestSetSpec> arraySliceTestCases() {
                                         + "ARRAY_SLICE(<ARRAY>, <INTEGER>, <INTEGER>)")
                         .testSqlValidationError(
                                 "ARRAY_SLICE()",
-                                " No match found for function signature ARRAY_SLICE()")
+                                "No match found for function signature ARRAY_SLICE().\n"
+                                        + "Supported signatures are:\n"
+                                        + "ARRAY_SLICE(<ARRAY>, <INTEGER>, <INTEGER>)\n"
+                                        + "ARRAY_SLICE(<ARRAY>, <INTEGER>)")
+                        .testSqlValidationError(
+                                "ARRAY_SLICE(1)",
+                                "No match found for function signature ARRAY_SLICE(<NUMERIC>).\n"
+                                        + "Supported signatures are:\n"
+                                        + "ARRAY_SLICE(<ARRAY>, <INTEGER>, <INTEGER>)\n"
+                                        + "ARRAY_SLICE(<ARRAY>, <INTEGER>)")
                         .testSqlValidationError("ARRAY_SLICE(null)", "Illegal use of 'NULL'"));
     }
 }

flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/CalcITCase.scala

@@ -659,9 +659,12 @@ class CalcITCase extends StreamingTestBase {
     } catch {
       case e: Exception =>
         assertEquals(
-          "SQL validation failed. From line 1, column 12 to line 1, column 30: " +
-            "No match found for function signature CURRENT_WATERMARK()",
-          e.getMessage)
+          "SQL validation failed. From line 1, column 12 to line 1, column 30: No match found for function signature CURRENT_WATERMARK().\n" +
+            "Supported signatures are:\n" +
+            "CURRENT_WATERMARK(<TIMESTAMP_WITHOUT_TIME_ZONE *ROWTIME*>)\n" +
+            "CURRENT_WATERMARK(<TIMESTAMP_WITH_LOCAL_TIME_ZONE *ROWTIME*>)",
+          e.getMessage
+        )
     }
   }