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();