feat!: support refreshing Iceberg tables

Util/src/main/java/io/deephaven/util/datastructures/SubscriptionSet.java

@@ -7,6 +7,7 @@
 import io.deephaven.base.reference.WeakReferenceWrapper;
 import io.deephaven.base.verify.Assert;
 import io.deephaven.base.verify.Require;
+import org.apache.commons.lang3.function.TriConsumer;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -219,6 +220,36 @@ public final <NOTIFICATION_TYPE> boolean deliverNotification(
         return initialSize > 0 && size == 0;
     }
 
+    /**
+     * Dispatch a notification to all subscribers. Clean up any GC'd subscriptions.
+     *
+     * @param procedure The notification procedure to invoke
+     * @param firstNotification The first item to deliver
+     * @param secondNotification The second item to deliver (must be of the same type as {@code firstNotification})
+     * @param activeOnly Whether to restrict this notification to active subscriptions only
+     * @return Whether this operation caused the set to become <b>empty</b>
+     */
+    public final <NOTIFICATION_TYPE> boolean deliverNotification(
+            @NotNull final TriConsumer<LISTENER_TYPE, NOTIFICATION_TYPE, NOTIFICATION_TYPE> procedure,
+            @Nullable final NOTIFICATION_TYPE firstNotification,
+            @Nullable final NOTIFICATION_TYPE secondNotification,
+            final boolean activeOnly) {
+        final int initialSize = size;
+        for (int si = 0; si < size;) {
+            final Entry currentEntry = subscriptions[si];
+            final LISTENER_TYPE currentListener = currentEntry.getListener();
+            if (currentListener == null) {
+                removeAt(si);
+                continue; // si is not incremented in this case - we'll reconsider the same slot if necessary.
+            }
+            if (!activeOnly || currentEntry.isActive()) {
+                procedure.accept(currentListener, firstNotification, secondNotification);
+            }
+            ++si;
+        }
+        return initialSize > 0 && size == 0;
+    }
+
     private void removeAt(final int subscriptionIndex) {
         final int lastSubscriptionIndex = --size;
         subscriptions[subscriptionIndex] = subscriptions[lastSubscriptionIndex];

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

@@ -27,20 +27,89 @@ 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(@NotNull Object token);
+
+        /**
+         * End the transaction and process the location changes.
+         *
+         * @param token A token to identify the transaction.
+         */
+        void endTransaction(@NotNull Object token);
+
+        /**
+         * <p>
          * 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.
+         * </p>
+         * <p>
+         * If transactionToken is {@code null}, the key will be added to the pending additions immediately.
+         * </p>
+         *
+         * @param tableLocationKey The new table location key.
+         * @param transactionToken The token identifying the transaction, or {@code null} if this addition is not part
+         *        of a transaction.
+         */
+        void handleTableLocationKeyAdded(
+                @NotNull ImmutableTableLocationKey tableLocationKey,
+                @Nullable Object transactionToken);
+
+        /**
+         * 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. This addition is not part of any transaction, and is equivalent to
+         * {@code handleTableLocationKeyAdded(tableLocationKey, null);} by default.
          *
-         * @param tableLocationKey The new table location key
+         * @param tableLocationKey The new table location key.
          */
-        void handleTableLocationKey(@NotNull ImmutableTableLocationKey tableLocationKey);
+        default void handleTableLocationKeyAdded(@NotNull ImmutableTableLocationKey tableLocationKey) {
+            handleTableLocationKeyAdded(tableLocationKey, null);
+        }
 
         /**
+         * <p>
          * Notify the listener of a {@link TableLocationKey} that has been removed.
+         * </p>
+         * <p>
+         * If transactionToken is {@code null}, the key will be added to the pending removals immediately.
+         * </p>
+         *
+         * @param tableLocationKey The table location key that was removed.
+         * @param transactionToken The token identifying the transaction, or {@code null} if this addition is not part
+         *        of a transaction.
+         */
+        void handleTableLocationKeyRemoved(
+                @NotNull ImmutableTableLocationKey tableLocationKey,
+                @Nullable Object transactionToken);
+
+        /**
+         * Notify the listener of a {@link TableLocationKey} that has been removed. This addition is not part of any
+         * transaction, and is equivalent to {@code handleTableLocationKeyAdded(tableLocationKey, null);} by default.
+         *
+         * @param tableLocationKey The table location key that was removed.
+         */
+        @SuppressWarnings("unused")
+        default void handleTableLocationKeyRemoved(@NotNull ImmutableTableLocationKey tableLocationKey) {
+            handleTableLocationKeyRemoved(tableLocationKey, null);
+        }
+
+        /**
+         * <p>
+         * Notify the listener of collections of {@link TableLocationKey TableLocationKeys} added or removed 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.
+         * </p>
          *
-         * @param tableLocationKey The table location key that was removed
+         * @param addedKeys Collection of table location keys that were added.
+         * @param removedKeys Collection of table location keys that were removed.
          */
-        void handleTableLocationKeyRemoved(@NotNull ImmutableTableLocationKey tableLocationKey);
+        void handleTableLocationKeysUpdate(
+                @Nullable Collection<ImmutableTableLocationKey> addedKeys,
+                @Nullable Collection<ImmutableTableLocationKey> removedKeys);
     }
 
     /**
@@ -86,7 +155,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 +164,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);
+    boolean hasTableLocationKey(@NotNull 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 +191,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/AbstractTableLocation.java

@@ -12,6 +12,7 @@
 import io.deephaven.hash.KeyedObjectHashMap;
 import io.deephaven.hash.KeyedObjectKey;
 import io.deephaven.util.annotations.InternalUseOnly;
+import io.deephaven.util.referencecounting.ReferenceCounted;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -35,6 +36,8 @@ public abstract class AbstractTableLocation
     private final KeyedObjectHashMap<CharSequence, ColumnLocation> columnLocations =
             new KeyedObjectHashMap<>(StringUtils.charSequenceKey());
 
+    private final ReferenceCounted referenceCounted;
+
     @SuppressWarnings("rawtypes")
     private static final AtomicReferenceFieldUpdater<AbstractTableLocation, KeyedObjectHashMap> CACHED_DATA_INDEXES_UPDATER =
             AtomicReferenceFieldUpdater.newUpdater(
@@ -58,6 +61,15 @@ protected AbstractTableLocation(@NotNull final TableKey tableKey,
         super(supportsSubscriptions);
         this.tableKey = Require.neqNull(tableKey, "tableKey").makeImmutable();
         this.tableLocationKey = Require.neqNull(tableLocationKey, "tableLocationKey").makeImmutable();
+
+        referenceCounted = new ReferenceCounted() {
+            @Override
+            protected void onReferenceCountAtZero() {
+                // Call the location's onReferenceCountAtZero method
+                AbstractTableLocation.this.onReferenceCountAtZero();
+            }
+        };
+
     }
 
     @Override
@@ -158,7 +170,7 @@ public final ColumnLocation getColumnLocation(@NotNull final CharSequence name)
      * Clear all column locations (usually because a truncated location was observed).
      */
     @SuppressWarnings("unused")
-    protected final void clearColumnLocations() {
+    public final void clearColumnLocations() {
         columnLocations.clear();
     }
 
@@ -229,4 +241,34 @@ public final BasicDataIndex getDataIndex(@NotNull final String... columns) {
     @InternalUseOnly
     @Nullable
     public abstract BasicDataIndex loadDataIndex(@NotNull String... columns);
+
+    // ------------------------------------------------------------------------------------------------------------------
+    // Reference counting implementation
+    // ------------------------------------------------------------------------------------------------------------------
+
+    /**
+     * Increment the reference count by one.
+     *
+     * @throws IllegalStateException If the reference count was not successfully incremented
+     */
+    public final void incrementReferenceCount() {
+        referenceCounted.incrementReferenceCount();
+    }
+
+    /**
+     * Decrement the reference count by one, when the reference count reaches zero this location will be cleared.
+     *
+     * @throws IllegalStateException If the reference count was not successfully incremented
+     */
+    public void decrementReferenceCount() {
+        referenceCounted.decrementReferenceCount();
+    }
+
+    /**
+     * The reference count has reached zero, we can clear this location and release any resources.
+     */
+    private void onReferenceCountAtZero() {
+        handleUpdate(null, System.currentTimeMillis());
+        clearColumnLocations();
+    }
 }