fix

Author: zjw1111

src/paimon/fs/local/local_file.cpp

@@ -151,15 +151,8 @@ Result<bool> LocalFile::Mkdir() const {
             dir.resize(len - 1);
         }
     }
-    size_t pos = dir.rfind('/');
-    if (pos == std::string::npos) {
-        return mkdir(dir.c_str(), 0755) == 0;
-    }
-    std::string parent_dir = dir.substr(0, pos);
-    if (!parent_dir.empty() && access(parent_dir.c_str(), F_OK) != 0) {
-        PAIMON_RETURN_NOT_OK(MkNestDir(parent_dir));
-    }
-    return mkdir(dir.c_str(), 0755) == 0;
+    PAIMON_ASSIGN_OR_RAISE(bool success, MkNestDir(dir));
+    return success;
 }
 
 Result<std::unique_ptr<LocalFileStatus>> LocalFile::GetFileStatus() const {

src/paimon/fs/local/local_file_system.cpp

@@ -76,7 +76,7 @@ Status LocalFileSystem::Mkdirs(const std::string& path) const {
 }
 
 Status LocalFileSystem::MkdirsInternal(const LocalFile& file) const {
-    // Important: The 'Exists()' check above must come before the 'IsDirectory()'
+    // Important: The 'Exists()' check above must come before the 'IsDir()'
     // check to be safe when multiple parallel instances try to create the directory
     PAIMON_ASSIGN_OR_RAISE(bool is_exist, file.Exists());
     if (is_exist) {
@@ -90,6 +90,10 @@ Status LocalFileSystem::MkdirsInternal(const LocalFile& file) const {
         }
     }
 
+    auto parent = file.GetParentFile();
+    if (!parent.IsEmpty()) {
+        PAIMON_RETURN_NOT_OK(MkdirsInternal(parent));
+    }
     PAIMON_ASSIGN_OR_RAISE(bool success, file.Mkdir());
     if (!success) {
         PAIMON_ASSIGN_OR_RAISE(bool is_dir, file.IsDir());