fix(postgres): avoid json marshal roundtrip for json columns

[itsbilal]

Previously we did a json unmarshal/marshal with a "relaxed number" extension to convert any large numbers that don't fit in a float into a string. Other than the large number case there was no reason to parse the whole document, and for large json values this was starting to greatly increase CDC time spent per row.

This change significantly reduces this processing overhead by walking through the json value token-by-token instead and just swapping out any numbers that can't be cast to float with a stringified version of them instead.

[dtunikov]

can we add a simple benchmark showing that this new implementation is more performant that the previous one?
yeah, I know that it must be faster since we don't do unmarshal/marshal roundtrip anymore
but still would be good to be sure that encoding/json/jsontext doesn't have any weird hidden costs

[dtunikov]

we don't even have to merge this benchmark code (since we probably want to get rid of jsoniter dependency)
but would be nice to see some results

[itsbilal]

I wrote one, will put it up in a separate PR as it's the same overall structure as the one I used in #4722. There's a ~17% improvement in end-to-end observed sync times, but since that includes all of sync and normalize too, the actual performance gain in PullRecords is more significant:

                         │ ../pg_before2.bench │          ../pg_after2.bench          │
                         │       sec/op        │   sec/op     vs base                 │
PostgresClickHouseCDC-18            88.20 ± 4%   74.78 ± 14%  -15.21% (p=0.005 n=7+6)

                         │    ../pg_before2.bench    │                 ../pg_after2.bench                 │
                         │ catchup_after_insert_s/op │ catchup_after_insert_s/op  vs base                 │
PostgresClickHouseCDC-18                  72.51 ± 6%                 58.89 ± 18%  -18.79% (p=0.002 n=7+6)

                         │ ../pg_before2.bench │           ../pg_after2.bench            │
                         │   replicate_s/op    │ replicate_s/op  vs base                 │
PostgresClickHouseCDC-18            88.20 ± 4%      74.78 ± 14%  -15.22% (p=0.005 n=7+6)

                         │ ../pg_before2.bench │          ../pg_after2.bench           │
                         │       rows/s        │    rows/s     vs base                 │
PostgresClickHouseCDC-18           22.68k ± 4%   26.75k ± 12%  +17.95% (p=0.005 n=7+6)

[dtunikov]

Oh, I think you mean e2e benchmark, right?
I meant a simple unit test like benchmark that compares:
call convertWithRelaxedNumbers + strings.Builder for arrays vs call json.Marshal + json.Unmarshal
(on a raw bytes stream)

[ilidemi]

+1 to decode+encode benchmark. With just the relaxed number addition it was 5x difference for me, but curious what is the hit from unicode normalization (which we have to do, it seems) and the object decoding would show up

[itsbilal]

Added this benchmark to this PR - see flow/connectors/postgres/json_test.go for the code. Here's the difference I see when running this on my laptop. Looks like deeper JSON objects benefit more than shallower ones do, but we're still not at 5x (guessing it might be the unicode normalization?):

bilal.akhtar@Mohammeds-MacBook-Pro flow % benchstat ./bench_oldmod.bench ./bench_new.bench
goos: darwin
goarch: arm64
pkg: github.com/PeerDB-io/peerdb/flow/connectors/postgres
cpu: Apple M5 Max
                                                │ ./bench_oldmod.bench │         ./bench_new.bench          │
                                                │        sec/op        │   sec/op     vs base               │
ConvertRelaxedNumber/numFields=4/maxDepth=32-18           40.31m ±  6%   23.31m ± 3%  -42.18% (p=0.002 n=6)
ConvertRelaxedNumber/numFields=8/maxDepth=4-18            31.73µ ±  1%   23.86µ ± 1%  -24.78% (p=0.002 n=6)
ConvertRelaxedNumber/numFields=8/maxDepth=8-18            7.507m ±  3%   4.816m ± 3%  -35.84% (p=0.002 n=6)
ConvertRelaxedNumber/numFields=8/maxDepth=16-18            9.692 ±  6%    6.314 ± 3%  -34.85% (p=0.002 n=6)
ConvertRelaxedNumber/numFields=64/maxDepth=2-18           9.303m ± 16%   8.645m ± 1%        ~ (p=0.065 n=6)
ConvertRelaxedNumber/numFields=64/maxDepth=4-18            3.351 ±  1%    3.232 ± 2%   -3.55% (p=0.002 n=6)

[ilidemi]

Yeah, looks like it's unicode normalization (and encoding overall, which only does normalization for us), and also seems different data results in different perf.

On the same data as above but copying bytes and inserting quotes instead of calling the encoder, 1.75-2.59x become 2.97-3.85x. Data that resulted in around 5x for me (GH issues/commits/events), comes to just below 3x on this impl. The difficulty with copying bytes is that bad unicode becomes unqueryable in CH, but thanks to encoder we can do AllowInvalidUTF8(false) on the decoder, do the bytes approach, and if decoder fails, start over with AllowInvalidUTF8(true) and the encoder, at a cost of being slower than this impl and in really bad cases even compared to main (but imo it's fine and we can put in telemetry).

Full results (Legacy = main, Committed = this PR, Raw patch = just copy bytes, Raw + repair = copy bytes and restart on bad unicode):

Payload	Size	Legacy	Committed	Raw patch	Raw + repair
Small	55 B	969.0 ns	525.5 ns · 1.84×	312.5 ns · 3.10×	313.5 ns · 3.09×
Small + long number	1.0 KB	4.25 µs	2.84 µs · 1.49×	2.32 µs · 1.83×	2.23 µs · 1.90×
1 MiB	1.05 MB	990.0 µs	921.5 µs · 1.07×	598.7 µs · 1.65×	589.0 µs · 1.68×
1 MiB + long number	1.05 MB	965.9 µs	967.2 µs · 1.00×	605.1 µs · 1.60×	572.4 µs · 1.69×
Issues	51.8 KB	246.1 µs	82.6 µs · 2.98×	44.7 µs · 5.50×	44.5 µs · 5.53×
Issues + long number	52.8 KB	254.7 µs	84.9 µs · 3.00×	48.1 µs · 5.29×	48.1 µs · 5.30×
Issues + invalid Unicode	51.8 KB	244.7 µs	84.5 µs · 2.90×	45.4 µs · 5.39×	109.3 µs · 2.24×
Commits	80.3 KB	293.9 µs	102.1 µs · 2.88×	52.6 µs · 5.59×	52.5 µs · 5.60×
Commits + long number	81.3 KB	294.6 µs	102.9 µs · 2.86×	58.1 µs · 5.07×	58.3 µs · 5.05×
Commits + invalid Unicode	80.3 KB	297.0 µs	112.2 µs · 2.65×	57.9 µs · 5.13×	141.4 µs · 2.10×
Events	411.2 KB	1.75 ms	654.8 µs · 2.68×	353.7 µs · 4.96×	353.2 µs · 4.96×
Events + long number	412.2 KB	1.79 ms	656.3 µs · 2.73×	370.1 µs · 4.84×	369.3 µs · 4.85×
Events + invalid Unicode	411.2 KB	1.75 ms	652.4 µs · 2.69×	355.7 µs · 4.92×	856.8 µs · 2.04×
Escaped Unicode	1.05 MB	2.19 ms	2.92 ms · 0.75×	886.1 µs · 2.47×	888.6 µs · 2.46×
Invalid UTF-8	1.05 MB	976.9 µs	2.22 ms · 0.44×	696.3 µs · 1.40×	2.59 ms · 0.38×
Lone surrogate	1.05 MB	3.07 ms	2.07 ms · 1.48×	624.3 µs · 4.91×	2.30 ms · 1.34×
Generated 4/32	13.47 MB	44.24 ms	21.41 ms · 2.07×	13.80 ms · 3.21×	13.81 ms · 3.20×
Generated 8/4	10.5 KB	35.2 µs	20.1 µs · 1.75×	11.8 µs · 2.99×	11.6 µs · 3.04×
Generated 8/8	2.54 MB	8.07 ms	4.07 ms · 1.98×	2.72 ms · 2.97×	2.71 ms · 2.98×
Generated 8/16	3.36 GB	15.66 s	6.34 s · 2.47×	4.12 s · 3.80×	4.93 s · 3.18×
Generated 64/2	2.30 MB	9.41 ms	3.73 ms · 2.53×	2.53 ms · 3.72×	2.54 ms · 3.70×
Generated 64/4	930.15 MB	3.92 s	1.51 s · 2.59×	1.03 s · 3.80×	1.02 s · 3.85×

Can be reproduced by rerunning benchmark-postgres-json.sh on this branch (takes a few minutes)

Also robot flagged that jsontext has a fast path for *bytes.Buffer and the branch is using strings, so a microoptimization is possible.

[github-actions]

❌ Test Failure

Analysis: Not flaky — the build failed deterministically because unresolved git merge conflict markers were committed into flow/connectors/postgres/cdc.go (lines ~657-697), causing Go syntax errors that broke the flow-api/flow-worker image builds on every one of 3 retries in both matrix jobs, so no tests ever ran.
Confidence: 0.99

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[github-actions]

❌ Test Failure

Analysis: Not flaky — unresolved git merge conflict markers committed in flow/connectors/postgres/cdc.go cause a Go compile error, failing the peer-flow image build so flow-api never starts and no tests run, reproducing identically across all three matrix jobs and all retries.
Confidence: 0.99

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[claude]

Code review

Found 1 issue (compile error):

flow/connectors/postgres/json_test.go line 91 — convertWithRelaxedNumbers is declared with two parameters (func convertWithRelaxedNumbers(input io.Reader, sizeHint int) ([]byte, error)), but the call in testRelaxedNumber passes only one argument:

transformed, err2 := convertWithRelaxedNumbers(strings.NewReader(tc.input))

This fails to compile (not enough arguments in call to convertWithRelaxedNumbers), breaking the build of every test in the connpostgres package. All other call sites (cdc.go, qrep_query_executor.go) pass the size hint as the second argument. Suggested fix:

transformed, err2 := convertWithRelaxedNumbers(strings.NewReader(tc.input), len(tc.input))

Checked for bugs and CLAUDE.md compliance; no other issues found. (An earlier revision of this PR, 4daa598, contained unresolved merge-conflict markers in flow/connectors/postgres/cdc.go — those are already resolved in the current head 58c82ad.)

[claude]

Code review — found 2 issues (checked for bugs and CLAUDE.md compliance; no CLAUDE.md violations):

1. Compile error in flow/connectors/postgres/json_test.go line 91

convertWithRelaxedNumbers takes two arguments (input io.Reader, sizeHint int), but the call at json_test.go#L90-L92 passes only one. Since this file is in package connpostgres, the whole test package fails to build — none of the tests in flow/connectors/postgres (including the new json_roundtrip_test.go tests) can compile or run. Fix: convertWithRelaxedNumbers(strings.NewReader(tc.input), len(tc.input)).

2. Duplicate JSON object keys hard-fail the fast path — flow/connectors/postgres/json.go lines 98 and 105

jsontext defaults to AllowDuplicateNames == false (per RFC 7493), so dec.ReadToken() returns ErrDuplicateName on the second occurrence of a repeated object member name — and the encoder enforces the same on write (json.go#L97-L106). Postgres json columns store the input text verbatim and legally contain duplicate keys; the jsoniter path this replaces tolerates them last-wins, and this PR itself adds a DuplicateJsonKeysCounter because duplicates occur in real data.

With PEERDB_POSTGRES_FAST_PROCESS_JSON_COLUMNS=true, such a row makes convertWithRelaxedNumbers error, which propagates as failed to process json and aborts the CDC pull (cdc.go#L407-L412) or the QRep partition (qrep_query_executor.go#L539-L544), stalling the mirror on data the old path handled fine. This affects json and json[] columns (jsonb dedupes keys on write, so it cannot produce duplicates).

Suggested fix: pass jsontext.AllowDuplicateNames(true) to both the decoder (L98) and the encoder (L105). Note the resulting behavior difference: the fast path will then preserve duplicate keys verbatim, while the old jsoniter path collapsed them last-wins — probably worth adding a repeated-key case to jsonRoundtripCases() to pin down the intended semantics.

[github-actions]

❌ Test Failure

Analysis: Not flaky — a deterministic Go compile error (connectors/postgres/json_test.go:91:52: not enough arguments in call to convertWithRelaxedNumbers) broke the build identically in all three matrix jobs and on retry, so the tests never executed.
Confidence: 0.98

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[github-actions]

🔄 Flaky Test Detected

Analysis: The e2e binary hit its 20m wall-clock timeout (panic: test timed out after 20m0s) with zero assertion failures — all 270 reported failures are collateral from the panic, only 16 tests were mid-flight and progressing normally, and the other two matrix legs on the same commit passed at 947s/1012s against the 1200s limit, leaving too little headroom for a slower runner.
Confidence: 0.8

✅ Automatically retrying the workflow

View workflow run

[ilidemi]

Would that not regress us back where we started with full deserialization and reflection? I think it's ok to be submitting duplicate keys and let CH switch from last to first. The query is showing data that customer sent anyway, we're just delivering more of it.

[itsbilal]

This only applies in the old code path where we're already doing the full deserialization. I imagine you meant that we want to add telemetry to see how common this is in the wild, and if it's not common at all, then we don't need to worry about the subtle change in semantics when we switch to the new code path? Or if we don't really care about the semantics then we may as well go ahead and just do the AllowDuplicateKeys(true) option.

[ilidemi]

I am an idiot, apologies. Let's just go with AllowDuplicateKeys(true).

[ilidemi]

+1 to decode+encode benchmark. With just the relaxed number addition it was 5x difference for me, but curious what is the hit from unicode normalization (which we have to do, it seems) and the object decoding would show up

[github-actions]

❌ Test Failure

Analysis: A real, deterministic Go compile error in the PR's new flow/connectors/postgres/json_test.go ("declared and not used: i" at lines 164 and 174) fails the build identically across all three matrix jobs.
Confidence: 0.97

⚠️ This appears to be a real bug - manual intervention needed

View workflow run

[github-actions]

❌ Test Failure

Analysis: Deterministic build failure — connectors/postgres/json_test.go:164 and :174 use := with no new variables on the left, so the postgres connector test package fails to compile identically across all matrix jobs and on retry.
Confidence: 0.98

⚠️ This appears to be a real bug - manual intervention needed

View workflow run