-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: hmoazzem <[email protected]>
- Loading branch information
Showing
8 changed files
with
432 additions
and
0 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,188 @@ | ||
| { | ||
| "schema": { | ||
| "type": "struct", | ||
| "fields": [ | ||
| { | ||
| "type": "struct", | ||
| "fields": [ | ||
| { | ||
| "type": "int32", | ||
| "optional": false, | ||
| "field": "id" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "first_name" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "last_name" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "email" | ||
| } | ||
| ], | ||
| "optional": true, | ||
| "name": "PostgreSQL_server.inventory.customers.Value", | ||
| "field": "before" | ||
| }, | ||
| { | ||
| "type": "struct", | ||
| "fields": [ | ||
| { | ||
| "type": "int32", | ||
| "optional": false, | ||
| "field": "id" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "first_name" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "last_name" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "email" | ||
| } | ||
| ], | ||
| "optional": true, | ||
| "name": "PostgreSQL_server.inventory.customers.Value", | ||
| "field": "after" | ||
| }, | ||
| { | ||
| "type": "struct", | ||
| "fields": [ | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "version" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "connector" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "name" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": false, | ||
| "field": "ts_ms" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": false, | ||
| "field": "ts_us" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": false, | ||
| "field": "ts_ns" | ||
| }, | ||
| { | ||
| "type": "boolean", | ||
| "optional": true, | ||
| "default": false, | ||
| "field": "snapshot" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "db" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "schema" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "table" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": true, | ||
| "field": "txId" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": true, | ||
| "field": "lsn" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": true, | ||
| "field": "xmin" | ||
| } | ||
| ], | ||
| "optional": false, | ||
| "name": "io.debezium.connector.postgresql.Source", | ||
| "field": "source" | ||
| }, | ||
| { | ||
| "type": "string", | ||
| "optional": false, | ||
| "field": "op" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": true, | ||
| "field": "ts_ms" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": true, | ||
| "field": "ts_us" | ||
| }, | ||
| { | ||
| "type": "int64", | ||
| "optional": true, | ||
| "field": "ts_ns" | ||
| } | ||
| ], | ||
| "optional": false, | ||
| "name": "PostgreSQL_server.inventory.customers.Envelope" | ||
| }, | ||
| "payload": { | ||
| "before": null, | ||
| "after": { | ||
| "id": 1, | ||
| "first_name": "Anne", | ||
| "last_name": "Kretchmar", | ||
| "email": "[email protected]" | ||
| }, | ||
| "source": { | ||
| "version": "3.0.5.Final", | ||
| "connector": "postgresql", | ||
| "name": "PostgreSQL_server", | ||
| "ts_ms": 1559033904863, | ||
| "ts_us": 1559033904863123, | ||
| "ts_ns": 1559033904863123000, | ||
| "snapshot": true, | ||
| "db": "postgres", | ||
| "sequence": "[\"24023119\",\"24023128\"]", | ||
| "schema": "public", | ||
| "table": "customers", | ||
| "txId": 555, | ||
| "lsn": 24023128, | ||
| "xmin": null | ||
| }, | ||
| "op": "c", | ||
| "ts_ms": 1559033904863, | ||
| "ts_us": 1559033904863841, | ||
| "ts_ns": 1559033904863841257 | ||
| } | ||
| } |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package testutil | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "os" | ||
| "path/filepath" | ||
| "runtime" | ||
|
|
||
| "github.com/edgeflare/pgo/pkg/pglogrepl" | ||
| ) | ||
|
|
||
| // LoadCDC returns a CDC object in Debezium format | ||
| func LoadCDC() (pglogrepl.CDC, error) { | ||
| var cdc pglogrepl.CDC | ||
|
|
||
| // Get the directory containing this file | ||
| _, currentFile, _, _ := runtime.Caller(0) | ||
| dir := filepath.Dir(currentFile) | ||
|
|
||
| data, err := os.ReadFile(filepath.Join(dir, "sample.cdc.json")) | ||
| if err != nil { | ||
| return cdc, err | ||
| } | ||
|
|
||
| err = json.Unmarshal(data, &cdc) | ||
| if err != nil { | ||
| return cdc, err | ||
| } | ||
|
|
||
| return cdc, nil | ||
| } |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| // Package transform provides utilities for applying transformations to change data capture (CDC) events in pipelines. | ||
| // It's inspired by Debezium's [Single Message Transformations (SMTs)](https://docs.confluent.io/platform/current/connect/transforms/overview.html) usage. | ||
|
|
||
| package transform |
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| package transform | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/edgeflare/pgo/pkg/pglogrepl" | ||
| "github.com/tidwall/gjson" | ||
| ) | ||
|
|
||
| // ExtractConfig holds the configuration for the extract transformation | ||
| type ExtractConfig struct { | ||
| Fields []string `json:"fields"` | ||
| } | ||
|
|
||
| // Validate validates the ExtractConfig | ||
| func (c *ExtractConfig) Validate() error { | ||
| if len(c.Fields) == 0 { | ||
| return fmt.Errorf("at least one field is required") | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Type returns the type of the transformation | ||
| func (c *ExtractConfig) Type() string { | ||
| return "extract" | ||
| } | ||
|
|
||
| // Extract creates a TransformFunc that extracts specified fields from the CDC event | ||
| func Extract(config *ExtractConfig) TransformFunc { | ||
| return func(cdc *pglogrepl.CDC) (*pglogrepl.CDC, error) { | ||
| if err := config.Validate(); err != nil { | ||
| return cdc, fmt.Errorf("invalid extract configuration: %w", err) | ||
| } | ||
| current := cdc | ||
| fields := config.Fields | ||
| newBefore := make(map[string]interface{}) | ||
| newAfter := make(map[string]interface{}) | ||
|
|
||
| if before, ok := current.Payload.Before.(map[string]interface{}); ok { | ||
| for _, field := range fields { | ||
| if value, exists := before[field]; exists { | ||
| newBefore[field] = value | ||
| } | ||
| } | ||
| } | ||
| if after, ok := current.Payload.After.(map[string]interface{}); ok { | ||
| for _, field := range fields { | ||
| if value, exists := after[field]; exists { | ||
| newAfter[field] = value | ||
| } | ||
| } | ||
| } | ||
|
|
||
| current.Payload.Before = newBefore | ||
| current.Payload.After = newAfter | ||
|
|
||
| return current, nil | ||
| } | ||
| } | ||
|
|
||
| // extractValueFromMap extracts a value from a map using a dot-notated field | ||
| func extractValueFromMap(data map[string]interface{}, field string) (string, bool) { | ||
| // Try direct access to the map | ||
| if value, exists := data[field]; exists { | ||
| return fmt.Sprintf("%v", value), true | ||
| } | ||
|
|
||
| // Fallback to gjson for dot-notated nested fields | ||
| result := gjson.Get(fmt.Sprintf("%v", data), field) | ||
| if result.Exists() { | ||
| return result.String(), true | ||
| } | ||
|
|
||
| return "", false | ||
| } |
Oops, something went wrong.