diff --git a/superset/assets/src/datasource/DatasourceEditor.jsx b/superset/assets/src/datasource/DatasourceEditor.jsx
index f16c1e53496..8df3fd9002c 100644
--- a/superset/assets/src/datasource/DatasourceEditor.jsx
+++ b/superset/assets/src/datasource/DatasourceEditor.jsx
@@ -349,13 +349,13 @@ export class DatasourceEditor extends React.PureComponent {
             control={<TextControl />}
           />}
         <Field
-          fieldKey="owner"
-          label={t('Owner')}
-          descr={t('Owner of the datasource')}
+          fieldKey="owners"
+          label={t('Owners')}
+          descr={t('Owners of the datasource')}
           control={
             <SelectAsyncControl
               dataEndpoint="/users/api/read"
-              multi={false}
+              multi
               mutator={data => data.pks.map((pk, i) => ({
                 value: pk,
                 label: `${data.result[i].first_name} ${data.result[i].last_name}`,
diff --git a/superset/connectors/base/models.py b/superset/connectors/base/models.py
index 216ed9ec63e..6876fa09585 100644
--- a/superset/connectors/base/models.py
+++ b/superset/connectors/base/models.py
@@ -25,6 +25,7 @@ class BaseDatasource(AuditMixinNullable, ImportMixin):
     baselink = None  # url portion pointing to ModelView endpoint
     column_class = None  # link to derivative of BaseColumn
     metric_class = None  # link to derivative of BaseMetric
+    owner_class = None
 
     # Used to do code highlighting when displaying the query in the UI
     query_language = None
@@ -45,7 +46,7 @@ class BaseDatasource(AuditMixinNullable, ImportMixin):
     perm = Column(String(1000))
 
     sql = None
-    owner = None
+    owners = None
     update_from_object_fields = None
 
     @declared_attr
@@ -205,7 +206,7 @@ class BaseDatasource(AuditMixinNullable, ImportMixin):
             'metrics': [o.data for o in self.metrics],
             'metrics_combo': self.metrics_combo,
             'order_by_choices': order_by_choices,
-            'owner': self.owner.id if self.owner else None,
+            'owners': [owner.id for owner in self.owners],
             'verbose_map': verbose_map,
             'select_star': self.select_star,
         }
@@ -325,7 +326,7 @@ class BaseDatasource(AuditMixinNullable, ImportMixin):
         for attr in self.update_from_object_fields:
             setattr(self, attr, obj.get(attr))
 
-        self.user_id = obj.get('owner')
+        self.owners = obj.get('owners', [])
 
         # Syncing metrics
         metrics = self.get_fk_many_from_list(
diff --git a/superset/connectors/druid/models.py b/superset/connectors/druid/models.py
index 937c8d854c1..ae53246c8fd 100644
--- a/superset/connectors/druid/models.py
+++ b/superset/connectors/druid/models.py
@@ -26,7 +26,7 @@ from pydruid.utils.postaggregator import (
 import requests
 import sqlalchemy as sa
 from sqlalchemy import (
-    Boolean, Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint,
+    Boolean, Column, DateTime, ForeignKey, Integer, String, Table, Text, UniqueConstraint,
 )
 from sqlalchemy.orm import backref, relationship
 
@@ -43,6 +43,7 @@ from superset.utils.core import (
 
 DRUID_TZ = conf.get('DRUID_TZ')
 POST_AGG_TYPE = 'postagg'
+metadata = Model.metadata  # pylint: disable=no-member
 
 
 # Function wrapper because bound methods cannot
@@ -446,6 +447,14 @@ class DruidMetric(Model, BaseMetric):
         return import_datasource.import_simple_obj(db.session, i_metric, lookup_obj)
 
 
+druiddatasource_user = Table(
+    'druiddatasource_user', metadata,
+    Column('id', Integer, primary_key=True),
+    Column('user_id', Integer, ForeignKey('ab_user.id')),
+    Column('datasource_id', Integer, ForeignKey('datasources.id')),
+)
+
+
 class DruidDatasource(Model, BaseDatasource):
 
     """ORM object referencing Druid datasources (tables)"""
@@ -458,6 +467,7 @@ class DruidDatasource(Model, BaseDatasource):
     cluster_class = DruidCluster
     metric_class = DruidMetric
     column_class = DruidColumn
+    owner_class = security_manager.user_model
 
     baselink = 'druiddatasourcemodelview'
 
@@ -470,11 +480,8 @@ class DruidDatasource(Model, BaseDatasource):
         String(250), ForeignKey('clusters.cluster_name'))
     cluster = relationship(
         'DruidCluster', backref='datasources', foreign_keys=[cluster_name])
-    user_id = Column(Integer, ForeignKey('ab_user.id'))
-    owner = relationship(
-        security_manager.user_model,
-        backref=backref('datasources', cascade='all, delete-orphan'),
-        foreign_keys=[user_id])
+    owners = relationship(owner_class, secondary=druiddatasource_user,
+                          backref='druiddatasources')
     UniqueConstraint('cluster_name', 'datasource_name')
 
     export_fields = (
@@ -657,7 +664,7 @@ class DruidDatasource(Model, BaseDatasource):
             datasource = cls(
                 datasource_name=druid_config['name'],
                 cluster=cluster,
-                owner=user,
+                owners=[user],
                 changed_by_fk=user.id,
                 created_by_fk=user.id,
             )
diff --git a/superset/connectors/druid/views.py b/superset/connectors/druid/views.py
index 18c1aef0d19..eda7ce400c0 100644
--- a/superset/connectors/druid/views.py
+++ b/superset/connectors/druid/views.py
@@ -214,12 +214,12 @@ class DruidDatasourceModelView(DatasourceModelView, DeleteMixin, YamlExportMixin
     order_columns = ['datasource_link', 'modified']
     related_views = [DruidColumnInlineView, DruidMetricInlineView]
     edit_columns = [
-        'datasource_name', 'cluster', 'description', 'owner',
+        'datasource_name', 'cluster', 'description', 'owners',
         'is_hidden',
         'filter_select_enabled', 'fetch_values_from',
         'default_endpoint', 'offset', 'cache_timeout']
     search_columns = (
-        'datasource_name', 'cluster', 'description', 'owner',
+        'datasource_name', 'cluster', 'description', 'owners',
     )
     add_columns = edit_columns
     show_columns = add_columns + ['perm', 'slices']
@@ -263,7 +263,7 @@ class DruidDatasourceModelView(DatasourceModelView, DeleteMixin, YamlExportMixin
         'datasource_link': _('Data Source'),
         'cluster': _('Cluster'),
         'description': _('Description'),
-        'owner': _('Owner'),
+        'owners': _('Owners'),
         'is_hidden': _('Is Hidden'),
         'filter_select_enabled': _('Enable Filter Select'),
         'default_endpoint': _('Default Endpoint'),
diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py
index cf22add6282..779e4a313c9 100644
--- a/superset/connectors/sqla/models.py
+++ b/superset/connectors/sqla/models.py
@@ -9,7 +9,7 @@ import pandas as pd
 import sqlalchemy as sa
 from sqlalchemy import (
     and_, asc, Boolean, Column, DateTime, desc, ForeignKey, Integer, or_,
-    select, String, Text,
+    select, String, Table, Text,
 )
 from sqlalchemy.exc import CompileError
 from sqlalchemy.orm import backref, relationship
@@ -27,6 +27,7 @@ from superset.models.helpers import QueryResult
 from superset.utils import core as utils, import_datasource
 
 config = app.config
+metadata = Model.metadata  # pylint: disable=no-member
 
 
 class AnnotationDatasource(BaseDatasource):
@@ -250,6 +251,14 @@ class SqlMetric(Model, BaseMetric):
         return import_datasource.import_simple_obj(db.session, i_metric, lookup_obj)
 
 
+sqlatable_user = Table(
+    'sqlatable_user', metadata,
+    Column('id', Integer, primary_key=True),
+    Column('user_id', Integer, ForeignKey('ab_user.id')),
+    Column('table_id', Integer, ForeignKey('tables.id')),
+)
+
+
 class SqlaTable(Model, BaseDatasource):
 
     """An ORM object for SqlAlchemy table references"""
@@ -258,6 +267,7 @@ class SqlaTable(Model, BaseDatasource):
     query_language = 'sql'
     metric_class = SqlMetric
     column_class = TableColumn
+    owner_class = security_manager.user_model
 
     __tablename__ = 'tables'
     __table_args__ = (UniqueConstraint('database_id', 'table_name'),)
@@ -266,11 +276,7 @@ class SqlaTable(Model, BaseDatasource):
     main_dttm_col = Column(String(250))
     database_id = Column(Integer, ForeignKey('dbs.id'), nullable=False)
     fetch_values_predicate = Column(String(1000))
-    user_id = Column(Integer, ForeignKey('ab_user.id'))
-    owner = relationship(
-        security_manager.user_model,
-        backref='tables',
-        foreign_keys=[user_id])
+    owners = relationship(owner_class, secondary=sqlatable_user, backref='tables')
     database = relationship(
         'Database',
         backref=backref('tables', cascade='all, delete-orphan'),
diff --git a/superset/connectors/sqla/views.py b/superset/connectors/sqla/views.py
index c085958f5b5..734a2c0ac5c 100644
--- a/superset/connectors/sqla/views.py
+++ b/superset/connectors/sqla/views.py
@@ -162,7 +162,7 @@ class TableModelView(DatasourceModelView, DeleteMixin, YamlExportMixin):  # noqa
     edit_columns = [
         'table_name', 'sql', 'filter_select_enabled',
         'fetch_values_predicate', 'database', 'schema',
-        'description', 'owner',
+        'description', 'owners',
         'main_dttm_col', 'default_endpoint', 'offset', 'cache_timeout',
         'is_sqllab_view', 'template_params',
     ]
@@ -171,7 +171,7 @@ class TableModelView(DatasourceModelView, DeleteMixin, YamlExportMixin):  # noqa
     related_views = [TableColumnInlineView, SqlMetricInlineView]
     base_order = ('changed_on', 'desc')
     search_columns = (
-        'database', 'schema', 'table_name', 'owner', 'is_sqllab_view',
+        'database', 'schema', 'table_name', 'owners', 'is_sqllab_view',
     )
     description_columns = {
         'slices': _(
@@ -233,7 +233,7 @@ class TableModelView(DatasourceModelView, DeleteMixin, YamlExportMixin):  # noqa
         'cache_timeout': _('Cache Timeout'),
         'table_name': _('Table Name'),
         'fetch_values_predicate': _('Fetch Values Predicate'),
-        'owner': _('Owner'),
+        'owners': _('Owners'),
         'main_dttm_col': _('Main Datetime Column'),
         'description': _('Description'),
         'is_sqllab_view': _('SQL Lab View'),
diff --git a/superset/migrations/versions/3e1b21cd94a4_change_owner_to_m2m_relation_on_.py b/superset/migrations/versions/3e1b21cd94a4_change_owner_to_m2m_relation_on_.py
new file mode 100644
index 00000000000..e087b35384d
--- /dev/null
+++ b/superset/migrations/versions/3e1b21cd94a4_change_owner_to_m2m_relation_on_.py
@@ -0,0 +1,105 @@
+"""change_owner_to_m2m_relation_on_datasources.py
+
+Revision ID: 3e1b21cd94a4
+Revises: 4ce8df208545
+Create Date: 2018-12-15 12:34:47.228756
+
+"""
+
+# revision identifiers, used by Alembic.
+from superset import db
+from superset.utils.core import generic_find_fk_constraint_name
+
+revision = '3e1b21cd94a4'
+down_revision = '6c7537a6004a'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+sqlatable_user = sa.Table(
+    'sqlatable_user', sa.MetaData(),
+    sa.Column('id', sa.Integer, primary_key=True),
+    sa.Column('user_id', sa.Integer, sa.ForeignKey('ab_user.id')),
+    sa.Column('table_id', sa.Integer, sa.ForeignKey('tables.id')),
+)
+
+SqlaTable = sa.Table(
+    'tables', sa.MetaData(),
+    sa.Column('id', sa.Integer, primary_key=True),
+    sa.Column('user_id', sa.Integer, sa.ForeignKey('ab_user.id')),
+)
+
+druiddatasource_user = sa.Table(
+    'druiddatasource_user', sa.MetaData(),
+    sa.Column('id', sa.Integer, primary_key=True),
+    sa.Column('user_id', sa.Integer, sa.ForeignKey('ab_user.id')),
+    sa.Column('datasource_id', sa.Integer, sa.ForeignKey('datasources.id')),
+)
+
+DruidDatasource = sa.Table(
+    'datasources', sa.MetaData(),
+    sa.Column('id', sa.Integer, primary_key=True),
+    sa.Column('user_id', sa.Integer, sa.ForeignKey('ab_user.id')),
+)
+
+
+def upgrade():
+    op.create_table('sqlatable_user',
+                    sa.Column('id', sa.Integer(), nullable=False),
+                    sa.Column('user_id', sa.Integer(), nullable=True),
+                    sa.Column('table_id', sa.Integer(), nullable=True),
+                    sa.ForeignKeyConstraint(['table_id'], ['tables.id'], ),
+                    sa.ForeignKeyConstraint(['user_id'], ['ab_user.id'], ),
+                    sa.PrimaryKeyConstraint('id')
+                    )
+    op.create_table('druiddatasource_user',
+                    sa.Column('id', sa.Integer(), nullable=False),
+                    sa.Column('user_id', sa.Integer(), nullable=True),
+                    sa.Column('datasource_id', sa.Integer(), nullable=True),
+                    sa.ForeignKeyConstraint(['datasource_id'], ['datasources.id'], ),
+                    sa.ForeignKeyConstraint(['user_id'], ['ab_user.id'], ),
+                    sa.PrimaryKeyConstraint('id')
+                    )
+
+    bind = op.get_bind()
+    insp = sa.engine.reflection.Inspector.from_engine(bind)
+    session = db.Session(bind=bind)
+
+    tables = session.query(SqlaTable).all()
+    for table in tables:
+        if table.user_id is not None:
+            session.execute(
+                sqlatable_user.insert().values(user_id=table.user_id, table_id=table.id)
+            )
+
+    druiddatasources = session.query(DruidDatasource).all()
+    for druiddatasource in druiddatasources:
+        if druiddatasource.user_id is not None:
+            session.execute(
+                druiddatasource_user.insert().values(user_id=druiddatasource.user_id, datasource_id=druiddatasource.id)
+            )
+
+    session.close()
+    with op.batch_alter_table('tables') as batch_op:
+        batch_op.drop_constraint('user_id', type_='foreignkey')
+        batch_op.drop_column('user_id')
+    with op.batch_alter_table('datasources') as batch_op:
+        batch_op.drop_constraint(generic_find_fk_constraint_name(
+            'datasources',
+            {'id'},
+            'ab_user',
+            insp,
+        ), type_='foreignkey')
+        batch_op.drop_column('user_id')
+
+
+def downgrade():
+    op.drop_table('sqlatable_user')
+    op.drop_table('druiddatasource_user')
+    with op.batch_alter_table('tables') as batch_op:
+        batch_op.add_column(sa.Column('user_id', sa.INTEGER(), nullable=True))
+        batch_op.create_foreign_key('user_id', 'ab_user', ['user_id'], ['id'])
+    with op.batch_alter_table('datasources') as batch_op:
+        batch_op.add_column(sa.Column('user_id', sa.INTEGER(), nullable=True))
+        batch_op.create_foreign_key('fk_datasources_user_id_ab_user', 'ab_user', ['user_id'], ['id'])
diff --git a/superset/views/datasource.py b/superset/views/datasource.py
index 5df20a29776..9d3d3419eab 100644
--- a/superset/views/datasource.py
+++ b/superset/views/datasource.py
@@ -29,6 +29,10 @@ class Datasource(BaseSupersetView):
                     'this data source configuration'),
                 status='401',
             )
+
+        if 'owners' in datasource:
+            datasource['owners'] = db.session.query(orm_datasource.owner_class).filter(
+                orm_datasource.owner_class.id.in_(datasource['owners'])).all()
         orm_datasource.update_from_object(datasource)
         data = orm_datasource.data
         db.session.commit()