feat: expose point-query primitives in paimon

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))]

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]

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))
     }
 

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 =

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
         )
     }
 }

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 =

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()

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))
     }