Method WriteTableBatch

plugins/destination/sqlite/client/write.go:24–56

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

}

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

	tx, err := c.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}

	defer func() {
		if err == nil {
			err = tx.Commit()
			if err != nil {
				c.logger.Error().Err(err).Msg("failed to commit transaction")
			}
		}
		if err != nil {
			if rollbackErr := tx.Rollback(); rollbackErr != nil {
				c.logger.Error().Err(rollbackErr).Str("table", msgs[0].GetTable().Name).Msg("Failed to rollback transaction")
			}
		}
	}()

	for _, msg := range msgs {
		err = c.insertMessage(ctx, tx, msg)
		if err != nil {
			return err
		}
	}

	return nil
}

func (c *Client) insertMessage(ctx context.Context, tx *sql.Tx, m *message.WriteInsert) error {
	table := m.GetTable()