laskoviymishka commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4362896382
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -656,8 +666,9 @@ impl FileScanTaskDeleteFile {
&self.file_path
}
- /// Returns the total size of the delete file in bytes.
- pub fn file_size_in_bytes(&self) -> u64 {
+ /// Returns the total size of the delete file in bytes, or `None` if the
task does not
+ /// carry it.
+ pub fn file_size_in_bytes(&self) -> Option<u64> {
Review Comment:
Could we mark this as breaking, with `feat(reader)!:` in the title or a
breaking-change note in the description, as #3263 did? On `main` this getter
[returns
`u64`](https://github.com/apache/iceberg-rust/blob/ba849a6ed3030c237e46be3e1d39e67cebf035a7/crates/iceberg/src/scan/task.rs#L659-L662),
so downstream callers that use it as one stop compiling. A task serialized
with `None` also can't be read by an older build, since the base serde proxy
[requires the
field](https://github.com/apache/iceberg-rust/blob/ba849a6ed3030c237e46be3e1d39e67cebf035a7/crates/iceberg/src/scan/task.rs#L613-L615).
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -1129,6 +1154,285 @@ mod tests {
assert!(result.is_none()); // no pos dels for file 3
}
+ /// Local filesystem storage that counts metadata requests.
+ #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+ struct MetadataCountingStorage {
+ #[serde(skip)]
+ inner: LocalFsStorage,
+ #[serde(skip)]
+ calls: Arc<AtomicUsize>,
+ /// Fail every metadata request with a retryable error.
+ #[serde(skip)]
+ fail_metadata: bool,
+ }
+
+ #[async_trait::async_trait]
+ #[typetag::serde(name = "test_metadata_counting_storage")]
+ impl Storage for MetadataCountingStorage {
+ async fn metadata(&self, path: &str) -> Result<FileMetadata> {
+ self.calls.fetch_add(1, Ordering::SeqCst);
+ if self.fail_metadata {
+ return Err(
+ Error::new(ErrorKind::Unexpected, "metadata temporarily
unavailable")
+ .with_retryable(true),
+ );
+ }
+ self.inner.metadata(path).await
+ }
+
+ fn new_input(&self, path: &str) -> Result<InputFile> {
+ Ok(InputFile::new(Arc::new(self.clone()), path.to_string()))
+ }
+
+ fn new_output(&self, path: &str) -> Result<OutputFile> {
+ Ok(OutputFile::new(Arc::new(self.clone()), path.to_string()))
+ }
+
+ async fn exists(&self, path: &str) -> Result<bool> {
+ self.inner.exists(path).await
+ }
+
+ async fn read(&self, path: &str) -> Result<Bytes> {
+ self.inner.read(path).await
+ }
+
+ async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
+ self.inner.reader(path).await
+ }
+
+ async fn write(&self, path: &str, bs: Bytes) -> Result<()> {
+ self.inner.write(path, bs).await
+ }
+
+ async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> {
+ self.inner.writer(path).await
+ }
+
+ async fn delete(&self, path: &str) -> Result<()> {
+ self.inner.delete(path).await
+ }
+
+ async fn delete_prefix(&self, path: &str) -> Result<()> {
+ self.inner.delete_prefix(path).await
+ }
+
+ async fn delete_stream(&self, paths: BoxStream<'static, String>) ->
Result<()> {
+ self.inner.delete_stream(paths).await
+ }
+ }
+
+ #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+ struct MetadataCountingStorageFactory {
+ #[serde(skip)]
+ calls: Arc<AtomicUsize>,
+ #[serde(skip)]
+ fail_metadata: bool,
+ }
+
+ #[typetag::serde(name = "test_metadata_counting_storage_factory")]
+ impl StorageFactory for MetadataCountingStorageFactory {
+ fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
+ Ok(Arc::new(MetadataCountingStorage {
+ inner: LocalFsStorage,
+ calls: Arc::clone(&self.calls),
+ fail_metadata: self.fail_metadata,
+ }))
+ }
+ }
+
+ /// Rebuilds `delete` at `path` without a file size.
+ fn without_size(delete: &FileScanTaskDeleteFile, path: &str) ->
FileScanTaskDeleteFile {
+ FileScanTaskDeleteFile::builder()
+ .with_file_path(path.to_string())
+ .with_file_type(delete.file_type())
+ .with_file_format(delete.file_format())
+ .with_partition_spec_id(delete.partition_spec_id())
+ .with_equality_ids(delete.equality_ids().map(<[i32]>::to_vec))
+
.with_referenced_data_file(delete.referenced_data_file().map(str::to_string))
+ .with_content_offset(delete.content_offset())
+ .with_content_size_in_bytes(delete.content_size_in_bytes())
+ .with_record_count(delete.record_count())
+ .with_key_metadata(delete.key_metadata().map(Box::from))
+ .build()
+ .unwrap()
+ }
+
+ #[tokio::test]
+ async fn test_unknown_delete_sizes_are_resolved_once_per_file() {
+ let tmp_dir = TempDir::new().unwrap();
+ let file_scan_tasks = setup(tmp_dir.path());
+ let factory = Arc::new(MetadataCountingStorageFactory::default());
+ let calls = Arc::clone(&factory.calls);
+ let file_io = FileIOBuilder::new(factory).build();
+ let task = &file_scan_tasks[0];
+ let unknown: Vec<_> = task
+ .deletes()
+ .iter()
+ .map(|delete| without_size(delete, delete.file_path()))
+ .collect();
+ let positions = |filter: &DeleteFilter| -> Vec<u64> {
+ filter
+ .get_delete_vector(task)
+ .unwrap()
+ .lock()
+ .unwrap()
+ .iter()
+ .collect()
+ };
+
+ // Known sizes never request metadata.
+ let known = CachingDeleteFileLoader::new(file_io.clone(), 10,
Runtime::current())
+ .load_deletes(task.deletes(), task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+ // Unknown sizes: one request per delete file, and none again when
+ // another task references an already loaded delete file.
+ let loader = CachingDeleteFileLoader::new(file_io, 10,
Runtime::current());
+ let lazy = loader
+ .load_deletes(&unknown, task.schema_ref())
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(calls.load(Ordering::SeqCst), unknown.len());
+ loader
+ .load_deletes(&unknown[1..], task.schema_ref())
Review Comment:
Could this also run two `load_deletes` on the same unknown-size files
concurrently, say with `futures::join!`, and check `calls` stays at
`unknown.len()`? Here the second load starts after the first finishes, so it
takes the `AlreadyLoaded` path. The only `WaitFor` test is
[state-level](https://github.com/apache/iceberg-rust/blob/aa189cce61d0eb51febb1e228e738486b28219f8/crates/iceberg/src/arrow/delete_filter.rs#L310-L337)
and doesn't go through `load_deletes`. A succeeding stat is enough, since a
failing one hangs today (#3434).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
