3 files changed, 24 insertions, 2 deletions
diff --git a/lib/sqlalchemy/engine/base.py b/lib/sqlalchemy/engine/base.py
index 88558df5d..462e5f9ec 100644
--- a/lib/sqlalchemy/engine/base.py
+++ b/lib/sqlalchemy/engine/base.py
@@ -16,6 +16,7 @@ from .. import exc
 from .. import inspection
 from .. import log
 from .. import util
+from ..sql import compiler
 from ..sql import schema
 from ..sql import util as sql_util
 
@@ -1083,6 +1084,8 @@ class Connection(Connectable):
                     schema_translate_map=self.schema_for_object
                     if not self.schema_for_object.is_default
                     else None,
+                    linting=self.dialect.compiler_linting
+                    | compiler.WARN_LINTING,
                 )
                 self._execution_options["compiled_cache"][key] = compiled_sql
         else:
@@ -1093,6 +1096,7 @@ class Connection(Connectable):
                 schema_translate_map=self.schema_for_object
                 if not self.schema_for_object.is_default
                 else None,
+                linting=self.dialect.compiler_linting | compiler.WARN_LINTING,
             )
 
         ret = self._execute_context(
diff --git a/lib/sqlalchemy/engine/create.py b/lib/sqlalchemy/engine/create.py
index 58fe91c7e..5198c8cd6 100644
--- a/lib/sqlalchemy/engine/create.py
+++ b/lib/sqlalchemy/engine/create.py
@@ -13,6 +13,7 @@ from .. import event
 from .. import exc
 from .. import pool as poollib
 from .. import util
+from ..sql import compiler
 
 
 @util.deprecated_params(
@@ -142,6 +143,16 @@ def create_engine(url, **kwargs):
     :param empty_in_strategy:   No longer used; SQLAlchemy now uses
         "empty set" behavior for IN in all cases.
 
+    :param enable_from_linting: defaults to True.  Will emit a warning
+        if a given SELECT statement is found to have un-linked FROM elements
+        which would cause a cartesian product.
+
+        .. versionadded:: 1.4
+
+        .. seealso::
+
+            :ref:`change_4737`
+
     :param encoding: Defaults to ``utf-8``.  This is the string
         encoding used by SQLAlchemy for string encode/decode
         operations which occur within SQLAlchemy, **outside of
@@ -446,6 +457,11 @@ def create_engine(url, **kwargs):
 
     dialect_args["dbapi"] = dbapi
 
+    dialect_args.setdefault("compiler_linting", compiler.NO_LINTING)
+    enable_from_linting = kwargs.pop("enable_from_linting", True)
+    if enable_from_linting:
+        dialect_args["compiler_linting"] ^= compiler.COLLECT_CARTESIAN_PRODUCTS
+
     for plugin in plugins:
         plugin.handle_dialect_kwargs(dialect_cls, dialect_args)
 
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 1c995f05f..378890444 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -31,7 +31,6 @@ from ..sql import expression
 from ..sql import schema
 from ..sql.elements import quoted_name
 
-
 AUTOCOMMIT_REGEXP = re.compile(
     r"\s*(?:UPDATE|INSERT|CREATE|DELETE|DROP|ALTER)", re.I | re.UNICODE
 )
@@ -214,6 +213,9 @@ class DefaultDialect(interfaces.Dialect):
         supports_native_boolean=None,
         max_identifier_length=None,
         label_length=None,
+        # int() is because the @deprecated_params decorator cannot accommodate
+        # the direct reference to the "NO_LINTING" object
+        compiler_linting=int(compiler.NO_LINTING),
         **kwargs
     ):
 
@@ -249,7 +251,7 @@ class DefaultDialect(interfaces.Dialect):
                 self._user_defined_max_identifier_length
             )
         self.label_length = label_length
-
+        self.compiler_linting = compiler_linting
         if self.description_encoding == "use_encoding":
             self._description_decoder = (
                 processors.to_unicode_processor_factory