fix: re-executed parameterized queries returned stale rows on the cached-plan fast path (#877)

src/include/common/counter.h

@@ -17,6 +17,9 @@ class LimitCounter {
 
     bool exceedLimit() const { return counter.load() >= limitNumber; }
 
+    // Re-arm the counter for another execution of a cached physical plan.
+    void reset() { counter.store(0); }
+
 private:
     common::offset_t limitNumber;
     std::atomic<common::offset_t> counter;

src/include/main/client_config.h

@@ -8,6 +8,24 @@
 namespace lbug {
 namespace main {
 
+// Scope of statements for which the cached-physical-plan fast path (see
+// CachedPreparedStatement::physicalPlanCache) may be used when a parameterized prepared
+// statement is re-executed. This is a safety valve for latent state-reuse bugs in the plan
+// cache: setting it to READS, WRITES or NONE trades the optimization for protection.
+enum class CachedPreparedStatementScope {
+    READS = 0,  // cache and reuse plans of read-only statements only
+    WRITES = 1, // cache and reuse plans of write statements only
+    BOTH = 2,   // cache and reuse plans of reads and writes
+    NONE = 3,   // disable plan caching entirely (re-map the plan on every execution)
+};
+
+struct CachedPreparedStatementScopeUtils {
+    // Parses one of [READS, WRITES, BOTH, NONE] (case-insensitive); throws a
+    // RuntimeException otherwise. Defined in settings.cpp.
+    static CachedPreparedStatementScope fromString(const std::string& str);
+    static std::string toString(CachedPreparedStatementScope scope);
+};
+
 struct ClientConfigDefault {
     // 0 means timeout is disabled by default.
     static constexpr uint64_t TIMEOUT_IN_MS = 0;
@@ -24,6 +42,10 @@ struct ClientConfigDefault {
     static constexpr bool ENABLE_PLAN_OPTIMIZER = true;
     static constexpr bool ENABLE_INTERNAL_CATALOG = false;
     static constexpr bool ENABLE_PACKED_PATH_EXTEND = false;
+    // Statement kinds for which the cached-physical-plan fast path is enabled.
+    // BOTH preserves the historical behaviour (all parameterized statements).
+    static constexpr CachedPreparedStatementScope CACHED_PREPARED_STATEMENT_SCOPE =
+        CachedPreparedStatementScope::BOTH;
     // Memory budget (in bytes) for the in-memory primary-key uniqueness buffer used when COPY-ing
     // into a primary-key node table that has no hash index. Once the buffer exceeds this budget it
     // is sorted and spilled to disk as a sorted run; cross-run duplicates are detected during a
@@ -69,6 +91,10 @@ struct ClientConfig {
     // Memory budget (bytes) for the no-hash-index COPY primary-key validator before it spills
     // sorted runs to disk. See ClientConfigDefault::PK_VALIDATOR_SPILL_THRESHOLD.
     uint64_t pkValidatorSpillThreshold = ClientConfigDefault::PK_VALIDATOR_SPILL_THRESHOLD;
+    // Which statement kinds may reuse a cached physical plan when a parameterized prepared
+    // statement is re-executed. See CachedPreparedStatementScope for the safety rationale.
+    CachedPreparedStatementScope cachedPreparedStatementScope =
+        ClientConfigDefault::CACHED_PREPARED_STATEMENT_SCOPE;
 };
 
 } // namespace main

src/include/main/client_context.h

@@ -226,6 +226,9 @@ class LBUG_API ClientContext {
         CachedPreparedStatement* cachedPreparedStatement,
         std::optional<uint64_t> queryID = std::nullopt, QueryConfig config = {},
         bool cachePhysicalPlan = false);
+    // Whether the cached-physical-plan fast path may be used for the given statement, per the
+    // `enable_cached_prepared_statement` setting.
+    bool isCachedPlanAllowedFor(const PreparedStatement& preparedStatement) const;
     std::unique_ptr<QueryResult> queryNoLock(std::string_view query,
         std::optional<uint64_t> queryID = std::nullopt, QueryConfig config = {});
 

src/include/main/settings.h

@@ -166,5 +166,15 @@ struct EnablePackedPathExtendSetting {
     static common::Value getSetting(const ClientContext* context);
 };
 
+struct EnableCachedPreparedStatementSetting {
+    static constexpr auto name = "enable_cached_prepared_statement";
+    static constexpr auto inputType = common::LogicalTypeID::STRING;
+    // One of [READS, WRITES, BOTH, NONE] (case-insensitive): which statement kinds may
+    // reuse a cached physical plan when a parameterized prepared statement is re-executed.
+    // Safety valve for latent plan-cache state-reuse bugs; NONE disables the optimization.
+    static void setContext(ClientContext* context, const common::Value& parameter);
+    static common::Value getSetting(const ClientContext* context);
+};
+
 } // namespace main
 } // namespace lbug

src/include/processor/operator/cross_product.h

@@ -50,8 +50,12 @@ class CrossProduct final : public PhysicalOperator {
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<CrossProduct>(info.copy(), localState.copy(), children[0]->copy(),
-            id, printInfo->copy());
+        auto result = std::make_unique<CrossProduct>(info.copy(), localState.copy(),
+            children[0]->copy(), id, printInfo->copy());
+        for (auto i = 1u; i < children.size(); ++i) {
+            result->addChild(children[i]->copy());
+        }
+        return result;
     }
 
 private:

src/include/processor/operator/hash_join/hash_join_build.h

@@ -47,6 +47,11 @@ class HashJoinSharedState {
 
     JoinHashTable* getHashTable() { return hashTable.get(); }
 
+    // Re-arm the shared state for another execution of a cached physical plan: drop the rows
+    // and hash slots accumulated by the previous execution. Without this, later executions
+    // probe stale rows from earlier executions.
+    void resetForReuse() { hashTable->resetForReuse(); }
+
 protected:
     std::mutex mtx;
     std::unique_ptr<JoinHashTable> hashTable;
@@ -83,6 +88,8 @@ class HashJoinBuild : public Sink {
 
     std::shared_ptr<HashJoinSharedState> getSharedState() const { return sharedState; }
 
+    void initGlobalStateInternal(ExecutionContext* context) override;
+
     void initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) override;
 
     void executeInternal(ExecutionContext* context) override;

src/include/processor/operator/hash_join/hash_join_probe.h

@@ -79,8 +79,12 @@ class HashJoinProbe : public PhysicalOperator, public SelVectorOverWriter {
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return make_unique<HashJoinProbe>(sharedState, joinType, flatProbe, probeDataInfo,
+        auto result = make_unique<HashJoinProbe>(sharedState, joinType, flatProbe, probeDataInfo,
             children[0]->copy(), id, printInfo->copy());
+        for (auto i = 1u; i < children.size(); ++i) {
+            result->addChild(children[i]->copy());
+        }
+        return result;
     }
 
 private:

src/include/processor/operator/hash_join/join_hash_table.h

@@ -43,6 +43,14 @@ class JoinHashTable : public BaseHashTable {
         factorizedTable->lookup(vectors, colIdxesToScan, tuplesToRead, startPos, numTuplesToRead);
     }
     void merge(JoinHashTable& other) { factorizedTable->merge(*other.factorizedTable); }
+    // Drop all entries and hash slots so the table can be re-built on the next execution of a
+    // cached physical plan. Nothing outside the owning HashJoinSharedState references the
+    // factorized table, so clearing in place is safe.
+    void resetForReuse() {
+        factorizedTable->clear();
+        hashSlotsBlocks.clear();
+        maxNumHashSlots = 0;
+    }
     uint8_t** getPrevTuple(const uint8_t* tuple) const {
         return (uint8_t**)(tuple + prevPtrColOffset);
     }

src/include/processor/operator/intersect/intersect.h

@@ -47,8 +47,12 @@ class Intersect : public PhysicalOperator {
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<Intersect>(outputDataPos, intersectDataInfos, sharedHTs,
+        auto result = std::make_unique<Intersect>(outputDataPos, intersectDataInfos, sharedHTs,
             children[0]->copy(), id, printInfo->copy());
+        for (auto i = 1u; i < children.size(); ++i) {
+            result->addChild(children[i]->copy());
+        }
+        return result;
     }
 
 private:

src/include/processor/operator/limit.h

@@ -37,6 +37,12 @@ class Limit final : public PhysicalOperator {
 
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
+    void initGlobalStateInternal(ExecutionContext* /*context*/) override {
+        // Runs once per execution. Reset the shared counter so a re-executed cached plan
+        // starts counting from zero instead of staying exhausted at limitNumber.
+        counter->store(0);
+    }
+
     std::unique_ptr<PhysicalOperator> copy() override {
         return make_unique<Limit>(limitNumber, counter, dataChunkToSelectPos, dataChunksPosInScope,
             children[0]->copy(), id, printInfo->copy());

src/include/processor/operator/order_by/order_by_merge.h

@@ -25,7 +25,10 @@ class OrderByMerge final : public Sink {
     void executeInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<OrderByMerge>(sharedState, sharedDispatcher, id, printInfo->copy());
+        auto result =
+            std::make_unique<OrderByMerge>(sharedState, sharedDispatcher, id, printInfo->copy());
+        result->addChild(children[0]->copy());
+        return result;
     }
 
 private:

src/include/processor/operator/order_by/order_by_scan.h

@@ -44,7 +44,10 @@ class OrderByScan final : public PhysicalOperator {
     void initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<OrderByScan>(outVectorPos, sharedState, id, printInfo->copy());
+        auto result =
+            std::make_unique<OrderByScan>(outVectorPos, sharedState, id, printInfo->copy());
+        result->addChild(children[0]->copy());
+        return result;
     }
 
     double getProgress(ExecutionContext* context) const override;

src/include/processor/operator/order_by/top_k_scanner.h

@@ -34,7 +34,9 @@ class TopKScan final : public PhysicalOperator {
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<TopKScan>(outVectorPos, sharedState, id, printInfo->copy());
+        auto result = std::make_unique<TopKScan>(outVectorPos, sharedState, id, printInfo->copy());
+        result->addChild(children[0]->copy());
+        return result;
     }
 
 private:

src/include/processor/operator/path_property_probe.h

@@ -85,8 +85,12 @@ class PathPropertyProbe : public PhysicalOperator {
     bool getNextTuplesInternal(ExecutionContext* context) final;
 
     std::unique_ptr<PhysicalOperator> copy() final {
-        return std::make_unique<PathPropertyProbe>(info.copy(), sharedState, children[0]->copy(),
-            id, printInfo->copy());
+        auto result = std::make_unique<PathPropertyProbe>(info.copy(), sharedState,
+            children[0]->copy(), id, printInfo->copy());
+        for (auto i = 1u; i < children.size(); ++i) {
+            result->addChild(children[i]->copy());
+        }
+        return result;
     }
 
 private:

src/include/processor/operator/profile.h

@@ -23,7 +23,11 @@ class Profile final : public SimpleSink {
     void executeInternal(ExecutionContext* context) override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<Profile>(info, messageTable, id, printInfo->copy());
+        auto result = std::make_unique<Profile>(info, messageTable, id, printInfo->copy());
+        for (auto& child : children) {
+            result->addChild(child->copy());
+        }
+        return result;
     }
 
 private:

src/include/processor/operator/recursive_extend.h

@@ -40,9 +40,22 @@ class RecursiveExtend : public Sink {
 
     void executeInternal(ExecutionContext* context) override;
 
+    // The reset must happen in prepareForReuse() rather than initGlobalStateInternal(): this
+    // operator's task only starts after the semi-masker child pipeline has filled the node
+    // offset masks, so resetting there would run too late. prepareForReuse() runs on the whole
+    // plan before any task of the new execution starts.
+    void prepareForReuse(storage::MemoryManager* memoryManager) override {
+        sharedState->resetForReuse();
+        PhysicalOperator::prepareForReuse(memoryManager);
+    }
+
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<RecursiveExtend>(function->copy(), bindData, sharedState, id,
+        auto result = std::make_unique<RecursiveExtend>(function->copy(), bindData, sharedState, id,
             printInfo->copy());
+        for (auto& child : children) {
+            result->addChild(child->copy());
+        }
+        return result;
     }
 
 private:

src/include/processor/operator/recursive_extend_shared_state.h

@@ -20,6 +20,17 @@ struct RecursiveExtendSharedState {
         }
     }
 
+    // Re-arm the state for another execution of a cached physical plan: drop the rows
+    // accumulated in the factorized table pool and the limit counter of the previous
+    // execution. The node offset masks don't need resetting here: the semi masker that fills
+    // them replaces their contents wholesale on every execution.
+    void resetForReuse() {
+        if (counter != nullptr) {
+            counter->reset();
+        }
+        factorizedTablePool.resetForReuse();
+    }
+
     void setInputNodeMask(std::unique_ptr<common::NodeOffsetMaskMap> maskMap) {
         inputNodeMask = std::move(maskMap);
     }

src/include/processor/operator/result_collector.h

@@ -87,9 +87,18 @@ class ResultCollector final : public Sink {
 
     std::unique_ptr<main::QueryResult> getQueryResult() const override;
 
+    // Marks this collector as the statement's result collector, i.e. the plan root whose
+    // FactorizedTable is handed to the client via getQueryResult(). Collectors created for
+    // intermediate pipelines (union branches, cross-product/accumulate/SIP builds) default to
+    // internal: their tables are only read by other operators of the same plan, so reuse can
+    // clear them in place even though the plan itself keeps extra references to them.
+    void setResultExposedToClient() { internalResultTable = false; }
+
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<ResultCollector>(info.copy(), sharedState, children[0]->copy(), id,
-            printInfo->copy());
+        auto result = std::make_unique<ResultCollector>(info.copy(), sharedState,
+            children[0]->copy(), id, printInfo->copy());
+        result->internalResultTable = internalResultTable;
+        return result;
     }
 
 private:
@@ -100,6 +109,7 @@ class ResultCollector final : public Sink {
 private:
     ResultCollectorInfo info;
     std::shared_ptr<ResultCollectorSharedState> sharedState;
+    bool internalResultTable = true;
     std::vector<common::ValueVector*> payloadVectors;
     std::vector<common::ValueVector*> payloadAndMarkVectors;
 

src/include/processor/operator/semi_masker.h

@@ -32,6 +32,12 @@ class SemiMaskerSharedState {
 
     void mergeToGlobal();
 
+    // Re-arm the state for another execution of a cached physical plan. Local states (and the
+    // masks they fill) are recreated by every worker thread via appendLocalState(); keeping
+    // entries from a previous execution would re-merge that execution's node offsets into the
+    // global masks.
+    void resetForReuse() { localInfos.clear(); }
+
 private:
     common::table_id_map_t<std::vector<common::SemiMask*>> masksPerTable;
     std::vector<std::shared_ptr<SemiMaskerLocalState>> localInfos;
@@ -65,6 +71,12 @@ class BaseSemiMasker : public PhysicalOperator {
         : PhysicalOperator{type_, std::move(child), id, std::move(printInfo)}, keyPos{keyPos},
           keyVector{nullptr}, sharedState{std::move(sharedState)}, localState{nullptr} {}
 
+    void initGlobalStateInternal(ExecutionContext* /*context*/) override {
+        // Runs once per execution, before worker threads call initLocalStateInternal(). Drop
+        // local mask states left over from the previous execution of a cached plan.
+        sharedState->resetForReuse();
+    }
+
     void initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) override;
 
     void finalizeInternal(ExecutionContext* context) final;

src/include/processor/operator/sink.h

@@ -116,7 +116,11 @@ class DummySimpleSink final : public SimpleSink {
     void executeInternal(ExecutionContext*) override {}
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<DummySimpleSink>(messageTable, id);
+        auto result = std::make_unique<DummySimpleSink>(messageTable, id);
+        for (auto& child : children) {
+            result->addChild(child->copy());
+        }
+        return result;
     }
 };
 

src/include/processor/operator/skip.h

@@ -39,6 +39,12 @@ class Skip final : public PhysicalOperator, public SelVectorOverWriter {
 
     bool getNextTuplesInternal(ExecutionContext* context) override;
 
+    void initGlobalStateInternal(ExecutionContext* /*context*/) override {
+        // Runs once per execution. Reset the shared counter so a re-executed cached plan
+        // starts skipping from zero again.
+        counter->store(0);
+    }
+
     std::unique_ptr<PhysicalOperator> copy() override {
         return make_unique<Skip>(skipNumber, counter, dataChunkToSelectPos, dataChunksPosInScope,
             children[0]->copy(), id, printInfo->copy());

src/include/processor/operator/table_function_call.h

@@ -74,7 +74,12 @@ class LBUG_API TableFunctionCall final : public PhysicalOperator {
     double getProgress(ExecutionContext* context) const override;
 
     std::unique_ptr<PhysicalOperator> copy() override {
-        return std::make_unique<TableFunctionCall>(info.copy(), sharedState, id, printInfo->copy());
+        auto result =
+            std::make_unique<TableFunctionCall>(info.copy(), sharedState, id, printInfo->copy());
+        for (auto& child : children) {
+            result->addChild(child->copy());
+        }
+        return result;
     }
 
 private: