diff --git a/CHANGELOG.md b/CHANGELOG.md index 0d8b1c9dfc..0896cedde9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -60,6 +60,7 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti ### Deprecations ### Fixes +- The NoSQL persistence commit log (`Commits.commitLog`) no longer stops early when a commit's recent-ancestor tail is shorter than the internal fetch page size. With a `polaris.persistence.reference-previous-head-count` smaller than the page size, the natural-order commit log previously truncated at the first short tail because trailing null entries in the fetch page were treated as end-of-history, which could also drop still-referenced objects during maintenance. - Async task execution (table cleanup, manifest and batch file cleanup) now retries when a handler returns false on transient errors (e.g. IO or delete failures). Previously `false` was swallowed with only a warning log and the task was never retried via the existing retry mechanism. - `TableCleanupTaskHandler` now clamps `TABLE_METADATA_CLEANUP_BATCH_SIZE` to at least 1. Previously a non-positive realm override caused an infinite loop (0) or `IllegalArgumentException` (<0) when splitting metadata files for cleanup. - `ManifestFileCleanupTaskHandler` now handles Iceberg v2 delete manifests in addition to data manifests. Previously, `DROP TABLE PURGE` on a v2 table that had been updated via merge-on-read DML left position-delete files and their manifests as orphans in object storage; the cleanup task failed silently because `ManifestFiles.read()` rejects delete manifests. diff --git a/persistence/nosql/persistence/impl/build.gradle.kts b/persistence/nosql/persistence/impl/build.gradle.kts index 58711d0e97..28b4a207a2 100644 --- a/persistence/nosql/persistence/impl/build.gradle.kts +++ b/persistence/nosql/persistence/impl/build.gradle.kts @@ -83,6 +83,7 @@ dependencies { testFixturesCompileOnly(project(":polaris-immutables")) testFixturesAnnotationProcessor(project(":polaris-immutables", configuration = "processor")) + testFixturesImplementation(project(":polaris-idgen-api")) testFixturesImplementation(libs.guava) testFixturesImplementation(libs.junit.pioneer) diff --git a/persistence/nosql/persistence/impl/src/main/java/org/apache/polaris/persistence/nosql/impl/commits/CommitsImpl.java b/persistence/nosql/persistence/impl/src/main/java/org/apache/polaris/persistence/nosql/impl/commits/CommitsImpl.java index 3edc12779b..985e3bddc1 100644 --- a/persistence/nosql/persistence/impl/src/main/java/org/apache/polaris/persistence/nosql/impl/commits/CommitsImpl.java +++ b/persistence/nosql/persistence/impl/src/main/java/org/apache/polaris/persistence/nosql/impl/commits/CommitsImpl.java @@ -216,8 +216,9 @@ protected C computeNext() { } lastCommit = null; - var ids = new ObjRef[REVERSE_COMMIT_FETCH_SIZE]; - for (var i = 0; i < REVERSE_COMMIT_FETCH_SIZE && i < tail.length; i++) { + var pageSize = Math.min(REVERSE_COMMIT_FETCH_SIZE, tail.length); + var ids = new ObjRef[pageSize]; + for (var i = 0; i < pageSize; i++) { ids[i] = objRef(type, tail[i], 1); } diff --git a/persistence/nosql/persistence/impl/src/testFixtures/java/org/apache/polaris/persistence/nosql/impl/commits/BaseTestCommitLogImpl.java b/persistence/nosql/persistence/impl/src/testFixtures/java/org/apache/polaris/persistence/nosql/impl/commits/BaseTestCommitLogImpl.java index b440078359..823f41d132 100644 --- a/persistence/nosql/persistence/impl/src/testFixtures/java/org/apache/polaris/persistence/nosql/impl/commits/BaseTestCommitLogImpl.java +++ b/persistence/nosql/persistence/impl/src/testFixtures/java/org/apache/polaris/persistence/nosql/impl/commits/BaseTestCommitLogImpl.java @@ -18,15 +18,22 @@ */ package org.apache.polaris.persistence.nosql.impl.commits; +import static java.util.function.Function.identity; + import java.util.ArrayList; import java.util.Collections; import java.util.Iterator; import java.util.List; import java.util.Optional; import java.util.OptionalLong; +import java.util.UUID; import java.util.stream.Collectors; import java.util.stream.IntStream; +import org.apache.polaris.ids.api.IdGenerator; +import org.apache.polaris.ids.api.MonotonicClock; import org.apache.polaris.persistence.nosql.api.Persistence; +import org.apache.polaris.persistence.nosql.api.PersistenceParams; +import org.apache.polaris.persistence.nosql.api.backend.Backend; import org.apache.polaris.persistence.nosql.testextension.PersistenceTestExtension; import org.apache.polaris.persistence.nosql.testextension.PolarisPersistence; import org.assertj.core.api.SoftAssertions; @@ -41,6 +48,9 @@ public abstract class BaseTestCommitLogImpl { @InjectSoftAssertions protected SoftAssertions soft; @PolarisPersistence protected Persistence persistence; + @PolarisPersistence protected Backend backend; + @PolarisPersistence protected MonotonicClock clock; + @PolarisPersistence protected IdGenerator idGenerator; @ParameterizedTest @ValueSource(ints = {0, 1, 3, 19, 20, 21, 39, 40, 41, 255}) @@ -131,6 +141,66 @@ public void commitLogOffsets(int offsetIndex, TestInfo testInfo) throws Exceptio .containsExactlyElementsOf(chronological); } + /** + * When a commit's {@code tail} is shorter than the number of commits fetched per page, the + * "natural" commit log must not stop early. This exercises {@code referencePreviousHeadCount} + * values below the internal reverse-fetch page size, so pages contain fewer entries than the page + * size. That previously caused {@code commitLog} to truncate the history at the first commit of a + * short tail. + */ + @ParameterizedTest + @ValueSource(ints = {2, 3, 5}) + public void commitLogShortTail(int referencePreviousHeadCount, TestInfo testInfo) + throws Exception { + var refName = + testInfo.getTestMethod().orElseThrow().getName() + "-" + referencePreviousHeadCount; + var numCommits = 50; + + var reducedPersistence = + backend.newPersistence( + identity(), + PersistenceParams.BuildablePersistenceParams.builder() + .referencePreviousHeadCount(referencePreviousHeadCount) + .build(), + UUID.randomUUID().toString(), + clock, + idGenerator); + + reducedPersistence.createReference(refName, Optional.empty()); + + var committer = + reducedPersistence.createCommitter(refName, SimpleCommitTestObj.class, String.class); + for (int i = 0; i < numCommits; i++) { + var payload = "commit #" + i; + committer.commit( + (state, refObjSupplier) -> + state.commitResult( + "foo", + ImmutableSimpleCommitTestObj.builder().payload(payload), + refObjSupplier.get())); + } + + var commits = reducedPersistence.commits(); + var expectedPayloads = + IntStream.range(0, numCommits).mapToObj(i -> "commit #" + i).collect(Collectors.toList()); + + // "reversed" (most recent commit last) already used a growing page list and returned the full + // history; assert it as a baseline. + soft.assertThatIterator(commits.commitLogReversed(refName, 0L, SimpleCommitTestObj.class)) + .toIterable() + .extracting(SimpleCommitTestObj::payload) + .containsExactlyElementsOf(expectedPayloads); + + // "natural" (most recent commit first) must return every commit, not stop at the first short + // tail. + Collections.reverse(expectedPayloads); + soft.assertThatIterator( + commits.commitLog(refName, OptionalLong.empty(), SimpleCommitTestObj.class)) + .toIterable() + .extracting(SimpleCommitTestObj::payload) + .containsExactlyElementsOf(expectedPayloads); + } + private static List toList(Iterator iterator) { var result = new ArrayList(); iterator.forEachRemaining(result::add);