diff --git a/crates/integrations/datafusion/README.md b/crates/integrations/datafusion/README.md index 1b6523e22..5c7904b90 100644 --- a/crates/integrations/datafusion/README.md +++ b/crates/integrations/datafusion/README.md @@ -27,13 +27,16 @@ This crate contains the integration of [Apache DataFusion](https://datafusion.ap ## External engine snapshots Implement `TableEngineResolver::resolve_table_snapshot` to support -`VERSION AS OF ` on external data tables (integer IDs only). +`VERSION AS OF ` on external data tables (integer IDs only), and +`resolve_table_at_timestamp` to support `TIMESTAMP AS OF ''`. +The timestamp is read in the local time zone and passed as epoch milliseconds; +the engine picks the snapshot. Providers must use the snapshot's data and schema. Missing snapshots are errors; `Ok(None)` means the table is missing. Resolved providers remain read-only. `SQLContext` needs no dialect setting and resolves snapshots independently of the current provider. Raw `SessionContext` needs `register_catalog_table_engine` and a -dialect supporting `VERSION AS OF`; it preloads the current provider and propagates +dialect supporting these clauses; it preloads the current provider and propagates loading errors. ## Reader memory diff --git a/crates/integrations/datafusion/src/catalog.rs b/crates/integrations/datafusion/src/catalog.rs index 4232ae2b0..19770788a 100644 --- a/crates/integrations/datafusion/src/catalog.rs +++ b/crates/integrations/datafusion/src/catalog.rs @@ -93,6 +93,22 @@ pub trait TableEngineResolver: Debug + Send + Sync { )) } + /// Resolve the table as of `timestamp_millis` (epoch milliseconds) and its + /// schema; the engine picks the snapshot. A time with no snapshot is an error; + /// `Ok(None)` means the table is missing. + async fn resolve_table_at_timestamp( + &self, + request: &EngineTableRequest, + _timestamp_millis: i64, + ) -> DFResult>> { + Err(plan_datafusion_err!( + "timestamp time travel is not supported for routed '{}' tables ('{}.{}')", + request.declared, + request.database, + request.table + )) + } + /// Resolves `$`; returns `Ok(None)` if missing or unsupported. /// The request names the base table; `metadata_table` is ASCII-lowercase. /// Session scan options are checked before this call. @@ -117,25 +133,50 @@ pub(crate) struct ReadOnlyTableProvider { dynamic_options: DynamicOptions, } +/// The historical state requested from an engine. +#[derive(Debug, Clone, Copy)] +pub(crate) enum EngineTimeTravel { + Snapshot(i64), + TimestampMillis(i64), +} + +impl std::fmt::Display for EngineTimeTravel { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Snapshot(id) => write!(f, "snapshot {id}"), + Self::TimestampMillis(millis) => write!(f, "timestamp-millis {millis}"), + } + } +} + +pub(crate) fn unsupported_engine_time_travel( + declared: PaimonTableType, + table_name: &str, +) -> datafusion::error::DataFusionError { + plan_datafusion_err!( + "time travel is not supported for routed '{declared}' tables ('{table_name}'); \ + only VERSION AS OF with an integer snapshot id and TIMESTAMP AS OF \ + are supported on data tables" + ) +} + impl ReadOnlyTableProvider { pub(crate) fn unsupported_time_travel(&self) -> datafusion::error::DataFusionError { - plan_datafusion_err!( - "time travel is not supported for routed '{}' tables ('{}'); \ - only VERSION AS OF with an integer snapshot id is supported on data tables", - self.declared, - self.table_name - ) + unsupported_engine_time_travel(self.declared, &self.table_name) } - pub(crate) async fn at_snapshot(&self, snapshot_id: i64) -> DFResult> { + pub(crate) async fn at_time_travel( + &self, + travel: EngineTimeTravel, + ) -> DFResult> { if self.metadata_table { return Err(self.unsupported_time_travel()); } - resolve_engine_snapshot( + resolve_engine_time_travel( self.request.clone(), Arc::clone(&self.resolver), Arc::clone(&self.dynamic_options), - snapshot_id, + travel, ) .await } @@ -149,11 +190,11 @@ pub(crate) fn parse_engine_snapshot_id(version: &str) -> DFResult { }) } -async fn resolve_engine_snapshot( +async fn resolve_engine_time_travel( request: EngineTableRequest, resolver: Arc, dynamic_options: DynamicOptions, - snapshot_id: i64, + travel: EngineTimeTravel, ) -> DFResult> { await_with_runtime(async move { let table_name = format!("{}.{}", request.database, request.table); @@ -164,12 +205,17 @@ async fn resolve_engine_snapshot( paimon::spec::CoreOptions::new(&options) .ensure_engine_can_serve(&table_name) .map_err(to_datafusion_error)?; - let inner = resolver - .resolve_table_snapshot(&request, snapshot_id) - .await? - .ok_or_else(|| { - plan_datafusion_err!("table '{table_name}' was not found at snapshot {snapshot_id}") - })?; + let resolved = match travel { + EngineTimeTravel::Snapshot(id) => resolver.resolve_table_snapshot(&request, id).await?, + EngineTimeTravel::TimestampMillis(millis) => { + resolver + .resolve_table_at_timestamp(&request, millis) + .await? + } + }; + let inner = resolved.ok_or_else(|| { + plan_datafusion_err!("table '{table_name}' was not found at {travel}") + })?; Ok(Arc::new(ReadOnlyTableProvider { inner, declared: request.declared, @@ -423,10 +469,10 @@ impl PaimonCatalogProvider { Arc::clone(&self.table_engines) } - pub(crate) async fn table_at_snapshot( + pub(crate) async fn table_at_time_travel( &self, request: EngineTableRequest, - snapshot_id: i64, + travel: EngineTimeTravel, ) -> DFResult> { let resolver = self .table_engines @@ -442,13 +488,8 @@ impl PaimonCatalogProvider { request.table ) })?; - resolve_engine_snapshot( - request, - resolver, - Arc::clone(&self.dynamic_options), - snapshot_id, - ) - .await + resolve_engine_time_travel(request, resolver, Arc::clone(&self.dynamic_options), travel) + .await } fn paimon_schema(&self, name: &str) -> Option> { diff --git a/crates/integrations/datafusion/src/relation_planner.rs b/crates/integrations/datafusion/src/relation_planner.rs index 2078d48ec..10b2679db 100644 --- a/crates/integrations/datafusion/src/relation_planner.rs +++ b/crates/integrations/datafusion/src/relation_planner.rs @@ -34,11 +34,12 @@ use datafusion::sql::sqlparser::ast::{self, TableFactor, TableVersion}; use paimon::spec::{SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION}; use crate::catalog::{ - parse_engine_snapshot_id, ReadOnlyTableProvider, UnavailableEngineTableProvider, + parse_engine_snapshot_id, EngineTimeTravel, ReadOnlyTableProvider, + UnavailableEngineTableProvider, }; use crate::table::{ObjectTableProvider, PaimonTableProvider}; -/// Resolves Paimon time travel and registered engine snapshot IDs. +/// Resolves Paimon time travel, and registered engine snapshot IDs and timestamps. #[derive(Debug)] pub struct PaimonRelationPlanner; @@ -84,15 +85,22 @@ impl RelationPlanner for PaimonRelationPlanner { } if let Some(routed) = provider.downcast_ref::() { - let TableVersion::VersionAsOf(expr) = version else { - return Err(routed.unsupported_time_travel()); + let travel = match version { + TableVersion::VersionAsOf(expr) => { + let options = resolve_version_as_of(expr)?; + EngineTimeTravel::Snapshot(parse_engine_snapshot_id( + &options[SCAN_VERSION_OPTION], + )?) + } + TableVersion::TimestampAsOf(expr) => { + EngineTimeTravel::TimestampMillis(timestamp_as_of_millis(expr)?) + } + _ => return Err(routed.unsupported_time_travel()), }; - let options = resolve_version_as_of(expr)?; - let snapshot_id = parse_engine_snapshot_id(&options[SCAN_VERSION_OPTION])?; let routed = routed.clone(); let new_provider = crate::runtime::block_on_with_runtime( - async move { routed.at_snapshot(snapshot_id).await }, - "engine snapshot resolution thread panicked", + async move { routed.at_time_travel(travel).await }, + "engine time travel resolution thread panicked", )?; return plan_scan(relation, table_ref, new_provider); } @@ -210,14 +218,19 @@ fn resolve_version_as_of(expr: &ast::Expr) -> DFResult> /// Resolve `TIMESTAMP AS OF ` into `scan.timestamp-millis` option. fn resolve_timestamp_as_of(expr: &ast::Expr) -> DFResult> { + let millis = timestamp_as_of_millis(expr)?; + Ok(HashMap::from([( + SCAN_TIMESTAMP_MILLIS_OPTION.to_string(), + millis.to_string(), + )])) +} + +/// Parse the timestamp string of `TIMESTAMP AS OF ` to epoch millis. +fn timestamp_as_of_millis(expr: &ast::Expr) -> DFResult { match expr { ast::Expr::Value(v) => match &v.value { ast::Value::SingleQuotedString(s) | ast::Value::DoubleQuotedString(s) => { - let millis = parse_timestamp_to_millis(s)?; - Ok(HashMap::from([( - SCAN_TIMESTAMP_MILLIS_OPTION.to_string(), - millis.to_string(), - )])) + parse_timestamp_to_millis(s) } _ => Err(datafusion::error::DataFusionError::Plan(format!( "Unsupported TIMESTAMP AS OF expression: {expr}. Expected a timestamp string." diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index 140a424f5..f4c02dc1b 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -773,8 +773,6 @@ impl SQLContext { sql: &str, sql_options: SQLOptions, ) -> DFResult { - use paimon::spec::{SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION}; - let mut tracker = crate::merge_into::TempTableTracker::new(self); let version_clauses = extract_all_version_as_of(sql); @@ -797,8 +795,7 @@ impl SQLContext { &catalog, &catalog_name, &identifier, - SCAN_VERSION_OPTION, - info.version.clone(), + TimeTravelClause::Version(info.version.clone()), ) .await?; @@ -818,8 +815,7 @@ impl SQLContext { &catalog, &catalog_name, &identifier, - SCAN_TIMESTAMP_MILLIS_OPTION, - millis.to_string(), + TimeTravelClause::TimestampMillis(millis), ) .await?; @@ -865,9 +861,11 @@ impl SQLContext { catalog: &Arc, catalog_name: &str, identifier: &Identifier, - option: &str, - value: String, + clause: TimeTravelClause, ) -> DFResult> { + use crate::catalog::EngineTimeTravel; + use paimon::spec::{SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION}; + let parsed = identifier .parsed_object_name() .map_err(to_datafusion_error)?; @@ -887,15 +885,20 @@ impl SQLContext { identifier.full_name() )); } - if parsed.system_table().is_some() || option != paimon::spec::SCAN_VERSION_OPTION { - return Err(plan_datafusion_err!( - "time travel is not supported for routed '{}' tables ('{}'); \ - only VERSION AS OF with an integer snapshot id is supported on data tables", + if parsed.system_table().is_some() { + return Err(crate::catalog::unsupported_engine_time_travel( external.declared(), - identifier.full_name() + &identifier.full_name(), )); } - let snapshot_id = crate::catalog::parse_engine_snapshot_id(&value)?; + let travel = match clause { + TimeTravelClause::Version(version) => { + EngineTimeTravel::Snapshot(crate::catalog::parse_engine_snapshot_id(&version)?) + } + TimeTravelClause::TimestampMillis(millis) => { + EngineTimeTravel::TimestampMillis(millis) + } + }; let provider = self .ctx .catalog(catalog_name) @@ -906,19 +909,25 @@ impl SQLContext { plan_datafusion_err!("Catalog '{catalog_name}' is not a Paimon catalog") })?; return provider - .table_at_snapshot( + .table_at_time_travel( crate::EngineTableRequest::new( base.database().to_string(), base.object().to_string(), external.declared(), ), - snapshot_id, + travel, ) .await; } let (table, base, system_name) = crate::table_loader::loaded_table_for_read(loaded, identifier).await?; + let (option, value) = match clause { + TimeTravelClause::Version(version) => (SCAN_VERSION_OPTION, version), + TimeTravelClause::TimestampMillis(millis) => { + (SCAN_TIMESTAMP_MILLIS_OPTION, millis.to_string()) + } + }; let mut options = self.dynamic_options.read().unwrap().clone(); options.insert(option.to_string(), value); let table = table @@ -3584,6 +3593,12 @@ struct TimestampAsOfInfo { clause_range: (usize, usize), } +/// A parsed `VERSION AS OF` or `TIMESTAMP AS OF` selector. +enum TimeTravelClause { + Version(String), + TimestampMillis(i64), +} + /// Check whether a SQL string contains a time-travel keyword (`VERSION AS OF` or /// `TIMESTAMP AS OF`) **outside** of single-quoted string literals, `--` line /// comments, and `/* */` block comments. diff --git a/crates/integrations/datafusion/tests/table_type_routing.rs b/crates/integrations/datafusion/tests/table_type_routing.rs index 243140aab..9e2914308 100644 --- a/crates/integrations/datafusion/tests/table_type_routing.rs +++ b/crates/integrations/datafusion/tests/table_type_routing.rs @@ -17,7 +17,7 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use async_trait::async_trait; use datafusion::arrow::array::{Array, Int32Array, StringArray}; @@ -311,9 +311,24 @@ struct SnapshotEngineResolver { calls: AtomicUsize, current_calls: AtomicUsize, current_unavailable: bool, + timestamps: Mutex>, } const LARGE_SNAPSHOT_ID: i64 = 6_251_097_304_546_047_273; +/// Local commit times of the fake engine's snapshots 10 and 20. +const SNAPSHOT_10_TIME: &str = "2026-10-01 00:00:00"; +const SNAPSHOT_20_TIME: &str = "2026-10-02 00:00:00"; +const BACKEND_FAILURE_TIME: &str = "2000-01-01 00:00:00"; + +fn local_millis(timestamp: &str) -> i64 { + use chrono::TimeZone; + let naive = chrono::NaiveDateTime::parse_from_str(timestamp, "%Y-%m-%d %H:%M:%S").unwrap(); + chrono::Local + .from_local_datetime(&naive) + .single() + .unwrap() + .timestamp_millis() +} #[async_trait] impl TableEngineResolver for SnapshotEngineResolver { @@ -379,6 +394,27 @@ impl TableEngineResolver for SnapshotEngineResolver { vec![vec![batch]], )?))) } + + async fn resolve_table_at_timestamp( + &self, + request: &EngineTableRequest, + timestamp_millis: i64, + ) -> DFResult>> { + self.timestamps.lock().unwrap().push(timestamp_millis); + // Pick the latest snapshot committed at or before the timestamp. + let snapshot_id = if timestamp_millis == local_millis(BACKEND_FAILURE_TIME) { + 500 + } else if timestamp_millis >= local_millis(SNAPSHOT_20_TIME) { + 20 + } else if timestamp_millis >= local_millis(SNAPSHOT_10_TIME) { + 10 + } else { + return Err(DataFusionError::Plan(format!( + "no snapshot exists at or before {timestamp_millis}" + ))); + }; + self.resolve_table_snapshot(request, snapshot_id).await + } } #[async_trait] @@ -582,20 +618,71 @@ async fn engine_snapshot_uses_historical_data_and_schema() { } } +#[tokio::test] +async fn engine_timestamp_passes_exact_millis_and_uses_the_resolved_snapshot() { + let (env, resolver) = snapshot_env().await; + for native in [false, true] { + for (timestamp, expected) in [ + (SNAPSHOT_10_TIME, ["old-a", "old-b"]), + ("2026-10-01 23:59:59", ["old-a", "old-b"]), + (SNAPSHOT_20_TIME, ["new-a", "new-d"]), + ("2026-10-03 00:00:00", ["new-a", "new-d"]), + ] { + resolver.timestamps.lock().unwrap().clear(); + let batches = routed_sql( + &env, + native, + &format!( + "SELECT h.historical FROM {CATALOG}.{DB}.it TIMESTAMP AS OF '{timestamp}' AS h \ + ORDER BY h.id" + ), + ) + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!( + string_column_values(&batches, "historical"), + expected, + "{timestamp}" + ); + assert_eq!( + *resolver.timestamps.lock().unwrap(), + [local_millis(timestamp)] + ); + } + + let err = routed_sql( + &env, + native, + &format!("SELECT payload FROM {CATALOG}.{DB}.it TIMESTAMP AS OF '{SNAPSHOT_20_TIME}'"), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("payload"), "{err}"); + } +} + #[tokio::test] async fn sql_context_snapshot_remains_readable_when_current_resolution_fails() { let (env, resolver) = current_unavailable_env().await; - let batches = env - .ctx - .sql(&format!( - "SELECT h.historical FROM {CATALOG}.{DB}.it VERSION AS OF 10 AS h WHERE h.id = 2" - )) - .await - .unwrap() - .collect() - .await - .unwrap(); - assert_eq!(string_column_values(&batches, "historical"), ["old-b"]); + for selector in [ + "VERSION AS OF 10".to_string(), + format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"), + ] { + let batches = env + .ctx + .sql(&format!( + "SELECT h.historical FROM {CATALOG}.{DB}.it {selector} AS h WHERE h.id = 2" + )) + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!(string_column_values(&batches, "historical"), ["old-b"]); + } let err = env .ctx @@ -629,21 +716,27 @@ async fn sql_context_snapshot_remains_readable_when_current_resolution_fails() { #[tokio::test] async fn raw_session_snapshot_propagates_current_resolution_errors() { let (env, resolver) = current_unavailable_env().await; - for suffix in ["", " WHERE FALSE", " LIMIT 0"] { - let err = env - .ctx - .ctx() - .sql(&format!( - "SELECT * FROM {CATALOG}.{DB}.it VERSION AS OF 10{suffix}" - )) - .await - .unwrap_err(); - assert!( - err.to_string().contains("current table is unavailable"), - "{err}" - ); + for selector in [ + "VERSION AS OF 10".to_string(), + format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"), + ] { + for suffix in ["", " WHERE FALSE", " LIMIT 0"] { + let err = env + .ctx + .ctx() + .sql(&format!( + "SELECT * FROM {CATALOG}.{DB}.it {selector}{suffix}" + )) + .await + .unwrap_err(); + assert!( + err.to_string().contains("current table is unavailable"), + "{err}" + ); + } } assert_eq!(resolver.calls.load(Ordering::Relaxed), 0); + assert!(resolver.timestamps.lock().unwrap().is_empty()); } async fn current_unavailable_env() -> (TestEnv, Arc) { @@ -701,7 +794,7 @@ async fn engine_errors_propagate_to_context_table() { #[tokio::test] async fn engine_snapshots_can_be_joined_with_each_other_and_paimon() { - let (env, _) = snapshot_env().await; + let (env, resolver) = snapshot_env().await; for native in [false, true] { let batches = routed_sql( &env, @@ -742,6 +835,50 @@ async fn engine_snapshots_can_be_joined_with_each_other_and_paimon() { .await .unwrap(); assert_eq!(string_column_values(&batches, "historical"), ["old-a"]); + + let batches = routed_sql( + &env, + native, + &format!( + "SELECT a.historical AS old_value, b.historical AS new_value \ + FROM {CATALOG}.{DB}.it TIMESTAMP AS OF '{SNAPSHOT_10_TIME}' AS a \ + JOIN {CATALOG}.{DB}.it VERSION AS OF 20 AS b ON a.id = b.id" + ), + ) + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!(string_column_values(&batches, "old_value"), ["old-a"]); + assert_eq!(string_column_values(&batches, "new_value"), ["new-a"]); + + // Paimon tables keep Paimon's own timestamp resolution. + resolver.timestamps.lock().unwrap().clear(); + let batches = routed_sql( + &env, + native, + &format!( + "SELECT p.id, h.historical \ + FROM {CATALOG}.{DB}.it TIMESTAMP AS OF '{SNAPSHOT_10_TIME}' AS h \ + JOIN {CATALOG}.{DB}.pt TIMESTAMP AS OF '2999-01-01 00:00:00' AS p ON h.id = p.id \ + ORDER BY h.id" + ), + ) + .await + .unwrap() + .collect() + .await + .unwrap(); + assert_eq!( + string_column_values(&batches, "historical"), + ["old-a", "old-b"] + ); + assert_eq!(column_i32(&batches), [1, 2]); + assert_eq!( + *resolver.timestamps.lock().unwrap(), + [local_millis(SNAPSHOT_10_TIME)] + ); } } @@ -785,21 +922,42 @@ async fn engine_snapshot_supports_count_and_empty_results() { async fn engine_snapshot_errors_never_fall_back_to_current_data() { let (env, _) = snapshot_env().await; for native in [false, true] { - for (table, version, message) in [ - ("it", 404, "snapshot 404 does not exist or has expired"), - ("it", 500, "snapshot backend failed"), - ("ghost", 10, "not found"), + for (table, selector, message) in [ + ( + "it", + "VERSION AS OF 404".to_string(), + "snapshot 404 does not exist or has expired", + ), + ( + "it", + "VERSION AS OF 500".to_string(), + "snapshot backend failed", + ), + ("ghost", "VERSION AS OF 10".to_string(), "not found"), + ( + "it", + "TIMESTAMP AS OF '2026-09-30 23:59:59'".to_string(), + "no snapshot exists at or before", + ), + ( + "it", + format!("TIMESTAMP AS OF '{BACKEND_FAILURE_TIME}'"), + "snapshot backend failed", + ), + ( + "ghost", + format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"), + "not found", + ), ] { let err = routed_sql( &env, native, - &format!( - "SELECT * FROM {CATALOG}.{DB}.{table} VERSION AS OF {version} WHERE FALSE" - ), + &format!("SELECT * FROM {CATALOG}.{DB}.{table} {selector} WHERE FALSE"), ) .await .unwrap_err(); - assert!(err.to_string().contains(message), "{err}"); + assert!(err.to_string().contains(message), "{selector}: {err}"); } } } @@ -813,33 +971,38 @@ async fn engine_snapshot_providers_remain_read_only() { let (env, _) = snapshot_env().await; for native in [false, true] { - let df = routed_sql( - &env, - native, - &format!("SELECT * FROM {CATALOG}.{DB}.it VERSION AS OF 10"), - ) - .await - .unwrap(); - let mut provider = None; - df.logical_plan() - .apply(|node| { - if let LogicalPlan::TableScan(scan) = node { - provider = Some(source_as_provider(&scan.source)?); - return Ok(TreeNodeRecursion::Stop); - } - Ok(TreeNodeRecursion::Continue) - }) - .unwrap(); - let provider = provider.unwrap(); - let err = provider - .insert_into( - &env.ctx.ctx().state(), - Arc::new(EmptyExec::new(provider.schema())), - InsertOp::Append, + for selector in [ + "VERSION AS OF 10".to_string(), + format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"), + ] { + let df = routed_sql( + &env, + native, + &format!("SELECT * FROM {CATALOG}.{DB}.it {selector}"), ) .await - .unwrap_err(); - assert!(err.to_string().contains("write is not supported"), "{err}"); + .unwrap(); + let mut provider = None; + df.logical_plan() + .apply(|node| { + if let LogicalPlan::TableScan(scan) = node { + provider = Some(source_as_provider(&scan.source)?); + return Ok(TreeNodeRecursion::Stop); + } + Ok(TreeNodeRecursion::Continue) + }) + .unwrap(); + let provider = provider.unwrap(); + let err = provider + .insert_into( + &env.ctx.ctx().state(), + Arc::new(EmptyExec::new(provider.schema())), + InsertOp::Append, + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("write is not supported"), "{err}"); + } } } @@ -856,15 +1019,21 @@ async fn engine_snapshot_resolution_checks_session_options() { .await .unwrap(); for native in [false, true] { - let err = routed_sql( - &env, - native, - &format!("SELECT * FROM {CATALOG}.{DB}.it VERSION AS OF 10"), - ) - .await - .unwrap_err(); - assert!(err.to_string().contains(message), "{err}"); - assert_eq!(resolver.calls.load(Ordering::Relaxed), 0); + for selector in [ + "VERSION AS OF 10".to_string(), + format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"), + ] { + let err = routed_sql( + &env, + native, + &format!("SELECT * FROM {CATALOG}.{DB}.it {selector}"), + ) + .await + .unwrap_err(); + assert!(err.to_string().contains(message), "{selector}: {err}"); + assert_eq!(resolver.calls.load(Ordering::Relaxed), 0); + assert!(resolver.timestamps.lock().unwrap().is_empty()); + } } env.ctx.sql(&format!("RESET 'paimon.{key}'")).await.unwrap(); } @@ -882,6 +1051,8 @@ async fn engine_snapshot_rejects_unsupported_selectors_and_metadata_tables() { } else { ("VERSION AS OF 'tag-name'", "integer snapshot id") }; + let timestamp = format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"); + let timestamp = timestamp.as_str(); for (table, selector, message) in [ ("it", selector, message), ( @@ -891,19 +1062,21 @@ async fn engine_snapshot_rejects_unsupported_selectors_and_metadata_tables() { ), ( "it", - "TIMESTAMP AS OF '2020-01-01 00:00:00'", - "time travel is not supported", + "TIMESTAMP AS OF '2026-10-01'", + "Cannot parse time travel timestamp", ), ( "it$history", "VERSION AS OF 10", "time travel is not supported", ), + ("it$history", timestamp, "time travel is not supported"), ( "it$branch_b1", "VERSION AS OF 10", "branches are not supported", ), + ("it$branch_b1", timestamp, "branches are not supported"), ] { let err = routed_sql( &env, @@ -914,6 +1087,7 @@ async fn engine_snapshot_rejects_unsupported_selectors_and_metadata_tables() { .unwrap_err(); assert!(err.to_string().contains(message), "{selector}: {err}"); assert_eq!(resolver.calls.load(Ordering::Relaxed), 0); + assert!(resolver.timestamps.lock().unwrap().is_empty()); } } } @@ -926,19 +1100,24 @@ async fn engine_snapshot_requires_explicit_resolver_support() { let env = setup_with_engine(resolver).await; enable_version_syntax(&env).await; for native in [false, true] { - let err = routed_sql( - &env, - native, - &format!("SELECT * FROM {CATALOG}.{DB}.it VERSION AS OF 10 WHERE FALSE"), - ) - .await - .unwrap_err(); - let expected = if registered { - "time travel is not supported" - } else { - "no table engine is registered" - }; - assert!(err.to_string().contains(expected), "{err}"); + for selector in [ + "VERSION AS OF 10".to_string(), + format!("TIMESTAMP AS OF '{SNAPSHOT_10_TIME}'"), + ] { + let err = routed_sql( + &env, + native, + &format!("SELECT * FROM {CATALOG}.{DB}.it {selector} WHERE FALSE"), + ) + .await + .unwrap_err(); + let expected = if registered { + "time travel is not supported" + } else { + "no table engine is registered" + }; + assert!(err.to_string().contains(expected), "{selector}: {err}"); + } } } }