[datafusion-cli] Implement average LIST duration for object store profiling (#19127)

## Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->

- Closes #18138

## Rationale for this change

The `list` operation returns a stream, so it previously recorded
`duration: None`, missing performance insights. Time-to-first-item is a
useful metric for list operations, indicating how quickly results start.
This adds duration tracking by measuring time until the first item is
yielded (or the stream ends).

## What changes are included in this PR?

1. Added `TimeToFirstItemStream`: A stream wrapper that measures elapsed
time from creation until the first item is yielded (or the stream ends
if empty).
2. Updated `instrumented_list`: Wraps the inner stream with
`TimeToFirstItemStream` to record duration.
3. Changed `requests` field: Switched from `Mutex<Vec<RequestDetails>>`
to `Arc<Mutex<Vec<RequestDetails>>>` to allow sharing across async
boundaries (needed for the stream wrapper).
4. Updated tests: Modified `instrumented_store_list` to consume at least
one stream item and verify that `duration` is now `Some(Duration)`
instead of `None`.

## Are these changes tested?

Yes. The existing test `instrumented_store_list` was updated to:
- Consume at least one item from the stream using `stream.next().await`
- Assert that `request.duration.is_some()` (previously `is_none()`)

All tests pass, including the updated list test and other instrumented
operation tests.

## Are there any user-facing changes?

Users with profiling enabled will see duration values for `list`
operations instead of nothing.

Author: peterxcli

datafusion-cli/src/object_storage/instrumented.rs

@@ -35,7 +35,7 @@ use datafusion::{
     error::DataFusionError,
     execution::object_store::{DefaultObjectStoreRegistry, ObjectStoreRegistry},
 };
-use futures::stream::BoxStream;
+use futures::stream::{BoxStream, Stream};
 use object_store::{
     GetOptions, GetRange, GetResult, ListResult, MultipartUpload, ObjectMeta,
     ObjectStore, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result,
@@ -44,6 +44,58 @@ use object_store::{
 use parking_lot::{Mutex, RwLock};
 use url::Url;
 
+/// A stream wrapper that measures the time until the first response(item or end of stream) is yielded
+struct TimeToFirstItemStream<S> {
+    inner: S,
+    start: Instant,
+    request_index: usize,
+    requests: Arc<Mutex<Vec<RequestDetails>>>,
+    first_item_yielded: bool,
+}
+
+impl<S> TimeToFirstItemStream<S> {
+    fn new(
+        inner: S,
+        start: Instant,
+        request_index: usize,
+        requests: Arc<Mutex<Vec<RequestDetails>>>,
+    ) -> Self {
+        Self {
+            inner,
+            start,
+            request_index,
+            requests,
+            first_item_yielded: false,
+        }
+    }
+}
+
+impl<S> Stream for TimeToFirstItemStream<S>
+where
+    S: Stream<Item = Result<ObjectMeta>> + Unpin,
+{
+    type Item = Result<ObjectMeta>;
+
+    fn poll_next(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let poll_result = std::pin::Pin::new(&mut self.inner).poll_next(cx);
+
+        if !self.first_item_yielded && poll_result.is_ready() {
+            self.first_item_yielded = true;
+            let elapsed = self.start.elapsed();
+
+            let mut requests = self.requests.lock();
+            if let Some(request) = requests.get_mut(self.request_index) {
+                request.duration = Some(elapsed);
+            }
+        }
+
+        poll_result
+    }
+}
+
 /// The profiling mode to use for an [`InstrumentedObjectStore`] instance. Collecting profiling
 /// data will have a small negative impact on both CPU and memory usage. Default is `Disabled`
 #[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
@@ -92,7 +144,7 @@ impl From<u8> for InstrumentedObjectStoreMode {
 pub struct InstrumentedObjectStore {
     inner: Arc<dyn ObjectStore>,
     instrument_mode: AtomicU8,
-    requests: Mutex<Vec<RequestDetails>>,
+    requests: Arc<Mutex<Vec<RequestDetails>>>,
 }
 
 impl InstrumentedObjectStore {
@@ -101,7 +153,7 @@ impl InstrumentedObjectStore {
         Self {
             inner: object_store,
             instrument_mode,
-            requests: Mutex::new(Vec::new()),
+            requests: Arc::new(Mutex::new(Vec::new())),
         }
     }
 
@@ -219,19 +271,31 @@ impl InstrumentedObjectStore {
         prefix: Option<&Path>,
     ) -> BoxStream<'static, Result<ObjectMeta>> {
         let timestamp = Utc::now();
-        let ret = self.inner.list(prefix);
+        let start = Instant::now();
+        let inner_stream = self.inner.list(prefix);
+
+        let request_index = {
+            let mut requests = self.requests.lock();
+            requests.push(RequestDetails {
+                op: Operation::List,
+                path: prefix.cloned().unwrap_or_else(|| Path::from("")),
+                timestamp,
+                duration: None,
+                size: None,
+                range: None,
+                extra_display: None,
+            });
+            requests.len() - 1
+        };
 
-        self.requests.lock().push(RequestDetails {
-            op: Operation::List,
-            path: prefix.cloned().unwrap_or_else(|| Path::from("")),
-            timestamp,
-            duration: None, // list returns a stream, so the duration isn't meaningful
-            size: None,
-            range: None,
-            extra_display: None,
-        });
+        let wrapped_stream = TimeToFirstItemStream::new(
+            inner_stream,
+            start,
+            request_index,
+            Arc::clone(&self.requests),
+        );
 
-        ret
+        Box::pin(wrapped_stream)
     }
 
     async fn instrumented_list_with_delimiter(
@@ -759,6 +823,7 @@ impl ObjectStoreRegistry for InstrumentedObjectStoreRegistry {
 
 #[cfg(test)]
 mod tests {
+    use futures::StreamExt;
     use object_store::WriteMultipart;
 
     use super::*;
@@ -899,13 +964,15 @@ mod tests {
 
         instrumented.set_instrument_mode(InstrumentedObjectStoreMode::Trace);
         assert!(instrumented.requests.lock().is_empty());
-        let _ = instrumented.list(Some(&path));
+        let mut stream = instrumented.list(Some(&path));
+        // Consume at least one item from the stream to trigger duration measurement
+        let _ = stream.next().await;
         assert_eq!(instrumented.requests.lock().len(), 1);
 
         let request = instrumented.take_requests().pop().unwrap();
         assert_eq!(request.op, Operation::List);
         assert_eq!(request.path, path);
-        assert!(request.duration.is_none());
+        assert!(request.duration.is_some());
         assert!(request.size.is_none());
         assert!(request.range.is_none());
         assert!(request.extra_display.is_none());