JeonDaehong commented on code in PR #3435:
URL: https://github.com/apache/iceberg-rust/pull/3435#discussion_r4413476399
##########
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,
+ ) {
+ let task = setup(tmp_dir.path()).remove(0);
+ let pos_del = task
+ .deletes()
+ .iter()
+ .find(|delete| delete.file_type() ==
DataContentType::PositionDeletes)
+ .unwrap()
+ .clone();
+ let loader = CachingDeleteFileLoader::new(FileIO::new_with_fs(), 10,
Runtime::current());
+
+ let PosDelLoadAction::Load(guard) = loader
+ .delete_filter
+ .try_start_pos_del_load(pos_del.file_path())
+ else {
+ panic!("expected Load for a new file");
+ };
+ let path = pos_del.file_path().to_string();
+ let mut waiter = Box::pin(async move {
+ CachingDeleteFileLoader::load_file_for_task(
+ &pos_del,
+ loader.basic_delete_file_loader.clone(),
+ loader.delete_filter.clone(),
+ task.schema_ref(),
+ )
+ .await
+ });
+ assert!(futures::poll!(&mut waiter).is_pending());
+ (guard, waiter, path)
+ }
+
+ #[tokio::test]
+ async fn test_pos_del_waiter_loads_file_after_abandoned_load() {
Review Comment:
correction on the sibling test: on the current-thread test runtime it didn't
actually overlap the two loads, the valid file only loaded on the second call,
same thing @mbutrovich found on #3433. 5fa6051 adds a test storage that yields
before reads and holds `gated.parquet` until the test opens it. with a probe
the first call now goes Load invalid, Load gated, fail invalid, abandon gated,
and the second call loads gated.
--
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]
