From e785d47d903d87f3b44892e1670c2b7cf827622b Mon Sep 17 00:00:00 2001
From: vkorukanti <venki.korukanti@gmail.com>
Date: Tue, 30 Apr 2024 22:10:02 -0700
Subject: [PATCH] f

---
 .../delta/kernel/client/FileSystemClient.java  | 10 ++++++++++
 .../kernel/test/MockTableClientUtils.scala     |  3 +++
 .../client/DefaultFileSystemClient.java        |  7 +++++++
 .../client/DefaultFileSystemClientSuite.scala  | 18 ++++++++++++++++++
 4 files changed, 38 insertions(+)

diff --git a/kernel/kernel-api/src/main/java/io/delta/kernel/client/FileSystemClient.java b/kernel/kernel-api/src/main/java/io/delta/kernel/client/FileSystemClient.java
index da119adb206..9153cebc014 100644
--- a/kernel/kernel-api/src/main/java/io/delta/kernel/client/FileSystemClient.java
+++ b/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;
 }
diff --git a/kernel/kernel-api/src/test/scala/io/delta/kernel/test/MockTableClientUtils.scala b/kernel/kernel-api/src/test/scala/io/delta/kernel/test/MockTableClientUtils.scala
index 5d2fc8c92da..7de85a809a8 100644
--- a/kernel/kernel-api/src/test/scala/io/delta/kernel/test/MockTableClientUtils.scala
+++ b/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")
 }
diff --git a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/client/DefaultFileSystemClient.java b/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/client/DefaultFileSystemClient.java
index a7b7aca9586..9924b4b227d 100644
--- a/kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/client/DefaultFileSystemClient.java
+++ b/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 {
diff --git a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/client/DefaultFileSystemClientSuite.scala b/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/client/DefaultFileSystemClientSuite.scala
index 738fbeb4db2..97ee2a959db 100644
--- a/kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/client/DefaultFileSystemClientSuite.scala
+++ b/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)))
+    }
+  }
 }