feat(planning): add finite-byte admission for retained snapshots

docs/source/user_guide.rst

@@ -39,6 +39,7 @@ User Guide
    user_guide/commit
    user_guide/compaction
    user_guide/read
+   user_guide/bounded_planning
    user_guide/metrics
    user_guide/clean
    user_guide/prefetch

docs/source/user_guide/bounded_planning.rst

@@ -0,0 +1,158 @@
+.. 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.
+
+Bounded metadata planning
+=========================
+
+``TableScan::CreateWithPlanningBudget`` opts into synchronous, finite-byte metadata
+planning. ``TableScan::Create`` retains its existing behavior. This API admits
+allocations before requesting storage, including allocations outside ``MemoryPool``.
+It does not plan first and compare a memory estimate afterwards.
+
+Budget contract
+---------------
+
+A ``PlanningBudget`` holds a cumulative requested-byte limit. Every admitted
+allocation consumes bytes permanently; freeing storage does not replenish the
+budget. Consequently temporary input, parser, decompression and output storage
+cannot each reuse the whole limit. This conservative policy bounds their combined
+peak and the total admitted allocation work; it can reject a query whose peak live
+storage would fit. It measures requested storage, including allocator bookkeeping
+added by this planner, rather than process RSS, malloc metadata, alignment padding,
+stack storage or filesystem page cache.
+
+The budget covers retained caller-provided path/schema/options, scan and plan
+objects, metadata file buffers, RapidJSON DOM and parser stacks, Avro schema lookup
+and decoded records, Zstandard stream workspace and output, duplicate detection,
+file metadata, split vectors, strings and shared ownership control blocks.
+Zstandard frames with unknown decoded length grow output only after admission.
+Compressed and decoded storage coexist within the same budget.
+
+Create the budget once per query. Scans may share it; atomic admission never exceeds
+the shared limit. Admission does not serialize upstream ``Malloc``/``Free`` calls.
+Overlapping scans or destruction of escaped plans/splits require a thread-safe
+upstream, such as ``GetDefaultPool()``, or caller synchronization covering both
+operations and cleanup. Individual scan objects must not be called concurrently.
+The first allocation failure makes the budget terminal.
+Exhaustion returns ``CapacityError``; upstream allocation failure returns
+``OutOfMemory``. Malformed inputs return an error and unsupported profiles return
+``NotImplemented``. No failure returns a partial plan or invokes the unrestricted
+planner. A failed scan returns its original failure on subsequent calls; a
+successful batch scan produces one plan and then returns end of scan.
+
+Callers create and validate the ``ScanContext`` and retained schema before invoking
+this API. Their construction, the fixed budget control object, fixed error Status
+messages, and process-owned immutable empty row/statistics constants are outside
+the variable planning payload. Existing shared executor, allocator and cache
+services supplied through the context are also outside it; this planner schedules
+no work on them and reads/writes no shared metadata cache. Creating or growing
+those services is not covered. Data reading, metrics retrieval, split serialization
+and user copies made after planning require their own allocation policy.
+
+Successful plans and separately retained nonempty splits retain the budget and its
+upstream allocator through file-metadata shared owners. Destroying the scan or plan
+therefore cannot invalidate an escaped split. Failed planning releases all temporary
+storage, including JSON children whose parent allocation failed. Freed storage
+still counts in ``BytesAdmitted()``.
+
+Admitted profile
+----------------
+
+The first implementation supports Linux with libstdc++ 9 or later, its C++11
+string ABI, and statically linked Zstandard with matching headers and static
+workspace APIs (including the default bundled Zstandard). Other builds return
+``NotImplemented`` from the bounded factory and retain ordinary scan support.
+It supports local filesystem tables with:
+
+* An explicitly retained schema and positive numeric ``scan.snapshot-id``.
+* Unpartitioned append-only data, ``bucket=-1``, main branch and matching schema ID.
+* Avro OCF manifest lists and version-2 ADD-only entries, null or Zstandard codec.
+* Parquet or ORC data files without external paths, extra files, embedded indexes,
+  deletion vectors or data-evolution metadata.
+* No streaming, planning predicates, partition/bucket filters, limit, realtime
+  context, catalog/format table or global-index result.
+
+DELETE entries, duplicate ADD identities, different file schemas, unsupported scan
+selectors and unsupported codecs fail explicitly. Missing/expired snapshots fail.
+Snapshot and schema IDs must match the bound request. Metadata names cannot escape
+one filesystem component. Inputs must remain available and immutable throughout
+planning, as with a retained snapshot read; this API does not pin snapshot retention
+or provide an atomic filesystem snapshot.
+
+Planning emits one ordinary native split per file and omits pruning statistics.
+The normal native data reader consumes these splits. This profile deliberately
+avoids generic Avro/Arrow decoding, schema discovery, asynchronous planning and
+shared caches, because those paths do not route all variable allocations through
+an admission allocator. Enabling this opt-in control can change planning latency,
+I/O and split grouping; it is not a performance-equivalent switch for the cached
+unrestricted planner. It does not change normal scans.
+
+Allocation qualification
+------------------------
+
+Parser and decompressor buffers use a PMR resource backed by the supplied budget.
+Shared metadata and plan blocks use ``allocate_shared`` with that allocator.
+Ordinary SDK publication constructors still require their existing STL types.
+Before invoking them the planner reserves GNU C++11 ABI string forward-range/copy
+storage (at most length + 1), exact-size vector element storage, the split object
+and its GNU shared-pointer control block. Small-string optimization may cause
+conservative over-admission. No approximate per-row multiplier is used.
+
+This qualification relies on the GNU implementations of ``basic_string::_M_construct``,
+``vector(size_type)`` and ``shared_ptr(Y*)``, not on a portable standard guarantee
+about capacities. The Linux static test target links libstdc++ statically so that its out-of-line
+string constructors are included in the ``operator new`` hook. A long-string
+calibration proves observation and failure injection. The test observes upstream
+pool requests during the complete public factory and planning call, and consumes
+all retained-input admission before observing publication allocations so those
+credits cannot mask a missing reservation. A new standard-library implementation or ABI needs requalification
+before expanding this support gate.
+
+Usage
+-----
+
+.. code-block:: cpp
+
+   #include "paimon/memory/planning_budget.h"
+   #include "paimon/scan_context.h"
+   #include "paimon/table/source/table_scan.h"
+
+   // schema_json is retained, validated table schema JSON bound by the caller.
+   auto pool = paimon::GetDefaultPool();
+   PAIMON_ASSIGN_OR_RAISE(auto budget,
+                          paimon::PlanningBudget::Create(32 * 1024 * 1024, pool));
+   paimon::ScanContextBuilder builder(local_table_path);
+   builder.SetTableSchema(schema_json)
+       .WithMemoryPool(pool)
+       .AddOption("scan.snapshot-id", "1");
+   PAIMON_ASSIGN_OR_RAISE(auto context, builder.Finish());
+   PAIMON_ASSIGN_OR_RAISE(auto scan,
+                          paimon::TableScan::CreateWithPlanningBudget(
+                              std::move(context), budget));
+   PAIMON_ASSIGN_OR_RAISE(auto plan, scan->CreatePlan());
+   // Feed plan->Splits() to the normal TableRead. Data-reader bytes are separate.
+
+Tests
+-----
+
+``paimon-bounded-table-scan-test`` exercises real Parquet full-field and ORC
+``id``-projection plan/read parity (existing ORC TIME decoding is outside this change),
+exact and insufficient quotas, shared concurrent admission, allocation failures at
+every arena and publication allocation, escaped-split ownership, malformed metadata,
+wrong writer field types, duplicate fields/files and unsupported profile rejection.
+The GNU static target additionally observes actual allocation requests. The existing
+core tests exercise the unchanged ordinary scan path.

include/paimon/memory/planning_budget.h

@@ -0,0 +1,60 @@
+/*
+ * 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 <cstdint>
+#include <memory>
+
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class PlanningBudgetAccess;
+
+/// Query-owned cumulative allocation admission for bounded metadata planning.
+/// Frees do not replenish admission: this deliberately also bounds temporary peaks.
+/// The limit covers requested storage, not allocator overhead or process RSS.
+/// Admission is atomic, but upstream Malloc/Free calls are not serialized. Overlapping
+/// scans or destruction of escaped plans/splits require a thread-safe upstream (such
+/// as GetDefaultPool()), or caller synchronization covering operations and cleanup.
+class PAIMON_EXPORT PlanningBudget {
+ public:
+    static Result<std::shared_ptr<PlanningBudget>> Create(
+        uint64_t byte_limit, const std::shared_ptr<MemoryPool>& upstream = GetDefaultPool());
+
+    uint64_t ByteLimit() const {
+        return limit_;
+    }
+    uint64_t BytesAdmitted() const {
+        return admitted_.load();
+    }
+
+ private:
+    friend class PlanningBudgetAccess;
+    PlanningBudget(uint64_t limit, std::shared_ptr<MemoryPool> upstream);
+    const uint64_t limit_;
+    std::shared_ptr<MemoryPool> upstream_;
+    std::atomic<uint64_t> admitted_{0};
+    // The first allocation failure is terminal across scans sharing this budget.
+    std::atomic<uint8_t> failure_{0};
+};
+}  // namespace paimon

include/paimon/table/source/table_scan.h

@@ -32,6 +32,7 @@
 
 namespace paimon {
 class ScanContext;
+class PlanningBudget;
 
 /// A scanner interface for reading table's meta and create a plan.
 class PAIMON_EXPORT TableScan {
@@ -42,6 +43,13 @@ class PAIMON_EXPORT TableScan {
     /// @return A Result containing a unique pointer to the `TableScan` instance.
     static Result<std::unique_ptr<TableScan>> Create(std::unique_ptr<ScanContext> context);
 
+    /// Create a synchronous scan with pre-allocation planning-byte admission.
+    /// See docs/source/user_guide/bounded_planning.rst for the admitted profile,
+    /// cumulative-byte contract, allocator qualification and ownership boundaries.
+    /// Unsupported inputs return an error without invoking the unrestricted planner.
+    static Result<std::unique_ptr<TableScan>> CreateWithPlanningBudget(
+        std::unique_ptr<ScanContext> context, const std::shared_ptr<PlanningBudget>& budget);
+
     virtual ~TableScan() = default;
 
     /// Create a scan plan.

src/paimon/CMakeLists.txt

@@ -110,6 +110,7 @@ set(PAIMON_COMMON_SRCS
     common/lookup/lookup_store_factory.cpp
     common/memory/bytes.cpp
     common/memory/memory_pool.cpp
+    common/memory/planning_budget.cpp
     common/memory/memory_segment.cpp
     common/memory/memory_segment_utils.cpp
     common/memory/memory_slice.cpp
@@ -481,6 +482,7 @@ set(PAIMON_CORE_SRCS
     core/table/source/startup_mode.cpp
     core/table/source/table_read.cpp
     core/table/source/table_scan.cpp
+    core/table/source/bounded_table_scan.cpp
     core/table/source/data_evolution_batch_scan.cpp
     core/table/source/data_evolution_full_text_read.cpp
     core/table/source/data_evolution_full_text_scan.cpp
@@ -529,6 +531,46 @@ if(PAIMON_ENABLE_REST)
          rest/rest_util.cpp)
 endif()
 
+# The bounded decoder uses Zstd's static-only workspace API. Ordinary scans also
+# support shared Zstd, so disable only the bounded entry point for that build.
+set(PAIMON_BOUNDED_PLANNING_ZSTD OFF)
+set(_bounded_zstd_target zstd)
+get_target_property(_bounded_zstd_type ${_bounded_zstd_target} TYPE)
+if(_bounded_zstd_type STREQUAL "INTERFACE_LIBRARY")
+    get_target_property(_bounded_zstd_links zstd INTERFACE_LINK_LIBRARIES)
+    list(LENGTH _bounded_zstd_links _bounded_zstd_link_count)
+    if(_bounded_zstd_link_count EQUAL 1 AND TARGET "${_bounded_zstd_links}")
+        set(_bounded_zstd_target ${_bounded_zstd_links})
+        get_target_property(_bounded_zstd_type ${_bounded_zstd_target} TYPE)
+    endif()
+endif()
+get_target_property(_bounded_zstd_location ${_bounded_zstd_target} IMPORTED_LOCATION)
+if(_bounded_zstd_type STREQUAL "STATIC_LIBRARY"
+   OR (_bounded_zstd_type STREQUAL "UNKNOWN_LIBRARY" AND _bounded_zstd_location MATCHES
+                                                         "\\.a$"))
+    if(TARGET zstd_ep)
+        # The pinned bundled static library is built after configuration.
+        set(PAIMON_BOUNDED_PLANNING_ZSTD ON)
+    else()
+        include(CMakePushCheckState)
+        include(CheckCXXSourceCompiles)
+        cmake_push_check_state(RESET)
+        set(CMAKE_REQUIRED_LIBRARIES zstd)
+        # Reconfiguration may select a different system library in the same build.
+        unset(PAIMON_BOUNDED_ZSTD_STATIC_API CACHE)
+        check_cxx_source_compiles("#define ZSTD_STATIC_LINKING_ONLY
+#include <zstd.h>
+int main() {
+    auto volatile estimate = &ZSTD_estimateDStreamSize_fromFrame;
+    auto volatile initialize = &ZSTD_initStaticDStream;
+    return !estimate || !initialize;
+}"
+                                  PAIMON_BOUNDED_ZSTD_STATIC_API)
+        cmake_pop_check_state()
+        set(PAIMON_BOUNDED_PLANNING_ZSTD ${PAIMON_BOUNDED_ZSTD_STATIC_API})
+    endif()
+endif()
+
 add_paimon_lib(paimon
                SOURCES
                ${PAIMON_COMMON_SRCS}
@@ -538,6 +580,7 @@ add_paimon_lib(paimon
                tbb
                glog
                fmt
+               zstd
                roaring_bitmap
                xxhash
                Threads::Threads
@@ -550,6 +593,7 @@ add_paimon_lib(paimon
                tbb
                glog
                fmt
+               zstd
                roaring_bitmap
                xxhash
                Threads::Threads
@@ -559,6 +603,11 @@ add_paimon_lib(paimon
                ${PAIMON_REST_LINK_LIBS}
                SHARED_LINK_FLAGS
                ${PAIMON_VERSION_SCRIPT_FLAGS})
+set_property(SOURCE core/table/source/bounded_table_scan.cpp
+             APPEND
+             PROPERTY COMPILE_DEFINITIONS
+                      PAIMON_BOUNDED_PLANNING_ZSTD=$<BOOL:${PAIMON_BOUNDED_PLANNING_ZSTD}>
+)
 
 add_subdirectory(common/file_index)
 add_subdirectory(common/global_index)
@@ -579,6 +628,44 @@ if(PAIMON_BUILD_TESTS)
         set_tests_properties(paimon-executor-submission-test PROPERTIES TIMEOUT 30)
     endif()
 
+    set(PAIMON_BOUNDED_TEST_LIB paimon_shared)
+    if(PAIMON_BUILD_STATIC)
+        set(PAIMON_BOUNDED_TEST_LIB paimon_static)
+    endif()
+    add_paimon_test(bounded_table_scan_test
+                    SOURCES
+                    core/table/source/bounded_table_scan_test.cpp
+                    STATIC_LINK_LIBS
+                    ${PAIMON_BOUNDED_TEST_LIB}
+                    test_utils_static
+                    zstd
+                    ${PAIMON_LOCAL_FILE_SYSTEM_STATIC_LINK_LIBS}
+                    ${PAIMON_PARQUET_FILE_FORMAT_STATIC_LINK_LIBS}
+                    ${PAIMON_ORC_FILE_FORMAT_STATIC_LINK_LIBS}
+                    ${PAIMON_AVRO_FILE_FORMAT_STATIC_LINK_LIBS}
+                    ${GTEST_LINK_TOOLCHAIN})
+    target_compile_definitions(paimon-bounded-table-scan-test
+                               PRIVATE PAIMON_BOUNDED_PLANNING_ZSTD=$<BOOL:${PAIMON_BOUNDED_PLANNING_ZSTD}>
+    )
+    include(CheckCXXSourceCompiles)
+    check_cxx_source_compiles("#include <string>
+#if !defined(__GLIBCXX__) || !_GLIBCXX_USE_CXX11_ABI || _GLIBCXX_RELEASE < 9
+#error Unsupported allocation observation ABI
+#endif
+int main() { return 0; }"
+                              PAIMON_PLANNING_TEST_GNU_ABI)
+    if(PAIMON_BUILD_STATIC
+       AND CMAKE_SYSTEM_NAME STREQUAL "Linux"
+       AND CMAKE_SIZEOF_VOID_P EQUAL 8
+       AND PAIMON_PLANNING_TEST_GNU_ABI)
+        target_compile_definitions(paimon-bounded-table-scan-test
+                                   PRIVATE PAIMON_TEST_PLANNING_ALLOCATIONS)
+        # Include out-of-line basic_string constructors in the allocation hook.
+        target_link_options(paimon-bounded-table-scan-test PRIVATE "-static-libstdc++"
+                            "LINKER:--wrap=_Znwm")
+    endif()
+    set_tests_properties(paimon-bounded-table-scan-test PROPERTIES TIMEOUT 120)
+
     # The shared HTTP client is compiled only for the components that need libcurl,
     # so its test follows the same gate.
     set(PAIMON_COMMON_HTTP_CLIENT_TEST_SRCS)