mbutrovich commented on code in PR #3435:
URL: https://github.com/apache/iceberg-rust/pull/3435#discussion_r4444225222
##########
crates/iceberg/src/arrow/delete_filter.rs:
##########
@@ -61,18 +61,86 @@ pub(crate) struct DeleteFilter {
runtime: Runtime,
}
+/// One attempt to load a positional delete file, shared with the tasks
waiting on it.
+#[derive(Debug, Default)]
+pub(crate) struct PosDelAttempt {
+ notify: Arc<Notify>,
+ /// Set when the load fails, so the tasks waiting on it get the error
instead of
+ /// each opening the file again.
+ failure: OnceLock<(ErrorKind, String)>,
+}
+
+/// A wait on another task's positional delete load, from
[`PosDelLoadAction::WaitFor`].
+pub(crate) struct PosDelWait {
+ notified: OwnedNotified,
+ attempt: Arc<PosDelAttempt>,
+}
+
+impl PosDelWait {
+ /// Waits for the load to end. Returns its error if it failed. `Ok` means
it finished
+ /// or was dropped without an error, so the caller must call
+ /// [`DeleteFilter::try_start_pos_del_load`] again rather than assume the
file is loaded.
+ pub(crate) async fn wait(self) -> Result<()> {
+ self.notified.await;
+ match self.attempt.failure.get() {
+ Some((kind, message)) => Err(Error::new(*kind, message.clone())),
+ None => Ok(()),
+ }
+ }
+}
+
/// Action to take when trying to start loading a positional delete file
pub(crate) enum PosDelLoadAction {
- /// The file is not loaded, the caller should load it.
- Load,
+ /// The file is not loaded, the caller should load it and call
+ /// [`PosDelLoadGuard::finish`] once its delete vectors are merged.
+ Load(PosDelLoadGuard),
/// The file is already loaded, nothing to do.
AlreadyLoaded,
/// The file is currently being loaded by another task.
- /// The caller *must* await this future to ensure data availability before
+ /// The caller *must* await [`PosDelWait::wait`] to ensure data
availability before
/// returning, as subsequent access (get_delete_vector) is synchronous. The
- /// future is created under the state lock so it cannot miss the loader's
+ /// wait is created under the state lock so it cannot miss the loader's
/// `notify_waiters()` (which stores no permit).
- WaitFor(OwnedNotified),
+ WaitFor(PosDelWait),
+}
+
+/// Ownership of an in-progress positional delete load, returned by
+/// [`DeleteFilter::try_start_pos_del_load`].
+///
+/// [`finish`](Self::finish) marks the file loaded. [`fail`](Self::fail) hands
the error to
+/// the tasks waiting on this attempt. Dropping the guard without finishing
removes the
+/// `Loading` marker and wakes the waiters, so a later attempt can load the
file instead of
+/// waiting forever.
+pub(crate) struct PosDelLoadGuard {
+ filter: DeleteFilter,
+ file_path: String,
+ attempt: Arc<PosDelAttempt>,
+ finished: bool,
+}
+
+impl PosDelLoadGuard {
+ /// Marks the file as loaded and wakes any waiters.
+ pub(crate) fn finish(mut self) {
+ self.finished = true;
+ self.filter.finish_pos_del_load(&self.file_path);
+ }
+
+ /// Ends the load with `error`, which the tasks waiting on it return.
Later calls to
+ /// [`DeleteFilter::try_start_pos_del_load`] start a new attempt.
+ pub(crate) fn fail(self, error: &Error) {
+ let _ = self
+ .attempt
+ .failure
+ .set((error.kind(), error.message().to_string()));
+ }
Review Comment:
Should the waiters get the whole error, including its context and source?
`fail` keeps only `kind` and `message`, so a waiter's error drops both. With
the OpenDAL storage, the cause of an I/O failure is in the source:
[`from_opendal_error`](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/storage/opendal/src/utils.rs#L23-L29)
sets the message to "Failure in doing io operation" and puts the OpenDAL error
in `with_source`. I failed a guard with an error built that way and printed
what the leader and a waiter return:
```
leader: Unexpected, context: { path: s3://bucket/pos-delete.parquet } =>
Failure in doing io operation, source: 403 Forbidden
waiter: Unexpected => Failure in doing io operation
```
In the 200-task example from the earlier thread, 199 tasks return the second
line. With more than one data file in flight, the scan [buffers tasks
unordered](https://github.com/apache/iceberg-rust/blob/896f9d6c50ff52fa9e43aec170d3f14bfb40b512/crates/iceberg/src/arrow/reader/pipeline.rs#L97-L104),
so as I read it the error a user sees can be a waiter's, with no path and no
cause.
Keeping the `Display` output as the message carries both. The waiter's error
then starts with the kind twice (`DataInvalid => DataInvalid => bad file`), and
the two tests that compare `err.message()` exactly would check with `contains`
instead. Could we also add a source to the error in
`test_failed_pos_del_load_returns_its_error_to_waiters` and check that the
waiter's error mentions it?
```suggestion
pub(crate) fn fail(self, error: &Error) {
let _ = self
.attempt
.failure
.set((error.kind(), error.to_string()));
}
```
##########
crates/iceberg/src/arrow/caching_delete_file_loader.rs:
##########
@@ -1791,4 +1808,394 @@ 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));
+ }
+
+ /// The load of a positional delete file taken by the test, and tasks
polled until
+ /// they wait on it.
+ struct PendingPosDelLoad {
+ guard: PosDelLoadGuard,
+ waiters: Vec<futures::future::BoxFuture<'static,
Result<DeleteFileContext>>>,
+ path: String,
+ }
+
+ /// Takes the load of the first test task's positional delete file, then
polls
+ /// `load_file_for_task` for the same file from `waiters` tasks until each
waits on it.
+ async fn wait_on_pos_del_load(tmp_dir: &TempDir, waiters: usize) ->
PendingPosDelLoad {
+ 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 mut pending = Vec::with_capacity(waiters);
+ for _ in 0..waiters {
+ let (pos_del, schema) = (pos_del.clone(), task.schema_ref());
+ let (basic, filter) = (
+ loader.basic_delete_file_loader.clone(),
+ loader.delete_filter.clone(),
+ );
+ let mut waiter: futures::future::BoxFuture<'static, _> =
Box::pin(async move {
+ CachingDeleteFileLoader::load_file_for_task(&pos_del, basic,
filter, schema).await
+ });
+ assert!(futures::poll!(&mut waiter).is_pending());
+ pending.push(waiter);
+ }
+ PendingPosDelLoad {
+ guard,
+ waiters: pending,
+ path: pos_del.file_path().to_string(),
+ }
+ }
+
+ #[tokio::test]
+ async fn test_pos_del_waiter_loads_file_after_cancelled_load() {
+ let tmp_dir = TempDir::new().unwrap();
+ let PendingPosDelLoad {
+ guard, mut waiters, ..
+ } = wait_on_pos_del_load(&tmp_dir, 1).await;
+
+ // Dropped without an error, as when a sibling's error stops
`load_deletes`.
+ drop(guard);
+
+ let ctx = tokio::time::timeout(std::time::Duration::from_secs(5),
waiters.remove(0))
+ .await
+ .expect("waiter did not wake after the load was cancelled")
+ .unwrap();
+ assert!(matches!(ctx, DeleteFileContext::PosDels { .. }));
+ }
+
+ #[tokio::test]
+ async fn test_pos_del_waiters_return_the_error_of_a_failed_load() {
+ let tmp_dir = TempDir::new().unwrap();
+ let PendingPosDelLoad { guard, waiters, .. } =
wait_on_pos_del_load(&tmp_dir, 2).await;
+
+ // The file is still there, so a waiter that opened it again would
succeed.
+ guard.fail(&Error::new(ErrorKind::Unexpected, "load failed"));
+
+ for waiter in waiters {
+ let err = tokio::time::timeout(std::time::Duration::from_secs(5),
waiter)
+ .await
+ .expect("waiter did not wake after the load failed")
+ .err()
+ .unwrap();
+ assert_eq!(err.kind(), ErrorKind::Unexpected);
+ assert_eq!(err.message(), "load failed");
+ }
+ }
+
+ #[tokio::test]
+ async fn test_pos_del_waiters_retry_after_cancelled_load() {
+ let tmp_dir = TempDir::new().unwrap();
+ let PendingPosDelLoad {
+ guard,
+ waiters,
+ path,
+ } = wait_on_pos_del_load(&tmp_dir, 2).await;
+
+ // Each waiter loads the file itself after a cancelled load, and gets
its own error.
+ std::fs::remove_file(&path).unwrap();
+ drop(guard);
+
+ for waiter in waiters {
+ let err = tokio::time::timeout(std::time::Duration::from_secs(5),
waiter)
+ .await
+ .expect("waiter did not wake after the load was cancelled")
+ .err()
+ .unwrap();
+ assert!(err.to_string().contains("Failed to open file"), "{err}");
+ }
+ }
+
+ /// Local filesystem storage that yields before every read, so the loads
of several
+ /// delete files in one `load_deletes` call start together, and holds
reads of
+ /// `gated.parquet` until the gate is opened, so one of them is still in
flight when
+ /// another fails. `LocalFsStorage` never yields, so on the current-thread
test
+ /// runtime each load would otherwise finish before the next one starts.
+ mod gated_storage {
+ use std::sync::Arc;
+ use std::sync::atomic::{AtomicBool, Ordering};
+
+ use bytes::Bytes;
+ use futures::stream::BoxStream;
+
+ use crate::Result;
+ use crate::io::{
+ FileMetadata, FileRead, FileWrite, InputFile, LocalFsStorage,
OutputFile, Storage,
+ StorageConfig, StorageFactory,
+ };
+
+ #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
+ pub(super) struct GatedStorage {
+ #[serde(skip)]
+ inner: LocalFsStorage,
+ #[serde(skip)]
+ open: Arc<AtomicBool>,
+ }
+
+ impl GatedStorage {
+ async fn wait_for_gate(&self, path: &str) {
+ tokio::task::yield_now().await;
+ while path.ends_with("gated.parquet") &&
!self.open.load(Ordering::SeqCst) {
+ tokio::task::yield_now().await;
+ }
+ }
+ }
+
+ #[async_trait::async_trait]
+ #[typetag::serde(name = "test_gated_storage")]
+ impl Storage for GatedStorage {
+ async fn metadata(&self, path: &str) -> Result<FileMetadata> {
+ 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.wait_for_gate(path).await;
+ self.inner.read(path).await
+ }
+
+ async fn reader(&self, path: &str) -> Result<Box<dyn FileRead>> {
+ self.wait_for_gate(path).await;
+ 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)]
+ pub(super) struct GatedStorageFactory {
+ #[serde(skip)]
+ pub(super) open: Arc<AtomicBool>,
+ }
+
+ #[typetag::serde(name = "test_gated_storage_factory")]
+ impl StorageFactory for GatedStorageFactory {
+ fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn
Storage>> {
+ Ok(Arc::new(GatedStorage {
+ inner: LocalFsStorage,
+ open: Arc::clone(&self.open),
+ }))
+ }
+ }
+ }
+
+ /// A failing positional delete file ends `load_deletes` early while the
load of the
+ /// other file in the same call is still in flight, dropping its guard.
That file
+ /// still loads later.
+ #[tokio::test]
+ async fn
test_pos_del_load_dropped_by_a_failing_sibling_can_be_loaded_later() {
+ let tmp_dir = TempDir::new().unwrap();
+ let schema = Arc::new(Schema::builder().build().unwrap());
+ let write = |name: &str, positions: Vec<i64>| {
+ let path = tmp_dir.path().join(name);
+ let path = path.to_str().unwrap().to_string();
+ let batch = RecordBatch::try_new(
+ crate::arrow::delete_filter::tests::create_pos_del_schema(),
+ vec![
+ Arc::new(StringArray::from_iter_values(vec![
+ "data.parquet";
+ positions.len()
+ ])),
+ Arc::new(Int64Array::from_iter_values(positions)),
+ ],
+ )
+ .unwrap();
+ let mut writer =
+ ArrowWriter::try_new(File::create(&path).unwrap(),
batch.schema(), None).unwrap();
+ writer.write(&batch).unwrap();
+ writer.close().unwrap();
+ FileScanTaskDeleteFile::builder()
+ .with_file_path(path.clone())
+
.with_file_size_in_bytes(std::fs::metadata(&path).unwrap().len())
+ .with_file_type(DataContentType::PositionDeletes)
+ .with_file_format(DataFileFormat::Parquet)
+ .with_partition_spec_id(0)
+ .build()
+ .unwrap()
+ };
Review Comment:
Could this closure and the one in
`test_pos_del_load_can_be_retried_after_parse_failure` (lines 2145-2169) become
one test helper, such as `fn write_pos_del_file(path: &str, positions:
Vec<i64>) -> FileScanTaskDeleteFile`? They write the same batch and build the
same `FileScanTaskDeleteFile`, and differ only in how the path is formed. One
helper keeps the two tests' fixtures from drifting apart.
--
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]
