You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2021/03/28 15:24:51 UTC
[flink] branch master updated: [FLINK-21738][table] Caching
built-in functions for CoreModule and HiveModule
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 912fc8d [FLINK-21738][table] Caching built-in functions for CoreModule and HiveModule
912fc8d is described below
commit 912fc8defc82c3509e498356743f6a0ff493e413
Author: zoucao <zh...@hotmail.com>
AuthorDate: Thu Mar 18 00:50:08 2021 +0800
[FLINK-21738][table] Caching built-in functions for CoreModule and HiveModule
This closes #15267
---
.../apache/flink/table/module/hive/HiveModule.java | 11 ++++++++---
.../apache/flink/table/module/ModuleManager.java | 18 +++++++----------
.../org/apache/flink/table/module/CoreModule.java | 23 +++++++++++++++++-----
3 files changed, 33 insertions(+), 19 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/module/hive/HiveModule.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/module/hive/HiveModule.java
index d82a6da..a7d70a3 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/module/hive/HiveModule.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/module/hive/HiveModule.java
@@ -74,6 +74,7 @@ public class HiveModule implements Module {
private final HiveFunctionDefinitionFactory factory;
private final String hiveVersion;
private final HiveShim hiveShim;
+ private Set<String> functionNames;
public HiveModule() {
this(HiveShimLoader.getHiveVersion());
@@ -86,13 +87,17 @@ public class HiveModule implements Module {
this.hiveVersion = hiveVersion;
this.hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
this.factory = new HiveFunctionDefinitionFactory(hiveShim);
+ this.functionNames = new HashSet<>();
}
@Override
public Set<String> listFunctions() {
- Set<String> builtInFuncs = hiveShim.listBuiltInFunctions();
- builtInFuncs.removeAll(BUILT_IN_FUNC_BLACKLIST);
- return builtInFuncs;
+ // lazy initialize
+ if (functionNames.isEmpty()) {
+ functionNames = hiveShim.listBuiltInFunctions();
+ functionNames.removeAll(BUILT_IN_FUNC_BLACKLIST);
+ }
+ return functionNames;
}
@Override
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/module/ModuleManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/module/ModuleManager.java
index b0f5b5f..28ac675 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/module/ModuleManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/module/ModuleManager.java
@@ -173,19 +173,15 @@ public class ModuleManager {
* @return an optional of {@link FunctionDefinition}
*/
public Optional<FunctionDefinition> getFunctionDefinition(String name) {
- Optional<String> module =
- usedModules.stream()
- .filter(
- n ->
- loadedModules.get(n).listFunctions().stream()
- .anyMatch(e -> e.equalsIgnoreCase(name)))
- .findFirst();
- if (module.isPresent()) {
- LOG.debug("Got FunctionDefinition '{}' from '{}' module.", name, module.get());
- return loadedModules.get(module.get()).getFunctionDefinition(name);
+ for (String moduleName : usedModules) {
+ if (loadedModules.get(moduleName).listFunctions().stream()
+ .anyMatch(name::equalsIgnoreCase)) {
+ LOG.debug("Got FunctionDefinition '{}' from '{}' module.", name, moduleName);
+ return loadedModules.get(moduleName).getFunctionDefinition(name);
+ }
}
- LOG.debug("Cannot find FunctionDefinition '{}' from any loaded modules.", name);
+ LOG.debug("Cannot find FunctionDefinition '{}' from any loaded modules.", name);
return Optional.empty();
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/module/CoreModule.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/module/CoreModule.java
index 5dcaa80..e3ef58e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/module/CoreModule.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/module/CoreModule.java
@@ -18,9 +18,12 @@
package org.apache.flink.table.module;
+import org.apache.flink.table.functions.BuiltInFunctionDefinition;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.table.functions.FunctionDefinition;
+import java.util.HashSet;
+import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.function.Function;
@@ -29,19 +32,29 @@ import java.util.stream.Collectors;
/** Module of default core metadata in Flink. */
public class CoreModule implements Module {
public static final CoreModule INSTANCE = new CoreModule();
+ private final List<BuiltInFunctionDefinition> functionDefinitions;
+ private Set<String> functionNames;
- private CoreModule() {}
+ private CoreModule() {
+ this.functionDefinitions = BuiltInFunctionDefinitions.getDefinitions();
+ this.functionNames = new HashSet<>();
+ }
@Override
public Set<String> listFunctions() {
- return BuiltInFunctionDefinitions.getDefinitions().stream()
- .map(f -> f.getName())
- .collect(Collectors.toSet());
+ // lazy initialize
+ if (functionNames.isEmpty()) {
+ functionNames =
+ functionDefinitions.stream()
+ .map(BuiltInFunctionDefinition::getName)
+ .collect(Collectors.toSet());
+ }
+ return functionNames;
}
@Override
public Optional<FunctionDefinition> getFunctionDefinition(String name) {
- return BuiltInFunctionDefinitions.getDefinitions().stream()
+ return functionDefinitions.stream()
.filter(f -> f.getName().equalsIgnoreCase(name))
.findFirst()
.map(Function.identity());