From 8983b9657f28cf70e732d05022ab9f8ec7481221 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Tue, 13 Oct 2026 21:18:52 -0700 Subject: [PATCH] fix(datafusion): resolve temporary objects first in SHOW CREATE TABLE A temporary table or view shadows a catalog table in queries, but SHOW CREATE TABLE still returned the shadowed Paimon table's DDL. Temporary objects were also described through information_schema.views, which loads every table in the catalog, so an unrelated failing table broke the statement. Look up the temporary object first, with names normalized as in queries, and return its definition directly (NULL when it has none). information_schema keeps resolving to DataFusion's metadata tables. --- crates/integrations/datafusion/src/catalog.rs | 18 ++ .../datafusion/src/sql_context.rs | 74 +++++++- .../datafusion/tests/sql_context_tests.rs | 159 ++++++++++++++++++ 3 files changed, 248 insertions(+), 3 deletions(-) diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index 19770788a..e5093f140 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -718,6 +718,24 @@ impl PaimonCatalogProvider { .get(database) .is_some_and(|db| db.table_exist(table_name)) } + + /// Returns the temp table or view with the given name in the specified database. + pub(crate) async fn temp_table( + &self, + database: &str, + table_name: &str, + ) -> DFResult>> { + let temp = self + .temp_tables + .read() + .unwrap_or_else(|e| e.into_inner()) + .get(database) + .cloned(); + match temp { + Some(temp) => temp.table(table_name).await, + None => Ok(None), + } + } } /// Represents a [`SchemaProvider`] for the Paimon [`Catalog`], managing diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index f4c02dc1b..3c8d34dd0 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -1262,6 +1262,17 @@ impl SQLContext { name: &ObjectName, sql_options: SQLOptions, ) -> DFResult { + // Temporary objects shadow catalog tables; as in DataFusion, one without a + // definition shows NULL. + if let Some((catalog_name, database, table, temp)) = self.temp_table_for(name).await? { + return self.show_create_result( + catalog_name, + database, + table, + temp.get_table_definition(), + true, + ); + } let (catalog, catalog_name, identifier) = self.resolve_catalog_and_table(name)?; let table = match catalog.get_table(&identifier).await { Ok(table) => table, @@ -1272,19 +1283,76 @@ impl SQLContext { }; crate::table_loader::ensure_paimon_served(&table, &identifier)?; let definition = crate::table::build_table_definition(&table)?; + self.show_create_result( + catalog_name, + identifier.database().to_string(), + identifier.object().to_string(), + Some(&definition), + false, + ) + } + + /// The temporary table or view `name` resolves to in queries. + async fn temp_table_for( + &self, + name: &ObjectName, + ) -> DFResult)>> { + let enable_ident_normalization = self.ctx.enable_ident_normalization(); + let parts: Vec = name + .0 + .iter() + .filter_map(|part| part.as_ident()) + .map(|ident| normalize_schema_identifier(ident, enable_ident_normalization)) + .collect(); + let table_ref = match parts.as_slice() { + [table] => TableReference::bare(table.as_str()), + [database, table] => TableReference::partial(database.as_str(), table.as_str()), + [catalog, database, table] => { + TableReference::full(catalog.as_str(), database.as_str(), table.as_str()) + } + _ => return Ok(None), + }; + let (catalog_name, database, table) = self.resolve_temp_table_name(table_ref)?; + // Queries resolve information_schema before consulting the catalog. + if database == datafusion::catalog::information_schema::INFORMATION_SCHEMA + && self.ctx.copied_config().information_schema() + { + return Ok(None); + } + let Some(catalog_provider) = self.ctx.catalog(&catalog_name) else { + return Ok(None); + }; + let Some(provider) = + catalog_provider.downcast_ref::() + else { + return Ok(None); + }; + Ok(provider + .temp_table(&database, &table) + .await? + .map(|temp| (catalog_name, database, table, temp))) + } + fn show_create_result( + &self, + catalog_name: String, + database: String, + table: String, + definition: Option<&str>, + nullable: bool, + ) -> DFResult { let schema = Arc::new(Schema::new(vec![ Field::new("table_catalog", ArrowDataType::Utf8, false), Field::new("table_schema", ArrowDataType::Utf8, false), Field::new("table_name", ArrowDataType::Utf8, false), - Field::new("definition", ArrowDataType::Utf8, false), + Field::new("definition", ArrowDataType::Utf8, nullable), ])); let batch = RecordBatch::try_new( schema, vec![ Arc::new(StringArray::from(vec![catalog_name])), - Arc::new(StringArray::from(vec![identifier.database().to_string()])), - Arc::new(StringArray::from(vec![identifier.object().to_string()])), + Arc::new(StringArray::from(vec![database])), + Arc::new(StringArray::from(vec![table])), Arc::new(StringArray::from(vec![definition])), ], )?; diff --git a/crates/integrations/datafusion/tests/sql_context_tests.rs b/crates/integrations/datafusion/tests/sql_context_tests.rs index 8e0eccea2..b4ddf07ce 100644 --- a/crates/integrations/datafusion/tests/sql_context_tests.rs +++ b/crates/integrations/datafusion/tests/sql_context_tests.rs @@ -2852,6 +2852,165 @@ async fn test_show_create_table_round_trip_with_quoted_identifiers_and_options() assert_schema_equivalent(&original_schema, recreated.schema()); } +/// The `table_name` and `definition` (`None` for NULL) of a one-row `SHOW CREATE TABLE` result. +async fn show_create_row(sql_context: &SQLContext, table_ref: &str) -> (String, Option) { + let batches = sql_context + .sql(&format!("SHOW CREATE TABLE {table_ref}")) + .await + .unwrap() + .collect() + .await + .unwrap(); + let rows: Vec<_> = batches.iter().filter(|b| b.num_rows() > 0).collect(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].num_rows(), 1); + let column = |name| { + rows[0] + .column_by_name(name) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + }; + let definition = column("definition"); + ( + column("table_name").value(0).to_string(), + definition + .is_valid(0) + .then(|| definition.value(0).to_string()), + ) +} + +#[tokio::test] +async fn test_show_create_table_prefers_temporary_objects() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + for table in ["t", "u"] { + sql_context + .sql(&format!("CREATE TABLE paimon.test_db.{table} (id INT)")) + .await + .unwrap(); + } + let definition = collect_definition(&sql_context, "paimon.test_db.t").await; + assert!( + definition.contains("CREATE TABLE \"test_db\".\"t\""), + "{definition}" + ); + + sql_context + .sql("CREATE TEMPORARY VIEW paimon.test_db.t AS SELECT 1 AS id") + .await + .unwrap(); + sql_context + .sql("CREATE TEMPORARY TABLE paimon.test_db.u (id INT)") + .await + .unwrap(); + assert_eq!( + show_create_row(&sql_context, "paimon.test_db.t").await, + ("t".to_string(), Some("SELECT 1 AS id".to_string())) + ); + assert_eq!( + show_create_row(&sql_context, "paimon.test_db.u").await, + ("u".to_string(), None) + ); +} + +#[tokio::test] +async fn test_show_create_table_names_temporary_objects_like_queries() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + let schema = Arc::new(datafusion::arrow::datatypes::Schema::empty()); + sql_context + .register_temp_table( + "paimon.test_db.\"Foo\"", + Arc::new(MemTable::try_new(schema, vec![vec![]]).unwrap()), + ) + .unwrap(); + sql_context + .sql("CREATE TEMPORARY VIEW paimon.test_db.foo AS SELECT 1 AS id") + .await + .unwrap(); + + assert_eq!( + show_create_row(&sql_context, "paimon.test_db.Foo").await, + ("foo".to_string(), Some("SELECT 1 AS id".to_string())) + ); + assert_eq!( + show_create_row(&sql_context, "paimon.test_db.\"Foo\"").await, + ("Foo".to_string(), None) + ); +} + +#[tokio::test] +async fn test_show_create_temporary_object_ignores_unrelated_tables() { + let (tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + sql_context + .sql("CREATE SCHEMA paimon.test_db") + .await + .unwrap(); + sql_context + .sql("CREATE TABLE paimon.test_db.broken (id INT)") + .await + .unwrap(); + std::fs::write( + tmp.path().join("test_db.db/broken/schema/schema-0"), + "not a schema", + ) + .unwrap(); + sql_context + .sql("SELECT * FROM information_schema.views") + .await + .unwrap() + .collect() + .await + .expect_err("a catalog scan should load the broken table"); + + sql_context + .sql("CREATE TEMPORARY VIEW paimon.test_db.v AS SELECT 1 AS id") + .await + .unwrap(); + assert_eq!( + show_create_row(&sql_context, "paimon.test_db.v").await, + ("v".to_string(), Some("SELECT 1 AS id".to_string())) + ); +} + +#[tokio::test] +async fn test_show_create_table_resolves_information_schema_like_queries() { + let (_tmp, catalog) = create_test_env(); + let sql_context = create_sql_context(catalog).await; + sql_context + .sql("CREATE TEMPORARY VIEW paimon.information_schema.tables AS SELECT 1 AS id") + .await + .unwrap(); + + let df = sql_context + .sql("SELECT * FROM paimon.information_schema.tables") + .await + .unwrap(); + assert!(df.schema().has_column_with_unqualified_name("table_name")); + let batches = sql_context + .sql("SHOW CREATE TABLE paimon.information_schema.tables") + .await + .unwrap() + .collect() + .await + .unwrap(); + let shown = datafusion::arrow::util::pretty::pretty_format_batches(&batches) + .unwrap() + .to_string(); + assert!(!shown.contains("SELECT 1 AS id"), "{shown}"); +} + #[tokio::test] async fn test_show_create_table_rejects_non_round_trippable_types() { let (_tmp, catalog) = create_test_env();