From 8f43fceecd76665445aedf633a3c63e14302a852 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Thu, 15 Oct 2026 02:07:39 -0700 Subject: [PATCH] feat(datafusion): support SHOW CREATE TABLE for external engines Return the definition reported by the engine provider for external data tables, resolved through the catalog provider so session checks and engine errors apply. Paimon tables keep the generated DDL. --- crates/integrations/datafusion/README.md | 6 + .../datafusion/src/sql_context.rs | 56 +++- .../datafusion/tests/table_type_routing.rs | 251 ++++++++++++++++-- 3 files changed, 294 insertions(+), 19 deletions(-) diff --git a/crates/integrations/datafusion/README.md b/crates/integrations/datafusion/README.md index 5c7904b90..b12615c40 100644 --- a/crates/integrations/datafusion/README.md +++ b/crates/integrations/datafusion/README.md @@ -39,6 +39,12 @@ current provider. Raw `SessionContext` needs `register_catalog_table_engine` and dialect supporting these clauses; it preloads the current provider and propagates loading errors. +## External engine DDL + +`SQLContext` answers `SHOW CREATE TABLE` on an external data table with the resolved +provider's `get_table_definition()`, unchanged. A missing or blank definition is +reported as unsupported. + ## Reader memory Paimon Parquet scans use the memory pool from the executing DataFusion `TaskContext`. diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index 3c8d34dd0..bad1203fa 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -1274,15 +1274,35 @@ impl SQLContext { ); } let (catalog, catalog_name, identifier) = self.resolve_catalog_and_table(name)?; - let table = match catalog.get_table(&identifier).await { - Ok(table) => table, + let definition = match catalog.load_table(&identifier).await { + Ok(paimon::catalog::LoadedTable::External(external)) => { + let catalog_provider = self + .ctx + .catalog(&catalog_name) + .ok_or_else(|| plan_datafusion_err!("Unknown catalog '{catalog_name}'"))?; + engine_table_definition(catalog_provider, &identifier, external.declared()).await? + } + Ok(paimon::catalog::LoadedTable::Paimon(table)) => { + crate::table_loader::ensure_paimon_served(&table, &identifier)?; + crate::table::build_table_definition(&table)? + } + Ok(paimon::catalog::LoadedTable::Object(_)) => { + return Err(plan_datafusion_err!( + "table '{}' is declared 'object-table' and cannot be read as a Paimon table", + identifier.full_name() + )); + } + Ok(_) => { + return Err(plan_datafusion_err!( + "table '{}' cannot be read as a Paimon table", + identifier.full_name() + )); + } Err(paimon::Error::TableNotExist { .. }) => { return self.ctx.sql_with_options(sql, sql_options).await; } Err(e) => return Err(to_datafusion_error(e)), }; - 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(), @@ -2456,6 +2476,34 @@ impl SQLContext { } } +/// DDL from the engine provider of an external table. +async fn engine_table_definition( + catalog_provider: Arc, + identifier: &Identifier, + declared: paimon::spec::TableType, +) -> DFResult { + let table_name = identifier.full_name(); + let schema = catalog_provider + .schema(identifier.database()) + .ok_or_else(|| plan_datafusion_err!("database '{}' not found", identifier.database()))?; + let provider = schema + .table(identifier.object()) + .await? + .ok_or_else(|| plan_datafusion_err!("table '{table_name}' not found"))?; + if let Some(unavailable) = + provider.downcast_ref::() + { + return Err(unavailable.unavailable_error()); + } + match provider.get_table_definition() { + Some(definition) if !definition.trim().is_empty() => Ok(definition.to_string()), + _ => Err(plan_datafusion_err!( + "SHOW CREATE TABLE is not supported for '{declared}' table '{table_name}': \ + the table engine provides no definition" + )), + } +} + fn validate_immutable_scalar_plan(plan: &LogicalPlan) -> DFResult<()> { let mut violation = None; plan.apply(|node| { diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs b/crates/integrations/datafusion/tests/table_type_routing.rs index 9e2914308..45e3c27f2 100644 --- a/crates/integrations/datafusion/tests/table_type_routing.rs +++ b/crates/integrations/datafusion/tests/table_type_routing.rs @@ -477,6 +477,13 @@ async fn setup() -> TestEnv { } async fn setup_with_engine(resolver: Option>) -> TestEnv { + setup_with_external_tables(resolver, &["it", "ghost", "ft"]).await +} + +async fn setup_with_external_tables( + resolver: Option>, + external_tables: &[&str], +) -> TestEnv { let paimon_dir = TempDir::new().unwrap(); let warehouse = format!("file://{}", paimon_dir.path().display()); let mut options = Options::new(); @@ -484,11 +491,10 @@ async fn setup_with_engine(resolver: Option>) -> Te let fs_catalog = Arc::new(FileSystemCatalog::new(options).unwrap()); let typed_catalog = Arc::new(TypedTestCatalog { inner: Arc::clone(&fs_catalog), - declared_types: HashMap::from([ - ("it".to_string(), TableType::IcebergTable), - ("ghost".to_string(), TableType::IcebergTable), - ("ft".to_string(), TableType::IcebergTable), - ]), + declared_types: external_tables + .iter() + .map(|name| (name.to_string(), TableType::IcebergTable)) + .collect(), }); let mut ctx = SQLContext::new(); ctx.register_catalog(CATALOG, typed_catalog).await.unwrap(); @@ -2021,19 +2027,234 @@ async fn every_version_clause_on_routed_tables_is_rejected() { } } +#[derive(Debug)] +struct DefinedTable { + inner: MemTable, + definition: Option, +} + +#[async_trait] +impl TableProvider for DefinedTable { + fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef { + self.inner.schema() + } + + fn table_type(&self) -> datafusion::logical_expr::TableType { + datafusion::logical_expr::TableType::Base + } + + fn get_table_definition(&self) -> Option<&str> { + self.definition.as_deref() + } + + async fn scan( + &self, + state: &dyn datafusion::catalog::Session, + projection: Option<&Vec>, + filters: &[datafusion::logical_expr::Expr], + limit: Option, + ) -> DFResult> { + self.inner.scan(state, projection, filters, limit).await + } +} + +const QUOTED_TABLE: &str = "Lake Events"; + +fn engine_ddl(database: &str, table: &str) -> String { + format!("CREATE TABLE lake.{database}.`{table}` (id INT) USING iceberg\n") +} + +#[derive(Debug, Default)] +struct DdlEngineResolver { + calls: AtomicUsize, +} + +#[async_trait] +impl TableEngineResolver for DdlEngineResolver { + async fn resolve_table( + &self, + request: &EngineTableRequest, + ) -> DFResult>> { + self.calls.fetch_add(1, Ordering::Relaxed); + let definition = match request.table.as_str() { + "it" | QUOTED_TABLE => Some(engine_ddl(&request.database, &request.table)), + "no_ddl" => None, + "blank_ddl" => Some(" \n\t".to_string()), + "broken" => { + return Err(DataFusionError::Execution( + "ddl backend exploded".to_string(), + )) + } + _ => return Ok(None), + }; + let schema = Arc::new(ArrowSchema::new(vec![Field::new( + "id", + DataType::Int32, + false, + )])); + Ok(Some(Arc::new(DefinedTable { + inner: MemTable::try_new(schema, vec![vec![]])?, + definition, + }))) + } +} + +async fn ddl_env(external_tables: &[&str]) -> (TestEnv, Arc) { + let resolver = Arc::new(DdlEngineResolver::default()); + let env = setup_with_external_tables(Some(resolver.clone()), external_tables).await; + (env, resolver) +} + +async fn show_create(env: &TestEnv, table: &str) -> DFResult> { + show_create_in(env, DB, table).await +} + +async fn show_create_in(env: &TestEnv, db: &str, table: &str) -> DFResult> { + env.ctx + .sql(&format!("SHOW CREATE TABLE {CATALOG}.{db}.{table}")) + .await? + .collect() + .await +} + +fn show_create_definition(batches: &[RecordBatch]) -> Option { + 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 = rows[0] + .column_by_name("definition") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + column.is_valid(0).then(|| column.value(0).to_string()) +} + #[tokio::test] -async fn show_create_on_routed_tables_reports_the_declared_type() { - let env = setup().await; - let Err(err) = env +async fn show_create_returns_the_engine_definition() { + let (env, resolver) = ddl_env(&["it", QUOTED_TABLE]).await; + env.ctx + .sql(&format!("CREATE SCHEMA {CATALOG}.other_db")) + .await + .unwrap(); + for (db, name, table) in [ + (DB, "it", "it"), + (DB, "\"Lake Events\"", QUOTED_TABLE), + ("other_db", "it", "it"), + ] { + let batches = show_create_in(&env, db, name).await.unwrap(); + assert_eq!(string_column_values(&batches, "table_catalog"), [CATALOG]); + assert_eq!(string_column_values(&batches, "table_schema"), [db]); + assert_eq!(string_column_values(&batches, "table_name"), [table]); + assert_eq!( + show_create_definition(&batches), + Some(engine_ddl(db, table)) + ); + } + assert_eq!(resolver.calls.load(Ordering::Relaxed), 3); +} + +#[tokio::test] +async fn show_create_reports_external_tables_it_cannot_describe() { + let (env, _) = ddl_env(&["no_ddl", "blank_ddl", "broken", "ghost"]).await; + for (table, message) in [ + ( + "no_ddl", + "SHOW CREATE TABLE is not supported for 'iceberg-table' table 'shared_db.no_ddl'", + ), + ( + "blank_ddl", + "SHOW CREATE TABLE is not supported for 'iceberg-table' table 'shared_db.blank_ddl'", + ), + ("broken", "ddl backend exploded"), + ("ghost", "table 'shared_db.ghost' not found"), + ("missing", "not found"), + ] { + let err = show_create(&env, table).await.unwrap_err(); + assert!(err.to_string().contains(message), "{table}: {err}"); + } + + let err = show_create(&setup_with_engine(None).await, "it") + .await + .unwrap_err(); + assert!( + err.to_string() + .contains("no table engine is registered for 'iceberg-table' tables"), + "{err}" + ); + let err = show_create(&setup().await, "it").await.unwrap_err(); + assert!( + err.to_string() + .contains("SHOW CREATE TABLE is not supported for 'iceberg-table'"), + "{err}" + ); +} + +#[tokio::test] +async fn show_create_resolves_only_the_target_table() { + let (env, resolver) = ddl_env(&["it", "broken"]).await; + let err = env .ctx - .sql(&format!("SHOW CREATE TABLE {CATALOG}.{DB}.it")) + .sql("SELECT * FROM information_schema.views") .await - else { - panic!("Paimon DDL would misrepresent an engine-served table"); - }; - let msg = err.to_string(); - assert!(msg.contains("iceberg-table"), "{msg}"); - assert!(msg.contains("cannot be read as a Paimon table"), "{msg}"); + .unwrap() + .collect() + .await + .unwrap_err(); + assert!(err.to_string().contains("ddl backend exploded"), "{err}"); + + let calls = resolver.calls.load(Ordering::Relaxed); + let batches = show_create(&env, "it").await.unwrap(); + assert_eq!(show_create_definition(&batches), Some(engine_ddl(DB, "it"))); + assert_eq!(resolver.calls.load(Ordering::Relaxed), calls + 1); +} + +#[tokio::test] +async fn show_create_checks_session_options_before_calling_the_engine() { + let (env, resolver) = ddl_env(&["it"]).await; + for (key, value, message) in [ + ("query-auth.enabled", "true", "query-auth"), + ("scan.version", "20", "time travel"), + ("incremental-between", "1,5", "incremental-between"), + ] { + env.ctx + .sql(&format!("SET 'paimon.{key}' = '{value}'")) + .await + .unwrap(); + let err = show_create(&env, "it").await.unwrap_err(); + assert!(err.to_string().contains(message), "{key}: {err}"); + env.ctx.sql(&format!("RESET 'paimon.{key}'")).await.unwrap(); + } + assert_eq!(resolver.calls.load(Ordering::Relaxed), 0); +} + +#[tokio::test] +async fn show_create_follows_temporary_shadowing_and_keeps_paimon_ddl() { + // `broken` would fail a catalog scan. + let (env, resolver) = ddl_env(&["it", "broken"]).await; + let paimon_ddl = show_create_definition(&show_create(&env, "pt").await.unwrap()).unwrap(); + assert!( + paimon_ddl.contains("CREATE TABLE \"shared_db\".\"pt\""), + "{paimon_ddl}" + ); + assert!(paimon_ddl.contains("\"id\" INT NOT NULL"), "{paimon_ddl}"); + + let schema = Arc::new(ArrowSchema::new(vec![Field::new( + "id", + DataType::Int32, + false, + )])); + env.ctx + .register_temp_table( + format!("{CATALOG}.{DB}.it"), + Arc::new(MemTable::try_new(schema, vec![vec![]]).unwrap()), + ) + .unwrap(); + + let batches = show_create(&env, "it").await.unwrap(); + assert_eq!(string_column_values(&batches, "table_name"), ["it"]); + assert_eq!(show_create_definition(&batches), None); + assert_eq!(resolver.calls.load(Ordering::Relaxed), 0); } #[tokio::test]