(ctx context.Context, tmpTableName string, table *schema.Table)
| 60 | } |
| 61 | |
| 62 | func (c *Client) upsert(ctx context.Context, tmpTableName string, table *schema.Table) error { |
| 63 | var sb strings.Builder |
| 64 | sb.WriteString("INSERT INTO ") |
| 65 | sb.WriteString(table.Name) |
| 66 | sb.WriteString("(" + strings.Join(sanitized(table.Columns.Names()), ", ") + ")") |
| 67 | sb.WriteString(" SELECT ") |
| 68 | sb.WriteString(strings.Join(sanitized(table.Columns.Names()), ", ")) |
| 69 | sb.WriteString(" FROM ") |
| 70 | sb.WriteString(tmpTableName) |
| 71 | sb.WriteString(" ON CONFLICT (" + strings.Join(table.PrimaryKeys(), ", ") + ")") |
| 72 | indices := nonPkIndices(table) |
| 73 | if len(indices) == 0 { |
| 74 | sb.WriteString(" DO NOTHING") |
| 75 | return c.exec(ctx, sb.String()) |
| 76 | } |
| 77 | |
| 78 | sb.WriteString(" DO UPDATE SET ") |
| 79 | |
| 80 | written := 0 |
| 81 | for _, index := range nonPkIndices(table) { |
| 82 | col := table.Columns[index] |
| 83 | if col.Unique { |
| 84 | // we skip this stuff, as unique constraint can't be updated by DuckDB |
| 85 | continue |
| 86 | } |
| 87 | if written > 0 { |
| 88 | sb.WriteString(", ") |
| 89 | } |
| 90 | sb.WriteString(sanitizeID(col.Name)) |
| 91 | sb.WriteString(" = excluded.") |
| 92 | sb.WriteString(sanitizeID(col.Name)) |
| 93 | written++ |
| 94 | } |
| 95 | query := sb.String() |
| 96 | |
| 97 | // return c.exec(ctx, query) |
| 98 | // per https://duckdb.org/docs/sql/indexes#over-eager-unique-constraint-checking we might need some retries |
| 99 | // as the upsert for tables with PKs is transformed into delete + insert internally |
| 100 | _, err := backoff.Retry(ctx, func() (any, error) { |
| 101 | return nil, c.exec(ctx, query) |
| 102 | }, backoff.WithBackOff(backoff.NewConstantBackOff(50*time.Millisecond)), backoff.WithMaxTries(3)) |
| 103 | return err |
| 104 | } |
| 105 | |
| 106 | func (c *Client) deleteByPK(ctx context.Context, tmpTableName string, table *schema.Table) error { |
| 107 | var sb strings.Builder |
no test coverage detected