fix: reject Paimon reads and writes on iceberg/object/lance tables

crates/integrations/datafusion/src/procedures.rs

@@ -276,10 +276,7 @@ async fn get_table(
 ) -> DFResult<Table> {
     let table_str = require_arg(args, "table")?;
     let identifier = resolve_table_identifier(table_str, catalog_name)?;
-    catalog
-        .get_table(&identifier)
-        .await
-        .map_err(to_datafusion_error)
+    crate::table_loader::get_paimon_table(catalog, &identifier).await
 }
 
 fn managers(table: &Table) -> (SnapshotManager, TagManager) {

crates/integrations/datafusion/src/sql_context.rs

@@ -1173,6 +1173,7 @@ impl SQLContext {
             Err(paimon::Error::TableNotExist { .. }) => return self.ctx.sql(sql).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)?;
 
         let schema = Arc::new(Schema::new(vec![

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

@@ -121,6 +121,8 @@ impl PaimonTableProvider {
         table: Table,
         table_definition: Option<String>,
     ) -> DFResult<Self> {
+        let identifier = table.identifier().clone();
+        crate::table_loader::ensure_paimon_served(&table, &identifier)?;
         let fields = datafusion_read_fields(&table);
         let schema = datafusion_arrow_schema(&fields, true)?;
         Ok(Self {
@@ -134,6 +136,8 @@ impl PaimonTableProvider {
         table: Table,
         blob_reader_registry: BlobReaderRegistry,
     ) -> DFResult<Self> {
+        let identifier = table.identifier().clone();
+        crate::table_loader::ensure_paimon_served(&table, &identifier)?;
         blob_reader_registry
             .register_if_absent(table.location().to_string(), table.file_io().clone());
         Self::try_new(table)
@@ -144,6 +148,8 @@ impl PaimonTableProvider {
         blob_reader_registry: BlobReaderRegistry,
         table_definition: Option<String>,
     ) -> DFResult<Self> {
+        let identifier = table.identifier().clone();
+        crate::table_loader::ensure_paimon_served(&table, &identifier)?;
         blob_reader_registry
             .register_if_absent(table.location().to_string(), table.file_io().clone());
         Self::try_new_with_table_definition(table, table_definition)

crates/integrations/datafusion/src/table_loader.rs

@@ -23,6 +23,38 @@ use paimon::table::Table;
 
 use crate::error::to_datafusion_error;
 
+/// [`Catalog::get_table`] for paths that need a `Table` rather than a
+/// [`paimon::catalog::LoadedTable`], rejecting an engine-served declared
+/// type: `get_table` is a trait method, so a catalog outside this repository
+/// can return a table for any type.
+pub(crate) async fn get_paimon_table(
+    catalog: &Arc<dyn Catalog>,
+    identifier: &Identifier,
+) -> DFResult<Table> {
+    let table = catalog
+        .get_table(identifier)
+        .await
+        .map_err(to_datafusion_error)?;
+    ensure_paimon_served(&table, identifier)?;
+    Ok(table)
+}
+
+/// The check from [`get_paimon_table`], for callers that already hold the
+/// table or special-case the `get_table` error.
+pub(crate) fn ensure_paimon_served(table: &Table, identifier: &Identifier) -> DFResult<()> {
+    let declared = paimon::spec::CoreOptions::new(table.schema().options())
+        .table_type()
+        .map_err(to_datafusion_error)?;
+    if declared.requires_table_engine() {
+        return Err(DataFusionError::Plan(format!(
+            "table '{}' is declared '{}' and cannot be read as a Paimon table",
+            identifier.full_name(),
+            declared
+        )));
+    }
+    Ok(())
+}
+
 pub(crate) async fn load_table_for_read(
     catalog: &Arc<dyn Catalog>,
     identifier: &Identifier,