diff --git a/CHANGELOG.md b/CHANGELOG.md index d6e7e32846..200c5743fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,9 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti ## [Unreleased] ### Highlights +- Table and view commits no longer delete newly written metadata when the metastore write outcome + is unknown (e.g. a dropped connection); they still clean up on known failures. JDBC reports these + errors as `PersistenceCommitStateUnknownException` (HTTP 500). ### Upgrade notes diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java index aaeaff117c..85093c6a3c 100644 --- a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java +++ b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperations.java @@ -60,6 +60,14 @@ public class DatasourceOperations { private static final String UNIQUENESS_CONSTRAINT_VIOLATION_SQL_CODE = "23505"; private static final String RELATION_DOES_NOT_EXIST = "42P01"; + // SQLSTATE codes treated as ambiguous commit outcomes (the write may already have committed). + // Class 08 (connection exception) is SQL-standard and portable across databases. + private static final String CONNECTION_EXCEPTION_SQL_STATE_CLASS = "08"; + // POSTGRES: query_canceled, e.g. statement_timeout (also CockroachDB via PG compatibility) + private static final String POSTGRES_QUERY_CANCELED_SQL_STATE = "57014"; + // ODBC: timeout expired (used by some drivers, e.g. jTDS and older MySQL connectors) + private static final String ODBC_TIMEOUT_SQL_STATE = "HYT00"; + // H2 STATUS CODES // 90079 = Schema not found, 42S02 = Table or view not found private static final String H2_SCHEMA_DOES_NOT_EXIST = "90079"; @@ -178,7 +186,8 @@ public void executeSelectOverStream( executeSelectOverStreamWithConnection(query, converterInstance, consumer, connection); return null; } - }); + }, + false); } /** Connection-aware version for use inside runWithinTransaction. */ @@ -192,7 +201,8 @@ public void executeSelectOverStream( () -> { executeSelectOverStreamWithConnection(query, converterInstance, consumer, connection); return null; - }); + }, + false); } /** @@ -358,7 +368,67 @@ public Integer execute(Connection connection, QueryGenerator.PreparedQuery prepa } } - private boolean isRetryable(SQLException e) { + /** + * Whether a SQLException indicates the statement may already have been applied on the server + * while the client cannot confirm the outcome (connection loss, timeout, cancellation). + * + *

Callers must not treat these as definite failures for cleanup or safe retry of CAS writes. + */ + public boolean isAmbiguousCommitOutcome(SQLException e) { + if (e == null) { + return false; + } + if (e instanceof java.sql.SQLTimeoutException) { + return true; + } + if (e instanceof java.sql.SQLTransientConnectionException + || e instanceof java.sql.SQLNonTransientConnectionException) { + return true; + } + String sqlState = e.getSQLState(); + if (sqlState != null) { + if (sqlState.startsWith(CONNECTION_EXCEPTION_SQL_STATE_CLASS) + || sqlState.equals(POSTGRES_QUERY_CANCELED_SQL_STATE) + || sqlState.equals(ODBC_TIMEOUT_SQL_STATE)) { + return true; + } + } + return messageContainsAny( + e, + "connection reset", + "connection refused", + "connection is closed", + "broken pipe", + "query canceled", + "canceling statement due to statement timeout"); + } + + /** + * Whether the exception message contains any of the given lowercase needles. The message is + * lowercased once; a null message matches nothing. + */ + private static boolean messageContainsAny(SQLException e, String... needles) { + String message = e.getMessage(); + if (message == null) { + return false; + } + String lower = message.toLowerCase(Locale.ROOT); + for (String needle : needles) { + if (lower.contains(needle)) { + return true; + } + } + return false; + } + + private boolean isRetryable(SQLException e, boolean mutating) { + // For mutating operations an ambiguous outcome (connection loss, timeout, cancellation) may + // mean the write already committed under auto-commit, so retrying risks a double-apply or a + // misreported result. Reads have no commit outcome to protect, so they stay retryable. + if (mutating && isAmbiguousCommitOutcome(e)) { + return false; + } + String sqlState = e.getSQLState(); if (sqlState != null) { @@ -366,14 +436,20 @@ private boolean isRetryable(SQLException e) { } // Additionally, one might check for specific error messages or other conditions - return e.getMessage().toLowerCase(Locale.ROOT).contains("connection refused") - || e.getMessage().toLowerCase(Locale.ROOT).contains("connection reset"); + return messageContainsAny(e, "connection refused", "connection reset"); } // TODO: consider refactoring to use a retry library, inorder to have fair retries // and more knobs for tuning retry pattern. @VisibleForTesting T withRetries(Operation operation) throws SQLException { + // Default to the mutating policy: it is the safe choice when the caller does not state whether + // the operation writes. + return withRetries(operation, true); + } + + @VisibleForTesting + T withRetries(Operation operation, boolean mutating) throws SQLException { int attempts = 0; // maximum number of retries. int maxAttempts = relationalJdbcConfiguration.maxRetries().orElse(1); @@ -409,7 +485,7 @@ T withRetries(Operation operation) throws SQLException { attempts++; long timeLeft = Math.max((maxRetryTime - TimeUnit.NANOSECONDS.toMillis(System.nanoTime())), 0L); - if (timeLeft == 0 || attempts >= maxAttempts || !isRetryable(sqlException)) { + if (timeLeft == 0 || attempts >= maxAttempts || !isRetryable(sqlException, mutating)) { String exceptionMessage = String.format( "Failed due to '%s' (error code %d, sql-state '%s'), after %s attempts and %s milliseconds", diff --git a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java index 1709c8b535..9aec4697b5 100644 --- a/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java +++ b/persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcBasePersistenceImpl.java @@ -56,6 +56,7 @@ import org.apache.polaris.core.persistence.BasePersistence; import org.apache.polaris.core.persistence.EntityAlreadyExistsException; import org.apache.polaris.core.persistence.IntegrationPersistence; +import org.apache.polaris.core.persistence.PersistenceCommitStateUnknownException; import org.apache.polaris.core.persistence.PolicyMappingAlreadyExistsException; import org.apache.polaris.core.persistence.PrincipalSecretsGenerator; import org.apache.polaris.core.persistence.RetryOnConcurrencyException; @@ -144,7 +145,7 @@ public void writeEntity( return datasourceOperations.executeUpdate(preparedQuery); }); } catch (SQLException e) { - throw new RuntimeException("Error persisting entity", e); + throw wrapEntityWriteFailure(e); } } @@ -169,10 +170,7 @@ public void writeEntities( return true; }); } catch (SQLException e) { - throw new RuntimeException( - String.format( - "Error executing the transaction for writing entities due to %s", e.getMessage()), - e); + throw wrapEntityWriteFailure(e, "Error executing the transaction for writing entities"); } } @@ -236,8 +234,7 @@ private void persistEntity( throw new RetryOnConcurrencyException( e, "Conflicting entity is not visible in the current transaction snapshot; retry"); } - throw new RuntimeException( - String.format("Failed to write entity due to %s", e.getMessage()), e); + throw wrapEntityWriteFailure(e); } } else { // CAS on both entity_version and grant_records_version because grant operations only @@ -274,12 +271,23 @@ private void persistEntity( originalEntity.getGrantRecordsVersion()); } } catch (SQLException e) { - throw new RuntimeException( - String.format("Failed to write entity due to %s", e.getMessage()), e); + throw wrapEntityWriteFailure(e); } } } + private RuntimeException wrapEntityWriteFailure(SQLException e) { + return wrapEntityWriteFailure(e, "Failed to write entity"); + } + + private RuntimeException wrapEntityWriteFailure(SQLException e, String context) { + if (datasourceOperations.isAmbiguousCommitOutcome(e)) { + return new PersistenceCommitStateUnknownException( + String.format("%s due to %s; commit outcome is unknown", context, e.getMessage()), e); + } + return new RuntimeException(String.format("%s due to %s", context, e.getMessage()), e); + } + @Override public void writeToGrantRecords( @NonNull PolarisCallContext callCtx, @NonNull PolarisGrantRecord grantRec) { diff --git a/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperationsTest.java b/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperationsTest.java index 02b1382cb5..6c14bf5beb 100644 --- a/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperationsTest.java +++ b/persistence/relational-jdbc/src/test/java/org/apache/polaris/persistence/relational/jdbc/DatasourceOperationsTest.java @@ -228,18 +228,19 @@ void testSuccessfulExecutionOnFirstAttempt() throws SQLException { @Test void testSuccessfulExecutionAfterOneRetry() throws SQLException { - when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(4)); + when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(3)); when(relationalJdbcConfiguration.maxDurationInMs()).thenReturn(Optional.of(2000L)); when(relationalJdbcConfiguration.initialDelayInMs()).thenReturn(Optional.of(0L)); + // Only serialization failures (definite rollback) are retryable; connection-class errors are + // treated as ambiguous commit outcomes and must not be retried. when(mockOperation.execute()) .thenThrow(new SQLException("Retryable error", "40001")) - .thenThrow(new SQLException("connection refused")) - .thenThrow(new SQLException("connection reset")) + .thenThrow(new SQLException("Retryable error", "40001")) .thenReturn("Success!"); String result = datasourceOperations.withRetries(mockOperation); assertEquals("Success!", result); - verify(mockOperation, times(4)).execute(); + verify(mockOperation, times(3)).execute(); } @Test @@ -323,4 +324,48 @@ void testDefaultConfigurationValues() throws SQLException { assertThrows(SQLException.class, () -> datasourceOperations.withRetries(mockOperation)); verify(mockOperation, times(1)).execute(); } + + @Test + void isAmbiguousCommitOutcome_classifiesConnectionAndTimeoutFailures() { + assertTrue(datasourceOperations.isAmbiguousCommitOutcome(new SQLException("reset", "08006"))); + assertTrue( + datasourceOperations.isAmbiguousCommitOutcome(new SQLException("canceled", "57014"))); + assertTrue( + datasourceOperations.isAmbiguousCommitOutcome( + new SQLException("Connection reset by peer"))); + assertTrue( + datasourceOperations.isAmbiguousCommitOutcome(new java.sql.SQLTimeoutException("timeout"))); + // Serialization failure is a definite rollback — retryable, not ambiguous commit success. + assertTrue( + !datasourceOperations.isAmbiguousCommitOutcome(new SQLException("serialization", "40001"))); + // Constraint violations are definite failures. + assertTrue(!datasourceOperations.isAmbiguousCommitOutcome(new SQLException("unique", "23505"))); + } + + @Test + void withRetries_doesNotRetryAmbiguousConnectionFailure() throws Exception { + when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(3)); + when(relationalJdbcConfiguration.maxDurationInMs()).thenReturn(Optional.of(5000L)); + when(relationalJdbcConfiguration.initialDelayInMs()).thenReturn(Optional.of(1L)); + when(mockOperation.execute()).thenThrow(new SQLException("Connection reset", "08006")); + + assertThrows(SQLException.class, () -> datasourceOperations.withRetries(mockOperation)); + verify(mockOperation, times(1)).execute(); + } + + @Test + void withRetries_retriesAmbiguousConnectionFailureForReads() throws Exception { + when(relationalJdbcConfiguration.maxRetries()).thenReturn(Optional.of(3)); + when(relationalJdbcConfiguration.maxDurationInMs()).thenReturn(Optional.of(5000L)); + when(relationalJdbcConfiguration.initialDelayInMs()).thenReturn(Optional.of(1L)); + // A read has no commit outcome to protect, so a transient connection failure is retried. + when(mockOperation.execute()) + .thenThrow(new SQLException("Connection reset by peer")) + .thenThrow(new SQLException("Connection reset by peer")) + .thenReturn("Success!"); + + String result = datasourceOperations.withRetries(mockOperation, false); + assertEquals("Success!", result); + verify(mockOperation, times(3)).execute(); + } } diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/PersistenceCommitStateUnknownException.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/PersistenceCommitStateUnknownException.java new file mode 100644 index 0000000000..26d7d88a27 --- /dev/null +++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/PersistenceCommitStateUnknownException.java @@ -0,0 +1,51 @@ +/* + * 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.persistence; + +import jakarta.ws.rs.core.Response; +import org.apache.polaris.core.exceptions.PolarisException; + +/** + * Thrown when a persistence write may have committed but the client cannot confirm the outcome (for + * example a connection drop or timeout after the database applied an auto-commit update). Named to + * distinguish it from Iceberg's table-commit {@code CommitStateUnknownException}. + * + *

Callers must not treat this as a definite failure: newly written metadata files must not be + * deleted, and clients should not blindly retry the same commit. + * + *

Note: the simple class name is surfaced to clients as the {@code type} field of the REST error + * payload (see {@code PolarisExceptionMapper}). Renaming this class changes that wire value, which + * clients may use to recognize an unknown commit outcome. + */ +public class PersistenceCommitStateUnknownException extends PolarisException { + + public PersistenceCommitStateUnknownException(String message, Throwable cause) { + super(message, cause); + } + + public PersistenceCommitStateUnknownException(String message) { + super(message); + } + + @Override + public int httpStatusCode() { + // Iceberg REST: unknown commit outcome must be 5xx so clients do not treat it as cleanable. + return Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(); + } +} diff --git a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java index 2f3d998879..d55594ddc5 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java +++ b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java @@ -72,6 +72,7 @@ import org.apache.iceberg.encryption.EncryptionManager; import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.BadRequestException; +import org.apache.iceberg.exceptions.CleanableFailure; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.ForbiddenException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; @@ -119,6 +120,7 @@ import org.apache.polaris.core.entity.table.IcebergTableLikeEntity; import org.apache.polaris.core.exceptions.CommitConflictException; import org.apache.polaris.core.exceptions.PolarisServiceUnavailableException; +import org.apache.polaris.core.persistence.PersistenceCommitStateUnknownException; import org.apache.polaris.core.persistence.PolarisMetaStoreManager; import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper; import org.apache.polaris.core.persistence.dao.entity.BaseResult; @@ -1704,6 +1706,32 @@ public ViewBuilder withLocation(String newLocation) { } } + /** + * Whether newly written metadata files may be deleted after a table/view commit failure. + * + *

Follows Iceberg's rule: cleanup is only safe when the commit outcome is a known failure + * ({@link CleanableFailure} and related definite-failure types). When the outcome is unknown — + * connection loss after the metastore may already have applied the pointer update — the metadata + * file must be retained so a published pointer cannot dangle. + */ + @VisibleForTesting + static boolean shouldCleanupMetadataOnCommitFailure(Throwable failure) { + if (failure == null) { + return false; + } + // Never delete metadata when the commit may already have been applied. + if (failure instanceof PersistenceCommitStateUnknownException) { + return false; + } + if (failure instanceof CleanableFailure) { + return true; + } + // Definite pre- or post-CAS failures that do not implement CleanableFailure. + return failure instanceof AlreadyExistsException + || failure instanceof NotFoundException + || failure instanceof CommitConflictException; + } + /** * An implementation of {@link TableOperations} that integrates with {@link LocalIcebergCatalog}. * Much of this code was originally copied from {@link @@ -1967,6 +1995,8 @@ public void doCommit(TableMetadata base, TableMetadata metadata) { String newLocation = writeResult.location(); String oldLocation = base == null ? null : base.metadataFileLocation(); boolean writeSucceeded = false; + boolean persistenceAttempted = false; + RuntimeException commitFailure = null; try { // TODO: Consider using the entity from doRefresh() directly to do the conflict detection // instead of a two-layer CAS (checking metadataLocation to detect concurrent modification @@ -2032,6 +2062,7 @@ public void doCommit(TableMetadata base, TableMetadata metadata) { tableIdentifier, oldLocation, newLocation, existingLocation); } + persistenceAttempted = true; if (null == existingLocation) { createTableLike(tableIdentifier, entity, false); } else { @@ -2039,8 +2070,7 @@ public void doCommit(TableMetadata base, TableMetadata metadata) { } // We diverge from `BaseMetastoreTableOperations`: only update the in-memory state after // the metastore persistence succeeds. If we updated it before and persistence threw, - // the finally-block cleanup would delete newLocation while this ops instance still - // pointed at it — leaving a dangling reference until the caller refreshes. + // cleanup could delete newLocation while this ops instance still pointed at it. if (makeMetadataCurrentOnCommit) { currentMetadata = TableMetadata.buildFrom(metadata) @@ -2050,10 +2080,21 @@ public void doCommit(TableMetadata base, TableMetadata metadata) { currentMetadataLocation = newLocation; } writeSucceeded = true; + } catch (RuntimeException e) { + commitFailure = e; + throw e; } finally { + // Pre-write failures leave a definite orphan and are always cleaned up. Once the + // persistence call has been attempted, only delete on known failures: an ambiguous outcome + // (connection drop after the pointer may already have been applied) must leave the file in + // place — see Iceberg SnapshotProducer / TableOperations commit contract. if (!writeSucceeded && writeResult.written()) { - IcebergCatalogHandler.cleanupWrittenMetadataFiles( - List.of(new IcebergCatalogHandler.FileToDelete(io(), newLocation))); + boolean cleanup = + !persistenceAttempted || shouldCleanupMetadataOnCommitFailure(commitFailure); + if (cleanup) { + IcebergCatalogHandler.cleanupWrittenMetadataFiles( + List.of(new IcebergCatalogHandler.FileToDelete(io(), newLocation))); + } } } } @@ -2430,6 +2471,8 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) { String newLocation = writeResult.location(); String oldLocation = base == null ? null : currentMetadataLocation; boolean writeSucceeded = false; + boolean persistenceAttempted = false; + RuntimeException commitFailure = null; try { IcebergTableLikeEntity entity = IcebergTableLikeEntity.of( @@ -2465,6 +2508,7 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) { + "because it has been concurrently modified to %s", identifier, oldLocation, newLocation, existingLocation); } + persistenceAttempted = true; if (null == existingLocation) { createTableLike(identifier, entity, true); } else { @@ -2476,10 +2520,20 @@ public void doCommit(ViewMetadata base, ViewMetadata metadata) { currentMetadataLocation = newLocation; } writeSucceeded = true; + } catch (RuntimeException e) { + commitFailure = e; + throw e; } finally { + // Pre-write failures leave a definite orphan and are always cleaned up. Once the + // persistence call has been attempted, only delete on known failures (not unknown/ambiguous + // outcomes) so a possibly-published pointer cannot be left dangling. if (!writeSucceeded && writeResult.written()) { - IcebergCatalogHandler.cleanupWrittenMetadataFiles( - List.of(new IcebergCatalogHandler.FileToDelete(io(), newLocation))); + boolean cleanup = + !persistenceAttempted || shouldCleanupMetadataOnCommitFailure(commitFailure); + if (cleanup) { + IcebergCatalogHandler.cleanupWrittenMetadataFiles( + List.of(new IcebergCatalogHandler.FileToDelete(io(), newLocation))); + } } } } diff --git a/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitMetadataCleanupTest.java b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitMetadataCleanupTest.java new file mode 100644 index 0000000000..ae49edee50 --- /dev/null +++ b/runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/CommitMetadataCleanupTest.java @@ -0,0 +1,300 @@ +/* + * 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.sql.SQLException; +import java.util.EnumSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +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.AlreadyExistsException; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.NotFoundException; +import org.apache.iceberg.exceptions.ValidationException; +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.exceptions.CommitConflictException; +import org.apache.polaris.core.persistence.PersistenceCommitStateUnknownException; +import org.apache.polaris.core.persistence.PolarisMetaStoreManager; +import org.apache.polaris.core.persistence.dao.entity.BaseResult; +import org.apache.polaris.core.persistence.dao.entity.EntityResult; +import org.apache.polaris.service.TestServices; +import org.apache.polaris.service.catalog.AccessDelegationMode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mockito; + +/** + * Tests that single-table commit cleanup of newly written metadata files only runs for known + * failures, not when the metastore write may already have succeeded (ambiguous outcome). + */ +public class CommitMetadataCleanupTest { + private static final String NAMESPACE = "ns"; + private static final String CATALOG = "test-catalog"; + private static final String PROPERTY_NAME = "custom-property-1"; + private static final UUID IDEMPOTENCY_KEY = new UUID(116617318654508422L, -7820829973016961092L); + + @Test + void retainsMetadataWhenPersistenceThrowsAfterSuccessfulWrite(@TempDir Path tempDir) + throws Exception { + String location = catalogBaseLocation(tempDir); + 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 -> { + Object result = invocation.callRealMethod(); + if (shouldFail.get()) { + // Simulate: DB applied the CAS, then client lost the ack. + throw new RuntimeException( + "boom", new SQLException("Connection reset", "08006")); + } + return result; + }) + .when(spy) + .updateEntityPropertiesIfNotChanged( + Mockito.any(), Mockito.any(), Mockito.any()); + return spy; + }) + .build(); + + createCatalogAndNamespace(testServices, location); + String tableName = "ambiguous-cleanup-table"; + createTable(testServices, tableName, location); + TableIdentifier tableId = TableIdentifier.of(NAMESPACE, tableName); + + Set metadataFilesBefore = metadataFiles(tempDir); + shouldFail.set(true); + + assertThatThrownBy(() -> updateTable(testServices, tableId, "value-after")) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("boom"); + + Set metadataFilesAfter = metadataFiles(tempDir); + assertThat(metadataFilesAfter) + .as("New metadata must be retained when commit outcome may be success") + .hasSizeGreaterThan(metadataFilesBefore.size()); + assertThat(metadataFilesAfter).containsAll(metadataFilesBefore); + + // Catalog pointer was applied before the thrown error; load must still resolve. + shouldFail.set(false); + String metadataLocation = + testServices + .catalogAdapter() + .newHandler(testServices.securityContext(), CATALOG) + .loadTable( + tableId, "all", null, EnumSet.noneOf(AccessDelegationMode.class), Optional.empty()) + .get() + .tableMetadata() + .metadataFileLocation(); + assertThat(Path.of(java.net.URI.create(metadataLocation).getPath())).exists(); + } + + @Test + void cleansUpMetadataOnKnownConcurrentModificationFailure(@TempDir Path tempDir) + throws Exception { + String location = catalogBaseLocation(tempDir); + 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 EntityResult( + BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, + "simulated concurrent modification"); + } + return invocation.callRealMethod(); + }) + .when(spy) + .updateEntityPropertiesIfNotChanged( + Mockito.any(), Mockito.any(), Mockito.any()); + return spy; + }) + .build(); + + createCatalogAndNamespace(testServices, location); + String tableName = "known-failure-cleanup-table"; + createTable(testServices, tableName, location); + TableIdentifier tableId = TableIdentifier.of(NAMESPACE, tableName); + + Set metadataFilesBefore = metadataFiles(tempDir); + shouldFail.set(true); + + assertThatThrownBy(() -> updateTable(testServices, tableId, "value-conflict")) + .isInstanceOf(CommitConflictException.class); + + Set metadataFilesAfter = metadataFiles(tempDir); + assertThat(metadataFilesAfter) + .as("Orphan metadata from a known CAS failure should be cleaned up") + .isEqualTo(metadataFilesBefore); + } + + @ParameterizedTest + @MethodSource("cleanupDecisionCases") + void shouldCleanupMetadataOnCommitFailure(Throwable failure, boolean expectedCleanup) { + assertThat(LocalIcebergCatalog.shouldCleanupMetadataOnCommitFailure(failure)) + .isEqualTo(expectedCleanup); + } + + static Stream cleanupDecisionCases() { + return Stream.of( + Arguments.of(null, false), + Arguments.of(new RuntimeException("opaque boom"), false), + Arguments.of( + new RuntimeException("boom", new SQLException("Connection reset", "08006")), false), + Arguments.of( + new PersistenceCommitStateUnknownException( + "unknown", new SQLException("timeout", "57014")), + false), + Arguments.of( + new org.apache.iceberg.exceptions.CommitStateUnknownException( + new RuntimeException("db timeout")), + false), + Arguments.of(new CommitFailedException("concurrent metadata location"), true), + Arguments.of(new CommitConflictException("concurrent entity version"), true), + Arguments.of(new AlreadyExistsException("table exists"), true), + Arguments.of(new NotFoundException("missing"), true), + Arguments.of(new ValidationException("invalid"), true)); + } + + private static String catalogBaseLocation(Path tempDir) { + String location = tempDir.toAbsolutePath().toUri().toString(); + if (location.endsWith("/")) { + location = location.substring(0, location.length() - 1); + } + return location; + } + + 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 void updateTable(TestServices services, TableIdentifier tableId, String propertyValue) { + UpdateTableRequest request = + UpdateTableRequest.create( + tableId, + List.of(), + List.of(new MetadataUpdate.SetProperties(Map.of(PROPERTY_NAME, propertyValue)))); + services + .catalogAdapter() + .newHandler(services.securityContext(), CATALOG) + .updateTable(tableId, request); + } + + 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()); + } +} diff --git a/runtime/service/src/test/java/org/apache/polaris/service/exception/ExceptionMapperTest.java b/runtime/service/src/test/java/org/apache/polaris/service/exception/ExceptionMapperTest.java index 7ab4284a28..bcddd7c696 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/exception/ExceptionMapperTest.java +++ b/runtime/service/src/test/java/org/apache/polaris/service/exception/ExceptionMapperTest.java @@ -41,6 +41,7 @@ import org.apache.polaris.core.exceptions.FileIOUnknownHostException; import org.apache.polaris.core.exceptions.PolarisException; import org.apache.polaris.core.exceptions.PolarisServiceUnavailableException; +import org.apache.polaris.core.persistence.PersistenceCommitStateUnknownException; import org.apache.polaris.core.persistence.PolicyMappingAlreadyExistsException; import org.apache.polaris.core.policy.exceptions.NoSuchPolicyException; import org.apache.polaris.core.policy.exceptions.PolicyAttachException; @@ -112,6 +113,16 @@ public void testNamespaceException() { assertThat(response.getStatus()).isEqualTo(409); } + @Test + public void testCommitStateUnknownExceptionIsInternalServerError() { + PolarisExceptionMapper mapper = new PolarisExceptionMapper(); + Response response = + mapper.toResponse( + new PersistenceCommitStateUnknownException( + "commit outcome unknown", new RuntimeException("connection reset"))); + assertThat(response.getStatus()).isEqualTo(500); + } + @Test public void testServiceUnavailableWithRetryAfter() { PolarisExceptionMapper mapper = new PolarisExceptionMapper();