diff --git a/UPDATING.md b/UPDATING.md
index 55923ec8f209..69535cd5233f 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -36,6 +36,14 @@ using `ENABLE_PROXY_FIX = True`, review the newly-introducted variable,
 backend serialization. To disable set `RESULTS_BACKEND_USE_MSGPACK = False`
 in your configuration.
 
+* [5449](https://github.com/apache/incubator-superset/pull/5449): a change which
+adds a uniqueness criterion to the tables table. Depending on the integrity of
+the data, manual intervention may be required.
+
+* [8332](https://github.com/apache/incubator-superset/pull/8332): makes
+`tables.table_name`, `dbs.database_name`, and `clusters.cluster_name` non-nullable.
+Depending on the integrity of the data, manual intervention may be required.
+
 ## 0.34.0
 
 * [7848](https://github.com/apache/incubator-superset/pull/7848): If you are
diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py
index 980e607772b3..68e13faa531e 100644
--- a/superset/connectors/sqla/models.py
+++ b/superset/connectors/sqla/models.py
@@ -41,9 +41,11 @@
     Table,
     Text,
 )
-from sqlalchemy.exc import CompileError
+from sqlalchemy.engine.base import Connection
+from sqlalchemy.exc import CompileError, SQLAlchemyError
 from sqlalchemy.orm import backref, Query, relationship, RelationshipProperty, Session
 from sqlalchemy.orm.exc import NoResultFound
+from sqlalchemy.orm.mapper import Mapper
 from sqlalchemy.schema import UniqueConstraint
 from sqlalchemy.sql import column, ColumnElement, literal_column, table, text
 from sqlalchemy.sql.expression import Label, Select, TextAsFrom
@@ -304,7 +306,11 @@ class SqlaTable(Model, BaseDatasource):
     owner_class = security_manager.user_model
 
     __tablename__ = "tables"
-    __table_args__ = (UniqueConstraint("database_id", "table_name"),)
+
+    # Note this unqiuness constraint is not part of the physicalschema, i.e., it doesn't
+    # exist in the migrations, but is required by `import_from_dict` to ensure the
+    # correct filters are applied in order to identify uniqueness.
+    __table_args__ = (UniqueConstraint("database_id", "schema", "table_name"),)
 
     table_name = Column(String(250))
     main_dttm_col = Column(String(250))
@@ -1112,6 +1118,82 @@ def get_extra_cache_keys(self, query_obj: Dict) -> List[Any]:
             return extra_cache_keys
         return []
 
+    @staticmethod
+    def before_insert(
+        mapper: Mapper, connection: Connection, target: "SqlaTable"
+    ) -> None:
+        """
+        Check whether before insert if the target table already exists.
+
+        :param mapper: The table mappper
+        :param connection: The DB-API connection
+        :param target: The mapped instance being persisted
+        :raises Exception: If the target table is not unique
+        """
+
+        from superset.views.base import get_datasource_exist_error_msg
+
+        if SqlaTable.exists(target):
+            raise SQLAlchemyError(get_datasource_exist_error_msg(target.full_name))
+
+    @staticmethod
+    def before_update(
+        mapper: Mapper, connection: Connection, target: "SqlaTable"
+    ) -> None:
+        """
+        Check whether before update if the target table already exists.
+
+        Note this listener is called when any fields are being updated and thus it is
+        necessary to first check whether the reference table is being updated.
+
+        :param mapper: The table mapper
+        :param connection: The DB-API connection
+        :param target: The mapped instance being persisted
+        :raises Exception: If the target table is not unique
+        """
+
+        from superset.views.base import get_datasource_exist_error_msg
+
+        # Check whether the relevant attributes have changed.
+        state = db.inspect(target)  # pylint: disable=no-member
+
+        for attr in ["database_id", "schema", "table_name"]:
+            history = state.get_history(attr, True)
+
+            if history.has_changes():
+                break
+        else:
+            return None
+
+        if SqlaTable.exists(target):
+            raise SQLAlchemyError(get_datasource_exist_error_msg(target.full_name))
+
+    @staticmethod
+    def exists(record: "SqlaTable") -> bool:
+        """
+        Return True if the table exists, False otherwise.
+
+        A table is deemed to already exist based on the uniqueness of the database,
+        schema, and name.
+
+        :param record: The table record
+        :returns: Whether the record exists
+        """
+
+        count = (
+            db.session.query(SqlaTable)
+            .filter_by(
+                database_id=record.database_id,
+                schema=record.schema,
+                table_name=record.table_name,
+            )
+            .count()
+        )
+
+        return count != 0
+
 
 sa.event.listen(SqlaTable, "after_insert", security_manager.set_perm)
 sa.event.listen(SqlaTable, "after_update", security_manager.set_perm)
+sa.event.listen(SqlaTable, "before_insert", SqlaTable.before_insert)
+sa.event.listen(SqlaTable, "before_update", SqlaTable.before_update)
diff --git a/superset/connectors/sqla/views.py b/superset/connectors/sqla/views.py
index 3b242ec8ed4d..a70dcc2d10f2 100644
--- a/superset/connectors/sqla/views.py
+++ b/superset/connectors/sqla/views.py
@@ -27,6 +27,7 @@
 from flask_babel import gettext as __
 from flask_babel import lazy_gettext as _
 from wtforms.ext.sqlalchemy.fields import QuerySelectField
+from wtforms.validators import Regexp
 
 from superset import appbuilder, db, security_manager
 from superset.connectors.base.views import DatasourceModelView
@@ -34,7 +35,6 @@
 from superset.views.base import (
     DatasourceFilter,
     DeleteMixin,
-    get_datasource_exist_error_msg,
     ListWidgetWithCheckboxes,
     SupersetModelView,
     YamlExportMixin,
@@ -303,6 +303,11 @@ class TableModelView(DatasourceModelView, DeleteMixin, YamlExportMixin):  # noqa
         "template_params": _("Template parameters"),
         "modified": _("Modified"),
     }
+    validators_columns = {
+        "table_name": [
+            Regexp(r"^[^\.]+$", message=_("Table name cannot contain a schema"))
+        ]
+    }
 
     edit_form_extra_fields = {
         "database": QuerySelectField(
@@ -313,14 +318,10 @@ class TableModelView(DatasourceModelView, DeleteMixin, YamlExportMixin):  # noqa
     }
 
     def pre_add(self, table):
-        with db.session.no_autoflush:
-            table_query = db.session.query(models.SqlaTable).filter(
-                models.SqlaTable.table_name == table.table_name,
-                models.SqlaTable.schema == table.schema,
-                models.SqlaTable.database_id == table.database.id,
-            )
-            if db.session.query(table_query.exists()).scalar():
-                raise Exception(get_datasource_exist_error_msg(table.full_name))
+
+        # Although the listener exists this is necessary to ensure that FAB flashes the
+        # specific message as opposed to "General Error <class.Exception>".
+        models.SqlaTable.before_insert(None, None, table)
 
         # Fail before adding if the table can't be found
         try:
diff --git a/superset/migrations/versions/1d73b15530e7_update_tables.py b/superset/migrations/versions/1d73b15530e7_update_tables.py
new file mode 100644
index 000000000000..2c3c30d45dd4
--- /dev/null
+++ b/superset/migrations/versions/1d73b15530e7_update_tables.py
@@ -0,0 +1,57 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""update tables
+
+Revision ID: 1d73b15530e7
+Revises: b6fa807eac07
+Create Date: 2018-07-20 11:36:04.535859
+
+"""
+from alembic import op
+from sqlalchemy import engine
+from sqlalchemy.exc import OperationalError, ProgrammingError
+
+from superset.utils.core import generic_find_uq_constraint_name
+
+# revision identifiers, used by Alembic.
+revision = "1d73b15530e7"
+down_revision = "b6fa807eac07"
+
+conv = {"uq": "uq_%(table_name)s_%(column_0_name)s"}
+
+
+def upgrade():
+    bind = op.get_bind()
+    insp = engine.reflection.Inspector.from_engine(bind)
+
+    # Drop the uniqueness constraint if it exists.
+    try:
+        with op.batch_alter_table("tables", naming_convention=conv) as batch_op:
+            batch_op.drop_constraint(
+                generic_find_uq_constraint_name("tables", {"table_name"}, insp)
+                or "uq_tables_table_name",
+                type_="unique",
+            )
+    except (ProgrammingError, OperationalError):
+        pass
+
+
+def downgrade():
+
+    # Re-add the uniqueness constraint.
+    with op.batch_alter_table("tables", naming_convention=conv) as batch_op:
+        batch_op.create_unique_constraint("uq_tables_table_name", ["table_name"])