comments

Author: lihaosky

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Model.java

@@ -19,6 +19,7 @@
 package org.apache.flink.table.api;
 
 import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.table.api.config.MLPredictRuntimeConfigOptions;
 import org.apache.flink.table.catalog.ResolvedSchema;
 import org.apache.flink.types.ColumnList;
 
@@ -104,13 +105,7 @@ public interface Model {
      * runtime configuration options such as max-concurrent-operations, timeout, and execution mode
      * settings.
      *
-     * <p>Common runtime options include:
-     *
-     * <ul>
-     *   <li>"max-concurrent-operations" - Number of records to process in each batch
-     *   <li>"timeout" - Maximum time to wait for model inference
-     *   <li>"async" - Whether to enable asynchronous execution
-     * </ul>
+     * <p>For Common runtime options, see {@link MLPredictRuntimeConfigOptions}.
      *
      * <p>Example:
      *

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ModelImpl.java

@@ -77,11 +77,11 @@ public Table predict(Table table, ColumnList inputColumns) {
 
     @Override
     public Table predict(Table table, ColumnList inputColumns, Map<String, String> options) {
-        // Use Expressions.map() instead of Expression.lit() to create a MAP literal since
+        // Use Expressions.map() instead of Expressions.lit() to create a MAP literal since
         // lit() is not serializable to sql.
         if (options.isEmpty()) {
             return tableEnvironment.fromCall(
-                    "ML_PREDICT",
+                    BuiltInFunctionDefinitions.ML_PREDICT.getName(),
                     table.asArgument("INPUT"),
                     this.asArgument("MODEL"),
                     Expressions.descriptor(inputColumns).asArgument("ARGS"));
@@ -93,7 +93,7 @@ public Table predict(Table table, ColumnList inputColumns, Map<String, String> o
                     configKVs.add(v);
                 });
         return tableEnvironment.fromCall(
-                "ML_PREDICT",
+                BuiltInFunctionDefinitions.ML_PREDICT.getName(),
                 table.asArgument("INPUT"),
                 this.asArgument("MODEL"),
                 Expressions.descriptor(inputColumns).asArgument("ARGS"),

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ApiExpressionVisitor.java

@@ -29,6 +29,8 @@ public final R visit(Expression other) {
             return visit((UnresolvedReferenceExpression) other);
         } else if (other instanceof TableReferenceExpression) {
             return visit((TableReferenceExpression) other);
+        } else if (other instanceof ModelReferenceExpression) {
+            return visit((ModelReferenceExpression) other);
         } else if (other instanceof LocalReferenceExpression) {
             return visit((LocalReferenceExpression) other);
         } else if (other instanceof LookupCallExpression) {

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ModelReferenceExpression.java

@@ -21,6 +21,7 @@
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.table.api.Model;
 import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.catalog.ContextResolvedModel;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.utils.DataTypeUtils;
@@ -103,6 +104,10 @@ public List<ResolvedExpression> getResolvedChildren() {
 
     @Override
     public String asSerializableString(SqlFactory sqlFactory) {
+        if (model.isAnonymous()) {
+            throw new ValidationException("Anonymous models cannot be serialized.");
+        }
+
         return "MODEL " + model.getIdentifier().asSerializableString();
     }
 

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/ResolvedExpressionVisitor.java

@@ -32,6 +32,8 @@ public abstract class ResolvedExpressionVisitor<R> implements ExpressionVisitor<
     public final R visit(Expression other) {
         if (other instanceof TableReferenceExpression) {
             return visit((TableReferenceExpression) other);
+        } else if (other instanceof ModelReferenceExpression) {
+            return visit((ModelReferenceExpression) other);
         } else if (other instanceof LocalReferenceExpression) {
             return visit((LocalReferenceExpression) other);
         } else if (other instanceof ResolvedExpression) {
@@ -42,6 +44,8 @@ public final R visit(Expression other) {
 
     public abstract R visit(TableReferenceExpression tableReference);
 
+    public abstract R visit(ModelReferenceExpression modelReferenceExpression);
+
     public abstract R visit(LocalReferenceExpression localReference);
 
     /** For resolved expressions created by the planner. */

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/utils/ResolvedExpressionDefaultVisitor.java

@@ -22,6 +22,7 @@
 import org.apache.flink.table.expressions.CallExpression;
 import org.apache.flink.table.expressions.FieldReferenceExpression;
 import org.apache.flink.table.expressions.LocalReferenceExpression;
+import org.apache.flink.table.expressions.ModelReferenceExpression;
 import org.apache.flink.table.expressions.NestedFieldReferenceExpression;
 import org.apache.flink.table.expressions.ResolvedExpression;
 import org.apache.flink.table.expressions.ResolvedExpressionVisitor;
@@ -41,6 +42,10 @@ public T visit(TableReferenceExpression tableReference) {
         return defaultMethod(tableReference);
     }
 
+    public T visit(ModelReferenceExpression modelReferenceExpression) {
+        return defaultMethod(modelReferenceExpression);
+    }
+
     @Override
     public T visit(LocalReferenceExpression localReference) {
         return defaultMethod(localReference);

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/FailingTableApiTestStep.java

@@ -0,0 +1,142 @@
+/*
+ * 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.test.program;
+
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.test.program.TableApiTestStep.TableEnvAccessor;
+import org.apache.flink.util.Preconditions;
+
+import java.util.function.Function;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Test step for executing Table API query that will fail eventually with either {@link
+ * ValidationException} (during planning time) or {@link TableRuntimeException} (during execution
+ * time).
+ *
+ * <p>Similar to {@link FailingSqlTestStep} but uses Table API instead of SQL.
+ */
+public final class FailingTableApiTestStep implements TestStep {
+
+    private final Function<TableEnvAccessor, Table> tableQuery;
+    private final String sinkName;
+    public final Class<? extends Exception> expectedException;
+    public final String expectedErrorMessage;
+
+    FailingTableApiTestStep(
+            Function<TableEnvAccessor, Table> tableQuery,
+            String sinkName,
+            Class<? extends Exception> expectedException,
+            String expectedErrorMessage) {
+        Preconditions.checkArgument(
+                expectedException == ValidationException.class
+                        || expectedException == TableRuntimeException.class,
+                "Usually a Table API query should fail with either validation or runtime exception. "
+                        + "Otherwise this might require an update to the exception design.");
+        this.tableQuery = tableQuery;
+        this.sinkName = sinkName;
+        this.expectedException = expectedException;
+        this.expectedErrorMessage = expectedErrorMessage;
+    }
+
+    @Override
+    public TestKind getKind() {
+        return TestKind.FAILING_TABLE_API;
+    }
+
+    public Table toTable(TableEnvironment env) {
+        return tableQuery.apply(
+                new TableEnvAccessor() {
+                    @Override
+                    public Table from(String path) {
+                        return env.from(path);
+                    }
+
+                    @Override
+                    public Table fromCall(String path, Object... arguments) {
+                        return env.fromCall(path, arguments);
+                    }
+
+                    @Override
+                    public Table fromCall(
+                            Class<? extends org.apache.flink.table.functions.UserDefinedFunction>
+                                    function,
+                            Object... arguments) {
+                        return env.fromCall(function, arguments);
+                    }
+
+                    @Override
+                    public Table fromValues(Object... values) {
+                        return env.fromValues(values);
+                    }
+
+                    @Override
+                    public Table fromValues(
+                            org.apache.flink.table.types.AbstractDataType<?> dataType,
+                            Object... values) {
+                        return env.fromValues(dataType, values);
+                    }
+
+                    @Override
+                    public Table sqlQuery(String query) {
+                        return env.sqlQuery(query);
+                    }
+
+                    @Override
+                    public org.apache.flink.table.api.Model fromModel(String modelPath) {
+                        return env.fromModelPath(modelPath);
+                    }
+
+                    @Override
+                    public org.apache.flink.table.api.Model from(
+                            org.apache.flink.table.api.ModelDescriptor modelDescriptor) {
+                        return env.from(modelDescriptor);
+                    }
+                });
+    }
+
+    public void apply(TableEnvironment env) {
+        assertThatThrownBy(
+                        () -> {
+                            final Table table = toTable(env);
+                            table.executeInsert(sinkName).await();
+                        })
+                .satisfies(anyCauseMatches(expectedException, expectedErrorMessage));
+    }
+
+    public void applyAsSql(TableEnvironment env) {
+        assertThatThrownBy(
+                        () -> {
+                            final Table table = toTable(env);
+                            final String query =
+                                    table.getQueryOperation()
+                                            .asSerializableString(
+                                                    org.apache.flink.table.expressions
+                                                            .DefaultSqlFactory.INSTANCE);
+                            env.executeSql(String.format("INSERT INTO %s %s", sinkName, query))
+                                    .await();
+                        })
+                .satisfies(anyCauseMatches(expectedException, expectedErrorMessage));
+    }
+}

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableApiTestStep.java

@@ -19,6 +19,7 @@
 package org.apache.flink.table.test.program;
 
 import org.apache.flink.table.api.Model;
+import org.apache.flink.table.api.ModelDescriptor;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.TableResult;
@@ -81,6 +82,11 @@ public Table sqlQuery(String query) {
                     public Model fromModel(String modelPath) {
                         return env.fromModelPath(modelPath);
                     }
+
+                    @Override
+                    public Model from(ModelDescriptor modelDescriptor) {
+                        return env.from(modelDescriptor);
+                    }
                 });
     }
 
@@ -120,5 +126,8 @@ public interface TableEnvAccessor {
 
         /** See {@link TableEnvironment#fromModelPath(String)}. */
         Model fromModel(String modelPath);
+
+        /** See {@link TableEnvironment#from(ModelDescriptor)}. */
+        Model from(ModelDescriptor modelDescriptor);
     }
 }

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestProgram.java

@@ -21,6 +21,7 @@
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.table.api.Table;
 import org.apache.flink.table.api.TableRuntimeException;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.expressions.Expression;
 import org.apache.flink.table.functions.UserDefinedFunction;
 import org.apache.flink.table.test.program.FunctionTestStep.FunctionBehavior;
@@ -355,6 +356,22 @@ public Builder runFailingSql(
             return this;
         }
 
+        /**
+         * Run step for executing a Table API query that will fail eventually with either {@link
+         * ValidationException} (during planning time) or {@link TableRuntimeException} (during
+         * execution time).
+         */
+        public Builder runFailingTableApi(
+                Function<TableEnvAccessor, Table> toTable,
+                String sinkName,
+                Class<? extends Exception> expectedException,
+                String expectedErrorMessage) {
+            this.runSteps.add(
+                    new FailingTableApiTestStep(
+                            toTable, sinkName, expectedException, expectedErrorMessage));
+            return this;
+        }
+
         public Builder runTableApi(Function<TableEnvAccessor, Table> toTable, String sinkName) {
             this.runSteps.add(new TableApiTestStep(toTable, sinkName));
             return this;

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TestStep.java

@@ -51,7 +51,8 @@ enum TestKind {
         SINK_WITHOUT_DATA,
         SINK_WITH_DATA,
         SINK_WITH_RESTORE_DATA,
-        FAILING_SQL
+        FAILING_SQL,
+        FAILING_TABLE_API
     }
 
     TestKind getKind();