From eb5e96ab06166def5c09fe2ebdf3ddf6cc521319 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Mon, 14 Sep 2026 16:32:46 +0800 Subject: [PATCH 1/5] perf(read): reduce metadata round trips during scan planning --- src/paimon/core/manifest/manifest_file.cpp | 7 +- src/paimon/core/manifest/manifest_file.h | 7 +- .../core/manifest/manifest_file_test.cpp | 53 +++++++++ src/paimon/core/manifest/manifest_list.h | 9 +- .../core/manifest/manifest_list_test.cpp | 112 +++++++++++++++++- src/paimon/core/operation/file_store_scan.cpp | 39 ++++-- src/paimon/core/snapshot_file_scan.cpp | 6 +- src/paimon/core/snapshot_file_scan_test.cpp | 57 +++++++++ src/paimon/core/utils/file_utils.cpp | 19 +-- src/paimon/core/utils/file_utils_test.cpp | 48 ++++++++ src/paimon/core/utils/objects_file.h | 59 ++++++--- src/paimon/fs/jindo/jindo_file_system.cpp | 15 ++- 12 files changed, 386 insertions(+), 45 deletions(-) diff --git a/src/paimon/core/manifest/manifest_file.cpp b/src/paimon/core/manifest/manifest_file.cpp index 0fdb78d70..e3d1e6bc8 100644 --- a/src/paimon/core/manifest/manifest_file.cpp +++ b/src/paimon/core/manifest/manifest_file.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include "arrow/c/abi.h" @@ -90,7 +91,8 @@ Result> ManifestFile::Create( } Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t bucket, - std::vector* entries) const { + std::vector* entries, + std::optional file_size) const { return ReadArrowBatches( file_name, [this, bucket, entries](const std::shared_ptr& batch) -> Status { @@ -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> ManifestFile::Write( diff --git a/src/paimon/core/manifest/manifest_file.h b/src/paimon/core/manifest/manifest_file.h index 528bb7bf5..189e229e6 100644 --- a/src/paimon/core/manifest/manifest_file.h +++ b/src/paimon/core/manifest/manifest_file.h @@ -20,6 +20,7 @@ #include #include +#include #include #include @@ -63,8 +64,12 @@ class ManifestFile : public ObjectsFile { Result> Write(const std::vector& 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. Status ReadBucketEntries(const std::string& file_name, int32_t bucket, - std::vector* entries) const; + std::vector* entries, + std::optional file_size = std::nullopt) const; private: ManifestFile(const std::shared_ptr& file_system, diff --git a/src/paimon/core/manifest/manifest_file_test.cpp b/src/paimon/core/manifest/manifest_file_test.cpp index 5cd264aa3..765ed3541 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> Open(const FileStatus& file_status) const override { + opened_lengths.push_back(file_status.GetLen()); + return local_.Open(file_status.GetPath()); + } + Result> 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 opened_lengths; private: LocalFileSystem local_; @@ -366,6 +374,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(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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 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 all_entries; + ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr, &all_entries, kRecordedSize)); + ASSERT_EQ(2, all_entries.size()); + ASSERT_EQ(std::vector({kRecordedSize}), counting_file_system->opened_lengths); + ASSERT_EQ(0, counting_file_system->open_count); + + std::vector bucket_one_entries; + ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1, &bucket_one_entries, + kRecordedSize)); + ASSERT_EQ(std::vector({all_entries[0]}), bucket_one_entries); + ASSERT_EQ(std::vector({kRecordedSize, kRecordedSize}), + counting_file_system->opened_lengths); + ASSERT_EQ(0, counting_file_system->open_count); +} + TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) { auto pool = GetDefaultPool(); std::vector source_entries = diff --git a/src/paimon/core/manifest/manifest_list.h b/src/paimon/core/manifest/manifest_list.h index 6bf64e8ef..9516b3332 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 { /// @return Status indicating whether the operation was successful or not. Status ReadBaseManifests(const Snapshot& snapshot, std::vector* manifests) const { - return Read(snapshot.BaseManifestList(), /*filter=*/nullptr, manifests); + return Read(snapshot.BaseManifestList(), /*filter=*/nullptr, manifests, + snapshot.BaseManifestListSize()); } /// Return a `ManifestFileMeta` for each delta manifest in this snapshot. @@ -100,7 +101,8 @@ class ManifestList : public ObjectsFile { /// @return Status indicating whether the operation was successful or not. Status ReadDeltaManifests(const Snapshot& snapshot, std::vector* manifests) const { - return Read(snapshot.DeltaManifestList(), /*filter=*/nullptr, manifests); + return Read(snapshot.DeltaManifestList(), /*filter=*/nullptr, manifests, + snapshot.DeltaManifestListSize()); } /// Return a `ManifestFileMeta` for each changelog manifest in this snapshot. @@ -113,7 +115,8 @@ class ManifestList : public ObjectsFile { const std::optional& 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, manifests, + snapshot.ChangelogManifestListSize()); } 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 97b420cf5..7021f321a 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 +#include #include #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> Open(const std::string& path) const override { + ++open_without_length_count; + return LocalFileSystem::Open(path); + } + + Result> 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 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& base_size, + const std::string& delta_list, const std::optional& 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 CreateManifestList( const std::string& file_format_str, const std::string& root_path, const std::shared_ptr& pool) const { - std::shared_ptr file_system = std::make_shared(); + return CreateManifestList(std::make_shared(), file_format_str, root_path, + pool); + } + + std::unique_ptr CreateManifestList( + const std::shared_ptr& file_system, const std::string& file_format_str, + const std::string& root_path, const std::shared_ptr& pool) const { EXPECT_OK_AND_ASSIGN(std::shared_ptr file_format, FileFormatFactory::Get(file_format_str, {})); auto unused_schema = arrow::schema(arrow::FieldVector({arrow::field("f0", arrow::utf8())})); @@ -178,6 +234,60 @@ TEST_F(ManifestListTest, TestReadChangelogManifests) { ASSERT_EQ(std::vector({expected_meta}), actual_metas); } +TEST_F(ManifestListTest, TestReadDataManifestsOpensWithSizeFromSnapshot) { + auto pool = GetDefaultPool(); + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto fs = std::make_shared(); + 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 actual_metas; + ASSERT_OK(manifest_list->ReadDataManifests(snapshot, &actual_metas)); + ASSERT_EQ(std::vector({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({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(); + 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 actual_metas; + ASSERT_OK(manifest_list->ReadDataManifests(snapshot, &actual_metas)); + ASSERT_EQ(std::vector({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/file_store_scan.cpp b/src/paimon/core/operation/file_store_scan.cpp index 2b182107c..7e66269f6 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_ptr, Status FileStoreScan::ReadManifestsWithSnapshot(const Snapshot& snapshot, std::vector* 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 metas; + PAIMON_RETURN_NOT_OK(base ? manifest_list_->ReadBaseManifests(snapshot, &metas) + : manifest_list_->ReadDeltaManifests(snapshot, &metas)); + return metas; + }; + std::vector>>> 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& 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, + &tmp_entries, meta.FileSize())); } 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 bucket_entries; - PAIMON_RETURN_NOT_OK( - manifest_file_->ReadBucketEntries(meta.FileName(), bucket, &bucket_entries)); + PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries( + meta.FileName(), bucket, &bucket_entries, meta.FileSize())); 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 { return FilterManifestEntry(entry); }, - &unfiltered_entries)); + &unfiltered_entries, manifest.FileSize())); entries->reserve(entries->size() + unfiltered_entries.size()); for (auto& entry : unfiltered_entries) { entries->emplace_back(std::move(entry)); diff --git a/src/paimon/core/snapshot_file_scan.cpp b/src/paimon/core/snapshot_file_scan.cpp index 700f20288..86e2f4178 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 entries; if (bucket_id_) { PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries( - meta.FileName(), bucket_id_.value(), &entries)); + meta.FileName(), bucket_id_.value(), &entries, meta.FileSize())); } else { - PAIMON_RETURN_NOT_OK( - manifest_file_->Read(meta.FileName(), /*filter=*/nullptr, &entries)); + PAIMON_RETURN_NOT_OK(manifest_file_->Read(meta.FileName(), /*filter=*/nullptr, + &entries, meta.FileSize())); } 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 751b39aaa..fbc680919 100644 --- a/src/paimon/core/snapshot_file_scan_test.cpp +++ b/src/paimon/core/snapshot_file_scan_test.cpp @@ -26,11 +26,14 @@ #include #include #include +#include #include #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 +66,25 @@ std::set 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> Open(const std::string& path) const override { + opens.emplace_back(path, std::nullopt); + return LocalFileSystem::Open(path); + } + + Result> Open(const FileStatus& file_status) const override { + opens.emplace_back(file_status.GetPath(), file_status.GetLen()); + return LocalFileSystem::Open(file_status.GetPath()); + } + + mutable std::vector>> opens; +}; + } // namespace TEST(SnapshotFileScanTest, TestLatestSnapshotAndBucketFilter) { @@ -314,4 +336,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(); + + ASSERT_OK_AND_ASSIGN( + std::set 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/utils/file_utils.cpp b/src/paimon/core/utils/file_utils.cpp index 6b0877307..012bb44d9 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& Status FileUtils::ListVersionedFileStatus(const std::shared_ptr& fs, const std::string& dir, const std::string& prefix, std::vector* file_status_list) { - PAIMON_ASSIGN_OR_RAISE(bool exist, fs->Exists(dir)); - if (exist) { - std::vector 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 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/file_utils_test.cpp b/src/paimon/core/utils/file_utils_test.cpp index ef3db4d6d..3cb31b33f 100644 --- a/src/paimon/core/utils/file_utils_test.cpp +++ b/src/paimon/core/utils/file_utils_test.cpp @@ -27,6 +27,30 @@ namespace paimon::test { +namespace { + +/// Counts what a listing asks of the file system, so the round trips it saves stay saved. On a +/// remote store an `Exists()` in front of a `ListDir()` is a trip of its own, and the listing +/// already answers the same question: a directory that is not there comes back empty. +class ListingCountingFileSystem : public LocalFileSystem { + public: + Result Exists(const std::string& path) const override { + ++exists_count; + return LocalFileSystem::Exists(path); + } + + Status ListDir(const std::string& directory, + std::vector* file_status_list) const override { + ++list_dir_count; + return LocalFileSystem::ListDir(directory, file_status_list); + } + + mutable int exists_count = 0; + mutable int list_dir_count = 0; +}; + +} // namespace + TEST(FileUtilsTest, TestSimple) { std::string test_data_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/snapshot/"; @@ -53,4 +77,28 @@ TEST(FileUtilsTest, TestNotNumber) { ASSERT_NOK(FileUtils::ListVersionedFiles(std::move(fs), test_data_path, "manifest-", &files)); } +TEST(FileUtilsTest, TestListVersionedFilesListsWithoutProbingExistence) { + std::string test_data_path = + paimon::test::GetDataDir() + "/orc/append_09.db/append_09/snapshot/"; + auto fs = std::make_shared(); + std::vector files; + ASSERT_OK(FileUtils::ListVersionedFiles(fs, test_data_path, "snapshot-", &files)); + ASSERT_EQ(files.size(), 5u); + ASSERT_EQ(fs->list_dir_count, 1); + ASSERT_EQ(fs->exists_count, 0); +} + +// The missing directory is the case the probe used to be justified by, so it is the one that has +// to reach the same answer without it. +TEST(FileUtilsTest, TestListVersionedFilesMissingDirListsWithoutProbingExistence) { + std::string test_data_path = + paimon::test::GetDataDir() + "/orc/append_09.db/append_09/not_exist/"; + auto fs = std::make_shared(); + std::vector files; + ASSERT_OK(FileUtils::ListVersionedFiles(fs, test_data_path, "snapshot-", &files)); + ASSERT_EQ(files.size(), 0u); + ASSERT_EQ(fs->list_dir_count, 1); + ASSERT_EQ(fs->exists_count, 0); +} + } // namespace paimon::test diff --git a/src/paimon/core/utils/objects_file.h b/src/paimon/core/utils/objects_file.h index 56ba6e319..2b9011af1 100644 --- a/src/paimon/core/utils/objects_file.h +++ b/src/paimon/core/utils/objects_file.h @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -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. Leave it + /// unset when the length is not known; the read then discovers it itself. Status Read(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result) const; + std::vector* result, std::optional file_size = std::nullopt) const; Status ReadIfFileExist(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result) const; + std::vector* result, + std::optional file_size = std::nullopt) 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&)>& consumer) const; + const std::function&)>& consumer, + std::optional file_size = std::nullopt) const; std::shared_ptr path_factory_; std::shared_ptr pool_; @@ -105,7 +111,12 @@ class ObjectsFile { std::string compression_; std::shared_ptr cache_; - Result ReadFileSegment(const std::string& file_path) const; + Result ReadFileSegment(const std::string& file_path, + const std::optional& file_size) const; + + /// Opens the file for reading, handing over the length when the caller already has it. + Result> OpenForRead(const std::string& file_path, + const std::optional& file_size) const; }; template @@ -132,11 +143,12 @@ ObjectsFile::ObjectsFile(const std::shared_ptr& file_system, template Status ObjectsFile::ReadIfFileExist(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result) const { + std::vector* result, + std::optional file_size) 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, result, file_size); } return Status::OK(); } @@ -144,7 +156,7 @@ Status ObjectsFile::ReadIfFileExist(const std::string& file_name, template Status ObjectsFile::Read(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result) const { + std::vector* result, std::optional file_size) const { return ReadArrowBatches( file_name, [this, &filter, result](const std::shared_ptr& struct_array) -> Status { @@ -164,13 +176,15 @@ Status ObjectsFile::Read(const std::string& file_name, } } return Status::OK(); - }); + }, + file_size); } template Status ObjectsFile::ReadArrowBatches( const std::string& file_name, - const std::function&)>& consumer) const { + const std::function&)>& consumer, + std::optional file_size) const { std::string file_path = path_factory_->ToPath(file_name); std::shared_ptr file_input_stream; std::shared_ptr cached_bytes; @@ -180,9 +194,9 @@ Status ObjectsFile::ReadArrowBatches( auto cache_key = CacheKey::ForKind(file_path, /*position=*/0, /*length=*/-1, CacheKind::MANIFEST); auto supplier = - [this, - &file_path](const std::shared_ptr&) -> Result> { - PAIMON_ASSIGN_OR_RAISE(MemorySegment segment, ReadFileSegment(file_path)); + [this, &file_path, + &file_size](const std::shared_ptr&) -> Result> { + PAIMON_ASSIGN_OR_RAISE(MemorySegment segment, ReadFileSegment(file_path, file_size)); return std::make_shared(segment, CacheCallback()); }; Result> cache_result = cache_->Get(cache_key, supplier); @@ -195,7 +209,7 @@ Status ObjectsFile::ReadArrowBatches( } if (!file_input_stream) { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr unique_file_input_stream, - file_system_->Open(file_path)); + OpenForRead(file_path, file_size)); file_input_stream = std::shared_ptr(std::move(unique_file_input_stream)); } @@ -223,9 +237,24 @@ Status ObjectsFile::ReadArrowBatches( } template -Result ObjectsFile::ReadFileSegment(const std::string& file_path) const { +Result> ObjectsFile::OpenForRead( + const std::string& file_path, const std::optional& 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 +Result ObjectsFile::ReadFileSegment( + const std::string& file_path, const std::optional& file_size) const { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr 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 793f2b2fe..866cbade8 100644 --- a/src/paimon/fs/jindo/jindo_file_system.cpp +++ b/src/paimon/fs/jindo/jindo_file_system.cpp @@ -173,11 +173,18 @@ Result JindoFileSystem::GetFileStatus(const std::string& path) const Status JindoFileSystem::ListDir(const std::string& directory, std::vector* 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 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())); From 7b172c354ce5009d8601506a820167bfe8135dad Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Mon, 14 Sep 2026 21:21:41 +0800 Subject: [PATCH 2/5] fix --- src/paimon/core/index/index_file_handler.cpp | 4 +- .../manifest/index_manifest_file_handler.cpp | 4 +- .../index_manifest_file_handler_test.cpp | 18 +++++--- src/paimon/core/manifest/manifest_file.cpp | 4 +- src/paimon/core/manifest/manifest_file.h | 7 ++-- .../core/manifest/manifest_file_test.cpp | 42 +++++++++++-------- src/paimon/core/manifest/manifest_list.h | 12 +++--- .../core/manifest/manifest_list_test.cpp | 3 +- .../core/operation/commit/commit_scanner.cpp | 8 ++-- .../core/operation/expire_snapshots.cpp | 10 +++-- .../core/operation/file_store_commit_impl.cpp | 6 ++- .../operation/file_store_commit_impl_test.cpp | 21 ++++++---- src/paimon/core/operation/file_store_scan.cpp | 6 +-- .../core/operation/manifest_file_merger.cpp | 4 +- .../operation/manifest_file_merger_test.cpp | 4 +- .../operation/orphan_files_cleaner_impl.cpp | 12 ++++-- src/paimon/core/snapshot_file_scan.cpp | 4 +- .../table/system/global_system_tables.cpp | 5 ++- .../table/system/metadata_system_tables.cpp | 4 +- src/paimon/core/utils/objects_file.h | 20 ++++----- 20 files changed, 118 insertions(+), 80 deletions(-) diff --git a/src/paimon/core/index/index_file_handler.cpp b/src/paimon/core/index/index_file_handler.cpp index 9b45e748a..d614d85e6 100644 --- a/src/paimon/core/index/index_file_handler.cpp +++ b/src/paimon/core/index/index_file_handler.cpp @@ -55,8 +55,8 @@ Result> IndexFileHandler::Scan( return std::vector(); } std::vector 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 415437298..55261de95 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 #include #include #include @@ -201,7 +202,8 @@ Result IndexManifestFileHandler::Write( std::vector 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 9661697c0..42048dc75 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 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 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 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 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 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 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 e3d1e6bc8..ecbcb6bdc 100644 --- a/src/paimon/core/manifest/manifest_file.cpp +++ b/src/paimon/core/manifest/manifest_file.cpp @@ -91,8 +91,8 @@ Result> ManifestFile::Create( } Status ManifestFile::ReadBucketEntries(const std::string& file_name, int32_t bucket, - std::vector* entries, - std::optional file_size) const { + std::optional file_size, + std::vector* entries) const { return ReadArrowBatches( file_name, [this, bucket, entries](const std::shared_ptr& batch) -> Status { diff --git a/src/paimon/core/manifest/manifest_file.h b/src/paimon/core/manifest/manifest_file.h index 189e229e6..2f3a60cb3 100644 --- a/src/paimon/core/manifest/manifest_file.h +++ b/src/paimon/core/manifest/manifest_file.h @@ -66,10 +66,11 @@ class ManifestFile : public ObjectsFile { /// 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. + /// 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::vector* entries, - std::optional file_size = std::nullopt) const; + std::optional file_size, + std::vector* entries) const; private: ManifestFile(const std::shared_ptr& file_system, diff --git a/src/paimon/core/manifest/manifest_file_test.cpp b/src/paimon/core/manifest/manifest_file_test.cpp index 765ed3541..0ce9cc728 100644 --- a/src/paimon/core/manifest/manifest_file_test.cpp +++ b/src/paimon/core/manifest/manifest_file_test.cpp @@ -132,10 +132,12 @@ class ManifestFileTest : public testing::Test { /*target_file_size=*/1024, pool, options, unused_schema)); std::vector 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; @@ -271,7 +273,7 @@ TEST_F(ManifestFileTest, TestManifestCacheIsDisabledWithoutInjectedCache) { std::vector 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); @@ -280,7 +282,7 @@ TEST_F(ManifestFileTest, TestManifestCacheIsDisabledWithoutInjectedCache) { ASSERT_OK(manifest_file->Read( "manifest-3ea5ee21-d399-4f1c-a749-2fc63dbf0852-1", [](const ManifestEntry& entry) -> Result { 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); @@ -313,10 +315,10 @@ TEST_F(ManifestFileTest, TestManifestCacheReusesCachedBytes) { std::vector 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 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); @@ -353,20 +355,24 @@ TEST_F(ManifestFileTest, TestReadBucketEntriesMaterializesOnlySelectedBucket) { const std::string manifest_name = "manifest-3a44a0da-1008-463c-914e-28d271375e24-0"; std::vector 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 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({all_entries[0]}), bucket_one_entries); std::vector 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({all_entries[1]}), bucket_zero_entries); std::vector 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); @@ -405,14 +411,14 @@ TEST_F(ManifestFileTest, TestReadPassesKnownSizeToOpen) { constexpr int64_t kRecordedSize = 2617; std::vector all_entries; - ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr, &all_entries, kRecordedSize)); + ASSERT_OK(manifest_file->Read(manifest_name, /*filter=*/nullptr, kRecordedSize, &all_entries)); ASSERT_EQ(2, all_entries.size()); ASSERT_EQ(std::vector({kRecordedSize}), counting_file_system->opened_lengths); ASSERT_EQ(0, counting_file_system->open_count); std::vector bucket_one_entries; - ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1, &bucket_one_entries, - kRecordedSize)); + ASSERT_OK(manifest_file->ReadBucketEntries(manifest_name, /*bucket=*/1, kRecordedSize, + &bucket_one_entries)); ASSERT_EQ(std::vector({all_entries[0]}), bucket_one_entries); ASSERT_EQ(std::vector({kRecordedSize, kRecordedSize}), counting_file_system->opened_lengths); @@ -459,11 +465,13 @@ TEST_F(ManifestFileTest, TestReadBucketEntriesSkipsDeserializingOtherBuckets) { manifest_file->WriteWithoutRolling({invalid_other_bucket, valid_target_bucket})); std::vector 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 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({valid_target_bucket}), bucket_entries); } diff --git a/src/paimon/core/manifest/manifest_list.h b/src/paimon/core/manifest/manifest_list.h index 9516b3332..918d04918 100644 --- a/src/paimon/core/manifest/manifest_list.h +++ b/src/paimon/core/manifest/manifest_list.h @@ -90,8 +90,8 @@ class ManifestList : public ObjectsFile { /// @return Status indicating whether the operation was successful or not. Status ReadBaseManifests(const Snapshot& snapshot, std::vector* manifests) const { - return Read(snapshot.BaseManifestList(), /*filter=*/nullptr, manifests, - snapshot.BaseManifestListSize()); + return Read(snapshot.BaseManifestList(), /*filter=*/nullptr, + snapshot.BaseManifestListSize(), manifests); } /// Return a `ManifestFileMeta` for each delta manifest in this snapshot. @@ -101,8 +101,8 @@ class ManifestList : public ObjectsFile { /// @return Status indicating whether the operation was successful or not. Status ReadDeltaManifests(const Snapshot& snapshot, std::vector* manifests) const { - return Read(snapshot.DeltaManifestList(), /*filter=*/nullptr, manifests, - snapshot.DeltaManifestListSize()); + return Read(snapshot.DeltaManifestList(), /*filter=*/nullptr, + snapshot.DeltaManifestListSize(), manifests); } /// Return a `ManifestFileMeta` for each changelog manifest in this snapshot. @@ -115,8 +115,8 @@ class ManifestList : public ObjectsFile { const std::optional& changelog_manifest_list = snapshot.ChangelogManifestList(); if (changelog_manifest_list) { - return Read(changelog_manifest_list.value(), /*filter=*/nullptr, manifests, - snapshot.ChangelogManifestListSize()); + 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 7021f321a..d8d255380 100644 --- a/src/paimon/core/manifest/manifest_list_test.cpp +++ b/src/paimon/core/manifest/manifest_list_test.cpp @@ -121,7 +121,8 @@ class ManifestListTest : public testing::Test { const std::string& file_name, const std::shared_ptr& pool) const { auto manifest_list = CreateManifestList(file_format_str, root_path, pool); std::vector 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; } }; diff --git a/src/paimon/core/operation/commit/commit_scanner.cpp b/src/paimon/core/operation/commit/commit_scanner.cpp index 448bd7e47..1a14ac2c6 100644 --- a/src/paimon/core/operation/commit/commit_scanner.cpp +++ b/src/paimon/core/operation/commit/commit_scanner.cpp @@ -118,8 +118,8 @@ Result> CommitScanner::ReadIncrementalEntries( std::vector incremental_entries; for (const ManifestFileMeta& manifest_meta : delta_manifests) { std::vector 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> 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 a7103e65a..e1c05c501 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& skipping_sets) { std::vector 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 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& skipping_data_files, DataFilePathFactoryCache* data_file_path_factory_cache) { std::vector 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 data_files_to_delete; for (const auto& manifest_file_meta : manifest_file_metas) { std::vector 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", diff --git a/src/paimon/core/operation/file_store_commit_impl.cpp b/src/paimon/core/operation/file_store_commit_impl.cpp index 3fa60e984..06a1eba43 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 #include #include +#include #include #include #include @@ -253,7 +254,8 @@ Result> FileStoreCommitImpl::ReadAddManifestEntries( for (const auto& meta : data_manifests) { std::vector entries; PAIMON_RETURN_NOT_OK(manifest_file_->Read( - meta.FileName(), [](const ManifestEntry&) -> Result { return true; }, &entries)); + meta.FileName(), [](const ManifestEntry&) -> Result { 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> FileStoreCommitImpl::MaxSequenceNumber( std::vector entries; PAIMON_RETURN_NOT_OK(manifest_file_->Read( manifest.FileName(), [](const ManifestEntry&) -> Result { return true; }, - &entries)); + /*file_size=*/std::nullopt, &entries)); std::optional 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 dfada9443..b6710ba81 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 #include #include +#include #include #include #include @@ -3372,7 +3373,8 @@ TEST_F(FileStoreCommitImplTest, TestCommitWithIndexFiles) { std::vector 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 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 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 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 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 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 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 7e66269f6..89c137f30 100644 --- a/src/paimon/core/operation/file_store_scan.cpp +++ b/src/paimon/core/operation/file_store_scan.cpp @@ -325,7 +325,7 @@ Status FileStoreScan::ReadFileEntries(const std::vector& manif PAIMON_RETURN_NOT_OK(ReadManifestFileMeta(meta, &tmp_entries)); } else { PAIMON_RETURN_NOT_OK(manifest_file_->Read(meta.FileName(), /*filter=*/nullptr, - &tmp_entries, meta.FileSize())); + meta.FileSize(), &tmp_entries)); } return tmp_entries; }; @@ -455,7 +455,7 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries( auto read_meta_task = [this, meta, bucket]() -> Result> { std::vector bucket_entries; PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries( - meta.FileName(), bucket, &bucket_entries, meta.FileSize())); + meta.FileName(), bucket, meta.FileSize(), &bucket_entries)); return bucket_entries; }; futures.push_back(Via(executor_.get(), read_meta_task)); @@ -576,7 +576,7 @@ Status FileStoreScan::ReadManifestFileMeta(const ManifestFileMeta& manifest, PAIMON_RETURN_NOT_OK(manifest_file_->Read( manifest.FileName(), [this](const ManifestEntry& entry) -> Result { return FilterManifestEntry(entry); }, - &unfiltered_entries, manifest.FileSize())); + 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 aa308e096..f268e5cda 100644 --- a/src/paimon/core/operation/manifest_file_merger.cpp +++ b/src/paimon/core/operation/manifest_file_merger.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include @@ -175,7 +176,8 @@ Result> ManifestFileMerger::MergeEntries( } std::vector 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 result; PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries(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 89a78a5a2..af0830aad 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 #include #include +#include #include #include #include @@ -158,7 +159,8 @@ class ManifestFileMergerTest : public testing::Test { const std::vector>& entry_file_name_expected) { std::vector 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> 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 f42708c89..0af493ae8 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> OrphanFilesCleanerImpl::GetUsedFilesBySnapshot( } std::vector 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 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& extra_file : diff --git a/src/paimon/core/snapshot_file_scan.cpp b/src/paimon/core/snapshot_file_scan.cpp index 86e2f4178..9d80ec860 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 entries; if (bucket_id_) { PAIMON_RETURN_NOT_OK(manifest_file_->ReadBucketEntries( - meta.FileName(), bucket_id_.value(), &entries, meta.FileSize())); + meta.FileName(), bucket_id_.value(), meta.FileSize(), &entries)); } else { PAIMON_RETURN_NOT_OK(manifest_file_->Read(meta.FileName(), /*filter=*/nullptr, - &entries, meta.FileSize())); + meta.FileSize(), &entries)); } PAIMON_RETURN_NOT_OK(ApplyPartitionFilter(&entries)); return ManifestReadResult{meta, std::move(entries)}; diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 68ec9ac05..bd484edc2 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 #include #include +#include #include #include #include @@ -230,8 +231,8 @@ Result AggregateFileStats(const std::shared_ptr std::vector 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 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 093994317..5279e6871 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> ReadLatestManifestEntries( CreateManifestFile(context, path_factory, core_options, pool)); std::vector 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/objects_file.h b/src/paimon/core/utils/objects_file.h index 2b9011af1..845efd41f 100644 --- a/src/paimon/core/utils/objects_file.h +++ b/src/paimon/core/utils/objects_file.h @@ -65,14 +65,14 @@ 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. Leave it - /// unset when the length is not known; the read then discovers it itself. + /// 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(const T&)>& filter, - std::vector* result, std::optional file_size = std::nullopt) const; + std::optional file_size, std::vector* result) const; Status ReadIfFileExist(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result, - std::optional file_size = std::nullopt) const; + std::optional file_size, std::vector* result) const; void DeleteQuietly(const std::string& file_name) { std::string path = path_factory_->ToPath(file_name); @@ -95,7 +95,7 @@ class ObjectsFile { Status ReadArrowBatches( const std::string& file_name, const std::function&)>& consumer, - std::optional file_size = std::nullopt) const; + std::optional file_size) const; std::shared_ptr path_factory_; std::shared_ptr pool_; @@ -143,12 +143,12 @@ ObjectsFile::ObjectsFile(const std::shared_ptr& file_system, template Status ObjectsFile::ReadIfFileExist(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result, - std::optional file_size) const { + std::optional file_size, + std::vector* 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, file_size); + return Read(file_name, filter, file_size, result); } return Status::OK(); } @@ -156,7 +156,7 @@ Status ObjectsFile::ReadIfFileExist(const std::string& file_name, template Status ObjectsFile::Read(const std::string& file_name, const std::function(const T&)>& filter, - std::vector* result, std::optional file_size) const { + std::optional file_size, std::vector* result) const { return ReadArrowBatches( file_name, [this, &filter, result](const std::shared_ptr& struct_array) -> Status { From 3ef9d1fdcacc4e64142abe03fc304a6941e732ac Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Mon, 14 Sep 2026 21:35:57 +0800 Subject: [PATCH 3/5] fix --- src/paimon/core/snapshot_file_scan_test.cpp | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/paimon/core/snapshot_file_scan_test.cpp b/src/paimon/core/snapshot_file_scan_test.cpp index fbc680919..e6629fbde 100644 --- a/src/paimon/core/snapshot_file_scan_test.cpp +++ b/src/paimon/core/snapshot_file_scan_test.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -73,15 +74,24 @@ std::set ExpectedFiles(const std::string& table_path, class OpenRecordingFileSystem : public LocalFileSystem { public: Result> Open(const std::string& path) const override { - opens.emplace_back(path, std::nullopt); + { + std::lock_guard lock(opens_mutex); + opens.emplace_back(path, std::nullopt); + } return LocalFileSystem::Open(path); } Result> Open(const FileStatus& file_status) const override { - opens.emplace_back(file_status.GetPath(), file_status.GetLen()); + { + std::lock_guard 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>> opens; }; From ef8741f9dde2a9f5032aec0b9574e6296d39c2f9 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Mon, 14 Sep 2026 21:55:26 +0800 Subject: [PATCH 4/5] fix --- src/paimon/core/utils/file_utils_test.cpp | 48 ------------------- .../fs/jindo/jindo_file_system_test.cpp | 16 +++++++ 2 files changed, 16 insertions(+), 48 deletions(-) diff --git a/src/paimon/core/utils/file_utils_test.cpp b/src/paimon/core/utils/file_utils_test.cpp index 3cb31b33f..ef3db4d6d 100644 --- a/src/paimon/core/utils/file_utils_test.cpp +++ b/src/paimon/core/utils/file_utils_test.cpp @@ -27,30 +27,6 @@ namespace paimon::test { -namespace { - -/// Counts what a listing asks of the file system, so the round trips it saves stay saved. On a -/// remote store an `Exists()` in front of a `ListDir()` is a trip of its own, and the listing -/// already answers the same question: a directory that is not there comes back empty. -class ListingCountingFileSystem : public LocalFileSystem { - public: - Result Exists(const std::string& path) const override { - ++exists_count; - return LocalFileSystem::Exists(path); - } - - Status ListDir(const std::string& directory, - std::vector* file_status_list) const override { - ++list_dir_count; - return LocalFileSystem::ListDir(directory, file_status_list); - } - - mutable int exists_count = 0; - mutable int list_dir_count = 0; -}; - -} // namespace - TEST(FileUtilsTest, TestSimple) { std::string test_data_path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09/snapshot/"; @@ -77,28 +53,4 @@ TEST(FileUtilsTest, TestNotNumber) { ASSERT_NOK(FileUtils::ListVersionedFiles(std::move(fs), test_data_path, "manifest-", &files)); } -TEST(FileUtilsTest, TestListVersionedFilesListsWithoutProbingExistence) { - std::string test_data_path = - paimon::test::GetDataDir() + "/orc/append_09.db/append_09/snapshot/"; - auto fs = std::make_shared(); - std::vector files; - ASSERT_OK(FileUtils::ListVersionedFiles(fs, test_data_path, "snapshot-", &files)); - ASSERT_EQ(files.size(), 5u); - ASSERT_EQ(fs->list_dir_count, 1); - ASSERT_EQ(fs->exists_count, 0); -} - -// The missing directory is the case the probe used to be justified by, so it is the one that has -// to reach the same answer without it. -TEST(FileUtilsTest, TestListVersionedFilesMissingDirListsWithoutProbingExistence) { - std::string test_data_path = - paimon::test::GetDataDir() + "/orc/append_09.db/append_09/not_exist/"; - auto fs = std::make_shared(); - std::vector files; - ASSERT_OK(FileUtils::ListVersionedFiles(fs, test_data_path, "snapshot-", &files)); - ASSERT_EQ(files.size(), 0u); - ASSERT_EQ(fs->list_dir_count, 1); - ASSERT_EQ(fs->exists_count, 0); -} - } // namespace paimon::test diff --git a/src/paimon/fs/jindo/jindo_file_system_test.cpp b/src/paimon/fs/jindo/jindo_file_system_test.cpp index e8b26dc6c..714d14157 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 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 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"; From e84a81da50ae2467191aecd82c5e006fd3677a74 Mon Sep 17 00:00:00 2001 From: "yonghao.fyh" Date: Tue, 15 Sep 2026 07:12:34 +0800 Subject: [PATCH 5/5] fix --- src/paimon/core/operation/expire_snapshots.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/paimon/core/operation/expire_snapshots.cpp b/src/paimon/core/operation/expire_snapshots.cpp index e1c05c501..9915efd46 100644 --- a/src/paimon/core/operation/expire_snapshots.cpp +++ b/src/paimon/core/operation/expire_snapshots.cpp @@ -410,7 +410,8 @@ Status ExpireSnapshots::GetDataFileSkippingSet( LinkedHashMap boundary_entries; for (const ManifestFileMeta& manifest : manifests) { std::vector 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 {