[FLINK-38257][SQL] Add module dynamic function loading possibility #26933
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| /* | ||
| * 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.state.table.module; | ||
|
|
||
| import org.apache.flink.table.functions.BuiltInFunctionDefinition; | ||
| import org.apache.flink.table.functions.DynamicBuiltInFunctionDefinitionFactory; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public class ExampleDynamicBuiltInFunctionDefinitionFactory | ||
|
There was a problem hiding this comment. api looks minimal and clear. There was a problem hiding this comment. This API is now fully experimental and later subject to change. If the time comes and There was a problem hiding this comment. makes sense, thanks for clarifying bottom line: +1 to keep the api minimal as proposed |
||
| implements DynamicBuiltInFunctionDefinitionFactory { | ||
| @Override | ||
| public String factoryIdentifier() { | ||
| return StateModule.IDENTIFIER + ".example_function_factory"; | ||
| } | ||
|
|
||
| @Override | ||
| public List<BuiltInFunctionDefinition> getBuiltInFunctionDefinitions() { | ||
| return List.of(ExampleDynamicTableFunction.FUNCTION_DEFINITION); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| /* | ||
| * 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.state.table.module; | ||
|
|
||
| import org.apache.flink.annotation.Internal; | ||
| import org.apache.flink.table.annotation.DataTypeHint; | ||
| import org.apache.flink.table.annotation.FunctionHint; | ||
| import org.apache.flink.table.api.DataTypes; | ||
| import org.apache.flink.table.functions.BuiltInFunctionDefinition; | ||
| import org.apache.flink.table.functions.SpecializedFunction; | ||
| import org.apache.flink.table.functions.TableFunction; | ||
| import org.apache.flink.table.types.inference.TypeStrategies; | ||
| import org.apache.flink.types.Row; | ||
|
|
||
| import static org.apache.flink.table.functions.FunctionKind.TABLE; | ||
|
|
||
| @Internal | ||
| @FunctionHint(output = @DataTypeHint("ROW<my-column STRING NOT NULL>")) | ||
| public class ExampleDynamicTableFunction extends TableFunction<Row> { | ||
|
|
||
| public static final BuiltInFunctionDefinition FUNCTION_DEFINITION = | ||
| BuiltInFunctionDefinition.newBuilder() | ||
| .name("example_dynamic_table_function") | ||
| .kind(TABLE) | ||
| .runtimeClass(ExampleDynamicTableFunction.class.getName()) | ||
| .outputTypeStrategy( | ||
| TypeStrategies.explicit( | ||
| DataTypes.ROW( | ||
| DataTypes.FIELD( | ||
| "my-column", DataTypes.STRING().notNull())))) | ||
| .build(); | ||
|
|
||
| public ExampleDynamicTableFunction(SpecializedFunction.SpecializedContext context) {} | ||
|
|
||
| public void eval() { | ||
| Row row = Row.withNames(); | ||
| row.setField("my-column", "my-value"); | ||
| collect(row); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| /* | ||
| * 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.state.table.module; | ||
|
|
||
| import org.apache.flink.api.common.RuntimeExecutionMode; | ||
| import org.apache.flink.configuration.Configuration; | ||
| import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; | ||
| import org.apache.flink.table.api.Table; | ||
| import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; | ||
| import org.apache.flink.types.Row; | ||
|
|
||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import java.util.Iterator; | ||
| import java.util.List; | ||
|
|
||
| import static org.apache.flink.configuration.ExecutionOptions.RUNTIME_MODE; | ||
| import static org.assertj.core.api.Assertions.assertThat; | ||
| import static org.assertj.core.api.Assertions.assertThatThrownBy; | ||
|
|
||
| /** Unit tests for the savepoint SQL reader. */ | ||
| public class StateModuleTest { | ||
| @Test | ||
| public void testDynamicBuiltinFunctionShouldBeLoaded() throws Exception { | ||
| Configuration config = new Configuration(); | ||
| config.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH); | ||
| StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config); | ||
| StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); | ||
|
|
||
| tEnv.executeSql("LOAD MODULE state"); | ||
| Table table = tEnv.sqlQuery("SELECT * FROM example_dynamic_table_function()"); | ||
| List<Row> result = tEnv.toDataStream(table).executeAndCollect(100); | ||
|
|
||
| assertThat(result.size()).isEqualTo(1); | ||
| Iterator<Row> it = result.iterator(); | ||
| assertThat(it.next().toString()).isEqualTo("+I[my-value]"); | ||
| } | ||
|
|
||
| @Test | ||
| public void testMultipleFunctionsWithSameNameShouldThrow() { | ||
| assertThatThrownBy( | ||
| () -> | ||
| StateModule.checkDuplicatedFunctions( | ||
| List.of( | ||
| ExampleDynamicTableFunction.FUNCTION_DEFINITION, | ||
| ExampleDynamicTableFunction.FUNCTION_DEFINITION))) | ||
| .isInstanceOf(IllegalStateException.class) | ||
| .hasMessage("Duplicate function names found: example_dynamic_table_function"); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| # 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. | ||
|
|
||
| org.apache.flink.state.table.module.ExampleDynamicBuiltInFunctionDefinitionFactory |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| /* | ||
| * 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.functions; | ||
|
|
||
| import org.apache.flink.annotation.Experimental; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| /** | ||
| * {@link BuiltInFunctionDefinition} factory for dynamic function registration. This could be useful | ||
| * when functions wanted to be automatically loaded from SQL modules. A good example usage is that | ||
| * state processor API SQL module is providing extra table functions from Kafka connector. | ||
| */ | ||
| @Experimental | ||
| public interface DynamicBuiltInFunctionDefinitionFactory { | ||
| /** | ||
| * Returns the unique identifier of the factory. The suggested pattern is the following: | ||
| * [module-name].[factory-name]. Such case modules can load all [module-name] prefixed functions | ||
| * which belong to them. | ||
| */ | ||
| String factoryIdentifier(); | ||
|
|
||
| /** Returns list of {@link BuiltInFunctionDefinition} which can be registered dynamically. */ | ||
| List<BuiltInFunctionDefinition> getBuiltInFunctionDefinitions(); | ||
| } |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nice to keep the definition near the code. do you think we should also mark that the function should not be pre-calculated by the optimizer? i mean to make it clear that flink must always run it at query time, not try to be clever and “pre-compute” it
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This one is not yet clear🤷🏻♂️
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
maybe i misunderstood what happens here with
savepoint_metadata, but flink's planner can constant-fold and cache function results at plan time if it believes a function is deterministic. that is correct for pure math/string functions, but could be problematic for something like thissavepoint_metadata, which read external, changing state (fs/checkpoint files). and between planning and exec, or between two exec, the metadata could change. so i just thought whether we should tell the planner this function is non-deterministic, so it will not pre-evaluate or cache it, and will always run it at exec time. idk whether theBuiltInFunctionDefinition.newBuilder()can supports to call.deterministic(false), or use and override the UDFisDeterministic()There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That's fair point. I'm not fan of fixing something which is unrelated but this is so tiny that adding
notDeterministic()is fine.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Though it worth the discussion separate whether the planner caches it with input params or not. I would guess yes, otherwise it just doesn't make sense from usage point of view. If the params are stored together with the result then adding
notDeterministic()doesn't hurt but also not helping.