diff --git a/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java b/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java index ee99b07684..c8ab2305fb 100644 --- a/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java +++ b/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java @@ -53,6 +53,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableMetadataParser; +import org.apache.iceberg.TableProperties; import org.apache.iceberg.Transaction; import org.apache.iceberg.UpdatePartitionSpec; import org.apache.iceberg.UpdateSchema; @@ -1489,6 +1490,159 @@ public void multipleDiffsAgainstMultipleTablesLastFails() { assertThat(schema2.columns()).hasSize(1); } + /** + * With {@code write.metadata.delete-after-commit.enabled=true}, old metadata files evicted from + * the metadata log must NOT be deleted while a multi-table transaction is in flight: if a later + * table in the transaction fails, the catalog rolls back to metadata pointers that must still + * resolve to existing files. + */ + @Test + public void testCommitTransactionDoesNotDeleteOldMetadataFilesOnFailure() { + Namespace namespace = Namespace.of("txnMetadataRetainNs"); + TableIdentifier identifier1 = TableIdentifier.of(namespace, "txnMetadataRetainTable1"); + TableIdentifier identifier2 = TableIdentifier.of(namespace, "txnMetadataRetainTable2"); + + if (requiresNamespaceCreate()) { + catalog().createNamespace(namespace); + } + + // Keep only one previous metadata version, so every commit makes the oldest metadata file a + // deletion candidate. + Map deleteAfterCommitProps = + Map.of( + TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED, "true", + TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, "1"); + + catalog().buildTable(identifier1, SCHEMA).withProperties(deleteAfterCommitProps).create(); + Table table2 = + catalog().buildTable(identifier2, SCHEMA).withProperties(deleteAfterCommitProps).create(); + + // Move table1 from v1 to v2 so the single previous-versions slot is filled with v1. + catalog() + .loadTable(identifier1) + .updateSchema() + .addColumn("extra_col", Types.LongType.get()) + .commit(); + + // Capture all metadata file locations of table1 before the failing transaction (v2 + v1). + TableMetadata preTxnMetadata = + ((BaseTable) catalog().loadTable(identifier1)).operations().current(); + List preTxnMetadataLocations = + Stream.concat( + Stream.of(preTxnMetadata.metadataFileLocation()), + preTxnMetadata.previousFiles().stream().map(TableMetadata.MetadataLogEntry::file)) + .toList(); + assertThat(preTxnMetadataLocations).hasSizeGreaterThanOrEqualTo(2); + + // Transaction: a benign update on table1 (would evict v1 from the metadata log) and a + // conflicting update on table2 that makes the transaction fail as a whole. + Transaction t1Transaction = catalog().loadTable(identifier1).newTransaction(); + t1Transaction.updateSchema().addColumn("new_col1", Types.LongType.get()).commit(); + + Transaction t2Transaction = catalog().loadTable(identifier2).newTransaction(); + t2Transaction.updateSchema().renameColumn("data", "new-column").commit(); + + // delete the column that is being renamed in the above TX to cause a conflict + table2.updateSchema().deleteColumn("data").commit(); + + TableCommit tableCommit1 = + TableCommit.create( + identifier1, + ((BaseTransaction) t1Transaction).startMetadata(), + ((BaseTransaction) t1Transaction).currentMetadata()); + TableCommit tableCommit2 = + TableCommit.create( + identifier2, + ((BaseTransaction) t2Transaction).startMetadata(), + ((BaseTransaction) t2Transaction).currentMetadata()); + + assertThatThrownBy(() -> restCatalog.commitTransaction(tableCommit1, tableCommit2)) + .isInstanceOf(CommitFailedException.class); + + // The transaction failed, so every metadata file that existed before it must still exist. + try (ResolvingFileIO resolvingFileIO = new ResolvingFileIO()) { + initializeClientFileIO(resolvingFileIO); + resolvingFileIO.setConf(new Configuration()); + for (String location : preTxnMetadataLocations) { + assertThat(resolvingFileIO.newInputFile(location).exists()) + .as("Metadata file %s must not be deleted by a failed transaction", location) + .isTrue(); + } + } + } + + /** + * Positive counterpart of {@link #testCommitTransactionDoesNotDeleteOldMetadataFilesOnFailure}: + * once a multi-table transaction commits successfully, metadata files evicted from the metadata + * log are deleted (deferred until after the atomic commit). + */ + @Test + public void testCommitTransactionDeletesOldMetadataFilesAfterSuccess() { + Namespace namespace = Namespace.of("txnMetadataDeleteNs"); + TableIdentifier identifier1 = TableIdentifier.of(namespace, "txnMetadataDeleteTable1"); + TableIdentifier identifier2 = TableIdentifier.of(namespace, "txnMetadataDeleteTable2"); + + if (requiresNamespaceCreate()) { + catalog().createNamespace(namespace); + } + + Map deleteAfterCommitProps = + Map.of( + TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED, "true", + TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, "1"); + + catalog().buildTable(identifier1, SCHEMA).withProperties(deleteAfterCommitProps).create(); + catalog().buildTable(identifier2, SCHEMA).withProperties(deleteAfterCommitProps).create(); + + // Move table1 from v1 to v2; v1 now occupies the single previous-versions slot and will be + // evicted (and become deletable) by the next commit. + catalog() + .loadTable(identifier1) + .updateSchema() + .addColumn("extra_col", Types.LongType.get()) + .commit(); + + TableMetadata preTxnMetadata = + ((BaseTable) catalog().loadTable(identifier1)).operations().current(); + assertThat(preTxnMetadata.previousFiles()).isNotEmpty(); + String evictedMetadataLocation = preTxnMetadata.previousFiles().get(0).file(); + + Transaction t1Transaction = catalog().loadTable(identifier1).newTransaction(); + t1Transaction.updateSchema().addColumn("new_col1", Types.LongType.get()).commit(); + + Transaction t2Transaction = catalog().loadTable(identifier2).newTransaction(); + t2Transaction.updateSchema().addColumn("new_col2", Types.LongType.get()).commit(); + + TableCommit tableCommit1 = + TableCommit.create( + identifier1, + ((BaseTransaction) t1Transaction).startMetadata(), + ((BaseTransaction) t1Transaction).currentMetadata()); + TableCommit tableCommit2 = + TableCommit.create( + identifier2, + ((BaseTransaction) t2Transaction).startMetadata(), + ((BaseTransaction) t2Transaction).currentMetadata()); + + restCatalog.commitTransaction(tableCommit1, tableCommit2); + + TableMetadata postTxnMetadata = + ((BaseTable) catalog().loadTable(identifier1)).operations().current(); + + try (ResolvingFileIO resolvingFileIO = new ResolvingFileIO()) { + initializeClientFileIO(resolvingFileIO); + resolvingFileIO.setConf(new Configuration()); + assertThat(resolvingFileIO.newInputFile(evictedMetadataLocation).exists()) + .as( + "Metadata file %s evicted by the committed transaction must be deleted", + evictedMetadataLocation) + .isFalse(); + assertThat(resolvingFileIO.newInputFile(postTxnMetadata.metadataFileLocation()).exists()) + .as("Current metadata file must exist after the committed transaction") + .isTrue(); + } + } + @Test public void testMultipleConflictingCommitsToSingleTableInTransaction() { Namespace namespace = Namespace.of("ns1"); diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java index 66c1d63f5a..e996e16cea 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionEventTest.java @@ -24,23 +24,15 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import jakarta.ws.rs.core.Response; -import java.nio.file.Files; import java.nio.file.Path; -import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.UUID; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Collectors; -import java.util.stream.Stream; import org.apache.iceberg.MetadataUpdate; import org.apache.iceberg.TableMetadata; -import org.apache.iceberg.TableProperties; import org.apache.iceberg.UpdateRequirement; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; -import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.rest.requests.CommitTransactionRequest; import org.apache.iceberg.rest.requests.CreateNamespaceRequest; import org.apache.iceberg.rest.requests.CreateTableRequest; @@ -50,8 +42,6 @@ import org.apache.polaris.core.admin.model.CreateCatalogRequest; import org.apache.polaris.core.admin.model.FileStorageConfigInfo; import org.apache.polaris.core.admin.model.StorageConfigInfo; -import org.apache.polaris.core.persistence.dao.entity.BaseResult; -import org.apache.polaris.core.persistence.dao.entity.EntitiesResult; import org.apache.polaris.service.TestServices; import org.apache.polaris.service.events.EventAttributes; import org.apache.polaris.service.events.PolarisEvent; @@ -60,7 +50,6 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; -import org.mockito.Mockito; public class CommitTransactionEventTest { private static final String namespace = "ns"; @@ -165,128 +154,6 @@ void testLoadTableResponsesInCommitTransaction() { assertThat(metadata.properties()).containsEntry(propertyName, "value2"); } - @Test - void testCommitTransactionDoesNotDeleteOldMetadataFilesDuringTransaction() throws Exception { - TestServices testServices = createTestServices(); - createCatalogAndNamespace(testServices, Map.of(), catalogLocation); - - String table1Name = "txn-nodelete-table1"; - String table2Name = "txn-nodelete-table2"; - - // Create tables with metadata deletion enabled and max previous versions = 1. - // This means after the second commit, the original (v1) metadata file becomes - // eligible for deletion. - Map tableProperties = new HashMap<>(); - tableProperties.put(TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED, "true"); - tableProperties.put(TableProperties.METADATA_PREVIOUS_VERSIONS_MAX, "1"); - createTableWithProperties(testServices, table1Name, catalogLocation, tableProperties); - createTableWithProperties(testServices, table2Name, catalogLocation, tableProperties); - - // Perform an initial update on table1 via commitTransaction to move it from v1 to v2. - // After this, v1 is in the previous-files log (max=1 slot filled). - testServices - .restApi() - .commitTransaction( - catalog, - new CommitTransactionRequest( - List.of( - UpdateTableRequest.create( - TableIdentifier.of(namespace, table1Name), - List.of(), - List.of( - new MetadataUpdate.SetProperties( - Map.of("initial-prop", "initial-value")))))), - IDEMPOTENCY_KEY, - testServices.realmContext(), - testServices.securityContext()); - - // Capture the set of metadata files on disk for table1 before the next transaction. - // At this point we should have v1.metadata.json and v2.metadata.json. - Path table1MetadataDir = - Path.of( - catalogLocation.replaceFirst("^file:", ""), catalog, namespace, table1Name, "metadata"); - Set metadataFilesBefore = metadataFiles(table1MetadataDir); - assertThat(metadataFilesBefore) - .as("Should have at least 2 metadata files (v1 and v2) before the transaction") - .hasSizeGreaterThanOrEqualTo(2); - - // Now perform a multi-table commitTransaction that updates table1 (v2 -> v3). - // This will evict v1 from the previous-files log, making it a deletion candidate. - // The second table update will fail validation (bad schema ID), causing the - // transaction to abort. - // BUG (without PR #4920): CatalogUtil.deleteRemovedMetadataFiles runs eagerly - // inside the per-table commit, deleting v1 even though the overall transaction - // has not committed atomically yet. - try { - testServices - .restApi() - .commitTransaction( - catalog, - new CommitTransactionRequest( - List.of( - UpdateTableRequest.create( - TableIdentifier.of(namespace, table1Name), - List.of(), - List.of( - new MetadataUpdate.SetProperties( - Map.of("second-prop", "second-value")))), - UpdateTableRequest.create( - TableIdentifier.of(namespace, table2Name), - // This requirement will fail: schema ID -1 does not exist - List.of(new UpdateRequirement.AssertCurrentSchemaID(-1)), - List.of(new MetadataUpdate.SetProperties(Map.of("will-fail", "true")))))), - IDEMPOTENCY_KEY, - testServices.realmContext(), - testServices.securityContext()); - } catch (Exception ignored) { - // Expected: second table's requirement fails - } - - // Assert: all metadata files that existed before the transaction must still be present. - // The transaction failed, so no metadata should have been deleted. - Set metadataFilesAfter = metadataFiles(table1MetadataDir); - assertThat(metadataFilesAfter) - .as( - "Old metadata files must NOT be deleted during a transaction that has not " - + "committed atomically — premature deletion causes data corruption on rollback") - .containsAll(metadataFilesBefore); - } - - private Set metadataFiles(Path metadataDir) throws Exception { - if (!Files.exists(metadataDir)) { - return Set.of(); - } - try (Stream stream = Files.list(metadataDir)) { - return stream - .filter(p -> p.getFileName().toString().endsWith(".metadata.json")) - .collect(Collectors.toSet()); - } - } - - private void createTableWithProperties( - TestServices services, - String tableName, - String baseLocation, - Map properties) { - CreateTableRequest createTableRequest = - CreateTableRequest.builder() - .withName(tableName) - .withLocation(String.format("%s/%s/%s/%s", baseLocation, catalog, namespace, tableName)) - .withSchema(SCHEMA) - .setProperties(properties) - .build(); - services - .restApi() - .createTable( - catalog, - namespace, - createTableRequest, - null, - IDEMPOTENCY_KEY, - services.realmContext(), - services.securityContext()); - } - private void createCatalogAndNamespace( TestServices services, Map catalogConfig, String catalogLocation) { CatalogProperties.Builder propertiesBuilder = @@ -406,74 +273,4 @@ private CommitTransactionRequest generateCommitTransactionRequest( updateRequirements, List.of(new MetadataUpdate.SetProperties(Map.of(propertyName, "value2")))))); } - - @Test - void testCommitTransactionCleansUpMetadataOnFailure(@TempDir Path tempDir) throws Exception { - String location = tempDir.toAbsolutePath().toUri().toString(); - if (location.endsWith("/")) { - location = location.substring(0, location.length() - 1); - } - - // Create TestServices with a spy that will fail on updateEntitiesPropertiesIfNotChanged - // but only AFTER initial setup (table creation) succeeds. - AtomicBoolean shouldFail = new AtomicBoolean(false); - TestServices testServices = - TestServices.builder() - .config( - Map.of( - "ALLOW_INSECURE_STORAGE_TYPES", - "true", - "SUPPORTED_CATALOG_STORAGE_TYPES", - List.of("FILE"))) - .metaStoreManagerDecorator( - msm -> { - org.apache.polaris.core.persistence.PolarisMetaStoreManager spy = - Mockito.spy(msm); - Mockito.doAnswer( - invocation -> { - if (shouldFail.get()) { - return new EntitiesResult( - BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, - "simulated CAS failure"); - } - return invocation.callRealMethod(); - }) - .when(spy) - .updateEntitiesPropertiesIfNotChanged(Mockito.any(), Mockito.any()); - return spy; - }) - .build(); - - createCatalogAndNamespace(testServices, Map.of(), location); - - String table1Name = "cleanup-table-1"; - String table2Name = "cleanup-table-2"; - createTable(testServices, table1Name, location); - createTable(testServices, table2Name, location); - - // Capture exact set of metadata file paths before the failing transaction - Set metadataFilesBefore = metadataFiles(tempDir); - - // Now enable the CAS failure and attempt a commitTransaction - shouldFail.set(true); - assertThatThrownBy( - () -> - testServices - .restApi() - .commitTransaction( - catalog, - generateCommitTransactionRequest(false, table1Name, table2Name), - IDEMPOTENCY_KEY, - testServices.realmContext(), - testServices.securityContext())) - .isInstanceOf(CommitFailedException.class) - .hasMessageContaining("Transaction commit failed"); - - // After the failed transaction, no new metadata files should remain (they were cleaned up). - Set metadataFilesAfter = metadataFiles(tempDir); - assertThat(metadataFilesAfter).isEqualTo(metadataFilesBefore); - } - - // Positive test added in AbstractLocalIcebergCatalogTest to avoid making this event-focused test - // class depend on full transaction setup that can be slow/heavy in some envs. } diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionMetadataCleanupTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionMetadataCleanupTest.java new file mode 100644 index 0000000000..1789afb730 --- /dev/null +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitTransactionMetadataCleanupTest.java @@ -0,0 +1,210 @@ +/* + * 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.iceberg; + +import static org.apache.polaris.service.admin.PolarisAuthzTestBase.SCHEMA; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import jakarta.ws.rs.core.Response; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.iceberg.MetadataUpdate; +import org.apache.iceberg.catalog.Namespace; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.rest.requests.CommitTransactionRequest; +import org.apache.iceberg.rest.requests.CreateNamespaceRequest; +import org.apache.iceberg.rest.requests.CreateTableRequest; +import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.polaris.core.admin.model.Catalog; +import org.apache.polaris.core.admin.model.CatalogProperties; +import org.apache.polaris.core.admin.model.CreateCatalogRequest; +import org.apache.polaris.core.admin.model.FileStorageConfigInfo; +import org.apache.polaris.core.admin.model.StorageConfigInfo; +import org.apache.polaris.core.persistence.PolarisMetaStoreManager; +import org.apache.polaris.core.persistence.dao.entity.BaseResult; +import org.apache.polaris.core.persistence.dao.entity.EntitiesResult; +import org.apache.polaris.service.TestServices; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.Mockito; + +/** + * Tests that {@code commitTransaction} cleans up newly written metadata files when the final atomic + * batch update against the metastore fails. + */ +public class CommitTransactionMetadataCleanupTest { + private static final String namespace = "ns"; + private static final String catalog = "test-catalog"; + private static final String propertyName = "custom-property-1"; + + // UUID v7 + private static final UUID IDEMPOTENCY_KEY = new UUID(116617318654508422L, -7820829973016961092L); + + @Test + void testCommitTransactionCleansUpMetadataOnFailure(@TempDir Path tempDir) throws Exception { + String location = tempDir.toAbsolutePath().toUri().toString(); + if (location.endsWith("/")) { + location = location.substring(0, location.length() - 1); + } + + // Create TestServices with a spy that will fail on updateEntitiesPropertiesIfNotChanged + // but only AFTER initial setup (table creation) succeeds. + AtomicBoolean shouldFail = new AtomicBoolean(false); + TestServices testServices = + TestServices.builder() + .config( + Map.of( + "ALLOW_INSECURE_STORAGE_TYPES", + "true", + "SUPPORTED_CATALOG_STORAGE_TYPES", + List.of("FILE"))) + .metaStoreManagerDecorator( + msm -> { + PolarisMetaStoreManager spy = Mockito.spy(msm); + Mockito.doAnswer( + invocation -> { + if (shouldFail.get()) { + return new EntitiesResult( + BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, + "simulated CAS failure"); + } + return invocation.callRealMethod(); + }) + .when(spy) + .updateEntitiesPropertiesIfNotChanged(Mockito.any(), Mockito.any()); + return spy; + }) + .build(); + + createCatalogAndNamespace(testServices, location); + + String table1Name = "cleanup-table-1"; + String table2Name = "cleanup-table-2"; + createTable(testServices, table1Name, location); + createTable(testServices, table2Name, location); + + // Capture exact set of metadata file paths before the failing transaction + Set metadataFilesBefore = metadataFiles(tempDir); + + // Now enable the CAS failure and attempt a commitTransaction + shouldFail.set(true); + assertThatThrownBy( + () -> + testServices + .restApi() + .commitTransaction( + catalog, + generateCommitTransactionRequest(table1Name, table2Name), + IDEMPOTENCY_KEY, + testServices.realmContext(), + testServices.securityContext())) + .isInstanceOf(CommitFailedException.class) + .hasMessageContaining("Transaction commit failed"); + + // After the failed transaction, no new metadata files should remain (they were cleaned up). + Set metadataFilesAfter = metadataFiles(tempDir); + assertThat(metadataFilesAfter).isEqualTo(metadataFilesBefore); + } + + private static Set metadataFiles(Path directory) throws Exception { + try (Stream files = Files.walk(directory)) { + return files.filter(p -> p.toString().endsWith(".metadata.json")).collect(Collectors.toSet()); + } + } + + private CommitTransactionRequest generateCommitTransactionRequest( + String table1Name, String table2Name) { + return new CommitTransactionRequest( + List.of( + UpdateTableRequest.create( + TableIdentifier.of(namespace, table1Name), + List.of(), + List.of(new MetadataUpdate.SetProperties(Map.of(propertyName, "value1")))), + UpdateTableRequest.create( + TableIdentifier.of(namespace, table2Name), + List.of(), + List.of(new MetadataUpdate.SetProperties(Map.of(propertyName, "value2")))))); + } + + private void createCatalogAndNamespace(TestServices services, String catalogLocation) { + CatalogProperties.Builder propertiesBuilder = + CatalogProperties.builder() + .setDefaultBaseLocation(String.format("%s/%s", catalogLocation, catalog)); + + StorageConfigInfo config = + FileStorageConfigInfo.builder() + .setStorageType(StorageConfigInfo.StorageTypeEnum.FILE) + .build(); + Catalog catalogObject = + new Catalog( + Catalog.TypeEnum.INTERNAL, catalog, propertiesBuilder.build(), 0L, 0L, 1, config); + try (Response response = + services + .catalogsApi() + .createCatalog( + new CreateCatalogRequest(catalogObject), + services.realmContext(), + services.securityContext())) { + assertThat(response.getStatus()).isEqualTo(Response.Status.CREATED.getStatusCode()); + } + + CreateNamespaceRequest createNamespaceRequest = + CreateNamespaceRequest.builder().withNamespace(Namespace.of(namespace)).build(); + try (Response response = + services + .restApi() + .createNamespace( + catalog, + createNamespaceRequest, + IDEMPOTENCY_KEY, + services.realmContext(), + services.securityContext())) { + assertThat(response.getStatus()).isEqualTo(Response.Status.OK.getStatusCode()); + } + } + + private void createTable(TestServices services, String tableName, String baseLocation) { + CreateTableRequest createTableRequest = + CreateTableRequest.builder() + .withName(tableName) + .withLocation(String.format("%s/%s/%s/%s", baseLocation, catalog, namespace, tableName)) + .withSchema(SCHEMA) + .build(); + services + .restApi() + .createTable( + catalog, + namespace, + createTableRequest, + null, + IDEMPOTENCY_KEY, + services.realmContext(), + services.securityContext()); + } +}