Sink from-proto through a disk spool, and load without constraints - #869
Merged
Conversation
Contributor
Author
🔍 Vulnerabilities of
|
| digest | sha256:93f147d56dc616e61f337d89cc77fb85180fd9259c54698a30436efe055ef17d |
| vulnerabilities |     |
| platform | linux/amd64 |
| size | 123 MB |
| packages | 382 |
📦 Base Image oisupport/staging-amd64:24.04
| also known as |
|
| digest | sha256:019e8eb29a85e74d64925745884f2ec79aa27e3feab36353d24656f4d6b89467 |
| vulnerabilities |   |
sduchesneau
force-pushed
the
feature/sink-sql-local-cache
branch
from
August 10, 2026 19:48
4ddddca to
db6e303
Compare
sduchesneau
force-pushed
the
feature/sink-sql-local-cache
branch
from
August 11, 2026 17:02
db6e303 to
fb0b784
Compare
maoueh
force-pushed
the
feature/sink-sql-local-cache
branch
from
August 11, 2026 18:52
6408e54 to
c810ac2
Compare
maoueh
previously requested changes
Aug 11, 2026
| // parameterised INSERT of the same value produces. Binary COPY does no coercion, so an | ||
| // encoder bug shows up as either a hard COPY failure or, worse, silently wrong data. | ||
| func TestPgCopyBinaryRoundTrip(t *testing.T) { | ||
| requireDocker(t) |
Contributor
There was a problem hiding this comment.
Tests that are relatively fast to perform, even within Docker, should all be run by default. If this can easily launch testcontainers to make the test, it should be enabled automatically.
Comment on lines
+233
to
+241
| func sortedKeys[V any](in map[string]V) []string { | ||
| out := make([]string, 0, len(in)) | ||
| for key := range in { | ||
| out = append(out, key) | ||
| } | ||
| sort.Strings(out) | ||
|
|
||
| return out | ||
| } |
Contributor
There was a problem hiding this comment.
Doesnt maps + slices could do that in one line?
| // sinker is unchanged: what differs is that a "flush" only seals a segment, and the | ||
| // database is written to later, by the buffer's own goroutine. That is the whole point — | ||
| // the stream stops waiting on PostgreSQL. | ||
| type bufferInserter struct { |
Contributor
There was a problem hiding this comment.
We have also now BufferedInserter, not sure if we could not find a better name
Comment on lines
+129
to
+139
| func humanBytes(n int64) string { | ||
| value := float64(n) | ||
| for _, unit := range []string{"B", "KiB", "MiB", "GiB"} { | ||
| if value < 1024 { | ||
| return fmt.Sprintf("%.1f%s", value, unit) | ||
| } | ||
| value /= 1024 | ||
| } | ||
|
|
||
| return fmt.Sprintf("%.1fTiB", value) | ||
| } |
Contributor
There was a problem hiding this comment.
I'm adding a formatx package to hold all this, once existing, will point Agents.md to it.
sduchesneau
force-pushed
the
feature/sink-sql-local-cache
branch
2 times, most recently
from
August 12, 2026 02:08
bcc27da to
048215b
Compare
sduchesneau
force-pushed
the
feature/sink-sql-local-cache
branch
2 times, most recently
from
August 12, 2026 19:16
3b39425 to
64d1b76
Compare
sduchesneau
changed the title
A writer that turns a walk's []any into the bytes `COPY ... FROM STDIN (FORMAT BINARY)` expects, and the normalisation that gets there: binary COPY does no coercion, so every value has to match the column type the server reported.
Substreams throughput is paid for, so a slow or stalled database should cost disk rather than download progress, and blocks already paid for should survive a restart rather than be streamed again. The package owns the segment lifecycle, the disk budget and the sizing. What the bytes on disk look like is a Codec, how a sealed segment reaches the server is an Applier, and both are the driver's business. Segments are sized by measured commit duration rather than by a block count: block payloads differ by orders of magnitude across chains, so a fixed count gives wildly unstable durations. The disk budget is the only ceiling on how far ahead of the database the stream may run, and the open segment counts against it. An idle window commits what is open when the stream goes quiet, so a stall cannot leave the cursor where it was.
Two things the from-proto sink used to do inline now happen off the stream's path, and the flags say which resource each one spends. Rows go to the spool rather than to the database, in the format the chosen --write-mode can send unchanged: binary COPY files, rendered SQL tuples per table, an interleaved log replayed in walk order, or typed values on ClickHouse, whose inserts are columnar. Which path runs used to be derived from whether a directory flag was set and whether the schema's foreign keys happened to form a cycle, and an unorderable schema was silently downgraded to row-at-a-time inserts — an order of magnitude slower, announced only in a log line. An explicit mode the driver or the schema cannot support is now an error that names the fix. Constraints are created after the load rather than during it: measured through binary COPY, loading with foreign keys in place costs 27.7x against 3.3x for building the same ones afterwards. --apply-constraints says when, defaulting to auto — a backfill that ends with no primary keys and no foreign keys is a silent wrong result that looks like success, where the stop-the-world pass is at least a visible one. `constraints apply` and `constraints drop` are how that pass goes into a maintenance window instead, and what setup recorded is what they use. The flags are grouped by what they spend: --decode-* is CPU, --db-write-* is one commit to the database, --spool-* is how far ahead of it the stream may run. Both mode vocabularies still land on one command, the mode being read from the module long after the flags are registered, so the help groups them under headings and a flag typed for the other mode fails rather than being ignored. At the chain head the spool is drained and left behind: a block should be queryable when it arrives, not when the segment it lands in is full.
The container-backed tests: every write mode produces the same table, a spooled ClickHouse run matches an unspooled one, a bounded run writes what it held at the stop block, an undo removes the right rows with and without constraints, and the switch to direct inserts at the chain head leaves nothing behind. Plus the two that need no container: the constraint policy round-trips through what setup records, and the table order really is a topological sort over the foreign keys.
The numbers the defaults rest on, so they can be re-run rather than trusted: binary COPY against multi-row INSERT under each set of constraints, the cost of loading with constraints in place against building them afterwards, and where the client's own ceiling is.
The reference gets a table per flag group and the constraint timings; the plan records the decisions and what they rejected, and the local-buffer plan it supersedes says so at the top.
sduchesneau
force-pushed
the
feature/sink-sql-local-cache
branch
from
August 12, 2026 19:19
64d1b76 to
2e999da
Compare
A bounded run that spooled panicked at its stop block: the spool owns the transactions while it is open, so BeginTransaction was a no-op and the constraint pass dereferenced a nil one. Draining first also puts the rows in before the constraints go on, which is the whole point of creating them afterwards — the tail of the range was otherwise loaded against foreign keys that were already there.
A run interrupted before the backfill ended, or whose constraint pass was killed part-way, leaves a database with no primary keys and no foreign keys. It answers queries, slowly and rejecting nothing, and looks exactly like a database that is fine — so the sink now compares the policy against the catalog on every start and names what is absent. One indexed query, against a pass that takes minutes. It only reports. Creating them stays the timing's business: auto does it when the backfill ends, manual leaves it to `constraints apply`. Also drops the constraint policy that setup recorded in _sink_info_. setup is optional — the run path creates the schema when it finds none — so what it stored was there or not depending on a step nobody has to take, and `constraints apply` would quietly mean different things on two identical deployments.
Every index build and foreign key validation was held open in a single transaction, so a schema large enough to exhaust memory lost the whole pass and the next run started from nothing. They are committed one at a time now, which bounds the memory and leaves a killed run's finished work in place. --constraints-per-transaction raises that on the `constraints` commands; the run keeps to one, more deliberate scheduling being what the command is for. Two bugs the batching exposed, both from keying constraints by name alone. Names are unique per table, not per schema, and every table carries a foreign key called fk_block: one present on any table counted as present on all of them, so the rest were never created, and `constraints drop` removed one and then failed on the primary key the others still referenced. They are keyed by relation and name now, and reported that way too.
`constraints apply|drop` and `setup` inferred the output module from the package and had no way to be told otherwise, so on a package with more than one candidate they derived a schema from a different module than the run they accompany — and then created constraints for tables that are not there. They take it as an optional second argument now, the same shape the run command has always had.
An output without schema.proto annotations gets tables derived from the message structure alone: no primary keys, no unique constraints, no foreign keys, and so no indexes at all. The run said nothing, and `constraints apply` said only that there was nothing to apply, which reads like the schema is already fine.
Every table carries _block_number_ and every reorg deletes from every table by it, but nothing indexed it: a foreign key indexes its referenced side only. Each undo was a sequential scan per table, and the cascade the undo used to rely on would have been worse still, PostgreSQL looking the child rows up once per deleted parent row. The index is created in the same pass as the constraints, being the same kind of expensive and wanted at the same moment. --disable-block-number-index leaves it out for a run that can never reorg. It is also the one thing that pass can do for an output with no schema annotations, so `constraints apply` now warns there instead of refusing: nothing declares a primary key, but the reorg path still deletes by block number. Starting from an unannotated package meant finding the right proto and the right extension names by hand, so `substreams tools extract-proto --sql` writes the output message back out with every option commented out beside the message and field it applies to, and the annotations file beside it so the result parses.
10GiB of erc20-balance-changes-shaped rows, locally: building the index after the load costs 2.6s on a 44s load and 218MiB against 10GiB, and having it in place while the rows arrive costs nothing measurable — _block_number_ only increases during a backfill, so every insert lands on the rightmost page of the btree. It takes one table's undo from 1.296s to 15ms. The reorg path deletes from every table by that column. ANALYZE is part of the method rather than incidental: a COPY leaves no statistics, and without them the planner sequentially scans a predicate matching 0.2% of the table. The first version of this measurement reported 4.7s for an indexed undo for exactly that reason, so the chosen plan is now reported next to every number.
A backslash carries no meaning inside a standard string literal, and standard_conforming_strings has been on by default since PostgreSQL 9.1, so escaping it stored two backslashes where the value had one. That also put the write mode in the data: COPY hands pgtype the string as it is, so only batch-insert and row-insert came through quoteLiteral. The write-mode test now carries a backslash and a quote, which is what pins the three modes to the same bytes. The two protojson branches escaped the same way, over output that is mostly backslashes. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Binary COPY resolves the encode plan from the Go type rather than from the column, so retyping an empty array to []string reached no plan at all on a numeric[], bigint[], bytea[], bool[] or timestamp[] column: the first block carrying an empty list refused the row and took the segment with it. The walker's own []any encodes against every array OID, so it is left alone. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The walk hands an enum over as EnumValue and MapFieldType declares the column TEXT[], but normalizeSlice had no case for it, so the row was refused and the segment with it. Scalar enums were already fine, pgtype rendering them through their String method. The elements keep that type rather than being flattened to their names, which is the same text the rendered write modes emit for them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
VerifySchemaCompatibility runs ahead of CreateDatabase and read the columns over a connection naming the schema, which on a first run does not exist yet. ClickHouse refuses that with UNKNOWN_DATABASE and newClient retries a failed dial forever on a context of its own, so setup never returned. It reads through 'default' now, as CreateDatabase already did. system.columns is global, so which database the connection names does not change the answer. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Recovery answered "already applied?" from the segment's first block alone, so a record left behind by an earlier run could answer for a different segment that happens to start on the same block. Comparing the last block too narrows that, and needs no change to the table. It is not enough on its own: the sizer restarts at its floor every run, so a re-streamed segment can end on the same block as the record it collides with. What lets such a record exist at all is dealt with separately, by sealing a segment only once its own cursor is recorded. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Scaling the current target by target/elapsed ignored how much the commit moved, so an idle or drain seal — fast because it was small — doubled the target while the database may well have been slow.
runFromProtoSetup built its policy from --apply-constraints, a flag only the run command declared, so the value read was always the empty string and the schema came out bare however the --disable-* flags were set. The flag stays off setup. There the question is not when the constraints are created but whether, since setup creates the schema and exits — two of the three values would collapse onto the same answer. What it creates is said by the --disable-* flags alone, and --disable-all-constraints is the shorthand for turning all three off at once, which is what --no-constraints said before it was deprecated. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
db.Close was only reached when a bounded range completed, so a Ctrl-C during a backfill — the run the spool exists for — left the open segment unsealed and its blocks to be streamed again.
Map fields named an entry type nothing declared and nested enums were dropped altogether, so the file the command tells the operator to feed back through --proto-file-override did not parse. Field hints are now whole declarations, since an option has to sit inside the field's own brackets.
Both branches fixed the empty array and the repeated enum. The enum is rendered here rather than handed over as it is. Both encode identically today — pgtype picks the Stringer, which is what ValueToString renders with too — but pgtype chooses between TextValuer, driver.Valuer and fmt.Stringer in that order, so the day EnumValue gains an earlier one, binary COPY would start writing something the rendered write modes do not, and nothing would fail. Rendering here keeps that choice ours. The empty array is passed through untouched: with no element there is no value whose representation could drift, only a plan that has to resolve, and the walker's []any resolves against every array OID where a mapped one leaves any OID it does not enumerate. Takes both test files from that branch: they cover 14 array types against the real encoder, where the ones dropped here covered 10. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
A flush writes its rows and then its cursor, but the seal ran between the two: a segment closed carrying the cursor the previous flush stored, and committed claiming a range that cursor stops short of. The next run resumes there, Run's undo deletes every row above it — the segment's own — and the record stays behind to answer AlreadyApplied for the segment carrying exactly those rows. Recovery discards that one without marking a hole, replays the segments behind it, and moves the cursor over a gap it just created. Sealing when the cursor is recorded rather than at flush keeps every segment covered by its own cursor. Records reaching past the stored cursor are dropped at startup, which clears what a database synced by an earlier build is still carrying. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Every segment applied recorded a row and nothing ever removed them, so a sink that stays live grew the table for as long as it ran: each restart backfills the little it is behind, seals a segment or two on the way, and reaches the head again. Reaching the head drains the spool and closes it for good, which leaves no directory on disk for a record to answer for — the records only ever told recovery whether a directory it found was already applied. A run that ends at its stop block keeps them instead, having no next restart to grow them. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The reader refused any record over 64MiB while the writer had no limit at all, so a large rendered tuple — a wide bytes column, a long repeated field — was written and counted into the manifest, verified intact, and then failed only at apply. Recovery replayed that same segment on every start, so the sink could not come up again until the directory was removed by hand. A record cannot be longer than what is left of the file holding it, which catches a corrupt length just as well without putting a ceiling on a row the writer accepted. The four-byte prefix is now a write-time error rather than a silent truncation, and a segment that genuinely cannot be applied says what removing it costs. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Fix the spool's critical bugs before it ships
A stop block ends the run; it does not say the backfill is over. A range is routinely one chunk of several, and creating the constraints at the end of one left every chunk after it loading into a constrained schema — measured at 27.7x, which is the case this whole arrangement exists to avoid. Reaching chain HEAD is the one signal that says there is nothing left to load, so that is now the only place 'auto' acts. A bounded run says on the way out what it left behind, since a schema with no primary keys and no foreign keys answers queries and looks exactly like one that is fine. Two log lines named things that do not exist while they were at it: an --apply-constraints=head value, and an 'apply-constraints' subcommand where it is 'constraints apply'. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
--apply-constraints goes back on setup, which reads it: 'always' creates them with the schema, while 'auto' and 'manual' both leave the tables bare, the first for the run to constrain at chain HEAD and the second for `constraints apply`. --disable-all-constraints goes away with it. The three --disable-* switches already say what the schema declares, and --no-constraints deprecates to them again rather than to a fourth spelling of the same thing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The "falling behind" warning tripped on how many blocks the buffer spanned, which says nothing once rows are on disk: the sparse start of a large backfill covers millions of blocks holding a few hundred kilobytes, so a healthy run warned continuously from its first minute. It now measures the spool against the disk budget it was given, which is the number the operator set. The startup warning also promised constraints "will be created once the backfill reaches chain HEAD", which a run given a stop block may never do. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
--constraints-per-transaction read as an execution knob and only bundled statements into one transaction, which is not what it says and not what the pass needed. It becomes --constraints-parallelism, and now does what the name promised: statements inside a wave go to the server together. Constraints sit on independent relations, so the only ordering that matters is that a foreign key needs the key it references — the keys are one wave and the foreign keys the next, the reverse for a drop. Each statement still commits on its own, so a killed pass keeps what it finished. --constraints-work-mem sets maintenance_work_mem for the duration of each statement. Most servers default to 64MB, at which an index build over a large table spills to an external merge sort; measured over 370M rows the pass took as long as loading every row. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Wait for the relayer, not the reader node, in devenv The container wait matched "serving gRPC", which the reader node logs for its own port well before the relayer serves the one tier1 connects to, and the host-side dial that followed proves nothing: Docker binds the mapped port when the container starts. Tier1 starting in that window gets a live stream with no partial blocks at all, which TestPartialBlocksWithStores saw as every block arriving full, with no undos.
maoueh
approved these changes
Aug 17, 2026
Comment on lines
+22
to
+29
| # tests_e2e is its own Go module, so `go test ./...` above never reaches it. It stays | ||
| # out of this script until it passes on Linux: the node writes root-owned files into | ||
| # the bind-mounted t.TempDir, which fails every container test's cleanup on a CI | ||
| # runner. See the fix/tests-e2e-on-linux branch. | ||
| # | ||
| # pushd tests_e2e &> /dev/null | ||
| # go test ./... "$@" | ||
| # popd | ||
| # popd &> /dev/null |
Contributor
There was a problem hiding this comment.
Useless, still not running ...
| setupFlags.Bool("system-tables-only", false, "[DatabaseChanges mode] will only create/update the systems tables (cursors, substreams_history) and ignore the schema from the manifest") | ||
| setupFlags.Bool("ignore-duplicate-table-errors", false, "[DatabaseChanges mode][Dev] Use this if you want to ignore duplicate table errors, take caution that this means the 'schema.sql' file will not have run fully!") | ||
| setupFlags.Bool("postgraphile", false, "Will append the necessary 'comments' on cursors table to fully support postgraphile") | ||
| setupFlags.Bool("system-tables-only", false, "will only create/update the systems tables (cursors, substreams_history) and ignore the schema from the manifest") |
Contributor
There was a problem hiding this comment.
Why we lost [DatabaseChanges] note, if it's because it's not adding system tables in both, we should list all possible system tables for both mode.
Comment on lines
+133
to
+139
| func stringFlag(cmd *cobra.Command, name string) string { | ||
| if sflags.FlagDefined(cmd, name) { | ||
| return sflags.MustGetString(cmd, name) | ||
| } | ||
|
|
||
| return "" | ||
| } |
Contributor
There was a problem hiding this comment.
I'm thinking of adding those generated version on sflags now, maybe OptionalGetString with default, thinking about it.
| manifestPath, | ||
| outputModule, | ||
| "sink_database_changes", | ||
| sinkUserAgent("sink_database_changes", sinkPostgresDriver), |
Contributor
There was a problem hiding this comment.
sink_database_changes and even sinkUserAgent don't really use more "standard" agent formats which uses a lot the / for namespacing.
We should do the same + show full name postgres, clickhouse.
Contributor
There was a problem hiding this comment.
We should also embedded CLI version in the user agent too
| // addDatabaseChangesModeRunFlags registers the run flags that only apply when the | ||
| // selected module outputs DatabaseChanges. | ||
| func addDatabaseChangesModeRunFlags(flags *pflag.FlagSet) { | ||
| flags.Int("undo-buffer-size", 0, "[DatabaseChanges mode] If non-zero, handling of reorgs in the database is disabled. Instead, a buffer is introduced to only process blocks once they have been confirmed by that many blocks, introducing a latency but slightly reducing the load on the database when close to head. Set to 0 to enable reorg handling in the database (required for some databases like Postgres).") |
Contributor
There was a problem hiding this comment.
Are we sure some of them [DatabaseChanges mode] should keep those?
Comment on lines
+845
to
+846
| "This one outputs DatabaseChanges, where rows are written from the module's own database changes "+ | ||
| "and the schema is the 'schema.sql' bundled in the manifest") |
Contributor
There was a problem hiding this comment.
Each rejectFlags because should have a link to some docs reference.
Comment on lines
+824
to
+826
| if err := dropDatabaseConstraints(database); err != nil { | ||
| return err | ||
| } |
Contributor
There was a problem hiding this comment.
A confirmation should be asked just in case IMO
Comment on lines
+935
to
+937
| zlog.Warn("the module's output carries no schema.proto annotations, so its tables get no primary keys, no unique constraints and no foreign keys. " + | ||
| "Queries against them are sequential scans and duplicate ids are not rejected. " + | ||
| "`substreams tools extract-proto --sql` writes the annotated proto to start from, and --proto-file-override points the sink at it") |
Contributor
There was a problem hiding this comment.
Same here, revisit those integrity warnings/errors and ensure we have proper backlinking to docs or README or some other material.
Comment on lines
+1279
to
+1287
| func parseByteSize(cmd *cobra.Command, name string) (int64, error) { | ||
| raw := sflags.MustGetString(cmd, name) | ||
| parsed, err := humanize.ParseBytes(raw) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("invalid --%s %q: %w", name, raw, err) | ||
| } | ||
|
|
||
| return int64(parsed), nil | ||
| } |
Contributor
There was a problem hiding this comment.
Very good candidate for sflags addition too.
Comment on lines
+1289
to
+1296
| // flagChanged reports whether the operator actually typed the flag, as opposed to it | ||
| // carrying its default. Every mode and deprecation check keys off this: a default value | ||
| // must never be read as an instruction. | ||
| func flagChanged(cmd *cobra.Command, name string) bool { | ||
| flag := cmd.Flags().Lookup(name) | ||
|
|
||
| return flag != nil && flag.Changed | ||
| } |
Contributor
There was a problem hiding this comment.
sflags has Provided variant on all getter MustGetBoolProvided returns two value T, bool where second is if the flag.Changed
(Maybe Changed would be a better word in sflags actually).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The from-proto SQL sink wrote to the database on the stream's own goroutine, with the
constraints in place while it loaded, through a flag surface that had grown to say
different things depending on the driver and on where the stream was. This changes all
three, and adds the index the reorg path always needed.
What changes
Rows go through a spool. They land on local disk, pre-encoded into whatever the target
can take unchanged, and a background goroutine applies whole segments. The stream stops
waiting on the database, and blocks already downloaded survive a restart instead of being
streamed — and paid for — twice. Against
erc20-balance-changesover 50,000 Ethereummainnet blocks (17M rows) into a containerised PostgreSQL 17, this took the run from 190.9s
to 70.9s and the stream from 262 to 673 messages per second, with an identical table.
Both drivers spool. ClickHouse keeps the guarantees it already had — no transactions, its
cursor in a file — so applying a segment there is the same inserts followed by the same
cursor write, in the same order.
Constraints are created after the load, not during it. Measured through binary COPY
over 500k rows, loading with foreign keys in place costs 27.7x against 3.3x for building
the same ones afterwards.
--apply-constraintssays when:auto(default) has the sink doit when the stream reaches chain HEAD,
manualleaves it tosink postgres constraints apply,alwayskeeps the old behaviour.autois the default despite being astop-the-world pass — a backfill that ends with no primary keys and no foreign keys is a
silent wrong result that looks like success.
A stop block deliberately does not trigger it. Reaching
-tends the run without sayingthe backfill is done — a range is routinely one chunk of several — and building the
constraints there would leave every later chunk loading into a constrained schema, the
27.7x case this exists to avoid. The run says on the way out what it left behind.
Each statement commits on its own, so a run killed by the OOM killer keeps what it
finished.
--constraints-parallelismruns them side by side (keys first, then the foreignkeys that reference them), and
--constraints-work-memsetsmaintenance_work_memfor theduration of each: at the usual 64MB default an index build over a large table spills to an
external merge sort.
_block_number_is indexed, at startup, concurrently. Every table carries that columnand every reorg deletes from every table by it, and a foreign key indexes only its
referenced side — so each undo was a sequential scan per table. Measured over 10GiB: 2.6s
to build, 218MiB against 10GiB, and one table's undo goes from 1.296s to 15ms. It is
deliberately not part of the constraint pass: those describe the schema and are the
operator's to schedule, where this one the sink depends on.
--write-modereplaces guessing. Which write path ran used to be derived from whethera directory flag was set and whether the schema's foreign keys happened to form a cycle,
and an unorderable schema was silently downgraded to row-at-a-time inserts — an order of
magnitude slower, announced only in a log line. It is now
copy,batch-insert,row-insertorauto, and an explicit mode the driver or schema cannot support is anerror that names the fix.
Flags are named after what they spend.
--decode-*is CPU,--db-write-*is onecommit to the database,
--spool-*is how far ahead of it the stream may run. Commit sizeis not a setting: the sink measures each one and sizes the next segment toward
--db-write-target-duration, block payloads differing by orders of magnitude across chainsbeing exactly what makes a fixed block count give unstable durations.
substreams tools extract-proto --sqlwrites a module's output proto back out withevery SQL annotation commented out beside the message and field it applies to, and the
annotations file next to it so the result parses as-is. Starting from a package whose
author never annotated it was otherwise a matter of finding the right proto and the right
extension names by hand.
ClickHouse takes a package with no annotations. It required
order_by_fieldson everytable, so a Substreams not written for the sink could neither be set up nor run. Tables now
default to
ORDER BY (_block_number_, _row_id_),PRIMARY KEY (_block_number_)andPARTITION BY (toYYYYMM(_block_timestamp_))._row_id_numbers the rows one block writesto one table: the tables are
ReplacingMergeTree, so a sorting key that cannot tell thoserows apart collapses a whole block into one row at merge time. It is added only where no
order_by_fieldsis declared, and PostgreSQL is untouched. Three things were in the waybehind the error message — the ClickHouse database was built with "use proto options"
hardcoded to true, so an unannotated package inserted nothing at all;
setupexited withFlag "sink-info-folder" does not exist, those flags being registered on the run commandonly; and an undo wrote tombstones without
_row_id_, landing them on a different sortingkey where they removed nothing.
erc20-balance-changes/map_balance_changesnow fills aClickHouse database as-is.
Reviewing
Fourteen commits, each building and vetting clean on its own. The first four are the shape
of it:
Encode rows in PostgreSQL's binary COPY wire formatHold rows on disk and apply whole segments in the backgroundCodecandApplierare the driver's businessWrite through the spool, load without constraintsCover the spool, the write modes and the constraint timingsThe rest are fixes and follow-ups found while using it, each self-contained: draining the
spool before the stop block, the startup report of missing constraints, committing the
constraint pass in pieces, the module argument on the schema commands, and the block number
index with its measurement.
Compatibility
--block-batch-sizeand--no-constraintsstill work for a release and warn. Everythingelse here is new, so nothing else needs an alias. A from-proto flag typed for a
DatabaseChanges Substreams is now an error rather than silently ignored, and so is the
reverse — a DatabaseChanges Substreams owns its schema, so
sink postgres constraintsrefuses it outright.
setupandconstraintstake the output module as an optional second argument, as the runcommand always has; a package with more than one candidate had no way to say which.
Exhaustive list of changes
Commands
Flags added
Flags deprecated / moved
Mode guards
Spool
Constraints
Block number index
Behaviour fixes
Observability
Tests and measurements
Docs
ClickHouse without proto annotations
Measured end to end
Uniswap v4 on Base,
map_events, blocks 25,350,988 → 36,712,209 (11,361,221 blocks), intocontainerised ClickHouse and PostgreSQL 17. Both engines wrote 370,507,376 rows and
agreed on every table to the row, which is the correctness result as much as the timing one.
The constraint pass costing as much as loading all 370M rows is what makes the stop-block
change worth having: a ten-chunk backfill used to pay that at every chunk boundary, and
have every chunk after the first load the 27.7x path.
Per-block sink cost is now small enough that the run is usually delivery-bound: block
insert 0.02ms, entities 0.01ms, flush 0.00ms, with 77% of wall time waiting on the stream
over the full range.
Fixes folded in
#881 merged into this branch: twelve bugs found reviewing the work above, each with a
regression test checked to fail without its fix — among them the ClickHouse cursor never
reaching its file while spooling,
_blocks_rows racing the applier, a descriptor leakedper sealed COPY segment, reorgs failing on cyclic schemas, backslashes doubled in rendered
literals, and empty or enum arrays aborting a segment in the default write mode.