fix(planning): drain accepted manifest tasks on planning failure

docs/source/user_guide/read.rst

@@ -41,6 +41,28 @@ cross-language task scheduling and interaction (e.g., Java and C++), substantial
 across the two language ecosystems.
 
 
+Planning with a custom memory pool
+----------------------------------
+
+Supply an application-owned pool through ``ScanContextBuilder::WithMemoryPool``.
+In a function returning ``Status`` or ``Result``, use the ordinary scan API:
+
+.. code-block:: cpp
+
+    ScanContextBuilder builder(table_path);
+    builder.WithMemoryPool(pool);
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ScanContext> context, builder.Finish());
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<TableScan> scan, TableScan::Create(std::move(context)));
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> plan, scan->CreatePlan());
+
+* The pool controls only allocations routed through it, not all planning memory
+  or process RSS.
+* Handle both exceptions such as ``std::bad_alloc`` and failed ``Result`` values;
+  a failure is not an empty or partial successful plan.
+* Keep the pool alive until all objects using its allocations have been released,
+  including retained plans, splits, cache entries and executor task closures.
+  Use a thread-safe pool or synchronize its operations when they can overlap.
+
 Schema Evolution
 -----------------------
 Scope and Compatibility

include/paimon/executor.h

@@ -54,6 +54,7 @@ class PAIMON_EXPORT Executor {
     ///
     /// @note This method should be thread-safe and can be called from multiple threads
     /// simultaneously.
+    /// If submission throws, the task must not have been accepted or run.
     virtual void Add(std::function<void()> func) = 0;
 
     /// Shutdown the executor immediately, discarding all pending tasks.

include/paimon/scan_context.h

@@ -225,6 +225,9 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// @param memory_pool The memory pool to use.
     /// @return Reference to this builder for method chaining.
     /// @note if not set, memory_pool is default pool
+    /// A custom pool controls only allocations routed through it, not all planning memory.
+    /// Keep it alive while any plan, split, cache entry, or executor task can retain its
+    /// allocations; destroying the scan alone does not release all such objects.
     ScanContextBuilder& WithMemoryPool(const std::shared_ptr<MemoryPool>& memory_pool);
     /// Set custom executor for task execution.
     /// @param executor The executor to use.

src/paimon/CMakeLists.txt

@@ -1040,6 +1040,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/source/snapshot/snapshot_reader_test.cpp
                     core/table/source/startup_mode_test.cpp
                     core/table/source/table_scan_test.cpp
+                    core/table/source/scan_exception_safety_test.cpp
                     core/table/system/system_table_test.cpp
                     core/tag/tag_test.cpp
                     core/utils/blob_view_lookup_test.cpp

src/paimon/core/operation/file_store_scan.cpp

@@ -37,6 +37,7 @@
 #include "paimon/common/predicate/literal_converter.h"
 #include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/field_type_utils.h"
+#include "paimon/common/utils/scope_guard.h"
 #include "paimon/core/io/data_file_meta.h"
 #include "paimon/core/io/file_index_evaluator.h"
 #include "paimon/core/io/file_index_options.h"
@@ -323,6 +324,13 @@ 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();
+                    }
+                }
+            });
             futures.reserve(2);
             futures.push_back(
                 Via(executor_.get(), [read_list]() { return read_list(/*base=*/true); }));
@@ -352,6 +360,14 @@ Status FileStoreScan::ReadFileEntries(const std::vector<ManifestFileMeta>& manif
                                       bool apply_scan_filter,
                                       std::vector<ManifestEntry>* manifest_entries) const {
     std::vector<std::future<Result<std::vector<ManifestEntry>>>> futures;
+    ScopeGuard drain([&futures] {
+        for (auto& future : futures) {
+            if (future.valid()) {
+                future.wait();
+            }
+        }
+    });
+    futures.reserve(manifest_metas.size());
     for (const auto& meta : manifest_metas) {
         auto read_meta_task = [this, meta,
                                apply_scan_filter]() -> Result<std::vector<ManifestEntry>> {
@@ -511,6 +527,13 @@ Status FileStoreScan::ReadAndMergeBucketFileEntries(
     const bool inferred_bucket = !bucket_filter_ && bucket_selector_ != nullptr;
     if (core_options_.ScanManifestEntryLazyDecodeEnabled()) {
         std::vector<std::future<Result<std::vector<ManifestEntry>>>> futures;
+        ScopeGuard drain([&futures] {
+            for (auto& future : futures) {
+                if (future.valid()) {
+                    future.wait();
+                }
+            }
+        });
         futures.reserve(manifest_metas.size());
         for (const auto& meta : manifest_metas) {
             auto read_meta_task = [this, meta, bucket,