Initial commit of Iceberg integration.

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

@@ -7,7 +7,6 @@
 import io.deephaven.engine.table.impl.locations.TableDataException;
 import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder;
 import io.deephaven.iceberg.location.IcebergTableLocationKey;
-import io.deephaven.util.channel.SeekableChannelsProvider;
 import org.apache.iceberg.*;
 import org.apache.iceberg.io.FileIO;
 import org.jetbrains.annotations.NotNull;
@@ -23,7 +22,7 @@
  * Iceberg {@link TableLocationKeyFinder location finder} for tables without partitions that will discover data files
  * from a {@link org.apache.iceberg.Snapshot}
  */
-public final class IcebergPartitionedLayout implements TableLocationKeyFinder<IcebergTableLocationKey> {
+public final class IcebergKeyValuePartitionedLayout implements TableLocationKeyFinder<IcebergTableLocationKey> {
     private final Snapshot tableSnapshot;
     private final FileIO fileIO;
     private final String[] partitionColumns;
@@ -44,7 +43,7 @@ private static IcebergTableLocationKey locationKey(
      * @param partitionColumns The columns to use for partitioning.
      * @param readInstructions The instructions for customizations while reading.
      */
-    public IcebergPartitionedLayout(
+    public IcebergKeyValuePartitionedLayout(
             @NotNull final Snapshot tableSnapshot,
             @NotNull final FileIO fileIO,
             @NotNull final String[] partitionColumns,

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

@@ -25,7 +25,7 @@ public class IcebergTableLocation implements TableLocation {
     private final ImmutableTableKey tableKey;
     private final ImmutableTableLocationKey tableLocationKey;
 
-    AbstractTableLocation internalTableLocation;
+    private final AbstractTableLocation internalTableLocation;
 
     public IcebergTableLocation(@NotNull final TableKey tableKey,
             @NotNull final IcebergTableLocationKey tableLocationKey,
@@ -54,7 +54,8 @@ public ImmutableTableKey getTableKey() {
     }
 
     @Override
-    public @NotNull ImmutableTableLocationKey getKey() {
+    @NotNull
+    public ImmutableTableLocationKey getKey() {
         return tableLocationKey;
     }
 
@@ -79,12 +80,14 @@ public void refresh() {
     }
 
     @Override
-    public @NotNull List<SortColumn> getSortedColumns() {
+    @NotNull
+    public List<SortColumn> getSortedColumns() {
         return internalTableLocation.getSortedColumns();
     }
 
     @Override
-    public @NotNull List<String[]> getDataIndexColumns() {
+    @NotNull
+    public List<String[]> getDataIndexColumns() {
         return internalTableLocation.getDataIndexColumns();
     }
 
@@ -94,17 +97,20 @@ public boolean hasDataIndex(@NotNull String... columns) {
     }
 
     @Override
-    public @Nullable BasicDataIndex getDataIndex(@NotNull String... columns) {
+    @Nullable
+    public BasicDataIndex getDataIndex(@NotNull String... columns) {
         return internalTableLocation.getDataIndex(columns);
     }
 
     @Override
-    public @NotNull ColumnLocation getColumnLocation(@NotNull CharSequence name) {
+    @NotNull
+    public ColumnLocation getColumnLocation(@NotNull CharSequence name) {
         return internalTableLocation.getColumnLocation(name);
     }
 
     @Override
-    public @NotNull Object getStateLock() {
+    @NotNull
+    public Object getStateLock() {
         return internalTableLocation.getStateLock();
     }
 

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

@@ -31,8 +31,8 @@ public IcebergTableLocationFactory(@NotNull final Object readInstructions) {
     public TableLocation makeLocation(@NotNull final TableKey tableKey,
             @NotNull final IcebergTableLocationKey locationKey,
             @Nullable final TableDataRefreshService refreshService) {
-        final URI parquetFileURI = locationKey.getURI();
-        if (!FILE_URI_SCHEME.equals(parquetFileURI.getScheme()) || new File(parquetFileURI).exists()) {
+        final URI fileURI = locationKey.getURI();
+        if (!FILE_URI_SCHEME.equals(fileURI.getScheme()) || new File(fileURI).exists()) {
             return new IcebergTableLocation(tableKey, locationKey, readInstructions);
         } else {
             return new NonexistentTableLocation(tableKey, locationKey);

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

@@ -22,7 +22,6 @@ public class IcebergTableLocationKey extends URITableLocationKey {
     private static final String IMPLEMENTATION_NAME = IcebergTableLocationKey.class.getSimpleName();
 
     final FileFormat format;
-
     final URITableLocationKey internalTableLocationKey;
 
     /**
@@ -59,19 +58,7 @@ public String getImplementationName() {
     }
 
     /**
-     * Returns {@code true} if a previous file reader has been created, or if one was successfully created on-demand.
-     *
-     * <p>
-     * When {@code false}, this may mean that the file:
-     * <ol>
-     * <li>does not exist, or is otherwise inaccessible</li>
-     * <li>is in the process of being written, and is not yet a valid parquet file</li>
-     * <li>is _not_ a parquet file</li>
-     * <li>is a corrupt parquet file</li>
-     * </ol>
-     *
-     *
-     * @return true if the file reader exists or was successfully created
+     * See {@link ParquetTableLocationKey#verifyFileReader()}.
      */
     public synchronized boolean verifyFileReader() {
         if (format == FileFormat.PARQUET) {

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

@@ -17,7 +17,7 @@
 import io.deephaven.extensions.s3.Credentials;
 import io.deephaven.extensions.s3.S3Instructions;
 import io.deephaven.iceberg.layout.IcebergFlatLayout;
-import io.deephaven.iceberg.layout.IcebergPartitionedLayout;
+import io.deephaven.iceberg.layout.IcebergKeyValuePartitionedLayout;
 import io.deephaven.iceberg.location.IcebergTableLocationFactory;
 import io.deephaven.iceberg.location.IcebergTableLocationKey;
 import io.deephaven.parquet.table.ParquetInstructions;
@@ -44,13 +44,13 @@ public class IcebergCatalog {
     private final S3Instructions s3Instructions;
 
     /**
-     * Construct an IcebergCatalog given a set of configurable instructions..
+     * Construct an IcebergCatalog given a set of configurable instructions.
      *
      * @param name The optional service name
      */
     IcebergCatalog(final @Nullable String name, final IcebergInstructions instructions) {
         // Set up the properties map for the Iceberg catalog
-        Map<String, String> properties = new HashMap<>();
+        final Map<String, String> properties = new HashMap<>();
 
         final Configuration conf = new Configuration();
 
@@ -181,7 +181,7 @@ private Table readTableInternal(
                     partitionSpec.fields().stream().map(PartitionField::name).toArray(String[]::new);
 
             // Create the partitioning column location key finder
-            keyFinder = new IcebergPartitionedLayout(
+            keyFinder = new IcebergKeyValuePartitionedLayout(
                     snapshot,
                     fileIO,
                     partitionColumns,

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

@@ -5,6 +5,7 @@
 
 import io.deephaven.engine.table.ColumnDefinition;
 import io.deephaven.engine.table.TableDefinition;
+import io.deephaven.engine.table.impl.locations.TableDataException;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -35,20 +36,14 @@ static TableDefinition fromSchema(final Schema schema, PartitionSpec partitionSp
         for (final Types.NestedField field : schema.columns()) {
             final String name = field.name();
             final Type type = field.type();
-            try {
-                final io.deephaven.qst.type.Type<?> qstType = convertPrimitiveType(type);
-                final ColumnDefinition<?> column;
-                if (partitionNames.contains(name)) {
-                    column = ColumnDefinition.of(name, qstType).withPartitioning();
-                } else {
-                    column = ColumnDefinition.of(name, qstType);
-                }
-                columns.add(column);
-            } catch (UnsupportedOperationException e) {
-                // TODO: Currently will silently skip the column. Would it be better to skip and warn the user or
-                // break and declare failure? We don't have a mechanism for skipping columns, do we need an overload
-                // with a supplied table definition?
+            final io.deephaven.qst.type.Type<?> qstType = convertPrimitiveType(type);
+            final ColumnDefinition<?> column;
+            if (partitionNames.contains(name)) {
+                column = ColumnDefinition.of(name, qstType).withPartitioning();
+            } else {
+                column = ColumnDefinition.of(name, qstType);
             }
+            columns.add(column);
         }
 
         return TableDefinition.of(columns);
@@ -84,7 +79,9 @@ static io.deephaven.qst.type.Type<?> convertPrimitiveType(final Type icebergType
                 return io.deephaven.qst.type.Type.find(byte[].class);
             }
         }
-        throw new UnsupportedOperationException("Unsupported type: " + typeId);
+        throw new TableDataException(
+                "Unsupported iceberg column type " + typeId.name() +
+                        " with logical type " + typeId.javaClass());
     }
 
     private IcebergTools() {}

extensions/iceberg/src/test/java/io.deephaven.iceberg/IcebergToolsTest.java

@@ -32,6 +32,11 @@ public void setUp() {
                 .build();
     }
 
+    @Test
+    public void testNothing() {
+        // Dummy to prevent JUnit from complaining about no tests
+    }
+
     // TODO: discuss how to perform tests since they require a full MiniIO + Iceberg setup
 
     // @Test