CDC to S3/GCS #507
CDC to S3/GCS #507
Conversation
Amogh-Bharadwaj
force-pushed
the
s3-cdc
branch
from
18dac94
to
3b62078
Compare
flow/connectors/s3/metadata.go
Outdated
| pool *pgxpool.Pool | ||
| } | ||
|
|
||
| func NewPostgresMetadataStore(ctx context.Context, pgConfig *protos.PostgresConfig) (*PostgresMetadataStore, error) { |
There was a problem hiding this comment.
can we share this impl with eventhubs somehow?
flow/connectors/s3/s3.go
Outdated
| var firstCP int64 = 0 | ||
| tableNameRowsMapping := make(map[string]uint32) | ||
|
|
||
| for _, record := range req.Records.Records { |
There was a problem hiding this comment.
don't we do this elsewhere too? can we refactor to share the code?
Amogh-Bharadwaj
force-pushed
the
s3-cdc
branch
from
be6ad20
to
9ff768b
Compare
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
flow/connectors/eventhub/eventhub.go
Outdated
| @@ -40,7 +41,9 @@ func NewEventHubConnector( | |||
| } | |||
|
|
|||
| hubManager := NewEventHubManager(ctx, defaultAzureCreds, config) | |||
| pgMetadata, err := NewPostgresMetadataStore(ctx, config.GetMetadataDb()) | |||
| metadataSchemaName := "peerdb_eventhub_metadata" | |||
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
G101: Potential hardcoded credentials (gosec)
| @@ -167,11 +161,10 @@ func (c *EventHubConnector) GetLastSyncBatchID(jobName string) (int64, error) { | |||
| } | |||
|
|
|||
| // update offset for a job | |||
| func (c *EventHubConnector) updateLastOffset(jobName string, offset int64) error { | |||
| ms := c.pgMetadata | |||
| func (p *PostgresMetadataStore) UpdateLastOffset(jobName string, offset int64) error { | |||
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
unnecessary leading newline (whitespace)
| @@ -206,14 +199,13 @@ func (c *EventHubConnector) updateLastOffset(jobName string, offset int64) error | |||
| } | |||
|
|
|||
| // update offset for a job | |||
| func (c *EventHubConnector) incrementSyncBatchID(jobName string) error { | |||
| ms := c.pgMetadata | |||
| func (p *PostgresMetadataStore) IncrementID(jobName string) error { | |||
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
unnecessary leading newline (whitespace)
flow/connectors/utils/stream.go
Outdated
| } | ||
|
|
||
| first := true | ||
| var firstCP int64 = req.CP |
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
ST1023: should omit type int64 from declaration; it will be inferred from the right-hand side (stylecheck)
flow/connectors/eventhub/eventhub.go
Outdated
| tableSchemas map[string]*protos.TableSchema | ||
| creds *azidentity.DefaultAzureCredential | ||
| hubManager *EventHubManager | ||
| } | ||
|
|
||
| const ( | ||
| metadataSchemaName = "peerdb_eventhub_metadata" |
There was a problem hiding this comment.
🚫 [golangci] reported by reviewdog 🐶
G101: Potential hardcoded credentials (gosec)
Amogh-Bharadwaj
force-pushed
the
s3-cdc
branch
from
af9e493
to
6e4b5e0
Compare
Initial Load: One folder per table
CDC is a separate folder
Tests added for both S3 and GCS versions of the peer (CDC)