This is an automated email from the ASF dual-hosted git repository.
jerry-024 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 793ab42 fix(datafusion): skip view lookup for table functions (#518)
793ab42 is described below
commit 793ab4287386af1923d119ef56ba86708824a413
Author: shyjsarah <[email protected]>
AuthorDate: Mon Jul 13 16:23:07 2026 +0800
fix(datafusion): skip view lookup for table functions (#518)
---
crates/integrations/datafusion/src/catalog.rs | 10 ++++
crates/integrations/datafusion/src/sql_context.rs | 70 ++++++++++++++++++++++-
2 files changed, 79 insertions(+), 1 deletion(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 5bcdd10..8ec6f86 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -447,6 +447,16 @@ impl SchemaProvider for PaimonSchemaProvider {
if branch.is_some() {
return Ok(None);
}
+ // DataFusion preloads every relation name before
planning, including
+ // registered table functions. Do not reinterpret a
missing UDTF name as
+ // a REST view; the planner will resolve it through the
UDTF registry.
+ if session_state
+ .as_ref()
+ .and_then(|provider| provider())
+ .is_some_and(|state|
state.table_functions().contains_key(identifier.object()))
+ {
+ return Ok(None);
+ }
let view = match catalog.get_view(&identifier).await {
Ok(view) => view,
Err(paimon::Error::ViewNotExist { .. })
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index b1e8729..5523229 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -5275,6 +5275,8 @@ mod tests {
struct ProbeTrackingCatalog {
get_calls: std::sync::atomic::AtomicUsize,
create_calls: std::sync::atomic::AtomicUsize,
+ get_view_calls: std::sync::atomic::AtomicUsize,
+ table_identifiers: std::sync::Mutex<Vec<String>>,
}
impl ProbeTrackingCatalog {
@@ -5282,6 +5284,8 @@ mod tests {
Self {
get_calls: std::sync::atomic::AtomicUsize::new(0),
create_calls: std::sync::atomic::AtomicUsize::new(0),
+ get_view_calls: std::sync::atomic::AtomicUsize::new(0),
+ table_identifiers: std::sync::Mutex::new(Vec::new()),
}
}
fn get_count(&self) -> usize {
@@ -5290,6 +5294,13 @@ mod tests {
fn create_count(&self) -> usize {
self.create_calls.load(std::sync::atomic::Ordering::SeqCst)
}
+ fn get_view_count(&self) -> usize {
+ self.get_view_calls
+ .load(std::sync::atomic::Ordering::SeqCst)
+ }
+ fn table_identifiers(&self) -> Vec<String> {
+ self.table_identifiers.lock().unwrap().clone()
+ }
}
#[async_trait]
@@ -5307,9 +5318,12 @@ mod tests {
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
Ok(())
}
- async fn get_database(&self, _name: &str) -> paimon::Result<Database> {
+ async fn get_database(&self, name: &str) -> paimon::Result<Database> {
self.get_calls
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ if name == "analytics" {
+ return Ok(Database::new(name.to_string(), HashMap::new(),
None));
+ }
Err(paimon::Error::Unsupported {
message: "simulated Forbidden".to_string(),
})
@@ -5323,6 +5337,10 @@ mod tests {
Ok(())
}
async fn get_table(&self, identifier: &Identifier) ->
paimon::Result<Table> {
+ self.table_identifiers
+ .lock()
+ .unwrap()
+ .push(identifier.to_string());
Err(paimon::Error::TableNotExist {
full_name: identifier.to_string(),
})
@@ -5361,6 +5379,19 @@ mod tests {
) -> paimon::Result<()> {
Ok(())
}
+
+ async fn get_view(
+ &self,
+ _identifier: &Identifier,
+ ) -> paimon::Result<paimon::catalog::View> {
+ self.get_view_calls
+ .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ Err(paimon::Error::RestApi {
+ source: paimon::api::RestError::Forbidden {
+ message: "view access is forbidden".to_string(),
+ },
+ })
+ }
}
#[tokio::test]
@@ -5444,6 +5475,43 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn
registered_table_function_skips_view_lookup_during_relation_preload() {
+ let catalog = Arc::new(ProbeTrackingCatalog::new());
+ let mut ctx = SQLContext::new();
+ ctx.register_catalog_with_default_db("paimon", catalog.clone(), None)
+ .await
+ .unwrap();
+ ctx.set_current_database("analytics").await.unwrap();
+
+ let _ = ctx
+ .sql(
+ "SELECT *
+ FROM vector_search(
+ 'paimon.analytics.documents',
+ 'embedding',
+ '[0.1, 0.2]',
+ 3
+ )",
+ )
+ .await
+ .expect_err("the target table is intentionally absent");
+
+ assert_eq!(
+ catalog.table_identifiers(),
+ vec![
+ "analytics.vector_search",
+ "analytics.documents"
+ ],
+ "DataFusion may preload the UDTF name, then the function must
resolve its table argument"
+ );
+ assert_eq!(
+ catalog.get_view_count(),
+ 0,
+ "registered table functions must not be resolved as REST catalog
views"
+ );
+ }
+
#[tokio::test]
async fn register_catalog_default_wrapper_uses_default_db() {
// The bare register_catalog() must delegate with Some("default"), so
the
