feat(catalog): Add register_table to Catalog trait (#1509)

## Which issue does this PR close?



- Part of #1508 

## What changes are included in this PR?

- Added `register_table` to `Catalog` trait
- Implemented `register_table` for `MemoryCatalog`

## Are these changes tested?
Added an unit test for the implementation in `MemoryCatalog`

Author: CTTY

crates/catalog/glue/src/catalog.rs

@@ -628,6 +628,17 @@ impl Catalog for GlueCatalog {
         }
     }
 
+    async fn register_table(
+        &self,
+        _table_ident: &TableIdent,
+        _metadata_location: String,
+    ) -> Result<Table> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Registering a table is not supported yet",
+        ))
+    }
+
     async fn update_table(&self, _commit: TableCommit) -> Result<Table> {
         Err(Error::new(
             ErrorKind::FeatureUnsupported,

crates/catalog/hms/src/catalog.rs

@@ -504,6 +504,17 @@ impl Catalog for HmsCatalog {
         Ok(())
     }
 
+    async fn register_table(
+        &self,
+        _table_ident: &TableIdent,
+        _metadata_location: String,
+    ) -> Result<Table> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Registering a table is not supported yet",
+        ))
+    }
+
     async fn update_table(&self, _commit: TableCommit) -> Result<Table> {
         Err(Error::new(
             ErrorKind::FeatureUnsupported,

crates/catalog/rest/src/catalog.rs

@@ -740,6 +740,17 @@ impl Catalog for RestCatalog {
         }
     }
 
+    async fn register_table(
+        &self,
+        _table_ident: &TableIdent,
+        _metadata_location: String,
+    ) -> Result<Table> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Registering a table is not supported yet",
+        ))
+    }
+
     async fn update_table(&self, mut commit: TableCommit) -> Result<Table> {
         let context = self.context().await?;
 

crates/catalog/s3tables/src/catalog.rs

@@ -473,6 +473,17 @@ impl Catalog for S3TablesCatalog {
         Ok(())
     }
 
+    async fn register_table(
+        &self,
+        _table_ident: &TableIdent,
+        _metadata_location: String,
+    ) -> Result<Table> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Registering a table is not supported yet",
+        ))
+    }
+
     /// Updates an existing table within the s3tables catalog.
     ///
     /// This function is still in development and will always return an error.

crates/catalog/sql/src/catalog.rs

@@ -769,6 +769,17 @@ impl Catalog for SqlCatalog {
         Ok(())
     }
 
+    async fn register_table(
+        &self,
+        _table_ident: &TableIdent,
+        _metadata_location: String,
+    ) -> Result<Table> {
+        Err(Error::new(
+            ErrorKind::FeatureUnsupported,
+            "Registering a table is not supported yet",
+        ))
+    }
+
     async fn update_table(&self, _commit: TableCommit) -> Result<Table> {
         Err(Error::new(
             ErrorKind::FeatureUnsupported,

crates/iceberg/src/catalog/memory/catalog.rs

@@ -276,6 +276,26 @@ impl Catalog for MemoryCatalog {
         Ok(())
     }
 
+    async fn register_table(
+        &self,
+        table_ident: &TableIdent,
+        metadata_location: String,
+    ) -> Result<Table> {
+        let mut root_namespace_state = self.root_namespace_state.lock().await;
+        root_namespace_state.insert_new_table(&table_ident.clone(), metadata_location.clone())?;
+
+        let input_file = self.file_io.new_input(metadata_location.clone())?;
+        let metadata_content = input_file.read().await?;
+        let metadata = serde_json::from_slice::<TableMetadata>(&metadata_content)?;
+
+        Table::builder()
+            .file_io(self.file_io.clone())
+            .metadata_location(metadata_location)
+            .metadata(metadata)
+            .identifier(table_ident.clone())
+            .build()
+    }
+
     /// Update a table to the catalog.
     async fn update_table(&self, _commit: TableCommit) -> Result<Table> {
         Err(Error::new(
@@ -1697,4 +1717,48 @@ mod tests {
             ),
         );
     }
+
+    #[tokio::test]
+    async fn test_register_table() {
+        // Create a catalog and namespace
+        let catalog = new_memory_catalog();
+        let namespace_ident = NamespaceIdent::new("test_namespace".into());
+        create_namespace(&catalog, &namespace_ident).await;
+
+        // Create a table to get a valid metadata file
+        let source_table_ident = TableIdent::new(namespace_ident.clone(), "source_table".into());
+        create_table(&catalog, &source_table_ident).await;
+
+        // Get the metadata location from the source table
+        let source_table = catalog.load_table(&source_table_ident).await.unwrap();
+        let metadata_location = source_table.metadata_location().unwrap().to_string();
+
+        // Register a new table using the same metadata location
+        let register_table_ident =
+            TableIdent::new(namespace_ident.clone(), "register_table".into());
+        let registered_table = catalog
+            .register_table(&register_table_ident, metadata_location.clone())
+            .await
+            .unwrap();
+
+        // Verify the registered table has the correct identifier
+        assert_eq!(registered_table.identifier(), &register_table_ident);
+
+        // Verify the registered table has the correct metadata location
+        assert_eq!(
+            registered_table.metadata_location().unwrap().to_string(),
+            metadata_location
+        );
+
+        // Verify the table exists in the catalog
+        assert!(catalog.table_exists(&register_table_ident).await.unwrap());
+
+        // Verify we can load the registered table
+        let loaded_table = catalog.load_table(&register_table_ident).await.unwrap();
+        assert_eq!(loaded_table.identifier(), &register_table_ident);
+        assert_eq!(
+            loaded_table.metadata_location().unwrap().to_string(),
+            metadata_location
+        );
+    }
 }

crates/iceberg/src/catalog/mod.rs

@@ -100,6 +100,9 @@ pub trait Catalog: Debug + Sync + Send {
     /// Rename a table in the catalog.
     async fn rename_table(&self, src: &TableIdent, dest: &TableIdent) -> Result<()>;
 
+    /// Register an existing table to the catalog.
+    async fn register_table(&self, table: &TableIdent, metadata_location: String) -> Result<Table>;
+
     /// Update a table to the catalog.
     async fn update_table(&self, commit: TableCommit) -> Result<Table>;
 }