Method WriteTableBatch

plugins/destination/duckdb/client/write.go:137–186

(ctx context.Context, name string, msgs message.WriteInserts)

}

func (c *Client) WriteTableBatch(ctx context.Context, name string, msgs message.WriteInserts) error {
	if len(msgs) == 0 {
		return nil
	}

	table := msgs[0].GetTable()

	writeStart := time.Now()
	tmpFile, err := writeTMPFile(table, msgs)
	if err != nil {
		return err
	}
	c.logger.Debug().Str("table", table.Name).Str("duration", time.Since(writeStart).String()).Msg("write tmp file")
	defer os.Remove(tmpFile)

	if len(table.PrimaryKeys()) == 0 {
		copyStart := time.Now()
		defer func() {
			c.logger.Debug().Str("table", table.Name).Str("duration", time.Since(copyStart).String()).Msg("copy file to table")
		}()
		return c.copyFromFile(ctx, name, tmpFile, table)
	}

	tmpTableName := name + strings.ReplaceAll(uuid.New().String(), "-", "_")
	// we skip constraints here, as copy doesn't allow us to handle conflicts, we handle those during upsert
	if err := c.createTableIfNotExist(ctx, tmpTableName, table, true); err != nil {
		return fmt.Errorf("failed to create table %s: %w", tmpTableName, err)
	}
	defer func() {
		e := c.exec(ctx, "drop table "+tmpTableName)
		if err == nil {
			// we preserve original error, so update only on nil err
			err = e
		}
	}()

	if err := c.copyFromFile(ctx, tmpTableName, tmpFile, table); err != nil {
		return fmt.Errorf("failed to copy from file %s: %w", tmpFile, err)
	}

	// At time of writing (March 2023), duckdb does not support updating list columns.
	// As a workaround, we delete the row and insert it again. This makes it non-atomic, unfortunately,
	// but this is unavoidable until support is added to duckdb itself.
	// See https://github.com/duckdb/duckdb/blob/c5d9afb97bbf0be12216f3b89ae3131afbbc3156/src/storage/table/list_column_data.cpp#L243-L251
	if containsList(table) {
		return c.deleteInsert(ctx, tmpTableName, table)
	}

	return c.upsert(ctx, tmpTableName, table)
}

func writeTMPFile(table *schema.Table, msgs []*message.WriteInsert) (fileName string, err error) {
	sc := transformSchemaForWriting(table)