[FLINK-18419] Create catalog in TableEnvironment using user ClassLoader

deepfield · Jul 10, 2020 · 69ef4b7 · 69ef4b7
1 parent abf3091
commit 69ef4b7
Show file tree

Hide file tree

Showing 5 changed files with 90 additions and 22 deletions.
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
@@ -71,7 +71,9 @@
 import org.apache.flink.table.descriptors.StreamTableDescriptor;
 import org.apache.flink.table.expressions.ApiExpressionUtils;
 import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.CatalogFactory;
 import org.apache.flink.table.factories.ComponentFactoryService;
+import org.apache.flink.table.factories.TableFactoryService;
 import org.apache.flink.table.functions.ScalarFunction;
 import org.apache.flink.table.functions.UserDefinedFunction;
 import org.apache.flink.table.functions.UserDefinedFunctionHelper;
@@ -983,15 +985,7 @@ private TableResult executeOperation(Operation operation) {
  } else if (operation instanceof AlterCatalogFunctionOperation) {
  return alterCatalogFunction((AlterCatalogFunctionOperation) operation);
  } else if (operation instanceof CreateCatalogOperation) {
- CreateCatalogOperation createCatalogOperation = (CreateCatalogOperation) operation;
- String exMsg = getDDLOpExecuteErrorMsg(createCatalogOperation.asSummaryString());
- try {
- catalogManager.registerCatalog(
- createCatalogOperation.getCatalogName(), createCatalogOperation.getCatalog());
- return TableResultImpl.TABLE_RESULT_OK;
- } catch (CatalogException e) {
- throw new ValidationException(exMsg, e);
- }
+ return createCatalog((CreateCatalogOperation) operation);
  } else if (operation instanceof DropCatalogOperation) {
  DropCatalogOperation dropCatalogOperation = (DropCatalogOperation) operation;
  String exMsg = getDDLOpExecuteErrorMsg(dropCatalogOperation.asSummaryString());
@@ -1047,6 +1041,24 @@ private TableResult executeOperation(Operation operation) {
  }
  }
 
+ private TableResult createCatalog(CreateCatalogOperation operation) {
+ String exMsg = getDDLOpExecuteErrorMsg(operation.asSummaryString());
+ try {
+ String catalogName = operation.getCatalogName();
+ Map<String, String> properties = operation.getProperties();
+ final CatalogFactory factory = TableFactoryService.find(
+ CatalogFactory.class,
+ properties,
+ userClassLoader);
+
+ Catalog catalog = factory.createCatalog(catalogName, properties);
+ catalogManager.registerCatalog(catalogName, catalog);
+ return TableResultImpl.TABLE_RESULT_OK;
+ } catch (CatalogException e) {
+ throw new ValidationException(exMsg, e);
+ }
+ }
+
  private TableResult buildShowResult(String columnName, String[] objects) {
  return buildResult(
  new String[]{columnName},

diff --git a/...-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogOperation.java b/...-api-java/src/main/java/org/apache/flink/table/operations/ddl/CreateCatalogOperation.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.table.operations.ddl;
 
-import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.operations.Operation;
 import org.apache.flink.table.operations.OperationUtils;
 
@@ -33,25 +32,26 @@
  */
 public class CreateCatalogOperation implements CreateOperation {
  private final String catalogName;
- private final Catalog catalog;
+ private final Map<String, String> properties;
 
- public CreateCatalogOperation(String catalogName, Catalog catalog) {
+ public CreateCatalogOperation(String catalogName, Map<String, String> properties) {
  this.catalogName = checkNotNull(catalogName);
- this.catalog = checkNotNull(catalog);
+ this.properties = checkNotNull(properties);
  }
 
  public String getCatalogName() {
  return catalogName;
  }
 
- public Catalog getCatalog() {
- return catalog;
+ public Map<String, String> getProperties() {
+ return Collections.unmodifiableMap(properties);
  }
 
  @Override
  public String asSummaryString() {
  Map<String, Object> params = new LinkedHashMap<>();
  params.put("catalogName", catalogName);
+ params.put("properties", properties);
 
  return OperationUtils.formatWithChildren(
  "CREATE CATALOG",

diff --git a/flink-table/flink-table-planner-blink/pom.xml b/flink-table/flink-table-planner-blink/pom.xml
@@ -223,6 +223,14 @@ under the License.
  <scope>test</scope>
  </dependency>
 
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
  <!-- SuccessException used in TestValuesTableFactory -->
  <dependency>
  <groupId>org.apache.flink</groupId>

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
@@ -74,8 +74,6 @@
 import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.catalog.UnresolvedIdentifier;
 import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
-import org.apache.flink.table.factories.CatalogFactory;
-import org.apache.flink.table.factories.TableFactoryService;
 import org.apache.flink.table.operations.CatalogSinkModifyOperation;
 import org.apache.flink.table.operations.DescribeTableOperation;
 import org.apache.flink.table.operations.ExplainOperation;
@@ -553,11 +551,7 @@ private Operation convertCreateCatalog(SqlCreateCatalog sqlCreateCatalog) {
  sqlCreateCatalog.getPropertyList().getList().forEach(p ->
  properties.put(((SqlTableOption) p).getKeyString(), ((SqlTableOption) p).getValueString()));
 
- final CatalogFactory factory =
- TableFactoryService.find(CatalogFactory.class, properties, this.getClass().getClassLoader());
-
- Catalog catalog = factory.createCatalog(catalogName, properties);
- return new CreateCatalogOperation(catalogName, catalog);
+ return new CreateCatalogOperation(catalogName, properties);
  }
 
  /** Convert DROP CATALOG statement. */

diff --git a/...ble-planner-blink/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java b/...ble-planner-blink/src/test/java/org/apache/flink/table/planner/catalog/CatalogITCase.java
@@ -23,8 +23,14 @@
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.table.catalog.GenericInMemoryCatalog;
+import org.apache.flink.testutils.ClassLoaderUtils;
+import org.apache.flink.util.TemporaryClassLoaderContext;
 
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.net.URLClassLoader;
 
 import static org.apache.flink.table.descriptors.GenericInMemoryCatalogValidator.CATALOG_TYPE_VALUE_GENERIC_IN_MEMORY;
 import static org.junit.Assert.assertFalse;
@@ -35,6 +41,9 @@
  */
 public class CatalogITCase {
 
+ @Rule
+ public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
  @Test
  public void testCreateCatalog() {
  String name = "c1";
@@ -61,9 +70,54 @@ public void testDropCatalog() {
  assertFalse(tableEnv.getCatalog(name).isPresent());
  }
 
+ @Test
+ public void testCreateCatalogFromUserClassLoader() throws Exception {
+ final String className = "UserCatalogFactory";
+ URLClassLoader classLoader = ClassLoaderUtils.withRoot(temporaryFolder.newFolder())
+ .addResource("META-INF/services/org.apache.flink.table.factories.TableFactory", "UserCatalogFactory")
+ .addClass(
+ className,
+ "import org.apache.flink.table.catalog.GenericInMemoryCatalog;\n" +
+ "import org.apache.flink.table.factories.CatalogFactory;\n" +
+ "import java.util.Collections;\n" +
+ "import org.apache.flink.table.catalog.Catalog;\n" +
+ "import java.util.HashMap;\n" +
+ "import java.util.List;\n" +
+ "import java.util.Map;\n" +
+ "\tpublic class UserCatalogFactory implements CatalogFactory {\n" +
+ "\t\t@Override\n" +
+ "\t\tpublic Catalog createCatalog(\n" +
+ "\t\t\t\tString name,\n" +
+ "\t\t\t\tMap<String, String> properties) {\n" +
+ "\t\t\treturn new GenericInMemoryCatalog(name);\n" +
+ "\t\t}\n" +
+ "\n" +
+ "\t\t@Override\n" +
+ "\t\tpublic Map<String, String> requiredContext() {\n" +
+ "\t\t\tHashMap<String, String> hashMap = new HashMap<>();\n" +
+ "\t\t\thashMap.put(\"type\", \"userCatalog\");\n" +
+ "\t\t\treturn hashMap;\n" +
+ "\t\t}\n" +
+ "\n" +
+ "\t\t@Override\n" +
+ "\t\tpublic List<String> supportedProperties() {\n" +
+ "\t\t\treturn Collections.emptyList();\n" +
+ "\t\t}\n" +
+ "\t}"
+ ).build();
+
+ try (TemporaryClassLoaderContext context = TemporaryClassLoaderContext.of(classLoader)) {
+ TableEnvironment tableEnvironment = getTableEnvironment();
+ tableEnvironment.executeSql("CREATE CATALOG cat WITH ('type'='userCatalog')");
+
+ assertTrue(tableEnvironment.getCatalog("cat").isPresent());
+ }
+ }
+
  private TableEnvironment getTableEnvironment() {
  EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  return StreamTableEnvironment.create(env, settings);
  }
+
 }