fix: return full commit log when a commit tail is shorter than the fetch page

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.

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)

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);
           }
 

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<SimpleCommitTestObj> toList(Iterator<SimpleCommitTestObj> iterator) {
     var result = new ArrayList<SimpleCommitTestObj>();
     iterator.forEachRemaining(result::add);