Repository: sqlfluff/sqlfluff

Issue:
Rule: Warn if the overall query (i.e. final query, ignoring CTEs) produces an indeterminate number of columns
Use of `SELECT *` in production queries is generally considered a bad practice, e.g. [this StackOverflow post](https://stackoverflow.com/questions/3639861/why-is-select-considered-harmful). SQLFluff should warn if a query does this.

In our team, we often allow an exception, i.e.`SELECT *` is okay *if* it's reading from a CTE (i.e. another "table" defined elsewhere in the same query), since in that case, the schema of the CTE may be fixed, so `SELECT *` from a CTE wouldn't allow any new upstream columns from other tables or queries to leak through.

This mistake caused one of our production jobs to fail recently because the query was being used to add new records to an existing table, and it failed in a cryptic way because the `SELECT *` picked up a new column that wasn't present in the existing table being appended to.

Discussion:

Patch:
diff --git a/src/sqlfluff/core/dialects/dialect_ansi.py b/src/sqlfluff/core/dialects/dialect_ansi.py
index 91a6d2d24..2c1f7a626 100644
--- a/src/sqlfluff/core/dialects/dialect_ansi.py
+++ b/src/sqlfluff/core/dialects/dialect_ansi.py
@@ -12,7 +12,7 @@ grammar. Check out their docs, they're awesome.
 https://www.cockroachlabs.com/docs/stable/sql-grammar.html#select_stmt
 """
 
-from typing import List, Tuple, NamedTuple, Optional
+from typing import Generator, List, Tuple, NamedTuple, Optional
 
 from sqlfluff.core.parser import (
     Matchable,
@@ -480,14 +480,20 @@ class ObjectReferenceSegment(BaseSegment):
         allow_gaps=False,
     )
 
-    @staticmethod
-    def _iter_reference_parts(elem):
+    class ObjectReferencePart(NamedTuple):
+        """Details about a table alias."""
+
+        part: str  # Name of the part
+        segment: BaseSegment  # Segment containing the part
+
+    @classmethod
+    def _iter_reference_parts(cls, elem) -> Generator[ObjectReferencePart, None, None]:
         """Extract the elements of a reference and yield."""
         # trim on quotes and split out any dots.
         for part in elem.raw_trimmed().split("."):
-            yield part, elem
+            yield cls.ObjectReferencePart(part, elem)
 
-    def iter_raw_references(self):
+    def iter_raw_references(self) -> Generator[ObjectReferencePart, None, None]:
         """Generate a list of reference strings and elements.
 
         Each element is a tuple of (str, segment). If some are
@@ -506,7 +512,7 @@ class ObjectReferenceSegment(BaseSegment):
         """Return the qualification type of this reference."""
         return "qualified" if self.is_qualified() else "unqualified"
 
-    def extract_reference(self, level):
+    def extract_reference(self, level: int) -> Optional[ObjectReferencePart]:
         """Extract a reference of a given level.
 
         e.g. level 1 = the object.
diff --git a/src/sqlfluff/core/parser/segments/base.py b/src/sqlfluff/core/parser/segments/base.py
index 700805b61..f1cf04a9a 100644
--- a/src/sqlfluff/core/parser/segments/base.py
+++ b/src/sqlfluff/core/parser/segments/base.py
@@ -683,19 +683,25 @@ class BaseSegment:
                 buff.append(seg)
         return buff
 
-    def recursive_crawl(self, *seg_type):
+    def recursive_crawl(self, *seg_type, recurse_into=True):
         """Recursively crawl for segments of a given type.
 
         Args:
             seg_type: :obj:`str`: one or more type of segment
                 to look for.
+            recurse_into: :obj:`bool`: When an element of type "seg_type" is
+                found, whether to recurse into it.
         """
         # Check this segment
         if self.is_type(*seg_type):
+            match = True
             yield self
-        # Recurse
-        for seg in self.segments:
-            yield from seg.recursive_crawl(*seg_type)
+        else:
+            match = False
+        if recurse_into or not match:
+            # Recurse
+            for seg in self.segments:
+                yield from seg.recursive_crawl(*seg_type, recurse_into=recurse_into)
 
     def path_to(self, other):
         """Given a segment which is assumed within self, get the intermediate segments.
diff --git a/src/sqlfluff/core/rules/analysis/__init__.py b/src/sqlfluff/core/rules/analysis/__init__.py
new file mode 100644
index 000000000..7725788b9
--- /dev/null
+++ b/src/sqlfluff/core/rules/analysis/__init__.py
@@ -0,0 +1,1 @@
+"""Code analysis tools to support development of more complex rules."""
diff --git a/src/sqlfluff/core/rules/analysis/select.py b/src/sqlfluff/core/rules/analysis/select.py
new file mode 100644
index 000000000..30d701305
--- /dev/null
+++ b/src/sqlfluff/core/rules/analysis/select.py
@@ -0,0 +1,140 @@
+"""Basic code analysis tools for SELECT statements."""
+from typing import List, NamedTuple, Optional
+
+from sqlfluff.core.dialects.base import Dialect
+from sqlfluff.core.dialects.dialect_ansi import AliasInfo
+from sqlfluff.core.parser.segments.base import BaseSegment
+
+
+class SelectStatementColumnsAndTables(NamedTuple):
+    """Structure returned by get_select_statement_info()."""
+
+    select_statement: BaseSegment
+    table_aliases: List[AliasInfo]
+    value_table_function_aliases: List[AliasInfo]
+    reference_buffer: List[BaseSegment]
+    select_targets: List[BaseSegment]
+    col_aliases: List[str]
+    using_cols: List[str]
+
+
+def get_select_statement_info(
+    segment: BaseSegment, dialect: Optional[Dialect], early_exit: bool = True
+) -> Optional[SelectStatementColumnsAndTables]:
+    """Analyze a select statement: targets, aliases, etc. Return info."""
+    assert segment.is_type("select_statement")
+    table_aliases, value_table_function_aliases = get_aliases_from_select(
+        segment, dialect
+    )
+    if early_exit and not table_aliases and not value_table_function_aliases:
+        return None
+
+    # Iterate through all the references, both in the select clause, but also
+    # potential others.
+    sc = segment.get_child("select_clause")
+    reference_buffer = list(sc.recursive_crawl("object_reference"))
+    # Add any wildcard references
+    reference_buffer += list(sc.recursive_crawl("wildcard_identifier"))
+    for potential_clause in (
+        "where_clause",
+        "groupby_clause",
+        "having_clause",
+        "orderby_clause",
+    ):
+        clause = segment.get_child(potential_clause)
+        if clause:
+            reference_buffer += list(clause.recursive_crawl("object_reference"))
+    # PURGE any references which are in nested select statements
+    for ref in reference_buffer.copy():
+        ref_path = segment.path_to(ref)
+        # is it in a subselect? i.e. a select which isn't this one.
+        if any(
+            seg.is_type("select_statement") and seg is not segment for seg in ref_path
+        ):
+            reference_buffer.remove(ref)
+
+    # Get all select targets.
+    select_targets = segment.get_child("select_clause").get_children(
+        "select_target_element"
+    )
+
+    # Get all column aliases
+    col_aliases = []
+    for col_seg in list(sc.recursive_crawl("alias_expression")):
+        for seg in col_seg.segments:
+            if seg.is_type("identifier"):
+                col_aliases.append(seg.raw)
+
+    # Get any columns referred to in a using clause, and extract anything
+    # from ON clauses.
+    using_cols = []
+    fc = segment.get_child("from_clause")
+    if fc:
+        for join_clause in fc.recursive_crawl("join_clause"):
+            in_using_brackets = False
+            seen_using = False
+            for seg in join_clause.segments:
+                if seg.is_type("keyword") and seg.name == "USING":
+                    seen_using = True
+                elif seg.is_type("join_on_condition"):
+                    for on_seg in seg.segments:
+                        if on_seg.is_type("expression"):
+                            # Deal with expressions
+                            reference_buffer += list(
+                                seg.recursive_crawl("object_reference")
+                            )
+                elif seen_using and seg.is_type("start_bracket"):
+                    in_using_brackets = True
+                elif seen_using and seg.is_type("end_bracket"):
+                    in_using_brackets = False
+                    seen_using = False
+                elif in_using_brackets and seg.is_type("identifier"):
+                    using_cols.append(seg.raw)
+
+    return SelectStatementColumnsAndTables(
+        select_statement=segment,
+        table_aliases=table_aliases or [],
+        value_table_function_aliases=value_table_function_aliases or [],
+        reference_buffer=reference_buffer,
+        select_targets=select_targets,
+        col_aliases=col_aliases,
+        using_cols=using_cols,
+    )
+
+
+def get_aliases_from_select(segment, dialect=None):
+    """Gets the aliases referred to in the FROM clause.
+
+    Returns a tuple of two lists:
+    - Table aliases
+    - Value table function aliases
+    """
+    fc = segment.get_child("from_clause")
+    if not fc:
+        # If there's no from clause then just abort.
+        return None, None
+    aliases = fc.get_eventual_aliases()
+
+    # We only want table aliases, so filter out aliases for value table
+    # functions.
+    table_aliases = []
+    value_table_function_aliases = []
+    for table_expr, alias_info in aliases:
+        if not _has_value_table_function(table_expr, dialect):
+            table_aliases.append(alias_info)
+        else:
+            value_table_function_aliases.append(alias_info)
+    return table_aliases, value_table_function_aliases
+
+
+def _has_value_table_function(table_expr, dialect):
+    if not dialect:
+        # We need the dialect to get the value table function names. If
+        # we don't have it, assume the clause does not have a value table
+        # function.
+        return False
+
+    for function_name in table_expr.recursive_crawl("function_name"):
+        if function_name.raw.lower() in dialect.sets("value_table_functions"):
+            return True
+    return False
diff --git a/src/sqlfluff/core/rules/analysis/select_crawler.py b/src/sqlfluff/core/rules/analysis/select_crawler.py
new file mode 100644
index 000000000..004751a03
--- /dev/null
+++ b/src/sqlfluff/core/rules/analysis/select_crawler.py
@@ -0,0 +1,141 @@
+"""Tools for more complex analysis of SELECT statements."""
+from collections import defaultdict
+from typing import Dict, List, NamedTuple, Optional, Union
+
+from cached_property import cached_property
+
+from sqlfluff.core.dialects.dialect_ansi import AliasInfo
+from sqlfluff.core.dialects.base import Dialect
+from sqlfluff.core.parser.segments.base import BaseSegment
+from sqlfluff.core.rules.analysis.select import get_select_statement_info
+
+
+class WildcardInfo(NamedTuple):
+    """Structure returned by SelectCrawler.get_wildcard_info()."""
+
+    segment: BaseSegment
+    tables: List[str]
+
+
+class SelectCrawler:
+    """Class for recursive dependency analysis related to SELECT statements.
+
+    This class is a wrapper for select.get_select_statement_info(), but it adds
+    recursive dependency walking.
+    """
+
+    @classmethod
+    def gather(
+        cls, segment: BaseSegment, dialect: Dialect
+    ) -> Dict[Optional[str], List["SelectCrawler"]]:
+        """Find top-level SELECTs and CTEs, return info."""
+        queries = defaultdict(list)
+        # We specify recurse_into=False because we only want top-level select
+        # statmeents and CTEs. We'll deal with nested selects later as needed,
+        # when processing their top-level parent.
+        for select_statement in segment.recursive_crawl(
+            "select_statement", recurse_into=False
+        ):
+            select_name = cls._get_name_if_cte(select_statement, segment)
+            queries[select_name].append(SelectCrawler(select_statement, dialect))
+        return dict(queries)
+
+    @classmethod
+    def get(
+        cls,
+        segment: BaseSegment,
+        queries: Dict[str, List["SelectCrawler"]],
+        dialect: Dialect,
+    ) -> Union[str, List["SelectCrawler"]]:
+        """Find SELECTs, table refs, or value table function calls in segment.
+
+        If we find a SELECT, return info list. Otherwise, return table name
+        or function call string.
+        """
+        buff = []
+        for seg in segment.recursive_crawl(
+            "table_reference", "select_statement", recurse_into=False
+        ):
+            if seg is segment:
+                # If we are starting with a select_statement, recursive_crawl()
+                # returns the statement itself. Skip that.
+                continue
+
+            if seg.type == "table_reference":
+                if not seg.is_qualified() and seg.raw in queries:
+                    # It's a CTE.
+                    return queries[seg.raw]
+                else:
+                    # It's an external table.
+                    return seg.raw
+            else:
+                assert seg.type == "select_statement"
+                buff.append(SelectCrawler(seg, dialect))
+        if not buff:
+            # If we reach here, the SELECT may be querying from a value table
+            # function, e.g. UNNEST(). For our purposes, this is basically the
+            # same as an external table. Return the "table" part as a string.
+            table_expr = segment.get_child("main_table_expression")
+            if table_expr:
+                return table_expr.raw
+        return buff
+
+    def __init__(self, select_statement, dialect):
+        self.select_statement = select_statement
+        self.dialect = dialect
+
+    @cached_property
+    def select_info(self):
+        """Returns SelectStatementColumnsAndTables on the SELECT."""
+        result = get_select_statement_info(
+            self.select_statement, self.dialect, early_exit=False
+        )
+        return result
+
+    def find_alias(self, table: str) -> Optional[AliasInfo]:
+        """Find corresponding table_aliases entry (if any) matching "table"."""
+        alias_info = [
+            t
+            for t in self.select_info.table_aliases
+            if t.aliased and t.ref_str == table
+        ]
+        assert len(alias_info) <= 1
+        return alias_info[0] if alias_info else None
+
+    def get_wildcard_info(self) -> List[WildcardInfo]:
+        """Find wildcard (*) targets in the SELECT."""
+        buff = []
+        for seg in self.select_info.select_targets:
+            if seg.get_child("wildcard_expression"):
+                if "." in seg.raw:
+                    # The wildcard specifies a target table.
+                    table = seg.raw.rsplit(".", 1)[0]
+                    buff.append(WildcardInfo(seg, [table]))
+                else:
+                    # The wildcard is unqualified (i.e. does not specify a
+                    # table). This means to include all columns from all the
+                    # tables in the query.
+                    buff.append(
+                        WildcardInfo(
+                            seg,
+                            [
+                                alias_info.ref_str
+                                for alias_info in self.select_info.table_aliases
+                            ],
+                        )
+                    )
+        return buff
+
+    @staticmethod
+    def _get_name_if_cte(
+        select_statement: BaseSegment, ancestor_segment: BaseSegment
+    ) -> Optional[str]:
+        """Return name if CTE. If top-level, return None."""
+        cte = None
+        path_to = ancestor_segment.path_to(select_statement)
+        for seg in path_to:
+            if seg.is_type("common_table_expression"):
+                cte = seg
+                break
+        select_name = cte.segments[0].raw if cte else None
+        return select_name
diff --git a/src/sqlfluff/core/rules/std/L020.py b/src/sqlfluff/core/rules/std/L020.py
index 905fa6321..c00e8d9de 100644
--- a/src/sqlfluff/core/rules/std/L020.py
+++ b/src/sqlfluff/core/rules/std/L020.py
@@ -3,6 +3,7 @@
 import itertools
 
 from sqlfluff.core.rules.base import BaseCrawler, LintResult
+from sqlfluff.core.rules.analysis.select import get_select_statement_info
 
 
 class Rule_L020(BaseCrawler):
@@ -40,45 +41,7 @@ class Rule_L020(BaseCrawler):
                 ]
         return None
 
-    @staticmethod
-    def _has_value_table_function(table_expr, dialect):
-        if not dialect:
-            # We need the dialect to get the value table function names. If
-            # we don't have it, assume the clause does not have a value table
-            # function.
-            return False
-
-        for function_name in table_expr.recursive_crawl("function_name"):
-            if function_name.raw.lower() in dialect.sets("value_table_functions"):
-                return True
-        return False
-
-    @classmethod
-    def _get_aliases_from_select(cls, segment, dialect=None):
-        """Gets the aliases referred to in the FROM clause.
-
-        Returns a tuple of two lists:
-        - Table aliases
-        - Value table function aliases
-        """
-        fc = segment.get_child("from_clause")
-        if not fc:
-            # If there's no from clause then just abort.
-            return None, None
-        aliases = fc.get_eventual_aliases()
-
-        # We only want table aliases, so filter out aliases for value table
-        # functions.
-        table_aliases = []
-        value_table_function_aliases = []
-        for table_expr, alias_info in aliases:
-            if not cls._has_value_table_function(table_expr, dialect):
-                table_aliases.append(alias_info)
-            else:
-                value_table_function_aliases.append(alias_info)
-        return table_aliases, value_table_function_aliases
-
-    def _eval(self, segment, parent_stack, **kwargs):
+    def _eval(self, segment, parent_stack, dialect, **kwargs):
         """Get References and Aliases and allow linting.
 
         This rule covers a lot of potential cases of odd usages of
@@ -88,69 +51,10 @@ class Rule_L020(BaseCrawler):
         `_lint_references_and_aliases` method.
         """
         if segment.is_type("select_statement"):
-            table_aliases, value_table_function_aliases = self._get_aliases_from_select(
-                segment, kwargs.get("dialect")
-            )
-            if not table_aliases and not value_table_function_aliases:
+            select_info = get_select_statement_info(segment, dialect)
+            if not select_info:
                 return None
 
-            # Iterate through all the references, both in the select clause, but also
-            # potential others.
-            sc = segment.get_child("select_clause")
-            reference_buffer = list(sc.recursive_crawl("object_reference"))
-            # Add any wildcard references
-            reference_buffer += list(sc.recursive_crawl("wildcard_identifier"))
-            for potential_clause in (
-                "where_clause",
-                "groupby_clause",
-                "having_clause",
-                "orderby_clause",
-            ):
-                clause = segment.get_child(potential_clause)
-                if clause:
-                    reference_buffer += list(clause.recursive_crawl("object_reference"))
-            # PURGE any references which are in nested select statements
-            for ref in reference_buffer.copy():
-                ref_path = segment.path_to(ref)
-                # is it in a subselect? i.e. a select which isn't this one.
-                if any(
-                    seg.is_type("select_statement") and seg is not segment
-                    for seg in ref_path
-                ):
-                    reference_buffer.remove(ref)
-
-            # Get all column aliases
-            col_aliases = []
-            for col_seg in list(sc.recursive_crawl("alias_expression")):
-                for seg in col_seg.segments:
-                    if seg.is_type("identifier"):
-                        col_aliases.append(seg.raw)
-
-            # Get any columns referred to in a using clause, and extract anything
-            # from ON clauses.
-            using_cols = []
-            fc = segment.get_child("from_clause")
-            for join_clause in fc.recursive_crawl("join_clause"):
-                in_using_brackets = False
-                seen_using = False
-                for seg in join_clause.segments:
-                    if seg.is_type("keyword") and seg.name == "USING":
-                        seen_using = True
-                    elif seg.is_type("join_on_condition"):
-                        for on_seg in seg.segments:
-                            if on_seg.is_type("expression"):
-                                # Deal with expressions
-                                reference_buffer += list(
-                                    seg.recursive_crawl("object_reference")
-                                )
-                    elif seen_using and seg.is_type("start_bracket"):
-                        in_using_brackets = True
-                    elif seen_using and seg.is_type("end_bracket"):
-                        in_using_brackets = False
-                        seen_using = False
-                    elif in_using_brackets and seg.is_type("identifier"):
-                        using_cols.append(seg.raw)
-
             # Work out if we have a parent select function
             parent_select = None
             for seg in reversed(parent_stack):
@@ -161,11 +65,11 @@ class Rule_L020(BaseCrawler):
             # Pass them all to the function that does all the work.
             # NB: Subclasses of this rules should override the function below
             return self._lint_references_and_aliases(
-                table_aliases,
-                value_table_function_aliases,
-                reference_buffer,
-                col_aliases,
-                using_cols,
+                select_info.table_aliases,
+                select_info.value_table_function_aliases,
+                select_info.reference_buffer,
+                select_info.col_aliases,
+                select_info.using_cols,
                 parent_select,
             )
         return None
diff --git a/src/sqlfluff/core/rules/std/L026.py b/src/sqlfluff/core/rules/std/L026.py
index 7d7fddf4f..ac8a2461f 100644
--- a/src/sqlfluff/core/rules/std/L026.py
+++ b/src/sqlfluff/core/rules/std/L026.py
@@ -1,5 +1,6 @@
 """Implementation of Rule L026."""
 
+from sqlfluff.core.rules.analysis.select import get_aliases_from_select
 from sqlfluff.core.rules.base import LintResult
 from sqlfluff.core.rules.std.L025 import Rule_L025
 
@@ -46,7 +47,7 @@ class Rule_L026(Rule_L025):
             if tbl_ref and tbl_ref[0] not in [a.ref_str for a in table_aliases]:
                 # Last check, this *might* be a correlated subquery reference.
                 if parent_select:
-                    parent_aliases, _ = self._get_aliases_from_select(parent_select)
+                    parent_aliases, _ = get_aliases_from_select(parent_select)
                     if parent_aliases and tbl_ref[0] in [a[0] for a in parent_aliases]:
                         continue
 
diff --git a/src/sqlfluff/core/rules/std/L044.py b/src/sqlfluff/core/rules/std/L044.py
new file mode 100644
index 000000000..d735b041e
--- /dev/null
+++ b/src/sqlfluff/core/rules/std/L044.py
@@ -0,0 +1,140 @@
+"""Implementation of Rule L044."""
+from typing import Dict, List
+
+from sqlfluff.core.rules.analysis.select_crawler import SelectCrawler
+from sqlfluff.core.dialects.base import Dialect
+from sqlfluff.core.rules.base import BaseCrawler, LintResult
+
+
+class RuleFailure(Exception):
+    """Exception class for reporting lint failure inside deeply nested code."""
+
+    pass
+
+
+class Rule_L044(BaseCrawler):
+    """Query produces an unknown number of result columns.
+
+    | **Anti-pattern**
+    | Querying all columns using `*` produces a query result where the number
+    | or ordering of columns changes if the upstream table's schema changes.
+    | This should generally be avoided because it can cause slow performance,
+    | cause important schema changes to go undetected, or break production code.
+    | For example:
+    | * If a query does `SELECT t.*` and is expected to return columns `a`, `b`,
+    |   and `c`, the actual columns returned will be wrong/different if columns
+    |   are added to or deleted from the input table.
+    | * `UNION` and `DIFFERENCE` clauses require the inputs have the same number
+    |   of columns (and compatible types).
+    | * `JOIN` queries may break due to new column name conflicts, e.g. the
+    |   query references a column "c" which initially existed in only one input
+    |   table but a column of the same name is added to another table.
+    | * `CREATE TABLE (<<column schema>>) AS SELECT *`
+
+
+    .. code-block::
+
+        WITH cte AS (
+            SELECT * FROM foo
+        )
+
+        SELECT * FROM cte
+        UNION
+        SELECT a, b FROM t
+
+    | **Best practice**
+    | Somewhere along the "path" to the source data, specify columns explicitly.
+
+    .. code-block::
+
+        WITH cte AS (
+            SELECT * FROM foo
+        )
+
+        SELECT a, b FROM cte
+        UNION
+        SELECT a, b FROM t
+
+    """
+
+    _works_on_unparsable = False
+
+    def _handle_alias(self, alias_info, dialect, queries):
+        select_info_target = SelectCrawler.get(
+            alias_info.table_expression, queries, dialect
+        )
+        if isinstance(select_info_target, str):
+            # It's an alias to an external table whose
+            # number of columns could vary without our
+            # knowledge. Thus, warn.
+            self.logger.debug(
+                f"Query target {select_info_target} is external. Generating warning."
+            )
+            raise RuleFailure()
+        else:
+            # Handle nested SELECT.
+            self._analyze_result_columns(select_info_target, dialect, queries)
+
+    def _analyze_result_columns(
+        self,
+        select_info_list: List[SelectCrawler],
+        dialect: Dialect,
+        queries: Dict[str, List[SelectCrawler]],
+    ):
+        """Given info on a list of SELECTs, determine whether to warn."""
+        # Recursively walk from the given query (select_info_list) to any
+        # wildcard columns in the select targets. If every wildcard evdentually
+        # resolves to a query without wildcards, all is well. Otherwise, warn.
+        for select_info in select_info_list:
+            self.logger.debug(f"Analyzing query: {select_info.select_statement.raw}")
+            for wildcard in select_info.get_wildcard_info():
+                if wildcard.tables:
+                    for wildcard_table in wildcard.tables:
+                        self.logger.debug(
+                            f"Wildcard: {wildcard.segment.raw} has target {wildcard_table}"
+                        )
+                        # Is it an alias?
+                        alias_info = select_info.find_alias(wildcard_table)
+                        if alias_info:
+                            # Found the alias matching the wildcard. Recurse,
+                            # analyzing the query associated with that alias.
+                            self._handle_alias(alias_info, dialect, queries)
+                        else:
+                            # Not an alias. Is it a CTE?
+                            if wildcard_table in queries:
+                                # Wildcard refers to a CTE. Analyze it.
+                                self._analyze_result_columns(
+                                    queries[wildcard_table], dialect, queries
+                                )
+                            else:
+                                # Not CTE, not table alias. Presumably an
+                                # external table. Warn.
+                                self.logger.debug(
+                                    f"Query target {wildcard_table} is external. Generating warning."
+                                )
+                                raise RuleFailure()
+                else:
+                    # No table was specified with the wildcard. Assume we're
+                    # querying from a nested select in FROM.
+                    select_info_target = SelectCrawler.get(
+                        select_info.select_statement, queries, dialect
+                    )
+                    assert isinstance(select_info_target, list)
+                    self._analyze_result_columns(
+                        select_info_target,
+                        dialect,
+                        queries,
+                    )
+
+    def _eval(self, segment, dialect, **kwargs):
+        """Outermost query should produce known number of columns."""
+        if segment.is_type("statement"):
+            queries = SelectCrawler.gather(segment, dialect)
+
+            # Begin analysis at the final, outer query (key=None).
+            select_info = queries[None]
+            try:
+                return self._analyze_result_columns(select_info, dialect, queries)
+            except RuleFailure:
+                return LintResult(anchor=queries[None][0].select_info.select_statement)
+        return None
diff --git a/src/sqlfluff/core/rules/std/L045.py b/src/sqlfluff/core/rules/std/L045.py
new file mode 100644
index 000000000..50d260f70
--- /dev/null
+++ b/src/sqlfluff/core/rules/std/L045.py
@@ -0,0 +1,73 @@
+"""Implementation of Rule L045."""
+from typing import Dict, List
+
+from sqlfluff.core.dialects.base import Dialect
+from sqlfluff.core.rules.base import BaseCrawler, LintResult
+from sqlfluff.core.rules.analysis.select_crawler import SelectCrawler
+from sqlfluff.core.rules.doc_decorators import document_fix_compatible
+
+
+@document_fix_compatible
+class Rule_L045(BaseCrawler):
+    """Query defines a CTE (common-table expression) but does not use it.
+
+    | **Anti-pattern**
+    | Defining a CTE that is not used by the query is harmless, but it means
+    | the code is unnecesary and could be removed.
+
+    .. code-block::
+
+        WITH cte1 AS (
+          SELECT a
+          FROM t
+        ),
+        cte2 AS (
+          SELECT b
+          FROM u
+        )
+
+        SELECT *
+        FROM cte1
+
+    | **Best practice**
+    | Remove unused CTEs.
+
+    .. code-block::
+
+        WITH cte1 AS (
+          SELECT a
+          FROM t
+        )
+
+        SELECT *
+        FROM cte1
+    """
+
+    @classmethod
+    def _visit_sources(
+        cls,
+        select_info_list: List[SelectCrawler],
+        dialect: Dialect,
+        queries: Dict[str, List[SelectCrawler]],
+    ):
+        for select_info in select_info_list:
+            for alias_info in select_info.select_info.table_aliases:
+                # Does the query read from a CTE? If so, visit the CTE.
+                for target_segment in alias_info.table_expression.get_children(
+                    "main_table_expression", "join_clause"
+                ):
+                    target = target_segment.raw
+                    if target in queries:
+                        select_info_target = queries.pop(target)
+                        if isinstance(select_info_target, list):
+                            cls._visit_sources(select_info_target, dialect, queries)
+
+    def _eval(self, segment, dialect, **kwargs):
+        if segment.is_type("statement"):
+            queries = SelectCrawler.gather(segment, dialect)
+
+            # Begin analysis at the final, outer query (key=None).
+            self._visit_sources(queries.pop(None), dialect, queries)
+            if queries:
+                return LintResult(anchor=segment)
+        return None


Test patch:
diff --git a/test/api/simple_test.py b/test/api/simple_test.py
index 789b03bab..352bb7b0a 100644
--- a/test/api/simple_test.py
+++ b/test/api/simple_test.py
@@ -20,6 +20,12 @@ lint_result = [
         "line_no": 1,
         "line_pos": 1,
     },
+    {
+        "code": "L044",
+        "description": "Query produces an unknown number of result columns.",
+        "line_no": 1,
+        "line_pos": 1,
+    },
     {
         "code": "L039",
         "description": "Unnecessary whitespace found.",
diff --git a/test/core/rules/test_cases/L044.yml b/test/core/rules/test_cases/L044.yml
new file mode 100644
index 000000000..3ca762d9d
--- /dev/null
+++ b/test/core/rules/test_cases/L044.yml
@@ -0,0 +1,287 @@
+rule: L044
+
+test_1:
+  pass_str: select a, b from t
+
+test_2:
+  fail_str: select * from t
+
+test_3:
+  pass_str: |
+    with cte as (
+        select
+            a, b
+        from
+            t
+    )
+    select * from cte
+
+test_4:
+  fail_str: |
+    with cte as (
+        select
+            *
+        from
+            t
+    )
+    select * from cte
+
+
+test_5:
+  pass_str: |
+    with cte as (
+        select
+            *
+        from
+            t
+    )
+    select a, b from cte
+
+test_6:
+  pass_str: |
+    with cte1 as (
+        select
+            *
+        from
+            t
+    ),
+    cte2 as (
+        select
+            a, b
+        from
+            t
+    )
+    select
+        cte1.a,
+        cte2.*
+    from cte1
+    join cte2
+    using (a)
+
+test_7:
+  fail_str: |
+    with cte1 as (
+        select
+            *
+        from
+            t
+    ),
+    cte2 as (
+        select
+            a, b
+        from
+            t
+    )
+    select
+        cte1.*,
+        cte2.*
+    from cte1
+    join cte2
+    using (a)
+
+test_8:
+  pass_str: |
+    select * from (
+        select
+            a, b
+        from
+            t
+    )
+
+test_9:
+  fail_str: |
+    with cte as (
+        select
+            a, b
+        from
+            t
+    )
+    select
+        cte.*,
+        t.*
+    from cte1
+    join t
+    using (a)
+
+test_10:
+  fail_str: |
+    with cte as (
+        select
+            a, b
+        from
+            t
+    )
+    select
+        cte.*,
+        t_alias.*
+    from cte1
+    join (select * from t) as t_alias
+    using (a)
+
+test_11:
+  pass_str: |
+    select
+        t_alias.*
+    from cte1
+    join (select a from t) as t_alias
+    using (a)
+
+test_12:
+  fail_str: |
+    select
+        t_alias.*
+    from t1
+    join (select * from t) as t_alias
+    using (a)
+
+test_13:
+  pass_str: |
+    with cte as (
+        select
+            a, b
+        from
+            t
+    )
+    select
+        cte.*,
+        t_alias.a
+    from cte1
+    join (select * from t) as t_alias
+    using (a)
+
+test_14:
+  fail_str: select *, t.*, t.a, b from t
+
+test_15:
+  pass_str: |
+    select a from t1
+    union all
+    select b from t2
+
+test_16:
+  fail_str: |
+    select a from t1
+    union all
+    select * from t2
+
+test_17:
+  fail_str: |
+    select * from t1
+    union all
+    select b from t2
+
+test_18:
+  fail_str: |
+    with cte as (
+      select * from t1
+      union all
+      select b from t2
+    )
+
+    select * from cte
+    union all
+    select b from t3
+
+test_19:
+  pass_str: |
+    with cte as (
+      select a from t1
+      union all
+      select b from t2
+    )
+
+    select * from cte
+    union all
+    select b from t3
+
+test_20:
+  pass_str: |
+    WITH cte_orders AS (
+      SELECT customer_id, total
+      FROM orders
+    )
+
+    SELECT customers.name, cte_orders.*
+    FROM
+      customers,
+      cte_orders
+    WHERE
+      clients.id = orders.clientId
+
+test_21:
+  pass_str: |
+    WITH cte_orders AS (
+      SELECT customer_id, total
+      FROM orders
+    )
+
+    SELECT *
+    FROM
+      cte_orders AS orders
+
+test_22:
+  fail_str: |
+    WITH cte_orders AS (
+      SELECT *
+      FROM orders
+    )
+
+    SELECT *
+    FROM
+      cte_orders AS orders
+
+test_23:
+  fail_str: |
+    WITH cte_orders AS (
+      SELECT customer_id, total
+      FROM orders
+    )
+
+    SELECT *
+    FROM
+      orders AS cte_orders
+
+test_24:
+  fail_str: SELECT p.* FROM races, UNNEST(participants) AS p
+
+test_25:
+  pass_str: SELECT p FROM races, UNNEST(participants) AS p
+
+test_26:
+  fail_str: SELECT * FROM a JOIN b
+
+test_27:
+  fail_str: |
+    WITH cte AS (
+      SELECT a
+      FROM t
+    )
+    SELECT *
+    FROM cte
+    JOIN b
+
+test_28:
+  pass_str: |
+    WITH cte1 AS (
+      SELECT a
+      FROM t
+    ),
+    cte2 AS (
+      SELECT b
+      FROM u
+    )
+    SELECT *
+    FROM cte1
+    JOIN cte2
+
+test_29:
+  pass_str: select a, b from `d.t`
+  configs:
+    core:
+      dialect: bigquery
+
+test_30:
+  fail_str: select * from `d.t`
+  configs:
+    core:
+      dialect: bigquery
diff --git a/test/core/rules/test_cases/L045.yml b/test/core/rules/test_cases/L045.yml
new file mode 100644
index 000000000..5ad8fa642
--- /dev/null
+++ b/test/core/rules/test_cases/L045.yml
@@ -0,0 +1,78 @@
+rule: L045
+
+test_1:
+  pass_str: select * from t
+
+test_2:
+  pass_str: |
+    with cte as (
+        select
+            a, b
+        from
+            t
+    )
+    select * from cte
+
+test_3:
+  pass_str: |
+    WITH cte1 AS (
+      SELECT a
+      FROM t
+    ),
+    cte2 AS (
+      SELECT b
+      FROM u
+    )
+    SELECT *
+    FROM cte1
+    JOIN cte2
+
+test_4:
+  fail_str: |
+    WITH cte1 AS (
+      SELECT a
+      FROM t
+    ),
+    cte2 AS (
+      SELECT b
+      FROM u
+    )
+    SELECT *
+    FROM cte1
+
+test_5:
+  fail_str: |
+    WITH cte_orders AS (
+      SELECT customer_id, total
+      FROM orders
+    )
+
+    SELECT *
+    FROM
+      orders AS cte_orders
+
+test_6:
+  pass_str: |
+    WITH cte1 AS (
+      SELECT a
+      FROM t
+    ),
+    cte2 AS (
+      SELECT b
+      FROM cte1
+    )
+    SELECT *
+    FROM cte2
+
+test_7:
+  fail_str: |
+    WITH cte1 AS (
+      SELECT a
+      FROM t
+    ),
+    cte2 AS (
+      SELECT b
+      FROM cte1
+    )
+    SELECT *
+    FROM cte1
\ No newline at end of file