This is an automated email from the ASF dual-hosted git repository.
emaynard 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 c2209cf0f Integration tests for Catalog Federation (#2344)
c2209cf0f is described below
commit c2209cf0fcd92d7843ebde33be0bbbb560615c9b
Author: Pooja Nilangekar <[email protected]>
AuthorDate: Thu Aug 21 13:39:13 2025 -0400
Integration tests for Catalog Federation (#2344)
Adds a Junit5 integration test for catalog federation.
---
.../it/test/CatalogFederationIntegrationTest.java | 239 +++++++++++++++++++++
.../service/spark/it/CatalogFederationIT.java | 40 ++++
2 files changed, 279 insertions(+)
diff --git
a/integration-tests/src/main/java/org/apache/polaris/service/it/test/CatalogFederationIntegrationTest.java
b/integration-tests/src/main/java/org/apache/polaris/service/it/test/CatalogFederationIntegrationTest.java
new file mode 100644
index 000000000..890e92cae
--- /dev/null
+++
b/integration-tests/src/main/java/org/apache/polaris/service/it/test/CatalogFederationIntegrationTest.java
@@ -0,0 +1,239 @@
+/*
+ * 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.it.test;
+
+import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.net.URI;
+import java.util.List;
+import org.apache.polaris.core.admin.model.AuthenticationParameters;
+import org.apache.polaris.core.admin.model.Catalog;
+import org.apache.polaris.core.admin.model.CatalogGrant;
+import org.apache.polaris.core.admin.model.CatalogPrivilege;
+import org.apache.polaris.core.admin.model.CatalogProperties;
+import org.apache.polaris.core.admin.model.CatalogRole;
+import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.ExternalCatalog;
+import org.apache.polaris.core.admin.model.FileStorageConfigInfo;
+import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
+import org.apache.polaris.core.admin.model.PolarisCatalog;
+import org.apache.polaris.core.admin.model.PrincipalWithCredentials;
+import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.service.it.env.ClientCredentials;
+import org.apache.polaris.service.it.env.ManagementApi;
+import org.apache.polaris.service.it.env.PolarisApiEndpoints;
+import org.apache.polaris.service.it.env.PolarisClient;
+import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.apache.polaris.service.it.ext.SparkSessionBuilder;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Integration test for catalog federation functionality. This test verifies
that an external
+ * catalog can be created that federates with an internal catalog.
+ */
+@ExtendWith(PolarisIntegrationTestExtension.class)
+public class CatalogFederationIntegrationTest {
+
+ private static PolarisClient client;
+ private static ManagementApi managementApi;
+ private static PolarisApiEndpoints endpoints;
+ private static ClientCredentials adminCredentials;
+ private static SparkSession spark;
+ private static String sparkToken;
+
+ private static final String PRINCIPAL_NAME = "test-catalog-federation-user";
+ private static final String LOCAL_CATALOG_NAME = "test_catalog_local";
+ private static final String EXTERNAL_CATALOG_NAME = "test_catalog_external";
+ private static final String CATALOG_ROLE_NAME = "catalog_admin";
+ private static final String PRINCIPAL_ROLE_NAME = "service_admin";
+
+ @TempDir static java.nio.file.Path warehouseDir;
+
+ private URI baseLocation;
+ private PrincipalWithCredentials newUserCredentials;
+
+ @BeforeAll
+ static void setup(PolarisApiEndpoints apiEndpoints, ClientCredentials
credentials) {
+ endpoints = apiEndpoints;
+ client = polarisClient(endpoints);
+ managementApi = client.managementApi(credentials);
+ adminCredentials = credentials;
+ sparkToken = client.obtainToken(credentials);
+ }
+
+ @AfterAll
+ static void close() throws Exception {
+ if (client != null) {
+ client.close();
+ }
+ }
+
+ @BeforeEach
+ void before() {
+ this.baseLocation = URI.create("file:///tmp/warehouse");
+ }
+
+ @AfterEach
+ void after() {
+ if (spark != null) {
+ SparkSession.clearDefaultSession();
+ SparkSession.clearActiveSession();
+ spark.close();
+ }
+ managementApi.dropCatalog(EXTERNAL_CATALOG_NAME);
+ managementApi.dropCatalog(LOCAL_CATALOG_NAME);
+ managementApi.deletePrincipal(PRINCIPAL_NAME);
+ }
+
+ @Test
+ void testCatalogFederation() {
+ newUserCredentials = managementApi.createPrincipal(PRINCIPAL_NAME);
+
+ FileStorageConfigInfo storageConfig =
+ FileStorageConfigInfo.builder()
+ .setStorageType(StorageConfigInfo.StorageTypeEnum.FILE)
+ .setAllowedLocations(List.of(baseLocation.toString()))
+ .build();
+
+ CatalogProperties catalogProperties = new
CatalogProperties(baseLocation.toString());
+
+ Catalog localCatalog =
+ PolarisCatalog.builder()
+ .setType(Catalog.TypeEnum.INTERNAL)
+ .setName(LOCAL_CATALOG_NAME)
+ .setProperties(catalogProperties)
+ .setStorageConfigInfo(storageConfig)
+ .build();
+ managementApi.createCatalog(localCatalog);
+
+ CatalogGrant catalogGrant =
+ CatalogGrant.builder()
+ .setType(CatalogGrant.TypeEnum.CATALOG)
+ .setPrivilege(CatalogPrivilege.TABLE_WRITE_DATA)
+ .build();
+ managementApi.addGrant(LOCAL_CATALOG_NAME, CATALOG_ROLE_NAME,
catalogGrant);
+ managementApi.assignPrincipalRole(PRINCIPAL_NAME, PRINCIPAL_ROLE_NAME);
+ CatalogRole localCatalogAdminRole =
+ managementApi.getCatalogRole(LOCAL_CATALOG_NAME, CATALOG_ROLE_NAME);
+ managementApi.grantCatalogRoleToPrincipalRole(
+ PRINCIPAL_ROLE_NAME, LOCAL_CATALOG_NAME, localCatalogAdminRole);
+
+ AuthenticationParameters authParams =
+ OAuthClientCredentialsParameters.builder()
+
.setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.OAUTH)
+ .setTokenUri(endpoints.catalogApiEndpoint().toString() +
"/v1/oauth/tokens")
+ .setClientId(newUserCredentials.getCredentials().getClientId())
+
.setClientSecret(newUserCredentials.getCredentials().getClientSecret())
+ .setScopes(List.of("PRINCIPAL_ROLE:ALL"))
+ .build();
+ ConnectionConfigInfo connectionConfig =
+ IcebergRestConnectionConfigInfo.builder()
+
.setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.ICEBERG_REST)
+ .setUri(endpoints.catalogApiEndpoint().toString())
+ .setRemoteCatalogName(LOCAL_CATALOG_NAME)
+ .setAuthenticationParameters(authParams)
+ .build();
+ ExternalCatalog externalCatalog =
+ ExternalCatalog.builder()
+ .setType(Catalog.TypeEnum.EXTERNAL)
+ .setName(EXTERNAL_CATALOG_NAME)
+ .setConnectionConfigInfo(connectionConfig)
+ .setProperties(catalogProperties)
+ .setStorageConfigInfo(storageConfig)
+ .build();
+ managementApi.createCatalog(externalCatalog);
+
+ managementApi.addGrant(EXTERNAL_CATALOG_NAME, CATALOG_ROLE_NAME,
catalogGrant);
+ CatalogRole externalCatalogAdminRole =
+ managementApi.getCatalogRole(EXTERNAL_CATALOG_NAME, CATALOG_ROLE_NAME);
+ managementApi.grantCatalogRoleToPrincipalRole(
+ PRINCIPAL_ROLE_NAME, EXTERNAL_CATALOG_NAME, externalCatalogAdminRole);
+
+ spark =
+ SparkSessionBuilder.buildWithTestDefaults()
+ .withWarehouse(warehouseDir.toUri())
+ .addCatalog(
+ LOCAL_CATALOG_NAME, "org.apache.iceberg.spark.SparkCatalog",
endpoints, sparkToken)
+ .addCatalog(
+ EXTERNAL_CATALOG_NAME,
+ "org.apache.iceberg.spark.SparkCatalog",
+ endpoints,
+ sparkToken)
+ .getOrCreate();
+
+ spark.sql("USE " + LOCAL_CATALOG_NAME);
+ spark.sql("CREATE NAMESPACE IF NOT EXISTS ns1");
+ spark.sql("CREATE TABLE IF NOT EXISTS ns1.test_table (id int, name
string)");
+ spark.sql("INSERT INTO ns1.test_table VALUES (1, 'Alice')");
+ spark.sql("INSERT INTO ns1.test_table VALUES (2, 'Bob')");
+
+ spark.sql("CREATE NAMESPACE IF NOT EXISTS ns2");
+ spark.sql("CREATE TABLE IF NOT EXISTS ns2.test_table (id int, name
string)");
+ spark.sql("INSERT INTO ns2.test_table VALUES (1, 'Apache Spark')");
+ spark.sql("INSERT INTO ns2.test_table VALUES (2, 'Apache Iceberg')");
+
+ spark.sql("USE " + EXTERNAL_CATALOG_NAME);
+ List<Row> namespaces = spark.sql("SHOW NAMESPACES").collectAsList();
+ assertThat(namespaces).hasSize(2);
+
+ List<Row> ns1Data = spark.sql("SELECT * FROM ns1.test_table ORDER BY
id").collectAsList();
+ assertThat(ns1Data).hasSize(2);
+ assertThat(ns1Data.get(0).getInt(0)).isEqualTo(1);
+ assertThat(ns1Data.get(0).getString(1)).isEqualTo("Alice");
+ assertThat(ns1Data.get(1).getInt(0)).isEqualTo(2);
+ assertThat(ns1Data.get(1).getString(1)).isEqualTo("Bob");
+ spark.sql("INSERT INTO ns1.test_table VALUES (3, 'Charlie')");
+ List<Row> ns2Data = spark.sql("SELECT * FROM ns2.test_table ORDER BY
id").collectAsList();
+ assertThat(ns2Data).hasSize(2);
+ assertThat(ns2Data.get(0).getInt(0)).isEqualTo(1);
+ assertThat(ns2Data.get(0).getString(1)).isEqualTo("Apache Spark");
+ assertThat(ns2Data.get(1).getInt(0)).isEqualTo(2);
+ assertThat(ns2Data.get(1).getString(1)).isEqualTo("Apache Iceberg");
+ spark.sql("INSERT INTO ns2.test_table VALUES (3, 'Apache Polaris')");
+
+ spark.sql("USE " + LOCAL_CATALOG_NAME);
+ spark.sql("REFRESH TABLE ns1.test_table");
+ spark.sql("REFRESH TABLE ns2.test_table");
+ List<Row> updatedNs1Data =
+ spark.sql("SELECT * FROM ns1.test_table ORDER BY id").collectAsList();
+ assertThat(updatedNs1Data).hasSize(3);
+ assertThat(updatedNs1Data.get(2).getInt(0)).isEqualTo(3);
+ assertThat(updatedNs1Data.get(2).getString(1)).isEqualTo("Charlie");
+ List<Row> updatedNs2Data =
+ spark.sql("SELECT * FROM ns2.test_table ORDER BY id").collectAsList();
+ assertThat(updatedNs2Data).hasSize(3);
+ assertThat(updatedNs2Data.get(2).getInt(0)).isEqualTo(3);
+ assertThat(updatedNs2Data.get(2).getString(1)).isEqualTo("Apache Polaris");
+
+ spark.sql("DROP TABLE ns1.test_table");
+ spark.sql("DROP TABLE ns2.test_table");
+ spark.sql("DROP NAMESPACE ns1");
+ spark.sql("DROP NAMESPACE ns2");
+ }
+}
diff --git
a/runtime/spark-tests/src/intTest/java/org/apache/polaris/service/spark/it/CatalogFederationIT.java
b/runtime/spark-tests/src/intTest/java/org/apache/polaris/service/spark/it/CatalogFederationIT.java
new file mode 100644
index 000000000..a9e3fbd79
--- /dev/null
+++
b/runtime/spark-tests/src/intTest/java/org/apache/polaris/service/spark/it/CatalogFederationIT.java
@@ -0,0 +1,40 @@
+/*
+ * 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.spark.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import io.quarkus.test.junit.QuarkusTestProfile;
+import io.quarkus.test.junit.TestProfile;
+import java.util.Map;
+import org.apache.polaris.service.it.test.CatalogFederationIntegrationTest;
+
+@TestProfile(CatalogFederationIT.CatalogFederationProfile.class)
+@QuarkusIntegrationTest
+public class CatalogFederationIT extends CatalogFederationIntegrationTest {
+
+ public static class CatalogFederationProfile implements QuarkusTestProfile {
+ @Override
+ public Map<String, String> getConfigOverrides() {
+ return Map.of(
+ "polaris.features.\"ENABLE_CATALOG_FEDERATION\"", "true",
+ "polaris.features.\"SUPPORTED_CATALOG_CONNECTION_TYPES\"",
"[\"ICEBERG_REST\"]",
+ "polaris.features.\"ALLOW_OVERLAPPING_CATALOG_URLS\"", "true");
+ }
+ }
+}
