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 c3c3423e feat(datafusion): support SHOW CREATE TABLE for external
engines (#1131)
c3c3423e is described below
commit c3c3423ef385ba88999f94e09dd31db2328e3335
Author: Jiajia Li <[email protected]>
AuthorDate: Fri Oct 16 12:11:08 2026 +0800
feat(datafusion): support SHOW CREATE TABLE for external engines (#1131)
---
crates/integrations/datafusion/README.md | 6 +
crates/integrations/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 5c7904b9..b12615c4 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 3c8d34dd..bad1203f 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<dyn datafusion::catalog::CatalogProvider>,
+ identifier: &Identifier,
+ declared: paimon::spec::TableType,
+) -> DFResult<String> {
+ 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::<crate::catalog::UnavailableEngineTableProvider>()
+ {
+ 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 9e291430..45e3c27f 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<Arc<dyn TableEngineResolver>>) ->
TestEnv {
+ setup_with_external_tables(resolver, &["it", "ghost", "ft"]).await
+}
+
+async fn setup_with_external_tables(
+ resolver: Option<Arc<dyn TableEngineResolver>>,
+ 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<Arc<dyn
TableEngineResolver>>) -> 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<String>,
+}
+
+#[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<usize>>,
+ filters: &[datafusion::logical_expr::Expr],
+ limit: Option<usize>,
+ ) -> DFResult<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
+ 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<Option<Arc<dyn TableProvider>>> {
+ 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<DdlEngineResolver>) {
+ 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<Vec<RecordBatch>>
{
+ show_create_in(env, DB, table).await
+}
+
+async fn show_create_in(env: &TestEnv, db: &str, table: &str) ->
DFResult<Vec<RecordBatch>> {
+ env.ctx
+ .sql(&format!("SHOW CREATE TABLE {CATALOG}.{db}.{table}"))
+ .await?
+ .collect()
+ .await
+}
+
+fn show_create_definition(batches: &[RecordBatch]) -> Option<String> {
+ 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::<StringArray>()
+ .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]
