From ad42523ff637a5975c2bce6c4aaba91466bd2ec7 Mon Sep 17 00:00:00 2001 From: Vaibhave Sekar Date: Sun, 9 Aug 2026 06:07:32 +0000 Subject: [PATCH] fix: scope information schema to catalog --- datafusion/catalog/src/information_schema.rs | 164 +++++++----------- .../core/src/execution/session_state.rs | 27 +-- .../information_schema_multiple_catalogs.slt | 31 ++-- 3 files changed, 96 insertions(+), 126 deletions(-) diff --git a/datafusion/catalog/src/information_schema.rs b/datafusion/catalog/src/information_schema.rs index d9ad7791af67c..d399f49e4ed7e 100644 --- a/datafusion/catalog/src/information_schema.rs +++ b/datafusion/catalog/src/information_schema.rs @@ -21,7 +21,7 @@ use crate::streaming::StreamingTable; use crate::table::TableFunction; -use crate::{CatalogProviderList, SchemaProvider, TableProvider}; +use crate::{CatalogProvider, SchemaProvider, TableProvider}; use arrow::array::builder::{BooleanBuilder, UInt8Builder}; use arrow::{ array::{StringBuilder, UInt64Builder}, @@ -79,11 +79,15 @@ pub struct InformationSchemaProvider { } impl InformationSchemaProvider { - /// Creates a new [`InformationSchemaProvider`] for the provided `catalog_list` - pub fn new(catalog_list: Arc) -> Self { + /// Creates a new [`InformationSchemaProvider`] for the specified catalog. + pub fn new( + catalog_name: impl Into, + catalog: Arc, + ) -> Self { Self { config: InformationSchemaConfig { - catalog_list, + catalog_name: catalog_name.into(), + catalog, table_functions: HashMap::new(), }, } @@ -102,7 +106,8 @@ impl InformationSchemaProvider { #[derive(Clone, Debug)] struct InformationSchemaConfig { - catalog_list: Arc, + catalog_name: String, + catalog: Arc, table_functions: HashMap>, } @@ -114,54 +119,44 @@ impl InformationSchemaConfig { ) -> Result<(), DataFusionError> { // create a mem table with the names of tables - for catalog_name in self.catalog_list.catalog_names() { - let catalog = self.catalog_list.catalog(&catalog_name).unwrap(); - - for schema_name in catalog.schema_names() { - if schema_name != INFORMATION_SCHEMA { - // 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_type) = - schema.table_type(&table_name).await? - { - builder.add_table( - &catalog_name, - &schema_name, - &table_name, - table_type, - ); - } + for schema_name in self.catalog.schema_names() { + if schema_name != INFORMATION_SCHEMA { + // schema name may not exist in the catalog, so we need to check + if let Some(schema) = self.catalog.schema(&schema_name) { + for table_name in schema.table_names() { + if let Some(table_type) = schema.table_type(&table_name).await? { + builder.add_table( + &self.catalog_name, + &schema_name, + &table_name, + table_type, + ); } } } } + } - // Add a final list for the information schema tables themselves - for table_name in INFORMATION_SCHEMA_TABLES { - builder.add_table( - &catalog_name, - INFORMATION_SCHEMA, - table_name, - TableType::View, - ); - } + // Add a final list for the information schema tables themselves + for table_name in INFORMATION_SCHEMA_TABLES { + builder.add_table( + &self.catalog_name, + INFORMATION_SCHEMA, + table_name, + TableType::View, + ); } Ok(()) } fn make_schemata(&self, builder: &mut InformationSchemataBuilder) { - for catalog_name in self.catalog_list.catalog_names() { - let catalog = self.catalog_list.catalog(&catalog_name).unwrap(); - - for schema_name in catalog.schema_names() { - if schema_name != INFORMATION_SCHEMA - && let Some(schema) = catalog.schema(&schema_name) - { - let schema_owner = schema.owner_name(); - builder.add_schemata(&catalog_name, &schema_name, schema_owner); - } + for schema_name in self.catalog.schema_names() { + if schema_name != INFORMATION_SCHEMA + && let Some(schema) = self.catalog.schema(&schema_name) + { + let schema_owner = schema.owner_name(); + builder.add_schemata(&self.catalog_name, &schema_name, schema_owner); } } } @@ -170,22 +165,18 @@ impl InformationSchemaConfig { &self, builder: &mut InformationSchemaViewBuilder, ) -> Result<(), DataFusionError> { - for catalog_name in self.catalog_list.catalog_names() { - let catalog = self.catalog_list.catalog(&catalog_name).unwrap(); - - for schema_name in catalog.schema_names() { - if schema_name != INFORMATION_SCHEMA { - // 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? { - builder.add_view( - &catalog_name, - &schema_name, - &table_name, - table.get_table_definition(), - ) - } + for schema_name in self.catalog.schema_names() { + if schema_name != INFORMATION_SCHEMA { + // schema name may not exist in the catalog, so we need to check + if let Some(schema) = self.catalog.schema(&schema_name) { + for table_name in schema.table_names() { + if let Some(table) = schema.table(&table_name).await? { + builder.add_view( + &self.catalog_name, + &schema_name, + &table_name, + table.get_table_definition(), + ) } } } @@ -200,26 +191,22 @@ impl InformationSchemaConfig { &self, builder: &mut InformationSchemaColumnsBuilder, ) -> Result<(), DataFusionError> { - for catalog_name in self.catalog_list.catalog_names() { - let catalog = self.catalog_list.catalog(&catalog_name).unwrap(); - - for schema_name in catalog.schema_names() { - if schema_name != INFORMATION_SCHEMA { - // 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? { - for (field_position, field) in - table.schema().fields().iter().enumerate() - { - builder.add_column( - &catalog_name, - &schema_name, - &table_name, - field_position, - field, - ) - } + for schema_name in self.catalog.schema_names() { + if schema_name != INFORMATION_SCHEMA { + // schema name may not exist in the catalog, so we need to check + if let Some(schema) = self.catalog.schema(&schema_name) { + for table_name in schema.table_names() { + if let Some(table) = schema.table(&table_name).await? { + for (field_position, field) in + table.schema().fields().iter().enumerate() + { + builder.add_column( + &self.catalog_name, + &schema_name, + &table_name, + field_position, + field, + ) } } } @@ -1564,7 +1551,8 @@ mod tests { #[tokio::test] async fn make_tables_uses_table_type() { let config = InformationSchemaConfig { - catalog_list: Arc::new(Fixture), + catalog_name: "acatalog".to_string(), + catalog: Arc::new(Fixture), table_functions: HashMap::new(), }; let mut builder = InformationSchemaTablesBuilder { @@ -1607,24 +1595,6 @@ mod tests { } } - impl CatalogProviderList for Fixture { - fn register_catalog( - &self, - _: String, - _: Arc, - ) -> Option> { - unimplemented!("not required for these tests") - } - - fn catalog_names(&self) -> Vec { - vec!["acatalog".to_string()] - } - - fn catalog(&self, _: &str) -> Option> { - Some(Arc::new(Self)) - } - } - impl CatalogProvider for Fixture { fn schema_names(&self) -> Vec { vec!["aschema".to_string()] diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index bfd38faacf816..782205dc9c003 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -369,26 +369,27 @@ impl SessionState { table_ref: impl Into, ) -> datafusion_common::Result> { let resolved_ref = self.resolve_table_ref(table_ref); + let catalog = self + .catalog_list + .catalog(&resolved_ref.catalog) + .ok_or_else(|| { + plan_datafusion_err!( + "failed to resolve catalog: {}", + resolved_ref.catalog + ) + })?; + if self.config.information_schema() && *resolved_ref.schema == *INFORMATION_SCHEMA { return Ok(Arc::new( - InformationSchemaProvider::new(Arc::clone(&self.catalog_list)) + InformationSchemaProvider::new(resolved_ref.catalog.to_string(), catalog) .with_table_functions(self.table_functions.clone()), )); } - self.catalog_list - .catalog(&resolved_ref.catalog) - .ok_or_else(|| { - plan_datafusion_err!( - "failed to resolve catalog: {}", - resolved_ref.catalog - ) - })? - .schema(&resolved_ref.schema) - .ok_or_else(|| { - plan_datafusion_err!("failed to resolve schema: {}", resolved_ref.schema) - }) + catalog.schema(&resolved_ref.schema).ok_or_else(|| { + plan_datafusion_err!("failed to resolve schema: {}", resolved_ref.schema) + }) } /// Add `analyzer_rule` to the end of the list of diff --git a/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt b/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt index f57ecbc32a57a..8a204a1df0763 100644 --- a/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt +++ b/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt @@ -77,25 +77,24 @@ set datafusion.catalog.default_schema = my_other_schema; statement ok create table t3 as values(1); +query TTT rowsort +SELECT table_catalog, table_schema, table_name +FROM my_catalog.information_schema.tables +WHERE table_schema <> 'information_schema'; +---- +my_catalog my_schema t1 +my_catalog my_schema t2 + +query TTT rowsort +SELECT table_catalog, table_schema, table_name +FROM my_other_catalog.information_schema.tables +WHERE table_schema <> 'information_schema'; +---- +my_other_catalog my_other_schema t3 + query TTTT rowsort SELECT * from information_schema.tables; ---- -datafusion information_schema columns VIEW -datafusion information_schema df_settings VIEW -datafusion information_schema parameters VIEW -datafusion information_schema routines VIEW -datafusion information_schema schemata VIEW -datafusion information_schema tables VIEW -datafusion information_schema views VIEW -my_catalog information_schema columns VIEW -my_catalog information_schema df_settings VIEW -my_catalog information_schema parameters VIEW -my_catalog information_schema routines VIEW -my_catalog information_schema schemata VIEW -my_catalog information_schema tables VIEW -my_catalog information_schema views VIEW -my_catalog my_schema t1 BASE TABLE -my_catalog my_schema t2 BASE TABLE my_other_catalog information_schema columns VIEW my_other_catalog information_schema df_settings VIEW my_other_catalog information_schema parameters VIEW