feat!: new Engine StorageHandler head API (#1465)

## What changes are proposed in this pull request?
In CCv2 we need a head API to fetch metadata about commits after they
are published. This is similar to the
[getFileStatus](https://github.com/delta-io/delta/blob/master/kernel/kernel-api/src/main/java/io/delta/kernel/engine/FileSystemClient.java#L96)
API in the java kernel. This PR (1) adds the new `head` method to the
`StorageHandler` trait, and (2) implements it for the default engine


### This PR affects the following public APIs
New `head` API for `StorageHandler`

## How was this change tested?
New UT for default engine impl

Author: zachschuermann

kernel/src/engine/default/filesystem.rs

@@ -200,6 +200,23 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
             Ok(())
         })
     }
+
+    fn head(&self, path: &Url) -> DeltaResult<FileMeta> {
+        let store = self.inner.clone();
+        let url = path.clone();
+        let path = Path::from_url_path(path.path())?;
+        self.task_executor.block_on(async move {
+            store
+                .head(&path)
+                .await
+                .map_err(Into::into)
+                .map(|meta| FileMeta {
+                    location: url,
+                    last_modified: meta.last_modified.timestamp_millis(),
+                    size: meta.size,
+                })
+        })
+    }
 }
 
 #[cfg(test)]
@@ -357,4 +374,45 @@ mod tests {
         let new_dest_url = Url::from_file_path(tmp.path().join("new_dest.txt")).unwrap();
         assert!(handler.copy_atomic(&missing_url, &new_dest_url).is_err());
     }
+
+    #[tokio::test]
+    async fn test_head() {
+        let tmp = tempfile::tempdir().unwrap();
+        let store = Arc::new(LocalFileSystem::new());
+        let executor = Arc::new(TokioBackgroundExecutor::new());
+        let handler = ObjectStoreStorageHandler::new(store.clone(), executor);
+
+        let data = Bytes::from("test-content");
+        let file_path = Path::from_absolute_path(tmp.path().join("test.txt")).unwrap();
+        let write_time = current_time_duration().unwrap();
+        store.put(&file_path, data.clone().into()).await.unwrap();
+
+        let file_url = Url::from_file_path(tmp.path().join("test.txt")).unwrap();
+        let file_meta = handler.head(&file_url).unwrap();
+
+        assert_eq!(file_meta.location, file_url);
+        assert_eq!(file_meta.size, data.len() as u64);
+
+        // Verify timestamp is within the expected range
+        let meta_time = Duration::from_millis(file_meta.last_modified as u64);
+        assert!(
+            meta_time.abs_diff(write_time) < Duration::from_millis(100),
+            "last_modified timestamp should be around {} ms, but was {} ms",
+            write_time.as_millis(),
+            meta_time.as_millis()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_head_non_existent() {
+        let tmp = tempfile::tempdir().unwrap();
+        let store = Arc::new(LocalFileSystem::new());
+        let executor = Arc::new(TokioBackgroundExecutor::new());
+        let handler = ObjectStoreStorageHandler::new(store, executor);
+
+        let missing_url = Url::from_file_path(tmp.path().join("missing.txt")).unwrap();
+        let result = handler.head(&missing_url);
+
+        assert!(matches!(result, Err(Error::FileNotFound(_))));
+    }
 }

kernel/src/engine/sync/storage.rs

@@ -74,6 +74,10 @@ impl StorageHandler for SyncStorageHandler {
     fn copy_atomic(&self, _src: &Url, _dest: &Url) -> DeltaResult<()> {
         unimplemented!("SyncStorageHandler does not implement copy");
     }
+
+    fn head(&self, _path: &Url) -> DeltaResult<FileMeta> {
+        unimplemented!("head is not implemented for SyncStorageHandler")
+    }
 }
 
 #[cfg(test)]

kernel/src/lib.rs

@@ -539,6 +539,11 @@ pub trait StorageHandler: AsAny {
     /// Copy a file atomically from source to destination. If the destination file already exists,
     /// it must return Err(Error::FileAlreadyExists).
     fn copy_atomic(&self, src: &Url, dest: &Url) -> DeltaResult<()>;
+
+    /// Perform a HEAD request for the given file at a Url, returning the file metadata.
+    ///
+    /// If the file does not exist, this must return an `Err` with [`Error::FileNotFound`].
+    fn head(&self, path: &Url) -> DeltaResult<FileMeta>;
 }
 
 /// Provides JSON handling functionality to Delta Kernel.

kernel/src/listed_log_files.rs

@@ -637,6 +637,9 @@ mod list_log_files_with_log_tail_tests {
             fn copy_atomic(&self, src: &Url, dest: &Url) -> DeltaResult<()> {
                 panic!("copy used from {src} to {dest}");
             }
+            fn head(&self, _path: &Url) -> DeltaResult<crate::FileMeta> {
+                panic!("head used");
+            }
         }
 
         // when log_tail covers the entire requested range, no filesystem listing should occur