MCPcopy Create free account
hub / github.com/cloudquery/cloudquery / upsert

Method upsert

plugins/destination/duckdb/client/write.go:62–104  ·  view source on GitHub ↗
(ctx context.Context, tmpTableName string, table *schema.Table)

Source from the content-addressed store, hash-verified

60}
61
62func (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
106func (c *Client) deleteByPK(ctx context.Context, tmpTableName string, table *schema.Table) error {
107 var sb strings.Builder

Callers 1

WriteTableBatchMethod · 0.95

Calls 5

execMethod · 0.95
sanitizedFunction · 0.85
nonPkIndicesFunction · 0.85
sanitizeIDFunction · 0.70
StringMethod · 0.45

Tested by

no test coverage detected