@@ -216,6 +216,11 @@ impl SqlCatalog {
216216 }
217217 }
218218 }
219+
220+ /// Get the catalogs `FileIO`
221+ pub fn file_io ( & self ) -> FileIO {
222+ self . fileio . clone ( )
223+ }
219224}
220225
221226#[ async_trait]
@@ -765,13 +770,30 @@ impl Catalog for SqlCatalog {
765770
766771 async fn register_table (
767772 & self ,
768- _table_ident : & TableIdent ,
769- _metadata_location : String ,
773+ table_ident : & TableIdent ,
774+ metadata_location : String ,
770775 ) -> Result < Table > {
771- Err ( Error :: new (
772- ErrorKind :: FeatureUnsupported ,
773- "Registering a table is not supported yet" ,
774- ) )
776+ if !self . table_exists ( table_ident) . await ? {
777+ return no_such_table_err ( table_ident) ;
778+ }
779+
780+ let metadata = TableMetadata :: read_from ( & self . fileio , metadata_location. clone ( ) ) . await ?;
781+
782+ let namespace = table_ident. namespace ( ) ;
783+ let tbl_name = table_ident. name ( ) . to_string ( ) ;
784+
785+ self . execute ( & format ! (
786+ "INSERT INTO {CATALOG_TABLE_NAME}
787+ ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}, {CATALOG_FIELD_METADATA_LOCATION_PROP}, {CATALOG_FIELD_RECORD_TYPE})
788+ VALUES (?, ?, ?, ?, ?)
789+ " ) , vec ! [ Some ( & self . name) , Some ( & namespace. join( "." ) ) , Some ( & tbl_name) , Some ( & metadata_location) , Some ( CATALOG_FIELD_TABLE_RECORD_TYPE ) ] , None ) . await ?;
790+
791+ Ok ( Table :: builder ( )
792+ . identifier ( table_ident. clone ( ) )
793+ . metadata_location ( metadata_location)
794+ . metadata ( metadata)
795+ . file_io ( self . file_io ( ) )
796+ . build ( ) ?)
775797 }
776798
777799 async fn update_table ( & self , _commit : TableCommit ) -> Result < Table > {
0 commit comments