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/include/storage/table/column_chunk_data.h b/src/include/storage/table/column_chunk_data.h
index 38b283336..acc851c99 100644
--- a/src/include/storage/table/column_chunk_data.h
+++ b/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;
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/main/client_context.cpp b/src/main/client_context.cpp
index 6b21252e9..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);
@@ -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*/,
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/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/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/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/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/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/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/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/storage/table/column_chunk_data.cpp b/src/storage/table/column_chunk_data.cpp
index cd48c84b0..fded2466e 100644
--- a/src/storage/table/column_chunk_data.cpp
+++ b/src/storage/table/column_chunk_data.cpp
@@ -181,18 +181,21 @@ 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);
 }
 
-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 +211,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);
     }
     }
 }
@@ -363,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) {
@@ -415,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;
@@ -1142,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 fa4980c9b..09445b81a 100644
--- a/src/storage/table/rel_table.cpp
+++ b/src/storage/table/rel_table.cpp
@@ -220,13 +220,29 @@ 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()) {
+        // 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;
+    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 +251,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 +310,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);
 }