[FLINK-xxx][1/N] introduce MODEL keyword for model as argument in function

Author: lihaosky

flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd

@@ -666,6 +666,7 @@
   # Example: DateFunctionCall().
   builtinFunctionCallMethods: [
     "TryCastFunctionCall()"
+    "ExplicitModel()"
   ]
 
   # List of methods for parsing extensions to "ALTER <scope>" calls.

flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl

@@ -3089,6 +3089,22 @@ SqlNode TryCastFunctionCall() :
     }
 }
 
+/**
+ * Parses an explicit Model m reference.
+ */
+SqlNode ExplicitModel() :
+{
+    SqlNode modelRef;
+    final Span s;
+}
+{
+    <MODEL> modelRef = CompoundIdentifier()
+    {
+        s = span();
+        return new SqlExplicitModelOperator(2, null, null, null).createCall(s.pos(), modelRef);
+    }
+}
+
 /**
 * Parses a partition key/value,
 * e.g. p or p = '10'.

flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/SqlExplicitModelCall.java

@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+/** SqlExplicitModelCall is a SQL call that represents an explicit model. */
+public class SqlExplicitModelCall extends SqlBasicCall {
+
+    public SqlExplicitModelCall(
+            SqlOperator operator,
+            List<? extends @Nullable SqlNode> operandList,
+            SqlParserPos pos,
+            @Nullable SqlLiteral functionQualifier) {
+        super(operator, operandList, pos, functionQualifier);
+    }
+}

flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/SqlExplicitModelOperator.java

@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.SqlOperandTypeChecker;
+import org.apache.calcite.sql.type.SqlOperandTypeInference;
+import org.apache.calcite.sql.type.SqlReturnTypeInference;
+import org.apache.calcite.util.ImmutableNullableList;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** SqlExplicitModelOperator is a SQL operator that represents an explicit model. */
+public class SqlExplicitModelOperator extends SqlPrefixOperator {
+
+    public SqlExplicitModelOperator(
+            int prec,
+            @Nullable SqlReturnTypeInference returnTypeInference,
+            @Nullable SqlOperandTypeInference operandTypeInference,
+            @Nullable SqlOperandTypeChecker operandTypeChecker) {
+        super(
+                "MODEL",
+                SqlKind.OTHER_FUNCTION,
+                prec,
+                returnTypeInference,
+                operandTypeInference,
+                operandTypeChecker);
+    }
+
+    @Override
+    public SqlCall createCall(
+            @Nullable SqlLiteral functionQualifier,
+            SqlParserPos pos,
+            @Nullable SqlNode... operands) {
+        pos = pos.plusAll(operands);
+        return new SqlExplicitModelCall(
+                this, ImmutableNullableList.copyOf(operands), pos, functionQualifier);
+    }
+}

flink-table/flink-sql-parser/src/test/java/org/apache/flink/sql/parser/FlinkSqlParserImplTest.java

@@ -3360,6 +3360,22 @@ void testCreateModelAsWithOutput() {
                                         "CREATE MODEL AS SELECT syntax does not support to specify explicit output columns."));
     }
 
+    @Test
+    void testModelInFunction() {
+        sql("select * from table(ml_predict(TABLE my_table, MODEL my_model))")
+                .ok(
+                        "SELECT *\n"
+                                + "FROM TABLE(`ML_PREDICT`((TABLE `MY_TABLE`), MODEL `MY_MODEL`))");
+    }
+
+    @Test
+    void testModelInFunctionWithoutTable() {
+        sql("select * from func(TABLE my_table, MODEL cat.db.my_model)")
+                .ok(
+                        "SELECT *\n"
+                                + "FROM TABLE(`FUNC`((TABLE `MY_TABLE`), MODEL `CAT`.`DB`.`MY_MODEL`))");
+    }
+
     /*
      * This test was backported from Calcite 1.38 (CALCITE-6266).
      * Remove it together with upgrade to Calcite 1.38.