Copy passthrough during snapshot mode if no transformers are configured - #1129
Closed
kvch wants to merge 1 commit into
Closed
Copy passthrough during snapshot mode if no transformers are configured#1129kvch wants to merge 1 commit into
kvch wants to merge 1 commit into
Conversation
CoverageTotal: 62.8% (+0.1% vs Coverage in packages changed by this PR:
|
tsg
approved these changes
Sep 3, 2026
| if cfg.RetryPolicy.DisableRetries { | ||
| targetConn, err = pglib.NewConnPool(ctx, cfg.TargetURL, poolOpts...) | ||
| } else { | ||
| targetConn, err = pglibretrier.NewQuerier(ctx, cfg.RetryPolicy, func(ctx context.Context) (pglib.Querier, error) { |
Member
There was a problem hiding this comment.
Do we need retries in this case? From my agent:
The target pool is wrapped in the retrier, which re-runs the ExecInTx closure on a retriable error. The pipe reader is already partially consumed by then, so the retry sends a truncated COPY and fails with the row count mismatch, then retries again until the backoff gives up. Nothing corrupt lands because the target tx rolls back, but the retry loop burns time and logs misleading warnings. The passthrough should probably disable retries on the target conn and let the existing page range failure abort the table, or retry the whole range from the source.
kvch
force-pushed
the
snapshot-copy-passthrough
branch
from
September 10, 2026 12:11
efa8c09 to
57ab6a3
Compare
kvch
force-pushed
the
snapshot-copy-passthrough
branch
from
September 10, 2026 12:57
57ab6a3 to
51908de
Compare
Collaborator
Author
|
The PR has diverged a lot from main. I am putting up new stacked prs, and closing this one. |
9 tasks
kvch
added a commit
that referenced
this pull request
Sep 10, 2026
### Description Three small extractions so that a second component can reuse the postgres target behaviour the bulk ingest writer already implements: how a failure is classified as retriable, which retry policy applies once the default is resolved, and how large the concurrent COPY budget may be. Nothing changes about what any of them decide. Each is a move or an extraction, verified by the existing tests. This is the first of four stacked PRs splitting #1129. On its own it adds no functionality; the consumer arrives in the branch above it. ##### Related Issue(s) - Related to #1129 #### Type of Change - [x] 🔧 Refactoring (no functional changes) #### Changes Made - Extract the retrying querier's classification into an exported `retrier.IsRetriableError` - Move `CopyBudgetReserve` and `CopyBudgetSize` from `pkg/wal/processor/postgres` to `internal/sync`, next to the weighted semaphore they size - Add `Config.EffectiveRetryPolicy`, returning the postgres writer's retry policy once its default has been applied. `retryPolicy` stays unexported and remains the single definition. #### Testing - [ ] Unit tests added/updated - [ ] Integration tests added/updated - [ ] Manual testing performed - [ ] All existing tests pass #### Checklist - [x] Code follows project style guidelines - [x] Self-review completed - [x] Code is well-commented - [ ] Documentation updated where necessary
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
2 participants
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.
Description
This PR adds support for snapshotting a table by piping the source's
COPY ... TO STDOUTstraight into the target'sCOPY ... FROM STDINwhen the rows do not need to be parsed. When a transformer, filter, injector or sanitizer is configured, we fall back to the previous behaviour.Previously, we read read rows, decoded every value into Go values via pgx, wrapped them in
wal.Eventstructs, passed them through the processor chain to the bulk ingest writer, which re-encoded them into aCOPY. Both ends were alreadyCOPY; only the round trip through Go in the middle is removed. I measured improvement on ~17% on a text heavy table. It is a workload-specific improvement, since the saving is largest where decoding allocates most. So it can be higher for some cases.Furthermore, #801 (cube) and #1035 (enum) exist because pgx cannot produce those types' binary encodings, which is why
textOnlyCopyTypesandneedsTextCopyexist. Server-to-server COPY never touches a pgx codec, so that family of bugs cannot occur on this path.Builds on the raw COPY stream primitives from #1126.
Please review commit by commit.
feat: stream snapshot rows without decoding them: the new logicrefactor: make the page range reader a strategy:pg_snapshot_range_snapshotter.golooks like a new file but isprocessTableRangerelocated: three receiver renames and one inlined variableType of Change
Changes Made
Commit 1: the passthrough
io.Pipe, so memory is bounded by the pipe rather than the page range, and each side closes its end with its error so neither can block on the otherCommit 2: the strategy extraction
newSnapshotGenerator, which takes the reader as an argument, so no path can leave it unset.Testing
Checklist