ingest
Load a collection from a source larger than memory, and be able to carry on after being killed.
upsert(name, flow) sends what it is given, one batch after another, and everything above it is the caller's: how big a batch is, how many are in flight, what happens when one times out, and where to start again after the process died at point four hundred thousand. That code is written once per project, it is written by someone who has just started using a vector database, and it is where the data goes missing.
var token = loadCheckpoint() // null on a first run
val report = qdrant.ingest("docs", points, resumeFrom = token) { saveCheckpoint(it) }What it owns
Batching, by point count and by serialized size, so one oversized point cannot make a request the server refuses.
Bounded concurrency: at most concurrency requests in flight. Unbounded parallelism is how an ingest that was fine for raw vectors starts being rejected the day the server does the embedding.
Per-batch retry, so one failed batch is retried rather than the stream restarted. Only KdrantException.retryable failures are retried; a malformed point is returned at once, because retrying it wastes the window in which the rest could still have been written.
A resume token, handed to onCheckpoint as it advances and returned in IngestReport.
What it does not own
Storing the token. Where it goes — a file, a row, an object store — is a question about the caller's deployment, and a client that picked one would be wrong somewhere.
It does not embed anything either. If the points carry dev.kdrant.model.VectorData.Inference the embedding happens on the server, and that is exactly the case where concurrency wants to be small: the ingest stops being upload-bound and starts competing for the server's inference capacity.
Parameters
the source. Cold, replayable, and emitted in a stable order if resumeFrom is ever going to be used against it.
most points per request.
soft ceiling on a request's serialized size; a batch is flushed before crossing it. A single point larger than this is still sent, alone, because splitting a point is not a thing.
requests in flight at once. 1 makes the whole ingest sequential.
attempts after the first, for a retryable failure.
delay before the first retry; it doubles per attempt.
a token from an earlier run. Its points are dropped as they arrive rather than never produced: the source is replayed in full and the acknowledged prefix does not go over the network again. Where reading the source is itself the expensive part, have the source start after the checkpoint and pass no token.
called as the acknowledged prefix grows, on the ingest's own coroutine. Keep it quick: it runs between batches, and a slow write here is throughput the ingest does not have.
Throws
if a bound is not positive.
if a batch fails and cannot be retried. The exception is thrown after the last onCheckpoint call, so the token in hand is the prefix that was written.