perf(format): format files in parallel and skip unchanged files

Makefile

@@ -161,6 +161,7 @@ E2E_DUCKDB_CLI_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/databricks \
 	tests/e2e/src/sqlbuild/cli/commands/main/dbt \
 	tests/e2e/src/sqlbuild/cli/commands/main/debug \
+	tests/e2e/src/sqlbuild/cli/commands/main/format/test_format_cache.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/format/test_format_example_projects.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/format/test_rule_fixes.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/format/test_token_preservation.py \

crates/sqlbuild-cli/src/argv/command_table.json

@@ -2385,6 +2385,18 @@
         "help": "only remove redundant typed-null SQL test fixture columns",
         "group": 1
        },
+       {
+        "kind": "store_true",
+        "options": [
+         "--no-cache"
+        ],
+        "dest": "no_cache",
+        "nargs": 0,
+        "const": true,
+        "default": false,
+        "help": "format every file instead of skipping files a previous run found clean",
+        "group": 1
+       },
        {
         "kind": "store",
         "dest": "format_paths",

crates/sqlbuild-cli/src/dispatch/constants.rs

@@ -629,14 +629,15 @@ pub(crate) const SKILLS_ROUTE: Route = routed(
     },
 );
 
-const FORMAT_ARGUMENTS: [Field; 9] = [
+const FORMAT_ARGUMENTS: [Field; 10] = [
     PROJECT_DIR,
     field("select", Source::FormatSelect),
     EXCLUDE,
     attr("check", "format_check"),
     attr("diff", "format_diff"),
     attr("fixtures_only", "format_fixtures_only"),
     attr("fix", "format_fix"),
+    same("no_cache"),
     JSON_OUTPUT,
     NO_COLOR,
 ];

crates/sqlbuild-python/src/bindings/_helpers/compile_driver/stage_work.rs

@@ -23,6 +23,8 @@ pub(crate) const LINEAGE_GRAPH_BUILDS: &str = "lineage_graph_builds";
 pub(crate) const PROJECT_GRAPHS: &str = "build_project_graph";
 pub(crate) const FORMAT_LINT_BATCHES: &str = "format_lint_batches";
 pub(crate) const FORMAT_LINT_FILES: &str = "format_lint_files";
+pub(crate) const FORMAT_CACHE_HITS: &str = "format_cache_hits";
+pub(crate) const FORMAT_CACHE_MISSES: &str = "format_cache_misses";
 pub(crate) const SQL_TEST_ARTIFACT_PLANS: &str = "plan_compiled_sql_test_artifacts";
 pub(crate) const SQL_TEST_MODEL_CHAINS: &str = "sql_test_model_chain_names_by_key";
 pub(crate) const BACKGROUND_SQL_TEST_PLANS: &str = "background_sql_test_plans";

crates/sqlbuild-python/src/bindings/_helpers/format_command/cache.rs

@@ -0,0 +1,84 @@
+//! Files already formatted and fault-free, kept in the shared native store to skip next time.
+
+use std::path::{Path, PathBuf};
+
+use pyo3::exceptions::PyOSError;
+use pyo3::prelude::{PyResult, Python};
+use sqlbuild_cache::digest::main::content_digest::content_digest;
+use sqlbuild_cache::digest::main::hex_digest::hex_digest;
+use sqlbuild_cache::digest::types::ContentDigest;
+use sqlbuild_cache::store::models::NativeStore;
+
+use crate::bindings::_helpers::cache::native_store::{
+    current_code_identity, open_store, save_store,
+};
+use crate::bindings::_helpers::compile_command::constants::TARGET_DIRECTORY_NAME;
+use crate::bindings::_helpers::frontier::constants::{
+    CACHE_DIRECTORY_NAME, COMPILER_CACHE_DIRECTORY_NAME, NATIVE_CACHE_NAMESPACE_SUFFIX,
+};
+
+/// The native store kind holding clean format results.
+const FORMAT_RESULT_STORE_KIND: &str = "format-results";
+/// The store file below the compile cache folder.
+pub(crate) const FORMAT_RESULT_STORE_FILE_NAME: &str = "format-results.bin";
+/// Bumped whenever what a clean result means changes; older stores open empty.
+const FORMAT_RESULT_STORE_VERSION: &str = "format-result-store-v1";
+/// The one value stored: formatting left the file unchanged and reported nothing about it.
+const CLEAN: &[u8] = b"clean";
+
+/// One run's view of the stored clean results, keyed by file, text and effective settings.
+pub(crate) struct FormatCache {
+    store: NativeStore,
+    path: PathBuf,
+    settings: String,
+}
+
+impl FormatCache {
+    /// The project's store for this SQLBuild build, or `None` when it cannot be read.
+    pub(crate) fn open(
+        py: Python<'_>,
+        project_dir: &Path,
+        settings: String,
+    ) -> PyResult<Option<Self>> {
+        let path: PathBuf = project_dir
+            .join(TARGET_DIRECTORY_NAME)
+            .join(CACHE_DIRECTORY_NAME)
+            .join(format!(
+                "{COMPILER_CACHE_DIRECTORY_NAME}{NATIVE_CACHE_NAMESPACE_SUFFIX}"
+            ))
+            .join(FORMAT_RESULT_STORE_FILE_NAME);
+        let environment: String = hex_digest(&content_digest(&[
+            FORMAT_RESULT_STORE_VERSION,
+            &current_code_identity(py)?,
+        ]));
+        match open_store(py, &path, FORMAT_RESULT_STORE_KIND, &environment) {
+            Ok(store) => Ok(Some(Self {
+                store,
+                path,
+                settings,
+            })),
+            Err(error) if error.is_instance_of::<PyOSError>(py) => Ok(None),
+            Err(error) => Err(error),
+        }
+    }
+
+    /// The key of one file: its project-relative path, its exact text and the run's settings.
+    pub(crate) fn key(&self, relative_path: &str, contents: &str) -> ContentDigest {
+        content_digest(&[relative_path, contents, &self.settings])
+    }
+
+    /// Whether a previous run found the file under `key` clean.
+    pub(crate) fn is_clean(&mut self, key: &ContentDigest) -> bool {
+        self.store.get(key) == Some(CLEAN)
+    }
+
+    /// Remember that the file under `key` formatted cleanly.
+    pub(crate) fn record_clean(&mut self, key: ContentDigest) {
+        self.store.put(key, CLEAN.to_vec());
+    }
+
+    /// Save newly recorded results; a store that cannot be written is left as it was.
+    pub(crate) fn save(&self, py: Python<'_>) {
+        let _ = save_store(py, &self.store, &self.path, &[]);
+    }
+}

crates/sqlbuild-python/src/bindings/_helpers/format_command/command.rs

@@ -41,6 +41,7 @@ struct FormatFlags {
     fix: bool,
     json_output: bool,
     use_color: bool,
+    no_cache: bool,
 }
 
 /// The adapter, selected files and discovered inputs a run formats with.
@@ -227,6 +228,7 @@ fn run_command(
         fixtures_only: flags.fixtures_only,
         write: !(flags.check || flags.diff),
         fix: flags.fix,
+        cache: !(flags.no_cache || flags.fix),
     };
     let (outcome, status, elapsed) = format_with_status(py, &request, flags.use_color)?;
     let root: PathBuf = project_dir.extract()?;
@@ -279,6 +281,7 @@ pub(crate) fn run_format_command<'py>(
                 "fix",
                 "json_output",
                 "no_color",
+                "no_cache",
             ],
         ),
     )?;
@@ -296,6 +299,7 @@ pub(crate) fn run_format_command<'py>(
         fix: arguments.flag("fix")?,
         json_output: arguments.flag("json_output")?,
         use_color,
+        no_cache: arguments.flag("no_cache")?,
     };
     let select: Vec<String> = string_tuple(&arguments.value("select")?)?;
     let exclude: Vec<String> = string_tuple(&arguments.value("exclude")?)?;

crates/sqlbuild-python/src/bindings/_helpers/format_command/files.rs

@@ -1,51 +1,37 @@
-//! Authored SQL files as `{Path: str}` dictionaries, and writing formatted files back.
+//! Writing formatted files back atomically, in the authored newline convention.
 
 use std::fs::{self, File, OpenOptions};
 use std::io::Write;
 use std::path::{Path, PathBuf};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use pyo3::exceptions::PyOSError;
-use pyo3::prelude::{Bound, PyAny, PyAnyMethods, PyModule, PyModuleMethods, PyResult, Python};
-use pyo3::types::{PyDict, PyDictMethods};
-use pyo3::{pyfunction, wrap_pyfunction};
+use pyo3::prelude::{Bound, PyAny, PyModule, PyModuleMethods, PyResult, Python};
+use pyo3::{IntoPyObject, pyfunction, wrap_pyfunction};
+use rayon::iter::{IndexedParallelIterator, IntoParallelRefIterator, ParallelIterator};
+
+use crate::bindings::_helpers::format_command::stages::on_format_workers;
 
 const LINE_FEED: &str = "\n";
 const CARRIAGE_RETURN_LINE_FEED: &str = "\r\n";
 
 /// One file the formatter would change: its path and its text before and after.
 pub(crate) struct FormatChange<'py> {
     pub(crate) path: Bound<'py, PyAny>,
+    /// The run's index of the changed file.
+    pub(crate) entry: usize,
     pub(crate) before: String,
     pub(crate) after: String,
 }
 
-/// The entries of a `{Path: value}` dictionary, sorted as Python sorts paths.
-pub(crate) fn sorted_items<'py>(
-    dictionary: &Bound<'py, PyDict>,
-) -> PyResult<Vec<(Bound<'py, PyAny>, Bound<'py, PyAny>)>> {
-    let mut items: Vec<(PathBuf, Bound<'py, PyAny>, Bound<'py, PyAny>)> = dictionary
-        .iter()
-        .map(|(key, value)| Ok((key.extract::<PathBuf>()?, key, value)))
-        .collect::<PyResult<_>>()?;
-    items.sort_by(|left, right| left.0.cmp(&right.0));
-    Ok(items
-        .into_iter()
-        .map(|(_, key, value)| (key, value))
-        .collect())
-}
-
-/// `{path: overrides.get(path, contents) for path, contents in files.items()}`.
-pub(crate) fn overlaid<'py>(
-    files: &Bound<'py, PyDict>,
-    overrides: &Bound<'py, PyDict>,
-) -> PyResult<Bound<'py, PyDict>> {
-    let merged: Bound<'py, PyDict> = PyDict::new(files.py());
-    for (path, contents) in files.iter() {
-        let value: Bound<'py, PyAny> = overrides.get_item(&path)?.unwrap_or(contents);
-        merged.set_item(path, value)?;
-    }
-    Ok(merged)
+/// One file a format run would leave in its final text.
+pub(crate) struct PendingWrite<'t> {
+    pub(crate) entry: usize,
+    pub(crate) path: &'t Path,
+    /// The text read, or empty for a file the run did not collect.
+    pub(crate) authored: &'t str,
+    pub(crate) contents: &'t str,
+    pub(crate) newline: &'static str,
 }
 
 /// Generated contents in the authored newline convention.
@@ -68,24 +54,54 @@ fn temporary_path(path: &Path) -> PathBuf {
     path.with_file_name(format!(".{name}.{}{nanos:x}", std::process::id()))
 }
 
-fn write_replacing(path: &Path, temporary: &Path, contents: &str) -> std::io::Result<()> {
+/// Write and sync `contents` to a new `temporary` beside `path`, with its permissions.
+fn stage_replacement(path: &Path, temporary: &Path, contents: &str) -> std::io::Result<()> {
     let permissions: fs::Permissions = fs::metadata(path)?.permissions();
     let mut handle: File = OpenOptions::new()
         .write(true)
         .create_new(true)
         .open(temporary)?;
     handle.set_permissions(permissions)?;
     handle.write_all(contents.as_bytes())?;
-    handle.sync_all()?;
-    fs::rename(temporary, path)
+    handle.sync_all()
+}
+
+fn write_failure(path: &Path, error: &std::io::Error) -> String {
+    format!("{}: {error}", path.display())
 }
 
 /// Replace a text file atomically, keeping its permissions.
 fn write_atomically(path: &Path, contents: &str) -> PyResult<()> {
     let temporary: PathBuf = temporary_path(path);
-    let written: std::io::Result<()> = write_replacing(path, &temporary, contents);
+    let written: std::io::Result<()> =
+        stage_replacement(path, &temporary, contents).and_then(|()| fs::rename(&temporary, path));
     let _ = fs::remove_file(&temporary);
-    written.map_err(|error| PyOSError::new_err(format!("{}: {error}", path.display())))
+    written.map_err(|error| PyOSError::new_err(write_failure(path, &error)))
+}
+
+/// Replace `files` in order after staging all in parallel; the earliest failure stops the rest.
+fn replace_in_order(files: &[(&Path, &str)]) -> Result<(), String> {
+    let temporaries: Vec<PathBuf> = files.iter().map(|(path, _)| temporary_path(path)).collect();
+    let staged: Vec<std::io::Result<()>> = on_format_workers(|| {
+        Ok(files
+            .par_iter()
+            .zip(temporaries.par_iter())
+            .map(|((path, contents), temporary)| stage_replacement(path, temporary, contents))
+            .collect())
+    })?;
+    let mut outcome: Result<(), String> = Ok(());
+    let mut renamed: usize = 0;
+    for (((path, _), temporary), staging) in files.iter().zip(&temporaries).zip(staged) {
+        if let Err(error) = staging.and_then(|()| fs::rename(temporary, path)) {
+            outcome = Err(write_failure(path, &error));
+            break;
+        }
+        renamed += 1;
+    }
+    for temporary in &temporaries[renamed..] {
+        let _ = fs::remove_file(temporary);
+    }
+    outcome
 }
 
 /// `write_atomically(*, path, contents)`: replace a text file atomically, keeping its mode.
@@ -99,45 +115,37 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     module.add_function(wrap_pyfunction!(write_file_atomically, module)?)
 }
 
-/// The changed files in path order, written in place when `write` is set.
+/// The changed files of `writes` in order, replaced in that order when `write` is set.
 pub(crate) fn persist_changes<'py>(
-    (files, updated): (&Bound<'py, PyDict>, &Bound<'py, PyDict>),
-    newlines: &Bound<'py, PyDict>,
+    py: Python<'py>,
+    writes: &[PendingWrite<'_>],
     write: bool,
 ) -> PyResult<Vec<FormatChange<'py>>> {
-    let mut changes: Vec<FormatChange<'py>> = Vec::new();
-    for (path, contents) in sorted_items(updated)? {
-        let newline: String = newlines
-            .get_item(&path)?
-            .map_or_else(|| Ok(LINE_FEED.to_owned()), |value| value.extract())?;
-        let authored: String = files
-            .get_item(&path)?
-            .map_or_else(|| Ok(String::new()), |value| value.extract())?;
-        let after: String = with_newline_style(&contents.extract::<String>()?, &newline);
-        let before: String = with_newline_style(&authored, &newline);
-        if after == before {
-            continue;
-        }
-        if write {
-            write_atomically(&path.extract::<PathBuf>()?, &after)?;
-        }
-        changes.push(FormatChange {
-            path,
-            before,
-            after,
-        });
-    }
-    Ok(changes)
-}
-
-/// `{Path: contents}` of the collected files, in collection order.
-pub(crate) fn file_dictionary<'py>(
-    py: Python<'py>,
-    files: Vec<(PathBuf, String)>,
-) -> PyResult<Bound<'py, PyDict>> {
-    let dictionary: Bound<'py, PyDict> = PyDict::new(py);
-    for (path, contents) in files {
-        dictionary.set_item(path, contents)?;
+    let changed: Vec<(&PendingWrite<'_>, String, String)> = writes
+        .iter()
+        .filter_map(|pending| {
+            let after: String = with_newline_style(pending.contents, pending.newline);
+            let before: String = with_newline_style(pending.authored, pending.newline);
+            (after != before).then_some((pending, before, after))
+        })
+        .collect();
+    if write {
+        let files: Vec<(&Path, &str)> = changed
+            .iter()
+            .map(|(pending, _, after)| (pending.path, after.as_str()))
+            .collect();
+        py.detach(|| replace_in_order(&files))
+            .map_err(PyOSError::new_err)?;
     }
-    Ok(dictionary)
+    changed
+        .into_iter()
+        .map(|(pending, before, after)| {
+            Ok(FormatChange {
+                path: pending.path.into_pyobject(py)?.into_any(),
+                entry: pending.entry,
+                before,
+                after,
+            })
+        })
+        .collect()
 }

crates/sqlbuild-python/src/bindings/_helpers/format_command/mod.rs

@@ -1,7 +1,9 @@
+pub(crate) mod cache;
 pub(crate) mod command;
 pub(crate) mod config;
 pub(crate) mod files;
 pub(crate) mod registration;
 pub(crate) mod report;
 pub(crate) mod run;
 pub(crate) mod selection;
+pub(crate) mod stages;