Repository: sqlfluff/sqlfluff

Issue:
L042 loop limit on fixes reached when CTE itself contains a subquery
### Search before asking

- [X] I searched the [issues](https://github.com/sqlfluff/sqlfluff/issues) and found no similar issues.


### What Happened

While running `sqlfluff fix --dialect snowflake` on a sql file, I get 
```
==== finding fixable violations ====
WARNING    Loop limit on fixes reached [10].                                                                                                                                                              
==== no fixable linting violations found ====                                                                                                                                                             
All Finished 📜 🎉!
  [22 unfixable linting violations found]
```

```
INSERT OVERWRITE INTO dwh.test_table

WITH cte1 AS (
	SELECT *
	FROM (SELECT
		*,
		ROW_NUMBER() OVER (PARTITION BY r ORDER BY updated_at DESC) AS latest
		FROM mongo.temp
	WHERE latest = 1
))

SELECT * FROM cte1 WHERE 1=1;
```

All of the 22  violations are a mix of L002, L003 and L004.

### Expected Behaviour

`sqlfluff` should be able to fix the violations

### Observed Behaviour

Even if I try to fix the violations manually, it still shows the same error.

### How to reproduce

I will try to generate a sql file that will be able to reproduce the issue

### Dialect

Snowflake

### Version

1.1.0

### Configuration

```
# https://docs.sqlfluff.com/en/stable/rules.html

[sqlfluff]
exclude_rules = L029, L031, L034

[sqlfluff:indentation]
indented_joins = true
indented_using_on = true

[sqlfluff:rules:L002]
tab_space_size = 4

[sqlfluff:rules:L003]
hanging_indents = true
indent_unit = tab
tab_space_size = 4

[sqlfluff:rules:L004]
indent_unit = tab
tab_space_size = 4

[sqlfluff:rules:L010]
capitalisation_policy = upper

[sqlfluff:rules:L011]
aliasing = explicit

[sqlfluff:rules:L012]
aliasing = explicit

[sqlfluff:rules:L014]
extended_capitalisation_policy = lower

[sqlfluff:rules:L016]
ignore_comment_clauses = true
ignore_comment_lines = true
indent_unit = tab
tab_space_size = 4

[sqlfluff:rules:L019]
comma_style = trailing

[sqlfluff:rules:L022]
comma_style = trailing

[sqlfluff:rules:L028]
single_table_references = unqualified

[sqlfluff:rules:L030]
extended_capitalisation_policy = upper

[sqlfluff:rules:L040]
capitalisation_policy = upper

[sqlfluff:rules:L042]
forbid_subquery_in = both

[sqlfluff:rules:L054]
group_by_and_order_by_style = explicit

[sqlfluff:rules:L063]
extended_capitalisation_policy = upper

[sqlfluff:rules:L066]
min_alias_length = 3
max_alias_length = 15

[sqlfluff:templater:jinja:context]
params = {"DB": "DEMO"}
```

### Are you willing to work on and submit a PR to address the issue?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/sqlfluff/sqlfluff/blob/main/CODE_OF_CONDUCT.md)


Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) Report
> Merging [#3700](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) (81d3061) into [main](https://codecov.io/gh/sqlfluff/sqlfluff/commit/213dd62198d16ebc3bab6a7e73b6654e3141036b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) (213dd62) will **not change** coverage.
> The diff coverage is `100.00%`.

```diff
@@            Coverage Diff            @@
##              main     #3700   +/-   ##
=========================================
  Coverage   100.00%   100.00%           
=========================================
  Files          176       176           
  Lines        13413     13413           
=========================================
  Hits         13413     13413           
```


| [Impacted Files](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff) | Coverage Δ | |
|---|---|---|
| [src/sqlfluff/rules/L042.py](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff#diff-c3JjL3NxbGZsdWZmL3J1bGVzL0wwNDIucHk=) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff). Last update [213dd62...81d3061](https://codecov.io/gh/sqlfluff/sqlfluff/pull/3700?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sqlfluff).

alanmcruickshank: @barrywhart - are you ok incorporating #3717 into this or do you need a hand?
barrywhart: @alanmcruickshank: I'm okay, thanks! This is a slow-moving, slightly experimental PR anyway.

Patch:
diff --git a/src/sqlfluff/core/parser/segments/base.py b/src/sqlfluff/core/parser/segments/base.py
index d92e451ca..9775d060b 100644
--- a/src/sqlfluff/core/parser/segments/base.py
+++ b/src/sqlfluff/core/parser/segments/base.py
@@ -1749,6 +1749,11 @@ class IdentitySet(MutableSet):
         """Add an element."""
         self.map[self.key(value)] = value
 
+    def update(self, value):
+        """Add elements in 'value'."""
+        for v in value:
+            self.add(v)
+
     def discard(self, value):  # MutableSet
         """Remove an element.  Do not raise an exception if absent."""
         self.map.pop(self.key(value), None)  # pragma: no cover
diff --git a/src/sqlfluff/rules/L028.py b/src/sqlfluff/rules/L028.py
index 30434b347..607f9c4ff 100644
--- a/src/sqlfluff/rules/L028.py
+++ b/src/sqlfluff/rules/L028.py
@@ -3,8 +3,9 @@
 from typing import Iterator, List, Optional, Set
 
 from sqlfluff.core.dialects.common import AliasInfo, ColumnAliasInfo
-from sqlfluff.core.parser.segments.base import BaseSegment
+from sqlfluff.core.parser.segments.base import BaseSegment, IdentitySet
 from sqlfluff.core.parser.segments.raw import SymbolSegment
+from sqlfluff.utils.analysis.select import SelectStatementColumnsAndTables
 from sqlfluff.utils.analysis.select_crawler import Query, SelectCrawler
 from sqlfluff.core.rules import (
     BaseRule,
@@ -99,12 +100,16 @@ class Rule_L028(BaseRule):
 
         if not FunctionalContext(context).parent_stack.any(sp.is_type(*_START_TYPES)):
             crawler = SelectCrawler(context.segment, context.dialect)
+            visited: IdentitySet = IdentitySet()
             if crawler.query_tree:
                 # Recursively visit and check each query in the tree.
-                return list(self._visit_queries(crawler.query_tree))
+                return list(self._visit_queries(crawler.query_tree, visited))
         return None
 
-    def _visit_queries(self, query: Query) -> Iterator[LintResult]:
+    def _visit_queries(
+        self, query: Query, visited: IdentitySet
+    ) -> Iterator[LintResult]:
+        select_info: Optional[SelectStatementColumnsAndTables] = None
         if query.selectables:
             select_info = query.selectables[0].select_info
             # How many table names are visible from here? If more than one then do
@@ -138,8 +143,24 @@ class Rule_L028(BaseRule):
                     self._fix_inconsistent_to,
                     fixable,
                 )
-        for child in query.children:
-            yield from self._visit_queries(child)
+        children = list(query.children)
+        # 'query.children' includes CTEs and "main" queries, but not queries in
+        # the "FROM" list. We want to visit those as well.
+        if select_info:
+            for a in select_info.table_aliases:
+                for q in SelectCrawler.get(query, a.from_expression_element):
+                    if not isinstance(q, Query):
+                        continue
+                    # Check for previously visited selectables to avoid possible
+                    # infinite recursion, e.g.:
+                    #   WITH test1 AS (SELECT i + 1, j + 1 FROM test1)
+                    #   SELECT * FROM test1;
+                    if any(s.selectable in visited for s in q.selectables):
+                        continue
+                    visited.update(s.selectable for s in q.selectables)
+                    children.append(q)
+        for child in children:
+            yield from self._visit_queries(child, visited)
 
 
 def _check_references(
diff --git a/src/sqlfluff/rules/L042.py b/src/sqlfluff/rules/L042.py
index 711f05b64..8c3eebcbd 100644
--- a/src/sqlfluff/rules/L042.py
+++ b/src/sqlfluff/rules/L042.py
@@ -2,7 +2,7 @@
 import copy
 from functools import partial
 from typing import (
-    Generator,
+    Iterator,
     List,
     NamedTuple,
     Optional,
@@ -12,7 +12,9 @@ from typing import (
     TypeVar,
     cast,
 )
+
 from sqlfluff.core.dialects.base import Dialect
+from sqlfluff.core.dialects.common import AliasInfo
 from sqlfluff.core.parser.segments.base import BaseSegment
 from sqlfluff.core.parser.segments.raw import (
     CodeSegment,
@@ -21,8 +23,15 @@ from sqlfluff.core.parser.segments.raw import (
     SymbolSegment,
     WhitespaceSegment,
 )
-from sqlfluff.core.rules import BaseRule, LintFix, LintResult, RuleContext
+from sqlfluff.core.rules import (
+    BaseRule,
+    EvalResultType,
+    LintFix,
+    LintResult,
+    RuleContext,
+)
 from sqlfluff.utils.analysis.select import get_select_statement_info
+from sqlfluff.utils.analysis.select_crawler import Query, Selectable, SelectCrawler
 from sqlfluff.core.rules.crawlers import SegmentSeekerCrawler
 from sqlfluff.core.rules.doc_decorators import (
     document_configuration,
@@ -51,10 +60,11 @@ _SELECT_TYPES = [
 
 
 class _NestedSubQuerySummary(NamedTuple):
-    parent_clause_type: str
-    parent_select_segments: Segments
-    clause_segments: Segments
-    subquery: BaseSegment
+    query: Query
+    selectable: Selectable
+    table_alias: AliasInfo
+    sc: SelectCrawler
+    select_source_names: Set[str]
 
 
 @document_groups
@@ -107,137 +117,164 @@ class Rule_L042(BaseRule):
         "both": ["join_clause", "from_expression_element"],
     }
 
-    def _eval(self, context: RuleContext) -> Optional[List[LintResult]]:
+    def _eval(self, context: RuleContext) -> EvalResultType:
         """Join/From clauses should not contain subqueries. Use CTEs instead."""
         self.forbid_subquery_in: str
-        parent_types = self._config_mapping[self.forbid_subquery_in]
-        segment = FunctionalContext(context).segment
-        parent_stack = FunctionalContext(context).parent_stack
+        functional_context = FunctionalContext(context)
+        segment = functional_context.segment
+        parent_stack = functional_context.parent_stack
+        is_select = segment.all(is_type(*_SELECT_TYPES))
         is_select_child = parent_stack.any(is_type(*_SELECT_TYPES))
-        if is_select_child:
+        if not is_select or is_select_child:
             # Nothing to do.
             return None
 
-        # Gather all possible offending Elements in one crawl
-        nested_subqueries: List[_NestedSubQuerySummary] = []
-        selects = segment.recursive_crawl(*_SELECT_TYPES, recurse_into=True)
-        for select in selects.iterate_segments():
-            for res in _find_nested_subqueries(select, context.dialect):
-                if res.parent_clause_type not in parent_types:
-                    continue
-                nested_subqueries.append(res)
+        crawler = SelectCrawler(context.segment, context.dialect)
+        assert crawler.query_tree
+
+        # generate an instance which will track and shape our output CTE
+        ctes = _CTEBuilder()
+        # Init the output/final select &
+        # populate existing CTEs
+        for cte in crawler.query_tree.ctes.values():
+            ctes.insert_cte(cte.cte_definition_segment)  # type: ignore
+
+        is_with = segment.all(is_type("with_compound_statement"))
+        # TODO: consider if we can fix recursive CTEs
+        is_recursive = is_with and len(segment.children(is_keyword("recursive"))) > 0
+        case_preference = _get_case_preference(segment)
+        output_select = segment
+        if is_with:
+            output_select = segment.children(
+                is_type(
+                    "set_expression",
+                    "select_statement",
+                )
+            )
 
-        if not nested_subqueries:
-            return None
         # If there are offending elements calculate fixes
-        return _calculate_fixes(
+        clone_map = SegmentCloneMap(segment[0])
+        result = self._lint_query(
             dialect=context.dialect,
-            root_select=segment,
-            nested_subqueries=nested_subqueries,
-            parent_stack=parent_stack,
+            query=crawler.query_tree,
+            ctes=ctes,
+            case_preference=case_preference,
+            clone_map=clone_map,
         )
 
-
-def _calculate_fixes(
-    dialect: Dialect,
-    root_select: Segments,
-    nested_subqueries: List[_NestedSubQuerySummary],
-    parent_stack: Segments,
-) -> List[LintResult]:
-    """Given the Root select and the offending subqueries calculate fixes."""
-    is_with = root_select.all(is_type("with_compound_statement"))
-    # TODO: consider if we can fix recursive CTEs
-    is_recursive = is_with and len(root_select.children(is_keyword("recursive"))) > 0
-    case_preference = _get_case_preference(root_select)
-    # generate an instance which will track and shape our output CTE
-    ctes = _CTEBuilder()
-    # Init the output/final select &
-    # populate existing CTEs
-    for cte in root_select.children(is_type("common_table_expression")):
-        assert isinstance(cte, CTEDefinitionSegment), "TypeGuard"
-        ctes.insert_cte(cte)
-
-    output_select = root_select
-    if is_with:
-        output_select = root_select.children(
-            is_type(
-                "set_expression",
-                "select_statement",
+        if result:
+            lint_result, from_expression, alias_name, subquery_parent = result
+            assert any(
+                from_expression is seg for seg in subquery_parent.recursive_crawl_all()
             )
-        )
+            this_seg_clone = clone_map[from_expression]
+            new_table_ref = _create_table_ref(alias_name, context.dialect)
+            this_seg_clone.segments = [new_table_ref]
+            ctes.replace_with_clone(subquery_parent, clone_map)
+
+            # Issue 3617: In T-SQL (and possibly other dialects) the automated fix
+            # leaves parentheses in a location that causes a syntax error. This is an
+            # unusual corner case. For simplicity, we still generate the lint warning
+            # but don't try to generate a fix. Someone could look at this later (a
+            # correct fix would involve removing the parentheses.)
+            bracketed_ctas = [seg.type for seg in parent_stack[-2:]] == [
+                "create_table_statement",
+                "bracketed",
+            ]
+            if bracketed_ctas or ctes.has_duplicate_aliases() or is_recursive:
+                # If we have duplicate CTE names just don't fix anything
+                # Return the lint warnings anyway
+                return lint_result
+
+            # Compute fix.
+            edit = [
+                ctes.compose_select(
+                    clone_map[output_select[0]],
+                    case_preference=case_preference,
+                ),
+            ]
+            lint_result.fixes = [
+                LintFix.replace(
+                    segment[0],
+                    edit_segments=edit,
+                )
+            ]
+            return lint_result
+        return None
 
-    lint_results: List[LintResult] = []
-    clone_map = SegmentCloneMap(root_select[0])
-    is_new_name = False
-    new_table_ref = None
-    for parent_type, _, this_seg, subquery in nested_subqueries:
-        alias_name, is_new_name = ctes.create_cte_alias(
-            this_seg.children(is_type("alias_expression"))
-        )
-        new_cte = _create_cte_seg(
-            alias_name=alias_name,
-            subquery=clone_map[subquery],
-            case_preference=case_preference,
-            dialect=dialect,
-        )
-        ctes.insert_cte(new_cte)
-        this_seg_clone = clone_map[this_seg[0]]
-        assert this_seg_clone.pos_marker, "TypeGuard"
-        new_table_ref = _create_table_ref(alias_name, dialect)
-        this_seg_clone.segments = (new_table_ref,)
-        anchor = subquery
-        # Grab the first keyword or symbol in the subquery to use as the
-        # anchor. This makes the lint warning less likely to be filtered out
-        # if a bit of the subquery happens to be templated.
-        for seg in subquery.recursive_crawl("keyword", "symbol"):
-            anchor = seg
-            break
-        res = LintResult(
-            anchor=anchor,
-            description=f"{parent_type} clauses should not contain "
-            "subqueries. Use CTEs instead",
-            fixes=[],
-        )
-        lint_results.append(res)
-
-    # Issue 3617: In T-SQL (and possibly other dialects) the automated fix
-    # leaves parentheses in a location that causes a syntax error. This is an
-    # unusual corner case. For simplicity, we still generate the lint warning
-    # but don't try to generate a fix. Someone could look at this later (a
-    # correct fix would involve removing the parentheses.)
-    bracketed_ctas = [seg.type for seg in parent_stack[-2:]] == [
-        "create_table_statement",
-        "bracketed",
-    ]
-    if bracketed_ctas or ctes.has_duplicate_aliases() or is_recursive:
-        # If we have duplicate CTE names just don't fix anything
-        # Return the lint warnings anyway
-        return lint_results
-
-    # Add fixes to the last result only
-    edit = [
-        ctes.compose_select(
-            clone_map[output_select[0]],
-            case_preference=case_preference,
-        ),
-    ]
-    lint_results[-1].fixes = [
-        LintFix.replace(
-            root_select[0],
-            edit_segments=edit,
-        )
-    ]
-    if is_new_name:
-        assert lint_results[0].fixes[0].edit
-        assert new_table_ref
-        # If we're creating a new CTE name but the CTE name does not appear in
-        # the fix, discard the lint error. This prevents the rule from looping,
-        # i.e. making the same fix repeatedly.
-        if not any(
-            seg.uuid == new_table_ref.uuid for seg in edit[0].recursive_crawl_all()
-        ):
-            lint_results[-1].fixes = []
-    return lint_results
+    def _nested_subqueries(
+        self, query: Query, dialect: Dialect
+    ) -> Iterator[_NestedSubQuerySummary]:
+        parent_types = self._config_mapping[self.forbid_subquery_in]
+        for q in [query] + list(query.ctes.values()):
+            for selectable in q.selectables:
+                if not selectable.select_info:
+                    continue  # pragma: no cover
+                select_source_names = set()
+                for a in selectable.select_info.table_aliases:
+                    # For each table in FROM, return table name and any alias.
+                    if a.ref_str:
+                        select_source_names.add(a.ref_str)
+                    if a.object_reference:
+                        select_source_names.add(a.object_reference.raw)
+                for table_alias in selectable.select_info.table_aliases:
+                    sc = SelectCrawler(table_alias.from_expression_element, dialect)
+                    if sc.query_tree:
+                        path_to = selectable.selectable.path_to(
+                            table_alias.from_expression_element
+                        )
+                        if not any(seg.is_type(*parent_types) for seg in path_to):
+                            continue
+                        if _is_correlated_subquery(
+                            Segments(sc.query_tree.selectables[0].selectable),
+                            select_source_names,
+                            dialect,
+                        ):
+                            continue
+                        yield _NestedSubQuerySummary(
+                            q, selectable, table_alias, sc, select_source_names
+                        )
+
+    def _lint_query(
+        self,
+        dialect: Dialect,
+        query: Query,
+        ctes: "_CTEBuilder",
+        case_preference,
+        clone_map,
+    ) -> Optional[Tuple[LintResult, BaseSegment, str, BaseSegment]]:
+        """Given the root query, compute lint warnings."""
+        nsq: _NestedSubQuerySummary
+        for nsq in self._nested_subqueries(query, dialect):
+            alias_name, is_new_name = ctes.create_cte_alias(nsq.table_alias)
+            anchor = nsq.table_alias.from_expression_element.segments[0]
+            new_cte = _create_cte_seg(
+                alias_name=alias_name,
+                subquery=clone_map[anchor],
+                case_preference=case_preference,
+                dialect=dialect,
+            )
+            ctes.insert_cte(new_cte)
+
+            # Grab the first keyword or symbol in the subquery to
+            # use as the anchor. This makes the lint warning less
+            # likely to be filtered out if a bit of the subquery
+            # happens to be templated.
+            anchor = next(anchor.recursive_crawl("keyword", "symbol"))
+            res = LintResult(
+                anchor=anchor,
+                description=f"{nsq.query.selectables[0].selectable.type} clauses "
+                "should not contain subqueries. Use CTEs instead",
+                fixes=[],
+            )
+            if len(nsq.query.selectables) == 1:
+                return (
+                    res,
+                    nsq.table_alias.from_expression_element,
+                    alias_name,
+                    nsq.query.selectables[0].selectable,
+                )
+        return None
 
 
 def _get_first_select_statement_descendant(
@@ -252,27 +289,6 @@ def _get_first_select_statement_descendant(
     return None  # pragma: no cover
 
 
-def _get_sources_from_select(segment: BaseSegment, dialect: Dialect) -> Set[str]:
-    """Given segment, return set of table or alias names it queries from."""
-    result = set()
-    select = None
-    if segment.is_type("select_statement"):
-        select = segment
-    elif segment.is_type("with_compound_statement"):
-        # For WITH statement, process the main query underneath.
-        select = _get_first_select_statement_descendant(segment)
-    if select and select.is_type("select_statement"):
-        select_info = get_select_statement_info(select, dialect)
-        if select_info:
-            for a in select_info.table_aliases:
-                # For each table in FROM, return table name and any alias.
-                if a.ref_str:
-                    result.add(a.ref_str)
-                if a.object_reference:
-                    result.add(a.object_reference.raw)
-    return result
-
-
 def _is_correlated_subquery(
     nested_select: Segments, select_source_names: Set[str], dialect: Dialect
 ):
@@ -280,8 +296,6 @@ def _is_correlated_subquery(
 
     https://en.wikipedia.org/wiki/Correlated_subquery
     """
-    if not nested_select:
-        return False  # pragma: no cover
     select_statement = _get_first_select_statement_descendant(nested_select[0])
     if not select_statement:
         return False  # pragma: no cover
@@ -298,51 +312,6 @@ def _is_correlated_subquery(
     return False
 
 
-def _find_nested_subqueries(
-    select: Segments,
-    dialect: Dialect,
-) -> Generator[_NestedSubQuerySummary, None, None]:
-    """Find possible offending elements and return enough to fix them."""
-    select_types = [
-        "with_compound_statement",
-        "set_expression",
-        "select_statement",
-    ]
-    from_clause = select.children().first(is_type("from_clause")).children()
-    offending_types = ["join_clause", "from_expression_element"]
-    select_source_names = _get_sources_from_select(select[0], dialect)
-
-    # Match any of the types we care about
-    for this_seg in from_clause.children(is_type(*offending_types)).iterate_segments():
-        parent_type = this_seg[0].get_type()
-        # Ensure we are at the right depth (from_expression_element)
-        if not this_seg.all(is_type("from_expression_element")):
-            this_seg = this_seg.children(
-                is_type("from_expression_element"),
-            )
-
-        table_expression_el = this_seg.children(
-            is_type("table_expression"),
-        )
-
-        # Is it bracketed? If so, lint that instead.
-        bracketed_expression = table_expression_el.children(
-            is_type("bracketed"),
-        )
-        nested_select = bracketed_expression or table_expression_el
-        # If we find a child with a "problem" type, raise an issue.
-        # If not, we're fine.
-        seg = nested_select.children(is_type(*select_types))
-        if not seg:
-            # If there is no match there is no error
-            continue
-        # Type, parent_select, parent_sequence
-        if not _is_correlated_subquery(nested_select, select_source_names, dialect):
-            yield _NestedSubQuerySummary(
-                parent_type, select, this_seg, table_expression_el[0]
-            )
-
-
 class _CTEBuilder:
     """Gather CTE parts, maintain order and track naming/aliasing."""
 
@@ -369,7 +338,9 @@ class _CTEBuilder:
     def insert_cte(self, cte: CTEDefinitionSegment):
         """Add a new CTE to the list as late as possible but before all its parents."""
         # This should still have the position markers of its true position
-        inbound_subquery = Segments(cte).children().last()
+        inbound_subquery = (
+            Segments(cte).children().last(lambda seg: bool(seg.pos_marker))
+        )
         insert_position = next(
             (
                 i
@@ -381,14 +352,11 @@ class _CTEBuilder:
 
         self.ctes.insert(insert_position, cte)
 
-    def create_cte_alias(
-        self, alias_segment: Optional[Segments] = None
-    ) -> Tuple[str, bool]:
+    def create_cte_alias(self, alias: Optional[AliasInfo]) -> Tuple[str, bool]:
         """Find or create the name for the next CTE."""
-        if alias_segment:
+        if alias and alias.aliased and alias.ref_str:
             # If we know the name use it
-            name = alias_segment.children().last()[0].raw
-            return name, False
+            return alias.ref_str, False
 
         self.name_idx = self.name_idx + 1
         name = f"prep_{self.name_idx}"
@@ -398,7 +366,7 @@ class _CTEBuilder:
         return name, True
 
     def get_cte_segments(self) -> List[BaseSegment]:
-        """Return a valid list of CTES with required padding Segements."""
+        """Return a valid list of CTES with required padding segments."""
         cte_segments: List[BaseSegment] = []
         for cte in self.ctes:
             cte_segments = cte_segments + [
@@ -439,16 +407,24 @@ class _CTEBuilder:
         )
         return new_select
 
+    def replace_with_clone(self, segment, clone_map):
+        for idx, cte in enumerate(self.ctes):
+            if any(segment is seg for seg in cte.recursive_crawl_all()):
+                self.ctes[idx] = clone_map[self.ctes[idx]]
+                return
+
 
 def _is_child(maybe_parent: Segments, maybe_child: Segments) -> bool:
     """Is the child actually between the start and end markers of the parent."""
-    assert len(maybe_child) == 1, "Cannot assess Childness of multiple Segments"
-    assert len(maybe_parent) == 1, "Cannot assess Childness of multiple Parents"
+    assert (
+        len(maybe_child) == 1
+    ), "Cannot assess child relationship of multiple segments"
+    assert (
+        len(maybe_parent) == 1
+    ), "Cannot assess child relationship of multiple parents"
     child_markers = maybe_child[0].pos_marker
     parent_pos = maybe_parent[0].pos_marker
-    if not parent_pos or not child_markers:
-        return False  # pragma: no cover
-
+    assert parent_pos and child_markers
     if child_markers < parent_pos.start_point_marker():
         return False  # pragma: no cover
 
diff --git a/src/sqlfluff/utils/analysis/select_crawler.py b/src/sqlfluff/utils/analysis/select_crawler.py
index f3f5ae4d7..5d3641fba 100644
--- a/src/sqlfluff/utils/analysis/select_crawler.py
+++ b/src/sqlfluff/utils/analysis/select_crawler.py
@@ -33,8 +33,13 @@ class Selectable:
     """A "SELECT" query segment."""
 
     selectable: BaseSegment
+    parent: Optional[BaseSegment]
     dialect: Dialect
 
+    def as_str(self) -> str:
+        """String representation for logging/testing."""
+        return self.selectable.raw
+
     @cached_property
     def select_info(self):
         """Returns SelectStatementColumnsAndTables on the SELECT."""
@@ -112,7 +117,7 @@ class Selectable:
         """Find corresponding table_aliases entry (if any) matching "table"."""
         alias_info = [
             t
-            for t in self.select_info.table_aliases
+            for t in (self.select_info.table_aliases if self.select_info else [])
             if t.aliased and t.ref_str == table
         ]
         assert len(alias_info) <= 1
@@ -131,8 +136,24 @@ class Query:
     parent: Optional["Query"] = field(default=None)
     # Children (could be CTE, subselect, or other).
     children: List["Query"] = field(default_factory=list)
+    cte_definition_segment: Optional[BaseSegment] = field(default=None)
     cte_name_segment: Optional[BaseSegment] = field(default=None)
 
+    def as_json(self) -> Dict:
+        """JSON representation for logging/testing."""
+        result = {}
+        if self.query_type != QueryType.Simple:
+            result["query_type"] = self.query_type.name
+        if self.selectables:
+            result["selectables"] = [
+                s.as_str() for s in self.selectables
+            ]  # type: ignore
+        if self.ctes:
+            result["ctes"] = {
+                k: v.as_json() for k, v in self.ctes.items()
+            }  # type: ignore
+        return result
+
     def lookup_cte(self, name: str, pop: bool = True) -> Optional["Query"]:
         """Look up a CTE by name, in the current or any parent scope."""
         cte = self.ctes.get(name.upper())
@@ -146,7 +167,7 @@ class Query:
             return None
 
     def crawl_sources(
-        self, segment: BaseSegment, recurse_into=True, pop=False
+        self, segment: BaseSegment, recurse_into=True, pop=False, lookup_cte=True
     ) -> Generator[Union[str, "Query"], None, None]:
         """Find SELECTs, table refs, or value table function calls in segment.
 
@@ -154,20 +175,26 @@ class Query:
         references or function call strings, yield those.
         """
         found_nested_select = False
-        for seg in segment.recursive_crawl(
+        types = [
             "table_reference",
             "set_expression",
             "select_statement",
             "values_clause",
-            recurse_into=recurse_into,
+        ]
+        for event, path in SelectCrawler.visit_segments(
+            segment, recurse_into=recurse_into
         ):
+            seg = path[-1]
+            if event == "end" or not seg.is_type(*types):
+                continue
+
             if seg is segment:
                 # If the starting segment itself matches the list of types we're
                 # searching for, recursive_crawl() will return it. Skip that.
                 continue
 
             if seg.is_type("table_reference"):
-                if not seg.is_qualified():
+                if not seg.is_qualified() and lookup_cte:
                     cte = self.lookup_cte(seg.raw, pop=pop)
                     if cte:
                         # It's a CTE.
@@ -179,7 +206,15 @@ class Query:
                     "set_expression", "select_statement", "values_clause"
                 )
                 found_nested_select = True
-                crawler = SelectCrawler(seg, self.dialect, parent=self)
+                seg_ = Segments(*path[1:]).first(
+                    sp.is_type(
+                        "from_expression_element",
+                        "set_expression",
+                        "select_statement",
+                        "values_clause",
+                    )
+                )[0]
+                crawler = SelectCrawler(seg_, self.dialect, parent=self)
                 # We know this will pass because we specified parent=self above.
                 assert crawler.query_tree
                 yield crawler.query_tree
@@ -234,9 +269,10 @@ class SelectCrawler:
             except ValueError:
                 pass
 
-        # Stores the last CTE name we saw, so we can associate it with the
-        # corresponding Query.
-        cte_name_segment: Optional[BaseSegment] = None
+        # Stacks for CTE definition & names we've seen but haven't consumed yet,
+        # so we can associate with the corresponding Query.
+        cte_definition_segment_stack: List[BaseSegment] = []
+        cte_name_segment_stack: List[BaseSegment] = []
 
         # Visit segment and all its children
         for event, path in SelectCrawler.visit_segments(segment):
@@ -263,9 +299,17 @@ class SelectCrawler:
                             # added to this Query later.
                             query = self.query_class(QueryType.Simple, dialect)
                             append_query(query)
-                        else:
+                        # Ignore segments under a from_expression_element.
+                        # Those will be nested queries, and we're only
+                        # interested in CTEs and "main" queries, i.e.
+                        # standalones or those following a block of CTEs.
+                        elif not any(
+                            seg.is_type("from_expression_element") for seg in path[1:]
+                        ):
                             # It's a select_statement or values_clause.
-                            selectable = Selectable(path[-1], dialect)
+                            selectable = Selectable(
+                                path[-1], path[-2] if len(path) >= 2 else None, dialect
+                            )
                             # Determine if this is part of a set_expression.
                             if len(path) >= 2 and path[-2].is_type("set_expression"):
                                 # It's part of a set_expression. Append to the
@@ -280,27 +324,37 @@ class SelectCrawler:
                                 append_query(query)
                     else:
                         # We're processing a "with" statement.
-                        if cte_name_segment:
+                        if cte_name_segment_stack:
                             # If we have a CTE name, this is the Query for that
                             # name.
                             query = self.query_class(
                                 QueryType.Simple,
                                 dialect,
-                                cte_name_segment=cte_name_segment,
+                                cte_definition_segment=cte_definition_segment_stack[-1],
+                                cte_name_segment=cte_name_segment_stack[-1],
                             )
                             if path[-1].is_type(
                                 "select_statement", "values_clause", "update_statement"
                             ):
                                 # Add to the Query object we just created.
-                                query.selectables.append(Selectable(path[-1], dialect))
+                                query.selectables.append(
+                                    Selectable(
+                                        path[-1],
+                                        path[-2] if len(path) >= 2 else None,
+                                        dialect,
+                                    )
+                                )
                             else:
                                 # Processing a set_expression. Nothing
                                 # additional to do here; we'll add selectables
                                 # to the Query later when we encounter those
                                 # child segments.
                                 pass
-                            query_stack[-1].ctes[cte_name_segment.raw_upper] = query
-                            cte_name_segment = None
+                            query_stack[-1].ctes[
+                                cte_name_segment_stack[-1].raw_upper
+                            ] = query
+                            cte_definition_segment_stack.pop()
+                            cte_name_segment_stack.pop()
                             append_query(query)
                         else:
                             # There's no CTE name, so we're probably processing
@@ -311,7 +365,8 @@ class SelectCrawler:
                             # interested in CTEs and "main" queries, i.e.
                             # standalones or those following a block of CTEs.
                             if not any(
-                                seg.is_type("from_expression_element") for seg in path
+                                seg.is_type("from_expression_element")
+                                for seg in path[1:]
                             ):
                                 if path[-1].is_type(
                                     "select_statement", "update_statement"
@@ -319,7 +374,11 @@ class SelectCrawler:
                                     # Processing a select_statement. Add it to the
                                     # Query object on top of the stack.
                                     query_stack[-1].selectables.append(
-                                        Selectable(path[-1], dialect)
+                                        Selectable(
+                                            path[-1],
+                                            path[-2] if len(path) >= 2 else None,
+                                            dialect,
+                                        )
                                     )
                                 else:
                                     # Processing a set_expression. Nothing
@@ -328,13 +387,19 @@ class SelectCrawler:
                 elif path[-1].is_type("with_compound_statement"):
                     # Beginning a "with" statement, i.e. a block of CTEs.
                     query = self.query_class(QueryType.WithCompound, dialect)
-                    if cte_name_segment:
-                        query_stack[-1].ctes[cte_name_segment.raw_upper] = query
-                        cte_name_segment = None
+                    if cte_name_segment_stack:
+                        query_stack[-1].ctes[
+                            cte_name_segment_stack[-1].raw_upper
+                        ] = query
+                        query.cte_definition_segment = cte_definition_segment_stack[-1]
+                        cte_definition_segment_stack.pop()
+                        cte_name_segment_stack.pop()
                     append_query(query)
                 elif path[-1].is_type("common_table_expression"):
-                    # This is a "<<cte name>> AS". Grab the name for later.
-                    cte_name_segment = path[-1].segments[0]
+                    # This is a "<<cte name>> AS". Save definition segment and
+                    # name for later.
+                    cte_definition_segment_stack.append(path[-1])
+                    cte_name_segment_stack.append(path[-1].segments[0])
             elif event == "end":
                 finish_segment()
 
@@ -355,13 +420,14 @@ class SelectCrawler:
         return list(query.crawl_sources(segment, True))
 
     @classmethod
-    def visit_segments(cls, seg, path=None):
+    def visit_segments(cls, seg, path=None, recurse_into=True):
         """Recursively visit all segments."""
         if path is None:
             path = []
         path.append(seg)
         yield "start", path
-        for seg in seg.segments:
-            yield from cls.visit_segments(seg, path)
+        if recurse_into:
+            for seg in seg.segments:
+                yield from cls.visit_segments(seg, path, recurse_into)
         yield "end", path
         path.pop()


Test patch:
diff --git a/test/fixtures/rules/std_rule_cases/L042.yml b/test/fixtures/rules/std_rule_cases/L042.yml
index 643bbe0f1..642798380 100644
--- a/test/fixtures/rules/std_rule_cases/L042.yml
+++ b/test/fixtures/rules/std_rule_cases/L042.yml
@@ -99,7 +99,7 @@ double_nested_fail:
       L042:
         forbid_subquery_in: both
 
-double_nested_unfixable_cte_clash:
+double_nested_fail_2:
   fail_str: |
     select
         a.x, a.y, b.z
@@ -109,6 +109,20 @@ double_nested_unfixable_cte_clash:
         select x, z from p_cte
       ) as b
     ) as b on (a.x = b.x)
+  fix_str: |
+    with b as (
+      select x, z from (
+        select x, z from p_cte
+      ) as b
+    )
+    select
+        a.x, a.y, b.z
+    from a
+    join b on (a.x = b.x)
+  violations_after_fix:
+    - description: select_statement clauses should not contain subqueries. Use CTEs instead
+      line_no: 2
+      line_pos: 20
   configs:
     rules:
       L042:
@@ -127,6 +141,23 @@ unfixable_cte_clash:
         select 1
       ) as b
     ) as c on (a.x = b.x)
+  fix_str: |
+    with "b" as (
+      select x, z from p_cte
+    ),
+    c as (
+      select x, z from (
+        select 1
+      ) as b
+    )
+    select
+        a.x, a.y, b.z
+    from a
+    join c on (a.x = b.x)
+  violations_after_fix:
+    - description: select_statement clauses should not contain subqueries. Use CTEs instead
+      line_no: 5
+      line_pos: 20
   configs:
     rules:
       L042:
@@ -458,10 +489,16 @@ issue_3572_correlated_subquery_3:
 issue_3598_avoid_looping_1:
   fail_str: |
     WITH cte1 AS (
-    	SELECT a
-    	FROM (SELECT a)
+        SELECT a
+        FROM (SELECT a)
+    )
+    SELECT a FROM cte1
+  fix_str: |
+    WITH prep_1 AS (SELECT a),
+    cte1 AS (
+        SELECT a
+        FROM prep_1
     )
-
     SELECT a FROM cte1
   configs:
     rules:
@@ -474,8 +511,37 @@ issue_3598_avoid_looping_2:
         SELECT *
         FROM (SELECT * FROM mongo.temp)
     )
-
     SELECT * FROM cte1
+  fix_str: |
+    WITH prep_1 AS (SELECT * FROM mongo.temp),
+    cte1 AS (
+        SELECT *
+        FROM prep_1
+    )
+    SELECT * FROM cte1
+  configs:
+    rules:
+      L042:
+        forbid_subquery_in: both
+
+test_fail_subquery_in_cte:
+  fail_str: |
+    with b as (
+      select x, z from (
+        select x, z from p_cte
+      )
+    )
+    select b.z
+    from b
+  fix_str: |
+    with prep_1 as (
+        select x, z from p_cte
+      ),
+    b as (
+      select x, z from prep_1
+    )
+    select b.z
+    from b
   configs:
     rules:
       L042:
diff --git a/test/utils/analysis/test_select_crawler.py b/test/utils/analysis/test_select_crawler.py
new file mode 100644
index 000000000..c87270dad
--- /dev/null
+++ b/test/utils/analysis/test_select_crawler.py
@@ -0,0 +1,197 @@
+"""Test the select_crawler module."""
+import pytest
+
+from sqlfluff.core.linter.linter import Linter
+from sqlfluff.utils.analysis import select_crawler
+
+
+@pytest.mark.parametrize(
+    "sql, expected_json",
+    [
+        (
+            # Test trivial query.
+            "select 1",
+            {"selectables": ["select 1"]},
+        ),
+        (
+            # Test set expression.
+            "select 1 union select 2",
+            {"selectables": ["select 1", "select 2"]},
+        ),
+        (
+            # Test multiple CTEs.
+            "with cte1 as (select 1 as x), cte2 as (select 2 as y) "
+            "select * from cte1 join cte2 using (x)",
+            {
+                "ctes": {
+                    "CTE1": {"selectables": ["select 1 as x"]},
+                    "CTE2": {"selectables": ["select 2 as y"]},
+                },
+                "query_type": "WithCompound",
+                "selectables": ["select * from cte1 join cte2 using (x)"],
+            },
+        ),
+        (
+            # Nested CTEs (from L044 test suite)
+            """
+        with a as (
+            with b as (select 1 from c)
+            select * from b
+        )
+        select * from a
+        """,
+            {
+                "ctes": {
+                    "A": {
+                        "ctes": {"B": {"selectables": ["select 1 from c"]}},
+                        "query_type": "WithCompound",
+                        "selectables": ["select * from b"],
+                    }
+                },
+                "query_type": "WithCompound",
+                "selectables": ["select * from a"],
+            },
+        ),
+        (
+            # Nested CTEs (from L044 test suite)
+            """
+        with b as (select 1 from c)
+        select * from (
+            with a as (select * from b)
+            select * from a
+        )
+        """,
+            {
+                "ctes": {"B": {"selectables": ["select 1 from c"]}},
+                "query_type": "WithCompound",
+                "selectables": [
+                    "select * from (\n"
+                    "            with a as (select * from b)\n"
+                    "            select * from a\n"
+                    "        )"
+                ],
+            },
+        ),
+        (
+            # Test that subquery in "from" not included.
+            "select a.x from (select z from b)",
+            {"selectables": ["select a.x from (select z from b)"]},
+        ),
+        (
+            # Test that subquery in "from" / "join" not included.
+            "select a.x from a join (select z from b) as b on (a.x = b.x)",
+            {
+                "selectables": [
+                    "select a.x from a join (select z from b) as b on (a.x = b.x)"
+                ]
+            },
+        ),
+        (
+            # In CTE main query, test that subquery in "from" not included.
+            "with prep as (select 1) select a.x from (select z from b)",
+            {
+                "ctes": {"PREP": {"selectables": ["select 1"]}},
+                "query_type": "WithCompound",
+                "selectables": ["select a.x from (select z from b)"],
+            },
+        ),
+        (
+            # In CTE main query, test that subquery in "from" / "join" not included.
+            "with prep as (select 1) "
+            "select a.x from a join (select z from b) as b on (a.x = b.x)",
+            {
+                "ctes": {"PREP": {"selectables": ["select 1"]}},
+                "query_type": "WithCompound",
+                "selectables": [
+                    "select a.x from a join (select z from b) as b on (a.x = " "b.x)"
+                ],
+            },
+        ),
+        (
+            """with prep_1 as (
+    with d as (
+        select x, z from b
+    )
+    select * from d
+)
+select
+    a.x, a.y, b.z
+from a
+join prep_1 using (x)
+""",
+            {
+                "ctes": {
+                    "PREP_1": {
+                        "ctes": {
+                            "D": {"selectables": ["select x, z from b"]},
+                        },
+                        "query_type": "WithCompound",
+                        "selectables": ["select * from d"],
+                    }
+                },
+                "query_type": "WithCompound",
+                "selectables": [
+                    "select\n    a.x, a.y, b.z\nfrom a\njoin prep_1 using (x)"
+                ],
+            },
+        ),
+    ],
+)
+def test_select_crawler_constructor(sql, expected_json):
+    """Test SelectCrawler when created using constructor."""
+    linter = Linter(dialect="ansi")
+    parsed = linter.parse_string(sql)
+    segments = list(
+        parsed.tree.recursive_crawl(
+            "with_compound_statement",
+            "set_expression",
+            "select_statement",
+        )
+    )
+    segment = segments[0]
+    crawler = select_crawler.SelectCrawler(segment, linter.dialect)
+    assert all(
+        cte.cte_definition_segment is not None
+        for cte in crawler.query_tree.ctes.values()
+    )
+    json_query_tree = crawler.query_tree.as_json()
+    assert expected_json == json_query_tree
+
+
+def test_select_crawler_nested():
+    """Test invoking with an outer from_expression_segment."""
+    sql = """
+select
+    a.x, a.y, b.z
+from a
+join (
+    with d as (
+        select x, z from b
+    )
+    select * from d
+) using (x)
+    """
+    linter = Linter(dialect="ansi")
+    parsed = linter.parse_string(sql)
+    segments = list(
+        parsed.tree.recursive_crawl(
+            "with_compound_statement",
+            "set_expression",
+            "select_statement",
+        )
+    )
+    segment = segments[0]
+    crawler = select_crawler.SelectCrawler(segment, linter.dialect)
+    sc = select_crawler.SelectCrawler(
+        crawler.query_tree.selectables[0]
+        .select_info.table_aliases[1]
+        .from_expression_element,
+        linter.dialect,
+    )
+    assert sc.query_tree.as_json() == {
+        "selectables": [
+            "select * from d",
+        ],
+        "ctes": {"D": {"selectables": ["select x, z from b"]}},
+        "query_type": "WithCompound",
+    }