Initial commit of Iceberg integration.

extensions/iceberg/build.gradle

@@ -0,0 +1,65 @@
+plugins {
+    id 'java-library'
+    id 'io.deephaven.project.register'
+}
+
+description 'Iceberg: Support to read iceberg catalogs.'
+
+ext {
+    icebergVersion = '1.4.3'
+    hadoopVersion = '3.3.6'
+}
+
+dependencies {
+    api project(':engine-api')
+    api project(':engine-stringset')
+    api project(':engine-table')
+
+    implementation project(':engine-table')
+    implementation project(':engine-base')
+    implementation project(':log-factory')
+    implementation project(':Configuration')
+    implementation depTrove3
+
+    compileOnly depAnnotations
+
+    Classpaths.inheritAutoService(project)
+    Classpaths.inheritImmutables(project)
+
+    Classpaths.inheritJUnitPlatform(project)
+    Classpaths.inheritAssertJ(project)
+
+    Classpaths.inheritParquetHadoop(project)
+    implementation project(':extensions-parquet-base')
+    implementation project(':extensions-parquet-table')
+    implementation project(':extensions-s3')
+
+    implementation "org.apache.iceberg:iceberg-api:${icebergVersion}"
+    implementation "org.apache.iceberg:iceberg-core:${icebergVersion}"
+    implementation "org.apache.iceberg:iceberg-core:${icebergVersion}:tests"
+    implementation "org.apache.iceberg:iceberg-aws:${icebergVersion}"
+    // implementation "org.apache.iceberg:iceberg-azure:${icebergVersion}"
+    // implementation "org.apache.iceberg:iceberg-gcp:${icebergVersion}"
+    implementation "org.apache.iceberg:iceberg-bundled-guava:${icebergVersion}"
+
+    implementation "org.apache.hadoop:hadoop-common:${hadoopVersion}"
+    implementation "org.apache.hadoop:hadoop-hdfs-client:${hadoopVersion}"
+
+    runtimeOnly "org.apache.iceberg:iceberg-aws-bundle:${icebergVersion}"
+    // runtimeOnly "org.apache.iceberg:iceberg-azure-bundle:${icebergVersion}"
+    // runtimeOnly "org.apache.iceberg:iceberg-gcp-bundle:${icebergVersion}"
+
+    testImplementation depCommonsCompress
+    testImplementation depCommonsIo
+    testImplementation project(':engine-test-utils'),
+            TestTools.projectDependency(project, 'engine-base')
+
+    Classpaths.inheritJUnitClassic(project, 'testImplementation')
+
+    testRuntimeOnly project(':log-to-slf4j'),
+            project(path: ':configs'),
+            project(path: ':test-configs')
+    Classpaths.inheritSlf4j(project, 'slf4j-simple', 'testRuntimeOnly')
+}
+
+TestTools.addEngineOutOfBandTest(project)

extensions/iceberg/gradle.properties

@@ -0,0 +1 @@
+io.deephaven.project.ProjectType=JAVA_PUBLIC

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

@@ -0,0 +1,80 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.iceberg.layout;
+
+import io.deephaven.base.FileUtils;
+import io.deephaven.engine.table.impl.locations.TableDataException;
+import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder;
+import io.deephaven.iceberg.location.IcebergTableLocationKey;
+import org.apache.iceberg.*;
+import org.apache.iceberg.io.FileIO;
+import org.jetbrains.annotations.NotNull;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+/**
+ * Iceberg {@link TableLocationKeyFinder location finder} for tables without partitions that will discover data files
+ * from a {@link org.apache.iceberg.Snapshot}
+ */
+public final class IcebergFlatLayout implements TableLocationKeyFinder<IcebergTableLocationKey> {
+    private final Snapshot tableSnapshot;
+    private final FileIO fileIO;
+    private final Map<URI, IcebergTableLocationKey> cache;
+    private final Object readInstructions;
+
+    private static IcebergTableLocationKey locationKey(
+            final FileFormat format,
+            final URI fileUri,
+            @NotNull final Object readInstructions) {
+        return new IcebergTableLocationKey(format, fileUri, 0, null, readInstructions);
+    }
+
+    /**
+     * @param tableSnapshot The {@link Snapshot} from which to discover data files.
+     * @param readInstructions the instructions for customizations while reading.
+     */
+    public IcebergFlatLayout(
+            @NotNull final Snapshot tableSnapshot,
+            @NotNull final FileIO fileIO,
+            @NotNull final Object readInstructions) {
+        this.tableSnapshot = tableSnapshot;
+        this.fileIO = fileIO;
+        this.readInstructions = readInstructions;
+
+        this.cache = new HashMap<>();
+    }
+
+    public String toString() {
+        return IcebergFlatLayout.class.getSimpleName() + '[' + tableSnapshot + ']';
+    }
+
+    @Override
+    public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKey> locationKeyObserver) {
+        try {
+            // Retrieve the manifest files from the snapshot
+            final List<ManifestFile> manifestFiles = tableSnapshot.allManifests(fileIO);
+            for (final ManifestFile manifestFile : manifestFiles) {
+                final ManifestReader<DataFile> reader = ManifestFiles.read(manifestFile, fileIO);
+                for (DataFile df : reader) {
+                    final URI fileUri = FileUtils.convertToURI(df.path().toString(), false);
+                    IcebergTableLocationKey locationKey = cache.get(fileUri);
+                    if (locationKey == null) {
+                        locationKey = locationKey(df.format(), fileUri, readInstructions);
+                        if (!locationKey.verifyFileReader()) {
+                            continue;
+                        }
+                        cache.put(fileUri, locationKey);
+                    }
+                    locationKeyObserver.accept(locationKey);
+                }
+            }
+        } catch (final Exception e) {
+            throw new TableDataException("Error finding Iceberg locations under " + tableSnapshot, e);
+        }
+    }
+}

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

@@ -0,0 +1,92 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.iceberg.layout;
+
+import io.deephaven.base.FileUtils;
+import io.deephaven.engine.table.impl.locations.TableDataException;
+import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder;
+import io.deephaven.iceberg.location.IcebergTableLocationKey;
+import org.apache.iceberg.*;
+import org.apache.iceberg.io.FileIO;
+import org.jetbrains.annotations.NotNull;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+/**
+ * Iceberg {@link TableLocationKeyFinder location finder} for tables without partitions that will discover data files
+ * from a {@link org.apache.iceberg.Snapshot}
+ */
+public final class IcebergKeyValuePartitionedLayout implements TableLocationKeyFinder<IcebergTableLocationKey> {
+    private final Snapshot tableSnapshot;
+    private final FileIO fileIO;
+    private final String[] partitionColumns;
+    private final Object readInstructions;
+    private final Map<URI, IcebergTableLocationKey> cache;
+
+    private static IcebergTableLocationKey locationKey(
+            final FileFormat format,
+            final URI fileUri,
+            final Map<String, Comparable<?>> partitions,
+            @NotNull final Object readInstructions) {
+        return new IcebergTableLocationKey(format, fileUri, 0, partitions, readInstructions);
+    }
+
+    /**
+     * @param tableSnapshot The {@link Snapshot} from which to discover data files.
+     * @param fileIO The file IO to use for reading manifest data files.
+     * @param partitionColumns The columns to use for partitioning.
+     * @param readInstructions The instructions for customizations while reading.
+     */
+    public IcebergKeyValuePartitionedLayout(
+            @NotNull final Snapshot tableSnapshot,
+            @NotNull final FileIO fileIO,
+            @NotNull final String[] partitionColumns,
+            @NotNull final Object readInstructions) {
+        this.tableSnapshot = tableSnapshot;
+        this.fileIO = fileIO;
+        this.partitionColumns = partitionColumns;
+        this.readInstructions = readInstructions;
+
+        this.cache = new HashMap<>();
+    }
+
+    public String toString() {
+        return IcebergFlatLayout.class.getSimpleName() + '[' + tableSnapshot + ']';
+    }
+
+    @Override
+    public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKey> locationKeyObserver) {
+        final Map<String, Comparable<?>> partitions = new LinkedHashMap<>();
+        try {
+            // Retrieve the manifest files from the snapshot
+            final List<ManifestFile> manifestFiles = tableSnapshot.allManifests(fileIO);
+            for (final ManifestFile manifestFile : manifestFiles) {
+                final ManifestReader<DataFile> reader = ManifestFiles.read(manifestFile, fileIO);
+                for (DataFile df : reader) {
+                    final URI fileUri = FileUtils.convertToURI(df.path().toString(), false);
+                    IcebergTableLocationKey locationKey = cache.get(fileUri);
+                    if (locationKey == null) {
+                        final PartitionData partitionData = (PartitionData) df.partition();
+                        for (int ii = 0; ii < partitionColumns.length; ++ii) {
+                            partitions.put(partitionColumns[ii], (Comparable<?>) partitionData.get(ii));
+                        }
+                        locationKey = locationKey(df.format(), fileUri, partitions, readInstructions);
+                        if (!locationKey.verifyFileReader()) {
+                            continue;
+                        }
+                        cache.put(fileUri, locationKey);
+                    }
+                    locationKeyObserver.accept(locationKey);
+                }
+            }
+        } catch (final Exception e) {
+            throw new TableDataException("Error finding Iceberg locations under " + tableSnapshot, e);
+        }
+    }
+}

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

@@ -0,0 +1,131 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.iceberg.location;
+
+import io.deephaven.api.SortColumn;
+import io.deephaven.base.verify.Require;
+import io.deephaven.engine.rowset.RowSet;
+import io.deephaven.engine.table.BasicDataIndex;
+import io.deephaven.engine.table.impl.locations.*;
+import io.deephaven.engine.table.impl.locations.impl.AbstractTableLocation;
+import io.deephaven.parquet.table.ParquetInstructions;
+import io.deephaven.parquet.table.location.ParquetTableLocation;
+import io.deephaven.parquet.table.location.ParquetTableLocationKey;
+import org.apache.iceberg.FileFormat;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.util.List;
+
+public class IcebergTableLocation implements TableLocation {
+
+    private static final String IMPLEMENTATION_NAME = IcebergTableLocation.class.getSimpleName();
+
+    private final ImmutableTableKey tableKey;
+    private final ImmutableTableLocationKey tableLocationKey;
+
+    private final AbstractTableLocation internalTableLocation;
+
+    public IcebergTableLocation(@NotNull final TableKey tableKey,
+            @NotNull final IcebergTableLocationKey tableLocationKey,
+            @NotNull final Object readInstructions) {
+        this.tableKey = Require.neqNull(tableKey, "tableKey").makeImmutable();
+        this.tableLocationKey = Require.neqNull(tableLocationKey, "tableLocationKey").makeImmutable();
+
+        if (tableLocationKey.format == FileFormat.PARQUET) {
+            this.internalTableLocation = new ParquetTableLocation(tableKey,
+                    (ParquetTableLocationKey) tableLocationKey.internalTableLocationKey,
+                    (ParquetInstructions) readInstructions);
+        } else {
+            throw new IllegalArgumentException("Unsupported file format: " + tableLocationKey.format);
+        }
+    }
+
+    @Override
+    public String getImplementationName() {
+        return IMPLEMENTATION_NAME;
+    }
+
+    @Override
+    @NotNull
+    public ImmutableTableKey getTableKey() {
+        return tableKey;
+    }
+
+    @Override
+    @NotNull
+    public ImmutableTableLocationKey getKey() {
+        return tableLocationKey;
+    }
+
+    @Override
+    public boolean supportsSubscriptions() {
+        return internalTableLocation.supportsSubscriptions();
+    }
+
+    @Override
+    public void subscribe(@NotNull Listener listener) {
+        internalTableLocation.subscribe(listener);
+    }
+
+    @Override
+    public void unsubscribe(@NotNull Listener listener) {
+        internalTableLocation.unsubscribe(listener);
+    }
+
+    @Override
+    public void refresh() {
+        internalTableLocation.refresh();
+    }
+
+    @Override
+    @NotNull
+    public List<SortColumn> getSortedColumns() {
+        return internalTableLocation.getSortedColumns();
+    }
+
+    @Override
+    @NotNull
+    public List<String[]> getDataIndexColumns() {
+        return internalTableLocation.getDataIndexColumns();
+    }
+
+    @Override
+    public boolean hasDataIndex(@NotNull String... columns) {
+        return internalTableLocation.hasDataIndex(columns);
+    }
+
+    @Override
+    @Nullable
+    public BasicDataIndex getDataIndex(@NotNull String... columns) {
+        return internalTableLocation.getDataIndex(columns);
+    }
+
+    @Override
+    @NotNull
+    public ColumnLocation getColumnLocation(@NotNull CharSequence name) {
+        return internalTableLocation.getColumnLocation(name);
+    }
+
+    @Override
+    @NotNull
+    public Object getStateLock() {
+        return internalTableLocation.getStateLock();
+    }
+
+    @Override
+    public RowSet getRowSet() {
+        return internalTableLocation.getRowSet();
+    }
+
+    @Override
+    public long getSize() {
+        return internalTableLocation.getSize();
+    }
+
+    @Override
+    public long getLastModifiedTimeMillis() {
+        return internalTableLocation.getLastModifiedTimeMillis();
+    }
+}