@@ -765,13 +765,30 @@ impl Catalog for SqlCatalog {
765765
766766 async fn register_table (
767767 & self ,
768- _table_ident : & TableIdent ,
769- _metadata_location : String ,
768+ table_ident : & TableIdent ,
769+ metadata_location : String ,
770770 ) -> Result < Table > {
771- Err ( Error :: new (
772- ErrorKind :: FeatureUnsupported ,
773- "Registering a table is not supported yet" ,
774- ) )
771+ if self . table_exists ( table_ident) . await ? {
772+ return table_already_exists_err ( table_ident) ;
773+ }
774+
775+ let metadata = TableMetadata :: read_from ( & self . fileio , & metadata_location) . await ?;
776+
777+ let namespace = table_ident. namespace ( ) ;
778+ let tbl_name = table_ident. name ( ) . to_string ( ) ;
779+
780+ self . execute ( & format ! (
781+ "INSERT INTO {CATALOG_TABLE_NAME}
782+ ({CATALOG_FIELD_CATALOG_NAME}, {CATALOG_FIELD_TABLE_NAMESPACE}, {CATALOG_FIELD_TABLE_NAME}, {CATALOG_FIELD_METADATA_LOCATION_PROP}, {CATALOG_FIELD_RECORD_TYPE})
783+ VALUES (?, ?, ?, ?, ?)
784+ " ) , vec ! [ Some ( & self . name) , Some ( & namespace. join( "." ) ) , Some ( & tbl_name) , Some ( & metadata_location) , Some ( CATALOG_FIELD_TABLE_RECORD_TYPE ) ] , None ) . await ?;
785+
786+ Ok ( Table :: builder ( )
787+ . identifier ( table_ident. clone ( ) )
788+ . metadata_location ( metadata_location)
789+ . metadata ( metadata)
790+ . file_io ( self . fileio . clone ( ) )
791+ . build ( ) ?)
775792 }
776793
777794 async fn update_table ( & self , _commit : TableCommit ) -> Result < Table > {
@@ -1908,4 +1925,63 @@ mod tests {
19081925 "Unexpected => No such table: TableIdent { namespace: NamespaceIdent([\" a\" ]), name: \" tbl1\" }"
19091926 ) ;
19101927 }
1928+
1929+ #[ tokio:: test]
1930+ async fn test_register_table_throws_error_if_table_with_same_name_already_exists ( ) {
1931+ let warehouse_loc = temp_path ( ) ;
1932+ let catalog = new_sql_catalog ( warehouse_loc. clone ( ) ) . await ;
1933+ let namespace_ident = NamespaceIdent :: new ( "a" . into ( ) ) ;
1934+ create_namespace ( & catalog, & namespace_ident) . await ;
1935+ let table_name = "tbl1" ;
1936+ let table_ident = TableIdent :: new ( namespace_ident. clone ( ) , table_name. into ( ) ) ;
1937+ create_table ( & catalog, & table_ident) . await ;
1938+
1939+ assert_eq ! (
1940+ catalog
1941+ . register_table( & table_ident, warehouse_loc)
1942+ . await
1943+ . unwrap_err( )
1944+ . to_string( ) ,
1945+ format!( "Unexpected => Table {:?} already exists." , & table_ident)
1946+ ) ;
1947+ }
1948+
1949+ #[ tokio:: test]
1950+ async fn test_register_table ( ) {
1951+ let warehouse_loc = temp_path ( ) ;
1952+ let catalog = new_sql_catalog ( warehouse_loc. clone ( ) ) . await ;
1953+ let namespace_ident = NamespaceIdent :: new ( "a" . into ( ) ) ;
1954+ create_namespace ( & catalog, & namespace_ident) . await ;
1955+
1956+ let table_name = "abc" ;
1957+ let location = warehouse_loc. clone ( ) ;
1958+ let table_creation = TableCreation :: builder ( )
1959+ . name ( table_name. into ( ) )
1960+ . location ( location. clone ( ) )
1961+ . schema ( simple_table_schema ( ) )
1962+ . build ( ) ;
1963+
1964+ let table_ident = TableIdent :: new ( namespace_ident. clone ( ) , table_name. into ( ) ) ;
1965+ let expected_table = catalog
1966+ . create_table ( & namespace_ident, table_creation)
1967+ . await
1968+ . unwrap ( ) ;
1969+
1970+ let metadata_location = expected_table
1971+ . metadata_location ( )
1972+ . expect ( "Expected metadata location to be set" )
1973+ . to_string ( ) ;
1974+
1975+ assert_table_eq ( & expected_table, & table_ident, & simple_table_schema ( ) ) ;
1976+
1977+ let _ = catalog. drop_table ( & table_ident) . await ;
1978+
1979+ let table = catalog
1980+ . register_table ( & table_ident, metadata_location. clone ( ) )
1981+ . await
1982+ . unwrap ( ) ;
1983+
1984+ assert_eq ! ( table. identifier( ) , expected_table. identifier( ) ) ;
1985+ assert_eq ! ( table. metadata_location( ) , Some ( metadata_location. as_str( ) ) ) ;
1986+ }
19111987}
0 commit comments