Repository: dbt-labs/dbt-core
Pull request: clean up block parsing to make it dumber and more effective (#1547)
Fixes #1547 

The block parser is now two-tiered: one part extracts tags `{% x %}` and is pretty purely linear, and one part tries to make sense out of begin/end sequencing and maintains a stack. I think this version is much more maintainable and readable, and I feel way more confident releasing this. The diff is pretty big, but I got to delete a lot of code, and "un-teach" dbt a lot of things about jinja.

I think it's pretty unlikely that this is even remotely close to perfect, but it is going to be so much easier to fix the bugs we find!

Problem statement:
Tell the block parser to ignore tags/nesting by default
## Feature
Replace recursive block parsing with extracting a flat list of tags and some metadata (position, block type/name)

### Feature description
The block parser currently has a list of special tags that don't/might not have a matching `{% endtag %}`, and recursively descends into blocks and parsers them. Instead, the block parser should just return a list of tags with positions and names. This is all fine as long as any tags we don't support are of the form `{% x %}` or `{% x y %}` and come with a matching `{% endx %}`. In hindsight, that's kind of risky and silly. Instead, we should special case the blocks we care about, match them up with their end-statements up later, and extract the data that way.

At that point we can do things like try to detect macros defined inside if statements, or really anything else, without having the structure of the AST impact the low-level block extraction work. As it is I kind of anticipate weird block extraction bugs where we're looking for tags that don't exist.

The raw tag has to remain special, like comments/expr blocks (which should be unchanged!).

### Who will this benefit?
Me: I will be less uncomfortable with block extraction
Future developers (probably me...): tag extraction will be its own phase, so it'll be easier to handle that vs actual block extraction when fixing bugs/adding features - right now they're pretty intertwined
Users: Users will be able to use tags we haven't added support for, whatever they may be

Hints from issue discussion:

Commit message: clean up block parsing to make it dumber and more effective

Patch to review:
diff --git a/core/dbt/clients/_jinja_blocks.py b/core/dbt/clients/_jinja_blocks.py
index a0e567e0ca2..766b451a57d 100644
--- a/core/dbt/clients/_jinja_blocks.py
+++ b/core/dbt/clients/_jinja_blocks.py
@@ -1,4 +1,5 @@
 import re
+from collections import namedtuple
 
 import dbt.exceptions
 
@@ -45,6 +46,9 @@ def end_pat(self):
         return regex(pattern)
 
 
+Tag = namedtuple('Tag', 'block_type_name block_name start end')
+
+
 _NAME_PATTERN = r'[A-Za-z_][A-Za-z_0-9]*'
 
 COMMENT_START_PATTERN = regex(r'(?:(?P<comment_start>(\s*\{\#)))')
@@ -63,31 +67,14 @@ def end_pat(self):
 )))
 
 
-TAG_CLOSE_PATTERN = regex(r'(?:(?P<tag_close>(\-\%\}\s*|\%\})))')
-# if you do {% materialization foo, adapter="myadapter' %} and end up with
-# mismatched quotes this will still match, but jinja will fail somewhere
-# since the adapter= argument has to be an adapter name, and none have quotes
-# or anything else in them. So this should be fine.
-MATERIALIZATION_ARGS_PATTERN = regex(
-    r'\s*,\s*'
-    r'''(?P<adpater_arg>(adapter=(?:['"]{}['"])|default))'''
-    .format(_NAME_PATTERN)
-)
-# macros an stuff like macros get open parents, followed by a very complicated
-# argument spec! In fact, it's easiest to parse it in tiny little chunks
-# because we have to handle awful stuff like string parsing ;_;
-MACRO_ARGS_START_PATTERN = regex(r'\s*(?P<macro_start>\()\s*')
-MACRO_ARGS_END_PATTERN = regex(r'\s*(?P<macro_end>(\)))\s*')
-
-# macros can be like {% macro foo(bar) %} or {% macro foo(bar, baz) %} or
-# {% macro foo(bar, baz="quux") %} or ...
-# I think jinja disallows default values after required (like Python), but we
-# can ignore that and let jinja deal
-MACRO_ARG_PATTERN = regex(''.join((
-    r'\s*(?P<macro_arg_name>({}))\s*',
-    r'((?P<value>=)|(?P<more_args>,)?)\s*'.format(_NAME_PATTERN),
+RAW_BLOCK_PATTERN = regex(''.join((
+    r'(?:\s*\{\%\-|\{\%)\s*raw\s*(?:\-\%\}\s*|\%\})',
+    r'(?:.*)',
+    r'(?:\s*\{\%\-|\{\%)\s*endraw\s*(?:\-\%\}\s*|\%\})',
 )))
 
+TAG_CLOSE_PATTERN = regex(r'(?:(?P<tag_close>(\-\%\}\s*|\%\})))')
+
 # stolen from jinja's lexer. Note that we've consumed all prefix whitespace by
 # the time we want to use this.
 STRING_PATTERN = regex(
@@ -97,26 +84,8 @@ def end_pat(self):
 
 QUOTE_START_PATTERN = regex(r'''(?P<quote>(['"]))''')
 
-# any number of non-quote characters, followed by:
-# - quote: a quote mark indicating start of a string (you'll want to backtrack
-#          the regex end on quotes and then match with the string pattern)
-# - a comma (so there will be another full argument)
-# - a closing parenthesis (you can now expect a closing tag)
-NON_STRING_MACRO_ARGS_PATTERN = regex(
-    # anything, followed by a quote, open/close paren, or comma
-    r'''(.*?)'''
-    r'''((?P<quote>(['"]))|(?P<open>(\())|(?P<close>(\)))|(?P<comma>(\,)))'''
-)
-
-
-NON_STRING_DO_BLOCK_MEMBER_PATTERN = regex(
-    # anything, followed by a quote, paren, or a tag end
-    r'''(.*?)'''
-    r'''((?P<quote>(['"]))|(?P<open>(\())|(?P<close>(\))))'''
-)
-
 
-class BlockIterator(object):
+class TagIterator(object):
     def __init__(self, data):
         self.data = data
         self.blocks = []
@@ -135,17 +104,6 @@ def _search(self, pattern):
     def _match(self, pattern):
         return pattern.match(self.data, self.pos)
 
-    def handle_raw(self, match):
-        start = match.end()
-        end_pat = BlockTag('raw', None).end_pat()
-        end_match = self._search(end_pat)
-        if end_match is None:
-            dbt.exceptions.raise_compiler_error(
-                'unexpected EOF, expected {% endraw %}'
-            )
-        self.advance(end_match.end())
-        return BlockData(self.data[start:end_match.start()])
-
     def _first_match(self, *patterns, **kwargs):
         matches = []
         for pattern in patterns:
@@ -199,216 +157,87 @@ def handle_expr(self, match):
 
         self.advance(match.end())
 
-    def _advance_to_tag_close(self):
-        endtag = self._expect_match('%}', TAG_CLOSE_PATTERN)
-        self.advance(endtag.end())
-
-    def _consume_to_tag(self, tag_name, start):
-        def block_handler(block_match):
-            block_type_name = block_match.groupdict()['block_type_name']
-            if block_type_name == tag_name:
-                self.advance(block_match.end())
-                self._advance_to_tag_close()
-                return self.data[start:block_match.start()]
-            else:
-                # must be a new block
-                self.handle_block(block_match)
-
-        return self.next_block(block_handler)
-
-    def _consume_block_contents(self, match):
-        """Find a block's end. The current state of the parser should be after
-        the open block is completed:
+    def handle_comment(self, match):
+        self.advance(match.end())
+        match = self._expect_match('#}', COMMENT_END_PATTERN)
+        self.advance(match.end())
 
-            {% blk foo %}my data {% endblk %}
-                         ^ right here
+    def _expect_block_close(self):
+        """Search for the tag close marker.
+        To the right of the type name, there are a few possiblities:
+           - a name (handled by the regex's 'block_name')
+           - any number of: `=`, `(`, `)`, strings, etc (arguments)
+           - nothing
 
-        This will recurse!
+        followed eventually by a %}
 
-        :param re.Match match: The regex match for the block tag to look for
-            the end of.
-        :return re.Match: An end pattern match for the block. Should be an
-            re.Match correspondding to
-            `BlockTag(**match.groupdict()).end_pat()`
+        So the only characters we actually have to worry about in this context
+        are quote and `%}` - nothing else can hide the %} and be valid jinja.
         """
-        block = BlockTag(**match.groupdict())
-
-        contents = self._consume_to_tag(block.end_block_type_name, self.pos)
-        if contents is None:
-            dbt.exceptions.raise_compiler_error(
-                'Never found a close tag for the open tag {} ("{}")'
-                .format(block.block_type_name, block.end_block_type_name)
+        while True:
+            end_match = self._expect_match(
+                'tag close ("%}")',
+                QUOTE_START_PATTERN,
+                TAG_CLOSE_PATTERN
             )
-        block.contents = contents
-        return block
-
-    def handle_top_block(self, match):
-        """Handle a block. The current state of the parser should be after the
-        block open tag is completed:
-            {% blk foo %}my data {% endblk %}
-                         ^ right here
-        """
-        # we have to handle comments inside blocks because you could do this:
-        # {% blk foo %}asdf {# {% endblk %} #} {%endblk%}
-        # they still end up in the data/raw_data of the block itself, but we
-        # have to know to ignore stuff until the end comment marker!
-
-        # the full block started at the given match start, which may include
-        # prefixed whitespace! we'll strip it later
-        block_start = match.start()
-        found = self.handle_block(match)
-        found.full_block = self.data[block_start:self.pos]
-        return found
-
-    def handle_materialization_block(self, match):
-        """Handle a materialization block. The current status of the parser
-        should be:
-            {% materialization custom, default %}
-            ^ right here
-
-        And the match should be for "{% materialization custom"
-        """
-        self.advance(match.end())
-        self._expect_match('materialization args',
-                           MATERIALIZATION_ARGS_PATTERN)
-        self._advance_to_tag_close()
-        # Now we're just looking for {% endmaterialization %}
-        return self._consume_block_contents(match)
-
-    def handle_do_block(self, match, expect_block):
-        if expect_block:
-            # we might be wrong to expect a block ({% do (...) %}, for example)
-            # so see if there's more data before the tag closes. if there
-            # isn't, we expect a block.
-            close_match = self._expect_match('%}', TAG_CLOSE_PATTERN)
-            unprocessed = self.data[match.end():close_match.start()].strip()
-            expect_block = not unprocessed
-
-        if expect_block:
-            # if we're here, expect_block is True and we must have set
-            # close_match
-            self.advance(close_match.end())
-            block = self._consume_block_contents(match)
-        else:
-            # we have a do-statement like {% do thing() %}, so no {% enddo %}
-            # also, we don't want to advance to the end of the match, as it
-            # might be inside a string or something! So go back and figure out
-            # this should get us past "{% do"
-            self.advance(match.end())
-            self._process_rval_components()
-            block = BlockTag('do', None,
-                             full_block=self.data[match.start():self.pos])
-        return block
-
-    def handle_set_block(self, match):
-        self.advance(match.end())
-        equal_or_close = self._expect_match('%} or =',
-                                            TAG_CLOSE_PATTERN, regex(r'='))
-        self.advance(equal_or_close.end())
-        if equal_or_close.groupdict().get('tag_close') is None:
-            # it's an equals sign, must be like {% set x = 1 %}
-            self._process_rval_components()
-            # watch out, order matters here on python 2
-            block = BlockTag(full_block=self.data[match.start():self.pos],
-                             **match.groupdict())
-        else:
-            # it's a tag close, must be like {% set x %}...{% endset %}
-            block = self._consume_block_contents(match)
-        return block
-
-    def handle_if_block(self, match):
+            self.advance(end_match.end())
+            if end_match.groupdict().get('tag_close') is not None:
+                return
+            # must be a string. Rewind to its start and advance past it.
+            self.rewind()
+            string_match = self._expect_match('string', STRING_PATTERN)
+            self.advance(string_match.end())
+
+    def handle_raw(self):
+        # raw blocks are super special, they are a single complete regex
+        match = self._expect_match('{% raw %}...{% endraw %}',
+                                   RAW_BLOCK_PATTERN)
         self.advance(match.end())
-        self._process_rval_components()
-        return self._consume_block_contents(match)
+        return match.end()
 
-    def handle_elif_block(self, match):
-        self.advance(match.end())
-        self._process_rval_components()
+    def handle_tag(self, match):
+        """The tag could be one of a few things:
 
-    def handle_else_block(self, match):
-        self.advance(match.end())
-        self._advance_to_tag_close()
+            {% mytag %}
+            {% mytag x = y %}
+            {% mytag x = "y" %}
+            {% mytag x.y() %}
+            {% mytag foo("a", "b", c="d") %}
 
-    def handle_comment(self, match):
-        self.advance(match.end())
-        match = self._expect_match('#}', COMMENT_END_PATTERN)
-        self.advance(match.end())
-
-    def handle_normal_block(self, match):
-        # once we've advanced to the end of the given match, we're somewhere
-        # like this: {% block_type_name block_type
-        # we've either got arguments, a close of tag (%}), or bad input.
-        # we've handled materializations already (they're weird!)
-        # thankfully, comments aren't allowed *inside* a block def...
-        self.advance(match.end())
-        block_end_match = self._expect_match('%} or (...)',
-                                             TAG_CLOSE_PATTERN,
-                                             MACRO_ARGS_START_PATTERN)
-        self.advance(block_end_match.end())
-        if block_end_match.groupdict().get('macro_start') is not None:
-            # we've hit our first parenthesis!
-            self._parenthesis_stack = [True]
-            self._process_macro_args()
-            self._advance_to_tag_close()
-
-        # tag close time!
-        return self._consume_block_contents(match)
-
-    def handle_block(self, match):
-        matchgroups = match.groupdict()
-        if 'block_type_name' not in matchgroups:
-            raise dbt.exceptions.InternalException(
-                'Got bad matchgroups {} in handle_block'.format(matchgroups)
-            )
-        block_type_name = matchgroups['block_type_name']
+        But the key here is that it's always going to be `{% mytag`!
+        """
+        groups = match.groupdict()
+        # always a value
+        block_type_name = groups['block_type_name']
+        # might be None
+        block_name = groups.get('block_name')
+        start_pos = self.pos
         if block_type_name == 'raw':
-            block = self.handle_raw(match)
-        elif block_type_name == 'materialization':
-            block = self.handle_materialization_block(match)
-        elif block_type_name == 'do':
-            # if there is a "block_name" in the match groups, we don't expect a
-            # block as the "block name" is actually part of the do-statement.
-            # we need to do this to handle the (weird and probably wrong!) case
-            # of a do-statement that is only a single identifier - techincally
-            # allowed in jinja. (for example, {% do thing %})
-            expect_block = matchgroups.get('block_name') is None
-            block = self.handle_do_block(match, expect_block=expect_block)
-        elif block_type_name == 'set':
-            block = self.handle_set_block(match)
-        elif block_type_name == 'if':
-            block = self.handle_if_block(match)
-        elif block_type_name == 'elif':
-            block = self.handle_elif_block(match)
-        elif block_type_name == 'else':
-            block = self.handle_else_block(match)
+            match = self._expect_match('{% raw %}...{% endraw %}',
+                                       RAW_BLOCK_PATTERN)
+            self.advance(match.end())
         else:
-            block = self.handle_normal_block(match)
-        return block
-
-    def next_block(self, block_handler, include_toplevel=False):
-        """Return the next block, advancing the position as appropriate.
-
-        If include_toplevel is True, next_block will return a BlockData
-        containing any extra pre-match space
-
-        """
+            self.advance(match.end())
+            self._expect_block_close()
+        return Tag(
+            block_type_name=block_type_name,
+            block_name=block_name,
+            start=start_pos,
+            end=self.pos
+        )
+
+    def find_tags(self):
         while True:
             match = self._first_match(
                 BLOCK_START_PATTERN,
                 COMMENT_START_PATTERN,
-                EXPR_START_PATTERN)
-
+                EXPR_START_PATTERN
+            )
             if match is None:
-                # raise StopIteration?
-                return None
-
-            raw_toplevel = self.data[self.pos:match.start()]
-            if len(raw_toplevel) > 0:
-                self.advance(match.start())
-                if include_toplevel:
-                    return BlockData(raw_toplevel)
+                break
 
-            start = self.pos
+            self.advance(match.start())
+            # start = self.pos
 
             groups = match.groupdict()
             comment_start = groups.get('comment_start')
@@ -417,156 +246,122 @@ def next_block(self, block_handler, include_toplevel=False):
 
             if comment_start is not None:
                 self.handle_comment(match)
-                if include_toplevel:
-                    return BlockData(self.data[start:self.pos])
             elif expr_start is not None:
                 self.handle_expr(match)
-                if include_toplevel:
-                    return BlockData(self.data[start:self.pos])
             elif block_type_name is not None:
-                block = block_handler(match)
-                if block is not None:
-                    return block
-                elif include_toplevel:
-                    return BlockData(self.data[start:self.pos])
+                yield self.handle_tag(match)
             else:
                 raise dbt.exceptions.InternalException(
                     'Invalid regex match in next_block, expected block start, '
                     'expr start, or comment start'
                 )
 
-    def find_blocks(self):
-        def block_handler(match):
-            block_start = match.start()
-            block = self.handle_block(match)
-            block.full_block = self.data[block_start:self.pos]
-            return block
+    def __iter__(self):
+        return self.find_tags()
 
-        while self.data[self.pos:]:
-            block = self.next_block(block_handler, include_toplevel=True)
-            if block is None:
-                break
-            yield block
 
-        raw_toplevel = self.data[self.pos:]
-        if len(raw_toplevel) > 0:
-            yield BlockData(raw_toplevel)
+duplicate_tags = (
+    'Got nested tags: {outer.block_type_name} (started at {outer.start}) did '
+    'not have a matching {{% end{outer.block_type_name} %}} before a '
+    'subsequent {inner.block_type_name} was found (started at {inner.start})'
+)
 
-    def _process_rval_components(self):
-        """This is suspiciously similar to _process_macro_default_arg, probably
-        want to figure out how to merge the two.
 
-        Process the rval of an assignment statement or a do-block
-        """
-        while True:
-            match = self._expect_match(
-                'do block component',
-                # you could have a string, though that would be weird
-                STRING_PATTERN,
-                # a quote or an open/close parenthesis
-                NON_STRING_DO_BLOCK_MEMBER_PATTERN,
-                # a tag close
-                TAG_CLOSE_PATTERN
-            )
-            matchgroups = match.groupdict()
-            self.advance(match.end())
-            if matchgroups.get('string') is not None:
-                continue
-            elif matchgroups.get('quote') is not None:
-                self.rewind()
-                # now look for a string
-                match = self._expect_match('any string', STRING_PATTERN)
-                self.advance(match.end())
-            elif matchgroups.get('open'):
-                self._parenthesis_stack.append(True)
-            elif matchgroups.get('close'):
-                self._parenthesis_stack.pop()
-            elif matchgroups.get('tag_close'):
-                if self._parenthesis_stack:
-                    msg = ('Found "%}", expected ")"')
-                    dbt.exceptions.raise_compiler_error(msg)
-                return
-            # else whitespace
+_CONTROL_FLOW_TAGS = {
+    'if': 'endif',
+    'for': 'endfor',
+}
 
-    def _process_macro_default_arg(self):
-        """Handle the bit after an '=' in a macro default argument. This is
-        probably the trickiest thing. The goal here is to accept all strings
-        jinja would accept and always handle block start/end correctly: It's
-        fine to have false positives, jinja can fail later.
+_CONTROL_FLOW_END_TAGS = {
+    v: k
+    for k, v in _CONTROL_FLOW_TAGS.items()
+}
 
-        Return True if there are more arguments expected.
-        """
-        while self._parenthesis_stack:
-            match = self._expect_match(
-                'macro argument',
-                # you could have a string
-                STRING_PATTERN,
-                # a quote, a comma, or a open/close parenthesis
-                NON_STRING_MACRO_ARGS_PATTERN,
-                # we want to "match", not "search"
-                method='match'
-            )
-            matchgroups = match.groupdict()
-            self.advance(match.end())
-            if matchgroups.get('string') is not None:
-                # we got a string value. There could be more data.
-                continue
-            elif matchgroups.get('quote') is not None:
-                # we got a bunch of data and then a string opening value.
-                # put the quote back on the menu
-                self.rewind()
-                # now look for a string
-                match = self._expect_match('any string', STRING_PATTERN)
-                self.advance(match.end())
-            elif matchgroups.get('comma') is not None:
-                # small hack: if we hit a comma and there is one parenthesis
-                # left, return to look for a new name. otherwise we're still
-                # looking for the parameter close.
-                if len(self._parenthesis_stack) == 1:
-                    return
-            elif matchgroups.get('open'):
-                self._parenthesis_stack.append(True)
-            elif matchgroups.get('close'):
-                self._parenthesis_stack.pop()
-            else:
-                raise dbt.exceptions.InternalException(
-                    'unhandled regex in _process_macro_default_arg(), no match'
-                    ': {}'.format(matchgroups)
-                )
 
-    def _process_macro_args(self):
-        """Macro args are pretty tricky! Arg names themselves are simple, but
-        you can set arbitrary default values, including doing stuff like:
-        {% macro my_macro(arg="x" + ("}% {# {% endmacro %}" * 2)) %}
+class BlockIterator(object):
+    def __init__(self, data):
+        self.tag_parser = TagIterator(data)
+        self.current = None
+        self.stack = []
+        self.last_position = 0
+
+    @property
+    def current_end(self):
+        if self.current is None:
+            return 0
+        else:
+            return self.current.end
 
-        Which makes you a jerk, but is valid jinja.
-        """
-        # we are currently after the first parenthesis (+ any whitespace) after
-        # the macro args started. You can either have the close paren, or a
-        # name.
-        while self._parenthesis_stack:
-            match = self._expect_match('macro arguments',
-                                       MACRO_ARGS_END_PATTERN,
-                                       MACRO_ARG_PATTERN)
-            self.advance(match.end())
-            matchgroups = match.groupdict()
-            if matchgroups.get('macro_end') is not None:
-                self._parenthesis_stack.pop()
-            # we got an argument. let's see what it has
-            elif matchgroups.get('value') is not None:
-                # we have to process a single macro argument. This mutates
-                # the parenthesis stack! If it finds a comma, it will continue
-                # the loop.
-                self._process_macro_default_arg()
-            elif matchgroups.get('more_args') is not None:
-                continue
-            else:
-                raise dbt.exceptions.InternalException(
-                    'unhandled regex in _process_macro_args(), no match: {}'
-                    .format(matchgroups)
+    @property
+    def data(self):
+        return self.tag_parser.data
+
+    def is_current_end(self, tag):
+        return (
+            tag.block_type_name.startswith('end') and
+            self.current is not None and
+            tag.block_type_name[3:] == self.current.block_type_name
+        )
+
+    def find_blocks(self, allowed=None, collect_raw_data=True):
+        """Find all top-level blocks in the data."""
+        if allowed is None:
+            allowed = {'snapshot', 'macro', 'materialization', 'docs'}
+
+        for tag in self.tag_parser.find_tags():
+            if tag.block_type_name in _CONTROL_FLOW_TAGS:
+                self.stack.append(tag.block_type_name)
+            elif tag.block_type_name in _CONTROL_FLOW_END_TAGS:
+                found = None
+                if self.stack:
+                    found = self.stack.pop()
+                expected = None
+                if found:
+                    expected = _CONTROL_FLOW_TAGS[found]
+                if expected != tag.block_type_name:
+                    dbt.exceptions.raise_compiler_error((
+                        'Got unexpected control flow end tag, got {} but '
+                        'expected {} next (@ {})'
+                    ).format(tag.block_type_name, expected, tag.start))
+
+            if tag.block_type_name in allowed:
+                if self.stack:
+                    dbt.exceptions.raise_compiler_error((
+                        'Got a block definition inside control flow at {}. '
+                        'All dbt block definitions must be at the top level'
+                    ).format(tag.start))
+                if self.current is not None:
+                    dbt.exceptions.raise_compiler_error(
+                        duplicate_tags.format(outer=self.current, inner=tag)
+                    )
+                if collect_raw_data:
+                    raw_data = self.data[self.last_position:tag.start]
+                    self.last_position = tag.start
+                    if raw_data:
+                        yield BlockData(raw_data)
+                self.current = tag
+
+            elif self.is_current_end(tag):
+                self.last_position = tag.end
+                yield BlockTag(
+                    block_type_name=self.current.block_type_name,
+                    block_name=self.current.block_name,
+                    contents=self.data[self.current.end:tag.start],
+                    full_block=self.data[self.current.start:tag.end]
                 )
-            # if there are more arguments or a macro arg end we'll catch them
-            # on the next loop around
-
-    def lex_for_blocks(self):
-        return list(self.find_blocks())
+                self.current = None
+
+        if self.current:
+            dbt.exceptions.raise_compiler_error((
+                'Reached EOF without finding a close block for '
+                '{0.block_type_name} (from {0.end})'
+            ).format(self.current))
+
+        if collect_raw_data:
+            raw_data = self.data[self.last_position:]
+            if raw_data:
+                yield BlockData(raw_data)
+
+    def lex_for_blocks(self, allowed=None, collect_raw_data=True):
+        return list(self.find_blocks(allowed=allowed,
+                                     collect_raw_data=collect_raw_data))
diff --git a/core/dbt/clients/jinja.py b/core/dbt/clients/jinja.py
index 4ed0a083e4b..73bb8a872d9 100644
--- a/core/dbt/clients/jinja.py
+++ b/core/dbt/clients/jinja.py
@@ -354,5 +354,7 @@ def undefined_error(msg):
     raise jinja2.exceptions.UndefinedError(msg)
 
 
-def extract_toplevel_blocks(data):
-    return BlockIterator(data).lex_for_blocks()
+def extract_toplevel_blocks(data, allowed=None, collect_raw_data=True):
+    bi = BlockIterator(data)
+    return bi.lex_for_blocks(allowed=allowed,
+                             collect_raw_data=collect_raw_data)
diff --git a/core/dbt/parser/docs.py b/core/dbt/parser/docs.py
index 5d22aa526d7..dbc4e36ed9e 100644
--- a/core/dbt/parser/docs.py
+++ b/core/dbt/parser/docs.py
@@ -44,16 +44,17 @@ def load_file(cls, package_name, root_dir, relative_dirs):
 
     def parse(self, docfile):
         try:
-            blocks = extract_toplevel_blocks(docfile.file_contents)
+            blocks = extract_toplevel_blocks(
+                docfile.file_contents,
+                allowed={'docs'},
+                collect_raw_data=False
+            )
         except dbt.exceptions.CompilationException as exc:
             if exc.node is None:
                 exc.node = docfile
             raise
 
         for block in blocks:
-            if block.block_type_name != NodeType.Documentation:
-                continue
-
             try:
                 template = get_template(block.full_block, {})
             except dbt.exceptions.CompilationException as e:
diff --git a/core/dbt/parser/snapshots.py b/core/dbt/parser/snapshots.py
index 84e3d6b8c90..ee193a86ab1 100644
--- a/core/dbt/parser/snapshots.py
+++ b/core/dbt/parser/snapshots.py
@@ -26,16 +26,15 @@ def parse_snapshots_from_file(self, file_node, tags=None):
         # (we hope!) `snapshot` blocks
         try:
             blocks = dbt.clients.jinja.extract_toplevel_blocks(
-                file_node['raw_sql']
+                file_node['raw_sql'],
+                allowed={'snapshot'},
+                collect_raw_data=False
             )
         except dbt.exceptions.CompilationException as exc:
             if exc.node is None:
                 exc.node = file_node
             raise
         for block in blocks:
-            if block.block_type_name != NodeType.Snapshot:
-                # non-snapshot blocks are just ignored
-                continue
             name = block.block_name
             raw_sql = block.contents
             updates = {
diff --git a/core/dbt/rpc.py b/core/dbt/rpc.py
index 87305d89ee7..f6999e328ed 100644
--- a/core/dbt/rpc.py
+++ b/core/dbt/rpc.py
@@ -267,12 +267,28 @@ def handle(self, kwargs):
         self.started = time.time()
         self.timeout = kwargs.pop('timeout', None)
         self.queue = multiprocessing.Queue()
-        self.process = multiprocessing.Process(
-            target=_task_bootstrap,
-            args=(self.task, self.queue, kwargs)
-        )
-        self.process.start()
-        result = self.get_result()
+        if self.task.args.single_threaded:
+            _task_bootstrap(self.task, self.queue, kwargs)
+            try:
+                msgtype, result = self._wait_for_results()
+            except dbt.exceptions.RPCTimeoutException:
+                raise timeout_error(self.timeout)
+            except dbt.exceptions.Exception as exc:
+                raise dbt_error(exc)
+            except Exception as exc:
+                raise server_error(exc)
+
+            if msgtype == QueueMessageType.Error:
+                raise RPCException.from_error(result)
+
+            return result
+        else:
+            self.process = multiprocessing.Process(
+                target=_task_bootstrap,
+                args=(self.task, self.queue, kwargs)
+            )
+            self.process.start()
+            result = self.get_result()
         return result
 
     @property
diff --git a/core/dbt/task/rpc_server.py b/core/dbt/task/rpc_server.py
index 0dfdfa4c277..370e10b80d4 100644
--- a/core/dbt/task/rpc_server.py
+++ b/core/dbt/task/rpc_server.py
@@ -60,7 +60,7 @@ def run(self):
         # easier as well). The alternative involves tracking metadata+state in
         # a multiprocessing.Manager, adds polling the manager to the request
         # task handler and in general gets messy fast.
-        run_simple(host, port, app, threaded=True)
+        run_simple(host, port, app, threaded=not self.args.single_threaded)
 
     @Request.application
     def handle_jsonrpc_request(self, request):
diff --git a/test/unit/test_jinja.py b/test/unit/test_jinja.py
index 79d5b885937..03da585e7f4 100644
--- a/test/unit/test_jinja.py
+++ b/test/unit/test_jinja.py
@@ -18,7 +18,7 @@ class TestBlockLexer(unittest.TestCase):
     def test_basic(self):
         body = '{{ config(foo="bar") }}\r\nselect * from this.that\r\n'
         block_data = '  \n\r\t{%- mytype foo %}'+body+'{%endmytype -%}'
-        blocks = extract_toplevel_blocks(block_data)
+        blocks = extract_toplevel_blocks(block_data, allowed={'mytype'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'mytype')
         self.assertEqual(blocks[0].block_name, 'foo')
@@ -36,16 +36,14 @@ def test_multiple(self):
             '  {% mytype foo %}' + body_one + '{% endmytype %}' +
             '\r\n{% othertype bar %}' + body_two + '{% endothertype %}'
         )
-        all_blocks = extract_toplevel_blocks(block_data)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(block_data, allowed={'mytype', 'othertype'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 2)
 
     def test_comments(self):
         body = '{{ config(foo="bar") }}\r\nselect * from this.that\r\n'
         comment = '{# my comment #}'
         block_data = '  \n\r\t{%- mytype foo %}'+body+'{%endmytype -%}'
-        all_blocks = extract_toplevel_blocks(comment+block_data)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(comment+block_data, allowed={'mytype'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'mytype')
         self.assertEqual(blocks[0].block_name, 'foo')
@@ -56,8 +54,7 @@ def test_evil_comments(self):
         body = '{{ config(foo="bar") }}\r\nselect * from this.that\r\n'
         comment = '{# external comment {% othertype bar %} select * from thing.other_thing{% endothertype %} #}'
         block_data = '  \n\r\t{%- mytype foo %}'+body+'{%endmytype -%}'
-        all_blocks = extract_toplevel_blocks(comment+block_data)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(comment+block_data, allowed={'mytype'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'mytype')
         self.assertEqual(blocks[0].block_name, 'foo')
@@ -68,8 +65,7 @@ def test_nested_comments(self):
         body = '{# my comment #} {{ config(foo="bar") }}\r\nselect * from {# my other comment embedding {% endmytype %} #} this.that\r\n'
         block_data = '  \n\r\t{%- mytype foo %}'+body+'{% endmytype -%}'
         comment = '{# external comment {% othertype bar %} select * from thing.other_thing{% endothertype %} #}'
-        all_blocks = extract_toplevel_blocks(comment+block_data)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(comment+block_data, allowed={'mytype'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'mytype')
         self.assertEqual(blocks[0].block_name, 'foo')
@@ -77,8 +73,7 @@ def test_nested_comments(self):
         self.assertEqual(blocks[0].full_block, block_data)
 
     def test_complex_file(self):
-        all_blocks = extract_toplevel_blocks(complex_snapshot_file)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(complex_snapshot_file, allowed={'mytype', 'myothertype'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 3)
         self.assertEqual(blocks[0].block_type_name, 'mytype')
         self.assertEqual(blocks[0].block_name, 'foo')
@@ -94,21 +89,35 @@ def test_complex_file(self):
         self.assertEqual(blocks[2].contents, x_block[len('\n{% myothertype x %}'):-len('{% endmyothertype %}\n')])
 
     def test_peaceful_macro_coexistence(self):
-        body = '{# my macro #} {% macro foo(a, b) %} do a thing {%- endmacro %} {# my model #} {% a b %} {% enda %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 2)
-        self.assertEqual(blocks[0].block_type_name, 'macro')
-        self.assertEqual(blocks[0].block_name, 'foo')
-        self.assertEqual(blocks[0].contents, ' do a thing')
-        self.assertEqual(blocks[1].block_type_name, 'a')
-        self.assertEqual(blocks[1].block_name, 'b')
-        self.assertEqual(blocks[1].contents, ' ')
+        body = '{# my macro #} {% macro foo(a, b) %} do a thing {%- endmacro %} {# my model #} {% a b %} test {% enda %}'
+        blocks = extract_toplevel_blocks(body, allowed={'macro', 'a'}, collect_raw_data=True)
+        self.assertEqual(len(blocks), 4)
+        self.assertEqual(blocks[0].full_block, '{# my macro #} ')
+        self.assertEqual(blocks[1].block_type_name, 'macro')
+        self.assertEqual(blocks[1].block_name, 'foo')
+        self.assertEqual(blocks[1].contents, ' do a thing')
+        self.assertEqual(blocks[2].full_block, ' {# my model #} ')
+        self.assertEqual(blocks[3].block_type_name, 'a')
+        self.assertEqual(blocks[3].block_name, 'b')
+        self.assertEqual(blocks[3].contents, ' test ')
+
+    def test_macro_with_trailing_data(self):
+        body = '{# my macro #} {% macro foo(a, b) %} do a thing {%- endmacro %} {# my model #} {% a b %} test {% enda %} raw data so cool'
+        blocks = extract_toplevel_blocks(body, allowed={'macro', 'a'}, collect_raw_data=True)
+        self.assertEqual(len(blocks), 5)
+        self.assertEqual(blocks[0].full_block, '{# my macro #} ')
+        self.assertEqual(blocks[1].block_type_name, 'macro')
+        self.assertEqual(blocks[1].block_name, 'foo')
+        self.assertEqual(blocks[1].contents, ' do a thing')
+        self.assertEqual(blocks[2].full_block, ' {# my model #} ')
+        self.assertEqual(blocks[3].block_type_name, 'a')
+        self.assertEqual(blocks[3].block_name, 'b')
+        self.assertEqual(blocks[3].contents, ' test ')
+        self.assertEqual(blocks[4].full_block, ' raw data so cool')
 
     def test_macro_with_crazy_args(self):
         body = '''{% macro foo(a, b=asdf("cool this is 'embedded'" * 3) + external_var, c)%}cool{# block comment with {% endmacro %} in it #} stuff here {% endmacro %}'''
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'macro'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'macro')
         self.assertEqual(blocks[0].block_name, 'foo')
@@ -116,43 +125,36 @@ def test_macro_with_crazy_args(self):
 
     def test_materialization_parse(self):
         body = '{% materialization xxx, default %} ... {% endmaterialization %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'materialization'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'materialization')
         self.assertEqual(blocks[0].block_name, 'xxx')
         self.assertEqual(blocks[0].full_block, body)
 
         body = '{% materialization xxx, adapter="other" %} ... {% endmaterialization %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'materialization'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'materialization')
         self.assertEqual(blocks[0].block_name, 'xxx')
         self.assertEqual(blocks[0].full_block, body)
 
-    def test_nested_ok(self):
+    def test_nested_not_ok(self):
         # we don't allow nesting same blocks
-        # ideally we would not allow nesting any, but that's much harder
         body = '{% myblock a %} {% myblock b %} {% endmyblock %} {% endmyblock %}'
-        blocks = extract_toplevel_blocks(body)
-        self.assertEqual(len(blocks), 1)
-        self.assertEqual(blocks[0].block_type_name, 'myblock')
-        self.assertEqual(blocks[0].block_name, 'a')
-        self.assertEqual(blocks[0].contents, ' {% myblock b %} {% endmyblock %} ')
-        self.assertEqual(blocks[0].full_block, body)
+        with self.assertRaises(CompilationException):
+            extract_toplevel_blocks(body, allowed={'myblock'})
 
     def test_incomplete_block_failure(self):
-        fullbody = '{% myblock foo %} {% endblock %}'
-        for length in range(1, len(fullbody)-1):
+        fullbody = '{% myblock foo %} {% endmyblock %}'
+        for length in range(len('{% myblock foo %}'), len(fullbody)-1):
             body = fullbody[:length]
-        with self.assertRaises(CompilationException):
-            extract_toplevel_blocks(body)
+            with self.assertRaises(CompilationException):
+                extract_toplevel_blocks(body, allowed={'myblock'})
 
     def test_wrong_end_failure(self):
         body = '{% myblock foo %} {% endotherblock %}'
         with self.assertRaises(CompilationException):
-            extract_toplevel_blocks(body)
+            extract_toplevel_blocks(body, allowed={'myblock', 'otherblock'})
 
     def test_comment_no_end_failure(self):
         body = '{# '
@@ -161,124 +163,102 @@ def test_comment_no_end_failure(self):
 
     def test_comment_only(self):
         body = '{# myblock #}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body)
+        self.assertEqual(len(blocks), 1)
+        blocks = extract_toplevel_blocks(body, collect_raw_data=False)
         self.assertEqual(len(blocks), 0)
 
     def test_comment_block_self_closing(self):
         # test the case where a comment start looks a lot like it closes itself
         # (but it doesn't in jinja!)
         body = '{#} {% myblock foo %} {#}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, collect_raw_data=False)
         self.assertEqual(len(blocks), 0)
 
     def test_embedded_self_closing_comment_block(self):
         body = '{% myblock foo %} {#}{% endmyblock %} {#}{% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'myblock'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].full_block, body)
         self.assertEqual(blocks[0].contents, ' {#}{% endmyblock %} {#}')
 
     def test_set_statement(self):
         body = '{% set x = 1 %}{% myblock foo %}hi{% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 2)
-        self.assertEqual(blocks[0].full_block, '{% set x = 1 %}')
-        self.assertEqual(blocks[1].full_block, '{% myblock foo %}hi{% endmyblock %}')
+        blocks = extract_toplevel_blocks(body, allowed={'myblock'}, collect_raw_data=False)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, '{% myblock foo %}hi{% endmyblock %}')
 
     def test_set_block(self):
         body = '{% set x %}1{% endset %}{% myblock foo %}hi{% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 2)
-        self.assertEqual(blocks[0].contents, '1')
-        self.assertEqual(blocks[0].block_type_name, 'set')
-        self.assertEqual(blocks[0].block_name, 'x')
-        self.assertEqual(blocks[1].full_block, '{% myblock foo %}hi{% endmyblock %}')
+        blocks = extract_toplevel_blocks(body, allowed={'myblock'}, collect_raw_data=False)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, '{% myblock foo %}hi{% endmyblock %}')
 
     def test_crazy_set_statement(self):
         body = '{% set x = (thing("{% myblock foo %}")) %}{% otherblock bar %}x{% endotherblock %}{% set y = otherthing("{% myblock foo %}") %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 3)
-        self.assertEqual(blocks[0].full_block, '{% set x = (thing("{% myblock foo %}")) %}')
-        self.assertEqual(blocks[0].block_type_name, 'set')
-        self.assertEqual(blocks[1].full_block, '{% otherblock bar %}x{% endotherblock %}')
-        self.assertEqual(blocks[1].block_type_name, 'otherblock')
-        self.assertEqual(blocks[2].full_block, '{% set y = otherthing("{% myblock foo %}") %}')
-        self.assertEqual(blocks[2].block_type_name, 'set')
+        blocks = extract_toplevel_blocks(body, allowed={'otherblock'}, collect_raw_data=False)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, '{% otherblock bar %}x{% endotherblock %}')
+        self.assertEqual(blocks[0].block_type_name, 'otherblock')
 
     def test_do_statement(self):
         body = '{% do thing.update() %}{% myblock foo %}hi{% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 2)
-        self.assertEqual(blocks[0].full_block, '{% do thing.update() %}')
-        self.assertEqual(blocks[1].full_block, '{% myblock foo %}hi{% endmyblock %}')
+        blocks = extract_toplevel_blocks(body, allowed={'myblock'}, collect_raw_data=False)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, '{% myblock foo %}hi{% endmyblock %}')
 
     def test_deceptive_do_statement(self):
         body = '{% do thing %}{% myblock foo %}hi{% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 2)
-        self.assertEqual(blocks[0].full_block, '{% do thing %}')
-        self.assertEqual(blocks[1].full_block, '{% myblock foo %}hi{% endmyblock %}')
+        blocks = extract_toplevel_blocks(body, allowed={'myblock'}, collect_raw_data=False)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, '{% myblock foo %}hi{% endmyblock %}')
 
     def test_do_block(self):
         body = '{% do %}thing.update(){% enddo %}{% myblock foo %}hi{% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'do', 'myblock'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 2)
         self.assertEqual(blocks[0].contents, 'thing.update()')
         self.assertEqual(blocks[0].block_type_name, 'do')
         self.assertEqual(blocks[1].full_block, '{% myblock foo %}hi{% endmyblock %}')
 
     def test_crazy_do_statement(self):
-        body = '{% do (thing("{% myblock foo %}")) %}{% otherblock bar %}x{% endotherblock %}{% do otherthing("{% myblock foo %}") %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 3)
-        self.assertEqual(blocks[0].full_block, '{% do (thing("{% myblock foo %}")) %}')
-        self.assertEqual(blocks[0].block_type_name, 'do')
-        self.assertEqual(blocks[1].full_block, '{% otherblock bar %}x{% endotherblock %}')
-        self.assertEqual(blocks[1].block_type_name, 'otherblock')
-        self.assertEqual(blocks[2].full_block, '{% do otherthing("{% myblock foo %}") %}')
-        self.assertEqual(blocks[2].block_type_name, 'do')
+        body = '{% do (thing("{% myblock foo %}")) %}{% otherblock bar %}x{% endotherblock %}{% do otherthing("{% myblock foo %}") %}{% myblock x %}hi{% endmyblock %}'
+        blocks = extract_toplevel_blocks(body, allowed={'myblock', 'otherblock'}, collect_raw_data=False)
+        self.assertEqual(len(blocks), 2)
+        self.assertEqual(blocks[0].full_block, '{% otherblock bar %}x{% endotherblock %}')
+        self.assertEqual(blocks[0].block_type_name, 'otherblock')
+        self.assertEqual(blocks[1].full_block, '{% myblock x %}hi{% endmyblock %}')
+        self.assertEqual(blocks[1].block_type_name, 'myblock')
 
     def test_awful_jinja(self):
-        all_blocks = extract_toplevel_blocks(if_you_do_this_you_are_awful)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
-        self.assertEqual(len(blocks), 4)
-        self.assertEqual(blocks[0].block_type_name, 'do')
-        self.assertEqual(blocks[0].full_block, '''{% do\n    set('foo="bar"')\n%}''')
-        self.assertEqual(blocks[1].block_type_name, 'set')
-        self.assertEqual(blocks[1].full_block, '''{% set x = ("100" + "hello'" + '%}') %}''')
-        self.assertEqual(blocks[2].block_type_name, 'snapshot')
-        self.assertEqual(blocks[2].contents, '\n    '.join([
+        blocks = extract_toplevel_blocks(
+                if_you_do_this_you_are_awful,
+                allowed={'snapshot', 'materialization'},
+                collect_raw_data=False
+        )
+        self.assertEqual(len(blocks), 2)
+        self.assertEqual(len([b for b in blocks if b.block_type_name == '__dbt__data']), 0)
+        self.assertEqual(blocks[0].block_type_name, 'snapshot')
+        self.assertEqual(blocks[0].contents, '\n    '.join([
             '''{% set x = ("{% endsnapshot %}" + (40 * '%})')) %}''',
             '{# {% endsnapshot %} #}',
             '{% embedded %}',
             '    some block data right here',
             '{% endembedded %}'
         ]))
-        self.assertEqual(blocks[3].block_type_name, 'materialization')
-        self.assertEqual(blocks[3].contents, '\nhi\n')
+        self.assertEqual(blocks[1].block_type_name, 'materialization')
+        self.assertEqual(blocks[1].contents, '\nhi\n')
 
     def test_quoted_endblock_within_block(self):
         body = '{% myblock something -%}  {% set x = ("{% endmyblock %}") %}  {% endmyblock %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'myblock'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'myblock')
         self.assertEqual(blocks[0].contents, '{% set x = ("{% endmyblock %}") %}  ')
 
     def test_docs_block(self):
         body = '{% docs __my_doc__ %} asdf {# nope {% enddocs %}} #} {% enddocs %} {% docs __my_other_doc__ %} asdf "{% enddocs %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'docs'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 2)
         self.assertEqual(blocks[0].block_type_name, 'docs')
         self.assertEqual(blocks[0].contents, ' asdf {# nope {% enddocs %}} #} ')
@@ -289,8 +269,7 @@ def test_docs_block(self):
 
     def test_docs_block_expr(self):
         body = '{% docs more_doc %} asdf {{ "{% enddocs %}" ~ "}}" }}{% enddocs %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body, allowed={'docs'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'docs')
         self.assertEqual(blocks[0].contents, ' asdf {{ "{% enddocs %}" ~ "}}" }}')
@@ -299,13 +278,36 @@ def test_docs_block_expr(self):
     def test_unclosed_model_quotes(self):
         # test case for https://github.com/fishtown-analytics/dbt/issues/1533
         body = '{% model my_model -%} select * from "something"."something_else{% endmodel %}'
-        all_blocks = extract_toplevel_blocks(body)
-        blocks = [b for b in all_blocks if b.block_type_name != '__dbt__data']
+        blocks = extract_toplevel_blocks(body,allowed={'model'}, collect_raw_data=False)
         self.assertEqual(len(blocks), 1)
         self.assertEqual(blocks[0].block_type_name, 'model')
         self.assertEqual(blocks[0].contents, 'select * from "something"."something_else')
         self.assertEqual(blocks[0].block_name, 'my_model')
 
+    def test_if(self):
+        # if you conditionally define your macros/models, don't
+        body = '{% if true %}{% macro my_macro() %} adsf {% endmacro %}{% endif %}'
+        with self.assertRaises(CompilationException):
+            extract_toplevel_blocks(body)
+
+    def test_if_innocuous(self):
+        body = '{% if true %}{% something %}asdfasd{% endsomething %}{% endif %}'
+        blocks = extract_toplevel_blocks(body)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, body)
+
+    def test_for(self):
+        # no for-loops over macros.
+        body = '{% for x in range(10) %}{% macro my_macro() %} adsf {% endmacro %}{% endfor %}'
+        with self.assertRaises(CompilationException):
+            extract_toplevel_blocks(body)
+
+    def test_for_innocuous(self):
+        # no for-loops over macros.
+        body = '{% for x in range(10) %}{% something my_something %} adsf {% endsomething %}{% endfor %}'
+        blocks = extract_toplevel_blocks(body)
+        self.assertEqual(len(blocks), 1)
+        self.assertEqual(blocks[0].full_block, body)
 
 bar_block = '''{% mytype bar %}
 {# a comment
@@ -359,4 +361,3 @@ def test_unclosed_model_quotes(self):
 hi
 {% endmaterialization %}
 '''
-


Code review comments:
diff_hunk: @@ -354,5 +354,7 @@ def undefined_error(msg):
     raise jinja2.exceptions.UndefinedError(msg)
 
 
-def extract_toplevel_blocks(data):
-    return BlockIterator(data).lex_for_blocks()
+def extract_toplevel_blocks(data, allowed=None, collect_raw_data=True):
+    bi = BlockIterator(data)
+    return bi.lex_for_blocks(allowed=allowed,
+                             collect_raw_data=collect_raw_data)
path: core/dbt/clients/jinja.py
text: @user1:
can you either collapse this back or name `bi` something else? so minor but don't use acronyms/single letter variables

@user1:
I'm having trouble understanding the utility of `collect_raw_data`. i mean, I understand what it does, but I'm not sure when we'd use it in practice. all the (non-test) calls to `extract_toplevel_blocks` in this PR use collect_raw_data=False

@author:
The key is in the things I didn't change in this PR! We use it here during RPC parsing: https://github.com/fishtown-analytics/dbt/blob/fix/block-parser-ignore-more/core/dbt/task/compile.py#L64

@user1:
@author thanks, that makes sense. and the more code i read here the more it makes sense why we'd want it. it's nice to be able to turn this on and off for RPC use cases, tests, and also potentially for the evolution of the dbt parser. it gives us lots of options.

can you rename `allowed` to `allowed_blocks` and add a docstring to this fn?