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 d6890564 fix(datafusion): keep metadata queries available without
table engines (#747)
d6890564 is described below
commit d6890564fd5ec35da7672210e45f91b3d45a94bb
Author: shyjsarah <[email protected]>
AuthorDate: Sun Aug 30 21:02:33 2026 +0800
fix(datafusion): keep metadata queries available without table engines
(#747)
---
crates/integrations/datafusion/src/catalog.rs | 82 ++++++++++++--
crates/integrations/datafusion/src/table/mod.rs | 2 +-
.../datafusion/tests/table_type_routing.rs | 126 +++++++++++++++++++--
crates/paimon/src/catalog/filesystem.rs | 3 +-
crates/paimon/src/catalog/mod.rs | 39 ++++++-
crates/paimon/src/catalog/rest/rest_catalog.rs | 3 +-
6 files changed, 231 insertions(+), 24 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index d8893243..340cf07c 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<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`.
///
@@ -693,8 +745,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<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())
@@ -702,13 +770,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(),
@@ -907,7 +968,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 262c3217..9683bf58 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -63,7 +63,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> {
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<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"),
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index 45628ad6..6a57cd86 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -389,8 +389,9 @@ impl Catalog for FileSystemCatalog {
.map(crate::catalog::LoadedTable::Object);
}
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 cd3a8901..f16fd25a 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -278,11 +278,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<Vec<crate::spec::DataField>>,
}
impl ExternalTableMetadata {
@@ -290,6 +291,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 {
@@ -303,6 +309,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 {
@@ -312,7 +338,7 @@ impl LoadedTable {
});
}
options.ensure_engine_can_serve(full_name)?;
- Ok(Self::External(ExternalTableMetadata { declared }))
+ Ok(Self::External(ExternalTableMetadata { declared, fields }))
}
}
@@ -402,7 +428,12 @@ pub trait Catalog: Send + Sync {
.map(LoadedTable::Object);
}
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 7559a369..db167c3d 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -233,8 +233,9 @@ impl Catalog for RESTCatalog {
.map(crate::catalog::LoadedTable::Object);
}
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(),
);
