fix: report unsupported time travel instead of returning current data

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::<PaimonTableProvider>() 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::<PaimonTableProvider>() 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,

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

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;