fix(datafusion): keep metadata queries available without table engines

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<usize>>,
+        _filters: &[Expr],
+        _limit: Option<usize>,
+    ) -> DFResult<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
+        Err(self.unavailable_error())
+    }
+
+    async fn insert_into(
+        &self,
+        _state: &dyn datafusion::catalog::Session,
+        _input: Arc<dyn datafusion::physical_plan::ExecutionPlan>,
+        _insert_op: datafusion::logical_expr::dml::InsertOp,
+    ) -> DFResult<Arc<dyn datafusion::physical_plan::ExecutionPlan>> {
+        Err(self.unavailable_error())
+    }
+}
+
 /// Register `resolver` as the engine for `table_type` on the Paimon catalog
 /// named `catalog_name`.
 ///
@@ -674,22 +726,31 @@ 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<dyn TableProvider>));
+                    };
                     // 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())
                         .clone();
                     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)) => {

crates/integrations/datafusion/src/table/mod.rs

@@ -60,7 +60,7 @@ pub(crate) fn datafusion_read_fields(table: &Table) -> Vec<DataField> {
     fields
 }
 
-fn datafusion_arrow_schema(
+pub(crate) fn datafusion_arrow_schema(
     fields: &[DataField],
     schema_force_view_types: bool,
 ) -> DFResult<ArrowSchemaRef> {

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<String> {
+    batches
+        .iter()
+        .flat_map(|batch| {
+            let values = batch
+                .column_by_name(column)
+                .unwrap()
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .unwrap();
+            (0..values.len())
+                .map(|row| values.value(row).to_string())
+                .collect::<Vec<_>>()
+        })
+        .collect()
+}
+
 #[derive(Debug)]
 struct TypedTestCatalog {
     inner: Arc<FileSystemCatalog>,
@@ -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<LegacyTestCatalog>) {
     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"),

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

crates/paimon/src/catalog/mod.rs

@@ -275,18 +275,24 @@ 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<Vec<crate::spec::DataField>>,
 }
 
 impl ExternalTableMetadata {
     /// The type the table's metadata declares.
     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> {
+        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<crate::spec::DataField>,
+        options: &crate::spec::CoreOptions<'_>,
+        full_name: &str,
+    ) -> Result<Self> {
+        Self::external_impl(declared, Some(fields), options, full_name)
+    }
+
+    fn external_impl(
+        declared: TableType,
+        fields: Option<Vec<crate::spec::DataField>>,
+        options: &crate::spec::CoreOptions<'_>,
+        full_name: &str,
     ) -> Result<Self> {
         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)))
     }

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