Skip to content

fix: scope information schema to catalog #24200

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
VaibhaveS wants to merge 1 commit into
base: main
Choose a base branch
from
+96 −126
Open

fix: scope information schema to catalog #24200

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 67 additions & 97 deletions datafusion/catalog/src/information_schema.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand Down Expand Up @@ -79,11 +79,15 @@ pub struct InformationSchemaProvider {
}

impl InformationSchemaProvider {
/// Creates a new [`InformationSchemaProvider`] for the provided `catalog_list`
pub fn new(catalog_list: Arc<dyn CatalogProviderList>) -> Self {
/// Creates a new [`InformationSchemaProvider`] for the specified catalog.
pub fn new(
catalog_name: impl Into<String>,
catalog: Arc<dyn CatalogProvider>,
) -> Self {
Self {
config: InformationSchemaConfig {
catalog_list,
catalog_name: catalog_name.into(),
catalog,
table_functions: HashMap::new(),
},
}
Expand All @@ -102,7 +106,8 @@ impl InformationSchemaProvider {

#[derive(Clone, Debug)]
struct InformationSchemaConfig {
catalog_list: Arc<dyn CatalogProviderList>,
catalog_name: String,
catalog: Arc<dyn CatalogProvider>,
table_functions: HashMap<String, Arc<TableFunction>>,
}

Expand All @@ -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);
}
}
}
Expand All @@ -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(),
)
}
}
}
Expand All @@ -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,
)
}
}
}
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -1607,24 +1595,6 @@ mod tests {
}
}

impl CatalogProviderList for Fixture {
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 schema_names(&self) -> Vec<String> {
vec!["aschema".to_string()]
Expand Down
27 changes: 14 additions & 13 deletions datafusion/core/src/execution/session_state.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,26 +369,27 @@ impl SessionState {
table_ref: impl Into<TableReference>,
) -> datafusion_common::Result<Arc<dyn SchemaProvider>> {
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
Expand Down
31 changes: 15 additions & 16 deletions datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading