You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lihaosky (via GitHub)" <gi...@apache.org> on 2025/06/05 23:55:15 UTC
[PR] [FLINK-37794] ml_evaluate function [flink]
lihaosky opened a new pull request, #26644:
URL: https://github.com/apache/flink/pull/26644
## What is the purpose of the change
Introduce `ml_evaluate` function in FLIP-525
## Brief change log
* Add `ml_evaluate` function and related test
* Refactor `ml_predict` to put common validation logic to base class
## Verifying this change
Unit test
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (JavaDocs)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2135791031
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide input column names from the input table
Review Comment:
should this be feature column names?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2143097859
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,210 @@
+/*
+ * 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.flink.table.api.ValidationException;
+import org.apache.flink.table.ml.TaskType;
+import org.apache.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
Review Comment:
it would be worth giving an example or more description as to what this one is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2135785588
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table
Review Comment:
nit: table name ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2135813362
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide input column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
+ return SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTaskOrConfig(callBinding.operand(4)), throwOnFailure);
+ }
+ return true;
+ }
+
+ @Override
+ public SqlOperandCountRange getOperandCountRange() {
+ return SqlOperandCountRanges.between(MANDATORY_PARAM_NAMES.size(), PARAM_NAMES.size());
+ }
+
+ @Override
+ public boolean isOptional(int i) {
+ return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
Review Comment:
should the minimum check be greater or equal to in line with the max check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148937572
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLPredictTableFunction.java:
##########
@@ -134,100 +123,13 @@ public SqlOperandCountRange getOperandCountRange() {
@Override
public boolean isOptional(int i) {
- return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
+ return i >= getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
Review Comment:
Calctie uses 0 to indicate the first operand here. It's better we can add a case to prevent other break this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26644:
URL: https://github.com/apache/flink/pull/26644#issuecomment-2947066450
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f7e78e7e976a0781e2f271d83a858336bf72fca4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f7e78e7e976a0781e2f271d83a858336bf72fca4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f7e78e7e976a0781e2f271d83a858336bf72fca4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "lihaosky (via GitHub)" <gi...@apache.org>.
lihaosky commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2143197785
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/ml/TaskType.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.ml;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Arrays;
+
+/**
+ * Enum representing different types of machine learning tasks. Each task type has a corresponding
+ * name that can be used for identification.
+ */
+@PublicEvolving
+public enum TaskType {
Review Comment:
This is what we support for default evaluation. It's not needed for just `ml_predict`. Also user can implement their own evaluation udaf after predict if they want. We just provide a convenient function. We can add more if Flink support more types for evaluation. For embedding, maybe use one model output as source of truth and evaluate another model?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148952969
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
Review Comment:
The current behaviour is similar that we have a function
```
func([int], [long])
```
When users only specify `func(1)` and framework uses `1` as the second parameter. If users doesn't use named arguments, we should only use the location to map the parameter and value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148955561
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
Review Comment:
BTW, we can implement the second one first and then use another PR to derive the task from model pojo object.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2139479428
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLPredictTableFunction.java:
##########
@@ -134,100 +123,13 @@ public SqlOperandCountRange getOperandCountRange() {
@Override
public boolean isOptional(int i) {
- return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
+ return i >= getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
Review Comment:
i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax()
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
Review Comment:
I think it is part of public API. It's better we can add an enum class to illustrate.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
Review Comment:
In ML_PREDICT, we use `ARGS` to represent PredictFunction's input, but here we use `FEATURE`. It's better we can have the same name.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
+ return SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTaskOrConfig(callBinding.operand(4)), throwOnFailure);
+ }
+ return true;
+ }
+
+ @Override
+ public SqlOperandCountRange getOperandCountRange() {
+ return SqlOperandCountRanges.between(MANDATORY_PARAM_NAMES.size(), PARAM_NAMES.size());
+ }
+
+ @Override
+ public boolean isOptional(int i) {
+ return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
+ }
+
+ @Override
+ public String getAllowedSignatures(SqlOperator op, String opName) {
+ return opName
+ + "(TABLE table_name, MODEL model_name, DESCRIPTOR(label_column), DESCRIPTOR(feature_columns), [task], [MAP[]])";
+ }
+
+ private static Optional<RuntimeException> checkTask(SqlNode node) {
+ // Check if the task is a valid string
+ if (!(node instanceof SqlCharStringLiteral)) {
+ return Optional.of(
+ new RuntimeException(
+ "Expected a valid task string, but got: " + node + "."));
+ }
+
+ String task = ((SqlCharStringLiteral) node).getValueAs(NlsString.class).getValue();
+ if (SUPPORTED_TASKS.stream()
+ .noneMatch(supportedTask -> supportedTask.equalsIgnoreCase(task))) {
+ return Optional.of(
+ new RuntimeException(
Review Comment:
Use ValidationException
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLTableFunction.java:
##########
@@ -103,4 +115,132 @@ public SqlReturnTypeInference getRowTypeInference() {
}
protected abstract RelDataType inferRowType(SqlOperatorBinding opBinding);
+
+ protected static Optional<RuntimeException> checkModelSignature(
+ SqlCallBinding callBinding, int inputDescriptorIndex, int outputDescriptorIndex) {
+ SqlValidator validator = callBinding.getValidator();
+
+ // Check second operand is SqlModelCall
+ if (!(callBinding.operand(1) instanceof SqlModelCall)) {
+ return Optional.of(
+ new ValidationException("Second operand must be a model identifier."));
+ }
+
+ // Get input descriptor columns
+ SqlCall descriptorCall = (SqlCall) callBinding.operand(inputDescriptorIndex);
+ List<SqlNode> descriptCols = descriptorCall.getOperandList();
+
+ // Get model input size
+ SqlModelCall modelCall = (SqlModelCall) callBinding.operand(1);
+ RelDataType modelInputType = modelCall.getInputType(validator);
+
+ // Check sizes match
+ if (descriptCols.size() != modelInputType.getFieldCount()) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Number of input descriptor columns (%d) does not match model input size (%d).",
+ descriptCols.size(), modelInputType.getFieldCount())));
+ }
+
+ // Check input types match
+ final RelDataType tableType = validator.getValidatedNodeType(callBinding.operand(0));
+ final SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
+ for (int i = 0; i < descriptCols.size(); i++) {
+ Tuple3<Boolean, LogicalType, LogicalType> result =
+ checkModelDescriptorType(
+ tableType,
+ modelInputType.getFieldList().get(i).getType(),
+ descriptCols.get(i),
+ matcher);
+ if (!result.f0) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Input descriptor column type %s cannot be assigned to model input type %s at position %d.",
+ result.f1, result.f2, i)));
+ }
+ }
+
+ // Check output
+ if (outputDescriptorIndex > 0) {
+ descriptorCall = (SqlCall) callBinding.operand(outputDescriptorIndex);
+ descriptCols = descriptorCall.getOperandList();
+ if (descriptCols.size() != 1) {
+ return Optional.of(
+ new ValidationException(
+ "Label descriptor must have exactly one column for evaluation."));
+ }
+ RelDataType modelOutputType = modelCall.getOutputType(validator);
+ if (modelOutputType.getFieldCount() != 1) {
+ return Optional.of(
+ new ValidationException(
+ "Model output must have exactly one field for evaluation."));
+ }
+
+ Tuple3<Boolean, LogicalType, LogicalType> result =
+ checkModelDescriptorType(
+ tableType,
+ modelOutputType.getFieldList().get(0).getType(),
+ descriptCols.get(0),
+ matcher);
+ if (!result.f0) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Label descriptor column type %s cannot be assigned to model output type %s for evaluation.",
+ result.f1, result.f2)));
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ private static Tuple3<Boolean, LogicalType, LogicalType> checkModelDescriptorType(
+ RelDataType tableType,
+ RelDataType modelType,
+ SqlNode descriptorNode,
+ SqlNameMatcher matcher) {
+ SqlIdentifier columnName = (SqlIdentifier) descriptorNode;
+ String descriptColName =
+ columnName.isSimple() ? columnName.getSimple() : Util.last(columnName.names);
+ int index = matcher.indexOf(tableType.getFieldNames(), descriptColName);
+ RelDataType sourceType = tableType.getFieldList().get(index).getType();
+
+ LogicalType sourceLogicalType = toLogicalType(sourceType);
+ LogicalType targetLogicalType = toLogicalType(modelType);
+
+ return Tuple3.of(
+ LogicalTypeCasts.supportsImplicitCast(sourceLogicalType, targetLogicalType),
+ sourceLogicalType,
+ targetLogicalType);
+ }
+
+ protected static Optional<RuntimeException> checkConfig(SqlNode configNode) {
Review Comment:
I have already merged PR about runtime config usage. It's better we can use the latest check logic.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java:
##########
@@ -81,8 +81,8 @@
name = "stream-exec-ml-predict-table-function",
version = 1,
producedTransformations = StreamExecMLPredictTableFunction.ML_PREDICT_TRANSFORMATION,
- minPlanVersion = FlinkVersion.V2_1,
Review Comment:
Do we need to modify this?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
+ return SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTaskOrConfig(callBinding.operand(4)), throwOnFailure);
Review Comment:
It's better we use `MANDATORY_PARAM_NAMES.size() + 1` here because if condition doesn't use hard-code number
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
Review Comment:
I think task is required here. Because we don't require our users to create a model with option named `'task'`.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
Review Comment:
I am confused about the comment. Considering the current function signature,
```
return opName
+ "(TABLE table_name, MODEL model_name, DESCRIPTOR(label_column), DESCRIPTOR(feature_columns), [task], [MAP[]])";
```
The fifth parameter is always task. If not, we should throw a exception to notify users?
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
+ return SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTaskOrConfig(callBinding.operand(4)), throwOnFailure);
+ }
+ return true;
+ }
+
+ @Override
+ public SqlOperandCountRange getOperandCountRange() {
+ return SqlOperandCountRanges.between(MANDATORY_PARAM_NAMES.size(), PARAM_NAMES.size());
+ }
+
+ @Override
+ public boolean isOptional(int i) {
+ return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
+ }
+
+ @Override
+ public String getAllowedSignatures(SqlOperator op, String opName) {
+ return opName
+ + "(TABLE table_name, MODEL model_name, DESCRIPTOR(label_column), DESCRIPTOR(feature_columns), [task], [MAP[]])";
+ }
+
+ private static Optional<RuntimeException> checkTask(SqlNode node) {
+ // Check if the task is a valid string
+ if (!(node instanceof SqlCharStringLiteral)) {
+ return Optional.of(
+ new RuntimeException(
+ "Expected a valid task string, but got: " + node + "."));
+ }
+
+ String task = ((SqlCharStringLiteral) node).getValueAs(NlsString.class).getValue();
+ if (SUPPORTED_TASKS.stream()
+ .noneMatch(supportedTask -> supportedTask.equalsIgnoreCase(task))) {
+ return Optional.of(
+ new RuntimeException(
+ "Unsupported task: "
+ + task
+ + ". Supported tasks are: "
+ + String.join(", ", SUPPORTED_TASKS)
+ + "."));
+ }
+ return Optional.empty();
+ }
+
+ private static Optional<RuntimeException> checkTaskOrConfig(SqlNode node) {
+ if (node.getKind().equals(SqlKind.MAP_VALUE_CONSTRUCTOR)) {
+ // Check if the map is valid
+ return checkConfig(node);
+ } else if (node instanceof SqlCharStringLiteral) {
+ return checkTask(node);
+ } else {
+ return Optional.of(
+ new RuntimeException(
Review Comment:
Use ValidationException
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "davidradl (via GitHub)" <gi...@apache.org>.
davidradl commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2143095690
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/ml/TaskType.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.ml;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Arrays;
+
+/**
+ * Enum representing different types of machine learning tasks. Each task type has a corresponding
+ * name that can be used for identification.
+ */
+@PublicEvolving
+public enum TaskType {
Review Comment:
I wonder whetehr we should be less locked down for the types of Tasks. Hard coding in an enum is very locked down and does not leave scope for easy additions. I was curious to see embedding here, where is the source of truth for the list?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "lihaosky (via GitHub)" <gi...@apache.org>.
lihaosky commented on PR #26644:
URL: https://github.com/apache/flink/pull/26644#issuecomment-2960861298
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "lihaosky (via GitHub)" <gi...@apache.org>.
lihaosky commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2140558849
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
Review Comment:
Sure
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
Review Comment:
Sure
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
Review Comment:
5th param is optional, it can be task or config map. 6th param is also optional and it must be config if exist
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLPredictTableFunction.java:
##########
@@ -134,100 +123,13 @@ public SqlOperandCountRange getOperandCountRange() {
@Override
public boolean isOptional(int i) {
- return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
+ return i >= getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
Review Comment:
I'm confused. If `i` is operand index, it should be `>=` min count? But it's `>` in `SqlWindowTableFunction`.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
Review Comment:
`task` is optional here. It can be used to override the task in model or if task doesn't appear in model.
```
SELECT * FROM ML_EVALUATE(TABLE `eval_data`, MODEL `classifier_model`, DESCRIPTOR(label), DESCRIPTOR(f1, f2))
```
```
SELECT * FROM ML_EVALUATE(input => TABLE `eval_data`, model_input => MODEL `classifier_model`, label => DESCRIPTOR(label), features => DESCRIPTOR(f1, f2), task => 'classification')
```
are both supported in the FLIP. First one will use task in model option. If it doesn't exist, it's an error
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecMLPredictTableFunction.java:
##########
@@ -81,8 +81,8 @@
name = "stream-exec-ml-predict-table-function",
version = 1,
producedTransformations = StreamExecMLPredictTableFunction.ML_PREDICT_TRANSFORMATION,
- minPlanVersion = FlinkVersion.V2_1,
Review Comment:
It's fixed in https://github.com/apache/flink/pull/26661. I may need to rebase
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148954278
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
+ List.of(PARAM_INPUT, PARAM_MODEL, PARAM_LABEL, PARAM_FEATURE);
+
+ EvaluateOperandMetadata() {}
+
+ @Override
+ public List<RelDataType> paramTypes(RelDataTypeFactory typeFactory) {
+ return Collections.nCopies(
+ PARAM_NAMES.size(), typeFactory.createSqlType(SqlTypeName.ANY));
+ }
+
+ @Override
+ public List<String> paramNames() {
+ return PARAM_NAMES;
+ }
+
+ @Override
+ public boolean checkOperandTypes(SqlCallBinding callBinding, boolean throwOnFailure) {
+ if (!SqlValidatorUtils.checkTableAndDescriptorOperands(callBinding, 2, 3)) {
+ return SqlValidatorUtils.throwValidationSignatureErrorOrReturnFalse(
+ callBinding, throwOnFailure);
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkModelSignature(callBinding, 3, 2), throwOnFailure)) {
+ return false;
+ }
+
+ if (callBinding.getOperandCount() == PARAM_NAMES.size()) {
+ // Check task and config parameters
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkTask(callBinding.operand(4)), throwOnFailure)) {
+ return false;
+ }
+
+ if (!SqlValidatorUtils.throwExceptionOrReturnFalse(
+ checkConfig(callBinding.operand(5)), throwOnFailure)) {
+ return false;
+ }
+ }
+
+ if (callBinding.getOperandCount() == MANDATORY_PARAM_NAMES.size() + 1) {
+ // Last param can be config or task
Review Comment:
In SQL, we use BIGINT to parse the literal value if the value is a number without float.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 merged PR #26644:
URL: https://github.com/apache/flink/pull/26644
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148936515
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLPredictTableFunction.java:
##########
@@ -134,100 +123,13 @@ public SqlOperandCountRange getOperandCountRange() {
@Override
public boolean isOptional(int i) {
- return i > getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
+ return i >= getOperandCountRange().getMin() && i <= getOperandCountRange().getMax();
Review Comment:
I think you are right. Can you add a seperate commit to fix the problem in SqlWindowTableFunction?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148944468
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLEvaluateTableFunction.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.table.planner.functions.utils.SqlValidatorUtils;
+
+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.SqlCharStringLiteral;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+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 org.apache.calcite.util.NlsString;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * SqlMLEvaluateTableFunction implements an operator for model evaluation.
+ *
+ * <p>It allows six parameters:
+ *
+ * <ol>
+ * <li>a table name
+ * <li>a model name
+ * <li>a descriptor to provide label column names from the input table
+ * <li>a descriptor to provide feature column names from the input table
+ * <li>an optional task override string
+ * <li>an optional config map
+ * </ol>
+ *
+ * <p>The function returns a MAP type containing evaluation metrics and results.
+ */
+public class SqlMLEvaluateTableFunction extends SqlMLTableFunction {
+
+ public static final String PARAM_LABEL = "LABEL";
+ public static final String PARAM_FEATURE = "FEATURE";
+ public static final String PARAM_TASK = "TASK";
+
+ public static final List<String> SUPPORTED_TASKS =
+ List.of("regression", "clustering", "classification", "embedding", "text_generation");
+
+ public SqlMLEvaluateTableFunction() {
+ super("ML_EVALUATE", new EvaluateOperandMetadata());
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * <p>Overrides because the first parameter of ML table-value function is an explicit TABLE
+ * parameter, which is not scalar.
+ */
+ @Override
+ public boolean argumentMustBeScalar(int ordinal) {
+ return ordinal != 0;
+ }
+
+ @Override
+ protected RelDataType inferRowType(SqlOperatorBinding opBinding) {
+ final RelDataTypeFactory typeFactory = opBinding.getTypeFactory();
+ final RelDataType inputRowType = opBinding.getOperandType(0);
+
+ // Create a MAP type for evaluation results
+ RelDataType keyType = typeFactory.createSqlType(SqlTypeName.VARCHAR);
+ RelDataType valueType = typeFactory.createSqlType(SqlTypeName.DOUBLE);
+ RelDataType mapType = typeFactory.createMapType(keyType, valueType);
+
+ return typeFactory
+ .builder()
+ .kind(inputRowType.getStructKind())
+ .add("result", mapType)
+ .build();
+ }
+
+ private static class EvaluateOperandMetadata implements SqlOperandMetadata {
+ private static final List<String> PARAM_NAMES =
+ List.of(
+ PARAM_INPUT,
+ PARAM_MODEL,
+ PARAM_LABEL,
+ PARAM_FEATURE,
+ PARAM_TASK,
+ PARAM_CONFIG);
+ private static final List<String> MANDATORY_PARAM_NAMES =
Review Comment:
I understand your meaning.
1. If the task derivation from the model is obscure, so I think it's not friendly to users.
2. Currently, our `ModelProviderFactoryHelper#validate` request users to specify options that are subset of the union of the optional and required options. If we want to derive the task from model, we also need to relax the condition here. It introduces a new constraint here: task is a framework option and model provider should not provide an option named `task` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148945775
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/ml/TaskType.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.ml;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Arrays;
+
+/**
+ * Enum representing different types of machine learning tasks. Each task type has a corresponding
+ * name that can be used for identification.
+ */
+@PublicEvolving
+public enum TaskType {
Review Comment:
I think it's better we can use `@Experimental` to tell users the API is unstable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37794] ml_evaluate function [flink]
Posted by "fsk119 (via GitHub)" <gi...@apache.org>.
fsk119 commented on code in PR #26644:
URL: https://github.com/apache/flink/pull/26644#discussion_r2148969047
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/ml/TaskType.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.ml;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+import java.util.Arrays;
+
+/**
+ * Enum representing different types of machine learning tasks. Each task type has a corresponding
+ * name that can be used for identification.
+ */
+@PublicEvolving
+public enum TaskType {
+ REGRESSION("regression"),
+ CLUSTERING("clustering"),
+ CLASSIFICATION("classification"),
+ EMBEDDING("embedding"),
+ TEXT_GENERATION("text_generation");
+
+ private final String name;
+
+ TaskType(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public static TaskType fromName(String name) {
+ return Arrays.stream(values())
+ .filter(taskType -> taskType.name.equalsIgnoreCase(name))
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("Unknown task type: " + name));
+ }
+
+ public static boolean isValidTaskType(String name) {
Review Comment:
I think flink is case-sensitve.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLTableFunction.java:
##########
@@ -103,4 +126,193 @@ public SqlReturnTypeInference getRowTypeInference() {
}
protected abstract RelDataType inferRowType(SqlOperatorBinding opBinding);
+
+ protected static Optional<RuntimeException> checkModelSignature(
+ SqlCallBinding callBinding, int inputDescriptorIndex, int outputDescriptorIndex) {
+ SqlValidator validator = callBinding.getValidator();
+
+ // Check second operand is SqlModelCall
+ if (!(callBinding.operand(1) instanceof SqlModelCall)) {
+ return Optional.of(
+ new ValidationException("Second operand must be a model identifier."));
+ }
+
+ // Get input descriptor columns
+ SqlCall descriptorCall = (SqlCall) callBinding.operand(inputDescriptorIndex);
+ List<SqlNode> descriptCols = descriptorCall.getOperandList();
+
+ // Get model input size
+ SqlModelCall modelCall = (SqlModelCall) callBinding.operand(1);
+ RelDataType modelInputType = modelCall.getInputType(validator);
+
+ // Check sizes match
+ if (descriptCols.size() != modelInputType.getFieldCount()) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Number of input descriptor columns (%d) does not match model input size (%d).",
+ descriptCols.size(), modelInputType.getFieldCount())));
+ }
+
+ // Check input types match
+ final RelDataType tableType = validator.getValidatedNodeType(callBinding.operand(0));
+ final SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
+ for (int i = 0; i < descriptCols.size(); i++) {
+ Tuple3<Boolean, LogicalType, LogicalType> result =
+ checkModelDescriptorType(
+ tableType,
+ modelInputType.getFieldList().get(i).getType(),
+ descriptCols.get(i),
+ matcher);
+ if (!result.f0) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Input descriptor column type %s cannot be assigned to model input type %s at position %d.",
+ result.f1, result.f2, i)));
+ }
+ }
+
+ // Check output
+ if (outputDescriptorIndex > 0) {
+ descriptorCall = (SqlCall) callBinding.operand(outputDescriptorIndex);
+ descriptCols = descriptorCall.getOperandList();
+ if (descriptCols.size() != 1) {
+ return Optional.of(
+ new ValidationException(
+ "Label descriptor must have exactly one column for evaluation."));
+ }
+ RelDataType modelOutputType = modelCall.getOutputType(validator);
+ if (modelOutputType.getFieldCount() != 1) {
+ return Optional.of(
+ new ValidationException(
+ "Model output must have exactly one field for evaluation."));
+ }
+
+ Tuple3<Boolean, LogicalType, LogicalType> result =
+ checkModelDescriptorType(
+ tableType,
+ modelOutputType.getFieldList().get(0).getType(),
+ descriptCols.get(0),
+ matcher);
+ if (!result.f0) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Label descriptor column type %s cannot be assigned to model output type %s for evaluation.",
+ result.f1, result.f2)));
+ }
+ }
+
+ return Optional.empty();
+ }
+
+ private static Tuple3<Boolean, LogicalType, LogicalType> checkModelDescriptorType(
+ RelDataType tableType,
+ RelDataType modelType,
+ SqlNode descriptorNode,
+ SqlNameMatcher matcher) {
+ SqlIdentifier columnName = (SqlIdentifier) descriptorNode;
+ String descriptColName =
Review Comment:
descriptorColName
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/ml/SqlMLTableFunction.java:
##########
@@ -103,4 +126,193 @@ public SqlReturnTypeInference getRowTypeInference() {
}
protected abstract RelDataType inferRowType(SqlOperatorBinding opBinding);
+
+ protected static Optional<RuntimeException> checkModelSignature(
+ SqlCallBinding callBinding, int inputDescriptorIndex, int outputDescriptorIndex) {
+ SqlValidator validator = callBinding.getValidator();
+
+ // Check second operand is SqlModelCall
+ if (!(callBinding.operand(1) instanceof SqlModelCall)) {
+ return Optional.of(
+ new ValidationException("Second operand must be a model identifier."));
+ }
+
+ // Get input descriptor columns
+ SqlCall descriptorCall = (SqlCall) callBinding.operand(inputDescriptorIndex);
+ List<SqlNode> descriptCols = descriptorCall.getOperandList();
+
+ // Get model input size
+ SqlModelCall modelCall = (SqlModelCall) callBinding.operand(1);
+ RelDataType modelInputType = modelCall.getInputType(validator);
+
+ // Check sizes match
+ if (descriptCols.size() != modelInputType.getFieldCount()) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Number of input descriptor columns (%d) does not match model input size (%d).",
+ descriptCols.size(), modelInputType.getFieldCount())));
+ }
+
+ // Check input types match
+ final RelDataType tableType = validator.getValidatedNodeType(callBinding.operand(0));
+ final SqlNameMatcher matcher = validator.getCatalogReader().nameMatcher();
+ for (int i = 0; i < descriptCols.size(); i++) {
+ Tuple3<Boolean, LogicalType, LogicalType> result =
+ checkModelDescriptorType(
+ tableType,
+ modelInputType.getFieldList().get(i).getType(),
+ descriptCols.get(i),
+ matcher);
+ if (!result.f0) {
+ return Optional.of(
+ new ValidationException(
+ String.format(
+ "Input descriptor column type %s cannot be assigned to model input type %s at position %d.",
+ result.f1, result.f2, i)));
+ }
+ }
+
+ // Check output
+ if (outputDescriptorIndex > 0) {
Review Comment:
Actually, only ML_EVALUTE need to check this? It's better we can mode ML_EVALUTE logic to subclass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org