[FLINK-17345][python][table] Support register and get Python UDF in C…

…atalog (apache#11884)
silenceland · Apr 24, 2020 · 90fed72 · 90fed72
1 parent efd22a7
commit 90fed72
Show file tree

Hide file tree

Showing 11 changed files with 82 additions and 28 deletions.
diff --git a/...s/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java b/...s/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java
@@ -133,6 +133,7 @@ public class HiveCatalog extends AbstractCatalog {
 	// It's appended to Flink function's class name
 	// because Hive's Function object doesn't have properties or other place to store the flag for Flink functions.
 	private static final String FLINK_FUNCTION_PREFIX = "flink:";
+	private static final String FLINK_PYTHON_FUNCTION_PREFIX = FLINK_FUNCTION_PREFIX + "python:";
 
 	private final HiveConf hiveConf;
 	private final String hiveVersion;
@@ -1155,7 +1156,13 @@ public CatalogFunction getFunction(ObjectPath functionPath) throws FunctionNotEx
 			Function function = client.getFunction(functionPath.getDatabaseName(), functionPath.getObjectName());
 
 			if (function.getClassName().startsWith(FLINK_FUNCTION_PREFIX)) {
-				return new CatalogFunctionImpl(function.getClassName().substring(FLINK_FUNCTION_PREFIX.length()));
+				if (function.getClassName().startsWith(FLINK_PYTHON_FUNCTION_PREFIX)) {
+					return new CatalogFunctionImpl(
+						function.getClassName().substring(FLINK_PYTHON_FUNCTION_PREFIX.length()),
+						FunctionLanguage.PYTHON);
+				} else {
+					return new CatalogFunctionImpl(function.getClassName().substring(FLINK_FUNCTION_PREFIX.length()));
+				}
 			} else {
 				return new CatalogFunctionImpl(function.getClassName());
 			}
@@ -1187,13 +1194,16 @@ private static Function instantiateHiveFunction(ObjectPath functionPath, Catalog
 
 		// Hive Function does not have properties map
 		// thus, use a prefix in class name to distinguish Flink and Hive functions
-		String functionClassName = isGeneric ?
-			FLINK_FUNCTION_PREFIX + function.getClassName() :
-			function.getClassName();
-
-		if (!function.getFunctionLanguage().equals(FunctionLanguage.JAVA)) {
+		String functionClassName;
+		if (function.getFunctionLanguage().equals(FunctionLanguage.JAVA)) {
+			functionClassName = isGeneric ?
+				FLINK_FUNCTION_PREFIX + function.getClassName() :
+				function.getClassName();
+		} else if (function.getFunctionLanguage().equals(FunctionLanguage.PYTHON)) {
+			functionClassName = FLINK_PYTHON_FUNCTION_PREFIX + function.getClassName();
+		} else {
 			throw new UnsupportedOperationException("HiveCatalog supports only creating" +
-				" JAVA based function for now");
+				" JAVA or PYTHON based function for now");
 		}
 
 		return new Function(

diff --git a/...r-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogMetadataTestBase.java b/...r-hive/src/test/java/org/apache/flink/table/catalog/hive/HiveCatalogMetadataTestBase.java
@@ -21,6 +21,7 @@
 import org.apache.flink.table.catalog.CatalogFunction;
 import org.apache.flink.table.catalog.CatalogFunctionImpl;
 import org.apache.flink.table.catalog.CatalogTestBase;
+import org.apache.flink.table.catalog.FunctionLanguage;
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.functions.hive.HiveGenericUDF;
 import org.apache.flink.table.functions.hive.HiveSimpleUDF;
@@ -67,4 +68,9 @@ protected CatalogFunction createFunction() {
 	protected CatalogFunction createAnotherFunction() {
 		return new CatalogFunctionImpl(HiveGenericUDF.class.getCanonicalName());
 	}
+
+	@Override
+	protected CatalogFunction createPythonFunction() {
+		return new CatalogFunctionImpl("test.func1", FunctionLanguage.PYTHON);
+	}
 }
diff --git a/flink-python/src/test/java/org/apache/flink/client/python/PythonFunctionFactoryTest.java b/flink-python/src/test/java/org/apache/flink/client/python/PythonFunctionFactoryTest.java
@@ -83,6 +83,14 @@ public static void cleanEnvironment() throws Exception {
 	}
 
 	public static void testPythonFunctionFactory() {
+		// flink catalog
+		flinkTableEnv.sqlUpdate("create function func1 as 'test1.func1' language python");
+		verifyPlan(flinkSourceTable.select("func1(str)"), flinkTableEnv);
+
+		// flink catalog
+		flinkTableEnv.sqlUpdate("alter function func1 as 'test1.func1' language python");
+		verifyPlan(flinkSourceTable.select("func1(str)"), flinkTableEnv);
+
 		// flink temporary catalog
 		flinkTableEnv.sqlUpdate("create temporary function func1 as 'test1.func1' language python");
 		verifyPlan(flinkSourceTable.select("func1(str)"), flinkTableEnv);
@@ -91,6 +99,14 @@ public static void testPythonFunctionFactory() {
 		flinkTableEnv.sqlUpdate("create temporary system function func1 as 'test1.func1' language python");
 		verifyPlan(flinkSourceTable.select("func1(str)"), flinkTableEnv);
 
+		// blink catalog
+		blinkTableEnv.sqlUpdate("create function func1 as 'test1.func1' language python");
+		verifyPlan(blinkSourceTable.select("func1(str)"), blinkTableEnv);
+
+		// blink catalog
+		blinkTableEnv.sqlUpdate("alter function func1 as 'test1.func1' language python");
+		verifyPlan(blinkSourceTable.select("func1(str)"), blinkTableEnv);
+
 		// blink temporary catalog
 		blinkTableEnv.sqlUpdate("create temporary function func1 as 'test1.func1' language python");
 		verifyPlan(blinkSourceTable.select("func1(str)"), blinkTableEnv);

diff --git a/...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/...able-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -41,7 +41,6 @@
 import org.apache.flink.table.catalog.CatalogManager;
 import org.apache.flink.table.catalog.ConnectorCatalogTable;
 import org.apache.flink.table.catalog.FunctionCatalog;
-import org.apache.flink.table.catalog.FunctionLanguage;
 import org.apache.flink.table.catalog.GenericInMemoryCatalog;
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.QueryOperationCatalogView;
@@ -999,9 +998,6 @@ private TableResult alterCatalogFunction(AlterCatalogFunctionOperation alterCata
 			if (alterCatalogFunctionOperation.isTemporary()) {
 				throw new ValidationException(
 					"Alter temporary catalog function is not supported");
-			} else if (function.getFunctionLanguage() == FunctionLanguage.PYTHON) {
-				throw new ValidationException(
-					"Alter Python catalog function is not supported");
 			} else {
 				Catalog catalog = getCatalogOrThrowException(
 					alterCatalogFunctionOperation.getFunctionIdentifier().getCatalogName());

diff --git a/...link-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogFunctionImpl.java b/...link-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogFunctionImpl.java
@@ -67,6 +67,9 @@ public Optional<String> getDetailedDescription() {
 
 	@Override
 	public boolean isGeneric() {
+		if (functionLanguage == FunctionLanguage.PYTHON) {
+			return true;
+		}
 		try {
 			Class c = Class.forName(className);
 			if (UserDefinedFunction.class.isAssignableFrom(c)) {

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
@@ -567,13 +567,13 @@ private Optional<FunctionLookup.Result> resolvePreciseFunctionReference(ObjectId
 					new ObjectPath(oi.getDatabaseName(), oi.getObjectName()));
 
 				FunctionDefinition fd;
-				if (catalog.getFunctionDefinitionFactory().isPresent()) {
+				if (catalog.getFunctionDefinitionFactory().isPresent() &&
+					catalogFunction.getFunctionLanguage() != FunctionLanguage.PYTHON) {
 					fd = catalog.getFunctionDefinitionFactory().get()
 						.createFunctionDefinition(oi.getObjectName(), catalogFunction);
 				} else {
 					// TODO update the FunctionDefinitionUtil once we drop the old function stack in DDL
-					fd = FunctionDefinitionUtil.createFunctionDefinition(
-						oi.getObjectName(), catalogFunction.getClassName());
+					fd = getFunctionDefinition(oi.getObjectName(), catalogFunction);
 				}
 
 				return Optional.of(

diff --git a/...ble-api-java/src/test/java/org/apache/flink/table/catalog/GenericInMemoryCatalogTest.java b/...ble-api-java/src/test/java/org/apache/flink/table/catalog/GenericInMemoryCatalogTest.java
@@ -164,4 +164,9 @@ protected CatalogFunction createFunction() {
 	protected CatalogFunction createAnotherFunction() {
 		return new CatalogFunctionImpl(TestSimpleUDF.class.getCanonicalName(), FunctionLanguage.SCALA);
 	}
+
+	@Override
+	protected CatalogFunction createPythonFunction() {
+		return new CatalogFunctionImpl("test.func1", FunctionLanguage.PYTHON);
+	}
 }
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/catalog/CatalogTest.java
@@ -627,6 +627,16 @@ public void testCreateFunction() throws Exception {
 		assertTrue(catalog.functionExists(path1));
 	}
 
+	@Test
+	public void testCreatePythonFunction() throws Exception {
+		catalog.createDatabase(db1, createDb(), false);
+		CatalogFunction pythonFunction = createPythonFunction();
+		catalog.createFunction(path1, createPythonFunction(), false);
+
+		CatalogFunction actual = catalog.getFunction(path1);
+		checkEquals(pythonFunction, actual);
+	}
+
 	@Test
 	public void testCreateFunction_DatabaseNotExistException() throws Exception {
 		assertFalse(catalog.databaseExists(db1));
@@ -668,6 +678,23 @@ public void testAlterFunction() throws Exception {
 		checkEquals(newFunc, actual);
 	}
 
+	@Test
+	public void testAlterPythonFunction() throws Exception {
+		catalog.createDatabase(db1, createDb(), false);
+
+		CatalogFunction func = createFunction();
+		catalog.createFunction(path1, func, false);
+
+		checkEquals(func, catalog.getFunction(path1));
+
+		CatalogFunction newFunc = createPythonFunction();
+		catalog.alterFunction(path1, newFunc, false);
+		CatalogFunction actual = catalog.getFunction(path1);
+
+		assertFalse(func.getClassName().equals(actual.getClassName()));
+		checkEquals(newFunc, actual);
+	}
+
 	@Test
 	public void testAlterFunction_FunctionNotExistException() throws Exception {
 		exception.expect(FunctionNotExistException.class);
@@ -1219,6 +1246,11 @@ public void testAlterTableStats_TableNotExistException_ignore() throws Exception
 	 */
 	protected abstract CatalogFunction createFunction();
 
+	/**
+	 * Create a Python CatalogFunction instance by specific catalog implementation.
+	 */
+	protected abstract CatalogFunction createPythonFunction();
+
 	/**
 	 * Create another CatalogFunction instance by specific catalog implementation.
 	 *

diff --git a/...link/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java b/...link/src/main/java/org/apache/flink/table/planner/operations/SqlToOperationConverter.java
@@ -292,12 +292,6 @@ private Operation convertCreateFunction(SqlCreateFunction sqlCreateFunction) {
 			);
 		} else {
 			FunctionLanguage language = parseLanguage(sqlCreateFunction.getFunctionLanguage());
-			if (language == FunctionLanguage.PYTHON && !sqlCreateFunction.isTemporary()) {
-				throw new ValidationException(String.format(
-					"Unsupported function type for Python function %s, " +
-					"only temporary function is supported.",
-					sqlCreateFunction.getFunctionClassName().getValueAs(String.class)));
-			}
 			CatalogFunction catalogFunction = new CatalogFunctionImpl(
 				sqlCreateFunction.getFunctionClassName().getValueAs(String.class),
 				language);

diff --git a/...k-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java b/...k-table-planner/src/main/java/org/apache/flink/table/sqlexec/SqlToOperationConverter.java
@@ -288,12 +288,6 @@ private Operation convertCreateFunction(SqlCreateFunction sqlCreateFunction) {
 			);
 		} else {
 			FunctionLanguage language = parseLanguage(sqlCreateFunction.getFunctionLanguage());
-			if (language == FunctionLanguage.PYTHON && !sqlCreateFunction.isTemporary()) {
-				throw new ValidationException(String.format(
-					"Unsupported function type for Python function %s, " +
-						"only temporary function is supported.",
-					sqlCreateFunction.getFunctionClassName().getValueAs(String.class)));
-			}
 			CatalogFunction catalogFunction = new CatalogFunctionImpl(
 				sqlCreateFunction.getFunctionClassName().getValueAs(String.class),
 				language);

diff --git a/...flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala b/...flink-table-planner/src/main/scala/org/apache/flink/table/api/internal/TableEnvImpl.scala
@@ -907,8 +907,6 @@ abstract class TableEnvImpl(
       val function = alterFunctionOperation.getCatalogFunction
       if (alterFunctionOperation.isTemporary) {
         throw new ValidationException("Alter temporary catalog function is not supported")
-      } else if (function.getFunctionLanguage eq FunctionLanguage.PYTHON) {
-        throw new ValidationException("Alter Python catalog function is not supported");
       } else {
         val catalog = getCatalogOrThrowException(
           alterFunctionOperation.getFunctionIdentifier.getCatalogName)