Skip to content

add option for specifying timeout in minutes and gtid position together with last known pk #139

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maxenglander merged 4 commits into from
Apr 7, 2025
Merged

add option for specifying timeout in minutes and gtid position together with last known pk #139

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
dae88cc
add option for specifying timeout in seconds
maxenglander Apr 5, 2025
ec6f4ac
fix typo
maxenglander Apr 5, 2025
d3d0904
default to 5 minutes if source opt is undefined
maxenglander Apr 5, 2025
aa659f2
add option to use gtid position together with last known pk (#140)
maxenglander Apr 7, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions cmd/airbyte-source/spec.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,21 @@
"description": "The max number of times we continue syncing after potential errors",
"order": 8
},
"timeout_seconds": {
"type": "integer",
"title": "Timeout (in seconds)",
"default": 300,
"minimum": 300,
"description": "Timeout in seconds for a sync attempt",
"order": 9
},
"use_gtid_with_table_pks": {
"type": "boolean",
"title": "Use GTID with table primary keys",
"default": false,
"description": "Use GTID position together with table primary keys",
"order": 10
},
"options": {
"type": "object",
"title": "Customize serialization",
Expand Down
22 changes: 12 additions & 10 deletions cmd/internal/planetscale_connection.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,16 +12,18 @@ import (

// PlanetScaleSource defines a configured Airbyte Source for a PlanetScale database
type PlanetScaleSource struct {
Host string `json:"host"`
Database string `json:"database"`
Username string `json:"username"`
Password string `json:"password"`
Shards string `json:"shards"`
UseReplica bool `json:"use_replica"`
UseRdonly bool `json:"use_rdonly"`
StartingGtids string `json:"starting_gtids"`
Options CustomSourceOptions `json:"options"`
MaxRetries uint `json:"max_retries"`
Host string `json:"host"`
Database string `json:"database"`
Username string `json:"username"`
Password string `json:"password"`
Shards string `json:"shards"`
UseReplica bool `json:"use_replica"`
UseRdonly bool `json:"use_rdonly"`
StartingGtids string `json:"starting_gtids"`
Options CustomSourceOptions `json:"options"`
MaxRetries uint `json:"max_retries"`
TimeoutSeconds *int `json:"timeout_seconds"`
UseGTIDWithTablePKs bool `json:"use_gtid_with_table_pks"`
}

type CustomSourceOptions struct {
Expand Down
15 changes: 9 additions & 6 deletions cmd/internal/planetscale_edge_database.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,7 +224,10 @@ func (p PlanetScaleEdgeDatabase) Read(ctx context.Context, w io.Writer, ps Plane
}

table := s.Stream
readDuration := 5 * time.Minute
timeout := 5 * time.Minute
if timeoutSeconds := ps.TimeoutSeconds; timeoutSeconds != nil {
Copy link

Copilot AI Apr 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The current implementation does not enforce the minimum timeout of 5 minutes as required. Consider adding a condition to set the timeout to 5 minutes if the provided value is less than that.

Copilot uses AI. Check for mistakes.
Copy link
Copy Markdown
Collaborator Author

@maxenglander maxenglander Apr 5, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

no thanks, it's useful to be able to use values less than 5m in testing

timeout = time.Duration(*timeoutSeconds) * time.Second
}
maxRetries := ps.MaxRetries

preamble := fmt.Sprintf("[%v:%v:%v shard : %v] ", table.Namespace, TabletTypeToString(tabletType), table.Name, currentPosition.Shard)
Expand All @@ -245,15 +248,15 @@ func (p PlanetScaleEdgeDatabase) Read(ctx context.Context, w io.Writer, ps Plane
p.Logger.Log(LOGLEVEL_INFO, preamble+"No new GTIDs found, exiting")
return TableCursorToSerializedCursor(currentPosition)
}
p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf(preamble+"New GTIDs found, syncing for %v", readDuration))
p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf(preamble+"New GTIDs found, syncing for %v", timeout))

var syncCount uint = 0
totalRecordCount := 0

for {
syncCount += 1
p.Logger.Log(LOGLEVEL_INFO, fmt.Sprintf("%sStarting sync #%v", preamble, syncCount))
newPosition, recordCount, err := p.sync(ctx, syncMode, currentPosition, stopPosition, table, ps, tabletType, readDuration)
newPosition, recordCount, err := p.sync(ctx, syncMode, currentPosition, stopPosition, table, ps, tabletType, timeout)
totalRecordCount += recordCount
currentSerializedCursor, sErr = TableCursorToSerializedCursor(currentPosition)
if sErr != nil {
Expand All @@ -279,11 +282,11 @@ func (p PlanetScaleEdgeDatabase) Read(ctx context.Context, w io.Writer, ps Plane
}
}

func (p PlanetScaleEdgeDatabase) sync(ctx context.Context, syncMode string, tc *psdbconnect.TableCursor, stopPosition string, s Stream, ps PlanetScaleSource, tabletType psdbconnect.TabletType, readDuration time.Duration) (*psdbconnect.TableCursor, int, error) {
func (p PlanetScaleEdgeDatabase) sync(ctx context.Context, syncMode string, tc *psdbconnect.TableCursor, stopPosition string, s Stream, ps PlanetScaleSource, tabletType psdbconnect.TabletType, timeout time.Duration) (*psdbconnect.TableCursor, int, error) {
preamble := fmt.Sprintf("[%v:%v:%v shard : %v] ", s.Namespace, TabletTypeToString(tabletType), s.Name, tc.Shard)

defer p.Logger.Flush()
ctx, cancel := context.WithTimeout(ctx, readDuration)
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()

var (
Expand All @@ -300,7 +303,7 @@ func (p PlanetScaleEdgeDatabase) sync(ctx context.Context, syncMode string, tc *
defer conn.Close()
}

if tc.LastKnownPk != nil {
if tc.LastKnownPk != nil && !ps.UseGTIDWithTablePKs {
tc.Position = ""
}

Expand Down
Loading