Clusters

A cluster is nodes started on the same lake. There is nothing else to set up: no coordinator service and no member list. The nodes find each other through the lake, one of them orders the commits, and any of them takes over if it dies. Use a cluster when one node isn’t enough writers or readers, or when the lake shouldn’t stop if a machine does.

# on each machine: the same lake, and the address the other nodes reach it at
pondra serve s3://my-bucket/lake --addr 10.0.0.1:8080
pondra serve s3://my-bucket/lake --addr 10.0.0.2:8080
pondra serve s3://my-bucket/lake --addr 10.0.0.3:8080 --reader

The first node up leads. The others follow. A --reader node answers SQL and never leads, so read-only bucket keys are enough for it. Nodes started at the same instant agree on one leader.

Role	What it does
Leader	Puts the writes in order (one number per commit), tiers, compacts, and cleans up.
Follower	Takes writes and queries like the leader and passes the ordering to it. Runs its share of streaming tasks and of big queries.
Reader	SQL only. Any number of them, next to the writers.

Try it on one machine

A folder is a lake too, so a cluster on your laptop is the same command with a port for each node. This starts three nodes and a reader in the background, with --ack replicated (see Failover):

mkdir -p cluster
for i in 1 2 3; do
  nohup pondra serve cluster/lake --addr 127.0.0.1:808$i --ack replicated > cluster/node$i.log 2>&1 &
  echo $! > cluster/node$i.pid
  sleep 1
done
nohup pondra serve cluster/lake --addr 127.0.0.1:8084 --reader > cluster/reader.log 2>&1 &
echo $! > cluster/reader.pid
until curl -sf localhost:8084/stats > /dev/null; do sleep 0.2; done

Every node’s GET /stats says who it is and who leads:

import requests

def roles(ports=(8081, 8082, 8083, 8084)):
    found = {}
    for port in ports:
        try:
            stats = requests.get(f"http://localhost:{port}/stats", timeout=2).json()
            found[port] = (stats["role"], stats["term"])
        except requests.RequestException:
            found[port] = ("down", None)
    return found

roles()

{8081: ('leader', 1), 8082: ('follower', 1), 8083: ('follower', 1), 8084: ('reader', 1)}

Write to any node

Every node takes writes, so point producers and clients at any of them, or at a name that reaches all. A follower does the work of the write and the leader only records where it went. A write is visible on every node a few milliseconds after it is acknowledged, and a follower shows its own client’s writes at once: a SELECT after an INSERT on the same node sees the row.

import pondra

nodes = [pondra.connect(f"http://localhost:{port}") for port in (8081, 8082, 8083, 8084)]
nodes[2].sql("CREATE TABLE events (id BIGINT, user VARCHAR, amount DOUBLE)")
nodes[2].sql("INSERT INTO events SELECT i, 'u' || (i % 3), i * 1.5 FROM generate_series(1, 30000) t(i)")
[node.sql("SELECT count(*) FROM events").item() for node in nodes]

[30000, 30000, 30000, 30000]

That wrote through the third node and read on all four. A reader says no to a write:

try:
    nodes[3].sql("INSERT INTO events VALUES (0, 'x', 0.0)")
except RuntimeError as e:
    print(e)

500: read-only node

Big queries use every node

A query over big tables can run across the nodes: each reads its share of the files, and the answer is the same as on one node. A node spreads a query only when that is faster. It weighs the bytes the plan would move, at the speed of its slowest link to the other nodes, against the time the other nodes would save, so a slow network never makes a query slower. ?spread=1 spreads a query anyway and ?spread=0 keeps it on the node. GET /metrics counts them.

query = "SELECT user, count(*) AS n, round(sum(amount)) AS total FROM events GROUP BY user ORDER BY user"
rows = requests.post("http://localhost:8083/sql?spread=1", data=query).json()
counts = [line for line in requests.get("http://localhost:8083/metrics").text.splitlines()
          if line.startswith("pondra_spread_queries_total")]
rows, counts

([{'user': 'u0', 'n': 10000, 'total': 225022500.0},
  {'user': 'u1', 'n': 10000, 'total': 224992500.0},
  {'user': 'u2', 'n': 10000, 'total': 225007500.0}],
 ['pondra_spread_queries_total 1'])

Big writes use every node too. An INSERT … SELECT or a CREATE TABLE … AS over a big table is written by every node at once, each node writing its own share’s files. The leader records them in one commit, so the rows, their row ids and a retried job’s answer are what one node would give. A query whose rows have to be brought together first (a GROUP BY, an ORDER BY, a LIMIT, a join that shuffles) is written from one node.

Caution

A query that can’t be split correctly runs on one node instead: NOT IN over a subquery on big data, a LIMIT inside such a subquery, and a window over all rows. A keyed table is read whole on every node, since a key’s versions are spread over the files.

Serverless, from any machine

pondra sql needs no node: any machine with the binary and access to the lake reads it directly, and writes do their own work. When a cluster is running, the leader only records what the write made. When nobody is running, pondra sql leads for the moment it takes.

pondra sql --dir cluster/lake "INSERT INTO events VALUES (30001, 'u1', 1.5)"
pondra sql --dir cluster/lake "SELECT count(*) AS n FROM events"

{"committed":true}
+-------+
| n     |
+-------+
| 30001 |
+-------+

A machine that can reach the bucket but not the leader, in another network or another company, still writes: its request goes into the lake’s inbox/ and the leader records it within a second or so. A read-only key still reads.

Failover

If a leader dies, the nodes that are left ask each other whether they still hear it. If none do, one claims the next term and leads, about five seconds after a crash and at once after a clean stop (Ctrl-C or SIGTERM). A leader that comes back finds it has been replaced and follows.

--ack chooses when a write is acknowledged, and every node needs the same setting:

--ack	Acknowledged when	Time to ack
durable (default)	the write is in the bucket	a millisecond on local disk, 0.25 to 0.7 s on S3 or R2
replicated	--replicas nodes hold it (default 2: the leader’s memory and a follower’s disk)	about 4 ms on any storage

With replicated the bucket gets the write a moment later. It survives any one node dying, but not the leader and every holder dying within that moment. Kill the leader mid-flight and see:

kill -9 $(cat cluster/node1.pid)

import time

for _ in range(30):                      # wait until both know the new leader
    time.sleep(1)
    now = roles((8082, 8083)).values()
    if sorted(role for role, term in now) == ["follower", "leader"] and len({term for role, term in now}) == 1:
        break
sorted(roles((8081, 8082, 8083)).values(), key=str)

[('down', None), ('follower', 2), ('leader', 2)]

Node 1 was the leader. Now one of the other two is, on term 2, and nothing acknowledged was lost. A write that reached a follower in the seconds before it learned of the new leader was answered with an error, not acknowledged: send it again (with the same producer and seq, or job, it is applied once). Writes go through either node:

alive = nodes[1:3]
alive[1].sql("INSERT INTO events VALUES (30002, 'u2', 3.0)")
[node.sql("SELECT count(*) FROM events").item() for node in alive]

[30002, 30002]

A reader node follows the new leader within about a second after a failover. Give your clients every node’s address so they carry on by themselves, and see What Pondra survives for the bucket, disks and each door.

Stop what is left when you’re done:

kill $(cat cluster/node2.pid cluster/node3.pid cluster/reader.pid)

Other things clusters do

• Attached lakes: each cluster leads its own lake, and every cluster reads the others (--attach sales=s3://my-bucket/sales, or ATTACH in SQL), which is how one lake’s commit rate is split over several. A write to sales.orders is recorded by that lake’s leader. See tables.
• Many lakes on one machine can share a port: pondra serve on a folder of lakes serves each as a database. Each database can lead a cluster of its own (clusters of one database).
• Laptops can join with pondra serve or --reader and leave again. After a full shutdown the first node started leads at once.
• Streaming tasks with shards are dealt to the live nodes, and move when membership changes (streaming).
• Every door is on every node: --pg, --kafka and --flight are per node, so give each node the ones its clients use.

Caution

All commits of one lake go through one leader in one sequence. That work is only metadata, but past many thousands of commits a second the answer is several lakes, not more nodes. Clusters are tested as several processes on one machine and against a slow simulated bucket, not yet measured across real machines, so expect the numbers on your network to differ. There is no transaction across lakes, and Kafka consumer-group membership lives in the leader’s memory, so after a failover consumers rejoin.