fix: ignore NotFound error of the non-first list during iter dir (#4891)

## Changes Made

Ignore the NotFound error during iter dir if got empty response after
the first list operation.

## Related Issues

<img width="1345" height="234" alt="image"
src="https://github.com/user-attachments/assets/ae5a8b0d-885c-453d-b1c9-208c0023cc97"
/>
As the above picture shows, globed the parent folder got a `Not found`
error, but globed the sub folder succeed.

The Not Found error was thrown during the second list request with next
continue token from the previous list response, but got empty response,
and then throw Not Found error, and then pop up the error to downstream,
cause the glob process failed.

The reason why got empty response of the second list request with a
continue token from prev response is that the listV2 of S3-like object
store is trying to solve the timeout problem of listing behavior
comparing to listv1.
Assume we are trying to list 1000 keys among abundant objects,
especially if we did much delete operations before listing which might
lead to delete holes problem via delete tombstone that will impact the
list performance.
- The listV1 might be hanged and waiting for 1000 objects return in this
situation and then the list http request will failed since timeout.
- the listV2 will try to retrieve objects matched the request condition
as much as possible within timeout, which means the response objects
might less than 1000, the response contains the `next continue token`
and `is_truncated=true` flag indicate the response is truncate and then
client should continue to list the remaining objects, but listv2 doesn't
ensure the existence of remaining objects because the previous scan
operation of kv store is cut down, so the later list request might get
empty response.

## Checklist

- [ ] Documented in API Docs (if applicable)
- [ ] Documented in User Guide (if applicable)
- [ ] If adding a new documentation page, doc is added to
`docs/mkdocs.yml` navigation
- [ ] Documentation builds and is formatted properly (tag @/ccmao1130
for docs review)

---------

Co-authored-by: wuzhiping <[email protected]>
Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: Desmond Cheong <[email protected]>

Author: 4 people

src/daft-io/src/object_io.rs

@@ -235,10 +235,24 @@ pub trait ObjectSource: Sync + Send {
 
             let mut continuation_token = lsr.continuation_token.clone();
             while continuation_token.is_some() {
-                let lsr = self.ls(&uri, posix, continuation_token.as_deref(), page_size, io_stats.clone()).await?;
-                continuation_token.clone_from(&lsr.continuation_token);
-                for fm in lsr.files {
-                    yield Ok(fm);
+                // Note: There might some race conditions here that the list response is empty
+                // even though the continuation token of previous response is not empty, so skip NotFound error here.
+                // TODO(desmond): Ideally we should patch how `ls` produces NotFound errors. See issue #4982
+                let lsr_result = self.ls(&uri, posix, continuation_token.as_deref(), page_size, io_stats.clone()).await;
+                match lsr_result {
+                    Ok(lsr) => {
+                        continuation_token.clone_from(&lsr.continuation_token);
+                        for fm in lsr.files {
+                            yield Ok(fm);
+                        }
+                    },
+                    Err(err) => {
+                        if matches!(err, super::Error::NotFound { .. }) {
+                            continuation_token = None;
+                        } else {
+                            yield Err(err);
+                        }
+                    }
                 }
             }
         };