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 0d880959 feat: expose point-query primitives in paimon (#739)
0d880959 is described below
commit 0d880959f86d7c29c5d90d9ebf07dd8ebe1a7424
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Aug 23 18:22:30 2026 +0800
feat: expose point-query primitives in paimon (#739)
---
crates/paimon/src/error.rs | 2 ++
crates/paimon/src/spec/blob_descriptor.rs | 52 ++++++++++++++++++++++------
crates/paimon/src/table/format_table_scan.rs | 1 +
crates/paimon/src/table/mod.rs | 29 ++++++++++++++--
crates/paimon/src/table/scan_trace.rs | 7 ++--
crates/paimon/src/table/snapshot_manager.rs | 5 +--
crates/paimon/src/table/source.rs | 30 ++++++++++++++++
crates/paimon/src/table/table_scan.rs | 1 +
8 files changed, 107 insertions(+), 20 deletions(-)
diff --git a/crates/paimon/src/error.rs b/crates/paimon/src/error.rs
index 39969ee8..a8be68f1 100644
--- a/crates/paimon/src/error.rs
+++ b/crates/paimon/src/error.rs
@@ -102,6 +102,8 @@ pub enum Error {
TableAlreadyExist { full_name: String },
#[snafu(display("Table {} does not exist.", full_name))]
TableNotExist { full_name: String },
+ #[snafu(display("Snapshot {} does not exist.", snapshot_id))]
+ SnapshotNotExist { snapshot_id: i64 },
#[snafu(display("View {} already exists.", full_name))]
ViewAlreadyExist { full_name: String },
#[snafu(display("View {} does not exist.", full_name))]
diff --git a/crates/paimon/src/spec/blob_descriptor.rs
b/crates/paimon/src/spec/blob_descriptor.rs
index f65b293d..8f32f04b 100644
--- a/crates/paimon/src/spec/blob_descriptor.rs
+++ b/crates/paimon/src/spec/blob_descriptor.rs
@@ -70,6 +70,10 @@ impl BlobDescriptor {
&self.uri
}
+ pub fn version(&self) -> u8 {
+ self.version
+ }
+
pub fn offset(&self) -> i64 {
self.offset
}
@@ -78,7 +82,12 @@ impl BlobDescriptor {
self.length
}
- pub(crate) fn range_spec(&self) -> crate::Result<BlobRangeSpec> {
+ /// Validate that this descriptor represents a usable byte range.
+ ///
+ /// A length of `-1` means reading from `offset` to the end of the object.
+ /// Other negative lengths, negative offsets, and overflowing bounded
ranges
+ /// are rejected.
+ pub fn validate(&self) -> crate::Result<()> {
if self.offset < 0 {
return Err(Error::DataInvalid {
message: format!(
@@ -98,22 +107,29 @@ impl BlobDescriptor {
});
}
- let offset = self.offset as u64;
- let length = if self.length == -1 {
- None
- } else {
- Some(self.length as u64)
- };
- if let Some(length) = length {
- offset
- .checked_add(length)
+ if self.length >= 0 {
+ (self.offset as u64)
+ .checked_add(self.length as u64)
.ok_or_else(|| Error::DataInvalid {
message: format!(
- "BlobDescriptor range overflows u64: offset={offset},
length={length}"
+ "BlobDescriptor range overflows u64: offset={},
length={}",
+ self.offset, self.length
),
source: None,
})?;
}
+ Ok(())
+ }
+
+ pub(crate) fn range_spec(&self) -> crate::Result<BlobRangeSpec> {
+ self.validate()?;
+
+ let offset = self.offset as u64;
+ let length = if self.length == -1 {
+ None
+ } else {
+ Some(self.length as u64)
+ };
Ok(BlobRangeSpec { offset, length })
}
@@ -231,6 +247,20 @@ mod tests {
let bytes = desc.serialize();
let deserialized = BlobDescriptor::deserialize(&bytes).unwrap();
assert_eq!(desc, deserialized);
+ assert_eq!(deserialized.version(), CURRENT_VERSION);
+ deserialized.validate().unwrap();
+ }
+
+ #[test]
+ fn test_validate_rejects_invalid_ranges() {
+ let negative_offset = BlobDescriptor::new("file:///tmp/a".to_string(),
-1, 1);
+ assert!(negative_offset.validate().is_err());
+
+ let negative_length = BlobDescriptor::new("file:///tmp/a".to_string(),
0, -2);
+ assert!(negative_length.validate().is_err());
+
+ let to_end = BlobDescriptor::new("file:///tmp/a".to_string(),
i64::MAX, -1);
+ to_end.validate().unwrap();
}
#[test]
diff --git a/crates/paimon/src/table/format_table_scan.rs
b/crates/paimon/src/table/format_table_scan.rs
index 224ef509..5a6a00c8 100644
--- a/crates/paimon/src/table/format_table_scan.rs
+++ b/crates/paimon/src/table/format_table_scan.rs
@@ -64,6 +64,7 @@ impl<'a> FormatTableScan<'a> {
self.ensure_query_auth_allowed()?;
let mut trace = ScanTrace::default();
let plan = self.plan_inner(Some(&mut trace)).await?;
+ trace.planned_data_file_bytes = plan.planned_data_file_bytes();
Ok((plan, trace))
}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 31f7be47..163bd0b1 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -434,19 +434,42 @@ impl Table {
/// schema (the `if let Ok` below swallows them); an invalid selector
/// still fails later at scan planning.
pub async fn copy_with_time_travel(&self, extra: HashMap<String, String>)
-> Result<Self> {
+ self.copy_with_time_travel_mode(extra, false).await
+ }
+
+ /// Like [`Self::copy_with_time_travel`], but propagates selector
resolution
+ /// failures. Services should use this variant so a missing or unreadable
+ /// snapshot cannot silently fall back to the current schema.
+ pub async fn copy_with_time_travel_strict(
+ &self,
+ extra: HashMap<String, String>,
+ ) -> Result<Self> {
+ self.copy_with_time_travel_mode(extra, true).await
+ }
+
+ async fn copy_with_time_travel_mode(
+ &self,
+ extra: HashMap<String, String>,
+ strict: bool,
+ ) -> Result<Self> {
let mut table = self.copy_with_options(extra);
// Reject unimplemented scan options on the merged view before any IO,
so
// both table-level and per-read options are covered.
CoreOptions::new(table.schema().options()).validate_scan_options()?;
// travel_to_snapshot returns Ok(None) without IO when the merged
// options contain no selector.
- if let Ok(Some(snapshot)) = time_travel::travel_to_snapshot(
+ let resolved = time_travel::travel_to_snapshot(
&table.snapshot_manager(),
&table.tag_manager(),
table.schema.options(),
)
- .await
- {
+ .await;
+ let snapshot = if strict {
+ resolved?
+ } else {
+ resolved.ok().flatten()
+ };
+ if let Some(snapshot) = snapshot {
if snapshot.schema_id() != table.schema.id() {
let snapshot_schema =
table.schema_manager.schema(snapshot.schema_id()).await?;
table.schema =
diff --git a/crates/paimon/src/table/scan_trace.rs
b/crates/paimon/src/table/scan_trace.rs
index 5aec3ac9..45063800 100644
--- a/crates/paimon/src/table/scan_trace.rs
+++ b/crates/paimon/src/table/scan_trace.rs
@@ -53,6 +53,8 @@ pub struct ScanTrace {
pub splits_after_limit: usize,
pub final_splits: usize,
pub final_files: usize,
+ /// Sum of known data-file sizes referenced by the final plan.
+ pub planned_data_file_bytes: u64,
pub limit: Option<usize>,
}
@@ -99,7 +101,7 @@ impl fmt::Display for ScanTrace {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
- "snapshot={:?}, manifests={}/{}, manifest_row_range_pruned={},
entries_read={}, bucket_pruned={}, partition_pruned={},
entry_row_range_pruned={}, data_stats_pruned={}, cross_schema_pruned={},
split_candidates_built={}, limit_early_stopped={}, splits_before_limit={},
splits_after_limit={}, files={}",
+ "snapshot={:?}, manifests={}/{}, manifest_row_range_pruned={},
entries_read={}, bucket_pruned={}, partition_pruned={},
entry_row_range_pruned={}, data_stats_pruned={}, cross_schema_pruned={},
split_candidates_built={}, limit_early_stopped={}, splits_before_limit={},
splits_after_limit={}, files={}, data_file_bytes={}",
self.snapshot_id,
self.manifest_files_after_partition_pruning,
self.manifest_files_before_partition_pruning,
@@ -114,7 +116,8 @@ impl fmt::Display for ScanTrace {
self.limit_early_stopped,
self.splits_before_limit,
self.splits_after_limit,
- self.final_files
+ self.final_files,
+ self.planned_data_file_bytes
)
}
}
diff --git a/crates/paimon/src/table/snapshot_manager.rs
b/crates/paimon/src/table/snapshot_manager.rs
index cb71b31c..1de8baa9 100644
--- a/crates/paimon/src/table/snapshot_manager.rs
+++ b/crates/paimon/src/table/snapshot_manager.rs
@@ -209,10 +209,7 @@ impl SnapshotManager {
let snapshot_path = self.snapshot_path(snapshot_id);
let snap_input = self.file_io.new_input(&snapshot_path)?;
if !snap_input.exists().await? {
- return Err(crate::Error::DataInvalid {
- message: format!("snapshot file does not exist:
{snapshot_path}"),
- source: None,
- });
+ return Err(crate::Error::SnapshotNotExist { snapshot_id });
}
let snap_bytes = snap_input.read().await?;
let snapshot: Snapshot =
diff --git a/crates/paimon/src/table/source.rs
b/crates/paimon/src/table/source.rs
index 2c1e39bc..c1e697ec 100644
--- a/crates/paimon/src/table/source.rs
+++ b/crates/paimon/src/table/source.rs
@@ -1295,6 +1295,20 @@ impl Plan {
&self.splits
}
+ /// Sum of data-file bytes referenced by this plan.
+ ///
+ /// Negative file sizes are treated as unknown and do not contribute. The
+ /// result is therefore a lower bound when a connector cannot provide every
+ /// file size. Totals larger than [`u64::MAX`] saturate at that value so
+ /// callers cannot under-count an oversized plan due to integer overflow.
+ pub fn planned_data_file_bytes(&self) -> u64 {
+ self.splits
+ .iter()
+ .flat_map(DataSplit::data_files)
+ .filter_map(|file| u64::try_from(file.file_size).ok())
+ .fold(0, u64::saturating_add)
+ }
+
/// Consume this plan and return its splits without cloning their file
metadata.
#[must_use = "consuming a plan without using its splits drops the planned
work"]
pub fn into_splits(self) -> Vec<DataSplit> {
@@ -1390,6 +1404,22 @@ mod tests {
assert!(Arc::ptr_eq(&data_files, &splits[0].data_files));
}
+ #[test]
+ fn planned_data_file_bytes_saturates_on_overflow() {
+ let mut first = file("a.parquet", 10, Some(0));
+ first.file_size = i64::MAX;
+ let mut second = file("b.parquet", 10, Some(10));
+ second.file_size = i64::MAX;
+ let mut overflow = file("c.parquet", 10, Some(20));
+ overflow.file_size = 2;
+ let mut unknown = file("unknown.parquet", 10, Some(30));
+ unknown.file_size = -1;
+
+ let plan = Plan::new(vec![split(vec![first, second, overflow,
unknown], true)]);
+
+ assert_eq!(plan.planned_data_file_bytes(), u64::MAX);
+ }
+
#[test]
fn data_split_serde_json_round_trip() {
let split = DataSplit::builder()
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 5c121922..b28915f6 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1101,6 +1101,7 @@ impl<'a> PaimonTableScan<'a> {
Some(&mut trace),
)
.await?;
+ trace.planned_data_file_bytes = plan.planned_data_file_bytes();
Ok((plan, trace))
}
