JeonDaehong commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4413475903
##########
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:
good catch, it didn't. took your yield in 514ee83. with a probe on the arms,
the joined section now prints Load for both files and then WaitFor for both,
and without the yield it's AlreadyLoaded for both like you saw.
--
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]
