fix(jindo): make async reads concurrency-safe

Author: zjw1111

CMakeLists.txt

@@ -57,6 +57,8 @@ option(PAIMON_ENABLE_AVRO "Whether to enable avro file format" ON)
 option(PAIMON_ENABLE_ORC "Whether to enable orc file format" ON)
 option(PAIMON_ENABLE_LANCE "Whether to enable lance file format" OFF)
 option(PAIMON_ENABLE_JINDO "Whether to enable jindo file system" OFF)
+option(PAIMON_ENABLE_JINDO_TESTS "Whether to enable Jindo tests that access remote OSS"
+       OFF)
 option(PAIMON_ENABLE_LUMINA "Whether to enable lumina vector index" OFF)
 option(PAIMON_ENABLE_LUCENE "Whether to enable lucene index" OFF)
 option(PAIMON_ENABLE_TANTIVY
@@ -70,6 +72,14 @@ endif()
 if(PAIMON_ENABLE_JINDO)
     add_definitions(-DPAIMON_ENABLE_JINDO)
 endif()
+if(PAIMON_ENABLE_JINDO_TESTS)
+    if(NOT PAIMON_BUILD_TESTS)
+        message(FATAL_ERROR "PAIMON_ENABLE_JINDO_TESTS requires PAIMON_BUILD_TESTS=ON")
+    endif()
+    if(NOT PAIMON_ENABLE_JINDO)
+        message(FATAL_ERROR "PAIMON_ENABLE_JINDO_TESTS requires PAIMON_ENABLE_JINDO=ON")
+    endif()
+endif()
 if(PAIMON_USE_CXX11_ABI)
     add_definitions(-D_GLIBCXX_USE_CXX11_ABI=1)
 else()

cmake_modules/DefineOptions.cmake

@@ -107,6 +107,9 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
 
     define_option(PAIMON_BUILD_TESTS "Build the Paimon googletest unit tests" OFF)
 
+    define_option(PAIMON_ENABLE_JINDO_TESTS "Enable Jindo tests that access remote OSS"
+                  OFF)
+
     define_option(PAIMON_BUILD_BENCHMARKS
                   "Build the Paimon Google Benchmark performance benchmarks" OFF)
 

include/paimon/fs/file_system.h

@@ -98,6 +98,8 @@ class PAIMON_EXPORT InputStream : public Stream {
     /// @param callback The callback function to be invoked upon completion of the read operation.
     ///                 The callback will receive a Status object indicating the success or failure
     ///                 of the read operation.
+    /// @note The caller must keep the input stream, buffer, and any externally referenced callback
+    ///       state alive until the callback is invoked.
     virtual void ReadAsync(char* buffer, int64_t size, int64_t offset,
                            std::function<void(Status)>&& callback) = 0;
 

src/paimon/CMakeLists.txt

@@ -764,18 +764,31 @@ if(PAIMON_BUILD_TESTS)
                     ${TEST_STATIC_LINK_LIBS}
                     ${GTEST_LINK_TOOLCHAIN})
 
+    set(PAIMON_JINDO_FS_TEST_SOURCES)
+    set(PAIMON_JINDO_FS_TEST_LINK_LIBS)
+    if(PAIMON_ENABLE_JINDO_TESTS)
+        list(APPEND PAIMON_JINDO_FS_TEST_SOURCES
+             fs/jindo/jindo_file_system_factory_test.cpp
+             fs/jindo/jindo_file_system_test.cpp)
+        list(APPEND PAIMON_JINDO_FS_TEST_LINK_LIBS
+             ${PAIMON_JINDO_FILE_SYSTEM_STATIC_LINK_LIBS})
+    endif()
+
     add_paimon_test(fs_test
                     SOURCES
                     common/fs/file_system_test.cpp
                     common/fs/resolving_file_system_test.cpp
                     fs/local/local_file_test.cpp
-                    # fs/jindo/jindo_file_system_factory_test.cpp
-                    # fs/jindo/jindo_file_system_test.cpp
+                    ${PAIMON_JINDO_FS_TEST_SOURCES}
                     STATIC_LINK_LIBS
                     paimon_shared
                     ${PAIMON_LOCAL_FILE_SYSTEM_STATIC_LINK_LIBS}
-                    # ${PAIMON_JINDO_FILE_SYSTEM_STATIC_LINK_LIBS}
+                    ${PAIMON_JINDO_FS_TEST_LINK_LIBS}
                     test_utils_static
                     ${GTEST_LINK_TOOLCHAIN})
 
+    if(PAIMON_ENABLE_JINDO_TESTS)
+        target_compile_definitions(paimon-fs-test PRIVATE PAIMON_ENABLE_JINDO_TESTS)
+    endif()
+
 endif()

src/paimon/common/fs/file_system_test.cpp

@@ -1474,7 +1474,9 @@ TEST_P(FileSystemTest, TestAtomicStoreAlreadyExist) {
 std::vector<std::string> GetTestValuesForFileSystemTest() {
     std::vector<std::string> values;
     values.emplace_back("local");
-    // values.emplace_back("jindo");
+#ifdef PAIMON_ENABLE_JINDO_TESTS
+    values.emplace_back("jindo");
+#endif
     return values;
 }
 

src/paimon/fs/jindo/jindo_file_system.cpp

@@ -16,6 +16,7 @@
 
 #include "paimon/fs/jindo/jindo_file_system.h"
 
+#include <atomic>
 #include <cassert>
 #include <utility>
 
@@ -52,6 +53,29 @@ class JindoFileSystemImpl {
     std::unique_ptr<JdoFileSystem> fs_;
 };
 
+namespace {
+
+class AsyncReadState {
+ public:
+    explicit AsyncReadState(std::function<void(Status)>&& callback)
+        : callback_(std::move(callback)) {}
+
+    void Complete(JdoStatus status) {
+        if (completed_.exchange(true)) {
+            return;
+        }
+        callback_(status.ok() ? Status::OK() : Status::IOError(status.errMsg()));
+    }
+
+    std::string_view result;
+
+ private:
+    std::atomic<bool> completed_{false};
+    std::function<void(Status)> callback_;
+};
+
+}  // namespace
+
 JindoFileSystem::JindoFileSystem(std::unique_ptr<JdoFileSystem>&& fs)
     : impl_(std::make_shared<JindoFileSystemImpl>(std::move(fs))) {}
 
@@ -215,15 +239,17 @@ Result<int64_t> JindoInputStream::Length() const {
 
 Result<int64_t> JindoInputStream::Read(char* buffer, int64_t size) {
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "read length"));
-    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->read(size, &result_, buffer));
-    return result_.length();
+    std::string_view result;
+    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->read(size, &result, buffer));
+    return result.length();
 }
 
 Result<int64_t> JindoInputStream::Read(char* buffer, int64_t size, int64_t offset) {
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(size, "read length"));
     PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(offset, "read offset"));
-    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->pread(offset, size, &result_, buffer));
-    return result_.length();
+    std::string_view result;
+    PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->pread(offset, size, &result, buffer));
+    return result.length();
 }
 
 void JindoInputStream::ReadAsync(char* buffer, int64_t size, int64_t offset,
@@ -238,12 +264,16 @@ void JindoInputStream::ReadAsync(char* buffer, int64_t size, int64_t offset,
         callback(validate_status);
         return;
     }
-    auto outer_callback = [=](JdoStatus status) {
-        callback(status.ok() ? Status::OK() : Status::IOError(status.errMsg()));
-    };
-    auto task = reader_->preadAsync(offset, size, &result_, buffer, outer_callback);
+    std::shared_ptr<AsyncReadState> state = std::make_shared<AsyncReadState>(std::move(callback));
+    auto task = reader_->preadAsync(offset, size, &state->result, buffer,
+                                    [state](JdoStatus status) { state->Complete(status); });
     assert(task);
-    [[maybe_unused]] auto perform_status = task->perform();
+
+    auto perform_status = task->perform();
+    if (!perform_status.ok()) {
+        state->Complete(perform_status);
+        return;
+    }
 }
 
 Status JindoInputStream::Close() {

src/paimon/fs/jindo/jindo_file_system.h

@@ -20,7 +20,6 @@
 #include <functional>
 #include <memory>
 #include <string>
-#include <string_view>
 #include <vector>
 
 #include "JdoFileSystem.hpp"  // NOLINT(build/include_subdir)
@@ -80,7 +79,6 @@ class JindoInputStream : public InputStream {
     // the Jindo Reader.
     std::shared_ptr<JindoFileSystemImpl> fs_;
     std::unique_ptr<JdoReader> reader_;
-    std::string_view result_;
 };
 
 class JindoOutputStream : public OutputStream {

src/paimon/fs/jindo/jindo_file_system_test.cpp

@@ -14,10 +14,19 @@
  * limitations under the License.
  */
 
+#include <chrono>
+#include <cstdint>
+#include <future>
+#include <string>
+#include <unordered_set>
+#include <vector>
+
 #include "gtest/gtest.h"
 #include "paimon/fs/jindo/jindo_file_system_factory.h"
 #include "paimon/testing/utils/testharness.h"
+
 namespace paimon::jindo::test {
+
 // This test shows inconsistent behavior with the local file system in some abnormal scenarios.
 class JindoFileSystemTest : public ::testing::Test {
  public:
@@ -124,4 +133,73 @@ TEST_F(JindoFileSystemTest, TestSeek) {
     ASSERT_OK(in_stream->Close());
 }
 
+TEST(JindoFileSystemPaginationTest, TestListDirAcrossOssPageBoundary) {
+    constexpr int32_t kFileCount = 1234;
+    const std::string test_dir = "oss://paimon-unittest/test_data/jindo_listdir_truncated_1234/";
+    std::map<std::string, std::string> options = paimon::test::GetJindoTestOptions();
+
+    auto fs_factory = std::make_shared<JindoFileSystemFactory>();
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileSystem> fs, fs_factory->Create(test_dir, options));
+
+    std::vector<std::unique_ptr<BasicFileStatus>> file_statuses;
+    ASSERT_OK(fs->ListDir(test_dir, &file_statuses));
+    ASSERT_EQ(file_statuses.size(), kFileCount);
+
+    std::unordered_set<std::string> actual_paths;
+    for (const std::unique_ptr<BasicFileStatus>& file_status : file_statuses) {
+        ASSERT_TRUE(actual_paths.insert(file_status->GetPath()).second)
+            << "duplicate path: " << file_status->GetPath();
+    }
+    for (int32_t i = 0; i < kFileCount; ++i) {
+        std::string index = std::to_string(i);
+        index.insert(/*pos=*/0, /*count=*/4 - index.size(), /*ch=*/'0');
+        ASSERT_NE(actual_paths.find(test_dir + "file-" + index + ".txt"), actual_paths.end());
+    }
+}
+
+TEST(JindoFileSystemAsyncReadTest, TestConcurrentReadAsyncAndReadFromOss) {
+    constexpr int32_t kConcurrentReads = 64;
+    constexpr int64_t kAsyncReadSize = 7;
+    const std::string file_path = "oss://paimon-unittest/test_data/jindo_read_async_128mb.bin";
+    std::map<std::string, std::string> options = paimon::test::GetJindoTestOptions();
+    auto fs_factory = std::make_shared<JindoFileSystemFactory>();
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileSystem> fs, fs_factory->Create(file_path, options));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input_stream, fs->Open(file_path));
+
+    std::vector<std::vector<char>> async_buffers(kConcurrentReads,
+                                                 std::vector<char>(kAsyncReadSize));
+    std::vector<std::promise<Status>> promises(kConcurrentReads);
+    std::vector<std::future<Status>> futures;
+    futures.reserve(kConcurrentReads);
+    for (std::promise<Status>& promise : promises) {
+        futures.push_back(promise.get_future());
+    }
+
+    std::vector<Status> sync_read_statuses;
+    std::vector<int64_t> sync_read_sizes;
+    sync_read_statuses.reserve(kConcurrentReads);
+    sync_read_sizes.reserve(kConcurrentReads);
+    for (int32_t i = 0; i < kConcurrentReads; ++i) {
+        input_stream->ReadAsync(
+            async_buffers[i].data(), async_buffers[i].size(), /*offset=*/0,
+            [&promises, i](Status status) { promises[i].set_value(std::move(status)); });
+
+        char sync_buffer = 0;
+        Result<int64_t> sync_read_result =
+            input_stream->Read(&sync_buffer, /*size=*/1, /*offset=*/i);
+        sync_read_statuses.push_back(sync_read_result.status());
+        sync_read_sizes.push_back(sync_read_result.ok() ? sync_read_result.value() : -1);
+    }
+
+    for (int32_t i = 0; i < kConcurrentReads; ++i) {
+        ASSERT_EQ(futures[i].wait_for(std::chrono::seconds(60)), std::future_status::ready)
+            << "async read=" << i;
+        ASSERT_OK(futures[i].get());
+    }
+    for (int32_t i = 0; i < kConcurrentReads; ++i) {
+        ASSERT_OK(sync_read_statuses[i]) << "sync read=" << i;
+        ASSERT_EQ(sync_read_sizes[i], 1) << "sync read=" << i;
+    }
+}
+
 }  // namespace paimon::jindo::test

test/inte/CMakeLists.txt

@@ -40,6 +40,10 @@ if(PAIMON_BUILD_TESTS)
                     ${TEST_STATIC_LINK_LIBS}
                     test_utils_static
                     ${GTEST_LINK_TOOLCHAIN})
+    if(PAIMON_ENABLE_JINDO_TESTS)
+        target_compile_definitions(paimon-write-and-read-inte-test
+                                   PRIVATE PAIMON_ENABLE_JINDO_TESTS)
+    endif()
 
     add_paimon_test(clean_inte_test
                     STATIC_LINK_LIBS

test/inte/write_and_read_inte_test.cpp

@@ -1096,7 +1096,9 @@ TEST_P(WriteAndReadInteTest, TestCharVarcharBinaryVarbinaryTypes) {
 
 std::vector<std::pair<std::string, std::string>> GetTestValuesForWriteAndReadInteTest() {
     std::vector<std::pair<std::string, std::string>> values = {{"parquet", "local"}};
-    // values.emplace_back("parquet", "jindo");
+#ifdef PAIMON_ENABLE_JINDO_TESTS
+    values.emplace_back("parquet", "jindo");
+#endif
 #ifdef PAIMON_ENABLE_ORC
     values.emplace_back("orc", "local");
 #endif