fix(datafusion): resolve temporary objects first in SHOW CREATE TABLE

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

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])),
             ],
         )?;

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