This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new df007728 perf(read): reduce metadata round trips during scan planning
(#338)
df007728 is described below
commit df007728e468b49805a314484ed8cd3b1a3c66ff
Author: Yonghao Fang <[email protected]>
AuthorDate: Tue Sep 15 13:16:42 2026 +0800
perf(read): reduce metadata round trips during scan planning (#338)
---
src/paimon/core/index/index_file_handler.cpp | 4 +-
.../core/manifest/index_manifest_file_handler.cpp | 4 +-
.../manifest/index_manifest_file_handler_test.cpp | 18 ++--
src/paimon/core/manifest/manifest_file.cpp | 5 +-
src/paimon/core/manifest/manifest_file.h | 6 ++
src/paimon/core/manifest/manifest_file_test.cpp | 89 +++++++++++++---
src/paimon/core/manifest/manifest_list.h | 9 +-
src/paimon/core/manifest/manifest_list_test.cpp | 115 ++++++++++++++++++++-
.../core/operation/commit/commit_scanner.cpp | 8 +-
src/paimon/core/operation/expire_snapshots.cpp | 13 ++-
.../core/operation/file_store_commit_impl.cpp | 6 +-
.../core/operation/file_store_commit_impl_test.cpp | 21 ++--
src/paimon/core/operation/file_store_scan.cpp | 39 +++++--
src/paimon/core/operation/manifest_file_merger.cpp | 4 +-
.../core/operation/manifest_file_merger_test.cpp | 4 +-
.../core/operation/orphan_files_cleaner_impl.cpp | 12 ++-
src/paimon/core/snapshot_file_scan.cpp | 6 +-
src/paimon/core/snapshot_file_scan_test.cpp | 67 ++++++++++++
.../core/table/system/global_system_tables.cpp | 5 +-
.../core/table/system/metadata_system_tables.cpp | 4 +-
src/paimon/core/utils/file_utils.cpp | 19 ++--
src/paimon/core/utils/objects_file.h | 57 +++++++---
src/paimon/fs/jindo/jindo_file_system.cpp | 15 ++-
src/paimon/fs/jindo/jindo_file_system_test.cpp | 16 +++
24 files changed, 452 insertions(+), 94 deletions(-)
diff --git a/src/paimon/core/index/index_file_handler.cpp
b/src/paimon/core/index/index_file_handler.cpp
index 9b45e748..d614d85e 100644
--- a/src/paimon/core/index/index_file_handler.cpp
+++ b/src/paimon/core/index/index_file_handler.cpp
@@ -55,8 +55,8 @@ Result<std::vector<IndexManifestEntry>>
IndexFileHandler::Scan(
return std::vector<IndexManifestEntry>();
}
std::vector<IndexManifestEntry> index_entries;
- PAIMON_RETURN_NOT_OK(
- index_manifest_file_->Read(index_manifest.value(), filter,
&index_entries));
+ PAIMON_RETURN_NOT_OK(index_manifest_file_->Read(index_manifest.value(),
filter,
+
/*file_size=*/std::nullopt, &index_entries));
return index_entries;
}
diff --git a/src/paimon/core/manifest/index_manifest_file_handler.cpp
b/src/paimon/core/manifest/index_manifest_file_handler.cpp
index 41543729..55261de9 100644
--- a/src/paimon/core/manifest/index_manifest_file_handler.cpp
+++ b/src/paimon/core/manifest/index_manifest_file_handler.cpp
@@ -18,6 +18,7 @@
#include "paimon/core/manifest/index_manifest_file_handler.h"
+#include <optional>
#include <set>
#include <string>
#include <unordered_map>
@@ -201,7 +202,8 @@ Result<std::string> IndexManifestFileHandler::Write(
std::vector<IndexManifestEntry> entries;
if (previous_index_manifest != std::nullopt) {
PAIMON_RETURN_NOT_OK(index_manifest_file->Read(previous_index_manifest.value(),
- /*filter=*/nullptr,
&entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &entries));
}
for (const auto& entry : entries) {
if (!(entry.kind == FileKind::Add())) {
diff --git a/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
b/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
index 9661697c..42048dc7 100644
--- a/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
+++ b/src/paimon/core/manifest/index_manifest_file_handler_test.cpp
@@ -123,7 +123,8 @@ TEST_F(IndexManifestFileHandlerTest,
GlobalCombinerDeletesThenAddsByFileName) {
/*bucket_mode=*/4,
index_manifest_file.get()));
std::vector<IndexManifestEntry> written_entries;
- ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
&written_entries));
+ ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&written_entries));
ASSERT_EQ(written_entries.size(), 1);
ASSERT_EQ(written_entries[0].index_file->FileName(), "global-0");
ASSERT_EQ(written_entries[0].index_file->RowCount(), 2);
@@ -156,7 +157,8 @@ TEST_F(IndexManifestFileHandlerTest,
BucketedCombinerUsesPartitionBucketAndIndex
/*bucket_mode=*/2,
index_manifest_file.get()));
std::vector<IndexManifestEntry> written_entries;
- ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
&written_entries));
+ ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&written_entries));
ASSERT_EQ(written_entries.size(), 2);
bool found_bucket0 = false;
@@ -192,7 +194,8 @@ TEST_F(IndexManifestFileHandlerTest,
BucketedCombinerOverwritesDuplicateAddedEnt
/*bucket_mode=*/2, index_manifest_file.get()));
std::vector<IndexManifestEntry> written_entries;
- ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
&written_entries));
+ ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&written_entries));
ASSERT_EQ(written_entries.size(), 1);
ASSERT_EQ(written_entries[0].index_file->FileName(), "dv-0-new");
ASSERT_EQ(written_entries[0].index_file->RowCount(), 20);
@@ -213,7 +216,8 @@ TEST_F(IndexManifestFileHandlerTest,
GlobalCombinerOverwritesDuplicateAddedEntri
/*bucket_mode=*/4, index_manifest_file.get()));
std::vector<IndexManifestEntry> written_entries;
- ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
&written_entries));
+ ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&written_entries));
ASSERT_EQ(written_entries.size(), 1);
ASSERT_EQ(written_entries[0].index_file->FileName(), "global-0");
ASSERT_EQ(written_entries[0].index_file->RowCount(), 20);
@@ -257,7 +261,8 @@ TEST_F(IndexManifestFileHandlerTest,
GlobalDvCombinerReplacesIndexFileInBucketUn
/*bucket_mode=*/-1,
index_manifest_file.get()));
std::vector<IndexManifestEntry> written_entries;
- ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
&written_entries));
+ ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&written_entries));
ASSERT_EQ(written_entries.size(), 2);
std::set<std::string> written_file_names;
for (const auto& entry : written_entries) {
@@ -278,7 +283,8 @@ TEST_F(IndexManifestFileHandlerTest,
GlobalDvCombinerReplacesIndexFileInBucketUn
/*bucket_mode=*/-1,
index_manifest_file.get()));
written_entries.clear();
- ASSERT_OK(index_manifest_file->Read(reordered_manifest,
/*filter=*/nullptr, &written_entries));
+ ASSERT_OK(index_manifest_file->Read(reordered_manifest, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&written_entries));
ASSERT_EQ(written_entries.size(), 2);
written_file_names.clear();
for (const auto& entry : written_entries) {
diff --git a/src/paimon/core/manifest/manifest_file.cpp
b/src/paimon/core/manifest/manifest_file.cpp
index 0fdb78d7..ecbcb6bd 100644
--- a/src/paimon/core/manifest/manifest_file.cpp
+++ b/src/paimon/core/manifest/manifest_file.cpp
@@ -20,6 +20,7 @@
#include <cassert>
#include <limits>
+#include <optional>
#include <utility>
#include "arrow/c/abi.h"
@@ -90,6 +91,7 @@ Result<std::unique_ptr<ManifestFile>> ManifestFile::Create(
}
Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t
bucket,
+ std::optional<int64_t> file_size,
std::vector<ManifestEntry>* entries)
const {
return ReadArrowBatches(
file_name,
@@ -106,7 +108,8 @@ Status ManifestFile::ReadBucketEntries(const std::string&
file_name, int32_t buc
entries->push_back(std::move(entry));
}
return Status::OK();
- });
+ },
+ file_size);
}
Result<std::vector<ManifestFileMeta>> ManifestFile::Write(
diff --git a/src/paimon/core/manifest/manifest_file.h
b/src/paimon/core/manifest/manifest_file.h
index 528bb7bf..2f3a60cb 100644
--- a/src/paimon/core/manifest/manifest_file.h
+++ b/src/paimon/core/manifest/manifest_file.h
@@ -20,6 +20,7 @@
#include <cstdint>
#include <memory>
+#include <optional>
#include <string>
#include <vector>
@@ -63,7 +64,12 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
Result<std::vector<ManifestFileMeta>> Write(const
std::vector<ManifestEntry>& entries);
/// Read a manifest file and deserialize only entries for the specified
bucket.
+ ///
+ /// @param file_size Length of the manifest when the caller already has it
from the manifest
+ /// list, which saves the read a metadata request on a
remote store. Pass
+ /// std::nullopt when the length is not known.
Status ReadBucketEntries(const std::string& file_name, int32_t bucket,
+ std::optional<int64_t> file_size,
std::vector<ManifestEntry>* entries) const;
private:
diff --git a/src/paimon/core/manifest/manifest_file_test.cpp
b/src/paimon/core/manifest/manifest_file_test.cpp
index 5cd264aa..0ce9cc72 100644
--- a/src/paimon/core/manifest/manifest_file_test.cpp
+++ b/src/paimon/core/manifest/manifest_file_test.cpp
@@ -55,6 +55,13 @@ class CountingFileSystem : public FileSystem {
return local_.Open(path);
}
+ /// Overridden because the base implementation forwards to `Open(path)`,
which would fold the
+ /// two ways of opening into one counter and hide whether a known length
reached the store.
+ Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status)
const override {
+ opened_lengths.push_back(file_status.GetLen());
+ return local_.Open(file_status.GetPath());
+ }
+
Result<std::unique_ptr<OutputStream>> Create(const std::string& path,
bool overwrite) const
override {
return local_.Create(path, overwrite);
@@ -93,6 +100,7 @@ class CountingFileSystem : public FileSystem {
mutable int open_count = 0;
mutable int get_file_status_count = 0;
+ mutable std::vector<int64_t> opened_lengths;
private:
LocalFileSystem local_;
@@ -124,10 +132,12 @@ class ManifestFileTest : public testing::Test {
/*target_file_size=*/1024, pool, options,
unused_schema));
std::vector<ManifestEntry> manifest_entries;
if (bucket) {
- EXPECT_OK(
- manifest_file->ReadBucketEntries(file_name, bucket.value(),
&manifest_entries));
+ EXPECT_OK(manifest_file->ReadBucketEntries(file_name,
bucket.value(),
+
/*file_size=*/std::nullopt,
+ &manifest_entries));
} else {
- EXPECT_OK(manifest_file->Read(file_name, /*filter=*/nullptr,
&manifest_entries));
+ EXPECT_OK(manifest_file->Read(file_name, /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&manifest_entries));
}
return manifest_entries;
@@ -263,7 +273,7 @@ TEST_F(ManifestFileTest,
TestManifestCacheIsDisabledWithoutInjectedCache) {
std::vector<ManifestEntry> first_read;
ASSERT_OK(manifest_file->Read("manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1",
- /*filter=*/nullptr, &first_read));
+ /*filter=*/nullptr,
/*file_size=*/std::nullopt, &first_read));
ASSERT_EQ(5, first_read.size());
ASSERT_EQ(1, counting_file_system->open_count);
ASSERT_EQ(0, counting_file_system->get_file_status_count);
@@ -272,7 +282,7 @@ TEST_F(ManifestFileTest,
TestManifestCacheIsDisabledWithoutInjectedCache) {
ASSERT_OK(manifest_file->Read(
"manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1",
[](const ManifestEntry& entry) -> Result<bool> { return entry.Kind()
== FileKind::Add(); },
- &filtered_read));
+ /*file_size=*/std::nullopt, &filtered_read));
ASSERT_EQ(1, filtered_read.size());
ASSERT_EQ(2, counting_file_system->open_count);
ASSERT_EQ(0, counting_file_system->get_file_status_count);
@@ -305,10 +315,10 @@ TEST_F(ManifestFileTest,
TestManifestCacheReusesCachedBytes) {
std::vector<ManifestEntry> first_read;
ASSERT_OK(manifest_file->Read("manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1",
- /*filter=*/nullptr, &first_read));
+ /*filter=*/nullptr,
/*file_size=*/std::nullopt, &first_read));
std::vector<ManifestEntry> second_read;
ASSERT_OK(manifest_file->Read("manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1",
- /*filter=*/nullptr, &second_read));
+ /*filter=*/nullptr,
/*file_size=*/std::nullopt, &second_read));
ASSERT_EQ(first_read, second_read);
ASSERT_EQ(1, counting_file_system->open_count);
@@ -345,20 +355,24 @@ TEST_F(ManifestFileTest,
TestReadBucketEntriesMaterializesOnlySelectedBucket) {
const std::string manifest_name =
"manifest-3a44a0da-1008-463c-914e-28d271375e24-0";
std::vector<ManifestEntry> all_entries;
- ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr,
&all_entries));
+ ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr,
/*file_size=*/std::nullopt,
+ &all_entries));
ASSERT_EQ(2, all_entries.size());
std::vector<ManifestEntry> bucket_one_entries;
- ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1,
&bucket_one_entries));
+ ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1,
+ /*file_size=*/std::nullopt,
&bucket_one_entries));
ASSERT_EQ(std::vector<ManifestEntry>({all_entries[0]}),
bucket_one_entries);
std::vector<ManifestEntry> bucket_zero_entries;
- ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/0,
&bucket_zero_entries));
+ ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/0,
+ /*file_size=*/std::nullopt,
&bucket_zero_entries));
ASSERT_EQ(std::vector<ManifestEntry>({all_entries[1]}),
bucket_zero_entries);
std::vector<ManifestEntry> missing_bucket_entries;
- ASSERT_OK(
- manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/2,
&missing_bucket_entries));
+ ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/2,
+ /*file_size=*/std::nullopt,
+ &missing_bucket_entries));
ASSERT_TRUE(missing_bucket_entries.empty());
ASSERT_EQ(1, counting_file_system->open_count);
@@ -366,6 +380,51 @@ TEST_F(ManifestFileTest,
TestReadBucketEntriesMaterializesOnlySelectedBucket) {
ASSERT_EQ(1, manifest_cache->SupplierCallCount());
}
+// A scan reads manifests whose length the manifest list already recorded.
Handing that length over
+// is what lets the store skip the metadata request a bare open issues, which
on a remote store is
+// a round trip paid before any of the file is read.
+TEST_F(ManifestFileTest, TestReadPassesKnownSizeToOpen) {
+ auto pool = GetDefaultPool();
+ auto counting_file_system = std::make_shared<CountingFileSystem>();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
+ FileFormatFactory::Get("orc", {}));
+ std::string root_path = paimon::test::GetDataDir() +
"/orc/append_09.db/append_09";
+ auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0",
arrow::utf8())}));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ FileStorePathFactory::Create(root_path, unused_schema,
/*partition_keys=*/{},
+ /*default_part_value=*/"",
file_format->Identifier(),
+ /*data_file_prefix=*/"data-",
+ /*legacy_partition_name_enabled=*/true,
/*external_paths=*/{},
+
/*global_index_external_path=*/std::nullopt,
+ /*index_file_in_data_file_dir=*/false,
pool));
+ ASSERT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT,
"orc"}}));
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<ManifestFile> manifest_file,
+ ManifestFile::Create(counting_file_system, file_format, "zstd",
path_factory,
+ /*target_file_size=*/1024, pool, options,
unused_schema));
+
+ const std::string manifest_name =
"manifest-3a44a0da-1008-463c-914e-28d271375e24-0";
+ // The length the checked-in manifest list records for this manifest, and
the length the file
+ // on disk actually has.
+ constexpr int64_t kRecordedSize = 2617;
+
+ std::vector<ManifestEntry> all_entries;
+ ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr,
kRecordedSize, &all_entries));
+ ASSERT_EQ(2, all_entries.size());
+ ASSERT_EQ(std::vector<int64_t>({kRecordedSize}),
counting_file_system->opened_lengths);
+ ASSERT_EQ(0, counting_file_system->open_count);
+
+ std::vector<ManifestEntry> bucket_one_entries;
+ ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1,
kRecordedSize,
+ &bucket_one_entries));
+ ASSERT_EQ(std::vector<ManifestEntry>({all_entries[0]}),
bucket_one_entries);
+ ASSERT_EQ(std::vector<int64_t>({kRecordedSize, kRecordedSize}),
+ counting_file_system->opened_lengths);
+ ASSERT_EQ(0, counting_file_system->open_count);
+}
+
TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
auto pool = GetDefaultPool();
std::vector<ManifestEntry> source_entries =
@@ -406,11 +465,13 @@ TEST_F(ManifestFileTest,
TestReadBucketEntriesSkipsDeserializingOtherBuckets) {
manifest_file->WriteWithoutRolling({invalid_other_bucket,
valid_target_bucket}));
std::vector<ManifestEntry> all_entries;
- ASSERT_NOK_WITH_MSG(manifest_file->Read(written_file.first,
/*filter=*/nullptr, &all_entries),
+ ASSERT_NOK_WITH_MSG(manifest_file->Read(written_file.first,
/*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&all_entries),
"Unsupported byte value 2 for file kind.");
std::vector<ManifestEntry> bucket_entries;
- ASSERT_OK(manifest_file->ReadBucketEntries(written_file.first,
/*bucket=*/0, &bucket_entries));
+ ASSERT_OK(manifest_file->ReadBucketEntries(written_file.first,
/*bucket=*/0,
+ /*file_size=*/std::nullopt,
&bucket_entries));
ASSERT_EQ(std::vector<ManifestEntry>({valid_target_bucket}),
bucket_entries);
}
diff --git a/src/paimon/core/manifest/manifest_list.h
b/src/paimon/core/manifest/manifest_list.h
index 6bf64e8e..918d0491 100644
--- a/src/paimon/core/manifest/manifest_list.h
+++ b/src/paimon/core/manifest/manifest_list.h
@@ -90,7 +90,8 @@ class ManifestList : public ObjectsFile<ManifestFileMeta> {
/// @return Status indicating whether the operation was successful or not.
Status ReadBaseManifests(const Snapshot& snapshot,
std::vector<ManifestFileMeta>* manifests) const {
- return Read(snapshot.BaseManifestList(), /*filter=*/nullptr,
manifests);
+ return Read(snapshot.BaseManifestList(), /*filter=*/nullptr,
+ snapshot.BaseManifestListSize(), manifests);
}
/// Return a `ManifestFileMeta` for each delta manifest in this snapshot.
@@ -100,7 +101,8 @@ class ManifestList : public ObjectsFile<ManifestFileMeta> {
/// @return Status indicating whether the operation was successful or not.
Status ReadDeltaManifests(const Snapshot& snapshot,
std::vector<ManifestFileMeta>* manifests) const {
- return Read(snapshot.DeltaManifestList(), /*filter=*/nullptr,
manifests);
+ return Read(snapshot.DeltaManifestList(), /*filter=*/nullptr,
+ snapshot.DeltaManifestListSize(), manifests);
}
/// Return a `ManifestFileMeta` for each changelog manifest in this
snapshot.
@@ -113,7 +115,8 @@ class ManifestList : public ObjectsFile<ManifestFileMeta> {
const std::optional<std::string>& changelog_manifest_list =
snapshot.ChangelogManifestList();
if (changelog_manifest_list) {
- return Read(changelog_manifest_list.value(), /*filter=*/nullptr,
manifests);
+ return Read(changelog_manifest_list.value(), /*filter=*/nullptr,
+ snapshot.ChangelogManifestListSize(), manifests);
} else {
return Status::OK();
}
diff --git a/src/paimon/core/manifest/manifest_list_test.cpp
b/src/paimon/core/manifest/manifest_list_test.cpp
index 97b420cf..d8d25538 100644
--- a/src/paimon/core/manifest/manifest_list_test.cpp
+++ b/src/paimon/core/manifest/manifest_list_test.cpp
@@ -19,6 +19,7 @@
#include "paimon/core/manifest/manifest_list.h"
#include <map>
+#include <optional>
#include <variant>
#include "arrow/type.h"
@@ -36,12 +37,67 @@
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
+
+namespace {
+
+/// Records which `Open` overload a read took. A length that planning already
knows should reach
+/// the file system instead of being rediscovered on open, which on a remote
store costs a round
+/// trip of its own before a single byte is read. `Open(FileStatus)` has to be
overridden here:
+/// the base implementation forwards to `Open(path)`, which would leave the
two paths
+/// indistinguishable.
+class OpenRecordingFileSystem : public LocalFileSystem {
+ public:
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ ++open_without_length_count;
+ return LocalFileSystem::Open(path);
+ }
+
+ Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status)
const override {
+ opened_lengths.push_back(file_status.GetLen());
+ return LocalFileSystem::Open(file_status.GetPath());
+ }
+
+ mutable int open_without_length_count = 0;
+ mutable std::vector<int64_t> opened_lengths;
+};
+
+/// Builds a snapshot carrying manifest list sizes the way a commit records
them, or leaving them
+/// unset the way a snapshot written before those fields existed would.
+Snapshot MakeSnapshot(const std::string& base_list, const
std::optional<int64_t>& base_size,
+ const std::string& delta_list, const
std::optional<int64_t>& delta_size) {
+ return Snapshot(
+ /*id=*/1, /*schema_id=*/0, /*base_manifest_list=*/base_list,
+ /*base_manifest_list_size=*/base_size,
/*delta_manifest_list=*/delta_list,
+ /*delta_manifest_list_size=*/delta_size,
/*changelog_manifest_list=*/std::nullopt,
+ /*changelog_manifest_list_size=*/std::nullopt,
/*index_manifest=*/std::nullopt,
+ /*commit_user=*/"user", /*commit_identifier=*/1,
Snapshot::CommitKind::Append(),
+ /*time_millis=*/0, /*total_record_count=*/1, /*delta_record_count=*/1,
+ /*changelog_record_count=*/1, /*watermark=*/std::nullopt,
/*statistics=*/std::nullopt,
+ /*properties=*/std::nullopt, /*next_row_id=*/std::nullopt);
+}
+
+ManifestFileMeta MakeMeta(const std::string& name, int64_t file_size, int64_t
num_added_files,
+ int64_t num_deleted_files) {
+ return ManifestFileMeta(name, file_size, num_added_files,
num_deleted_files,
+ SimpleStats::EmptyStats(), /*schema_id=*/0,
/*min_bucket=*/0,
+ /*max_bucket=*/0, /*min_level=*/0, /*max_level=*/0,
+ /*min_row_id=*/std::nullopt,
/*max_row_id=*/std::nullopt);
+}
+
+} // namespace
+
class ManifestListTest : public testing::Test {
public:
std::unique_ptr<ManifestList> CreateManifestList(
const std::string& file_format_str, const std::string& root_path,
const std::shared_ptr<MemoryPool>& pool) const {
- std::shared_ptr<FileSystem> file_system =
std::make_shared<LocalFileSystem>();
+ return CreateManifestList(std::make_shared<LocalFileSystem>(),
file_format_str, root_path,
+ pool);
+ }
+
+ std::unique_ptr<ManifestList> CreateManifestList(
+ const std::shared_ptr<FileSystem>& file_system, const std::string&
file_format_str,
+ const std::string& root_path, const std::shared_ptr<MemoryPool>& pool)
const {
EXPECT_OK_AND_ASSIGN(std::shared_ptr<FileFormat> file_format,
FileFormatFactory::Get(file_format_str, {}));
auto unused_schema =
arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())}));
@@ -65,7 +121,8 @@ class ManifestListTest : public testing::Test {
const std::string& file_name, const std::shared_ptr<MemoryPool>& pool)
const {
auto manifest_list = CreateManifestList(file_format_str, root_path,
pool);
std::vector<ManifestFileMeta> manifest_file_metas;
- EXPECT_OK(manifest_list->Read(file_name, /*filter=*/nullptr,
&manifest_file_metas));
+ EXPECT_OK(manifest_list->Read(file_name, /*filter=*/nullptr,
/*file_size=*/std::nullopt,
+ &manifest_file_metas));
return manifest_file_metas;
}
};
@@ -178,6 +235,60 @@ TEST_F(ManifestListTest, TestReadChangelogManifests) {
ASSERT_EQ(std::vector<ManifestFileMeta>({expected_meta}), actual_metas);
}
+TEST_F(ManifestListTest, TestReadDataManifestsOpensWithSizeFromSnapshot) {
+ auto pool = GetDefaultPool();
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fs = std::make_shared<OpenRecordingFileSystem>();
+ auto manifest_list = CreateManifestList(fs, "avro", dir->Str(), pool);
+ ManifestFileMeta base_meta = MakeMeta("manifest-base", /*file_size=*/100,
/*num_added_files=*/1,
+ /*num_deleted_files=*/0);
+ ManifestFileMeta delta_meta =
+ MakeMeta("manifest-delta", /*file_size=*/200, /*num_added_files=*/2,
+ /*num_deleted_files=*/1);
+ ASSERT_OK_AND_ASSIGN(auto base_list, manifest_list->Write({base_meta}));
+ ASSERT_OK_AND_ASSIGN(auto delta_list, manifest_list->Write({base_meta,
delta_meta}));
+ // Writing only creates files, so everything recorded from here on comes
from the reads.
+ ASSERT_EQ(fs->open_without_length_count, 0);
+ ASSERT_TRUE(fs->opened_lengths.empty());
+
+ Snapshot snapshot =
+ MakeSnapshot(base_list.first, base_list.second, delta_list.first,
delta_list.second);
+ std::vector<ManifestFileMeta> actual_metas;
+ ASSERT_OK(manifest_list->ReadDataManifests(snapshot, &actual_metas));
+ ASSERT_EQ(std::vector<ManifestFileMeta>({base_meta, base_meta,
delta_meta}), actual_metas);
+ // Each list was opened with the length the snapshot carried for it, so
neither read had to ask
+ // the store how long the file is.
+ ASSERT_EQ(std::vector<int64_t>({base_list.second, delta_list.second}),
fs->opened_lengths);
+ ASSERT_EQ(fs->open_without_length_count, 0);
+}
+
+// The sizes are optional in the snapshot format, so a snapshot written before
they were recorded
+// has to keep reading. This is what makes handing the size over an
optimization rather than a new
+// requirement on the metadata.
+TEST_F(ManifestListTest, TestReadDataManifestsWithoutSizesStillReads) {
+ auto pool = GetDefaultPool();
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto fs = std::make_shared<OpenRecordingFileSystem>();
+ auto manifest_list = CreateManifestList(fs, "avro", dir->Str(), pool);
+ ManifestFileMeta base_meta = MakeMeta("manifest-base", /*file_size=*/100,
/*num_added_files=*/1,
+ /*num_deleted_files=*/0);
+ ManifestFileMeta delta_meta =
+ MakeMeta("manifest-delta", /*file_size=*/200, /*num_added_files=*/2,
+ /*num_deleted_files=*/1);
+ ASSERT_OK_AND_ASSIGN(auto base_list, manifest_list->Write({base_meta}));
+ ASSERT_OK_AND_ASSIGN(auto delta_list, manifest_list->Write({base_meta,
delta_meta}));
+
+ Snapshot snapshot = MakeSnapshot(base_list.first,
/*base_size=*/std::nullopt, delta_list.first,
+ /*delta_size=*/std::nullopt);
+ std::vector<ManifestFileMeta> actual_metas;
+ ASSERT_OK(manifest_list->ReadDataManifests(snapshot, &actual_metas));
+ ASSERT_EQ(std::vector<ManifestFileMeta>({base_meta, base_meta,
delta_meta}), actual_metas);
+ ASSERT_TRUE(fs->opened_lengths.empty());
+ ASSERT_EQ(fs->open_without_length_count, 2);
+}
+
TEST_F(ManifestListTest, TestManifestListCompatibleWithJavaPaimon09) {
auto pool = GetDefaultPool();
auto manifest_file_metas = ReadManifestFileMeta("avro",
paimon::test::GetDataDir() + "/avro",
diff --git a/src/paimon/core/operation/commit/commit_scanner.cpp
b/src/paimon/core/operation/commit/commit_scanner.cpp
index 448bd7e4..1a14ac2c 100644
--- a/src/paimon/core/operation/commit/commit_scanner.cpp
+++ b/src/paimon/core/operation/commit/commit_scanner.cpp
@@ -118,8 +118,8 @@ Result<std::vector<ManifestEntry>>
CommitScanner::ReadIncrementalEntries(
std::vector<ManifestEntry> incremental_entries;
for (const ManifestFileMeta& manifest_meta : delta_manifests) {
std::vector<ManifestEntry> manifest_entries;
- PAIMON_RETURN_NOT_OK(
- manifest_file_->Read(manifest_meta.FileName(), /*filter=*/nullptr,
&manifest_entries));
+ PAIMON_RETURN_NOT_OK(manifest_file_->Read(manifest_meta.FileName(),
/*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&manifest_entries));
for (const ManifestEntry& entry : manifest_entries) {
if (changed_partition_set.find(entry.Partition()) !=
changed_partition_set.end()) {
const bool drop_stats =
core_options_.ManifestDeleteFileDropStats() &&
@@ -194,8 +194,8 @@ Result<std::vector<IndexManifestEntry>>
CommitScanner::ReadAllIndexEntriesFromPa
return false;
};
- PAIMON_RETURN_NOT_OK(
- index_manifest_file_->Read(snapshot.IndexManifest().value(), filter,
&index_entries));
+
PAIMON_RETURN_NOT_OK(index_manifest_file_->Read(snapshot.IndexManifest().value(),
filter,
+
/*file_size=*/std::nullopt, &index_entries));
return index_entries;
}
diff --git a/src/paimon/core/operation/expire_snapshots.cpp
b/src/paimon/core/operation/expire_snapshots.cpp
index a7103e65..9915efd4 100644
--- a/src/paimon/core/operation/expire_snapshots.cpp
+++ b/src/paimon/core/operation/expire_snapshots.cpp
@@ -306,7 +306,8 @@ bool ExpireSnapshots::TryDeleteEmptyDirectory(const
std::string& path) const {
Status ExpireSnapshots::CleanUnusedManifests(const std::string&
manifest_list_name,
const std::set<std::string>&
skipping_sets) {
std::vector<ManifestFileMeta> manifest_file_metas;
- auto status = manifest_list_->Read(manifest_list_name, nullptr,
&manifest_file_metas);
+ auto status = manifest_list_->Read(manifest_list_name, nullptr,
/*file_size=*/std::nullopt,
+ &manifest_file_metas);
if (status.ok()) {
std::vector<std::string> to_delete_manifests;
// TODO(jinli.zjw): optimize for async
@@ -326,13 +327,14 @@ Status ExpireSnapshots::CleanUnusedDataFiles(
const std::string& manifest_list_name, const std::set<std::string>&
skipping_data_files,
DataFilePathFactoryCache* data_file_path_factory_cache) {
std::vector<ManifestFileMeta> manifest_file_metas;
- auto status = manifest_list_->Read(manifest_list_name, nullptr,
&manifest_file_metas);
+ auto status = manifest_list_->Read(manifest_list_name, nullptr,
/*file_size=*/std::nullopt,
+ &manifest_file_metas);
if (status.ok()) {
std::map<std::string, ManifestEntry> data_files_to_delete;
for (const auto& manifest_file_meta : manifest_file_metas) {
std::vector<ManifestEntry> manifest_entries;
- auto status =
- manifest_file_->Read(manifest_file_meta.FileName(), nullptr,
&manifest_entries);
+ auto status = manifest_file_->Read(manifest_file_meta.FileName(),
nullptr,
+ /*file_size=*/std::nullopt,
&manifest_entries);
if (!status.ok()) {
// cancel deletion if any exception occurs
PAIMON_LOG_WARN(logger_, "Failed to read some manifest files.
Cancel deletion. %s",
@@ -408,7 +410,8 @@ Status ExpireSnapshots::GetDataFileSkippingSet(
LinkedHashMap<FileEntry::Identifier, ManifestEntry> boundary_entries;
for (const ManifestFileMeta& manifest : manifests) {
std::vector<ManifestEntry> entries;
- PAIMON_RETURN_NOT_OK(manifest_file_->Read(manifest.FileName(),
nullptr, &entries));
+ PAIMON_RETURN_NOT_OK(manifest_file_->Read(manifest.FileName(),
nullptr,
+
/*file_size=*/std::nullopt, &entries));
if (i == 0) {
PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries(entries,
&boundary_entries));
} else {
diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp
b/src/paimon/core/operation/file_store_commit_impl.cpp
index 3fa60e98..06a1eba4 100644
--- a/src/paimon/core/operation/file_store_commit_impl.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl.cpp
@@ -25,6 +25,7 @@
#include <future>
#include <limits>
#include <list>
+#include <optional>
#include <set>
#include <unordered_map>
#include <unordered_set>
@@ -253,7 +254,8 @@ Result<std::vector<ManifestEntry>>
FileStoreCommitImpl::ReadAddManifestEntries(
for (const auto& meta : data_manifests) {
std::vector<ManifestEntry> entries;
PAIMON_RETURN_NOT_OK(manifest_file_->Read(
- meta.FileName(), [](const ManifestEntry&) -> Result<bool> { return
true; }, &entries));
+ meta.FileName(), [](const ManifestEntry&) -> Result<bool> { return
true; },
+ /*file_size=*/std::nullopt, &entries));
unmerged_entries.insert(unmerged_entries.end(),
std::make_move_iterator(entries.begin()),
std::make_move_iterator(entries.end()));
}
@@ -1133,7 +1135,7 @@ Result<std::optional<int64_t>>
FileStoreCommitImpl::MaxSequenceNumber(
std::vector<ManifestEntry> entries;
PAIMON_RETURN_NOT_OK(manifest_file_->Read(
manifest.FileName(), [](const ManifestEntry&) -> Result<bool> {
return true; },
- &entries));
+ /*file_size=*/std::nullopt, &entries));
std::optional<int64_t> current_max =
SequenceSnapshotProperties::MaxSequenceNumberFromFiles(entries);
if (current_max) {
diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp
b/src/paimon/core/operation/file_store_commit_impl_test.cpp
index dfada944..b6710ba8 100644
--- a/src/paimon/core/operation/file_store_commit_impl_test.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp
@@ -23,6 +23,7 @@
#include <cstddef>
#include <filesystem>
#include <iostream>
+#include <optional>
#include <set>
#include <tuple>
#include <utility>
@@ -3372,7 +3373,8 @@ TEST_F(FileStoreCommitImplTest, TestCommitWithIndexFiles)
{
std::vector<IndexManifestEntry> index_entries;
ASSERT_OK(commit_impl->index_manifest_file_->Read(snapshot.IndexManifest().value(),
- /*filter=*/nullptr,
&index_entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &index_entries));
ASSERT_EQ(1u, index_entries.size());
ASSERT_EQ("bitmap-index-commit-1",
index_entries[0].index_file->FileName());
}
@@ -3436,7 +3438,8 @@ TEST_F(FileStoreCommitImplTest,
TestCommitWithCompactIndexFiles) {
std::vector<IndexManifestEntry> index_entries;
ASSERT_OK(commit_impl->index_manifest_file_->Read(snapshot.IndexManifest().value(),
- /*filter=*/nullptr,
&index_entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &index_entries));
ASSERT_EQ(1u, index_entries.size());
ASSERT_EQ("bitmap-index-commit-compact-1",
index_entries[0].index_file->FileName());
}
@@ -3474,7 +3477,8 @@ TEST_F(FileStoreCommitImplTest,
TestCommitWithDeletedIndexFiles) {
std::vector<IndexManifestEntry> index_entries;
ASSERT_OK(commit_impl->index_manifest_file_->Read(snapshot.IndexManifest().value(),
- /*filter=*/nullptr,
&index_entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &index_entries));
ASSERT_TRUE(index_entries.empty());
}
@@ -3513,7 +3517,8 @@ TEST_F(FileStoreCommitImplTest,
TestCommitWithCompactDeletedIndexFiles) {
std::vector<IndexManifestEntry> index_entries;
ASSERT_OK(commit_impl->index_manifest_file_->Read(snapshot.IndexManifest().value(),
- /*filter=*/nullptr,
&index_entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &index_entries));
ASSERT_TRUE(index_entries.empty());
}
@@ -3545,7 +3550,8 @@ TEST_F(FileStoreCommitImplTest,
TestOverwriteWithCompactIndexFiles) {
std::vector<IndexManifestEntry> index_entries;
ASSERT_OK(commit_impl->index_manifest_file_->Read(compact_snapshot.IndexManifest().value(),
- /*filter=*/nullptr,
&index_entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &index_entries));
ASSERT_EQ(1u, index_entries.size());
ASSERT_EQ("bitmap-index-compact-1",
index_entries[0].index_file->FileName());
}
@@ -3638,7 +3644,8 @@ TEST_F(FileStoreCommitImplTest,
TestFilterAndOverwriteWithCompactIndexFiles) {
std::vector<IndexManifestEntry> index_entries;
ASSERT_OK(commit_impl->index_manifest_file_->Read(compact_snapshot.IndexManifest().value(),
- /*filter=*/nullptr,
&index_entries));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &index_entries));
ASSERT_EQ(1u, index_entries.size());
ASSERT_EQ("bitmap-index-filter-compact-1",
index_entries[0].index_file->FileName());
}
@@ -3946,7 +3953,7 @@ TEST_F(FileStoreCommitImplTest,
TestOverwriteDropsDeleteFileStats) {
std::vector<ManifestEntry> delta_entries;
for (const ManifestFileMeta& manifest : delta_manifests) {
ASSERT_OK(commit_impl->manifest_file_->Read(manifest.FileName(),
/*filter=*/nullptr,
- &delta_entries));
+
/*file_size=*/std::nullopt, &delta_entries));
}
int32_t add_count = 0;
diff --git a/src/paimon/core/operation/file_store_scan.cpp
b/src/paimon/core/operation/file_store_scan.cpp
index 2b182107..89c137f3 100644
--- a/src/paimon/core/operation/file_store_scan.cpp
+++ b/src/paimon/core/operation/file_store_scan.cpp
@@ -276,8 +276,33 @@ Status
FileStoreScan::ReadManifests(std::optional<Snapshot>* snapshot_ptr,
Status FileStoreScan::ReadManifestsWithSnapshot(const Snapshot& snapshot,
std::vector<ManifestFileMeta>*
manifests) const {
switch (scan_mode_) {
- case ScanMode::ALL:
- return manifest_list_->ReadDataManifests(snapshot, manifests);
+ case ScanMode::ALL: {
+ // The base and the delta manifest list are two independent files
and neither read
+ // depends on the other, so issue both together instead of paying
the two metadata
+ // round trips one after the other. The result keeps the
base-then-delta order that
+ // ReadDataManifests produced.
+ auto read_list = [this, &snapshot](bool base) ->
Result<std::vector<ManifestFileMeta>> {
+ std::vector<ManifestFileMeta> metas;
+ PAIMON_RETURN_NOT_OK(base ?
manifest_list_->ReadBaseManifests(snapshot, &metas)
+ :
manifest_list_->ReadDeltaManifests(snapshot, &metas));
+ return metas;
+ };
+ std::vector<std::future<Result<std::vector<ManifestFileMeta>>>>
futures;
+ futures.reserve(2);
+ futures.push_back(
+ Via(executor_.get(), [read_list]() { return
read_list(/*base=*/true); }));
+ futures.push_back(
+ Via(executor_.get(), [read_list]() { return
read_list(/*base=*/false); }));
+ for (auto& metas : CollectAll(futures)) {
+ if (!metas.ok()) {
+ return metas.status();
+ }
+ for (auto& meta : metas.value()) {
+ manifests->emplace_back(std::move(meta));
+ }
+ }
+ return Status::OK();
+ }
case ScanMode::DELTA:
return manifest_list_->ReadDeltaManifests(snapshot, manifests);
case ScanMode::CHANGELOG:
@@ -299,8 +324,8 @@ Status FileStoreScan::ReadFileEntries(const
std::vector<ManifestFileMeta>& manif
if (apply_scan_filter) {
PAIMON_RETURN_NOT_OK(ReadManifestFileMeta(meta, &tmp_entries));
} else {
- PAIMON_RETURN_NOT_OK(
- manifest_file_->Read(meta.FileName(), /*filter=*/nullptr,
&tmp_entries));
+ PAIMON_RETURN_NOT_OK(manifest_file_->Read(meta.FileName(),
/*filter=*/nullptr,
+ meta.FileSize(),
&tmp_entries));
}
return tmp_entries;
};
@@ -429,8 +454,8 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries(
for (const auto& meta : manifest_metas) {
auto read_meta_task = [this, meta, bucket]() ->
Result<std::vector<ManifestEntry>> {
std::vector<ManifestEntry> bucket_entries;
- PAIMON_RETURN_NOT_OK(
- manifest_file_->ReadBucketEntries(meta.FileName(), bucket,
&bucket_entries));
+ PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries(
+ meta.FileName(), bucket, meta.FileSize(),
&bucket_entries));
return bucket_entries;
};
futures.push_back(Via(executor_.get(), read_meta_task));
@@ -551,7 +576,7 @@ Status FileStoreScan::ReadManifestFileMeta(const
ManifestFileMeta& manifest,
PAIMON_RETURN_NOT_OK(manifest_file_->Read(
manifest.FileName(),
[this](const ManifestEntry& entry) -> Result<bool> { return
FilterManifestEntry(entry); },
- &unfiltered_entries));
+ manifest.FileSize(), &unfiltered_entries));
entries->reserve(entries->size() + unfiltered_entries.size());
for (auto& entry : unfiltered_entries) {
entries->emplace_back(std::move(entry));
diff --git a/src/paimon/core/operation/manifest_file_merger.cpp
b/src/paimon/core/operation/manifest_file_merger.cpp
index aa308e09..f268e5cd 100644
--- a/src/paimon/core/operation/manifest_file_merger.cpp
+++ b/src/paimon/core/operation/manifest_file_merger.cpp
@@ -22,6 +22,7 @@
#include <cstddef>
#include <functional>
#include <list>
+#include <optional>
#include <string>
#include <utility>
@@ -175,7 +176,8 @@ Result<std::vector<ManifestFileMeta>>
ManifestFileMerger::MergeEntries(
}
std::vector<ManifestEntry> entries;
for (const auto& meta : metas) {
- PAIMON_RETURN_NOT_OK(manifest_file->Read(meta.FileName(),
/*filter=*/nullptr, &entries));
+ PAIMON_RETURN_NOT_OK(manifest_file->Read(meta.FileName(),
/*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&entries));
}
std::vector<ManifestEntry> result;
PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries<ManifestEntry>(entries,
&result));
diff --git a/src/paimon/core/operation/manifest_file_merger_test.cpp
b/src/paimon/core/operation/manifest_file_merger_test.cpp
index 89a78a5a..af0830aa 100644
--- a/src/paimon/core/operation/manifest_file_merger_test.cpp
+++ b/src/paimon/core/operation/manifest_file_merger_test.cpp
@@ -23,6 +23,7 @@
#include <functional>
#include <limits>
#include <map>
+#include <optional>
#include <set>
#include <string>
#include <utility>
@@ -158,7 +159,8 @@ class ManifestFileMergerTest : public testing::Test {
const std::vector<std::pair<std::string, FileKind>>&
entry_file_name_expected) {
std::vector<ManifestEntry> entries;
for (const auto& meta : metas) {
- ASSERT_OK(manifest_file_->Read(meta.FileName(),
/*filter=*/nullptr, &entries));
+ ASSERT_OK(manifest_file_->Read(meta.FileName(), /*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&entries));
}
std::vector<std::pair<std::string, FileKind>> entry_file_name_actual;
for (const auto& entry : entries) {
diff --git a/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
b/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
index f42708c8..0af493ae 100644
--- a/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
+++ b/src/paimon/core/operation/orphan_files_cleaner_impl.cpp
@@ -318,15 +318,19 @@ Result<std::set<std::string>>
OrphanFilesCleanerImpl::GetUsedFilesBySnapshot(
}
std::vector<ManifestFileMeta> manifests;
PAIMON_RETURN_NOT_OK(manifest_list_->ReadIfFileExist(snapshot.BaseManifestList(),
- /*filter=*/nullptr,
&manifests));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &manifests));
PAIMON_RETURN_NOT_OK(manifest_list_->ReadIfFileExist(snapshot.DeltaManifestList(),
- /*filter=*/nullptr,
&manifests));
+ /*filter=*/nullptr,
+
/*file_size=*/std::nullopt, &manifests));
for (const auto& manifest : manifests) {
used_files.insert(manifest.FileName());
std::vector<ManifestEntry> manifest_entries;
- PAIMON_RETURN_NOT_OK(manifest_file_->ReadIfFileExist(
- manifest.FileName(), /*filter=*/nullptr, &manifest_entries));
+
PAIMON_RETURN_NOT_OK(manifest_file_->ReadIfFileExist(manifest.FileName(),
+
/*filter=*/nullptr,
+
/*file_size=*/std::nullopt,
+
&manifest_entries));
for (const auto& manifest_entry : manifest_entries) {
used_files.insert(manifest_entry.FileName());
for (const std::optional<std::string>& extra_file :
diff --git a/src/paimon/core/snapshot_file_scan.cpp
b/src/paimon/core/snapshot_file_scan.cpp
index 700f2028..9d80ec86 100644
--- a/src/paimon/core/snapshot_file_scan.cpp
+++ b/src/paimon/core/snapshot_file_scan.cpp
@@ -184,10 +184,10 @@ class SnapshotFileCollector {
std::vector<ManifestEntry> entries;
if (bucket_id_) {
PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries(
- meta.FileName(), bucket_id_.value(), &entries));
+ meta.FileName(), bucket_id_.value(), meta.FileSize(),
&entries));
} else {
- PAIMON_RETURN_NOT_OK(
- manifest_file_->Read(meta.FileName(),
/*filter=*/nullptr, &entries));
+ PAIMON_RETURN_NOT_OK(manifest_file_->Read(meta.FileName(),
/*filter=*/nullptr,
+ meta.FileSize(),
&entries));
}
PAIMON_RETURN_NOT_OK(ApplyPartitionFilter(&entries));
return ManifestReadResult{meta, std::move(entries)};
diff --git a/src/paimon/core/snapshot_file_scan_test.cpp
b/src/paimon/core/snapshot_file_scan_test.cpp
index 751b39aa..e6629fbd 100644
--- a/src/paimon/core/snapshot_file_scan_test.cpp
+++ b/src/paimon/core/snapshot_file_scan_test.cpp
@@ -23,14 +23,18 @@
#include <initializer_list>
#include <map>
#include <memory>
+#include <mutex>
#include <optional>
#include <set>
#include <string>
+#include <utility>
#include <vector>
#include "gtest/gtest.h"
+#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/string_utils.h"
#include "paimon/defs.h"
+#include "paimon/fs/local/local_file_system.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/scan_context.h"
#include "paimon/testing/utils/testharness.h"
@@ -63,6 +67,34 @@ std::set<std::string> ExpectedFiles(const std::string&
table_path,
return paths;
}
+/// Records how each file was opened, so a scan can be shown to hand over the
lengths that planning
+/// already has instead of leaving the store to answer them again.
`Open(FileStatus)` has to be
+/// overridden here: the base implementation forwards to `Open(path)`, which
would erase the
+/// difference between the two.
+class OpenRecordingFileSystem : public LocalFileSystem {
+ public:
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ {
+ std::lock_guard<std::mutex> lock(opens_mutex);
+ opens.emplace_back(path, std::nullopt);
+ }
+ return LocalFileSystem::Open(path);
+ }
+
+ Result<std::unique_ptr<InputStream>> Open(const FileStatus& file_status)
const override {
+ {
+ std::lock_guard<std::mutex> lock(opens_mutex);
+ opens.emplace_back(file_status.GetPath(), file_status.GetLen());
+ }
+ return LocalFileSystem::Open(file_status.GetPath());
+ }
+
+ // A scan reads manifests concurrently through the executor, so recording
the opens has to be
+ // synchronized; the reads of `opens` happen only after the scan has
collected every task.
+ mutable std::mutex opens_mutex;
+ mutable std::vector<std::pair<std::string, std::optional<int64_t>>> opens;
+};
+
} // namespace
TEST(SnapshotFileScanTest, TestLatestSnapshotAndBucketFilter) {
@@ -314,4 +346,39 @@ TEST(SnapshotFileScanTest, TestInvalidArguments) {
"snapshot file scan does not support predicate
filter");
}
+// The metadata that leads a scan to a manifest already records how long that
manifest is, in the
+// manifest list for a manifest and in the snapshot for a manifest list.
Handing those lengths over
+// is what saves a metadata round trip per file on a remote store, and it is
the whole point of the
+// size fields being carried through planning.
+TEST(SnapshotFileScanTest, TestListFilesOpensManifestsWithKnownLength) {
+ std::string table_path = GetDataDir() + "/orc/append_09.db/append_09";
+ auto file_system = std::make_shared<OpenRecordingFileSystem>();
+
+ ASSERT_OK_AND_ASSIGN(
+ std::set<std::string> all_files,
+ SnapshotFileScan::ListFiles(table_path, /*branch=*/"main",
/*snapshot_id=*/std::nullopt,
+ /*scan_filter=*/nullptr, /*options=*/{},
file_system,
+ /*executor=*/nullptr,
/*memory_pool=*/nullptr));
+ ASSERT_FALSE(all_files.empty());
+
+ // Every manifest the scan read has to be opened with the length the
manifest list recorded for
+ // it. The two manifest lists are not held to that here: this snapshot was
written before the
+ // size fields existed, so they fall back to discovering their own length,
and
+ // ManifestListTest.TestReadDataManifests* covers both sides of that
choice.
+ int manifest_opens = 0;
+ for (const auto& [path, length] : file_system->opens) {
+ const std::string name = PathUtil::GetName(path);
+ if (name.rfind("manifest-list-", 0) == 0) {
+ continue;
+ }
+ if (path.find("/manifest/") == std::string::npos) {
+ continue;
+ }
+ ++manifest_opens;
+ ASSERT_TRUE(length.has_value()) << "opened without a known length: "
<< path;
+ }
+ // The latest snapshot of this table references five manifests.
+ ASSERT_EQ(5, manifest_opens);
+}
+
} // namespace paimon::test
diff --git a/src/paimon/core/table/system/global_system_tables.cpp
b/src/paimon/core/table/system/global_system_tables.cpp
index 68ec9ac0..bd484edc 100644
--- a/src/paimon/core/table/system/global_system_tables.cpp
+++ b/src/paimon/core/table/system/global_system_tables.cpp
@@ -22,6 +22,7 @@
#include <functional>
#include <map>
#include <memory>
+#include <optional>
#include <set>
#include <string>
#include <utility>
@@ -230,8 +231,8 @@ Result<AggregatedFileStats> AggregateFileStats(const
std::shared_ptr<FileSystem>
std::vector<ManifestEntry> entries;
for (const auto& manifest : manifests) {
- PAIMON_RETURN_NOT_OK(
- manifest_file->Read(manifest.FileName(), /*filter=*/nullptr,
&entries));
+ PAIMON_RETURN_NOT_OK(manifest_file->Read(manifest.FileName(),
/*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&entries));
}
std::vector<ManifestEntry> merged_entries;
diff --git a/src/paimon/core/table/system/metadata_system_tables.cpp
b/src/paimon/core/table/system/metadata_system_tables.cpp
index 09399431..5279e687 100644
--- a/src/paimon/core/table/system/metadata_system_tables.cpp
+++ b/src/paimon/core/table/system/metadata_system_tables.cpp
@@ -270,8 +270,8 @@ Result<std::vector<ManifestEntry>>
ReadLatestManifestEntries(
CreateManifestFile(context, path_factory,
core_options, pool));
std::vector<ManifestEntry> entries;
for (const auto& manifest : manifests) {
- PAIMON_RETURN_NOT_OK(
- manifest_file->Read(manifest.FileName(), /*filter=*/nullptr,
&entries));
+ PAIMON_RETURN_NOT_OK(manifest_file->Read(manifest.FileName(),
/*filter=*/nullptr,
+ /*file_size=*/std::nullopt,
&entries));
}
return entries;
}
diff --git a/src/paimon/core/utils/file_utils.cpp
b/src/paimon/core/utils/file_utils.cpp
index 6b087730..012bb44d 100644
--- a/src/paimon/core/utils/file_utils.cpp
+++ b/src/paimon/core/utils/file_utils.cpp
@@ -60,15 +60,16 @@ Status FileUtils::ListOriginalVersionedFiles(const
std::shared_ptr<FileSystem>&
Status FileUtils::ListVersionedFileStatus(const std::shared_ptr<FileSystem>&
fs,
const std::string& dir, const
std::string& prefix,
std::vector<BasicFileStatus>*
file_status_list) {
- PAIMON_ASSIGN_OR_RAISE(bool exist, fs->Exists(dir));
- if (exist) {
- std::vector<BasicFileStatus> file_statuses;
- PAIMON_RETURN_NOT_OK(fs->ListDir(dir, &file_statuses));
- for (auto& file_status : file_statuses) {
- std::string file_name = PathUtil::GetName(file_status.GetPath());
- if (StringUtils::StartsWith(file_name, prefix)) {
- file_status_list->emplace_back(std::move(file_status));
- }
+ // No Exists() probe in front of the listing. Every FileSystem lists a
directory that is not
+ // there as an empty result rather than an error, so the probe only
decided whether to make a
+ // call that answers the same question. On a remote store it costs a round
trip of its own,
+ // which is what a schema or snapshot lookup pays before it can read
anything.
+ std::vector<BasicFileStatus> file_statuses;
+ PAIMON_RETURN_NOT_OK(fs->ListDir(dir, &file_statuses));
+ for (auto& file_status : file_statuses) {
+ std::string file_name = PathUtil::GetName(file_status.GetPath());
+ if (StringUtils::StartsWith(file_name, prefix)) {
+ file_status_list->emplace_back(std::move(file_status));
}
}
return Status::OK();
diff --git a/src/paimon/core/utils/objects_file.h
b/src/paimon/core/utils/objects_file.h
index 56ba6e31..845efd41 100644
--- a/src/paimon/core/utils/objects_file.h
+++ b/src/paimon/core/utils/objects_file.h
@@ -20,6 +20,7 @@
#include <functional>
#include <memory>
+#include <optional>
#include <string>
#include <utility>
#include <vector>
@@ -63,11 +64,15 @@ class ObjectsFile {
virtual ~ObjectsFile() = default;
+ /// @param file_size Length of the file when planning already knows it,
which lets the read
+ /// skip the metadata request a bare `Open` issues on a
remote store. Pass
+ /// std::nullopt when the length is not known; the read
then discovers it
+ /// itself.
Status Read(const std::string& file_name, const
std::function<Result<bool>(const T&)>& filter,
- std::vector<T>* result) const;
+ std::optional<int64_t> file_size, std::vector<T>* result)
const;
Status ReadIfFileExist(const std::string& file_name,
const std::function<Result<bool>(const T&)>& filter,
- std::vector<T>* result) const;
+ std::optional<int64_t> file_size, std::vector<T>*
result) const;
void DeleteQuietly(const std::string& file_name) {
std::string path = path_factory_->ToPath(file_name);
@@ -89,7 +94,8 @@ class ObjectsFile {
Status ReadArrowBatches(
const std::string& file_name,
- const std::function<Status(const
std::shared_ptr<arrow::StructArray>&)>& consumer) const;
+ const std::function<Status(const
std::shared_ptr<arrow::StructArray>&)>& consumer,
+ std::optional<int64_t> file_size) const;
std::shared_ptr<PathFactory> path_factory_;
std::shared_ptr<MemoryPool> pool_;
@@ -105,7 +111,12 @@ class ObjectsFile {
std::string compression_;
std::shared_ptr<Cache> cache_;
- Result<MemorySegment> ReadFileSegment(const std::string& file_path) const;
+ Result<MemorySegment> ReadFileSegment(const std::string& file_path,
+ const std::optional<int64_t>&
file_size) const;
+
+ /// Opens the file for reading, handing over the length when the caller
already has it.
+ Result<std::unique_ptr<InputStream>> OpenForRead(const std::string&
file_path,
+ const
std::optional<int64_t>& file_size) const;
};
template <typename T>
@@ -132,11 +143,12 @@ ObjectsFile<T>::ObjectsFile(const
std::shared_ptr<FileSystem>& file_system,
template <typename T>
Status ObjectsFile<T>::ReadIfFileExist(const std::string& file_name,
const std::function<Result<bool>(const
T&)>& filter,
+ std::optional<int64_t> file_size,
std::vector<T>* result) const {
std::string file_path = path_factory_->ToPath(file_name);
PAIMON_ASSIGN_OR_RAISE(bool path_exist, file_system_->Exists(file_path));
if (path_exist) {
- return Read(file_name, filter, result);
+ return Read(file_name, filter, file_size, result);
}
return Status::OK();
}
@@ -144,7 +156,7 @@ Status ObjectsFile<T>::ReadIfFileExist(const std::string&
file_name,
template <typename T>
Status ObjectsFile<T>::Read(const std::string& file_name,
const std::function<Result<bool>(const T&)>&
filter,
- std::vector<T>* result) const {
+ std::optional<int64_t> file_size, std::vector<T>*
result) const {
return ReadArrowBatches(
file_name,
[this, &filter, result](const std::shared_ptr<arrow::StructArray>&
struct_array) -> Status {
@@ -164,13 +176,15 @@ Status ObjectsFile<T>::Read(const std::string& file_name,
}
}
return Status::OK();
- });
+ },
+ file_size);
}
template <typename T>
Status ObjectsFile<T>::ReadArrowBatches(
const std::string& file_name,
- const std::function<Status(const std::shared_ptr<arrow::StructArray>&)>&
consumer) const {
+ const std::function<Status(const std::shared_ptr<arrow::StructArray>&)>&
consumer,
+ std::optional<int64_t> file_size) const {
std::string file_path = path_factory_->ToPath(file_name);
std::shared_ptr<InputStream> file_input_stream;
std::shared_ptr<Bytes> cached_bytes;
@@ -180,9 +194,9 @@ Status ObjectsFile<T>::ReadArrowBatches(
auto cache_key =
CacheKey::ForKind(file_path, /*position=*/0, /*length=*/-1,
CacheKind::MANIFEST);
auto supplier =
- [this,
- &file_path](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
- PAIMON_ASSIGN_OR_RAISE(MemorySegment segment,
ReadFileSegment(file_path));
+ [this, &file_path,
+ &file_size](const std::shared_ptr<CacheKey>&) ->
Result<std::shared_ptr<CacheValue>> {
+ PAIMON_ASSIGN_OR_RAISE(MemorySegment segment,
ReadFileSegment(file_path, file_size));
return std::make_shared<CacheValue>(segment, CacheCallback());
};
Result<std::shared_ptr<CacheValue>> cache_result =
cache_->Get(cache_key, supplier);
@@ -195,7 +209,7 @@ Status ObjectsFile<T>::ReadArrowBatches(
}
if (!file_input_stream) {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream>
unique_file_input_stream,
- file_system_->Open(file_path));
+ OpenForRead(file_path, file_size));
file_input_stream =
std::shared_ptr<InputStream>(std::move(unique_file_input_stream));
}
@@ -223,9 +237,24 @@ Status ObjectsFile<T>::ReadArrowBatches(
}
template <typename T>
-Result<MemorySegment> ObjectsFile<T>::ReadFileSegment(const std::string&
file_path) const {
+Result<std::unique_ptr<InputStream>> ObjectsFile<T>::OpenForRead(
+ const std::string& file_path, const std::optional<int64_t>& file_size)
const {
+ if (file_size.has_value()) {
+ // Planning already read this length out of the manifest metadata, and
`Open(FileStatus)` is
+ // documented to let the file system skip the metadata request a bare
open issues. That
+ // request is a round trip of its own on a remote store, paid before a
single byte of the
+ // file is read. The files here are written once and never rewritten,
so a length recorded
+ // at planning time cannot go stale underneath the read.
+ return file_system_->Open(FileStatus(file_path, file_size.value()));
+ }
+ return file_system_->Open(file_path);
+}
+
+template <typename T>
+Result<MemorySegment> ObjectsFile<T>::ReadFileSegment(
+ const std::string& file_path, const std::optional<int64_t>& file_size)
const {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
- file_system_->Open(file_path));
+ OpenForRead(file_path, file_size));
PAIMON_ASSIGN_OR_RAISE(int64_t input_length, input_stream->Length());
PAIMON_RETURN_NOT_OK(input_stream->Seek(0, FS_SEEK_SET));
diff --git a/src/paimon/fs/jindo/jindo_file_system.cpp
b/src/paimon/fs/jindo/jindo_file_system.cpp
index 793f2b2f..866cbade 100644
--- a/src/paimon/fs/jindo/jindo_file_system.cpp
+++ b/src/paimon/fs/jindo/jindo_file_system.cpp
@@ -173,11 +173,18 @@ Result<FileStatus> JindoFileSystem::GetFileStatus(const
std::string& path) const
Status JindoFileSystem::ListDir(const std::string& directory,
std::vector<BasicFileStatus>*
file_status_list) const {
- PAIMON_ASSIGN_OR_RAISE(bool exist, Exists(directory));
- if (!exist) {
- return Status::OK();
+ // One status call answers what Exists() followed by GetFileStatus() asked
the store twice:
+ // whether the path is there at all, and whether it is a directory.
PAIMON_RETURN_NOT_OK_FROM_
+ // JINDO maps the SDK's not-found to Status::NotExist, which is what tells
a missing directory
+ // (listed as empty, as the other file systems do) from a call that
genuinely failed.
+ Result<FileStatus> dir_status = GetFileStatus(directory);
+ if (!dir_status.ok()) {
+ if (dir_status.status().IsNotExist()) {
+ return Status::OK();
+ }
+ return dir_status.status();
}
- PAIMON_ASSIGN_OR_RAISE(FileStatus file_status, GetFileStatus(directory));
+ const FileStatus& file_status = dir_status.value();
if (!file_status.IsDir()) {
return Status::Invalid(
fmt::format("file {} already exists and is not a directory",
file_status.GetPath()));
diff --git a/src/paimon/fs/jindo/jindo_file_system_test.cpp
b/src/paimon/fs/jindo/jindo_file_system_test.cpp
index e8b26dc6..714d1415 100644
--- a/src/paimon/fs/jindo/jindo_file_system_test.cpp
+++ b/src/paimon/fs/jindo/jindo_file_system_test.cpp
@@ -83,6 +83,22 @@ TEST_F(JindoFileSystemTest, TestRename) {
ASSERT_TRUE(is_exist);
}
+// Pins the contract the planning optimizations lean on: the SDK answers a
target that is not
+// there with its file-not-found error, the mapping surfaces that as Paimon's
NotExist (so callers
+// can tell "absent" apart from "the call failed"), and listing a missing
directory yields an empty
+// result instead of an error.
+TEST_F(JindoFileSystemTest,
TestMissingPathReportsNotExistAndMissingDirListsEmpty) {
+ const std::string missing_file = test_dir_ +
"no_such_dir/no_such_file.data";
+ Result<FileStatus> missing_status = fs_->GetFileStatus(missing_file);
+ ASSERT_FALSE(missing_status.ok());
+ ASSERT_TRUE(missing_status.status().IsNotExist()) <<
missing_status.status().ToString();
+
+ const std::string missing_dir = test_dir_ + "no_such_dir/";
+ std::vector<BasicFileStatus> file_statuses;
+ ASSERT_OK(fs_->ListDir(missing_dir, &file_statuses));
+ ASSERT_TRUE(file_statuses.empty());
+}
+
TEST_F(JindoFileSystemTest, TestSeek) {
std::string content = "abcdefghijk";
std::string file_path = test_dir_ + "file.data";
