Repository: sqlfluff/sqlfluff
Pull request: Draft: Resolve parallel dbt templating issue
This PR has a couple of things to still shake out. Namely that one test is failing for a reason I don't understand, and it looks like the CLI output is treading on toes when running in parallel. Those issues aside, I think it's ready for a review, at least from an architecture standpoint.

This resolves #1088 .

I've tested it on our large dbt project with the dbt templater and it's FAST.

It also builds on #1103 so will be easier to review once that's merged.

In short:
- We render the sql files in the main thread and then pass them to the child threads for parsing and linting.
- The rest of the changes are making enough methods in the linter _static_ or _class_ methods to do that in a way that we don't have to create a new `Linter` object each time.

Problem statement:
Dbt templater re-runs compile for each file while running parallel
While debugging slow running using the new `--parallel` feature, I've found that the dbt templater is running the compile step again for each file.

I *think* this is because each file uses it's own `Linter` object and so doesn't utilise the cached methods on the dbt templater.

The solution to this I think is to run the templating step in the main thread before dispatching out for parsing and linting. This should be possible by modifying the runner classes and the method calls available on `Linter`. The tricky part will be splitting out the templating step from `Linter.lint_string()` and `Linter.parse_string()`.

@barrywhart , having thought about this today, I was going to suggest this:
- We *don't* create a new `Linter` for each file.
- Instead, we abstract the parsing and linting methods out into `staticmethods` which can be safely called by the threads. Behind the existing methods of `parse_string()` and `lint_string()` there will be private methods of `_render_string()`, `_parse()` and `_lint()` which are static methods and handle all of the meaningful logic. These can be called via the existing methods _or directly from the runner classes_.
- In the parallel runners, the templating step (`_render_string()`) is called in the main thread, and then the resulting `TemplatedFile` in handed off to a child process to parse and lint (`_parse()` and `_lint()`). Those child processes never need their own `Linter()` object because they're just using the static methods, but might need a copy of the `FluffConfig` or similar objects.

Reckon that could work? I'm happy to have a go at this, I should probably get more familiar with the parallel code as is.

Hints from issue discussion:
That sounds like a good plan. I like the design of using static methods in the worker processes because it limits the number of object instances being passed to the workers. With `multiprocessing.Pool`, I've seen things get pretty confusing if you call normal instance methods from the workers, because Python serializes those objects automatically, but of course in the child it's not the same object, etc. etc. Far better to pass entire objects that you deliberately want to pass.

@alanmcruickshank: Go for it! I appreciate the help. Suddenly, work has gotten pretty busy for me with new hires, etc. etc.

Commit message: Fix filename passing

Patch to review:
diff --git a/src/sqlfluff/core/linter/common.py b/src/sqlfluff/core/linter/common.py
index 03ed5428a3a..1ad7506f466 100644
--- a/src/sqlfluff/core/linter/common.py
+++ b/src/sqlfluff/core/linter/common.py
@@ -5,9 +5,10 @@
     NamedTuple,
     Optional,
     Tuple,
+    Dict,
 )
 
-from sqlfluff.core.errors import SQLBaseError
+from sqlfluff.core.errors import SQLBaseError, SQLTemplaterError
 from sqlfluff.core.templaters import TemplatedFile
 from sqlfluff.core.config import FluffConfig
 from sqlfluff.core.parser.segments.base import BaseSegment
@@ -28,6 +29,19 @@ class NoQaDirective(NamedTuple):
     action: Optional[str]  # "enable", "disable", or "None"
 
 
+class RenderedFile(NamedTuple):
+    """An object to store the result of a templated file/string.
+
+    This is notable as it's the intermediate state between what happens
+    in the main thread and the child threads when running in parallel mode.
+    """
+
+    templated_file: TemplatedFile
+    templater_violations: List[SQLTemplaterError]
+    config: FluffConfig
+    time_dict: Dict[str, float]
+
+
 class ParsedString(NamedTuple):
     """An object to store the result of parsing a string.
 
diff --git a/src/sqlfluff/core/linter/linter.py b/src/sqlfluff/core/linter/linter.py
index ec041724d76..0604562dcce 100644
--- a/src/sqlfluff/core/linter/linter.py
+++ b/src/sqlfluff/core/linter/linter.py
@@ -36,7 +36,12 @@
 from sqlfluff.core.parser.segments.raw import RawSegment
 from sqlfluff.core.rules.base import BaseRule
 
-from sqlfluff.core.linter.common import RuleTuple, ParsedString, NoQaDirective
+from sqlfluff.core.linter.common import (
+    RuleTuple,
+    ParsedString,
+    NoQaDirective,
+    RenderedFile,
+)
 from sqlfluff.core.linter.linted_file import LintedFile
 from sqlfluff.core.linter.linted_dir import LintedDir
 from sqlfluff.core.linter.linting_result import LintingResult
@@ -88,6 +93,22 @@ def rule_tuples(self) -> List[RuleTuple]:
         rs = self.get_ruleset()
         return [RuleTuple(rule.code, rule.description) for rule in rs]
 
+    # #### Static methods
+    # These are the building blocks of the linting process.
+
+    @staticmethod
+    def _load_raw_file_and_config(fname, root_config):
+        """Load a raw file and the associated config."""
+        file_config = root_config.make_child_from_path(fname)
+        with open(
+            fname, "r", encoding="utf8", errors="backslashreplace"
+        ) as target_file:
+            raw_file = target_file.read()
+        # Scan the raw file for config commands.
+        file_config.process_raw_file_for_config(raw_file)
+        # Return the raw file and config
+        return raw_file, file_config
+
     @staticmethod
     def _lex_templated_file(
         templated_file: TemplatedFile, config: FluffConfig
@@ -191,76 +212,8 @@ def _parse_tokens(
                 linter_logger.info(unparsable.stringify())
         return parsed, violations
 
-    def render_string(self, in_str: str, fname: Optional[str], config: FluffConfig):
-        """Template the file."""
-        linter_logger.info("TEMPLATING RAW [%s] (%s)", self.templater.name, fname)
-
-        try:
-            templated_file, templater_violations = self.templater.process(
-                in_str=in_str, fname=fname, config=config, formatter=self.formatter
-            )
-        except SQLTemplaterSkipFile as s:
-            linter_logger.warning(str(s))
-            templated_file = None
-            templater_violations = []
-
-        if not templated_file:
-            linter_logger.info("TEMPLATING FAILED: %s", templater_violations)
-
-        return templated_file, templater_violations
-
-    def parse_string(
-        self,
-        in_str: str,
-        fname: Optional[str] = None,
-        recurse: bool = True,
-        config: Optional[FluffConfig] = None,
-    ) -> ParsedString:
-        """Parse a string."""
-        violations: List[SQLBaseError] = []
-        t0 = time.monotonic()
-
-        # Dispatch the output for the template header (including the config diff)
-        if self.formatter:
-            self.formatter.dispatch_template_header(fname, self.config, config)
-
-        # Just use the local config from here:
-        config = config or self.config
-
-        # Scan the raw file for config commands.
-        config.process_raw_file_for_config(in_str)
-
-        templated_file, templater_violations = self.render_string(in_str, fname, config)
-        violations += templater_violations
-
-        t1 = time.monotonic()
-
-        # Dispatch the output for the parse header
-        if self.formatter:
-            self.formatter.dispatch_parse_header(fname)
-
-        tokens: Optional[Sequence[BaseSegment]]
-        if templated_file:
-            tokens, lvs, config = self._lex_templated_file(templated_file, config)
-            violations += lvs
-        else:
-            tokens = None
-
-        t2 = time.monotonic()
-        linter_logger.info("PARSING (%s)", fname)
-
-        if tokens:
-            parsed, pvs = self._parse_tokens(tokens, config, recurse=recurse)
-            violations += pvs
-        else:
-            parsed = None
-
-        t3 = time.monotonic()
-        time_dict = {"templating": t1 - t0, "lexing": t2 - t1, "parsing": t3 - t2}
-        return ParsedString(parsed, violations, time_dict, templated_file, config)
-
-    @classmethod
-    def parse_noqa(cls, comment: str, line_no: int):
+    @staticmethod
+    def parse_noqa(comment: str, line_no: int):
         """Extract ignore mask entries from a comment string."""
         # Also trim any whitespace afterward
         if comment.startswith("noqa"):
@@ -299,6 +252,66 @@ def parse_noqa(cls, comment: str, line_no: int):
             return NoQaDirective(line_no, None, None)
         return None
 
+    @staticmethod
+    def remove_templated_errors(
+        linting_errors: List[SQLLintError],
+    ) -> List[SQLLintError]:
+        """Filter a list of lint errors, removing those which only occur in templated slices."""
+        # Filter out any linting errors in templated sections if relevant.
+        linting_errors = list(
+            filter(
+                lambda e: (
+                    # Is it in a literal section?
+                    e.segment.pos_marker.is_literal()
+                    # Is it a rule that is designed to work on templated sections?
+                    or e.rule.targets_templated
+                ),
+                linting_errors,
+            )
+        )
+        return linting_errors
+
+    @staticmethod
+    def _warn_unfixable(code: str):
+        linter_logger.warning(
+            f"One fix for {code} not applied, it would re-cause the same error."
+        )
+
+    # ### Class Methods
+    # These compose the base static methods into useful recipes.
+
+    @classmethod
+    def parse_rendered(cls, rendered: RenderedFile, recurse: bool = True):
+        """Parse a rendered file."""
+        t0 = time.monotonic()
+        violations = cast(List[SQLBaseError], rendered.templater_violations)
+        tokens: Optional[Sequence[BaseSegment]]
+        if rendered.templated_file:
+            tokens, lvs, config = cls._lex_templated_file(
+                rendered.templated_file, rendered.config
+            )
+            violations += lvs
+        else:
+            tokens = None
+
+        t1 = time.monotonic()
+        linter_logger.info("PARSING (%s)", rendered.templated_file.fname)
+
+        if tokens:
+            parsed, pvs = cls._parse_tokens(tokens, rendered.config, recurse=recurse)
+            violations += pvs
+        else:
+            parsed = None
+
+        time_dict = {
+            **rendered.time_dict,
+            "lexing": t1 - t0,
+            "parsing": time.monotonic() - t1,
+        }
+        return ParsedString(
+            parsed, violations, time_dict, rendered.templated_file, rendered.config
+        )
+
     @classmethod
     def extract_ignore_from_comment(cls, comment: RawSegment):
         """Extract ignore mask entries from a comment segment."""
@@ -310,22 +323,36 @@ def extract_ignore_from_comment(cls, comment: RawSegment):
             result.segment = comment
         return result
 
-    @staticmethod
-    def _warn_unfixable(code: str):
-        linter_logger.warning(
-            f"One fix for {code} not applied, it would re-cause the same error."
-        )
+    @classmethod
+    def extract_ignore_mask(cls, tree: Optional[BaseSegment]):
+        """Look for inline ignore comments and return NoQaDirectives."""
+        ignore_buff: List[NoQaDirective] = []
+        violations: List[SQLParseError] = []
+        if not tree:
+            return ignore_buff, violations
+        for comment in tree.recursive_crawl("comment"):
+            if comment.name == "inline_comment":
+                ignore_entry = cls.extract_ignore_from_comment(comment)
+                if isinstance(ignore_entry, SQLParseError):
+                    violations.append(ignore_entry)
+                elif ignore_entry:
+                    ignore_buff.append(ignore_entry)
+        if ignore_buff:
+            linter_logger.info("Parsed noqa directives from file: %r", ignore_buff)
+        return ignore_buff, violations
 
-    def lint_fix(
-        self,
+    @classmethod
+    def lint_fix_parsed(
+        cls,
         tree: BaseSegment,
-        config: Optional[FluffConfig] = None,
+        config: FluffConfig,
+        rule_set: List[BaseRule],
         fix: bool = False,
         fname: Optional[str] = None,
         templated_file: Optional[TemplatedFile] = None,
+        formatter: Any = None,
     ) -> Tuple[BaseSegment, List[SQLLintError]]:
         """Lint and optionally fix a tree object."""
-        config = config or self.config
         # Keep track of the linting errors
         all_linting_errors = []
         # A placeholder for the fixes we had on the previous loop
@@ -337,12 +364,12 @@ def lint_fix(
         loop_limit = config.get("runaway_limit") if fix else 1
 
         # Dispatch the output for the lint header
-        if self.formatter:
-            self.formatter.dispatch_lint_header(fname)
+        if formatter:
+            formatter.dispatch_lint_header(fname)
 
         for loop in range(loop_limit):
             changed = False
-            for crawler in self.get_ruleset(config=config):
+            for crawler in rule_set:
                 # fixes should be a dict {} with keys edit, delete, create
                 # delete is just a list of segments to delete
                 # edit and create are list of tuples. The first element is the
@@ -360,7 +387,7 @@ def lint_fix(
                     linter_logger.info(f"Applying Fixes: {fixes}")
                     # Do some sanity checks on the fixes before applying.
                     if fixes == last_fixes:
-                        self._warn_unfixable(crawler.code)
+                        cls._warn_unfixable(crawler.code)
                     else:
                         last_fixes = fixes
                         new_tree, _ = tree.apply_fixes(fixes)
@@ -374,7 +401,7 @@ def lint_fix(
                         else:
                             # Applying these fixes took us back to a state which we've
                             # seen before. Abort.
-                            self._warn_unfixable(crawler.code)
+                            cls._warn_unfixable(crawler.code)
 
             if loop == 0:
                 # Keep track of initial errors for reporting.
@@ -391,29 +418,150 @@ def lint_fix(
             linter_logger.warning(f"Loop limit on fixes reached [{loop_limit}].")
 
         if config.get("ignore_templated_areas", default=True):
-            initial_linting_errors = self.remove_templated_errors(
-                initial_linting_errors
-            )
+            initial_linting_errors = cls.remove_templated_errors(initial_linting_errors)
 
         return tree, initial_linting_errors
 
-    def remove_templated_errors(
-        self, linting_errors: List[SQLLintError]
-    ) -> List[SQLLintError]:
-        """Filter a list of lint errors, removing those which only occur in templated slices."""
-        # Filter out any linting errors in templated sections if relevant.
-        linting_errors = list(
-            filter(
-                lambda e: (
-                    # Is it in a literal section?
-                    e.segment.pos_marker.is_literal()
-                    # Is it a rule that is designed to work on templated sections?
-                    or e.rule.targets_templated
-                ),
-                linting_errors,
+    @classmethod
+    def lint_parsed(
+        cls,
+        parsed: ParsedString,
+        rule_set: List[BaseRule],
+        fix: bool = False,
+        formatter: Any = None,
+    ):
+        """Lint a ParsedString and return a LintedFile."""
+        violations = parsed.violations
+        time_dict = parsed.time_dict
+        # Look for comment segments which might indicate lines to ignore.
+        ignore_buff, ivs = cls.extract_ignore_mask(parsed.tree)
+        violations += ivs
+
+        if parsed.tree:
+            t0 = time.monotonic()
+            linter_logger.info("LINTING (%s)", parsed.templated_file.fname)
+            tree, initial_linting_errors = cls.lint_fix_parsed(
+                parsed.tree,
+                config=parsed.config,
+                rule_set=rule_set,
+                fix=fix,
+                fname=parsed.templated_file.fname,
+                templated_file=parsed.templated_file,
+                formatter=formatter,
             )
+            # Update the timing dict
+            time_dict["linting"] = time.monotonic() - t0
+
+            # We're only going to return the *initial* errors, rather
+            # than any generated during the fixing cycle.
+            violations += initial_linting_errors
+
+        # We process the ignore config here if appropriate
+        for violation in violations:
+            violation.ignore_if_in(parsed.config.get("ignore"))
+
+        linted_file = LintedFile(
+            parsed.templated_file.fname,
+            violations,
+            time_dict,
+            tree,
+            ignore_mask=ignore_buff,
+            templated_file=parsed.templated_file,
         )
-        return linting_errors
+
+        # This is the main command line output from linting.
+        if formatter:
+            formatter.dispatch_file_violations(
+                parsed.templated_file.fname, linted_file, only_fixable=fix
+            )
+
+        # Safety flag for unset dialects
+        if parsed.config.get("dialect") == "ansi" and linted_file.get_violations(
+            fixable=True if fix else None, types=SQLParseError
+        ):
+            if formatter:
+                formatter.dispatch_dialect_warning()
+
+        return linted_file
+
+    @classmethod
+    def lint_rendered(
+        cls,
+        rendered: RenderedFile,
+        rule_set: List[BaseRule],
+        fix: bool = False,
+        formatter: Any = None,
+    ) -> LintedFile:
+        """Take a RenderedFile and return a LintedFile."""
+        parsed = cls.parse_rendered(rendered)
+        return cls.lint_parsed(parsed, rule_set=rule_set, fix=fix, formatter=formatter)
+
+    # ### Instance Methods
+    # These are tied to a specific instance and so are not necessarily
+    # safe to use in parallel operations.
+
+    def render_string(
+        self, in_str: str, fname: Optional[str], config: FluffConfig
+    ) -> RenderedFile:
+        """Template the file."""
+        linter_logger.info("TEMPLATING RAW [%s] (%s)", self.templater.name, fname)
+
+        # Start the templating timer
+        t0 = time.monotonic()
+
+        try:
+            templated_file, templater_violations = self.templater.process(
+                in_str=in_str, fname=fname, config=config, formatter=self.formatter
+            )
+        except SQLTemplaterSkipFile as s:
+            linter_logger.warning(str(s))
+            templated_file = None
+            templater_violations = []
+
+        if not templated_file:
+            linter_logger.info("TEMPLATING FAILED: %s", templater_violations)
+
+        # Record time
+        time_dict = {"templating": time.monotonic() - t0}
+
+        return RenderedFile(templated_file, templater_violations, config, time_dict)
+
+    def render_file(self, fname: str, root_config: FluffConfig) -> RenderedFile:
+        """Load and render a file with relevant config."""
+        if self.formatter:
+            self.formatter.dispatch_path(fname)
+        # Load the raw file.
+        raw_file, config = self._load_raw_file_and_config(fname, root_config)
+        # Render the file
+        return self.render_string(raw_file, fname, config)
+
+    def parse_string(
+        self,
+        in_str: str,
+        fname: Optional[str] = None,
+        recurse: bool = True,
+        config: Optional[FluffConfig] = None,
+    ) -> ParsedString:
+        """Parse a string."""
+        violations: List[SQLBaseError] = []
+
+        # Dispatch the output for the template header (including the config diff)
+        if self.formatter:
+            self.formatter.dispatch_template_header(fname, self.config, config)
+
+        # Just use the local config from here:
+        config = config or self.config
+
+        # Scan the raw file for config commands.
+        config.process_raw_file_for_config(in_str)
+        rendered = self.render_string(in_str, fname, config)
+        violations += rendered.templater_violations
+
+        # Dispatch the output for the parse header
+        if self.formatter:
+            self.formatter.dispatch_parse_header(fname)
+
+        return self.parse_rendered(rendered, recurse=recurse)
 
     def fix(
         self,
@@ -423,8 +571,16 @@ def fix(
         templated_file: Optional[TemplatedFile] = None,
     ) -> Tuple[BaseSegment, List[SQLLintError]]:
         """Return the fixed tree and violations from lintfix when we're fixing."""
-        fixed_tree, violations = self.lint_fix(
-            tree, config, fix=True, fname=fname, templated_file=templated_file
+        config = config or self.config
+        rule_set = self.get_ruleset(config=config)
+        fixed_tree, violations = self.lint_fix_parsed(
+            tree,
+            config,
+            rule_set,
+            fix=True,
+            fname=fname,
+            templated_file=templated_file,
+            formatter=self.formatter,
         )
         return fixed_tree, violations
 
@@ -436,8 +592,16 @@ def lint(
         templated_file: Optional[TemplatedFile] = None,
     ) -> List[SQLLintError]:
         """Return just the violations from lintfix when we're only linting."""
-        _, violations = self.lint_fix(
-            tree, config, fix=False, fname=fname, templated_file=templated_file
+        config = config or self.config
+        rule_set = self.get_ruleset(config=config)
+        _, violations = self.lint_fix_parsed(
+            tree,
+            config,
+            rule_set,
+            fix=False,
+            fname=fname,
+            templated_file=templated_file,
+            formatter=self.formatter,
         )
         return violations
 
@@ -456,72 +620,12 @@ def lint_string(
         """
         # Sort out config, defaulting to the built in config if no override
         config = config or self.config
-
-        # Using the new parser, read the file object.
+        # Parse the string.
         parsed = self.parse_string(in_str=in_str, fname=fname, config=config)
-        time_dict = parsed.time_dict
-        vs = parsed.violations
-        tree = parsed.tree
-
-        # Look for comment segments which might indicate lines to ignore.
-        ignore_buff = []
-        if tree:
-            for comment in tree.recursive_crawl("comment"):
-                if comment.name == "inline_comment":
-                    ignore_entry = self.extract_ignore_from_comment(comment)
-                    if isinstance(ignore_entry, SQLParseError):
-                        vs.append(ignore_entry)
-                    elif ignore_entry:
-                        ignore_buff.append(ignore_entry)
-            if ignore_buff:
-                linter_logger.info("Parsed noqa directives from file: %r", ignore_buff)
-
-        if tree:
-            t0 = time.monotonic()
-            linter_logger.info("LINTING (%s)", fname)
-            tree, initial_linting_errors = self.lint_fix(
-                tree,
-                config=config,
-                fix=fix,
-                fname=fname,
-                templated_file=parsed.templated_file,
-            )
-            # Update the timing dict
-            t1 = time.monotonic()
-            time_dict["linting"] = t1 - t0
-
-            # We're only going to return the *initial* errors, rather
-            # than any generated during the fixing cycle.
-            vs += initial_linting_errors
-
-        # We process the ignore config here if appropriate
-        if config:
-            for violation in vs:
-                violation.ignore_if_in(config.get("ignore"))
-
-        linted_file = LintedFile(
-            fname,
-            vs,
-            time_dict,
-            tree,
-            ignore_mask=ignore_buff,
-            templated_file=parsed.templated_file,
-        )
-
-        # This is the main command line output from linting.
-        if self.formatter:
-            self.formatter.dispatch_file_violations(
-                fname, linted_file, only_fixable=fix
-            )
-
-        # Safety flag for unset dialects
-        if config.get("dialect") == "ansi" and linted_file.get_violations(
-            fixable=True if fix else None, types=SQLParseError
-        ):
-            if self.formatter:
-                self.formatter.dispatch_dialect_warning()
-
-        return linted_file
+        # Get rules as appropriate
+        rule_set = self.get_ruleset(config=config)
+        # Lint the file and return the LintedFile
+        return self.lint_parsed(parsed, rule_set, fix=fix, formatter=self.formatter)
 
     def paths_from_path(
         self,
@@ -712,11 +816,8 @@ def parse_path(
         for fname in self.paths_from_path(path):
             if self.formatter:
                 self.formatter.dispatch_path(path)
-            config = self.config.make_child_from_path(fname)
-            # Handle unicode issues gracefully
-            with open(
-                fname, "r", encoding="utf8", errors="backslashreplace"
-            ) as target_file:
-                yield self.parse_string(
-                    target_file.read(), fname=fname, recurse=recurse, config=config
-                )
+            # Load the file with the config and yield the result.
+            raw_file, config = self._load_raw_file_and_config(fname, self.config)
+            yield self.parse_string(
+                raw_file, fname=fname, recurse=recurse, config=config
+            )
diff --git a/src/sqlfluff/core/linter/runner.py b/src/sqlfluff/core/linter/runner.py
index c796c990dd0..262e11c1ea8 100644
--- a/src/sqlfluff/core/linter/runner.py
+++ b/src/sqlfluff/core/linter/runner.py
@@ -15,9 +15,6 @@
 import traceback
 from typing import Callable, List
 
-from sqlfluff.core.dialects import dialect_selector
-from sqlfluff.core.parser import Lexer
-
 
 linter_logger: logging.Logger = logging.getLogger("sqlfluff.linter")
 
@@ -33,6 +30,34 @@ def __init__(
         self.linter = linter
         self.config = config
 
+    pass_formatter = True
+
+    def iter_rendered(self, fnames):
+        """Iterate through rendered files ready for linting."""
+        for fname in fnames:
+            yield fname, self.linter.render_file(fname, self.config)
+
+    def iter_partials(self, fnames, fix: bool = False):
+        """Iterate through partials for linted files.
+
+        Generates filenames and objects which return LintedFiles.
+        """
+        for fname, rendered in self.iter_rendered(fnames):
+            # Generate a fresh ruleset
+            rule_set = self.linter.get_ruleset(config=rendered.config)
+            yield (
+                fname,
+                functools.partial(
+                    self.linter.lint_rendered,
+                    rendered,
+                    rule_set,
+                    fix,
+                    # Formatters may or may not be passed. They don't pickle
+                    # nicely so aren't appropriate in a multiprocessing world.
+                    self.linter.formatter if self.pass_formatter else None,
+                ),
+            )
+
     def run(self, fnames: List[str], fix: bool):
         """Run linting on the specified list of files."""
         raise NotImplementedError
@@ -44,20 +69,7 @@ def _init_global(cls, config):
         May be overridden by subclasses to apply global configuration, initialize
         logger state in child processes, etc.
         """
-        Lexer(dialect=config.get("dialect"))
-        dialect_selector(config.get("dialect"))
-
-    @classmethod
-    def _base_run(cls, config, linter, fname, fix):
-        """Core linting functionality."""
-        config = config.make_child_from_path(fname)
-        # Handle unicode issues gracefully
-        with open(
-            fname, "r", encoding="utf8", errors="backslashreplace"
-        ) as target_file:
-            return linter.lint_string(
-                target_file.read(), fname=fname, fix=fix, config=config
-            )
+        pass
 
     @staticmethod
     def _handle_lint_path_exception(fname, e):
@@ -77,9 +89,9 @@ class SequentialRunner(BaseRunner):
 
     def run(self, fnames, fix):
         """Sequential implementation."""
-        for fname in fnames:
+        for fname, partial in self.iter_partials(fnames, fix=fix):
             try:
-                yield self._base_run(self.config, self.linter, fname, fix)
+                yield partial()
             except Exception as e:
                 self._handle_lint_path_exception(fname, e)
 
@@ -89,31 +101,31 @@ class ParallelRunner(BaseRunner):
 
     POOL_TYPE: Callable
     MAP_FUNCTION_NAME: str
+    # Don't pass the formatter in a parallel world, they
+    # don't pickle well.
+    pass_formatter = False
 
     def __init__(self, linter, config, parallel):
         super().__init__(linter, config)
         self.parallel = parallel
 
     def run(self, fnames, fix):
-        """Parallel implementation."""
-        jobs = []
-        for fname in fnames:
-            jobs.append(
-                functools.partial(
-                    self._lint_path,
-                    type(self.linter),
-                    self.config,
-                    fname,
-                    fix,
-                )
-            )
+        """Parallel implementation.
+
+        Note that the partials are generated one at a time then
+        passed directly into the pool as they're ready. This means
+        the main thread can do the IO work while passing the parsing
+        and linting work out to the threads.
+        """
         with self._create_pool(
             self.parallel,
             self._init_global,
             (self.config,),
         ) as pool:
             try:
-                for lint_result in self._map(pool, self._apply, jobs):
+                for lint_result in self._map(
+                    pool, self._apply, self.iter_partials(fnames, fix=fix)
+                ):
                     if isinstance(lint_result, DelayedException):
                         try:
                             lint_result.reraise()
@@ -133,26 +145,17 @@ def run(self, fnames, fix):
                 print("Received keyboard interrupt. Cleaning up and shutting down...")
                 pool.terminate()
 
-    @classmethod
-    def _lint_path(cls, linter_cls, config, fname, fix=False):
-        """Lint a file in parallel mode.
-
-        Creates new Linter object to avoid potential issues, e.g.
-        - Multiprocessing-related pickling errors
-        - Shared state in general (to avoid race conditions)
-        """
-        try:
-            linter = linter_cls(config=config)
-            return cls._base_run(config, linter, fname, fix)
-        except Exception as e:
-            result = DelayedException(e)
-            result.fname = fname
-            return result
-
     @staticmethod
-    def _apply(f):
+    def _apply(partial_tuple):
         """Shim function used in parallel mode."""
-        return f()
+        # Unpack the tuple and ditch the filename in this case.
+        fname, partial = partial_tuple
+        try:
+            return partial()
+        # Capture any exceptions and return as delayed exception to handle
+        # in the main thread.
+        except Exception as e:
+            return DelayedException(e, fname=fname)
 
     @classmethod
     def _create_pool(cls, *args, **kwargs):
@@ -195,7 +198,7 @@ class MultiThreadRunner(ParallelRunner):
 class DelayedException(Exception):
     """Multiprocessing process pool uses this to propagate exceptions."""
 
-    def __init__(self, ee):
+    def __init__(self, ee, fname=None):
         self.ee = ee
         __, __, self.tb = sys.exc_info()
         self.fname = None
diff --git a/src/sqlfluff/core/rules/base.py b/src/sqlfluff/core/rules/base.py
index 2acb9efc6d4..4dca41d48c5 100644
--- a/src/sqlfluff/core/rules/base.py
+++ b/src/sqlfluff/core/rules/base.py
@@ -561,7 +561,7 @@ def eval(self, **kwargs):
         # Make sure we actually return the original class
         return cls
 
-    def get_rulelist(self, config):
+    def get_rulelist(self, config) -> List[BaseRule]:
         """Use the config to return the appropriate rules.
 
         We use the config both for whitelisting and blacklisting, but also
diff --git a/src/sqlfluff/core/templaters/base.py b/src/sqlfluff/core/templaters/base.py
index fb9781928c0..3be8a07ac12 100644
--- a/src/sqlfluff/core/templaters/base.py
+++ b/src/sqlfluff/core/templaters/base.py
@@ -88,8 +88,8 @@ class TemplatedFile:
     def __init__(
         self,
         source_str: str,
+        fname: str,
         templated_str: Optional[str] = None,
-        fname: Optional[str] = None,
         sliced_file: Optional[List[TemplatedFileSlice]] = None,
         raw_sliced: Optional[List[RawFileSlice]] = None,
     ):
@@ -123,7 +123,7 @@ def __init__(
     @classmethod
     def from_string(cls, raw):
         """Create TemplatedFile from a string."""
-        return cls(source_str=raw)
+        return cls(source_str=raw, fname="<string>")
 
     def __bool__(self):
         """Return true if there's a templated file."""
@@ -383,7 +383,7 @@ def __init__(self, **kwargs):
         """
 
     def process(
-        self, *, in_str: str, fname: Optional[str] = None, config=None, formatter=None
+        self, *, in_str: str, fname: str, config=None, formatter=None
     ) -> Tuple[Optional[TemplatedFile], list]:
         """Process a string and return a TemplatedFile.
 
diff --git a/src/sqlfluff/core/templaters/jinja.py b/src/sqlfluff/core/templaters/jinja.py
index ea3cd1b565a..7b208320d76 100644
--- a/src/sqlfluff/core/templaters/jinja.py
+++ b/src/sqlfluff/core/templaters/jinja.py
@@ -174,7 +174,7 @@ def _get_jinja_env():
         )
 
     def process(
-        self, *, in_str: str, fname: Optional[str] = None, config=None, formatter=None
+        self, *, in_str: str, fname: str, config=None, formatter=None
     ) -> Tuple[Optional[TemplatedFile], list]:
         """Process a string and return the new string.
 
diff --git a/src/sqlfluff/core/templaters/python.py b/src/sqlfluff/core/templaters/python.py
index eb0666cedd1..26c27c9db25 100644
--- a/src/sqlfluff/core/templaters/python.py
+++ b/src/sqlfluff/core/templaters/python.py
@@ -200,7 +200,7 @@ def get_context(self, fname=None, config=None):
         return live_context
 
     def process(
-        self, *, in_str: str, fname: Optional[str] = None, config=None, formatter=None
+        self, *, in_str: str, fname: str, config=None, formatter=None
     ) -> Tuple[Optional[TemplatedFile], list]:
         """Process a string and return a TemplatedFile.
 
diff --git a/test/cli/commands_test.py b/test/cli/commands_test.py
index dee62766ea2..aaede1b15a7 100644
--- a/test/cli/commands_test.py
+++ b/test/cli/commands_test.py
@@ -370,9 +370,10 @@ def test__cli__command_fix_stdin_logging_to_stderr(monkeypatch):
     perfect_sql = "select col from table"
 
     class MockLinter(sqlfluff.core.Linter):
-        def lint_fix(self, *args, **kwargs):
-            self._warn_unfixable("<FAKE CODE>")
-            return super().lint_fix(*args, **kwargs)
+        @classmethod
+        def lint_fix_parsed(cls, *args, **kwargs):
+            cls._warn_unfixable("<FAKE CODE>")
+            return super().lint_fix_parsed(*args, **kwargs)
 
     monkeypatch.setattr(sqlfluff.cli.commands, "Linter", MockLinter)
     result = invoke_assert_code(
diff --git a/test/core/linter_test.py b/test/core/linter_test.py
index 808999b94fe..a9e01bd27ef 100644
--- a/test/core/linter_test.py
+++ b/test/core/linter_test.py
@@ -240,23 +240,28 @@ def imap_unordered(self, *args, **kwargs):
     all([type(v) == SQLLintError for v in result.get_violations()])
 
 
-@patch("sqlfluff.core.linter.runner.BaseRunner._base_run")
-def test_lint_path_parallel_wrapper_exception(patched_base_run):
-    """Tests the error catching behavior of _lint_path_parallel_wrapper()."""
-    patched_base_run.side_effect = ValueError("Something unexpected happened")
-    result = runner.MultiProcessRunner._lint_path(Linter, FluffConfig(), "")
-    assert isinstance(result, runner.DelayedException)
-    with pytest.raises(ValueError):
-        result.reraise()
+@patch("sqlfluff.core.linter.Linter.lint_rendered")
+def test_lint_path_parallel_wrapper_exception(patched_lint):
+    """Tests the error catching behavior of _lint_path_parallel_wrapper().
+
+    Test on MultiThread runner because otherwise we have pickling issues.
+    """
+    patched_lint.side_effect = ValueError("Something unexpected happened")
+    for result in runner.MultiThreadRunner(Linter(), FluffConfig(), parallel=1).run(
+        ["test/fixtures/linter/passing.sql"], fix=False
+    ):
+        assert isinstance(result, runner.DelayedException)
+        with pytest.raises(ValueError):
+            result.reraise()
 
 
 @patch("sqlfluff.core.linter.runner.linter_logger")
-@patch("sqlfluff.core.Linter.lint_string")
+@patch("sqlfluff.core.linter.Linter.lint_rendered")
 def test__linter__linting_unexpected_error_handled_gracefully(
-    patched_lint_string, patched_logger
+    patched_lint, patched_logger
 ):
     """Test that an unexpected internal error is handled gracefully and returns the issue-surfacing file."""
-    patched_lint_string.side_effect = Exception("Something unexpected happened")
+    patched_lint.side_effect = Exception("Something unexpected happened")
     lntr = Linter()
     lntr.lint_paths(("test/fixtures/linter/passing.sql",))
     assert (
@@ -496,7 +501,7 @@ def test_linted_file_ignore_masked_violations(
         time_dict={},
         tree=None,
         ignore_mask=ignore_mask,
-        templated_file=TemplatedFile(""),
+        templated_file=TemplatedFile.from_string(""),
     )
     result = lf._ignore_masked_violations(violations)
     expected_violations = [v for i, v in enumerate(violations) if i in expected]
diff --git a/test/core/templaters/base_test.py b/test/core/templaters/base_test.py
index 714dd90e89c..71594a03dd0 100644
--- a/test/core/templaters/base_test.py
+++ b/test/core/templaters/base_test.py
@@ -45,7 +45,7 @@ def test__templater_raw():
     """Test the raw templater."""
     t = RawTemplater()
     instr = "SELECT * FROM {{blah}}"
-    outstr, _ = t.process(in_str=instr)
+    outstr, _ = t.process(in_str=instr, fname="test")
     assert instr == str(outstr)
 
 
@@ -143,7 +143,10 @@ def test__templated_file_get_line_pos_of_char_pos(
 ):
     """Test TemplatedFile.get_line_pos_of_char_pos."""
     file = TemplatedFile(
-        source_str=source_str, templated_str=templated_str, sliced_file=file_slices
+        source_str=source_str,
+        templated_str=templated_str,
+        sliced_file=file_slices,
+        fname="test",
     )
     res_line_no, res_line_pos = file.get_line_pos_of_char_pos(in_charpos)
     assert res_line_no == out_line_no
@@ -167,7 +170,9 @@ def test__templated_file_find_slice_indices_of_templated_pos(
     templated_position, inclusive, file_slices, sliced_idx_start, sliced_idx_stop
 ):
     """Test TemplatedFile._find_slice_indices_of_templated_pos."""
-    file = TemplatedFile(source_str="Dummy String", sliced_file=file_slices)
+    file = TemplatedFile(
+        source_str="Dummy String", sliced_file=file_slices, fname="test"
+    )
     res_start, res_stop = file._find_slice_indices_of_templated_pos(
         templated_position, inclusive=inclusive
     )
@@ -288,7 +293,10 @@ def test__templated_file_templated_slice_to_source_slice(
 ):
     """Test TemplatedFile.templated_slice_to_source_slice."""
     file = TemplatedFile(
-        source_str="Dummy String", sliced_file=file_slices, raw_sliced=raw_slices
+        source_str="Dummy String",
+        sliced_file=file_slices,
+        raw_sliced=raw_slices,
+        fname="test",
     )
     source_slice = file.templated_slice_to_source_slice(in_slice)
     literal_test = file.is_source_slice_literal(source_slice)
@@ -299,6 +307,7 @@ def test__templated_file_source_only_slices():
     """Test TemplatedFile.source_only_slices."""
     file = TemplatedFile(
         source_str=" Dummy String again ",  # NB: has length 20
+        fname="test",
         raw_sliced=[
             RawFileSlice("a" * 10, "literal", 0),
             RawFileSlice("b" * 7, "comment", 10),
diff --git a/test/core/templaters/jinja_test.py b/test/core/templaters/jinja_test.py
index 89f2d42f4ec..075f2d85c27 100644
--- a/test/core/templaters/jinja_test.py
+++ b/test/core/templaters/jinja_test.py
@@ -37,7 +37,7 @@
 def test__templater_jinja(instr, expected_outstr):
     """Test jinja templating and the treatment of whitespace."""
     t = JinjaTemplater(override_context=dict(blah="foo", condition="a < 10"))
-    outstr, _ = t.process(in_str=instr, config=FluffConfig())
+    outstr, _ = t.process(in_str=instr, fname="test", config=FluffConfig())
     assert str(outstr) == expected_outstr
 
 
@@ -45,7 +45,7 @@ def test__templater_jinja_error_variable():
     """Test missing variable error handling in the jinja templater."""
     t = JinjaTemplater(override_context=dict(blah="foo"))
     instr = JINJA_STRING
-    outstr, vs = t.process(in_str=instr, config=FluffConfig())
+    outstr, vs = t.process(in_str=instr, fname="test", config=FluffConfig())
     assert str(outstr) == "SELECT * FROM f, o, o WHERE \n\n"
     # Check we have violations.
     assert len(vs) > 0
@@ -57,7 +57,7 @@ def test__templater_jinja_error_syntax():
     """Test syntax problems in the jinja templater."""
     t = JinjaTemplater()
     instr = "SELECT {{foo} FROM jinja_error\n"
-    outstr, vs = t.process(in_str=instr, config=FluffConfig())
+    outstr, vs = t.process(in_str=instr, fname="test", config=FluffConfig())
     # Check we just skip templating.
     assert str(outstr) == instr
     # Check we have violations.
@@ -70,7 +70,7 @@ def test__templater_jinja_error_catatrophic():
     """Test error handling in the jinja templater."""
     t = JinjaTemplater(override_context=dict(blah=7))
     instr = JINJA_STRING
-    outstr, vs = t.process(in_str=instr, config=FluffConfig())
+    outstr, vs = t.process(in_str=instr, fname="test", config=FluffConfig())
     assert not outstr
     assert len(vs) > 0
 
diff --git a/test/core/templaters/python_test.py b/test/core/templaters/python_test.py
index 94d8f12dd2c..f9f9d73e69c 100644
--- a/test/core/templaters/python_test.py
+++ b/test/core/templaters/python_test.py
@@ -17,7 +17,7 @@ def test__templater_python():
     """Test the python templater."""
     t = PythonTemplater(override_context=dict(blah="foo"))
     instr = PYTHON_STRING
-    outstr, _ = t.process(in_str=instr)
+    outstr, _ = t.process(in_str=instr, fname="test")
     assert str(outstr) == "SELECT * FROM foo"
 
 
@@ -26,7 +26,7 @@ def test__templater_python_error():
     t = PythonTemplater(override_context=dict(noblah="foo"))
     instr = PYTHON_STRING
     with pytest.raises(SQLTemplaterError):
-        t.process(in_str=instr)
+        t.process(in_str=instr, fname="test")
 
 
 @pytest.mark.parametrize(
diff --git a/test/fixtures/linter/autofix/bigquery/004_templating/after.sql b/test/fixtures/linter/autofix/bigquery/004_templating/after.sql
index 66e7f8b0d50..74c90040823 100644
--- a/test/fixtures/linter/autofix/bigquery/004_templating/after.sql
+++ b/test/fixtures/linter/autofix/bigquery/004_templating/after.sql
@@ -10,16 +10,16 @@ raw_effect_sizes AS (
     SELECT
         COUNT(1) AS campaign_count,
         {{corr_states}}
-    {% for action in considered_actions %}
+        {% for action in considered_actions %}
         , SAFE_DIVIDE(SAFE_MULTIPLY(CORR({{metric}}_rate_su, {{action}}), STDDEV_POP({{metric}}_rate_su)), STDDEV_POP({{action}})) AS {{metric}}_{{action}}
-    {% endfor %}
+        {% endfor %}
     FROM
         `{{gcp_project}}.{{dataset}}.global_actions_states`
     GROUP BY
         {{corr_states}}
 ),
 
-  {% for action in considered_actions %}
+{% for action in considered_actions %}
 {{action}}_raw_effect_sizes AS (
     SELECT
         COUNT(1) AS campaign_count_{{action}},
@@ -41,15 +41,15 @@ raw_effect_sizes AS (
 new_raw_effect_sizes AS (
     SELECT
         {{corr_states}}
-    {% for action in considered_actions %}
+        {% for action in considered_actions %}
         , {{metric}}_{{action}}
         , campaign_count_{{action}}
-    {% endfor %}
+        {% endfor %}
     FROM
     {% for action in considered_actions %}
     {% if loop.first %}
     {{action}}_raw_effect_sizes
-    {% else %}
+        {% else %}
     JOIN
         {{action}}_raw_effect_sizes
         USING
@@ -62,11 +62,11 @@ imputed_effect_sizes AS (
     SELECT
         {{corr_states}}
         , o.campaign_count AS campaign_count
-    {% for action in considered_actions %}
+        {% for action in considered_actions %}
         , COALESCE(IF(IS_NAN(o.{{metric}}_{{action}}), 0, o.{{metric}}_{{action}}), 0) AS {{metric}}_{{action}}
         , COALESCE(IF(IS_NAN(n.{{metric}}_{{action}}), 0, n.{{metric}}_{{action}}), 0) AS new_{{metric}}_{{action}}
         , n.campaign_count_{{action}}
-{% endfor %}
+    {% endfor %}
     FROM
         raw_effect_sizes o
     JOIN


Code review comments:
diff_hunk: @@ -28,6 +29,19 @@ class NoQaDirective(NamedTuple):
     action: Optional[str]  # "enable", "disable", or "None"
 
 
+class RenderedFile(NamedTuple):
+    """An object to store the result of a templated file/string.
+
+    This is notable as it's the intermediate state between what happens
+    in the main thread and the child threads when running in parallel mode.
path: src/sqlfluff/core/linter/common.py
text: @user1:
```suggestion
    in the main process and the child processes when running in parallel mode.

```

diff_hunk: @@ -211,3 +207,28 @@ def __init__(self, ee):
     def reraise(self):
         """Reraise the encapsulated exception."""
         raise self.ee.with_traceback(self.tb)
+
+
+def get_runner(
+    linter,
+    config,
+    parallel: int,
+    allow_process_parallelism: bool = True,
+) -> BaseRunner:
+    """Generate a runner instance based on parallel and sytem configuration."""
+    if parallel > 1 and sys.version_info > (3, 7):
path: src/sqlfluff/core/linter/runner.py
text: @user1:
I'm thinking now that the version check should use `>=`, i.e.: `sys.version_info >= (3, 7)` because it's 3.6.x that is not supporting parallel.

@author:
I've updated the comparison. Do you have something I can link to and put in a comment about *why* 3.6 isn't supported. I had a quick google just now and couldn't work it out.

@user1:
TBH, I don't remember. I know it wasn't working. I recall thinking we could possibly get it working, but 3.6 is so old, was it worth the effort.

I'll see if I can reproduce the error on 3.6 again.

@user1:
This is the error I'm seeing on my Mac. My guess is that this could be a bug in older versions of the `multiprocessing` package, since the way we're using it is very "vanilla" AFAIK. I googled for the error message, and found [this Stack Overflow post](imap_unordered) which may be relevant, although it talks about using `Queue`. We're not explicitly using `Queue`, but presumably `multiprocessing.Pool.imap_unordered()` uses it internally to dispatch tasks to the workers.

```
sqlfluff lint --parallel 4
Traceback (most recent call last):
  File "/Users/bhart/.pyenv/versions/sqlfluff-3.6.4/bin/sqlfluff", line 11, in <module>
    load_entry_point('sqlfluff', 'console_scripts', 'sqlfluff')()
  File "/Users/bhart/.pyenv/versions/3.6.4/envs/sqlfluff-3.6.4/lib/python3.6/site-packages/click/core.py", line 1137, in __call__
    return self.main(*args, **kwargs)
  File "/Users/bhart/.pyenv/versions/3.6.4/envs/sqlfluff-3.6.4/lib/python3.6/site-packages/click/core.py", line 1062, in main
    rv = self.invoke(ctx)
  File "/Users/bhart/.pyenv/versions/3.6.4/envs/sqlfluff-3.6.4/lib/python3.6/site-packages/click/core.py", line 1668, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/bhart/.pyenv/versions/3.6.4/envs/sqlfluff-3.6.4/lib/python3.6/site-packages/click/core.py", line 1404, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/bhart/.pyenv/versions/3.6.4/envs/sqlfluff-3.6.4/lib/python3.6/site-packages/click/core.py", line 763, in invoke
    return __callback(*args, **kwargs)
  File "/Users/bhart/dev/sqlfluff/src/sqlfluff/cli/commands.py", line 346, in lint
    parallel=parallel,
  File "/Users/bhart/dev/sqlfluff/src/sqlfluff/core/linter/linter.py", line 710, in lint_paths
    parallel=parallel,
  File "/Users/bhart/dev/sqlfluff/src/sqlfluff/core/linter/linter.py", line 679, in lint_path
    for linted_file in runner.run(fnames, fix):
  File "/Users/bhart/dev/sqlfluff/src/sqlfluff/core/linter/runner.py", line 123, in run
    for lint_result in self._map(pool, self._apply, jobs):
  File "/Users/bhart/.pyenv/versions/3.6.4/lib/python3.6/multiprocessing/pool.py", line 735, in next
    raise value
multiprocessing.pool.MaybeEncodingError: Error sending result: 'LintedFile(path='/Users/bhart/dev/datascience_revenue_prediction/script/queries/test_query.sql', violations=[<SQLLintError: rule L003 pos:(2, 8), #fixes: 1, description: Indentation not hanging or a multiple of 4 spaces>, <SQLLintError: rule L003 pos:(4, 6), #fixes: 1, description: Indentation not hanging or a multiple of 4 spaces>, <SQLLintError: rule L036 pos:(1, 1), #fixes: 2, description: Select targets should be on a new line unless there is only one select target.>], time_dict={'templating': 0.015655905997846276, 'lexing': 0.0008951539930421859, 'parsing': 0.011481983005069196, 'linting': 0.03299699199851602}, tree=<FileSegment: ([L:  1, P:  1])>, ignore_mask=[], templated_file=<TemplatedFile>)'. Reason: 'TypeError("can't pickle _thread.RLock objects",)'
```