[FLINK-37780][2/N] ml builtin sql functions

Author: lihaosky

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMlPredictTableFunction.java

@@ -0,0 +1,96 @@
+/*
+ * 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.flink.table.planner.functions.sql.ml;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlCallBinding;
+import org.apache.calcite.sql.SqlOperandCountRange;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.type.SqlOperandCountRanges;
+import org.apache.calcite.sql.type.SqlOperandMetadata;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * SqlMlPredictTableFunction implements an operator for prediction.
+ *
+ * <p>It allows four parameters:
+ *
+ * <ol>
+ *   <li>a table
+ *   <li>a model name
+ *   <li>a descriptor to provide a column name from the input table
+ *   <li>an optional config map
+ * </ol>
+ */
+public class SqlMlPredictTableFunction extends SqlMlTableFunction {
+
+    public SqlMlPredictTableFunction(String name) {
+        super(name, new PredictOperandMetadata());
+    }
+
+    @Override
+    protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+        // TODO: output type based on table schema and model output schema
+        // model output schema to be available after integrated with SqlExplicitModelCall
+        return null;
+    }
+
+    private static class PredictOperandMetadata implements SqlOperandMetadata {
+        private final List<String> paramNames;
+        private final int mandatoryParamCount;
+
+        PredictOperandMetadata() {
+            paramNames = List.of(PARAM_DATA, PARAM_MODEL, PARAM_COLUMN, PARAM_CONFIG);
+            // Config is optional
+            mandatoryParamCount = 3;
+        }
+
+        @Override
+        public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+            return Collections.nCopies(
+                    paramNames.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+        }
+
+        @Override
+        public List<String> paramNames() {
+            return paramNames;
+        }
+
+        @Override
+        public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+            // TODO: Check operand types after integrated with SqlExplicitModelCall in validator
+            return false;
+        }
+
+        @Override
+        public SqlOperandCountRange getOperandCountRange() {
+            return SqlOperandCountRanges.between(mandatoryParamCount, paramNames.size());
+        }
+
+        @Override
+        public String getAllowedSignatures(SqlOperator op, String opName) {
+            return opName
+                    + "(TABLE table_name, MODEL model_name, DESCRIPTOR(input_columns), [MAP[]]";
+        }
+    }
+}

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMlTableFunction.java

@@ -0,0 +1,57 @@
+/*
+ * 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.flink.table.planner.functions.sql.ml;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperatorBinding;
+import org.apache.calcite.sql.SqlTableFunction;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.sql.type.SqlOperandMetadata;
+import org.apache.calcite.sql.type.SqlReturnTypeInference;
+
+/**
+ * Base class for a table-valued function that works with models. Examples include {@code
+ * ML_PREDICT}.
+ */
+public abstract class SqlMlTableFunction extends SqlFunction implements SqlTableFunction {
+
+    protected static final String PARAM_DATA = "data";
+    protected static final String PARAM_MODEL = "model";
+    protected static final String PARAM_COLUMN = "input_column";
+    protected static final String PARAM_CONFIG = "config";
+
+    public SqlMlTableFunction(String name, SqlOperandMetadata operandMetadata) {
+        super(
+                name,
+                SqlKind.OTHER_FUNCTION,
+                ReturnTypes.CURSOR,
+                null,
+                operandMetadata,
+                SqlFunctionCategory.SYSTEM);
+    }
+
+    @Override
+    public SqlReturnTypeInference getRowTypeInference() {
+        return this::inferRowType;
+    }
+
+    protected abstract RelDataType inferRowType(SqlOperatorBinding opBinding);
+}