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 44cda804 [Core] Preserve captured snapshot metadata and empty read
views (#1098)
44cda804 is described below
commit 44cda80452c9624ba4af6bb174a2bbaed1349960
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 10:15:05 2026 +0800
[Core] Preserve captured snapshot metadata and empty read views (#1098)
---
.../python/python/pypaimon_rust/datafusion.pyi | 3 +
bindings/python/src/table.rs | 12 ++
.../datafusion/src/partition_count_pushdown.rs | 13 +-
.../datafusion/src/system_tables/manifests.rs | 9 +-
.../datafusion/src/system_tables/mod.rs | 94 +++++++++++++
.../datafusion/src/system_tables/partitions.rs | 2 +-
.../datafusion/src/system_tables/table_indexes.rs | 9 +-
crates/integrations/datafusion/src/table/mod.rs | 2 +-
.../datafusion/tests/partition_count_pushdown.rs | 32 +++++
crates/paimon/src/catalog/partition_listing.rs | 9 +-
crates/paimon/src/table/branch_write_tests.rs | 2 +-
crates/paimon/src/table/data_evolution_writer.rs | 4 +-
crates/paimon/src/table/de_vector_scan.rs | 2 +-
crates/paimon/src/table/de_vector_scan/tests.rs | 71 ++++++++++
.../src/table/full_text_search_builder/scan.rs | 2 +-
crates/paimon/src/table/hybrid_search_builder.rs | 4 +-
crates/paimon/src/table/mod.rs | 83 +++++++++--
crates/paimon/src/table/pk_vector_scan.rs | 151 ++++++++++++++++-----
crates/paimon/src/table/table_merge_input.rs | 13 +-
crates/paimon/src/table/table_merge_into.rs | 14 +-
crates/paimon/src/table/table_scan.rs | 3 +
crates/paimon/src/table/table_update_predicate.rs | 10 +-
crates/paimon/src/table/time_travel.rs | 130 ++++++++++++++++--
crates/paimon/src/table/vector_search_result.rs | 2 +-
24 files changed, 557 insertions(+), 119 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 15ca03ce..a0c81bfa 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -214,6 +214,9 @@ class Table:
def copy_with_resolved_schema(self, schema_json: str, *, branch:
Optional[str] = None) -> "Table":
"""Replace all fields/options, preserving FileIO, REST context and
branch."""
...
+ def copy_with_pinned_snapshot(self, snapshot_json: Optional[str]) ->
"Table":
+ """Preserve the read schema and pin Java Snapshot JSON, or an empty
view with None."""
+ ...
def identifier(self) -> str: ...
def branch(self) -> str: ...
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 2c33c4b0..6eb49675 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -206,6 +206,18 @@ impl PyTable {
crate::hybrid_search::PyHybridSearchBuilder::new(Arc::clone(&self.inner))
}
+ /// Forward one captured Java-format snapshot, or an explicitly empty view.
+ #[pyo3(signature = (snapshot_json))]
+ fn copy_with_pinned_snapshot(&self, snapshot_json: Option<&str>) ->
PyResult<Self> {
+ let snapshot = snapshot_json
+ .map(serde_json::from_str::<paimon::spec::Snapshot>)
+ .transpose()
+ .map_err(|error| PyValueError::new_err(format!("Invalid snapshot
JSON: {error}")))?;
+ Ok(Self {
+ inner:
Arc::new(self.inner.copy_with_pinned_snapshot(snapshot.as_ref())),
+ })
+ }
+
fn new_vector_search_builder(&self) ->
crate::vector_search::PyVectorSearchBuilder {
crate::vector_search::PyVectorSearchBuilder::new(Arc::clone(&self.inner))
}
diff --git a/crates/integrations/datafusion/src/partition_count_pushdown.rs
b/crates/integrations/datafusion/src/partition_count_pushdown.rs
index 0dc73a53..f7fc0cbc 100644
--- a/crates/integrations/datafusion/src/partition_count_pushdown.rs
+++ b/crates/integrations/datafusion/src/partition_count_pushdown.rs
@@ -308,19 +308,12 @@ impl TableProvider for PartitionRowCountProvider {
return Ok(Some(table));
}
let selected =
table.copy_with_time_travel_strict(HashMap::new()).await?;
- let snapshot = match selected.travel_snapshot() {
- Some(snapshot) => snapshot.clone(),
- None => {
- let Some(snapshot) =
table.snapshot_manager().get_latest_snapshot().await?
- else {
- return Ok(None);
- };
- snapshot
- }
+ let Some(snapshot) = selected.resolve_read_snapshot().await? else {
+ return Ok(None);
};
// Pin data, not the snapshot's schema: DDL may have changed field
// positions since the latest data commit or since logical
planning.
- Ok(Some(table.copy_with_pinned_snapshot(&snapshot)))
+ Ok(Some(table.copy_with_pinned_snapshot(Some(&snapshot))))
})
.await
.map_err(to_datafusion_error)?;
diff --git a/crates/integrations/datafusion/src/system_tables/manifests.rs
b/crates/integrations/datafusion/src/system_tables/manifests.rs
index 9380b316..1acf7a52 100644
--- a/crates/integrations/datafusion/src/system_tables/manifests.rs
+++ b/crates/integrations/datafusion/src/system_tables/manifests.rs
@@ -160,15 +160,10 @@ impl TableProvider for ManifestsTable {
async fn collect_manifests(table: &Table) ->
paimon::Result<Vec<ManifestFileMeta>> {
let file_io = table.file_io();
- let snapshot_sm = table.snapshot_manager();
let manifest_sm =
paimon::table::SnapshotManager::new(file_io.clone(),
table.location().to_string());
- let snapshot = match table.travel_snapshot().cloned() {
- Some(snapshot) => snapshot,
- None => match snapshot_sm.get_latest_snapshot().await? {
- Some(snapshot) => snapshot,
- None => return Ok(Vec::new()),
- },
+ let Some(snapshot) = table.resolve_read_snapshot().await? else {
+ return Ok(Vec::new());
};
let base_path = manifest_sm.manifest_path(snapshot.base_manifest_list());
diff --git a/crates/integrations/datafusion/src/system_tables/mod.rs
b/crates/integrations/datafusion/src/system_tables/mod.rs
index f3fd36f9..7cbafb44 100644
--- a/crates/integrations/datafusion/src/system_tables/mod.rs
+++ b/crates/integrations/datafusion/src/system_tables/mod.rs
@@ -221,6 +221,100 @@ pub(crate) async fn load(
mod tests {
use super::{is_registered, parse_object_name_for_datafusion,
SYSTEM_TABLE_NAMES, TABLES};
+ #[tokio::test]
+ async fn pinned_empty_view_has_no_manifest_index_or_partition_rows() {
+ use std::sync::Arc;
+
+ use datafusion::arrow::array::{Int32Array, RecordBatch};
+ use datafusion::execution::context::SessionContext;
+ use paimon::catalog::Identifier;
+ use paimon::spec::{DataType, IntType, Schema};
+ use paimon::table::TableCommit;
+ use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
+
+ let temp = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set(
+ CatalogOptions::WAREHOUSE,
+ format!("file://{}", temp.path().display()),
+ );
+ let catalog = Arc::new(FileSystemCatalog::new(options).unwrap());
+ catalog
+ .create_database("db", true, Default::default())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db", "indexed");
+ catalog
+ .create_table(
+ &identifier,
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("pt", DataType::Int(IntType::new()))
+ .partition_keys(["pt"])
+ .option("row-tracking.enabled", "true")
+ .option("data-evolution.enabled", "true")
+ .option("global-index.enabled", "true")
+ .build()
+ .unwrap(),
+ false,
+ )
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ let batch = RecordBatch::try_from_iter(vec![
+ ("id", Arc::new(Int32Array::from(vec![1, 2])) as _),
+ ("pt", Arc::new(Int32Array::from(vec![0, 1])) as _),
+ ])
+ .unwrap();
+ let builder =
table.new_write_builder().with_commit_user("test").unwrap();
+ let mut writer = builder.new_write().unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "test".to_string())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ table
+ .new_global_index_build_builder()
+ .with_index_column("id")
+ .with_index_type("btree")
+ .execute()
+ .await
+ .unwrap();
+
+ let ctx = SessionContext::new();
+ for (name, builder) in [
+ ("manifests", super::manifests::build as super::Builder),
+ (
+ "table_indexes",
+ super::table_indexes::build as super::Builder,
+ ),
+ ] {
+ for empty in [false, true] {
+ let view = if empty {
+ table.copy_with_pinned_snapshot(None)
+ } else {
+ table.clone()
+ };
+ let provider = builder(view).unwrap();
+ let batches =
ctx.read_table(provider).unwrap().collect().await.unwrap();
+ let count: usize =
batches.iter().map(RecordBatch::num_rows).sum();
+ assert_eq!(count == 0, empty, "{name}, empty={empty}");
+ }
+ }
+ for empty in [false, true] {
+ let view = if empty {
+ table.copy_with_pinned_snapshot(None)
+ } else {
+ table.clone()
+ };
+ let provider =
+ super::partitions::build(catalog.clone(), identifier.clone(),
view).unwrap();
+ let batches =
ctx.read_table(provider).unwrap().collect().await.unwrap();
+ let count: usize = batches.iter().map(RecordBatch::num_rows).sum();
+ assert_eq!(count, if empty { 0 } else { 2 });
+ }
+ }
+
/// Guards against the two registries drifting: anything in `TABLES` must
/// also be in `SYSTEM_TABLE_NAMES`, and the only name allowed to be in
/// `SYSTEM_TABLE_NAMES` but not `TABLES` is `partitions` (routed via the
diff --git a/crates/integrations/datafusion/src/system_tables/partitions.rs
b/crates/integrations/datafusion/src/system_tables/partitions.rs
index 749bb282..b87db4f8 100644
--- a/crates/integrations/datafusion/src/system_tables/partitions.rs
+++ b/crates/integrations/datafusion/src/system_tables/partitions.rs
@@ -122,7 +122,7 @@ impl TableProvider for PartitionsTable {
_limit: Option<usize>,
) -> DFResult<Arc<dyn ExecutionPlan>> {
let table = self.table.clone();
- let partitions = if table.travel_snapshot().is_some() {
+ let partitions = if table.travel_snapshot().is_some() ||
table.is_time_traveled() {
crate::runtime::await_with_runtime(async move {
list_partitions_from_file_system(&table).await
})
diff --git a/crates/integrations/datafusion/src/system_tables/table_indexes.rs
b/crates/integrations/datafusion/src/system_tables/table_indexes.rs
index cbd1c2c1..c51d3e55 100644
--- a/crates/integrations/datafusion/src/system_tables/table_indexes.rs
+++ b/crates/integrations/datafusion/src/system_tables/table_indexes.rs
@@ -187,15 +187,10 @@ impl TableProvider for TableIndexesTable {
async fn collect_index_entries(table: &Table) ->
paimon::Result<Vec<IndexManifestEntry>> {
let file_io = table.file_io();
- let snapshot_sm = table.snapshot_manager();
let manifest_sm =
paimon::table::SnapshotManager::new(file_io.clone(),
table.location().to_string());
- let snapshot = match table.travel_snapshot().cloned() {
- Some(snapshot) => snapshot,
- None => match snapshot_sm.get_latest_snapshot().await? {
- Some(snapshot) => snapshot,
- None => return Ok(Vec::new()),
- },
+ let Some(snapshot) = table.resolve_read_snapshot().await? else {
+ return Ok(Vec::new());
};
let Some(index_manifest_name) = snapshot.index_manifest() else {
return Ok(Vec::new());
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index 5d04761e..0e296fcc 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -175,7 +175,7 @@ impl PaimonTableProvider {
}
pub(crate) fn with_pinned_snapshot(mut self, snapshot: &Snapshot) -> Self {
- self.table = self.table.copy_with_pinned_snapshot(snapshot);
+ self.table = self.table.copy_with_pinned_snapshot(Some(snapshot));
self
}
}
diff --git a/crates/integrations/datafusion/tests/partition_count_pushdown.rs
b/crates/integrations/datafusion/tests/partition_count_pushdown.rs
index f20674f2..0684a6bd 100644
--- a/crates/integrations/datafusion/tests/partition_count_pushdown.rs
+++ b/crates/integrations/datafusion/tests/partition_count_pushdown.rs
@@ -622,6 +622,38 @@ async fn
test_deletion_vector_zero_groups_and_unknown_cardinality_fallback() {
);
}
+#[tokio::test]
+async fn test_pinned_empty_view_is_used_by_count_pushdown_and_row_scan() {
+ let (_tmp, catalog, ctx) = setup().await;
+ let table = catalog
+ .get_table(&Identifier::new("test_db", "t"))
+ .await
+ .unwrap();
+ let pinned = table.copy_with_pinned_snapshot(None);
+ ctx.register_temp_table(
+ "paimon.test_db.pinned",
+
Arc::new(paimon_datafusion::PaimonTableProvider::try_new(pinned).unwrap()),
+ )
+ .unwrap();
+ assert_eq!(
+ rows(&ctx, "SELECT COUNT(*) FROM paimon.test_db.pinned").await,
+ vec![row("", 0)]
+ );
+ assert_eq!(
+ rows(&ctx, "SELECT COUNT(id) FROM paimon.test_db.pinned").await,
+ vec![row("", 0)]
+ );
+ let grouped = "SELECT dt, COUNT(*) FROM paimon.test_db.pinned GROUP BY dt";
+ assert!(!scans_table(&ctx, grouped).await);
+ assert!(rows(&ctx, grouped).await.is_empty());
+ assert!(!rows(
+ &ctx,
+ "SELECT dt, COUNT(*) FROM paimon.test_db.t GROUP BY dt"
+ )
+ .await
+ .is_empty());
+}
+
#[tokio::test]
async fn test_ungrouped_count_on_unpartitioned_and_empty_tables() {
let (_tmp, _catalog, ctx) = setup().await;
diff --git a/crates/paimon/src/catalog/partition_listing.rs
b/crates/paimon/src/catalog/partition_listing.rs
index 9bb28d30..dec7702b 100644
--- a/crates/paimon/src/catalog/partition_listing.rs
+++ b/crates/paimon/src/catalog/partition_listing.rs
@@ -35,14 +35,9 @@ pub async fn list_partitions_from_file_system(table: &Table)
-> Result<Vec<Parti
// Manifests carry partition values and per-column stats.
crate::spec::CoreOptions::new(table.schema().options()).ensure_read_authorized()?;
let file_io = table.file_io();
- let snapshot_sm = table.snapshot_manager();
let manifest_sm = SnapshotManager::new(file_io.clone(),
table.location().to_string());
- let snapshot = match table.travel_snapshot().cloned() {
- Some(snapshot) => snapshot,
- None => match snapshot_sm.get_latest_snapshot().await? {
- Some(snapshot) => snapshot,
- None => return Ok(Vec::new()),
- },
+ let Some(snapshot) = table.resolve_read_snapshot().await? else {
+ return Ok(Vec::new());
};
let base_path = manifest_sm.manifest_path(snapshot.base_manifest_list());
diff --git a/crates/paimon/src/table/branch_write_tests.rs
b/crates/paimon/src/table/branch_write_tests.rs
index 283705dd..612df694 100644
--- a/crates/paimon/src/table/branch_write_tests.rs
+++ b/crates/paimon/src/table/branch_write_tests.rs
@@ -467,7 +467,7 @@ async fn branch_write_historical_handles_remain_read_only()
{
append(&main, "seed", batch(vec![1], vec![10])).await;
let branch = tagged_branch(&main).await;
let snapshot = latest(&branch).await;
- let mut historical = branch.copy_with_pinned_snapshot(&snapshot);
+ let mut historical = branch.copy_with_pinned_snapshot(Some(&snapshot));
// The pinned snapshot remains read-only even with no selector in its
options.
historical.schema = historical
.schema
diff --git a/crates/paimon/src/table/data_evolution_writer.rs
b/crates/paimon/src/table/data_evolution_writer.rs
index 6c6a4bcd..05bf3ee5 100644
--- a/crates/paimon/src/table/data_evolution_writer.rs
+++ b/crates/paimon/src/table/data_evolution_writer.rs
@@ -611,7 +611,7 @@ impl DataEvolutionDeleteWriter {
message: "No files with row tracking found in target
table".into(),
source: None,
})?;
- let scan_table = self.table.copy_with_pinned_snapshot(&snapshot);
+ let scan_table = self.table.copy_with_pinned_snapshot(Some(&snapshot));
let scan = scan_table
.new_read_builder()
.new_scan()
@@ -1256,7 +1256,7 @@ impl RowIdFileIndex {
files: Vec::new(),
});
};
- let read_table = table.copy_with_pinned_snapshot(&snapshot);
+ let read_table = table.copy_with_pinned_snapshot(Some(&snapshot));
let plan = read_table.new_read_builder().new_scan().plan().await?;
Self::from_splits(read_table, plan.splits())
}
diff --git a/crates/paimon/src/table/de_vector_scan.rs
b/crates/paimon/src/table/de_vector_scan.rs
index 883dd077..1b2af332 100644
--- a/crates/paimon/src/table/de_vector_scan.rs
+++ b/crates/paimon/src/table/de_vector_scan.rs
@@ -63,7 +63,7 @@ impl Table {
CoreOptions::new(self.schema().options()).ensure_read_authorized()?;
let Some(snapshot) =
crate::table::time_travel::resolve_snapshot(self).await? else {
return Ok(PreparedVectorSearchFilter {
- table: self.clone(),
+ table: self.copy_with_pinned_snapshot(None),
include_row_ids: Arc::new(RoaringTreemap::new()),
});
};
diff --git a/crates/paimon/src/table/de_vector_scan/tests.rs
b/crates/paimon/src/table/de_vector_scan/tests.rs
index 5033ed9b..dc35c06f 100644
--- a/crates/paimon/src/table/de_vector_scan/tests.rs
+++ b/crates/paimon/src/table/de_vector_scan/tests.rs
@@ -32,6 +32,77 @@ use roaring::RoaringTreemap;
use std::collections::HashMap;
use std::sync::Arc;
+#[tokio::test]
+async fn prepared_empty_filter_keeps_its_read_table_empty_after_commit() {
+ let table = vector_test_table().copy_with_options(HashMap::from([
+ ("row-tracking.enabled".to_string(), "true".to_string()),
+ ("data-evolution.enabled".to_string(), "true".to_string()),
+ ("vector-index.search-mode".to_string(), "full".to_string()),
+ ]));
+ let prepared = table
+ .prepare_vector_search_filter(id_gt_filter(&table, 0))
+ .await
+ .unwrap();
+ let field = Arc::new(ArrowField::new("element", ArrowDataType::Float32,
true));
+ let mut vectors =
ListBuilder::new(Float32Builder::new()).with_field(field.clone());
+ vectors.values().append_value(1.0);
+ vectors.values().append_value(0.0);
+ vectors.append(true);
+ let batch = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("embedding", ArrowDataType::List(field), true),
+ ])),
+ vec![
+ Arc::new(Int32Array::from(vec![1])),
+ Arc::new(vectors.finish()),
+ ],
+ )
+ .unwrap();
+ let mut writer = TableWrite::new(&table, "test".to_string()).unwrap();
+ writer.write_arrow_batch(&batch).await.unwrap();
+ TableCommit::new(table.clone(), "test".to_string())
+ .commit(writer.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+
+ assert!(table.resolve_read_snapshot().await.unwrap().is_some());
+ assert!(prepared
+ .table()
+ .resolve_read_snapshot()
+ .await
+ .unwrap()
+ .is_none());
+ assert!(prepared
+ .table()
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap()
+ .splits()
+ .is_empty());
+ assert!(prepared
+ .table()
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1.0, 0.0])
+ .with_limit(2)
+ .execute()
+ .await
+ .unwrap()
+ .is_empty());
+ assert!(!table
+ .new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector(vec![1.0, 0.0])
+ .with_limit(2)
+ .execute()
+ .await
+ .unwrap()
+ .is_empty());
+}
+
#[tokio::test]
async fn reader_uses_planned_snapshot_after_snapshot_files_are_removed() {
let table = de_vector_table().await;
diff --git a/crates/paimon/src/table/full_text_search_builder/scan.rs
b/crates/paimon/src/table/full_text_search_builder/scan.rs
index 8e87a3cd..68e94f22 100644
--- a/crates/paimon/src/table/full_text_search_builder/scan.rs
+++ b/crates/paimon/src/table/full_text_search_builder/scan.rs
@@ -164,7 +164,7 @@ impl FullTextScan {
plan.next_row_id = snapshot.next_row_id();
// Pin the data without changing the schema already bound by the query.
// A schema-only rename does not create a new data snapshot.
- plan.table = self.table.copy_with_pinned_snapshot(&snapshot);
+ plan.table = self.table.copy_with_pinned_snapshot(Some(&snapshot));
if let Some(name) = snapshot.index_manifest() {
let path = self.table.snapshot_manager().manifest_path(name);
plan.entries = IndexManifest::read(self.table.file_io(),
&path).await?;
diff --git a/crates/paimon/src/table/hybrid_search_builder.rs
b/crates/paimon/src/table/hybrid_search_builder.rs
index e8838214..d6edb496 100644
--- a/crates/paimon/src/table/hybrid_search_builder.rs
+++ b/crates/paimon/src/table/hybrid_search_builder.rs
@@ -629,11 +629,11 @@ impl<'a> HybridSearchBuilder<'a> {
if let Some(snapshot) = &self.pinned_snapshot {
return Ok(snapshot
.as_ref()
- .map(|snapshot|
self.table.copy_with_pinned_snapshot(snapshot)));
+ .map(|snapshot|
self.table.copy_with_pinned_snapshot(Some(snapshot))));
}
Ok(super::time_travel::resolve_snapshot(self.table)
.await?
- .map(|snapshot| self.table.copy_with_pinned_snapshot(&snapshot)))
+ .map(|snapshot|
self.table.copy_with_pinned_snapshot(Some(&snapshot))))
}
/// Consume the vector search's scored positions, retaining its source
files
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 11a902ed..ce1b01f9 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -246,6 +246,23 @@ use crate::spec::{
SCAN_WATERMARK_OPTION,
};
use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+/// A captured read view. Absence of this enum means that the view is
unresolved.
+#[derive(Debug, Clone)]
+enum ReadSnapshot {
+ Empty,
+ Snapshot(Arc<Snapshot>),
+}
+
+impl ReadSnapshot {
+ fn snapshot(&self) -> Option<&Snapshot> {
+ match self {
+ Self::Empty => None,
+ Self::Snapshot(snapshot) => Some(snapshot),
+ }
+ }
+}
/// Table represents a table in the catalog.
#[derive(Debug, Clone)]
@@ -264,10 +281,9 @@ pub struct Table {
/// 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.
+ /// Captured snapshot or empty view, so scans don't resolve latest again.
/// Cleared when [`Table::copy_with_options`] changes the selector.
- travel_snapshot: Option<Snapshot>,
+ travel_snapshot: Option<ReadSnapshot>,
/// Explicit options remain overrides across later time-travel copies.
applied_dynamic_option_keys: HashSet<String>,
}
@@ -594,6 +610,8 @@ impl Table {
|| k == crate::spec::SCAN_WATERMARK_OPTION
|| k == crate::spec::SCAN_SNAPSHOT_ID_OPTION
|| k == crate::spec::SCAN_TAG_NAME_OPTION
+ || k == "scan.mode"
+ || k == "incremental-between-timestamp"
});
Self {
file_io: self.file_io.clone(),
@@ -653,7 +671,7 @@ impl Table {
/// same snapshot. The snapshot's schema is loaded when it differs from the
/// current table schema.
pub(crate) async fn copy_with_resolved_snapshot(&self, snapshot:
&Snapshot) -> Result<Self> {
- let mut table = self.copy_with_pinned_snapshot(snapshot);
+ let mut table = self.copy_with_pinned_snapshot(Some(snapshot));
if snapshot.schema_id() != self.schema.id() {
let schema =
self.schema_manager.schema(snapshot.schema_id()).await?;
let options = table.historical_field_options(&schema);
@@ -667,7 +685,8 @@ impl Table {
///
/// Unlike time travel, pinning must not change the fields used by an
already
/// planned query. The resolved snapshot is cached without additional I/O.
- pub fn copy_with_pinned_snapshot(&self, snapshot: &Snapshot) -> Self {
+ /// `None` captures an empty view, which remains empty after later commits.
+ pub fn copy_with_pinned_snapshot(&self, snapshot: Option<&Snapshot>) ->
Self {
let mut options = self.schema.options().clone();
for selector in [
SCAN_TIMESTAMP_MILLIS_OPTION,
@@ -676,18 +695,35 @@ impl Table {
SCAN_VERSION_OPTION,
SCAN_SNAPSHOT_ID_OPTION,
SCAN_TAG_NAME_OPTION,
+ "incremental-between-timestamp",
+ "scan.file-creation-time-millis",
+ "scan.creation-time-millis",
] {
options.remove(selector);
}
options.insert(
- SCAN_SNAPSHOT_ID_OPTION.to_string(),
- snapshot.id().to_string(),
+ "scan.mode".to_string(),
+ if snapshot.is_some() {
+ "from-snapshot"
+ } else {
+ "default"
+ }
+ .to_string(),
);
+ if let Some(snapshot) = snapshot {
+ options.insert(
+ SCAN_SNAPSHOT_ID_OPTION.to_string(),
+ snapshot.id().to_string(),
+ );
+ }
Self {
schema: self.schema.copy_with_replaced_options(options),
time_traveled: true,
- travel_snapshot: Some(snapshot.clone()),
+ travel_snapshot: Some(match snapshot {
+ Some(snapshot) =>
ReadSnapshot::Snapshot(Arc::new(snapshot.clone())),
+ None => ReadSnapshot::Empty,
+ }),
..self.clone()
}
}
@@ -807,7 +843,7 @@ impl Table {
table.schema =
snapshot_schema.copy_with_replaced_options(options);
table.time_traveled = true;
}
- table.travel_snapshot = Some(snapshot);
+ table.travel_snapshot =
Some(ReadSnapshot::Snapshot(Arc::new(snapshot)));
}
Ok(table)
}
@@ -911,8 +947,8 @@ impl Table {
})
}
- /// Whether this table copy reads a historical snapshot with its
- /// historical schema (see [`Table::copy_with_time_travel`]).
+ /// Whether this copy is read-only because its schema was time-traveled or
+ /// its read view was explicitly pinned (including an empty view).
pub fn is_time_traveled(&self) -> bool {
self.time_traveled
}
@@ -922,13 +958,34 @@ impl Table {
/// "selector set but unresolved" (silent fallback to latest) from a real
/// travelled read, so they can reject the former instead of reading
latest.
pub fn has_resolved_travel_snapshot(&self) -> bool {
- self.travel_snapshot.is_some()
+ self.travel_snapshot().is_some()
}
/// The snapshot resolved by [`Table::copy_with_time_travel`] from this
/// copy's options, if any. Lets scans skip re-resolving the selector.
pub fn travel_snapshot(&self) -> Option<&Snapshot> {
- self.travel_snapshot.as_ref()
+ self.travel_snapshot
+ .as_ref()
+ .and_then(ReadSnapshot::snapshot)
+ }
+
+ /// Resolve the selected read view, reusing complete cached metadata or an
+ /// explicitly empty view. Read latest only when no view has been captured.
+ pub async fn resolve_read_snapshot(&self) -> Result<Option<Snapshot>> {
+ time_travel::resolve_snapshot(self).await
+ }
+
+ /// Reject a captured schema whose read selection was subsequently changed.
+ pub(crate) fn ensure_read_snapshot_current(&self) -> Result<()> {
+ if self.is_time_traveled() && self.travel_snapshot.is_none() {
+ 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,
+ });
+ }
+ Ok(())
}
}
diff --git a/crates/paimon/src/table/pk_vector_scan.rs
b/crates/paimon/src/table/pk_vector_scan.rs
index 36f460ea..3a17c807 100644
--- a/crates/paimon/src/table/pk_vector_scan.rs
+++ b/crates/paimon/src/table/pk_vector_scan.rs
@@ -304,48 +304,55 @@ impl Scan for PkVectorScan {
async fn plan(&self) -> crate::Result<PkVectorScanPlan> {
let snapshot_manager = self.table.snapshot_manager();
- // Data splits first, via the table's own scan resolution (which honors
- // time travel / scan.snapshot-id). Deriving the snapshot from the
scan's
- // own output — rather than resolving `get_latest_snapshot()`
separately —
- // keeps the index manifest and the data splits on ONE snapshot,
matching
- // Java `PrimaryKeyVectorScan` (resolve one snapshot up front, read
data and
- // index from it). It also avoids a time-travel mismatch (data from the
- // travelled snapshot, index from latest) and a TOCTOU where a
concurrent
- // commit lands between two independent resolutions.
- //
- // The residual scalar filter, when set, is pushed into the read
builder so
- // scan planning drops files whose stats cannot match the predicate,
mirroring
- // Java `PrimaryKeyVectorScan` applying the filter at scan time. Files
that
- // survive are still residual-filtered per row downstream; this only
avoids
- // re-reading files the predicate already excludes.
- let mut read_builder = self.table.new_read_builder();
+ let options = self.table.schema().core_options();
+
options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
+ options.ensure_read_authorized()?;
+ options.validate_scan_options()?;
+ options.validate_data_file_path_directory()?;
+ self.table.ensure_read_snapshot_current()?;
+ // Point reads reuse one complete snapshot as Java PrimaryKeyVectorScan
+ // does, including tags whose original snapshot JSON has expired.
+ let incremental = options.incremental_timestamp_window()?.is_some();
+ let snapshot = if incremental {
+ None
+ } else {
+ self.table.resolve_read_snapshot().await?
+ };
+ let table = if incremental {
+ self.table.clone()
+ } else {
+ self.table.copy_with_pinned_snapshot(snapshot.as_ref())
+ };
+ // Residual filtering prunes files at scan time and checks each
+ // surviving candidate downstream.
+ let mut read_builder = table.new_read_builder();
if let Some(filter) = &self.filter {
read_builder.with_filter(filter.clone());
}
- // Plan the data splits and capture the snapshot the scan pinned in one
- // pass. The trace carries the resolved snapshot id even when the scan
- // yields zero data splits, so the plan reports its real snapshot id
- // (required by the cross-route snapshot-consistency guard) instead of
- // deriving it from a first split that may not exist.
- let (data_plan, trace) = read_builder
- .new_scan()
- .with_scan_all_files()
- .plan_with_trace()
- .await?;
+ let data_plan =
read_builder.new_scan().with_scan_all_files().plan().await?;
let data_splits = data_plan.splits().to_vec();
-
- // No snapshot at all (table never written): nothing to search and no
- // snapshot to pin. The empty split list makes every downstream
consumer
- // treat this as "no candidates", so this is the only plan without a
real
- // snapshot id.
- let Some(snapshot_id) = trace.snapshot_id else {
+ // Java accepts only a snapshot scan plan, or an empty incremental
plan.
+ // Never widen an incremental selection into a full-snapshot search.
+ if incremental && !data_splits.is_empty() {
+ return Err(data_invalid(
+ "Primary-key vector search requires a snapshot scan, not an
incremental scan",
+ ));
+ }
+ if incremental {
+ return Ok(PkVectorScanPlan {
+ snapshot_id: 0,
+ splits: Vec::new(),
+ physical_row_ranges_by_split: None,
+ });
+ }
+ let Some(snapshot) = snapshot else {
return Ok(PkVectorScanPlan {
snapshot_id: 0,
splits: Vec::new(),
physical_row_ranges_by_split: None,
});
};
- let snapshot = snapshot_manager.get_snapshot(snapshot_id).await?;
+ let snapshot_id = snapshot.id();
// Index-manifest scan into filtered ANN payload tuples.
let table_path = self.table.location().trim_end_matches('/');
@@ -1161,6 +1168,86 @@ mod tests {
.unwrap()
}
+ #[tokio::test]
+ async fn
plan_validates_original_scan_mode_and_rejects_nonempty_incremental_search() {
+ let (_tmp, table) = build_pruning_test_table(false).await;
+ for extra in [
+ HashMap::from([("scan.mode".to_string(),
"latest-full".to_string())]),
+ HashMap::from([(
+ "incremental-between-timestamp".to_string(),
+ "0,9999999999999".to_string(),
+ )]),
+ ] {
+ let view = table.copy_with_options(extra);
+ assert!(PkVectorScan::new(
+ &view,
+ prune_vector_field_id(&view),
+ PRUNE_INDEX_TYPE.to_string(),
+ None
+ )
+ .plan()
+ .await
+ .is_err());
+ }
+ let empty_window = table.copy_with_options(HashMap::from([(
+ "incremental-between-timestamp".to_string(),
+ "0,1".to_string(),
+ )]));
+ assert!(PkVectorScan::new(
+ &empty_window,
+ prune_vector_field_id(&empty_window),
+ PRUNE_INDEX_TYPE.to_string(),
+ None
+ )
+ .plan()
+ .await
+ .unwrap()
+ .splits
+ .is_empty());
+ }
+
+ #[tokio::test]
+ async fn
plan_uses_complete_retained_tag_metadata_and_explicit_empty_view() {
+ let (_tmp, table) = build_pruning_test_table(false).await;
+ let snapshot =
table.resolve_read_snapshot().await.unwrap().unwrap();
+ table.tag_manager().create("kept", &snapshot).await.unwrap();
+ let pinned = table.copy_with_pinned_snapshot(Some(&snapshot));
+ let tagged = table.copy_with_options(HashMap::from([(
+ "scan.tag-name".to_string(),
+ "kept".to_string(),
+ )]));
+ table
+ .file_io()
+
.delete_file(&table.snapshot_manager().snapshot_path(snapshot.id()))
+ .await
+ .unwrap();
+ for view in [pinned, tagged] {
+ let plan = PkVectorScan::new(
+ &view,
+ prune_vector_field_id(&view),
+ PRUNE_INDEX_TYPE.to_string(),
+ None,
+ )
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id, snapshot.id());
+ assert_eq!(plan.splits.len(), 1);
+ }
+ let empty = table.copy_with_pinned_snapshot(None);
+ let plan = PkVectorScan::new(
+ &empty,
+ prune_vector_field_id(&empty),
+ PRUNE_INDEX_TYPE.to_string(),
+ None,
+ )
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id, 0);
+ assert!(plan.splits.is_empty());
+ }
+
#[tokio::test]
async fn plan_prunes_rust_pk_file_using_key_stats() {
// Real PK table, one data file with id in [0, PRUNE_ROWS). A
predicate
diff --git a/crates/paimon/src/table/table_merge_input.rs
b/crates/paimon/src/table/table_merge_input.rs
index 67b910d6..7512718b 100644
--- a/crates/paimon/src/table/table_merge_input.rs
+++ b/crates/paimon/src/table/table_merge_input.rs
@@ -120,14 +120,11 @@ impl MergeInput {
};
// 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(
- || table.clone(),
- |snapshot| {
- table
- .copy_with_pinned_snapshot(snapshot)
- .copy_with_options(HashMap::from([("scan.mode".into(),
"default".into())]))
- },
- );
+ let table = if incremental {
+ table.clone()
+ } else {
+ table.copy_with_pinned_snapshot(snapshot.as_ref())
+ };
let mut builder = table.new_read_builder();
builder.with_projection(&names)?;
let plan = if incremental || snapshot.is_some() {
diff --git a/crates/paimon/src/table/table_merge_into.rs
b/crates/paimon/src/table/table_merge_into.rs
index a37ed906..0004e3b7 100644
--- a/crates/paimon/src/table/table_merge_into.rs
+++ b/crates/paimon/src/table/table_merge_into.rs
@@ -275,14 +275,12 @@ pub(super) async fn merge_into(
}
let _ = DataEvolutionWriter::new(table, update_columns.clone())?;
let snapshot = super::time_travel::resolve_snapshot(table).await?;
- let scan_table = match &snapshot {
- Some(snapshot) => table.copy_with_pinned_snapshot(snapshot),
- None => table.clone(),
- }
- .copy_with_options(HashMap::from([
- ("scalar-index.search-mode".into(), "FULL".into()),
- ("scan.mode".into(), "default".into()),
- ]));
+ let scan_table = table
+ .copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ "FULL".into(),
+ )]))
+ .copy_with_pinned_snapshot(snapshot.as_ref());
let self_merge = matches!(source, MergeSource::SelfTable);
if self_merge {
if on != [(ROW_ID.into(), ROW_ID.into())] || !not_matched.is_empty() {
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index dd2fc4bb..fa84bf5e 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1939,6 +1939,9 @@ impl<'a> PaimonTableScan<'a> {
core_options.ensure_type_paimon_served(&self.table.identifier().full_name())?;
core_options.validate_scan_options()?;
core_options.validate_data_file_path_directory()?;
+ // Timestamp windows bypass point-snapshot resolution but must still
+ // reject a schema whose captured selection was invalidated by a copy.
+ self.table.ensure_read_snapshot_current()?;
// File paths and stats, not table columns: the endpoint cannot rule
on them.
let query_auth =
CoreOptions::new(self.table.schema().options()).query_auth_enabled();
diff --git a/crates/paimon/src/table/table_update_predicate.rs
b/crates/paimon/src/table/table_update_predicate.rs
index 675db998..13c22bb0 100644
--- a/crates/paimon/src/table/table_update_predicate.rs
+++ b/crates/paimon/src/table/table_update_predicate.rs
@@ -114,11 +114,11 @@ pub(super) async fn update(
// Updates must include unindexed rows even when normal reads request the
// partial/detail scalar-index search mode.
let scan_table = table
- .copy_with_options(HashMap::from([
- ("scalar-index.search-mode".into(), "FULL".into()),
- ("scan.mode".into(), "default".into()),
- ]))
- .copy_with_pinned_snapshot(&snapshot);
+ .copy_with_options(HashMap::from([(
+ "scalar-index.search-mode".into(),
+ "FULL".into(),
+ )]))
+ .copy_with_pinned_snapshot(Some(&snapshot));
let mut read_builder = scan_table.new_read_builder();
let mut projection = Vec::new();
for name in &read_columns {
diff --git a/crates/paimon/src/table/time_travel.rs
b/crates/paimon/src/table/time_travel.rs
index fadb1715..e2604cef 100644
--- a/crates/paimon/src/table/time_travel.rs
+++ b/crates/paimon/src/table/time_travel.rs
@@ -134,17 +134,10 @@ async fn resolve_watermark(
/// subsequently changed is rejected instead of mixing that stale schema with a
/// different snapshot.
pub(crate) async fn resolve_snapshot(table: &Table) ->
crate::Result<Option<Snapshot>> {
- if let Some(snapshot) = table.travel_snapshot() {
- return Ok(Some(snapshot.clone()));
- }
- if table.is_time_traveled() {
- return Err(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,
- });
+ if let Some(view) = &table.travel_snapshot {
+ return Ok(view.snapshot().cloned());
}
+ table.ensure_read_snapshot_current()?;
match travel_to_snapshot(
&table.snapshot_manager(),
@@ -451,7 +444,7 @@ mod tests {
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());
+ let pinned =
traveled.copy_with_pinned_snapshot(traveled.travel_snapshot());
assert!(!pinned.schema().options().contains_key("scan.timestamp"));
}
for timestamp in ["2024-01-02 12:00:01", "2024-01-02 12:00:02"] {
@@ -513,7 +506,7 @@ mod tests {
.delete_file(&manager.snapshot_path(1))
.await
.unwrap();
- let pinned = table.copy_with_pinned_snapshot(&snapshot);
+ let pinned = table.copy_with_pinned_snapshot(Some(&snapshot));
assert_eq!(pinned.schema().id(), table.schema().id());
assert_eq!(pinned.schema().fields(), table.schema().fields());
assert_eq!(pinned.schema().options().get("custom").unwrap(), "value");
@@ -530,6 +523,119 @@ mod tests {
assert!(table.travel_snapshot().is_none());
}
+ #[tokio::test]
+ async fn test_pinning_replaces_prior_supported_scan_modes() {
+ use futures::TryStreamExt;
+ let (io, path) = setup_evolved_table().await;
+ let table = latest_table(&io, &path);
+ let snapshot = table.resolve_read_snapshot().await.unwrap().unwrap();
+ for extra in [
+ options(&[("scan.mode", "from-snapshot"), ("scan.snapshot-id",
"1")]),
+ options(&[
+ ("scan.mode", "from-timestamp"),
+ ("scan.timestamp-millis", "0"),
+ ]),
+ options(&[
+ ("scan.mode", "incremental"),
+ ("incremental-between-timestamp", "0,9999999999999"),
+ ]),
+ ] {
+ let configured = table.copy_with_options(extra);
+ for source in [None, Some(&snapshot)] {
+ let view = configured.copy_with_pinned_snapshot(source);
+ let plan =
view.new_read_builder().new_scan().plan().await.unwrap();
+ let builder = view.new_read_builder();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let rows: usize =
batches.iter().map(RecordBatch::num_rows).sum();
+ assert_eq!(rows, if source.is_some() { 5 } else { 0 });
+ assert!(!view
+ .schema()
+ .options()
+ .contains_key("incremental-between-timestamp"));
+ }
+ }
+ }
+
+ #[tokio::test]
+ async fn test_pinned_empty_view_survives_commits_and_option_copies() {
+ let io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/pinned_empty";
+ write_schema_file(&io, path, &schema_v0()).await;
+ let table = make_table(&io, path, schema_v0());
+ let pinned = table.copy_with_pinned_snapshot(None);
+ write_and_commit(&table, &batch_v0(vec![1, 2], vec![10, 20])).await;
+
+ assert_eq!(
+ table.resolve_read_snapshot().await.unwrap().unwrap().id(),
+ 1
+ );
+ for view in [
+ pinned.clone(),
+ pinned.copy_with_options(options(&[("read.batch-size", "1")])),
+ ] {
+ assert!(view.resolve_read_snapshot().await.unwrap().is_none());
+ assert!(view
+ .new_read_builder()
+ .new_scan()
+ .plan()
+ .await
+ .unwrap()
+ .splits()
+ .is_empty());
+ assert!(view.new_write_builder().new_write().is_err());
+ assert!(view.is_time_traveled());
+ assert!(!view.has_resolved_travel_snapshot());
+ assert!(crate::catalog::list_partitions_from_file_system(&view)
+ .await
+ .unwrap()
+ .is_empty());
+ assert!(view.partition_row_counts().await.unwrap().is_empty());
+ }
+ for extra in [
+ options(&[("scan.mode", "default")]),
+ options(&[("incremental-between-timestamp", "0,9999999999999")]),
+ ] {
+ let changed = pinned.copy_with_options(extra);
+ assert!(changed.resolve_read_snapshot().await.is_err());
+
assert!(changed.new_read_builder().new_scan().plan().await.is_err());
+ assert!(changed
+ .new_read_builder()
+ .new_scan()
+ .plan_with_trace()
+ .await
+ .is_err());
+ }
+ // Changing the selector must not silently reuse the empty view or the
+ // schema captured by that view. Explicitly re-resolve the pair.
+ let changed = pinned.copy_with_options(options(&[("scan.snapshot-id",
"1")]));
+ assert!(changed.resolve_read_snapshot().await.is_err());
+ let selected =
changed.copy_with_time_travel(HashMap::new()).await.unwrap();
+ assert_eq!(
+ selected
+ .resolve_read_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 1
+ );
+ let reset = pinned
+ .copy_with_resolved_schema(schema_v0(), "main")
+ .unwrap();
+ assert_eq!(
+ reset.resolve_read_snapshot().await.unwrap().unwrap().id(),
+ 1
+ );
+ assert!(!reset.is_time_traveled());
+ }
+
#[tokio::test]
async fn test_copy_with_time_travel_same_schema_still_rejects_write() {
let (file_io, table_path) = setup_evolved_table().await;
diff --git a/crates/paimon/src/table/vector_search_result.rs
b/crates/paimon/src/table/vector_search_result.rs
index b2fb1f6b..01610637 100644
--- a/crates/paimon/src/table/vector_search_result.rs
+++ b/crates/paimon/src/table/vector_search_result.rs
@@ -73,7 +73,7 @@ impl SearchResult {
vector_column: String,
hits: ScoredRowIds,
) -> Self {
- let snapshot_id = table.travel_snapshot.as_ref().map(|snapshot|
snapshot.id());
+ let snapshot_id = table.travel_snapshot().map(|snapshot|
snapshot.id());
Self {
table,
snapshot_id,
