feat: support refreshing Iceberg tables

extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergBaseLayout.java

@@ -170,7 +170,7 @@ public synchronized void findKeys(@NotNull final Consumer<IcebergTableLocationKe
  /**
  * Update the snapshot to the latest snapshot from the catalog if
  */
- protected boolean maybeUpdateSnapshot() {
+ protected synchronized boolean maybeUpdateSnapshot() {
  final Snapshot latestSnapshot = tableAdapter.currentSnapshot();
  if (latestSnapshot == null) {
  return false;
@@ -183,11 +183,8 @@ protected boolean maybeUpdateSnapshot() {
  }
 
  /**
- * Update the snapshot to the user specified snapshot. This will fail with an {@link IllegalArgumentException} if
- * the input snapshot is not newer (higher in sequence number) than the current snapshot or if the snapshot cannot
- * be found.
- *
- * @param snapshotId the id of the snapshot to load
+ * Update the snapshot to the user specified snapshot. See
+ * {@link io.deephaven.iceberg.util.IcebergTable#update(long)} for more details.
  */
  protected void updateSnapshot(long snapshotId) {
  final List<Snapshot> snapshots = tableAdapter.listSnapshots();
@@ -205,23 +202,18 @@ protected void updateSnapshot(long snapshotId) {
  }
 
  /**
- * Update the snapshot to the user specified snapshot. This will fail with an {@link IllegalArgumentException} if
- * the input snapshot is not newer (higher in sequence number) than the current snapshot.
- *
- * @param snapshot the snapshot to load
+ * Update the snapshot to the user specified snapshot. See
+ * {@link io.deephaven.iceberg.util.IcebergTable#update(Snapshot)} for more details.
  */
- protected void updateSnapshot(Snapshot snapshot) {
+ protected void updateSnapshot(@NotNull final Snapshot updateSnapshot) {
  // Validate that we are not trying to update to an older snapshot.
- if (snapshot == null) {
- throw new IllegalArgumentException("Input snapshot cannot be null");
- }
- if (this.snapshot != null && snapshot.sequenceNumber() <= this.snapshot.sequenceNumber()) {
+ if (snapshot != null && updateSnapshot.sequenceNumber() <= snapshot.sequenceNumber()) {
  throw new IllegalArgumentException(
- "Update snapshot sequence number (" + snapshot.sequenceNumber()
+ "Update snapshot sequence number (" + updateSnapshot.sequenceNumber()
  + ") must be higher than the current snapshot sequence number ("
- + this.snapshot.sequenceNumber() + ") for table " + tableAdapter);
+ + snapshot.sequenceNumber() + ") for table " + tableAdapter);
  }
 
- this.snapshot = snapshot;
+ snapshot = updateSnapshot;
  }
 }

extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergManualRefreshTableLocationProvider.java

@@ -51,9 +51,8 @@ public String getImplementationName() {
  }
 
  @Override
- public synchronized void refresh() {
- // There is no refresh service for this provider, but this is called as part of the initialization process.
- refreshLocations();
+ public void refresh() {
+ ensureInitialized();
  }
 
  @Override
@@ -72,11 +71,19 @@ public synchronized void update(final long snapshotId) {
 
  @Override
  public synchronized void update(final Snapshot snapshot) {
+ if (snapshot == null) {
+ throw new IllegalArgumentException("Input snapshot cannot be null");
+ }
  // Update the snapshot.
  locationKeyFinder.updateSnapshot(snapshot);
  refreshLocations();
  }
 
+ @Override
+ protected synchronized void doInitialization() {
+ refreshLocations();
+ }
+
  // ------------------------------------------------------------------------------------------------------------------
  // SubscriptionAggregator implementation
  // ------------------------------------------------------------------------------------------------------------------

extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java

@@ -7,13 +7,16 @@
 import io.deephaven.engine.table.*;
 import io.deephaven.engine.table.impl.QueryTable;
 import io.deephaven.engine.table.impl.sources.InMemoryColumnSource;
+import io.deephaven.iceberg.internal.DataInstructionsProviderLoader;
+import io.deephaven.iceberg.internal.DataInstructionsProviderPlugin;
 import io.deephaven.util.annotations.VisibleForTesting;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.jetbrains.annotations.NotNull;
-import io.deephaven.iceberg.internal.DataInstructionsProviderLoader;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.ResourcePaths;
 
 import java.util.*;
 

extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergInstructions.java

@@ -3,6 +3,7 @@
 //
 package io.deephaven.iceberg.util;
 
+import io.deephaven.annotations.CopyableStyle;
 import io.deephaven.engine.table.TableDefinition;
 import org.immutables.value.Value;
 import org.immutables.value.Value.Immutable;
@@ -15,11 +16,7 @@
  * this class may change in the future. As such, callers may wish to explicitly set the values.
  */
 @Immutable
-@Value.Style(visibility = Value.Style.ImplementationVisibility.PACKAGE,
- strictBuilder = true,
- weakInterning = true,
- jdkOnly = true,
- includeHashCode = "getClass().hashCode()")
+@CopyableStyle
 public abstract class IcebergInstructions {
  /**
  * The default {@link IcebergInstructions} to use when reading Iceberg data files. Providing this will use system
@@ -50,7 +47,7 @@ public static Builder builder() {
  public abstract Map<String, String> columnRenames();
 
  /**
- * Return a copy of this instructions object with the column renames updated to the provided entries.
+ * Return a copy of this instructions object with the column renames replaced by {@code entries}.
  */
  public abstract IcebergInstructions withColumnRenames(Map<String, ? extends String> entries);
 

extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableAdapter.java

@@ -77,13 +77,22 @@ public synchronized Snapshot currentSnapshot() {
  }
 
  /**
- * List all {@link Snapshot snapshots} of the Iceberg table.
+ * Get the current list of all {@link Snapshot snapshots} of the Iceberg table.
  *
  * @return A list of all snapshots of the given table.
  */
  public synchronized List<Snapshot> listSnapshots() {
  // Refresh the table to update the snapshot list.
  refresh();
+ return getSnapshots();
+ }
+
+ /**
+ * Get a list of all {@link Snapshot snapshots} of the Iceberg table (without refreshing).
+ *
+ * @return A list of all snapshots of the given table.
+ */
+ private List<Snapshot> getSnapshots() {
  final List<Snapshot> snapshots = new ArrayList<>();
  table.snapshots().forEach(snapshots::add);
  return snapshots;
@@ -124,6 +133,7 @@ public synchronized List<Snapshot> listSnapshots() {
  * @return A Table containing a list of all tables in the given namespace.
  */
  public Table snapshots() {
+ // Retrieve the current list of snapshots
  final List<Snapshot> snapshots = listSnapshots();
  final long size = snapshots.size();
 
@@ -173,9 +183,17 @@ public Table snapshots() {
  * @return An Optional<Snapshot> containing the requested snapshot if it exists.
  */
  private Optional<Snapshot> snapshot(final long snapshotId) {
- return listSnapshots().stream()
+ Optional<Snapshot> found = getSnapshots().stream()
  .filter(snapshot -> snapshot.snapshotId() == snapshotId)
  .findFirst();
+ if (found.isEmpty()) {
+ // Refresh the table to update the snapshot list, then try again.
+ refresh();
+ found = getSnapshots().stream()
+ .filter(snapshot -> snapshot.snapshotId() == snapshotId)
+ .findFirst();
+ }
+ return found;
  }
 
  /**
@@ -186,17 +204,27 @@ public synchronized Schema currentSchema() {
  return table.schema();
  }
 
+ /**
+ * Retrieve the current {@link Schema schema} of an Iceberg table.
+ */
+ public synchronized Map<Integer, Schema> schemas() {
+ refresh();
+ return table.schemas();
+ }
+
  /**
  * Retrieve a specific {@link Schema schema} of an Iceberg table.
  *
  * @param schemaId The identifier of the schema to load.
  */
- public synchronized Schema schema(final int schemaId) {
- // TODO: discuss refresh() strategy for this and other functions:
- // 1) ALWAYS refresh() before searching for a match (safe, might be slow)
- // 2) NEVER refresh() before searching (user should call refresh() manually)
- // 3) HYBRID, refresh() if search fails, then re-search
- return table.schemas().get(schemaId);
+ public synchronized Optional<Schema> schema(final int schemaId) {
+ Schema found = table.schemas().get(schemaId);
+ if (found == null) {
+ // Refresh the table to update the snapshot list, then try again.
+ refresh();
+ found = table.schemas().get(schemaId);
+ }
+ return Optional.ofNullable(found);
  }
 
  /**
@@ -252,21 +280,19 @@ public TableDefinition definition(
  @Nullable final Snapshot tableSnapshot,
  @Nullable final IcebergInstructions instructions) {
 
- final Snapshot snapshot;
  final Schema schema;
  final org.apache.iceberg.PartitionSpec partitionSpec;
 
  if (tableSnapshot == null) {
  synchronized (this) {
- // Refresh only once and record the current snapshot, using its schema and spec.
+ // Refresh only once and record the current schema and partition spec.
  refresh();
- snapshot = table.currentSnapshot();
- schema = snapshot != null ? schema(snapshot.schemaId()) : table.schema();
+ schema = table.schema();
  partitionSpec = table.spec();
  }
  } else {
- snapshot = tableSnapshot;
- schema = schema(tableSnapshot.schemaId());
+ // Use the schema from the snapshot
+ schema = schema(tableSnapshot.schemaId()).get();
  partitionSpec = table.spec();
  }
 
@@ -388,15 +414,17 @@ public IcebergTable table(
 
  if (tableSnapshot == null) {
  synchronized (this) {
- // Refresh only once and record the current snapshot, using its schema and spec.
+ // Refresh only once and record the current snapshot, schema (which may be newer than the
+ // snapshot schema), and partition spec.
  refresh();
  snapshot = table.currentSnapshot();
- schema = snapshot != null ? schema(snapshot.schemaId()) : table.schema();
+ schema = table.schema();
  partitionSpec = table.spec();
  }
  } else {
  snapshot = tableSnapshot;
- schema = schema(tableSnapshot.schemaId());
+ // Use the schema from the snapshot
+ schema = schema(tableSnapshot.schemaId()).get();
  partitionSpec = table.spec();
  }
 
@@ -427,8 +455,7 @@ public IcebergTable table(
  finalInstructions, dataInstructionsProviderLoader);
  }
 
- if (instructions == null
- || instructions.updateMode().updateType() == IcebergUpdateMode.IcebergUpdateType.STATIC) {
+ if (finalInstructions.updateMode().updateType() == IcebergUpdateMode.IcebergUpdateType.STATIC) {
  final IcebergTableLocationProviderBase<TableKey, IcebergTableLocationKey> locationProvider =
  new IcebergStaticTableLocationProvider<>(
  StandaloneTableKey.getInstance(),
@@ -447,7 +474,7 @@ public IcebergTable table(
  final UpdateSourceRegistrar updateSourceRegistrar = ExecutionContext.getContext().getUpdateGraph();
  final IcebergTableLocationProviderBase<TableKey, IcebergTableLocationKey> locationProvider;
 
- if (instructions.updateMode().updateType() == IcebergUpdateMode.IcebergUpdateType.MANUAL_REFRESHING) {
+ if (finalInstructions.updateMode().updateType() == IcebergUpdateMode.IcebergUpdateType.MANUAL_REFRESHING) {
  locationProvider = new IcebergManualRefreshTableLocationProvider<>(
  StandaloneTableKey.getInstance(),
  keyFinder,
@@ -460,7 +487,7 @@ public IcebergTable table(
  keyFinder,
  new IcebergTableLocationFactory(),
  TableDataRefreshService.getSharedRefreshService(),
- instructions.updateMode().autoRefreshMs(),
+ finalInstructions.updateMode().autoRefreshMs(),
  this,
  tableIdentifier);
  }

py/server/deephaven/experimental/iceberg.py

@@ -229,14 +229,9 @@ def table(self, instructions: Optional[IcebergInstructions] = None, snapshot_id:
  Table: the table read from the catalog.
  """
 
- if instructions is not None:
- instructions_object = instructions.j_object
- else:
- instructions_object = _JIcebergInstructions.DEFAULT
-
- if snapshot_id is not None:
- return IcebergTable(self.j_object.table(snapshot_id, instructions_object))
- return IcebergTable(self.j_object.readTable(instructions_object))
+ if snapshot_id:
+ return IcebergTable(self.j_object.table(snapshot_id, instructions))
+ return IcebergTable(self.j_object.table(instructions))
 
  @property
  def j_object(self) -> jpy.JType:
@@ -317,33 +312,6 @@ def snapshots(self, table_identifier: str) -> Table:
 
  return self.j_object.snapshots(table_identifier)
 
- def read_table(self, table_identifier: str, instructions: Optional[IcebergInstructions] = None, snapshot_id: Optional[int] = None) -> IcebergTable:
- """
- NOTE: deprecated, use `load_table(table_identifier).table()` instead.
-
- Reads the table from the catalog using the provided instructions. Optionally, a snapshot id can be provided to
- read a specific snapshot of the table.
-
- Args:
- table_identifier (str): the table to read.
- instructions (Optional[IcebergInstructions]): the instructions for reading the table. These instructions
- can include column renames, table definition, and specific data instructions for reading the data files
- from the provider. If omitted, the table will be read with default instructions.
- snapshot_id (Optional[int]): the snapshot id to read; if omitted the most recent snapshot will be selected.
-
- Returns:
- Table: the table read from the catalog.
- """
-
- if instructions is not None:
- instructions_object = instructions.j_object
- else:
- instructions_object = _JIcebergInstructions.DEFAULT
-
- if snapshot_id is not None:
- return IcebergTable(self.j_object.readTable(table_identifier, snapshot_id, instructions_object))
- return IcebergTable(self.j_object.readTable(table_identifier, instructions_object))
-
  @property
  def j_object(self) -> jpy.JType:
  return self.j_catalog_adapter