refactor: update CatalogInfo signatures (#162)

Author: sunng87

datafusion-postgres/src/pg_catalog/catalog_info.rs

@@ -10,11 +10,15 @@ use datafusion::{
 /// Define the interface for retrieve catalog data for pg_catalog tables
 #[async_trait]
 pub trait CatalogInfo: Clone + Send + Sync + Debug + 'static {
-    fn catalog_names(&self) -> Vec<String>;
+    fn catalog_names(&self) -> Result<Vec<String>, DataFusionError>;
 
-    fn schema_names(&self, catalog_name: &str) -> Option<Vec<String>>;
+    fn schema_names(&self, catalog_name: &str) -> Result<Option<Vec<String>>, DataFusionError>;
 
-    fn table_names(&self, catalog_name: &str, schema_name: &str) -> Option<Vec<String>>;
+    fn table_names(
+        &self,
+        catalog_name: &str,
+        schema_name: &str,
+    ) -> Result<Option<Vec<String>>, DataFusionError>;
 
     async fn table_schema(
         &self,
@@ -33,18 +37,23 @@ pub trait CatalogInfo: Clone + Send + Sync + Debug + 'static {
 
 #[async_trait]
 impl CatalogInfo for Arc<dyn CatalogProviderList> {
-    fn catalog_names(&self) -> Vec<String> {
-        CatalogProviderList::catalog_names(self.as_ref())
+    fn catalog_names(&self) -> Result<Vec<String>, DataFusionError> {
+        Ok(CatalogProviderList::catalog_names(self.as_ref()))
     }
 
-    fn schema_names(&self, catalog_name: &str) -> Option<Vec<String>> {
-        self.catalog(catalog_name).map(|c| c.schema_names())
+    fn schema_names(&self, catalog_name: &str) -> Result<Option<Vec<String>>, DataFusionError> {
+        Ok(self.catalog(catalog_name).map(|c| c.schema_names()))
     }
 
-    fn table_names(&self, catalog_name: &str, schema_name: &str) -> Option<Vec<String>> {
-        self.catalog(catalog_name)
+    fn table_names(
+        &self,
+        catalog_name: &str,
+        schema_name: &str,
+    ) -> Result<Option<Vec<String>>, DataFusionError> {
+        Ok(self
+            .catalog(catalog_name)
             .and_then(|c| c.schema(schema_name))
-            .map(|s| s.table_names())
+            .map(|s| s.table_names()))
     }
 
     async fn table_schema(

datafusion-postgres/src/pg_catalog/pg_attribute.rs

@@ -105,11 +105,11 @@ impl<C: CatalogInfo> PgAttributeTable<C> {
         // original one in case that schemas or tables were dropped.
         let mut swap_cache = HashMap::new();
 
-        for catalog_name in this.catalog_list.catalog_names() {
-            if let Some(schema_names) = this.catalog_list.schema_names(&catalog_name) {
+        for catalog_name in this.catalog_list.catalog_names()? {
+            if let Some(schema_names) = this.catalog_list.schema_names(&catalog_name)? {
                 for schema_name in schema_names {
                     if let Some(table_names) =
-                        this.catalog_list.table_names(&catalog_name, &schema_name)
+                        this.catalog_list.table_names(&catalog_name, &schema_name)?
                     {
                         // Process all tables in this schema
                         for table_name in table_names {

datafusion-postgres/src/pg_catalog/pg_class.rs

@@ -117,7 +117,7 @@ impl<C: CatalogInfo> PgClassTable<C> {
         let mut swap_cache = HashMap::new();
 
         // Iterate through all catalogs and schemas
-        for catalog_name in this.catalog_list.catalog_names() {
+        for catalog_name in this.catalog_list.catalog_names()? {
             let cache_key = OidCacheKey::Catalog(catalog_name.clone());
             let catalog_oid = if let Some(oid) = oid_cache.get(&cache_key) {
                 *oid
@@ -126,7 +126,7 @@ impl<C: CatalogInfo> PgClassTable<C> {
             };
             swap_cache.insert(cache_key, catalog_oid);
 
-            if let Some(schema_names) = this.catalog_list.schema_names(&catalog_name) {
+            if let Some(schema_names) = this.catalog_list.schema_names(&catalog_name)? {
                 for schema_name in schema_names {
                     let cache_key = OidCacheKey::Schema(catalog_name.clone(), schema_name.clone());
                     let schema_oid = if let Some(oid) = oid_cache.get(&cache_key) {
@@ -141,7 +141,7 @@ impl<C: CatalogInfo> PgClassTable<C> {
 
                     // Now process all tables in this schema
                     if let Some(table_names) =
-                        this.catalog_list.table_names(&catalog_name, &schema_name)
+                        this.catalog_list.table_names(&catalog_name, &schema_name)?
                     {
                         for table_name in table_names {
                             let cache_key = OidCacheKey::Table(

datafusion-postgres/src/pg_catalog/pg_database.rs

@@ -80,7 +80,7 @@ impl<C: CatalogInfo> PgDatabaseTable<C> {
         let mut oid_cache = this.oid_cache.write().await;
 
         // Add a record for each catalog (treating catalogs as "databases")
-        for catalog_name in this.catalog_list.catalog_names() {
+        for catalog_name in this.catalog_list.catalog_names()? {
             let cache_key = OidCacheKey::Catalog(catalog_name.clone());
             let catalog_oid = if let Some(oid) = oid_cache.get(&cache_key) {
                 *oid

datafusion-postgres/src/pg_catalog/pg_namespace.rs

@@ -62,8 +62,8 @@ impl<C: CatalogInfo> PgNamespaceTable<C> {
         let mut oid_cache = this.oid_cache.write().await;
 
         // Now add all schemas from DataFusion catalogs
-        for catalog_name in this.catalog_list.catalog_names() {
-            if let Some(schema_names) = this.catalog_list.schema_names(&catalog_name) {
+        for catalog_name in this.catalog_list.catalog_names()? {
+            if let Some(schema_names) = this.catalog_list.schema_names(&catalog_name)? {
                 for schema_name in schema_names {
                     let cache_key = OidCacheKey::Schema(catalog_name.clone(), schema_name.clone());
                     let schema_oid = if let Some(oid) = oid_cache.get(&cache_key) {

datafusion-postgres/src/pg_catalog/pg_tables.rs

@@ -49,11 +49,11 @@ impl<C: CatalogInfo> PgTablesTable<C> {
         let mut row_security = Vec::new();
 
         // Iterate through all catalogs and schemas
-        for catalog_name in this.catalog_list.catalog_names() {
-            if let Some(catalog_schema_names) = this.catalog_list.schema_names(&catalog_name) {
+        for catalog_name in this.catalog_list.catalog_names()? {
+            if let Some(catalog_schema_names) = this.catalog_list.schema_names(&catalog_name)? {
                 for schema_name in catalog_schema_names {
                     if let Some(catalog_table_names) =
-                        this.catalog_list.table_names(&catalog_name, &schema_name)
+                        this.catalog_list.table_names(&catalog_name, &schema_name)?
                     {
                         // Now process all tables in this schema
                         for table_name in catalog_table_names {