f

kernel/kernel-api/src/main/java/io/delta/kernel/client/FileSystemClient.java

@@ -67,4 +67,14 @@ public interface FileSystemClient {
     CloseableIterator<ByteArrayInputStream> readFiles(
         CloseableIterator<FileReadRequest> readRequests)
         throws IOException;
+
+    /**
+     * Create a directory at the given path including parent directories. This mimicks the behavior
+     * of `mkdir -p` in Unix.
+     *
+     * @param path Full qualified path to create a directory at.
+     * @return true if the directory was created successfully, false otherwise.
+     * @throws IOException for any IO error.
+     */
+    boolean mkdirs(String path) throws IOException;
 }

kernel/kernel-api/src/test/scala/io/delta/kernel/test/MockTableClientUtils.scala

@@ -160,4 +160,7 @@ trait BaseMockFileSystemClient extends FileSystemClient {
   override def readFiles(
       readRequests: CloseableIterator[FileReadRequest]): CloseableIterator[ByteArrayInputStream] =
     throw new UnsupportedOperationException("not supported in this test suite")
+
+  override def mkdirs(path: String): Boolean =
+    throw new UnsupportedOperationException("not supported in this test suite")
 }

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/client/DefaultFileSystemClient.java

@@ -97,6 +97,13 @@ public CloseableIterator<ByteArrayInputStream> readFiles(
                 getStream(elem.getPath(), elem.getStartOffset(), elem.getReadLength()));
     }
 
+    @Override
+    public boolean mkdirs(String path) throws IOException {
+        Path pathObject = new Path(path);
+        FileSystem fs = pathObject.getFileSystem(hadoopConf);
+        return fs.mkdirs(pathObject);
+    }
+
     private ByteArrayInputStream getStream(String filePath, int offset, int size) {
         Path path = new Path(filePath);
         try {

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/client/DefaultFileSystemClientSuite.scala

@@ -20,11 +20,13 @@ import java.io.FileNotFoundException
 import scala.collection.mutable.ArrayBuffer
 
 import io.delta.kernel.defaults.utils.TestUtils
+import org.apache.hadoop.fs.{FileSystem, Path}
 import org.scalatest.funsuite.AnyFunSuite
 
 class DefaultFileSystemClientSuite extends AnyFunSuite with TestUtils {
 
   val fsClient = defaultTableClient.getFileSystemClient
+  val fs = FileSystem.get(configuration)
 
   test("list from file") {
     val basePath = fsClient.resolvePath(getTestResourceFilePath("json-files"))
@@ -62,4 +64,20 @@ class DefaultFileSystemClientSuite extends AnyFunSuite with TestUtils {
     val resolvedPath = fsClient.resolvePath(inputPath)
     assert("file:" + inputPath === resolvedPath)
   }
+
+  test("mkdirs") {
+    withTempDir { tempdir =>
+      val dir1 = tempdir + "/test"
+      assert(fsClient.mkdirs(dir1))
+      assert(fs.exists(new Path(dir1)))
+
+      val dir2 = tempdir + "/test1/test2" // nested
+      assert(fsClient.mkdirs(dir2))
+      assert(fs.exists(new Path(dir2)))
+
+      val dir3 = "/non-existentfileTable/sfdsd"
+      assert(!fsClient.mkdirs(dir3))
+      assert(!fs.exists(new Path(dir3)))
+    }
+  }
 }