[FLINK-39915][table] Reject mixing positional and named function arguments

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidator.java

@@ -380,6 +380,7 @@ protected void addToSelectList(
         }
 
         final SqlBasicCall call = (SqlBasicCall) node;
+        checkNoNamedAndPositionalMixedArgs(call);
 
         // Special case for MODEL
         if (node instanceof SqlExplicitModelCall) {
@@ -432,6 +433,27 @@ protected void addToSelectList(
         return rewritten;
     }
 
+    /** Mixing positional and named arguments is not supported and crashes operand permutation. */
+    private static void checkNoNamedAndPositionalMixedArgs(SqlBasicCall call) {
+        if (!(call.getOperator() instanceof SqlFunction)) {
+            return;
+        }
+        final List<SqlNode> operands = call.getOperandList();
+        final boolean anyNamed =
+                operands.stream()
+                        .anyMatch(op -> op != null && op.getKind() == SqlKind.ARGUMENT_ASSIGNMENT);
+        final boolean anyPositional =
+                operands.stream()
+                        .anyMatch(op -> op != null && op.getKind() != SqlKind.ARGUMENT_ASSIGNMENT);
+        if (anyNamed && anyPositional) {
+            throw new ValidationException(
+                    "Cannot mix positional and named arguments when calling function '"
+                            + call.getOperator().getName()
+                            + "'. Use either all positional arguments or all named arguments "
+                            + "(e.g. arg => value).");
+        }
+    }
+
     @Override
     public SqlNode maybeCast(SqlNode node, RelDataType currentType, RelDataType desiredType) {
         return super.maybeCast(node, currentType, desiredType);

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/calcite/FlinkCalciteSqlValidatorTest.java

@@ -18,9 +18,13 @@
 
 package org.apache.flink.table.planner.calcite;
 
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.Schema;
 import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.functions.ScalarFunction;
 import org.apache.flink.table.planner.utils.PlannerMocks;
 
 import org.junit.jupiter.api.Test;
@@ -138,4 +142,32 @@ void testExplainUpsertInto() {
                 .hasMessageContaining(
                         "UPSERT INTO statement is not supported. Please use INSERT INTO instead.");
     }
+
+    @Test
+    void testMixedPositionalAndNamedArguments() {
+        plannerMocks
+                .getFunctionCatalog()
+                .registerTemporarySystemFunction("myFunc", new NamedArgsScalarFunction(), false);
+
+        assertDoesNotThrow(() -> plannerMocks.getParser().parse("SELECT myFunc(1, 2) FROM t1"));
+        assertDoesNotThrow(
+                () -> plannerMocks.getParser().parse("SELECT myFunc(in1 => 1, in2 => 2) FROM t1"));
+        assertThatThrownBy(
+                        () -> plannerMocks.getParser().parse("SELECT myFunc(1, in2 => 2) FROM t1"))
+                .isInstanceOf(ValidationException.class)
+                .hasMessageContaining("Cannot mix positional and named arguments");
+    }
+
+    /** Scalar function with named arguments for the mixed-argument validation test. */
+    public static class NamedArgsScalarFunction extends ScalarFunction {
+        @FunctionHint(
+                output = @DataTypeHint("INT"),
+                arguments = {
+                    @ArgumentHint(name = "in1", type = @DataTypeHint("INT")),
+                    @ArgumentHint(name = "in2", type = @DataTypeHint("INT"))
+                })
+        public Integer eval(Integer in1, Integer in2) {
+            return null;
+        }
+    }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ProcessTableFunctionTest.java

@@ -349,6 +349,11 @@ void testErrorBehavior(ErrorSpec spec) {
 
     private static Stream<ErrorSpec> errorSpecs() {
         return Stream.of(
+                ErrorSpec.ofSelect(
+                        "mixed positional and named arguments",
+                        ScalarArgsFunction.class,
+                        "SELECT * FROM f(1, b => true)",
+                        "Cannot mix positional and named arguments when calling function 'f'"),
                 ErrorSpec.ofSelect(
                         "invalid uid",
                         ScalarArgsFunction.class,