mbutrovich commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4412698043
##########
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:
> added in 765f199: two `load_deletes` on a fresh loader with
`futures::join!`. the size requests go up by `unknown.len()` once, and both
filters delete the same positions.
Does the second load reach `WaitFor` here? I added an `eprintln!` to each
`PosDelLoadAction` arm in `load_file_for_task` and ran this test on the head
commit. The joined section prints `Load` for `pos-del-1.parquet` and
`pos-del-2.parquet`, then `AlreadyLoaded` for both. So the second load starts
after the first has finished, which is the same path as the sequential check
above it. As I read it,
[`LocalFsStorage::metadata`](https://github.com/apache/iceberg-rust/blob/beba39c8d62810fa101740bb2dbb03ab0d8f07ac/crates/iceberg/src/io/storage/local_fs.rs#L100-L107)
never yields, so on the current-thread runtime that `#[tokio::test]` uses, the
first spawned load runs to completion before the second one is polled.
Could the counting storage yield after it counts? With this change in
[`MetadataCountingStorage::metadata`](https://github.com/apache/iceberg-rust/blob/765f199a740f29ab5d7694ecd4e4864d702ea135/crates/iceberg/src/arrow/caching_delete_file_loader.rs#L1172-L1173),
the same probe prints `WaitFor` for both files in the joined section, and the
test still passes.
```rust
self.calls.fetch_add(1, Ordering::SeqCst);
// `LocalFsStorage::metadata` doesn't yield, so without this the first of two
// concurrent loads finishes before the second starts.
tokio::task::yield_now().await;
```
--
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]
