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 7911ecc fix(table): use the snapshot's schema version for time travel
reads (#379)
7911ecc is described below
commit 7911ecca89192b36748602a02bf854ea3d2ba4b2
Author: chaoyang <[email protected]>
AuthorDate: Sat Jun 13 11:06:52 2026 +0800
fix(table): use the snapshot's schema version for time travel reads (#379)
Time travel previously only switched which snapshot was scanned while the
table, scan pruning, read evolution target, and the DataFusion provider
all kept using the latest schema, so historical reads lost the historical
shape (columns dropped later were invisible, type updates were applied
retroactively).
Mirror Java AbstractFileStoreTable.copy/tryTimeTravel: the new async
Table::copy_with_time_travel merges options and, when they select a
snapshot with a different schema id, replaces the table schema with that
snapshot's schema (options stay the merged ones, matching Java
TableSchema.copy semantics via the new copy_with_replaced_options).
Resolution failures fall back silently like Java; invalid selectors still
fail at scan planning.
Snapshot resolution is extracted from TableScan::resolve_snapshot into
table::time_travel::travel_to_snapshot (Java TimeTravelUtil) and shared.
DataFusion entry points are wired up: the SQLContext time-travel rewrite,
the catalog provider's dynamic options path (SET 'paimon.scan.*'), and
PaimonRelationPlanner (bridged through block_on_with_runtime since the
planner hook is synchronous).
Writing through a time-travelled table copy is rejected explicitly; Java
avoids this structurally by using copyWithoutTimeTravel on write paths,
which the shared provider here cannot do.
---
crates/integrations/datafusion/src/catalog.rs | 8 +-
.../datafusion/src/relation_planner.rs | 10 +-
crates/integrations/datafusion/src/sql_context.rs | 39 +-
.../datafusion/tests/time_travel_schema_tests.rs | 294 ++++++++++++++
crates/paimon/src/spec/schema.rs | 41 ++
crates/paimon/src/table/mod.rs | 73 +++-
crates/paimon/src/table/table_scan.rs | 66 ++--
crates/paimon/src/table/time_travel.rs | 424 +++++++++++++++++++++
crates/paimon/src/table/write_builder.rs | 17 +
9 files changed, 932 insertions(+), 40 deletions(-)
diff --git a/crates/integrations/datafusion/src/catalog.rs
b/crates/integrations/datafusion/src/catalog.rs
index 018ef6b..0ab84f6 100644
--- a/crates/integrations/datafusion/src/catalog.rs
+++ b/crates/integrations/datafusion/src/catalog.rs
@@ -395,7 +395,13 @@ impl SchemaProvider for PaimonSchemaProvider {
let table = if opts.is_empty() {
table
} else {
- table.copy_with_options(opts)
+ // Dynamic options may select a historical snapshot
+ // (e.g. `SET 'paimon.scan.version'`); switch to its
+ // schema so planning sees the snapshot's columns.
+ table
+ .copy_with_time_travel(opts)
+ .await
+ .map_err(to_datafusion_error)?
};
let provider =
PaimonTableProvider::try_new_with_blob_reader_registry(
table,
diff --git a/crates/integrations/datafusion/src/relation_planner.rs
b/crates/integrations/datafusion/src/relation_planner.rs
index 80b4665..fa6e0ad 100644
--- a/crates/integrations/datafusion/src/relation_planner.rs
+++ b/crates/integrations/datafusion/src/relation_planner.rs
@@ -88,7 +88,15 @@ impl RelationPlanner for PaimonRelationPlanner {
return Ok(RelationPlanning::Original(Box::new(relation)));
};
- let new_table =
paimon_provider.table().copy_with_options(extra_options);
+ // Resolving time travel may switch the table to the snapshot's schema,
+ // which requires async IO; this planner hook is synchronous, so bridge
+ // through the shared runtime like other sync DataFusion callbacks.
+ let table = paimon_provider.table().clone();
+ let new_table = crate::runtime::block_on_with_runtime(
+ async move { table.copy_with_time_travel(extra_options).await },
+ "paimon time travel resolution thread panicked",
+ )
+ .map_err(crate::to_datafusion_error)?;
let new_provider = PaimonTableProvider::try_new(new_table)?;
let new_source = provider_as_source(Arc::new(new_provider));
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 62e2b7b..feec128 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -509,7 +509,10 @@ impl SQLContext {
let mut options = self.dynamic_options.read().unwrap().clone();
options.insert(SCAN_VERSION_OPTION.to_string(),
info.version.clone());
- let table_with_options = paimon_table.copy_with_options(options);
+ let table_with_options = paimon_table
+ .copy_with_time_travel(options)
+ .await
+ .map_err(|e| DataFusionError::External(Box::new(e)))?;
let provider =
Arc::new(PaimonTableProvider::try_new_with_blob_reader_registry(
table_with_options,
self.blob_reader_registry.clone(),
@@ -538,7 +541,10 @@ impl SQLContext {
let mut options = self.dynamic_options.read().unwrap().clone();
options.insert(SCAN_TIMESTAMP_MILLIS_OPTION.to_string(),
millis.to_string());
- let table_with_options = paimon_table.copy_with_options(options);
+ let table_with_options = paimon_table
+ .copy_with_time_travel(options)
+ .await
+ .map_err(|e| DataFusionError::External(Box::new(e)))?;
let provider =
Arc::new(PaimonTableProvider::try_new_with_blob_reader_registry(
table_with_options,
self.blob_reader_registry.clone(),
@@ -944,7 +950,32 @@ impl SQLContext {
ok_result(&self.ctx)
}
+ /// Reject write statements while a session-level time-travel selector is
+ /// active.
+ ///
+ /// Writes always operate on the latest table state, but in the same
+ /// session reads resolve through the time-travelled snapshot schema (and
+ /// INSERT through the provider is rejected by the write builder), so
+ /// silently ignoring the selector here would be inconsistent. Failing
+ /// with a clear message is safer than writing against a different schema
+ /// than concurrent reads observe.
+ fn ensure_no_time_travel_for_write(&self, operation: &str) -> DFResult<()>
{
+ use paimon::spec::{SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION};
+
+ let options = self.dynamic_options.read().unwrap();
+ for key in [SCAN_VERSION_OPTION, SCAN_TIMESTAMP_MILLIS_OPTION] {
+ if options.contains_key(key) {
+ return Err(DataFusionError::Plan(format!(
+ "Cannot execute {operation} while time-travel option
'{key}' is set; \
+ RESET 'paimon.{key}' first"
+ )));
+ }
+ }
+ Ok(())
+ }
+
async fn handle_merge_into(&self, merge: &Merge) -> DFResult<DataFrame> {
+ self.ensure_no_time_travel_for_write("MERGE INTO")?;
let table_name = match &merge.table {
TableFactor::Table { name, .. } => name.clone(),
other => {
@@ -964,6 +995,7 @@ impl SQLContext {
}
async fn handle_update(&self, update: &Update) -> DFResult<DataFrame> {
+ self.ensure_no_time_travel_for_write("UPDATE")?;
let table_name = match &update.table.relation {
TableFactor::Table { name, .. } => name.clone(),
other => {
@@ -983,6 +1015,7 @@ impl SQLContext {
}
async fn handle_delete(&self, delete: &Delete) -> DFResult<DataFrame> {
+ self.ensure_no_time_travel_for_write("DELETE")?;
let tables = match &delete.from {
FromTable::WithFromKeyword(t) | FromTable::WithoutKeyword(t) => t,
};
@@ -1010,6 +1043,7 @@ impl SQLContext {
}
async fn handle_insert_overwrite_partition(&self, insert: &Insert) ->
DFResult<DataFrame> {
+ self.ensure_no_time_travel_for_write("INSERT OVERWRITE")?;
let table_name = match &insert.table {
TableObject::TableName(name) => name.clone(),
other => {
@@ -1141,6 +1175,7 @@ impl SQLContext {
}
async fn handle_truncate_table(&self, truncate: &Truncate) ->
DFResult<DataFrame> {
+ self.ensure_no_time_travel_for_write("TRUNCATE TABLE")?;
if truncate.table_names.len() > 1 {
return Err(DataFusionError::Plan(
"TRUNCATE TABLE does not support multiple tables".to_string(),
diff --git a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
new file mode 100644
index 0000000..71e50c9
--- /dev/null
+++ b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
@@ -0,0 +1,294 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Time travel must read old snapshots with the snapshot's schema version,
+//! not the latest one.
+
+use std::sync::Arc;
+
+use paimon::{CatalogOptions, FileSystemCatalog, Options};
+use paimon_datafusion::SQLContext;
+use tempfile::TempDir;
+
+async fn create_sql_context(catalog: Arc<FileSystemCatalog>) -> SQLContext {
+ let mut ctx = SQLContext::new();
+ ctx.register_catalog("paimon", catalog).await.unwrap();
+ ctx
+}
+
+/// Build a table with two schema versions and one snapshot per version:
+/// snapshot 1 (schema 0: id, name) and snapshot 2 (schema 1: + age).
+///
+/// The second schema version is written directly as a `schema-1` file (column
+/// DDL beyond options is not needed for this test and keeps it independent of
+/// ALTER TABLE support).
+async fn setup_evolved_table() -> (TempDir, SQLContext) {
+ let temp_dir = TempDir::new().expect("Failed to create temp dir");
+ let warehouse = format!("file://{}", temp_dir.path().display());
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse);
+ let catalog = Arc::new(FileSystemCatalog::new(options).unwrap());
+ let sql_context = create_sql_context(catalog).await;
+
+ sql_context
+ .sql("CREATE TABLE paimon.default.t (id INT, name STRING)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ sql_context
+ .sql("INSERT INTO paimon.default.t VALUES (1, 'a'), (2, 'b'), (3,
'c')")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ // Evolve the schema: append an `age INT` column as schema-1.
+ let schema_dir =
temp_dir.path().join("default.db").join("t").join("schema");
+ let schema0: serde_json::Value =
+
serde_json::from_str(&std::fs::read_to_string(schema_dir.join("schema-0")).unwrap())
+ .unwrap();
+ let mut schema1 = schema0.clone();
+ schema1["id"] = serde_json::json!(1);
+ schema1["fields"]
+ .as_array_mut()
+ .unwrap()
+ .push(serde_json::json!({"id": 2, "name": "age", "type": "INT"}));
+ schema1["highestFieldId"] = serde_json::json!(2);
+ std::fs::write(
+ schema_dir.join("schema-1"),
+ serde_json::to_string(&schema1).unwrap(),
+ )
+ .unwrap();
+
+ sql_context
+ .sql("INSERT INTO paimon.default.t VALUES (4, 'd', 14), (5, 'e', 15)")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ (temp_dir, sql_context)
+}
+
+fn column_names(batches: &[datafusion::arrow::record_batch::RecordBatch]) ->
Vec<String> {
+ batches[0]
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().to_string())
+ .collect()
+}
+
+fn total_rows(batches: &[datafusion::arrow::record_batch::RecordBatch]) ->
usize {
+ batches.iter().map(|b| b.num_rows()).sum()
+}
+
+#[tokio::test]
+async fn test_version_as_of_uses_snapshot_schema() {
+ let (_tmp, sql_context) = setup_evolved_table().await;
+
+ // Old snapshot: only the old columns, even with SELECT *.
+ let batches = sql_context
+ .sql("SELECT * FROM paimon.default.t VERSION AS OF 1")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&batches), vec!["id", "name"]);
+ assert_eq!(total_rows(&batches), 3);
+
+ // A column added later does not exist at snapshot 1.
+ let err = sql_context
+ .sql("SELECT age FROM paimon.default.t VERSION AS OF 1")
+ .await
+ .expect_err("selecting a column added after the snapshot should fail
at planning");
+ assert!(
+ err.to_string().contains("age"),
+ "error should mention the missing column: {err}"
+ );
+
+ // Latest read still sees the evolved schema and all rows.
+ let batches = sql_context
+ .sql("SELECT * FROM paimon.default.t")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&batches), vec!["id", "name", "age"]);
+ assert_eq!(total_rows(&batches), 5);
+}
+
+#[tokio::test]
+async fn test_session_scan_version_uses_snapshot_schema() {
+ let (_tmp, sql_context) = setup_evolved_table().await;
+
+ sql_context
+ .sql("SET 'paimon.scan.version' = '1'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ let batches = sql_context
+ .sql("SELECT * FROM paimon.default.t")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&batches), vec!["id", "name"]);
+ assert_eq!(total_rows(&batches), 3);
+
+ // Writing through a time-travelled table is rejected.
+ let result = sql_context
+ .sql("INSERT INTO paimon.default.t VALUES (6, 'f')")
+ .await;
+ let err = match result {
+ Err(e) => e.to_string(),
+ Ok(df) => match df.collect().await {
+ Err(e) => e.to_string(),
+ Ok(_) => panic!("INSERT into a time-travelled table should fail"),
+ },
+ };
+ assert!(
+ err.contains("time-travel option"),
+ "error should mention time travel: {err}"
+ );
+
+ // Other write statements are rejected up front instead of silently
+ // ignoring the active selector and writing to the latest state.
+ for sql in [
+ "UPDATE paimon.default.t SET name = 'x' WHERE id = 1",
+ "DELETE FROM paimon.default.t WHERE id = 1",
+ "TRUNCATE TABLE paimon.default.t",
+ ] {
+ let err = match sql_context.sql(sql).await {
+ Err(e) => e.to_string(),
+ Ok(df) => match df.collect().await {
+ Err(e) => e.to_string(),
+ Ok(_) => panic!("{sql} should fail while scan.version is set"),
+ },
+ };
+ assert!(
+ err.contains("time-travel option"),
+ "{sql} should mention the active time-travel option: {err}"
+ );
+ }
+
+ // A selector resolving to a snapshot with the current schema id pins the
+ // read just the same, so INSERT stays rejected.
+ sql_context
+ .sql("SET 'paimon.scan.version' = '2'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let result = sql_context
+ .sql("INSERT INTO paimon.default.t VALUES (6, 'f', 16)")
+ .await;
+ let err = match result {
+ Err(e) => e.to_string(),
+ Ok(df) => match df.collect().await {
+ Err(e) => e.to_string(),
+ Ok(_) => panic!("INSERT should fail while scan.version pins a
same-schema snapshot"),
+ },
+ };
+ assert!(
+ err.contains("time-travel option"),
+ "error should mention time travel: {err}"
+ );
+
+ sql_context
+ .sql("RESET 'paimon.scan.version'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let batches = sql_context
+ .sql("SELECT * FROM paimon.default.t")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&batches), vec!["id", "name", "age"]);
+}
+
+#[tokio::test]
+async fn test_relation_planner_version_as_of_uses_snapshot_schema() {
+ let (_tmp, sql_context) = setup_evolved_table().await;
+
+ // Going through the raw SessionContext exercises PaimonRelationPlanner's
+ // synchronous hook (and its runtime bridge) instead of the SQLContext
+ // rewrite path. `VERSION AS OF` needs a dialect with table versioning.
+ sql_context
+ .ctx()
+ .sql("SET datafusion.sql_parser.dialect = 'databricks'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ let batches = sql_context
+ .ctx()
+ .sql("SELECT * FROM paimon.\"default\".t VERSION AS OF 1")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&batches), vec!["id", "name"]);
+ assert_eq!(total_rows(&batches), 3);
+}
+
+#[tokio::test]
+async fn test_timestamp_as_of_uses_snapshot_schema() {
+ let (tmp, sql_context) = setup_evolved_table().await;
+
+ // Both snapshots commit within milliseconds while `TIMESTAMP AS OF` has
+ // second precision; rewrite snapshot 1's commit time to a fixed old value
+ // so a timestamp between the two snapshots exists.
+ let snapshot1_path = tmp
+ .path()
+ .join("default.db")
+ .join("t")
+ .join("snapshot")
+ .join("snapshot-1");
+ let mut snapshot1: serde_json::Value =
+
serde_json::from_str(&std::fs::read_to_string(&snapshot1_path).unwrap()).unwrap();
+ snapshot1["timeMillis"] = serde_json::json!(86_400_000u64); // 1970-01-02
+ std::fs::write(&snapshot1_path,
serde_json::to_string(&snapshot1).unwrap()).unwrap();
+
+ let batches = sql_context
+ .sql("SELECT * FROM paimon.default.t TIMESTAMP AS OF '1970-01-03
00:00:00'")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+ assert_eq!(column_names(&batches), vec!["id", "name"]);
+ assert_eq!(total_rows(&batches), 3);
+}
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 76d09b4..9d8a95e 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -126,6 +126,18 @@ impl TableSchema {
new_schema
}
+ /// Create a copy of this schema with the options replaced entirely,
+ /// keeping id, fields, keys, comment, and timestamps.
+ ///
+ /// Corresponds to Java `TableSchema.copy(Map<String, String> newOptions)`,
+ /// which constructs a new schema with the given options rather than
+ /// merging them.
+ pub fn copy_with_replaced_options(&self, options: HashMap<String, String>)
-> Self {
+ let mut new_schema = self.clone();
+ new_schema.options = options;
+ new_schema
+ }
+
/// Apply a list of schema changes and return a new schema with
incremented ID.
pub fn apply_changes(&self, changes: Vec<crate::spec::SchemaChange>) ->
crate::Result<Self> {
let mut new_schema = self.clone();
@@ -839,6 +851,35 @@ mod tests {
);
}
+ #[test]
+ fn test_copy_with_replaced_options() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("old-key", "old-value")
+ .comment(Some("c".into()))
+ .build()
+ .unwrap();
+ let table_schema = TableSchema::new(3, &schema);
+
+ let mut new_options = HashMap::new();
+ new_options.insert("new-key".to_string(), "new-value".to_string());
+ let copied = table_schema.copy_with_replaced_options(new_options);
+
+ // Options are replaced entirely, not merged.
+ assert_eq!(copied.options().get("old-key"), None);
+ assert_eq!(
+ copied.options().get("new-key"),
+ Some(&"new-value".to_string())
+ );
+ // Everything else is preserved.
+ assert_eq!(copied.id(), table_schema.id());
+ assert_eq!(copied.fields(), table_schema.fields());
+ assert_eq!(copied.primary_keys(), table_schema.primary_keys());
+ assert_eq!(copied.comment(), table_schema.comment());
+ assert_eq!(copied.time_millis(), table_schema.time_millis());
+ }
+
#[test]
fn test_schema_validation() {
// Duplicate field names
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 3872c6a..005af90 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -57,6 +57,7 @@ mod table_scan;
mod table_update;
pub(crate) mod table_write;
mod tag_manager;
+pub(crate) mod time_travel;
mod vector_search_builder;
mod write_builder;
@@ -88,7 +89,7 @@ pub use write_builder::WriteBuilder;
use crate::catalog::Identifier;
use crate::io::FileIO;
-use crate::spec::{DataField, TableSchema};
+use crate::spec::{DataField, Snapshot, TableSchema};
use std::collections::HashMap;
/// Table represents a table in the catalog.
@@ -100,6 +101,13 @@ pub struct Table {
schema: TableSchema,
schema_manager: SchemaManager,
rest_env: Option<RESTEnv>,
+ /// True when this table copy was switched to a historical schema by
+ /// [`Table::copy_with_time_travel`]. Such a copy is read-only.
+ time_traveled: bool,
+ /// Snapshot resolved by [`Table::copy_with_time_travel`] from this copy's
+ /// options, so scans don't have to resolve the same selector again.
+ /// Cleared when [`Table::copy_with_options`] changes the selector.
+ travel_snapshot: Option<Snapshot>,
}
impl Table {
@@ -119,6 +127,8 @@ impl Table {
schema,
schema_manager,
rest_env,
+ time_traveled: false,
+ travel_snapshot: None,
}
}
@@ -179,7 +189,19 @@ impl Table {
}
/// Create a copy of this table with extra options merged into the schema.
+ ///
+ /// This never switches the schema version; it corresponds to Java
+ /// `FileStoreTable.copyWithoutTimeTravel`. Use
+ /// [`Table::copy_with_time_travel`] when the options may select a
+ /// historical snapshot whose schema should be used for reading.
pub fn copy_with_options(&self, extra: HashMap<String, String>) -> Self {
+ // Changing the time-travel selector invalidates the resolved snapshot
+ // (a time-travelled schema then has no matching snapshot anymore, and
+ // scans of such a copy fail until `copy_with_time_travel` re-resolves
+ // it). Unrelated options keep the snapshot/schema pair intact.
+ let selector_changed = extra.keys().any(|k| {
+ k == crate::spec::SCAN_VERSION_OPTION || k ==
crate::spec::SCAN_TIMESTAMP_MILLIS_OPTION
+ });
Self {
file_io: self.file_io.clone(),
identifier: self.identifier.clone(),
@@ -187,7 +209,56 @@ impl Table {
schema: self.schema.copy_with_options(extra),
schema_manager: self.schema_manager.clone(),
rest_env: self.rest_env.clone(),
+ time_traveled: self.time_traveled,
+ travel_snapshot: if selector_changed {
+ None
+ } else {
+ self.travel_snapshot.clone()
+ },
+ }
+ }
+
+ /// Create a copy of this table with extra options merged in, switching to
+ /// the schema of the time-travelled snapshot when the merged options
+ /// select one.
+ ///
+ /// Mirrors Java `AbstractFileStoreTable.copy(dynamicOptions)` →
+ /// `tryTimeTravel`: if the merged options contain a time-travel selector
+ /// (`scan.version` / `scan.timestamp-millis`) that resolves to a snapshot,
+ /// the table's fields and keys come from that snapshot's schema while the
+ /// options stay the merged ones (Java `TableSchema.copy(newOptions)`).
+ /// Like Java, resolution failures fall back silently to the current
+ /// schema (the `if let Ok` below swallows them); an invalid selector
+ /// still fails later at scan planning.
+ pub async fn copy_with_time_travel(&self, extra: HashMap<String, String>)
-> Result<Self> {
+ let mut table = self.copy_with_options(extra);
+ // travel_to_snapshot returns Ok(None) without IO when the merged
+ // options contain no selector.
+ if let Ok(Some(snapshot)) =
+ time_travel::travel_to_snapshot(&table.file_io, &table.location,
table.schema.options())
+ .await
+ {
+ if snapshot.schema_id() != table.schema.id() {
+ let snapshot_schema =
table.schema_manager.schema(snapshot.schema_id()).await?;
+ table.schema =
+
snapshot_schema.copy_with_replaced_options(table.schema.options().clone());
+ table.time_traveled = true;
+ }
+ table.travel_snapshot = Some(snapshot);
}
+ Ok(table)
+ }
+
+ /// Whether this table copy reads a historical snapshot with its
+ /// historical schema (see [`Table::copy_with_time_travel`]).
+ pub fn is_time_traveled(&self) -> bool {
+ self.time_traveled
+ }
+
+ /// The snapshot resolved by [`Table::copy_with_time_travel`] from this
+ /// copy's options, if any. Lets scans skip re-resolving the selector.
+ pub(crate) fn travel_snapshot(&self) -> Option<&Snapshot> {
+ self.travel_snapshot.as_ref()
}
}
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 3135e77..170405b 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -32,7 +32,6 @@ use crate::io::FileIO;
use crate::spec::{
avro::SharedSchemaCache, bucket_dir_name, BinaryRow, CoreOptions,
DataField, DataFileMeta,
FileKind, IndexManifest, ManifestEntry, PartitionComputer, Predicate,
Snapshot,
- TimeTravelSelector,
};
use crate::table::bin_pack::split_for_batch;
use crate::table::merge_tree_split_generator::{
@@ -43,8 +42,6 @@ use crate::table::source::{
DataSplitBuilder, DeletionFile, PartitionBucket, Plan, RowRange,
};
use crate::table::SnapshotManager;
-use crate::table::TagManager;
-use crate::Error;
use futures::{StreamExt, TryStreamExt};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
@@ -375,42 +372,41 @@ impl<'a> TableScan<'a> {
}
async fn resolve_snapshot(&self) -> crate::Result<Option<Snapshot>> {
+ // A table copy produced by `copy_with_time_travel` already resolved
+ // the selector in its options; reuse it instead of re-reading
+ // tag/snapshot files on every plan.
+ if let Some(snapshot) = self.table.travel_snapshot() {
+ return Ok(Some(snapshot.clone()));
+ }
+ // A time-travelled schema without its resolved snapshot means the
+ // selector was changed after the travel (`copy_with_options`).
+ // Resolving the new selector here would evolve a different snapshot's
+ // files to the stale historical schema, so fail instead.
+ if self.table.is_time_traveled() {
+ return Err(crate::Error::DataInvalid {
+ message: "Table options changed after time travel; \
+ use copy_with_time_travel to re-resolve the snapshot
and schema"
+ .to_string(),
+ source: None,
+ });
+ }
+
let file_io = self.table.file_io();
let table_path = self.table.location();
- let snapshot_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
- let core_options = CoreOptions::new(self.table.schema().options());
- match core_options.try_time_travel_selector()? {
- Some(TimeTravelSelector::TimestampMillis(ts)) => {
- match snapshot_manager.earlier_or_equal_time_millis(ts).await?
{
- Some(s) => Ok(Some(s)),
- None => Err(Error::DataInvalid {
- message: format!("No snapshot found with timestamp <=
{ts}"),
- source: None,
- }),
- }
- }
- Some(TimeTravelSelector::Version(v)) => {
- // Tag first, then snapshot id, else error.
- let tag_manager = TagManager::new(file_io.clone(),
table_path.to_string());
- if tag_manager.tag_exists(v).await? {
- match tag_manager.get(v).await? {
- Some(s) => Ok(Some(s)),
- None => Err(Error::DataInvalid {
- message: format!("Tag '{v}' doesn't exist."),
- source: None,
- }),
- }
- } else if let Ok(id) = v.parse::<i64>() {
- snapshot_manager.get_snapshot(id).await.map(Some)
- } else {
- Err(Error::DataInvalid {
- message: format!("Version '{v}' is not a valid tag
name or snapshot id."),
- source: None,
- })
- }
+ match super::time_travel::travel_to_snapshot(
+ file_io,
+ table_path,
+ self.table.schema().options(),
+ )
+ .await?
+ {
+ Some(snapshot) => Ok(Some(snapshot)),
+ None => {
+ let snapshot_manager =
+ SnapshotManager::new(file_io.clone(),
table_path.to_string());
+ snapshot_manager.get_latest_snapshot().await
}
- None => snapshot_manager.get_latest_snapshot().await,
}
}
diff --git a/crates/paimon/src/table/time_travel.rs
b/crates/paimon/src/table/time_travel.rs
new file mode 100644
index 0000000..7a5d07d
--- /dev/null
+++ b/crates/paimon/src/table/time_travel.rs
@@ -0,0 +1,424 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Snapshot resolution for time travel, mirroring Java `TimeTravelUtil`.
+
+use crate::io::FileIO;
+use crate::spec::{CoreOptions, Snapshot, TimeTravelSelector};
+use crate::table::SnapshotManager;
+use crate::table::TagManager;
+use crate::Error;
+use std::collections::HashMap;
+
+/// Resolve the snapshot selected by the time-travel options, if any.
+///
+/// Returns `Ok(None)` when no time-travel selector is configured. Returns an
+/// error for invalid or conflicting selectors, and when the selector does not
+/// match any snapshot — callers that need Java `tryTravelToSnapshot`'s silent
+/// fallback (keep the current schema on failure) handle the `Err` themselves.
+pub(crate) async fn travel_to_snapshot(
+ file_io: &FileIO,
+ table_path: &str,
+ options: &HashMap<String, String>,
+) -> crate::Result<Option<Snapshot>> {
+ let core_options = CoreOptions::new(options);
+ let snapshot_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
+
+ match core_options.try_time_travel_selector()? {
+ Some(TimeTravelSelector::TimestampMillis(ts)) => {
+ match snapshot_manager.earlier_or_equal_time_millis(ts).await? {
+ Some(s) => Ok(Some(s)),
+ None => Err(Error::DataInvalid {
+ message: format!("No snapshot found with timestamp <=
{ts}"),
+ source: None,
+ }),
+ }
+ }
+ Some(TimeTravelSelector::Version(v)) => {
+ // Tag first, then snapshot id, else error.
+ let tag_manager = TagManager::new(file_io.clone(),
table_path.to_string());
+ if tag_manager.tag_exists(v).await? {
+ match tag_manager.get(v).await? {
+ Some(s) => Ok(Some(s)),
+ None => Err(Error::DataInvalid {
+ message: format!("Tag '{v}' doesn't exist."),
+ source: None,
+ }),
+ }
+ } else if let Ok(id) = v.parse::<i64>() {
+ snapshot_manager.get_snapshot(id).await.map(Some)
+ } else {
+ Err(Error::DataInvalid {
+ message: format!("Version '{v}' is not a valid tag name or
snapshot id."),
+ source: None,
+ })
+ }
+ }
+ None => Ok(None),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::catalog::Identifier;
+ use crate::io::{FileIO, FileIOBuilder};
+ use crate::spec::{DataType, IntType, Schema, TableSchema};
+ use crate::table::{SnapshotManager, Table, TableCommit, TableWrite,
TagManager};
+ use arrow_array::{Int32Array, RecordBatch};
+ use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
+ use std::collections::HashMap;
+ use std::sync::Arc;
+
+ fn schema_v0() -> TableSchema {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .build()
+ .unwrap();
+ TableSchema::new(0, &schema)
+ }
+
+ fn schema_v1() -> TableSchema {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .column("age", DataType::Int(IntType::new()))
+ .build()
+ .unwrap();
+ TableSchema::new(1, &schema)
+ }
+
+ fn make_table(file_io: &FileIO, table_path: &str, schema: TableSchema) ->
Table {
+ Table::new(
+ file_io.clone(),
+ Identifier::new("default", "evolved"),
+ table_path.to_string(),
+ schema,
+ None,
+ )
+ }
+
+ async fn write_schema_file(file_io: &FileIO, table_path: &str, schema:
&TableSchema) {
+ file_io
+ .mkdirs(&format!("{table_path}/schema/"))
+ .await
+ .unwrap();
+ let path = format!("{table_path}/schema/schema-{}", schema.id());
+ let content = serde_json::to_string(schema).unwrap();
+ file_io
+ .new_output(&path)
+ .unwrap()
+ .write(content.into())
+ .await
+ .unwrap();
+ }
+
+ fn batch_v0(ids: Vec<i32>, values: Vec<i32>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap()
+ }
+
+ fn batch_v1(ids: Vec<i32>, values: Vec<i32>, ages: Vec<i32>) ->
RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ArrowField::new("age", ArrowDataType::Int32, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ Arc::new(Int32Array::from(ages)),
+ ],
+ )
+ .unwrap()
+ }
+
+ async fn write_and_commit(table: &Table, batch: &RecordBatch) {
+ let mut write = TableWrite::new(table,
"test-user".to_string()).unwrap();
+ write.write_arrow_batch(batch).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ let commit = TableCommit::new(table.clone(), "test-user".to_string());
+ commit.commit(messages).await.unwrap();
+ }
+
+ /// Table with two schema versions and one snapshot per version:
+ /// snapshot 1 (schema 0: id, value) and snapshot 2 (schema 1: + age).
+ async fn setup_evolved_table() -> (FileIO, String) {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table_path = "memory:/evolved_table";
+ for dir in ["snapshot", "manifest"] {
+ file_io
+ .mkdirs(&format!("{table_path}/{dir}/"))
+ .await
+ .unwrap();
+ }
+ write_schema_file(&file_io, table_path, &schema_v0()).await;
+ let table_v0 = make_table(&file_io, table_path, schema_v0());
+ write_and_commit(&table_v0, &batch_v0(vec![1, 2, 3], vec![10, 20,
30])).await;
+
+ write_schema_file(&file_io, table_path, &schema_v1()).await;
+ let table_v1 = make_table(&file_io, table_path, schema_v1());
+ write_and_commit(&table_v1, &batch_v1(vec![4, 5], vec![40, 50],
vec![14, 15])).await;
+
+ (file_io, table_path.to_string())
+ }
+
+ fn latest_table(file_io: &FileIO, table_path: &str) -> Table {
+ make_table(file_io, table_path, schema_v1())
+ }
+
+ fn options(pairs: &[(&str, &str)]) -> HashMap<String, String> {
+ pairs
+ .iter()
+ .map(|(k, v)| (k.to_string(), v.to_string()))
+ .collect()
+ }
+
+ #[tokio::test]
+ async fn test_copy_with_time_travel_switches_to_snapshot_schema() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "1")]))
+ .await
+ .unwrap();
+
+ assert_eq!(traveled.schema().id(), 0);
+ let names: Vec<&str> = traveled
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name())
+ .collect();
+ assert_eq!(names, vec!["id", "value"]);
+ assert!(traveled.is_time_traveled());
+ // Options stay the merged ones, not the historical schema's options.
+ assert_eq!(
+ traveled.schema().options().get("scan.version"),
+ Some(&"1".to_string())
+ );
+ // The resolved snapshot is cached for scans, and invalidated when the
+ // selector changes.
+ assert_eq!(traveled.travel_snapshot().map(|s| s.id()), Some(1));
+ let recopied = traveled.copy_with_options(options(&[("scan.version",
"2")]));
+ assert!(recopied.travel_snapshot().is_none());
+ }
+
+ #[tokio::test]
+ async fn test_copy_with_time_travel_same_schema_still_rejects_write() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ // Snapshot 2 carries the current schema, so the schema is not
+ // switched — but the copy still reads a pinned snapshot, so writing
+ // through it is rejected like any other time-travelled copy.
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "2")]))
+ .await
+ .unwrap();
+
+ assert_eq!(traveled.schema().id(), 1);
+ assert!(!traveled.is_time_traveled());
+ assert!(traveled.new_write_builder().new_write().is_err());
+ }
+
+ #[tokio::test]
+ async fn test_copy_with_time_travel_without_selector_is_noop() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let copied =
table.copy_with_time_travel(HashMap::new()).await.unwrap();
+
+ assert_eq!(copied.schema(), table.schema());
+ assert!(!copied.is_time_traveled());
+ }
+
+ #[tokio::test]
+ async fn test_copy_with_time_travel_invalid_selector_falls_back_silently()
{
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ // Not a tag and not a snapshot id: kept latest, error deferred to
scan.
+ let copied = table
+ .copy_with_time_travel(options(&[("scan.version",
"no-such-version")]))
+ .await
+ .unwrap();
+ assert_eq!(copied.schema().id(), 1);
+ assert!(!copied.is_time_traveled());
+
+ // Conflicting selectors behave the same.
+ let copied = table
+ .copy_with_time_travel(options(&[
+ ("scan.version", "1"),
+ ("scan.timestamp-millis", "123"),
+ ]))
+ .await
+ .unwrap();
+ assert_eq!(copied.schema().id(), 1);
+ assert!(!copied.is_time_traveled());
+ }
+
+ #[tokio::test]
+ async fn test_copy_with_time_travel_by_timestamp_and_tag() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let snapshot_manager = SnapshotManager::new(file_io.clone(),
table_path.clone());
+ let snapshot1 = snapshot_manager.get_snapshot(1).await.unwrap();
+
+ let traveled = table
+ .copy_with_time_travel(options(&[(
+ "scan.timestamp-millis",
+ &snapshot1.time_millis().to_string(),
+ )]))
+ .await
+ .unwrap();
+ assert_eq!(traveled.schema().id(), 0);
+
+ let tag_manager = TagManager::new(file_io.clone(), table_path.clone());
+ tag_manager.create("v1-tag", &snapshot1).await.unwrap();
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "v1-tag")]))
+ .await
+ .unwrap();
+ assert_eq!(traveled.schema().id(), 0);
+ assert!(traveled.is_time_traveled());
+ }
+
+ #[tokio::test]
+ async fn test_time_traveled_table_rejects_write() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "1")]))
+ .await
+ .unwrap();
+
+ let err = match traveled.new_write_builder().new_write() {
+ Err(e) => e,
+ Ok(_) => panic!("expected write rejection on time-travelled
table"),
+ };
+ assert!(
+ matches!(err, crate::Error::Unsupported { ref message }
+ if message.contains("time-travel option")),
+ "expected write rejection on time-travelled table, got {err:?}"
+ );
+ // The latest table is unaffected.
+ assert!(table.new_write_builder().new_write().is_ok());
+ }
+
+ #[tokio::test]
+ async fn test_changing_selector_after_travel_fails_scan() {
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "1")]))
+ .await
+ .unwrap();
+
+ // Merging unrelated options keeps the resolved snapshot/schema pair.
+ let recopied = traveled.copy_with_options(options(&[("k", "v")]));
+ assert_eq!(recopied.travel_snapshot().map(|s| s.id()), Some(1));
+
+ // Changing the selector without re-resolving leaves a historical
+ // schema with no matching snapshot; scanning such a copy must fail
+ // instead of evolving another snapshot's files to the stale schema.
+ let stale = traveled.copy_with_options(options(&[("scan.version",
"2")]));
+ assert!(stale.travel_snapshot().is_none());
+ let err = stale
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .expect_err("scan after selector change must fail");
+ assert!(
+ matches!(err, crate::Error::DataInvalid { ref message, .. }
+ if message.contains("copy_with_time_travel")),
+ "expected stale time-travel state error, got {err:?}"
+ );
+
+ // Re-resolving through copy_with_time_travel is the supported path.
+ let retraveled = traveled
+ .copy_with_time_travel(options(&[("scan.version", "2")]))
+ .await
+ .unwrap();
+ assert_eq!(retraveled.schema().id(), 1);
+ assert_eq!(retraveled.travel_snapshot().map(|s| s.id()), Some(2));
+ }
+
+ #[tokio::test]
+ async fn test_time_travel_read_uses_snapshot_schema() {
+ use futures::TryStreamExt;
+
+ let (file_io, table_path) = setup_evolved_table().await;
+ let table = latest_table(&file_io, &table_path);
+
+ let traveled = table
+ .copy_with_time_travel(options(&[("scan.version", "1")]))
+ .await
+ .unwrap();
+ let builder = traveled.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let names: Vec<String> = batches[0]
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().to_string())
+ .collect();
+ assert_eq!(names, vec!["id", "value"]);
+ let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(rows, 3);
+
+ // Latest table sees both snapshots' data and the evolved schema.
+ let builder = table.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(batches[0].schema().fields().len(), 3);
+ let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(rows, 5);
+ }
+}
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index e3a4dbc..34ae27d 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -80,6 +80,23 @@ impl<'a> WriteBuilder<'a> {
/// For primary-key tables, sequence numbers are lazily scanned per
partition
/// when the first writer for that partition is created.
pub fn new_write(&self) -> crate::Result<TableWrite> {
+ // A table with a time-travel selector reads a pinned snapshot (and may
+ // carry that snapshot's historical schema), so writing through the
+ // same copy would be inconsistent with what its reads observe — even
+ // when the pinned snapshot happens to share the current schema id.
+ // Java avoids this structurally (write paths use
copyWithoutTimeTravel);
+ // here the same table copy can serve both reads and writes, so reject
+ // explicitly. Conflicting selectors (`Err`) cannot be valid for writes
+ // either. Commit-only flows (new_commit) stay untouched.
+ let selector =
+
crate::spec::CoreOptions::new(self.table.schema().options()).try_time_travel_selector();
+ if !matches!(selector, Ok(None)) {
+ return Err(crate::Error::Unsupported {
+ message: "Cannot write to a table with a time-travel option
set \
+ (scan.version / scan.timestamp-millis)"
+ .to_string(),
+ });
+ }
let write = TableWrite::new(self.table, self.commit_user.clone())?;
Ok(if self.overwrite {
write.with_overwrite()
