hackergin commented on code in PR #22994: URL: https://github.com/apache/flink/pull/22994#discussion_r1264626431
########## flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ContextResolvedProcedure.java: ########## @@ -0,0 +1,50 @@ +/* + * 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.catalog; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.functions.FunctionIdentifier; +import org.apache.flink.table.procedures.Procedure; + +/** + * This class contains information about a procedure and its relationship with a {@link Catalog}. + * This class is meant for internal usages. Review Comment: I think this line is unnecessary since we already include an `Internal` annotation. ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java: ########## @@ -0,0 +1,246 @@ +/* + * 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.operations; + +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.annotation.ProcedureHint; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test cases for the call statements for {@link SqlNodeToOperationConversion}. */ +public class SqlNodeToCallOperationTest extends SqlNodeToOperationConversionTestBase { + + @BeforeEach + public void before() { + CatalogWithBuiltInProcedure procedureCatalog = + new CatalogWithBuiltInProcedure("procedure_catalog"); + catalogManager.registerCatalog("p1", procedureCatalog); + catalogManager.setCurrentCatalog("p1"); + } + + @Test + public void testCallStatement() { Review Comment: we can add a test case for unsupported expression. ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlProcedure.java: ########## @@ -0,0 +1,86 @@ +/* + * 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.bridging; + +import org.apache.flink.table.catalog.ContextResolvedProcedure; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.table.procedures.ProcedureDefinition; +import org.apache.flink.table.types.extraction.TypeInferenceExtractor; +import org.apache.flink.table.types.inference.TypeInference; + +import org.apache.calcite.sql.SqlFunction; +import org.apache.calcite.sql.SqlFunctionCategory; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.type.SqlOperandTypeChecker; +import org.apache.calcite.sql.type.SqlOperandTypeInference; +import org.apache.calcite.sql.type.SqlReturnTypeInference; +import org.checkerframework.checker.nullness.qual.Nullable; + +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createName; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlIdentifier; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlOperandTypeChecker; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlOperandTypeInference; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlReturnTypeInference; + +/** Bridges {@link Procedure} to Calcite's representation of a function. */ +public class BridgingSqlProcedure extends SqlFunction { + + private final ContextResolvedProcedure contextResolvedProcedure; + + public BridgingSqlProcedure( Review Comment: nit: private ########## flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/environment/SqlGatewayStreamExecutionEnvironment.java: ########## @@ -0,0 +1,38 @@ +/* + * 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.gateway.environment; + +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory; + +/** + * The SqlGatewayStreamExecutionEnvironment is a StreamExecutionEnvironment that runs the program Review Comment: ```suggestion * The SqlGatewayStreamExecutionEnvironment is a {@link StreamExecutionEnvironment} that runs the program ``` ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlProcedure.java: ########## @@ -0,0 +1,86 @@ +/* + * 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.bridging; + +import org.apache.flink.table.catalog.ContextResolvedProcedure; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.table.procedures.ProcedureDefinition; +import org.apache.flink.table.types.extraction.TypeInferenceExtractor; +import org.apache.flink.table.types.inference.TypeInference; + +import org.apache.calcite.sql.SqlFunction; +import org.apache.calcite.sql.SqlFunctionCategory; +import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.type.SqlOperandTypeChecker; +import org.apache.calcite.sql.type.SqlOperandTypeInference; +import org.apache.calcite.sql.type.SqlReturnTypeInference; +import org.checkerframework.checker.nullness.qual.Nullable; + +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createName; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlIdentifier; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlOperandTypeChecker; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlOperandTypeInference; +import static org.apache.flink.table.planner.functions.bridging.BridgingUtils.createSqlReturnTypeInference; + +/** Bridges {@link Procedure} to Calcite's representation of a function. */ +public class BridgingSqlProcedure extends SqlFunction { + + private final ContextResolvedProcedure contextResolvedProcedure; + + public BridgingSqlProcedure( + String name, + SqlIdentifier sqlIdentifier, + @Nullable SqlReturnTypeInference returnTypeInference, + @Nullable SqlOperandTypeInference operandTypeInference, + @Nullable SqlOperandTypeChecker operandTypeChecker, + SqlFunctionCategory category, + ContextResolvedProcedure contextResolvedProcedure) { + super( + name, + sqlIdentifier, + SqlKind.OTHER_FUNCTION, + returnTypeInference, + operandTypeInference, + operandTypeChecker, + category); + this.contextResolvedProcedure = contextResolvedProcedure; + } + + public ContextResolvedProcedure getContextResolveProcedure() { + return contextResolvedProcedure; + } + + public static BridgingSqlProcedure of( Review Comment: nit: no java doc. ########## flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/operations/PlannerCallProcedureOperation.java: ########## @@ -0,0 +1,352 @@ +/* + * 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.operations; + +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.ResultKind; +import org.apache.flink.table.api.TableConfig; +import org.apache.flink.table.api.TableException; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.config.ExecutionConfigOptions; +import org.apache.flink.table.api.internal.ResultProvider; +import org.apache.flink.table.api.internal.TableResultImpl; +import org.apache.flink.table.api.internal.TableResultInternal; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.conversion.DataStructureConverter; +import org.apache.flink.table.data.conversion.DataStructureConverters; +import org.apache.flink.table.operations.CallProcedureOperation; +import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.operations.OperationUtils; +import org.apache.flink.table.planner.functions.casting.RowDataToStringConverterImpl; +import org.apache.flink.table.procedure.DefaultProcedureContext; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.table.procedures.ProcedureDefinition; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.extraction.ExtractionUtils; +import org.apache.flink.table.types.logical.utils.LogicalTypeChecks; +import org.apache.flink.table.types.utils.DataTypeUtils; +import org.apache.flink.table.utils.print.RowDataToStringConverter; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.lang.reflect.Array; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.time.ZoneId; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static org.apache.flink.table.types.extraction.ExtractionUtils.isAssignable; + +/** Wrapper for valid call procedure operation generated by Planner. */ +public class PlannerCallProcedureOperation implements CallProcedureOperation { + + private final ObjectIdentifier procedureIdentifier; + private final Procedure procedure; + + /** The internal represent for input arguments. */ + private final Object[] internalInputArguments; + + private final DataType[] inputTypes; + private final DataType outputType; + + public PlannerCallProcedureOperation( + ObjectIdentifier procedureIdentifier, + Procedure procedure, + Object[] internalInputArguments, + DataType[] inputTypes, + DataType outputType) { + this.procedureIdentifier = procedureIdentifier; + this.procedure = procedure; + this.internalInputArguments = internalInputArguments; + this.inputTypes = inputTypes; + this.outputType = outputType; + } + + @Override + public TableResultInternal execute(Context ctx) { + TableConfig tableConfig = ctx.getTableConfig(); + ClassLoader userClassLoader = ctx.getResourceManager().getUserClassLoader(); + + // get the class for the args + Class<?>[] argumentClz = new Class[1 + inputTypes.length]; + argumentClz[0] = ProcedureContext.class; + for (int i = 0; i < inputTypes.length; i++) { + argumentClz[i + 1] = inputTypes[i].getConversionClass(); + } + + // get the value for the args + Object[] argumentVal = new Object[1 + internalInputArguments.length]; + StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(tableConfig.getConfiguration()); + argumentVal[0] = new DefaultProcedureContext(env); + for (int i = 0; i < internalInputArguments.length; i++) { + argumentVal[i + 1] = + toExternal(internalInputArguments[i], inputTypes[i], userClassLoader); + } + + // call the procedure, get result + Object procedureResult = callProcedure(procedure, argumentClz, argumentVal); + + // get result converter + ZoneId zoneId = tableConfig.getLocalTimeZone(); + DataType tableResultType = outputType; + // if is not composite type, wrap it to composited type + if (!LogicalTypeChecks.isCompositeType(outputType.getLogicalType())) { + tableResultType = DataTypes.ROW(DataTypes.FIELD("result", tableResultType)); + } + + ResolvedSchema resultSchema = DataTypeUtils.expandCompositeTypeToSchema(tableResultType); + RowDataToStringConverter rowDataToStringConverter = + new RowDataToStringConverterImpl( + tableResultType, + zoneId, + userClassLoader, + tableConfig + .get(ExecutionConfigOptions.TABLE_EXEC_LEGACY_CAST_BEHAVIOUR) + .isEnabled()); + // create DataStructure converters + DataStructureConverter<Object, Object> converter = + DataStructureConverters.getConverter(outputType); + converter.open(userClassLoader); + + return TableResultImpl.builder() + .resultProvider( + new CallProcedureResultProvider( + converter, rowDataToStringConverter, procedureResult)) + .schema(resultSchema) + .resultKind(ResultKind.SUCCESS_WITH_CONTENT) + .build(); + } + + private Object toExternal(Object internalValue, DataType inputType, ClassLoader classLoader) { + if (!(DataTypeUtils.isInternal(inputType))) { + // if the expected input type of the procedure is not internal type, + // which means the converted Flink internal value doesn't + // match the expected input type, then we need to convert the Flink + // internal value to external value + DataStructureConverter<Object, Object> converter = + DataStructureConverters.getConverter(inputType); + converter.open(classLoader); + return converter.toExternal(internalValue); + } else { + return internalValue; + } + } + + private Object callProcedure(Procedure procedure, Class<?>[] inputClz, Object[] inputArgs) { + String callMethodName = ProcedureDefinition.PROCEDURE_CALL; + + final List<Method> methods = + ExtractionUtils.collectMethods(procedure.getClass(), callMethodName); + Optional<Method> optionalCallMethod = + methods.stream() + .filter( + method -> + ExtractionUtils.isInvokable(method, inputClz) + && method.getReturnType().isArray() + && isAssignable( + outputType.getConversionClass(), + method.getReturnType().getComponentType(), + true)) + .findAny(); Review Comment: Is it possible to have multiple methods here? If so, we can use `findFirst` to ensure the stability of the results. Or in such situation, an exception should be thrown directly. ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestProcedureCatalogFactory.java: ########## @@ -0,0 +1,217 @@ +/* + * 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.factories; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** A factory to create a catalog with some built-in procedures for test purpose. */ +public class TestProcedureCatalogFactory implements CatalogFactory { + private static final String IDENTIFIER = "test_procedure_catalog"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } + + @Override + public Catalog createCatalog(Context context) { + final FactoryUtil.CatalogFactoryHelper helper = + FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + return new CatalogWithBuiltInProcedure(context.getName()); + } + + /** A catalog with some built-in procedures for test purpose. */ + public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog { + + private static final Map<ObjectPath, Procedure> PROCEDURE_MAP = new HashMap<>(); + + static { + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.sum_n"), new SumProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.get_year"), new GetYearProcedure()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_user"), new GenerateUserProcedure()); + } + + public CatalogWithBuiltInProcedure(String name) { + super(name); + } + + @Override + public List<String> listProcedures(String dbName) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(dbName)) { + throw new DatabaseNotExistException(getName(), dbName); + } + return PROCEDURE_MAP.keySet().stream() + .filter(procedurePath -> procedurePath.getDatabaseName().equals(dbName)) + .map(ObjectPath::getObjectName) + .collect(Collectors.toList()); + } + + @Override + public Procedure getProcedure(ObjectPath procedurePath) + throws ProcedureNotExistException, CatalogException { + if (PROCEDURE_MAP.containsKey(procedurePath)) { + return PROCEDURE_MAP.get(procedurePath); + } else { + throw new ProcedureNotExistException(getName(), procedurePath); + } + } + } + + /** A procedure to a sequence from 0 to n for test purpose. */ Review Comment: ```suggestion /** A procedure to a sequence from 0 to n for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestProcedureCatalogFactory.java: ########## @@ -0,0 +1,217 @@ +/* + * 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.factories; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** A factory to create a catalog with some built-in procedures for test purpose. */ +public class TestProcedureCatalogFactory implements CatalogFactory { + private static final String IDENTIFIER = "test_procedure_catalog"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } + + @Override + public Catalog createCatalog(Context context) { + final FactoryUtil.CatalogFactoryHelper helper = + FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + return new CatalogWithBuiltInProcedure(context.getName()); + } + + /** A catalog with some built-in procedures for test purpose. */ + public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog { + + private static final Map<ObjectPath, Procedure> PROCEDURE_MAP = new HashMap<>(); + + static { + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.sum_n"), new SumProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.get_year"), new GetYearProcedure()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_user"), new GenerateUserProcedure()); + } + + public CatalogWithBuiltInProcedure(String name) { + super(name); + } + + @Override + public List<String> listProcedures(String dbName) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(dbName)) { + throw new DatabaseNotExistException(getName(), dbName); + } + return PROCEDURE_MAP.keySet().stream() + .filter(procedurePath -> procedurePath.getDatabaseName().equals(dbName)) + .map(ObjectPath::getObjectName) + .collect(Collectors.toList()); + } + + @Override + public Procedure getProcedure(ObjectPath procedurePath) + throws ProcedureNotExistException, CatalogException { + if (PROCEDURE_MAP.containsKey(procedurePath)) { + return PROCEDURE_MAP.get(procedurePath); + } else { + throw new ProcedureNotExistException(getName(), procedurePath); + } + } + } + + /** A procedure to a sequence from 0 to n for test purpose. */ + public static class GenerateSequenceProcedure implements Procedure { + public long[] call(ProcedureContext procedureContext, int n) throws Exception { + return generate(procedureContext.getExecutionEnvironment(), n); + } + + public long[] call(ProcedureContext procedureContext, int n, String runTimeMode) + throws Exception { + StreamExecutionEnvironment env = procedureContext.getExecutionEnvironment(); + env.setRuntimeMode(RuntimeExecutionMode.valueOf(runTimeMode)); + return generate(env, n); + } + + private long[] generate(StreamExecutionEnvironment env, int n) throws Exception { + env.setParallelism(1); + long[] sequenceN = new long[n]; + int i = 0; + try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) { + while (result.hasNext()) { + sequenceN[i++] = result.next(); + } + } + return sequenceN; + } + } + + /** A procedure to sum decimal values for test purpose. */ + public static class SumProcedure implements Procedure { + public @DataTypeHint("ROW< sum_value decimal(10, 2), count INT >") Row[] call( + ProcedureContext procedureContext, + @DataTypeHint("DECIMAL(10, 2)") BigDecimal... inputs) { + if (inputs.length == 0) { + return new Row[] {Row.of(null, 0)}; + } + int counts = inputs.length; + BigDecimal result = inputs[0]; + for (int i = 1; i < inputs.length; i++) { + result = result.add(inputs[i]); + } + return new Row[] {Row.of(result, counts)}; + } + } + + /** A procedure to get year from the passed timestamp parameter for test purpose. */ + public static class GetYearProcedure implements Procedure { + public String[] call(ProcedureContext procedureContext, LocalDateTime... timestamps) { + String[] results = new String[timestamps.length]; + for (int i = 0; i < results.length; i++) { + results[i] = String.valueOf(timestamps[i].getYear()); + } + return results; + } + } + + /** A procedure to generate a user according to the passed parameters for test purpose. */ Review Comment: ```suggestion /** A procedure to generate a user according to the passed parameters for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/sql/ProcedureITCase.java: ########## @@ -110,40 +111,72 @@ void testShowProcedures() { CollectionUtil.iteratorToList( tEnv().executeSql("show procedures in `system` not like 'generate%'") .collect()); - assertThat(rows.toString()) - .isEqualTo("[+I[get_year], +I[miss_procedure_context], +I[sum_n]]"); + assertThat(rows.toString()).isEqualTo("[+I[get_year], +I[sum_n]]"); // show procedure with not ilike rows = CollectionUtil.iteratorToList( tEnv().executeSql("show procedures in `system` not ilike 'generaTe%'") .collect()); - assertThat(rows.toString()) - .isEqualTo("[+I[get_year], +I[miss_procedure_context], +I[sum_n]]"); + assertThat(rows.toString()).isEqualTo("[+I[get_year], +I[sum_n]]"); + } + + @Test + void testCallProcedure() { Review Comment: I believe one of the main scenarios for calling the producer is to submit Flink jobs to handle certain tasks, so I think we can add a test case for submitting Flink jobs. ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java: ########## @@ -0,0 +1,246 @@ +/* + * 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.operations; + +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.annotation.ProcedureHint; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test cases for the call statements for {@link SqlNodeToOperationConversion}. */ +public class SqlNodeToCallOperationTest extends SqlNodeToOperationConversionTestBase { + + @BeforeEach + public void before() { + CatalogWithBuiltInProcedure procedureCatalog = + new CatalogWithBuiltInProcedure("procedure_catalog"); + catalogManager.registerCatalog("p1", procedureCatalog); + catalogManager.setCurrentCatalog("p1"); + } + + @Test + public void testCallStatement() { + // test call the procedure which accepts primitive types as arguments + String sql = "call `system`.primitive_arg(1, 2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`primitive_arg`]," + + " inputTypes: [INT NOT NULL, BIGINT NOT NULL], outputTypes: [INT NOT NULL], arguments: [1, 2])"); + + // test call the procedure which has different type mapping for single method + // call with int + sql = "call `system`.different_type_mapping(1)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`different_type_mapping`]," + + " inputTypes: [INT], outputTypes: [INT], arguments: [1])"); + // call with bigint + sql = "call `system`.different_type_mapping(cast(1 as bigint))"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`different_type_mapping`]," + + " inputTypes: [BIGINT], outputTypes: [BIGINT], arguments: [1])"); + + // test call the procedure which has var arguments + sql = "call `system`.var_arg(1)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`var_arg`]," + + " inputTypes: [INT NOT NULL], outputTypes: [STRING], arguments: [1])"); + sql = "call `system`.var_arg(1, 2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`var_arg`]," + + " inputTypes: [INT NOT NULL, INT NOT NULL], outputTypes: [STRING], arguments: [1, 2])"); + sql = "call `system`.var_arg(1, 2, 1 + 2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`var_arg`]," + + " inputTypes: [INT NOT NULL, INT NOT NULL, INT NOT NULL], outputTypes: [STRING], arguments: [1, 2, 3])"); + + // test call the procedure with row as result and decimal as argument as well as + // explict/implicit cast + sql = "call `system`.row_result(cast(1.2 as decimal))"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`row_result`]," + + " inputTypes: [DECIMAL(10, 2)], outputTypes: [ROW<`i` INT>], arguments: [1.20])"); + sql = "call `system`.row_result(1.2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`row_result`]," + + " inputTypes: [DECIMAL(10, 2)], outputTypes: [ROW<`i` INT>], arguments: [1.20])"); + + // test call the procedure with pojo as result + sql = "call p1.`system`.pojo_result('name', 1)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`pojo_result`]," + + " inputTypes: [STRING, BIGINT NOT NULL]," + + " outputTypes: [*org.apache.flink.table.planner.operations.SqlNodeToCallOperationTest$MyPojo<`name` STRING, `id` BIGINT NOT NULL>*]," + + " arguments: [name, 1])"); + + // test call the procedure with timestamp as arguments + sql = + "call p1.`system`.timestamp_arg(timestamp '2023-04-22 00:00:00.300', " + + "timestamp '2023-04-22 00:00:00.300' + INTERVAL '1' day ) "; + verifyCallOperation( + sql, + String.format( + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`timestamp_arg`]," + + " inputTypes: [TIMESTAMP(3), TIMESTAMP(3)], outputTypes: [TIMESTAMP(3)]," + + " arguments: [%s, %s])", + LocalDateTime.parse("2023-04-22T00:00:00.300"), + LocalDateTime.parse("2023-04-23T00:00:00.300"))); + // should throw exception when the signature doesn't match + assertThatThrownBy(() -> parse("call `system`.primitive_arg(1)")) + .hasMessageContaining( + "No match found for function signature primitive_arg(<NUMERIC>)"); + } + + private void verifyCallOperation(String sql, String expectSummary) { + Operation operation = parse(sql); + assertThat(operation).isInstanceOf(PlannerCallProcedureOperation.class); + assertThat(parse(sql).asSummaryString()).isEqualTo(expectSummary); + } + + /** A catalog with some built-in procedures for test purpose. */ + private static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog { + + private static final Map<ObjectPath, Procedure> PROCEDURE_MAP = new HashMap<>(); + + static { + PROCEDURE_MAP.put( + ObjectPath.fromString("system.primitive_arg"), new ProcedureWithPrimitiveArg()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.different_type_mapping"), + new DifferentTypeMappingProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.var_arg"), new VarArgProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.row_result"), new RowResultProcedure()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.pojo_result"), new PojoResultProcedure()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.timestamp_arg"), new TimeStampArgProcedure()); + } + + public CatalogWithBuiltInProcedure(String name) { + super(name); + } + + @Override + public Procedure getProcedure(ObjectPath procedurePath) + throws ProcedureNotExistException, CatalogException { + if (PROCEDURE_MAP.containsKey(procedurePath)) { + return PROCEDURE_MAP.get(procedurePath); + } else { + throw new ProcedureNotExistException(getName(), procedurePath); + } + } + } + + private static class ProcedureWithPrimitiveArg implements Procedure { + public int[] call(ProcedureContext context, int arg1, long arg2) { + return null; + } + } + + @ProcedureHint(input = @DataTypeHint("INT"), output = @DataTypeHint("INT")) + @ProcedureHint(input = @DataTypeHint("BIGINT"), output = @DataTypeHint("BIGINT")) + private static class DifferentTypeMappingProcedure implements Procedure { + public Number[] call(ProcedureContext procedureContext, Number n) { + return null; + } + } + + private static class VarArgProcedure implements Procedure { + public String[] call(ProcedureContext procedureContext, int i, int... more) { + return null; + } + } + + private static class RowResultProcedure implements Procedure { + public @DataTypeHint("ROW<i INT>") Row[] call( + ProcedureContext procedureContext, + @DataTypeHint("DECIMAL(10, 2)") BigDecimal decimal) { + return null; + } + } + + private static class PojoResultProcedure implements Procedure { + public MyPojo[] call(ProcedureContext procedureContext, String name, long id) { + return new MyPojo[0]; + } + } + + private static class TimeStampArgProcedure implements Procedure { + public @DataTypeHint("TIMESTAMP(3)") LocalDateTime[] call( + ProcedureContext procedureContext, + @DataTypeHint("TIMESTAMP(3)") LocalDateTime localDateTime, + @DataTypeHint("TIMESTAMP(3)") TimestampData timestampData) { + return null; + } + } + + /** A simple pojo class for test purpose. */ Review Comment: ```suggestion /** A simple pojo class for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/calcite/FlinkPlannerImpl.scala: ########## @@ -182,6 +182,18 @@ class FlinkPlannerImpl( case compileAndExecute: SqlCompileAndExecutePlan => compileAndExecute.setOperand(0, validate(compileAndExecute.getOperandList.get(0))) compileAndExecute + // for call procedure statement + case sqlCallNode Review Comment: nit: Can we use sqlNode.getKind here ? ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestProcedureCatalogFactory.java: ########## @@ -0,0 +1,217 @@ +/* + * 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.factories; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** A factory to create a catalog with some built-in procedures for test purpose. */ +public class TestProcedureCatalogFactory implements CatalogFactory { + private static final String IDENTIFIER = "test_procedure_catalog"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } + + @Override + public Catalog createCatalog(Context context) { + final FactoryUtil.CatalogFactoryHelper helper = + FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + return new CatalogWithBuiltInProcedure(context.getName()); + } + + /** A catalog with some built-in procedures for test purpose. */ Review Comment: ```suggestion /** A catalog with some built-in procedures for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestProcedureCatalogFactory.java: ########## @@ -0,0 +1,217 @@ +/* + * 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.factories; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** A factory to create a catalog with some built-in procedures for test purpose. */ +public class TestProcedureCatalogFactory implements CatalogFactory { + private static final String IDENTIFIER = "test_procedure_catalog"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } + + @Override + public Catalog createCatalog(Context context) { + final FactoryUtil.CatalogFactoryHelper helper = + FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + return new CatalogWithBuiltInProcedure(context.getName()); + } + + /** A catalog with some built-in procedures for test purpose. */ + public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog { + + private static final Map<ObjectPath, Procedure> PROCEDURE_MAP = new HashMap<>(); + + static { + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.sum_n"), new SumProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.get_year"), new GetYearProcedure()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_user"), new GenerateUserProcedure()); + } + + public CatalogWithBuiltInProcedure(String name) { + super(name); + } + + @Override + public List<String> listProcedures(String dbName) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(dbName)) { + throw new DatabaseNotExistException(getName(), dbName); + } + return PROCEDURE_MAP.keySet().stream() + .filter(procedurePath -> procedurePath.getDatabaseName().equals(dbName)) + .map(ObjectPath::getObjectName) + .collect(Collectors.toList()); + } + + @Override + public Procedure getProcedure(ObjectPath procedurePath) + throws ProcedureNotExistException, CatalogException { + if (PROCEDURE_MAP.containsKey(procedurePath)) { + return PROCEDURE_MAP.get(procedurePath); + } else { + throw new ProcedureNotExistException(getName(), procedurePath); + } + } + } + + /** A procedure to a sequence from 0 to n for test purpose. */ + public static class GenerateSequenceProcedure implements Procedure { + public long[] call(ProcedureContext procedureContext, int n) throws Exception { + return generate(procedureContext.getExecutionEnvironment(), n); + } + + public long[] call(ProcedureContext procedureContext, int n, String runTimeMode) + throws Exception { + StreamExecutionEnvironment env = procedureContext.getExecutionEnvironment(); + env.setRuntimeMode(RuntimeExecutionMode.valueOf(runTimeMode)); + return generate(env, n); + } + + private long[] generate(StreamExecutionEnvironment env, int n) throws Exception { + env.setParallelism(1); + long[] sequenceN = new long[n]; + int i = 0; + try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) { + while (result.hasNext()) { + sequenceN[i++] = result.next(); + } + } + return sequenceN; + } + } + + /** A procedure to sum decimal values for test purpose. */ + public static class SumProcedure implements Procedure { + public @DataTypeHint("ROW< sum_value decimal(10, 2), count INT >") Row[] call( + ProcedureContext procedureContext, + @DataTypeHint("DECIMAL(10, 2)") BigDecimal... inputs) { + if (inputs.length == 0) { + return new Row[] {Row.of(null, 0)}; + } + int counts = inputs.length; + BigDecimal result = inputs[0]; + for (int i = 1; i < inputs.length; i++) { + result = result.add(inputs[i]); + } + return new Row[] {Row.of(result, counts)}; + } + } + + /** A procedure to get year from the passed timestamp parameter for test purpose. */ Review Comment: ```suggestion /** A procedure to get year from the passed timestamp parameter for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestProcedureCatalogFactory.java: ########## @@ -0,0 +1,217 @@ +/* + * 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.factories; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** A factory to create a catalog with some built-in procedures for test purpose. */ +public class TestProcedureCatalogFactory implements CatalogFactory { + private static final String IDENTIFIER = "test_procedure_catalog"; + + @Override + public String factoryIdentifier() { + return IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } + + @Override + public Catalog createCatalog(Context context) { + final FactoryUtil.CatalogFactoryHelper helper = + FactoryUtil.createCatalogFactoryHelper(this, context); + helper.validate(); + return new CatalogWithBuiltInProcedure(context.getName()); + } + + /** A catalog with some built-in procedures for test purpose. */ + public static class CatalogWithBuiltInProcedure extends GenericInMemoryCatalog { + + private static final Map<ObjectPath, Procedure> PROCEDURE_MAP = new HashMap<>(); + + static { + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_n"), new GenerateSequenceProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.sum_n"), new SumProcedure()); + PROCEDURE_MAP.put(ObjectPath.fromString("system.get_year"), new GetYearProcedure()); + PROCEDURE_MAP.put( + ObjectPath.fromString("system.generate_user"), new GenerateUserProcedure()); + } + + public CatalogWithBuiltInProcedure(String name) { + super(name); + } + + @Override + public List<String> listProcedures(String dbName) + throws DatabaseNotExistException, CatalogException { + if (!databaseExists(dbName)) { + throw new DatabaseNotExistException(getName(), dbName); + } + return PROCEDURE_MAP.keySet().stream() + .filter(procedurePath -> procedurePath.getDatabaseName().equals(dbName)) + .map(ObjectPath::getObjectName) + .collect(Collectors.toList()); + } + + @Override + public Procedure getProcedure(ObjectPath procedurePath) + throws ProcedureNotExistException, CatalogException { + if (PROCEDURE_MAP.containsKey(procedurePath)) { + return PROCEDURE_MAP.get(procedurePath); + } else { + throw new ProcedureNotExistException(getName(), procedurePath); + } + } + } + + /** A procedure to a sequence from 0 to n for test purpose. */ + public static class GenerateSequenceProcedure implements Procedure { + public long[] call(ProcedureContext procedureContext, int n) throws Exception { + return generate(procedureContext.getExecutionEnvironment(), n); + } + + public long[] call(ProcedureContext procedureContext, int n, String runTimeMode) + throws Exception { + StreamExecutionEnvironment env = procedureContext.getExecutionEnvironment(); + env.setRuntimeMode(RuntimeExecutionMode.valueOf(runTimeMode)); + return generate(env, n); + } + + private long[] generate(StreamExecutionEnvironment env, int n) throws Exception { + env.setParallelism(1); + long[] sequenceN = new long[n]; + int i = 0; + try (CloseableIterator<Long> result = env.fromSequence(0, n - 1).executeAndCollect()) { + while (result.hasNext()) { + sequenceN[i++] = result.next(); + } + } + return sequenceN; + } + } + + /** A procedure to sum decimal values for test purpose. */ Review Comment: ```suggestion /** A procedure to sum decimal values for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/operations/SqlNodeToCallOperationTest.java: ########## @@ -0,0 +1,246 @@ +/* + * 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.operations; + +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.annotation.ProcedureHint; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.operations.Operation; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Test cases for the call statements for {@link SqlNodeToOperationConversion}. */ +public class SqlNodeToCallOperationTest extends SqlNodeToOperationConversionTestBase { + + @BeforeEach + public void before() { + CatalogWithBuiltInProcedure procedureCatalog = + new CatalogWithBuiltInProcedure("procedure_catalog"); + catalogManager.registerCatalog("p1", procedureCatalog); + catalogManager.setCurrentCatalog("p1"); + } + + @Test + public void testCallStatement() { + // test call the procedure which accepts primitive types as arguments + String sql = "call `system`.primitive_arg(1, 2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`primitive_arg`]," + + " inputTypes: [INT NOT NULL, BIGINT NOT NULL], outputTypes: [INT NOT NULL], arguments: [1, 2])"); + + // test call the procedure which has different type mapping for single method + // call with int + sql = "call `system`.different_type_mapping(1)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`different_type_mapping`]," + + " inputTypes: [INT], outputTypes: [INT], arguments: [1])"); + // call with bigint + sql = "call `system`.different_type_mapping(cast(1 as bigint))"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`different_type_mapping`]," + + " inputTypes: [BIGINT], outputTypes: [BIGINT], arguments: [1])"); + + // test call the procedure which has var arguments + sql = "call `system`.var_arg(1)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`var_arg`]," + + " inputTypes: [INT NOT NULL], outputTypes: [STRING], arguments: [1])"); + sql = "call `system`.var_arg(1, 2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`var_arg`]," + + " inputTypes: [INT NOT NULL, INT NOT NULL], outputTypes: [STRING], arguments: [1, 2])"); + sql = "call `system`.var_arg(1, 2, 1 + 2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`var_arg`]," + + " inputTypes: [INT NOT NULL, INT NOT NULL, INT NOT NULL], outputTypes: [STRING], arguments: [1, 2, 3])"); + + // test call the procedure with row as result and decimal as argument as well as + // explict/implicit cast + sql = "call `system`.row_result(cast(1.2 as decimal))"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`row_result`]," + + " inputTypes: [DECIMAL(10, 2)], outputTypes: [ROW<`i` INT>], arguments: [1.20])"); + sql = "call `system`.row_result(1.2)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`row_result`]," + + " inputTypes: [DECIMAL(10, 2)], outputTypes: [ROW<`i` INT>], arguments: [1.20])"); + + // test call the procedure with pojo as result + sql = "call p1.`system`.pojo_result('name', 1)"; + verifyCallOperation( + sql, + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`pojo_result`]," + + " inputTypes: [STRING, BIGINT NOT NULL]," + + " outputTypes: [*org.apache.flink.table.planner.operations.SqlNodeToCallOperationTest$MyPojo<`name` STRING, `id` BIGINT NOT NULL>*]," + + " arguments: [name, 1])"); + + // test call the procedure with timestamp as arguments + sql = + "call p1.`system`.timestamp_arg(timestamp '2023-04-22 00:00:00.300', " + + "timestamp '2023-04-22 00:00:00.300' + INTERVAL '1' day ) "; + verifyCallOperation( + sql, + String.format( + "CALL PROCEDURE:" + + " (procedureIdentifier: [`p1`.`system`.`timestamp_arg`]," + + " inputTypes: [TIMESTAMP(3), TIMESTAMP(3)], outputTypes: [TIMESTAMP(3)]," + + " arguments: [%s, %s])", + LocalDateTime.parse("2023-04-22T00:00:00.300"), + LocalDateTime.parse("2023-04-23T00:00:00.300"))); + // should throw exception when the signature doesn't match + assertThatThrownBy(() -> parse("call `system`.primitive_arg(1)")) + .hasMessageContaining( + "No match found for function signature primitive_arg(<NUMERIC>)"); + } + + private void verifyCallOperation(String sql, String expectSummary) { + Operation operation = parse(sql); + assertThat(operation).isInstanceOf(PlannerCallProcedureOperation.class); + assertThat(parse(sql).asSummaryString()).isEqualTo(expectSummary); + } + + /** A catalog with some built-in procedures for test purpose. */ Review Comment: ```suggestion /** A catalog with some built-in procedures for testing purposes. */ ``` ########## flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestProcedureCatalogFactory.java: ########## @@ -0,0 +1,217 @@ +/* + * 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.factories; + +import org.apache.flink.api.common.RuntimeExecutionMode; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.annotation.DataTypeHint; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.GenericInMemoryCatalog; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.ProcedureNotExistException; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; +import org.apache.flink.table.procedure.ProcedureContext; +import org.apache.flink.table.procedures.Procedure; +import org.apache.flink.types.Row; +import org.apache.flink.util.CloseableIterator; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +/** A factory to create a catalog with some built-in procedures for test purpose. */ Review Comment: ```suggestion /** A factory to create a catalog with some built-in procedures for testing purposes. */ ``` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
