Merge branch 'main' into ctty/name-pos

Author: CTTY

CONTRIBUTING.md

@@ -104,8 +104,7 @@ cargo 1.69.0 (6e9a83356 2023-04-12)
 Currently, iceberg-rust uses Docker to set up environment for integration tests. Native Docker has some limitations, please check (https://github.com/apache/iceberg-rust/pull/748). Please use Orbstack or Podman.
 
 For MacOS users, you can install [OrbStack as a docker alternative](docs/contributing/orbstack.md).
-
-For podman, refer to [Using Podman instead of Docker](docs/contributing/podman.md)
+For Podman users, refer to [Using Podman instead of Docker](docs/contributing/podman.md)
 
 ## Build
 

crates/catalog/rest/src/catalog.rs

@@ -42,7 +42,7 @@ use crate::client::{
 use crate::types::{
     CatalogConfig, CommitTableRequest, CommitTableResponse, CreateTableRequest,
     ListNamespaceResponse, ListTableResponse, LoadTableResponse, NamespaceSerde,
-    RenameTableRequest,
+    RegisterTableRequest, RenameTableRequest,
 };
 
 const ICEBERG_REST_SPEC_VERSION: &str = "0.14.1";
@@ -101,6 +101,10 @@ impl RestCatalogConfig {
         self.url_prefixed(&["tables", "rename"])
     }
 
+    fn register_table_endpoint(&self, ns: &NamespaceIdent) -> String {
+        self.url_prefixed(&["namespaces", &ns.to_url_string(), "register"])
+    }
+
     fn table_endpoint(&self, table: &TableIdent) -> String {
         self.url_prefixed(&[
             "namespaces",
@@ -238,7 +242,7 @@ struct RestContext {
 pub struct RestCatalog {
     /// User config is stored as-is and never be changed.
     ///
-    /// It's could be different from the config fetched from the server and used at runtime.
+    /// It could be different from the config fetched from the server and used at runtime.
     user_config: RestCatalogConfig,
     ctx: OnceCell<RestContext>,
     /// Extensions for the FileIOBuilder.
@@ -755,13 +759,60 @@ impl Catalog for RestCatalog {
 
     async fn register_table(
         &self,
-        _table_ident: &TableIdent,
-        _metadata_location: String,
+        table_ident: &TableIdent,
+        metadata_location: String,
     ) -> Result<Table> {
-        Err(Error::new(
-            ErrorKind::FeatureUnsupported,
-            "Registering a table is not supported yet",
-        ))
+        let context = self.context().await?;
+
+        let request = context
+            .client
+            .request(
+                Method::POST,
+                context
+                    .config
+                    .register_table_endpoint(table_ident.namespace()),
+            )
+            .json(&RegisterTableRequest {
+                name: table_ident.name.clone(),
+                metadata_location: metadata_location.clone(),
+                overwrite: Some(false),
+            })
+            .build()?;
+
+        let http_response = context.client.query_catalog(request).await?;
+
+        let response: LoadTableResponse = match http_response.status() {
+            StatusCode::OK => {
+                deserialize_catalog_response::<LoadTableResponse>(http_response).await?
+            }
+            StatusCode::NOT_FOUND => {
+                return Err(Error::new(
+                    ErrorKind::NamespaceNotFound,
+                    "The namespace specified does not exist.",
+                ));
+            }
+            StatusCode::CONFLICT => {
+                return Err(Error::new(
+                    ErrorKind::TableAlreadyExists,
+                    "The given table already exists.",
+                ));
+            }
+            _ => return Err(deserialize_unexpected_catalog_error(http_response).await),
+        };
+
+        let metadata_location = response.metadata_location.as_ref().ok_or(Error::new(
+            ErrorKind::DataInvalid,
+            "Metadata location missing in `register_table` response!",
+        ))?;
+
+        let file_io = self.load_file_io(Some(metadata_location), None).await?;
+
+        Table::builder()
+            .identifier(table_ident.clone())
+            .file_io(file_io)
+            .metadata(response.metadata)
+            .metadata_location(metadata_location.clone())
+            .build()
     }
 
     async fn update_table(&self, mut commit: TableCommit) -> Result<Table> {
@@ -2470,4 +2521,87 @@ mod tests {
         update_table_mock.assert_async().await;
         load_table_mock.assert_async().await;
     }
+
+    #[tokio::test]
+    async fn test_register_table() {
+        let mut server = Server::new_async().await;
+
+        let config_mock = create_config_mock(&mut server).await;
+
+        let register_table_mock = server
+            .mock("POST", "/v1/namespaces/ns1/register")
+            .with_status(200)
+            .with_body_from_file(format!(
+                "{}/testdata/{}",
+                env!("CARGO_MANIFEST_DIR"),
+                "load_table_response.json"
+            ))
+            .create_async()
+            .await;
+
+        let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build());
+        let table_ident =
+            TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string());
+        let metadata_location = String::from(
+            "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
+        );
+
+        let table = catalog
+            .register_table(&table_ident, metadata_location)
+            .await
+            .unwrap();
+
+        assert_eq!(
+            &TableIdent::from_strs(vec!["ns1", "test1"]).unwrap(),
+            table.identifier()
+        );
+        assert_eq!(
+            "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
+            table.metadata_location().unwrap()
+        );
+
+        config_mock.assert_async().await;
+        register_table_mock.assert_async().await;
+    }
+
+    #[tokio::test]
+    async fn test_register_table_404() {
+        let mut server = Server::new_async().await;
+
+        let config_mock = create_config_mock(&mut server).await;
+
+        let register_table_mock = server
+            .mock("POST", "/v1/namespaces/ns1/register")
+            .with_status(404)
+            .with_body(
+                r#"
+{
+    "error": {
+        "message": "The namespace specified does not exist",
+        "type": "NoSuchNamespaceErrorException",
+        "code": 404
+    }
+}
+            "#,
+            )
+            .create_async()
+            .await;
+
+        let catalog = RestCatalog::new(RestCatalogConfig::builder().uri(server.url()).build());
+
+        let table_ident =
+            TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string());
+        let metadata_location = String::from(
+            "s3://warehouse/database/table/metadata/00001-5f2f8166-244c-4eae-ac36-384ecdec81fc.gz.metadata.json",
+        );
+        let table = catalog
+            .register_table(&table_ident, metadata_location)
+            .await;
+
+        assert!(table.is_err());
+        assert!(table.err().unwrap().message().contains("does not exist"));
+
+        config_mock.assert_async().await;
+        register_table_mock.assert_async().await;
+    }
 }

crates/catalog/rest/src/types.rs

@@ -191,3 +191,11 @@ pub(super) struct CommitTableResponse {
     pub(super) metadata_location: String,
     pub(super) metadata: TableMetadata,
 }
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub(super) struct RegisterTableRequest {
+    pub(super) name: String,
+    pub(super) metadata_location: String,
+    pub(super) overwrite: Option<bool>,
+}

crates/catalog/rest/tests/rest_catalog_test.rs

@@ -407,3 +407,39 @@ async fn test_list_empty_multi_level_namespace() {
         .unwrap();
     assert!(nss.is_empty());
 }
+
+#[tokio::test]
+async fn test_register_table() {
+    let catalog = get_catalog().await;
+
+    // Create namespace
+    let ns = NamespaceIdent::from_strs(["ns"]).unwrap();
+    catalog.create_namespace(&ns, HashMap::new()).await.unwrap();
+
+    // Create the table, store the metadata location, drop the table
+    let empty_schema = Schema::builder().build().unwrap();
+    let table_creation = TableCreation::builder()
+        .name("t1".to_string())
+        .schema(empty_schema)
+        .build();
+
+    let table = catalog.create_table(&ns, table_creation).await.unwrap();
+
+    let metadata_location = table.metadata_location().unwrap();
+    catalog.drop_table(table.identifier()).await.unwrap();
+
+    let new_table_identifier = TableIdent::from_strs(["ns", "t2"]).unwrap();
+    let table_registered = catalog
+        .register_table(&new_table_identifier, metadata_location.to_string())
+        .await
+        .unwrap();
+
+    assert_eq!(
+        table.metadata_location(),
+        table_registered.metadata_location()
+    );
+    assert_ne!(
+        table.identifier().to_string(),
+        table_registered.identifier().to_string()
+    );
+}

crates/integration_tests/testdata/docker-compose.yaml

@@ -80,6 +80,3 @@ services:
       - AWS_ACCESS_KEY_ID=admin
       - AWS_SECRET_ACCESS_KEY=password
       - AWS_REGION=us-east-1
-    links:
-      - rest:rest
-      - minio:minio

crates/integration_tests/testdata/spark/provision.py

@@ -30,6 +30,8 @@
         .getOrCreate()
 )
 
+spark.sql(f"""CREATE NAMESPACE IF NOT EXISTS rest.default""")
+
 spark.sql(
     f"""
 CREATE OR REPLACE TABLE rest.default.test_positional_merge_on_read_deletes (

crates/integrations/datafusion/src/lib.rs

@@ -21,7 +21,7 @@ pub use catalog::*;
 mod error;
 pub use error::*;
 
-mod physical_plan;
+pub mod physical_plan;
 mod schema;
 pub mod table;
 pub use table::table_provider_factory::IcebergTableProviderFactory;

crates/integrations/datafusion/src/physical_plan/mod.rs

@@ -18,3 +18,4 @@
 pub(crate) mod expr_to_predicate;
 pub(crate) mod metadata_scan;
 pub(crate) mod scan;
+pub use scan::IcebergTableScan;

crates/integrations/datafusion/src/physical_plan/scan.rs

@@ -39,7 +39,7 @@ use crate::to_datafusion_error;
 /// Manages the scanning process of an Iceberg [`Table`], encapsulating the
 /// necessary details and computed properties required for execution planning.
 #[derive(Debug)]
-pub(crate) struct IcebergTableScan {
+pub struct IcebergTableScan {
     /// A table in the catalog.
     table: Table,
     /// Snapshot of the table to scan.
@@ -79,6 +79,22 @@ impl IcebergTableScan {
         }
     }
 
+    pub fn table(&self) -> &Table {
+        &self.table
+    }
+
+    pub fn snapshot_id(&self) -> Option<i64> {
+        self.snapshot_id
+    }
+
+    pub fn projection(&self) -> Option<&[String]> {
+        self.projection.as_deref()
+    }
+
+    pub fn predicates(&self) -> Option<&Predicate> {
+        self.predicates.as_ref()
+    }
+
     /// Computes [`PlanProperties`] used in query optimization.
     fn compute_properties(schema: ArrowSchemaRef) -> PlanProperties {
         // TODO:

docs/contributing/podman.md

@@ -19,24 +19,30 @@
 
 # Using Podman instead of Docker
 
-Iceberg-rust does not require containerization, except for integration tests, where "docker" and "docker-compose" are used to start containers for minio and various catalogs. Below instructions setup "rootful podman" and docker's official docker-compose plugin to run integration tests as an alternative to docker or Orbstack. 
+Iceberg-rust does not require containerization, except for integration tests, where "docker" and "docker-compose" are used to start containers for minio and various catalogs. Below instructions setup "rootful podman" and docker's official docker-compose plugin to run integration tests as an alternative to docker or Orbstack.
 
 1. Have podman v4 or newer.
-    ```console
+    ```shell
     $ podman --version
     podman version 4.9.4-rhel
     ```
-
-2. Open file `/usr/bin/docker` and add the below contents:
+2. Create a docker wrapper script:
+   
+  * Create a fresh `/usr/bin/docker` file and add the below contents:
     ```bash
     #!/bin/sh
     [ -e /etc/containers/nodocker ] || \
     echo "Emulate Docker CLI using podman. Create /etc/containers/nodocker to quiet msg." >&2
     exec sudo /usr/bin/podman "$@"
     ```
 
+  * Set new `/usr/bin/docker` file to executable.
+    ```shell
+    sudo chmod +x /usr/bin/docker
+    ```
+
 3. Install the [docker compose plugin](https://docs.docker.com/compose/install/linux). Check for successful installation.
-    ```console
+    ```shell
     $ docker compose version
     Docker Compose version v2.28.1
     ```
@@ -53,7 +59,7 @@ Iceberg-rust does not require containerization, except for integration tests, wh
     ```
 
 6. Check that the following symlink exists.
-    ```console
+    ```shell
     $ ls -al /var/run/docker.sock
     lrwxrwxrwx 1 root root 27 Jul 24 12:18 /var/run/docker.sock -> /var/run/podman/podman.sock
     ```
@@ -83,3 +89,12 @@ As of podman v4, ["To be succinct and simple, when running rootless containers,
 
 * <https://www.redhat.com/sysadmin/container-ip-address-podman>
 * <https://github.com/containers/podman/blob/main/docs/tutorials/basic_networking.md>
+
+# Debugging Note:
+ - Fix for error: `Error: short-name "apache/iceberg-rest-fixture" did not resolve to an alias and no unqualified-search registries are defined in "/etc/containers/registries.conf"`
+    - Add or modify the `/etc/containers/registries.conf` file:
+    ```toml
+    [[registry]]
+    prefix = "docker.io"
+    location = "docker.io"
+    ```