[Core] Add mergeAppendTest to ensure consist distribution of data files in manifests

core/src/test/java/org/apache/iceberg/TestBase.java

@@ -29,6 +29,7 @@
 import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -784,18 +785,34 @@ static Iterator<ManifestEntry.Status> statuses(ManifestEntry.Status... statuses)
     return Iterators.forArray(statuses);
   }
 
+  static Iterator<ManifestEntry.Status> statusesRepeat(ManifestEntry.Status status, int count) {
+    return Iterators.limit(Iterators.cycle(Collections.singletonList(status)), count);
+  }
+
   static Iterator<Long> dataSeqs(Long... seqs) {
     return Iterators.forArray(seqs);
   }
 
+  static Iterator<Long> dataSeqsRepeat(Long value, int count) {
+    return Iterators.limit(Iterators.cycle(Collections.singletonList(value)), count);
+  }
+
   static Iterator<Long> fileSeqs(Long... seqs) {
     return Iterators.forArray(seqs);
   }
 
+  static Iterator<Long> fileSeqsRepeat(Long value, int count) {
+    return Iterators.limit(Iterators.cycle(Collections.singletonList(value)), count);
+  }
+
   static Iterator<Long> ids(Long... ids) {
     return Iterators.forArray(ids);
   }
 
+  static Iterator<Long> idsRepeat(Long value, int count) {
+    return Iterators.limit(Iterators.cycle(Collections.singletonList(value)), count);
+  }
+
   static Iterator<DataFile> files(DataFile... files) {
     return Iterators.forArray(files);
   }

core/src/test/java/org/apache/iceberg/TestMergeAppend.java

@@ -82,6 +82,56 @@ public void testAddManyFiles() {
     validateTableFiles(table, dataFiles);
   }
 
+  @TestTemplate
+  public void testAddManyFilesWithConsistentOrdering() {
+    assertThat(listManifestFiles()).as("Table should start empty").isEmpty();
+
+    int multiplier = 3;
+    int groupSize = SnapshotProducer.MIN_FILE_GROUP_SIZE;
+    List<DataFile> dataFiles = Lists.newArrayList();
+
+    for (int ordinal = 0; ordinal < multiplier * groupSize; ordinal++) {
+      StructLike partition = Row.of(ordinal % 2);
+      DataFile dataFile = FileGenerationUtil.generateDataFile(table, partition);
+      dataFiles.add(dataFile);
+    }
+
+    AppendFiles append = table.newAppend();
+    dataFiles.forEach(append::appendFile);
+    append.commit();
+
+    Snapshot snapshot = table.currentSnapshot();
+    List<ManifestFile> manifestsInSnapshot = snapshot.allManifests(table.io());
+    assertThat(manifestsInSnapshot).hasSize(multiplier);
+
+    // expect 1st manifest to contain first third of the data files
+    validateManifest(
+        manifestsInSnapshot.get(0),
+        dataSeqsRepeat(1L, groupSize),
+        fileSeqsRepeat(1L, groupSize),
+        idsRepeat(snapshot.snapshotId(), groupSize),
+        dataFiles.subList(0, groupSize).iterator(),
+        statusesRepeat(Status.ADDED, groupSize));
+
+    // expect 2nd manifest to contain second third of the data files
+    validateManifest(
+        manifestsInSnapshot.get(1),
+        dataSeqsRepeat(1L, groupSize),
+        fileSeqsRepeat(1L, groupSize),
+        idsRepeat(snapshot.snapshotId(), groupSize),
+        dataFiles.subList(groupSize, groupSize * 2).iterator(),
+        statusesRepeat(Status.ADDED, groupSize));
+
+    // expect 3rd manifest to contain last third of the data files
+    validateManifest(
+        manifestsInSnapshot.get(2),
+        dataSeqsRepeat(1L, groupSize),
+        fileSeqsRepeat(1L, groupSize),
+        idsRepeat(snapshot.snapshotId(), groupSize),
+        dataFiles.subList(groupSize * 2, groupSize * 3).iterator(),
+        statusesRepeat(Status.ADDED, groupSize));
+  }
+
   @TestTemplate
   public void testEmptyTableAppend() {
     assertThat(listManifestFiles()).isEmpty();