feat: Ensure data read order reflects commit sequence in Iceberg tables

extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergBaseLayout.java

@@ -20,6 +20,8 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -63,6 +65,16 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder<Iceber
      */
     ParquetInstructions parquetInstructions;
 
+    /**
+     * The number of files discovered, used for ordering the keys.
+     */
+    private int fileCount;
+
+    /**
+     * Create a new {@link IcebergTableLocationKey} for the given file URI.
+     * <p>
+     * This method assumes that keys are created in a specific order and that each key is created only once.
+     */
     protected IcebergTableLocationKey locationKey(
             final org.apache.iceberg.FileFormat format,
             final URI fileUri,
@@ -96,7 +108,7 @@ protected IcebergTableLocationKey locationKey(
 
                 parquetInstructions = builder.build();
             }
-            return new IcebergTableParquetLocationKey(fileUri, 0, partitions, parquetInstructions);
+            return new IcebergTableParquetLocationKey(fileUri, fileCount++, partitions, parquetInstructions);
         }
         throw new UnsupportedOperationException(String.format("%s:%d - an unsupported file format %s for URI '%s'",
                 tableAdapter, snapshot.snapshotId(), format, fileUri));
@@ -137,9 +149,11 @@ public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKe
             return;
         }
         final Table table = tableAdapter.icebergTable();
+        final List<DataFile> dataFiles = new ArrayList<>();
         try {
             // Retrieve the manifest files from the snapshot
             final List<ManifestFile> manifestFiles = snapshot.allManifests(table.io());
+            // TODO(deephaven-core#5989: Add unit tests for the ordering of manifest files
             for (final ManifestFile manifestFile : manifestFiles) {
                 // Currently only can process manifest files with DATA content type.
                 if (manifestFile.content() != ManifestContent.DATA) {
@@ -148,20 +162,28 @@ public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKe
                                     table, snapshot.snapshotId(), manifestFile.content()));
                 }
                 try (final ManifestReader<DataFile> reader = ManifestFiles.read(manifestFile, table.io())) {
-                    for (DataFile df : reader) {
-                        final URI fileUri = dataFileUri(df);
-                        final IcebergTableLocationKey locationKey =
-                                cache.computeIfAbsent(fileUri, uri -> keyFromDataFile(df, fileUri));
-                        if (locationKey != null) {
-                            locationKeyObserver.accept(locationKey);
-                        }
-                    }
+                    reader.forEach(dataFiles::add);
                 }
             }
         } catch (final Exception e) {
             throw new TableDataException(
                     String.format("%s:%d - error finding Iceberg locations", tableAdapter, snapshot.snapshotId()), e);
         }
+
+        // Sort manifest files to read data files in the correct commit order
+        dataFiles.sort(Comparator
+                .comparingLong(DataFile::dataSequenceNumber)
+                .thenComparingLong(DataFile::fileSequenceNumber)
+                .thenComparingLong(DataFile::pos));
+
+        for (final DataFile df : dataFiles) {
+            final URI fileUri = dataFileUri(df);
+            final IcebergTableLocationKey locationKey =
+                    cache.computeIfAbsent(fileUri, uri -> keyFromDataFile(df, fileUri));
+            if (locationKey != null) {
+                locationKeyObserver.accept(locationKey);
+            }
+        }
     }
 
     /**

extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergUpdateMode.java

@@ -29,14 +29,16 @@ public static IcebergUpdateMode staticMode() {
     }
 
     /**
-     * Set a manually-refreshing update mode for this table.
+     * Set a manually-refreshing update mode for this table. The ordering of the data in the table will depend on the
+     * order in which the data files are discovered on refresh.
      */
     public static IcebergUpdateMode manualRefreshingMode() {
         return MANUAL_REFRESHING;
     }
 
     /**
-     * Set a automatically-refreshing update mode for this table using the default refresh interval of 60 seconds.
+     * Set a automatically-refreshing update mode for this table using the default refresh interval of 60 seconds. The
+     * ordering of the data in the table will depend on the order in which the data files are discovered on refresh.
      */
     public static IcebergUpdateMode autoRefreshingMode() {
         return AUTO_REFRESHING;