(ctx context.Context, tx *sql.Tx, m *message.WriteInsert)
| 56 | } |
| 57 | |
| 58 | func (c *Client) insertMessage(ctx context.Context, tx *sql.Tx, m *message.WriteInsert) error { |
| 59 | table := m.GetTable() |
| 60 | sc := m.Record.Schema() |
| 61 | var sqlString string |
| 62 | if len(table.PrimaryKeys()) == 0 { |
| 63 | sqlString = c.insert(sc) |
| 64 | } else { |
| 65 | sqlString = c.upsert(sc) |
| 66 | } |
| 67 | vals := transformRecord(m.Record) |
| 68 | for _, v := range vals { |
| 69 | if _, err := tx.ExecContext(ctx, sqlString, v...); err != nil { |
| 70 | return fmt.Errorf("failed to execute '%s': %w", sqlString, err) |
| 71 | } |
| 72 | } |
| 73 | return nil |
| 74 | } |
| 75 | |
| 76 | func (*Client) insert(sc *arrow.Schema) string { |
| 77 | var sb strings.Builder |
no test coverage detected