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]