feat: support refreshing Iceberg tables

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

@@ -16,7 +16,7 @@
  * Partial {@link TableLocationProvider} implementation for standalone use or as part of a {@link TableDataService}.
  * <p>
  * Presents an interface similar to {@link TableLocationProvider.Listener} for subclasses to use when communicating with
- * the parent; see {@link #handleTableLocationKeyAdded(TableLocationKey, Object).
+ * the parent; see {@link #handleTableLocationKeyAdded(TableLocationKey, Object)}.
  * <p>
  * Note that subclasses are responsible for determining when it's appropriate to call {@link #setInitialized()} and/or
  * override {@link #doInitialization()}.
@@ -34,7 +34,7 @@ private static class Transaction {
  Set<ImmutableTableLocationKey> locationsAdded = EMPTY_TABLE_LOCATION_KEYS;
  Set<ImmutableTableLocationKey> locationsRemoved = EMPTY_TABLE_LOCATION_KEYS;
 
- void addlocationKey(ImmutableTableLocationKey locationKey) {
+ synchronized void addLocationKey(ImmutableTableLocationKey locationKey) {
  if (locationsAdded == EMPTY_TABLE_LOCATION_KEYS) {
  locationsAdded = new HashSet<>();
  }
@@ -49,7 +49,7 @@ void addlocationKey(ImmutableTableLocationKey locationKey) {
  }
  }
 
- void removeLocationKey(ImmutableTableLocationKey locationKey) {
+ synchronized void removeLocationKey(ImmutableTableLocationKey locationKey) {
  if (locationsRemoved == EMPTY_TABLE_LOCATION_KEYS) {
  locationsRemoved = new HashSet<>();
  }
@@ -72,7 +72,7 @@ void removeLocationKey(ImmutableTableLocationKey locationKey) {
  private final ImmutableTableKey tableKey;
 
  // Open transactions that are being accumulated
- private final Map<Object, Transaction> transactions = new HashMap<>();
+ private final Map<Object, Transaction> transactions = Collections.synchronizedMap(new HashMap<>());
 
  /**
  * Map from {@link TableLocationKey} to itself, or to a {@link TableLocation}. The values are {@link TableLocation}s
@@ -140,15 +140,13 @@ protected final void deliverInitialSnapshot(@NotNull final TableLocationProvider
  * @param token A token to identify the transaction
  */
  protected void beginTransaction(@NotNull final Object token) {
- synchronized (transactions) {
- // Verify that we can start a new transaction with this token.
- transactions.compute(token, (key, val) -> {
- if (val != null) {
- throw new IllegalStateException("A transaction with token " + token + " is already open.");
- }
- return new Transaction();
- });
- }
+ // Verify that we can start a new transaction with this token.
+ transactions.compute(token, (key, val) -> {
+ if (val != null) {
+ throw new IllegalStateException("A transaction with token " + token + " is already open.");
+ }
+ return new Transaction();
+ });
  }
 
  /**
@@ -157,40 +155,42 @@ protected void beginTransaction(@NotNull final Object token) {
  * @param token A token to identify the transaction
  */
  protected void endTransaction(@NotNull final Object token) {
- final Transaction transaction;
- synchronized (transactions) {
- // Verify that this transaction is open.
- transaction = transactions.remove(token);
- if (transaction == null) {
- throw new IllegalStateException("No transaction with token " + token + " is currently open.");
- }
+ // Verify that this transaction is open.
+ final Transaction transaction = transactions.remove(token);
+ if (transaction == null) {
+ throw new IllegalStateException("No transaction with token " + token + " is currently open.");
  }
 
- final Collection<ImmutableTableLocationKey> addedKeys =
- new ArrayList<>(transaction.locationsAdded.size());
- final Collection<ImmutableTableLocationKey> removedKeys =
- new ArrayList<>(transaction.locationsRemoved.size());
-
  // Return early if there are no changes to process.
  if (transaction.locationsAdded == EMPTY_TABLE_LOCATION_KEYS
  && transaction.locationsRemoved == EMPTY_TABLE_LOCATION_KEYS) {
  return;
  }
 
- for (ImmutableTableLocationKey locationKey : transaction.locationsRemoved) {
- final Object removedLocation = tableLocations.remove(locationKey);
- if (removedLocation != null) {
- removedKeys.add(locationKey);
+ final Collection<ImmutableTableLocationKey> addedKeys =
+ new ArrayList<>(transaction.locationsAdded.size());
+ final Collection<ImmutableTableLocationKey> removedKeys =
+ new ArrayList<>(transaction.locationsRemoved.size());
+
+ synchronized (tableLocations) {
+ for (ImmutableTableLocationKey locationKey : transaction.locationsRemoved) {
+ final Object removedLocation = tableLocations.remove(locationKey);
+ if (removedLocation != null) {
+ if (removedLocation instanceof AbstractTableLocation) {
+ ((AbstractTableLocation) removedLocation).decrementReferenceCount();
+ }
+ removedKeys.add(locationKey);
+ }
  }
- }
 
- for (ImmutableTableLocationKey locationKey : transaction.locationsAdded) {
- locationCreatedRecorder = false;
- tableLocations.putIfAbsent(locationKey, this::observeInsert);
- visitLocationKey(locationKey);
- if (locationCreatedRecorder) {
- verifyPartitionKeys(locationKey);
- addedKeys.add(locationKey);
+ for (ImmutableTableLocationKey locationKey : transaction.locationsAdded) {
+ locationCreatedRecorder = false;
+ tableLocations.putIfAbsent(locationKey, this::observeInsert);
+ visitLocationKey(locationKey);
+ if (locationCreatedRecorder) {
+ verifyPartitionKeys(locationKey);
+ addedKeys.add(locationKey);
+ }
  }
  }
 
@@ -230,16 +230,13 @@ protected final void handleTableLocationKeyAdded(
  @Nullable final Object transactionToken) {
 
  if (transactionToken != null) {
- final Transaction transaction;
- synchronized (transactions) {
- transaction = transactions.get(transactionToken);
- if (transaction == null) {
- throw new IllegalStateException(
- "No transaction with token " + transactionToken + " is currently open.");
- }
+ final Transaction transaction = transactions.get(transactionToken);
+ if (transaction == null) {
+ throw new IllegalStateException(
+ "No transaction with token " + transactionToken + " is currently open.");
  }
  // Store an immutable key
- transaction.addlocationKey(toKeyImmutable(locationKey));
+ transaction.addLocationKey(toKeyImmutable(locationKey));
  return;
  }
 
@@ -335,9 +332,11 @@ protected void doInitialization() {
  @Override
  @NotNull
  public final Collection<ImmutableTableLocationKey> getTableLocationKeys() {
- final List<ImmutableTableLocationKey> result = new ArrayList<>(tableLocations.size());
- tableLocations.keySet().forEach(key -> result.add((ImmutableTableLocationKey) key));
- return result;
+ synchronized (tableLocations) {
+ // Note that we never store keys that aren't immutable.
+ // noinspection unchecked,rawtypes
+ return new ArrayList<>((Set<ImmutableTableLocationKey>) (Set) tableLocations.keySet());
+ }
  }
 
  @Override
@@ -363,6 +362,9 @@ public TableLocation getTableLocationIfPresent(@NotNull final TableLocationKey t
  if (immutableKey == current) {
  // Note, this may contend for the lock on tableLocations
  tableLocations.add(current = makeTableLocation(immutableKey));
+ if (current instanceof AbstractTableLocation) {
+ ((AbstractTableLocation) current).incrementReferenceCount();
+ }
  }
  }
  }
@@ -418,27 +420,30 @@ protected void handleTableLocationKeyRemoved(
  @NotNull final TableLocationKey locationKey,
  @Nullable final Object transactionToken) {
  if (transactionToken != null) {
- final Transaction transaction;
- synchronized (transactions) {
- transaction = transactions.get(transactionToken);
- if (transaction == null) {
- throw new IllegalStateException(
- "No transaction with token " + transactionToken + " is currently open.");
- }
+ final Transaction transaction = transactions.get(transactionToken);
+ if (transaction == null) {
+ throw new IllegalStateException(
+ "No transaction with token " + transactionToken + " is currently open.");
  }
  transaction.removeLocationKey(toKeyImmutable(locationKey));
  return;
  }
 
  if (!supportsSubscriptions()) {
- tableLocations.remove(locationKey);
+ final Object removedLocation = tableLocations.remove(locationKey);
+ if (removedLocation instanceof AbstractTableLocation) {
+ ((AbstractTableLocation) removedLocation).decrementReferenceCount();
+ }
  return;
  }
 
  // If we're not in a transaction, we should push this key immediately.
  synchronized (subscriptions) {
  final Object removedLocation = tableLocations.remove(locationKey);
  if (removedLocation != null) {
+ if (removedLocation instanceof AbstractTableLocation) {
+ ((AbstractTableLocation) removedLocation).decrementReferenceCount();
+ }
  if (subscriptions.deliverNotification(
  Listener::handleTableLocationKeyRemoved,
  locationKey.makeImmutable(),

engine/table/src/main/java/io/deephaven/engine/table/impl/sources/regioned/RegionedColumnSourceManager.java

@@ -107,9 +107,6 @@ public class RegionedColumnSourceManager extends LivenessArtifact implements Col
  @ReferentialIntegrity
  private final Collection<DataIndex> retainedDataIndexes = new ArrayList<>();
 
- private final List<AbstractTableLocation> locationsToClear;
- private final UpdateCommitter<?> locationClearCommitter;
-
  /**
  * A reference to a delayed error notifier for the {@link #includedLocationsTable}, if one is pending.
  */
@@ -199,19 +196,6 @@ public class RegionedColumnSourceManager extends LivenessArtifact implements Col
  }
  }
 
-
- locationsToClear = new ArrayList<>();
- locationClearCommitter = new UpdateCommitter<>(this,
- ExecutionContext.getContext().getUpdateGraph(),
- (ignored) -> {
- locationsToClear.forEach(location -> {
- location.handleUpdate(null, System.currentTimeMillis());
- location.clearColumnLocations();
-
- });
- locationsToClear.clear();
- });
-
  invalidateCommitter = new UpdateCommitter<>(this,
  ExecutionContext.getContext().getUpdateGraph(),
  (ignored) -> {
@@ -222,6 +206,16 @@ public class RegionedColumnSourceManager extends LivenessArtifact implements Col
  });
  }
 
+ @Override
+ public void destroy() {
+ super.destroy();
+ includedTableLocations.keySet().forEach(location -> {
+ if (location instanceof AbstractTableLocation) {
+ ((AbstractTableLocation) location).decrementReferenceCount();
+ }
+ });
+ }
+
  @Override
  public synchronized void addLocation(@NotNull final TableLocation tableLocation) {
  final IncludedTableLocationEntry includedLocation = includedTableLocations.get(tableLocation.getKey());
@@ -232,9 +226,6 @@ public synchronized void addLocation(@NotNull final TableLocation tableLocation)
  log.debug().append("LOCATION_ADDED:").append(tableLocation.toString()).endl();
  }
  emptyTableLocations.add(new EmptyTableLocationEntry(tableLocation));
- if (tableLocation instanceof AbstractTableLocation) {
- ((AbstractTableLocation) tableLocation).incrementReferenceCount();
- }
  } else {
  // Duplicate location - not allowed
  final TableLocation duplicateLocation =
@@ -420,6 +411,9 @@ private TableUpdateImpl update(final boolean initializing) {
  includedTableLocations.add(entry);
  orderedIncludedTableLocations.add(entry);
  entry.processInitial(addedRowSetBuilder, entryToInclude.initialRowSet);
+ if (entry.location instanceof AbstractTableLocation) {
+ ((AbstractTableLocation) entry.location).incrementReferenceCount();
+ }
 
  // We have a new location, add the row set to the table and mark the row as added.
  // @formatter:off