feat: Use REST Catalog config properties for DataInstructionsProvider…

…Plugin (deephaven#6191)

The very first call to a REST Catalog is to fetch a
[CatalogConfig](https://github.com/apache/iceberg/blob/apache-iceberg-1.6.1/open-api/rest-catalog-open-api.yaml#L1564-L1583)
object via
[/v1/config](https://github.com/apache/iceberg/blob/apache-iceberg-1.6.1/open-api/rest-catalog-open-api.yaml#L65-L131).
This may include additional properties needed for further REST Catalog
API calls (Iceberg handles this at internally in
`org.apache.iceberg.rest.RESTSessionCatalog`). The additional properties
may also contain credentials to access a warehouse, like S3. This sort
of access control is implementation dependant, but this is the model
that [Polaris REST Catalog](https://polaris.apache.org/) uses.

Iceberg exposes the full set of properties after initialization via
`org.apache.iceberg.rest.RESTCatalog#properties`.

This PR passes along the full set of properties for REST Catalogs to
`DataInstructionsProviderPlugin`.

extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java

@@ -68,8 +68,7 @@ public static IcebergCatalogAdapter createS3Rest(
 
         final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI;
         catalog.initialize(catalogName, properties);
-
-        return new IcebergCatalogAdapter(catalog, properties);
+        return IcebergCatalogAdapter.of(catalog);
     }
 
     /**

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

@@ -19,6 +19,7 @@
 import io.deephaven.engine.updategraph.UpdateSourceRegistrar;
 import io.deephaven.engine.util.TableTools;
 import io.deephaven.iceberg.internal.DataInstructionsProviderLoader;
+import io.deephaven.iceberg.internal.DataInstructionsProviderPlugin;
 import io.deephaven.iceberg.layout.IcebergFlatLayout;
 import io.deephaven.iceberg.layout.IcebergKeyValuePartitionedLayout;
 import io.deephaven.iceberg.location.IcebergTableLocationFactory;
@@ -33,6 +34,8 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.ResourcePaths;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.jetbrains.annotations.NotNull;
@@ -68,9 +71,40 @@ public class IcebergCatalogAdapter {
 
     private final DataInstructionsProviderLoader dataInstructionsProvider;
 
+    /**
+     * Construct an IcebergCatalogAdapter from a Catalog. The properties supplied are provided to support
+     * {@link DataInstructionsProviderPlugin} resolution. In the case where {@code catalog} is a {@link RESTCatalog},
+     * {@link RESTCatalog#properties()} will be used instead.
+     *
+     * @param catalog the catalog
+     * @return the catalog adapter
+     */
+    static IcebergCatalogAdapter of(Catalog catalog, Map<String, String> properties) {
+        if (catalog instanceof RESTCatalog) {
+            return of((RESTCatalog) catalog);
+        }
+        return new IcebergCatalogAdapter(catalog, properties);
+    }
+
+    /**
+     * Construct an IcebergCatalogAdapter from a REST catalog. This passes along {@link RESTCatalog#properties()} which
+     * will include any additional properties the REST Catalog implementation sent back as part of the initial
+     * {@link ResourcePaths#config() config} call. These properties will be used for resolving
+     * {@link io.deephaven.iceberg.internal.DataInstructionsProviderPlugin}.
+     *
+     * @param restCatalog the rest catalog
+     * @return the catalog adapter
+     */
+    static IcebergCatalogAdapter of(RESTCatalog restCatalog) {
+        return new IcebergCatalogAdapter(restCatalog, restCatalog.properties());
+    }
+
     /**
      * Construct an IcebergCatalogAdapter from a catalog.
+     *
+     * @deprecated use a method or constructor which is more explicit about the properties it uses
      */
+    @Deprecated(forRemoval = true)
     IcebergCatalogAdapter(@NotNull final Catalog catalog) {
         this(catalog, Map.of());
     }

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

@@ -19,7 +19,7 @@ public abstract class IcebergTools {
     @SuppressWarnings("unused")
     public static IcebergCatalogAdapter createAdapter(
             final Catalog catalog) {
-        return new IcebergCatalogAdapter(catalog);
+        return IcebergCatalogAdapter.of(catalog, Map.of());
     }
 
     /**
@@ -125,7 +125,7 @@ public static IcebergCatalogAdapter createAdapter(
         // Create the Iceberg catalog from the properties
         final Catalog catalog = CatalogUtil.buildIcebergCatalog(catalogName, properties, hadoopConf);
 
-        return new IcebergCatalogAdapter(catalog, properties);
+        return IcebergCatalogAdapter.of(catalog, properties);
     }
 
 }