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

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/local/URITableLocationKey.java

@@ -28,7 +28,7 @@ public class URITableLocationKey extends PartitionedTableLocationKey {
     private static final String IMPLEMENTATION_NAME = URITableLocationKey.class.getSimpleName();
 
     protected final URI uri;
-    private final int order;
+    protected final int order;
 
     private int cachedHashCode;
 

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

@@ -20,6 +20,7 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.net.URI;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -61,13 +62,16 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder<Iceber
      * The {@link ParquetInstructions} object that will be used to read any Parquet data files in this table. Only
      * accessed while synchronized on {@code this}.
      */
-    ParquetInstructions parquetInstructions;
+    private ParquetInstructions parquetInstructions;
 
+    /**
+     * Create a new {@link IcebergTableLocationKey} for the given {@link DataFile} and {@link URI}.
+     */
     protected IcebergTableLocationKey locationKey(
-            final org.apache.iceberg.FileFormat format,
-            final URI fileUri,
+            @NotNull final DataFile dataFile,
+            @NotNull final URI fileUri,
             @Nullable final Map<String, Comparable<?>> partitions) {
-
+        final org.apache.iceberg.FileFormat format = dataFile.format();
         if (format == org.apache.iceberg.FileFormat.PARQUET) {
             if (parquetInstructions == null) {
                 // Start with user-supplied instructions (if provided).
@@ -78,7 +82,7 @@ protected IcebergTableLocationKey locationKey(
 
                 // Add any column rename mappings.
                 if (!instructions.columnRenames().isEmpty()) {
-                    for (Map.Entry<String, String> entry : instructions.columnRenames().entrySet()) {
+                    for (final Map.Entry<String, String> entry : instructions.columnRenames().entrySet()) {
                         builder.addColumnNameMapping(entry.getKey(), entry.getValue());
                     }
                 }
@@ -96,7 +100,7 @@ protected IcebergTableLocationKey locationKey(
 
                 parquetInstructions = builder.build();
             }
-            return new IcebergTableParquetLocationKey(fileUri, 0, partitions, parquetInstructions);
+            return new IcebergTableParquetLocationKey(dataFile, fileUri, 0, partitions, parquetInstructions);
         }
         throw new UnsupportedOperationException(String.format("%s:%d - an unsupported file format %s for URI '%s'",
                 tableAdapter, snapshot.snapshotId(), format, fileUri));
@@ -140,6 +144,9 @@ public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKe
         try {
             // Retrieve the manifest files from the snapshot
             final List<ManifestFile> manifestFiles = snapshot.allManifests(table.io());
+            // Sort manifest files by sequence number to read data files in the correct commit order
+            manifestFiles.sort(Comparator.comparingLong(ManifestFile::sequenceNumber));
+            // 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,10 +155,10 @@ 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);
+                    for (final DataFile dataFile : reader) {
+                        final URI fileUri = dataFileUri(dataFile);
                         final IcebergTableLocationKey locationKey =
-                                cache.computeIfAbsent(fileUri, uri -> keyFromDataFile(df, fileUri));
+                                cache.computeIfAbsent(fileUri, uri -> keyFromDataFile(dataFile, fileUri));
                         if (locationKey != null) {
                             locationKeyObserver.accept(locationKey);
                         }

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

@@ -35,7 +35,7 @@ public String toString() {
     }
 
     @Override
-    IcebergTableLocationKey keyFromDataFile(DataFile df, URI fileUri) {
-        return locationKey(df.format(), fileUri, null);
+    IcebergTableLocationKey keyFromDataFile(final DataFile dataFile, final URI fileUri) {
+        return locationKey(dataFile, fileUri, null);
     }
 }

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

@@ -80,10 +80,10 @@ public String toString() {
     }
 
     @Override
-    IcebergTableLocationKey keyFromDataFile(DataFile df, URI fileUri) {
+    IcebergTableLocationKey keyFromDataFile(@NotNull final DataFile dataFile, @NotNull final URI fileUri) {
         final Map<String, Comparable<?>> partitions = new LinkedHashMap<>();
 
-        final PartitionData partitionData = (PartitionData) df.partition();
+        final PartitionData partitionData = (PartitionData) dataFile.partition();
         for (final ColumnData colData : outputPartitioningColumns) {
             final String colName = colData.name;
             final Object colValue = partitionData.get(colData.index);
@@ -94,6 +94,6 @@ IcebergTableLocationKey keyFromDataFile(DataFile df, URI fileUri) {
             }
             partitions.put(colName, (Comparable<?>) colValue);
         }
-        return locationKey(df.format(), fileUri, partitions);
+        return locationKey(dataFile, fileUri, partitions);
     }
 }

extensions/iceberg/src/main/java/io/deephaven/iceberg/location/IcebergTableParquetLocationKey.java

@@ -6,6 +6,7 @@
 import io.deephaven.engine.table.impl.locations.TableLocationKey;
 import io.deephaven.parquet.table.ParquetInstructions;
 import io.deephaven.parquet.table.location.ParquetTableLocationKey;
+import org.apache.iceberg.DataFile;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -18,24 +19,36 @@
 public class IcebergTableParquetLocationKey extends ParquetTableLocationKey implements IcebergTableLocationKey {
     private static final String IMPLEMENTATION_NAME = IcebergTableParquetLocationKey.class.getSimpleName();
 
+    private final long dataSequenceNumber;
+    private final long fileSequenceNumber;
+    private final long pos;
+
     private final ParquetInstructions readInstructions;
 
     /**
      * Construct a new IcebergTableParquetLocationKey for the supplied {@code fileUri} and {@code partitions}.
      *
-     * @param fileUri The file that backs the keyed location
+     * @param dataFile The data file that backs the keyed location
+     * @param fileUri The {@link URI} for the file that backs the keyed location
      * @param order Explicit ordering index, taking precedence over other fields
      * @param partitions The table partitions enclosing the table location keyed by {@code this}. Note that if this
      *        parameter is {@code null}, the location will be a member of no partitions. An ordered copy of the map will
      *        be made, so the calling code is free to mutate the map after this call
      * @param readInstructions the instructions for customizations while reading
      */
     public IcebergTableParquetLocationKey(
+            @NotNull final DataFile dataFile,
             @NotNull final URI fileUri,
             final int order,
             @Nullable final Map<String, Comparable<?>> partitions,
             @NotNull final ParquetInstructions readInstructions) {
         super(fileUri, order, partitions, readInstructions);
+
+        // Following are used for ordering of data files. Files with unknown sequence numbers should be ordered last.
+        dataSequenceNumber = dataFile.dataSequenceNumber() != null ? dataFile.dataSequenceNumber() : Long.MAX_VALUE;
+        fileSequenceNumber = dataFile.fileSequenceNumber() != null ? dataFile.fileSequenceNumber() : Long.MAX_VALUE;
+        pos = dataFile.pos() != null ? dataFile.pos() : Long.MAX_VALUE;
+
         this.readInstructions = readInstructions;
     }
 
@@ -48,4 +61,36 @@ public String getImplementationName() {
     public ParquetInstructions readInstructions() {
         return readInstructions;
     }
+
+    /**
+     * Precedence-wise this implementation compares {@code order}, then {@code dataSequenceNumber}, then
+     * {@code fileSequenceNumber}, then {@code pos}, then applies a {@link PartitionsComparator} to {@code partitions}
+     * and finally {@code uri}.
+     *
+     * @inheritDoc
+     */
+    @Override
+    public int compareTo(@NotNull final TableLocationKey other) {
+        if (other instanceof IcebergTableParquetLocationKey) {
+            final IcebergTableParquetLocationKey otherTyped = (IcebergTableParquetLocationKey) other;
+            int comparisonResult;
+            if ((comparisonResult = Integer.compare(order, otherTyped.order)) != 0) {
+                return comparisonResult;
+            }
+            if ((comparisonResult = Long.compare(dataSequenceNumber, otherTyped.dataSequenceNumber)) != 0) {
+                return comparisonResult;
+            }
+            if ((comparisonResult = Long.compare(fileSequenceNumber, otherTyped.fileSequenceNumber)) != 0) {
+                return comparisonResult;
+            }
+            if ((comparisonResult = Long.compare(pos, otherTyped.pos)) != 0) {
+                return comparisonResult;
+            }
+            if ((comparisonResult = PartitionsComparator.INSTANCE.compare(partitions, otherTyped.partitions)) != 0) {
+                return comparisonResult;
+            }
+            return uri.compareTo(otherTyped.uri);
+        }
+        throw new ClassCastException("Cannot compare " + getClass() + " to " + other.getClass());
+    }
 }

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;