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

[adsharma]

Summary

Fixes #877.

Re-executing the same parameterized query string — the recommended execute(query, params) form, which reuses the cached physical plan — returned the first execution's rows whenever the plan contained a sort, top-k, join, cartesian product, OPTIONAL MATCH, UNION, EXISTS subquery, LIMIT/SKIP, or a recursive extend. The output was well-formed and no error was raised, so callers could not detect it.

Reproducer from the issue (three nodes with ids 1, 2, 3):

Q = "MATCH (a:N)-[:E]->(b:M) WHERE a.id = $v RETURN b.id"
await conn.execute(Q, {"v": 1})   # 1
await conn.execute(Q, {"v": 2})   # 1 — expected 2
await conn.execute(Q, {"v": 3})   # 1 — expected 3

shape (same three parameters)	before	after
single-table scan	[1, 2, 3]	[1, 2, 3]
ORDER BY / SKIP	[1, 1, 1] / no rows	[1, 2, 3]
traversal, OPTIONAL MATCH	[1, 1, 1]	[1, 2, 3]
LIMIT, UNION ALL	[1, None, None]	[1, 2, 3]

Root causes

Both are per-execution state surviving across executions of the cached operator tree — the same family as #841 and #870.

1. Whole sub-pipelines vanished from the cloned plan template

Several operator copy() implementations only cloned children[0] and dropped the children the plan mapper attaches afterward (build sides, sort sinks, union collectors). ProcessorTask::run() and the cached-plan fast path in ClientContext::executeNoLock() both clone through copy(), so the corresponding sink pipelines never ran again on later executions:

• OrderByScan / OrderByMerge / TopKScan — ORDER BY and top-k scans lost their sort sinks
• HashJoinProbe — dropped the build side (traversals, OPTIONAL MATCH, EXISTS, SIP-collector variants)
• Intersect, CrossProduct, PathPropertyProbe
• UnionAllScan — UNION / UNION ALL
• RecursiveExtend, TableFunctionCall (FTable scans), Profile, DummySimpleSink

Verified empirically: on the second execution of a top-k query the TOP_K sink task was never created at all — the sort was simply never re-run.

2. Shared states kept per-execution state that was never reset

• Limit / Skip counters stayed exhausted after the first execution → "first call right, later calls return no rows"
• HashJoinSharedState kept the previous execution's rows and hash slots (probe served stale rows)
• SortSharedState kept payload tables / sorted key blocks; strKeyColsInfo accumulated a duplicate entry per execution; KeyBlockMergeTaskDispatcher kept active merge tasks
• UnionAllScanSharedState kept the previous scan cursors → no rows after the first call
• SemiMaskerSharedState re-merged previous executions' local masks into the global node-offset masks (recursive extend)
• RecursiveExtendSharedState kept its limit counter and factorized-table pool contents
• ResultCollector: internal collectors (union branches, cross-product / accumulate / SIP builds) are only read by other operators of the same plan, which hold references to the same table object — so they must be cleared in place, never replaced. Replacing them with a fresh table (the Fix state reuse on the cached-physical-plan fast path (0-row scans, live-result clobbering) #870 use_count heuristic) left plan-internal readers (FTable scans, UnionAllScanSharedState, CrossProductLocalState) pointing at the stale table. Only the plan root collector — the one whose table is handed to the client via getQueryResult() — keeps the use_count-based fresh-table behavior for overlapping executions.

Note on parameter values: the cached-plan path already re-reads parameter values live at resolveResultVector time (Phase 1 work), so values propagate correctly once the plan structure and state resets are fixed.

Changes

• copy() implementations preserve all children
• Per-execution state is reset on hooks that run exactly once per execution:
  • initGlobalStateInternal(): Limit, Skip, HashJoinBuild, UnionAllScan, BaseSemiMasker, SortSharedState::init(), KeyBlockMergeTaskDispatcher::init()
  • prepareForReuse(): RecursiveExtend
• ResultCollector distinguishes internal vs client-facing result tables (flag propagated through copy())

The AsyncConnection.execute(PreparedStatement, ...) UnboundLocalError mentioned in the issue does not reproduce on current main (the conn_index handling in async_connection.py already covers it).

Testing

• New regression test ApiTest.RepeatedParameterizedCachedPlanExecution877 (covers ORDER BY, top-k, LIMIT, SKIP, joins, cartesian, OPTIONAL MATCH, UNION ALL, EXISTS, var-length extend, repeated through the cached-plan fast path). Fails on unpatched main, passes with this change.
• Full Python suite: 153 passed (includes test_async_prepare_and_execute_concurrent, the Fix state reuse on the cached-physical-plan fast path (0-row scans, live-result clobbering) #870 regression test)
• Full e2e suite: 1961/1962 — the single failure (dictionary_bug~orb383_relationship_projection_obfuscated.AnonymousParquetDeleteReload) was verified to fail identically on clean main
• Concurrency: repeated AsyncConnection pool stress with overlapping executions and held live results — no corruption, no crashes (an early draft of this change that didn't propagate the internal-table flag through copy() segfaulted here; the flag propagation fixes it)

[adsharma]

The test failures are pre-existing in main and tracked by #881