perf(processor): remove per-execute mallocs on prepared-statement fast path

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();
 

src/include/storage/table/column_chunk_data.h

@@ -304,28 +304,25 @@ 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&)>;
-    flush_buffer_func_t initializeFlushBufferFunction(
-        std::shared_ptr<CompressionAlg> compression) const;
+    // 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,
+        StorageValue minValue, StorageValue maxValue) const;
     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 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;

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

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);
@@ -644,7 +651,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 +663,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 +758,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 +769,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 +817,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*/,

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) {

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);

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) {

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());

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());
     }

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];

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());
     }

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();

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());

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));

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) {

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);

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));

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);

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());
     }

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();

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());
     }

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());
     }

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());
     }