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 { + /// 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 { + 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) -> Result { + 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, + ) -> Result { + self.copy_with_time_travel_mode(extra, true).await + } + + async fn copy_with_time_travel_mode( + &self, + extra: HashMap, + strict: bool, + ) -> Result { 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, } @@ -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 { @@ -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)) }