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 f73ebe4e feat: Support Java-compatible scan.timestamp time travel
(#919)
f73ebe4e is described below
commit f73ebe4e36a99ff8bc4924b2413ee30b8eb23263
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Sep 23 09:08:15 2026 +0800
feat: Support Java-compatible scan.timestamp time travel (#919)
---
Cargo.lock | 1 +
bindings/c/src/table.rs | 35 ++--
bindings/python/README.md | 3 +-
bindings/python/src/read.rs | 3 +-
bindings/python/tests/test_read.py | 24 +++
crates/integrations/datafusion/src/sql_context.rs | 3 +-
.../datafusion/tests/time_travel_schema_tests.rs | 103 ++++++++++
crates/paimon/Cargo.toml | 1 +
crates/paimon/src/spec/core_options.rs | 223 ++++++++++++++++++++-
crates/paimon/src/table/mod.rs | 7 +-
crates/paimon/src/table/table_scan.rs | 3 +-
crates/paimon/src/table/time_travel.rs | 95 +++++++++
crates/paimon/src/table/write_builder.rs | 2 +-
docs/src/python-binding.md | 9 +
docs/src/sql.md | 13 ++
15 files changed, 500 insertions(+), 25 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 3bcd32ed..c46567fb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4670,6 +4670,7 @@ dependencies = [
"hex",
"hmac 0.12.1",
"indexmap 2.14.0",
+ "jiff",
"libloading 0.9.0",
"log",
"lru 0.18.2",
diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs
index 058c95d8..598cc63d 100644
--- a/bindings/c/src/table.rs
+++ b/bindings/c/src/table.rs
@@ -373,8 +373,9 @@ pub unsafe extern "C" fn paimon_table_latest_snapshot(
}
/// Time-travel selector option names, in the core's resolution priority order.
-const TIME_TRAVEL_SELECTORS: [&str; 5] = [
+const TIME_TRAVEL_SELECTORS: [&str; 6] = [
"scan.timestamp-millis",
+ "scan.timestamp",
"scan.watermark",
"scan.version",
"scan.snapshot-id",
@@ -446,7 +447,7 @@ pub unsafe extern "C" fn paimon_table_new_read_builder(
}
/// Create a ReadBuilder from a Table with scan options (e.g. time-travel
-/// selectors `scan.snapshot-id` / `scan.tag-name` / `scan.timestamp-millis` /
+/// selectors `scan.snapshot-id` / `scan.tag-name` / `scan.timestamp-millis` /
`scan.timestamp` /
/// `scan.watermark` / `scan.version`). At most one time-travel selector may be
/// set. A selector that does not resolve to a snapshot is an error (never a
/// silent read-of-latest).
@@ -2520,20 +2521,22 @@ mod tests {
fn malformed_selector_value_does_not_silently_read_latest() {
unsafe {
let table = boxed_test_table();
- let k = CString::new("scan.snapshot-id").unwrap();
- let v = CString::new("abc").unwrap();
- let opts = [opt(&k, &v)];
- // Core swallows the parse error and falls back; the binding
reports
- // the unified "did not resolve" error rather than building a
- // latest-reading builder.
- let (code, message) = assert_rb_err_code_message(
- paimon_table_new_read_builder_with_options(table,
opts.as_ptr(), 1),
- );
- assert_eq!(code, PaimonErrorCode::InvalidInput as i32);
- assert!(
- message.contains("did not resolve"),
- "message should report the selector did not resolve, got:
{message}"
- );
+ for selector in ["scan.snapshot-id", "scan.timestamp"] {
+ let k = CString::new(selector).unwrap();
+ let v = CString::new("abc").unwrap();
+ let opts = [opt(&k, &v)];
+ // Core swallows the parse error and falls back; the binding
reports
+ // the unified "did not resolve" error rather than building a
+ // latest-reading builder.
+ let (code, message) = assert_rb_err_code_message(
+ paimon_table_new_read_builder_with_options(table,
opts.as_ptr(), 1),
+ );
+ assert_eq!(code, PaimonErrorCode::InvalidInput as i32);
+ assert!(
+ message.contains("did not resolve"),
+ "message should report the selector did not resolve, got:
{message}"
+ );
+ }
paimon_table_free(table);
}
}
diff --git a/bindings/python/README.md b/bindings/python/README.md
index c55c4010..b6505d1d 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -97,7 +97,8 @@ commit_messages = writer.prepare_commit()
write_builder.new_commit().commit(commit_messages)
# --- Time travel: read a past version ---
-# Supported options: scan.version, scan.timestamp-millis, scan.snapshot-id, or
scan.tag-name
+# Supported selectors: scan.version, scan.timestamp, scan.timestamp-millis,
+# scan.snapshot-id, scan.tag-name, or scan.watermark
read_builder_tt = table.new_read_builder({"scan.snapshot-id": "1"})
scan_tt = read_builder_tt.new_scan()
plan_tt = scan_tt.plan()
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index f3c00ca4..c9c93fcb 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -39,8 +39,9 @@ const MAP_SELECTED_KEYS_PREFIX: &str =
"__PAIMON_MAP_SELECTED_KEYS:";
const MAP_SELECTED_KEYS_DELIMITER: char = ';';
/// Time-travel selector option names, in the core's resolution priority order.
-const TIME_TRAVEL_SELECTORS: [&str; 5] = [
+const TIME_TRAVEL_SELECTORS: [&str; 6] = [
"scan.timestamp-millis",
+ "scan.timestamp",
"scan.watermark",
"scan.version",
"scan.snapshot-id",
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 0892c245..17721de7 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -967,6 +967,30 @@ def test_time_travel_by_snapshot_id():
assert _rows(batches) == 1 # only snapshot 1's row
+def test_time_travel_by_timestamp_string():
+ from pathlib import Path
+
+ with tempfile.TemporaryDirectory() as warehouse:
+ _make_two_snapshot_table(warehouse)
+ snapshot_path = Path(warehouse) / "tdb.db/t/snapshot/snapshot-1"
+ snapshot = json.loads(snapshot_path.read_text())
+ snapshot["timeMillis"] = 86_400_000 # 1970-01-02 UTC
+ snapshot_path.write_text(json.dumps(snapshot))
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+ for mode in [{}, {"scan.mode": "from-timestamp"}]:
+ builder = table.new_read_builder(
+ {"scan.timestamp": "1970-01-03 00:00:00.123456789", **mode})
+ splits = builder.new_scan().plan().splits()
+ result = pa.Table.from_batches(builder.new_read().read(splits))
+ assert result.column("id").to_pylist() == [1]
+ for value in ["invalid", "1970-01-01 00:00:00"]:
+ with pytest.raises(ValueError, match="did not resolve"):
+ table.new_read_builder({"scan.timestamp": value})
+ with pytest.raises(ValueError, match="Only one"):
+ table.new_read_builder({"scan.timestamp": "1970-01-03",
+ "scan.timestamp-millis": "172800000"})
+
+
def test_time_travel_by_tag_name():
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_two_snapshot_table(warehouse)
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index aba77b15..34c33035 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -1461,13 +1461,14 @@ impl SQLContext {
fn ensure_no_time_travel_for_write(&self, operation: &str) -> DFResult<()>
{
use paimon::spec::{
SCAN_SNAPSHOT_ID_OPTION, SCAN_TAG_NAME_OPTION,
SCAN_TIMESTAMP_MILLIS_OPTION,
- SCAN_VERSION_OPTION,
+ SCAN_TIMESTAMP_OPTION, SCAN_VERSION_OPTION,
};
let options = self.dynamic_options.read().unwrap();
for key in [
SCAN_VERSION_OPTION,
SCAN_TIMESTAMP_MILLIS_OPTION,
+ SCAN_TIMESTAMP_OPTION,
SCAN_SNAPSHOT_ID_OPTION,
SCAN_TAG_NAME_OPTION,
] {
diff --git a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
index 747cd579..e8598fce 100644
--- a/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
+++ b/crates/integrations/datafusion/tests/time_travel_schema_tests.rs
@@ -397,4 +397,107 @@ async fn test_timestamp_as_of_uses_snapshot_schema() {
.unwrap();
assert_eq!(column_names(&batches), vec!["id", "name"]);
assert_eq!(total_rows(&batches), 3);
+
+ sql_context
+ .sql("SET 'paimon.scan.timestamp' = '1970-01-03 00:00:00.123'")
+ .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);
+ let error = match sql_context
+ .sql("DELETE FROM paimon.default.t WHERE id = 1")
+ .await
+ {
+ Err(error) => error,
+ Ok(_) => panic!("writes with scan.timestamp must fail"),
+ };
+ assert!(error.to_string().contains("scan.timestamp"), "{error}");
+ sql_context
+ .sql("RESET 'paimon.scan.timestamp'")
+ .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"]);
+ assert_eq!(total_rows(&batches), 5);
+}
+
+#[tokio::test]
+async fn test_session_scan_timestamp_overlap_reads_earlier_snapshot() {
+ const CHILD: &str = "PAIMON_SCAN_TIMESTAMP_SQL_OVERLAP_CHILD";
+ if std::env::var_os(CHILD).is_none() {
+ // The system local timezone is process-global. Isolate this case so
+ // parallel tests cannot change how the ambiguous time is resolved.
+ let output =
std::process::Command::new(std::env::current_exe().unwrap())
+ .args([
+ "--exact",
+ "test_session_scan_timestamp_overlap_reads_earlier_snapshot",
+ ])
+ .env("TZ", "America/New_York")
+ .env(CHILD, "1")
+ .output()
+ .unwrap();
+ assert!(
+ output.status.success(),
+ "child failed:\n{}{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ );
+ return;
+ }
+
+ let (tmp, sql_context) = setup_evolved_table().await;
+ for (snapshot_id, time_millis) in [
+ (1, 1_730_610_000_000_u64), // 2024-11-03T05:00:00Z
+ (2, 1_730_613_600_000_u64), // 2024-11-03T06:00:00Z
+ ] {
+ let path = tmp
+ .path()
+ .join("default.db/t/snapshot")
+ .join(format!("snapshot-{snapshot_id}"));
+ let mut snapshot: serde_json::Value =
+
serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
+ snapshot["timeMillis"] = serde_json::json!(time_millis);
+ std::fs::write(path, serde_json::to_vec(&snapshot).unwrap()).unwrap();
+ }
+
+ sql_context
+ .sql("SET 'paimon.scan.timestamp' = '2024-11-03 01:30:00'")
+ .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);
+
+ sql_context
+ .sql("SET 'paimon.scan.timestamp' = '2024-11-03 02:00:00'")
+ .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"]);
+ assert_eq!(total_rows(&batches), 5);
}
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index c970799d..e58c46b1 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -77,6 +77,7 @@ tokio = { version = "1.39.2", features = [
"time",
] }
chrono = { version = "0.4.38", features = ["serde"] }
+jiff = "0.2.34"
serde = { version = "1", features = ["derive", "rc"] }
serde_bytes = "0.11.15"
serde_json = "1.0.120"
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 41a9664c..477b7891 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -125,6 +125,8 @@ const DEFAULT_COMMIT_TIMEOUT_MS: u64 = u64::MAX;
const DEFAULT_COMMIT_MIN_RETRY_WAIT_MS: u64 = 10;
const DEFAULT_COMMIT_MAX_RETRY_WAIT_MS: u64 = 10_000;
pub const SCAN_TIMESTAMP_MILLIS_OPTION: &str = "scan.timestamp-millis";
+/// Local date/time string used for snapshot time travel, matching Java Paimon.
+pub const SCAN_TIMESTAMP_OPTION: &str = "scan.timestamp";
pub const SCAN_VERSION_OPTION: &str = "scan.version";
pub const SCAN_SNAPSHOT_ID_OPTION: &str = "scan.snapshot-id";
pub const SCAN_TAG_NAME_OPTION: &str = "scan.tag-name";
@@ -502,7 +504,7 @@ impl<'a> CoreOptions<'a> {
SCAN_WATERMARK_OPTION,
]
} else if mode.eq_ignore_ascii_case("from-timestamp") {
- &[SCAN_TIMESTAMP_MILLIS_OPTION]
+ &[SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_TIMESTAMP_OPTION]
} else {
return Err(crate::Error::Unsupported {
message: format!(
@@ -1060,10 +1062,13 @@ impl<'a> CoreOptions<'a> {
}
fn configured_time_travel_selectors(&self) -> Vec<&'static str> {
- let mut selectors = Vec::with_capacity(5);
+ let mut selectors = Vec::with_capacity(6);
if self.options.contains_key(SCAN_TIMESTAMP_MILLIS_OPTION) {
selectors.push(SCAN_TIMESTAMP_MILLIS_OPTION);
}
+ if self.options.contains_key(SCAN_TIMESTAMP_OPTION) {
+ selectors.push(SCAN_TIMESTAMP_OPTION);
+ }
if self.options.contains_key(SCAN_WATERMARK_OPTION) {
selectors.push(SCAN_WATERMARK_OPTION);
}
@@ -1124,6 +1129,10 @@ impl<'a> CoreOptions<'a> {
if let Some(ts) = self.parse_i64_option(SCAN_TIMESTAMP_MILLIS_OPTION)?
{
Ok(Some(TimeTravelSelector::TimestampMillis(ts)))
+ } else if let Some(value) = self.options.get(SCAN_TIMESTAMP_OPTION) {
+ Ok(Some(TimeTravelSelector::TimestampMillis(
+ parse_scan_timestamp(value, &jiff::tz::TimeZone::system())?,
+ )))
} else if let Some(watermark) =
self.parse_i64_option(SCAN_WATERMARK_OPTION)? {
Ok(Some(TimeTravelSelector::Watermark(watermark)))
} else if let Some(value) =
self.options.get(SCAN_VERSION_OPTION).map(String::as_str) {
@@ -1614,6 +1623,66 @@ impl<'a> CoreOptions<'a> {
}
}
+/// Java DateTimeUtils accepts a date, a space-separated timestamp, or an ISO
+/// local timestamp (whose seconds are optional). It truncates to milliseconds
+/// and resolves the date/time in the process's default time zone.
+fn parse_scan_timestamp(value: &str, zone: &jiff::tz::TimeZone) ->
crate::Result<i64> {
+ use chrono::{Datelike, NaiveDate, NaiveDateTime, Timelike};
+
+ let datetime = NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S%.f")
+ .or_else(|_| NaiveDateTime::parse_from_str(value,
"%Y-%m-%dT%H:%M:%S%.f"))
+ .or_else(|_| NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M"))
+ .or_else(|_| {
+ NaiveDate::parse_from_str(value, "%Y-%m-%d")
+ .map(|date| date.and_hms_opt(0, 0, 0).unwrap())
+ })
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("Invalid value for {SCAN_TIMESTAMP_OPTION}:
'{value}'"),
+ source: Some(Box::new(error)),
+ })?;
+ // Chrono also accepts leap seconds and fractions longer than nanoseconds;
+ // Java's local timestamp parser does not.
+ if datetime.nanosecond() >= 1_000_000_000
+ || value
+ .rsplit_once('.')
+ .is_some_and(|(_, fraction)| fraction.len() > 9)
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Invalid value for {SCAN_TIMESTAMP_OPTION}:
'{value}'"),
+ source: None,
+ });
+ }
+ // Java parses with precision 3 before applying the system time zone.
+ // Truncating here also preserves the correct millisecond for instants
+ // immediately before the Unix epoch.
+ let year = i16::try_from(datetime.year()).map_err(|error|
crate::Error::DataInvalid {
+ message: format!("Invalid value for {SCAN_TIMESTAMP_OPTION}:
'{value}'"),
+ source: Some(Box::new(error)),
+ })?;
+ let civil = jiff::civil::DateTime::new(
+ year,
+ datetime.month() as i8,
+ datetime.day() as i8,
+ datetime.hour() as i8,
+ datetime.minute() as i8,
+ datetime.second() as i8,
+ (datetime.nanosecond() / 1_000_000 * 1_000_000) as i32,
+ )
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("Invalid value for {SCAN_TIMESTAMP_OPTION}:
'{value}'"),
+ source: Some(Box::new(error)),
+ })?;
+ // Jiff's compatible disambiguation matches Java LocalDateTime.atZone:
+ // the earlier instant in a fold, and the later local time in a gap.
+ zone.to_ambiguous_timestamp(civil)
+ .compatible()
+ .map(|timestamp| timestamp.as_millisecond())
+ .map_err(|error| crate::Error::DataInvalid {
+ message: format!("Invalid local time for {SCAN_TIMESTAMP_OPTION}:
'{value}'"),
+ source: Some(Box::new(error)),
+ })
+}
+
/// Parse a memory size string to bytes using binary (1024-based) semantics,
/// mirroring Java Paimon's `MemorySize.parseBytes`.
///
@@ -2494,6 +2563,156 @@ mod tests {
assert_eq!(parallelism(Some("many")), 64);
}
+ #[test]
+ fn test_scan_timestamp_parses_local_time_and_truncates_to_millis() {
+ let zone = jiff::tz::db().get("Asia/Shanghai").unwrap();
+ let midnight = 1_704_124_800_000_i64; // 2024-01-01T16:00:00Z
+ let noon = 1_704_168_184_000_i64; // 2024-01-02T04:03:04Z
+ for (value, expected) in [
+ ("2024-01-02", midnight),
+ ("2024-1-2 12:3:4", noon),
+ ("2024-01-02 12:03:04", noon),
+ ("2024-01-02 12:03:04.123456789", noon + 123),
+ ("2024-01-02T12:03:04.9", noon + 900),
+ ("2024-01-02T12:03", noon - 4000),
+ ] {
+ assert_eq!(
+ parse_scan_timestamp(value, &zone).unwrap(),
+ expected,
+ "{value}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_scan_timestamp_matches_java_zone_transitions() {
+ for (zone, value, expected) in [
+ (
+ "America/New_York",
+ "2024-11-03 01:30:00",
+ "2024-11-03T05:30:00Z",
+ ),
+ (
+ "America/New_York",
+ "2024-11-03 02:00:00",
+ "2024-11-03T07:00:00Z",
+ ),
+ (
+ "America/New_York",
+ "2024-11-03 02:00:00.999",
+ "2024-11-03T07:00:00.999Z",
+ ),
+ (
+ "America/New_York",
+ "2024-03-10 02:30:00",
+ "2024-03-10T07:30:00Z",
+ ),
+ (
+ "Australia/Lord_Howe",
+ "2024-10-06 02:15:00",
+ "2024-10-05T15:45:00Z",
+ ),
+ (
+ "Pacific/Apia",
+ "2011-12-30 12:00:00",
+ "2011-12-30T22:00:00Z",
+ ),
+ (
+ "UTC",
+ "1969-12-31 23:59:59.999999999",
+ "1969-12-31T23:59:59.999Z",
+ ),
+ ] {
+ let zone = jiff::tz::db().get(zone).unwrap();
+ assert_eq!(
+ parse_scan_timestamp(value, &zone).unwrap(),
+ chrono::DateTime::parse_from_rfc3339(expected)
+ .unwrap()
+ .timestamp_millis(),
+ "{value} in {zone:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_scan_timestamp_local_overlap_uses_earlier_instant() {
+ const CHILD: &str = "PAIMON_SCAN_TIMESTAMP_OVERLAP_CHILD";
+ if std::env::var_os(CHILD).is_some() {
+ // Resolve the system zone in a fresh process, including on
Windows.
+ let actual =
+ parse_scan_timestamp("2024-11-03 01:30:00",
&jiff::tz::TimeZone::system()).unwrap();
+ assert_eq!(actual, 1_730_611_800_000); // 2024-11-03T05:30:00Z
+ let overlap_end =
+ parse_scan_timestamp("2024-11-03 02:00:00",
&jiff::tz::TimeZone::system()).unwrap();
+ assert_eq!(overlap_end, 1_730_617_200_000); // 2024-11-03T07:00:00Z
+ let gap =
+ parse_scan_timestamp("2024-03-10 02:30:00",
&jiff::tz::TimeZone::system()).unwrap();
+ assert_eq!(gap, 1_710_055_800_000); // 2024-03-10T07:30:00Z
+ return;
+ }
+ let output =
std::process::Command::new(std::env::current_exe().unwrap())
+ .args([
+ "--exact",
+
"spec::core_options::tests::test_scan_timestamp_local_overlap_uses_earlier_instant",
+ ])
+ .env("TZ", "America/New_York")
+ .env(CHILD, "1")
+ .output()
+ .unwrap();
+ assert!(
+ output.status.success(),
+ "child failed:\n{}{}",
+ String::from_utf8_lossy(&output.stdout),
+ String::from_utf8_lossy(&output.stderr)
+ );
+ }
+
+ #[test]
+ fn test_scan_timestamp_rejects_invalid_values_and_conflicts() {
+ for value in [
+ "",
+ "invalid",
+ "1700000000000",
+ "2024-13-01",
+ "2024-01-01 25:00:00",
+ "2024-01-01T00:00:00Z",
+ "2024-01-01 23:59:60",
+ "2024-01-01 00:00:00.1234567890",
+ ] {
+ let options = HashMap::from([("scan.timestamp".to_string(),
value.to_string())]);
+ let core = CoreOptions::new(&options);
+ assert!(core.has_time_travel_selector());
+ let err = core.try_time_travel_selector().unwrap_err();
+ assert!(
+ matches!(err, crate::Error::DataInvalid { message, .. } if
message.contains("scan.timestamp")),
+ "{value}"
+ );
+ }
+ for selector in [
+ SCAN_TIMESTAMP_MILLIS_OPTION,
+ SCAN_WATERMARK_OPTION,
+ SCAN_VERSION_OPTION,
+ SCAN_SNAPSHOT_ID_OPTION,
+ SCAN_TAG_NAME_OPTION,
+ ] {
+ let options = HashMap::from([
+ ("scan.timestamp".to_string(), "2024-01-02".to_string()),
+ (selector.to_string(), "1".to_string()),
+ ]);
+ let err = CoreOptions::new(&options)
+ .try_time_travel_selector()
+ .unwrap_err();
+ assert!(
+ matches!(err, crate::Error::DataInvalid { message, .. } if
message.contains("Only one") && message.contains("scan.timestamp") &&
message.contains(selector))
+ );
+ }
+ let options = HashMap::from([
+ ("scan.timestamp".to_string(), "2024-01-02".to_string()),
+ ("scan.mode".to_string(), "from-timestamp".to_string()),
+ ]);
+ assert!(CoreOptions::new(&options).validate_scan_options().is_ok());
+ }
+
#[test]
fn test_try_time_travel_selector_rejects_conflicting_selectors() {
let options = HashMap::from([
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index c7363f3d..f536f3b5 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -193,7 +193,8 @@ use crate::catalog::{validate_branch_name, Identifier,
DEFAULT_MAIN_BRANCH};
use crate::io::FileIO;
use crate::spec::{
CoreOptions, DataField, Snapshot, TableSchema, SCAN_SNAPSHOT_ID_OPTION,
SCAN_TAG_NAME_OPTION,
- SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_VERSION_OPTION, SCAN_WATERMARK_OPTION,
+ SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_TIMESTAMP_OPTION, SCAN_VERSION_OPTION,
+ SCAN_WATERMARK_OPTION,
};
use std::collections::HashMap;
@@ -456,6 +457,7 @@ impl Table {
let selector_changed = extra.keys().any(|k| {
k == crate::spec::SCAN_VERSION_OPTION
|| k == crate::spec::SCAN_TIMESTAMP_MILLIS_OPTION
+ || k == crate::spec::SCAN_TIMESTAMP_OPTION
|| k == crate::spec::SCAN_WATERMARK_OPTION
|| k == crate::spec::SCAN_SNAPSHOT_ID_OPTION
|| k == crate::spec::SCAN_TAG_NAME_OPTION
@@ -533,6 +535,7 @@ impl Table {
let mut options = self.schema.options().clone();
for selector in [
SCAN_TIMESTAMP_MILLIS_OPTION,
+ SCAN_TIMESTAMP_OPTION,
SCAN_WATERMARK_OPTION,
SCAN_VERSION_OPTION,
SCAN_SNAPSHOT_ID_OPTION,
@@ -559,7 +562,7 @@ impl Table {
///
/// Mirrors Java `AbstractFileStoreTable.copy(dynamicOptions)` →
/// `tryTimeTravel`: if the merged options contain a time-travel selector
- /// (`scan.version` / `scan.timestamp-millis` / `scan.watermark` /
+ /// (`scan.version` / `scan.timestamp-millis` / `scan.timestamp` /
`scan.watermark` /
/// `scan.snapshot-id` / `scan.tag-name`) 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)`).
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 2a84357a..d1c614b6 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1490,13 +1490,14 @@ impl<'a> PaimonTableScan<'a> {
///
/// Time travel is resolved from table options:
/// - `scan.version` is resolved first, overwriting the same selector kind;
- /// only one of `scan.timestamp-millis`, `scan.watermark`,
`scan.snapshot-id`,
+ /// only one of `scan.timestamp-millis`, `scan.timestamp`,
`scan.watermark`, `scan.snapshot-id`,
/// `scan.tag-name` may remain after resolution
/// - `scan.version` → tag name (if exists) → `watermark-<value>` →
snapshot
/// id (if parseable) → error (ambiguous by design, like SQL `VERSION AS
OF`)
/// - `scan.snapshot-id` → snapshot id only (never a tag lookup)
/// - `scan.tag-name` → tag name only (never parsed as a snapshot id)
/// - `scan.timestamp-millis` → find the latest snapshot <= that timestamp
+ /// - `scan.timestamp` → parse in the local time zone, then resolve like
`scan.timestamp-millis`
/// - `scan.watermark` → find the earliest snapshot with watermark >= that
/// value (snapshots without a watermark are skipped)
/// - otherwise → read the latest snapshot
diff --git a/crates/paimon/src/table/time_travel.rs
b/crates/paimon/src/table/time_travel.rs
index 838f0d3c..fadb1715 100644
--- a/crates/paimon/src/table/time_travel.rs
+++ b/crates/paimon/src/table/time_travel.rs
@@ -374,6 +374,101 @@ mod tests {
.collect()
}
+ #[tokio::test]
+ async fn test_scan_timestamp_reads_historical_schema_and_rows() {
+ use chrono::{Local, TimeZone};
+ use futures::TryStreamExt;
+
+ let (io, path) = setup_evolved_table().await;
+ let table = latest_table(&io, &path);
+ let sm = table.snapshot_manager();
+ let base = Local
+ .with_ymd_and_hms(2024, 1, 2, 12, 0, 0)
+ .unwrap()
+ .timestamp_millis();
+ // Fix commit times so the exact/between/before boundaries never depend
+ // on how quickly the fixture commits complete.
+ for (id, millis) in [(1, base + 123), (2, base + 1000)] {
+ let mut snapshot =
serde_json::to_value(sm.get_snapshot(id).await.unwrap()).unwrap();
+ snapshot["timeMillis"] = serde_json::json!(millis);
+ io.new_output(&sm.snapshot_path(id))
+ .unwrap()
+ .write(serde_json::to_vec(&snapshot).unwrap().into())
+ .await
+ .unwrap();
+ }
+
+ for timestamp in ["2024-01-02 12:00:00.123",
"2024-01-02T12:00:00.999999999"] {
+ let opts = options(&[
+ ("scan.timestamp", timestamp),
+ ("scan.mode", "from-timestamp"),
+ ]);
+ let traveled = table
+ .copy_with_time_travel_strict(opts.clone())
+ .await
+ .unwrap();
+ assert_eq!(traveled.schema().id(), 0);
+ assert_eq!(traveled.travel_snapshot().unwrap().id(), 1);
+ assert!(traveled.new_write_builder().new_write().is_err());
+ assert!(table
+ .copy_with_options(opts)
+ .new_write_builder()
+ .new_write()
+ .is_err());
+ 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 ids: Vec<i32> = batches
+ .iter()
+ .flat_map(|batch| {
+ assert_eq!(batch.num_columns(), 2);
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .values()
+ .iter()
+ .copied()
+ })
+ .collect();
+ assert_eq!(ids, vec![1, 2, 3]);
+
+ let changed =
+ traveled.copy_with_options(options(&[("scan.timestamp",
"2024-01-02 12:00:01")]));
+ assert!(changed.travel_snapshot().is_none());
+ let resolved = changed
+ .copy_with_time_travel_strict(HashMap::new())
+ .await
+ .unwrap();
+ assert_eq!(resolved.travel_snapshot().unwrap().id(), 2);
+ assert_eq!(resolved.schema().id(), 1);
+
+ let pinned =
traveled.copy_with_pinned_snapshot(traveled.travel_snapshot().unwrap());
+ assert!(!pinned.schema().options().contains_key("scan.timestamp"));
+ }
+ for timestamp in ["2024-01-02 12:00:01", "2024-01-02 12:00:02"] {
+ let selected = super::resolve_snapshot(
+ &table.copy_with_options(options(&[("scan.timestamp",
timestamp)])),
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(selected.id(), 2);
+ }
+ for timestamp in ["2024-01-02 12:00:00.122", "bad timestamp"] {
+ let selected =
table.copy_with_options(options(&[("scan.timestamp", timestamp)]));
+
assert!(selected.new_read_builder().new_scan().plan().await.is_err());
+ }
+ }
+
#[tokio::test]
async fn test_copy_with_time_travel_switches_to_snapshot_schema() {
let (file_io, table_path) = setup_evolved_table().await;
diff --git a/crates/paimon/src/table/write_builder.rs
b/crates/paimon/src/table/write_builder.rs
index 7f00059d..b2af61d0 100644
--- a/crates/paimon/src/table/write_builder.rs
+++ b/crates/paimon/src/table/write_builder.rs
@@ -222,7 +222,7 @@ pub(super) fn ensure_table_write_allowed(table: &Table) ->
crate::Result<()> {
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 / scan.snapshot-id /
scan.tag-name)"
+ (scan.version / scan.timestamp-millis / scan.timestamp /
scan.watermark / scan.snapshot-id / scan.tag-name)"
.to_string(),
});
}
diff --git a/docs/src/python-binding.md b/docs/src/python-binding.md
index 6e5ce569..5d5ee49e 100644
--- a/docs/src/python-binding.md
+++ b/docs/src/python-binding.md
@@ -443,6 +443,9 @@ rb = table.new_read_builder({"scan.snapshot-id": "1"})
# By timestamp (epoch millis)
rb = table.new_read_builder({"scan.timestamp-millis": "1700000000000"})
+# By timestamp string (process local time zone; fractional seconds are
supported)
+rb = table.new_read_builder({"scan.timestamp": "2024-01-01 12:00:00.123"})
+
# By version
rb = table.new_read_builder({"scan.version": "3"})
@@ -453,6 +456,12 @@ rb = table.new_read_builder({"scan.tag-name":
"release-1.0"})
!!! warning
Only one time-travel selector may be set. Providing multiple selectors
will raise a `ValueError`.
+`scan.timestamp` selects the latest snapshot committed at or before the given
+local time. It accepts `YYYY-MM-DD`, `YYYY-MM-DD HH:MM:SS[.fraction]`, or
+`YYYY-MM-DDTHH:MM[:SS[.fraction]]`, with up to nine fractional digits truncated
+to milliseconds. It can be combined with `scan.mode=from-timestamp`. Invalid
+timestamps and times before the earliest available snapshot raise an error.
+
## Table Inspection
Inspect snapshots, tags, and partition statistics on a table. A
branch-qualified
diff --git a/docs/src/sql.md b/docs/src/sql.md
index 6cad8cfe..031a38bc 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -2018,6 +2018,19 @@ SELECT * FROM paimon.default.my_table TIMESTAMP AS OF
'2024-01-01 00:00:00';
This finds the latest snapshot whose commit time is less than or equal to the
given timestamp. The timestamp is interpreted in the local timezone.
+The session option `scan.timestamp` also accepts a local timestamp string,
+including fractional seconds (up to nine digits, truncated to milliseconds):
+
+```sql
+SET 'paimon.scan.timestamp' = '2024-01-01 00:00:00.123';
+SELECT * FROM paimon.default.my_table;
+RESET 'paimon.scan.timestamp';
+```
+
+It can be combined with `scan.mode=from-timestamp`, but cannot be combined with
+another time-travel selector such as `scan.timestamp-millis` or
`scan.snapshot-id`.
+An invalid timestamp or a time before the earliest available snapshot fails
the read.
+
### By Watermark
Use `VERSION AS OF 'watermark-<value>'` syntax:
