This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 75dc7227 fix(datafusion): resolve temporary objects first in SHOW
CREATE TABLE (#1126)
75dc7227 is described below
commit 75dc72279b8f1dee379b63424abd25dbb87aed6f
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Oct 15 14:18:14 2026 +0800
fix(datafusion): resolve temporary objects first in SHOW CREATE TABLE
(#1126)
---
crates/integrations/datafusion/src/catalog.rs | 18 +++
crates/integrations/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 19770788..e5093f14 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<Option<Arc<dyn TableProvider>>> {
+ 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 f4c02dc1..3c8d34dd 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<DataFrame> {
+ // 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<Option<(String, String, String, Arc<dyn TableProvider>)>> {
+ let enable_ident_normalization = self.ctx.enable_ident_normalization();
+ let parts: Vec<String> = 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::<crate::catalog::PaimonCatalogProvider>()
+ 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<DataFrame> {
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 8e0eccea..b4ddf07c 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<String>) {
+ 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::<datafusion::arrow::array::StringArray>()
+ .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();
