feat: add SchemaProvider::table_type(table_name: &str)

datafusion/catalog/src/information_schema.rs

@@ -103,12 +103,14 @@ impl InformationSchemaConfig {
                     // schema name may not exist in the catalog, so we need to check
                     if let Some(schema) = catalog.schema(&schema_name) {
                         for table_name in schema.table_names() {
-                            if let Some(table) = schema.table(&table_name).await? {
+                            if let Some(table_type) =
+                                schema.table_type(&table_name).await?
+                            {
                                 builder.add_table(
                                     &catalog_name,
                                     &schema_name,
                                     &table_name,
-                                    table.table_type(),
+                                    table_type,
                                 );
                             }
                         }
@@ -1359,3 +1361,6 @@ impl PartitionStream for InformationSchemaParameters {
         ))
     }
 }
+
+#[cfg(test)]
+mod tests;

datafusion/catalog/src/information_schema/tests.rs

@@ -0,0 +1,88 @@
+use std::sync::Arc;
+
+use crate::CatalogProvider;
+
+use super::*;
+
+#[tokio::test]
+async fn make_tables_uses_table_type() {
+    let config = InformationSchemaConfig {
+        catalog_list: Arc::new(Fixture),
+    };
+    let mut builder = InformationSchemaTablesBuilder {
+        catalog_names: StringBuilder::new(),
+        schema_names: StringBuilder::new(),
+        table_names: StringBuilder::new(),
+        table_types: StringBuilder::new(),
+        schema: Arc::new(Schema::empty()),
+    };
+
+    assert!(config.make_tables(&mut builder).await.is_ok());
+
+    assert_eq!("BASE TABLE", builder.table_types.finish().value(0));
+}
+
+#[derive(Debug)]
+struct Fixture;
+
+#[async_trait]
+impl SchemaProvider for Fixture {
+    // InformationSchemaConfig::make_tables should use this.
+    async fn table_type(&self, _: &str) -> Result<Option<TableType>> {
+        Ok(Some(TableType::Base))
+    }
+
+    // InformationSchemaConfig::make_tables used this before `table_type`
+    // existed but should not, as it may be expensive.
+    async fn table(&self, _: &str) -> Result<Option<Arc<dyn TableProvider>>> {
+        panic!("InformationSchemaConfig::make_tables called SchemaProvider::table instead of table_type")
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        unimplemented!("not required for these tests")
+    }
+
+    fn table_names(&self) -> Vec<String> {
+        vec!["atable".to_string()]
+    }
+
+    fn table_exist(&self, _: &str) -> bool {
+        unimplemented!("not required for these tests")
+    }
+}
+
+impl CatalogProviderList for Fixture {
+    fn as_any(&self) -> &dyn Any {
+        unimplemented!("not required for these tests")
+    }
+
+    fn register_catalog(
+        &self,
+        _: String,
+        _: Arc<dyn CatalogProvider>,
+    ) -> Option<Arc<dyn CatalogProvider>> {
+        unimplemented!("not required for these tests")
+    }
+
+    fn catalog_names(&self) -> Vec<String> {
+        vec!["acatalog".to_string()]
+    }
+
+    fn catalog(&self, _: &str) -> Option<Arc<dyn CatalogProvider>> {
+        Some(Arc::new(Self))
+    }
+}
+
+impl CatalogProvider for Fixture {
+    fn as_any(&self) -> &dyn Any {
+        unimplemented!("not required for these tests")
+    }
+
+    fn schema_names(&self) -> Vec<String> {
+        vec!["aschema".to_string()]
+    }
+
+    fn schema(&self, _: &str) -> Option<Arc<dyn SchemaProvider>> {
+        Some(Arc::new(Self))
+    }
+}

datafusion/catalog/src/schema.rs

@@ -26,6 +26,7 @@ use std::sync::Arc;
 
 use crate::table::TableProvider;
 use datafusion_common::Result;
+use datafusion_expr::TableType;
 
 /// Represents a schema, comprising a number of named tables.
 ///
@@ -54,6 +55,14 @@ pub trait SchemaProvider: Debug + Sync + Send {
         name: &str,
     ) -> Result<Option<Arc<dyn TableProvider>>, DataFusionError>;
 
+    /// Retrieves the type of a specific table from the schema by name, if it exists, otherwise
+    /// returns `None`.  Implementations for which this operation is cheap but [Self::table] is
+    /// expensive can override this to improve operations that only need the type, e.g.
+    /// `SELECT * FROM information_schema.tables`.
+    async fn table_type(&self, name: &str) -> Result<Option<TableType>> {
+        self.table(name).await.map(|o| o.map(|t| t.table_type()))
+    }
+
     /// If supported by the implementation, adds a new table named `name` to
     /// this schema.
     ///