Skip to main content

Distributed queries

By the end of this page you will know what actually happens when a query touches data spread across more than one node, which query shapes stay cheap and which force data across the network, and how to see what a query actually did.

Connecting​

Connect to any node, exactly as described in Cluster overview: there is no separate coordinator, router, or proxy process to find or configure. Whichever node you connect to coordinates your statement: it plans it, runs the parts it can serve locally, and forwards or fetches the rest from whichever nodes actually hold or lead the data involved.

A client needs to know nothing cluster-specific to get this. No routing hint, no "which node owns this row" lookup, no special driver or connection string beyond the ordinary PostgreSQL ones you'd use against a single node. Learner nodes are the one case worth naming explicitly: nothing routes a read to a learner automatically or preferentially; you get whichever node's own answer the address you connected to happens to produce.

What changes, and what doesn't​

The SQL is the same SQL, and the wire protocol is the same wire protocol; nothing about how you write a query or which client library you use changes because a table's rows happen to live on more than one node. What's actually different:

  • Isolation is unchanged. A cluster transaction is still fully serializable, exactly as on one node; a conflicting transaction is aborted with SQLSTATE 40001, and your existing retry-on-serialization-failure logic handles it unmodified. See Cluster overview and Failover for the full contract.
  • A statement can now involve more than one node's worth of work. Single-node ScramDB never has to move a row over a network to answer a query; a cluster sometimes does. That is the entire added cost surface this page is about, and it is why two queries that look equally simple can have very different latency on a cluster even though neither would on a single node.
  • A statement confined to one bucket's worth of data still takes a fast, local path. ScramDB detects when everything a statement touches lives in a single shard group and, in that case, commits it without paying for the general cross-node transaction protocol. This is exactly why the guidance below centers on staying inside one bucket wherever you can.

How a distributed query actually executes​

At the level that matters to you: a query plan's individual operations, a scan, a filter, a join, an aggregation, run wherever the data they need already is, and intermediate results move between nodes only when a later step needs data that no single node already holds all of. Three shapes cover almost everything:

  • A scan can be pruned to the nodes that actually own the relevant buckets. If your filter identifies a specific value of a table's distribution key, the engine can compute exactly which bucket that value hashes to and only visit the node or nodes that own it, never touching the rest of the table's data at all.
  • A join or aggregation that needs rows from more than one node's buckets brought together picks between shipping the small side once, or repartitioning both sides by the join key. Both are genuine network transfers; the difference is how much data moves and to how many nodes. See the next section for exactly when each applies.
  • The engine can adapt some of this at runtime, not just at planning time, when you opt in. Under the materialize exchange mode (SET exchange_mode = 'materialize'; the default is streaming, which skips all of this), small transfers into the same destination can be coalesced into fewer, larger ones; a partition that turns out far larger than the others once execution is under way, a skewed key, can be split rather than left to bottleneck one node; a fragment running unusually slowly can get a speculative backup dispatched elsewhere. Each of these three behaviors has its own opt-in setting, covered in Practical guidance below. None of it changes a query's result, only how quickly it arrives, and none of it runs unless you turn it on.

The practical consequence: latency on a cluster is not just "how much work," it's "how much work, plus how many network hops the data involved needs to make, plus the size of what crosses on each hop." A query whose plan needs one hop over the low-latency network between nodes in the same cluster costs little beyond that hop's round trip; a query whose plan needs several rounds of data movement, or moves a large table across the network, costs accordingly.

Data distribution, as you experience it​

A new table with a primary key is distributed automatically the moment it's created in cluster mode, no special syntax required:

CREATE TABLE orders (
    order_id    bigint PRIMARY KEY,
    customer_id bigint NOT NULL,
    total       numeric NOT NULL,
    created_at  timestamptz NOT NULL DEFAULT now()
);

ScramDB hashes each row's primary key to place it in one of [cluster] default_buckets buckets (8 by default), and each bucket is owned by as many nodes as your replication_factor calls for (see Multi-zone). Both the hash key and the bucket count are cluster-wide settings today, not something you choose per table: CREATE TABLE accepts a WITH (distribution_key = ..., buckets = ...) clause without error, but it does not currently change where a table's rows actually land in a running cluster, so don't rely on it. Treat the primary-key-hashed, default_buckets-wide default as the real, current behavior. A table with no primary key is not distributed across buckets at all.

Cheap: a query answerable from one bucket.

SELECT * FROM orders WHERE order_id = 482913;

An equality filter on the primary key lets the engine compute the one bucket that row lives in and go straight to the node or nodes that own it. The same is true of a single-row UPDATE or DELETE keyed the same way, and it's why point lookups and point writes on the primary key stay fast on a cluster.

Costly: a query with nothing to narrow the search.

SELECT * FROM orders WHERE total > 1000;

Filtering on a column other than the primary key gives the engine nothing to compute a bucket from, so it has to visit every bucket the table has. This is a fully-supported query; it costs proportionally more than the point lookup above, exactly as a full scan costs more than an index lookup on a single node, plus a cluster's extra step of gathering results from every node holding a piece of the table.

Grouping and aggregating is often cheaper than the scan it sits on top of. For an aggregate whose math can be computed in pieces and merged, sum, count, avg, min, max, and similar, each node computes its own partial result locally and ships back only that small partial result, not the raw rows; the coordinator does one final merge.

SELECT customer_id, sum(total)
FROM orders
GROUP BY customer_id;

This still pays for the full scan above, since nothing narrows it to one bucket, but the network cost past that scan is proportional to the number of distinct customer_id values that come back, not the number of rows in orders. An aggregate that can't be split this way, count(distinct ...), string_agg, array_agg, or any aggregate called with DISTINCT, is rejected with an explicit error against a distributed table rather than run slowly or incorrectly; filter it down to a single bucket first if you need one of these.

Joins move data when the two sides aren't already together, which is the normal case. Joining a large distributed table against a small one is cheap: the engine ships the small side to every node holding a piece of the large one, once, rather than moving the large table at all. Joining two large, distributed tables is the expensive case: both sides have to be repartitioned by the join key over the network so matching rows end up together, real traffic proportional to both tables' filtered size. There is no way today to make two independently-distributed tables share a placement so a join between them costs nothing, even when they're distributed on identically-named or identically-typed columns, so plan for a large-to-large join to move data rather than trying to design around it.

How a read picks its route​

Underneath the scan, join, and aggregate shapes above sits one more decision, and it runs on every read the cluster can route: a cost-based choice between a point lane, a single replica, and a full scatter, cheapest first. Read freshness and routing covers the freshness side of that choice in full, learner_read and max_staleness; this section covers the mechanics every route shares.

A point lookup takes one bounded hop, never the general path. A SELECT that resolves to exactly one row of one distribution key is a point lookup. When the node you are connected to leads that key's bucket, it answers directly off its own index. When another node leads it, this node forwards the request over a dedicated internal channel and waits, bounded, for the leader's answer, instead of running the statement through the full parse, plan, and scatter machinery a less targeted query needs. A send failure, a timeout, or a malformed reply steps aside to the general path automatically rather than hang or guess, so the shortcut is never a way to get a wrong answer, only occasionally a slower one. It only ever carries reads: a point write keeps the general commit path unconditionally, since a write's atomicity comes from the same machinery every other write already uses.

Above a size ceiling, nothing rides a single node, ever. However small a statement's own answer looks, if the data it has to read to produce that answer is large, it scatters across bucket owners the same way the full-scan case above does: a large query never becomes one replica's problem to carry alone.

Every one of these decisions renders as one line in EXPLAIN ANALYZE's Distributed Route: output and increments exactly one counter in /metrics, covered next.

Practical guidance​

  • Favor point lookups and point writes on the primary key for latency-sensitive paths; they take the fast, single-bucket path described above.
  • Put the small side of a join on the small side. A join between a large fact table and a small dimension table is cheap however the optimizer orients it, so you don't have to hint it, but a query written so the optimizer can tell which side is actually small, filtered, narrow, helps it pick well.
  • Expect a JOIN between two large distributed tables, or a full scan with no primary-key filter, to move real data, and size your expectations accordingly; this isn't a bug or a missing optimization, it's the real cost of that query shape on data that isn't already together.
  • Reach for sum, count, avg, min, and max over count(distinct ...), string_agg, or array_agg on a distributed table where you have the choice: the former group stays cheap past the scan it sits on, the latter group is rejected outright against a distributed table rather than run slowly.
  • A transaction confined to rows in one bucket is materially cheaper than one that touches several. If your workload can be structured so a single transaction's writes usually land in one primary-key value's bucket, a single order's rows, a single tenant's rows, it consistently takes the fast local path instead of the general cross-node one.
  • Turn on the adaptive shuffle behaviors for a join- or aggregation-heavy workload that's genuinely moving data. SET exchange_mode = 'materialize' is the prerequisite (the default, streaming, skips all of it); aqe_coalesce, aqe_skew_split, and straggler_backup (each 'on' or 'off', all default 'off') opt into the specific behaviors from How a distributed query actually executes on top of it. None of them change a query's result.

Observing what happened​

EXPLAIN and EXPLAIN ANALYZE work exactly as they do on single-node ScramDB, and show the same plan tree: scan, filter, join, and aggregate operators, plus, for ANALYZE, planning time, execution time, and rows returned. Verify this before relying on it for cluster diagnosis: today's EXPLAIN output does not annotate which node a step runs on, or estimate how many bytes a join or aggregation will move across the network. It tells you the shape of the plan, genuinely useful for predicting whether a query is a point lookup or a full scan, but not which physical nodes did the work.

EXPLAIN ANALYZE (never plain EXPLAIN, which does not execute the statement and so does not know yet) adds one more line after the plan and the timings: Distributed Route: ..., naming the route the statement actually took as one of oltp, point(group N), local(strong) / local(stale), replica(NODE,strong) / replica(NODE,stale), or scatter(N fragments). See How a read picks its route above and Read freshness and routing for what each of those means and when a read qualifies for which.

The log is where the actual cross-node join decision shows up. Every join the engine places across the cluster logs one structured line naming which strategy it chose and the estimated bytes for every strategy it considered, for example:

distributed join placement: chosen=broadcast left=orders right=customers N=3 est_left=48000 est_right=1200 colocate=None broadcast=2400 shuffle=49200 preserved=[] alignment_excluded_broadcast=false

Grep your node logs for distributed join placement: to see exactly what a specific join did and why, including the estimated cost of the strategies it did not pick. The estimator always considers a same-node option too (colocate in that log line); for two independently created tables it is never actually viable today, so expect colocate=None and a choice between broadcast and shuffle in practice.

Prometheus metrics (/metrics, default port 9090) cover the cluster transport and the runtime adaptive behavior described above. The shuffle_aqe_* and shuffle_straggler_backups rows only move off zero once you've opted into the matching session setting from Practical guidance; seeing zeros there on a cluster that has never enabled them is expected, not a sign anything is broken:

MetricWhat it tells you
scramdb_cluster_bytes_sent_total / scramdb_cluster_bytes_received_totalTotal cross-node traffic, cumulative.
scramdb_cluster_frames_sent_total / scramdb_cluster_frames_received_totalMessage counts, cumulative.
scramdb_cluster_send_queue_bytesCurrent outbound backlog; sustained growth here means a peer isn't draining fast enough.
scramdb_cluster_routes_total{route="..."}Statements per routing decision from How a read picks its route: oltp, point, local_replica, forward_replica, or scatter. Fixed cardinality by construction, always exactly one of those five label values, never a node id or a group id, so this counter can never grow with the size of the cluster.
scramdb_cluster_shuffle_consumer_fragments_totalHow many shuffle transfers a join or aggregation actually dispatched.
scramdb_cluster_shuffle_aqe_coalesce_groups_totalHow often small transfers were merged into fewer, larger ones.
scramdb_cluster_shuffle_aqe_skew_split_flagged_total / _acted_totalHow often a disproportionately large partition was detected, and how often the engine actually split it in response.
scramdb_cluster_shuffle_aqe_broadcast_demote_flagged_totalHow often a broadcast side over its size budget was flagged as a candidate for a different strategy. Detection only today; no automatic correction follows yet.
scramdb_cluster_shuffle_straggler_backups_totalHow often a slow fragment got a speculative backup dispatched elsewhere.
scramdb_cluster_handshake_latency_secondsPeer connection setup latency; not per-query, but a useful health signal for the transport every distributed query rides on.

There is no SHOW SHARDS-style SQL statement reachable today; if you see one referenced, expect a normal SQL parse error, not a placement report. The log line and the metrics above are the real, working way to see what a distributed query did.

Distributed transactions and locking​

A transaction that touches rows on more than one node commits with the same serializable guarantee as a single-node transaction, and the same SQLSTATE 40001 retry contract on conflict; see Cluster overview for that guarantee stated in full and Failover for what happens if the node coordinating a commit fails partway through. Underneath, a distributed commit holds row-level locks briefly while it decides; if the coordinating node crashes mid-decision, those locks don't hold forever. They lease out after txn_lock_ttl (10 seconds by default), and another node resolves the transaction rather than leaving affected rows stuck.

Two real, SQL-level locking primitives are cluster-wide, not just local to the node you happen to be connected to:

  • LOCK TABLE, used inside a transaction block exactly as in PostgreSQL, takes a lock that every node in the cluster respects: a conflicting statement on any node, not just the one that issued the LOCK TABLE, waits for it (or fails immediately with NOWAIT), following the same lock-mode conflict rules PostgreSQL uses (a plain read takes ACCESS SHARE, a write ROW EXCLUSIVE, TRUNCATE / ALTER TABLE / DROP / CREATE INDEX take ACCESS EXCLUSIVE, and so on).
  • pg_advisory_lock and the rest of that family (pg_try_advisory_lock, pg_advisory_unlock, and their transaction-scoped variants) are cluster-wide the same way: a lock taken on one node is respected by every other node.

Ordinary DML never takes a cluster-wide lock on your behalf; only an explicit LOCK TABLE or advisory-lock call does, which is what keeps the normal write path fast. Both cluster-wide locking primitives block by default with no built-in deadline, the same as PostgreSQL itself; bound the wait with your client's statement_timeout if you need one.

Next​