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 ae62495d feat(table): honor numeric incremental timestamp scan options
(#1074)
ae62495d is described below
commit ae62495d275ccf3811c38128788be8cf7e144177
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 14:46:45 2026 +0800
feat(table): honor numeric incremental timestamp scan options (#1074)
---
bindings/python/README.md | 7 +
.../python/python/pypaimon_rust/datafusion.pyi | 8 +-
bindings/python/tests/test_read.py | 29 +-
crates/paimon/src/spec/core_options.rs | 157 +++++++-
crates/paimon/src/table/format_table_scan.rs | 1 +
crates/paimon/src/table/incremental_scan.rs | 103 +++++-
crates/paimon/src/table/scan_trace.rs | 34 ++
crates/paimon/src/table/source.rs | 5 +
crates/paimon/src/table/table_merge_input.rs | 13 +-
crates/paimon/src/table/table_merge_into.rs | 2 +-
crates/paimon/src/table/table_scan.rs | 264 +++++---------
.../tests/incremental_timestamp_scan_test.rs | 396 +++++++++++++++++++++
12 files changed, 803 insertions(+), 216 deletions(-)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 3effeb7f..6de276a0 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -129,6 +129,13 @@ logical rows through the ordinary reader, including
primary-key merge engines.
The source does not require data evolution or row tracking. Arrow batches keep
their chunk boundaries; matching gathers only the contributing source chunks.
+Numeric `incremental-between-timestamp` options select `(start, end]` in core,
+including for table-source MERGE. AUTO reads APPEND deltas without a changelog
+producer and physical changelogs otherwise; OVERWRITE snapshots are skipped.
+Equal bounds return an empty plan. Window plans and all their splits identify
+the ending snapshot, including when no selected files remain. The existing
+`new_scan()` honors these options; no new Python API is required.
+
### Custom Blob URI sources
Before the first Arrow write, call
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 52b26b71..adfa1c75 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -268,8 +268,8 @@ class BatchTableUpdate:
) -> List[CommitMessage]:
"""Prepare MERGE against one target snapshot. None selects self-merge
on _ROW_ID.
- A Table source reads its selected full snapshot in core, independently
- of the target. Arrow input preserves batches and source row order.
+ A Table source reads its selected snapshot or numeric timestamp window
+ in core, independently of the target. Arrow input preserves batches
and source row order.
Clauses carry assignments as (column, source/target/literal, value)
tuples, a delete flag, and optional SQL condition metadata.
@@ -310,8 +310,8 @@ class StreamTableUpdate:
) -> List[CommitMessage]:
"""Prepare MERGE against one target snapshot. None selects self-merge
on _ROW_ID.
- A Table source reads its selected full snapshot in core, independently
- of the target. Arrow input preserves batches and source row order.
+ A Table source reads its selected snapshot or numeric timestamp window
+ in core, independently of the target. Arrow input preserves batches
and source row order.
Clauses carry assignments as (column, source/target/literal, value)
tuples, a delete flag, and optional SQL condition metadata.
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 55d91b46..1a88cbe2 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -1315,11 +1315,11 @@ def test_explicit_scan_mode_without_selector_raises():
table.new_read_builder({"scan.mode": "from-snapshot"})
-def test_unimplemented_scan_mode_raises_not_implemented():
+def test_incremental_scan_mode_requires_timestamp_window():
with tempfile.TemporaryDirectory() as warehouse:
_make_two_snapshot_table(warehouse)
table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
- with pytest.raises(NotImplementedError):
+ with pytest.raises(ValueError, match="incremental-between-timestamp"):
table.new_read_builder({"scan.mode": "incremental"})
@@ -1730,3 +1730,28 @@ def
test_incremental_row_positions_use_combined_delta_batch():
builder.with_row_ranges([(1, 3)]).with_limit(2)
plan = builder.new_incremental_scan(0, 2).with_row_position_slice(2,
5).plan()
assert
pa.Table.from_batches(builder.new_read().read(plan.splits())).column("id").to_pylist()
== [2, 3]
+
+
[email protected]("window,snapshot,expected", [
+ ("0,100", 1, [dict(id=1, name="a")]),
+ ("100,200", 2, [dict(id=2, name="b"), dict(id=3, name="c")]),
+ ("100,150", 1, []),
+ ("100,100", None, []),
+])
[email protected]("mode", ["default", "incremental"])
+def test_numeric_timestamp_window_options_use_core_scan(tmp_path, window,
snapshot, expected, mode):
+ from pathlib import Path
+
+ _make_two_snapshot_table(str(tmp_path))
+ table = PaimonCatalog({"warehouse": str(tmp_path)}).get_table("tdb.t")
+ for identifier, timestamp in [(1, 100), (2, 200)]:
+ path = Path(table.location()) / "snapshot" / ("snapshot-%s" %
identifier)
+ data = json.loads(path.read_text())
+ data["timeMillis"] = timestamp
+ path.write_text(json.dumps(data))
+ builder = table.new_read_builder({"scan.mode": mode,
"incremental-between-timestamp": window})
+ plan = builder.new_scan().plan()
+ assert plan.snapshot_id() == snapshot
+ batches = list(builder.new_read().read(plan.splits()))
+ actual = [row for batch in batches for row in batch.to_pylist()]
+ assert sorted(actual, key=lambda row: row["id"]) == expected
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 3bacbaa3..d740b08e 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -527,11 +527,11 @@ impl<'a> CoreOptions<'a> {
/// that selector is present (the batch-read semantics are identical to
/// leaving the mode at `default`); an explicit mode without its selector
/// is malformed input (`Error::DataInvalid`), mirroring Java's
- /// `SchemaValidation`. All other non-default modes are unimplemented.
+ /// `SchemaValidation`. Incremental timestamp windows reuse the batch
+ /// Delta/Changelog scanner; snapshot/tag ranges remain unimplemented here.
pub fn validate_scan_options(&self) -> crate::Result<()> {
for key in [
INCREMENTAL_BETWEEN_OPTION,
- INCREMENTAL_BETWEEN_TIMESTAMP_OPTION,
INCREMENTAL_BETWEEN_SCAN_MODE_OPTION,
] {
if self.options.contains_key(key) {
@@ -540,9 +540,47 @@ impl<'a> CoreOptions<'a> {
});
}
}
+ let incremental = self.incremental_timestamp_window()?.is_some();
+ if incremental {
+ self.try_changelog_producer()?;
+ if self.is_format_table() {
+ return Err(crate::Error::Unsupported {
+ message: "Incremental timestamp scans require a Paimon
table".into(),
+ });
+ }
+ let selectors = self.configured_time_travel_selectors();
+ if !selectors.is_empty()
+ || self.options.contains_key("scan.file-creation-time-millis")
+ || self.options.contains_key("scan.creation-time-millis")
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "incremental-between-timestamp cannot be used
with point-in-time scan options".into(),
+ source: None,
+ });
+ }
+ }
if let Some(mode) = self.options.get(SCAN_MODE_OPTION) {
+ if incremental
+ && !mode.eq_ignore_ascii_case("default")
+ && !mode.eq_ignore_ascii_case("incremental")
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "incremental-between-timestamp cannot be used with
scan.mode={mode}"
+ ),
+ source: None,
+ });
+ }
let selector_keys: &[&str] = if
mode.eq_ignore_ascii_case("default") {
return Ok(());
+ } else if mode.eq_ignore_ascii_case("incremental") {
+ if incremental {
+ return Ok(());
+ }
+ return Err(crate::Error::DataInvalid {
+ message: "scan.mode=incremental requires
incremental-between-timestamp".into(),
+ source: None,
+ });
} else if mode.eq_ignore_ascii_case("from-snapshot") {
&[
SCAN_SNAPSHOT_ID_OPTION,
@@ -575,6 +613,22 @@ impl<'a> CoreOptions<'a> {
Ok(())
}
+ /// The numeric timestamp window supported by PyPaimon. Snapshot resolution
+ /// and ordering checks belong to the batch scanner, as in Java.
+ pub(crate) fn incremental_timestamp_window(&self) ->
crate::Result<Option<(i64, i64)>> {
+ let Some(raw) = self.options.get(INCREMENTAL_BETWEEN_TIMESTAMP_OPTION)
else {
+ return Ok(None);
+ };
+ let parse = || {
+ let (start, end) = raw.split_once(',')?;
+ Some((start.parse::<i64>().ok()?, end.parse::<i64>().ok()?))
+ };
+ parse().map(Some).ok_or_else(|| crate::Error::DataInvalid {
+ message: format!("incremental-between-timestamp requires
start(exclusive),end(inclusive) in milliseconds, got: {raw}"),
+ source: None,
+ })
+ }
+
pub fn deletion_vectors_enabled(&self) -> bool {
self.options
.get(DELETION_VECTORS_ENABLED_OPTION)
@@ -1286,7 +1340,7 @@ impl<'a> CoreOptions<'a> {
pub fn ensure_engine_can_serve(&self, full_name: &str) ->
crate::Result<()> {
self.ensure_query_auth_absent()?;
self.validate_scan_options()?;
- if self.has_time_travel_selector() {
+ if self.has_time_travel_selector() ||
self.incremental_timestamp_window()?.is_some() {
return Err(crate::Error::Unsupported {
message: format!(
"time travel is not supported for engine-served table
'{full_name}'"
@@ -3715,11 +3769,7 @@ mod tests {
#[test]
fn test_validate_scan_options_rejects_unsupported() {
- for key in [
- "incremental-between",
- "incremental-between-timestamp",
- "incremental-between-scan-mode",
- ] {
+ for key in ["incremental-between", "incremental-between-scan-mode"] {
let options = HashMap::from([(key.to_string(), "x".to_string())]);
let err = CoreOptions::new(&options)
.validate_scan_options()
@@ -3728,6 +3778,95 @@ mod tests {
}
}
+ #[test]
+ fn test_numeric_incremental_timestamp_options_match_java_long_rules() {
+ for value in [
+ "0,100",
+ "100,100",
+ "200,100",
+ "+0,+100",
+ "-100,100",
+ "-9223372036854775808,9223372036854775807",
+ ] {
+ for mode in [None, Some("default"), Some("incremental")] {
+ let mut options =
+ HashMap::from([("incremental-between-timestamp".into(),
value.into())]);
+ if let Some(mode) = mode {
+ options.insert("scan.mode".into(), mode.into());
+ }
+ assert!(
+ CoreOptions::new(&options).validate_scan_options().is_ok(),
+ "{options:?}"
+ );
+ }
+ }
+ for value in [
+ "100",
+ "one,200",
+ "100,200,300",
+ "100, 200",
+ "1_00,200",
+ "0,9223372036854775808",
+ "-9223372036854775809,100",
+ ] {
+ let options =
HashMap::from([("incremental-between-timestamp".into(), value.into())]);
+ assert!(
+ matches!(
+ CoreOptions::new(&options).validate_scan_options(),
+ Err(crate::Error::DataInvalid { .. })
+ ),
+ "{value}"
+ );
+ }
+ }
+
+ #[test]
+ fn
test_incremental_timestamp_options_reject_conflicting_and_unservable_routes() {
+ for selector in [
+ "scan.snapshot-id",
+ "scan.tag-name",
+ "scan.version",
+ "scan.watermark",
+ "scan.timestamp",
+ "scan.timestamp-millis",
+ "scan.file-creation-time-millis",
+ "scan.creation-time-millis",
+ ] {
+ let options = HashMap::from([
+ ("incremental-between-timestamp".into(), "0,100".into()),
+ (selector.into(), "1".into()),
+ ]);
+ assert!(
+ matches!(
+ CoreOptions::new(&options).validate_scan_options(),
+ Err(crate::Error::DataInvalid { .. })
+ ),
+ "{selector}"
+ );
+ }
+ let options = HashMap::from([("scan.mode".into(),
"incremental".into())]);
+ assert!(matches!(
+ CoreOptions::new(&options).validate_scan_options(),
+ Err(crate::Error::DataInvalid { .. })
+ ));
+ let options = HashMap::from([("incremental-between-timestamp".into(),
"0,100".into())]);
+ assert!(matches!(
+ CoreOptions::new(&options).ensure_engine_can_serve("db.t"),
+ Err(crate::Error::Unsupported { .. })
+ ));
+ let mut format_options = options.clone();
+ format_options.insert("type".into(), "format-table".into());
+ assert!(matches!(
+ CoreOptions::new(&format_options).validate_scan_options(),
+ Err(crate::Error::Unsupported { .. })
+ ));
+ let mut producer_options = options;
+ producer_options.insert("changelog-producer".into(), "invalid".into());
+ assert!(CoreOptions::new(&producer_options)
+ .validate_scan_options()
+ .is_err());
+ }
+
#[test]
fn test_validate_scan_options_scan_mode_whitelist() {
// absent OK
@@ -3738,7 +3877,7 @@ mod tests {
let ok = HashMap::from([("scan.mode".to_string(),
"default".to_string())]);
assert!(CoreOptions::new(&ok).validate_scan_options().is_ok());
// unimplemented modes Unsupported
- for mode in ["compacted-full", "incremental", "latest", "latest-full"]
{
+ for mode in ["compacted-full", "latest", "latest-full"] {
let bad = HashMap::from([("scan.mode".to_string(),
mode.to_string())]);
let err =
CoreOptions::new(&bad).validate_scan_options().unwrap_err();
assert!(
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index 6af2c158..b5a1d1a0 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -78,6 +78,7 @@ impl<'a> FormatTableScan<'a> {
}
fn ensure_query_auth_allowed(&self) -> crate::Result<()> {
+
CoreOptions::new(self.table.schema().options()).validate_scan_options()?;
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()
}
diff --git a/crates/paimon/src/table/incremental_scan.rs
b/crates/paimon/src/table/incremental_scan.rs
index e2e82e04..7d30fd73 100644
--- a/crates/paimon/src/table/incremental_scan.rs
+++ b/crates/paimon/src/table/incremental_scan.rs
@@ -15,9 +15,43 @@
// specific language governing permissions and limitations
// under the License.
-use super::{DataSplit, Plan, SnapshotManager, Table, TableScan};
+use super::{DataSplit, Plan, ScanTrace, SnapshotManager, Table, TableScan};
use crate::spec::{CommitKind, CoreOptions};
+/// Java IncrementalDeltaStartingScanner.betweenTimestamps. Resolve both bounds
+/// once; readers consume the resulting plan and never reselect a source
window.
+pub(super) async fn timestamp_snapshot_range(
+ manager: &SnapshotManager,
+ start: i64,
+ end: i64,
+) -> crate::Result<Option<(i64, i64)>> {
+ let Some(earliest_id) = manager.earliest_snapshot_id().await? else {
+ return Ok(None);
+ };
+ let earliest = manager.get_snapshot(earliest_id).await?;
+ let Some(latest) = manager.get_latest_snapshot().await? else {
+ return Ok(None);
+ };
+ if end < start {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Ending timestamp {end} should be >= starting
timestamp {start}"),
+ source: None,
+ });
+ }
+ if start == end || start > latest.time_millis() as i64 || end <
earliest.time_millis() as i64 {
+ return Ok(None);
+ }
+ let start_id = manager
+ .earlier_or_equal_time_millis(start)
+ .await?
+ .map_or(earliest.id() - 1, |snapshot| snapshot.id());
+ let end_id = manager
+ .earlier_or_equal_time_millis(end)
+ .await?
+ .map_or(latest.id(), |snapshot| snapshot.id());
+ Ok(Some((start_id, end_id)))
+}
+
/// Batch incremental scan mode.
///
/// Range semantics: `(start_exclusive, end_inclusive]` — start is exclusive
and
@@ -273,6 +307,20 @@ impl<'a> IncrementalScan<'a> {
}
pub async fn plan(&self) -> crate::Result<IncrementalPlan> {
+ let mode = self.resolve_mode();
+ self.validate_per_snapshot_plan(mode).await?;
+ if self.start_exclusive == self.end_inclusive {
+ return Ok(IncrementalPlan::new(mode, Vec::new()));
+ }
+ match mode {
+ IncrementalScanMode::Delta => self.plan_delta(mode).await,
+ IncrementalScanMode::Changelog => self.plan_changelog(mode,
None).await,
+ IncrementalScanMode::Auto => unreachable!("Auto must resolve
before planning"),
+ IncrementalScanMode::Diff => self.plan_diff(mode).await,
+ }
+ }
+
+ async fn validate_per_snapshot_plan(&self, mode: IncrementalScanMode) ->
crate::Result<()> {
crate::spec::CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
if self.scan.has_row_position_selection()
|| self.scan.has_chunk_shuffle()
@@ -283,17 +331,7 @@ impl<'a> IncrementalScan<'a> {
.into(),
});
}
- let mode = self.resolve_mode();
- self.validate_snapshot_range(mode).await?;
- if self.start_exclusive == self.end_inclusive {
- return Ok(IncrementalPlan::new(mode, Vec::new()));
- }
- match mode {
- IncrementalScanMode::Delta => self.plan_delta(mode).await,
- IncrementalScanMode::Changelog => self.plan_changelog(mode).await,
- IncrementalScanMode::Auto => unreachable!("Auto must resolve
before planning"),
- IncrementalScanMode::Diff => self.plan_diff(mode).await,
- }
+ self.validate_snapshot_range(mode).await
}
/// Plan a Delta or Changelog range as one ordinary [`Plan`].
@@ -305,14 +343,22 @@ impl<'a> IncrementalScan<'a> {
/// represented by an ordinary split list because each unit contains a
/// before/after pair.
pub async fn plan_combined(&self) -> crate::Result<Plan> {
+ self.plan_combined_with_trace(None).await
+ }
+
+ pub(crate) async fn plan_combined_with_trace(
+ &self,
+ trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<Plan> {
match self.resolve_mode() {
- IncrementalScanMode::Delta => self.plan_combined_delta().await,
+ IncrementalScanMode::Delta =>
self.plan_combined_delta_with_trace(trace).await,
IncrementalScanMode::Changelog => {
- let incremental = self.plan().await?;
+
self.validate_per_snapshot_plan(IncrementalScanMode::Changelog).await?;
+ let incremental =
self.plan_changelog(IncrementalScanMode::Changelog, trace).await?;
let mut splits =
Vec::with_capacity(incremental.splits().len());
for split in incremental.splits() {
match split {
- IncrementalSplit::Data(split) =>
splits.push(split.clone()),
+ IncrementalSplit::Data(split) =>
splits.push(split.clone().with_snapshot_id(self.end_inclusive)),
IncrementalSplit::DiffPair { .. } => {
return Err(crate::Error::UnexpectedError {
message: "DiffPair appeared in a Changelog
incremental plan"
@@ -341,6 +387,13 @@ impl<'a> IncrementalScan<'a> {
/// must exist and supplies snapshot metadata. Snapshot deletion vectors
and
/// automatic global-index pruning do not apply to these historical events.
pub async fn plan_combined_delta(&self) -> crate::Result<Plan> {
+ self.plan_combined_delta_with_trace(None).await
+ }
+
+ async fn plan_combined_delta_with_trace(
+ &self,
+ trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<Plan> {
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
let mode = self.resolve_mode();
if mode != IncrementalScanMode::Delta {
@@ -366,7 +419,7 @@ impl<'a> IncrementalScan<'a> {
snapshots.push(end_snapshot.clone());
}
self.scan
- .plan_snapshot_deltas(&snapshots, &end_snapshot)
+ .plan_snapshot_deltas_with_trace(&snapshots, &end_snapshot, trace)
.await
}
@@ -435,7 +488,11 @@ impl<'a> IncrementalScan<'a> {
IncrementalPlan::try_new(mode, splits)
}
- async fn plan_changelog(&self, mode: IncrementalScanMode) ->
crate::Result<IncrementalPlan> {
+ async fn plan_changelog(
+ &self,
+ mode: IncrementalScanMode,
+ mut trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<IncrementalPlan> {
let mut splits = Vec::new();
for snapshot_id in (self.start_exclusive + 1)..=self.end_inclusive {
let snapshot =
self.snapshot_manager.get_snapshot(snapshot_id).await?;
@@ -447,7 +504,17 @@ impl<'a> IncrementalScan<'a> {
if snapshot.changelog_manifest_list().is_none() {
continue;
}
- let plan = self.scan.plan_snapshot_changelog(&snapshot).await?;
+ let mut snapshot_trace = ScanTrace::default();
+ let plan = self
+ .scan
+ .plan_snapshot_changelog_with_trace(
+ &snapshot,
+ trace.as_ref().map(|_| &mut snapshot_trace),
+ )
+ .await?;
+ if let Some(trace) = trace.as_deref_mut() {
+ trace.accumulate(&snapshot_trace);
+ }
splits.extend(plan.splits().iter().cloned().map(IncrementalSplit::Data));
}
IncrementalPlan::try_new(mode, splits)
diff --git a/crates/paimon/src/table/scan_trace.rs
b/crates/paimon/src/table/scan_trace.rs
index 45063800..8cd8aa25 100644
--- a/crates/paimon/src/table/scan_trace.rs
+++ b/crates/paimon/src/table/scan_trace.rs
@@ -59,6 +59,40 @@ pub struct ScanTrace {
}
impl ScanTrace {
+ /// Add counters from another independently planned changelog snapshot.
+ pub(crate) fn accumulate(&mut self, other: &Self) {
+ self.base_manifest_files += other.base_manifest_files;
+ self.delta_manifest_files += other.delta_manifest_files;
+ self.manifest_files_before_partition_pruning +=
+ other.manifest_files_before_partition_pruning;
+ self.manifest_files_after_partition_pruning +=
other.manifest_files_after_partition_pruning;
+ self.manifest_files_pruned_by_row_ranges +=
other.manifest_files_pruned_by_row_ranges;
+ self.manifest_entries_read += other.manifest_entries_read;
+ self.manifest_entries_pruned_by_bucket +=
other.manifest_entries_pruned_by_bucket;
+ self.manifest_entries_pruned_by_partition +=
other.manifest_entries_pruned_by_partition;
+ self.manifest_entries_after_entry_pruning +=
other.manifest_entries_after_entry_pruning;
+ self.manifest_entries_pruned_by_level +=
other.manifest_entries_pruned_by_level;
+ self.manifest_entries_pruned_by_row_ranges +=
other.manifest_entries_pruned_by_row_ranges;
+ self.manifest_entries_pruned_by_data_stats +=
other.manifest_entries_pruned_by_data_stats;
+ self.manifest_entries_after_manifest_filters +=
+ other.manifest_entries_after_manifest_filters;
+ self.manifest_entries_after_merge +=
other.manifest_entries_after_merge;
+ self.manifest_entries_pruned_by_cross_schema_stats +=
+ other.manifest_entries_pruned_by_cross_schema_stats;
+ self.manifest_entries_after_cross_schema_stats +=
+ other.manifest_entries_after_cross_schema_stats;
+ self.data_evolution_groups_before_stats +=
other.data_evolution_groups_before_stats;
+ self.data_evolution_groups_pruned_by_stats +=
other.data_evolution_groups_pruned_by_stats;
+ self.data_evolution_groups_pruned_by_row_ranges +=
+ other.data_evolution_groups_pruned_by_row_ranges;
+ self.split_candidates_built += other.split_candidates_built;
+ self.splits_before_limit += other.splits_before_limit;
+ self.splits_after_limit += other.splits_after_limit;
+ self.final_splits += other.final_splits;
+ self.final_files += other.final_files;
+ self.limit_early_stopped |= other.limit_early_stopped;
+ }
+
pub(crate) fn record_manifest_lists(&mut self, base_count: usize,
delta_count: usize) {
self.base_manifest_files = base_count;
self.delta_manifest_files = delta_count;
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 90e12e37..fd7a8d13 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -552,6 +552,11 @@ impl DataSplit {
pub fn snapshot_id(&self) -> i64 {
self.snapshot_id
}
+
+ pub(crate) fn with_snapshot_id(mut self, snapshot_id: i64) -> Self {
+ self.snapshot_id = snapshot_id;
+ self
+ }
pub fn partition(&self) -> &BinaryRow {
&self.partition
}
diff --git a/crates/paimon/src/table/table_merge_input.rs
b/crates/paimon/src/table/table_merge_input.rs
index a0a03d4c..67b910d6 100644
--- a/crates/paimon/src/table/table_merge_input.rs
+++ b/crates/paimon/src/table/table_merge_input.rs
@@ -108,7 +108,16 @@ impl MergeInput {
.iter()
.map(|field| field.name().as_str())
.collect::<Vec<_>>();
- let snapshot = super::time_travel::resolve_snapshot(table).await?;
+ let incremental = table
+ .schema()
+ .core_options()
+ .incremental_timestamp_window()?
+ .is_some();
+ let snapshot = if incremental {
+ None
+ } else {
+ super::time_travel::resolve_snapshot(table).await?
+ };
// Pin the source independently, keeping branch/time-travel resolution
// and the caller's resolved schema. Include L0 as Python
plan_for_write.
let table = snapshot.as_ref().map_or_else(
@@ -121,7 +130,7 @@ impl MergeInput {
);
let mut builder = table.new_read_builder();
builder.with_projection(&names)?;
- let plan = if snapshot.is_some() {
+ let plan = if incremental || snapshot.is_some() {
builder.new_scan().with_scan_all_files().plan().await?
} else {
// Run the normal reader's policy checks even with no snapshot, but
diff --git a/crates/paimon/src/table/table_merge_into.rs
b/crates/paimon/src/table/table_merge_into.rs
index 1eaae977..47ae2147 100644
--- a/crates/paimon/src/table/table_merge_into.rs
+++ b/crates/paimon/src/table/table_merge_into.rs
@@ -72,7 +72,7 @@ pub struct WhenNotMatched {
/// Materialized input, a Paimon source table, or the target joined on its row
ID.
pub enum MergeSource {
Batches(Vec<RecordBatch>),
- /// Read a Paimon table at one independently pinned source snapshot.
+ /// Read a Paimon table at an independently selected snapshot or timestamp
window.
Table(Arc<Table>),
SelfTable,
}
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index edd67a42..e728c730 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -213,6 +213,7 @@ enum IncrementalSplitMode {
enum ManifestListSource<'a> {
Snapshot(&'a Snapshot),
AppendDeltas(&'a [Snapshot]),
+ IncrementalList(&'a str),
}
/// Reads all manifest entries for a snapshot (base + delta manifest lists,
then each manifest file).
@@ -244,12 +245,16 @@ async fn read_all_manifest_entries(
manifest_parallelism: usize,
trace: Option<&mut ScanTrace>,
) -> crate::Result<Vec<ManifestEntry>> {
- let incremental = matches!(&source, ManifestListSource::AppendDeltas(_));
+ let incremental = !matches!(&source, ManifestListSource::Snapshot(_));
let (mut manifest_files, delta) = match source {
ManifestListSource::Snapshot(snapshot) => futures::try_join!(
read_manifest_list(file_io, table_path,
snapshot.base_manifest_list()),
read_manifest_list(file_io, table_path,
snapshot.delta_manifest_list()),
)?,
+ ManifestListSource::IncrementalList(name) => (
+ Vec::new(),
+ read_manifest_list(file_io, table_path, name).await?,
+ ),
ManifestListSource::AppendDeltas(snapshots) => {
let names = snapshots
.iter()
@@ -1319,13 +1324,17 @@ impl<'a> TableScan<'a> {
}
}
- pub(crate) async fn plan_snapshot_deltas(
+ pub(crate) async fn plan_snapshot_deltas_with_trace(
&self,
snapshots: &[Snapshot],
end_snapshot: &Snapshot,
+ trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
match &self.0 {
- TableScanKind::Paimon(scan) =>
scan.plan_snapshot_deltas(snapshots, end_snapshot).await,
+ TableScanKind::Paimon(scan) => {
+ scan.plan_snapshot_deltas(snapshots, end_snapshot, trace)
+ .await
+ }
TableScanKind::Format(_) => Err(crate::Error::Unsupported {
message: "Format tables do not support incremental delta
scan".to_string(),
}),
@@ -1355,9 +1364,13 @@ impl<'a> TableScan<'a> {
}
/// Plan data splits from a snapshot's changelog manifest list only.
- pub(crate) async fn plan_snapshot_changelog(&self, snapshot: &Snapshot) ->
crate::Result<Plan> {
+ pub(crate) async fn plan_snapshot_changelog_with_trace(
+ &self,
+ snapshot: &Snapshot,
+ trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<Plan> {
match &self.0 {
- TableScanKind::Paimon(scan) =>
scan.plan_snapshot_changelog(snapshot).await,
+ TableScanKind::Paimon(scan) =>
scan.plan_snapshot_changelog(snapshot, trace).await,
TableScanKind::Format(_) => Err(crate::Error::Unsupported {
message: "Format tables do not support incremental changelog
scan".to_string(),
}),
@@ -1543,6 +1556,16 @@ impl<'a> PaimonTableScan<'a> {
.filter(|grant| !grant.is_unrestricted())
.map(|grant| self.restricted_by(grant));
let scan = restricted.as_ref().unwrap_or(self);
+ if let Some((start, end)) = scan
+ .table
+ .schema()
+ .core_options()
+ .incremental_timestamp_window()?
+ {
+ let plan = scan.plan_timestamp_window(start, end, None).await?;
+ scan.check_planned_files(&plan, grant.is_some()).await?;
+ return Ok(plan.planned(grant));
+ }
let data_evolution_read_field_ids = scan.projected_read_field_ids()?;
let plan = match
super::time_travel::resolve_snapshot(scan.table).await? {
Some(snapshot) => {
@@ -1567,6 +1590,20 @@ impl<'a> PaimonTableScan<'a> {
limit: scan.limit,
..Default::default()
};
+ if let Some((start, end)) = scan
+ .table
+ .schema()
+ .core_options()
+ .incremental_timestamp_window()?
+ {
+ let plan = scan
+ .plan_timestamp_window(start, end, Some(&mut trace))
+ .await?;
+ trace.snapshot_id = plan.snapshot_id();
+ trace.planned_data_file_bytes = plan.planned_data_file_bytes();
+ scan.check_planned_files(&plan, grant.is_some()).await?;
+ return Ok((plan.planned(grant), trace));
+ }
let data_evolution_read_field_ids = scan.projected_read_field_ids()?;
let plan = match
super::time_travel::resolve_snapshot(scan.table).await? {
Some(snapshot) => {
@@ -1644,6 +1681,32 @@ impl<'a> PaimonTableScan<'a> {
}
}
+ async fn plan_timestamp_window(
+ &self,
+ start: i64,
+ end: i64,
+ trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<Plan> {
+ let Some((start, end)) =
super::incremental_scan::timestamp_snapshot_range(
+ &self.table.snapshot_manager(),
+ start,
+ end,
+ )
+ .await?
+ else {
+ return Ok(Plan::new(Vec::new()));
+ };
+ super::IncrementalScan::new(
+ self.table,
+ TableScan(TableScanKind::Paimon(self.clone())),
+ super::IncrementalScanMode::Auto,
+ start,
+ end,
+ )
+ .plan_combined_with_trace(trace)
+ .await
+ }
+
/// The grant predates the manifest read, so the table can have been
/// re-created in between. Also refuses stats the schema no longer covers.
async fn check_planned_files(&self, plan: &Plan, query_auth: bool) ->
crate::Result<()> {
@@ -1670,6 +1733,7 @@ impl<'a> PaimonTableScan<'a> {
let core_options = CoreOptions::new(self.table.schema().options());
// Unconditional: unrelated to query-auth.
core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+ core_options.validate_scan_options()?;
core_options.validate_data_file_path_directory()?;
// File paths and stats, not table columns: the endpoint cannot rule
on them.
@@ -2095,6 +2159,7 @@ impl<'a> PaimonTableScan<'a> {
snapshot,
snapshot.delta_manifest_list(),
data_evolution_read_field_ids.as_ref(),
+ None,
)
.await
}
@@ -2105,6 +2170,7 @@ impl<'a> PaimonTableScan<'a> {
&self,
snapshots: &[Snapshot],
end_snapshot: &Snapshot,
+ trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
self.validate_read_options()?;
let data_evolution_read_field_ids = self.projected_read_field_ids()?;
@@ -2114,7 +2180,7 @@ impl<'a> PaimonTableScan<'a> {
end_snapshot,
ManifestListSource::AppendDeltas(snapshots),
data_evolution_read_field_ids.as_ref(),
- None,
+ trace,
)
.await
}
@@ -2124,7 +2190,11 @@ impl<'a> PaimonTableScan<'a> {
/// Reuses the same split-building path as a full snapshot plan, but only
/// reads the changelog manifest list and keeps ADD entries. Snapshots
/// without a changelog list yield an empty plan.
- pub(crate) async fn plan_snapshot_changelog(&self, snapshot: &Snapshot) ->
crate::Result<Plan> {
+ pub(crate) async fn plan_snapshot_changelog(
+ &self,
+ snapshot: &Snapshot,
+ trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<Plan> {
self.validate_read_options()?;
let Some(list_name) = snapshot.changelog_manifest_list() else {
return Ok(Plan::new(Vec::new()).with_snapshot_id(snapshot.id()));
@@ -2136,6 +2206,7 @@ impl<'a> PaimonTableScan<'a> {
snapshot,
list_name,
data_evolution_read_field_ids.as_ref(),
+ trace,
)
.await
}
@@ -2145,54 +2216,13 @@ impl<'a> PaimonTableScan<'a> {
snapshot: &Snapshot,
manifest_list_name: &str,
data_evolution_read_field_ids: Option<&HashSet<i32>>,
+ trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
- if matches!(self.limit, Some(0)) {
- return Ok(Plan::new(Vec::new()).with_snapshot_id(snapshot.id()));
- }
- let core_options = CoreOptions::new(self.table.schema().options());
- let data_evolution_enabled = core_options.data_evolution_enabled();
- let global_index_settings =
- self.global_index_scan_settings(&core_options,
data_evolution_enabled)?;
- let index_entries = self
- .read_index_manifest_entries(
- snapshot,
- global_index_settings.is_some(),
- core_options.deletion_vectors_enabled() &&
!self.is_streaming(),
- )
- .await?;
- let manifest_row_ranges = self
- .manifest_row_ranges(snapshot, index_entries.as_deref(),
global_index_settings)
- .await?;
- if manifest_row_ranges.as_ref().is_some_and(Vec::is_empty) {
- return Ok(Plan::new(Vec::new()).with_snapshot_id(snapshot.id()));
- }
- // Positional scans count all candidate row IDs. Pruning a preceding
- // manifest or file here would renumber the surviving rows; intersect
- // explicit/global-index ranges after assigning positions instead.
- let row_range_index = if data_evolution_enabled &&
self.row_position_selection().is_none() {
- manifest_row_ranges.clone().map(RowRangeIndex::create)
- } else {
- None
- };
- let entries = self
- .plan_manifest_list_entries(manifest_list_name,
row_range_index.as_ref())
- .await?;
- let effective_row_ranges = self
- .effective_row_ranges(
- snapshot,
- &entries,
- index_entries.as_deref(),
- global_index_settings,
- manifest_row_ranges,
- )
- .await?;
- self.plan_snapshot_from_entries(
- snapshot.clone(),
- entries,
+ self.plan_snapshot_from_lists(
+ snapshot,
+ ManifestListSource::IncrementalList(manifest_list_name),
data_evolution_read_field_ids,
- index_entries,
- effective_row_ranges,
- None,
+ trace,
)
.await
}
@@ -2289,132 +2319,6 @@ impl<'a> PaimonTableScan<'a> {
Ok(())
}
- /// Read entries from a single manifest list (delta or changelog) with
- /// partition / bucket filter pushdown matching the full scan path.
- async fn plan_manifest_list_entries(
- &self,
- manifest_list_name: &str,
- row_range_index: Option<&RowRangeIndex>,
- ) -> crate::Result<Vec<ManifestEntry>> {
- let file_io = self.table.file_io();
- let table_path = self.table.location();
- let core_options = CoreOptions::new(self.table.schema().options());
- let has_primary_keys = !self.table.schema().primary_keys().is_empty();
- let partition_fields = self.table.schema().partition_fields();
-
- let mut manifest_metas =
- read_manifest_list(file_io, table_path, manifest_list_name).await?;
-
- if let Some(pf) = self.partition_filter.as_ref() {
- manifest_metas.retain(|meta| pf.matches_manifest(meta,
&partition_fields));
- }
- if let Some(index) = row_range_index {
- retain_manifest_row_ranges(&mut manifest_metas, index);
- }
-
- let bucket_key_fields: Vec<DataField> = if
self.bucket_predicate.is_none() {
- Vec::new()
- } else {
- let bucket_keys = core_options.bucket_key().unwrap_or_else(|| {
- if has_primary_keys {
- self.table.schema().trimmed_primary_keys()
- } else {
- Vec::new()
- }
- });
- bucket_keys
- .iter()
- .filter_map(|key| {
- self.table
- .schema()
- .fields()
- .iter()
- .find(|f| f.name() == key)
- .cloned()
- })
- .collect::<Vec<_>>()
- };
- let bucket_function_type = core_options.bucket_function_type()?;
-
- retain_manifest_buckets(
- &mut manifest_metas,
- has_primary_keys && !self.scan_all_files,
- self.bucket_predicate.as_ref(),
- &bucket_key_fields,
- bucket_function_type,
- );
-
- let base_path = format!("{}/{}", table_path.trim_end_matches('/'),
MANIFEST_DIR);
- let shared_cache = SharedSchemaCache::new();
- let partition_filter = self.partition_filter.as_ref();
- let bucket_predicate = self.bucket_predicate.as_ref();
- let scan_all_files = self.scan_all_files;
-
- let mut entries = Vec::new();
- for meta in manifest_metas {
- let path = format!("{base_path}/{}", meta.file_name());
- // This path validates incremental/changelog manifests before its
- // existing post-read row-range pruning, so do not hide DELETEs at
- // the block layer.
- let bytes = read_manifest_bytes_with_sidecar(
- file_io,
- &path,
- &meta,
- core_options.manifest_sidecar_enabled(),
- ManifestSidecarPruning {
- row_range_index: None,
- partition_filter,
- partition_arity: partition_fields.len(),
- bucket_predicate,
- bucket_key_fields: &bucket_key_fields,
- bucket_function_type,
- },
- )
- .await?;
- let mut bucket_cache: HashMap<i32, Option<HashSet<i32>>> =
HashMap::new();
- let manifest_entries =
crate::spec::avro::from_manifest_bytes_filtered_shared(
- &bytes,
- &shared_cache,
- &mut |_kind, partition_bytes, bucket, total_buckets| {
- if has_primary_keys && !scan_all_files && bucket < 0 {
- return false;
- }
- if let Some(pred) = bucket_predicate {
- let targets =
bucket_cache.entry(total_buckets).or_insert_with(|| {
- compute_target_buckets(
- pred,
- &bucket_key_fields,
- bucket_function_type,
- total_buckets,
- )
- });
- if let Some(targets) = targets {
- if !targets.contains(&bucket) {
- return false;
- }
- }
- }
- if let Some(pf) = partition_filter {
- match pf.matches_entry(partition_bytes) {
- Ok(false) => return false,
- Ok(true) => {}
- Err(_) => {}
- }
- }
- true
- },
- )?;
- entries.extend(manifest_entries);
- }
- validate_incremental_entries(&entries)?;
- let entries = if let Some(index) = row_range_index {
- retain_manifest_entry_row_ranges(entries, index)
- } else {
- entries
- };
- Ok(entries)
- }
-
async fn plan_snapshot(
&self,
snapshot: Snapshot,
@@ -6927,7 +6831,7 @@ mod tests {
assert_eq!(delta.snapshot_id(), Some(1));
let changelog = builder
.new_scan()
- .plan_snapshot_changelog(&snapshot)
+ .plan_snapshot_changelog_with_trace(&snapshot, None)
.await
.unwrap();
assert!(changelog.splits().is_empty());
diff --git a/crates/paimon/tests/incremental_timestamp_scan_test.rs
b/crates/paimon/tests/incremental_timestamp_scan_test.rs
new file mode 100644
index 00000000..3d3dac33
--- /dev/null
+++ b/crates/paimon/tests/incremental_timestamp_scan_test.rs
@@ -0,0 +1,396 @@
+// 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.
+
+mod common;
+
+use arrow_array::RecordBatch;
+use common::incremental_helpers::{
+ make_batch, make_batch_with_kinds, memory_table, persist_table_schema,
setup_dirs, write_batch,
+};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, IntType, Schema, TableSchema};
+use paimon::table::{MergeAssignment, MergeSource, Table, WhenMatched,
WhenNotMatched};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+async fn table(name: &str, mode: &str, producer: &str) -> Table {
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("changelog-producer", producer);
+ if mode == "pk" {
+ schema = schema.primary_key(["id"]).option("bucket", "1");
+ } else if mode == "evolution" {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true");
+ }
+ let path = format!("memory:/timestamp/{name}");
+ let (io, table) = memory_table(&path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, &path).await;
+ persist_table_schema(&io, &path, table.schema()).await;
+ table
+}
+
+async fn set_time(table: &Table, time: i64) {
+ let snapshot = table
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let mut json = serde_json::to_value(&snapshot).unwrap();
+ json["timeMillis"] = time.into();
+ table
+ .file_io()
+ .new_output(&table.snapshot_manager().snapshot_path(snapshot.id()))
+ .unwrap()
+ .write(serde_json::to_vec(&json).unwrap().into())
+ .await
+ .unwrap();
+}
+
+async fn append(table: &Table, time: i64, ids: Vec<i32>, values: Vec<i32>) {
+ write_batch(table, &make_batch(ids, values)).await;
+ set_time(table, time).await;
+}
+
+fn window(table: &Table, start: i64, end: i64) -> Table {
+ table.copy_with_options(HashMap::from([
+ ("scan.mode".into(), "incremental".into()),
+ (
+ "incremental-between-timestamp".into(),
+ format!("{start},{end}"),
+ ),
+ ]))
+}
+
+async fn read(table: &Table, all_files: bool) -> (Option<i64>, Vec<(i32,
i32)>) {
+ let builder = table.new_read_builder();
+ let scan = builder.new_scan();
+ let scan = if all_files {
+ scan.with_scan_all_files()
+ } else {
+ scan
+ };
+ let plan = scan.plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<arrow_array::Int32Array>()
+ .unwrap();
+ rows.extend(
+ ids.values()
+ .iter()
+ .copied()
+ .zip(values.values().iter().copied()),
+ );
+ }
+ rows.sort_unstable();
+ (plan.snapshot_id(), rows)
+}
+
+#[tokio::test]
+async fn timestamp_options_select_events_and_preserve_end_snapshot() {
+ for mode in ["append", "pk", "evolution"] {
+ let table = table(mode, mode, "none").await;
+ append(&table, 100, vec![1, 2], vec![10, 20]).await;
+ append(&table, 200, vec![1, 3], vec![11, 30]).await;
+ append(&table, 200, vec![1, 4], vec![12, 40]).await;
+ let builder = table.new_write_builder().with_overwrite();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&make_batch(vec![9], vec![90]))
+ .await
+ .unwrap();
+ builder
+ .new_commit()
+ .overwrite(writer.prepare_commit().await.unwrap(), None)
+ .await
+ .unwrap();
+ writer.close().await;
+ set_time(&table, 300).await;
+ append(&table, 400, vec![5], vec![50]).await;
+ for (start, end, snapshot, expected) in [
+ (100, 200, Some(3), vec![(1, 11), (1, 12), (3, 30), (4, 40)]),
+ (200, 300, Some(4), vec![]),
+ (100, 300, Some(4), vec![(1, 11), (1, 12), (3, 30), (4, 40)]),
+ (100, 150, Some(1), vec![]),
+ (0, 100, Some(1), vec![(1, 10), (2, 20)]),
+ (200, 400, Some(5), vec![(5, 50)]),
+ (100, 100, None, vec![]),
+ (500, 600, None, vec![]),
+ (0, 50, None, vec![]),
+ ] {
+ let selected = window(&table, start, end);
+ for all_files in [false, true] {
+ let (actual_snapshot, rows) = read(&selected, all_files).await;
+ assert_eq!(actual_snapshot, snapshot, "{mode}:
({start},{end}]");
+ assert_eq!(rows, expected, "{mode}: ({start},{end}]");
+ }
+ let (plan, trace) = selected
+ .new_read_builder()
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .unwrap();
+ assert_eq!(trace.snapshot_id, snapshot);
+ assert_eq!(trace.final_splits, plan.splits().len());
+ assert_eq!(
+ trace.planned_data_file_bytes,
+ plan.planned_data_file_bytes()
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn timestamp_auto_uses_physical_changelog_and_preserves_retracts() {
+ let table = table("changelog", "pk", "input").await;
+ append(&table, 100, vec![1], vec![10]).await;
+ write_batch(
+ &table,
+ &make_batch_with_kinds(vec![1, 1, 2], vec![10, 11, 20], vec![1, 2, 0]),
+ )
+ .await;
+ set_time(&table, 200).await;
+ append(&table, 300, vec![3], vec![30]).await;
+ let selected = window(&table, 100, 300);
+ let builder = selected.new_read_builder();
+ let (plan, trace) = builder.new_scan().plan_with_trace().await.unwrap();
+ assert_eq!(plan.snapshot_id(), Some(3));
+ assert!(plan.splits().iter().all(|split| split.snapshot_id() == 3));
+ assert!(trace.manifest_entries_read >= 2);
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow_with_row_kind(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut kinds = Vec::new();
+ for batch in batches {
+ kinds.extend(
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::StringArray>()
+ .unwrap()
+ .iter()
+ .map(|kind| kind.unwrap().to_owned()),
+ );
+ }
+ kinds.sort();
+ assert_eq!(kinds, vec!["+I", "+I", "+U", "-U"]);
+ assert_eq!(
+ read(&window(&table, 100, 200), false).await,
+ (Some(2), vec![(1, 10), (1, 11), (2, 20)])
+ );
+ assert_eq!(read(&table, false).await.1, vec![(1, 11), (2, 20), (3, 30)]);
+}
+
+#[tokio::test]
+async fn timestamp_windows_are_validated_before_reading_current_data() {
+ let table = table("validation", "append", "none").await;
+ append(&table, 100, vec![1], vec![10]).await;
+ for options in [
+ HashMap::from([("incremental-between-timestamp".into(),
"200,100".into())]),
+ HashMap::from([("incremental-between-timestamp".into(),
"one,200".into())]),
+ HashMap::from([("incremental-between-timestamp".into(),
"100,200,300".into())]),
+ HashMap::from([("scan.mode".into(), "incremental".into())]),
+ HashMap::from([
+ ("incremental-between-timestamp".into(), "0,200".into()),
+ ("scan.snapshot-id".into(), "1".into()),
+ ]),
+ HashMap::from([
+ ("incremental-between-timestamp".into(), "0,200".into()),
+ ("scan.mode".into(), "from-timestamp".into()),
+ ]),
+ ] {
+ let selected = table.copy_with_options(options);
+ let error = selected
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap_err();
+ assert!(matches!(error, paimon::Error::DataInvalid { .. }));
+ }
+ let selected = table
+ .copy_with_time_travel(HashMap::from([(
+ "incremental-between-timestamp".into(),
+ "0,100".into(),
+ )]))
+ .await
+ .unwrap();
+ assert_eq!(read(&selected, false).await.1, vec![(1, 10)]);
+}
+
+#[tokio::test]
+async fn empty_and_equal_windows_do_not_select_latest_data() {
+ let table = table("empty", "append", "none").await;
+ for (start, end) in [(100, 100), (0, 100), (200, 100)] {
+ assert_eq!(
+ read(&window(&table, start, end), false).await,
+ (None, vec![])
+ );
+ }
+ append(&table, 100, vec![1], vec![10]).await;
+ assert_eq!(read(&window(&table, 100, 100), false).await, (None, vec![]));
+ let selected = window(&table, 100, 100).copy_with_options(HashMap::from([(
+ "query-auth.enabled".into(),
+ "true".into(),
+ )]));
+ assert!(selected.new_read_builder().new_scan().plan().await.is_err());
+}
+
+#[tokio::test]
+async fn
core_merge_reads_only_the_timestamp_window_and_keeps_source_independent() {
+ let target = table("merge_target", "evolution", "none").await;
+ let source = table("merge_source", "append", "none").await;
+ append(&target, 100, vec![1], vec![10]).await;
+ append(&source, 100, vec![1], vec![-1]).await;
+ append(&source, 200, vec![1, 2], vec![11, 20]).await;
+ append(&source, 300, vec![3], vec![30]).await;
+ let builder = target.new_write_builder();
+ let messages = builder
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Table(Arc::new(window(&source, 100, 200))),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: vec![
+ ("id".into(), MergeAssignment::SourceColumn("id".into())),
+ (
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ ),
+ ],
+ }],
+ )
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+ assert_eq!(read(&target, false).await.1, vec![(1, 11), (2, 20)]);
+ assert_eq!(
+ read(&source, false).await.1,
+ vec![(1, -1), (1, 11), (2, 20), (3, 30)]
+ );
+}
+
+#[tokio::test]
+async fn expired_prefix_is_excluded_from_timestamp_history() {
+ let table = table("expired_prefix", "append", "none").await;
+ for (time, id) in [(100, 1), (200, 2), (300, 3)] {
+ append(&table, time, vec![id], vec![id * 10]).await;
+ }
+ table
+ .file_io()
+ .delete_file(&table.snapshot_manager().snapshot_path(1))
+ .await
+ .unwrap();
+ assert_eq!(
+ read(&window(&table, 0, 250), false).await,
+ (Some(2), vec![(2, 20)])
+ );
+}
+
+#[tokio::test]
+async fn invalid_source_window_fails_before_target_file_staging() {
+ let target = table("failed_merge_target", "evolution", "none").await;
+ let source = table("failed_merge_source", "append", "none").await;
+ append(&target, 100, vec![1], vec![10]).await;
+ append(&source, 100, vec![1], vec![11]).await;
+ let builder = target.new_read_builder();
+ let plan = builder.new_scan().plan().await.unwrap();
+ let bucket_path = plan.splits()[0].bucket_path();
+ let before_files = target
+ .file_io()
+ .list_status(bucket_path)
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|s| s.path)
+ .collect::<std::collections::HashSet<_>>();
+ let error = target
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Table(Arc::new(window(&source, 200, 100))),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }],
+ Vec::new(),
+ )
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(error, paimon::Error::DataInvalid { message, .. } if
message.contains("Ending timestamp"))
+ );
+ assert_eq!(
+ target
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(1)
+ );
+ assert_eq!(
+ target
+ .file_io()
+ .list_status(bucket_path)
+ .await
+ .unwrap()
+ .into_iter()
+ .map(|s| s.path)
+ .collect::<std::collections::HashSet<_>>(),
+ before_files
+ );
+}
