Skip to content

Commit 3310af0

Browse files
mrdrivingduckcodex
andcommitted
perf(fs): avoid redundant object-store metadata requests
Read planning already has data file sizes, but opening those files still makes an extra metadata request on object stores. Add a metadata-aware file-opening interface. Implementations that do not need it validate the supplied metadata and fall back to the existing path-based behavior. Object stores override it to use trusted metadata directly, avoiding the extra request. Route the new interface through filesystem routing and document the trusted-metadata contract. Keep a follow-up for deletion vector index files. Update the object-store test to use the metadata-aware interface. Co-authored-by: GPT-5.6 Terra <codex@users.noreply.github.com>
1 parent 1389694 commit 3310af0

18 files changed

Lines changed: 202 additions & 90 deletions

include/paimon/fs/file_system.h

Lines changed: 48 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include <functional>
2323
#include <memory>
2424
#include <string>
25+
#include <utility>
2526
#include <vector>
2627

2728
#include "paimon/result.h"
@@ -151,7 +152,7 @@ class PAIMON_EXPORT BasicFileStatus {
151152
virtual std::string GetPath() const = 0;
152153
};
153154

154-
/// Extended file status information interface.
155+
/// Extended file status information.
155156
///
156157
/// This class extends BasicFileStatus to provide comprehensive file system metadata including file
157158
/// size, modification time, and other attributes. It's used for operations that require detailed
@@ -161,21 +162,45 @@ class PAIMON_EXPORT FileStatus {
161162
FileStatus() = default;
162163
virtual ~FileStatus() = default;
163164

165+
/// Sentinel returned by `GetModificationTime()` when the modification time is not known.
166+
static constexpr int64_t kUnknownModificationTime = -1;
167+
168+
/// Create a file status from caller-supplied metadata.
169+
/// @param path The path of the file or directory.
170+
/// @param length The size of the file in bytes. It may be negative only when the size is
171+
/// unknown.
172+
/// @param is_dir Whether the path represents a directory. Defaults to false.
173+
FileStatus(std::string path, int64_t length, bool is_dir = false)
174+
: path_(std::move(path)), length_(length), is_dir_(is_dir) {}
175+
164176
/// Get the size of the file in bytes.
165177
/// @note For directories, this method is undefined behavior.
166-
virtual int64_t GetLen() const = 0;
178+
virtual int64_t GetLen() const {
179+
return length_;
180+
}
167181

168182
/// Check if this entry represents a directory.
169-
virtual bool IsDir() const = 0;
183+
virtual bool IsDir() const {
184+
return is_dir_;
185+
}
170186

171187
/// Get the path of this file or directory.
172-
virtual std::string GetPath() const = 0;
188+
virtual std::string GetPath() const {
189+
return path_;
190+
}
173191

174192
/// Get the last modification time of the file.
175193
///
176194
/// @return A long value representing the time the file was last modified, measured in
177195
/// milliseconds since the epoch (UTC January 1, 1970).
178-
virtual int64_t GetModificationTime() const = 0;
196+
virtual int64_t GetModificationTime() const {
197+
return kUnknownModificationTime;
198+
}
199+
200+
private:
201+
std::string path_;
202+
int64_t length_ = -1;
203+
bool is_dir_ = false;
179204
};
180205

181206
/// Abstract file system interface.
@@ -193,6 +218,24 @@ class PAIMON_EXPORT FileSystem {
193218
/// failure (e.g., file not found, permission denied).
194219
virtual Result<std::unique_ptr<InputStream>> Open(const std::string& path) const = 0;
195220

221+
/// Open an existing regular file for reading with known file metadata.
222+
/// @param file_status The trusted status of the file to open. Its path and length must
223+
/// identify an existing regular file. Its length must be non-negative;
224+
/// zero is valid for an empty file.
225+
/// @return Result containing a unique pointer to `InputStream` on success, or error status on
226+
/// failure (e.g., invalid file size, file not found, permission denied).
227+
/// @note File systems may rely on `file_status` to skip metadata requests. The caller must
228+
/// not expect this method to validate the path, file type, or size. A stale or
229+
/// incorrect status, or a file removed after planning, can cause reads to end early or
230+
/// fail when read instead of failing at open time. Wrapping file systems should forward
231+
/// both `Open` overloads.
232+
virtual Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const {
233+
if (file_status.GetLen() < 0) {
234+
return Status::Invalid("file size must be non-negative");
235+
}
236+
return Open(file_status.GetPath());
237+
}
238+
196239
/// Create a new file for writing.
197240
/// @param path The file path to create.
198241
/// @param overwrite If true, overwrite existing file; if false, fail if file exists.

src/paimon/common/file_index/bitmap/apply_bitmap_index_batch_reader_test.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -93,8 +93,8 @@ class ApplyBitmapIndexBatchReaderTest : public ::testing::Test,
9393
ASSERT_OK_AND_ASSIGN(
9494
file_batch_reader,
9595
PrefetchFileBatchReaderImpl::Create(
96-
/*data_file_path=*/"DUMMY", &reader_builder, fs_, prefetch_batch_count,
97-
batch_size, prefetch_batch_count * 2,
96+
/*data_file_path=*/"DUMMY", /*data_file_size=*/0, &reader_builder, fs_,
97+
prefetch_batch_count, batch_size, prefetch_batch_count * 2,
9898
/*enable_adaptive_prefetch_strategy=*/false, executor_,
9999
/*initialize_read_ranges=*/true,
100100
/*prefetch_cache_mode=*/PrefetchCacheMode::ALWAYS, CacheConfig(), pool_));

src/paimon/common/fs/file_system_test.cpp

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -283,6 +283,20 @@ TEST_P(FileSystemTest, TestSimpleWriteAndRead) {
283283
ASSERT_OK(in_stream->Close());
284284
}
285285

286+
TEST_P(FileSystemTest, TestOpenWithKnownFileSize) {
287+
const std::string content = "abcdefghijk";
288+
const std::string file_path = test_root_ + "/file.data";
289+
ASSERT_OK(fs_->WriteFile(file_path, content, /*overwrite=*/true));
290+
291+
FileStatus file_status(file_path, static_cast<int64_t>(content.size()));
292+
ASSERT_OK_AND_ASSIGN(auto input_stream, fs_->Open(file_status));
293+
ASSERT_OK_AND_ASSIGN(int64_t file_size, input_stream->Length());
294+
ASSERT_EQ(file_size, content.size());
295+
ASSERT_OK(input_stream->Close());
296+
297+
ASSERT_TRUE(fs_->Open(FileStatus(file_path, /*length=*/-1)).status().IsInvalid());
298+
}
299+
286300
TEST_P(FileSystemTest, TestWriteMultipleTimes) {
287301
std::vector<std::string> content_vec = {"abc", "defg", "hi", "j", "k"};
288302
std::string content = "abcdefghijk";

src/paimon/common/fs/object_store_file_system.cpp

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -388,6 +388,19 @@ Result<std::unique_ptr<InputStream>> ObjectStoreFileSystem::Open(const std::stri
388388
ToUri(object_path), metadata.value().size);
389389
}
390390

391+
Result<std::unique_ptr<InputStream>> ObjectStoreFileSystem::Open(
392+
const FileStatus& file_status) const {
393+
const std::string path = file_status.GetPath();
394+
const int64_t file_size = file_status.GetLen();
395+
PAIMON_RETURN_NOT_OK(ValidateValueNonNegative(file_size, "file size"));
396+
PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path));
397+
if (object_path.key.empty()) {
398+
return Status::Invalid(fmt::format("{} is a directory", path));
399+
}
400+
return std::make_unique<ObjectStoreInputStream>(client_, read_ahead_limiter_, object_path,
401+
ToUri(object_path), file_size);
402+
}
403+
391404
Result<std::unique_ptr<FileStatus>> ObjectStoreFileSystem::GetFileStatus(
392405
const std::string& path) const {
393406
PAIMON_ASSIGN_OR_RAISE(ObjectStorePath object_path, ParsePath(path));

src/paimon/common/fs/object_store_file_system.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ class PAIMON_EXPORT ObjectStoreFileSystem : public FileSystem {
8585
~ObjectStoreFileSystem() override = default;
8686

8787
Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
88+
Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const override;
8889
Result<std::unique_ptr<FileStatus>> GetFileStatus(const std::string& path) const override;
8990
Status ListDir(const std::string& directory,
9091
std::vector<std::unique_ptr<BasicFileStatus>>* file_status_list) const override;

src/paimon/common/fs/object_store_file_system_test.cpp

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,15 @@ TEST(ObjectStoreFileSystemTest, TestOpenBucketRootIsDirectory) {
159159
ASSERT_EQ(client->list_calls_, 0);
160160
}
161161

162+
TEST(ObjectStoreFileSystemTest, TestOpenWithKnownLengthSkipsHead) {
163+
auto client = std::make_shared<MockObjectStoreClient>();
164+
client->objects_["file"] = "data";
165+
ObjectStoreFileSystem fs("s3", client);
166+
ASSERT_OK_AND_ASSIGN(auto stream, fs.Open(FileStatus("s3://bucket/file", 4)));
167+
ASSERT_EQ(stream->Length().value(), 4);
168+
ASSERT_EQ(client->head_calls_, 0);
169+
}
170+
162171
TEST(ObjectStoreFileSystemTest, TestPathWithLeadingSlashes) {
163172
auto client = std::make_shared<MockObjectStoreClient>();
164173
client->objects_["file"] = "data";

src/paimon/common/fs/resolving_file_system.cpp

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,13 @@ Result<std::unique_ptr<InputStream>> ResolvingFileSystem::Open(const std::string
7979
return fs->Open(path);
8080
}
8181

82+
Result<std::unique_ptr<InputStream>> ResolvingFileSystem::Open(
83+
const FileStatus& file_status) const {
84+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileSystem> fs,
85+
GetRealFileSystem(file_status.GetPath()));
86+
return fs->Open(file_status);
87+
}
88+
8289
Result<std::unique_ptr<OutputStream>> ResolvingFileSystem::Create(const std::string& path,
8390
bool overwrite) const {
8491
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileSystem> fs, GetRealFileSystem(path));

src/paimon/common/fs/resolving_file_system.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ class ResolvingFileSystem : public FileSystem {
4141
~ResolvingFileSystem() override = default;
4242

4343
Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override;
44+
Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status) const override;
4445
Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
4546
bool overwrite) const override;
4647

src/paimon/common/reader/prefetch_file_batch_reader_impl.cpp

Lines changed: 13 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ std::pair<int64_t, int64_t> ComputeBatchSliceByReadRange(
5656
} // namespace
5757

5858
Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl::Create(
59-
const std::string& data_file_path, const ReaderBuilder* reader_builder,
59+
const std::string& data_file_path, int64_t data_file_size, const ReaderBuilder* reader_builder,
6060
const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num, int32_t batch_size,
6161
uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
6262
const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
@@ -83,20 +83,22 @@ Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> PrefetchFileBatchReaderImpl
8383

8484
std::shared_ptr<ReadAheadCache> cache;
8585
if (prefetch_cache_mode != PrefetchCacheMode::NEVER) {
86-
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream, fs->Open(data_file_path));
86+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
87+
fs->Open(FileStatus(data_file_path, data_file_size)));
8788
cache = std::make_shared<ReadAheadCache>(input_stream, cache_config, pool);
8889
}
8990
std::vector<std::future<Result<std::unique_ptr<FileBatchReader>>>> futures;
9091
for (uint32_t i = 0; i < prefetch_max_parallel_num; i++) {
91-
futures.push_back(Via(executor.get(),
92-
[&fs, &data_file_path, &reader_builder,
93-
&cache]() -> Result<std::unique_ptr<FileBatchReader>> {
94-
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
95-
fs->Open(data_file_path));
96-
auto cache_input_stream = std::make_shared<CacheInputStream>(
97-
std::move(input_stream), cache);
98-
return reader_builder->Build(cache_input_stream);
99-
}));
92+
futures.push_back(
93+
Via(executor.get(),
94+
[&fs, &data_file_path, data_file_size, &reader_builder,
95+
&cache]() -> Result<std::unique_ptr<FileBatchReader>> {
96+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
97+
fs->Open(FileStatus(data_file_path, data_file_size)));
98+
auto cache_input_stream =
99+
std::make_shared<CacheInputStream>(std::move(input_stream), cache);
100+
return reader_builder->Build(cache_input_stream);
101+
}));
100102
}
101103
std::vector<std::shared_ptr<PrefetchFileBatchReader>> readers;
102104
for (auto& file_batch_reader : CollectAll(futures)) {

src/paimon/common/reader/prefetch_file_batch_reader_impl.h

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -56,12 +56,12 @@ class Metrics;
5656
class PrefetchFileBatchReaderImpl : public PrefetchFileBatchReader {
5757
public:
5858
static Result<std::unique_ptr<PrefetchFileBatchReaderImpl>> Create(
59-
const std::string& data_file_path, const ReaderBuilder* reader_builder,
60-
const std::shared_ptr<FileSystem>& fs, uint32_t prefetch_max_parallel_num,
61-
int32_t batch_size, uint32_t prefetch_batch_count, bool enable_adaptive_prefetch_strategy,
62-
const std::shared_ptr<Executor>& executor, bool initialize_read_ranges,
63-
PrefetchCacheMode prefetch_cache_mode, const CacheConfig& cache_config,
64-
const std::shared_ptr<MemoryPool>& pool);
59+
const std::string& data_file_path, int64_t data_file_size,
60+
const ReaderBuilder* reader_builder, const std::shared_ptr<FileSystem>& fs,
61+
uint32_t prefetch_max_parallel_num, int32_t batch_size, uint32_t prefetch_batch_count,
62+
bool enable_adaptive_prefetch_strategy, const std::shared_ptr<Executor>& executor,
63+
bool initialize_read_ranges, PrefetchCacheMode prefetch_cache_mode,
64+
const CacheConfig& cache_config, const std::shared_ptr<MemoryPool>& pool);
6565

6666
~PrefetchFileBatchReaderImpl() override;
6767

0 commit comments

Comments
 (0)