diff --git a/crates/integrations/datafusion/src/relation_planner.rs b/crates/integrations/datafusion/src/relation_planner.rs index 60978e3d..5c4d983a 100644 --- a/crates/integrations/datafusion/src/relation_planner.rs +++ b/crates/integrations/datafusion/src/relation_planner.rs @@ -93,15 +93,23 @@ impl RelationPlanner for PaimonRelationPlanner { )); } + // Not ours: relation planners are session-global, so another + // engine's planner may handle the clause for its own provider. + let Some(paimon_provider) = provider.downcast_ref::() else { + return Ok(RelationPlanning::Original(Box::new(relation))); + }; + let extra_options = match version { TableVersion::VersionAsOf(expr) => resolve_version_as_of(expr)?, TableVersion::TimestampAsOf(expr) => resolve_timestamp_as_of(expr)?, - _ => return Ok(RelationPlanning::Original(Box::new(relation))), - }; - - // Check if this is a Paimon table. - let Some(paimon_provider) = provider.downcast_ref::() else { - return Ok(RelationPlanning::Original(Box::new(relation))); + // Falling through would answer a historical clause with + // current rows. + _ => { + return Err(plan_datafusion_err!( + "this time-travel syntax is not supported for Paimon tables; \ + use VERSION AS OF or TIMESTAMP AS OF" + )) + } }; // Resolving time travel may switch the table to the snapshot's schema, diff --git a/crates/integrations/datafusion/src/table_loader.rs b/crates/integrations/datafusion/src/table_loader.rs index c72d0385..819c3488 100644 --- a/crates/integrations/datafusion/src/table_loader.rs +++ b/crates/integrations/datafusion/src/table_loader.rs @@ -66,10 +66,34 @@ pub(crate) async fn load_table_for_read( identifier.database().to_string(), parsed.table().to_string(), ); - let mut table = catalog - .get_table(&base_identifier) - .await - .map_err(to_datafusion_error)?; + let mut table = match catalog.load_table(&base_identifier).await { + Ok(paimon::catalog::LoadedTable::Paimon(table)) => *table, + Ok(paimon::catalog::LoadedTable::Object(_)) => { + // This path serves branches, time travel and system tables, none + // of which an object table has; `table()` turns them away too. + return Err(DataFusionError::Plan(format!( + "branches, time travel and system tables are not supported for \ + 'object-table' tables ('{}')", + base_identifier.full_name() + ))); + } + Ok(paimon::catalog::LoadedTable::External(external)) => { + return Err(DataFusionError::Plan(format!( + "table '{}' is declared '{}' and cannot be read as a Paimon table", + base_identifier.full_name(), + external.declared() + ))); + } + // `LoadedTable` is non_exhaustive: a variant added upstream is not a + // Paimon table until this path says how to read one. + Ok(_) => { + return Err(DataFusionError::Plan(format!( + "table '{}' cannot be read as a Paimon table", + base_identifier.full_name() + ))); + } + Err(err) => return Err(to_datafusion_error(err)), + }; let system_table = parsed.system_table().map(str::to_string); if let Some(branch) = parsed.branch() { let is_branches_table = system_table diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs b/crates/integrations/datafusion/tests/table_type_routing.rs index 5620d6bc..a4bd0b3d 100644 --- a/crates/integrations/datafusion/tests/table_type_routing.rs +++ b/crates/integrations/datafusion/tests/table_type_routing.rs @@ -647,6 +647,84 @@ async fn object_and_lance_tables_route_to_engines() { assert_eq!(column_i32(&batches), vec![1, 3]); } +#[tokio::test] +async fn an_unsupported_time_travel_clause_on_a_paimon_table_is_rejected() { + use datafusion::prelude::SessionContext; + use paimon_datafusion::PaimonCatalogProvider; + + 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()); + fs_catalog + .create_database(DB, false, HashMap::new()) + .await + .unwrap(); + let plain = PaimonSchema::builder() + .column( + "id", + paimon::spec::DataType::Int(paimon::spec::IntType::new()), + ) + .build() + .unwrap(); + fs_catalog + .create_table(&Identifier::new(DB, "pt"), plain, false) + .await + .unwrap(); + + let ctx = SessionContext::new(); + ctx.register_catalog( + CATALOG, + Arc::new(PaimonCatalogProvider::new( + Some(CATALOG.to_string()), + fs_catalog, + Default::default(), + Default::default(), + None, + )), + ); + paimon_datafusion::register_catalog_table_engine( + &ctx, + CATALOG, + TableType::IcebergTable, + Arc::new(FakeEngineResolver), + ) + .unwrap(); + // The default dialect does not parse a version clause at all; these are + // the dialects that do, and where the clause used to be dropped. + for dialect in ["databricks", "mssql", "bigquery", "snowflake"] { + ctx.sql(&format!("SET datafusion.sql_parser.dialect = '{dialect}'")) + .await + .unwrap() + .collect() + .await + .unwrap(); + + let sql = format!("SELECT * FROM {CATALOG}.{DB}.pt FOR SYSTEM_TIME AS OF '2020-01-01'"); + let outcome = match ctx.sql(&sql).await { + Err(err) => Err(err), + Ok(df) => df.collect().await.map(|_| ()), + }; + let Err(err) = outcome else { + panic!("[{dialect}] a historical clause must not answer with current rows"); + }; + let msg = err.to_string(); + assert!(msg.contains("VERSION AS OF"), "[{dialect}] {msg}"); + + // The supported syntax still resolves through the same planner. + let err = ctx + .sql(&format!("SELECT * FROM {CATALOG}.{DB}.pt VERSION AS OF 1")) + .await + .unwrap() + .collect() + .await + .err() + .unwrap_or_else(|| panic!("[{dialect}] snapshot 1 does not exist in this fixture")); + assert!(err.to_string().contains("Snapshot 1"), "[{dialect}] {err}"); + } +} + #[tokio::test] async fn time_travel_on_routed_tables_is_rejected() { let env = setup().await;