feat: support refreshing Iceberg tables

engine/table/src/main/java/io/deephaven/engine/table/impl/TableUpdateMode.java

@@ -15,7 +15,7 @@ public enum TableUpdateMode {
  STATIC, APPEND_ONLY, ADD_ONLY, ADD_REMOVE;
 
  /**
- * Returns true if the addition is allowed.
+ * Returns true if addition is allowed.
  */
  public boolean addAllowed() {
  switch (this) {
@@ -30,7 +30,7 @@ public boolean addAllowed() {
  }
 
  /**
- * Returns true if the removal is allowed.
+ * Returns true if removal is allowed.
  */
  public boolean removeAllowed() {
  switch (this) {
@@ -56,9 +56,6 @@ public boolean removeAllowed() {
  * @return the most permissive mode encountered in the stream
  */
  public static TableUpdateMode mostPermissiveMode(Stream<TableUpdateMode> modes) {
- // Analyze the location update modes of the input providers to determine the location update mode
- // of the composite provider. The resultant mode is the most permissive mode of the input provider
- // locations.
  final MutableBoolean anyRemoves = new MutableBoolean(false);
  final MutableBoolean anyAdditions = new MutableBoolean(false);
  final MutableBoolean anyAppends = new MutableBoolean(false);

engine/table/src/test/java/io/deephaven/engine/table/impl/AbstractTableLocationProviderTest.java

@@ -5,7 +5,6 @@
 
 import io.deephaven.api.SortColumn;
 import io.deephaven.base.verify.Assert;
-import io.deephaven.configuration.Configuration;
 import io.deephaven.engine.liveness.LiveSupplier;
 import io.deephaven.engine.liveness.StandaloneLivenessManager;
 import io.deephaven.engine.table.BasicDataIndex;
@@ -14,9 +13,7 @@
 import io.deephaven.engine.table.impl.locations.impl.AbstractTableLocationProvider;
 import io.deephaven.engine.table.impl.locations.local.URITableLocationKey;
 import io.deephaven.engine.testutil.testcase.RefreshingTableTestCase;
-import io.deephaven.io.logger.StreamLoggerImpl;
 import io.deephaven.test.types.OutOfBandTest;
-import io.deephaven.util.process.ProcessEnvironment;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.junit.Test;
@@ -92,17 +89,17 @@ protected void destroy() {
 
  @Override
  protected @NotNull ColumnLocation makeColumnLocation(@NotNull String name) {
- return null;
+ throw new UnsupportedOperationException();
  }
 
  @Override
  public @Nullable BasicDataIndex loadDataIndex(@NotNull String... columns) {
- return null;
+ throw new UnsupportedOperationException();
  }
 
  @Override
  public void refresh() {
- Assert.statementNeverExecuted();
+ throw new UnsupportedOperationException();
  }
 
  @Override
@@ -123,19 +120,10 @@ public boolean hasDataIndex(@NotNull String... columns) {
 
  @Override
  public void setUp() throws Exception {
- if (null == ProcessEnvironment.tryGet()) {
- ProcessEnvironment.basicServerInitialization(Configuration.getInstance(),
- "AbstractTableLocationProviderTest", new StreamLoggerImpl());
- }
  super.setUp();
  setExpectError(false);
  }
 
- @Override
- public void tearDown() throws Exception {
- super.tearDown();
- }
-
  private List<TableLocationKey> createKeys(final int count) {
  final List<TableLocationKey> keys = new ArrayList<>();
  for (int i = 0; i < count; i++) {

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

@@ -14,9 +14,6 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.jetbrains.annotations.NotNull;
 
-import java.util.HashSet;
-import java.util.Set;
-
 /**
  * <p>
  * Automatically refreshing {@link TableLocationProvider} implementation that delegates {@link TableLocationKey location
@@ -68,44 +65,24 @@ public String getImplementationName() {
 
  @Override
  public synchronized void refresh() {
- adapter.refresh();
- final Snapshot latestSnapshot = adapter.currentSnapshot();
- if (latestSnapshot.sequenceNumber() > locationKeyFinder.snapshot.sequenceNumber()) {
- locationKeyFinder.snapshot = latestSnapshot;
- refreshSnapshot();
+ if (locationKeyFinder.maybeUpdateSnapshot()) {
+ refreshLocations();
  }
  }
 
  @Override
  public void update() {
- throw new IllegalStateException("An automatically refreshing Iceberg table cannot be manually updated");
+ throw new UnsupportedOperationException("Automatically refreshing Iceberg tables cannot be manually updated");
  }
 
  @Override
  public void update(long snapshotId) {
- throw new IllegalStateException("An automatically refreshing Iceberg table cannot be manually updated");
+ throw new UnsupportedOperationException("Automatically refreshing Iceberg tables cannot be manually updated");
  }
 
  @Override
  public void update(Snapshot snapshot) {
- throw new IllegalStateException("An automatically refreshing Iceberg table cannot be manually updated");
- }
-
- /**
- * Refresh the table location provider with the latest snapshot from the catalog. This method will identify new
- * locations and removed locations.
- */
- private void refreshSnapshot() {
- beginTransaction(this);
- final Set<ImmutableTableLocationKey> missedKeys = new HashSet<>();
- getTableLocationKeys(ttlk -> missedKeys.add(ttlk.get()));
- locationKeyFinder.findKeys(tableLocationKey -> {
- missedKeys.remove(tableLocationKey);
- handleTableLocationKeyAdded(tableLocationKey, this);
- });
- missedKeys.forEach(tlk -> handleTableLocationKeyRemoved(tlk, this));
- endTransaction(this);
- setInitialized();
+ throw new UnsupportedOperationException("Automatically refreshing Iceberg tables cannot be manually updated");
  }
 
  // ------------------------------------------------------------------------------------------------------------------
@@ -114,7 +91,7 @@ private void refreshSnapshot() {
 
  @Override
  protected final void activateUnderlyingDataSource() {
- refreshSnapshot();
+ refresh();
  subscriptionToken = refreshService.scheduleTableLocationProviderRefresh(this, refreshIntervalMs);
  }
 

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

@@ -109,7 +109,7 @@ protected IcebergTableLocationKey locationKey(
  */
  public IcebergBaseLayout(
  @NotNull final IcebergTableAdapter tableAdapter,
- @NotNull final Snapshot tableSnapshot,
+ @Nullable final Snapshot tableSnapshot,
  @NotNull final IcebergInstructions instructions,
  @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) {
  this.tableAdapter = tableAdapter;
@@ -166,4 +166,62 @@ public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKe
  String.format("%s:%d - error finding Iceberg locations", tableAdapter, snapshot.snapshotId()), e);
  }
  }
+
+ /**
+ * Update the snapshot to the latest snapshot from the catalog if
+ */
+ protected boolean maybeUpdateSnapshot() {
+ final Snapshot latestSnapshot = tableAdapter.currentSnapshot();
+ if (latestSnapshot == null) {
+ return false;
+ }
+ if (snapshot == null || latestSnapshot.sequenceNumber() > snapshot.sequenceNumber()) {
+ snapshot = latestSnapshot;
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Update the snapshot to the user specified snapshot. This will fail with an {@link IllegalArgumentException} if
+ * the input snapshot is not newer (higher in sequence number) than the current snapshot or if the snapshot cannot
+ * be found.
+ *
+ * @param snapshotId the id of the snapshot to load
+ */
+ protected void updateSnapshot(long snapshotId) {
+ final List<Snapshot> snapshots = tableAdapter.listSnapshots();
+
+ final Snapshot snapshot = snapshots.stream()
+ .filter(s -> s.snapshotId() == snapshotId).findFirst()
+ .orElse(null);
+
+ if (snapshot == null) {
+ throw new IllegalArgumentException(
+ "Snapshot " + snapshotId + " was not found in the list of snapshots for table " + tableAdapter
+ + ". Snapshots: " + snapshots);
+ }
+ updateSnapshot(snapshot);
+ }
+
+ /**
+ * Update the snapshot to the user specified snapshot. This will fail with an {@link IllegalArgumentException} if
+ * the input snapshot is not newer (higher in sequence number) than the current snapshot.
+ *
+ * @param snapshot the snapshot to load
+ */
+ protected void updateSnapshot(Snapshot snapshot) {
+ // Validate that we are not trying to update to an older snapshot.
+ if (snapshot == null) {
+ throw new IllegalArgumentException("Input snapshot cannot be null");
+ }
+ if (this.snapshot != null && snapshot.sequenceNumber() <= this.snapshot.sequenceNumber()) {
+ throw new IllegalArgumentException(
+ "Update snapshot sequence number (" + snapshot.sequenceNumber()
+ + ") must be higher than the current snapshot sequence number ("
+ + this.snapshot.sequenceNumber() + ") for table " + tableAdapter);
+ }
+
+ this.snapshot = snapshot;
+ }
 }

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

@@ -10,6 +10,7 @@
 import io.deephaven.iceberg.util.IcebergTableAdapter;
 import org.apache.iceberg.*;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 
 import java.net.URI;
 
@@ -25,7 +26,7 @@ public final class IcebergFlatLayout extends IcebergBaseLayout {
  */
  public IcebergFlatLayout(
  @NotNull final IcebergTableAdapter tableAdapter,
- @NotNull final Snapshot tableSnapshot,
+ @Nullable final Snapshot tableSnapshot,
  @NotNull final IcebergInstructions instructions,
  @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) {
  super(tableAdapter, tableSnapshot, instructions, dataInstructionsProvider);

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

@@ -14,6 +14,7 @@
 import org.apache.commons.lang3.mutable.MutableInt;
 import org.apache.iceberg.*;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 
 import java.net.URI;
 import java.util.*;
@@ -46,7 +47,7 @@ public ColumnData(String name, Class<?> type, int index) {
  */
  public IcebergKeyValuePartitionedLayout(
  @NotNull final IcebergTableAdapter tableAdapter,
- @NotNull final org.apache.iceberg.Snapshot tableSnapshot,
+ @Nullable final Snapshot tableSnapshot,
  @NotNull final PartitionSpec partitionSpec,
  @NotNull final IcebergInstructions instructions,
  @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) {

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

@@ -12,10 +12,6 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.jetbrains.annotations.NotNull;
 
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
 /**
  * <p>
  * Manually refreshing {@link TableLocationProvider} implementation that delegates {@link TableLocationKey location key}
@@ -28,8 +24,6 @@ public class IcebergManualRefreshTableLocationProvider<TK extends TableKey, TLK
 
  private static final String IMPLEMENTATION_NAME = IcebergManualRefreshTableLocationProvider.class.getSimpleName();
 
- private boolean initialized = false;
-
  public IcebergManualRefreshTableLocationProvider(
  @NotNull final TK tableKey,
  @NotNull final IcebergBaseLayout locationKeyFinder,
@@ -58,62 +52,29 @@ public String getImplementationName() {
 
  @Override
  public synchronized void refresh() {
- // There should be no refresh service for this provider.
- throw new UnsupportedOperationException();
+ // There is no refresh service for this provider, but this is called as part of the initialization process.
+ refreshLocations();
  }
 
  @Override
  public synchronized void update() {
- adapter.refresh();
- update(adapter.currentSnapshot());
+ if (locationKeyFinder.maybeUpdateSnapshot()) {
+ refreshLocations();
+ }
  }
 
  @Override
  public synchronized void update(final long snapshotId) {
- adapter.refresh();
- final List<Snapshot> snapshots = adapter.listSnapshots();
-
- final Snapshot snapshot = snapshots.stream()
- .filter(s -> s.snapshotId() == snapshotId).findFirst()
- .orElse(null);
-
- if (snapshot == null) {
- throw new IllegalArgumentException(
- "Snapshot " + snapshotId + " was not found in the list of snapshots for table " + tableIdentifier
- + ". Snapshots: " + snapshots);
- }
- update(snapshot);
+ // delegate to the locationKeyFinder to update the snapshot
+ locationKeyFinder.updateSnapshot(snapshotId);
+ refreshLocations();
  }
 
  @Override
  public synchronized void update(final Snapshot snapshot) {
- // Verify that the input snapshot is newer (higher in sequence number) than the current snapshot.
- if (snapshot.sequenceNumber() <= locationKeyFinder.snapshot.sequenceNumber()) {
- throw new IllegalArgumentException(
- "Update snapshot sequence number (" + snapshot.sequenceNumber()
- + ") must be higher than the current snapshot sequence number ("
- + locationKeyFinder.snapshot.sequenceNumber() + ") for table " + tableIdentifier);
- }
  // Update the snapshot.
- locationKeyFinder.snapshot = snapshot;
- refreshSnapshot();
- }
-
- /**
- * Refresh the table location provider with the latest snapshot from the catalog. This method will identify new
- * locations and removed locations.
- */
- private void refreshSnapshot() {
- beginTransaction(this);
- final Set<ImmutableTableLocationKey> missedKeys = new HashSet<>();
- getTableLocationKeys(ttlk -> missedKeys.add(ttlk.get()));
- locationKeyFinder.findKeys(tlk -> {
- missedKeys.remove(tlk);
- handleTableLocationKeyAdded(tlk, this);
- });
- missedKeys.forEach(tlk -> handleTableLocationKeyRemoved(tlk, this));
- endTransaction(this);
- setInitialized();
+ locationKeyFinder.updateSnapshot(snapshot);
+ refreshLocations();
  }
 
  // ------------------------------------------------------------------------------------------------------------------
@@ -122,11 +83,8 @@ private void refreshSnapshot() {
 
  @Override
  protected void activateUnderlyingDataSource() {
- if (!initialized) {
- refreshSnapshot();
- activationSuccessful(this);
- initialized = true;
- }
+ ensureInitialized();
+ activationSuccessful(this);
  }
 
  @Override