Make extract_toplevel_blocks() Faster

.changes/unreleased/Under the Hood-20231109-131545.yaml

@@ -0,0 +1,6 @@
+kind: Under the Hood
+body: Make extract_toplevel_blocks() Faster
+time: 2023-11-09T13:15:45.338059-05:00
+custom:
+  Author: peterallenwebb qmalcolm
+  Issue: "9037"

core/dbt/clients/_jinja_blocks.py

@@ -1,5 +1,7 @@
+import dataclasses
 import re
 from collections import namedtuple
+from typing import Dict, Optional
 
 from dbt.exceptions import (
     BlockDefinitionNotAtTopError,
@@ -97,13 +99,26 @@ def end_pat(self):
 QUOTE_START_PATTERN = regex(r"""(?P<quote>(['"]))""")
 
 
+@dataclasses.dataclass
+class PositionedMatch:
+    """Used to accelerate TagIterator. Records the result of searching a string, starting
+    at start_pos and finding match (or None)."""
+
+    start_pos: int
+    match: Optional[re.Match]
+
+
 class TagIterator:
     def __init__(self, data):
         self.data = data
-        self.blocks = []
-        self._parenthesis_stack = []
+        self.blocks = []  # unused?
+        self._parenthesis_stack = []  # unused?
         self.pos = 0
 
+        # Performance enhancement: A cache of the most recent matches seen for each pattern.
+        # Includes the start position used for the search.
+        self._past_matches: Dict[re.Pattern, PositionedMatch] = {}
+
     def linepos(self, end=None) -> str:
         """Given an absolute position in the input data, return a pair of
         line number + relative position to the start of the line.
@@ -122,8 +137,33 @@ def advance(self, new_position):
     def rewind(self, amount=1):
         self.pos -= amount
 
-    def _search(self, pattern):
-        return pattern.search(self.data, self.pos)
+    def _search(self, pattern) -> Optional[re.Match]:
+
+        # Check to see if we have a cached search on this pattern.
+        positioned_match = self._past_matches.get(pattern)
+
+        if positioned_match is None or positioned_match.start_pos > self.pos:
+            # We did not have a cached search, or we did, but it was done at a location
+            # further along in the string. Do a new search and cache it.
+            match = pattern.search(self.data, self.pos)
+            self._past_matches[pattern] = PositionedMatch(self.pos, match)
+        else:
+            # We have a cached search and its start position falls before (or at) the
+            # current search position...
+            if positioned_match.match is None:
+                # ...but there is no match in the rest of the 'data'.
+                match = None
+            elif positioned_match.match.start() >= self.pos:
+                # ...and there is a match we can reuse, because we have not yet passed
+                # the start position of the match. It's still the next match.
+                match = positioned_match.match
+            else:
+                # ...but we have passed the start of the cached match, and need to do a
+                # new search from our current position and cache it.
+                match = pattern.search(self.data, self.pos)
+                self._past_matches[pattern] = PositionedMatch(self.pos, match)
+
+        return match
 
     def _match(self, pattern):
         return pattern.match(self.data, self.pos)