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

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/TableLocationKey.java

@@ -62,4 +62,16 @@ <PARTITION_VALUE_TYPE extends Comparable<PARTITION_VALUE_TYPE>> PARTITION_VALUE_
      * with any live TableLocation.
      */
     default void clear() {}
+
+    /**
+     * By default, compare fully qualified class names of the implementing classes. This method is a fallback where the
+     * implementing classes are not directly comparable, and should help establish a consistent ordering between
+     * distinct implementations.
+     * <p>
+     * {@inheritDoc}
+     */
+    @Override
+    default int compareTo(@NotNull final TableLocationKey other) {
+        return this.getClass().getName().compareTo(other.getClass().getName());
+    }
 }

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/impl/PartitionedTableLocationKey.java

@@ -5,6 +5,7 @@
 
 import io.deephaven.base.log.LogOutput;
 import io.deephaven.base.log.LogOutputAppendable;
+import io.deephaven.engine.table.impl.locations.TableLocationKey;
 import io.deephaven.util.compare.ObjectComparisons;
 import io.deephaven.engine.table.impl.locations.ImmutableTableLocationKey;
 import io.deephaven.engine.table.impl.locations.UnknownPartitionKeyException;
@@ -15,9 +16,9 @@
 
 /**
  * Base {@link ImmutableTableLocationKey} implementation for table locations that may be enclosed by partitions.
- * Sub-classes should be sure to invoke the partition-map comparator at higher priority than other comparisons when
- * implementing {@link #compareTo(Object)}, and to include the partitions in their {@link #equals(Object)}
- * implementations.
+ * Subclasses should consider invoking the partition-map comparator at higher priority than other comparisons when
+ * implementing {@link #compareTo(Object)}. Also, should include the partitions in their {@link #equals(Object)} and
+ * {@link #hashCode()} implementations if not calling {@code super.equals()} and {@code super.hashCode()}.
  */
 public abstract class PartitionedTableLocationKey implements ImmutableTableLocationKey {
 
@@ -26,6 +27,8 @@ public abstract class PartitionedTableLocationKey implements ImmutableTableLocat
 
     protected final Map<String, Comparable<?>> partitions;
 
+    private int cachedHashCode;
+
     /**
      * Construct a new PartitionedTableLocationKey for the supplied {@code partitions}.
      *
@@ -55,6 +58,46 @@ public final Set<String> getPartitionKeys() {
         return partitions.keySet();
     }
 
+    @Override
+    public int compareTo(@NotNull final TableLocationKey other) {
+        if (other instanceof PartitionedTableLocationKey) {
+            final PartitionedTableLocationKey otherTyped = (PartitionedTableLocationKey) other;
+            final int partitionComparisonResult =
+                    PartitionsComparator.INSTANCE.compare(partitions, otherTyped.partitions);
+            if (partitionComparisonResult != 0) {
+                return partitionComparisonResult;
+            }
+        }
+        return ImmutableTableLocationKey.super.compareTo(other);
+    }
+
+    @Override
+    public int hashCode() {
+        if (cachedHashCode == 0) {
+            final int computedHashCode = partitions.hashCode();
+            // Don't use 0; that's used by StandaloneTableLocationKey, and also our sentinel for the need to compute
+            if (computedHashCode == 0) {
+                final int fallbackHashCode = PartitionedTableLocationKey.class.hashCode();
+                cachedHashCode = fallbackHashCode == 0 ? 1 : fallbackHashCode;
+            } else {
+                cachedHashCode = computedHashCode;
+            }
+        }
+        return cachedHashCode;
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object other) {
+        if (this == other) {
+            return true;
+        }
+        if (!(other instanceof PartitionedTableLocationKey)) {
+            return false;
+        }
+        final PartitionedTableLocationKey otherTyped = (PartitionedTableLocationKey) other;
+        return partitions.equals(otherTyped.partitions);
+    }
+
     /**
      * Formats a map of partitions as key-value pairs.
      */

engine/table/src/main/java/io/deephaven/engine/table/impl/locations/impl/StandaloneTableLocationKey.java

@@ -48,7 +48,7 @@ public int compareTo(@NotNull final TableLocationKey other) {
         if (other instanceof StandaloneTableLocationKey) {
             return 0;
         }
-        throw new ClassCastException("Cannot compare " + getClass() + " to " + other.getClass());
+        return ImmutableTableLocationKey.super.compareTo(other);
     }
 
     @Override

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

@@ -66,10 +66,11 @@ public String toString() {
     }
 
     /**
-     * Precedence-wise this implementation compares {@code order}, then applies a {@link PartitionsComparator} to
-     * {@code partitions}, then compares {@code file}.
-     *
-     * @inheritDoc
+     * When comparing with another {@link FileTableLocationKey}, precedence-wise this implementation compares
+     * {@code order}, then applies a {@link PartitionsComparator} to {@code partitions}, then compares {@code file}.
+     * Otherwise, it delegates to parent class.
+     * <p>
+     * {@inheritDoc}
      */
     @Override
     public int compareTo(@NotNull final TableLocationKey other) {
@@ -86,7 +87,7 @@ public int compareTo(@NotNull final TableLocationKey other) {
             }
             return file.compareTo(otherTyped.file);
         }
-        throw new ClassCastException("Cannot compare " + getClass() + " to " + other.getClass());
+        return super.compareTo(other);
     }
 
     @Override

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;
 
@@ -72,10 +72,11 @@ public String toString() {
     }
 
     /**
-     * Precedence-wise this implementation compares {@code order}, then applies a {@link PartitionsComparator} to
-     * {@code partitions}, then compares {@code uri}.
-     *
-     * @inheritDoc
+     * When comparing with another {@link URITableLocationKey}, precedence-wise this implementation compares
+     * {@code order}, then applies a {@link PartitionsComparator} to {@code partitions}, then compares {@code uri}.
+     * Otherwise, it delegates to parent class.
+     * <p>
+     * {@inheritDoc}
      */
     @Override
     public int compareTo(@NotNull final TableLocationKey other) {
@@ -92,7 +93,7 @@ public int compareTo(@NotNull final TableLocationKey other) {
             }
             return uri.compareTo(otherTyped.uri);
         }
-        throw new ClassCastException("Cannot compare " + getClass() + " to " + other.getClass());
+        return super.compareTo(other);
     }
 
     @Override

engine/table/src/test/java/io/deephaven/engine/table/impl/locations/impl/SimpleTableLocationKey.java

@@ -4,8 +4,6 @@
 package io.deephaven.engine.table.impl.locations.impl;
 
 import io.deephaven.base.log.LogOutput;
-import io.deephaven.engine.table.impl.locations.TableLocationKey;
-import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
 import java.util.Map;
@@ -31,23 +29,4 @@ public LogOutput append(LogOutput logOutput) {
         return logOutput.append(getImplementationName()).append("[partitions=")
                 .append(PartitionsFormatter.INSTANCE, partitions).append(']');
     }
-
-    @Override
-    public int compareTo(@NotNull final TableLocationKey other) {
-        if (other instanceof SimpleTableLocationKey) {
-            return PartitionsComparator.INSTANCE.compare(partitions, ((SimpleTableLocationKey) other).partitions);
-        }
-        throw new ClassCastException("Cannot compare " + getClass() + " to " + other.getClass());
-    }
-
-    @Override
-    public int hashCode() {
-        return partitions.hashCode();
-    }
-
-    @Override
-    public boolean equals(final Object other) {
-        return other == this || (other instanceof SimpleTableLocationKey
-                && partitions.equals(((SimpleTableLocationKey) other).partitions));
-    }
 }

engine/test-utils/src/main/java/io/deephaven/engine/testutil/locations/TableBackedTableLocationKey.java

@@ -46,10 +46,16 @@ public String toString() {
 
     @Override
     public int compareTo(@NotNull final TableLocationKey other) {
-        // noinspection DataFlowIssue
-        return Integer.compare(
-                (int) table.getAttribute("ID"),
-                (int) ((TableBackedTableLocationKey) other).table.getAttribute("ID"));
+        if (other instanceof TableBackedTableLocationKey) {
+            final TableBackedTableLocationKey otherTyped = (TableBackedTableLocationKey) other;
+            // noinspection DataFlowIssue
+            final int idComparisonResult =
+                    Integer.compare((int) table.getAttribute("ID"), (int) otherTyped.table.getAttribute("ID"));
+            if (idComparisonResult != 0) {
+                return idComparisonResult;
+            }
+        }
+        return ImmutableTableLocationKey.super.compareTo(other);
     }
 
     @Override

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/PythonTableDataService.java

@@ -704,15 +704,6 @@ public int hashCode() {
             return cachedHashCode;
         }
 
-        @Override
-        public int compareTo(@NotNull final TableLocationKey other) {
-            if (getClass() != other.getClass()) {
-                throw new ClassCastException(String.format("Cannot compare %s to %s", getClass(), other.getClass()));
-            }
-            final TableLocationKeyImpl otherTableLocationKey = (TableLocationKeyImpl) other;
-            return PartitionsComparator.INSTANCE.compare(partitions, otherTableLocationKey.partitions);
-        }
-
         @Override
         public LogOutput append(@NotNull final LogOutput logOutput) {
             return logOutput.append(getImplementationName())

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

@@ -17,14 +17,15 @@
 import io.deephaven.util.channel.SeekableChannelsProvider;
 import io.deephaven.util.channel.SeekableChannelsProviderLoader;
 import org.apache.iceberg.*;
+import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.io.FileIO;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
 import java.net.URI;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.function.Consumer;
 
 public abstract class IcebergBaseLayout implements TableLocationKeyFinder<IcebergTableLocationKey> {
@@ -33,6 +34,21 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder<Iceber
      */
     final IcebergTableAdapter tableAdapter;
 
+    /**
+     * The UUID of the table, if available.
+     */
+    @Nullable
+    private final UUID tableUuid;
+
+    /**
+     * Name of the {@link Catalog} used to access this table.
+     */
+    private final String catalogName;
+
+    /**
+     * The table identifier string.
+     */
+    private final String tableIdentifier;
     /**
      * The {@link TableDefinition} that will be used for life of this table. Although Iceberg table schema may change,
      * schema changes are not supported in Deephaven.
@@ -45,9 +61,9 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder<Iceber
     private final String uriScheme;
 
     /**
-     * A cache of {@link IcebergTableLocationKey IcebergTableLocationKeys} keyed by the URI of the file they represent.
+     * The {@link Snapshot} from which to discover data files.
      */
-    private final Map<URI, IcebergTableLocationKey> cache;
+    Snapshot snapshot;
 
     /**
      * The {@link ParquetInstructions} object that will be used to read any Parquet data files in this table.
@@ -60,17 +76,28 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder<Iceber
      */
     private final SeekableChannelsProvider channelsProvider;
 
+
     /**
-     * The {@link Snapshot} from which to discover data files.
+     * Create a new {@link IcebergTableLocationKey} for the given {@link ManifestFile}, {@link DataFile} and
+     * {@link URI}.
+     *
+     * @param manifestFile The manifest file from which the data file was discovered
+     * @param dataFile The data file that backs the keyed location
+     * @param fileUri The {@link URI} for the file that backs the keyed location
+     * @param partitions The table partitions enclosing the table location keyed by the returned key. If {@code null},
+     *        the location will be a member of no partitions.
+     *
+     * @return A new {@link IcebergTableLocationKey}
      */
-    Snapshot snapshot;
-
     protected IcebergTableLocationKey locationKey(
-            final org.apache.iceberg.FileFormat format,
-            final URI fileUri,
+            @NotNull final ManifestFile manifestFile,
+            @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) {
-            return new IcebergTableParquetLocationKey(fileUri, 0, partitions, parquetInstructions, channelsProvider);
+            return new IcebergTableParquetLocationKey(tableUuid, catalogName, tableIdentifier, manifestFile, dataFile,
+                    fileUri, 0, partitions, parquetInstructions, channelsProvider);
         }
         throw new UnsupportedOperationException(String.format("%s:%d - an unsupported file format %s for URI '%s'",
                 tableAdapter, snapshot.snapshotId(), format, fileUri));
@@ -85,6 +112,20 @@ public IcebergBaseLayout(
             @NotNull final IcebergReadInstructions instructions,
             @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) {
         this.tableAdapter = tableAdapter;
+        {
+            UUID uuid;
+            try {
+                uuid = tableAdapter.icebergTable().uuid();
+            } catch (final RuntimeException e) {
+                // The UUID method is unsupported for v1 Iceberg tables since uuid is optional for v1 tables.
+                uuid = null;
+            }
+            this.tableUuid = uuid;
+        }
+
+        this.catalogName = tableAdapter.catalog().name();
+        this.tableIdentifier = tableAdapter.tableIdentifier().toString();
+
         this.snapshot = tableAdapter.getSnapshot(instructions);
         this.tableDef = tableAdapter.definition(instructions);
         this.uriScheme = locationUri(tableAdapter.icebergTable()).getScheme();
@@ -111,10 +152,9 @@ public IcebergBaseLayout(
             this.parquetInstructions = builder.build();
         }
         this.channelsProvider = SeekableChannelsProviderLoader.getInstance().load(uriScheme, specialInstructions);
-        this.cache = new HashMap<>();
     }
 
-    abstract IcebergTableLocationKey keyFromDataFile(DataFile df, URI fileUri);
+    abstract IcebergTableLocationKey keyFromDataFile(ManifestFile manifestFile, DataFile dataFile, URI fileUri);
 
     private static String path(String path, FileIO io) {
         return io instanceof RelativeFileIO ? ((RelativeFileIO) io).absoluteLocation(path) : path;
@@ -145,15 +185,14 @@ 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(table, df);
+                    for (final DataFile dataFile : reader) {
+                        final URI fileUri = dataFileUri(table, dataFile);
                         if (!uriScheme.equals(fileUri.getScheme())) {
                             throw new TableDataException(String.format(
                                     "%s:%d - multiple URI schemes are not currently supported. uriScheme=%s, fileUri=%s",
                                     table, snapshot.snapshotId(), uriScheme, fileUri));
                         }
-                        final IcebergTableLocationKey locationKey =
-                                cache.computeIfAbsent(fileUri, uri -> keyFromDataFile(df, fileUri));
+                        final IcebergTableLocationKey locationKey = keyFromDataFile(manifestFile, dataFile, fileUri);
                         if (locationKey != null) {
                             locationKeyObserver.accept(locationKey);
                         }

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

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