This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 6be17e8 feat: support catalog provider in datafusion (#176)
6be17e8 is described below
commit 6be17e859d656b412a1cb0dd7d336da4ebad0298
Author: yuxia Luo <[email protected]>
AuthorDate: Thu Apr 2 11:08:02 2026 +0800
feat: support catalog provider in datafusion (#176)
---
crates/integrations/datafusion/src/catalog.rs | 148 +++++++++++++++++++++
crates/integrations/datafusion/src/lib.rs | 2 +
.../integrations/datafusion/tests/read_tables.rs | 33 ++++-
crates/paimon/examples/rest_catalog_example.rs | 2 +-
4 files changed, 183 insertions(+), 2 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
new file mode 100644
index 0000000..af5aa3d
--- /dev/null
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -0,0 +1,148 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Paimon catalog integration for DataFusion.
+
+use std::any::Any;
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use datafusion::catalog::{CatalogProvider, SchemaProvider};
+use datafusion::datasource::TableProvider;
+use datafusion::error::Result as DFResult;
+use paimon::catalog::{Catalog, Identifier};
+
+use crate::error::to_datafusion_error;
+use crate::table::PaimonTableProvider;
+
+/// Provides an interface to manage and access multiple schemas (databases)
+/// within a Paimon [`Catalog`].
+///
+/// This provider uses lazy loading - databases and tables are fetched
+/// on-demand from the catalog, ensuring data is always fresh.
+pub struct PaimonCatalogProvider {
+ /// Reference to the Paimon catalog.
+ catalog: Arc<dyn Catalog>,
+}
+
+impl Debug for PaimonCatalogProvider {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PaimonCatalogProvider").finish()
+ }
+}
+
+impl PaimonCatalogProvider {
+ /// Creates a new [`PaimonCatalogProvider`].
+ ///
+ /// All data is loaded lazily when accessed.
+ pub fn new(catalog: Arc<dyn Catalog>) -> Self {
+ PaimonCatalogProvider { catalog }
+ }
+}
+
+impl CatalogProvider for PaimonCatalogProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn schema_names(&self) -> Vec<String> {
+ futures::executor::block_on(async {
+ self.catalog.list_databases().await.unwrap_or_default()
+ })
+ }
+
+ fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>> {
+ futures::executor::block_on(async {
+ match self.catalog.get_database(name).await {
+ Ok(_) => Some(Arc::new(PaimonSchemaProvider::new(
+ self.catalog.clone(),
+ name.to_string(),
+ )) as Arc<dyn SchemaProvider>),
+ Err(paimon::Error::DatabaseNotExist { .. }) => None,
+ Err(_) => None,
+ }
+ })
+ }
+}
+
+/// Represents a [`SchemaProvider`] for the Paimon [`Catalog`], managing
+/// access to table providers within a specific database.
+///
+/// Tables are loaded lazily when accessed via the `table()` method.
+pub struct PaimonSchemaProvider {
+ /// Reference to the Paimon catalog.
+ catalog: Arc<dyn Catalog>,
+ /// Database name this schema represents.
+ database: String,
+}
+
+impl Debug for PaimonSchemaProvider {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("PaimonSchemaProvider")
+ .field("database", &self.database)
+ .finish()
+ }
+}
+
+impl PaimonSchemaProvider {
+ /// Creates a new [`PaimonSchemaProvider`] for the given database.
+ pub fn new(catalog: Arc<dyn Catalog>, database: String) -> Self {
+ PaimonSchemaProvider { catalog, database }
+ }
+}
+
+#[async_trait]
+impl SchemaProvider for PaimonSchemaProvider {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn table_names(&self) -> Vec<String> {
+ // Use blocking call to fetch table names synchronously.
+ // This is acceptable as table name listing is lightweight.
+ futures::executor::block_on(async {
+ self.catalog
+ .list_tables(&self.database)
+ .await
+ .unwrap_or_default()
+ })
+ }
+
+ async fn table(&self, name: &str) -> DFResult<Option<Arc<dyn
TableProvider>>> {
+ let identifier = Identifier::new(self.database.clone(), name);
+ match self.catalog.get_table(&identifier).await {
+ Ok(table) => {
+ let provider = PaimonTableProvider::try_new(table)?;
+ Ok(Some(Arc::new(provider) as Arc<dyn TableProvider>))
+ }
+ Err(paimon::Error::TableNotExist { .. }) => Ok(None),
+ Err(e) => Err(to_datafusion_error(e)),
+ }
+ }
+
+ fn table_exist(&self, name: &str) -> bool {
+ let identifier = Identifier::new(self.database.clone(), name);
+ futures::executor::block_on(async {
+ match self.catalog.get_table(&identifier).await {
+ Ok(_) => true,
+ Err(paimon::Error::TableNotExist { .. }) => false,
+ Err(_) => false,
+ }
+ })
+ }
+}
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 6dd5272..e9f8955 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -35,11 +35,13 @@
//!
//! This version does not support write or predicate pushdown.
+mod catalog;
mod error;
mod physical_plan;
mod schema;
mod table;
+pub use catalog::{PaimonCatalogProvider, PaimonSchemaProvider};
pub use error::to_datafusion_error;
pub use physical_plan::PaimonTableScan;
pub use schema::paimon_schema_to_arrow;
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index 2ac12dc..be43ded 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -18,10 +18,11 @@
use std::sync::Arc;
use datafusion::arrow::array::{Int32Array, StringArray};
+use datafusion::catalog::CatalogProvider;
use datafusion::prelude::SessionContext;
use paimon::catalog::Identifier;
use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
-use paimon_datafusion::PaimonTableProvider;
+use paimon_datafusion::{PaimonCatalogProvider, PaimonTableProvider};
fn get_test_warehouse() -> String {
std::env::var("PAIMON_TEST_WAREHOUSE").unwrap_or_else(|_|
"/tmp/paimon-warehouse".to_string())
@@ -213,3 +214,33 @@ async fn
test_scan_partition_count_respects_session_config() {
"target_partitions=1 should coalesce all splits into exactly 1
partition"
);
}
+
+// ======================= Catalog Provider Tests =======================
+#[tokio::test]
+async fn test_query_via_catalog_provider() {
+ let catalog = create_catalog();
+ let provider = PaimonCatalogProvider::new(Arc::new(catalog));
+
+ let ctx = SessionContext::new();
+ ctx.register_catalog("paimon", Arc::new(provider));
+
+ let df = ctx
+ .sql("SELECT id, name FROM paimon.default.simple_log_table")
+ .await
+ .expect("Failed to execute query");
+
+ let batches = df.collect().await.expect("Failed to collect results");
+ let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(total_rows, 3, "Expected 3 rows from simple_log_table");
+}
+
+#[tokio::test]
+async fn test_missing_database_returns_no_schema() {
+ let catalog = create_catalog();
+ let provider = PaimonCatalogProvider::new(Arc::new(catalog));
+
+ assert!(
+ provider.schema("definitely_missing_database").is_none(),
+ "missing databases should not resolve to a schema provider"
+ );
+}
diff --git a/crates/paimon/examples/rest_catalog_example.rs
b/crates/paimon/examples/rest_catalog_example.rs
index e03f269..cca3cbd 100644
--- a/crates/paimon/examples/rest_catalog_example.rs
+++ b/crates/paimon/examples/rest_catalog_example.rs
@@ -105,7 +105,7 @@ async fn main() {
let catalog = match CatalogFactory::create(options).await {
Ok(catalog) => catalog,
Err(err) => {
- eprintln!("Failed to create catalog: {err}");
+ eprintln!("Failed to create RESTCatalog: {err}");
return;
}
};
