fix(jindo): make async reads concurrency-safe

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_NETWORK_TESTS
+       "Whether to enable tests that access real remote services over the network" 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,11 @@ endif()
 if(PAIMON_ENABLE_JINDO)
     add_definitions(-DPAIMON_ENABLE_JINDO)
 endif()
+if(PAIMON_ENABLE_NETWORK_TESTS)
+    if(NOT PAIMON_BUILD_TESTS)
+        message(FATAL_ERROR "PAIMON_ENABLE_NETWORK_TESTS requires PAIMON_BUILD_TESTS=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_NETWORK_TESTS
+                  "Enable tests that access real remote services over the network" OFF)
+
     define_option(PAIMON_BUILD_BENCHMARKS
                   "Build the Paimon Google Benchmark performance benchmarks" OFF)
 

include/paimon/fs/file_system.h

@@ -98,6 +98,7 @@ 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 and buffer 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

@@ -791,29 +791,36 @@ if(PAIMON_BUILD_TESTS)
                     ${TEST_STATIC_LINK_LIBS}
                     ${GTEST_LINK_TOOLCHAIN})
 
-    # jindo_utils_test only checks the in-memory status conversion and does not need an
-    # OSS cluster, so it runs whenever jindo is built. The other jindo tests need real
-    # OSS access and stay disabled.
-    set(FS_TEST_JINDO_SOURCES)
+    set(PAIMON_JINDO_FS_TEST_SOURCES)
+    set(PAIMON_JINDO_FS_TEST_LINK_LIBS)
     if(PAIMON_ENABLE_JINDO)
-        list(APPEND FS_TEST_JINDO_SOURCES fs/jindo/jindo_utils_test.cpp)
+        list(APPEND PAIMON_JINDO_FS_TEST_SOURCES fs/jindo/jindo_utils_test.cpp)
+        if(PAIMON_ENABLE_NETWORK_TESTS)
+            list(APPEND PAIMON_JINDO_FS_TEST_SOURCES
+                 fs/jindo/jindo_file_system_factory_test.cpp
+                 fs/jindo/jindo_file_system_test.cpp)
+        endif()
+        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_TEST_JINDO_SOURCES}
-                    # 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}
                     EXTRA_INCLUDES
                     ${JINDOSDK_INCLUDE_DIR})
 
+    if(PAIMON_ENABLE_NETWORK_TESTS)
+        target_compile_definitions(paimon-fs-test PRIVATE PAIMON_ENABLE_NETWORK_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");
+#if defined(PAIMON_ENABLE_NETWORK_TESTS) && defined(PAIMON_ENABLE_JINDO)
+    values.emplace_back("jindo");
+#endif
     return values;
 }
 

src/paimon/core/operation/key_value_file_store_write_test.cpp

@@ -320,7 +320,6 @@ TEST_F(KeyValueFileStoreWriteTest, TestSharedShreddingMapRestoreInitializesNextW
         {"fields.tags.map.shared-shredding.column-placement-policy", "plain"},
         {"write-only", "true"},
         {"bucket", "1"},
-        {"enable-pk-commit-in-inte-test", ""},
     };
     auto logical_schema = arrow::schema({
         arrow::field("id", arrow::int32(), /*nullable=*/false),

src/paimon/fs/jindo/jindo_file_system.cpp

@@ -16,7 +16,9 @@
 
 #include "paimon/fs/jindo/jindo_file_system.h"
 
+#include <atomic>
 #include <cassert>
+#include <string_view>
 #include <utility>
 
 #include "JdoFileInfo.hpp"    // NOLINT(build/include_subdir)
@@ -52,6 +54,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 +240,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 +265,17 @@ 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);
+        [[maybe_unused]] auto status = task->cancel();
+        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

src/paimon/testing/utils/test_helper.h

@@ -59,19 +59,16 @@ class TestHelper {
         const std::vector<std::string>& primary_keys,
         const std::map<std::string, std::string>& options, bool is_streaming_mode,
         bool ignore_if_exists = false, const std::string& temp_directory = "") {
-        // only for test && only check the key
-        auto new_options = options;
-        new_options["enable-object-store-catalog-in-inte-test"] = "";
-        PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, new_options));
-        PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", new_options, ignore_if_exists));
+        PAIMON_ASSIGN_OR_RAISE(auto catalog, Catalog::Create(root_path, options));
+        PAIMON_RETURN_NOT_OK(catalog->CreateDatabase("foo", options, ignore_if_exists));
         ::ArrowSchema c_schema;
         ScopeGuard guard([schema = &c_schema]() { ArrowSchemaRelease(schema); });
         PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &c_schema));
         PAIMON_RETURN_NOT_OK(catalog->CreateTable(Identifier("foo", "bar"), &c_schema,
-                                                  partition_keys, primary_keys, new_options,
+                                                  partition_keys, primary_keys, options,
                                                   ignore_if_exists));
         std::string table_path = PathUtil::JoinPath(root_path, "foo.db/bar");
-        return Create(table_path, new_options, is_streaming_mode, temp_directory);
+        return Create(table_path, options, is_streaming_mode, temp_directory);
     }
 
     static Result<std::unique_ptr<TestHelper>> Create(
@@ -96,14 +93,10 @@ class TestHelper {
                                    .Finish());
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreWrite> write,
                                FileStoreWrite::Create(std::move(write_context)));
-        std::map<std::string, std::string> new_options = options;
-        // only for test && only check the key
-        new_options["enable-pk-commit-in-inte-test"] = "";
-        new_options["enable-object-store-commit-in-inte-test"] = "";
         CommitContextBuilder commit_context_builder(table_path, commit_user);
         PAIMON_ASSIGN_OR_RAISE(
             std::unique_ptr<CommitContext> commit_context,
-            commit_context_builder.SetOptions(new_options).IgnoreEmptyCommit(false).Finish());
+            commit_context_builder.SetOptions(options).IgnoreEmptyCommit(false).Finish());
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> commit,
                                FileStoreCommit::Create(std::move(commit_context)));
         return std::unique_ptr<TestHelper>(new TestHelper(std::move(file_system), std::move(write),

src/paimon/testing/utils/testharness.cpp

@@ -105,6 +105,8 @@ std::map<std::string, std::string> GetJindoTestOptions() {
         {"fs.oss.bucket.paimon-unittest.accessKeyId", access_key_id},
         {"fs.oss.bucket.paimon-unittest.accessKeySecret", access_key_secret},
         {"fs.oss.user", "paimon"},
+        {"enable-object-store-catalog-in-inte-test", ""},
+        {"enable-object-store-commit-in-inte-test", ""},
     };
     return options;
 }

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_NETWORK_TESTS)
+        target_compile_definitions(paimon-write-and-read-inte-test
+                                   PRIVATE PAIMON_ENABLE_NETWORK_TESTS)
+    endif()
 
     add_paimon_test(clean_inte_test
                     STATIC_LINK_LIBS

test/inte/pk_compaction_inte_test.cpp

@@ -183,9 +183,6 @@ class PkCompactionInteTest : public ::testing::Test,
     Status Commit(const std::string& table_path,
                   const std::vector<std::shared_ptr<CommitMessage>>& commit_msgs) const {
         CommitContextBuilder commit_builder(table_path, "commit_user_1");
-        std::map<std::string, std::string> commit_options = {
-            {"enable-pk-commit-in-inte-test", ""}, {"enable-object-store-commit-in-inte-test", ""}};
-        commit_builder.SetOptions(commit_options);
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<CommitContext> commit_context,
                                commit_builder.Finish());
         PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreCommit> file_store_commit,

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");
+#if defined(PAIMON_ENABLE_NETWORK_TESTS) && defined(PAIMON_ENABLE_JINDO)
+    values.emplace_back("parquet", "jindo");
+#endif
 #ifdef PAIMON_ENABLE_ORC
     values.emplace_back("orc", "local");
 #endif