feat!: support refreshing Iceberg tables

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

@@ -221,18 +221,18 @@ public final <NOTIFICATION_TYPE> boolean deliverNotification(
     }
 
     /**
-     * Dispatch a binary notification to all subscribers. Clean up any GC'd subscriptions.
+     * Dispatch a notification to all subscribers. Clean up any GC'd subscriptions.
      *
      * @param procedure The notification procedure to invoke
-     * @param notification The notification to deliver
-     * @param token An additional token to deliver (usually a transaction token)
+     * @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, Object> procedure,
-            @Nullable final NOTIFICATION_TYPE notification,
-            @Nullable final Object token,
+            @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;) {
@@ -243,7 +243,7 @@ public final <NOTIFICATION_TYPE> boolean deliverNotification(
                 continue; // si is not incremented in this case - we'll reconsider the same slot if necessary.
             }
             if (!activeOnly || currentEntry.isActive()) {
-                procedure.accept(currentListener, notification, token);
+                procedure.accept(currentListener, firstNotification, secondNotification);
             }
             ++si;
         }

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

@@ -49,9 +49,10 @@ interface Listener extends BasicTableDataListener {
          * <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.
+         * @param transactionToken The token identifying the transaction, or {@code null} if this addition is not part
+         *        of a transaction.
          */
         void handleTableLocationKeyAdded(
                 @NotNull ImmutableTableLocationKey tableLocationKey,
@@ -60,7 +61,8 @@ void handleTableLocationKeyAdded(
         /**
          * 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. Uses {@code this} as the token.
+         * 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.
          */
@@ -77,21 +79,37 @@ default void handleTableLocationKeyAdded(@NotNull ImmutableTableLocationKey tabl
          * </p>
          *
          * @param tableLocationKey The table location key that was removed.
-         * @param transactionToken The token identifying the transaction.
+         * @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. Uses {@code this} as the token.
+         * 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 addedKeys Collection of table location keys that were added.
+         * @param removedKeys Collection of table location keys that were removed.
+         */
+        void handleTableLocationKeysUpdate(
+                @Nullable Collection<ImmutableTableLocationKey> addedKeys,
+                @Nullable Collection<ImmutableTableLocationKey> removedKeys);
     }
 
     /**

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();
+    }
 }