Fetch configuration from DB #3407
Open
Conversation
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
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
2 times, most recently
from
7f140e4 to
00660b8
Compare
iamKunalGupta
requested changes
Sep 9, 2025
serprex
reviewed
Sep 9, 2025
serprex
approved these changes
Sep 9, 2025
serprex
reviewed
Sep 9, 2025
| if flowConfigUpdate.SnapshotNumTablesInParallel > 0 { | ||
| state.SnapshotNumTablesInParallel = flowConfigUpdate.SnapshotNumTablesInParallel | ||
| } | ||
| cfg = syncStateToConfigProtoInCatalog(ctx, cfg, state.FlowConfigUpdate) |
There was a problem hiding this comment.
this > 0 behavior was added recently in #3435, this probably regresses
There was a problem hiding this comment.
given that we no longer have or rely on state, this should be a non-issue.
serprex
reviewed
Sep 9, 2025
serprex
reviewed
Sep 9, 2025
serprex
reviewed
Sep 9, 2025
flow/workflows/cdc_flow.go
Outdated
| ctx workflow.Context, | ||
| cfg *protos.FlowConnectionConfigs, | ||
| flowJobName string, | ||
| // cfg *protos.FlowConnectionConfigs, |
There was a problem hiding this comment.
this breaks backwards compatibility. Should still take config & pull name from that to maintain function signature
There was a problem hiding this comment.
The problem with keeing the protos.FlowConnectionConfig here, is that in cases where this is too large - it will still fail.
Is the suggestion to pass both and then remove the cfg from here?
There was a problem hiding this comment.
no, only pass cfg. new code should pass an empty cfg with only FlowJobName present
There was a problem hiding this comment.
OK understood, I'll go in this direction then :)
serprex
reviewed
Sep 9, 2025
| } | ||
| // because we have renamed the tables. | ||
| cfg.TableMappings = state.SyncFlowOptions.TableMappings | ||
| return nil, errors.New("cannot start CDCFlow with Resync enabled, please drop the flow and start again") |
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
2 times, most recently
from
20f7fd4 to
0b42cca
Compare
serprex
reviewed
Sep 15, 2025
Comment on lines
21
to
24
| func FetchConfigFromDB(flowName string) (*protos.FlowConnectionConfigs, error) { | ||
| var configBytes sql.RawBytes | ||
| dbCtx := context.Background() | ||
| pool, _ := GetCatalogConnectionPoolFromEnv(dbCtx) |
serprex
reviewed
Sep 15, 2025
| ) (*protos.SetupFlowOutput, error) { | ||
| s.Info("executing setup flow") | ||
| // gotta fetch the config from the catalog. | ||
| config, err := internal.FetchConfigFromDB(flowJobName) |
There was a problem hiding this comment.
can't call this in workflows
could maybe run as side effect, but then it ends up in history
There was a problem hiding this comment.
Why can this not be called in workflows? 🤔
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
from
0b42cca to
e7712b1
Compare
No longer passing the entire configuration object to the Temporal job as we can hit the 2MB hard limit imposed by Temporal. Instead, pass flowJobName and fetch the config from the DB in all the places required. There are some additional changes that were required to be made, as some of the jobs are used from multiple contexts - and I had to adapt some of the information passed to ensure that they continue to work as expected.
alexstoick
force-pushed
the
feat/fetch-config-from-catalog
branch
from
e7712b1 to
962a495
Compare
alexstoick
requested a deployment
to
external-contributor
GitHub Actions
Waiting
alexstoick
requested a deployment
to
external-contributor
GitHub Actions
Waiting
alexstoick
requested a deployment
to
external-contributor
GitHub Actions
Waiting
alexstoick
requested a deployment
to
external-contributor
GitHub Actions
Waiting
alexstoick
added a commit
to alexstoick/peerdb
that referenced
this pull request
Oct 10, 2025
This uses a new helper in `internal.FetchConfigFromDB` to fetch a fully hydrated `protos.FlowConnectionConfigs`. When passing this config to Temporal we strip the `tableMappings` array element which can cause it to go over the 2MB limit. This contains a subset of the changes which were originally proposed in: PeerDB-io#3407
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 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.
No longer passing the entire configuration object to the Temporal job as we can hit the 2MB hard limit imposed by Temporal. Instead, pass flowJobName and fetch the config from the DB in all the places required.
There are some additional changes that were required to be made, as some of the jobs are used from multiple contexts - and I had to adapt some of the information passed to ensure that they continue to work as expected.