Adds instrumentation to PUT ops in the CLI (#18139)

## Which issue does this PR close?

This does not fully close, but is an incremental building block
component for:
 - #17207

The full context of how this code is likely to progress can be seen in
the POC for this effort:
 - #17266

## Rationale for this change

Further fills out the missing methods that have yet to be instrumented
in the instrumented object store.

## What changes are included in this PR?

 - Adds instrumentation around put_opts
 - Adds instrumentation around put_multipart
 - Adds tests for newly instrumented methods
 
## Are these changes tested?

Yes. Unit tests have been added for the new methods

Example output:
```sql
DataFusion CLI v50.2.0
> CREATE EXTERNAL TABLE
test(a bigint, b bigint)
STORED AS parquet LOCATION '../../test_table/';
0 row(s) fetched.
Elapsed 0.003 seconds.

> \object_store_profiling trace
ObjectStore Profile mode set to Trace
> INSERT INTO test values (1, 2), (3, 4);
+-------+
| count |
+-------+
| 2     |
+-------+
1 row(s) fetched.
Elapsed 0.007 seconds.

Object Store Profiling
Instrumented Object Store: instrument_mode: Trace, inner: LocalFileSystem(file:///)
2025-10-17T19:02:15.440246215+00:00 operation=List path=home/blake/open_source_src/datafusion-BlakeOrth/test_table
2025-10-17T19:02:15.444096012+00:00 operation=Put duration=0.000249s size=815 path=home/blake/open_source_src/datafusion-BlakeOrth/test_table/a9pjKBxSOtXZobJO_0.parquet

Summaries:
List
count: 1

Put
count: 1
duration min: 0.000249s
duration max: 0.000249s
duration avg: 0.000249s
size min: 815 B
size max: 815 B
size avg: 815 B
size sum: 815 B

>
```
(note: I have no idea how to exercise/show a multi-part put operation,
or if DataFusion even utilizes multipart puts for large files)

## Are there any user-facing changes?

No-ish

cc @alamb

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: BlakeOrth

datafusion-cli/src/object_storage/instrumented.rs

@@ -121,6 +121,54 @@ impl InstrumentedObjectStore {
             != InstrumentedObjectStoreMode::Disabled as u8
     }
 
+    async fn instrumented_put_opts(
+        &self,
+        location: &Path,
+        payload: PutPayload,
+        opts: PutOptions,
+    ) -> Result<PutResult> {
+        let timestamp = Utc::now();
+        let start = Instant::now();
+        let size = payload.content_length();
+        let ret = self.inner.put_opts(location, payload, opts).await?;
+        let elapsed = start.elapsed();
+
+        self.requests.lock().push(RequestDetails {
+            op: Operation::Put,
+            path: location.clone(),
+            timestamp,
+            duration: Some(elapsed),
+            size: Some(size),
+            range: None,
+            extra_display: None,
+        });
+
+        Ok(ret)
+    }
+
+    async fn instrumented_put_multipart(
+        &self,
+        location: &Path,
+        opts: PutMultipartOptions,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        let timestamp = Utc::now();
+        let start = Instant::now();
+        let ret = self.inner.put_multipart_opts(location, opts).await?;
+        let elapsed = start.elapsed();
+
+        self.requests.lock().push(RequestDetails {
+            op: Operation::Put,
+            path: location.clone(),
+            timestamp,
+            duration: Some(elapsed),
+            size: None,
+            range: None,
+            extra_display: None,
+        });
+
+        Ok(ret)
+    }
+
     async fn instrumented_get_opts(
         &self,
         location: &Path,
@@ -209,6 +257,10 @@ impl ObjectStore for InstrumentedObjectStore {
         payload: PutPayload,
         opts: PutOptions,
     ) -> Result<PutResult> {
+        if self.enabled() {
+            return self.instrumented_put_opts(location, payload, opts).await;
+        }
+
         self.inner.put_opts(location, payload, opts).await
     }
 
@@ -217,6 +269,10 @@ impl ObjectStore for InstrumentedObjectStore {
         location: &Path,
         opts: PutMultipartOptions,
     ) -> Result<Box<dyn MultipartUpload>> {
+        if self.enabled() {
+            return self.instrumented_put_multipart(location, opts).await;
+        }
+
         self.inner.put_multipart_opts(location, opts).await
     }
 
@@ -269,7 +325,7 @@ pub enum Operation {
     Get,
     _Head,
     List,
-    _Put,
+    Put,
 }
 
 impl fmt::Display for Operation {
@@ -599,6 +655,8 @@ impl ObjectStoreRegistry for InstrumentedObjectStoreRegistry {
 
 #[cfg(test)]
 mod tests {
+    use object_store::WriteMultipart;
+
     use super::*;
     use insta::assert_snapshot;
 
@@ -741,6 +799,72 @@ mod tests {
         assert!(request.extra_display.is_none());
     }
 
+    #[tokio::test]
+    async fn instrumented_store_put_opts() {
+        // The `setup_test_store()` method comes with data already `put` into it, so we'll setup
+        // manually for this test
+        let store = Arc::new(object_store::memory::InMemory::new());
+        let mode = AtomicU8::new(InstrumentedObjectStoreMode::default() as u8);
+        let instrumented = InstrumentedObjectStore::new(store, mode);
+
+        let path = Path::from("test/data");
+        let payload = PutPayload::from_static(b"test_data");
+        let size = payload.content_length();
+
+        // By default no requests should be instrumented/stored
+        assert!(instrumented.requests.lock().is_empty());
+        instrumented.put(&path, payload.clone()).await.unwrap();
+        assert!(instrumented.requests.lock().is_empty());
+
+        instrumented.set_instrument_mode(InstrumentedObjectStoreMode::Trace);
+        assert!(instrumented.requests.lock().is_empty());
+        instrumented.put(&path, payload).await.unwrap();
+        assert_eq!(instrumented.requests.lock().len(), 1);
+
+        let request = instrumented.take_requests().pop().unwrap();
+        assert_eq!(request.op, Operation::Put);
+        assert_eq!(request.path, path);
+        assert!(request.duration.is_some());
+        assert_eq!(request.size.unwrap(), size);
+        assert!(request.range.is_none());
+        assert!(request.extra_display.is_none());
+    }
+
+    #[tokio::test]
+    async fn instrumented_store_put_multipart() {
+        // The `setup_test_store()` method comes with data already `put` into it, so we'll setup
+        // manually for this test
+        let store = Arc::new(object_store::memory::InMemory::new());
+        let mode = AtomicU8::new(InstrumentedObjectStoreMode::default() as u8);
+        let instrumented = InstrumentedObjectStore::new(store, mode);
+
+        let path = Path::from("test/data");
+
+        // By default no requests should be instrumented/stored
+        assert!(instrumented.requests.lock().is_empty());
+        let mp = instrumented.put_multipart(&path).await.unwrap();
+        let mut write = WriteMultipart::new(mp);
+        write.write(b"test_data");
+        write.finish().await.unwrap();
+        assert!(instrumented.requests.lock().is_empty());
+
+        instrumented.set_instrument_mode(InstrumentedObjectStoreMode::Trace);
+        assert!(instrumented.requests.lock().is_empty());
+        let mp = instrumented.put_multipart(&path).await.unwrap();
+        let mut write = WriteMultipart::new(mp);
+        write.write(b"test_data");
+        write.finish().await.unwrap();
+        assert_eq!(instrumented.requests.lock().len(), 1);
+
+        let request = instrumented.take_requests().pop().unwrap();
+        assert_eq!(request.op, Operation::Put);
+        assert_eq!(request.path, path);
+        assert!(request.duration.is_some());
+        assert!(request.size.is_none());
+        assert!(request.range.is_none());
+        assert!(request.extra_display.is_none());
+    }
+
     #[test]
     fn request_details() {
         let rd = RequestDetails {
@@ -819,7 +943,7 @@ mod tests {
 
         // Add Put requests to test grouping
         requests.push(RequestDetails {
-            op: Operation::_Put,
+            op: Operation::Put,
             path: Path::from("test4"),
             timestamp: chrono::DateTime::from_timestamp(3, 0).unwrap(),
             duration: Some(Duration::from_millis(200)),
@@ -834,8 +958,8 @@ mod tests {
         +-----------+----------+-----------+-----------+-----------+------------+-------+
         | Get       | duration | 2.000000s | 8.000000s | 5.000000s | 15.000000s | 3     |
         | Get       | size     | 50 B      | 150 B     | 100 B     | 300 B      | 3     |
-        | _Put      | duration | 0.200000s | 0.200000s | 0.200000s | 0.200000s  | 1     |
-        | _Put      | size     | 75 B      | 75 B      | 75 B      | 75 B       | 1     |
+        | Put       | duration | 0.200000s | 0.200000s | 0.200000s | 0.200000s  | 1     |
+        | Put       | size     | 75 B      | 75 B      | 75 B      | 75 B       | 1     |
         +-----------+----------+-----------+-----------+-----------+------------+-------+
         ");
     }