lxy-9602 commented on code in PR #462:
URL: https://github.com/apache/paimon-cpp/pull/462#discussion_r4449542844
##########
src/paimon/core/operation/file_store_scan.cpp:
##########
@@ -323,6 +324,11 @@ Status FileStoreScan::ReadManifestsWithSnapshot(const
Snapshot& snapshot,
return metas;
};
std::vector<std::future<Result<std::vector<ManifestFileMeta>>>>
futures;
+ ScopeGuard drain([&futures] {
+ for (auto& future : futures) {
+ if (future.valid()) future.wait();
Review Comment:
For code style consistency, could you please add braces `{}` for the
standalone `if` block as well, similar to `{ future.wait(); }`.
##########
docs/source/user_guide/read.rst:
##########
@@ -41,6 +41,57 @@ cross-language task scheduling and interaction (e.g., Java
and C++), substantial
across the two language ecosystems.
+Planning with a custom memory pool
+----------------------------------
+
+Use ``ScanContextBuilder::WithMemoryPool(pool)`` with ordinary
+``TableScan::Create`` and ``CreatePlan`` to supply an application-owned pool.
+Planning continues to use the normal metadata decoders, filters and caches.
+A pool that refuses an allocation can throw ``std::bad_alloc``. The caller must
+handle both an exception from the planning call and a failed ``Result``:
adapters
+such as the Arrow memory pool can translate allocation refusal into an error.
+A failed call must not be treated as an empty or partial successful plan.
+
+The budget covers only requests routed through the supplied pool. STL
containers,
+metadata parser storage, shared services and other allocations may bypass it.
+It is not a limit on total planning memory, data-reader memory or process RSS.
+Unaccounted allocations are not guaranteed to be small merely because the pool
+is bounded. Applications requiring a stronger bound need a separate policy and
+validation of the uncovered allocations.
+
+The application chooses the accounting policy in its pool. A **live-byte**
limit
+charges outstanding requested bytes and returns credit on ``Free``. A growing
+``Realloc`` must reserve the increase before requesting storage; on failure it
+must preserve the original allocation and its charge. A **cumulative** limit
+instead charges allocation work without refunding frees. These are different
+contracts: freeing temporary buffers permits another allocation under a
live-byte
+limit, but not necessarily under a cumulative limit. Neither policy accounts
for
+allocations that do not reach that pool. ``MemoryPool::CurrentUsage`` describes
+live usage; the scan API does not install either budget policy for the caller.
+
+Keep a custom pool alive until all objects using its allocations have been
+released, including retained plans, separately retained splits, cache entries
+and executor task closures. A plan or split does not automatically retain a
+shared owner of every custom pool. It can outlive its scan when the caller
keeps
+the pool alive. Use a thread-safe pool, or synchronize all its operations
including
+cleanup, when planning tasks or consumers can run concurrently. A scan itself
+must not be called concurrently.
+
+Cache entries remain charged to the pool that allocated them until they are
+released. A shared cache can retain allocations from multiple pools; keep each
+such pool alive until all entries using its allocations have been released.
+
+When ordinary manifest planning fails, it waits for already accepted metadata
+tasks before unwinding the task submission scope. For this guarantee, if
+``Executor::Add`` throws, it must not have accepted or run the submitted task.
+The default executor satisfies this condition.
+
+Waiting for a future to become ready does not guarantee that its task closure
+has been destroyed. Keep the pool alive until task closures and their retained
+results have been released, including when a shared or custom executor outlives
+the scan.
+Waiting for tasks does not cancel native I/O.
+
Review Comment:
Could we simplify the documentation section? I think a short usage example
and a few key notes would be sufficient: the custom pool only controls
allocations routed through it, callers should handle both exceptions and failed
`Result`s, and the pool must outlive objects using its allocations.
The detailed discussion of accounting policies, reallocation semantics, and
internal task cleanup seems beyond the scope of the reading guide and may add
maintenance overhead.
##########
src/paimon/core/table/source/planning_memory_pool_test.cpp:
##########
@@ -0,0 +1,402 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <fstream>
+#include <future>
+#include <iterator>
+#include <limits>
+#include <memory>
+#include <mutex>
+#include <new>
+#include <optional>
+
+#include "arrow/memory_pool.h"
+#include "gtest/gtest.h"
+#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/executor.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/scan_context.h"
+#include "paimon/table/source/table_scan.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+class InlinePlanningExecutor final : public Executor {
+ public:
+ void Add(std::function<void()> task) override {
+ task();
+ }
+ void ShutdownNow() override {}
+ uint32_t GetThreadNum() const override {
+ return 1;
+ }
+};
+
+class PlanningFaultPool final : public MemoryPool {
+ public:
+ void* Malloc(uint64_t size, uint64_t alignment) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ Check();
+ ++malloc_calls;
+ return upstream_->Malloc(size, alignment);
+ }
+ void* Realloc(void* pointer, size_t old_size, size_t new_size, uint64_t
alignment) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ Check();
+ ++realloc_calls;
+ return upstream_->Realloc(pointer, old_size, new_size, alignment);
+ }
+ void Free(void* pointer, uint64_t size) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ upstream_->Free(pointer, size);
+ }
+ void Free(void* pointer, uint64_t size, uint64_t alignment) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ upstream_->Free(pointer, size, alignment);
+ }
+ uint64_t CurrentUsage() const override {
+ return upstream_->CurrentUsage();
+ }
+ uint64_t MaxMemoryUsage() const override {
+ return upstream_->MaxMemoryUsage();
+ }
+ std::atomic<uint64_t> calls{0};
+ std::atomic<uint64_t> fail_at{std::numeric_limits<uint64_t>::max()};
+ std::atomic<uint64_t> malloc_calls{0};
+ std::atomic<uint64_t> realloc_calls{0};
+ std::atomic<bool> rejected{false};
+
+ private:
+ void Check() {
+ if (calls.fetch_add(1) == fail_at.load()) {
+ rejected = true;
+ throw std::bad_alloc();
+ }
+ }
+ std::unique_ptr<MemoryPool> upstream_ = GetMemoryPool();
+ std::mutex mutex_;
+};
+
+std::string PlanningFixture() {
+ return GetDataDir() +
"/parquet/append_types_compatibility.db/rust_time_types";
+}
+
+struct PlanningInput {
+ std::string path = PlanningFixture();
+ std::string snapshot = "1";
+ std::shared_ptr<Cache> cache;
+ std::optional<int32_t> bucket;
+ std::shared_ptr<FileSystem> file_system;
+};
+
+Result<std::unique_ptr<TableScan>> OrdinaryScan(
+ const std::shared_ptr<PlanningFaultPool>& pool,
+ const std::shared_ptr<Executor>& executor =
std::make_shared<InlinePlanningExecutor>(),
+ const PlanningInput& input = {}) {
+ std::ifstream schema_file(input.path + "/schema/schema-0");
+ std::string schema{std::istreambuf_iterator<char>(schema_file),
+ std::istreambuf_iterator<char>()};
+ ScanContextBuilder builder(input.path);
+
builder.SetTableSchema(schema).WithMemoryPool(pool).WithExecutor(executor).AddOption(
+ "scan.snapshot-id", input.snapshot);
+ if (input.cache) {
+ builder.WithCache(input.cache)
+ .AddOption("scan.manifest-entry-cache.max-snapshots", "8")
+ .AddOption("scan.manifest-entry.lazy-decode.enabled", "true");
+ }
+ if (input.bucket) builder.SetBucketFilter(*input.bucket);
+ if (input.file_system) builder.WithFileSystem(input.file_system);
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> context,
builder.Finish());
+ return TableScan::Create(std::move(context));
+}
+
+class RejectingPlanningExecutor final : public Executor {
+ public:
+ explicit RejectingPlanningExecutor(uint32_t rejected_submission = 2,
+ uint32_t blocked_submission = 0)
+ : rejected_submission_(rejected_submission),
+ blocked_submission_(blocked_submission ? blocked_submission :
rejected_submission - 1) {}
+ ~RejectingPlanningExecutor() override {
+ Release(false);
+ worker_.reset();
+ }
+ void Add(std::function<void()> task) override {
+ uint32_t submission = ++submissions_;
+ if (submission == rejected_submission_) {
+ rejected_ = true;
+ throw std::bad_alloc();
+ }
+ if (submission != blocked_submission_) {
+ worker_->Add([this, submission, task = std::move(task)] {
+ task();
+ if (submission == 1) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ first_completed_ = true;
+ changed_.notify_all();
+ }
+ });
+ return;
+ }
+ worker_->Add([this, task = std::move(task)] {
+ bool execute;
+ {
+ std::unique_lock<std::mutex> lock(mutex_);
+ blocked_ = true;
+ changed_.notify_all();
+ changed_.wait(lock, [this] { return released_; });
+ execute = execute_;
+ }
+ if (execute) task();
+ });
+ }
+ void ShutdownNow() override {
+ worker_->ShutdownNow();
+ }
+ uint32_t GetThreadNum() const override {
+ return 2;
+ }
+ bool WaitUntilBlocked() {
+ std::unique_lock<std::mutex> lock(mutex_);
+ return changed_.wait_for(lock, std::chrono::seconds(5), [this] {
return blocked_; });
+ }
+ void Release(bool execute) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ execute_ = execute;
+ released_ = true;
+ changed_.notify_all();
+ }
+ bool Rejected() const {
+ return rejected_;
+ }
+ bool WaitUntilFirstCompleted() {
+ std::unique_lock<std::mutex> lock(mutex_);
+ return changed_.wait_for(lock, std::chrono::seconds(5),
+ [this] { return first_completed_; });
+ }
+
+ private:
+ std::unique_ptr<Executor> worker_ = CreateDefaultExecutor(2).value();
+ std::atomic<uint32_t> submissions_{0};
+ uint32_t rejected_submission_;
+ uint32_t blocked_submission_;
+ std::atomic<bool> rejected_{false};
+ std::mutex mutex_;
+ std::condition_variable changed_;
+ bool blocked_ = false;
+ bool released_ = false;
+ bool execute_ = false;
+ bool first_completed_ = false;
+};
+
+class PoolRefusingManifestFileSystem final : public LocalFileSystem {
+ public:
+ explicit PoolRefusingManifestFileSystem(std::shared_ptr<PlanningFaultPool>
pool)
+ : pool_(std::move(pool)) {}
+ using LocalFileSystem::Open;
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ if (path.find("/manifest/") != std::string::npos &&
!refused_.exchange(true)) {
+ pool_->fail_at = pool_->calls.load();
+ // A supplied file system may allocate through the caller's pool
too.
+ // Force refusal inside the real manifest worker, before opening a
file.
+ void* allocation = pool_->Malloc(1, 64);
+ pool_->Free(allocation, 1);
+ }
+ return LocalFileSystem::Open(path);
+ }
+
+ private:
+ std::shared_ptr<PlanningFaultPool> pool_;
+ mutable std::atomic<bool> refused_{false};
+};
+} // namespace
+
+TEST(OrdinaryPlanningPoolTest, RejectEachPlanningAllocationWithoutLeaking) {
+ auto baseline_pool = std::make_shared<PlanningFaultPool>();
Review Comment:
Could we rename this file and test suite to reflect that they primarily test
scan exception safety? The custom memory pool is a fault-injection mechanism
here, rather than the component under test.
##########
src/paimon/core/table/source/planning_memory_pool_test.cpp:
##########
@@ -0,0 +1,402 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <fstream>
+#include <future>
+#include <iterator>
+#include <limits>
+#include <memory>
+#include <mutex>
+#include <new>
+#include <optional>
+
+#include "arrow/memory_pool.h"
+#include "gtest/gtest.h"
+#include "paimon/common/io/cache/lru_cache.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/executor.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/scan_context.h"
+#include "paimon/table/source/table_scan.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+class InlinePlanningExecutor final : public Executor {
+ public:
+ void Add(std::function<void()> task) override {
+ task();
+ }
+ void ShutdownNow() override {}
+ uint32_t GetThreadNum() const override {
+ return 1;
+ }
+};
+
+class PlanningFaultPool final : public MemoryPool {
+ public:
+ void* Malloc(uint64_t size, uint64_t alignment) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ Check();
+ ++malloc_calls;
+ return upstream_->Malloc(size, alignment);
+ }
+ void* Realloc(void* pointer, size_t old_size, size_t new_size, uint64_t
alignment) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ Check();
+ ++realloc_calls;
+ return upstream_->Realloc(pointer, old_size, new_size, alignment);
+ }
+ void Free(void* pointer, uint64_t size) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ upstream_->Free(pointer, size);
+ }
+ void Free(void* pointer, uint64_t size, uint64_t alignment) override {
+ std::lock_guard<std::mutex> lock(mutex_);
+ upstream_->Free(pointer, size, alignment);
+ }
+ uint64_t CurrentUsage() const override {
+ return upstream_->CurrentUsage();
+ }
+ uint64_t MaxMemoryUsage() const override {
+ return upstream_->MaxMemoryUsage();
+ }
+ std::atomic<uint64_t> calls{0};
+ std::atomic<uint64_t> fail_at{std::numeric_limits<uint64_t>::max()};
+ std::atomic<uint64_t> malloc_calls{0};
+ std::atomic<uint64_t> realloc_calls{0};
+ std::atomic<bool> rejected{false};
+
+ private:
+ void Check() {
+ if (calls.fetch_add(1) == fail_at.load()) {
+ rejected = true;
+ throw std::bad_alloc();
+ }
+ }
+ std::unique_ptr<MemoryPool> upstream_ = GetMemoryPool();
+ std::mutex mutex_;
+};
+
+std::string PlanningFixture() {
+ return GetDataDir() +
"/parquet/append_types_compatibility.db/rust_time_types";
+}
+
+struct PlanningInput {
+ std::string path = PlanningFixture();
+ std::string snapshot = "1";
+ std::shared_ptr<Cache> cache;
+ std::optional<int32_t> bucket;
+ std::shared_ptr<FileSystem> file_system;
+};
+
+Result<std::unique_ptr<TableScan>> OrdinaryScan(
+ const std::shared_ptr<PlanningFaultPool>& pool,
+ const std::shared_ptr<Executor>& executor =
std::make_shared<InlinePlanningExecutor>(),
+ const PlanningInput& input = {}) {
+ std::ifstream schema_file(input.path + "/schema/schema-0");
+ std::string schema{std::istreambuf_iterator<char>(schema_file),
+ std::istreambuf_iterator<char>()};
+ ScanContextBuilder builder(input.path);
+
builder.SetTableSchema(schema).WithMemoryPool(pool).WithExecutor(executor).AddOption(
+ "scan.snapshot-id", input.snapshot);
+ if (input.cache) {
+ builder.WithCache(input.cache)
+ .AddOption("scan.manifest-entry-cache.max-snapshots", "8")
+ .AddOption("scan.manifest-entry.lazy-decode.enabled", "true");
+ }
+ if (input.bucket) builder.SetBucketFilter(*input.bucket);
+ if (input.file_system) builder.WithFileSystem(input.file_system);
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> context,
builder.Finish());
+ return TableScan::Create(std::move(context));
+}
+
+class RejectingPlanningExecutor final : public Executor {
+ public:
+ explicit RejectingPlanningExecutor(uint32_t rejected_submission = 2,
+ uint32_t blocked_submission = 0)
+ : rejected_submission_(rejected_submission),
+ blocked_submission_(blocked_submission ? blocked_submission :
rejected_submission - 1) {}
+ ~RejectingPlanningExecutor() override {
+ Release(false);
+ worker_.reset();
+ }
+ void Add(std::function<void()> task) override {
+ uint32_t submission = ++submissions_;
+ if (submission == rejected_submission_) {
+ rejected_ = true;
+ throw std::bad_alloc();
+ }
+ if (submission != blocked_submission_) {
+ worker_->Add([this, submission, task = std::move(task)] {
+ task();
+ if (submission == 1) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ first_completed_ = true;
+ changed_.notify_all();
+ }
+ });
+ return;
+ }
+ worker_->Add([this, task = std::move(task)] {
+ bool execute;
+ {
+ std::unique_lock<std::mutex> lock(mutex_);
+ blocked_ = true;
+ changed_.notify_all();
+ changed_.wait(lock, [this] { return released_; });
+ execute = execute_;
+ }
+ if (execute) task();
+ });
+ }
+ void ShutdownNow() override {
+ worker_->ShutdownNow();
+ }
+ uint32_t GetThreadNum() const override {
+ return 2;
+ }
+ bool WaitUntilBlocked() {
+ std::unique_lock<std::mutex> lock(mutex_);
+ return changed_.wait_for(lock, std::chrono::seconds(5), [this] {
return blocked_; });
+ }
+ void Release(bool execute) {
+ std::lock_guard<std::mutex> lock(mutex_);
+ execute_ = execute;
+ released_ = true;
+ changed_.notify_all();
+ }
+ bool Rejected() const {
+ return rejected_;
+ }
+ bool WaitUntilFirstCompleted() {
+ std::unique_lock<std::mutex> lock(mutex_);
+ return changed_.wait_for(lock, std::chrono::seconds(5),
+ [this] { return first_completed_; });
+ }
+
+ private:
+ std::unique_ptr<Executor> worker_ = CreateDefaultExecutor(2).value();
+ std::atomic<uint32_t> submissions_{0};
+ uint32_t rejected_submission_;
+ uint32_t blocked_submission_;
+ std::atomic<bool> rejected_{false};
+ std::mutex mutex_;
+ std::condition_variable changed_;
+ bool blocked_ = false;
+ bool released_ = false;
+ bool execute_ = false;
+ bool first_completed_ = false;
+};
+
+class PoolRefusingManifestFileSystem final : public LocalFileSystem {
+ public:
+ explicit PoolRefusingManifestFileSystem(std::shared_ptr<PlanningFaultPool>
pool)
+ : pool_(std::move(pool)) {}
+ using LocalFileSystem::Open;
+ Result<std::unique_ptr<InputStream>> Open(const std::string& path) const
override {
+ if (path.find("/manifest/") != std::string::npos &&
!refused_.exchange(true)) {
+ pool_->fail_at = pool_->calls.load();
+ // A supplied file system may allocate through the caller's pool
too.
+ // Force refusal inside the real manifest worker, before opening a
file.
+ void* allocation = pool_->Malloc(1, 64);
+ pool_->Free(allocation, 1);
+ }
+ return LocalFileSystem::Open(path);
+ }
+
+ private:
+ std::shared_ptr<PlanningFaultPool> pool_;
+ mutable std::atomic<bool> refused_{false};
+};
+} // namespace
+
+TEST(OrdinaryPlanningPoolTest, RejectEachPlanningAllocationWithoutLeaking) {
+ auto baseline_pool = std::make_shared<PlanningFaultPool>();
+ auto scan_result = OrdinaryScan(baseline_pool);
+ ASSERT_TRUE(scan_result.ok()) << scan_result.status().ToString();
+ auto baseline_scan = std::move(scan_result).value();
+ RecordProperty("factory_pool_calls", baseline_pool->calls.load());
+ baseline_pool->calls = 0;
+ baseline_pool->malloc_calls = 0;
+ baseline_pool->realloc_calls = 0;
+ auto baseline_result = baseline_scan->CreatePlan();
+ ASSERT_TRUE(baseline_result.ok()) << baseline_result.status().ToString();
+ auto baseline = std::move(baseline_result).value();
+ ASSERT_FALSE(baseline->Splits().empty());
+ const uint64_t allocation_count = baseline_pool->calls;
+ ASSERT_GT(allocation_count, 0);
+ RecordProperty("planning_malloc_calls",
baseline_pool->malloc_calls.load());
+ RecordProperty("planning_realloc_calls",
baseline_pool->realloc_calls.load());
+ baseline.reset();
+ baseline_scan.reset();
+ ASSERT_EQ(baseline_pool->CurrentUsage(), 0);
+
+ for (uint64_t ordinal = 0; ordinal < allocation_count; ++ordinal) {
+ SCOPED_TRACE(ordinal);
+ auto pool = std::make_shared<PlanningFaultPool>();
+ auto result = OrdinaryScan(pool);
+ ASSERT_TRUE(result.ok()) << result.status().ToString();
+ auto scan = std::move(result).value();
+ pool->calls = 0;
+ pool->fail_at = ordinal;
+ bool failed = false;
+ try {
+ auto plan = scan->CreatePlan();
+ failed = !plan.ok();
+ } catch (const std::bad_alloc&) {
+ failed = true;
+ }
+ ASSERT_TRUE(pool->rejected);
+ ASSERT_TRUE(failed) << "allocation refusal returned a successful plan";
+ scan.reset();
+ EXPECT_EQ(pool->CurrentUsage(), 0) << "pool storage survived failed
planning";
+ pool->fail_at = std::numeric_limits<uint64_t>::max();
+ {
+ auto recovery_result = OrdinaryScan(pool);
+ ASSERT_TRUE(recovery_result.ok()) <<
recovery_result.status().ToString();
Review Comment:
Could you please use the test macros from `testharness.h`, e.g.
`ASSERT_OK_AND_ASSIGN`, `ASSERT_OK`, and `ASSERT_NOK_WITH_MSG`, instead of
writing things like `ASSERT_TRUE(recovery_result.ok())`. Please adjust the
entire test accordingly, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
