You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2024/11/28 14:59:48 UTC
(iceberg) branch main updated: Core: Propagate custom metrics reporter when table is created/replaced through Transaction (#11671)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new a95943e556 Core: Propagate custom metrics reporter when table is created/replaced through Transaction (#11671)
a95943e556 is described below
commit a95943e5561c78c18062852e7f8027a191562e08
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Thu Nov 28 15:59:42 2024 +0100
Core: Propagate custom metrics reporter when table is created/replaced through Transaction (#11671)
---
.../org/apache/iceberg/BaseMetastoreCatalog.java | 9 ++-
.../main/java/org/apache/iceberg/Transactions.java | 6 ++
.../apache/iceberg/inmemory/InMemoryCatalog.java | 7 ++
.../org/apache/iceberg/catalog/CatalogTests.java | 90 ++++++++++++++++++++++
.../iceberg/inmemory/TestInMemoryCatalog.java | 12 ++-
.../org/apache/iceberg/jdbc/TestJdbcCatalog.java | 50 +-----------
.../iceberg/jdbc/TestJdbcCatalogWithV1Schema.java | 32 +++++---
.../org/apache/iceberg/rest/TestRESTCatalog.java | 83 ++++----------------
.../org/apache/iceberg/hive/TestHiveCatalog.java | 28 ++++---
.../apache/iceberg/nessie/TestNessieCatalog.java | 18 ++++-
.../rest/RESTCompatibilityKitCatalogTests.java | 6 ++
11 files changed, 198 insertions(+), 143 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
index e794b3121d..e960fe2b63 100644
--- a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java
@@ -217,7 +217,8 @@ public abstract class BaseMetastoreCatalog implements Catalog, Closeable {
tableProperties.putAll(tableOverrideProperties());
TableMetadata metadata =
TableMetadata.newTableMetadata(schema, spec, sortOrder, baseLocation, tableProperties);
- return Transactions.createTableTransaction(identifier.toString(), ops, metadata);
+ return Transactions.createTableTransaction(
+ identifier.toString(), ops, metadata, metricsReporter());
}
@Override
@@ -249,9 +250,11 @@ public abstract class BaseMetastoreCatalog implements Catalog, Closeable {
}
if (orCreate) {
- return Transactions.createOrReplaceTableTransaction(identifier.toString(), ops, metadata);
+ return Transactions.createOrReplaceTableTransaction(
+ identifier.toString(), ops, metadata, metricsReporter());
} else {
- return Transactions.replaceTableTransaction(identifier.toString(), ops, metadata);
+ return Transactions.replaceTableTransaction(
+ identifier.toString(), ops, metadata, metricsReporter());
}
}
diff --git a/core/src/main/java/org/apache/iceberg/Transactions.java b/core/src/main/java/org/apache/iceberg/Transactions.java
index 7afed0573a..a8ea40a6b9 100644
--- a/core/src/main/java/org/apache/iceberg/Transactions.java
+++ b/core/src/main/java/org/apache/iceberg/Transactions.java
@@ -30,6 +30,12 @@ public final class Transactions {
return new BaseTransaction(tableName, ops, TransactionType.CREATE_OR_REPLACE_TABLE, start);
}
+ public static Transaction createOrReplaceTableTransaction(
+ String tableName, TableOperations ops, TableMetadata start, MetricsReporter reporter) {
+ return new BaseTransaction(
+ tableName, ops, TransactionType.CREATE_OR_REPLACE_TABLE, start, reporter);
+ }
+
public static Transaction replaceTableTransaction(
String tableName, TableOperations ops, TableMetadata start) {
return new BaseTransaction(tableName, ops, TransactionType.REPLACE_TABLE, start);
diff --git a/core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java b/core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java
index a880f94f43..ff71bde71f 100644
--- a/core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java
+++ b/core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java
@@ -70,6 +70,7 @@ public class InMemoryCatalog extends BaseMetastoreViewCatalog
private String catalogName;
private String warehouseLocation;
private CloseableGroup closeableGroup;
+ private Map<String, String> catalogProperties;
public InMemoryCatalog() {
this.namespaces = Maps.newConcurrentMap();
@@ -85,6 +86,7 @@ public class InMemoryCatalog extends BaseMetastoreViewCatalog
@Override
public void initialize(String name, Map<String, String> properties) {
this.catalogName = name != null ? name : InMemoryCatalog.class.getSimpleName();
+ this.catalogProperties = ImmutableMap.copyOf(properties);
String warehouse = properties.getOrDefault(CatalogProperties.WAREHOUSE_LOCATION, "");
this.warehouseLocation = warehouse.replaceAll("/*$", "");
@@ -368,6 +370,11 @@ public class InMemoryCatalog extends BaseMetastoreViewCatalog
}
}
+ @Override
+ protected Map<String, String> properties() {
+ return catalogProperties == null ? ImmutableMap.of() : catalogProperties;
+ }
+
private class InMemoryTableOperations extends BaseMetastoreTableOperations {
private final FileIO fileIO;
private final TableIdentifier tableIdentifier;
diff --git a/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java b/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
index a011578865..4df91a4903 100644
--- a/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
+++ b/core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java
@@ -30,11 +30,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.FilesTable;
import org.apache.iceberg.HasTableOperations;
@@ -56,6 +59,10 @@ import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.metrics.CommitReport;
+import org.apache.iceberg.metrics.MetricsReport;
+import org.apache.iceberg.metrics.MetricsReporter;
+import org.apache.iceberg.metrics.ScanReport;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -144,6 +151,8 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
protected abstract C catalog();
+ protected abstract C initCatalog(String catalogName, Map<String, String> additionalProperties);
+
protected boolean supportsNamespaceProperties() {
return true;
}
@@ -2695,6 +2704,87 @@ public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
assertThat(catalog.dropTable(identifier)).isTrue();
}
+ @Test
+ public void testCatalogWithCustomMetricsReporter() throws IOException {
+ C catalogWithCustomReporter =
+ initCatalog(
+ "catalog_with_custom_reporter",
+ ImmutableMap.of(
+ CatalogProperties.METRICS_REPORTER_IMPL, CustomMetricsReporter.class.getName()));
+
+ if (requiresNamespaceCreate()) {
+ catalogWithCustomReporter.createNamespace(TABLE.namespace());
+ }
+
+ catalogWithCustomReporter.buildTable(TABLE, SCHEMA).create();
+
+ Table table = catalogWithCustomReporter.loadTable(TABLE);
+ DataFile dataFile =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath(FileFormat.PARQUET.addExtension(UUID.randomUUID().toString()))
+ .withFileSizeInBytes(10)
+ .withRecordCount(2)
+ .build();
+
+ // append file through FastAppend and check and reset counter
+ table.newFastAppend().appendFile(dataFile).commit();
+ assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
+ CustomMetricsReporter.COMMIT_COUNTER.set(0);
+
+ TableIdentifier identifier = TableIdentifier.of(NS, "custom_metrics_reporter_table");
+ // append file through createTransaction() and check and reset counter
+ catalogWithCustomReporter
+ .buildTable(identifier, SCHEMA)
+ .createTransaction()
+ .newFastAppend()
+ .appendFile(dataFile)
+ .commit();
+ assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
+ CustomMetricsReporter.COMMIT_COUNTER.set(0);
+
+ // append file through createOrReplaceTransaction() and check and reset counter
+ catalogWithCustomReporter
+ .buildTable(identifier, SCHEMA)
+ .createOrReplaceTransaction()
+ .newFastAppend()
+ .appendFile(dataFile)
+ .commit();
+ assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
+ CustomMetricsReporter.COMMIT_COUNTER.set(0);
+
+ // append file through replaceTransaction() and check and reset counter
+ catalogWithCustomReporter
+ .buildTable(TABLE, SCHEMA)
+ .replaceTransaction()
+ .newFastAppend()
+ .appendFile(dataFile)
+ .commit();
+ assertThat(CustomMetricsReporter.COMMIT_COUNTER.get()).isEqualTo(1);
+ CustomMetricsReporter.COMMIT_COUNTER.set(0);
+
+ try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+ assertThat(tasks.iterator()).hasNext();
+ }
+
+ assertThat(CustomMetricsReporter.SCAN_COUNTER.get()).isEqualTo(1);
+ // reset counter in case subclasses run this test multiple times
+ CustomMetricsReporter.SCAN_COUNTER.set(0);
+ }
+
+ public static class CustomMetricsReporter implements MetricsReporter {
+ static final AtomicInteger SCAN_COUNTER = new AtomicInteger(0);
+ static final AtomicInteger COMMIT_COUNTER = new AtomicInteger(0);
+
+ @Override
+ public void report(MetricsReport report) {
+ if (report instanceof ScanReport) {
+ SCAN_COUNTER.incrementAndGet();
+ } else if (report instanceof CommitReport) {
+ COMMIT_COUNTER.incrementAndGet();
+ }
+ }
+ }
+
private static void assertEmpty(String context, Catalog catalog, Namespace ns) {
try {
assertThat(catalog.listTables(ns)).as(context).isEmpty();
diff --git a/core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryCatalog.java b/core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryCatalog.java
index 63cd24b4e2..2c8650d635 100644
--- a/core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/inmemory/TestInMemoryCatalog.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.inmemory;
+import java.util.Map;
import org.apache.iceberg.catalog.CatalogTests;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.BeforeEach;
@@ -27,8 +28,7 @@ public class TestInMemoryCatalog extends CatalogTests<InMemoryCatalog> {
@BeforeEach
public void before() {
- this.catalog = new InMemoryCatalog();
- this.catalog.initialize("in-memory-catalog", ImmutableMap.of());
+ this.catalog = initCatalog("in-memory-catalog", ImmutableMap.of());
}
@Override
@@ -36,6 +36,14 @@ public class TestInMemoryCatalog extends CatalogTests<InMemoryCatalog> {
return catalog;
}
+ @Override
+ protected InMemoryCatalog initCatalog(
+ String catalogName, Map<String, String> additionalProperties) {
+ InMemoryCatalog cat = new InMemoryCatalog();
+ cat.initialize(catalogName, additionalProperties);
+ return cat;
+ }
+
@Override
protected boolean requiresNamespaceCreate() {
return true;
diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
index d21605cace..2d4eb2f157 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java
@@ -39,7 +39,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.hadoop.conf.Configuration;
@@ -50,8 +49,6 @@ import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
@@ -68,9 +65,6 @@ import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.hadoop.Util;
-import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.metrics.MetricsReport;
-import org.apache.iceberg.metrics.MetricsReporter;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -139,7 +133,8 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
catalog = initCatalog("test_jdbc_catalog", Maps.newHashMap());
}
- private JdbcCatalog initCatalog(String catalogName, Map<String, String> props) {
+ @Override
+ protected JdbcCatalog initCatalog(String catalogName, Map<String, String> additionalProperties) {
Map<String, String> properties = Maps.newHashMap();
properties.put(
CatalogProperties.URI,
@@ -150,7 +145,7 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
warehouseLocation = this.tableDir.toAbsolutePath().toString();
properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation);
properties.put("type", "jdbc");
- properties.putAll(props);
+ properties.putAll(additionalProperties);
return (JdbcCatalog) CatalogUtil.buildIcebergCatalog(catalogName, properties, conf);
}
@@ -1059,36 +1054,6 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
assertThat(JdbcUtil.stringToNamespace(nsString)).isEqualTo(ns);
}
- @Test
- public void testCatalogWithCustomMetricsReporter() throws IOException {
- JdbcCatalog catalogWithCustomReporter =
- initCatalog(
- "test_jdbc_catalog_with_custom_reporter",
- ImmutableMap.of(
- CatalogProperties.METRICS_REPORTER_IMPL, CustomMetricsReporter.class.getName()));
- try {
- catalogWithCustomReporter.buildTable(TABLE, SCHEMA).create();
- Table table = catalogWithCustomReporter.loadTable(TABLE);
- table
- .newFastAppend()
- .appendFile(
- DataFiles.builder(PartitionSpec.unpartitioned())
- .withPath(FileFormat.PARQUET.addExtension(UUID.randomUUID().toString()))
- .withFileSizeInBytes(10)
- .withRecordCount(2)
- .build())
- .commit();
- try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
- assertThat(tasks.iterator()).hasNext();
- }
- } finally {
- catalogWithCustomReporter.dropTable(TABLE);
- }
- // counter of custom metrics reporter should have been increased
- // 1x for commit metrics / 1x for scan metrics
- assertThat(CustomMetricsReporter.COUNTER.get()).isEqualTo(2);
- }
-
@Test
public void testCommitExceptionWithoutMessage() {
TableIdentifier tableIdent = TableIdentifier.of("db", "tbl");
@@ -1129,15 +1094,6 @@ public class TestJdbcCatalog extends CatalogTests<JdbcCatalog> {
}
}
- public static class CustomMetricsReporter implements MetricsReporter {
- static final AtomicInteger COUNTER = new AtomicInteger(0);
-
- @Override
- public void report(MetricsReport report) {
- COUNTER.incrementAndGet();
- }
- }
-
private String createMetadataLocationViaJdbcCatalog(TableIdentifier identifier)
throws SQLException {
// temporary connection just to actually create a concrete metadata location
diff --git a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalogWithV1Schema.java b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalogWithV1Schema.java
index b47c216ffc..7586d880c1 100644
--- a/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalogWithV1Schema.java
+++ b/core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalogWithV1Schema.java
@@ -23,6 +23,7 @@ import java.util.UUID;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.catalog.CatalogTests;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.io.TempDir;
@@ -38,6 +39,24 @@ public class TestJdbcCatalogWithV1Schema extends CatalogTests<JdbcCatalog> {
return catalog;
}
+ @Override
+ protected JdbcCatalog initCatalog(String catalogName, Map<String, String> additionalProperties) {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(
+ CatalogProperties.URI,
+ "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
+ properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user");
+ properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
+ properties.put(CatalogProperties.WAREHOUSE_LOCATION, tableDir.toAbsolutePath().toString());
+ properties.put(JdbcUtil.SCHEMA_VERSION_PROPERTY, JdbcUtil.SchemaVersion.V1.name());
+ properties.putAll(additionalProperties);
+
+ JdbcCatalog cat = new JdbcCatalog();
+ cat.setConf(new Configuration());
+ cat.initialize(catalogName, properties);
+ return cat;
+ }
+
@Override
protected boolean supportsNamespaceProperties() {
return true;
@@ -50,17 +69,6 @@ public class TestJdbcCatalogWithV1Schema extends CatalogTests<JdbcCatalog> {
@BeforeEach
public void setupCatalog() {
- Map<String, String> properties = Maps.newHashMap();
- properties.put(
- CatalogProperties.URI,
- "jdbc:sqlite:file::memory:?ic" + UUID.randomUUID().toString().replace("-", ""));
- properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user");
- properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password");
- properties.put(CatalogProperties.WAREHOUSE_LOCATION, tableDir.toAbsolutePath().toString());
- properties.put(JdbcUtil.SCHEMA_VERSION_PROPERTY, JdbcUtil.SchemaVersion.V1.name());
-
- catalog = new JdbcCatalog();
- catalog.setConf(new Configuration());
- catalog.initialize("testCatalog", properties);
+ this.catalog = initCatalog("testCatalog", ImmutableMap.of());
}
}
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
index 06008761ea..232cfd31d1 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java
@@ -36,14 +36,12 @@ import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.BaseTransaction;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
-import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -64,9 +62,6 @@ import org.apache.iceberg.exceptions.NotFoundException;
import org.apache.iceberg.exceptions.ServiceFailureException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.inmemory.InMemoryCatalog;
-import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.metrics.MetricsReport;
-import org.apache.iceberg.metrics.MetricsReporter;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -114,7 +109,6 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
@BeforeEach
public void createCatalog() throws Exception {
File warehouse = temp.toFile();
- Configuration conf = new Configuration();
this.backendCatalog = new InMemoryCatalog();
this.backendCatalog.initialize(
@@ -164,6 +158,12 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
httpServer.setHandler(servletContext);
httpServer.start();
+ this.restCatalog = initCatalog("prod", ImmutableMap.of());
+ }
+
+ @Override
+ protected RESTCatalog initCatalog(String catalogName, Map<String, String> additionalProperties) {
+ Configuration conf = new Configuration();
SessionCatalog.SessionContext context =
new SessionCatalog.SessionContext(
UUID.randomUUID().toString(),
@@ -171,20 +171,26 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
ImmutableMap.of("credential", "user:12345"),
ImmutableMap.of());
- this.restCatalog =
+ RESTCatalog catalog =
new RESTCatalog(
context,
(config) -> HTTPClient.builder(config).uri(config.get(CatalogProperties.URI)).build());
- restCatalog.setConf(conf);
- restCatalog.initialize(
- "prod",
+ catalog.setConf(conf);
+ Map<String, String> properties =
ImmutableMap.of(
CatalogProperties.URI,
httpServer.getURI().toString(),
CatalogProperties.FILE_IO_IMPL,
"org.apache.iceberg.inmemory.InMemoryFileIO",
"credential",
- "catalog:12345"));
+ "catalog:12345");
+ catalog.initialize(
+ catalogName,
+ ImmutableMap.<String, String>builder()
+ .putAll(properties)
+ .putAll(additionalProperties)
+ .build());
+ return catalog;
}
@SuppressWarnings("unchecked")
@@ -1623,61 +1629,6 @@ public class TestRESTCatalog extends CatalogTests<RESTCatalog> {
});
}
- @Test
- public void testCatalogWithCustomMetricsReporter() throws IOException {
- this.restCatalog =
- new RESTCatalog(
- new SessionCatalog.SessionContext(
- UUID.randomUUID().toString(),
- "user",
- ImmutableMap.of("credential", "user:12345"),
- ImmutableMap.of()),
- (config) -> HTTPClient.builder(config).uri(config.get(CatalogProperties.URI)).build());
- restCatalog.setConf(new Configuration());
- restCatalog.initialize(
- "prod",
- ImmutableMap.of(
- CatalogProperties.URI,
- httpServer.getURI().toString(),
- "credential",
- "catalog:12345",
- CatalogProperties.METRICS_REPORTER_IMPL,
- CustomMetricsReporter.class.getName()));
-
- if (requiresNamespaceCreate()) {
- restCatalog.createNamespace(TABLE.namespace());
- }
-
- restCatalog.buildTable(TABLE, SCHEMA).create();
- Table table = restCatalog.loadTable(TABLE);
- table
- .newFastAppend()
- .appendFile(
- DataFiles.builder(PartitionSpec.unpartitioned())
- .withPath("/path/to/data-a.parquet")
- .withFileSizeInBytes(10)
- .withRecordCount(2)
- .build())
- .commit();
-
- try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
- assertThat(tasks.iterator()).hasNext();
- }
-
- // counter of custom metrics reporter should have been increased
- // 1x for commit metrics / 1x for scan metrics
- assertThat(CustomMetricsReporter.COUNTER.get()).isEqualTo(2);
- }
-
- public static class CustomMetricsReporter implements MetricsReporter {
- static final AtomicInteger COUNTER = new AtomicInteger(0);
-
- @Override
- public void report(MetricsReport report) {
- COUNTER.incrementAndGet();
- }
- }
-
@Test
public void testCatalogExpiredBearerTokenRefreshWithoutCredential() {
// expires at epoch second = 1
diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
index 7d0eb641a3..709bb1caaa 100644
--- a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
+++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCatalog.java
@@ -110,20 +110,30 @@ public class TestHiveCatalog extends CatalogTests<HiveCatalog> {
@BeforeEach
public void before() throws TException {
- catalog =
- (HiveCatalog)
- CatalogUtil.loadCatalog(
- HiveCatalog.class.getName(),
- CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE,
- ImmutableMap.of(
- CatalogProperties.CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS,
- String.valueOf(TimeUnit.SECONDS.toMillis(10))),
- HIVE_METASTORE_EXTENSION.hiveConf());
+ catalog = initCatalog("hive", ImmutableMap.of());
String dbPath = HIVE_METASTORE_EXTENSION.metastore().getDatabasePath(DB_NAME);
Database db = new Database(DB_NAME, "description", dbPath, Maps.newHashMap());
HIVE_METASTORE_EXTENSION.metastoreClient().createDatabase(db);
}
+ @Override
+ protected HiveCatalog initCatalog(String catalogName, Map<String, String> additionalProperties) {
+ Map<String, String> properties =
+ ImmutableMap.of(
+ CatalogProperties.CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS,
+ String.valueOf(TimeUnit.SECONDS.toMillis(10)));
+
+ return (HiveCatalog)
+ CatalogUtil.loadCatalog(
+ HiveCatalog.class.getName(),
+ catalogName,
+ ImmutableMap.<String, String>builder()
+ .putAll(properties)
+ .putAll(additionalProperties)
+ .build(),
+ HIVE_METASTORE_EXTENSION.hiveConf());
+ }
+
@AfterEach
public void cleanup() throws Exception {
HIVE_METASTORE_EXTENSION.metastore().reset();
diff --git a/nessie/src/test/java/org/apache/iceberg/nessie/TestNessieCatalog.java b/nessie/src/test/java/org/apache/iceberg/nessie/TestNessieCatalog.java
index 55be034221..dce8f7ff0f 100644
--- a/nessie/src/test/java/org/apache/iceberg/nessie/TestNessieCatalog.java
+++ b/nessie/src/test/java/org/apache/iceberg/nessie/TestNessieCatalog.java
@@ -78,7 +78,7 @@ public class TestNessieCatalog extends CatalogTests<NessieCatalog> {
initialHashOfDefaultBranch = api.getDefaultBranch().getHash();
uri = nessieUri.toASCIIString();
hadoopConfig = new Configuration();
- catalog = initNessieCatalog("main");
+ catalog = initCatalog("nessie", ImmutableMap.of());
}
@AfterEach
@@ -112,18 +112,28 @@ public class TestNessieCatalog extends CatalogTests<NessieCatalog> {
.assign();
}
- private NessieCatalog initNessieCatalog(String ref) {
+ @Override
+ protected NessieCatalog initCatalog(
+ String catalogName, Map<String, String> additionalProperties) {
Map<String, String> options =
ImmutableMap.of(
"type",
"nessie",
"ref",
- ref,
+ "main",
CatalogProperties.URI,
uri,
CatalogProperties.WAREHOUSE_LOCATION,
temp.toUri().toString());
- return (NessieCatalog) CatalogUtil.buildIcebergCatalog("nessie", options, hadoopConfig);
+
+ return (NessieCatalog)
+ CatalogUtil.buildIcebergCatalog(
+ catalogName,
+ ImmutableMap.<String, String>builder()
+ .putAll(options)
+ .putAll(additionalProperties)
+ .build(),
+ hadoopConfig);
}
@Override
diff --git a/open-api/src/test/java/org/apache/iceberg/rest/RESTCompatibilityKitCatalogTests.java b/open-api/src/test/java/org/apache/iceberg/rest/RESTCompatibilityKitCatalogTests.java
index 4c4860e88a..a709d81434 100644
--- a/open-api/src/test/java/org/apache/iceberg/rest/RESTCompatibilityKitCatalogTests.java
+++ b/open-api/src/test/java/org/apache/iceberg/rest/RESTCompatibilityKitCatalogTests.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.rest;
import static org.assertj.core.api.Assertions.assertThat;
+import java.util.Map;
import org.apache.iceberg.catalog.CatalogTests;
import org.apache.iceberg.util.PropertyUtil;
import org.junit.jupiter.api.AfterAll;
@@ -63,6 +64,11 @@ public class RESTCompatibilityKitCatalogTests extends CatalogTests<RESTCatalog>
return restCatalog;
}
+ @Override
+ protected RESTCatalog initCatalog(String catalogName, Map<String, String> additionalProperties) {
+ return RCKUtils.initCatalogClient(additionalProperties);
+ }
+
@Override
protected boolean requiresNamespaceCreate() {
return PropertyUtil.propertyAsBoolean(