laskoviymishka commented on code in PR #3435:
URL: https://github.com/apache/iceberg-rust/pull/3435#discussion_r4362870990
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -1791,4 +1797,185 @@ mod tests {
assert_eq!(err.kind(), ErrorKind::DataInvalid);
assert!(err.message().contains("expected 2 from record_count"));
}
+
+ #[tokio::test]
+ async fn test_pos_del_load_can_be_retried_after_failure() {
+ let tmp_dir = TempDir::new().unwrap();
+ let file_scan_tasks = setup(tmp_dir.path());
+ let task = &file_scan_tasks[0];
+ let pos_del = task
+ .deletes()
+ .iter()
+ .find(|delete| delete.file_type() ==
DataContentType::PositionDeletes)
+ .unwrap()
+ .clone();
+ let path = pos_del.file_path().to_string();
+ let hidden = format!("{path}.hidden");
+ let load = |loader: &CachingDeleteFileLoader| {
+ loader.load_deletes(std::slice::from_ref(&pos_del),
task.schema_ref())
+ };
+
+ // The first load fails because the file is not there yet.
+ std::fs::rename(&path, &hidden).unwrap();
+ let loader = CachingDeleteFileLoader::new(FileIO::new_with_fs(), 10,
Runtime::current());
+ assert!(load(&loader).await.unwrap().is_err());
+
+ // Once it is, the same loader loads it instead of waiting on the
failed load.
+ std::fs::rename(&hidden, &path).unwrap();
+ let retried = tokio::time::timeout(std::time::Duration::from_secs(5),
load(&loader))
+ .await
+ .expect("retry waited on the failed load")
+ .unwrap()
+ .unwrap();
+ let fresh = load(&CachingDeleteFileLoader::new(
+ FileIO::new_with_fs(),
+ 10,
+ Runtime::current(),
+ ))
+ .await
+ .unwrap()
+ .unwrap();
+
+ let positions = |filter: &DeleteFilter| -> Vec<u64> {
+ filter
+ .get_delete_vector(task)
+ .unwrap()
+ .lock()
+ .unwrap()
+ .iter()
+ .collect()
+ };
+ assert!(!positions(&fresh).is_empty());
+ assert_eq!(positions(&retried), positions(&fresh));
+ }
+
+ /// Takes the load of the first test task's positional delete file, then
polls
+ /// `load_file_for_task` for the same file until it waits on that load.
+ async fn wait_on_pos_del_load(
+ tmp_dir: &TempDir,
+ ) -> (
+ PosDelLoadGuard,
+ std::pin::Pin<Box<impl Future<Output = Result<DeleteFileContext>>>>,
+ String,
Review Comment:
nit: could `wait_on_pos_del_load` return a small struct instead of the
3-tuple? Only the open-error test uses the path. With named fields, the other
caller could skip it with `..` instead of `_`.
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -163,6 +200,21 @@ impl DeleteFilter {
}
}
+ /// Removes the `Loading` marker of a positional delete file whose load
failed or was
+ /// dropped, and wakes any waiting tasks so they can try the load again.
+ fn abandon_pos_del_load(&self, file_path: &str) {
+ let removed = self
+ .state
+ .write()
+ .unwrap()
Review Comment:
Could this recover from a poisoned lock instead of unwrapping?
[`get_delete_vector_for_path`](https://github.com/apache/iceberg-rust/blob/faa71d744eb9908c26938f1232f6131fa5ae2bb8/crates/iceberg/src/arrow/delete_filter.rs#L128-L137)
already tolerates poisoning with `.read().ok()`. This now runs from `Drop`.
Say the lock was poisoned by a panic inside
[`upsert_delete_vector`](https://github.com/apache/iceberg-rust/blob/faa71d744eb9908c26938f1232f6131fa5ae2bb8/crates/iceberg/src/arrow/delete_filter.rs#L291-L302)
in the `DelVecs` arm. The guard then drops during that unwind, and the second
panic aborts the process. On `main` that chain ends as a task panic. It takes
two panics in a row, so this is minor.
```suggestion
.unwrap_or_else(std::sync::PoisonError::into_inner)
```
--
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]
