I/O: Apache Iceberg (support more catalogs and storage backends)

Author: amotl

cratedb_toolkit/io/iceberg.py

@@ -9,7 +9,7 @@
 import logging
 import tempfile
 from copy import copy
-from typing import Optional
+from typing import Dict, List, Optional
 
 import pandas as pd
 import polars as pl
@@ -83,37 +83,45 @@ def load_catalog(self) -> Catalog:
         """
         Load the Iceberg catalog with appropriate configuration.
         """
-        # TODO: Consider accepting catalog configuration as parameters
-        #       to support different catalog types (Hive, REST, etc.).
         return load_catalog(self.catalog, **self.catalog_properties)
 
     @property
     def catalog_properties(self):
         """
         Provide Iceberg catalog properties.
+        https://py.iceberg.apache.org/configuration/#catalogs
         https://py.iceberg.apache.org/reference/pyiceberg/catalog/
         """
-        return {
+        opts = {
             "uri": self.url.query_params.get(
                 "catalog-uri", f"sqlite:///{self.temporary_catalog_location}/pyiceberg_catalog.db"
             ),
+            "credential": self.url.query_params.get("catalog-credential"),
             "token": self.url.query_params.get("catalog-token"),
+            "type": self.url.query_params.get("catalog-type"),
             "warehouse": self.location,  # TODO: Is the `file://` prefix faster when accessing the local filesystem?
         }
+        prefixes = ["dynamodb.", "gcp.", "glue."]
+        opts.update(self.collect_properties(self.url.query_params, prefixes))
+        return opts
 
     @property
     def storage_options(self):
-        opts = {
-            "s3.endpoint": self.url.query_params.get("s3.endpoint"),
-            "s3.region": self.url.query_params.get("s3.region"),
-            "s3.access-key-id": self.url.query_params.get("s3.access-key-id"),
-            "s3.secret-access-key": self.url.query_params.get("s3.secret-access-key"),
-        }
-        return {k: v for k, v in opts.items() if v is not None}
         """
         Provide Iceberg storage properties.
         https://py.iceberg.apache.org/configuration/#fileio
         """
+        prefixes = ["adls.", "gcs.", "hdfs.", "hf.", "s3."]
+        return self.collect_properties(self.url.query_params, prefixes)
+
+    @staticmethod
+    def collect_properties(query_params: Dict, prefixes: List) -> Dict[str, str]:
+        opts = {}
+        for name, value in query_params.items():
+            for prefix in prefixes:
+                if name.startswith(prefix) and value is not None:
+                    opts[name] = value
+        return opts
 
     @property
     def identifier(self):

doc/backlog/io.md

@@ -17,3 +17,7 @@
 
 - Auth!
   https://github.com/apache/iceberg/issues/13550
+
+- Can the postgresql catalog implementation be used with CrateDB?
+
+- Verify with other catalogs: https://py.iceberg.apache.org/configuration/#common-integrations-examples

doc/io/iceberg/index.md

@@ -6,17 +6,17 @@
 
 Import and export data into/from [Apache Iceberg] tables, for humans and machines.
 
-Iceberg works with the concept of a [FileIO] which is a pluggable module for
-reading, writing, and deleting files. It supports different backends like
-S3, HDFS, Azure Data Lake, Google Cloud Storage, Alibaba Cloud Object Storage,
-and Hugging Face.
-
 ## Synopsis
 
-- Load from Iceberg table: `ctk load table file+iceberg://...`,
-  `ctk load table s3+iceberg://...`
+Load from Iceberg table:
+```shell
+ctk load table {file,s3,abfs,gs,hdfs}+iceberg://...
+```
 
-- Export to Iceberg table: `ctk save table file+iceberg://...`
+Export to Iceberg table:
+```shell
+ctk save table {file,s3,abfs,gs,hdfs}+iceberg://...
+```
 
 ## Install
 
@@ -32,6 +32,14 @@ other operating systems.
 
 ## Usage
 
+Iceberg works with the concept of a [FileIO] which is a pluggable module for
+reading, writing, and deleting files. It supports different backends like
+S3, HDFS, Azure Data Lake, Google Cloud Storage, Alibaba Cloud Object Storage,
+and Hugging Face.
+
+Please look up available configuration parameters in the reference documentation,
+otherwise derive your ETL commands from the examples shared below.
+
 ### Load
 
 Load from metadata file on filesystem.
@@ -48,18 +56,55 @@ ctk load table \
     --cluster-url="crate://crate:crate@localhost:4200/demo/taxi-tiny"
 ```
 
-Load from REST catalog and AWS S3 storage.
+Use REST catalog and AWS S3 storage.
 ```shell
 ctk load table \
     "s3+iceberg://bucket1/?catalog-uri=http://iceberg-catalog.example.org:5000&catalog-token=foo&catalog=default&namespace=demo&table=taxi-tiny&s3.access-key-id=<your_access_key_id>&s3.secret-access-key=<your_secret_access_key>&s3.endpoint=<endpoint_url>&s3.region=<s3-region>" \
     --cluster-url="crate://crate:crate@localhost:4200/demo/taxi-tiny"
 ```
 
-Query data in CrateDB.
+Use catalog in Apache Hive.
+```shell
+ctk load table "s3+iceberg://bucket1/?catalog-uri=thrift://localhost:9083/&catalog-credential=t-1234:secret&..."
+```
+
+Use catalog in AWS Glue.
+```shell
+ctk load table "s3+iceberg://bucket1/?catalog-type=glue&glue.id=foo&glue.profile-name=bar&glue.region=region&glue.access-key-id=key&glue.secret-access-key=secret&..."
+```
+
+Use catalog in Google BigQuery.
+```shell
+ctk load table "s3+iceberg://bucket1/?catalog-type=bigquery&gcp.bigquery.project-id=foo&..."
+```
+
+Use catalog in DynamoDB.
+```shell
+ctk load table "s3+iceberg://bucket1/?catalog-type=dynamodb&dynamodb.profile-name=foo&dynamodb.region=bar&dynamodb.access-key-id=key&dynamodb.secret-access-key=secret&..."
+```
+
+Load data from Azure Data Lake Storage.
+```shell
+ctk load table "abfs+iceberg://container/path/?adls.account-name=devstoreaccount1&adls.account-key=foo&..."
+```
+
+Load data from Google Cloud Storage.
+```shell
+ctk load table "gs+iceberg://bucket?gcs.project-id=..."
+```
+
+Load data from HDFS Storage.
+```shell
+ctk load table "hdfs+iceberg://path?hdfs.host=https://10.0.19.25/&hdfs.port=9000&hdfs.user=&hdfs.kerberos_ticket="
+```
+
+:::{tip}
+After loading your data into CrateDB, query it.
 ```shell
 ctk shell --command 'SELECT * FROM demo."taxi-tiny";'
 ctk show table 'demo."taxi-tiny"'
 ```
+:::
 
 ### Save
 
@@ -77,6 +122,8 @@ ctk save table \
     "s3+iceberg://bucket1/?catalog=default&namespace=demo&table=taxi-tiny&s3.access-key-id=<your_access_key_id>&s3.secret-access-key=<your_secret_access_key>&s3.endpoint=<endpoint_url>&s3.region=<s3-region>"
 ```
 
+For other target URLs, see "Source" section.
+
 ### Cloud
 
 A canonical invocation for copying data from an Iceberg table on AWS S3 to CrateDB Cloud.
@@ -141,7 +188,16 @@ to a truthy value, save operations will append to an existing table.
 ctk save table "file+iceberg://./var/lib/iceberg/?...&append=true"
 ```
 
+#### PyIceberg
+
+The PyIceberg I/O adapters accept a plethora of options that can be used 1:1.
+For a list of all available options, please consult the [FileIO] documentation.
+For I/O adapters not part of the documentation yet, please consult the source
+code about [catalog options] and [storage options].
+
 
 [Apache Iceberg]: https://iceberg.apache.org/
+[catalog options]: https://github.com/apache/iceberg-python/tree/main/pyiceberg/catalog
 [FileIO]: https://py.iceberg.apache.org/configuration/#fileio
+[storage options]: https://github.com/apache/iceberg-python/tree/main/pyiceberg/io
 [uv]: https://docs.astral.sh/uv/

pyproject.toml

@@ -182,7 +182,7 @@ optional-dependencies.io-base = [
   "fsspec[http,s3]",
   "pandas>=1,<3.1",
   "polars<1.39",
-  "pyiceberg[adlfs,gcsfs,hive,pyarrow,s3fs]<0.12",
+  "pyiceberg[adlfs,bigquery,dynamodb,gcsfs,glue,hive,pyarrow,s3fs]<0.12",
   "universal-pathlib<0.4",
 ]
 optional-dependencies.io-ingestr = [