feat: support refreshing Iceberg tables

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

@@ -7,10 +7,7 @@
 import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.rowset.TrackingWritableRowSet;
 import io.deephaven.engine.rowset.WritableRowSet;
-import io.deephaven.engine.table.ColumnSource;
-import io.deephaven.engine.table.DataIndex;
-import io.deephaven.engine.table.Table;
-import io.deephaven.engine.table.TableListener;
+import io.deephaven.engine.table.*;
 import io.deephaven.engine.table.impl.locations.ImmutableTableLocationKey;
 import io.deephaven.engine.table.impl.locations.TableLocation;
 import org.jetbrains.annotations.NotNull;
@@ -53,7 +50,7 @@ public interface ColumnSourceManager extends LivenessReferent {
  *
  * @return The set of added row keys, to be owned by the caller
  */
- WritableRowSet refresh();
+ TableUpdate refresh();
 
  /**
  * Advise this ColumnSourceManager that an error has occurred, and that it will no longer be {@link #refresh()

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

@@ -7,18 +7,15 @@
 import io.deephaven.base.verify.Require;
 import io.deephaven.engine.liveness.LivenessScopeStack;
 import io.deephaven.engine.rowset.TrackingWritableRowSet;
-import io.deephaven.engine.table.Table;
 import io.deephaven.engine.table.TableDefinition;
+import io.deephaven.engine.table.TableUpdate;
 import io.deephaven.engine.table.TableUpdateListener;
 import io.deephaven.engine.table.impl.locations.ImmutableTableLocationKey;
 import io.deephaven.engine.table.impl.locations.TableDataException;
 import io.deephaven.engine.table.impl.locations.TableLocationProvider;
-import io.deephaven.engine.table.impl.locations.TableLocationRemovedException;
 import io.deephaven.engine.updategraph.UpdateSourceRegistrar;
 import io.deephaven.engine.table.impl.perf.QueryPerformanceRecorder;
 import io.deephaven.engine.table.impl.locations.impl.TableLocationSubscriptionBuffer;
-import io.deephaven.engine.rowset.RowSet;
-import io.deephaven.engine.rowset.RowSetFactory;
 import io.deephaven.util.SafeCloseable;
 import io.deephaven.util.annotations.TestUseOnly;
 import org.apache.commons.lang3.mutable.Mutable;
@@ -106,7 +103,6 @@ public abstract class SourceTable<IMPL_TYPE extends SourceTable<IMPL_TYPE>> exte
  }
 
  setRefreshing(isRefreshing);
- setAttribute(Table.ADD_ONLY_TABLE_ATTRIBUTE, Boolean.TRUE);
  }
 
  /**
@@ -219,10 +215,6 @@ protected void instrumentedRefresh() {
  final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate = locationBuffer.processPending();
  final ImmutableTableLocationKey[] removedKeys =
  maybeRemoveLocations(locationUpdate.getPendingRemovedLocationKeys());
- if (removedKeys.length > 0) {
- throw new TableLocationRemovedException("Source table does not support removed locations",
- removedKeys);
- }
  maybeAddLocations(locationUpdate.getPendingAddedLocationKeys());
 
  // This class previously had functionality to notify "location listeners", but it was never used.
@@ -232,13 +224,16 @@ protected void instrumentedRefresh() {
  return;
  }
 
- final RowSet added = columnSourceManager.refresh();
- if (added.isEmpty()) {
+ final TableUpdate update = columnSourceManager.refresh();
+ if (update.empty()) {
+ update.release();
  return;
  }
 
- rowSet.insert(added);
- notifyListeners(added, RowSetFactory.empty(), RowSetFactory.empty());
+ Assert.eqTrue(update.shifted().empty(), "update.shifted().empty()");
+ rowSet.remove(update.removed());
+ rowSet.insert(update.added());
+ notifyListeners(update);
  }
 
  @Override

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

@@ -26,21 +26,41 @@ public interface TableLocationProvider extends NamedImplementation {
  */
  interface Listener extends BasicTableDataListener {
 
+ /**
+ * Begin a transaction that collects location key additions and removals to be processed atomically.
+ *
+ * @param token A token to identify the transaction.
+ */
+ void beginTransaction(final Object token);
+
+ /**
+ * End the transaction and process the location changes.
+ *
+ * @param token A token to identify the transaction.
+ */
+ void endTransaction(final Object token);
+
  /**
  * Notify the listener of a {@link TableLocationKey} encountered while initiating or maintaining the location
  * subscription. This should occur at most once per location, but the order of delivery is <i>not</i>
  * guaranteed.
  *
- * @param tableLocationKey The new table location key
+ * @param tableLocationKey The new table location key.
+ * @param transactionToken The token identifying the transaction.
  */
- void handleTableLocationKey(@NotNull ImmutableTableLocationKey tableLocationKey);
+ void handleTableLocationKeyAdded(
+ @NotNull final ImmutableTableLocationKey tableLocationKey,
+ final Object transactionToken);
 
  /**
  * Notify the listener of a {@link TableLocationKey} that has been removed.
  *
- * @param tableLocationKey The table location key that was removed
+ * @param tableLocationKey The table location key that was removed.
+ * @param transactionToken The token identifying the transaction.
  */
- void handleTableLocationKeyRemoved(@NotNull ImmutableTableLocationKey tableLocationKey);
+ void handleTableLocationKeyRemoved(
+ @NotNull final ImmutableTableLocationKey tableLocationKey,
+ final Object transactionToken);
  }
 
  /**
@@ -86,7 +106,7 @@ interface Listener extends BasicTableDataListener {
  * that {@link #refresh()} or {@link #subscribe(Listener)} has been called prior to calls to the various table
  * location fetch methods.
  *
- * @return this, to allow method chaining
+ * @return this, to allow method chaining.
  */
  TableLocationProvider ensureInitialized();
 
@@ -95,21 +115,21 @@ interface Listener extends BasicTableDataListener {
  * size - that is, they may not "exist" for application purposes. {@link #getTableLocation(TableLocationKey)} is
  * guaranteed to succeed for all results.
  *
- * @return A collection of keys for locations available from this provider
+ * @return A collection of keys for locations available from this provider.
  */
  @NotNull
  Collection<ImmutableTableLocationKey> getTableLocationKeys();
 
  /**
  * Check if this provider knows the supplied location key.
  *
- * @param tableLocationKey The key to test for
- * @return Whether the key is known to this provider
+ * @param tableLocationKey The key to test.
+ * @return Whether the key is known to this provider.
  */
  boolean hasTableLocationKey(@NotNull final TableLocationKey tableLocationKey);
 
  /**
- * @param tableLocationKey A {@link TableLocationKey} specifying the location to get
+ * @param tableLocationKey A {@link TableLocationKey} specifying the location to get.
  * @return The {@link TableLocation} matching the given key
  */
  @NotNull
@@ -122,8 +142,8 @@ default TableLocation getTableLocation(@NotNull TableLocationKey tableLocationKe
  }
 
  /**
- * @param tableLocationKey A {@link TableLocationKey} specifying the location to get
- * @return The {@link TableLocation} matching the given key if present, else null
+ * @param tableLocationKey A {@link TableLocationKey} specifying the location to get.
+ * @return The {@link TableLocation} matching the given key if present, else null.
  */
  @Nullable
  TableLocation getTableLocationIfPresent(@NotNull TableLocationKey tableLocationKey);

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

@@ -84,7 +84,9 @@ public final ImmutableTableKey getKey() {
 
  @Override
  protected final void deliverInitialSnapshot(@NotNull final TableLocationProvider.Listener listener) {
- unmodifiableTableLocationKeys.forEach(listener::handleTableLocationKey);
+ listener.beginTransaction(this);
+ unmodifiableTableLocationKeys.forEach(tlk -> listener.handleTableLocationKeyAdded(tlk, this));
+ listener.endTransaction(this);
  }
 
  /**
@@ -110,13 +112,28 @@ protected final void handleTableLocationKey(@NotNull final TableLocationKey loca
  visitLocationKey(locationKey);
  if (locationCreatedRecorder) {
  verifyPartitionKeys(locationKey);
- if (subscriptions.deliverNotification(Listener::handleTableLocationKey, toKeyImmutable(result), true)) {
+ if (subscriptions.deliverNotification(
+ (listener, tlk) -> listener.handleTableLocationKeyAdded(tlk,
+ AbstractTableLocationProvider.this),
+ toKeyImmutable(result), true)) {
  onEmpty();
  }
  }
  }
  }
 
+ protected final void beginTransaction() {
+ if (subscriptions != null) {
+ subscriptions.deliverNotification(listener -> listener.beginTransaction(this), true);
+ }
+ }
+
+ protected final void endTransaction() {
+ if (subscriptions != null) {
+ subscriptions.deliverNotification(listener -> listener.endTransaction(this), true);
+ }
+ }
+
  /**
  * Called <i>after</i> a table location has been visited by {@link #handleTableLocationKey(TableLocationKey)}, but
  * before notifications have been delivered to any subscriptions, if applicable. The default implementation does
@@ -180,6 +197,13 @@ protected void doInitialization() {
  @Override
  @NotNull
  public final Collection<ImmutableTableLocationKey> getTableLocationKeys() {
+ // We need to prevent reading the map (and maybe mutating it?) during a transaction.
+ // We could transition between two maps, a stable copy and a shadow copy that is being mutated.
+ // Or we could hold a bigger lock while mutating the map, and hold the same lock here. Sounds like a job for a
+ // read-write lock (e.g. ReentrantReadWriteLock), maybe. If you want `FunctionalLock`, the pattern (but mostly
+ // not the code) from io.deephaven.engine.updategraph.UpdateGraphLock could help.
+ // I think we need the read-write lock for correctness, and I think we need to make it explicit. That is, the
+ // user needs to be able to get a read lock and hold it while it's operating on the returned collection.
  return unmodifiableTableLocationKeys;
  }
 
@@ -215,8 +239,8 @@ public TableLocation getTableLocationIfPresent(@NotNull final TableLocationKey t
  /**
  * Remove a {@link TableLocationKey} and its corresponding {@link TableLocation} (if it was created). All
  * subscribers to this TableLocationProvider will be
- * {@link TableLocationProvider.Listener#handleTableLocationKeyRemoved(ImmutableTableLocationKey) notified}. If the
- * TableLocation was created, all of its subscribers will additionally be
+ * {@link TableLocationProvider.Listener#handleTableLocationKeyRemoved(ImmutableTableLocationKey, Object)}
+ * notified}. If the TableLocation was created, all of its subscribers will additionally be
  * {@link TableLocation.Listener#handleUpdate() notified} that it no longer exists. This TableLocationProvider will
  * continue to update other locations and will no longer provide or request information about the removed location.
  *
@@ -257,7 +281,8 @@ public void removeTableLocationKey(@NotNull final TableLocationKey locationKey)
  protected void handleTableLocationKeyRemoved(@NotNull final ImmutableTableLocationKey locationKey) {
  if (supportsSubscriptions()) {
  synchronized (subscriptions) {
- if (subscriptions.deliverNotification(Listener::handleTableLocationKeyRemoved, locationKey, true)) {
+ if (subscriptions.deliverNotification(
+ (listener, tlk) -> listener.handleTableLocationKeyRemoved(tlk, this), locationKey, true)) {
  onEmpty();
  }
  }

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

@@ -115,7 +115,7 @@ public void subscribe(@NotNull final Listener listener) {
  p.subscribe(listener);
  } else {
  p.refresh();
- p.getTableLocationKeys().forEach(listener::handleTableLocationKey);
+ p.getTableLocationKeys().forEach(tlk -> listener.handleTableLocationKeyAdded(tlk, this));
  }
  });
  }

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

@@ -157,20 +157,42 @@ private FilteringListener(@NotNull final TableLocationProvider.Listener outputLi
  }
 
  @Override
- public void handleTableLocationKey(@NotNull final ImmutableTableLocationKey tableLocationKey) {
+ public void beginTransaction(final Object token) {
+ // Delegate to the wrapped listener.
+ final TableLocationProvider.Listener outputListener = getWrapped();
+ if (outputListener != null) {
+ outputListener.beginTransaction(token);
+ }
+ }
+
+ @Override
+ public void endTransaction(final Object token) {
+ // Delegate to the wrapped listener.
+ final TableLocationProvider.Listener outputListener = getWrapped();
+ if (outputListener != null) {
+ outputListener.endTransaction(token);
+ }
+ }
+
+ @Override
+ public void handleTableLocationKeyAdded(
+ @NotNull final ImmutableTableLocationKey tableLocationKey,
+ final Object transactionToken) {
  final TableLocationProvider.Listener outputListener = getWrapped();
  // We can't try to clean up null listeners here, the underlying implementation may not allow concurrent
  // unsubscribe operations.
  if (outputListener != null && locationKeyFilter.accept(tableLocationKey)) {
- outputListener.handleTableLocationKey(tableLocationKey);
+ outputListener.handleTableLocationKeyAdded(tableLocationKey, transactionToken);
  }
  }
 
  @Override
- public void handleTableLocationKeyRemoved(@NotNull final ImmutableTableLocationKey tableLocationKey) {
+ public void handleTableLocationKeyRemoved(
+ @NotNull final ImmutableTableLocationKey tableLocationKey,
+ final Object transactionToken) {
  final TableLocationProvider.Listener outputListener = getWrapped();
  if (outputListener != null && locationKeyFilter.accept(tableLocationKey)) {
- outputListener.handleTableLocationKeyRemoved(tableLocationKey);
+ outputListener.handleTableLocationKeyRemoved(tableLocationKey, transactionToken);
  }
  }
 

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

@@ -3,14 +3,14 @@
 //
 package io.deephaven.engine.table.impl.locations.impl;
 
-import io.deephaven.engine.table.impl.locations.TableKey;
-import io.deephaven.engine.table.impl.locations.TableLocation;
-import io.deephaven.engine.table.impl.locations.TableLocationKey;
-import io.deephaven.engine.table.impl.locations.TableLocationProvider;
+import io.deephaven.engine.table.impl.locations.*;
 import io.deephaven.engine.table.impl.locations.util.TableDataRefreshService;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
+import java.util.HashSet;
+import java.util.Set;
+
 /**
  * Polling-driven {@link TableLocationProvider} implementation that delegates {@link TableLocationKey location key}
  * discovery to a {@link TableLocationKeyFinder} and {@link TableLocation location} creation to a
@@ -46,9 +46,21 @@ public String getImplementationName() {
  return IMPLEMENTATION_NAME;
  }
 
+ // The simplest way to support "push" of new data availability is to provide a callback to the user that just calls
+ // `refresh`, which would need to become synchronized. Alternatively, we could make an Iceberg-specific aTLP
+ // implementation that exposes a more specific callback, e.g. with a snapshot ID, as well as the option to disable
+ // polling. We do need a mechanism to avoid going backwards, probably.
  @Override
  public void refresh() {
- locationKeyFinder.findKeys(this::handleTableLocationKey);
+ beginTransaction();
+ final Set<ImmutableTableLocationKey> missedKeys = new HashSet<>(getTableLocationKeys());
+ locationKeyFinder.findKeys(tableLocationKey -> {
+ // noinspection SuspiciousMethodCalls
+ missedKeys.remove(tableLocationKey);
+ handleTableLocationKey(tableLocationKey);
+ });
+ missedKeys.forEach(this::handleTableLocationKeyRemoved);
+ endTransaction();
  setInitialized();
  }