Skip to content

Streamline adaptive executor for single task queries#8607

Open
colm-mchugh wants to merge 9 commits into
mainfrom
colm/1-task-AE-take-2
Open

Streamline adaptive executor for single task queries#8607
colm-mchugh wants to merge 9 commits into
mainfrom
colm/1-task-AE-take-2

Conversation

@colm-mchugh

Copy link
Copy Markdown
Contributor

DESCRIPTION: New scan path for fast path queries that is a streamlined version of adaptive executor specialized for one task.

Introduces SingleTaskAdaptiveExecutor, a streamlined execution path for single-shard fast-path router queries that bypasses the full adaptive executor's WaitEventSet machinery, pool management, and slow-start logic.

Eligibility: fastPathRouterPlan && citus.enable_single_task_execution GUC (default on) && not multi-row INSERT && no dependent jobs && single-replicated table. EXPLAIN ANALYZE falls back to the regular AdaptiveExecutor at execution time.

Uses simple WaitLatchOrSocket polling on a single connection instead of the multi-connection WaitEventSet infrastructure.

Multi-replicated tables (shard_replication_factor > 1) are excluded because SingleTaskAdaptiveExecutor only uses the first placement, which would silently skip replica writes. This covers modifications to reference tables.

New GUC: citus.enable_single_task_execution (PGC_USERSET, default true)
New executor type: MULTI_EXECUTOR_SINGLE_TASK
New scan name: "Citus Single Task"

@colm-mchugh
colm-mchugh marked this pull request as draft June 4, 2026 08:39
@codecov

codecov Bot commented Jun 4, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 84.77612% with 51 lines in your changes missing coverage. Please review.
✅ Project coverage is 88.71%. Comparing base (f91592f) to head (812fb54).

Additional details and impacted files
@@            Coverage Diff             @@
##             main    #8607      +/-   ##
==========================================
- Coverage   88.73%   88.71%   -0.02%     
==========================================
  Files         288      288              
  Lines       64428    64725     +297     
  Branches     8113     8169      +56     
==========================================
+ Hits        57167    57424     +257     
- Misses       4917     4941      +24     
- Partials     2344     2360      +16     
🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@colm-mchugh
colm-mchugh force-pushed the colm/1-task-AE-take-2 branch 5 times, most recently from 295095c to dfdc70b Compare June 4, 2026 13:14
@colm-mchugh colm-mchugh self-assigned this Jun 4, 2026
@colm-mchugh
colm-mchugh force-pushed the colm/1-task-AE-take-2 branch 3 times, most recently from 894f941 to db3dd8a Compare June 5, 2026 15:03
if (PQisBusy(connection->pgConn))
{
int sock = PQsocket(connection->pgConn);
int rc = WaitLatchOrSocket(MyLatch,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Doesn't this create a wait event set in every call? If so, what's the gain here? Seems like a lot of code to maintain.

@colm-mchugh colm-mchugh Jun 8, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it creates a WaitEventSet per call. This path doesn't fire when results are buffered on entering the loop. But for large result sets it could accumulate overhead so will likely change to create lazily, when the socket is not immediately readable.

what's the gain here?

The gain comes from bypassing the pool/session/connection-state-machine infrastructure, not the wait mechanics. Running pgbench on CPU-bound workload shows TPS gain of 15% (SELECT), 12% (INSERT) and 12% (UPDATE) so for an in-memory OLTP workload there's a substantial gain.

@colm-mchugh
colm-mchugh force-pushed the colm/1-task-AE-take-2 branch 2 times, most recently from febc9ac to e1be8b7 Compare June 8, 2026 11:36
@colm-mchugh
colm-mchugh marked this pull request as ready for review June 8, 2026 12:45
@colm-mchugh
colm-mchugh requested a review from Copilot June 9, 2026 15:08

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

This PR introduces a new streamlined execution path (“SingleTaskAdaptiveExecutor”) for eligible single-shard fast-path router queries, aiming to reduce overhead by bypassing the full adaptive executor machinery while preserving EXPLAIN visibility and maintaining correctness via eligibility checks and fallbacks.

Changes:

  • Add a new executor type (MULTI_EXECUTOR_SINGLE_TASK) plus custom scan plumbing to run a dedicated single-task execution path for eligible fast-path router queries.
  • Introduce a new GUC (citus.enable_single_task_execution, default on) and update plan selection logic to choose the single-task executor when safe.
  • Expand regression coverage with a new single_task_execution test, update multiple expected files for the new scan name, and add normalization to reduce output flakiness.

Reviewed changes

Copilot reviewed 43 out of 45 changed files in this pull request and generated 1 comment.

Show a summary per file
File Description
src/backend/distributed/executor/adaptive_executor.c Implements SingleTaskExecutor() and supporting logic for single-connection polling and result ingestion.
src/backend/distributed/executor/citus_custom_scan.c Registers the new custom scan methods and routes execution to SingleTaskExecutor (or falls back for EXPLAIN ANALYZE).
src/backend/distributed/executor/multi_server_executor.c Chooses MULTI_EXECUTOR_SINGLE_TASK for eligible fast-path router plans (with replication/eligibility checks).
src/backend/distributed/planner/distributed_planner.c Wires MULTI_EXECUTOR_SINGLE_TASK to the new custom scan methods during plan finalization.
src/backend/distributed/shared_library_init.c Adds the citus.enable_single_task_execution GUC definition.
src/include/distributed/adaptive_executor.h Exposes the SingleTaskExecutor(CitusScanState *) entry point.
src/include/distributed/citus_custom_scan.h Declares SingleTaskExecutorCustomScanMethods.
src/include/distributed/multi_server_executor.h Adds MULTI_EXECUTOR_SINGLE_TASK and the EnableSingleTaskExecution GUC variable.
src/test/regress/bin/normalize.sed Normalizes repartition-related NOTICE output (hash boundaries, suffixes, shard id ranges) to reduce flakiness.
src/test/regress/citus_tests/run_test.py Adjusts test deps (adds multi_transaction_recovery to minimal_schedule).
src/test/regress/multi_1_schedule Schedules the new single_task_execution test in the non-parallelized block with other router prepared-statement tests.
src/test/regress/sql/single_task_execution.sql New regression test validating explain output and result equivalence with the GUC toggled.
src/test/regress/sql/multi_transaction_recovery.sql Makes the test more robust (cleanup + dynamic group IDs) and disables single-task exec for predictable 2PC behavior.
src/test/regress/sql/multi_router_planner_fast_path.sql Forces custom plans for prepared statements to avoid flaky debug output; resets afterward.
src/test/regress/sql/local_shard_execution.sql Makes colocations explicit; disables single-task for predictable error messages; wraps flaky error cases in DO/EXCEPTION checks.
src/test/regress/sql/failure_single_select.sql Disables single-task exec for predictable error behavior in failure tests.
src/test/regress/sql/failure_savepoints.sql Disables single-task exec for predictable error behavior in failure tests.
src/test/regress/sql/failure_multi_dml.sql Disables single-task exec for predictable error behavior; ensures worker-side script also disables it.
src/test/regress/sql/failure_connection_establishment.sql Disables single-task exec for predictable error behavior in connection-establishment tests.
src/test/regress/expected/single_task_execution.out New expected output file for the new regression test.
src/test/regress/expected/upgrade_basic_after.out Updates expected EXPLAIN scan name for single-task-eligible plans.
src/test/regress/expected/upgrade_basic_after_0.out Adds an additional expected output variant for upgrade coverage.
src/test/regress/expected/sqlancer_failures.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/multi_router_planner_fast_path.out Updates expected debug output + plan_cache_mode SET/RESET lines.
src/test/regress/expected/multi_transaction_recovery.out Updates expected output for cleanup, dynamic group IDs, and the single-task disable.
src/test/regress/expected/local_shard_execution_replicated.out Updates expected NOTICE output after normalization changes.
src/test/regress/expected/failure_single_select.out Mirrors the SQL change disabling single-task exec.
src/test/regress/expected/failure_savepoints.out Mirrors the SQL change disabling single-task exec.
src/test/regress/expected/failure_multi_dml.out Mirrors the SQL change disabling single-task exec; expected worker script output changes.
src/test/regress/expected/failure_connection_establishment.out Mirrors the SQL change disabling single-task exec.
src/test/regress/expected/multi_mx_insert_select_repartition.out Updates expected NOTICE output after normalization changes.
src/test/regress/expected/multi_mx_explain.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/multi_mx_explain_0.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/multi_explain.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/multi_explain_0.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/multi_tenant_isolation.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/multi_tenant_isolation_nonblocking.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/pg13.out Updates expected EXPLAIN/ANALYZE output formatting and scan name for single-task-eligible plan(s).
src/test/regress/expected/pg15.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/pg18.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable; aligns formatting for pg_get_loaded_modules output.
src/test/regress/expected/multi_data_types.out Updates expected EXPLAIN/ANALYZE scan name + spacing for single-task-eligible DML.
src/test/regress/expected/drop_column_partitioned_table.out Updates expected EXPLAIN scan name to “Citus Single Task” where applicable.
src/test/regress/expected/coordinator_shouldhaveshards.out Updates expected NOTICE output after normalization changes.
src/test/regress/expected/single_node.out Updates expected NOTICE output after normalization changes.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Comment on lines +122 to +126
Oid relationId = distributedPlan->targetRelationId;
if (!OidIsValid(relationId) && list_length(job->taskList) == 1)
{
Task *task = (Task *) linitial(job->taskList);
if (list_length(task->relationShardList) > 0)

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The !OidIsValid(relationId) check covers SELECT queries, which always have targetRelationId = InvalidOid. Reading from a single placement is safe regardless of replication factor. For DML, targetRelationId is always set, so the SingleReplicatedTable() check correctly routes multi-replicated tables to the adaptive executor for write replication.

@colm-mchugh
colm-mchugh force-pushed the colm/1-task-AE-take-2 branch 2 times, most recently from 0764a33 to 67fdd57 Compare July 6, 2026 13:06

@onurctirtir onurctirtir left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This review was produced by Claude Opus 4.8 (model ID claude-opus-4.8), driven by @onurctirtir.

Every finding below was reproduced empirically on an isolated 2-worker Citus cluster built from this PR branch (colm/1-task-AE-take-2).

Methodology: read the STE path against the README + the adaptive executor it forks from, then tried to break it — focusing on the control-flow the new SingleTaskExecutorStart fast path skips relative to the full adaptive executor.

Comment thread src/backend/distributed/executor/adaptive_executor.c
Comment thread src/backend/distributed/executor/adaptive_executor.c
Comment thread src/backend/distributed/executor/adaptive_executor.c Outdated
Comment thread src/backend/distributed/executor/adaptive_executor.c
Comment thread src/backend/distributed/executor/adaptive_executor.c Outdated
Comment thread src/test/regress/sql/single_task_execution.sql
@ihalatci ihalatci added this to the 14.2 & 13.4 milestone Jul 8, 2026
Comment thread src/backend/distributed/executor/citus_custom_scan.c Outdated
Comment thread src/test/regress/sql/failure_single_select.sql Outdated
Comment thread src/test/regress/sql/failure_connection_establishment.sql
Comment thread src/test/regress/sql/failure_savepoints.sql Outdated
Comment thread src/test/regress/sql/failure_multi_dml.sql Outdated
Comment thread src/backend/distributed/executor/adaptive_executor.c Outdated
Comment thread src/test/regress/sql/single_task_execution.sql
Comment thread src/backend/distributed/executor/multi_server_executor.c
Comment thread src/backend/distributed/executor/adaptive_executor.c Outdated
Comment thread src/backend/distributed/executor/adaptive_executor.c
@colm-mchugh
colm-mchugh force-pushed the colm/1-task-AE-take-2 branch 3 times, most recently from b542c70 to 36a778c Compare July 14, 2026 11:49
…path queries

Introduces SingleTaskAdaptiveExecutor, a streamlined execution path for
single-shard fast-path router queries that bypasses the full adaptive
executor's WaitEventSet machinery, pool management, and slow-start logic.

Eligibility: fastPathRouterPlan && citus.enable_single_task_execution GUC
(default on) && not multi-row INSERT && no dependent jobs && single-replicated
table. EXPLAIN ANALYZE falls back to the regular AdaptiveExecutor at execution
time.

Uses simple WaitLatchOrSocket polling on a single connection instead of
the multi-connection WaitEventSet infrastructure.

Multi-replicated tables (shard_replication_factor > 1) are excluded
because SingleTaskAdaptiveExecutor only uses the first placement, which
would silently skip replica writes. This covers modifications to
reference tables.

New GUC: citus.enable_single_task_execution (PGC_USERSET, default true)
New executor type: MULTI_EXECUTOR_ONE_TASK_ADAPTIVE
New scan name: "Citus Single-Task Adaptive"

Self Review fixes: guard stat counter, document SortTupleStore omission

and fix connection error text - use libpq error messages everywhere.
Also fix a parameter handling bug; `MarkUnreferencedExternParams()`
modified the params list which was then passed to local execution,
which expects unmodified params. Fixed by using unmodified params
for local execution.
Fix local_shard_execution and multi_transaction_recovery to pass reliably
under both standalone (flaky) and full-schedule CI runs.

local_shard_execution:
  Shard-to-worker placement varies depending on which cluster setup
  schedule runs before the test (minimal_cluster_management vs
  multi_cluster_management assign different group IDs, which changes
  round-robin shard placement). Add a normalize.sed rule to mask shard
  IDs in the 1470xxx range so the comparison succeeds regardless of
  placement order. Also use colocate_with => 'none' to prevent
  colocation group inheritance from prior tests.

multi_transaction_recovery:
  The test inserted fake pg_dist_transaction records using hardcoded
  group IDs (0 and 1), which only matched the multi_cluster_management
  setup. When the flaky runner uses minimal_cluster_management (group
  IDs 14 and 16), recover_prepared_transactions() could not find the
  matching nodes and left stale records. Fix by looking up the actual
  worker_1 group ID dynamically. Also DELETE FROM pg_dist_transaction
  at the start to clear residual records from prior schedule tests,
  and add a TestDeps entry so run_test.py uses minimal_schedule.
Refactor `SingleTaskExecutor` from a monolithic function into three phases:
- `SingleTaskExecutorStart()`: connection setup, query dispatch
- `SingleTaskExecutorRun()`: fetch up to maxBatchSize rows per call
- `SingleTaskExecutorEnd()`: cancel in-flight queries (LIMIT), cleanup

This enables batched execution for the STE path, matching the adaptive
executor's batching support. The STE can now stop fetching when a LIMIT
is satisfied.

Also change WaitEventSetWait timeout from 0 (non-blocking poll) to -1
(block until event). With a single connection there is nothing to
multiplex, so blocking is correct. The spin-wait at timeout=0 caused
severe CPU contention at high concurrency eliminating throughput
gains from the STE code path.
- Fix stat-counter undercounting
- Align connection-loss error messages with the Adaptive Executor (AE)
- Enforce max_shared_pool_size (WAIT_FOR_CONNECTION)
- Disable local execution after a remote modification to the local node
- Retry dead cached connections in a loop instead of a one-shot probe
- Failure tests: exercise STE in failure_single_select; keep AE for the
  savepoint / multi-DML failure tests (STE's in-transaction connection-loss
  error text is timing-dependent)
- Add STE/AE parity tests in single_task_execution: stat + connection
  counters, statement_timeout, SET LOCAL propagation, local/remote logging;
  plus dead-cached-connection reconnect coverage (single_task_execution,
  detect_conn_close)
- Refactor Start/Run into helpers (AcquireLiveConnectionForSingleTask,
  RecordSingleTaskModification)
@colm-mchugh
colm-mchugh force-pushed the colm/1-task-AE-take-2 branch from 36a778c to 812fb54 Compare July 15, 2026 08:46
@ihalatci ihalatci added cherry-pick-13.2 The tag to be removed once the PR is cherry-picked to 13.2 rel branch. cherry-pick-14 labels Jul 16, 2026

@codeforall codeforall left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No correctness concerns from me. Just a few nits, all take-it-or-leave-it:

  • The STE path sends only TaskQueryStringAtIndex(task, 0) with no loop over task->queryCount. It's safe today because every multi-query shape (multi-row INSERT, modifying CTEs, etc.) is excluded upstream, and I couldn't get a queryCount > 1 task onto the path in testing , but nothing enforces it, and if a future fast-path change ever slipped one through, the extra queries would be silently dropped. A one-line Assert(task->queryCount == 1) in SingleTaskExecutorStart would lock the invariant cheaply.

  • Two small test additions would be nice since I had to build a cluster to confirm the behavior: a prepared-statement param case (the deferred-pruning path Copilot asked about) and an EXPLAIN ANALYZE counter-is-exactly-one assertion. Both already work; it'd just pin them.

  • A tiny thing, the refactor dropped the old XXX: SendRemoteCommandParams … isolation_select_vs_all.spec breadcrumb in SendNextQuery, might be worth keeping in SendQueryStringToConnection.

Nothing blocking here, LGTM. Thanks for the thorough comments and the strong test coverage, made this an easy read.

Copilot AI review requested due to automatic review settings July 22, 2026 15:35

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 43 out of 45 changed files in this pull request and generated 2 comments.

Comment on lines +1871 to +1875
ste->waitEventSet = CreateWaitEventSet(WaitEventSetTracker_compat, 3);
AddWaitEventToSet(ste->waitEventSet, WL_LATCH_SET,
PGINVALID_SOCKET, MyLatch, NULL);
AddWaitEventToSet(ste->waitEventSet, WL_EXIT_ON_PM_DEATH,
PGINVALID_SOCKET, NULL, NULL);
{
CitusScanState *scanState = palloc0(sizeof(CitusScanState));

/* reuse MULTI_EXECUTOR_ADAPTIVE for stats bucketing (see Research §6.1) */
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

cherry-pick-13.2 The tag to be removed once the PR is cherry-picked to 13.2 rel branch. cherry-pick-14

Projects

None yet

Development

Successfully merging this pull request may close these issues.

6 participants