This is an automated email from the ASF dual-hosted git repository.
baiyangtx pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 922b45767 [AMORO-1794] Add cache catalog config for iceberg catalog
(#1795)
922b45767 is described below
commit 922b457672c9da99f5977a64c6435abfeb4dc6d6
Author: Qishang Zhong <[email protected]>
AuthorDate: Sun Apr 7 22:24:06 2024 +0800
[AMORO-1794] Add cache catalog config for iceberg catalog (#1795)
* [AMORO-1794] Add cache config for iceberg catalog
* Fix conflict
---
.../arctic/formats/iceberg/IcebergCatalog.java | 39 +++++++++++---------
.../arctic/mixed/BasicMixedIcebergCatalog.java | 30 ++--------------
.../netease/arctic/utils/ArcticCatalogUtil.java | 41 ++++++++++++++++++++++
3 files changed, 65 insertions(+), 45 deletions(-)
diff --git
a/core/src/main/java/com/netease/arctic/formats/iceberg/IcebergCatalog.java
b/core/src/main/java/com/netease/arctic/formats/iceberg/IcebergCatalog.java
index 0ccca71e4..ccfc17464 100644
--- a/core/src/main/java/com/netease/arctic/formats/iceberg/IcebergCatalog.java
+++ b/core/src/main/java/com/netease/arctic/formats/iceberg/IcebergCatalog.java
@@ -21,13 +21,13 @@ package com.netease.arctic.formats.iceberg;
import com.netease.arctic.AmoroTable;
import com.netease.arctic.FormatCatalog;
import com.netease.arctic.table.TableMetaStore;
+import com.netease.arctic.utils.ArcticCatalogUtil;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.NoSuchTableException;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import java.util.List;
import java.util.Map;
@@ -35,24 +35,25 @@ import java.util.stream.Collectors;
public class IcebergCatalog implements FormatCatalog {
+ private SupportsNamespaces asNamespaceCatalog;
private final Catalog icebergCatalog;
private final TableMetaStore metaStore;
private final Map<String, String> properties;
- public IcebergCatalog(
- Catalog icebergCatalog, Map<String, String> properties, TableMetaStore
metaStore) {
- this.icebergCatalog = icebergCatalog;
+ public IcebergCatalog(Catalog catalog, Map<String, String> properties,
TableMetaStore metaStore) {
+ this.icebergCatalog = ArcticCatalogUtil.buildCacheCatalog(catalog,
properties);
+ if (catalog instanceof SupportsNamespaces) {
+ this.asNamespaceCatalog = (SupportsNamespaces) catalog;
+ }
this.metaStore = metaStore;
this.properties = properties;
}
@Override
public List<String> listDatabases() {
- if (icebergCatalog instanceof SupportsNamespaces) {
- return ((SupportsNamespaces) icebergCatalog)
- .listNamespaces().stream().map(ns ->
ns.level(0)).collect(Collectors.toList());
- }
- return Lists.newArrayList();
+ return asNamespaceCatalog().listNamespaces().stream()
+ .map(ns -> ns.level(0))
+ .collect(Collectors.toList());
}
@Override
@@ -68,18 +69,12 @@ public class IcebergCatalog implements FormatCatalog {
@Override
public void createDatabase(String database) {
- if (icebergCatalog instanceof SupportsNamespaces) {
- Namespace ns = Namespace.of(database);
- ((SupportsNamespaces) icebergCatalog).createNamespace(ns);
- }
+ asNamespaceCatalog().createNamespace(Namespace.of(database));
}
@Override
public void dropDatabase(String database) {
- Namespace ns = Namespace.of(database);
- if (icebergCatalog instanceof SupportsNamespaces) {
- ((SupportsNamespaces) icebergCatalog).dropNamespace(ns);
- }
+ asNamespaceCatalog().dropNamespace(Namespace.of(database));
}
@Override
@@ -103,6 +98,16 @@ public class IcebergCatalog implements FormatCatalog {
}
}
+ private SupportsNamespaces asNamespaceCatalog() {
+ if (asNamespaceCatalog == null) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Iceberg catalog: %s doesn't implement SupportsNamespaces",
+ icebergCatalog.getClass().getName()));
+ }
+ return asNamespaceCatalog;
+ }
+
@Override
public boolean dropTable(String database, String table, boolean purge) {
return icebergCatalog.dropTable(TableIdentifier.of(database, table),
purge);
diff --git
a/core/src/main/java/com/netease/arctic/mixed/BasicMixedIcebergCatalog.java
b/core/src/main/java/com/netease/arctic/mixed/BasicMixedIcebergCatalog.java
index a7d09dba3..1fd9c5d8d 100644
--- a/core/src/main/java/com/netease/arctic/mixed/BasicMixedIcebergCatalog.java
+++ b/core/src/main/java/com/netease/arctic/mixed/BasicMixedIcebergCatalog.java
@@ -33,10 +33,9 @@ import com.netease.arctic.table.TableMetaStore;
import com.netease.arctic.table.TableProperties;
import com.netease.arctic.table.blocker.BasicTableBlockerManager;
import com.netease.arctic.table.blocker.TableBlockerManager;
+import com.netease.arctic.utils.ArcticCatalogUtil;
import com.netease.arctic.utils.TablePropertyUtil;
import org.apache.hadoop.conf.Configuration;
-import org.apache.iceberg.CachingCatalog;
-import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
@@ -49,7 +48,6 @@ import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.util.PropertyUtil;
import java.util.List;
import java.util.Map;
@@ -75,27 +73,6 @@ public class BasicMixedIcebergCatalog implements
ArcticCatalog {
@Override
public void initialize(String name, Map<String, String> properties,
TableMetaStore metaStore) {
- boolean cacheEnabled =
- PropertyUtil.propertyAsBoolean(
- properties, CatalogProperties.CACHE_ENABLED,
CatalogProperties.CACHE_ENABLED_DEFAULT);
-
- boolean cacheCaseSensitive =
- PropertyUtil.propertyAsBoolean(
- properties,
- CatalogProperties.CACHE_CASE_SENSITIVE,
- CatalogProperties.CACHE_CASE_SENSITIVE_DEFAULT);
-
- long cacheExpirationIntervalMs =
- PropertyUtil.propertyAsLong(
- properties,
- CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS,
- CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS_DEFAULT);
-
- // An expiration interval of 0ms effectively disables caching.
- // Do not wrap with CachingCatalog.
- if (cacheExpirationIntervalMs == 0) {
- cacheEnabled = false;
- }
Pattern databaseFilterPattern = null;
if (properties.containsKey(CatalogMetaProperties.KEY_DATABASE_FILTER)) {
String databaseFilter =
properties.get(CatalogMetaProperties.KEY_DATABASE_FILTER);
@@ -104,10 +81,7 @@ public class BasicMixedIcebergCatalog implements
ArcticCatalog {
Catalog catalog = buildIcebergCatalog(name, properties,
metaStore.getConfiguration());
this.name = name;
this.tableMetaStore = metaStore;
- this.icebergCatalog =
- cacheEnabled
- ? CachingCatalog.wrap(catalog, cacheCaseSensitive,
cacheExpirationIntervalMs)
- : catalog;
+ this.icebergCatalog = ArcticCatalogUtil.buildCacheCatalog(catalog,
properties);
if (catalog instanceof SupportsNamespaces) {
this.asNamespaceCatalog = (SupportsNamespaces) catalog;
}
diff --git a/core/src/main/java/com/netease/arctic/utils/ArcticCatalogUtil.java
b/core/src/main/java/com/netease/arctic/utils/ArcticCatalogUtil.java
index f1cafc52b..7e77c2391 100644
--- a/core/src/main/java/com/netease/arctic/utils/ArcticCatalogUtil.java
+++ b/core/src/main/java/com/netease/arctic/utils/ArcticCatalogUtil.java
@@ -32,9 +32,11 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.CachingCatalog;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.Table;
import org.apache.iceberg.aws.glue.GlueCatalog;
+import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.hadoop.HadoopTableOperations;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -321,4 +323,43 @@ public class ArcticCatalogUtil {
toProperties.put(toKey, (T) fromProperties.get(fromKey));
}
}
+
+ /**
+ * Build cache catalog.
+ *
+ * @param catalog The catalog of the wrap that needs to be cached
+ * @param properties table properties
+ * @return If Cache is enabled, CachingCatalog is returned, otherwise, the
input catalog is
+ * returned
+ */
+ public static Catalog buildCacheCatalog(Catalog catalog, Map<String, String>
properties) {
+ boolean cacheEnabled =
+ PropertyUtil.propertyAsBoolean(
+ properties, CatalogProperties.CACHE_ENABLED,
CatalogProperties.CACHE_ENABLED_DEFAULT);
+
+ boolean cacheCaseSensitive =
+ PropertyUtil.propertyAsBoolean(
+ properties,
+ CatalogProperties.CACHE_CASE_SENSITIVE,
+ CatalogProperties.CACHE_CASE_SENSITIVE_DEFAULT);
+
+ long cacheExpirationIntervalMs =
+ PropertyUtil.propertyAsLong(
+ properties,
+ CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS,
+ CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS_DEFAULT);
+
+ // An expiration interval of 0ms effectively disables caching.
+ // Do not wrap with CachingCatalog.
+ if (cacheExpirationIntervalMs <= 0) {
+ LOG.warn(
+ "Configuration `{}` is {}, less than or equal to 0, then the cache
will not take effect.",
+ CatalogProperties.CACHE_EXPIRATION_INTERVAL_MS,
+ cacheExpirationIntervalMs);
+ cacheEnabled = false;
+ }
+ return cacheEnabled
+ ? CachingCatalog.wrap(catalog, cacheCaseSensitive,
cacheExpirationIntervalMs)
+ : catalog;
+ }
}
