This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new a8cc843f9 [AMORO-3210] Support loading flink mixed catalog without ams
url (#3211)
a8cc843f9 is described below
commit a8cc843f98af7daadd47f80287a9c83586ab2807
Author: ZhouJinsong <[email protected]>
AuthorDate: Fri Sep 20 15:46:01 2024 +0800
[AMORO-3210] Support loading flink mixed catalog without ams url (#3211)
* Support loading flink mixed catalog without ams url
* Change some flink doc description
* Fix a spelling issue in method name
* Fix a style error
---
.../java/org/apache/amoro/utils/CatalogUtil.java | 2 +-
.../java/org/apache/amoro/mixed/CatalogLoader.java | 6 +-
.../apache/amoro/utils/MixedFormatCatalogUtil.java | 173 -------
.../amoro/catalog/BasicCatalogTestHelper.java | 4 +-
.../apache/amoro/catalog/BasicMixedCatalog.java | 3 +-
.../org/apache/amoro/catalog/TableTestBase.java | 4 +-
.../formats/IcebergHadoopCatalogTestHelper.java | 5 +-
.../MixedIcebergHadoopCatalogTestHelper.java | 6 +-
.../apache/amoro/flink/InternalCatalogBuilder.java | 87 +++-
.../amoro/flink/catalog/FlinkUnifiedCatalog.java | 7 +-
.../apache/amoro/flink/catalog/MixedCatalog.java | 6 +-
.../catalog/factories/CatalogFactoryOptions.java | 47 --
.../factories/FlinkUnifiedCatalogFactory.java | 64 ++-
.../factories/mixed/MixedCatalogFactory.java | 55 +-
.../flink/table/KafkaConnectorOptionsUtil.java | 567 ---------------------
...eFactory.java => MixedDynamicTableFactory.java} | 26 +-
.../org/apache/amoro/flink/table/OptionsUtil.java | 64 +++
...actory.java => UnifiedDynamicTableFactory.java} | 8 +-
.../org.apache.flink.table.factories.Factory | 2 +-
.../amoro/flink/catalog/TestMixedCatalog.java | 10 +-
.../org.apache.flink.table.factories.Factory | 2 +-
docs/engines/flink/flink-ddl.md | 13 +-
22 files changed, 250 insertions(+), 911 deletions(-)
diff --git a/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java
b/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java
index 89d3e7fdf..bd49ea931 100644
--- a/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java
+++ b/amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java
@@ -197,7 +197,7 @@ public class CatalogUtil {
tableMeta.getTableIdentifier().getTableName());
}
- public static org.apache.amoro.api.TableIdentifier amsTaleId(TableIdentifier
tableIdentifier) {
+ public static org.apache.amoro.api.TableIdentifier
amsTableId(TableIdentifier tableIdentifier) {
return new org.apache.amoro.api.TableIdentifier(
tableIdentifier.getCatalog(),
tableIdentifier.getDatabase(),
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/CatalogLoader.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/CatalogLoader.java
index 3842e8e6a..b285fec01 100644
---
a/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/CatalogLoader.java
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/mixed/CatalogLoader.java
@@ -31,6 +31,7 @@ import org.apache.amoro.properties.CatalogMetaProperties;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.apache.amoro.shade.thrift.org.apache.thrift.TException;
import org.apache.amoro.table.TableMetaStore;
+import org.apache.amoro.utils.CatalogUtil;
import org.apache.amoro.utils.MixedFormatCatalogUtil;
import org.apache.iceberg.common.DynConstructors;
@@ -86,8 +87,7 @@ public class CatalogLoader {
catalogImpl = MIXED_ICEBERG_CATALOG_IMP;
break;
case CatalogMetaProperties.CATALOG_TYPE_HIVE:
- Set<TableFormat> tableFormats =
- MixedFormatCatalogUtil.tableFormats(metastoreType,
catalogProperties);
+ Set<TableFormat> tableFormats =
CatalogUtil.tableFormats(metastoreType, catalogProperties);
if (tableFormats.contains(TableFormat.MIXED_HIVE)) {
catalogImpl = HIVE_CATALOG_IMPL;
} else {
@@ -142,7 +142,7 @@ public class CatalogLoader {
catalogName,
type,
catalogMeta.getCatalogProperties(),
- MixedFormatCatalogUtil.buildMetaStore(catalogMeta));
+ CatalogUtil.buildMetaStore(catalogMeta));
} catch (NoSuchObjectException e1) {
throw new IllegalArgumentException("catalog not found, please check
catalog name", e1);
} catch (Exception e) {
diff --git
a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/MixedFormatCatalogUtil.java
b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/MixedFormatCatalogUtil.java
index ba3a5e2e1..3508bfefc 100644
---
a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/MixedFormatCatalogUtil.java
+++
b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/MixedFormatCatalogUtil.java
@@ -18,20 +18,14 @@
package org.apache.amoro.utils;
-import org.apache.amoro.TableFormat;
import org.apache.amoro.api.CatalogMeta;
-import org.apache.amoro.api.TableMeta;
import org.apache.amoro.io.AuthenticatedFileIO;
import org.apache.amoro.op.MixedHadoopTableOperations;
import org.apache.amoro.op.MixedTableOperations;
import org.apache.amoro.properties.CatalogMetaProperties;
import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
-import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
-import org.apache.amoro.table.TableIdentifier;
-import org.apache.amoro.table.TableMetaStore;
import org.apache.amoro.table.TableProperties;
-import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.BaseTable;
@@ -46,49 +40,13 @@ import org.apache.iceberg.util.PropertyUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Arrays;
-import java.util.Locale;
import java.util.Map;
-import java.util.Set;
import java.util.stream.Collectors;
public class MixedFormatCatalogUtil {
private static final Logger LOG =
LoggerFactory.getLogger(MixedFormatCatalogUtil.class);
- /** Return table format set catalog supported. */
- public static Set<TableFormat> tableFormats(CatalogMeta meta) {
- return tableFormats(meta.getCatalogType(), meta.getCatalogProperties());
- }
-
- /** Return table format set catalog supported. */
- public static Set<TableFormat> tableFormats(
- String metastoreType, Map<String, String> catalogProperties) {
- if (catalogProperties != null
- && catalogProperties.containsKey(CatalogMetaProperties.TABLE_FORMATS))
{
- String tableFormatsProperty =
catalogProperties.get(CatalogMetaProperties.TABLE_FORMATS);
- return Arrays.stream(tableFormatsProperty.split(","))
- .map(
- tableFormatString ->
-
TableFormat.valueOf(tableFormatString.trim().toUpperCase(Locale.ROOT)))
- .collect(Collectors.toSet());
- } else {
- // Generate table format from catalog type for compatibility with older
versions
- switch (metastoreType) {
- case CatalogMetaProperties.CATALOG_TYPE_AMS:
- return Sets.newHashSet(TableFormat.MIXED_ICEBERG);
- case CatalogMetaProperties.CATALOG_TYPE_CUSTOM:
- case CatalogMetaProperties.CATALOG_TYPE_HADOOP:
- case CatalogMetaProperties.CATALOG_TYPE_GLUE:
- return Sets.newHashSet(TableFormat.ICEBERG);
- case CatalogMetaProperties.CATALOG_TYPE_HIVE:
- return Sets.newHashSet(TableFormat.MIXED_HIVE);
- default:
- throw new IllegalArgumentException("Unsupported catalog type:" +
metastoreType);
- }
- }
- }
-
/** Merge catalog properties in client side into catalog meta. */
public static void mergeCatalogProperties(CatalogMeta meta, Map<String,
String> properties) {
if (meta.getCatalogProperties() == null) {
@@ -135,90 +93,6 @@ public class MixedFormatCatalogUtil {
return icebergCatalogProperties;
}
- /** Build {@link TableMetaStore} from catalog meta. */
- public static TableMetaStore buildMetaStore(CatalogMeta catalogMeta) {
- // load storage configs
- TableMetaStore.Builder builder = TableMetaStore.builder();
- if (catalogMeta.getStorageConfigs() != null) {
- Map<String, String> storageConfigs = catalogMeta.getStorageConfigs();
- if
(CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HADOOP.equalsIgnoreCase(
- MixedFormatCatalogUtil.getCompatibleStorageType(storageConfigs))) {
- String coreSite =
storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE);
- String hdfsSite =
storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE);
- String hiveSite =
storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HIVE_SITE);
- builder
- .withBase64CoreSite(coreSite)
- .withBase64MetaStoreSite(hiveSite)
- .withBase64HdfsSite(hdfsSite);
- }
- }
-
- boolean loadAuthFromAMS =
- PropertyUtil.propertyAsBoolean(
- catalogMeta.getCatalogProperties(),
- CatalogMetaProperties.LOAD_AUTH_FROM_AMS,
- CatalogMetaProperties.LOAD_AUTH_FROM_AMS_DEFAULT);
- // load auth configs from ams
- if (loadAuthFromAMS) {
- if (catalogMeta.getAuthConfigs() != null) {
- Map<String, String> authConfigs = catalogMeta.getAuthConfigs();
- String authType =
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE);
- LOG.info("TableMetaStore use auth config in catalog meta, authType is
{}", authType);
- if
(CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE.equalsIgnoreCase(authType))
{
- String hadoopUsername =
-
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME);
- builder.withSimpleAuth(hadoopUsername);
- } else if
(CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_KERBEROS.equalsIgnoreCase(
- authType)) {
- String krb5 =
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KRB5);
- String keytab =
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KEYTAB);
- String principal =
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_PRINCIPAL);
- builder.withBase64KrbAuth(keytab, krb5, principal);
- } else if
(CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_AK_SK.equalsIgnoreCase(authType))
{
- String accessKey =
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_ACCESS_KEY);
- String secretKey =
authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_SECRET_KEY);
- builder.withAkSkAuth(accessKey, secretKey);
- }
- }
- }
-
- // cover auth configs from ams with auth configs in properties
- String authType =
-
catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE);
- if (StringUtils.isNotEmpty(authType)) {
- LOG.info("TableMetaStore use auth config in properties, authType is {}",
authType);
- if
(CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE.equalsIgnoreCase(authType))
{
- String hadoopUsername =
- catalogMeta
- .getCatalogProperties()
- .get(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME);
- builder.withSimpleAuth(hadoopUsername);
- } else if
(CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_KERBEROS.equalsIgnoreCase(
- authType)) {
- String krb5 =
-
catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KRB5);
- String keytab =
-
catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KEYTAB);
- String principal =
- catalogMeta
- .getCatalogProperties()
- .get(CatalogMetaProperties.AUTH_CONFIGS_KEY_PRINCIPAL);
- builder.withBase64KrbAuth(keytab, krb5, principal);
- } else if
(CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_AK_SK.equalsIgnoreCase(authType))
{
- String accessKey =
- catalogMeta
- .getCatalogProperties()
- .get(CatalogMetaProperties.AUTH_CONFIGS_KEY_ACCESS_KEY);
- String secretKey =
- catalogMeta
- .getCatalogProperties()
- .get(CatalogMetaProperties.AUTH_CONFIGS_KEY_SECRET_KEY);
- builder.withAkSkAuth(accessKey, secretKey);
- }
- }
- return builder.build();
- }
-
/** Wrap table operation with authorization logic for {@link Table}. */
public static Table useMixedTableOperations(
Table table,
@@ -295,53 +169,6 @@ public class MixedFormatCatalogUtil {
return mergedProperties;
}
- public static TableIdentifier tableId(TableMeta tableMeta) {
- return TableIdentifier.of(
- tableMeta.getTableIdentifier().getCatalog(),
- tableMeta.getTableIdentifier().getDatabase(),
- tableMeta.getTableIdentifier().getTableName());
- }
-
- public static org.apache.amoro.api.TableIdentifier amsTaleId(TableIdentifier
tableIdentifier) {
- return new org.apache.amoro.api.TableIdentifier(
- tableIdentifier.getCatalog(),
- tableIdentifier.getDatabase(),
- tableIdentifier.getTableName());
- }
-
- /**
- * Get storage type compatible with history storage type `hdfs`, which is
`Hadoop` now.
- *
- * @param conf - configurations containing `storage.type`
- * @return storage type, return `Hadoop` if `storage.type` is `hdfs`, return
null if
- * `storage.type` not exist.
- */
- public static String getCompatibleStorageType(Map<String, String> conf) {
- if (CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HDFS_LEGACY.equals(
- conf.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE))) {
- return CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HADOOP;
- }
- return conf.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE);
- }
-
- /**
- * Copy property from source properties to target properties, support
changing the key name.
- *
- * @param fromProperties - from these properties
- * @param toProperties - to these properties
- * @param fromKey - from key
- * @param toKey - to key
- */
- public static <T> void copyProperty(
- Map<String, String> fromProperties,
- Map<String, T> toProperties,
- String fromKey,
- String toKey) {
- if (StringUtils.isNotEmpty(fromProperties.get(fromKey))) {
- toProperties.put(toKey, (T) fromProperties.get(fromKey));
- }
- }
-
/**
* Build cache catalog.
*
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicCatalogTestHelper.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicCatalogTestHelper.java
index 599cfffd3..054864f99 100644
---
a/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicCatalogTestHelper.java
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicCatalogTestHelper.java
@@ -25,7 +25,6 @@ import org.apache.amoro.api.CatalogMeta;
import org.apache.amoro.properties.CatalogMetaProperties;
import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
-import org.apache.amoro.utils.MixedFormatCatalogUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.catalog.Catalog;
@@ -100,7 +99,8 @@ public class BasicCatalogTestHelper implements
CatalogTestHelper {
"Cannot build mixed-tables for table format:" + tableFormat);
}
return new MixedTables(
- catalogMeta.getCatalogProperties(),
MixedFormatCatalogUtil.buildMetaStore(catalogMeta));
+ catalogMeta.getCatalogProperties(),
+ org.apache.amoro.utils.CatalogUtil.buildMetaStore(catalogMeta));
}
@Override
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicMixedCatalog.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicMixedCatalog.java
index 9717db05f..16b7c0afa 100644
---
a/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicMixedCatalog.java
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/BasicMixedCatalog.java
@@ -47,6 +47,7 @@ import org.apache.amoro.table.TableMetaStore;
import org.apache.amoro.table.TableProperties;
import org.apache.amoro.table.blocker.BasicTableBlockerManager;
import org.apache.amoro.table.blocker.TableBlockerManager;
+import org.apache.amoro.utils.CatalogUtil;
import org.apache.amoro.utils.CompatiblePropertyUtil;
import org.apache.amoro.utils.ConvertStructUtil;
import org.apache.amoro.utils.MixedFormatCatalogUtil;
@@ -214,7 +215,7 @@ public class BasicMixedCatalog implements
MixedFormatCatalog {
protected TableMeta getMixedTableMeta(TableIdentifier identifier) {
TableMeta tableMeta;
try {
- tableMeta =
getClient().getTable(MixedFormatCatalogUtil.amsTaleId(identifier));
+ tableMeta = getClient().getTable(CatalogUtil.amsTableId(identifier));
return tableMeta;
} catch (NoSuchObjectException e) {
throw new NoSuchTableException(e, "load table failed %s.", identifier);
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/TableTestBase.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/TableTestBase.java
index 2dc49c9d9..7e6419256 100644
---
a/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/TableTestBase.java
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/catalog/TableTestBase.java
@@ -23,7 +23,7 @@ import org.apache.amoro.table.MixedTable;
import org.apache.amoro.table.TableBuilder;
import org.apache.amoro.table.TableMetaStore;
import org.apache.amoro.table.UnkeyedTable;
-import org.apache.amoro.utils.MixedFormatCatalogUtil;
+import org.apache.amoro.utils.CatalogUtil;
import org.apache.amoro.utils.MixedTableUtil;
import org.junit.After;
import org.junit.Before;
@@ -41,7 +41,7 @@ public abstract class TableTestBase extends CatalogTestBase {
@Before
public void setupTable() {
- this.tableMetaStore =
MixedFormatCatalogUtil.buildMetaStore(getCatalogMeta());
+ this.tableMetaStore = CatalogUtil.buildMetaStore(getCatalogMeta());
getUnifiedCatalog().createDatabase(TableTestHelper.TEST_DB_NAME);
switch (getTestFormat()) {
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/IcebergHadoopCatalogTestHelper.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/IcebergHadoopCatalogTestHelper.java
index 6fad3b3c7..34625cc2d 100644
---
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/IcebergHadoopCatalogTestHelper.java
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/IcebergHadoopCatalogTestHelper.java
@@ -24,6 +24,7 @@ import org.apache.amoro.formats.iceberg.IcebergCatalogFactory;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
import org.apache.amoro.table.TableMetaStore;
+import org.apache.amoro.utils.CatalogUtil;
import org.apache.amoro.utils.MixedFormatCatalogUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.PartitionSpec;
@@ -73,7 +74,7 @@ public class IcebergHadoopCatalogTestHelper extends
AbstractFormatCatalogTestHel
@Override
public AmoroCatalog amoroCatalog() {
IcebergCatalogFactory icebergCatalogFactory = new IcebergCatalogFactory();
- TableMetaStore metaStore =
MixedFormatCatalogUtil.buildMetaStore(getCatalogMeta());
+ TableMetaStore metaStore = CatalogUtil.buildMetaStore(getCatalogMeta());
Map<String, String> properties =
icebergCatalogFactory.convertCatalogProperties(
catalogName, getMetastoreType(), catalogProperties);
@@ -85,7 +86,7 @@ public class IcebergHadoopCatalogTestHelper extends
AbstractFormatCatalogTestHel
Map<String, String> props =
MixedFormatCatalogUtil.withIcebergCatalogInitializeProperties(
catalogName, getMetastoreType(), catalogProperties);
- TableMetaStore metaStore =
MixedFormatCatalogUtil.buildMetaStore(getCatalogMeta());
+ TableMetaStore metaStore = CatalogUtil.buildMetaStore(getCatalogMeta());
return org.apache.iceberg.CatalogUtil.buildIcebergCatalog(
catalogName, props, metaStore.getConfiguration());
}
diff --git
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/MixedIcebergHadoopCatalogTestHelper.java
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/MixedIcebergHadoopCatalogTestHelper.java
index da7b00c9d..ceb5642bd 100644
---
a/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/MixedIcebergHadoopCatalogTestHelper.java
+++
b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/MixedIcebergHadoopCatalogTestHelper.java
@@ -28,7 +28,7 @@ import
org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
import org.apache.amoro.table.TableIdentifier;
import org.apache.amoro.table.TableMetaStore;
-import org.apache.amoro.utils.MixedFormatCatalogUtil;
+import org.apache.amoro.utils.CatalogUtil;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.types.Types;
@@ -63,7 +63,7 @@ public class MixedIcebergHadoopCatalogTestHelper
@Override
public AmoroCatalog amoroCatalog() {
MixedIcebergCatalogFactory mixedIcebergCatalogFactory = new
MixedIcebergCatalogFactory();
- TableMetaStore metaStore =
MixedFormatCatalogUtil.buildMetaStore(getCatalogMeta());
+ TableMetaStore metaStore = CatalogUtil.buildMetaStore(getCatalogMeta());
Map<String, String> properties =
mixedIcebergCatalogFactory.convertCatalogProperties(
catalogName, getMetastoreType(), catalogProperties);
@@ -74,7 +74,7 @@ public class MixedIcebergHadoopCatalogTestHelper
@Override
public MixedFormatCatalog originalCatalog() {
CatalogMeta meta = getCatalogMeta();
- TableMetaStore metaStore = MixedFormatCatalogUtil.buildMetaStore(meta);
+ TableMetaStore metaStore = CatalogUtil.buildMetaStore(meta);
return CatalogLoader.createCatalog(
catalogName(), meta.getCatalogType(), meta.getCatalogProperties(),
metaStore);
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/InternalCatalogBuilder.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/InternalCatalogBuilder.java
index b286defac..089255df5 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/InternalCatalogBuilder.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/InternalCatalogBuilder.java
@@ -18,18 +18,31 @@
package org.apache.amoro.flink;
+import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE;
+import static org.apache.iceberg.flink.FlinkCatalogFactory.HADOOP_CONF_DIR;
+import static org.apache.iceberg.flink.FlinkCatalogFactory.HIVE_CONF_DIR;
+
import org.apache.amoro.mixed.CatalogLoader;
import org.apache.amoro.mixed.MixedFormatCatalog;
import org.apache.amoro.properties.CatalogMetaProperties;
+import org.apache.amoro.shade.guava32.com.google.common.base.Strings;
+import org.apache.amoro.table.TableMetaStore;
import org.apache.amoro.utils.ConfigurationFileUtil;
-import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.runtime.util.HadoopUtils;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.util.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.flink.FlinkCatalogFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.Serializable;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Map;
@@ -39,12 +52,68 @@ public class InternalCatalogBuilder implements Serializable
{
private String metastoreUrl;
private Map<String, String> properties = new HashMap<>(0);
+ private String catalogName;
private MixedFormatCatalog createMixedFormatCatalog() {
- Preconditions.checkArgument(
- StringUtils.isNotBlank(metastoreUrl),
- "metastoreUrl can not be empty. e.g:
thrift://127.0.0.1:port/catalogName");
- return CatalogLoader.load(metastoreUrl, properties);
+ if (metastoreUrl != null) {
+ return CatalogLoader.load(metastoreUrl, properties);
+ } else {
+ Preconditions.checkArgument(catalogName != null, "Catalog name cannot be
empty");
+ String metastoreType =
properties.get(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE);
+ Preconditions.checkArgument(metastoreType != null, "Catalog type cannot
be empty");
+ TableMetaStore tableMetaStore =
+ TableMetaStore.builder()
+ .withConfiguration(clusterHadoopConf(metastoreType, properties))
+ .build();
+ return CatalogLoader.createCatalog(catalogName, metastoreType,
properties, tableMetaStore);
+ }
+ }
+
+ public static Configuration clusterHadoopConf(
+ String metastoreType, Map<String, String> properties) {
+ Configuration configuration =
+
HadoopUtils.getHadoopConfiguration(GlobalConfiguration.loadConfiguration());
+ if (ICEBERG_CATALOG_TYPE_HIVE.equals(metastoreType)) {
+ String hiveConfDir = properties.get(HIVE_CONF_DIR);
+ String hadoopConfDir = properties.get(HADOOP_CONF_DIR);
+ configuration = mergeHiveConf(configuration, hiveConfDir, hadoopConfDir);
+ }
+ return configuration;
+ }
+
+ private static Configuration mergeHiveConf(
+ Configuration hadoopConf, String hiveConfDir, String hadoopConfDir) {
+ Configuration newConf = new Configuration(hadoopConf);
+ if (!Strings.isNullOrEmpty(hiveConfDir)) {
+ Preconditions.checkState(
+ Files.exists(Paths.get(hiveConfDir, "hive-site.xml")),
+ "There should be a hive-site.xml file under the directory %s",
+ hiveConfDir);
+ newConf.addResource(new Path(hiveConfDir, "hive-site.xml"));
+ } else {
+ // If don't provide the hive-site.xml path explicitly, it will try to
load resource from
+ // classpath. If still
+ // couldn't load the configuration file, then it will throw exception in
HiveCatalog.
+ URL configFile =
InternalCatalogBuilder.class.getClassLoader().getResource("hive-site.xml");
+ if (configFile != null) {
+ newConf.addResource(configFile);
+ }
+ }
+
+ if (!Strings.isNullOrEmpty(hadoopConfDir)) {
+ Preconditions.checkState(
+ Files.exists(Paths.get(hadoopConfDir, "hdfs-site.xml")),
+ "Failed to load Hadoop configuration: missing %s",
+ Paths.get(hadoopConfDir, "hdfs-site.xml"));
+ newConf.addResource(new Path(hadoopConfDir, "hdfs-site.xml"));
+ Preconditions.checkState(
+ Files.exists(Paths.get(hadoopConfDir, "core-site.xml")),
+ "Failed to load Hadoop configuration: missing %s",
+ Paths.get(hadoopConfDir, "core-site.xml"));
+ newConf.addResource(new Path(hadoopConfDir, "core-site.xml"));
+ }
+
+ return newConf;
}
public String getMetastoreUrl() {
@@ -66,9 +135,6 @@ public class InternalCatalogBuilder implements Serializable {
}
public InternalCatalogBuilder metastoreUrl(String metastoreUrl) {
- Preconditions.checkArgument(
- StringUtils.isNotBlank(metastoreUrl),
- "metastore url can not be empty e.g:
thrift://127.0.0.1:port/catalogName");
this.metastoreUrl = metastoreUrl;
return this;
}
@@ -113,4 +179,9 @@ public class InternalCatalogBuilder implements Serializable
{
this.properties = finalProperties;
return this;
}
+
+ public InternalCatalogBuilder catalogName(String catalogName) {
+ this.catalogName = catalogName;
+ return this;
+ }
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
index 1ce6aff3b..3d944adb7 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/FlinkUnifiedCatalog.java
@@ -33,7 +33,7 @@ import
org.apache.amoro.flink.catalog.factories.FlinkUnifiedCatalogFactory;
import
org.apache.amoro.flink.catalog.factories.iceberg.IcebergFlinkCatalogFactory;
import org.apache.amoro.flink.catalog.factories.mixed.MixedCatalogFactory;
import
org.apache.amoro.flink.catalog.factories.paimon.PaimonFlinkCatalogFactory;
-import org.apache.amoro.flink.table.AmoroDynamicTableFactory;
+import org.apache.amoro.flink.table.UnifiedDynamicTableFactory;
import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.apache.amoro.table.TableIdentifier;
@@ -62,8 +62,6 @@ import
org.apache.flink.table.catalog.stats.CatalogTableStatistics;
import org.apache.flink.table.expressions.Expression;
import org.apache.flink.table.factories.CatalogFactory;
import org.apache.flink.table.factories.Factory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.List;
@@ -72,7 +70,6 @@ import java.util.Optional;
/** This is a Flink catalog wrap a unified catalog. */
public class FlinkUnifiedCatalog extends AbstractCatalog {
- private static final Logger LOG =
LoggerFactory.getLogger(FlinkUnifiedCatalog.class);
private final UnifiedCatalog unifiedCatalog;
private final String amsUri;
@@ -329,7 +326,7 @@ public class FlinkUnifiedCatalog extends AbstractCatalog {
@Override
public Optional<Factory> getFactory() {
- return Optional.of(new AmoroDynamicTableFactory(availableCatalogs));
+ return Optional.of(new UnifiedDynamicTableFactory(availableCatalogs));
}
@Override
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/MixedCatalog.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/MixedCatalog.java
index d43d9a1f0..d8dba8a50 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/MixedCatalog.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/MixedCatalog.java
@@ -27,7 +27,7 @@ import static
org.apache.flink.util.Preconditions.checkNotNull;
import org.apache.amoro.NoSuchDatabaseException;
import org.apache.amoro.flink.InternalCatalogBuilder;
import org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions;
-import org.apache.amoro.flink.table.DynamicTableFactory;
+import org.apache.amoro.flink.table.MixedDynamicTableFactory;
import org.apache.amoro.flink.table.descriptors.MixedFormatValidator;
import org.apache.amoro.flink.util.MixedFormatUtils;
import org.apache.amoro.mixed.MixedFormatCatalog;
@@ -237,7 +237,7 @@ public class MixedCatalog extends AbstractCatalog {
return;
}
- properties.put(CONNECTOR.key(), DynamicTableFactory.IDENTIFIER);
+ properties.put(CONNECTOR.key(), MixedDynamicTableFactory.IDENTIFIER);
properties.put(MixedFormatValidator.MIXED_FORMAT_CATALOG.key(),
tableIdentifier.getCatalog());
properties.put(MixedFormatValidator.MIXED_FORMAT_TABLE.key(),
tableIdentifier.getTableName());
properties.put(MixedFormatValidator.MIXED_FORMAT_DATABASE.key(),
tableIdentifier.getDatabase());
@@ -611,7 +611,7 @@ public class MixedCatalog extends AbstractCatalog {
@Override
public Optional<Factory> getFactory() {
- return Optional.of(new DynamicTableFactory(this));
+ return Optional.of(new MixedDynamicTableFactory(this));
}
public InternalCatalogBuilder catalogBuilder() {
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/CatalogFactoryOptions.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/CatalogFactoryOptions.java
index 75f57737a..464e32eb0 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/CatalogFactoryOptions.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/CatalogFactoryOptions.java
@@ -18,16 +18,13 @@
package org.apache.amoro.flink.catalog.factories;
-import static org.apache.amoro.flink.catalog.MixedCatalog.DEFAULT_DB;
import static org.apache.amoro.properties.CatalogMetaProperties.TABLE_FORMATS;
import org.apache.amoro.flink.catalog.FlinkUnifiedCatalog;
import org.apache.amoro.flink.catalog.MixedCatalog;
-import org.apache.amoro.properties.CatalogMetaProperties;
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;
/** {@link ConfigOption}s for {@link MixedCatalog} and {@link
FlinkUnifiedCatalog}. */
@Internal
@@ -37,53 +34,9 @@ public class CatalogFactoryOptions {
@Deprecated public static final String LEGACY_MIXED_IDENTIFIER = "arctic";
public static final String UNIFIED_IDENTIFIER = "unified";
- public static final ConfigOption<String> DEFAULT_DATABASE =
- ConfigOptions.key(CommonCatalogOptions.DEFAULT_DATABASE_KEY)
- .stringType()
- .defaultValue(DEFAULT_DB);
-
- public static final String PROPERTIES_PREFIX = "properties";
-
public static final ConfigOption<String> METASTORE_URL =
ConfigOptions.key("metastore.url").stringType().noDefaultValue();
- // authorization configs
- public static final ConfigOption<String> AUTH_AMS_CONFIGS_DISABLE =
- ConfigOptions.key(PROPERTIES_PREFIX + "." +
CatalogMetaProperties.LOAD_AUTH_FROM_AMS)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> AUTH_METHOD =
- ConfigOptions.key(PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> SIMPLE_USER_NAME =
- ConfigOptions.key(
- PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> KEYTAB_LOGIN_USER =
- ConfigOptions.key(PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_PRINCIPAL)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> KRB5_CONF_PATH =
- ConfigOptions.key(PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_KRB_PATH)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> KRB5_CONF_ENCODE =
- ConfigOptions.key(PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_KRB_ENCODE)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> KEYTAB_PATH =
- ConfigOptions.key(
- PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_KEYTAB_PATH)
- .stringType()
- .noDefaultValue();
- public static final ConfigOption<String> KEYTAB_ENCODE =
- ConfigOptions.key(
- PROPERTIES_PREFIX + "." +
CatalogMetaProperties.AUTH_CONFIGS_KEY_KEYTAB_ENCODE)
- .stringType()
- .noDefaultValue();
-
public static final ConfigOption<String> FLINK_TABLE_FORMATS =
ConfigOptions.key(TABLE_FORMATS)
.stringType()
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/FlinkUnifiedCatalogFactory.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/FlinkUnifiedCatalogFactory.java
index 8c7fc50ff..5b1cfcf86 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/FlinkUnifiedCatalogFactory.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/FlinkUnifiedCatalogFactory.java
@@ -19,25 +19,30 @@
package org.apache.amoro.flink.catalog.factories;
import static org.apache.amoro.Constants.THRIFT_TABLE_SERVICE_NAME;
-import static
org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions.DEFAULT_DATABASE;
+import static org.apache.amoro.flink.table.OptionsUtil.getCatalogProperties;
import static org.apache.amoro.properties.CatalogMetaProperties.TABLE_FORMATS;
-import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
+import org.apache.amoro.CommonUnifiedCatalog;
import org.apache.amoro.TableFormat;
import org.apache.amoro.UnifiedCatalog;
import org.apache.amoro.UnifiedCatalogLoader;
import org.apache.amoro.client.AmsThriftUrl;
+import org.apache.amoro.flink.InternalCatalogBuilder;
import org.apache.amoro.flink.catalog.FlinkUnifiedCatalog;
-import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.apache.amoro.flink.catalog.MixedCatalog;
import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
-import org.apache.amoro.utils.MixedFormatCatalogUtil;
+import org.apache.amoro.table.TableMetaStore;
+import org.apache.amoro.utils.CatalogUtil;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
import org.apache.flink.table.factories.CatalogFactory;
-import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.Preconditions;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.flink.FlinkCatalogFactory;
-import java.util.HashSet;
+import java.util.Collections;
+import java.util.Map;
import java.util.Set;
/** Factory for {@link FlinkUnifiedCatalog}. */
@@ -57,37 +62,46 @@ public class FlinkUnifiedCatalogFactory implements
CatalogFactory {
@Override
public Set<ConfigOption<?>> requiredOptions() {
- Set<ConfigOption<?>> requiredOptions = new HashSet<>();
- requiredOptions.add(CatalogFactoryOptions.METASTORE_URL);
- return requiredOptions;
+ return Collections.emptySet();
}
@Override
public Set<ConfigOption<?>> optionalOptions() {
- final Set<ConfigOption<?>> options = new HashSet<>();
- options.add(PROPERTY_VERSION);
- options.add(DEFAULT_DATABASE);
- return options;
+ return Collections.emptySet();
}
@Override
public Catalog createCatalog(Context context) {
- final FactoryUtil.CatalogFactoryHelper helper =
- FactoryUtil.createCatalogFactoryHelper(this, context);
- helper.validate();
- final String defaultDatabase = helper.getOptions().get(DEFAULT_DATABASE);
- String metastoreUrl =
helper.getOptions().get(CatalogFactoryOptions.METASTORE_URL);
+ final String defaultDatabase =
+ context
+ .getOptions()
+ .getOrDefault(CommonCatalogOptions.DEFAULT_DATABASE_KEY,
MixedCatalog.DEFAULT_DB);
+ final String metastoreUrl =
context.getOptions().get(CatalogFactoryOptions.METASTORE_URL.key());
+ final Map<String, String> catalogProperties =
getCatalogProperties(context.getOptions());
- String amoroCatalogName =
- AmsThriftUrl.parse(metastoreUrl,
THRIFT_TABLE_SERVICE_NAME).catalogName();
- UnifiedCatalog unifiedCatalog =
- UnifiedCatalogLoader.loadUnifiedCatalog(metastoreUrl,
amoroCatalogName, Maps.newHashMap());
+ UnifiedCatalog unifiedCatalog;
+ if (metastoreUrl != null) {
+ String amoroCatalogName =
+ AmsThriftUrl.parse(metastoreUrl,
THRIFT_TABLE_SERVICE_NAME).catalogName();
+ unifiedCatalog =
+ UnifiedCatalogLoader.loadUnifiedCatalog(
+ metastoreUrl, amoroCatalogName, catalogProperties);
+ } else {
+ String metastoreType =
catalogProperties.get(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE);
+ Preconditions.checkArgument(metastoreType != null, "Catalog type cannot
be empty");
+ TableMetaStore tableMetaStore =
+ TableMetaStore.builder()
+ .withConfiguration(
+ InternalCatalogBuilder.clusterHadoopConf(metastoreType,
catalogProperties))
+ .build();
+ unifiedCatalog =
+ new CommonUnifiedCatalog(
+ context.getName(), metastoreType, catalogProperties,
tableMetaStore);
+ }
Configuration hadoopConf =
unifiedCatalog.authenticationContext().getConfiguration();
-
Set<TableFormat> tableFormats =
- MixedFormatCatalogUtil.tableFormats(
- unifiedCatalog.metastoreType(), unifiedCatalog.properties());
+ CatalogUtil.tableFormats(unifiedCatalog.metastoreType(),
unifiedCatalog.properties());
validate(tableFormats);
return new FlinkUnifiedCatalog(
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/mixed/MixedCatalogFactory.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/mixed/MixedCatalogFactory.java
index fd586e4d6..2298c859b 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/mixed/MixedCatalogFactory.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/catalog/factories/mixed/MixedCatalogFactory.java
@@ -18,22 +18,18 @@
package org.apache.amoro.flink.catalog.factories.mixed;
-import static
org.apache.amoro.flink.table.KafkaConnectorOptionsUtil.getKafkaParams;
-import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
+import static org.apache.amoro.flink.table.OptionsUtil.getCatalogProperties;
import org.apache.amoro.flink.InternalCatalogBuilder;
import org.apache.amoro.flink.catalog.MixedCatalog;
import org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions;
-import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
import org.apache.flink.table.factories.CatalogFactory;
-import org.apache.flink.table.factories.FactoryUtil;
import java.util.Collections;
-import java.util.HashSet;
import java.util.Map;
-import java.util.Optional;
import java.util.Set;
/** Factory for {@link MixedCatalog} */
@@ -47,25 +43,20 @@ public class MixedCatalogFactory implements CatalogFactory {
@Override
public Catalog createCatalog(Context context) {
- final FactoryUtil.CatalogFactoryHelper helper =
- FactoryUtil.createCatalogFactoryHelper(this, context);
- helper.validate();
+ final String defaultDatabase =
+ context
+ .getOptions()
+ .getOrDefault(CommonCatalogOptions.DEFAULT_DATABASE_KEY,
MixedCatalog.DEFAULT_DB);
+ final String metastoreUrl =
context.getOptions().get(CatalogFactoryOptions.METASTORE_URL.key());
+ final Map<String, String> catalogProperties =
getCatalogProperties(context.getOptions());
- final String defaultDatabase =
helper.getOptions().get(CatalogFactoryOptions.DEFAULT_DATABASE);
- String metastoreUrl =
helper.getOptions().get(CatalogFactoryOptions.METASTORE_URL);
- final Map<String, String> mixedCatalogProperties =
getKafkaParams(context.getOptions());
- final Map<String, String> catalogProperties =
Maps.newHashMap(mixedCatalogProperties);
+ final InternalCatalogBuilder catalogBuilder =
+ InternalCatalogBuilder.builder()
+ .metastoreUrl(metastoreUrl)
+ .catalogName(context.getName())
+ .properties(catalogProperties);
- Optional<String> tableFormatsOptional =
-
helper.getOptions().getOptional(CatalogFactoryOptions.FLINK_TABLE_FORMATS);
- tableFormatsOptional.ifPresent(
- tableFormats ->
-
catalogProperties.put(CatalogFactoryOptions.FLINK_TABLE_FORMATS.key(),
tableFormats));
-
- return new MixedCatalog(
- context.getName(),
- defaultDatabase,
-
InternalCatalogBuilder.builder().metastoreUrl(metastoreUrl).properties(catalogProperties));
+ return new MixedCatalog(context.getName(), defaultDatabase,
catalogBuilder);
}
@Override
@@ -75,22 +66,6 @@ public class MixedCatalogFactory implements CatalogFactory {
@Override
public Set<ConfigOption<?>> optionalOptions() {
- final Set<ConfigOption<?>> options = new HashSet<>();
- options.add(PROPERTY_VERSION);
- options.add(CatalogFactoryOptions.METASTORE_URL);
- options.add(CatalogFactoryOptions.DEFAULT_DATABASE);
-
- // authorization config
- options.add(CatalogFactoryOptions.AUTH_AMS_CONFIGS_DISABLE);
- options.add(CatalogFactoryOptions.AUTH_METHOD);
- options.add(CatalogFactoryOptions.SIMPLE_USER_NAME);
- options.add(CatalogFactoryOptions.KEYTAB_LOGIN_USER);
- options.add(CatalogFactoryOptions.KRB5_CONF_PATH);
- options.add(CatalogFactoryOptions.KRB5_CONF_ENCODE);
- options.add(CatalogFactoryOptions.KEYTAB_PATH);
- options.add(CatalogFactoryOptions.KEYTAB_ENCODE);
-
- options.add(CatalogFactoryOptions.FLINK_TABLE_FORMATS);
- return options;
+ return Collections.emptySet();
}
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/KafkaConnectorOptionsUtil.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/KafkaConnectorOptionsUtil.java
deleted file mode 100644
index 0cab7037a..000000000
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/KafkaConnectorOptionsUtil.java
+++ /dev/null
@@ -1,567 +0,0 @@
-/*
- * 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://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.amoro.flink.table;
-
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.DELIVERY_GUARANTEE;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.KEY_FIELDS;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.KEY_FIELDS_PREFIX;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.KEY_FORMAT;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.SCAN_STARTUP_MODE;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.SCAN_STARTUP_SPECIFIC_OFFSETS;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.SCAN_STARTUP_TIMESTAMP_MILLIS;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.SINK_PARTITIONER;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.TOPIC;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.TOPIC_PATTERN;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.TRANSACTIONAL_ID_PREFIX;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.VALUE_FIELDS_INCLUDE;
-import static
org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.VALUE_FORMAT;
-import static org.apache.flink.table.factories.FactoryUtil.FORMAT;
-
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.connector.base.DeliveryGuarantee;
-import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
-import
org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
-import
org.apache.flink.streaming.connectors.kafka.partitioner.FlinkFixedPartitioner;
-import
org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
-import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;
-import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.factories.DynamicTableFactory;
-import org.apache.flink.table.factories.FactoryUtil;
-import org.apache.flink.table.types.DataType;
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
-import org.apache.flink.table.types.logical.utils.LogicalTypeChecks;
-import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.InstantiationUtil;
-import org.apache.flink.util.Preconditions;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-public class KafkaConnectorOptionsUtil {
-
- private static final ConfigOption<String> SCHEMA_REGISTRY_SUBJECT =
-
ConfigOptions.key("schema-registry.subject").stringType().noDefaultValue();
-
- //
--------------------------------------------------------------------------------------------
- // Option enumerations
- //
--------------------------------------------------------------------------------------------
-
- // Sink partitioner.
- public static final String SINK_PARTITIONER_VALUE_DEFAULT = "default";
- public static final String SINK_PARTITIONER_VALUE_FIXED = "fixed";
- public static final String SINK_PARTITIONER_VALUE_ROUND_ROBIN =
"round-robin";
-
- // Prefix for Kafka specific properties.
- public static final String PROPERTIES_PREFIX = "properties.";
-
- // Other keywords.
- private static final String PARTITION = "partition";
- private static final String OFFSET = "offset";
- protected static final String AVRO_CONFLUENT = "avro-confluent";
- protected static final String DEBEZIUM_AVRO_CONFLUENT =
"debezium-avro-confluent";
- private static final List<String> SCHEMA_REGISTRY_FORMATS =
- Arrays.asList(AVRO_CONFLUENT, DEBEZIUM_AVRO_CONFLUENT);
-
- //
--------------------------------------------------------------------------------------------
- // Validation
- //
--------------------------------------------------------------------------------------------
-
- public static void validateTableSourceOptions(ReadableConfig tableOptions) {
- validateSourceTopic(tableOptions);
- validateScanStartupMode(tableOptions);
- }
-
- public static void validateTableSinkOptions(ReadableConfig tableOptions) {
- validateSinkTopic(tableOptions);
- validateSinkPartitioner(tableOptions);
- }
-
- public static void validateSourceTopic(ReadableConfig tableOptions) {
- Optional<List<String>> topic = tableOptions.getOptional(TOPIC);
- Optional<String> pattern = tableOptions.getOptional(TOPIC_PATTERN);
-
- if (topic.isPresent() && pattern.isPresent()) {
- throw new ValidationException(
- "Option 'topic' and 'topic-pattern' shouldn't be set together.");
- }
-
- if (!topic.isPresent() && !pattern.isPresent()) {
- throw new ValidationException("Either 'topic' or 'topic-pattern' must be
set.");
- }
- }
-
- public static void validateSinkTopic(ReadableConfig tableOptions) {
- String errorMessageTemp =
- "Flink Kafka sink currently only supports single topic, but got %s:
%s.";
- if (!isSingleTopic(tableOptions)) {
- if (tableOptions.getOptional(TOPIC_PATTERN).isPresent()) {
- throw new ValidationException(
- String.format(errorMessageTemp, "'topic-pattern'",
tableOptions.get(TOPIC_PATTERN)));
- } else {
- throw new ValidationException(
- String.format(errorMessageTemp, "'topic'",
tableOptions.get(TOPIC)));
- }
- }
- }
-
- private static void validateScanStartupMode(ReadableConfig tableOptions) {
- tableOptions
- .getOptional(SCAN_STARTUP_MODE)
- .ifPresent(
- mode -> {
- switch (mode) {
- case TIMESTAMP:
- if
(!tableOptions.getOptional(SCAN_STARTUP_TIMESTAMP_MILLIS).isPresent()) {
- throw new ValidationException(
- String.format(
- "'%s' is required in '%s' startup mode" + " but
missing.",
- SCAN_STARTUP_TIMESTAMP_MILLIS.key(),
- KafkaConnectorOptions.ScanStartupMode.TIMESTAMP));
- }
-
- break;
- case SPECIFIC_OFFSETS:
- if
(!tableOptions.getOptional(SCAN_STARTUP_SPECIFIC_OFFSETS).isPresent()) {
- throw new ValidationException(
- String.format(
- "'%s' is required in '%s' startup mode" + " but
missing.",
- SCAN_STARTUP_SPECIFIC_OFFSETS.key(),
-
KafkaConnectorOptions.ScanStartupMode.SPECIFIC_OFFSETS));
- }
- if (!isSingleTopic(tableOptions)) {
- throw new ValidationException(
- "Currently Kafka source only supports specific offset
for single topic.");
- }
- String specificOffsets =
tableOptions.get(SCAN_STARTUP_SPECIFIC_OFFSETS);
- parseSpecificOffsets(specificOffsets,
SCAN_STARTUP_SPECIFIC_OFFSETS.key());
-
- break;
- }
- });
- }
-
- private static void validateSinkPartitioner(ReadableConfig tableOptions) {
- tableOptions
- .getOptional(SINK_PARTITIONER)
- .ifPresent(
- partitioner -> {
- if (partitioner.equals(SINK_PARTITIONER_VALUE_ROUND_ROBIN)
- && tableOptions.getOptional(KEY_FIELDS).isPresent()) {
- throw new ValidationException(
- "Currently 'round-robin' partitioner only works when
option 'key.fields' is not specified.");
- } else if (partitioner.isEmpty()) {
- throw new ValidationException(
- String.format(
- "Option '%s' should be a non-empty string.",
SINK_PARTITIONER.key()));
- }
- });
- }
-
- //
--------------------------------------------------------------------------------------------
- // Utilities
- //
--------------------------------------------------------------------------------------------
-
- public static List<String> getSourceTopics(ReadableConfig tableOptions) {
- return tableOptions.getOptional(TOPIC).orElse(null);
- }
-
- public static Pattern getSourceTopicPattern(ReadableConfig tableOptions) {
- return
tableOptions.getOptional(TOPIC_PATTERN).map(Pattern::compile).orElse(null);
- }
-
- private static boolean isSingleTopic(ReadableConfig tableOptions) {
- // Option 'topic-pattern' is regarded as multi-topics.
- return tableOptions.getOptional(TOPIC).map(t -> t.size() ==
1).orElse(false);
- }
-
- public static StartupOptions getStartupOptions(ReadableConfig tableOptions) {
- final Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>();
- final StartupMode startupMode =
- tableOptions
- .getOptional(SCAN_STARTUP_MODE)
- .map(KafkaConnectorOptionsUtil::fromOption)
- .orElse(StartupMode.GROUP_OFFSETS);
- if (startupMode == StartupMode.SPECIFIC_OFFSETS) {
- // It will be refactored after support specific offset for multiple
topics in
- // FLINK-18602. We have already checked tableOptions.get(TOPIC) contains
one topic in
- // validateScanStartupMode().
- buildSpecificOffsets(tableOptions, tableOptions.get(TOPIC).get(0),
specificOffsets);
- }
-
- final StartupOptions options = new StartupOptions();
- options.startupMode = startupMode;
- options.specificOffsets = specificOffsets;
- if (startupMode == StartupMode.TIMESTAMP) {
- options.startupTimestampMillis =
tableOptions.get(SCAN_STARTUP_TIMESTAMP_MILLIS);
- }
- return options;
- }
-
- private static void buildSpecificOffsets(
- ReadableConfig tableOptions, String topic, Map<KafkaTopicPartition,
Long> specificOffsets) {
- String specificOffsetsStrOpt =
tableOptions.get(SCAN_STARTUP_SPECIFIC_OFFSETS);
- final Map<Integer, Long> offsetMap =
- parseSpecificOffsets(specificOffsetsStrOpt,
SCAN_STARTUP_SPECIFIC_OFFSETS.key());
- offsetMap.forEach(
- (partition, offset) -> {
- final KafkaTopicPartition topicPartition = new
KafkaTopicPartition(topic, partition);
- specificOffsets.put(topicPartition, offset);
- });
- }
-
- /**
- * Returns the {@link StartupMode} of Kafka Consumer by passed-in
table-specific {@link
- * KafkaConnectorOptions.ScanStartupMode}.
- */
- private static StartupMode fromOption(KafkaConnectorOptions.ScanStartupMode
scanStartupMode) {
- switch (scanStartupMode) {
- case EARLIEST_OFFSET:
- return StartupMode.EARLIEST;
- case LATEST_OFFSET:
- return StartupMode.LATEST;
- case GROUP_OFFSETS:
- return StartupMode.GROUP_OFFSETS;
- case SPECIFIC_OFFSETS:
- return StartupMode.SPECIFIC_OFFSETS;
- case TIMESTAMP:
- return StartupMode.TIMESTAMP;
-
- default:
- throw new TableException("Unsupported startup mode. Validator should
have checked that.");
- }
- }
-
- public static Properties getKafkaProperties(Map<String, String>
tableOptions) {
- final Properties kafkaProperties = new Properties();
-
- if (hasKafkaClientProperties(tableOptions)) {
- tableOptions.keySet().stream()
- .filter(key -> key.startsWith(PROPERTIES_PREFIX))
- .forEach(
- key -> {
- final String value = tableOptions.get(key);
- final String subKey =
key.substring((PROPERTIES_PREFIX).length());
- kafkaProperties.put(subKey, value);
- });
- }
- return kafkaProperties;
- }
-
- public static Map<String, String> getKafkaParams(Map<String, String>
tableOptions) {
- if (hasKafkaClientProperties(tableOptions)) {
- return tableOptions.entrySet().stream()
- .filter(e -> e.getKey().startsWith(PROPERTIES_PREFIX))
- .collect(
- Collectors.toMap(
- e -> e.getKey().substring(PROPERTIES_PREFIX.length()),
Map.Entry::getValue));
- }
- return Collections.emptyMap();
- }
-
- /**
- * The partitioner can be either "fixed", "round-robin" or a customized
partitioner full class
- * name.
- */
- public static Optional<FlinkKafkaPartitioner<RowData>>
getFlinkKafkaPartitioner(
- ReadableConfig tableOptions, ClassLoader classLoader) {
- return tableOptions
- .getOptional(SINK_PARTITIONER)
- .flatMap(
- (String partitioner) -> {
- switch (partitioner) {
- case SINK_PARTITIONER_VALUE_FIXED:
- return Optional.of(new FlinkFixedPartitioner<>());
- case SINK_PARTITIONER_VALUE_DEFAULT:
- case SINK_PARTITIONER_VALUE_ROUND_ROBIN:
- return Optional.empty();
- // Default fallback to full class name of the partitioner.
- default:
- return Optional.of(initializePartitioner(partitioner,
classLoader));
- }
- });
- }
-
- /**
- * Parses SpecificOffsets String to Map.
- *
- * <p>SpecificOffsets String format was given as following:
- *
- * <pre>
- * scan.startup.specific-offsets =
partition:0,offset:42;partition:1,offset:300
- * </pre>
- *
- * @return SpecificOffsets with Map format, key is partition, and value is
offset
- */
- public static Map<Integer, Long> parseSpecificOffsets(
- String specificOffsetsStr, String optionKey) {
- final Map<Integer, Long> offsetMap = new HashMap<>();
- final String[] pairs = specificOffsetsStr.split(";");
- final String validationExceptionMessage =
- String.format(
- "Invalid properties '%s' should follow the format "
- + "'partition:0,offset:42;partition:1,offset:300', but is
'%s'.",
- optionKey, specificOffsetsStr);
-
- if (pairs.length == 0) {
- throw new ValidationException(validationExceptionMessage);
- }
-
- for (String pair : pairs) {
- if (null == pair || pair.length() == 0 || !pair.contains(",")) {
- throw new ValidationException(validationExceptionMessage);
- }
-
- final String[] kv = pair.split(",");
- if (kv.length != 2 || !kv[0].startsWith(PARTITION + ':') ||
!kv[1].startsWith(OFFSET + ':')) {
- throw new ValidationException(validationExceptionMessage);
- }
-
- String partitionValue = kv[0].substring(kv[0].indexOf(":") + 1);
- String offsetValue = kv[1].substring(kv[1].indexOf(":") + 1);
- try {
- final Integer partition = Integer.valueOf(partitionValue);
- final Long offset = Long.valueOf(offsetValue);
- offsetMap.put(partition, offset);
- } catch (NumberFormatException e) {
- throw new ValidationException(validationExceptionMessage, e);
- }
- }
- return offsetMap;
- }
-
- /**
- * Decides if the table options contains Kafka client properties that start
with prefix
- * 'properties'.
- */
- private static boolean hasKafkaClientProperties(Map<String, String>
tableOptions) {
- return tableOptions.keySet().stream().anyMatch(k ->
k.startsWith(PROPERTIES_PREFIX));
- }
-
- /** Returns a class value with the given class name. */
- private static <T> FlinkKafkaPartitioner<T> initializePartitioner(
- String name, ClassLoader classLoader) {
- try {
- Class<?> clazz = Class.forName(name, true, classLoader);
- if (!FlinkKafkaPartitioner.class.isAssignableFrom(clazz)) {
- throw new ValidationException(
- String.format(
- "Sink partitioner class '%s' should extend from the required
class %s",
- name, FlinkKafkaPartitioner.class.getName()));
- }
- @SuppressWarnings("unchecked")
- final FlinkKafkaPartitioner<T> kafkaPartitioner =
- InstantiationUtil.instantiate(name, FlinkKafkaPartitioner.class,
classLoader);
-
- return kafkaPartitioner;
- } catch (ClassNotFoundException | FlinkException e) {
- throw new ValidationException(
- String.format("Could not find and instantiate partitioner class
'%s'", name), e);
- }
- }
-
- /**
- * Creates an array of indices that determine which physical fields of the
table schema to include
- * in the key format and the order that those fields have in the key format.
- *
- * <p>See {@link KafkaConnectorOptions#KEY_FORMAT}, {@link
KafkaConnectorOptions#KEY_FIELDS}, and
- * {@link KafkaConnectorOptions#KEY_FIELDS_PREFIX} for more information.
- */
- public static int[] createKeyFormatProjection(ReadableConfig options,
DataType physicalDataType) {
- final LogicalType physicalType = physicalDataType.getLogicalType();
- Preconditions.checkArgument(physicalType.is(LogicalTypeRoot.ROW), "Row
data type expected.");
- final Optional<String> optionalKeyFormat = options.getOptional(KEY_FORMAT);
- final Optional<List<String>> optionalKeyFields =
options.getOptional(KEY_FIELDS);
-
- if (!optionalKeyFormat.isPresent() && optionalKeyFields.isPresent()) {
- throw new ValidationException(
- String.format(
- "The option '%s' can only be declared if a key format is defined
using '%s'.",
- KEY_FIELDS.key(), KEY_FORMAT.key()));
- } else if (optionalKeyFormat.isPresent()
- && (!optionalKeyFields.isPresent() || optionalKeyFields.get().size()
== 0)) {
- throw new ValidationException(
- String.format(
- "A key format '%s' requires the declaration of one or more of
key fields using '%s'.",
- KEY_FORMAT.key(), KEY_FIELDS.key()));
- }
-
- if (!optionalKeyFormat.isPresent()) {
- return new int[0];
- }
-
- final String keyPrefix = options.getOptional(KEY_FIELDS_PREFIX).orElse("");
-
- final List<String> keyFields = optionalKeyFields.get();
- final List<String> physicalFields =
LogicalTypeChecks.getFieldNames(physicalType);
- return keyFields.stream()
- .mapToInt(
- keyField -> {
- final int pos = physicalFields.indexOf(keyField);
- // check that field name exists
- if (pos < 0) {
- throw new ValidationException(
- String.format(
- "Could not find the field '%s' in the table schema for
usage in the key format. "
- + "A key field must be a regular, physical column.
"
- + "The following columns can be selected in the
'%s' option:\n"
- + "%s",
- keyField, KEY_FIELDS.key(), physicalFields));
- }
- // check that field name is prefixed correctly
- if (!keyField.startsWith(keyPrefix)) {
- throw new ValidationException(
- String.format(
- "All fields in '%s' must be prefixed with '%s' when
option '%s' "
- + "is set but field '%s' is not prefixed.",
- KEY_FIELDS.key(), keyPrefix, KEY_FIELDS_PREFIX.key(),
keyField));
- }
- return pos;
- })
- .toArray();
- }
-
- /**
- * Creates an array of indices that determine which physical fields of the
table schema to include
- * in the value format.
- *
- * <p>See {@link KafkaConnectorOptions#VALUE_FORMAT}, {@link
- * KafkaConnectorOptions#VALUE_FIELDS_INCLUDE}, and {@link
- * KafkaConnectorOptions#KEY_FIELDS_PREFIX} for more information.
- */
- public static int[] createValueFormatProjection(
- ReadableConfig options, DataType physicalDataType) {
- final LogicalType physicalType = physicalDataType.getLogicalType();
- Preconditions.checkArgument(physicalType.is(LogicalTypeRoot.ROW), "Row
data type expected.");
- final int physicalFieldCount =
LogicalTypeChecks.getFieldCount(physicalType);
- final IntStream physicalFields = IntStream.range(0, physicalFieldCount);
-
- final String keyPrefix = options.getOptional(KEY_FIELDS_PREFIX).orElse("");
-
- final KafkaConnectorOptions.ValueFieldsStrategy strategy =
options.get(VALUE_FIELDS_INCLUDE);
- if (strategy == KafkaConnectorOptions.ValueFieldsStrategy.ALL) {
- if (keyPrefix.length() > 0) {
- throw new ValidationException(
- String.format(
- "A key prefix is not allowed when option '%s' is set to '%s'. "
- + "Set it to '%s' instead to avoid field overlaps.",
- VALUE_FIELDS_INCLUDE.key(),
- KafkaConnectorOptions.ValueFieldsStrategy.ALL,
- KafkaConnectorOptions.ValueFieldsStrategy.EXCEPT_KEY));
- }
- return physicalFields.toArray();
- } else if (strategy ==
KafkaConnectorOptions.ValueFieldsStrategy.EXCEPT_KEY) {
- final int[] keyProjection = createKeyFormatProjection(options,
physicalDataType);
- return physicalFields
- .filter(pos -> IntStream.of(keyProjection).noneMatch(k -> k == pos))
- .toArray();
- }
- throw new TableException("Unknown value fields strategy:" + strategy);
- }
-
- /**
- * Returns a new table context with a default schema registry subject value
in the options if the
- * format is a schema registry format (e.g. 'avro-confluent') and the
subject is not defined.
- */
- public static DynamicTableFactory.Context autoCompleteSchemaRegistrySubject(
- DynamicTableFactory.Context context) {
- Map<String, String> tableOptions = context.getCatalogTable().getOptions();
- Map<String, String> newOptions =
autoCompleteSchemaRegistrySubject(tableOptions);
- if (newOptions.size() > tableOptions.size()) {
- // build a new context
- return new FactoryUtil.DefaultDynamicTableContext(
- context.getObjectIdentifier(),
- context.getCatalogTable().copy(newOptions),
- context.getEnrichmentOptions(),
- context.getConfiguration(),
- context.getClassLoader(),
- context.isTemporary());
- } else {
- return context;
- }
- }
-
- private static Map<String, String> autoCompleteSchemaRegistrySubject(
- Map<String, String> options) {
- Configuration configuration = Configuration.fromMap(options);
- // the subject autoComplete should only be used in sink, check the topic
first
- validateSinkTopic(configuration);
- final Optional<String> valueFormat =
configuration.getOptional(VALUE_FORMAT);
- final Optional<String> keyFormat = configuration.getOptional(KEY_FORMAT);
- final Optional<String> format = configuration.getOptional(FORMAT);
- final String topic = configuration.get(TOPIC).get(0);
-
- if (format.isPresent() && SCHEMA_REGISTRY_FORMATS.contains(format.get())) {
- autoCompleteSubject(configuration, format.get(), topic + "-value");
- } else if (valueFormat.isPresent() &&
SCHEMA_REGISTRY_FORMATS.contains(valueFormat.get())) {
- autoCompleteSubject(configuration, "value." + valueFormat.get(), topic +
"-value");
- }
-
- if (keyFormat.isPresent() &&
SCHEMA_REGISTRY_FORMATS.contains(keyFormat.get())) {
- autoCompleteSubject(configuration, "key." + keyFormat.get(), topic +
"-key");
- }
- return configuration.toMap();
- }
-
- private static void autoCompleteSubject(
- Configuration configuration, String format, String subject) {
- ConfigOption<String> subjectOption =
- ConfigOptions.key(format + "." + SCHEMA_REGISTRY_SUBJECT.key())
- .stringType()
- .noDefaultValue();
- if (!configuration.getOptional(subjectOption).isPresent()) {
- configuration.setString(subjectOption, subject);
- }
- }
-
- static void validateDeliveryGuarantee(ReadableConfig tableOptions) {
- if (tableOptions.get(DELIVERY_GUARANTEE) == DeliveryGuarantee.EXACTLY_ONCE
- && !tableOptions.getOptional(TRANSACTIONAL_ID_PREFIX).isPresent()) {
- throw new ValidationException(
- TRANSACTIONAL_ID_PREFIX.key()
- + " must be specified when using
DeliveryGuarantee.EXACTLY_ONCE.");
- }
- }
-
- //
--------------------------------------------------------------------------------------------
- // Inner classes
- //
--------------------------------------------------------------------------------------------
-
- /** Kafka startup options. * */
- public static class StartupOptions {
- public StartupMode startupMode;
- public Map<KafkaTopicPartition, Long> specificOffsets;
- public long startupTimestampMillis;
- }
-
- private KafkaConnectorOptionsUtil() {}
-}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/DynamicTableFactory.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedDynamicTableFactory.java
similarity index 94%
rename from
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/DynamicTableFactory.java
rename to
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedDynamicTableFactory.java
index 6e62b83d5..60f8f05a6 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/DynamicTableFactory.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedDynamicTableFactory.java
@@ -61,6 +61,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
@@ -68,18 +69,19 @@ import java.util.Properties;
import java.util.Set;
/** A factory generates {@link MixedFormatDynamicSource} and {@link
MixedFormatDynamicSink} */
-public class DynamicTableFactory implements DynamicTableSourceFactory,
DynamicTableSinkFactory {
- private static final Logger LOG =
LoggerFactory.getLogger(DynamicTableFactory.class);
+public class MixedDynamicTableFactory
+ implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+ private static final Logger LOG =
LoggerFactory.getLogger(MixedDynamicTableFactory.class);
public static final String IDENTIFIER = "mixed-format";
private InternalCatalogBuilder internalCatalogBuilder;
private String internalCatalogName;
- public DynamicTableFactory(MixedCatalog mixedCatalog) {
+ public MixedDynamicTableFactory(MixedCatalog mixedCatalog) {
this.internalCatalogBuilder = mixedCatalog.catalogBuilder();
this.internalCatalogName = mixedCatalog.amsCatalogName();
}
- public DynamicTableFactory() {}
+ public MixedDynamicTableFactory() {}
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
@@ -95,15 +97,16 @@ public class DynamicTableFactory implements
DynamicTableSourceFactory, DynamicTa
// It denotes create table by ddl 'connector' option, not through
catalog.db.tableName
if (actualBuilder == null || actualCatalogName == null) {
- String metastoreUrl = options.get(CatalogFactoryOptions.METASTORE_URL);
- Preconditions.checkNotNull(
- metastoreUrl, String.format("%s should be set",
CatalogFactoryOptions.METASTORE_URL));
- actualBuilder =
InternalCatalogBuilder.builder().metastoreUrl(metastoreUrl);
-
actualCatalogName =
options.get(MixedFormatValidator.MIXED_FORMAT_CATALOG);
Preconditions.checkNotNull(
actualCatalogName,
String.format("%s should be set",
MixedFormatValidator.MIXED_FORMAT_CATALOG.key()));
+ String metastoreUrl = options.get(CatalogFactoryOptions.METASTORE_URL);
+ actualBuilder =
+ InternalCatalogBuilder.builder()
+ .metastoreUrl(metastoreUrl)
+ .catalogName(actualCatalogName)
+ .properties(options.toMap());
}
if (options.containsKey(MixedFormatValidator.MIXED_FORMAT_DATABASE.key())
@@ -211,7 +214,7 @@ public class DynamicTableFactory implements
DynamicTableSourceFactory, DynamicTa
@Override
public Set<ConfigOption<?>> requiredOptions() {
- return new HashSet<>();
+ return Collections.emptySet();
}
@Override
@@ -247,8 +250,7 @@ public class DynamicTableFactory implements
DynamicTableSourceFactory, DynamicTa
TableSchema physicalSchema =
TableSchemaUtils.getPhysicalSchema(catalogTable.getSchema());
Schema schema = FlinkSchemaUtil.convert(physicalSchema);
- final Properties properties =
- KafkaConnectorOptionsUtil.getKafkaProperties(mixedTable.properties());
+ final Properties properties =
OptionsUtil.getKafkaProperties(mixedTable.properties());
// add topic-partition discovery
final Optional<Long> partitionDiscoveryInterval =
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/OptionsUtil.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/OptionsUtil.java
new file mode 100644
index 000000000..38ef34666
--- /dev/null
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/OptionsUtil.java
@@ -0,0 +1,64 @@
+/*
+ * 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://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.amoro.flink.table;
+
+import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+
+import java.util.Map;
+import java.util.Properties;
+
+public class OptionsUtil {
+
+ // Prefix for specific properties.
+ public static final String PROPERTIES_PREFIX = "properties.";
+
+ public static Properties getKafkaProperties(Map<String, String>
tableOptions) {
+ final Properties kafkaProperties = new Properties();
+
+ if (hasProperties(tableOptions)) {
+ tableOptions.keySet().stream()
+ .filter(key -> key.startsWith(PROPERTIES_PREFIX))
+ .forEach(
+ key -> {
+ final String value = tableOptions.get(key);
+ final String subKey =
key.substring((PROPERTIES_PREFIX).length());
+ kafkaProperties.put(subKey, value);
+ });
+ }
+ return kafkaProperties;
+ }
+
+ public static Map<String, String> getCatalogProperties(Map<String, String>
options) {
+ Map<String, String> catalogProperties = Maps.newHashMap();
+ options.forEach(
+ (key, value) -> {
+ if (key.startsWith(PROPERTIES_PREFIX)) {
+ catalogProperties.put(key.substring((PROPERTIES_PREFIX).length()),
value);
+ } else {
+ catalogProperties.put(key, value);
+ }
+ });
+ return catalogProperties;
+ }
+
+ /** Decides if the table options contains table properties that start with
prefix 'properties'. */
+ private static boolean hasProperties(Map<String, String> tableOptions) {
+ return tableOptions.keySet().stream().anyMatch(k ->
k.startsWith(PROPERTIES_PREFIX));
+ }
+}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/AmoroDynamicTableFactory.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnifiedDynamicTableFactory.java
similarity index 94%
rename from
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/AmoroDynamicTableFactory.java
rename to
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnifiedDynamicTableFactory.java
index b929d1ddd..203a823f7 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/AmoroDynamicTableFactory.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnifiedDynamicTableFactory.java
@@ -39,15 +39,15 @@ import java.util.Optional;
import java.util.Set;
/**
- * AmoroDynamicTableFactory is a factory for creating dynamic table sources
and sinks. It implements
- * both DynamicTableSourceFactory and DynamicTableSinkFactory interfaces.
+ * UnifiedDynamicTableFactory is a factory for creating dynamic table sources
and sinks. It
+ * implements both DynamicTableSourceFactory and DynamicTableSinkFactory
interfaces.
*/
-public class AmoroDynamicTableFactory
+public class UnifiedDynamicTableFactory
implements DynamicTableSourceFactory, DynamicTableSinkFactory {
private final Map<TableFormat, AbstractCatalog> availableCatalogs;
- public AmoroDynamicTableFactory(Map<TableFormat, AbstractCatalog>
availableCatalogs) {
+ public UnifiedDynamicTableFactory(Map<TableFormat, AbstractCatalog>
availableCatalogs) {
this.availableCatalogs =
Preconditions.checkNotNull(availableCatalogs, "availableCatalogs
cannot be null");
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
index 0572d04b7..3125d8dfa 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -20,4 +20,4 @@
org.apache.amoro.flink.catalog.factories.mixed.MixedIcebergCatalogFactory
org.apache.amoro.flink.catalog.factories.mixed.MixedHiveCatalogFactory
org.apache.amoro.flink.catalog.factories.mixed.MixedCatalogFactory
org.apache.amoro.flink.catalog.factories.FlinkUnifiedCatalogFactory
-org.apache.amoro.flink.table.DynamicTableFactory
+org.apache.amoro.flink.table.MixedDynamicTableFactory
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
index 3c415a4cd..b9195a2e5 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
@@ -34,7 +34,7 @@ import org.apache.amoro.TableTestHelper;
import org.apache.amoro.catalog.BasicCatalogTestHelper;
import org.apache.amoro.catalog.CatalogTestBase;
import org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions;
-import org.apache.amoro.flink.table.DynamicTableFactory;
+import org.apache.amoro.flink.table.MixedDynamicTableFactory;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.apache.amoro.table.MixedTable;
@@ -406,11 +406,11 @@ public class TestMixedCatalog extends CatalogTestBase {
// create Table with compute columns under default catalog
props = Maps.newHashMap();
- props.put("connector", DynamicTableFactory.IDENTIFIER);
+ props.put("connector", MixedDynamicTableFactory.IDENTIFIER);
props.put(CatalogFactoryOptions.METASTORE_URL.key(), getCatalogUrl());
- props.put(DynamicTableFactory.IDENTIFIER + ".catalog", catalogName);
- props.put(DynamicTableFactory.IDENTIFIER + ".database", DB);
- props.put(DynamicTableFactory.IDENTIFIER + ".table", TABLE);
+ props.put(MixedDynamicTableFactory.IDENTIFIER + ".catalog", catalogName);
+ props.put(MixedDynamicTableFactory.IDENTIFIER + ".database", DB);
+ props.put(MixedDynamicTableFactory.IDENTIFIER + ".table", TABLE);
sql(
"CREATE TABLE default_catalog.default_database."
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
index 9911642b8..2a8b595d8 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -17,4 +17,4 @@
#
org.apache.amoro.flink.catalog.factories.mixed.MixedCatalogFactory
-org.apache.amoro.flink.table.DynamicTableFactory
\ No newline at end of file
+org.apache.amoro.flink.table.MixedDynamicTableFactory
\ No newline at end of file
diff --git a/docs/engines/flink/flink-ddl.md b/docs/engines/flink/flink-ddl.md
index f7bd199cd..3978e417a 100644
--- a/docs/engines/flink/flink-ddl.md
+++ b/docs/engines/flink/flink-ddl.md
@@ -40,12 +40,13 @@ CREATE CATALOG <catalog_name> WITH (
Where `<catalog_name>` is the user-defined name of the Flink catalog, and
`<config_key>`=`<config_value>` has the following configurations:
-| Key | Default Value | Type | Required |
Description
|
-|----------------------------------|---------------|---------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| type | N/A | String | Yes |
Catalog type, validate values are mixed_iceberg and mixed_hive
|
-| metastore.url | (none) | String | Yes | The
URL for Amoro Metastore is
thrift://`<ip>`:`<port>`/`<catalog_name_in_metastore>`.<br>If high availability
is enabled for AMS, it can also be specified in the form of
zookeeper://{zookeeper-server}/{cluster-name}/{catalog-name}. |
-| default-database<img width=100/> | default | String | No | The
default database to use
|
-| property-version | 1 | Integer | No |
Catalog properties version, this option is for future backward compatibility
|
+| Key | Default Value | Type | Required | Description
|
+|------------------|---------------|---------|----------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| type | N/A | String | Yes | Catalog type,
validate values are mixed_iceberg and mixed_hive
|
+| metastore.url | (none) | String | No | The URL for Amoro
Metastore is thrift://`<ip>`:`<port>`/`<catalog_name_in_metastore>`.<br>If high
availability is enabled for AMS, it can also be specified in the form of
zookeeper://{zookeeper-server}/{cluster-name}/{catalog-name}. |
+| default-database | default | String | No | The default database
to use
|
+| property-version | 1 | Integer | No | Catalog properties
version, this option is for future backward compatibility
|
+| catalog-type | N/A | String | No | Metastore type of
the catalog, validate values are hadoop, hive, rest, custom
|
The authentication information of AMS catalog can upload configuration files
on AMS website,
or specify the authentication information and configuration file paths when
creating catalogs with Flink DDL
