perf(json): stream CDC row cleaning to cut migration CPU#54
Open
niteshvijay1995 wants to merge 3 commits into
Open
perf(json): stream CDC row cleaning to cut migration CPU#54niteshvijay1995 wants to merge 3 commits into
niteshvijay1995 wants to merge 3 commits into
Conversation
Replace the per-row JsonDocument DOM parse in CdcJsonRowParser with a single forward-only Utf8JsonReader -> Utf8JsonWriter pass that strips the synthetic __sys_* system columns and extracts writetime/TTL without building a document tree, LINQ closures, or a per-row MemoryStream/ ToArray copy. - Reusable pooled buffers (ArrayBufferWriter + Utf8JsonWriter.Reset, scratch/transcode arrays) are allocated once per page instead of per row; PageReader now creates one parser per page and reuses it. - String/number values are raw-copied, preserving the source's exact representation and escaping with no managed string per value. - __sys_clttl array handling and the defensive __sys_* guard are column-order-independent (SELECT JSON * does not guarantee order). Behavior is unchanged: verified byte-for-byte identical cleaned output and identical metadata over 200k varied synthetic rows (unicode/4-byte emoji, embedded escapes, nested UDT/map/list, decimals, exponents, nulls, interleaved and unknown __sys_* columns), and end-to-end against a 100k-row source table (all primitive types) with full row parity. A parse-only microbenchmark shows ~32% less CPU time per row. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR optimizes the JSON CDC migration path by replacing JsonDocument-based per-row parsing with a streaming Utf8JsonReader → Utf8JsonWriter pass that strips __sys_* columns and extracts writetime/TTL metadata during the same traversal, aiming to reduce CPU and allocations during SELECT JSON * → clean → INSERT JSON.
Changes:
- Reworked
CdcJsonRowParserinto a reusable instance that performs a single-pass streaming clean+metadata extraction. - Updated
PageReader.ReadJsonPageAsyncto allocate one parser per page and reuse it across the row loop.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
| CassandraMigrationProcessor/DataTransfer/PageReader.cs | Creates one CdcJsonRowParser per JSON page and reuses it across rows. |
| CassandraMigrationProcessor/CassandraDriver/CdcJsonRowParser.cs | Replaces DOM parsing with streaming read/write, extracting metadata and stripping system columns in one pass. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+149
to
+151
| // One reusable parser per page: its pooled buffers are reset (not | ||
| // reallocated) per row, so a page of N rows amortizes the parser's | ||
| // allocations across the whole page instead of paying them per row. |
Comment on lines
+25
to
+30
| /// Performance: metadata extraction and system-column stripping are | ||
| /// done in a <em>single</em> traversal of the parsed row, writing the | ||
| /// cleaned envelope into a reusable pooled buffer instead of a fresh | ||
| /// <c>MemoryStream</c> per row. Instances are cheap and intended to be | ||
| /// reused across all rows in a page (they are not thread-safe — use one | ||
| /// per reading loop). |
Comment on lines
169
to
+171
|
|
||
| return new CdcRowMetadata(writetime, expiryEpochSeconds); | ||
| _writer.WriteEndObject(); | ||
| _writer.Flush(); |
| } | ||
|
|
||
| // Regular column: emit name + value into the cleaned envelope. | ||
| _writer.WritePropertyName(reader.GetString()!); |
| writer.WriteStartObject(); | ||
| while (reader.Read() && reader.TokenType != JsonTokenType.EndObject) | ||
| { | ||
| writer.WritePropertyName(reader.GetString()!); |
Address review feedback while preserving the CPU-reduction goal: - Property names are now written directly from the reader's UTF-8 span in the common (non-escaped, single-segment) case, removing a managed string allocation per property on the hot path. Escaped or multi-segment names fall back to GetString(). Both routes go through the writer's encoder, so output is byte-identical (verified over unicode, emoji, and HTML-escaped names). - Parse() now rejects trailing content after the root object, matching JsonDocument.Parse semantics instead of silently ignoring extra tokens. - Reworded 'pooled' buffer comments to 'reused' to accurately describe the reused ArrayBufferWriter/byte[] fields (no ArrayPool rental). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Collaborator
Author
|
Thanks for the review — addressed in 1c0013e:
Re-verified: byte-identical property-name encoding, order-independent |
Previously the parser silently fell back to reader.Skip() when a system metadata column had an unexpected shape (a non-numeric __sys_rw_ts, a non-array __sys_clttl, or a non-numeric __sys_clttl[0] expiry). That dropped writetime/TTL metadata without any signal, so a source-contract violation would produce silently-degraded rows on the destination. The parser now throws a JsonException in each of these cases. The single caller (PageReader.ReadJsonPageAsync) does not catch it, so the exception propagates and faults the job — surfacing the problem instead of hiding it. Intentional (non-error) skips are preserved and documented: the opaque detail object trailing __sys_clttl[0], and the defensive strip of any unknown future __sys_* column that the destination schema cannot accept. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
Comments suppressed due to low confidence (2)
CassandraMigrationProcessor/CassandraDriver/CdcJsonRowParser.cs:130
- This throws if
__sys_clttlis present but not an array. Previously, malformed TTL metadata was ignored (treated as absent) while still stripping the__sys_*column; throwing here can fail the whole migration on a single bad row. If strictness is desired, the PR description/behavior contract should be updated; otherwise consider tolerating unexpected shapes by skipping and continuing.
if (reader.TokenType != JsonTokenType.StartArray)
throw new JsonException(
$"Malformed row payload: '{SysCellLevelTtlColumn}' was " +
$"{reader.TokenType}, expected an array.");
CassandraMigrationProcessor/CassandraDriver/CdcJsonRowParser.cs:142
- This throws if
__sys_clttl[0]is not a Number. The previous parser treated unexpected metadata shapes as "no TTL"; throwing here can make migrations brittle if the source ever emits null/0/other unexpected tokens. Consider only extracting expiry when the first element is a Number and otherwise skipping any nested content and continuing.
if (reader.TokenType != JsonTokenType.Number)
throw new JsonException(
$"Malformed row payload: '{SysCellLevelTtlColumn}'[0] " +
$"was {reader.TokenType}, expected the expiry epoch as a Number.");
long expiry = reader.GetInt64();
Comment on lines
+112
to
+116
| if (reader.TokenType != JsonTokenType.Number) | ||
| throw new JsonException( | ||
| $"Malformed row payload: '{SysRwTimestampColumn}' was " + | ||
| $"{reader.TokenType}, expected a Number."); | ||
| writetime = reader.GetInt64(); |
Collaborator
Author
There was a problem hiding this comment.
Thanks — this is a deliberate design choice rather than an oversight, so I''ll keep the throw.
The distinction the change draws is absence vs. corruption:
- Absent
__sys_rw_ts/__sys_clttl(the common case — e.g. a latest-version change feed that doesn''t surface these columns) is still tolerated. The property name simply isn''t encountered, sowritetime/expirystay null, the cleaned row is returned normally, and the destination write falls back towritetime = now/TTL = 0. I verified this against a live change-feed source: neither column is emitted for standard rows and migration proceeds unaffected. - Present but malformed (e.g.
__sys_rw_tsas a string, or__sys_clttlnot an array) is the only case that now throws. When the source does emit these columns their shape is a hard contract (writetime = number, clttl =[epoch, {details}]). Silently coercing a malformed value to "missing" would drop per-row TTL/writetime without any signal, producing rows on the destination that look fine but have lost fidelity — exactly the kind of silent data-correctness regression we want to avoid.
So absence never fails a row; only a genuine contract violation does, and that failure is surfaced instead of hidden. I''ll update the PR description to make this explicit (the "behavior unchanged" note refers to the cleaned-envelope output for well-formed rows, not to malformed-metadata handling).
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.
Summary
Reduces the CPU cost of the JSON migration data path (
SELECT JSON *→ clean →INSERT JSON) by replacing the per-rowJsonDocumentDOM parse inCdcJsonRowParserwith a single forward-onlyUtf8JsonReader→Utf8JsonWriterpass. The parser strips the synthetic__sys_*system columns and extracts writetime / per-row TTL in one traversal, with no document tree, no LINQ closure, and no per-rowMemoryStream/ToArray()copy.What changed
CdcJsonRowParseris now a reusable instance class:ArrayBufferWriter<byte>+Utf8JsonWriter.Reset(), plus scratch/transcodebyte[]fields — reused, notArrayPool-rented) are allocated once per page rather than per row.ValueSpanin the common non-escaped, single-segment case, avoiding a managed-string allocation per property; escaped/multi-segment names fall back toGetString(). Both routes go through the writer's encoder, so output is byte-identical.Utf8JsonWriterruns withSkipValidation = true(input is already structurally valid).__sys_clttlarray handling and the defensive__sys_*guard are column-order-independent —SELECT JSON *does not guarantee that system columns come last.ParseOnceconvenience for one-shot callers.PageReader.ReadJsonPageAsynccreates one parser per page and reuses it across the row loop.Behavior
For well-formed rows the cleaned envelope and extracted metadata are byte-identical to the previous parser; the destination
INSERT JSONstores the same data. Two error paths are made stricter (previously silent), so a corrupt payload surfaces instead of being hidden:JsonDocument.Parse), rather than being silently ignored.__sys_rw_tsthat is present but not a JSON number, or a__sys_clttlthat is present but not the expected[epoch, {details}]array — now throws instead of silently discarding the value. Absence of these columns is still fully tolerated: the row is cleaned normally and the write falls back towritetime = now/TTL = 0. Only a present-but-corrupt value fails, so per-row TTL/writetime can never be dropped without a signal.Validation
__sys_*columns interleaved in the middle plus an unknown__sys_extra): output byte-for-byte identical to the previous parser, metadata identical. Property-name span vs.GetString()encoding verified identical across unicode, emoji, and HTML-escaped names.__sys_rw_ts/__sys_clttlshapes and trailing/second-object payloads throw; well-formed rows (including absent sys columns and empty__sys_clttl) parse with correct metadata.ExtractMetadata/StripSysColumns).Performance
A parse-only microbenchmark over the representative dataset shows ~32% less CPU time per row, with the DOM-tree allocation removed from the hot path. Wall-clock throughput is unchanged when a migration is source-read bound, so the win shows up as CPU headroom on CPU-bound deployments.