From ad42523ff637a5975c2bce6c4aaba91466bd2ec7 Mon Sep 17 00:00:00 2001 From: Vaibhave Sekar Date: Sun, 9 Aug 2026 06:07:32 +0000 Subject: [PATCH 1/7] 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 From ea2d35f32fe9549135a64e1d1a0211a2c7ac86a1 Mon Sep 17 00:00:00 2001 From: Pepijn Van Eeckhoudt Date: Wed, 9 Sep 2026 18:28:43 +0200 Subject: [PATCH 2/7] Add a 'system' catalog that provides a global information schema --- datafusion-cli/src/main.rs | 5 +- datafusion/catalog/src/information_schema.rs | 283 +++++++++++++----- datafusion/common/src/config.rs | 3 + datafusion/core/src/execution/context/mod.rs | 4 +- .../core/src/execution/session_state.rs | 112 +++++-- datafusion/execution/src/config.rs | 14 +- datafusion/sql/src/statement.rs | 70 +++-- 7 files changed, 366 insertions(+), 125 deletions(-) diff --git a/datafusion-cli/src/main.rs b/datafusion-cli/src/main.rs index f82206a5bd184..f08cb6eeb37ef 100644 --- a/datafusion-cli/src/main.rs +++ b/datafusion-cli/src/main.rs @@ -370,8 +370,9 @@ fn get_session_config(args: &Args) -> Result { config_options.sql_parser.dialect = Dialect::Spark; } - let mut session_config = - SessionConfig::from(config_options).with_information_schema(true); + let mut session_config = SessionConfig::from(config_options) + .with_system_catalog(Some(String::from("system"))) + .with_information_schema(true); if args.reads_sql_from_stdin() { // When stdin carries the session's SQL — the REPL (including any rc diff --git a/datafusion/catalog/src/information_schema.rs b/datafusion/catalog/src/information_schema.rs index d399f49e4ed7e..9fd70cf9f7d0b 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::{CatalogProvider, SchemaProvider, TableProvider}; +use crate::{CatalogProviderList, SchemaProvider, TableProvider}; use arrow::array::builder::{BooleanBuilder, UInt8Builder}; use arrow::{ array::{StringBuilder, UInt64Builder}, @@ -79,15 +79,12 @@ pub struct InformationSchemaProvider { } impl InformationSchemaProvider { - /// Creates a new [`InformationSchemaProvider`] for the specified catalog. - pub fn new( - catalog_name: impl Into, - catalog: Arc, - ) -> Self { + /// Creates a new [`InformationSchemaProvider`] for the provided `catalog_list` + pub fn new(catalog_list: Arc) -> Self { Self { config: InformationSchemaConfig { - catalog_name: catalog_name.into(), - catalog, + system_catalog_name: None, + catalog_list, table_functions: HashMap::new(), }, } @@ -102,12 +99,17 @@ impl InformationSchemaProvider { self.config.table_functions = table_functions; self } + + pub fn with_system_catalog(mut self, catalog_name: String) -> Self { + self.config.system_catalog_name = Some(catalog_name); + self + } } #[derive(Clone, Debug)] struct InformationSchemaConfig { - catalog_name: String, - catalog: Arc, + system_catalog_name: Option, + catalog_list: Arc, table_functions: HashMap>, } @@ -119,45 +121,74 @@ impl InformationSchemaConfig { ) -> Result<(), DataFusionError> { // create a mem table with the names of tables - 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, - ); + 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, + ); + } } } } } + + // 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, - ); + if let Some(system_catalog_name) = &self.system_catalog_name { + for table_name in INFORMATION_SCHEMA_TABLES { + builder.add_table( + system_catalog_name, + INFORMATION_SCHEMA, + table_name, + TableType::View, + ); + } } Ok(()) } fn make_schemata(&self, builder: &mut InformationSchemataBuilder) { - 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); + 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); + } } + + // Add the information schema itself + builder.add_schemata(&catalog_name, INFORMATION_SCHEMA, None); + } + + // Add the system information schema + if let Some(system_catalog_name) = &self.system_catalog_name { + builder.add_schemata(system_catalog_name, INFORMATION_SCHEMA, None); } } @@ -165,24 +196,55 @@ impl InformationSchemaConfig { &self, builder: &mut InformationSchemaViewBuilder, ) -> Result<(), DataFusionError> { - 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(), - ) - } + 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) { + Self::add_views( + builder, + &catalog_name, + &schema_name, + schema.as_ref(), + ) + .await?; } } + + // Add the information schema views themselves + Self::add_views(builder, &catalog_name, INFORMATION_SCHEMA, self).await?; } } + // Add the system information schema + if let Some(system_catalog_name) = &self.system_catalog_name { + Self::add_views(builder, system_catalog_name, INFORMATION_SCHEMA, self) + .await?; + } + + Ok(()) + } + + async fn add_views( + builder: &mut InformationSchemaViewBuilder, + catalog_name: &str, + schema_name: &str, + schema: &dyn SchemaProvider, + ) -> Result<(), DataFusionError> { + for table_name in schema.table_names() { + if let Some(table) = schema.table(&table_name).await? + && table.table_type() == TableType::View + { + builder.add_view( + catalog_name, + schema_name, + &table_name, + table.get_table_definition(), + ) + } + } Ok(()) } @@ -191,32 +253,60 @@ impl InformationSchemaConfig { &self, builder: &mut InformationSchemaColumnsBuilder, ) -> Result<(), DataFusionError> { - 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, - ) - } - } + 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) { + Self::add_columns( + builder, + &catalog_name, + &schema_name, + schema.as_ref(), + ) + .await?; } } } + + // Add the information schema view columns themselves + Self::add_columns(builder, &catalog_name, INFORMATION_SCHEMA, self).await?; + } + + // Add the system information schema view columns + if let Some(system_catalog_name) = &self.system_catalog_name { + Self::add_columns(builder, system_catalog_name, INFORMATION_SCHEMA, self) + .await?; } Ok(()) } + async fn add_columns( + builder: &mut InformationSchemaColumnsBuilder, + catalog_name: &str, + schema_name: &str, + schema: &dyn SchemaProvider, + ) -> Result<(), DataFusionError> { + 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, + ) + } + } + } + Ok(()) + } + /// Construct the `information_schema.df_settings` virtual table fn make_df_settings( &self, @@ -562,6 +652,24 @@ fn remove_native_type_prefix(native_type: &NativeType) -> String { #[async_trait] impl SchemaProvider for InformationSchemaProvider { + fn table_names(&self) -> Vec { + self.config.table_names() + } + + async fn table( + &self, + name: &str, + ) -> Result>, DataFusionError> { + self.config.get_table(name) + } + + fn table_exist(&self, name: &str) -> bool { + self.config.table_exist(name) + } +} + +#[async_trait] +impl SchemaProvider for InformationSchemaConfig { fn table_names(&self) -> Vec { INFORMATION_SCHEMA_TABLES .iter() @@ -573,7 +681,20 @@ impl SchemaProvider for InformationSchemaProvider { &self, name: &str, ) -> Result>, DataFusionError> { - let config = self.config.clone(); + self.get_table(name) + } + + fn table_exist(&self, name: &str) -> bool { + INFORMATION_SCHEMA_TABLES.contains(&name.to_ascii_lowercase().as_str()) + } +} + +impl InformationSchemaConfig { + fn get_table( + &self, + name: &str, + ) -> Result>, DataFusionError> { + let config = self.clone(); let table: Arc = match name.to_ascii_lowercase().as_str() { TABLES => Arc::new(InformationSchemaTables::new(config)), COLUMNS => Arc::new(InformationSchemaColumns::new(config)), @@ -589,10 +710,6 @@ impl SchemaProvider for InformationSchemaProvider { StreamingTable::try_new(Arc::clone(table.schema()), vec![table]).unwrap(), ))) } - - fn table_exist(&self, name: &str) -> bool { - INFORMATION_SCHEMA_TABLES.contains(&name.to_ascii_lowercase().as_str()) - } } #[derive(Debug)] @@ -1551,8 +1668,8 @@ mod tests { #[tokio::test] async fn make_tables_uses_table_type() { let config = InformationSchemaConfig { - catalog_name: "acatalog".to_string(), - catalog: Arc::new(Fixture), + system_catalog_name: None, + catalog_list: Arc::new(Fixture), table_functions: HashMap::new(), }; let mut builder = InformationSchemaTablesBuilder { @@ -1595,6 +1712,24 @@ 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/common/src/config.rs b/datafusion/common/src/config.rs index f770cebf88f95..6f92cae3c42eb 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -235,6 +235,9 @@ config_namespace! { /// The default schema name - this impacts what SQL queries use if not specified pub default_schema: String, default = "public".to_string() + /// The system schema name + pub system_catalog: Option, default = None + /// Should DataFusion provide access to `information_schema` /// virtual tables for displaying schema information pub information_schema: bool, default = false diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs index ff1ad25811440..1919a83bd7853 100644 --- a/datafusion/core/src/execution/context/mod.rs +++ b/datafusion/core/src/execution/context/mod.rs @@ -2764,7 +2764,7 @@ mod tests { async fn sql_create_catalog() -> Result<()> { // the information schema used to introduce cyclic Arcs let ctx = SessionContext::new_with_config( - SessionConfig::new().with_information_schema(true), + SessionConfig::new().with_information_schema(true).with_system_catalog(Some(String::from("system"))), ); // Create catalog @@ -2780,7 +2780,7 @@ mod tests { .await?; // Check table exists in schema - let results = ctx.sql("SELECT * FROM information_schema.tables WHERE table_catalog='test' AND table_schema='abc' AND table_name = 'y'").await.unwrap().collect().await.unwrap(); + let results = ctx.sql("SELECT * FROM system.information_schema.tables WHERE table_catalog='test' AND table_schema='abc' AND table_name = 'y'").await.unwrap().collect().await.unwrap(); assert_eq!(results[0].num_rows(), 1); Ok(()) diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index 869c3002a24d1..27114a89e1438 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -46,7 +46,7 @@ use datafusion_common::display::{PlanType, StringifiedPlan, ToStringifiedPlan}; use datafusion_common::tree_node::TreeNode; use datafusion_common::{ DFSchema, DataFusionError, ResolvedTableReference, TableReference, config_err, - exec_err, plan_datafusion_err, + exec_err, plan_datafusion_err, plan_err, }; use datafusion_execution::TaskContext; use datafusion_execution::config::SessionConfig; @@ -76,7 +76,9 @@ use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use datafusion_physical_optimizer::optimizer::PhysicalOptimizer; use datafusion_physical_plan::ExecutionPlan; use datafusion_physical_plan::operator_statistics::StatisticsRegistry; -use datafusion_session::{PhysicalOptimizerContext, PhysicalOptimizerRule, Session}; +use datafusion_session::{ + CatalogProvider, PhysicalOptimizerContext, PhysicalOptimizerRule, Session, +}; #[cfg(feature = "sql")] use datafusion_sql::{ parser::{DFParserBuilder, Statement}, @@ -404,28 +406,75 @@ impl SessionState { table_ref: impl Into, ) -> datafusion_common::Result> { let resolved_ref = self.resolve_table_ref(table_ref); - let catalog = self - .inner - .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 + let catalog_name = resolved_ref.catalog.as_ref(); + + if let Some(system_catalog) = self.inner.config.system_catalog() + && system_catalog == catalog_name { - return Ok(Arc::new( - InformationSchemaProvider::new(resolved_ref.catalog.to_string(), catalog) - .with_table_functions(self.table_functions.clone()), - )); + if *resolved_ref.schema == *INFORMATION_SCHEMA { + return Ok(Arc::new( + InformationSchemaProvider::new(Arc::clone(&self.inner.catalog_list)) + .with_table_functions(self.inner.table_functions.clone()) + .with_system_catalog(String::from(system_catalog)), + )); + } + } else { + let catalog = + self.inner + .catalog_list + .catalog(catalog_name) + .ok_or_else(|| { + plan_datafusion_err!("failed to resolve catalog: {catalog_name}") + })?; + + if self.inner.config.information_schema() + && *resolved_ref.schema == *INFORMATION_SCHEMA + { + #[derive(Debug)] + struct SingleCatalogList { + catalog_name: String, + catalog: Arc, + } + + impl CatalogProviderList for SingleCatalogList { + fn register_catalog( + &self, + _: String, + _: Arc, + ) -> Option> { + None + } + + fn catalog_names(&self) -> Vec { + vec![self.catalog_name.clone()] + } + + fn catalog(&self, name: &str) -> Option> { + if name == self.catalog_name { + Some(Arc::clone(&self.catalog)) + } else { + None + } + } + } + + let catalog_list = Arc::new(SingleCatalogList { + catalog_name: catalog_name.to_string(), + catalog, + }); + + return Ok(Arc::new( + InformationSchemaProvider::new(catalog_list) + .with_table_functions(self.inner.table_functions.clone()), + )); + } + + if let Some(schema) = catalog.schema(&resolved_ref.schema) { + return Ok(schema); + } } - catalog.schema(&resolved_ref.schema).ok_or_else(|| { - plan_datafusion_err!("failed to resolve schema: {}", resolved_ref.schema) - }) + plan_err!("failed to resolve schema: {}", resolved_ref.schema) } /// Add `analyzer_rule` to the end of the list of @@ -588,6 +637,27 @@ impl SessionState { statement, enable_ident_normalization, )?; + + let table_refs = if let Some(system_catalog) = self.config().system_catalog() { + table_refs + .iter() + .map(|table_ref| match table_ref { + TableReference::Partial { schema, table } + if schema.as_ref() == INFORMATION_SCHEMA => + { + TableReference::Full { + catalog: Arc::from(system_catalog), + schema: Arc::clone(schema), + table: Arc::clone(table), + } + } + r => r.clone(), + }) + .collect() + } else { + table_refs + }; + Ok(table_refs) } diff --git a/datafusion/execution/src/config.rs b/datafusion/execution/src/config.rs index efaedebdadb33..2d21772016ccd 100644 --- a/datafusion/execution/src/config.rs +++ b/datafusion/execution/src/config.rs @@ -219,6 +219,11 @@ impl SessionConfig { self.options.execution.target_partitions } + /// Is the information schema enabled? + pub fn system_catalog(&self) -> Option<&str> { + self.options.catalog.system_catalog.as_deref() + } + /// Is the information schema enabled? pub fn information_schema(&self) -> bool { self.options.catalog.information_schema @@ -285,7 +290,14 @@ impl SessionConfig { self } - /// Enables or disables the inclusion of `information_schema` virtual tables + /// Enables or disables the inclusion of the system virtual catalog + pub fn with_system_catalog(mut self, catalog_name: Option) -> Self { + self.options_mut().catalog.system_catalog = catalog_name; + self + } + + /// Enables or disables the inclusion of `information_schema` virtual tables in each user + /// defined catalog pub fn with_information_schema(mut self, enabled: bool) -> Self { self.options_mut().catalog.information_schema = enabled; self diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index 1a9072212f2f3..543c63eeb9e49 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -16,6 +16,7 @@ // under the License. use std::collections::{BTreeMap, HashMap, HashSet}; +use std::ops::Deref; use std::path::Path; use std::str::FromStr; use std::sync::Arc; @@ -1637,14 +1638,16 @@ impl SqlToRel<'_, S> { /// Generate a logical plan from a "SHOW TABLES" query fn show_tables_to_plan(&self) -> Result { - if self.has_table("information_schema", "tables") { - let query = "SELECT * FROM information_schema.tables;"; - let mut rewrite = DFParser::parse_sql(query)?; - assert_eq!(rewrite.len(), 1); - self.statement_to_plan(rewrite.pop_front().unwrap()) // length of rewrite is 1 - } else { - plan_err!("SHOW TABLES is not supported unless information_schema is enabled") - } + let Some(tables_table_ref) = self.resolve_info_table("tables") else { + return plan_err!( + "SHOW TABLES is not supported unless information_schema is enabled" + ); + }; + + let query = format!("SELECT * FROM {tables_table_ref};"); + let mut rewrite = DFParser::parse_sql(&query)?; + assert_eq!(rewrite.len(), 1); + self.statement_to_plan(rewrite.pop_front().unwrap()) // length of rewrite is 1 } fn describe_table_to_plan(&self, table_name: ObjectName) -> Result { @@ -2120,11 +2123,11 @@ impl SqlToRel<'_, S> { } fn show_variable_to_plan(&self, variable: &[Ident]) -> Result { - if !self.has_table("information_schema", "df_settings") { + let Some(df_settings_table_ref) = self.resolve_info_table("df_settings") else { return plan_err!( "SHOW [VARIABLE] is not supported unless information_schema is enabled" ); - } + }; let verbose = variable .last() @@ -2139,7 +2142,7 @@ impl SqlToRel<'_, S> { } let variable = object_name_to_string(&ObjectName::from(variable_vec)); - let base_query = format!("SELECT {columns} FROM information_schema.df_settings"); + let base_query = format!("SELECT {columns} FROM {df_settings_table_ref}"); let query = if variable == "all" { // Add an ORDER BY so the output comes out in a consistent order format!("{base_query} ORDER BY name") @@ -2988,11 +2991,11 @@ impl SqlToRel<'_, S> { self.options.enable_ident_normalization, )?; - if !self.has_table("information_schema", "columns") { + let Some(columns_table_ref) = self.resolve_info_table("columns") else { return plan_err!( "SHOW COLUMNS is not supported unless information_schema is enabled" ); - } + }; // Do a table lookup to verify the table exists let table_ref = self.object_name_to_table_reference(sql_table_name)?; @@ -3005,9 +3008,8 @@ impl SqlToRel<'_, S> { "table_catalog, table_schema, table_name, column_name, data_type, is_nullable" }; - let query = format!( - "SELECT {select_list} FROM information_schema.columns WHERE {where_clause}" - ); + let query = + format!("SELECT {select_list} FROM {columns_table_ref} WHERE {where_clause}"); let mut rewrite = DFParser::parse_sql(&query)?; assert_eq!(rewrite.len(), 1); @@ -3136,11 +3138,12 @@ FROM ( &self, sql_table_name: ObjectName, ) -> Result { - if !self.has_table("information_schema", "tables") { + let Some(tables_table_ref) = self.resolve_info_table("tables") else { return plan_err!( "SHOW CREATE TABLE is not supported unless information_schema is enabled" ); - } + }; + // Figure out the where clause let where_clause = object_name_to_qualifier( &sql_table_name, @@ -3152,7 +3155,7 @@ FROM ( let _ = self.context_provider.get_table_source(table_ref)?; let query = format!( - "SELECT table_catalog, table_schema, table_name, definition FROM information_schema.views WHERE {where_clause}" + "SELECT table_catalog, table_schema, table_name, definition FROM {tables_table_ref} WHERE {where_clause}" ); let mut rewrite = DFParser::parse_sql(&query)?; @@ -3161,14 +3164,31 @@ FROM ( } /// Return true if there is a table provider available for "schema.table" - fn has_table(&self, schema: &str, table: &str) -> bool { - let tables_reference = TableReference::Partial { - schema: schema.into(), - table: table.into(), + fn resolve_info_table(&self, table: &str) -> Option { + let table_reference = if let Some(system_catalog) = + &self.context_provider.options().catalog.system_catalog + { + TableReference::Full { + catalog: system_catalog.deref().into(), + schema: "information_schema".into(), + table: table.into(), + } + } else { + TableReference::Partial { + schema: "information_schema".into(), + table: table.into(), + } }; - self.context_provider - .get_table_source(tables_reference) + + if self + .context_provider + .get_table_source(table_reference.clone()) .is_ok() + { + Some(table_reference) + } else { + None + } } fn validate_transaction_kind( From db98eb912fbbc2e5dc51a85ccfd7881f80fd2fb6 Mon Sep 17 00:00:00 2001 From: Pepijn Van Eeckhoudt Date: Wed, 30 Sep 2026 00:22:49 +0200 Subject: [PATCH 3/7] Fix resolution of partial information_schema table references --- .../core/src/execution/session_state.rs | 21 ------------------- 1 file changed, 21 deletions(-) diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index 7959b171597e7..43cd46502c0bb 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -662,27 +662,6 @@ impl SessionState { statement, enable_ident_normalization, )?; - - let table_refs = if let Some(system_catalog) = self.config().system_catalog() { - table_refs - .iter() - .map(|table_ref| match table_ref { - TableReference::Partial { schema, table } - if schema.as_ref() == INFORMATION_SCHEMA => - { - TableReference::Full { - catalog: Arc::from(system_catalog), - schema: Arc::clone(schema), - table: Arc::clone(table), - } - } - r => r.clone(), - }) - .collect() - } else { - table_refs - }; - Ok(table_refs) } From 6da6d7224baa52c2f9a2a4a03a1913301583e33f Mon Sep 17 00:00:00 2001 From: Pepijn Van Eeckhoudt Date: Wed, 30 Sep 2026 00:23:13 +0200 Subject: [PATCH 4/7] Formatting --- datafusion/core/src/execution/context/mod.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs index 1d7efe1ae9eaf..edb6361340144 100644 --- a/datafusion/core/src/execution/context/mod.rs +++ b/datafusion/core/src/execution/context/mod.rs @@ -2846,7 +2846,9 @@ mod tests { async fn sql_create_catalog() -> Result<()> { // the information schema used to introduce cyclic Arcs let ctx = SessionContext::new_with_config( - SessionConfig::new().with_information_schema(true).with_system_catalog(Some(String::from("system"))), + SessionConfig::new() + .with_information_schema(true) + .with_system_catalog(Some(String::from("system"))), ); // Create catalog From 1b1ed047d6ef3fed81ea1f37dad147a68b07ce08 Mon Sep 17 00:00:00 2001 From: Pepijn Van Eeckhoudt Date: Fri, 2 Oct 2026 21:59:25 +0200 Subject: [PATCH 5/7] Update config documentation --- datafusion/common/src/config.rs | 2 +- docs/source/user-guide/configs.md | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 795cad6a5a9f7..2b22de90470fa 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -237,7 +237,7 @@ config_namespace! { /// The default schema name - this impacts what SQL queries use if not specified pub default_schema: String, default = "public".to_string() - /// The system schema name + /// The system schema name - when set a global system and `information_schema` catalog will be available pub system_catalog: Option, default = None /// Should DataFusion provide access to `information_schema` diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index ba8ea52bb0850..6537031c613a0 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -68,6 +68,7 @@ The following configuration settings are available: | datafusion.catalog.create_default_catalog_and_schema | true | Whether the default catalog and schema should be created automatically. | | datafusion.catalog.default_catalog | datafusion | The default catalog name - this impacts what SQL queries use if not specified | | datafusion.catalog.default_schema | public | The default schema name - this impacts what SQL queries use if not specified | +| datafusion.catalog.system_catalog | NULL | The system schema name - when set a global system and `information_schema` catalog will be available | | datafusion.catalog.information_schema | false | Should DataFusion provide access to `information_schema` virtual tables for displaying schema information | | datafusion.catalog.location | NULL | Location scanned to load tables for `default` schema | | datafusion.catalog.format | NULL | Type of `TableProvider` to use when loading `default` schema | From 1735fc569a5134a1e8661e3a648026d9a08ac662 Mon Sep 17 00:00:00 2001 From: Pepijn Van Eeckhoudt Date: Sat, 3 Oct 2026 00:17:37 +0200 Subject: [PATCH 6/7] Move information schema resolution to session state --- .../core/src/execution/session_state.rs | 35 +++++++++++-- datafusion/sql/src/resolve.rs | 50 ------------------- 2 files changed, 32 insertions(+), 53 deletions(-) diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index 43cd46502c0bb..31ce00f16b81d 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -37,7 +37,7 @@ use arrow_schema::DataType; use arrow_schema::FieldRef; use datafusion_catalog::MemoryCatalogProviderList; use datafusion_catalog::information_schema::{ - INFORMATION_SCHEMA, InformationSchemaProvider, + INFORMATION_SCHEMA, INFORMATION_SCHEMA_TABLES, InformationSchemaProvider, }; use datafusion_catalog::{TableFunction, TableFunctionImpl}; use datafusion_common::alias::AliasGenerator; @@ -95,7 +95,9 @@ use log::{debug, info}; use object_store::ObjectStore; #[cfg(feature = "sql")] use sqlparser::{ - ast::{Expr as SQLExpr, ExprWithAlias as SQLExprWithAlias}, + ast::{ + Expr as SQLExpr, ExprWithAlias as SQLExprWithAlias, Statement as SQLStatement, + }, dialect::dialect_from_str, }; use url::Url; @@ -671,7 +673,34 @@ impl SessionState { &self, statement: Statement, ) -> datafusion_common::Result { - let references = self.resolve_table_references(&statement)?; + let mut references = self.resolve_table_references(&statement)?; + let require_info_schema = match &statement { + Statement::Statement(s) => matches!( + s.as_ref(), + SQLStatement::ShowCreate { .. } + | SQLStatement::ShowTables { .. } + | SQLStatement::ShowColumns { .. } + | SQLStatement::ShowVariable { .. } + | SQLStatement::ShowFunctions { .. } + ), + _ => false, + }; + + if require_info_schema { + let info_tables = INFORMATION_SCHEMA_TABLES.iter(); + let info_table_refs: Vec<_> = if let Some(system_catalog) = + self.config().system_catalog() + { + info_tables + .map(|n| TableReference::full(system_catalog, INFORMATION_SCHEMA, *n)) + .collect() + } else { + info_tables + .map(|n| TableReference::partial(INFORMATION_SCHEMA, *n)) + .collect() + }; + references.extend(info_table_refs); + } let mut provider = SessionContextProvider { state: self, diff --git a/datafusion/sql/src/resolve.rs b/datafusion/sql/src/resolve.rs index b6b1a74849be0..9abc687a2c39b 100644 --- a/datafusion/sql/src/resolve.rs +++ b/datafusion/sql/src/resolve.rs @@ -25,28 +25,6 @@ use crate::planner::object_name_to_table_reference; use datafusion_common::TableReference; use sqlparser::ast::*; -// following constants are used in `resolve_table_references` -// and should be same as `datafusion/catalog/src/information_schema.rs` -const INFORMATION_SCHEMA: &str = "information_schema"; -const TABLES: &str = "tables"; -const VIEWS: &str = "views"; -const COLUMNS: &str = "columns"; -const DF_SETTINGS: &str = "df_settings"; -const SCHEMATA: &str = "schemata"; -const ROUTINES: &str = "routines"; -const PARAMETERS: &str = "parameters"; - -/// All information schema tables -const INFORMATION_SCHEMA_TABLES: &[&str] = &[ - TABLES, - VIEWS, - COLUMNS, - DF_SETTINGS, - SCHEMATA, - ROUTINES, - PARAMETERS, -]; - // Collect table/CTE references as `TableReference`s and normalize them during traversal. // This avoids a second normalization/conversion pass after visiting the AST. struct RelationVisitor { @@ -127,34 +105,6 @@ impl Visitor for RelationVisitor { self.insert_relation(obj_name)?; } - // SHOW statements will later be rewritten into a SELECT from the information_schema - let requires_information_schema = matches!( - statement, - Statement::ShowFunctions { .. } - | Statement::ShowVariable { .. } - | Statement::ShowStatus { .. } - | Statement::ShowVariables { .. } - | Statement::ShowCreate { .. } - | Statement::ShowColumns { .. } - | Statement::ShowTables { .. } - | Statement::ShowCollation { .. } - ); - if requires_information_schema { - for s in INFORMATION_SCHEMA_TABLES { - // Information schema references are synthesized here, so convert directly. - let obj = ObjectName::from(vec![ - Ident::new(INFORMATION_SCHEMA), - Ident::new(*s), - ]); - match object_name_to_table_reference(obj, self.enable_ident_normalization) - { - Ok(tbl_ref) => { - self.relations.insert(tbl_ref); - } - Err(e) => return ControlFlow::Break(e), - } - } - } ControlFlow::Continue(()) } } From 1e5b40e3c05cd1e6ab277e01a4d0b5a3f61a8473 Mon Sep 17 00:00:00 2001 From: Pepijn Van Eeckhoudt Date: Sat, 3 Oct 2026 15:19:16 +0200 Subject: [PATCH 7/7] Revert information schema content changes --- datafusion/catalog/src/information_schema.rs | 21 +------------------ datafusion/sql/src/statement.rs | 2 +- .../test_files/information_schema.slt | 2 ++ 3 files changed, 4 insertions(+), 21 deletions(-) diff --git a/datafusion/catalog/src/information_schema.rs b/datafusion/catalog/src/information_schema.rs index 9fd70cf9f7d0b..30ed1a3753a98 100644 --- a/datafusion/catalog/src/information_schema.rs +++ b/datafusion/catalog/src/information_schema.rs @@ -181,14 +181,6 @@ impl InformationSchemaConfig { builder.add_schemata(&catalog_name, &schema_name, schema_owner); } } - - // Add the information schema itself - builder.add_schemata(&catalog_name, INFORMATION_SCHEMA, None); - } - - // Add the system information schema - if let Some(system_catalog_name) = &self.system_catalog_name { - builder.add_schemata(system_catalog_name, INFORMATION_SCHEMA, None); } } @@ -234,9 +226,7 @@ impl InformationSchemaConfig { schema: &dyn SchemaProvider, ) -> Result<(), DataFusionError> { for table_name in schema.table_names() { - if let Some(table) = schema.table(&table_name).await? - && table.table_type() == TableType::View - { + if let Some(table) = schema.table(&table_name).await? { builder.add_view( catalog_name, schema_name, @@ -270,15 +260,6 @@ impl InformationSchemaConfig { } } } - - // Add the information schema view columns themselves - Self::add_columns(builder, &catalog_name, INFORMATION_SCHEMA, self).await?; - } - - // Add the system information schema view columns - if let Some(system_catalog_name) = &self.system_catalog_name { - Self::add_columns(builder, system_catalog_name, INFORMATION_SCHEMA, self) - .await?; } Ok(()) diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index a88048a0cbcf2..75a7e6170fe6f 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -3146,7 +3146,7 @@ FROM ( &self, sql_table_name: ObjectName, ) -> Result { - let Some(tables_table_ref) = self.resolve_info_table("tables") else { + let Some(tables_table_ref) = self.resolve_info_table("views") else { return plan_err!( "SHOW CREATE TABLE is not supported unless information_schema is enabled" ); diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index f6633a165df18..b34c772f618fe 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -213,6 +213,7 @@ datafusion.catalog.has_header true datafusion.catalog.information_schema true datafusion.catalog.location NULL datafusion.catalog.newlines_in_values false +datafusion.catalog.system_catalog NULL datafusion.execution.batch_size 8192 datafusion.execution.coalesce_batches true datafusion.execution.collect_statistics true @@ -375,6 +376,7 @@ datafusion.catalog.has_header true Default value for `format.has_header` for `CR datafusion.catalog.information_schema true Should DataFusion provide access to `information_schema` virtual tables for displaying schema information datafusion.catalog.location NULL Location scanned to load tables for `default` schema datafusion.catalog.newlines_in_values false Specifies whether newlines in (quoted) CSV values are supported. This is the default value for `format.newlines_in_values` for `CREATE EXTERNAL TABLE` if not specified explicitly in the statement. Parsing newlines in quoted values may be affected by execution behaviour such as parallel file scanning. Setting this to `true` ensures that newlines in values are parsed successfully, which may reduce performance. +datafusion.catalog.system_catalog NULL The system schema name - when set a global system and `information_schema` catalog will be available datafusion.execution.batch_size 8192 Default batch size while creating new batches, it's especially useful for buffer-in-memory batches since creating tiny batches would result in too much metadata memory consumption datafusion.execution.coalesce_batches true When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting datafusion.execution.collect_statistics true Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Defaults to true.