feat!: support refreshing Iceberg tables

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

@@ -3,13 +3,12 @@
 //
 package io.deephaven.engine.table.impl;
 
-import io.deephaven.engine.liveness.LivenessReferent;
+import io.deephaven.engine.liveness.LivenessNode;
 import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.rowset.TrackingWritableRowSet;
-import io.deephaven.engine.rowset.WritableRowSet;
 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;
 
@@ -19,7 +18,7 @@
 /**
  * Manager for ColumnSources in a Table.
  */
-public interface ColumnSourceManager extends LivenessReferent {
+public interface ColumnSourceManager extends LivenessNode {
 
     /**
      * Get a map of name to {@link ColumnSource} for the column sources maintained by this manager.
@@ -116,5 +115,5 @@ public interface ColumnSourceManager extends LivenessReferent {
      * @return true if the location key was actually removed
      * @param tableLocationKey the location key being removed
      */
-    boolean removeLocationKey(@NotNull ImmutableTableLocationKey tableLocationKey);
+    boolean removeLocationKey(@NotNull TrackedTableLocationKey tableLocationKey);
 }

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

@@ -7,6 +7,7 @@
 import io.deephaven.api.filter.Filter;
 import io.deephaven.base.verify.Assert;
 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;
@@ -204,7 +205,7 @@ protected final Table redefine(TableDefinition newDefinitionExternal, TableDefin
                 reference, null, viewColumns, null);
     }
 
-    private static final String LOCATION_KEY_COLUMN_NAME = "__PartitionAwareSourceTable_TableLocationKey__";
+    private static final String TRACKED_KEY_COLUMN_NAME = "__PartitionAwareSourceTable_TrackedTableLocationKey__";
 
     private static <T> ColumnSource<? super T> makePartitionSource(@NotNull final ColumnDefinition<T> columnDefinition,
             @NotNull final Collection<ImmutableTableLocationKey> locationKeys) {
@@ -221,30 +222,39 @@ private static <T> ColumnSource<? super T> makePartitionSource(@NotNull final Co
     }
 
     @Override
-    protected final Collection<ImmutableTableLocationKey> filterLocationKeys(
-            @NotNull final Collection<ImmutableTableLocationKey> foundLocationKeys) {
+    protected final Collection<TrackedTableLocationKey> filterLocationKeys(
+            @NotNull final Collection<TrackedTableLocationKey> foundLocationKeys) {
         if (partitioningColumnFilters.length == 0) {
             return foundLocationKeys;
         }
+
+        final Collection<ImmutableTableLocationKey> immutableTableLocationKeys = foundLocationKeys.stream()
+                .map(TrackedTableLocationKey::getKey)
+                .collect(Collectors.toList());
+
         // TODO (https://github.com/deephaven/deephaven-core/issues/867): Refactor around a ticking partition table
         final List<String> partitionTableColumnNames = Stream.concat(
                 partitioningColumnDefinitions.keySet().stream(),
-                Stream.of(LOCATION_KEY_COLUMN_NAME)).collect(Collectors.toList());
+                Stream.of(TRACKED_KEY_COLUMN_NAME)).collect(Collectors.toList());
         final List<ColumnSource<?>> partitionTableColumnSources =
                 new ArrayList<>(partitioningColumnDefinitions.size() + 1);
         for (final ColumnDefinition<?> columnDefinition : partitioningColumnDefinitions.values()) {
-            partitionTableColumnSources.add(makePartitionSource(columnDefinition, foundLocationKeys));
+            partitionTableColumnSources.add(makePartitionSource(columnDefinition, immutableTableLocationKeys));
         }
+        // Add the tracked keys to the table
         partitionTableColumnSources.add(ArrayBackedColumnSource.getMemoryColumnSource(foundLocationKeys,
-                ImmutableTableLocationKey.class, null));
+                TrackedTableLocationKey.class, null));
+
         final Table filteredColumnPartitionTable = TableTools
                 .newTable(foundLocationKeys.size(), partitionTableColumnNames, partitionTableColumnSources)
                 .where(Filter.and(partitioningColumnFilters));
         if (filteredColumnPartitionTable.size() == foundLocationKeys.size()) {
             return foundLocationKeys;
         }
-        final Iterable<ImmutableTableLocationKey> iterable =
-                () -> filteredColumnPartitionTable.columnIterator(LOCATION_KEY_COLUMN_NAME);
+
+        // Return the filtered keys
+        final Iterable<TrackedTableLocationKey> 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

@@ -163,9 +163,12 @@ protected void instrumentedRefresh() {
                 processNewLocationsUpdateRoot = null;
                 removedLocationsComitter = null;
                 tableLocationProvider.refresh();
-                try (final RowSet added = sortAndAddLocations(tableLocationProvider.getTableLocationKeys().stream()
-                        .filter(locationKeyMatcher)
-                        .map(tableLocationProvider::getTableLocation))) {
+
+                final Collection<TableLocation> locations = new ArrayList<>();
+                tableLocationProvider.getTableLocationKeys(tlk -> {
+                    locations.add(tableLocationProvider.getTableLocation(tlk.getKey()));
+                }, locationKeyMatcher);
+                try (final RowSet added = sortAndAddLocations(locations.stream())) {
                     resultRows.insert(added);
                 }
             }
@@ -213,9 +216,14 @@ private Table makeConstituentTable(@NotNull final TableLocation tableLocation) {
         }
 
         private void processPendingLocations(final boolean notifyListeners) {
-            final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate = subscriptionBuffer.processPending();
-            final RowSet removed = processRemovals(locationUpdate);
-            final RowSet added = processAdditions(locationUpdate);
+            final RowSet removed;
+            final RowSet added;
+
+            try (final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate =
+                    subscriptionBuffer.processPending()) {
+                removed = processRemovals(locationUpdate);
+                added = processAdditions(locationUpdate);
+            }
 
             resultRows.update(added, removed);
             if (notifyListeners) {
@@ -242,6 +250,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)
                     .filter(locationKeyMatcher)
                     .map(tableLocationProvider::getTableLocation)
                     .map(PendingLocationState::new)
@@ -263,6 +272,7 @@ private RowSet processRemovals(final TableLocationSubscriptionBuffer.LocationUpd
             final Set<ImmutableTableLocationKey> relevantRemovedLocations =
                     locationUpdate.getPendingRemovedLocationKeys()
                             .stream()
+                            .map(TrackedTableLocationKey::getKey)
                             .filter(locationKeyMatcher)
                             .collect(Collectors.toSet());
 

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

@@ -10,9 +10,7 @@
 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.*;
 import io.deephaven.engine.updategraph.UpdateSourceRegistrar;
 import io.deephaven.engine.table.impl.perf.QueryPerformanceRecorder;
 import io.deephaven.engine.table.impl.locations.impl.TableLocationSubscriptionBuffer;
@@ -22,6 +20,7 @@
 import org.apache.commons.lang3.mutable.MutableObject;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.ArrayList;
 import java.util.Collection;
 
 /**
@@ -140,39 +139,42 @@ private void initializeAvailableLocations() {
                 if (isRefreshing()) {
                     final TableLocationSubscriptionBuffer locationBuffer =
                             new TableLocationSubscriptionBuffer(locationProvider);
-                    final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate =
-                            locationBuffer.processPending();
-
-                    maybeRemoveLocations(locationUpdate.getPendingRemovedLocationKeys());
-                    maybeAddLocations(locationUpdate.getPendingAddedLocationKeys());
+                    manage(locationBuffer);
+                    try (final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate =
+                            locationBuffer.processPending()) {
+                        maybeRemoveLocations(locationUpdate.getPendingRemovedLocationKeys());
+                        maybeAddLocations(locationUpdate.getPendingAddedLocationKeys());
+                    }
                     updateSourceRegistrar.addSource(locationChangePoller = new LocationChangePoller(locationBuffer));
                 } else {
                     locationProvider.refresh();
-                    maybeAddLocations(locationProvider.getTableLocationKeys());
+                    final Collection<TrackedTableLocationKey> tableLocationKeys = new ArrayList<>();
+                    locationProvider.getTableLocationKeys(tableLocationKeys::add);
+                    maybeAddLocations(tableLocationKeys);
                 }
             });
             locationsInitialized = true;
         }
     }
 
-    private void maybeAddLocations(@NotNull final Collection<ImmutableTableLocationKey> locationKeys) {
+    private void maybeAddLocations(@NotNull final Collection<TrackedTableLocationKey> locationKeys) {
         if (locationKeys.isEmpty()) {
             return;
         }
         filterLocationKeys(locationKeys)
                 .parallelStream()
-                .forEach(lk -> columnSourceManager.addLocation(locationProvider.getTableLocation(lk)));
+                .forEach(lk -> columnSourceManager.addLocation(locationProvider.getTableLocation(lk.getKey())));
     }
 
-    private ImmutableTableLocationKey[] maybeRemoveLocations(
-            @NotNull final Collection<ImmutableTableLocationKey> removedKeys) {
+    private TrackedTableLocationKey[] maybeRemoveLocations(
+            @NotNull final Collection<TrackedTableLocationKey> removedKeys) {
         if (removedKeys.isEmpty()) {
-            return ImmutableTableLocationKey.ZERO_LENGTH_IMMUTABLE_TABLE_LOCATION_KEY_ARRAY;
+            return TrackedTableLocationKey.ZERO_LENGTH_TRACKED_TABLE_LOCATION_KEY_ARRAY;
         }
 
         return filterLocationKeys(removedKeys).stream()
                 .filter(columnSourceManager::removeLocationKey)
-                .toArray(ImmutableTableLocationKey[]::new);
+                .toArray(TrackedTableLocationKey[]::new);
     }
 
     private void initializeLocationSizes() {
@@ -212,10 +214,11 @@ private LocationChangePoller(@NotNull final TableLocationSubscriptionBuffer loca
 
         @Override
         protected void instrumentedRefresh() {
-            final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate = locationBuffer.processPending();
-            final ImmutableTableLocationKey[] removedKeys =
-                    maybeRemoveLocations(locationUpdate.getPendingRemovedLocationKeys());
-            maybeAddLocations(locationUpdate.getPendingAddedLocationKeys());
+            try (final TableLocationSubscriptionBuffer.LocationUpdate locationUpdate =
+                    locationBuffer.processPending()) {
+                maybeRemoveLocations(locationUpdate.getPendingRemovedLocationKeys());
+                maybeAddLocations(locationUpdate.getPendingAddedLocationKeys());
+            }
 
             // This class previously had functionality to notify "location listeners", but it was never used.
             // Resurrect from git history if needed.
@@ -252,8 +255,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<ImmutableTableLocationKey> filterLocationKeys(
-            @NotNull final Collection<ImmutableTableLocationKey> foundLocationKeys) {
+    protected Collection<TrackedTableLocationKey> filterLocationKeys(
+            @NotNull final Collection<TrackedTableLocationKey> foundLocationKeys) {
         return foundLocationKeys;
     }
 

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

@@ -12,9 +12,6 @@
  */
 @Immutable
 public interface ImmutableTableLocationKey extends TableLocationKey {
-
-    ImmutableTableLocationKey[] ZERO_LENGTH_IMMUTABLE_TABLE_LOCATION_KEY_ARRAY = new ImmutableTableLocationKey[0];
-
     @FinalDefault
     default ImmutableTableLocationKey makeImmutable() {
         return this;

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

@@ -6,6 +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.table.BasicDataIndex;
 import io.deephaven.engine.table.Table;
 import io.deephaven.io.log.impl.LogOutputStringImpl;
@@ -21,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 {
+public interface TableLocation extends NamedImplementation, LogOutputAppendable, TableLocationState, LivenessNode {
 
     /**
      * Listener interface for anything that wants to know about changes to a location.

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

@@ -8,6 +8,8 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.util.Collection;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
 
 /**
  * Discovery utility for {@link TableLocation}s for a given table.
@@ -41,23 +43,24 @@ interface Listener extends BasicTableDataListener {
         void endTransaction(@NotNull 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>
+         * Notify the listener of a {@link TrackedTableLocationKey} 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.
          */
-        void handleTableLocationKeyAdded(@NotNull ImmutableTableLocationKey tableLocationKey);
+        void handleTableLocationKeyAdded(@NotNull TrackedTableLocationKey tableLocationKey);
 
         /**
-         * Notify the listener of a {@link TableLocationKey} that has been removed. This removal is not part of any
-         * transaction, and is equivalent to {@code handleTableLocationKeyRemoved(tableLocationKey, null);} by default.
+         * Notify the listener of a {@link TrackedTableLocationKey} that has been removed. This removal is not part of
+         * any transaction, and is equivalent to {@code handleTableLocationKeyRemoved(tableLocationKey, null);} by
+         * default.
          *
          * @param tableLocationKey The table location key that was removed.
          */
         @SuppressWarnings("unused")
-        void handleTableLocationKeyRemoved(@NotNull ImmutableTableLocationKey tableLocationKey);
+        void handleTableLocationKeyRemoved(@NotNull TrackedTableLocationKey tableLocationKey);
 
         /**
          * <p>
@@ -70,8 +73,8 @@ interface Listener extends BasicTableDataListener {
          * @param removedKeys Collection of table location keys that were removed.
          */
         default void handleTableLocationKeysUpdate(
-                @NotNull Collection<ImmutableTableLocationKey> addedKeys,
-                @NotNull Collection<ImmutableTableLocationKey> removedKeys) {
+                @NotNull Collection<TrackedTableLocationKey> addedKeys,
+                @NotNull Collection<TrackedTableLocationKey> removedKeys) {
             removedKeys.forEach(this::handleTableLocationKeyRemoved);
             addedKeys.forEach(this::handleTableLocationKeyAdded);
         }
@@ -128,11 +131,17 @@ default void handleTableLocationKeysUpdate(
      * Get this provider's currently known location keys. The locations specified by the keys returned may have null
      * 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.
      */
-    @NotNull
-    Collection<ImmutableTableLocationKey> getTableLocationKeys();
+    default void getTableLocationKeys(Consumer<TrackedTableLocationKey> consumer) {
+        getTableLocationKeys(consumer, key -> true);
+    }
+
+    /**
+     * Get this provider's currently known location keys. The locations specified by the keys returned may have null
+     * size - that is, they may not "exist" for application purposes. {@link #getTableLocation(TableLocationKey)} is
+     * guaranteed to succeed for all results.
+     */
+    void getTableLocationKeys(Consumer<TrackedTableLocationKey> consumer, Predicate<ImmutableTableLocationKey> filter);
 
     /**
      * Check if this provider knows the supplied location key.