You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by ta...@apache.org on 2024/08/29 17:58:11 UTC
(polaris) branch main updated: Add the ability to intercept File IO by specifying a wrapper class (#155)
This is an automated email from the ASF dual-hosted git repository.
takidau pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new 07c84442 Add the ability to intercept File IO by specifying a wrapper class (#155)
07c84442 is described below
commit 07c8444213f0a212c28770448fa2754d0c230169
Author: Andrew Guterman <an...@gmail.com>
AuthorDate: Thu Aug 29 10:58:07 2024 -0700
Add the ability to intercept File IO by specifying a wrapper class (#155)
* Add FileIOWrapper to allow intercepting File IO
* Clean up issues after squash
* Clean up after rebase
* Update copyrights, other minor fixes from rebase
* Apply spotless
---
.../polaris/core/monitor/MetricRegistryAware.java | 24 ++++
polaris-server.yml | 2 +
.../apache/polaris/service/PolarisApplication.java | 18 ++-
.../service/catalog/BasePolarisCatalog.java | 10 +-
.../service/catalog/DefaultFileIOFactory.java | 34 +++++
.../polaris/service/catalog/FileIOFactory.java | 33 +++++
.../service/config/PolarisApplicationConfig.java | 12 ++
.../context/PolarisCallContextCatalogFactory.java | 14 +-
.../polaris/service/task/TaskFileIOSupplier.java | 10 +-
.../services/io.dropwizard.jackson.Discoverable | 3 +-
...g.apache.polaris.service.catalog.FileIOFactory} | 7 +-
.../service/admin/PolarisAuthzTestBase.java | 11 +-
.../service/catalog/BasePolarisCatalogTest.java | 154 +++++++++++++++------
.../catalog/BasePolarisCatalogViewTest.java | 7 +-
.../service/catalog/MeasuredFileIOFactory.java | 152 ++++++++++++++++++++
.../PolarisCatalogHandlerWrapperAuthzTest.java | 3 +-
.../resources/polaris-server-integrationtest.yml | 3 +
17 files changed, 432 insertions(+), 65 deletions(-)
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/monitor/MetricRegistryAware.java b/polaris-core/src/main/java/org/apache/polaris/core/monitor/MetricRegistryAware.java
new file mode 100644
index 00000000..2977d60a
--- /dev/null
+++ b/polaris-core/src/main/java/org/apache/polaris/core/monitor/MetricRegistryAware.java
@@ -0,0 +1,24 @@
+/*
+ * 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.polaris.core.monitor;
+
+/** Allows setting a configured instance of {@link PolarisMetricRegistry} */
+public interface MetricRegistryAware {
+ void setMetricRegistry(PolarisMetricRegistry metricRegistry);
+}
diff --git a/polaris-server.yml b/polaris-server.yml
index e74d8311..f6dd44ac 100644
--- a/polaris-server.yml
+++ b/polaris-server.yml
@@ -91,6 +91,8 @@ metaStoreManager:
# type: eclipse-link # uncomment to use eclipse-link as metastore
# persistence-unit: polaris
+io:
+ factoryType: default
# TODO - avoid duplicating token broker config
oauth2:
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/PolarisApplication.java b/polaris-service/src/main/java/org/apache/polaris/service/PolarisApplication.java
index 544c9f64..3c73d0a5 100644
--- a/polaris-service/src/main/java/org/apache/polaris/service/PolarisApplication.java
+++ b/polaris-service/src/main/java/org/apache/polaris/service/PolarisApplication.java
@@ -74,6 +74,7 @@ import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
import org.apache.polaris.core.auth.PolarisAuthorizer;
import org.apache.polaris.core.context.CallContext;
import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.monitor.MetricRegistryAware;
import org.apache.polaris.core.monitor.PolarisMetricRegistry;
import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
import org.apache.polaris.service.admin.PolarisServiceImpl;
@@ -81,6 +82,7 @@ import org.apache.polaris.service.admin.api.PolarisCatalogsApi;
import org.apache.polaris.service.admin.api.PolarisPrincipalRolesApi;
import org.apache.polaris.service.admin.api.PolarisPrincipalsApi;
import org.apache.polaris.service.auth.DiscoverableAuthenticator;
+import org.apache.polaris.service.catalog.FileIOFactory;
import org.apache.polaris.service.catalog.IcebergCatalogAdapter;
import org.apache.polaris.service.catalog.api.IcebergRestCatalogApi;
import org.apache.polaris.service.catalog.api.IcebergRestConfigurationApi;
@@ -185,10 +187,22 @@ public class PolarisApplication extends Application<PolarisApplicationConfig> {
"realmContext", new ContextResolverFilter(realmContextResolver, callContextResolver))
.addMappingForUrlPatterns(EnumSet.of(DispatcherType.REQUEST), true, "/*");
+ FileIOFactory fileIOFactory = configuration.getFileIOFactory();
+ if (fileIOFactory instanceof MetricRegistryAware mrAware) {
+ mrAware.setMetricRegistry(polarisMetricRegistry);
+ }
+ if (fileIOFactory instanceof OpenTelemetryAware otAware) {
+ otAware.setOpenTelemetry(openTelemetry);
+ }
+ if (fileIOFactory instanceof ConfigurationStoreAware csAware) {
+ csAware.setConfigurationStore(configurationStore);
+ }
+
TaskHandlerConfiguration taskConfig = configuration.getTaskHandler();
TaskExecutorImpl taskExecutor =
new TaskExecutorImpl(taskConfig.executorService(), metaStoreManagerFactory);
- TaskFileIOSupplier fileIOSupplier = new TaskFileIOSupplier(metaStoreManagerFactory);
+ TaskFileIOSupplier fileIOSupplier =
+ new TaskFileIOSupplier(metaStoreManagerFactory, fileIOFactory);
taskExecutor.addTaskHandler(
new TableCleanupTaskHandler(taskExecutor, metaStoreManagerFactory, fileIOSupplier));
taskExecutor.addTaskHandler(
@@ -199,7 +213,7 @@ public class PolarisApplication extends Application<PolarisApplicationConfig> {
"Initializing PolarisCallContextCatalogFactory for metaStoreManagerType {}",
metaStoreManagerFactory);
CallContextCatalogFactory catalogFactory =
- new PolarisCallContextCatalogFactory(entityManagerFactory, taskExecutor);
+ new PolarisCallContextCatalogFactory(entityManagerFactory, taskExecutor, fileIOFactory);
PolarisAuthorizer authorizer = new PolarisAuthorizer(configurationStore);
IcebergCatalogAdapter catalogAdapter =
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java b/polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java
index 552d6a93..2538caf5 100644
--- a/polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java
+++ b/polaris-service/src/main/java/org/apache/polaris/service/catalog/BasePolarisCatalog.java
@@ -42,11 +42,9 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.exception.ExceptionUtils;
-import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.BaseMetastoreTableOperations;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.CatalogProperties;
-import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
@@ -175,6 +173,7 @@ public class BasePolarisCatalog extends BaseMetastoreViewCatalog
private CloseableGroup closeableGroup;
private Map<String, String> catalogProperties;
private Map<String, String> tableDefaultProperties;
+ private final FileIOFactory fileIOFactory;
/**
* @param entityManager provides handle to underlying PolarisMetaStoreManager with which to
@@ -189,7 +188,8 @@ public class BasePolarisCatalog extends BaseMetastoreViewCatalog
CallContext callContext,
PolarisResolutionManifestCatalogView resolvedEntityView,
AuthenticatedPolarisPrincipal authenticatedPrincipal,
- TaskExecutor taskExecutor) {
+ TaskExecutor taskExecutor,
+ FileIOFactory fileIOFactory) {
this.entityManager = entityManager;
this.callContext = callContext;
this.resolvedEntityView = resolvedEntityView;
@@ -199,6 +199,7 @@ public class BasePolarisCatalog extends BaseMetastoreViewCatalog
this.taskExecutor = taskExecutor;
this.catalogId = catalogEntity.getId();
this.catalogName = catalogEntity.getName();
+ this.fileIOFactory = fileIOFactory;
}
@Override
@@ -1967,8 +1968,7 @@ public class BasePolarisCatalog extends BaseMetastoreViewCatalog
Map<String, String> propertiesWithS3CustomizedClientFactory = new HashMap<>(properties);
propertiesWithS3CustomizedClientFactory.put(
S3FileIOProperties.CLIENT_FACTORY, PolarisS3FileIOClientFactory.class.getName());
- return CatalogUtil.loadFileIO(
- ioImpl, propertiesWithS3CustomizedClientFactory, new Configuration());
+ return fileIOFactory.loadFileIO(ioImpl, propertiesWithS3CustomizedClientFactory);
}
private void blockedUserSpecifiedWriteLocation(Map<String, String> properties) {
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/catalog/DefaultFileIOFactory.java b/polaris-service/src/main/java/org/apache/polaris/service/catalog/DefaultFileIOFactory.java
new file mode 100644
index 00000000..0ff8034b
--- /dev/null
+++ b/polaris-service/src/main/java/org/apache/polaris/service/catalog/DefaultFileIOFactory.java
@@ -0,0 +1,34 @@
+/*
+ * 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.polaris.service.catalog;
+
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.io.FileIO;
+
+/** A simple FileIOFactory implementation that defers all the work to the Iceberg SDK */
+@JsonTypeName("default")
+public class DefaultFileIOFactory implements FileIOFactory {
+ @Override
+ public FileIO loadFileIO(String impl, Map<String, String> properties) {
+ return CatalogUtil.loadFileIO(impl, properties, new Configuration());
+ }
+}
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/catalog/FileIOFactory.java b/polaris-service/src/main/java/org/apache/polaris/service/catalog/FileIOFactory.java
new file mode 100644
index 00000000..173864e1
--- /dev/null
+++ b/polaris-service/src/main/java/org/apache/polaris/service/catalog/FileIOFactory.java
@@ -0,0 +1,33 @@
+/*
+ * 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.polaris.service.catalog;
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import io.dropwizard.jackson.Discoverable;
+import java.util.Map;
+import org.apache.iceberg.io.FileIO;
+
+/** Interface for providing a way to construct FileIO objects, such as for reading/writing S3. */
+@JsonTypeInfo(
+ use = JsonTypeInfo.Id.NAME,
+ include = JsonTypeInfo.As.PROPERTY,
+ property = "factoryType")
+public interface FileIOFactory extends Discoverable {
+ FileIO loadFileIO(String impl, Map<String, String> properties);
+}
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/config/PolarisApplicationConfig.java b/polaris-service/src/main/java/org/apache/polaris/service/config/PolarisApplicationConfig.java
index c205198b..f8b7371f 100644
--- a/polaris-service/src/main/java/org/apache/polaris/service/config/PolarisApplicationConfig.java
+++ b/polaris-service/src/main/java/org/apache/polaris/service/config/PolarisApplicationConfig.java
@@ -28,6 +28,7 @@ import org.apache.polaris.core.PolarisConfigurationStore;
import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
import org.apache.polaris.service.auth.DiscoverableAuthenticator;
+import org.apache.polaris.service.catalog.FileIOFactory;
import org.apache.polaris.service.context.CallContextResolver;
import org.apache.polaris.service.context.RealmContextResolver;
import org.slf4j.LoggerFactory;
@@ -52,6 +53,7 @@ public class PolarisApplicationConfig extends Configuration {
private List<String> defaultRealms;
private String awsAccessKey;
private String awsSecretKey;
+ private FileIOFactory fileIOFactory;
@JsonProperty("metaStoreManager")
public void setMetaStoreManagerFactory(MetaStoreManagerFactory metaStoreManagerFactory) {
@@ -63,6 +65,16 @@ public class PolarisApplicationConfig extends Configuration {
return metaStoreManagerFactory;
}
+ @JsonProperty("io")
+ public void setFileIOFactory(FileIOFactory fileIOFactory) {
+ this.fileIOFactory = fileIOFactory;
+ }
+
+ @JsonProperty("io")
+ public FileIOFactory getFileIOFactory() {
+ return fileIOFactory;
+ }
+
@JsonProperty("authenticator")
public void setPolarisAuthenticator(
DiscoverableAuthenticator<String, AuthenticatedPolarisPrincipal> polarisAuthenticator) {
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/context/PolarisCallContextCatalogFactory.java b/polaris-service/src/main/java/org/apache/polaris/service/context/PolarisCallContextCatalogFactory.java
index f82e7231..4fef6a84 100644
--- a/polaris-service/src/main/java/org/apache/polaris/service/context/PolarisCallContextCatalogFactory.java
+++ b/polaris-service/src/main/java/org/apache/polaris/service/context/PolarisCallContextCatalogFactory.java
@@ -31,6 +31,7 @@ import org.apache.polaris.core.entity.PolarisBaseEntity;
import org.apache.polaris.core.persistence.PolarisEntityManager;
import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
import org.apache.polaris.service.catalog.BasePolarisCatalog;
+import org.apache.polaris.service.catalog.FileIOFactory;
import org.apache.polaris.service.config.RealmEntityManagerFactory;
import org.apache.polaris.service.task.TaskExecutor;
import org.slf4j.Logger;
@@ -45,11 +46,15 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
private final RealmEntityManagerFactory entityManagerFactory;
private final TaskExecutor taskExecutor;
+ private final FileIOFactory fileIOFactory;
public PolarisCallContextCatalogFactory(
- RealmEntityManagerFactory entityManagerFactory, TaskExecutor taskExecutor) {
+ RealmEntityManagerFactory entityManagerFactory,
+ TaskExecutor taskExecutor,
+ FileIOFactory fileIOFactory) {
this.entityManagerFactory = entityManagerFactory;
this.taskExecutor = taskExecutor;
+ this.fileIOFactory = fileIOFactory;
}
@Override
@@ -70,7 +75,12 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
BasePolarisCatalog catalogInstance =
new BasePolarisCatalog(
- entityManager, context, resolvedManifest, authenticatedPrincipal, taskExecutor);
+ entityManager,
+ context,
+ resolvedManifest,
+ authenticatedPrincipal,
+ taskExecutor,
+ fileIOFactory);
context.contextVariables().put(CallContext.REQUEST_PATH_CATALOG_INSTANCE_KEY, catalogInstance);
diff --git a/polaris-service/src/main/java/org/apache/polaris/service/task/TaskFileIOSupplier.java b/polaris-service/src/main/java/org/apache/polaris/service/task/TaskFileIOSupplier.java
index 7b636d56..1d6a367e 100644
--- a/polaris-service/src/main/java/org/apache/polaris/service/task/TaskFileIOSupplier.java
+++ b/polaris-service/src/main/java/org/apache/polaris/service/task/TaskFileIOSupplier.java
@@ -22,21 +22,23 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
-import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.CatalogProperties;
-import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.io.FileIO;
import org.apache.polaris.core.context.CallContext;
import org.apache.polaris.core.entity.PolarisTaskConstants;
import org.apache.polaris.core.entity.TaskEntity;
import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.service.catalog.FileIOFactory;
public class TaskFileIOSupplier implements Function<TaskEntity, FileIO> {
private final MetaStoreManagerFactory metaStoreManagerFactory;
+ private final FileIOFactory fileIOFactory;
- public TaskFileIOSupplier(MetaStoreManagerFactory metaStoreManagerFactory) {
+ public TaskFileIOSupplier(
+ MetaStoreManagerFactory metaStoreManagerFactory, FileIOFactory fileIOFactory) {
this.metaStoreManagerFactory = metaStoreManagerFactory;
+ this.fileIOFactory = fileIOFactory;
}
@Override
@@ -60,6 +62,6 @@ public class TaskFileIOSupplier implements Function<TaskEntity, FileIO> {
String ioImpl =
properties.getOrDefault(
CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.io.ResolvingFileIO");
- return CatalogUtil.loadFileIO(ioImpl, properties, new Configuration());
+ return fileIOFactory.loadFileIO(ioImpl, properties);
}
}
diff --git a/polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable b/polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable
index cf170d64..e290634d 100644
--- a/polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable
+++ b/polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable
@@ -22,4 +22,5 @@ org.apache.polaris.core.persistence.MetaStoreManagerFactory
org.apache.polaris.service.config.OAuth2ApiService
org.apache.polaris.service.context.RealmContextResolver
org.apache.polaris.service.context.CallContextResolver
-org.apache.polaris.service.auth.TokenBrokerFactory
\ No newline at end of file
+org.apache.polaris.service.auth.TokenBrokerFactory
+org.apache.polaris.service.catalog.FileIOFactory
diff --git a/polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable b/polaris-service/src/main/resources/META-INF/services/org.apache.polaris.service.catalog.FileIOFactory
similarity index 70%
copy from polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable
copy to polaris-service/src/main/resources/META-INF/services/org.apache.polaris.service.catalog.FileIOFactory
index cf170d64..e4808ecc 100644
--- a/polaris-service/src/main/resources/META-INF/services/io.dropwizard.jackson.Discoverable
+++ b/polaris-service/src/main/resources/META-INF/services/org.apache.polaris.service.catalog.FileIOFactory
@@ -17,9 +17,4 @@
# under the License.
#
-org.apache.polaris.service.auth.DiscoverableAuthenticator
-org.apache.polaris.core.persistence.MetaStoreManagerFactory
-org.apache.polaris.service.config.OAuth2ApiService
-org.apache.polaris.service.context.RealmContextResolver
-org.apache.polaris.service.context.CallContextResolver
-org.apache.polaris.service.auth.TokenBrokerFactory
\ No newline at end of file
+org.apache.polaris.service.catalog.DefaultFileIOFactory
diff --git a/polaris-service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java b/polaris-service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java
index ad766c4e..4133082a 100644
--- a/polaris-service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java
+++ b/polaris-service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java
@@ -61,6 +61,7 @@ import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
import org.apache.polaris.core.storage.cache.StorageCredentialCache;
import org.apache.polaris.service.catalog.BasePolarisCatalog;
+import org.apache.polaris.service.catalog.DefaultFileIOFactory;
import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
import org.apache.polaris.service.config.DefaultConfigurationStore;
import org.apache.polaris.service.config.RealmEntityManagerFactory;
@@ -353,7 +354,12 @@ public abstract class PolarisAuthzTestBase {
callContext, entityManager, authenticatedRoot, CATALOG_NAME);
this.baseCatalog =
new BasePolarisCatalog(
- entityManager, callContext, passthroughView, authenticatedRoot, Mockito.mock());
+ entityManager,
+ callContext,
+ passthroughView,
+ authenticatedRoot,
+ Mockito.mock(),
+ new DefaultFileIOFactory());
this.baseCatalog.initialize(
CATALOG_NAME,
ImmutableMap.of(
@@ -369,7 +375,8 @@ public abstract class PolarisAuthzTestBase {
return entityManager;
}
},
- Mockito.mock());
+ Mockito.mock(),
+ new DefaultFileIOFactory());
}
@Override
diff --git a/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java b/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java
index 1652c4c0..c574e5f4 100644
--- a/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java
+++ b/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogTest.java
@@ -84,6 +84,7 @@ import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
import org.apache.polaris.core.storage.cache.StorageCredentialCache;
import org.apache.polaris.service.admin.PolarisAdminService;
import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
+import org.apache.polaris.service.task.TableCleanupTaskHandler;
import org.apache.polaris.service.task.TaskExecutor;
import org.apache.polaris.service.task.TaskFileIOSupplier;
import org.apache.polaris.service.types.NotificationRequest;
@@ -204,7 +205,12 @@ public class BasePolarisCatalogTest extends CatalogTests<BasePolarisCatalog> {
TaskExecutor taskExecutor = Mockito.mock();
this.catalog =
new BasePolarisCatalog(
- entityManager, callContext, passthroughView, authenticatedRoot, taskExecutor);
+ entityManager,
+ callContext,
+ passthroughView,
+ authenticatedRoot,
+ taskExecutor,
+ new DefaultFileIOFactory());
this.catalog.initialize(
CATALOG_NAME,
ImmutableMap.of(
@@ -256,6 +262,44 @@ public class BasePolarisCatalogTest extends CatalogTests<BasePolarisCatalog> {
return true;
}
+ private MetaStoreManagerFactory createMockMetaStoreManagerFactory() {
+ return new MetaStoreManagerFactory() {
+ @Override
+ public PolarisMetaStoreManager getOrCreateMetaStoreManager(RealmContext realmContext) {
+ return metaStoreManager;
+ }
+
+ @Override
+ public Supplier<PolarisMetaStoreSession> getOrCreateSessionSupplier(
+ RealmContext realmContext) {
+ return () -> polarisContext.getMetaStore();
+ }
+
+ @Override
+ public StorageCredentialCache getOrCreateStorageCredentialCache(RealmContext realmContext) {
+ return new StorageCredentialCache();
+ }
+
+ @Override
+ public void setMetricRegistry(PolarisMetricRegistry metricRegistry) {}
+
+ @Override
+ public Map<String, PolarisMetaStoreManager.PrincipalSecretsResult> bootstrapRealms(
+ List<String> realms) {
+ throw new NotImplementedException("Bootstrapping realms is not supported");
+ }
+
+ @Override
+ public void purgeRealms(List<String> realms) {
+ throw new NotImplementedException("Purging realms is not supported");
+ }
+
+ @Override
+ public void setStorageIntegrationProvider(
+ PolarisStorageIntegrationProvider storageIntegrationProvider) {}
+ };
+ }
+
@Test
public void testRenameTableMissingDestinationNamespace() {
Assumptions.assumeTrue(
@@ -614,7 +658,12 @@ public class BasePolarisCatalogTest extends CatalogTests<BasePolarisCatalog> {
TaskExecutor taskExecutor = Mockito.mock();
BasePolarisCatalog catalog =
new BasePolarisCatalog(
- entityManager, callContext, passthroughView, authenticatedRoot, taskExecutor);
+ entityManager,
+ callContext,
+ passthroughView,
+ authenticatedRoot,
+ taskExecutor,
+ new DefaultFileIOFactory());
catalog.initialize(
catalogWithoutStorage,
ImmutableMap.of(
@@ -667,7 +716,12 @@ public class BasePolarisCatalogTest extends CatalogTests<BasePolarisCatalog> {
TaskExecutor taskExecutor = Mockito.mock();
BasePolarisCatalog catalog =
new BasePolarisCatalog(
- entityManager, callContext, passthroughView, authenticatedRoot, taskExecutor);
+ entityManager,
+ callContext,
+ passthroughView,
+ authenticatedRoot,
+ taskExecutor,
+ new DefaultFileIOFactory());
catalog.initialize(
catalogName,
ImmutableMap.of(
@@ -1087,44 +1141,7 @@ public class BasePolarisCatalogTest extends CatalogTests<BasePolarisCatalog> {
.containsEntry(PolarisCredentialProperty.AWS_SECRET_KEY, SECRET_ACCESS_KEY)
.containsEntry(PolarisCredentialProperty.AWS_TOKEN, SESSION_TOKEN);
FileIO fileIO =
- new TaskFileIOSupplier(
- new MetaStoreManagerFactory() {
- @Override
- public PolarisMetaStoreManager getOrCreateMetaStoreManager(
- RealmContext realmContext) {
- return metaStoreManager;
- }
-
- @Override
- public Supplier<PolarisMetaStoreSession> getOrCreateSessionSupplier(
- RealmContext realmContext) {
- return () -> polarisContext.getMetaStore();
- }
-
- @Override
- public StorageCredentialCache getOrCreateStorageCredentialCache(
- RealmContext realmContext) {
- return new StorageCredentialCache();
- }
-
- @Override
- public void setMetricRegistry(PolarisMetricRegistry metricRegistry) {}
-
- @Override
- public Map<String, PolarisMetaStoreManager.PrincipalSecretsResult>
- bootstrapRealms(List<String> realms) {
- throw new NotImplementedException("Bootstrapping realms is not supported");
- }
-
- @Override
- public void purgeRealms(List<String> realms) {
- throw new NotImplementedException("Purging realms is not supported");
- }
-
- @Override
- public void setStorageIntegrationProvider(
- PolarisStorageIntegrationProvider storageIntegrationProvider) {}
- })
+ new TaskFileIOSupplier(createMockMetaStoreManagerFactory(), new DefaultFileIOFactory())
.apply(taskEntity);
Assertions.assertThat(fileIO).isNotNull().isInstanceOf(InMemoryFileIO.class);
}
@@ -1168,4 +1185,59 @@ public class BasePolarisCatalogTest extends CatalogTests<BasePolarisCatalog> {
Assertions.assertThat(BasePolarisCatalog.SHOULD_RETRY_REFRESH_PREDICATE.test(otherException))
.isTrue();
}
+
+ @Test
+ public void testFileIOWrapper() {
+ RealmContext realmContext = () -> "realm";
+ CallContext callContext = CallContext.of(realmContext, polarisContext);
+ CallContext.setCurrentContext(callContext);
+ PolarisPassthroughResolutionView passthroughView =
+ new PolarisPassthroughResolutionView(
+ callContext, entityManager, authenticatedRoot, CATALOG_NAME);
+
+ MeasuredFileIOFactory measured = new MeasuredFileIOFactory();
+ BasePolarisCatalog catalog =
+ new BasePolarisCatalog(
+ entityManager,
+ callContext,
+ passthroughView,
+ authenticatedRoot,
+ Mockito.mock(),
+ measured);
+ catalog.initialize(
+ CATALOG_NAME,
+ ImmutableMap.of(
+ CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO"));
+
+ Assertions.assertThat(measured.getNumOutputFiles() + measured.getInputBytes())
+ .as("Nothing was created yet")
+ .isEqualTo(0);
+
+ catalog.createNamespace(NS);
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ // Asserting greaterThan 0 is sufficient for validating that the wrapper works without making
+ // assumptions about the
+ // specific implementations of table operations.
+ Assertions.assertThat(measured.getNumOutputFiles()).as("A table was created").isGreaterThan(0);
+
+ table.updateProperties().set("foo", "bar").commit();
+ Assertions.assertThat(measured.getInputBytes())
+ .as("A table was read and written")
+ .isGreaterThan(0);
+
+ Assertions.assertThat(catalog.dropTable(TABLE)).as("Table deletion should succeed").isTrue();
+ TableCleanupTaskHandler handler =
+ new TableCleanupTaskHandler(
+ Mockito.mock(),
+ createMockMetaStoreManagerFactory(),
+ (task) -> measured.loadFileIO("org.apache.iceberg.inmemory.InMemoryFileIO", Map.of()));
+ handler.handleTask(
+ TaskEntity.of(
+ metaStoreManager
+ .loadTasks(polarisContext, "testExecutor", 1)
+ .getEntities()
+ .getFirst()));
+ Assertions.assertThat(measured.getNumDeletedFiles()).as("A table was deleted").isGreaterThan(0);
+ }
}
diff --git a/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogViewTest.java b/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogViewTest.java
index 5b1916d1..f8051aa0 100644
--- a/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogViewTest.java
+++ b/polaris-service/src/test/java/org/apache/polaris/service/catalog/BasePolarisCatalogViewTest.java
@@ -129,7 +129,12 @@ public class BasePolarisCatalogViewTest extends ViewCatalogTests<BasePolarisCata
callContext, entityManager, authenticatedRoot, CATALOG_NAME);
this.catalog =
new BasePolarisCatalog(
- entityManager, callContext, passthroughView, authenticatedRoot, Mockito.mock());
+ entityManager,
+ callContext,
+ passthroughView,
+ authenticatedRoot,
+ Mockito.mock(),
+ new DefaultFileIOFactory());
this.catalog.initialize(
CATALOG_NAME,
ImmutableMap.of(
diff --git a/polaris-service/src/test/java/org/apache/polaris/service/catalog/MeasuredFileIOFactory.java b/polaris-service/src/test/java/org/apache/polaris/service/catalog/MeasuredFileIOFactory.java
new file mode 100644
index 00000000..ba8107e6
--- /dev/null
+++ b/polaris-service/src/test/java/org/apache/polaris/service/catalog/MeasuredFileIOFactory.java
@@ -0,0 +1,152 @@
+/*
+ * 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.polaris.service.catalog;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+
+/** A FileIOFactory that measures the number of bytes read, files written, and files deleted. */
+public class MeasuredFileIOFactory implements FileIOFactory {
+ private final List<MeasuredFileIO> ios;
+
+ public MeasuredFileIOFactory() {
+ ios = new ArrayList<>();
+ }
+
+ @Override
+ public FileIO loadFileIO(String ioImpl, Map<String, String> properties) {
+ MeasuredFileIO wrapped =
+ new MeasuredFileIO(CatalogUtil.loadFileIO(ioImpl, properties, new Configuration()));
+ ios.add(wrapped);
+ return wrapped;
+ }
+
+ public long getInputBytes() {
+ long sum = 0;
+ for (MeasuredFileIO io : ios) {
+ sum += io.inputBytes;
+ }
+ return sum;
+ }
+
+ public long getNumOutputFiles() {
+ long sum = 0;
+ for (MeasuredFileIO io : ios) {
+ sum += io.numOutputFiles;
+ }
+ return sum;
+ }
+
+ public long getNumDeletedFiles() {
+ long sum = 0;
+ for (MeasuredFileIO io : ios) {
+ sum += io.numDeletedFiles;
+ }
+ return sum;
+ }
+
+ public static class MeasuredFileIO implements FileIO {
+ private final FileIO io;
+ private long inputBytes;
+ private int numOutputFiles;
+ private int numDeletedFiles;
+
+ public MeasuredFileIO(FileIO io) {
+ this.io = io;
+ }
+
+ private InputFile measureInputFile(InputFile inner) {
+ inputBytes += inner.getLength();
+ return inner;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ return measureInputFile(io.newInputFile(path));
+ }
+
+ @Override
+ public InputFile newInputFile(String path, long length) {
+ return measureInputFile(io.newInputFile(path, length));
+ }
+
+ @Override
+ public InputFile newInputFile(DataFile file) {
+ return measureInputFile(io.newInputFile(file));
+ }
+
+ @Override
+ public InputFile newInputFile(DeleteFile file) {
+ return measureInputFile(io.newInputFile(file));
+ }
+
+ @Override
+ public InputFile newInputFile(ManifestFile manifest) {
+ return measureInputFile(io.newInputFile(manifest));
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ numOutputFiles++;
+ return io.newOutputFile(path);
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ numDeletedFiles++;
+ io.deleteFile(path);
+ }
+
+ @Override
+ public void deleteFile(InputFile file) {
+ numDeletedFiles++;
+ io.deleteFile(file);
+ }
+
+ @Override
+ public void deleteFile(OutputFile file) {
+ numDeletedFiles++;
+ io.deleteFile(file);
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return io.properties();
+ }
+
+ @Override
+ public void initialize(Map<String, String> properties) {
+ io.initialize(properties);
+ }
+
+ @Override
+ public void close() {
+ io.close();
+ }
+ }
+}
diff --git a/polaris-service/src/test/java/org/apache/polaris/service/catalog/PolarisCatalogHandlerWrapperAuthzTest.java b/polaris-service/src/test/java/org/apache/polaris/service/catalog/PolarisCatalogHandlerWrapperAuthzTest.java
index 93e8ba3a..8566e6ac 100644
--- a/polaris-service/src/test/java/org/apache/polaris/service/catalog/PolarisCatalogHandlerWrapperAuthzTest.java
+++ b/polaris-service/src/test/java/org/apache/polaris/service/catalog/PolarisCatalogHandlerWrapperAuthzTest.java
@@ -1676,7 +1676,8 @@ public class PolarisCatalogHandlerWrapperAuthzTest extends PolarisAuthzTestBase
return entityManager;
}
},
- Mockito.mock()) {
+ Mockito.mock(),
+ new DefaultFileIOFactory()) {
@Override
public Catalog createCallContextCatalog(
CallContext context,
diff --git a/polaris-service/src/test/resources/polaris-server-integrationtest.yml b/polaris-service/src/test/resources/polaris-server-integrationtest.yml
index 08b5373c..fd4a9d53 100644
--- a/polaris-service/src/test/resources/polaris-server-integrationtest.yml
+++ b/polaris-service/src/test/resources/polaris-server-integrationtest.yml
@@ -85,6 +85,9 @@ metaStoreManager:
# type: eclipse-link # uncomment to use eclipse-link as metastore
# persistence-unit: polaris-dev
+io:
+ factoryType: default
+
oauth2:
type: default
tokenBroker: