The search engine

Pipestream Search is ProtoMolt's Rust distributed-search companion. ProtoMolt's mounted Java search role is Lucene-based; Pipestream Search serves the distributed engine path. It scans every vector instead of approximating, scores text with corpus-wide statistics instead of per-shard ones, and answers a question either with the same result a single machine would compute or with an error that names what stopped it.

What it is

The default way this engine searches vectors is a method called TurboQuant, which stores every vector in a compact four-bit form and reads all of them on every query. On its own that four-bit pass is fast and close to right: measured on the reference corpus, it returned 0.875 of the documents that truly belong in a top ten. Rescoring a deeper pool of its candidates against the exact original numbers closes the rest of the gap, to 0.991 at twice the depth and to all ten of ten at five times.

The index on disk is a set of segment files that are never edited once they are written, with each segment's parts covering the same rows, so a machine maps a segment into memory and reads it from the file instead of loading the index into the process. That layout gives sharding two properties a graph-based index does not have. Shards can collaborate while a query is running, because they can share one score cutoff. And a machine joining the cluster, or coming back after a failure, is given a verified copy of the source machine's segment files and then replays the writes that arrived during the copy, rather than reading the documents again and building an index structure from them.

flowchart TB
        client(["client"]) -->|"a question"| coord["coordinator
holds the only ranked list
counts the corpus for everyone"]
        coord -->|"gRPC stream, plus a signed UDP channel"| n0["shard 0
reads every vector it holds
word lists on disk"]
        coord -->|"gRPC stream, plus a signed UDP channel"| n1["shard 1"]
        coord -->|"gRPC stream, plus a signed UDP channel"| nn["shard N"]
        n0 -->|"candidates, and my own tenth best"| coord
        n1 -->|"candidates, and my own tenth best"| coord
        nn -->|"candidates, and my own tenth best"| coord
        n0 -.->|"analyse the text, turn a question into a vector"| sc["analysis sidecar
or the built-in Rust analyzer"]
        coord -.-> sc

The three parts

Three pieces do the work: a vector scan, a text index, and a coordinator that holds the answer together. Each of them takes a different road from the one most search systems take, so it is worth starting with the usual road in each case.

A scan that reads every vector. A vector search asks which stored items point in most nearly the same direction as the question, where each item has been turned into a list of a few hundred numbers by a language model. Comparing the question against every stored item would answer that question directly, and for most engines that is too slow to consider, so they build a graph instead: each item is linked to a few of its nearest neighbours, and a search starts somewhere in the graph and walks towards the question, following whichever link improves the score. The walk is very fast, and it is approximate, because a good answer sitting in a corner of the graph that the walk never reached is simply not found. How often that happens depends on the data, which is why an engine of that kind has an accuracy figure to characterise and defend.

This engine keeps the direct comparison and makes it cheap instead. Each vector is stored in a four-bit form that takes roughly an eighth of the room the original numbers do, and the scan reads every one of them. There is no graph to walk and no neighbour list to get wrong, so the order the scan produces is the true order of the compressed vectors, not a sample of it. The compression itself does cost some accuracy, and the answer to that is a second pass rather than a larger graph: ask for more candidates than you need, rescore those candidates against the exact original numbers kept in a separate file, and the exact answer comes back. How many candidates a given accuracy target needs is a table measured on the corpus and shipped with the index, not an estimate.

A text index scored by BM25. Keyword search has worked the same way for decades. For every word in the vocabulary, keep a list of the documents that contain it; to answer a query, walk the lists belonging to the query's words and add up a relevance number for each document you meet along the way. Those lists are called postings, and the formula almost everyone uses to score them is BM25, which rewards a word that appears often in a document and discounts a word that appears in a great many of them. This engine writes the lists into one binary file per shard and reads them through the operating system's memory mapping, so the file is never loaded into the process as a whole.

Stored alongside the lists, for every block of postings, is the highest score any document in that block could possibly reach. When the query already holds better answers than a whole block's best case, the scorer skips the block without reading it. On a word that appears in a large share of the corpus that is worth up to about seventy times the speed when ten results are wanted, and the results are identical, bit for bit, with the skipping switched off.

A coordinator that holds the only answer. When an index is split across several machines, the usual arrangement asks each machine for its own best ten and merges the lists that come back. That works, and it makes every machine finish a whole job on its own. Here the machines are not asked for ten of anything. The coordinator keeps the only ranked list in the query; shards stream candidates towards it, and the coordinator sends back the score of its tenth-best answer so far as that answer improves. A shard that has been told that number can discard anything below it without any risk, because ten documents already beat it.

The cutoff travels two ways at once. The authoritative path is the gRPC stream each node holds open for the length of the query. Beside it runs a plain UDP channel carrying the same cutoff updates, which arrive a little sooner and are signed with a key of their own, so a leaked client credential cannot be used to forge a cutoff and cut somebody's search short. Every message on the fast channel has a twin on the stream, so a lost datagram costs a little extra scanning and cannot change a result.

The answer one machine would give, or an error

One rule runs through the whole design. A query answered by the fleet returns what the same query would return on a single machine holding the entire corpus, under a fixed order for breaking ties, or it fails and says what stopped it. That is a structural promise rather than a statistical one. It does not mean the results are usually the same; it means a result that cannot be shown to be the same is not served.

The machinery behind it is ordinary. Each node ends its stream with a completion frame, and the coordinator answers only once every frame has arrived reporting success and the same scoring fingerprint. Candidates that a rising cutoff overtook while they were in flight are dropped when the lists are merged. A query that was cancelled is reported as incomplete rather than served as a shorter answer.

The same stance turns up in small places, as a habit of returning an error that names its cause instead of a result that looks fine. A phrase query on a field that never stored word positions returns an error naming that field, rather than guessing the positions from where the characters sat. A prefix search that expands past its limit returns an error naming the count, instead of a shortened list that looks complete. A request for more results than the coordinator allows returns an error naming both numbers, instead of being trimmed to fit. A sum of integers that leaves the range of a 64-bit integer returns an error suggesting the floating-point form, rather than wrapping around. A fleet whose shards disagree about which embedding model their vectors were built with is turned away before any shard is asked.

Why one shard's score means the same as another's

Merging ranked lists from several machines only works if the scores were computed in the same terms. The BM25 formula depends on facts about the corpus as a whole: how many documents there are, how many of them contain the word, how long the average document is. A machine that answers those questions about its own slice produces scores that are perfectly reasonable on their own and are not comparable with another machine's.

So the coordinator does the counting for everyone. It analyses the query text once, asks every shard how many documents it holds, how long they are in total, and how many of them contain each of the query's words, adds those answers together, and sends the totals back down with the query. Every shard then scores with the same view of the corpus that a single index over all of it would have. The vector side has an equivalent requirement: the fleet is configured with one set of TurboQuant parameters before any document is loaded, each shard reports a fingerprint of the settings it was built with, and a fleet whose fingerprints disagree is turned away rather than merged.

The widget below is an invented corpus of thirty documents split across three shards, with the query word appearing in six documents on shard A, two on shard B, and one on shard C. The counts are made up; the arithmetic on them is the engine's own. Switch between the two ways of gathering the statistics and watch the best document fall to fifth place.

There is a test for this. One case asserts that a query across the shards returns the same thing as the same query against a single index over all the documents. A second case asserts that per-shard statistics would have given a different answer, so the first case cannot decay into something that passes for the wrong reason.

The cutoff the shards share

The reasoning behind the shared cutoff is short enough to follow in full. Suppose a shard has already found ten documents of its own and its tenth-best has a score of 4.0. The best ten in the whole corpus are the best ten out of everything the shards hold together, so the corpus-wide tenth-best cannot be worse than any single shard's tenth-best. Which means 4.0 is a lower bound on the final answer. Take the highest such number any shard has published, send to all of them, and every shard can throw away anything scoring below it, with no chance of discarding a document that belonged in the answer. Documents that tie with the cutoff are kept, because the comparison is a strict one.

The saving was measured. On the reference corpus with eight shards, sharing the cutoff brought the number of candidates that had to be considered for the answer down from 392 to 209 per query when ten results were wanted, and from 646,560 to 247,442 when ten thousand were. The results are identical, bit for bit, whether the sharing is on or off. That equality is what makes the saving worth having: it is not a quality setting to be traded off, it is work that could not have changed the answer. The demo page animates the race lane by lane.

Adding a condition costs no results

Real searches usually carry conditions: only this court, only after 1990, only within twenty-five kilometres of a point. For an engine built on a neighbour graph, a condition is awkward to honour. Apply it after the walk and the walk will have spent its effort on documents the condition then removes, so when the condition is narrow the few survivors are not the best few. Apply it during the walk and links have to be cut out of the graph, and a graph with links missing no longer connects the way the search assumed it would.

A scan that reads everything does not have to choose. The condition is compiled once on the coordinator into a small typed tree, sent to each shard, and resolved there against that shard's own vocabulary, which turns column names into table positions and words into numbers. The result is a list of which rows are still allowed, and the scan takes that list as a mask, so a document the condition removed is never scored at all. The answer is the true best ten among the documents that survived, whether the condition keeps ninety percent of the corpus or a dozen rows out of it.

The fleet, and what it measured

The engine's reference deployment is eight shards over 86.6 million court-opinion chunks: six shards on one 32-core workstation and one shard on each of two Raspberry Pi 5 boards, with mutual TLS between every machine and bearer tokens at the public edge. Steady state is 674 GB across the fleet. Every number below was measured on that corpus.

Two of the three machines are worth describing before the numbers are. A Raspberry Pi 5 is a single-board computer about the size of a deck of cards, with four cores and, in this cluster, 7 GB of memory on one board and 15 GB on the other. Everything measured here would be considerably faster on server hardware, and the figures should be read with that in mind.

The small boards are in the cluster on purpose. This engine is meant to run as a node on a phone as well as in a rack, and a phone has roughly a Pi's share of processing power, so putting a Pi in the fleet is the standing test of the idea the whole design rests on: that one fast machine and several slow ones can work on the same question together and still return the exact answer. A cluster of identical servers would never exercise that. Machines of different sizes collaborating on one query is the strength of the design, so the reference cluster is arranged to make that collaboration do real work.

Latency over 36 queries, three repeats each, asking for ten results and leaving out the cost of turning the query text into a vector. The tables on this page write the number of results asked for as k, and p50 for the median, p90 and p99 for the times only one query in ten and one in a hundred exceed. The right-hand columns are the same run with mutual TLS between every machine.

Transport security costs two to three percent at the median, because the coordinator opens a connection to each node once and keeps it, so no handshake sits on the query path. Turning the question into a vector costs 11.5 ms, once per query. Two things in the table are worth a second look. The first is that the vector route's cost barely changes with the number of results requested, 280 ms for ten and 294 ms for a hundred, because the scan reads every vector either way. The second is in the per-shard timings: each Pi board spends 4.1 to 4.3 seconds of its own working time on a query, against 2.3 seconds on the workstation, so the fleet's answer waits on the Pis. That is the arrangement doing its job rather than a fault to engineer away. The small boards carry an eighth of the corpus each, they are slower at it, and the answer they take part in is still exact.

What recall means, and what this engine's is

Recall is the word for the question "did the search find the things it should have found". Work out, by comparing the query against every document in the corpus in exact arithmetic, which ten documents really are the ten best answers. Then look at the ten the engine returned. If eight of the true ten are in there, recall at ten is 0.8. A perfect score is 1.0, and every step below it is a document that belonged on the page and was left off it. Move the slider below to see what a given recall figure looks like from the reader's side of the screen, then use the three buttons to put the engine's own measurements on the same picture.

What a deeper pool costs

Those three measurements are three rungs of a ladder that runs further. The engine was profiled against brute-force ground truth over all 86.6 million rows, using 64 held-out queries, at a range of result counts and candidate depths.

The pattern holds at every depth tested. The four-bit ordering on its own is right about 87 to 90 percent of the time and can fall to 20 percent on an unlucky query, and asking for five times as many candidates and rescoring them recovers the exact answer. What that costs is the rescoring, not the search. Finding the candidates stays between 290 and 300 ms all the way out to ten thousand of them, and reaches 750 ms only at a hundred thousand, because the shared cutoff keeps the cost of the scan flat while the pool of candidates grows a hundredfold. The rescoring is what climbs, from about 2 ms on a pool of fifty to 1.6 seconds on a pool of fifty thousand.

What splitting the index buys, and what it does not

Splitting a corpus into shards is usually described as a way to make a system bigger. It is more accurate to say it makes a single question faster. A separate sweep put the same corpus on one machine at four different widths and asked the same forty queries at each: one shard answered in 6592 ms at the median, two shards in 1943 ms, four in 626 ms, and eight in 313 ms, a twenty-one-fold improvement. The corpus is the same at every width, and the correctness check holds the results identical across all four layouts.

The number of questions per second is a different curve, and a flatter one. On the same machine the eight-shard layout served 3.16 queries per second with one client asking and 3.24 with thirty-two asking at once, because the shards were already using the machine's cores on the first query. Splitting the work buys the time one person waits, not the number of people who can wait at the same time.

Related