[FLINK-21822][table] Migrate built-in and test catalog factories to n…

…ew stack

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java

@@ -45,7 +45,7 @@
 import org.apache.flink.table.catalog.hive.client.HiveMetastoreClientWrapper;
 import org.apache.flink.table.catalog.hive.client.HiveShim;
 import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
-import org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator;
+import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
 import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
 import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
@@ -129,7 +129,7 @@ public HiveTableSink(
  this.catalogTable = table;
  hiveVersion =
  Preconditions.checkNotNull(
- jobConf.get(HiveCatalogValidator.CATALOG_HIVE_VERSION),
+ jobConf.get(HiveCatalogFactoryOptions.HIVE_VERSION.key()),
  "Hive version is not defined");
  hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
  tableSchema = TableSchemaUtils.getPhysicalSchema(table.getSchema());

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java

@@ -37,7 +37,7 @@
 import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.flink.table.catalog.hive.client.HiveShim;
 import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
-import org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator;
+import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
 import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.connector.source.DynamicTableSource;
@@ -110,7 +110,7 @@ public HiveTableSource(
  this.catalogTable = Preconditions.checkNotNull(catalogTable);
  this.hiveVersion =
  Preconditions.checkNotNull(
- jobConf.get(HiveCatalogValidator.CATALOG_HIVE_VERSION),
+ jobConf.get(HiveCatalogFactoryOptions.HIVE_VERSION.key()),
  "Hive version is not defined");
  this.hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
  }

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/HiveCatalog.java

@@ -60,7 +60,7 @@
 import org.apache.flink.table.catalog.hive.client.HiveMetastoreClientWrapper;
 import org.apache.flink.table.catalog.hive.client.HiveShim;
 import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
-import org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator;
+import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
 import org.apache.flink.table.catalog.hive.factories.HiveFunctionDefinitionFactory;
 import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
 import org.apache.flink.table.catalog.hive.util.HiveStatsUtil;
@@ -203,7 +203,7 @@ protected HiveCatalog(
  hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);
  // add this to hiveConf to make sure table factory and source/sink see the same Hive version
  // as HiveCatalog
- this.hiveConf.set(HiveCatalogValidator.CATALOG_HIVE_VERSION, hiveVersion);
+ this.hiveConf.set(HiveCatalogFactoryOptions.HIVE_VERSION.key(), hiveVersion);
 
  LOG.info("Created HiveCatalog '{}'", catalogName);
  }

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactory.java

@@ -18,92 +18,88 @@
 
 package org.apache.flink.table.catalog.hive.factories;
 
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.hive.HiveCatalog;
-import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
-import org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator;
-import org.apache.flink.table.descriptors.DescriptorProperties;
 import org.apache.flink.table.factories.CatalogFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
 
-import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HADOOP_CONF_DIR;
-import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HIVE_CONF_DIR;
-import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_HIVE_VERSION;
-import static org.apache.flink.table.catalog.hive.descriptors.HiveCatalogValidator.CATALOG_TYPE_VALUE_HIVE;
-import static org.apache.flink.table.descriptors.CatalogDescriptorValidator.CATALOG_DEFAULT_DATABASE;
-import static org.apache.flink.table.descriptors.CatalogDescriptorValidator.CATALOG_PROPERTY_VERSION;
-import static org.apache.flink.table.descriptors.CatalogDescriptorValidator.CATALOG_TYPE;
+import static org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions.DEFAULT_DATABASE;
+import static org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions.HADOOP_CONF_DIR;
+import static org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions.HIVE_CONF_DIR;
+import static org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions.HIVE_VERSION;
+import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
 
 /** Catalog factory for {@link HiveCatalog}. */
 public class HiveCatalogFactory implements CatalogFactory {
  private static final Logger LOG = LoggerFactory.getLogger(HiveCatalogFactory.class);
 
  @Override
- public Map<String, String> requiredContext() {
- Map<String, String> context = new HashMap<>();
- context.put(CATALOG_TYPE, CATALOG_TYPE_VALUE_HIVE); // hive
- context.put(CATALOG_PROPERTY_VERSION, "1"); // backwards compatibility
- return context;
+ public String factoryIdentifier() {
+ return HiveCatalogFactoryOptions.IDENTIFIER;
  }
 
  @Override
- public List<String> supportedProperties() {
- List<String> properties = new ArrayList<>();
-
- // default database
- properties.add(CATALOG_DEFAULT_DATABASE);
-
- properties.add(CATALOG_HIVE_CONF_DIR);
-
- properties.add(CATALOG_HIVE_VERSION);
-
- properties.add(CATALOG_HADOOP_CONF_DIR);
-
- return properties;
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.emptySet();
  }
 
  @Override
- public Catalog createCatalog(String name, Map<String, String> properties) {
- final DescriptorProperties descriptorProperties = getValidatedProperties(properties);
-
- final String defaultDatabase =
- descriptorProperties
- .getOptionalString(CATALOG_DEFAULT_DATABASE)
- .orElse(HiveCatalog.DEFAULT_DB);
-
- final Optional<String> hiveConfDir =
- descriptorProperties.getOptionalString(CATALOG_HIVE_CONF_DIR);
-
- final Optional<String> hadoopConfDir =
- descriptorProperties.getOptionalString(CATALOG_HADOOP_CONF_DIR);
+ public Set<ConfigOption<?>> optionalOptions() {
+ final Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(DEFAULT_DATABASE);
+ options.add(PROPERTY_VERSION);
+ options.add(HIVE_CONF_DIR);
+ options.add(HIVE_VERSION);
+ options.add(HADOOP_CONF_DIR);
+ return options;
+ }
 
-  final String version =
-  descriptorProperties
-  .getOptionalString(CATALOG_HIVE_VERSION)
-  .orElse(HiveShimLoader.getHiveVersion());
+ @Override
+ public Catalog createCatalog(Context context) {
+ final Configuration configuration = Configuration.fromMap(context.getOptions());
+ validateConfiguration(configuration);
 
  return new HiveCatalog(
- name,
- defaultDatabase,
- hiveConfDir.orElse(null),
- hadoopConfDir.orElse(null),
- version);
+ context.getName(),
+ configuration.getString(DEFAULT_DATABASE),
+ configuration.getString(HIVE_CONF_DIR),
+ configuration.getString(HADOOP_CONF_DIR),
+ configuration.getString(HIVE_VERSION));
  }
 
- private static DescriptorProperties getValidatedProperties(Map<String, String> properties) {
- final DescriptorProperties descriptorProperties = new DescriptorProperties(true);
- descriptorProperties.putProperties(properties);
-
- new HiveCatalogValidator().validate(descriptorProperties);
-
- return descriptorProperties;
+ private void validateConfiguration(Configuration configuration) {
+ final String defaultDatabase = configuration.getString(DEFAULT_DATABASE);
+ if (defaultDatabase != null && defaultDatabase.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "Option '%s' was provided, but is empty", DEFAULT_DATABASE.key()));
+ }
+
+ final String hiveConfDir = configuration.getString(HIVE_CONF_DIR);
+ if (hiveConfDir != null && hiveConfDir.isEmpty()) {
+ throw new ValidationException(
+ String.format("Option '%s' was provided, but is empty", HIVE_CONF_DIR.key()));
+ }
+
+ final String hadoopConfDir = configuration.getString(HADOOP_CONF_DIR);
+ if (hadoopConfDir != null && hadoopConfDir.isEmpty()) {
+ throw new ValidationException(
+ String.format("Option '%s' was provided, but is empty", HADOOP_CONF_DIR.key()));
+ }
+
+ final String hiveVersion = configuration.getString(HIVE_VERSION);
+ if (hiveVersion != null && hiveVersion.isEmpty()) {
+ throw new ValidationException(
+ String.format("Option '%s' was provided, but is empty", HIVE_VERSION.key()));
+ }
  }
 }

flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/factories/HiveCatalogFactoryOptions.java

@@ -0,0 +1,51 @@
+/*
+ * 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:https://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.catalog.hive.factories;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.flink.table.catalog.hive.HiveCatalog;
+import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
+
+/** {@link ConfigOption}s for {@link HiveCatalog}. */
+@Internal
+public final class HiveCatalogFactoryOptions {
+
+ public static final String IDENTIFIER = "hive";
+
+ public static final ConfigOption<String> DEFAULT_DATABASE =
+ ConfigOptions.key(CommonCatalogOptions.DEFAULT_DATABASE_KEY)
+ .stringType()
+ .defaultValue(HiveCatalog.DEFAULT_DB);
+
+ public static final ConfigOption<String> HIVE_CONF_DIR =
+ ConfigOptions.key("hive-conf-dir").stringType().noDefaultValue();
+
+ public static final ConfigOption<String> HIVE_VERSION =
+ ConfigOptions.key("hive-version")
+ .stringType()
+ .defaultValue(HiveShimLoader.getHiveVersion());
+
+ public static final ConfigOption<String> HADOOP_CONF_DIR =
+ ConfigOptions.key("hadoop-conf-dir").stringType().noDefaultValue();
+
+ private HiveCatalogFactoryOptions() {}
+}

flink-connectors/flink-connector-hive/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory

@@ -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:https://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.table.catalog.hive.factories.HiveCatalogFactory

flink-connectors/flink-connector-hive/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory

@@ -13,6 +13,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-org.apache.flink.table.catalog.hive.factories.HiveCatalogFactory
 org.apache.flink.table.module.hive.HiveModuleFactory
 org.apache.flink.table.planner.delegation.hive.HiveParserFactory