This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 2104a029 feat(datafusion): support timestamp time travel for external
engines (#1121)
2104a029 is described below
commit 2104a02922dca25a7bdf5e1fb1a1f3d09eab62ee
Author: Jiajia Li <[email protected]>
AuthorDate: Wed Oct 14 08:32:45 2026 +0800
feat(datafusion): support timestamp time travel for external engines (#1121)
---
crates/integrations/datafusion/README.md | 7 +-
crates/integrations/datafusion/src/catalog.rs | 93 ++++--
.../datafusion/src/relation_planner.rs | 39 ++-
crates/integrations/datafusion/src/sql_context.rs | 47 ++-
.../datafusion/tests/table_type_routing.rs | 345 ++++++++++++++++-----
5 files changed, 391 insertions(+), 140 deletions(-)
diff --git a/crates/integrations/datafusion/README.md
b/crates/integrations/datafusion/README.md
index 1b6523e2..5c7904b9 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 <snapshot_id>` on external data tables (integer IDs only).
+`VERSION AS OF <snapshot_id>` on external data tables (integer IDs only), and
+`resolve_table_at_timestamp` to support `TIMESTAMP AS OF '<yyyy-MM-dd
HH:mm:ss>'`.
+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 4232ae2b..19770788 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<Option<Arc<dyn TableProvider>>> {
+ Err(plan_datafusion_err!(
+ "timestamp time travel is not supported for routed '{}' tables
('{}.{}')",
+ request.declared,
+ request.database,
+ request.table
+ ))
+ }
+
/// Resolves `<table>$<metadata_table>`; 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<Arc<dyn TableProvider>> {
+ pub(crate) async fn at_time_travel(
+ &self,
+ travel: EngineTimeTravel,
+ ) -> DFResult<Arc<dyn TableProvider>> {
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<i64> {
})
}
-async fn resolve_engine_snapshot(
+async fn resolve_engine_time_travel(
request: EngineTableRequest,
resolver: Arc<dyn TableEngineResolver>,
dynamic_options: DynamicOptions,
- snapshot_id: i64,
+ travel: EngineTimeTravel,
) -> DFResult<Arc<dyn TableProvider>> {
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<Arc<dyn TableProvider>> {
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<Arc<dyn SchemaProvider>> {
diff --git a/crates/integrations/datafusion/src/relation_planner.rs
b/crates/integrations/datafusion/src/relation_planner.rs
index 2078d48e..10b2679d 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::<ReadOnlyTableProvider>()
{
- 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<HashMap<String, String>>
/// Resolve `TIMESTAMP AS OF <expr>` into `scan.timestamp-millis` option.
fn resolve_timestamp_as_of(expr: &ast::Expr) -> DFResult<HashMap<String,
String>> {
+ 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 <expr>` to epoch millis.
+fn timestamp_as_of_millis(expr: &ast::Expr) -> DFResult<i64> {
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 140a424f..f4c02dc1 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<DataFrame> {
- 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<dyn Catalog>,
catalog_name: &str,
identifier: &Identifier,
- option: &str,
- value: String,
+ clause: TimeTravelClause,
) -> DFResult<Arc<dyn TableProvider>> {
+ 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 243140aa..9e291430 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<Vec<i64>>,
}
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<Option<Arc<dyn TableProvider>>> {
+ 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<SnapshotEngineResolver>) {
@@ -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}");
+ }
}
}
}
