fix(fs): create parent directories before opening Jindo writers (alibaba#430)

Author: Zouxxyy

src/paimon/CMakeLists.txt

@@ -796,7 +796,8 @@ if(PAIMON_BUILD_TESTS)
     # OSS access and stay disabled.
     set(FS_TEST_JINDO_SOURCES)
     if(PAIMON_ENABLE_JINDO)
-        list(APPEND FS_TEST_JINDO_SOURCES fs/jindo/jindo_utils_test.cpp)
+        list(APPEND FS_TEST_JINDO_SOURCES fs/jindo/jindo_file_system_unit_test.cpp
+             fs/jindo/jindo_utils_test.cpp)
     endif()
 
     add_paimon_test(fs_test

src/paimon/fs/jindo/jindo_file_system.cpp

@@ -26,6 +26,7 @@
 #include "fmt/format.h"
 #include "jdo_error.h"  // NOLINT(build/include_subdir)
 #include "paimon/common/utils/math.h"
+#include "paimon/common/utils/path_util.h"
 #include "paimon/fs/jindo/jindo_file_status.h"
 #include "paimon/fs/jindo/jindo_utils.h"
 
@@ -68,6 +69,18 @@ Result<std::unique_ptr<OutputStream>> JindoFileSystem::Create(const std::string&
         return Status::Invalid(
             fmt::format("do not allow overwrite, but the file {} already exists", path));
     }
+    const std::string parent_path = PathUtil::GetParentDirPath(path);
+    if (!parent_path.empty()) {
+        PAIMON_ASSIGN_OR_RAISE(Path parent, PathUtil::ToPath(parent_path));
+        // Do not issue mkdir for scheme-only or authority-only URI parents.
+        if (!parent.path.empty()) {
+            PAIMON_RETURN_NOT_OK(Mkdirs(parent_path));
+        }
+    }
+    return OpenWriter(path);
+}
+
+Result<std::unique_ptr<OutputStream>> JindoFileSystem::OpenWriter(const std::string& path) const {
     std::unique_ptr<JdoWriter> writer;
     PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->openWriter(path, &writer));
     return std::make_unique<JindoOutputStream>(impl_, std::move(writer));

src/paimon/fs/jindo/jindo_file_system.h

@@ -57,6 +57,9 @@ class JindoFileSystem : public FileSystem {
 
     Result<bool> Exists(const std::string& path) const override;
 
+ protected:
+    virtual Result<std::unique_ptr<OutputStream>> OpenWriter(const std::string& path) const;
+
  private:
     std::shared_ptr<JindoFileSystemImpl> impl_;
 };

src/paimon/fs/jindo/jindo_file_system_unit_test.cpp

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/fs/jindo/jindo_file_system.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class TestOutputStream : public OutputStream {
+ public:
+    Result<int64_t> GetPos() const override {
+        return 0;
+    }
+
+    Result<int64_t> Write(const char*, int64_t size) override {
+        return size;
+    }
+
+    Status Flush() override {
+        return Status::OK();
+    }
+
+    Status Close() override {
+        return Status::OK();
+    }
+
+    Result<std::string> GetUri() const override {
+        return std::string();
+    }
+};
+
+class TestJindoFileSystem : public jindo::JindoFileSystem {
+ public:
+    TestJindoFileSystem() : JindoFileSystem(std::make_unique<JdoFileSystem>()) {}
+
+    Result<bool> Exists(const std::string&) const override {
+        return false;
+    }
+
+    Status Mkdirs(const std::string& path) const override {
+        parent_path_ = path;
+        calls_.push_back("mkdirs");
+        return mkdirs_status_;
+    }
+
+    void SetMkdirsStatus(Status status) {
+        mkdirs_status_ = std::move(status);
+    }
+
+    const std::string& GetParentPath() const {
+        return parent_path_;
+    }
+
+    bool IsWriterOpened() const {
+        return writer_opened_;
+    }
+
+    const std::vector<std::string>& GetCalls() const {
+        return calls_;
+    }
+
+ protected:
+    Result<std::unique_ptr<OutputStream>> OpenWriter(const std::string&) const override {
+        calls_.push_back("open_writer");
+        writer_opened_ = true;
+        std::unique_ptr<OutputStream> output = std::make_unique<TestOutputStream>();
+        return output;
+    }
+
+ private:
+    mutable std::string parent_path_;
+    mutable bool writer_opened_ = false;
+    mutable std::vector<std::string> calls_;
+    Status mkdirs_status_ = Status::OK();
+};
+
+TEST(JindoFileSystemUnitTest, CreateMakesParentDirectoryBeforeOpeningWriter) {
+    TestJindoFileSystem fs;
+    const std::string path = "oss://bucket/table/bucket-24/data.orc";
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+                         fs.Create(path, /*overwrite=*/false));
+    ASSERT_TRUE(output);
+    ASSERT_EQ(fs.GetParentPath(), "oss://bucket/table/bucket-24");
+    ASSERT_TRUE(fs.IsWriterOpened());
+    ASSERT_EQ(fs.GetCalls().size(), 2);
+    ASSERT_EQ(fs.GetCalls()[0], "mkdirs");
+    ASSERT_EQ(fs.GetCalls()[1], "open_writer");
+}
+
+TEST(JindoFileSystemUnitTest, CreateSkipsObjectStoreRootParent) {
+    TestJindoFileSystem fs;
+    const std::string path = "oss://bucket/data.orc";
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+                         fs.Create(path, /*overwrite=*/false));
+    ASSERT_TRUE(output);
+    ASSERT_TRUE(fs.GetParentPath().empty());
+    ASSERT_TRUE(fs.IsWriterOpened());
+    ASSERT_EQ(fs.GetCalls().size(), 1);
+    ASSERT_EQ(fs.GetCalls()[0], "open_writer");
+}
+
+TEST(JindoFileSystemUnitTest, CreateReturnsParentDirectoryFailure) {
+    TestJindoFileSystem fs;
+    const std::string path = "oss://bucket/table/bucket-24/data.orc";
+    fs.SetMkdirsStatus(Status::IOError("failed to create parent directory"));
+
+    ASSERT_NOK_WITH_MSG(fs.Create(path, /*overwrite=*/false), "failed to create parent directory");
+    ASSERT_EQ(fs.GetParentPath(), "oss://bucket/table/bucket-24");
+    ASSERT_FALSE(fs.IsWriterOpened());
+    ASSERT_EQ(fs.GetCalls().size(), 1);
+    ASSERT_EQ(fs.GetCalls()[0], "mkdirs");
+}
+
+}  // namespace paimon::test