[FLINK-13296][table] FunctionCatalog.lookupFunction() should check in…

… memory functions if the target function doesn't exist in catalog Currently the logic to lookup a function is check either the catalog or the in memory function. But the correct logic is to 1st check the catalog, and if the function doesn't exist there, check in memory functions. There should be a resolution order. This closes apache#9135.
obermeier · Jul 17, 2019 · f4e71f9 · f4e71f9
1 parent 200a5bf
commit f4e71f9
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 33 deletions.
diff --git a/...client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java b/...client/src/test/java/org/apache/flink/table/client/gateway/local/LocalExecutorITCase.java
@@ -49,7 +49,6 @@
 
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
@@ -430,7 +429,6 @@ public void testStreamQueryExecutionSink() throws Exception {
 		}
 	}
 
-	@Ignore
 	@Test
 	public void testUseCatalogAndUseDatabase() throws Exception {
 		final String csvOutputPath = new File(tempFolder.newFolder().getAbsolutePath(), "test-out.csv").toURI().toString();

diff --git a/...le/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java b/...le/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/FunctionCatalog.java
@@ -159,56 +159,59 @@ public String[] getUserDefinedFunctions() {
 	public Optional<FunctionLookup.Result> lookupFunction(String name) {
 		String functionName = normalizeName(name);
 
-		FunctionDefinition userCandidate = null;
+		FunctionDefinition userCandidate;
 
 		Catalog catalog = catalogManager.getCatalog(catalogManager.getCurrentCatalog()).get();
 
-		if (catalog.getTableFactory().isPresent() &&
+		try {
+			CatalogFunction catalogFunction = catalog.getFunction(
+				new ObjectPath(catalogManager.getCurrentDatabase(), functionName));
+
+			if (catalog.getTableFactory().isPresent() &&
 				catalog.getTableFactory().get() instanceof FunctionDefinitionFactory) {
-			try {
-				CatalogFunction catalogFunction = catalog.getFunction(
-					new ObjectPath(catalogManager.getCurrentDatabase(), functionName));
 
 				FunctionDefinitionFactory factory = (FunctionDefinitionFactory) catalog.getTableFactory().get();
 
 				userCandidate = factory.createFunctionDefinition(functionName, catalogFunction);
-			} catch (FunctionNotExistException e) {
-				// Ignore
-			}
 
-			return Optional.of(
+				return Optional.of(
 					new FunctionLookup.Result(
 						ObjectIdentifier.of(catalogManager.getCurrentCatalog(), catalogManager.getCurrentDatabase(), name),
 						userCandidate)
 				);
-		} else {
-			// Else, check in-memory functions
-			userCandidate = userFunctions.get(functionName);
-
-			final Optional<FunctionDefinition> foundDefinition;
-			if (userCandidate != null) {
-				foundDefinition = Optional.of(userCandidate);
 			} else {
+				// TODO: should go thru function definition discover service
+			}
+		} catch (FunctionNotExistException e) {
+			// Ignore
+		}
 
-				// TODO once we connect this class with the Catalog APIs we need to make sure that
-				//  built-in functions are present in "root" built-in catalog. This allows to
-				//  overwrite built-in functions but also fallback to the "root" catalog. It should be
-				//  possible to disable the "root" catalog if that is desired.
+		// If no corresponding function is found in catalog, check in-memory functions
+		userCandidate = userFunctions.get(functionName);
 
-				foundDefinition = BuiltInFunctionDefinitions.getDefinitions()
-					.stream()
-					.filter(f -> functionName.equals(normalizeName(f.getName())))
-					.findFirst()
-					.map(Function.identity());
-			}
+		final Optional<FunctionDefinition> foundDefinition;
+		if (userCandidate != null) {
+			foundDefinition = Optional.of(userCandidate);
+		} else {
 
-			String defaultCatalogName = catalogManager.getDefaultCatalogName();
+			// TODO once we connect this class with the Catalog APIs we need to make sure that
+			//  built-in functions are present in "root" built-in catalog. This allows to
+			//  overwrite built-in functions but also fallback to the "root" catalog. It should be
+			//  possible to disable the "root" catalog if that is desired.
 
-			return foundDefinition.map(definition -> new FunctionLookup.Result(
-				ObjectIdentifier.of(defaultCatalogName, catalogManager.getCatalog(defaultCatalogName).get().getDefaultDatabase(), name),
-				definition)
-			);
+			foundDefinition = BuiltInFunctionDefinitions.getDefinitions()
+				.stream()
+				.filter(f -> functionName.equals(normalizeName(f.getName())))
+				.findFirst()
+				.map(Function.identity());
 		}
+
+		String defaultCatalogName = catalogManager.getDefaultCatalogName();
+
+		return foundDefinition.map(definition -> new FunctionLookup.Result(
+			ObjectIdentifier.of(defaultCatalogName, catalogManager.getCatalog(defaultCatalogName).get().getDefaultDatabase(), name),
+			definition)
+		);
 	}
 
 	@Override