JeonDaehong commented on code in PR #3433:
URL: https://github.com/apache/iceberg-rust/pull/3433#discussion_r4411738594
##########
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:
retitled to `feat(reader)!:` and added a breaking-change note to the
description, covering the getter and reading a `None` task on an older build.
##########
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.
--
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]
