Initial commit of Iceberg integration.

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

@@ -54,7 +54,8 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder<Iceber
     final Map<URI, IcebergTableLocationKey> cache;
 
     /**
-     * The {@link ParquetInstructions} object that will be used to read any Parquet data files in this table.
+     * 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;
 
@@ -68,6 +69,9 @@ protected IcebergTableLocationKey locationKey(
                 // Start with user-supplied instructions (if provided).
                 final ParquetInstructions.Builder builder = new ParquetInstructions.Builder();
 
+                // Add the table definition.
+                builder.setTableDefinition(tableDef);
+
                 // Add any column rename mappings.
                 if (!instructions.columnRenameMap().isEmpty()) {
                     for (Map.Entry<String, String> entry : instructions.columnRenameMap().entrySet()) {
@@ -76,9 +80,8 @@ protected IcebergTableLocationKey locationKey(
                 }
 
                 // Add the S3 instructions.
-                if (instructions.s3Instructions().isPresent()) {
-                    builder.setSpecialInstructions(instructions.s3Instructions().get());
-                }
+                instructions.s3Instructions().ifPresent(builder::setSpecialInstructions);
+
                 parquetInstructions = builder.build();
             }
             return new IcebergTableParquetLocationKey(fileUri, 0, partitions, parquetInstructions);

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

@@ -22,8 +22,21 @@
  * a {@link Snapshot}
  */
 public final class IcebergKeyValuePartitionedLayout extends IcebergBaseLayout {
-    private final String[] partitionColumns;
-    private final Class<?>[] partitionColumnTypes;
+    private final String[] inputPartitionColumnNames;
+
+    private class ColumnData {
+        final String name;
+        final Class<?> type;
+        final int index;
+
+        public ColumnData(String name, Class<?> type, int index) {
+            this.name = name;
+            this.type = type;
+            this.index = index;
+        }
+    }
+
+    private final List<ColumnData> outputPartitionColumns;
 
     /**
      * @param tableDef The {@link TableDefinition} that will be used for the table.
@@ -37,14 +50,33 @@ public IcebergKeyValuePartitionedLayout(
             @NotNull final org.apache.iceberg.Table table,
             @NotNull final org.apache.iceberg.Snapshot tableSnapshot,
             @NotNull final FileIO fileIO,
+            @NotNull final PartitionSpec partitionSpec,
             @NotNull final IcebergInstructions instructions) {
         super(tableDef, table, tableSnapshot, fileIO, instructions);
 
-        partitionColumns =
-                tableDef.getPartitioningColumns().stream().map(ColumnDefinition::getName).toArray(String[]::new);
-        partitionColumnTypes = Arrays.stream(partitionColumns)
-                .map(colName -> TypeUtils.getBoxedType(tableDef.getColumn(colName).getDataType()))
-                .toArray(Class<?>[]::new);
+        // Get the list of (potentially renamed) columns on which the Iceberg table is partitioned. This will be the
+        // order of the values in DataFile.partition() collection.
+        inputPartitionColumnNames = partitionSpec.fields().stream()
+                .map(PartitionField::name)
+                .map(col -> instructions.columnRenameMap().getOrDefault(col, col))
+                .toArray(String[]::new);
+
+        outputPartitionColumns = new ArrayList<>();
+
+        // Get the list of columns in the table definition that are included in the partition columns.
+        final List<String> outputCols = tableDef.getColumnNames();
+        outputCols.retainAll(List.of(inputPartitionColumnNames));
+
+        for (String col : outputCols) {
+            // Is this so inefficient that it's worth it to use a map?
+            for (int i = 0; i < inputPartitionColumnNames.length; i++) {
+                if (inputPartitionColumnNames[i].equals(col)) {
+                    outputPartitionColumns
+                            .add(new ColumnData(col, TypeUtils.getBoxedType(tableDef.getColumn(col).getDataType()), i));
+                    break;
+                }
+            }
+        }
     }
 
     @Override
@@ -57,14 +89,15 @@ IcebergTableLocationKey keyFromDataFile(DataFile df, URI fileUri) {
         final Map<String, Comparable<?>> partitions = new LinkedHashMap<>();
 
         final PartitionData partitionData = (PartitionData) df.partition();
-        for (int ii = 0; ii < partitionColumns.length; ++ii) {
-            final Object value = partitionData.get(ii);
-            if (value != null && !value.getClass().isAssignableFrom(partitionColumnTypes[ii])) {
-                throw new TableDataException("Partitioning column " + partitionColumns[ii]
+        for (ColumnData colData : outputPartitionColumns) {
+            final String colName = colData.name;
+            final Object value = partitionData.get(colData.index);
+            if (value != null && !value.getClass().isAssignableFrom(colData.type)) {
+                throw new TableDataException("Partitioning column " + colName
                         + " has type " + value.getClass().getName()
-                        + " but expected " + partitionColumnTypes[ii].getName());
+                        + " but expected " + colData.type.getName());
             }
-            partitions.put(partitionColumns[ii], (Comparable<?>) value);
+            partitions.put(colName, (Comparable<?>) value);
         }
         return locationKey(df.format(), fileUri, partitions);
     }

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

@@ -7,6 +7,9 @@
 import io.deephaven.engine.table.impl.locations.TableLocation;
 import io.deephaven.engine.table.impl.locations.impl.TableLocationFactory;
 import io.deephaven.engine.table.impl.locations.util.TableDataRefreshService;
+import io.deephaven.parquet.table.ParquetInstructions;
+import io.deephaven.parquet.table.location.ParquetTableLocation;
+import io.deephaven.parquet.table.location.ParquetTableLocationKey;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -22,7 +25,8 @@ public TableLocation makeLocation(@NotNull final TableKey tableKey,
             @NotNull final IcebergTableLocationKey locationKey,
             @Nullable final TableDataRefreshService refreshService) {
         if (locationKey instanceof IcebergTableParquetLocationKey) {
-            return new IcebergTableParquetLocation(tableKey, locationKey);
+            return new ParquetTableLocation(tableKey, (ParquetTableLocationKey) locationKey,
+                    (ParquetInstructions) locationKey.readInstructions());
         }
         throw new UnsupportedOperationException("Unsupported location key type: " + locationKey.getClass());
     }

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

@@ -45,7 +45,7 @@ public String getImplementationName() {
     }
 
     @Override
-    public Object readInstructions() {
+    public ParquetInstructions readInstructions() {
         return readInstructions;
     }
 }