diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index cbee1a6e..5b28c232 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -154,6 +154,58 @@ impl TableProvider for ReadOnlyTableProvider { } } +/// Metadata-only provider for an external table whose engine is not registered. +/// +/// DataFusion's information schema asks every catalog table for its schema. +/// Returning this provider keeps those metadata queries available without +/// weakening the fail-closed behavior for actual table reads or writes. +#[derive(Debug)] +struct UnavailableEngineTableProvider { + schema: datafusion::arrow::datatypes::SchemaRef, + declared: PaimonTableType, + table_name: String, +} + +impl UnavailableEngineTableProvider { + fn unavailable_error(&self) -> datafusion::error::DataFusionError { + plan_datafusion_err!( + "no table engine is registered for '{}' tables ('{}')", + self.declared, + self.table_name + ) + } +} + +#[async_trait] +impl TableProvider for UnavailableEngineTableProvider { + fn schema(&self) -> datafusion::arrow::datatypes::SchemaRef { + Arc::clone(&self.schema) + } + + fn table_type(&self) -> TableType { + TableType::Base + } + + async fn scan( + &self, + _state: &dyn datafusion::catalog::Session, + _projection: Option<&Vec>, + _filters: &[Expr], + _limit: Option, + ) -> DFResult> { + Err(self.unavailable_error()) + } + + async fn insert_into( + &self, + _state: &dyn datafusion::catalog::Session, + _input: Arc, + _insert_op: datafusion::logical_expr::dml::InsertOp, + ) -> DFResult> { + Err(self.unavailable_error()) + } +} + /// Register `resolver` as the engine for `table_type` on the Paimon catalog /// named `catalog_name`. /// @@ -674,8 +726,24 @@ impl SchemaProvider for PaimonSchemaProvider { identifier.full_name() )); } + let Some(resolver) = table_engines.get(&declared) else { + let schema = match external.fields() { + Some(fields) => crate::table::datafusion_arrow_schema( + fields, + schema_force_view_types, + )?, + None => Arc::new(datafusion::arrow::datatypes::Schema::empty()), + }; + return Ok(Some(Arc::new(UnavailableEngineTableProvider { + schema, + declared, + table_name: identifier.full_name(), + }) as Arc)); + }; // The Paimon arm below applies these; an engine would - // ignore them and answer from current data. + // ignore them and answer from current data. A missing + // engine only exposes catalog metadata, so read-specific + // session options do not apply to that fallback. let session_options = dynamic_options .read() .unwrap_or_else(|e| e.into_inner()) @@ -683,13 +751,6 @@ impl SchemaProvider for PaimonSchemaProvider { paimon::spec::CoreOptions::new(&session_options) .ensure_engine_can_serve(&identifier.full_name()) .map_err(to_datafusion_error)?; - let resolver = table_engines.get(&declared).ok_or_else(|| { - plan_datafusion_err!( - "no table engine is registered for '{}' tables ('{}')", - declared, - identifier.full_name() - ) - })?; let resolved = resolver .resolve_table(&EngineTableRequest::new( identifier.database().to_string(), @@ -881,7 +942,10 @@ impl SchemaProvider for PaimonSchemaProvider { true } }, - None => false, + // `table()` returns a metadata-only provider in this + // case, so the SchemaProvider existence contract + // requires the same answer here. + None => true, } } Ok(paimon::catalog::LoadedTable::Paimon(table)) => { diff --git a/crates/integrations/datafusion/src/table/mod.rs b/crates/integrations/datafusion/src/table/mod.rs index c55b97e7..3c59e4a0 100644 --- a/crates/integrations/datafusion/src/table/mod.rs +++ b/crates/integrations/datafusion/src/table/mod.rs @@ -60,7 +60,7 @@ pub(crate) fn datafusion_read_fields(table: &Table) -> Vec { fields } -fn datafusion_arrow_schema( +pub(crate) fn datafusion_arrow_schema( fields: &[DataField], schema_force_view_types: bool, ) -> DFResult { diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs b/crates/integrations/datafusion/tests/table_type_routing.rs index 3777ea97..5620d6bc 100644 --- a/crates/integrations/datafusion/tests/table_type_routing.rs +++ b/crates/integrations/datafusion/tests/table_type_routing.rs @@ -34,6 +34,23 @@ use tempfile::TempDir; const CATALOG: &str = "cat"; const DB: &str = "shared_db"; +fn string_column_values(batches: &[RecordBatch], column: &str) -> Vec { + batches + .iter() + .flat_map(|batch| { + let values = batch + .column_by_name(column) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + (0..values.len()) + .map(|row| values.value(row).to_string()) + .collect::>() + }) + .collect() +} + #[derive(Debug)] struct TypedTestCatalog { inner: Arc, @@ -88,8 +105,14 @@ impl Catalog for TypedTestCatalog { if let Some(declared) = self.declared_types.get(identifier.object()) { if declared.requires_table_engine() { let options = HashMap::new(); - return LoadedTable::external( + let fields = vec![paimon::spec::DataField::new( + 0, + "external_id".to_string(), + paimon::spec::DataType::Int(paimon::spec::IntType::new()), + )]; + return LoadedTable::external_with_fields( *declared, + fields, &paimon::spec::CoreOptions::new(&options), &identifier.full_name(), ); @@ -867,14 +890,97 @@ async fn an_external_type_without_an_engine_says_so() { .await .unwrap(); - let Err(err) = ctx.sql(&format!("SELECT * FROM {CATALOG}.{DB}.it")).await else { - panic!("an external table without an engine must not resolve"); - }; + let err = ctx + .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it")) + .await + .expect("metadata-only planning should succeed") + .collect() + .await + .expect_err("an external table without an engine must not be readable"); let msg = err.to_string(); assert!(msg.contains("no table engine is registered"), "{msg}"); assert!(msg.contains("iceberg-table"), "{msg}"); } +#[tokio::test] +async fn unregistered_external_table_does_not_break_information_schema_columns() { + let paimon_dir = TempDir::new().unwrap(); + let warehouse = format!("file://{}", paimon_dir.path().display()); + let mut options = Options::new(); + options.set(CatalogOptions::WAREHOUSE, warehouse); + let fs_catalog = Arc::new(FileSystemCatalog::new(options).unwrap()); + let typed_catalog = Arc::new(TypedTestCatalog { + inner: fs_catalog, + declared_types: HashMap::from([("external".to_string(), TableType::IcebergTable)]), + }); + let mut ctx = SQLContext::new(); + ctx.register_catalog(CATALOG, typed_catalog).await.unwrap(); + ctx.sql(&format!("CREATE SCHEMA {CATALOG}.{DB}")) + .await + .unwrap(); + ctx.sql(&format!( + "CREATE TABLE {CATALOG}.{DB}.pt (id INT NOT NULL, name STRING)" + )) + .await + .unwrap(); + ctx.sql("SET 'paimon.scan.version' = '1'").await.unwrap(); + + let provider = ctx.ctx().catalog(CATALOG).unwrap(); + let schema = provider.schema(DB).unwrap(); + assert!( + schema.table("external").await.unwrap().is_some(), + "metadata-only table loading should succeed without a registered engine" + ); + assert!( + schema.table_exist("external"), + "table_exist must agree with the metadata-only table provider" + ); + + let show_tables = ctx + .sql("SHOW TABLES") + .await + .expect("SHOW TABLES must not load an external table engine") + .collect() + .await + .expect("SHOW TABLES must remain queryable"); + let shown_names = string_column_values(&show_tables, "table_name"); + assert!( + shown_names.contains(&"external".to_string()), + "{shown_names:?}" + ); + + let batches = ctx + .sql(&format!( + "SELECT column_name FROM information_schema.columns \ + WHERE table_catalog = '{CATALOG}' \ + AND table_schema = '{DB}' \ + AND table_name = 'pt' \ + ORDER BY ordinal_position" + )) + .await + .expect("an unrelated unregistered engine table must not break planning") + .collect() + .await + .expect("information_schema.columns must remain queryable"); + let names = string_column_values(&batches, "column_name"); + assert_eq!(names, vec!["id", "name"]); + + let external_columns = ctx + .sql(&format!( + "SELECT column_name FROM information_schema.columns \ + WHERE table_catalog = '{CATALOG}' \ + AND table_schema = '{DB}' \ + AND table_name = 'external'" + )) + .await + .expect("the external table schema should be available from catalog metadata") + .collect() + .await + .expect("the external table schema should not require an engine"); + let external_names = string_column_values(&external_columns, "column_name"); + assert_eq!(external_names, vec!["external_id"]); +} + async fn legacy_catalog_with_iceberg_table() -> (TempDir, Arc) { let paimon_dir = TempDir::new().unwrap(); let warehouse = format!("file://{}", paimon_dir.path().display()); @@ -928,9 +1034,13 @@ async fn a_legacy_catalog_cannot_serve_an_external_table_as_paimon() { let mut ctx = SQLContext::new(); ctx.register_catalog(CATALOG, catalog).await.unwrap(); - let Err(err) = ctx.sql(&format!("SELECT * FROM {CATALOG}.{DB}.it")).await else { - panic!("a legacy catalog must not serve an iceberg table as Paimon"); - }; + let err = ctx + .sql(&format!("SELECT * FROM {CATALOG}.{DB}.it")) + .await + .expect("catalog metadata should be sufficient for planning") + .collect() + .await + .expect_err("a legacy catalog must not serve an iceberg table as Paimon"); let msg = err.to_string(); assert!(msg.contains("no table engine is registered"), "{msg}"); assert!(msg.contains("iceberg-table"), "{msg}"); @@ -961,7 +1071,7 @@ async fn a_legacy_catalog_refuses_every_destructive_statement() { let (_dir, ctx) = legacy_sql_context().await; for sql in [ - format!("INSERT INTO {CATALOG}.{DB}.it VALUES (1)"), + format!("INSERT INTO {CATALOG}.{DB}.it VALUES (1, 1)"), format!("INSERT OVERWRITE {CATALOG}.{DB}.it PARTITION (pt = 1) VALUES (1)"), format!("UPDATE {CATALOG}.{DB}.it SET id = 2"), format!("DELETE FROM {CATALOG}.{DB}.it"), diff --git a/crates/paimon/src/catalog/filesystem.rs b/crates/paimon/src/catalog/filesystem.rs index 0ebd8ab8..8b37e4a7 100644 --- a/crates/paimon/src/catalog/filesystem.rs +++ b/crates/paimon/src/catalog/filesystem.rs @@ -361,8 +361,9 @@ impl Catalog for FileSystemCatalog { let options = CoreOptions::new(schema.options()); let declared = options.table_type()?; if declared.requires_table_engine() { - return crate::catalog::LoadedTable::external( + return crate::catalog::LoadedTable::external_with_fields( declared, + schema.fields().to_vec(), &options, &identifier.full_name(), ); diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs index 2b91ff19..2dc8e903 100644 --- a/crates/paimon/src/catalog/mod.rs +++ b/crates/paimon/src/catalog/mod.rs @@ -275,11 +275,12 @@ pub enum LoadedTable { } /// What a caller needs to pick an engine for a table Paimon cannot construct. -/// Only [`LoadedTable::external`] can build one, so the stored-metadata checks -/// always run before a caller sees it. +/// Only [`LoadedTable::external`] and [`LoadedTable::external_with_fields`] can +/// build one, so the stored-metadata checks always run before a caller sees it. #[derive(Debug)] pub struct ExternalTableMetadata { declared: TableType, + fields: Option>, } impl ExternalTableMetadata { @@ -287,6 +288,11 @@ impl ExternalTableMetadata { pub fn declared(&self) -> TableType { self.declared } + + /// The table fields returned by the catalog, when available. + pub fn fields(&self) -> Option<&[crate::spec::DataField]> { + self.fields.as_deref() + } } impl LoadedTable { @@ -300,6 +306,26 @@ impl LoadedTable { declared: TableType, options: &crate::spec::CoreOptions<'_>, full_name: &str, + ) -> Result { + Self::external_impl(declared, None, options, full_name) + } + + /// Like [`Self::external`], preserving catalog metadata for schema-only + /// consumers such as DataFusion's information schema. + pub fn external_with_fields( + declared: TableType, + fields: Vec, + options: &crate::spec::CoreOptions<'_>, + full_name: &str, + ) -> Result { + Self::external_impl(declared, Some(fields), options, full_name) + } + + fn external_impl( + declared: TableType, + fields: Option>, + options: &crate::spec::CoreOptions<'_>, + full_name: &str, ) -> Result { if !declared.requires_table_engine() { return Err(Error::Unsupported { @@ -309,7 +335,7 @@ impl LoadedTable { }); } options.ensure_engine_can_serve(full_name)?; - Ok(Self::External(ExternalTableMetadata { declared })) + Ok(Self::External(ExternalTableMetadata { declared, fields })) } } @@ -390,7 +416,12 @@ pub trait Catalog: Send + Sync { let options = crate::spec::CoreOptions::new(table.schema().options()); let declared = options.table_type()?; if declared.requires_table_engine() { - return LoadedTable::external(declared, &options, &identifier.full_name()); + return LoadedTable::external_with_fields( + declared, + table.schema().fields().to_vec(), + &options, + &identifier.full_name(), + ); } Ok(LoadedTable::Paimon(Box::new(table))) } diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs b/crates/paimon/src/catalog/rest/rest_catalog.rs index be41cfc8..85d7c394 100644 --- a/crates/paimon/src/catalog/rest/rest_catalog.rs +++ b/crates/paimon/src/catalog/rest/rest_catalog.rs @@ -221,8 +221,9 @@ impl Catalog for RESTCatalog { let options = crate::spec::CoreOptions::new(schema.options()); let declared = options.table_type()?; if declared.requires_table_engine() { - return crate::catalog::LoadedTable::external( + return crate::catalog::LoadedTable::external_with_fields( declared, + schema.fields().to_vec(), &options, &identifier.full_name(), );