feat(scan): support cooperative append-only planning cancellation

docs/source/api/scan.rst

@@ -47,6 +47,39 @@ Decimal literals are rescaled to the bucket field's type only when the conversio
 is exact. NaN literals and decimals that cannot be represented exactly disable
 inferred bucket pruning.
 
+Cooperative batch planning cancellation
+=======================================
+
+Attach a query-owned ``std::shared_ptr<paimon::ScanCancellationToken>`` using
+``ScanContextBuilder::WithCancellationToken``. Another thread may call ``Cancel()``
+while the single planning caller runs ``TableScan::Create`` or ``CreatePlan()``.
+The scan retains the token after its context is consumed. Cancellation is one-way;
+use a new token for a new query. A recognized request returns the typed
+``Status::Cancelled`` rather than a partial or empty successful plan. An operation
+that already returned a native error retains that error. Hint failures remain best
+effort while a query is live; if the failed hint read also requests cancellation,
+that original error terminates planning instead of starting a retry.
+
+This opt-in path supports managed append-only tables with fixed positive or
+unaware buckets and Avro manifest metadata. Data files may use another supported
+format. Batch startup may be latest, latest-full, from-snapshot, or
+from-snapshot-full; the latter two require a numeric snapshot ID. Primary-key,
+data-evolution, deletion-vector, streaming, real-time, system-table, format-table,
+timestamp/tag-startup, supplied global-index results, and active embedded-index
+evaluation paths are refused rather than silently ignoring cancellation. Active
+embedded indexes are detected while reading matching manifest entries; disabling
+file-index reading avoids that evaluation path.
+
+Manifest work runs synchronously on the caller, without submitting to or shutting
+down its executor. Both shared manifest caches are bypassed for this query; existing
+cache contents and other queries are unchanged. Calls without a token keep their
+ordinary planning behavior. Cancellation does not interrupt blocking filesystem
+calls, one decoder operation, or CPU sorting. It provides neither a latency bound
+nor a planning-memory quota, and does not cancel subsequent data-file reading.
+Rebuild the library and paired native consumers together: the added context state
+and optional constructor argument preserve source usage, not binary ABI.
+Arrow import/export error paths also release acquired roots for calls without a token.
+
 Interface
 =========
 
@@ -62,6 +95,9 @@ Interface
    :members:
    :undoc-members:
 
+.. doxygenclass:: paimon::ScanCancellationToken
+   :members:
+
 .. doxygenclass:: paimon::Plan
    :members:
    :undoc-members:

include/paimon/scan_cancellation.h

@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <atomic>
+
+#include "paimon/status.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// A one-way, thread-safe cancellation request for one batch planning operation.
+/// Create a new token for a new query. Cancellation is observed cooperatively;
+/// it does not interrupt an in-progress filesystem or decoder call.
+class PAIMON_EXPORT ScanCancellationToken final {
+ public:
+    void Cancel() noexcept {
+        cancelled_.store(true, std::memory_order_relaxed);
+    }
+
+    bool IsCancelled() const noexcept {
+        return cancelled_.load(std::memory_order_relaxed);
+    }
+
+    Status CheckCancelled() const {
+        if (IsCancelled()) {
+            return Status::Cancelled("scan planning cancelled");
+        }
+        return Status::OK();
+    }
+
+ private:
+    std::atomic<bool> cancelled_{false};
+};
+
+}  // namespace paimon

include/paimon/scan_context.h

@@ -40,6 +40,7 @@ class Executor;
 class FormatTable;
 class MemoryPool;
 class Predicate;
+class ScanCancellationToken;
 
 /// `ScanContext` is some configuration for table scan operations.
 ///
@@ -58,7 +59,8 @@ class PAIMON_EXPORT ScanContext {
                 const std::optional<std::string>& table_schema,
                 const std::map<std::string, std::string>& options,
                 const std::shared_ptr<Cache>& cache,
-                const std::shared_ptr<FormatTable>& format_table);
+                const std::shared_ptr<FormatTable>& format_table,
+                const std::shared_ptr<ScanCancellationToken>& cancellation_token = nullptr);
 
     ~ScanContext();
 
@@ -88,6 +90,10 @@ class PAIMON_EXPORT ScanContext {
     std::shared_ptr<Executor> GetExecutor() const {
         return executor_;
     }
+
+    std::shared_ptr<ScanCancellationToken> GetCancellationToken() const {
+        return cancellation_token_;
+    }
     std::shared_ptr<GlobalIndexResult> GetGlobalIndexResult() const {
         return global_index_result_;
     }
@@ -129,6 +135,7 @@ class PAIMON_EXPORT ScanContext {
     std::map<std::string, std::string> options_;
     std::shared_ptr<Cache> cache_;
     std::shared_ptr<FormatTable> format_table_;
+    std::shared_ptr<ScanCancellationToken> cancellation_token_;
 };
 
 /// Filter configuration for table scan operations
@@ -212,6 +219,14 @@ class PAIMON_EXPORT ScanContextBuilder {
     /// @return Reference to this builder for method chaining.
     /// @note if not set, memory_pool is default pool
     ScanContextBuilder& WithMemoryPool(const std::shared_ptr<MemoryPool>& memory_pool);
+
+    /// Enable cooperative cancellation for managed append-only batch scans with Avro metadata.
+    /// Supports latest and numeric snapshot startup; unsupported modes are explicitly refused.
+    /// Manifest planning runs synchronously and bypasses shared manifest caches.
+    /// Blocking I/O, one decoder call and CPU sorting are not interrupted; no latency or memory
+    /// bound is implied. Cancel may run concurrently, but CreatePlan itself remains single-caller.
+    ScanContextBuilder& WithCancellationToken(
+        const std::shared_ptr<ScanCancellationToken>& cancellation_token);
     /// Set custom executor for task execution.
     /// @param executor The executor to use.
     /// @return Reference to this builder for method chaining.

src/paimon/common/utils/bin_packing.h

@@ -24,6 +24,8 @@
 #include <utility>
 #include <vector>
 
+#include "paimon/result.h"
+
 namespace paimon {
 /// Contains bin packing implementations.
 class BinPacking {
@@ -35,11 +37,28 @@ class BinPacking {
     static std::vector<std::vector<T>> PackForOrdered(
         std::vector<T>&& items, const std::function<int64_t(const T&)>& weight_func,
         int64_t target_weight) {
+        // With no checkpoint, packing cannot return a Status error.
+        return PackForOrdered<T>(std::move(items), weight_func, target_weight, nullptr).value();
+    }
+
+    template <typename T>
+    static Result<std::vector<std::vector<T>>> PackForOrdered(
+        std::vector<T>&& items, const std::function<int64_t(const T&)>& weight_func,
+        int64_t target_weight, const std::function<Status()>& checkpoint) {
+        if (checkpoint) {
+            PAIMON_RETURN_NOT_OK(checkpoint());
+        }
         std::vector<std::vector<T>> packed;
         std::vector<T> bin_items;
         int64_t bin_weight = 0;
         for (auto& item : items) {
+            if (checkpoint) {
+                PAIMON_RETURN_NOT_OK(checkpoint());
+            }
             int64_t weight = weight_func(item);
+            if (checkpoint) {
+                PAIMON_RETURN_NOT_OK(checkpoint());
+            }
             // when get a much big item or total weight enough, we check the binItems size. If
             // greater than zero, we pack it
             if (bin_weight + weight > target_weight && bin_items.size() > 0) {
@@ -53,6 +72,9 @@ class BinPacking {
         if (bin_items.size() > 0) {
             packed.push_back(std::move(bin_items));
         }
+        if (checkpoint) {
+            PAIMON_RETURN_NOT_OK(checkpoint());
+        }
         return packed;
     }
 };

src/paimon/common/utils/bin_packing_test.cpp

@@ -27,6 +27,35 @@
 
 namespace paimon::test {
 
+TEST(BinPackingTest, CheckedPackingPreservesOrderAndPropagatesRefusal) {
+    const auto weight = [](const int32_t& value) -> int64_t { return value; };
+    const std::vector<std::vector<int32_t>> expected = {{2, 3}, {7}, {1}};
+    int32_t checks = 0;
+    auto result = BinPacking::PackForOrdered<int32_t>({2, 3, 7, 1}, weight, 5, [&checks]() {
+        ++checks;
+        return Status::OK();
+    });
+    ASSERT_TRUE(result.ok());
+    auto packed = std::move(result).value();
+    ASSERT_EQ(packed, expected);
+    ASSERT_EQ(checks, 10);
+    ASSERT_EQ(BinPacking::PackForOrdered<int32_t>({2, 3, 7, 1}, weight, 5), expected);
+    for (int32_t stop = 1; stop <= checks; ++stop) {
+        SCOPED_TRACE(stop);
+        int32_t visits = 0;
+        auto result = BinPacking::PackForOrdered<int32_t>({2, 3, 7, 1}, weight, 5, [&]() {
+            return ++visits == stop ? Status::Cancelled("packing cancelled") : Status::OK();
+        });
+        ASSERT_FALSE(result.ok());
+        ASSERT_TRUE(result.status().IsCancelled());
+        ASSERT_EQ(visits, stop);
+    }
+    auto empty = BinPacking::PackForOrdered<int32_t>(
+        {}, weight, 5, []() { return Status::Cancelled("empty packing cancelled"); });
+    ASSERT_FALSE(empty.ok());
+    ASSERT_TRUE(empty.status().IsCancelled());
+}
+
 TEST(BinPackingTest, TestExactlyPack) {
     std::map<std::string, int64_t> items_map = {{"a", 1000ll}, {"b", 20ll}, {"c", 200ll}};
     auto weight_func = [&](const std::string& item) -> int64_t { return items_map[item]; };

src/paimon/common/utils/scan_cancellation.h

@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include "paimon/scan_cancellation.h"
+
+namespace paimon {
+
+inline Status CheckScanCancellation(const ScanCancellationToken* token) {
+    return token ? token->CheckCancelled() : Status::OK();
+}
+
+}  // namespace paimon

src/paimon/core/append/append_compact_coordinator.cpp

@@ -256,7 +256,7 @@ Result<LinkedHashMap<BinaryRow, std::vector<std::shared_ptr<DataFileMeta>>>> Sca
     }
 
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan, scan->CreatePlan());
-    std::vector<ManifestEntry> add_entries = plan->Files(FileKind::Add());
+    PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestEntry> add_entries, plan->Files(FileKind::Add()));
 
     int64_t compaction_file_size = core_options.GetCompactionFileSize(/*has_primary_key=*/false);
     LinkedHashMap<BinaryRow, std::vector<std::shared_ptr<DataFileMeta>>> partition_files;

src/paimon/core/manifest/manifest_file.cpp

@@ -138,14 +138,16 @@ Status ManifestFile::ReadSelectedEntries(
     const std::string& file_name, std::optional<int64_t> file_size, const RowRangeIndex* row_ranges,
     const std::function<Result<bool>(const InternalRow&)>& row_filter,
     const std::function<Result<bool>(const ManifestEntry&)>& entry_filter,
-    std::vector<ManifestEntry>* entries) const {
+    std::vector<ManifestEntry>* entries, const ScanCancellationToken* token) const {
+    PAIMON_RETURN_NOT_OK(CheckScanCancellation(token));
     if (!row_ranges && !row_filter) {
-        return Read(file_name, entry_filter, file_size, entries);
+        return Read(file_name, entry_filter, file_size, entries, token);
     }
     return ReadArrowBatches(
         file_name, file_size,
-        [this, row_ranges, &row_filter, &entry_filter,
-         entries](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
+        [this, row_ranges, &row_filter, &entry_filter, entries,
+         token](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
+            PAIMON_RETURN_NOT_OK(CheckScanCancellation(token));
             // ManifestMetaReader has aligned both the entry and its nested file schema.
             // Only row-range scans need to probe the file columns before entry decoding.
             std::shared_ptr<arrow::StructArray> files;
@@ -168,6 +170,7 @@ Status ManifestFile::ReadSelectedEntries(
             }
             ColumnarRow row(batch->fields(), pool_, /*row_id=*/0);
             for (int64_t i = 0; i < batch->length(); ++i) {
+                PAIMON_RETURN_NOT_OK(CheckScanCancellation(token));
                 row.SetRowId(i);
                 PAIMON_RETURN_NOT_OK(
                     ManifestEntrySerializer::ValidateVersion(row.GetInt(kVersionFieldIndex)));
@@ -190,22 +193,25 @@ Status ManifestFile::ReadSelectedEntries(
                 }
                 if (row_filter) {
                     PAIMON_ASSIGN_OR_RAISE(bool keep, row_filter(row));
+                    PAIMON_RETURN_NOT_OK(CheckScanCancellation(token));
                     if (!keep) {
                         continue;
                     }
                 }
                 PAIMON_ASSIGN_OR_RAISE(ManifestEntry entry, serializer_->FromRow(row));
+                PAIMON_RETURN_NOT_OK(CheckScanCancellation(token));
                 if (entry_filter) {
                     PAIMON_ASSIGN_OR_RAISE(bool keep, entry_filter(entry));
+                    PAIMON_RETURN_NOT_OK(CheckScanCancellation(token));
                     if (!keep) {
                         continue;
                     }
                 }
                 entries->push_back(std::move(entry));
             }
-            return Status::OK();
+            return CheckScanCancellation(token);
         },
-        /*prepare_reader=*/nullptr);
+        /*prepare_reader=*/nullptr, token);
 }
 
 Status ManifestFile::PrepareBucketRead(int32_t bucket,

src/paimon/core/manifest/manifest_file.h

@@ -87,7 +87,7 @@ class ManifestFile : public ObjectsFile<ManifestEntry> {
         const RowRangeIndex* row_ranges,
         const std::function<Result<bool>(const InternalRow&)>& row_filter,
         const std::function<Result<bool>(const ManifestEntry&)>& entry_filter,
-        std::vector<ManifestEntry>* entries) const;
+        std::vector<ManifestEntry>* entries, const ScanCancellationToken* token = nullptr) const;
 
  private:
     Status PrepareBucketRead(int32_t bucket, const std::optional<int32_t>& expected_total_buckets,