feat: support refreshing Iceberg tables

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

@@ -3,12 +3,12 @@
 //
 package io.deephaven.engine.table.impl;
 
-import io.deephaven.engine.liveness.LivenessNode;
+import io.deephaven.engine.liveness.LivenessReferent;
 import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.rowset.TrackingWritableRowSet;
 import io.deephaven.engine.table.*;
+import io.deephaven.engine.table.impl.locations.ImmutableTableLocationKey;
 import io.deephaven.engine.table.impl.locations.TableLocation;
-import io.deephaven.engine.table.impl.locations.TrackedTableLocationKey;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -18,7 +18,7 @@
 /**
  * Manager for ColumnSources in a Table.
  */
-public interface ColumnSourceManager extends LivenessNode {
+public interface ColumnSourceManager extends LivenessReferent {
 
     /**
      * Get a map of name to {@link ColumnSource} for the column sources maintained by this manager.
@@ -112,8 +112,7 @@ public interface ColumnSourceManager extends LivenessNode {
     /**
      * Remove a table location key from the sources.
      *
-     * @return true if the location key was actually removed
      * @param tableLocationKey the location key being removed
      */
-    boolean removeLocationKey(@NotNull TrackedTableLocationKey tableLocationKey);
+    void removeLocationKey(@NotNull ImmutableTableLocationKey tableLocationKey);
 }

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

@@ -6,8 +6,8 @@
 import io.deephaven.api.Selectable;
 import io.deephaven.api.filter.Filter;
 import io.deephaven.base.verify.Assert;
+import io.deephaven.engine.liveness.LiveSupplier;
 import io.deephaven.engine.table.*;
-import io.deephaven.engine.table.impl.locations.TrackedTableLocationKey;
 import io.deephaven.engine.table.impl.select.analyzers.SelectAndViewAnalyzer;
 import io.deephaven.engine.updategraph.UpdateSourceRegistrar;
 import io.deephaven.engine.table.impl.perf.QueryPerformanceRecorder;
@@ -222,14 +222,14 @@ private static <T> ColumnSource<? super T> makePartitionSource(@NotNull final Co
     }
 
     @Override
-    protected final Collection<TrackedTableLocationKey> filterLocationKeys(
-            @NotNull final Collection<TrackedTableLocationKey> foundLocationKeys) {
+    protected final Collection<LiveSupplier<ImmutableTableLocationKey>> filterLocationKeys(
+            @NotNull final Collection<LiveSupplier<ImmutableTableLocationKey>> foundLocationKeys) {
         if (partitioningColumnFilters.length == 0) {
             return foundLocationKeys;
         }
 
         final Collection<ImmutableTableLocationKey> immutableTableLocationKeys = foundLocationKeys.stream()
-                .map(TrackedTableLocationKey::getKey)
+                .map(LiveSupplier::get)
                 .collect(Collectors.toList());
 
         // TODO (https://github.com/deephaven/deephaven-core/issues/867): Refactor around a ticking partition table
@@ -242,8 +242,11 @@ protected final Collection<TrackedTableLocationKey> filterLocationKeys(
             partitionTableColumnSources.add(makePartitionSource(columnDefinition, immutableTableLocationKeys));
         }
         // Add the tracked keys to the table
-        partitionTableColumnSources.add(ArrayBackedColumnSource.getMemoryColumnSource(foundLocationKeys,
-                TrackedTableLocationKey.class, null));
+        // noinspection unchecked,rawtypes
+        partitionTableColumnSources.add(ArrayBackedColumnSource.getMemoryColumnSource(
+                (Collection<LiveSupplier>) (Collection) foundLocationKeys,
+                LiveSupplier.class,
+                null));
 
         final Table filteredColumnPartitionTable = TableTools
                 .newTable(foundLocationKeys.size(), partitionTableColumnNames, partitionTableColumnSources)
@@ -253,7 +256,7 @@ protected final Collection<TrackedTableLocationKey> filterLocationKeys(
         }
 
         // Return the filtered keys
-        final Iterable<TrackedTableLocationKey> iterable =
+        final Iterable<LiveSupplier<ImmutableTableLocationKey>> iterable =
                 () -> filteredColumnPartitionTable.columnIterator(TRACKED_KEY_COLUMN_NAME);
         return StreamSupport.stream(iterable.spliterator(), false).collect(Collectors.toList());
     }

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

@@ -4,6 +4,7 @@
 package io.deephaven.engine.table.impl;
 
 import io.deephaven.base.verify.Assert;
+import io.deephaven.engine.liveness.*;
 import io.deephaven.engine.primitive.iterator.CloseableIterator;
 import io.deephaven.engine.rowset.*;
 import io.deephaven.engine.table.*;
@@ -37,6 +38,26 @@ public class SourcePartitionedTable extends PartitionedTableImpl {
     private static final String KEY_COLUMN_NAME = "TableLocationKey";
     private static final String CONSTITUENT_COLUMN_NAME = "LocationTable";
 
+    /**
+     * Private constructor for a {@link SourcePartitionedTable}.
+     *
+     * @param table the locations table to use for this {@link SourcePartitionedTable}
+     * @param constituentDefinition The {@link TableDefinition} expected of constituent {@link Table tables}
+     * @param refreshLocations Whether the set of locations should be refreshed
+     */
+    private SourcePartitionedTable(
+            @NotNull final Table table,
+            @NotNull final TableDefinition constituentDefinition,
+            final boolean refreshLocations) {
+        super(table,
+                Set.of(KEY_COLUMN_NAME),
+                true,
+                CONSTITUENT_COLUMN_NAME,
+                constituentDefinition,
+                refreshLocations,
+                false);
+    }
+
     /**
      * Construct a {@link SourcePartitionedTable} from the supplied parameters.
      * <p>
@@ -51,29 +72,30 @@ public class SourcePartitionedTable extends PartitionedTableImpl {
      * @param refreshSizes Whether the locations found should be refreshed
      * @param locationKeyMatcher Function to filter desired location keys
      */
-    public SourcePartitionedTable(
+    public static SourcePartitionedTable create(
             @NotNull final TableDefinition constituentDefinition,
             @NotNull final UnaryOperator<Table> applyTablePermissions,
             @NotNull final TableLocationProvider tableLocationProvider,
             final boolean refreshLocations,
             final boolean refreshSizes,
             @NotNull final Predicate<ImmutableTableLocationKey> locationKeyMatcher) {
-        super(new UnderlyingTableMaintainer(
-                constituentDefinition,
+        final UnderlyingTableMaintainer maintainer = new UnderlyingTableMaintainer(constituentDefinition,
                 applyTablePermissions,
                 tableLocationProvider,
                 refreshLocations,
                 refreshSizes,
-                locationKeyMatcher).result(),
-                Set.of(KEY_COLUMN_NAME),
-                true,
-                CONSTITUENT_COLUMN_NAME,
+                locationKeyMatcher);
+
+        final SourcePartitionedTable sourcePartitionedTable = new SourcePartitionedTable(
+                maintainer.result(),
                 constituentDefinition,
-                refreshLocations,
-                false);
+                refreshLocations);
+
+        maintainer.assignLivenessManager(sourcePartitionedTable);
+        return sourcePartitionedTable;
     }
 
-    private static final class UnderlyingTableMaintainer {
+    private static final class UnderlyingTableMaintainer extends ReferenceCountedLivenessNode {
 
         private final TableDefinition constituentDefinition;
         private final UnaryOperator<Table> applyTablePermissions;
@@ -103,13 +125,17 @@ private UnderlyingTableMaintainer(
                 final boolean refreshLocations,
                 final boolean refreshSizes,
                 @NotNull final Predicate<ImmutableTableLocationKey> locationKeyMatcher) {
+            super(false);
+
+            // Increase the refcount of this liveness node to allow it to manage the subscription buffer.
+            retainReference();
+
             this.constituentDefinition = constituentDefinition;
             this.applyTablePermissions = applyTablePermissions;
             this.tableLocationProvider = tableLocationProvider;
             this.refreshSizes = refreshSizes;
             this.locationKeyMatcher = locationKeyMatcher;
 
-            // noinspection resource
             resultRows = RowSetFactory.empty().toTracking();
             resultTableLocationKeys = ArrayBackedColumnSource.getMemoryColumnSource(TableLocationKey.class, null);
             resultLocationTables = ArrayBackedColumnSource.getMemoryColumnSource(Table.class, null);
@@ -130,6 +156,7 @@ private UnderlyingTableMaintainer(
 
             if (needToRefreshLocations) {
                 subscriptionBuffer = new TableLocationSubscriptionBuffer(tableLocationProvider);
+                manage(subscriptionBuffer);
                 pendingLocationStates = new IntrusiveDoublyLinkedQueue<>(
                         IntrusiveDoublyLinkedNode.Adapter.<PendingLocationState>getInstance());
                 readyLocationStates = new IntrusiveDoublyLinkedQueue<>(
@@ -154,7 +181,6 @@ protected void instrumentedRefresh() {
                             removedConstituents.forEach(result::unmanage);
                             removedConstituents = null;
                         });
-
                 processPendingLocations(false);
             } else {
                 subscriptionBuffer = null;
@@ -166,7 +192,7 @@ protected void instrumentedRefresh() {
 
                 final Collection<TableLocation> locations = new ArrayList<>();
                 tableLocationProvider.getTableLocationKeys(tlk -> {
-                    locations.add(tableLocationProvider.getTableLocation(tlk.getKey()));
+                    locations.add(tableLocationProvider.getTableLocation(tlk.get()));
                 }, locationKeyMatcher);
                 try (final RowSet added = sortAndAddLocations(locations.stream())) {
                     resultRows.insert(added);
@@ -178,6 +204,15 @@ protected void instrumentedRefresh() {
             }
         }
 
+        /**
+         * Manage ourselves with the provided liveness manager and cleanup the incremented reference count from the
+         * constructor.
+         */
+        private void assignLivenessManager(final LivenessManager manager) {
+            manager.manage(this);
+            dropReference();
+        }
+
         private QueryTable result() {
             return result;
         }
@@ -207,7 +242,7 @@ private Table makeConstituentTable(@NotNull final TableLocation tableLocation) {
                     constituentDefinition,
                     "SingleLocationSourceTable-" + tableLocation,
                     RegionedTableComponentFactoryImpl.INSTANCE,
-                    new SingleTableLocationProvider(tableLocation),
+                    new SingleTableLocationProvider(tableLocation, TableLocationProvider.UpdateMode.ADD_REMOVE),
                     refreshSizes ? refreshCombiner : null);
 
             // Be careful to propagate the systemic attribute properly to child tables
@@ -250,7 +285,7 @@ private RowSet processAdditions(final TableLocationSubscriptionBuffer.LocationUp
              */
             // TODO (https://github.com/deephaven/deephaven-core/issues/867): Refactor around a ticking partition table
             locationUpdate.getPendingAddedLocationKeys().stream()
-                    .map(TrackedTableLocationKey::getKey)
+                    .map(LiveSupplier::get)
                     .filter(locationKeyMatcher)
                     .map(tableLocationProvider::getTableLocation)
                     .map(PendingLocationState::new)
@@ -272,7 +307,7 @@ private RowSet processRemovals(final TableLocationSubscriptionBuffer.LocationUpd
             final Set<ImmutableTableLocationKey> relevantRemovedLocations =
                     locationUpdate.getPendingRemovedLocationKeys()
                             .stream()
-                            .map(TrackedTableLocationKey::getKey)
+                            .map(LiveSupplier::get)
                             .filter(locationKeyMatcher)
                             .collect(Collectors.toSet());
 

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

@@ -5,8 +5,10 @@
 
 import io.deephaven.base.verify.Assert;
 import io.deephaven.base.verify.Require;
+import io.deephaven.engine.liveness.LiveSupplier;
 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;
@@ -101,7 +103,20 @@ public abstract class SourceTable<IMPL_TYPE extends SourceTable<IMPL_TYPE>> exte
             }
         }
 
-        setRefreshing(isRefreshing);
+        if (isRefreshing) {
+            setRefreshing(true);
+            if (locationProvider.getUpdateMode() == TableLocationProvider.UpdateMode.APPEND_ONLY
+                    && locationProvider.getLocationUpdateMode() == TableLocationProvider.UpdateMode.STATIC) {
+                // This table is APPEND_ONLY IFF the set of locations is APPEND_ONLY
+                // and the location contents are STATIC
+                setAttribute(Table.APPEND_ONLY_TABLE_ATTRIBUTE, Boolean.FALSE);
+            } else if (locationProvider.getUpdateMode() != TableLocationProvider.UpdateMode.ADD_REMOVE
+                    && locationProvider.getLocationUpdateMode() != TableLocationProvider.UpdateMode.ADD_REMOVE) {
+                // This table is ADD_ONLY IFF the set of locations is not allowed to remove locations (!ADD_REMOVE)
+                // and the locations contents are not allowed to remove rows (!ADD_REMOVE)
+                setAttribute(Table.ADD_ONLY_TABLE_ATTRIBUTE, Boolean.TRUE);
+            }
+        }
     }
 
     /**
@@ -148,33 +163,44 @@ private void initializeAvailableLocations() {
                     updateSourceRegistrar.addSource(locationChangePoller = new LocationChangePoller(locationBuffer));
                 } else {
                     locationProvider.refresh();
-                    final Collection<TrackedTableLocationKey> tableLocationKeys = new ArrayList<>();
-                    locationProvider.getTableLocationKeys(tableLocationKeys::add);
-                    maybeAddLocations(tableLocationKeys);
+                    final Collection<LiveSupplier<ImmutableTableLocationKey>> keySuppliers = new ArrayList<>();
+                    // Manage each of the location keys as we see them (since the TLP is not guaranteeing them outside
+                    // the callback)
+                    locationProvider.getTableLocationKeys(ttlk -> {
+                        if (isRefreshing()) {
+                            manage(ttlk);
+                        }
+                        keySuppliers.add(ttlk);
+                    });
+                    maybeAddLocations(keySuppliers);
+                    if (isRefreshing()) {
+                        // Now we can un-manage the location keys
+                        keySuppliers.forEach(this::unmanage);
+                    }
                 }
             });
             locationsInitialized = true;
         }
     }
 
-    private void maybeAddLocations(@NotNull final Collection<TrackedTableLocationKey> locationKeys) {
+    private void maybeAddLocations(@NotNull final Collection<LiveSupplier<ImmutableTableLocationKey>> locationKeys) {
         if (locationKeys.isEmpty()) {
             return;
         }
         filterLocationKeys(locationKeys)
                 .parallelStream()
-                .forEach(lk -> columnSourceManager.addLocation(locationProvider.getTableLocation(lk.getKey())));
+                .forEach(lk -> columnSourceManager.addLocation(locationProvider.getTableLocation(lk.get())));
     }
 
-    private TrackedTableLocationKey[] maybeRemoveLocations(
-            @NotNull final Collection<TrackedTableLocationKey> removedKeys) {
+    private void maybeRemoveLocations(
+            @NotNull final Collection<LiveSupplier<ImmutableTableLocationKey>> removedKeys) {
         if (removedKeys.isEmpty()) {
-            return TrackedTableLocationKey.ZERO_LENGTH_TRACKED_TABLE_LOCATION_KEY_ARRAY;
+            return;
         }
 
-        return filterLocationKeys(removedKeys).stream()
-                .filter(columnSourceManager::removeLocationKey)
-                .toArray(TrackedTableLocationKey[]::new);
+        filterLocationKeys(removedKeys).stream()
+                .map(LiveSupplier::get)
+                .forEach(columnSourceManager::removeLocationKey);
     }
 
     private void initializeLocationSizes() {
@@ -216,6 +242,14 @@ private LocationChangePoller(@NotNull final TableLocationSubscriptionBuffer loca
         protected void instrumentedRefresh() {
             try (final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate =
                     locationBuffer.processPending()) {
+                if (locationProvider.getUpdateMode() != TableLocationProvider.UpdateMode.ADD_REMOVE
+                        && !locationUpdate.getPendingRemovedLocationKeys().isEmpty()) {
+                    // This TLP doesn't support removed locations, we need to throw an exception.
+                    final ImmutableTableLocationKey[] keys = locationUpdate.getPendingRemovedLocationKeys().stream()
+                            .map(LiveSupplier::get).toArray(ImmutableTableLocationKey[]::new);
+                    throw new TableLocationRemovedException("Source table does not support removed locations", keys);
+                }
+
                 maybeRemoveLocations(locationUpdate.getPendingRemovedLocationKeys());
                 maybeAddLocations(locationUpdate.getPendingAddedLocationKeys());
             }
@@ -255,8 +289,8 @@ protected void onRefreshError(@NotNull final Exception error) {
      *        {@link TableLocationProvider}, but not yet incorporated into the table
      * @return A sub-collection of the input
      */
-    protected Collection<TrackedTableLocationKey> filterLocationKeys(
-            @NotNull final Collection<TrackedTableLocationKey> foundLocationKeys) {
+    protected Collection<LiveSupplier<ImmutableTableLocationKey>> filterLocationKeys(
+            @NotNull final Collection<LiveSupplier<ImmutableTableLocationKey>> foundLocationKeys) {
         return foundLocationKeys;
     }
 

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

@@ -6,7 +6,7 @@
 import io.deephaven.api.SortColumn;
 import io.deephaven.base.log.LogOutput;
 import io.deephaven.base.log.LogOutputAppendable;
-import io.deephaven.engine.liveness.LivenessNode;
+import io.deephaven.engine.liveness.LivenessReferent;
 import io.deephaven.engine.table.BasicDataIndex;
 import io.deephaven.engine.table.Table;
 import io.deephaven.io.log.impl.LogOutputStringImpl;
@@ -22,7 +22,7 @@
  * location allows access to columns, size, and possibly other metadata for a single partition that may be included in a
  * source table.
  */
-public interface TableLocation extends NamedImplementation, LogOutputAppendable, TableLocationState, LivenessNode {
+public interface TableLocation extends NamedImplementation, LogOutputAppendable, TableLocationState, LivenessReferent {
 
     /**
      * Listener interface for anything that wants to know about changes to a location.

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

@@ -56,4 +56,9 @@ <PARTITION_VALUE_TYPE extends Comparable<PARTITION_VALUE_TYPE>> PARTITION_VALUE_
      * @return An immutable version of this key
      */
     ImmutableTableLocationKey makeImmutable();
+
+    /**
+     * Release any cached data associated with this key. This would only be called at EOL for this key.
+     */
+    default void clear() {}
 }