perf(fs): skip getFileStatus on Jindo open when the file length is known (#332)

Author: lucasfang

src/paimon/common/fs/file_system_test.cpp

@@ -292,6 +292,13 @@ TEST_P(FileSystemTest, TestOpenWithKnownFileSize) {
     ASSERT_OK_AND_ASSIGN(auto input_stream, fs_->Open(file_status));
     ASSERT_OK_AND_ASSIGN(int64_t file_size, input_stream->Length());
     ASSERT_EQ(file_size, content.size());
+    // Reading through the trusted-length open must serve the object's real bytes, not just
+    // echo the length the caller passed in.
+    std::string read_content(content.size(), '\0');
+    ASSERT_OK_AND_ASSIGN(int64_t read_len,
+                         input_stream->Read(read_content.data(), read_content.size()));
+    ASSERT_EQ(read_len, static_cast<int64_t>(content.size()));
+    ASSERT_EQ(content, read_content);
     ASSERT_OK(input_stream->Close());
 
     ASSERT_TRUE(fs_->Open(FileStatus(file_path, /*length=*/-1)).status().IsInvalid());

src/paimon/fs/jindo/jindo_file_system.cpp

@@ -88,6 +88,18 @@ Result<std::unique_ptr<InputStream>> JindoFileSystem::Open(const std::string& pa
     return std::make_unique<JindoInputStream>(impl_, std::move(reader));
 }
 
+Result<std::unique_ptr<InputStream>> JindoFileSystem::Open(const FileStatus& file_status) const {
+    const int64_t file_length = file_status.GetLen();
+    PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(file_length, "file size"));
+    // The trusted length lets the store skip the getFileStatus it otherwise issues on open.
+    // The status is not re-validated here; a stale or incorrect length surfaces as a read
+    // error later rather than at open time.
+    std::unique_ptr<JdoReader> reader;
+    PAIMON_RETURN_NOT_OK_FROM_JINDO(
+        impl_->GetFileSystem()->openReader(file_status.GetPath(), file_length, &reader));
+    return std::make_unique<JindoInputStream>(impl_, std::move(reader));
+}
+
 Result<std::unique_ptr<OutputStream>> JindoFileSystem::Create(const std::string& path,
                                                               bool overwrite) const {
     PAIMON_ASSIGN_OR_RAISE(bool exist, Exists(path));

src/paimon/fs/jindo/jindo_file_system.h

@@ -41,6 +41,7 @@ class JindoFileSystem : public FileSystem {
     using FileSystem::Open;
 
     Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
+    Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const override;
     Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
                                                  bool overwrite) const override;
     Status Mkdirs(const std::string& path) const override;

src/paimon/fs/jindo/jindo_file_system_test.cpp

@@ -204,4 +204,32 @@ TEST(JindoFileSystemAsyncReadTest, TestConcurrentReadAsyncAndReadFromOss) {
     }
 }
 
+TEST_F(JindoFileSystemTest, TestOpenWithKnownFileSizeReadsBackContent) {
+    const std::string content = "abcdefghijk";
+    const std::string file_path = test_dir_ + "file.data";
+    ASSERT_OK(fs_->WriteFile(file_path, content, /*overwrite=*/true));
+
+    // The read path hands Open a length it already trusts; the stream must still serve the
+    // object's real bytes, not just echo the length back.
+    ASSERT_OK_AND_ASSIGN(auto in_stream,
+                         fs_->Open(FileStatus(file_path, static_cast<int64_t>(content.size()))));
+    ASSERT_OK_AND_ASSIGN(int64_t length, in_stream->Length());
+    ASSERT_EQ(length, static_cast<int64_t>(content.size()));
+    std::string read_content(content.size(), '\0');
+    ASSERT_OK_AND_ASSIGN(int64_t read_len,
+                         in_stream->Read(read_content.data(), read_content.size()));
+    ASSERT_EQ(read_len, static_cast<int64_t>(content.size()));
+    ASSERT_EQ(content, read_content);
+    ASSERT_OK(in_stream->Close());
+
+    // A stale length shorter than the object must never yield bytes beyond what it vouches
+    // for: the wrapper read is all-or-nothing, so over-reading the declared length fails with
+    // an EOF error instead of silently returning truncated data.
+    const int64_t short_length = static_cast<int64_t>(content.size()) - 3;
+    ASSERT_OK_AND_ASSIGN(auto short_stream, fs_->Open(FileStatus(file_path, short_length)));
+    std::string short_read(content.size(), '\0');
+    ASSERT_NOK_WITH_MSG(short_stream->Read(short_read.data(), short_read.size()), "EOF reached");
+    ASSERT_OK(short_stream->Close());
+}
+
 }  // namespace paimon::jindo::test

src/paimon/fs/jindo/jindo_file_system_unit_test.cpp

@@ -136,4 +136,14 @@ TEST(JindoFileSystemUnitTest, CreateReturnsParentDirectoryFailure) {
     ASSERT_EQ(fs.GetCalls()[0], "mkdirs");
 }
 
+TEST(JindoFileSystemUnitTest, OpenWithFileStatusRejectsNegativeLengthBeforeOpening) {
+    // The length is validated before the store is touched, so this holds without a live OSS:
+    // an uninitialized JdoFileSystem would report an init error, not a size error, if the
+    // validation were skipped.
+    jindo::JindoFileSystem fs(std::make_unique<JdoFileSystem>());
+
+    ASSERT_NOK_WITH_MSG(fs.Open(FileStatus("oss://bucket/data.parquet", /*length=*/-1)),
+                        "file size");
+}
+
 }  // namespace paimon::test

third_party/jindosdk-nextarch/include/JdoFileSystem.hpp

@@ -134,6 +134,13 @@ class JDO_EXPORT JdoFileSystem {
 
     JdoStatus openReader(const std::string &path, std::unique_ptr<JdoReader>* result);
 
+    // Open a read-only reader with a trusted file length. The length is handed to the store
+    // together with a "file status already resolved" hint, so open skips the getFileStatus
+    // request it would otherwise issue. The caller is responsible for the length being correct;
+    // a stale length may make reads end early or fail at read time instead of at open time.
+    JdoStatus openReader(const std::string &path, int64_t file_length,
+                         std::unique_ptr<JdoReader>* result);
+
     JdoStatus openWriter(const std::string &path, std::unique_ptr<JdoWriter>* result);
 
     JdoStatus mkdir(const std::string &path, bool recursive);

third_party/jindosdk-nextarch/src/JdoFileSystem.cpp

@@ -33,6 +33,7 @@
 #include "jdo_api.h"
 #include "jdo_defines.h"
 #include "jdo_error.h"
+#include "jdo_option_keys.h"
 #include "jdo_options.h"
 #include "jdo_file_status.h"
 #include "jdo_list_dir_result.h"
@@ -1148,6 +1149,33 @@ JdoStatus JdoFileSystem::openReader(const std::string &path,
     return JdoStatus::OK();
 }
 
+JdoStatus JdoFileSystem::openReader(const std::string &path, int64_t file_length,
+                                    std::unique_ptr<JdoReader> *result) {
+    auto [store, err] = GetJdoStore(path);
+    if (!err.ok()) {
+        return err;
+    }
+
+    // The hint is what actually suppresses the getFileStatus on open; the store only reads
+    // JDO_OPEN_OPTS_FILE_LENGTH when it is set, and otherwise resolves the status itself.
+    auto options = jdo_createOptions();
+    jdo_setOption(options, JDO_OPEN_OPTS_HAS_GET_FILE_STATUS, "true");
+    jdo_setOption(options, JDO_OPEN_OPTS_FILE_LENGTH, std::to_string(file_length).c_str());
+
+    START_CALL(store)
+    auto stream = jdo_open(ctx, path.c_str(), JDO_OPEN_FLAG_READ_ONLY, 0777, options);
+    END_CALL()
+    jdo_freeOptions(options);
+
+    if (errorCode != 0) {
+        return JdoStatus::InternalError(errorCode, "open file reader failed: ", errorMsg);
+    }
+
+    result->reset(new JindoReader(path, store, stream));
+
+    return JdoStatus::OK();
+}
+
 JdoStatus JdoFileSystem::openWriter(const std::string &path,
                                     std::unique_ptr<JdoWriter> *result) {
     auto [store, err] = GetJdoStore(path);