JeonDaehong commented on code in PR #3435:
URL: https://github.com/apache/iceberg-rust/pull/3435#discussion_r4411739182
##########
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:
done, it returns a `PendingPosDelLoad` now and takes the number of waiters.
##########
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:
added in d51812c: two waiters on a load that fails while the file is still
there both return the load's error (so neither reopened it), two waiters on a
cancelled load each return their own open error, and a load dropped by a
failing sibling's `item?` loads later on the same loader.
##########
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:
done, the guard's drop now uses `unwrap_or_else(PoisonError::into_inner)`.
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -271,25 +271,31 @@ impl CachingDeleteFileLoader {
return Self::load_deletion_vector(task,
basic_delete_file_loader).await;
}
- match del_filter.try_start_pos_del_load(task.file_path()) {
- PosDelLoadAction::AlreadyLoaded =>
Ok(DeleteFileContext::ExistingPosDel),
- PosDelLoadAction::WaitFor(notified) => {
- // Positional deletes are accessed synchronously by
ArrowReader.
- // We must wait here to ensure the data is ready
before returning,
- // otherwise ArrowReader might get an empty/partial
result.
- notified.await;
- Ok(DeleteFileContext::ExistingPosDel)
+ loop {
+ match del_filter.try_start_pos_del_load(task.file_path()) {
+ PosDelLoadAction::AlreadyLoaded => {
+ return Ok(DeleteFileContext::ExistingPosDel);
+ }
+ PosDelLoadAction::WaitFor(notified) => {
+ // Positional deletes are accessed synchronously
by ArrowReader.
+ // We must wait here to ensure the data is ready
before returning,
+ // otherwise ArrowReader might get an
empty/partial result.
+ // The wait also ends if that load fails, so check
again.
+ notified.await;
+ }
Review Comment:
went with the singleflight version. a failed load now keeps its error and
the tasks waiting on that attempt return it, so in your 200-task example the
file is opened once. the entry is removed either way, so a later call starts a
new attempt. a load dropped without an error (a sibling's error ending
`load_deletes`) still lets its waiters retry, since nothing failed for that
file.
--
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]
