From eecb60173ecbdf7187f42a9f67ec55c7ce5107e3 Mon Sep 17 00:00:00 2001
From: Arun Sharma <arun@sharma-home.net>
Date: Sat, 17 Oct 2026 12:16:33 -0700
Subject: [PATCH 1/6] perf(processor): remove per-execute mallocs on
 prepared-statement fast path

Three contained removals, measured with an LD_PRELOAD allocation profiler
on 120x cached-plan executions of LDBC Q1 (678 malloc+free pairs per
execute before, 573 after; Q1 hot exec p50 224us to 210us):

- PhysicalOperator::registerProfilingMetrics: bind thread-local dummy
  metrics when the profiler is disabled instead of registering into the
  profiler unordered_map (map nodes plus strings plus unique_ptr per
  operator per task, plus mutex traffic).
- RelTable::initScanState: reuse a scan-state positions buffer across
  per-morsel inits, skip the bound-ID sort when input is already ordered,
  and replace the stdmap node-group partition with a linear run scan
  over the sorted offsets.
- ResultSetPool: per-thread MRU cache of up to 8 descriptor slots instead
  of a single slot, which thrashed across pipelines and reallocated all
  vectors every execution. Steady-state pool hit rate 50pct to 100pct.
---
 src/include/processor/processor.h            |  9 ++-
 src/include/storage/table/rel_table.h        |  4 ++
 src/processor/operator/physical_operator.cpp | 10 +++
 src/processor/processor.cpp                  | 33 ++++++---
 src/storage/table/rel_table.cpp              | 75 ++++++++++++++------
 5 files changed, 98 insertions(+), 33 deletions(-)

diff --git a/src/include/processor/processor.h b/src/include/processor/processor.h
index 32ece3f8c..d10978873 100644
--- a/src/include/processor/processor.h
+++ b/src/include/processor/processor.h
@@ -39,8 +39,13 @@ class ResultSetPool {
 
 private:
     struct Entry {
-        uint64_t descriptorID = UINT64_MAX;
-        std::shared_ptr<ResultSet> resultSet;
+        // MRU-first cache of (descriptor id, ResultSet). A single slot thrashed when
+        // worker threads ran different pipelines on consecutive executions (each pipeline
+        // sink has its own descriptor id), so every execution reallocated all vectors.
+        // A small per-thread cache covers the working set; the cap bounds memory to
+        // MAX_CACHED_PER_THREAD ResultSets per thread.
+        static constexpr size_t MAX_CACHED_PER_THREAD = 8;
+        std::vector<std::pair<uint64_t, std::shared_ptr<ResultSet>>> cached;
     };
     std::mutex mtx;
     std::unordered_map<std::thread::id, Entry> entries;
diff --git a/src/include/storage/table/rel_table.h b/src/include/storage/table/rel_table.h
index 5ab61c25b..a34476454 100644
--- a/src/include/storage/table/rel_table.h
+++ b/src/include/storage/table/rel_table.h
@@ -43,6 +43,10 @@ struct RelTableScanState : TableScanState {
     // groups have been scanned (the local/uncommitted scan and downstream resume paths
     // iterate over the full batch).
     std::vector<common::sel_t> allBoundNodePositions;
+    // Reusable workspace for RelTable::initScanState: the batch's bound-node positions
+    // (copied from the input selection vector, then sorted by node offset). Kept on the
+    // scan state so repeated per-morsel inits reuse the buffer instead of allocating.
+    std::vector<common::sel_t> positionsBuffer;
 
     // Multi-parent packed scan support (see docs/multi_parent_lifetime.md). When
     // packedMultiParentScan is true, the CSR scan may serve children of MULTIPLE bound parents
diff --git a/src/processor/operator/physical_operator.cpp b/src/processor/operator/physical_operator.cpp
index 4ed4150d6..704eb4ac3 100644
--- a/src/processor/operator/physical_operator.cpp
+++ b/src/processor/operator/physical_operator.cpp
@@ -244,6 +244,16 @@ void PhysicalOperator::finalize(ExecutionContext* context) {
 }
 
 void PhysicalOperator::registerProfilingMetrics(Profiler* profiler) {
+    if (!profiler->enabled) {
+        // Fast path for normal (non-PROFILE) execution: bind to thread-local dummies
+        // instead of registering into the profiler's metric map. Registration allocates
+        // map nodes, strings and metric objects per operator per task and serializes on
+        // the profiler mutex -- pure overhead when nobody reads the metrics.
+        thread_local TimeMetric dummyTime{false};
+        thread_local NumericMetric dummyNum{false};
+        metrics = std::make_unique<OperatorMetrics>(dummyTime, dummyNum);
+        return;
+    }
     auto executionTime = profiler->registerTimeMetric(getTimeMetricKey());
     auto numOutputTuple = profiler->registerNumericMetric(getNumTupleMetricKey());
     metrics = std::make_unique<OperatorMetrics>(*executionTime, *numOutputTuple);
diff --git a/src/processor/processor.cpp b/src/processor/processor.cpp
index 6029643fa..b84e852c1 100644
--- a/src/processor/processor.cpp
+++ b/src/processor/processor.cpp
@@ -20,9 +20,19 @@ std::shared_ptr<ResultSet> ResultSetPool::getOrCreate(ResultSetDescriptor* descr
     std::shared_ptr<ResultSet> reusable;
     {
         std::lock_guard lock{mtx};
-        auto& entry = entries[threadID];
-        if (entry.resultSet && entry.descriptorID == descriptor->id) {
-            reusable = entry.resultSet;
+        auto& cached = entries[threadID].cached;
+        for (auto it = cached.begin(); it != cached.end(); ++it) {
+            if (it->first == descriptor->id && it->second != nullptr) {
+                // Move-to-front: the just-used descriptor is the most likely next hit.
+                // (Rotate first, then copy out: moving the pair after pilfering its
+                // shared_ptr would leave a null entry that matches forever but never hits.)
+                if (it != cached.begin()) {
+                    std::rotate(cached.begin(), it, it + 1);
+                    it = cached.begin();
+                }
+                reusable = it->second;
+                break;
+            }
         }
     }
     if (reusable) {
@@ -31,16 +41,19 @@ std::shared_ptr<ResultSet> ResultSetPool::getOrCreate(ResultSetDescriptor* descr
         reusable->resetForReuse();
         return reusable;
     }
-    // First time on this thread, or a different prepared statement: allocate fresh outside the
-    // lock, then swap it in and release the previous ResultSet after unlocking.
+    // First time on this thread, or an uncached descriptor: allocate fresh outside the
+    // lock, then install it (evicting the LRU entry past the cap) and release evicted
+    // ResultSets after unlocking.
     auto resultSet = std::make_shared<ResultSet>(descriptor, memoryManager);
-    std::shared_ptr<ResultSet> previous;
+    std::shared_ptr<ResultSet> evicted;
     {
         std::lock_guard lock{mtx};
-        auto& entry = entries[threadID];
-        previous = std::move(entry.resultSet);
-        entry.resultSet = resultSet;
-        entry.descriptorID = descriptor->id;
+        auto& cached = entries[threadID].cached;
+        cached.insert(cached.begin(), {descriptor->id, resultSet});
+        if (cached.size() > Entry::MAX_CACHED_PER_THREAD) {
+            evicted = std::move(cached.back().second);
+            cached.pop_back();
+        }
     }
     return resultSet;
 }
diff --git a/src/storage/table/rel_table.cpp b/src/storage/table/rel_table.cpp
index fa4980c9b..af6640aec 100644
--- a/src/storage/table/rel_table.cpp
+++ b/src/storage/table/rel_table.cpp
@@ -220,13 +220,24 @@ void RelTable::initScanState(Transaction* transaction, TableScanState& scanState
     // Collect the positions of this batch's bound nodes. On a fresh batch they come from
     // the input sel vector; when resuming a batch (e.g. a multi rel table scan moving to
     // the next rel table over the same bound nodes) they were cached by the previous init.
-    std::vector<common::sel_t> positions;
+    // Reusable workspace (no per-batch allocation): the batch's bound-node positions.
+    auto& positions = relScanState.positionsBuffer;
     const auto& sel = resetCachedBoundNodeSelVec ?
                           relScanState.nodeIDVector->state->getSelVector() :
                           relScanState.cachedBoundNodeSelVector;
     positions.resize(sel.getSelSize());
+    if (positions.empty()) {
+        return;
+    }
+    bool alreadySorted = true;
+    common::offset_t prevOffset = 0;
     for (auto i = 0u; i < sel.getSelSize(); i++) {
         positions[i] = sel[i];
+        const auto offset = relScanState.nodeIDVector->readNodeOffset(sel[i]);
+        if (i > 0 && offset < prevOffset) {
+            alreadySorted = false;
+        }
+        prevOffset = offset;
     }
     // The CSR scan processes the bound nodes of a batch in a single forward pass over the
     // CSR lists, which are laid out in ascending bound-node-offset order within each node
@@ -235,21 +246,24 @@ void RelTable::initScanState(Transaction* transaction, TableScanState& scanState
     // (a)->(b)->(c) as scan(a) -> extend(a->b) -> extend(b->c)), the bound nodes are the
     // children of a single parent and come in arbitrary order, so sort them by offset.
     // Parents whose CSR list lies before the current scan position would otherwise be
-    // silently skipped by CSRNodeGroup::scanCommittedPersistentWithCache.
-    std::sort(positions.begin(), positions.end(), [&](common::sel_t a, common::sel_t b) {
-        return relScanState.nodeIDVector->readNodeOffset(a) <
-               relScanState.nodeIDVector->readNodeOffset(b);
-    });
-    // Partition the bound nodes by node group. They form a contiguous sequence within a
-    // single node group when the input is a node table scan. However, the children of a
-    // single parent can span multiple node groups, in which case the batch is scanned one
+    // silently skipped by CSRNodeGroup::scanCommittedPersistentWithCache. Batches that
+    // already arrive in ascending offset order skip the sort.
+    if (!alreadySorted) {
+        std::sort(positions.begin(), positions.end(), [&](common::sel_t a, common::sel_t b) {
+            return relScanState.nodeIDVector->readNodeOffset(a) <
+                   relScanState.nodeIDVector->readNodeOffset(b);
+        });
+    }
+    // Partition the bound nodes by node group. Offsets are ascending now, so node groups
+    // form contiguous runs: comparing the first/last group detects the common
+    // single-group case without a tree or per-group vectors. The children of a single
+    // parent can still span multiple node groups, in which case the batch is scanned one
     // node group at a time; see RelTableScanState::boundNodeGroupPlan.
-    std::map<node_group_idx_t, std::vector<common::sel_t>> groups;
-    for (const auto pos : positions) {
-        const auto boundNodeID = relScanState.nodeIDVector->getValue<nodeID_t>(pos);
-        groups[StorageUtils::getNodeGroupIdx(boundNodeID.offset)].push_back(pos);
-    }
-    if (groups.size() <= 1) {
+    const auto firstGroupIdx =
+        StorageUtils::getNodeGroupIdx(relScanState.nodeIDVector->readNodeOffset(positions.front()));
+    const auto lastGroupIdx =
+        StorageUtils::getNodeGroupIdx(relScanState.nodeIDVector->readNodeOffset(positions.back()));
+    if (firstGroupIdx == lastGroupIdx) {
         // Single node group (the common case): historical fast path.
         relScanState.boundNodeGroupPlan.clear();
         relScanState.allBoundNodePositions.clear();
@@ -291,13 +305,32 @@ void RelTable::initScanState(Transaction* transaction, TableScanState& scanState
         }
         return;
     }
-    // Multi-group batch: scan one node group at a time.
-    relScanState.allBoundNodePositions = std::move(positions);
+    // Multi-group batch: scan one node group at a time. Offsets are ascending, so node
+    // groups form contiguous runs over the sorted positions: linear run partition, no map.
+    // allBoundNodePositions keeps a copy (reusing its capacity across batches); the
+    // workspace buffer stays intact for the next batch.
+    relScanState.allBoundNodePositions = positions;
     relScanState.boundNodeGroupPlan.clear();
-    for (auto& [groupIdx, groupPositions] : groups) {
-        relScanState.boundNodeGroupPlan.emplace_back(
-            getDirectedTableData(relScanState.direction)->getNodeGroup(groupIdx),
-            std::move(groupPositions));
+    auto runStart = 0u;
+    auto runGroupIdx = firstGroupIdx;
+    for (auto k = 1u;; ++k) {
+        const auto runEnds = (k == positions.size());
+        const auto groupIdx = runEnds ?
+                                  runGroupIdx :
+                                  StorageUtils::getNodeGroupIdx(
+                                      relScanState.nodeIDVector->readNodeOffset(positions[k]));
+        if (runEnds || groupIdx != runGroupIdx) {
+            std::vector<common::sel_t> runPositions(positions.begin() + runStart,
+                positions.begin() + k);
+            relScanState.boundNodeGroupPlan.emplace_back(
+                getDirectedTableData(relScanState.direction)->getNodeGroup(runGroupIdx),
+                std::move(runPositions));
+            if (runEnds) {
+                break;
+            }
+            runStart = k;
+            runGroupIdx = groupIdx;
+        }
     }
     relScanState.startNextBoundNodeGroup(transaction);
 }

From 46462122d18787ea6a7378552f5e91d3871fcace Mon Sep 17 00:00:00 2001
From: Arun Sharma <arun@sharma-home.net>
Date: Sat, 17 Oct 2026 12:50:32 -0700
Subject: [PATCH 2/6] perf(processor): pool cloned physical plans per thread on
 the fast path

Checkout of the cached physical-plan template clones hundreds of small allocations per execution (operators, evaluators, print infos, schemas, descriptors). Keep up to 10 ready plans per thread, reset via prepareForReuse, and reuse them. Descriptors stay attached on pooled plans; nested same-statement execution falls back to fresh clones.
---
 src/include/main/prepared_statement.h | 13 +++++++
 src/main/client_context.cpp           | 56 ++++++++++++++++++++++++---
 2 files changed, 64 insertions(+), 5 deletions(-)

diff --git a/src/include/main/prepared_statement.h b/src/include/main/prepared_statement.h
index 5509ba799..b52655e57 100644
--- a/src/include/main/prepared_statement.h
+++ b/src/include/main/prepared_statement.h
@@ -1,7 +1,9 @@
 #pragma once
 
 #include <memory>
+#include <mutex>
 #include <string>
+#include <thread>
 #include <unordered_map>
 #include <vector>
 
@@ -73,6 +75,17 @@ struct CachedPreparedStatement {
     // https://github.com/LadybugDB/ladybug/issues/1158).
     uint64_t planVersion = 0;
 
+    // Pool of cloned physical plans ready for reuse, keyed by executing thread.
+    // Cloning the cached template per execution costs hundreds of small allocations
+    // (operators, evaluators, print infos, schemas); pooled plans are reset via
+    // prepareForReuse() and reused instead. Capped per thread: deeper reentrancy
+    // (nested same-statement execution) falls back to fresh clones. Guarded by
+    // physicalPlanPoolMutex; entries are dropped with the statement.
+    static constexpr size_t PHYSICAL_PLAN_POOL_PER_THREAD = 10;
+    std::mutex physicalPlanPoolMutex;
+    std::unordered_map<std::thread::id, std::vector<std::unique_ptr<processor::PhysicalPlan>>>
+        physicalPlanPool;
+
     CachedPreparedStatement();
     ~CachedPreparedStatement();
 
diff --git a/src/main/client_context.cpp b/src/main/client_context.cpp
index 6b21252e9..75a9f500a 100644
--- a/src/main/client_context.cpp
+++ b/src/main/client_context.cpp
@@ -644,7 +644,9 @@ void attachSinkDescriptors(processor::PhysicalOperator* op,
     const std::vector<std::unique_ptr<processor::ResultSetDescriptor>>& sinks, common::idx_t& pos) {
     if (auto* sink = dynamic_cast<processor::Sink*>(op)) {
         auto& desc = sinks[pos];
-        if (desc != nullptr) {
+        if (desc != nullptr && sink->getDescriptor() == nullptr) {
+            // Only attach when missing: pooled plans reused across executions keep
+            // the descriptors attached on first use (same ids, content-identical).
             sink->setDescriptor(desc->copy());
         }
     }
@@ -654,6 +656,39 @@ void attachSinkDescriptors(processor::PhysicalOperator* op,
     }
 }
 
+// Pooled physical-plan checkout: pop a same-thread reusable plan, else clone the cached
+// template. Either way the returned plan is reset for reuse and carries sink descriptors.
+std::unique_ptr<processor::PhysicalPlan> checkoutPooledPhysicalPlan(
+    CachedPreparedStatement* cachedStatement) {
+    {
+        std::lock_guard lock{cachedStatement->physicalPlanPoolMutex};
+        auto it = cachedStatement->physicalPlanPool.find(std::this_thread::get_id());
+        if (it != cachedStatement->physicalPlanPool.end() && !it->second.empty()) {
+            auto plan = std::move(it->second.back());
+            it->second.pop_back();
+            if (std::getenv("POOL_DEBUG")) {
+                std::fprintf(stderr, "[planpool] HIT depth=%lu\n",
+                    (unsigned long)it->second.size());
+            }
+            return plan;
+        }
+        if (std::getenv("POOL_DEBUG")) {
+            std::fprintf(stderr, "[planpool] MISS\n");
+        }
+    }
+    return std::make_unique<processor::PhysicalPlan>(
+        cachedStatement->physicalPlanCache->lastOperator->copy());
+}
+
+void recyclePooledPhysicalPlan(CachedPreparedStatement* cachedStatement,
+    std::unique_ptr<processor::PhysicalPlan> plan) {
+    std::lock_guard lock{cachedStatement->physicalPlanPoolMutex};
+    auto& pool = cachedStatement->physicalPlanPool[std::this_thread::get_id()];
+    if (pool.size() < CachedPreparedStatement::PHYSICAL_PLAN_POOL_PER_THREAD) {
+        pool.push_back(std::move(plan));
+    }
+}
+
 } // namespace
 
 bool ClientContext::isCachedPlanAllowedFor(const PreparedStatement& preparedStatement) const {
@@ -716,6 +751,7 @@ std::unique_ptr<QueryResult> ClientContext::executeNoLock(PreparedStatement* pre
                 auto executionContext =
                     std::make_unique<ExecutionContext>(profiler.get(), this, *queryID);
                 std::unique_ptr<PhysicalPlan> physicalPlan;
+                bool pooledPlan = false;
                 // The `enable_cached_prepared_statement` setting gates both cache reuse and
                 // cache population, so a disabled scope never serves (or fills) the plan cache.
                 // Statements with parameters baked into a plan (e.g. parameter-valued
@@ -726,16 +762,19 @@ std::unique_ptr<QueryResult> ClientContext::executeNoLock(PreparedStatement* pre
                                                isCachedPlanAllowedFor(*preparedStatement) &&
                                                !cachedStatement->hasBakedParameters;
                 if (cachedPlanAllowed && cachedStatement->physicalPlanCache) {
-                    // Fast path: clone cached operator tree and refresh sink state.
-                    // Avoids the PlanMapper::mapOperator recursion entirely.
-                    physicalPlan = std::make_unique<PhysicalPlan>(
-                        cachedStatement->physicalPlanCache->lastOperator->copy());
+                    // Fast path: reuse a pooled plan when available, else clone the cached
+                    // operator tree template. Avoids the PlanMapper::mapOperator recursion
+                    // and, on reuse, the per-execution clone entirely.
+                    physicalPlan = checkoutPooledPhysicalPlan(cachedStatement);
                     physicalPlan->lastOperator->prepareForReuse(storage::MemoryManager::Get(*this));
                     // Re-attach the ResultSetDescriptors (not propagated by copy()) of every
                     // pipeline-head sink, so each sub-task can build its ResultSet.
+                    // Pooled plans keep the descriptors attached on first use, so this is a
+                    // no-op for them.
                     common::idx_t pos = 0;
                     attachSinkDescriptors(physicalPlan->lastOperator.get(),
                         cachedStatement->sinkResultSetDescriptors, pos);
+                    pooledPlan = true;
                 } else {
                     auto mapper = PlanMapper(executionContext.get());
                     physicalPlan = mapper.getPhysicalPlan(cachedStatement->logicalPlan.get(),
@@ -771,6 +810,13 @@ std::unique_ptr<QueryResult> ClientContext::executeNoLock(PreparedStatement* pre
                     result = localDatabase->queryProcessor->execute(physicalPlan.get(),
                         executionContext.get());
                 }
+                // Recycle fast-path plans for reuse by later executions on this thread.
+                // Mapper-built plans are always fresh (and may carry baked parameters),
+                // so only pooled/template-cloned plans go back. Skipped on exception:
+                // the plan is then destroyed, and the next execution clones afresh.
+                if (pooledPlan) {
+                    recyclePooledPhysicalPlan(cachedStatement, std::move(physicalPlan));
+                }
             },
             preparedStatement->isReadOnly(), isTransactionStatement,
             TransactionHelper::getAction(true /*shouldCommitNewTransaction*/,

From ad3f67af34a2f94956bd916f25c6642f3f23cab7 Mon Sep 17 00:00:00 2001
From: Arun Sharma <arun@sharma-home.net>
Date: Sat, 17 Oct 2026 13:18:35 -0700
Subject: [PATCH 3/6] perf(processor): make operator init paths idempotent plus
 SFO-bound chunk functors

Reused-operator hardening plus allocator fixes. No behavior change for fresh operators (all cleared vectors start empty; all SFO lambdas evaluate identically): operator initLocalState fills now clear member vectors first so re-execution starts clean (hash join build and probe, intersect, cross product, order by and top-k, hash and simple aggregates, scan tables, hash-aggregate local state), and ColumnChunkData keeps the compression algorithm on the chunk with capture-this stdfunction members that fit small-function optimization instead of heap-allocating 24-byte functors per chunk.
---
 src/include/storage/table/column_chunk_data.h | 13 ++++++-
 .../operator/aggregate/base_aggregate.cpp     |  1 +
 .../aggregate/base_aggregate_scan.cpp         |  1 +
 .../operator/aggregate/hash_aggregate.cpp     |  4 ++
 .../aggregate/hash_aggregate_scan.cpp         |  1 +
 .../operator/aggregate/simple_aggregate.cpp   |  2 +
 src/processor/operator/cross_product.cpp      |  1 +
 .../operator/hash_join/hash_join_build.cpp    |  5 +++
 .../operator/hash_join/hash_join_probe.cpp    |  3 ++
 .../operator/intersect/intersect.cpp          |  6 +++
 src/processor/operator/order_by/order_by.cpp  |  2 +
 src/processor/operator/order_by/top_k.cpp     |  2 +
 src/processor/operator/scan/scan_table.cpp    |  1 +
 src/storage/table/column_chunk_data.cpp       | 37 ++++++++++++++-----
 14 files changed, 67 insertions(+), 12 deletions(-)

diff --git a/src/include/storage/table/column_chunk_data.h b/src/include/storage/table/column_chunk_data.h
index 38b283336..79c7fee1d 100644
--- a/src/include/storage/table/column_chunk_data.h
+++ b/src/include/storage/table/column_chunk_data.h
@@ -306,8 +306,13 @@ class LBUG_API ColumnChunkData {
 private:
     using flush_buffer_func_t = std::function<ColumnChunkMetadata(const std::span<uint8_t>,
         FileHandle*, const PageRange&, const ColumnChunkMetadata&)>;
-    flush_buffer_func_t initializeFlushBufferFunction(
-        std::shared_ptr<CompressionAlg> compression) const;
+    // Stack-only functor dispatch for flushBufferFunction: constructs the (small)
+    // compression functor on the stack per call instead of heap-allocating it inside
+    // the std::function at chunk-construction time.
+    ColumnChunkMetadata flushBufferImpl(std::span<uint8_t> buffer, FileHandle* dataFH,
+        const PageRange& entry, const ColumnChunkMetadata& otherMetadata) const;
+    ColumnChunkMetadata getMetadataImpl(std::span<const uint8_t> buffer, uint64_t numValues,
+        StorageValue minValue, StorageValue maxValue) const;
     uint64_t getBufferSize(uint64_t capacity_) const;
 
 protected:
@@ -319,6 +324,10 @@ class LBUG_API ColumnChunkData {
     ResidencyState residencyState;
     common::LogicalType dataType;
     bool enableCompression;
+    // Owned compression algorithm, shared by the flush/metadata functors. Keeping it here
+    // lets both std::function members capture only `this` (8 bytes, fits small-function
+    // optimization) instead of heap-allocating 24-byte bound functors per chunk.
+    std::shared_ptr<CompressionAlg> compressionAlg;
     uint32_t numBytesPerValue;
     uint64_t capacity;
     std::unique_ptr<MemoryBuffer> buffer;
diff --git a/src/processor/operator/aggregate/base_aggregate.cpp b/src/processor/operator/aggregate/base_aggregate.cpp
index c3fee3099..7a9e91eef 100644
--- a/src/processor/operator/aggregate/base_aggregate.cpp
+++ b/src/processor/operator/aggregate/base_aggregate.cpp
@@ -23,6 +23,7 @@ BaseAggregateSharedState::BaseAggregateSharedState(
       readyForFinalization{false} {}
 
 void BaseAggregate::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* /*context*/) {
+    aggInputs.clear();
     for (auto& info : aggInfos) {
         auto aggregateInput = AggregateInput();
         if (info.aggVectorPos.dataChunkPos == INVALID_DATA_CHUNK_POS) {
diff --git a/src/processor/operator/aggregate/base_aggregate_scan.cpp b/src/processor/operator/aggregate/base_aggregate_scan.cpp
index 6cdc06f18..b19e8c395 100644
--- a/src/processor/operator/aggregate/base_aggregate_scan.cpp
+++ b/src/processor/operator/aggregate/base_aggregate_scan.cpp
@@ -8,6 +8,7 @@ namespace processor {
 
 void BaseAggregateScan::initLocalStateInternal(ResultSet* resultSet,
     ExecutionContext* /*context*/) {
+    aggregateVectors.clear();
     for (auto& dataPos : scanInfo.aggregatesPos) {
         auto valueVector = resultSet->getValueVector(dataPos);
         aggregateVectors.push_back(valueVector);
diff --git a/src/processor/operator/aggregate/hash_aggregate.cpp b/src/processor/operator/aggregate/hash_aggregate.cpp
index 0db1e670f..ed7070825 100644
--- a/src/processor/operator/aggregate/hash_aggregate.cpp
+++ b/src/processor/operator/aggregate/hash_aggregate.cpp
@@ -237,6 +237,10 @@ void HashAggregateSharedState::assertFinalized() const {
 void HashAggregateLocalState::init(HashAggregateSharedState* sharedState, ResultSet& resultSet,
     main::ClientContext* context, std::vector<function::AggregateFunction>& aggregateFunctions,
     std::vector<common::LogicalType> distinctKeyTypes) {
+    // Clear member vectors so re-execution on a reused operator starts clean.
+    keyVectors.clear();
+    dependentKeyVectors.clear();
+    leadingState = nullptr;
     auto& info = sharedState->getAggregateInfo();
     std::vector<LogicalType> keyDataTypes;
     for (auto& pos : info.flatKeysPos) {
diff --git a/src/processor/operator/aggregate/hash_aggregate_scan.cpp b/src/processor/operator/aggregate/hash_aggregate_scan.cpp
index c28f7626c..1391af298 100644
--- a/src/processor/operator/aggregate/hash_aggregate_scan.cpp
+++ b/src/processor/operator/aggregate/hash_aggregate_scan.cpp
@@ -7,6 +7,7 @@ namespace processor {
 
 void HashAggregateScan::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
     BaseAggregateScan::initLocalStateInternal(resultSet, context);
+    groupByKeyVectors.clear();
     for (auto& dataPos : groupByKeyVectorsPos) {
         auto valueVector = resultSet->getValueVector(dataPos);
         groupByKeyVectors.push_back(valueVector.get());
diff --git a/src/processor/operator/aggregate/simple_aggregate.cpp b/src/processor/operator/aggregate/simple_aggregate.cpp
index 9beeaeb97..64a06299b 100644
--- a/src/processor/operator/aggregate/simple_aggregate.cpp
+++ b/src/processor/operator/aggregate/simple_aggregate.cpp
@@ -242,6 +242,8 @@ void SimpleAggregateSharedState::finalizePartitions(storage::MemoryManager* memo
 }
 
 void SimpleAggregate::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
+    localAggregateStates.clear();
+    distinctHashTables.clear();
     BaseAggregate::initLocalStateInternal(resultSet, context);
     for (auto i = 0u; i < aggregateFunctions.size(); ++i) {
         auto& func = aggregateFunctions[i];
diff --git a/src/processor/operator/cross_product.cpp b/src/processor/operator/cross_product.cpp
index f1d8a1ec9..e2764bdf8 100644
--- a/src/processor/operator/cross_product.cpp
+++ b/src/processor/operator/cross_product.cpp
@@ -7,6 +7,7 @@ namespace lbug {
 namespace processor {
 
 void CrossProduct::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* /*context*/) {
+    vectorsToScan.clear();
     for (auto& pos : info.outVecPos) {
         vectorsToScan.push_back(resultSet->getValueVector(pos).get());
     }
diff --git a/src/processor/operator/hash_join/hash_join_build.cpp b/src/processor/operator/hash_join/hash_join_build.cpp
index 209d8daf7..598f9b951 100644
--- a/src/processor/operator/hash_join/hash_join_build.cpp
+++ b/src/processor/operator/hash_join/hash_join_build.cpp
@@ -26,6 +26,11 @@ void HashJoinSharedState::mergeLocalHashTable(JoinHashTable& localHashTable) {
 }
 
 void HashJoinBuild::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
+    // Clear member vectors so re-execution on a reused operator (e.g. pooled task roots)
+    // starts clean; fresh operators are unaffected (vectors start empty).
+    keyState = nullptr;
+    keyVectors.clear();
+    payloadVectors.clear();
     std::vector<LogicalType> keyTypes;
     for (auto i = 0u; i < info.keysPos.size(); ++i) {
         auto vector = resultSet->getValueVector(info.keysPos[i]).get();
diff --git a/src/processor/operator/hash_join/hash_join_probe.cpp b/src/processor/operator/hash_join/hash_join_probe.cpp
index 6b94f74cb..4a3c18fc4 100644
--- a/src/processor/operator/hash_join/hash_join_probe.cpp
+++ b/src/processor/operator/hash_join/hash_join_probe.cpp
@@ -16,6 +16,9 @@ std::string HashJoinProbePrintInfo::toString() const {
 }
 
 void HashJoinProbe::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
+    // Clear member vectors so re-execution on a reused operator starts clean.
+    keyVectors.clear();
+    vectorsToReadInto.clear();
     probeState = std::make_unique<ProbeState>();
     for (auto& keyDataPos : probeDataInfo.keysDataPos) {
         keyVectors.push_back(resultSet->getValueVector(keyDataPos).get());
diff --git a/src/processor/operator/intersect/intersect.cpp b/src/processor/operator/intersect/intersect.cpp
index 625628b4f..98b322ef1 100644
--- a/src/processor/operator/intersect/intersect.cpp
+++ b/src/processor/operator/intersect/intersect.cpp
@@ -19,6 +19,12 @@ std::string IntersectPrintInfo::toString() const {
 }
 
 void Intersect::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* /*context*/) {
+    // Clear member vectors so re-execution on a reused operator starts clean.
+    probeKeyVectors.clear();
+    payloadColumnIdxesToScanFrom.clear();
+    payloadVectorsToScanInto.clear();
+    intersectSelVectors.clear();
+    isIntersectListAFlatValue.clear();
     outKeyVector = resultSet->getValueVector(outputDataPos);
     for (auto& dataInfo : intersectDataInfos) {
         probeKeyVectors.push_back(resultSet->getValueVector(dataInfo.keyDataPos));
diff --git a/src/processor/operator/order_by/order_by.cpp b/src/processor/operator/order_by/order_by.cpp
index 5a97ec526..35300c515 100644
--- a/src/processor/operator/order_by/order_by.cpp
+++ b/src/processor/operator/order_by/order_by.cpp
@@ -18,6 +18,8 @@ std::string OrderByPrintInfo::toString() const {
 }
 
 void OrderBy::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
+    payloadVectors.clear();
+    orderByVectors.clear();
     localState = SortLocalState();
     localState.init(info, *sharedState, storage::MemoryManager::Get(*context->clientContext));
     for (auto& dataPos : info.payloadsPos) {
diff --git a/src/processor/operator/order_by/top_k.cpp b/src/processor/operator/order_by/top_k.cpp
index 99e2aa6c1..cdf477624 100644
--- a/src/processor/operator/order_by/top_k.cpp
+++ b/src/processor/operator/order_by/top_k.cpp
@@ -259,6 +259,8 @@ void TopKLocalState::append(const std::vector<common::ValueVector*>& keyVectors,
 }
 
 void TopK::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
+    payloadVectors.clear();
+    orderByVectors.clear();
     localState = TopKLocalState();
     localState.init(info, storage::MemoryManager::Get(*context->clientContext), *resultSet,
         skipNumber, limitNumber);
diff --git a/src/processor/operator/scan/scan_table.cpp b/src/processor/operator/scan/scan_table.cpp
index 0584f8654..c2674ac1d 100644
--- a/src/processor/operator/scan/scan_table.cpp
+++ b/src/processor/operator/scan/scan_table.cpp
@@ -139,6 +139,7 @@ common::sel_t ScanTable::applyNbrMaskFilter(common::SelectionVector& selVector,
 }
 
 void ScanTable::initLocalStateInternal(ResultSet*, ExecutionContext*) {
+    outVectors.clear();
     for (auto& pos : opInfo.outVectorsPos) {
         outVectors.push_back(resultSet->getValueVector(pos).get());
     }
diff --git a/src/storage/table/column_chunk_data.cpp b/src/storage/table/column_chunk_data.cpp
index cd48c84b0..7ccaf2e83 100644
--- a/src/storage/table/column_chunk_data.cpp
+++ b/src/storage/table/column_chunk_data.cpp
@@ -181,18 +181,32 @@ void ColumnChunkData::initializeBuffer(PhysicalTypeID physicalType, MemoryManage
 }
 
 void ColumnChunkData::initializeFunction() {
-    const auto compression = getCompression(dataType, enableCompression);
-    getMetadataFunction = GetCompressionMetadata(compression, dataType);
-    flushBufferFunction = initializeFlushBufferFunction(compression);
+    compressionAlg = getCompression(dataType, enableCompression);
+    // Capture only `this` so both closures fit std::function's inline storage (no heap
+    // allocation); the compression algorithm is owned by the chunk, and the small
+    // dispatch functors are constructed on the stack per call in the Impl helpers.
+    flushBufferFunction = [this](const std::span<uint8_t> buffer, FileHandle* dataFH,
+                              const PageRange& entry, const ColumnChunkMetadata& otherMetadata) {
+        return flushBufferImpl(buffer, dataFH, entry, otherMetadata);
+    };
+    getMetadataFunction = [this](const std::span<uint8_t> buffer, uint64_t numValues,
+                              StorageValue minValue, StorageValue maxValue) {
+        return getMetadataImpl(buffer, numValues, minValue, maxValue);
+    };
 }
 
-ColumnChunkData::flush_buffer_func_t ColumnChunkData::initializeFlushBufferFunction(
-    std::shared_ptr<CompressionAlg> compression) const {
+ColumnChunkMetadata ColumnChunkData::getMetadataImpl(std::span<const uint8_t> buffer,
+    uint64_t numValues, StorageValue minValue, StorageValue maxValue) const {
+    return GetCompressionMetadata(compressionAlg, dataType)(buffer, numValues, minValue, maxValue);
+}
+
+ColumnChunkMetadata ColumnChunkData::flushBufferImpl(std::span<uint8_t> buffer, FileHandle* dataFH,
+    const PageRange& entry, const ColumnChunkMetadata& otherMetadata) const {
     switch (dataType.getPhysicalType()) {
     case PhysicalTypeID::BOOL: {
         // Since we compress into memory, storage is the same as fixed-sized
         // values, but we need to mark it as being boolean compressed.
-        return uncompressedFlushBuffer;
+        return uncompressedFlushBuffer(buffer, dataFH, entry, otherMetadata);
     }
     case PhysicalTypeID::STRING:
     case PhysicalTypeID::JSON:
@@ -208,16 +222,19 @@ ColumnChunkData::flush_buffer_func_t ColumnChunkData::initializeFlushBufferFunct
     case PhysicalTypeID::UINT16:
     case PhysicalTypeID::UINT8:
     case PhysicalTypeID::INT128: {
-        return CompressedFlushBuffer(compression, dataType);
+        return CompressedFlushBuffer(compressionAlg, dataType)(buffer, dataFH, entry,
+            otherMetadata);
     }
     case PhysicalTypeID::DOUBLE: {
-        return CompressedFloatFlushBuffer<double>(compression, dataType);
+        return CompressedFloatFlushBuffer<double>(compressionAlg, dataType)(buffer, dataFH, entry,
+            otherMetadata);
     }
     case PhysicalTypeID::FLOAT: {
-        return CompressedFloatFlushBuffer<float>(compression, dataType);
+        return CompressedFloatFlushBuffer<float>(compressionAlg, dataType)(buffer, dataFH, entry,
+            otherMetadata);
     }
     default: {
-        return uncompressedFlushBuffer;
+        return uncompressedFlushBuffer(buffer, dataFH, entry, otherMetadata);
     }
     }
 }

From 3b181e6953809dd0ca66f86ae0faf197b5d03358 Mon Sep 17 00:00:00 2001
From: Arun Sharma <arun@sharma-home.net>
Date: Sat, 17 Oct 2026 22:20:36 -0700
Subject: [PATCH 4/6] fix(processor): clear pooled-plan state on rebind plus
 missed operator resets

- Clear physicalPlanPool when the physical-plan cache is invalidated
  on rebind (parameter type change / catalog version bump). Pooled trees
  cloned from the previous template reference stale evaluators and sink
  descriptors; the second execution after a type change reused them and
  returned wrong results (ApiTest.PrepareWithoutParamsReusesPlan).
- Reset per-execution vectors missed by the idempotency pass so pooled
  operator objects start clean: ResultCollector payload vectors,
  QueryPrimaryKeyLookup outVectors, PackedFilteredCount multiplicity
  states/local counts, UnionAllScan vectors, GroupedReachableCount key
  vectors/scan states, CopyTo input vectors.
- RelTable::initScanState: on empty batches drop the stale multi-group
  plan and mark the scan exhausted instead of rescanning the previous
  batch.
- ColumnChunkData: call flush/metadata impls directly instead of via
  this-capturing std::functions (use-after-move when chunks relocate;
  also removes the per-chunk allocation entirely).
- ResultSetPool: revert MRU-8 to the single per-thread slot. With unique
  descriptor ids per uncached query the 8-slot cache accumulates garbage
  until the cap, so buffer-manager memory keeps growing across identical
  queries (BufferManagerTest.TestBMUsageForIdenticalQueries).
---
 src/include/processor/processor.h             |  9 ++---
 src/include/storage/table/column_chunk_data.h | 18 ++--------
 src/main/client_context.cpp                   |  7 ++++
 .../aggregate/packed_filtered_count.cpp       |  4 +++
 src/processor/operator/persistent/copy_to.cpp |  2 ++
 .../operator/query_primary_key_lookup.cpp     |  2 ++
 src/processor/operator/result_collector.cpp   |  5 +++
 .../operator/scan/grouped_reachable_count.cpp |  3 ++
 .../operator/table_scan/union_all_scan.cpp    |  2 ++
 src/processor/processor.cpp                   | 33 ++++++-------------
 src/storage/table/column_chunk_data.cpp       | 17 ++--------
 src/storage/table/rel_table.cpp               |  5 +++
 12 files changed, 48 insertions(+), 59 deletions(-)

diff --git a/src/include/processor/processor.h b/src/include/processor/processor.h
index d10978873..32ece3f8c 100644
--- a/src/include/processor/processor.h
+++ b/src/include/processor/processor.h
@@ -39,13 +39,8 @@ class ResultSetPool {
 
 private:
     struct Entry {
-        // MRU-first cache of (descriptor id, ResultSet). A single slot thrashed when
-        // worker threads ran different pipelines on consecutive executions (each pipeline
-        // sink has its own descriptor id), so every execution reallocated all vectors.
-        // A small per-thread cache covers the working set; the cap bounds memory to
-        // MAX_CACHED_PER_THREAD ResultSets per thread.
-        static constexpr size_t MAX_CACHED_PER_THREAD = 8;
-        std::vector<std::pair<uint64_t, std::shared_ptr<ResultSet>>> cached;
+        uint64_t descriptorID = UINT64_MAX;
+        std::shared_ptr<ResultSet> resultSet;
     };
     std::mutex mtx;
     std::unordered_map<std::thread::id, Entry> entries;
diff --git a/src/include/storage/table/column_chunk_data.h b/src/include/storage/table/column_chunk_data.h
index 79c7fee1d..acc851c99 100644
--- a/src/include/storage/table/column_chunk_data.h
+++ b/src/include/storage/table/column_chunk_data.h
@@ -304,11 +304,8 @@ class LBUG_API ColumnChunkData {
     void resetInMemoryStats();
 
 private:
-    using flush_buffer_func_t = std::function<ColumnChunkMetadata(const std::span<uint8_t>,
-        FileHandle*, const PageRange&, const ColumnChunkMetadata&)>;
-    // Stack-only functor dispatch for flushBufferFunction: constructs the (small)
-    // compression functor on the stack per call instead of heap-allocating it inside
-    // the std::function at chunk-construction time.
+    // Direct dispatch helpers for flush/metadata (called without std::function indirection,
+    // so no per-chunk heap allocation and no `this`-capture lifetime issues under moves).
     ColumnChunkMetadata flushBufferImpl(std::span<uint8_t> buffer, FileHandle* dataFH,
         const PageRange& entry, const ColumnChunkMetadata& otherMetadata) const;
     ColumnChunkMetadata getMetadataImpl(std::span<const uint8_t> buffer, uint64_t numValues,
@@ -316,25 +313,16 @@ class LBUG_API ColumnChunkData {
     uint64_t getBufferSize(uint64_t capacity_) const;
 
 protected:
-    using get_metadata_func_t = std::function<ColumnChunkMetadata(const std::span<uint8_t>,
-        uint64_t, StorageValue, StorageValue)>;
-    using get_min_max_func_t =
-        std::function<std::pair<StorageValue, StorageValue>(const uint8_t*, uint64_t)>;
-
     ResidencyState residencyState;
     common::LogicalType dataType;
     bool enableCompression;
-    // Owned compression algorithm, shared by the flush/metadata functors. Keeping it here
-    // lets both std::function members capture only `this` (8 bytes, fits small-function
-    // optimization) instead of heap-allocating 24-byte bound functors per chunk.
+    // Owned compression algorithm for the flush/metadata dispatch helpers below.
     std::shared_ptr<CompressionAlg> compressionAlg;
     uint32_t numBytesPerValue;
     uint64_t capacity;
     std::unique_ptr<MemoryBuffer> buffer;
     std::unique_ptr<NullChunkData> nullData;
     uint64_t numValues;
-    flush_buffer_func_t flushBufferFunction;
-    get_metadata_func_t getMetadataFunction;
 
     // On-disk metadata for column chunk.
     ColumnChunkMetadata metadata;
diff --git a/src/main/client_context.cpp b/src/main/client_context.cpp
index 75a9f500a..d924a7eb5 100644
--- a/src/main/client_context.cpp
+++ b/src/main/client_context.cpp
@@ -444,6 +444,13 @@ std::unique_ptr<QueryResult> ClientContext::executeWithParams(PreparedStatement*
         cachedStatement->planVersion = newCachedStatement->planVersion;
         cachedStatement->physicalPlanCache.reset();
         cachedStatement->sinkResultSetDescriptors.clear();
+        // The pool holds operator trees cloned from the previous template (bound for the
+        // previous parameter types / catalog version). They reference stale evaluators and
+        // descriptors and must not be served for the rebound plan.
+        {
+            std::lock_guard poolLock{cachedStatement->physicalPlanPoolMutex};
+            cachedStatement->physicalPlanPool.clear();
+        }
         return executeNoLock(newPreparedStatement.get(), cachedStatement, queryID, {}, true);
     }
     return executeNoLock(newPreparedStatement.get(), newCachedStatement.get(), queryID, {}, true);
diff --git a/src/processor/operator/aggregate/packed_filtered_count.cpp b/src/processor/operator/aggregate/packed_filtered_count.cpp
index 213bbbdfb..f394a5e1c 100644
--- a/src/processor/operator/aggregate/packed_filtered_count.cpp
+++ b/src/processor/operator/aggregate/packed_filtered_count.cpp
@@ -61,6 +61,10 @@ void PackedFilteredCount::initLocalStateInternal(ResultSet* resultSet, Execution
     rhsValueVector = resultSet->getValueVector(info.rhsValuePos).get();
     selectState = resultSet->getDataChunk(info.selectChunkPos)->state.get();
     flatState = resultSet->getDataChunk(info.flatChunkPos)->state.get();
+    // Clear for pooled-plan reuse (fresh operators start empty). localCounts is moved out in
+    // executeInternal, but clear defensively in case of partial execution.
+    multiplicityStates.clear();
+    localCounts.clear();
     for (auto dataChunkPos : info.multiplicityChunkPos) {
         multiplicityStates.push_back(resultSet->getDataChunk(dataChunkPos)->state.get());
     }
diff --git a/src/processor/operator/persistent/copy_to.cpp b/src/processor/operator/persistent/copy_to.cpp
index f30467928..33eb2050d 100644
--- a/src/processor/operator/persistent/copy_to.cpp
+++ b/src/processor/operator/persistent/copy_to.cpp
@@ -18,6 +18,8 @@ std::string CopyToPrintInfo::toString() const {
 void CopyTo::initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) {
     localState.exportFuncLocalState =
         info.exportFunc.initLocalState(*context->clientContext, *info.bindData, info.isFlatVec);
+    // Clear for pooled-plan reuse (fresh operators start empty).
+    localState.inputVectors.clear();
     localState.inputVectors.reserve(info.inputVectorPoses.size());
     for (auto& inputVectorPos : info.inputVectorPoses) {
         localState.inputVectors.push_back(resultSet->getValueVector(inputVectorPos));
diff --git a/src/processor/operator/query_primary_key_lookup.cpp b/src/processor/operator/query_primary_key_lookup.cpp
index b09f225be..5b4e1fbfd 100644
--- a/src/processor/operator/query_primary_key_lookup.cpp
+++ b/src/processor/operator/query_primary_key_lookup.cpp
@@ -27,6 +27,8 @@ std::string QueryPrimaryKeyLookupPrintInfo::toString() const {
 void QueryPrimaryKeyLookup::initLocalStateInternal(ResultSet* resultSet,
     ExecutionContext* context) {
     nodeIDVector = resultSet->getValueVector(opInfo.nodeIDPos).get();
+    // Clear for pooled-plan reuse (fresh operators start empty).
+    outVectors.clear();
     for (auto& pos : opInfo.outVectorsPos) {
         outVectors.push_back(resultSet->getValueVector(pos).get());
     }
diff --git a/src/processor/operator/result_collector.cpp b/src/processor/operator/result_collector.cpp
index d57e8fd1a..a72139c9f 100644
--- a/src/processor/operator/result_collector.cpp
+++ b/src/processor/operator/result_collector.cpp
@@ -22,6 +22,11 @@ std::string ResultCollectorPrintInfo::toString() const {
 }
 
 void ResultCollector::initNecessaryLocalState(ResultSet* resultSet, ExecutionContext* context) {
+    // Pooled physical plans reuse this operator across executions: reset the vectors first
+    // (fresh operators start empty, so this is a no-op for them). finalizeInternal() also
+    // calls this on a different ResultSet, so without clearing each call would accumulate.
+    payloadVectors.clear();
+    payloadAndMarkVectors.clear();
     payloadVectors.reserve(info.payloadPositions.size());
     for (auto& pos : info.payloadPositions) {
         auto vec = resultSet->getValueVector(pos).get();
diff --git a/src/processor/operator/scan/grouped_reachable_count.cpp b/src/processor/operator/scan/grouped_reachable_count.cpp
index 4169bdf4e..56e06966d 100644
--- a/src/processor/operator/scan/grouped_reachable_count.cpp
+++ b/src/processor/operator/scan/grouped_reachable_count.cpp
@@ -21,6 +21,9 @@ namespace processor {
 
 void GroupedReachableCount::initLocalStateInternal(ResultSet* resultSet,
     ExecutionContext* context) {
+    // Clear for pooled-plan reuse (fresh operators start empty).
+    keyVectors.clear();
+    scanStates.clear();
     for (auto& pos : keyOutputPoss) {
         keyVectors.push_back(resultSet->getValueVector(pos).get());
     }
diff --git a/src/processor/operator/table_scan/union_all_scan.cpp b/src/processor/operator/table_scan/union_all_scan.cpp
index 446665e05..03c1ddc45 100644
--- a/src/processor/operator/table_scan/union_all_scan.cpp
+++ b/src/processor/operator/table_scan/union_all_scan.cpp
@@ -45,6 +45,8 @@ std::unique_ptr<UnionAllScanMorsel> UnionAllScanSharedState::getMorselNoLock(
 
 void UnionAllScan::initLocalStateInternal(ResultSet* /*resultSet_*/,
     ExecutionContext* /*context*/) {
+    // Clear for pooled-plan reuse (fresh operators start empty).
+    vectors.clear();
     for (auto& dataPos : info.outputPositions) {
         vectors.push_back(resultSet->getValueVector(dataPos).get());
     }
diff --git a/src/processor/processor.cpp b/src/processor/processor.cpp
index b84e852c1..6029643fa 100644
--- a/src/processor/processor.cpp
+++ b/src/processor/processor.cpp
@@ -20,19 +20,9 @@ std::shared_ptr<ResultSet> ResultSetPool::getOrCreate(ResultSetDescriptor* descr
     std::shared_ptr<ResultSet> reusable;
     {
         std::lock_guard lock{mtx};
-        auto& cached = entries[threadID].cached;
-        for (auto it = cached.begin(); it != cached.end(); ++it) {
-            if (it->first == descriptor->id && it->second != nullptr) {
-                // Move-to-front: the just-used descriptor is the most likely next hit.
-                // (Rotate first, then copy out: moving the pair after pilfering its
-                // shared_ptr would leave a null entry that matches forever but never hits.)
-                if (it != cached.begin()) {
-                    std::rotate(cached.begin(), it, it + 1);
-                    it = cached.begin();
-                }
-                reusable = it->second;
-                break;
-            }
+        auto& entry = entries[threadID];
+        if (entry.resultSet && entry.descriptorID == descriptor->id) {
+            reusable = entry.resultSet;
         }
     }
     if (reusable) {
@@ -41,19 +31,16 @@ std::shared_ptr<ResultSet> ResultSetPool::getOrCreate(ResultSetDescriptor* descr
         reusable->resetForReuse();
         return reusable;
     }
-    // First time on this thread, or an uncached descriptor: allocate fresh outside the
-    // lock, then install it (evicting the LRU entry past the cap) and release evicted
-    // ResultSets after unlocking.
+    // First time on this thread, or a different prepared statement: allocate fresh outside the
+    // lock, then swap it in and release the previous ResultSet after unlocking.
     auto resultSet = std::make_shared<ResultSet>(descriptor, memoryManager);
-    std::shared_ptr<ResultSet> evicted;
+    std::shared_ptr<ResultSet> previous;
     {
         std::lock_guard lock{mtx};
-        auto& cached = entries[threadID].cached;
-        cached.insert(cached.begin(), {descriptor->id, resultSet});
-        if (cached.size() > Entry::MAX_CACHED_PER_THREAD) {
-            evicted = std::move(cached.back().second);
-            cached.pop_back();
-        }
+        auto& entry = entries[threadID];
+        previous = std::move(entry.resultSet);
+        entry.resultSet = resultSet;
+        entry.descriptorID = descriptor->id;
     }
     return resultSet;
 }
diff --git a/src/storage/table/column_chunk_data.cpp b/src/storage/table/column_chunk_data.cpp
index 7ccaf2e83..fded2466e 100644
--- a/src/storage/table/column_chunk_data.cpp
+++ b/src/storage/table/column_chunk_data.cpp
@@ -182,17 +182,6 @@ void ColumnChunkData::initializeBuffer(PhysicalTypeID physicalType, MemoryManage
 
 void ColumnChunkData::initializeFunction() {
     compressionAlg = getCompression(dataType, enableCompression);
-    // Capture only `this` so both closures fit std::function's inline storage (no heap
-    // allocation); the compression algorithm is owned by the chunk, and the small
-    // dispatch functors are constructed on the stack per call in the Impl helpers.
-    flushBufferFunction = [this](const std::span<uint8_t> buffer, FileHandle* dataFH,
-                              const PageRange& entry, const ColumnChunkMetadata& otherMetadata) {
-        return flushBufferImpl(buffer, dataFH, entry, otherMetadata);
-    };
-    getMetadataFunction = [this](const std::span<uint8_t> buffer, uint64_t numValues,
-                              StorageValue minValue, StorageValue maxValue) {
-        return getMetadataImpl(buffer, numValues, minValue, maxValue);
-    };
 }
 
 ColumnChunkMetadata ColumnChunkData::getMetadataImpl(std::span<const uint8_t> buffer,
@@ -380,7 +369,7 @@ ColumnChunkMetadata ColumnChunkData::getMetadataToFlush() const {
         maxValue = max.value_or(StorageValue());
     }
     DASSERT(getBufferSize() == getBufferSize(capacity));
-    return getMetadataFunction(buffer->getBuffer(), numValues, minValue, maxValue);
+    return getMetadataImpl(buffer->getBuffer(), numValues, minValue, maxValue);
 }
 
 void ColumnChunkData::append(ValueVector* vector, const SelectionView& selView) {
@@ -432,7 +421,7 @@ ColumnChunkMetadata ColumnChunkData::flushBuffer(PageAllocator& pageAllocator,
     if (!otherMetadata.compMeta.isConstant() && bufferSizeToFlush != 0) {
         DASSERT(bufferSizeToFlush <= buffer->getBuffer().size_bytes());
         const auto bufferToFlush = buffer->getBuffer().subspan(0, bufferSizeToFlush);
-        return flushBufferFunction(bufferToFlush, pageAllocator.getDataFH(), entry, otherMetadata);
+        return flushBufferImpl(bufferToFlush, pageAllocator.getDataFH(), entry, otherMetadata);
     }
     DASSERT(otherMetadata.getNumPages() == 0);
     return otherMetadata;
@@ -1159,7 +1148,7 @@ uint64_t ColumnChunkData::getSizeOnDiskInMemoryStats() const {
     if (nullData) {
         nullSize = nullData->getSizeOnDiskInMemoryStats();
     }
-    auto metadata = getMetadataFunction(buffer->getBuffer(), numValues,
+    auto metadata = getMetadataImpl(buffer->getBuffer(), numValues,
         inMemoryStats.min.value_or(StorageValue{}), inMemoryStats.max.value_or(StorageValue{}));
     return metadata.getNumDataPages(dataType.getPhysicalType()) * common::LBUG_PAGE_SIZE + nullSize;
 }
diff --git a/src/storage/table/rel_table.cpp b/src/storage/table/rel_table.cpp
index af6640aec..09445b81a 100644
--- a/src/storage/table/rel_table.cpp
+++ b/src/storage/table/rel_table.cpp
@@ -227,6 +227,11 @@ void RelTable::initScanState(Transaction* transaction, TableScanState& scanState
                           relScanState.cachedBoundNodeSelVector;
     positions.resize(sel.getSelSize());
     if (positions.empty()) {
+        // No bound nodes in this batch: drop any stale multi-group plan from the previous
+        // batch so scanNext() observes an exhausted batch instead of rescanning it.
+        relScanState.boundNodeGroupPlan.clear();
+        relScanState.allBoundNodePositions.clear();
+        scanState.source = TableScanSource::NONE;
         return;
     }
     bool alreadySorted = true;

From cba24fbace0be99565ad802ec28f05bbb94ab728 Mon Sep 17 00:00:00 2001
From: Arun Sharma <arun@sharma-home.net>
Date: Mon, 19 Oct 2026 10:25:17 -0700
Subject: [PATCH 5/6] fix(processor): harden pooled-plan reuse for review

- metrics: use static disabled dummies instead of thread_local so pooled
  plans that outlive/migrate threads never hold dangling references.
  Sharing is safe: disabled metrics never mutate state.
- pool: drop POOL_DEBUG getenv/fprintf hot-path noise; cap per-thread to 2
  and total threads to 32 with empty-entry pruning so short-lived-thread
  churn cannot grow the map unboundedly; document thread-affinity as a
  hint and the descriptor-identity invariant (baked-param gate + pool
  clear on rebind).
- init idempotence: clear reused vectors in PathPropertyProbe,
  ReachableCount, MergeLocalState, TopK/OrderBy scan states.
---
 src/include/main/prepared_statement.h         |  9 +++-
 src/main/client_context.cpp                   | 45 ++++++++++++++-----
 .../operator/order_by/order_by_scan.cpp       |  2 +
 .../operator/order_by/top_k_scanner.cpp       |  2 +
 .../operator/path_property_probe.cpp          |  3 ++
 src/processor/operator/persistent/merge.cpp   |  2 +
 src/processor/operator/physical_operator.cpp  | 11 +++--
 .../operator/scan/reachable_count.cpp         |  2 +
 8 files changed, 62 insertions(+), 14 deletions(-)

diff --git a/src/include/main/prepared_statement.h b/src/include/main/prepared_statement.h
index b52655e57..a2513f5a5 100644
--- a/src/include/main/prepared_statement.h
+++ b/src/include/main/prepared_statement.h
@@ -81,7 +81,14 @@ struct CachedPreparedStatement {
     // prepareForReuse() and reused instead. Capped per thread: deeper reentrancy
     // (nested same-statement execution) falls back to fresh clones. Guarded by
     // physicalPlanPoolMutex; entries are dropped with the statement.
-    static constexpr size_t PHYSICAL_PLAN_POOL_PER_THREAD = 10;
+    // Thread affinity is a perf hint only: plans are interchangeable across threads.
+    // A stale entry from an exited thread is therefore safe to hand to a new thread
+    // whose std::thread::id happens to compare equal, and empty per-thread vectors
+    // are pruned on checkout so short-lived-thread churn cannot grow the map
+    // unboundedly (see checkoutPooledPhysicalPlan / recyclePooledPhysicalPlan).
+    // Total retained plans are bounded by MAX_POOLED_THREADS * PER_THREAD.
+    static constexpr size_t PHYSICAL_PLAN_POOL_PER_THREAD = 2;
+    static constexpr size_t PHYSICAL_PLAN_POOL_MAX_THREADS = 32;
     std::mutex physicalPlanPoolMutex;
     std::unordered_map<std::thread::id, std::vector<std::unique_ptr<processor::PhysicalPlan>>>
         physicalPlanPool;
diff --git a/src/main/client_context.cpp b/src/main/client_context.cpp
index d924a7eb5..38a59d92b 100644
--- a/src/main/client_context.cpp
+++ b/src/main/client_context.cpp
@@ -653,7 +653,12 @@ void attachSinkDescriptors(processor::PhysicalOperator* op,
         auto& desc = sinks[pos];
         if (desc != nullptr && sink->getDescriptor() == nullptr) {
             // Only attach when missing: pooled plans reused across executions keep
-            // the descriptors attached on first use (same ids, content-identical).
+            // the descriptors attached on first use.
+            // Invariant: descriptors are content-identical across executions of the same
+            // cached plan. This holds because (a) statements with parameters baked into
+            // the plan never populate the pool (hasBakedParameters gate in executeNoLock),
+            // and (b) any rebind (new parameter types / catalog version bump) clears the
+            // pool in executeWithParams before the new template can be served.
             sink->setDescriptor(desc->copy());
         }
     }
@@ -665,6 +670,8 @@ void attachSinkDescriptors(processor::PhysicalOperator* op,
 
 // Pooled physical-plan checkout: pop a same-thread reusable plan, else clone the cached
 // template. Either way the returned plan is reset for reuse and carries sink descriptors.
+// Empty per-thread vectors are pruned here so threads that have exited (or returned their
+// plans) do not leave the map growing unboundedly over a long-lived connection.
 std::unique_ptr<processor::PhysicalPlan> checkoutPooledPhysicalPlan(
     CachedPreparedStatement* cachedStatement) {
     {
@@ -673,14 +680,18 @@ std::unique_ptr<processor::PhysicalPlan> checkoutPooledPhysicalPlan(
         if (it != cachedStatement->physicalPlanPool.end() && !it->second.empty()) {
             auto plan = std::move(it->second.back());
             it->second.pop_back();
-            if (std::getenv("POOL_DEBUG")) {
-                std::fprintf(stderr, "[planpool] HIT depth=%lu\n",
-                    (unsigned long)it->second.size());
-            }
             return plan;
         }
-        if (std::getenv("POOL_DEBUG")) {
-            std::fprintf(stderr, "[planpool] MISS\n");
+        // Opportunistic prune: drop empty per-thread entries left behind by exited
+        // threads or fully checked-out pools. Keeps the map proportional to the number
+        // of threads currently retaining plans.
+        for (auto pit = cachedStatement->physicalPlanPool.begin();
+             pit != cachedStatement->physicalPlanPool.end();) {
+            if (pit->second.empty()) {
+                pit = cachedStatement->physicalPlanPool.erase(pit);
+            } else {
+                ++pit;
+            }
         }
     }
     return std::make_unique<processor::PhysicalPlan>(
@@ -690,10 +701,24 @@ std::unique_ptr<processor::PhysicalPlan> checkoutPooledPhysicalPlan(
 void recyclePooledPhysicalPlan(CachedPreparedStatement* cachedStatement,
     std::unique_ptr<processor::PhysicalPlan> plan) {
     std::lock_guard lock{cachedStatement->physicalPlanPoolMutex};
-    auto& pool = cachedStatement->physicalPlanPool[std::this_thread::get_id()];
-    if (pool.size() < CachedPreparedStatement::PHYSICAL_PLAN_POOL_PER_THREAD) {
-        pool.push_back(std::move(plan));
+    auto tid = std::this_thread::get_id();
+    auto it = cachedStatement->physicalPlanPool.find(tid);
+    if (it != cachedStatement->physicalPlanPool.end()) {
+        if (it->second.size() < CachedPreparedStatement::PHYSICAL_PLAN_POOL_PER_THREAD) {
+            it->second.push_back(std::move(plan));
+        }
+        return;
+    }
+    // New thread entry: refuse to grow the map past MAX_THREADS and drop the plan
+    // instead, so short-lived-thread churn cannot accumulate unbounded entries.
+    // Dropping is always safe (falls back to fresh clones); thread affinity is only
+    // a locality hint since pooled plans are interchangeable across threads.
+    if (cachedStatement->physicalPlanPool.size() >=
+        CachedPreparedStatement::PHYSICAL_PLAN_POOL_MAX_THREADS) {
+        return;
     }
+    auto& pool = cachedStatement->physicalPlanPool[tid];
+    pool.push_back(std::move(plan));
 }
 
 } // namespace
diff --git a/src/processor/operator/order_by/order_by_scan.cpp b/src/processor/operator/order_by/order_by_scan.cpp
index 9f0cff3b9..9516590ec 100644
--- a/src/processor/operator/order_by/order_by_scan.cpp
+++ b/src/processor/operator/order_by/order_by_scan.cpp
@@ -9,6 +9,8 @@ namespace processor {
 
 void OrderByScanLocalState::init(std::vector<DataPos>& outVectorPos, SortSharedState& sharedState,
     ResultSet& resultSet) {
+    // Clear for pooled-plan reuse (fresh states start empty).
+    vectorsToRead.clear();
     for (auto& dataPos : outVectorPos) {
         vectorsToRead.push_back(resultSet.getValueVector(dataPos).get());
     }
diff --git a/src/processor/operator/order_by/top_k_scanner.cpp b/src/processor/operator/order_by/top_k_scanner.cpp
index 239578bdb..f3cbee63b 100644
--- a/src/processor/operator/order_by/top_k_scanner.cpp
+++ b/src/processor/operator/order_by/top_k_scanner.cpp
@@ -5,6 +5,8 @@ namespace processor {
 
 void TopKLocalScanState::init(std::vector<DataPos>& outVectorPos, TopKSharedState& sharedState,
     ResultSet& resultSet) {
+    // Clear for pooled-plan reuse (fresh states start empty).
+    vectorsToScan.clear();
     for (auto& pos : outVectorPos) {
         vectorsToScan.push_back(resultSet.getValueVector(pos).get());
     }
diff --git a/src/processor/operator/path_property_probe.cpp b/src/processor/operator/path_property_probe.cpp
index 5f2d3036f..8351d1602 100644
--- a/src/processor/operator/path_property_probe.cpp
+++ b/src/processor/operator/path_property_probe.cpp
@@ -30,6 +30,9 @@ void PathPropertyProbe::initLocalStateInternal(ResultSet* /*resultSet_*/,
         StructVector::getFieldVectorRaw(*relsDataVector, InternalKeyword::SRC);
     pathDstNodeIDsDataVector =
         StructVector::getFieldVectorRaw(*relsDataVector, InternalKeyword::DST);
+    // Clear for pooled-plan reuse (fresh operators start empty).
+    pathNodesPropertyDataVectors.clear();
+    pathRelsPropertyDataVectors.clear();
     for (auto fieldIdx : info.nodeFieldIndices) {
         pathNodesPropertyDataVectors.push_back(
             StructVector::getFieldVector(nodesDataVector, fieldIdx).get());
diff --git a/src/processor/operator/persistent/merge.cpp b/src/processor/operator/persistent/merge.cpp
index abf211d8b..0b613ce8f 100644
--- a/src/processor/operator/persistent/merge.cpp
+++ b/src/processor/operator/persistent/merge.cpp
@@ -45,6 +45,8 @@ void Merge::initLocalStateInternal(ResultSet* resultSet_, ExecutionContext* cont
 
 void MergeLocalState::init(ResultSet& resultSet, main::ClientContext* context, MergeInfo& info) {
     std::vector<common::LogicalType> types;
+    // Clear for pooled-plan reuse (fresh states start empty).
+    keyVectors.clear();
     for (auto& evaluator : info.keyEvaluators) {
         auto keyVector = evaluator->resultVector.get();
         types.push_back(keyVector->dataType.copy());
diff --git a/src/processor/operator/physical_operator.cpp b/src/processor/operator/physical_operator.cpp
index 704eb4ac3..68ad79f30 100644
--- a/src/processor/operator/physical_operator.cpp
+++ b/src/processor/operator/physical_operator.cpp
@@ -245,12 +245,17 @@ void PhysicalOperator::finalize(ExecutionContext* context) {
 
 void PhysicalOperator::registerProfilingMetrics(Profiler* profiler) {
     if (!profiler->enabled) {
-        // Fast path for normal (non-PROFILE) execution: bind to thread-local dummies
+        // Fast path for normal (non-PROFILE) execution: bind to shared disabled dummies
         // instead of registering into the profiler's metric map. Registration allocates
         // map nodes, strings and metric objects per operator per task and serializes on
         // the profiler mutex -- pure overhead when nobody reads the metrics.
-        thread_local TimeMetric dummyTime{false};
-        thread_local NumericMetric dummyNum{false};
+        // These must have static (not thread_local) storage: pooled physical plans
+        // outlive threads and migrate across them, so a thread_local dummy would dangle
+        // once its thread exits. Sharing is safe because disabled metrics never mutate
+        // state (TimeMetric::start/stop and NumericMetric::increase return early when
+        // enabled==false), so there is no data race on the shared instances.
+        static TimeMetric dummyTime{false};
+        static NumericMetric dummyNum{false};
         metrics = std::make_unique<OperatorMetrics>(dummyTime, dummyNum);
         return;
     }
diff --git a/src/processor/operator/scan/reachable_count.cpp b/src/processor/operator/scan/reachable_count.cpp
index 3c69fb99b..955da0253 100644
--- a/src/processor/operator/scan/reachable_count.cpp
+++ b/src/processor/operator/scan/reachable_count.cpp
@@ -25,6 +25,8 @@ void ReachableCount::initLocalStateInternal(ResultSet* resultSet, ExecutionConte
     // Pre-compute the forward rel surfaces leaving from the bound node table and prepare one
     // scan state per physical rel table so they can be reused across all frontier nodes.
     relInfos = graph->getRelInfos(boundTableID);
+    // Clear for pooled-plan reuse (fresh operators start empty).
+    scanStates.clear();
     for (auto& relInfo : relInfos) {
         auto scanState = graph->prepareRelScan(*relInfo.relGroupEntry, relInfo.relTableID,
             relInfo.dstTableID, /*relProperties=*/std::vector<std::string>{});

From 74d6092db409cccd6bffa7e973f360e687f32f51 Mon Sep 17 00:00:00 2001
From: Arun Sharma <arun@sharma-home.net>
Date: Mon, 19 Oct 2026 11:01:45 -0700
Subject: [PATCH 6/6] feat(processor): make physical plan pool depth
 configurable

Add physical_plan_pool_per_thread (default 5), following the
checkpoint_threshold pattern: SystemConfig/DBConfig field, CALL setting
with validation (rejects negatives), and C API struct wiring. 0 disables
pooled-plan reuse. Checkout/recycle read the live setting, trim oversized
pools on limit changes, and drain the pool when disabled.

Adds SystemConfigTest.testPhysicalPlanPoolPerThreadSetting covering the
default, CALL get/set, negative rejection, and repeated prepared execution
with pooling both disabled and enabled.
---
 src/c_api/database.cpp                |  4 +-
 src/include/c_api/lbug.h              |  3 ++
 src/include/main/database.h           |  5 ++-
 src/include/main/db_config.h          |  1 +
 src/include/main/prepared_statement.h |  8 +++-
 src/include/main/settings.h           | 10 +++++
 src/main/client_context.cpp           | 62 ++++++++++++++++++---------
 src/main/database.cpp                 |  5 ++-
 src/main/db_config.cpp                |  6 ++-
 src/main/settings.cpp                 | 15 +++++++
 test/api/system_config_test.cpp       | 36 ++++++++++++++++
 11 files changed, 126 insertions(+), 29 deletions(-)

diff --git a/src/c_api/database.cpp b/src/c_api/database.cpp
index 7f43e73ac..61e593fde 100644
--- a/src/c_api/database.cpp
+++ b/src/c_api/database.cpp
@@ -15,7 +15,8 @@ lbug_state lbug_database_init(const char* database_path, lbug_system_config conf
         auto systemConfig = SystemConfig(config.buffer_pool_size, config.max_num_threads,
             config.enable_compression, config.read_only, config.max_db_size, config.auto_checkpoint,
             config.checkpoint_threshold, true, config.throw_on_wal_replay_failure,
-            config.enable_checksums, config.enable_multi_writes, config.enable_default_hash_index);
+            config.enable_checksums, config.enable_multi_writes, config.enable_default_hash_index,
+            config.physical_plan_pool_per_thread);
 
 #if defined(__APPLE__)
         systemConfig.threadQos = config.thread_qos;
@@ -54,6 +55,7 @@ lbug_system_config lbug_default_system_config() {
     cSystemConfig.enable_checksums = config.enableChecksums;
     cSystemConfig.enable_multi_writes = config.enableMultiWrites;
     cSystemConfig.enable_default_hash_index = config.enableDefaultHashIndex;
+    cSystemConfig.physical_plan_pool_per_thread = config.physicalPlanPoolPerThread;
 #if defined(__APPLE__)
     cSystemConfig.thread_qos = config.threadQos;
 #endif
diff --git a/src/include/c_api/lbug.h b/src/include/c_api/lbug.h
index d26425fe2..34439b143 100644
--- a/src/include/c_api/lbug.h
+++ b/src/include/c_api/lbug.h
@@ -140,6 +140,9 @@ typedef struct {
     bool enable_multi_writes;
     // If true, node tables create the default primary-key hash index.
     bool enable_default_hash_index;
+    // How many reusable cloned physical plans each thread retains per cached prepared
+    // statement. 0 disables pooled-plan reuse.
+    uint64_t physical_plan_pool_per_thread;
 
 #if defined(__APPLE__)
     // The thread quality of service (QoS) for the worker threads.
diff --git a/src/include/main/database.h b/src/include/main/database.h
index 5df2da2e1..f3553d7a1 100644
--- a/src/include/main/database.h
+++ b/src/include/main/database.h
@@ -73,13 +73,15 @@ struct LBUG_API SystemConfig {
      * Default to false.
      * @param enableDefaultHashIndex If true, node tables create the default primary-key hash
      * index.
+     * @param physicalPlanPoolPerThread How many reusable cloned physical plans each thread
+     * retains per cached prepared statement (pooled-plan fast path). 0 disables pooling.
      */
     explicit SystemConfig(uint64_t bufferPoolSize = -1u, uint64_t maxNumThreads = 0,
         bool enableCompression = true, bool readOnly = false, uint64_t maxDBSize = -1u,
         bool autoCheckpoint = true, uint64_t checkpointThreshold = 16777216 /* 16MB */,
         bool forceCheckpointOnClose = true, bool throwOnWalReplayFailure = false,
         bool enableChecksums = true, bool enableMultiWrites = false,
-        bool enableDefaultHashIndex = true
+        bool enableDefaultHashIndex = true, uint64_t physicalPlanPoolPerThread = 5
 #if defined(__APPLE__)
         ,
         uint32_t threadQos = QOS_CLASS_DEFAULT
@@ -98,6 +100,7 @@ struct LBUG_API SystemConfig {
     bool enableChecksums;
     bool enableMultiWrites;
     bool enableDefaultHashIndex;
+    uint64_t physicalPlanPoolPerThread;
 #if defined(__APPLE__)
     uint32_t threadQos;
 #endif
diff --git a/src/include/main/db_config.h b/src/include/main/db_config.h
index 0f6b4bbd7..9189bc8ca 100644
--- a/src/include/main/db_config.h
+++ b/src/include/main/db_config.h
@@ -29,6 +29,7 @@ struct DBConfig {
     bool enableChecksums;
     bool enableDefaultHashIndex;
     bool enableSpillingToDisk;
+    uint64_t physicalPlanPoolPerThread;
 #if defined(__APPLE__)
     uint32_t threadQos;
 #endif
diff --git a/src/include/main/prepared_statement.h b/src/include/main/prepared_statement.h
index a2513f5a5..65dcadfef 100644
--- a/src/include/main/prepared_statement.h
+++ b/src/include/main/prepared_statement.h
@@ -86,8 +86,12 @@ struct CachedPreparedStatement {
     // whose std::thread::id happens to compare equal, and empty per-thread vectors
     // are pruned on checkout so short-lived-thread churn cannot grow the map
     // unboundedly (see checkoutPooledPhysicalPlan / recyclePooledPhysicalPlan).
-    // Total retained plans are bounded by MAX_POOLED_THREADS * PER_THREAD.
-    static constexpr size_t PHYSICAL_PLAN_POOL_PER_THREAD = 2;
+    // Total retained plans are bounded by MAX_POOLED_THREADS * per-thread limit.
+    // The per-thread limit is runtime-configurable via the
+    // `physical_plan_pool_per_thread` setting (DBConfig); 0 disables pooling.
+    // Defaults below mirror the setting's default; the pool code reads the live
+    // setting rather than these constants.
+    static constexpr size_t PHYSICAL_PLAN_POOL_PER_THREAD_DEFAULT = 5;
     static constexpr size_t PHYSICAL_PLAN_POOL_MAX_THREADS = 32;
     std::mutex physicalPlanPoolMutex;
     std::unordered_map<std::thread::id, std::vector<std::unique_ptr<processor::PhysicalPlan>>>
diff --git a/src/include/main/settings.h b/src/include/main/settings.h
index 5ec556f12..f6c34f96d 100644
--- a/src/include/main/settings.h
+++ b/src/include/main/settings.h
@@ -186,5 +186,15 @@ struct EnableSubNodeGroupMorselsSetting {
     static common::Value getSetting(const ClientContext* context);
 };
 
+struct PhysicalPlanPoolPerThreadSetting {
+    static constexpr auto name = "physical_plan_pool_per_thread";
+    static constexpr auto inputType = common::LogicalTypeID::INT64;
+    // How many reusable cloned physical plans each thread retains per cached prepared
+    // statement. 0 disables pooled-plan reuse (every execution clones afresh).
+    // Safety valve for latent plan-reuse state bugs; lowering bounds retained memory.
+    static void setContext(ClientContext* context, const common::Value& parameter);
+    static common::Value getSetting(const ClientContext* context);
+};
+
 } // namespace main
 } // namespace lbug
diff --git a/src/main/client_context.cpp b/src/main/client_context.cpp
index 38a59d92b..ff38b694f 100644
--- a/src/main/client_context.cpp
+++ b/src/main/client_context.cpp
@@ -672,25 +672,38 @@ void attachSinkDescriptors(processor::PhysicalOperator* op,
 // template. Either way the returned plan is reset for reuse and carries sink descriptors.
 // Empty per-thread vectors are pruned here so threads that have exited (or returned their
 // plans) do not leave the map growing unboundedly over a long-lived connection.
+// maxPerThread is the live `physical_plan_pool_per_thread` setting; 0 disables pooling.
 std::unique_ptr<processor::PhysicalPlan> checkoutPooledPhysicalPlan(
-    CachedPreparedStatement* cachedStatement) {
+    CachedPreparedStatement* cachedStatement, uint64_t maxPerThread) {
     {
         std::lock_guard lock{cachedStatement->physicalPlanPoolMutex};
-        auto it = cachedStatement->physicalPlanPool.find(std::this_thread::get_id());
-        if (it != cachedStatement->physicalPlanPool.end() && !it->second.empty()) {
-            auto plan = std::move(it->second.back());
-            it->second.pop_back();
-            return plan;
-        }
-        // Opportunistic prune: drop empty per-thread entries left behind by exited
-        // threads or fully checked-out pools. Keeps the map proportional to the number
-        // of threads currently retaining plans.
-        for (auto pit = cachedStatement->physicalPlanPool.begin();
-             pit != cachedStatement->physicalPlanPool.end();) {
-            if (pit->second.empty()) {
-                pit = cachedStatement->physicalPlanPool.erase(pit);
-            } else {
-                ++pit;
+        if (maxPerThread == 0) {
+            // Pooling disabled: drop any retained plans so memory is released promptly
+            // and later executions cannot serve stale pooled state.
+            cachedStatement->physicalPlanPool.clear();
+        } else {
+            auto it = cachedStatement->physicalPlanPool.find(std::this_thread::get_id());
+            if (it != cachedStatement->physicalPlanPool.end() && !it->second.empty()) {
+                // Eagerly trim entries that exceed a recently lowered limit.
+                while (it->second.size() > maxPerThread) {
+                    it->second.pop_back();
+                }
+                if (!it->second.empty()) {
+                    auto plan = std::move(it->second.back());
+                    it->second.pop_back();
+                    return plan;
+                }
+            }
+            // Opportunistic prune: drop empty per-thread entries left behind by exited
+            // threads or fully checked-out pools. Keeps the map proportional to the number
+            // of threads currently retaining plans.
+            for (auto pit = cachedStatement->physicalPlanPool.begin();
+                 pit != cachedStatement->physicalPlanPool.end();) {
+                if (pit->second.empty()) {
+                    pit = cachedStatement->physicalPlanPool.erase(pit);
+                } else {
+                    ++pit;
+                }
             }
         }
     }
@@ -699,12 +712,15 @@ std::unique_ptr<processor::PhysicalPlan> checkoutPooledPhysicalPlan(
 }
 
 void recyclePooledPhysicalPlan(CachedPreparedStatement* cachedStatement,
-    std::unique_ptr<processor::PhysicalPlan> plan) {
+    std::unique_ptr<processor::PhysicalPlan> plan, uint64_t maxPerThread) {
+    if (maxPerThread == 0) {
+        return;
+    }
     std::lock_guard lock{cachedStatement->physicalPlanPoolMutex};
     auto tid = std::this_thread::get_id();
     auto it = cachedStatement->physicalPlanPool.find(tid);
     if (it != cachedStatement->physicalPlanPool.end()) {
-        if (it->second.size() < CachedPreparedStatement::PHYSICAL_PLAN_POOL_PER_THREAD) {
+        if (it->second.size() < maxPerThread) {
             it->second.push_back(std::move(plan));
         }
         return;
@@ -796,8 +812,10 @@ std::unique_ptr<QueryResult> ClientContext::executeNoLock(PreparedStatement* pre
                 if (cachedPlanAllowed && cachedStatement->physicalPlanCache) {
                     // Fast path: reuse a pooled plan when available, else clone the cached
                     // operator tree template. Avoids the PlanMapper::mapOperator recursion
-                    // and, on reuse, the per-execution clone entirely.
-                    physicalPlan = checkoutPooledPhysicalPlan(cachedStatement);
+                    // and, on reuse, the per-execution clone entirely. The pool depth is
+                    // the live `physical_plan_pool_per_thread` setting (0 disables reuse).
+                    const auto poolPerThread = getDBConfig()->physicalPlanPoolPerThread;
+                    physicalPlan = checkoutPooledPhysicalPlan(cachedStatement, poolPerThread);
                     physicalPlan->lastOperator->prepareForReuse(storage::MemoryManager::Get(*this));
                     // Re-attach the ResultSetDescriptors (not propagated by copy()) of every
                     // pipeline-head sink, so each sub-task can build its ResultSet.
@@ -846,8 +864,10 @@ std::unique_ptr<QueryResult> ClientContext::executeNoLock(PreparedStatement* pre
                 // Mapper-built plans are always fresh (and may carry baked parameters),
                 // so only pooled/template-cloned plans go back. Skipped on exception:
                 // the plan is then destroyed, and the next execution clones afresh.
+                // A zero `physical_plan_pool_per_thread` drops the plan instead.
                 if (pooledPlan) {
-                    recyclePooledPhysicalPlan(cachedStatement, std::move(physicalPlan));
+                    recyclePooledPhysicalPlan(cachedStatement, std::move(physicalPlan),
+                        getDBConfig()->physicalPlanPoolPerThread);
                 }
             },
             preparedStatement->isReadOnly(), isTransactionStatement,
diff --git a/src/main/database.cpp b/src/main/database.cpp
index ede6c3041..fca32b6a0 100644
--- a/src/main/database.cpp
+++ b/src/main/database.cpp
@@ -98,7 +98,7 @@ uint64_t getAvailableSystemMemory() {
 SystemConfig::SystemConfig(uint64_t bufferPoolSize_, uint64_t maxNumThreads, bool enableCompression,
     bool readOnly, uint64_t maxDBSize, bool autoCheckpoint, uint64_t checkpointThreshold,
     bool forceCheckpointOnClose, bool throwOnWalReplayFailure, bool enableChecksums,
-    bool enableMultiWrites, bool enableDefaultHashIndex
+    bool enableMultiWrites, bool enableDefaultHashIndex, uint64_t physicalPlanPoolPerThread
 #if defined(__APPLE__)
     ,
     uint32_t threadQos
@@ -108,7 +108,8 @@ SystemConfig::SystemConfig(uint64_t bufferPoolSize_, uint64_t maxNumThreads, boo
       autoCheckpoint{autoCheckpoint}, checkpointThreshold{checkpointThreshold},
       forceCheckpointOnClose{forceCheckpointOnClose},
       throwOnWalReplayFailure(throwOnWalReplayFailure), enableChecksums(enableChecksums),
-      enableMultiWrites{enableMultiWrites}, enableDefaultHashIndex{enableDefaultHashIndex} {
+      enableMultiWrites{enableMultiWrites}, enableDefaultHashIndex{enableDefaultHashIndex},
+      physicalPlanPoolPerThread{physicalPlanPoolPerThread} {
 #if defined(__APPLE__)
     this->threadQos = threadQos;
 #endif
diff --git a/src/main/db_config.cpp b/src/main/db_config.cpp
index fde975298..92c988109 100644
--- a/src/main/db_config.cpp
+++ b/src/main/db_config.cpp
@@ -27,7 +27,8 @@ static ConfigurationOption options[] = { // NOLINT(cert-err58-cpp):
     GET_CONFIGURATION(EnableInternalCatalogSetting),
     GET_CONFIGURATION(EnablePackedPathExtendSetting),
     GET_CONFIGURATION(EnableCachedPreparedStatementSetting),
-    GET_CONFIGURATION(EnableSubNodeGroupMorselsSetting)};
+    GET_CONFIGURATION(EnableSubNodeGroupMorselsSetting),
+    GET_CONFIGURATION(PhysicalPlanPoolPerThreadSetting)};
 
 DBConfig::DBConfig(const SystemConfig& systemConfig)
     : bufferPoolSize{systemConfig.bufferPoolSize}, maxNumThreads{systemConfig.maxNumThreads},
@@ -38,7 +39,8 @@ DBConfig::DBConfig(const SystemConfig& systemConfig)
       forceCheckpointOnClose{systemConfig.forceCheckpointOnClose},
       throwOnWalReplayFailure(systemConfig.throwOnWalReplayFailure),
       enableChecksums(systemConfig.enableChecksums),
-      enableDefaultHashIndex{systemConfig.enableDefaultHashIndex}, enableSpillingToDisk{true} {
+      enableDefaultHashIndex{systemConfig.enableDefaultHashIndex}, enableSpillingToDisk{true},
+      physicalPlanPoolPerThread{systemConfig.physicalPlanPoolPerThread} {
 #if defined(__APPLE__)
     this->threadQos = systemConfig.threadQos;
 #endif
diff --git a/src/main/settings.cpp b/src/main/settings.cpp
index 7c1901c43..3382027db 100644
--- a/src/main/settings.cpp
+++ b/src/main/settings.cpp
@@ -337,6 +337,21 @@ common::Value EnableSubNodeGroupMorselsSetting::getSetting(const ClientContext*
         context->getClientConfig()->subNodeGroupMorselScope));
 }
 
+void PhysicalPlanPoolPerThreadSetting::setContext(ClientContext* context,
+    const common::Value& parameter) {
+    parameter.validateType(inputType);
+    const auto value = parameter.getValue<int64_t>();
+    if (value < 0) {
+        throw common::RuntimeException(
+            std::format("physical_plan_pool_per_thread must be >= 0, got {}.", value));
+    }
+    context->getDBConfigUnsafe()->physicalPlanPoolPerThread = static_cast<uint64_t>(value);
+}
+
+common::Value PhysicalPlanPoolPerThreadSetting::getSetting(const ClientContext* context) {
+    return common::Value(static_cast<int64_t>(context->getDBConfig()->physicalPlanPoolPerThread));
+}
+
 void SpillToDiskSetting::setContext(ClientContext* context, const common::Value& parameter) {
     parameter.validateType(inputType);
     context->getDBConfigUnsafe()->enableSpillingToDisk = parameter.getValue<bool>();
diff --git a/test/api/system_config_test.cpp b/test/api/system_config_test.cpp
index 7c39d2de6..96beeec99 100644
--- a/test/api/system_config_test.cpp
+++ b/test/api/system_config_test.cpp
@@ -172,6 +172,42 @@ TEST_F(SystemConfigTest, testDisableDefaultHashIndexFromDBConfig) {
     ASSERT_EQ(nodeTable.tryGetPKIndex(), nullptr);
 }
 
+TEST_F(SystemConfigTest, testPhysicalPlanPoolPerThreadSetting) {
+    // Default comes from SystemConfig.
+    ASSERT_EQ(systemConfig->physicalPlanPoolPerThread, 5);
+    auto db = std::make_unique<Database>(databasePath, *systemConfig);
+    auto con = std::make_unique<Connection>(db.get());
+    auto result = con->query("CALL current_setting('physical_plan_pool_per_thread') RETURN *");
+    ASSERT_TRUE(result->isSuccess());
+    ASSERT_EQ(TestHelper::convertResultToString(*result), std::vector<std::string>{"5"});
+    // Runtime update via CALL; 0 disables pooled-plan reuse (fresh clone path).
+    assertQuery(*con->query("CALL physical_plan_pool_per_thread=0;"));
+    result = con->query("CALL current_setting('physical_plan_pool_per_thread') RETURN *");
+    ASSERT_TRUE(result->isSuccess());
+    ASSERT_EQ(TestHelper::convertResultToString(*result), std::vector<std::string>{"0"});
+    assertQuery(*con->query("CREATE NODE TABLE P(id INT64, PRIMARY KEY(id));"));
+    assertQuery(*con->query("UNWIND range(1, 10) AS i CREATE (:P {id: i});"));
+    auto prepared = con->prepare("MATCH (p:P) WHERE p.id = $x RETURN p.id;");
+    ASSERT_TRUE(prepared->isSuccess());
+    for (auto i = 1; i <= 10; ++i) {
+        auto r = con->execute(prepared.get(), std::make_pair(std::string("x"), (int64_t)i));
+        ASSERT_TRUE(r->isSuccess()) << r->toString();
+        ASSERT_EQ(TestHelper::convertResultToString(*r),
+            std::vector<std::string>{std::to_string(i)});
+    }
+    // Negative values are rejected.
+    auto bad = con->query("CALL physical_plan_pool_per_thread=-1;");
+    ASSERT_FALSE(bad->isSuccess());
+    // Restore pooling and verify the reuse path still returns correct results.
+    assertQuery(*con->query("CALL physical_plan_pool_per_thread=5;"));
+    for (auto i = 1; i <= 10; ++i) {
+        auto r = con->execute(prepared.get(), std::make_pair(std::string("x"), (int64_t)i));
+        ASSERT_TRUE(r->isSuccess()) << r->toString();
+        ASSERT_EQ(TestHelper::convertResultToString(*r),
+            std::vector<std::string>{std::to_string(i)});
+    }
+}
+
 // Regression: the per-thread reusable ResultSet used to live in a process-wide thread_local, so
 // on single-threaded builds (where the calling thread runs ProcessorTask::run, e.g. the
 // single-threaded Wasm package) it outlived Database::~Database and released (or reused)