From a302aa37ed87447948a0bb53f7866ae9518da59b Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sun, 20 Sep 2026 18:08:37 +0100 Subject: [PATCH] fix: batch clone fingerprint finalization --- .../cli/commands/_helpers/clone/execution.py | 37 ++-- .../_helpers/clone/fingerprint_progress.py | 58 ++++++ .../cli/commands/_helpers/clone/output.py | 29 +++ .../classes/native_progress_projector.py | 1 + .../compiler/fingerprints/_helpers/sql.py | 58 ++++-- .../fingerprints/_helpers/write_many.py | 165 ++++++++++++++++++ .../compiler/fingerprints/constants.py | 2 + .../compiler/fingerprints/main/write_many.py | 40 +++++ src/sqlbuild/compiler/fingerprints/types.py | 9 + .../executor/clone/_helpers/fingerprinting.py | 120 ++++++++++--- .../executor/clone/main/fingerprinting.py | 3 + src/sqlbuild/executor/clone/types.py | 10 ++ .../compiler/fingerprints/main/_test_types.py | 7 + .../fingerprints/main/test_read_write.py | 60 +++++++ .../commands/_helpers/clone/_test_types.py | 27 +++ .../clone/test_fingerprint_progress.py | 56 ++++++ .../commands/_helpers/clone/test_output.py | 56 ++++++ .../classes/test_native_progress_projector.py | 36 ++++ .../compiler/fingerprints/main/_test_types.py | 17 ++ .../compiler/fingerprints/main/test_write.py | 105 +++++++++++ .../clone/_helpers/fingerprinting/helpers.py | 2 +- .../fingerprinting/test_fingerprinting.py | 27 +-- 22 files changed, 863 insertions(+), 62 deletions(-) create mode 100644 src/sqlbuild/cli/commands/_helpers/clone/fingerprint_progress.py create mode 100644 src/sqlbuild/compiler/fingerprints/_helpers/write_many.py create mode 100644 src/sqlbuild/compiler/fingerprints/main/write_many.py create mode 100644 src/sqlbuild/compiler/fingerprints/types.py create mode 100644 tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_fingerprint_progress.py diff --git a/src/sqlbuild/cli/commands/_helpers/clone/execution.py b/src/sqlbuild/cli/commands/_helpers/clone/execution.py index 9250fec285..56e20f2e63 100644 --- a/src/sqlbuild/cli/commands/_helpers/clone/execution.py +++ b/src/sqlbuild/cli/commands/_helpers/clone/execution.py @@ -6,6 +6,9 @@ from typing import TextIO from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.cli.commands._helpers.clone.fingerprint_progress import ( + create_clone_fingerprint_progress_reporter, +) from sqlbuild.cli.commands._helpers.clone.output import ( clone_relation_flow_text, render_clone_item_line, @@ -37,7 +40,7 @@ CloneItemResult, CloneSourceEntries, ) -from sqlbuild.executor.clone.types import CloneItemCallback +from sqlbuild.executor.clone.types import CloneFingerprintProgressReporter, CloneItemCallback from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle from sqlbuild.spec.contracts.main.resolve_target_config import resolve_target_config from sqlbuild.spec.contracts.models import SourceEntry @@ -104,18 +107,28 @@ def execute_clone_plan( ) finally: event_writer.close() - with OperationLifecycle(operation_kind="clone", operation_name="clone_finalization"): - copy_clone_fingerprints( - result=result, - origin_model_entries=preparation.pipeline_result.origin_model_entries, - destination_model_entries=preparation.pipeline_result.destination_model_entries, - origin_seed_entries=preparation.pipeline_result.origin_seed_entries, - destination_seed_entries=preparation.pipeline_result.destination_seed_entries, - adapter=invocation.adapter, - destination_connection=connection_context.destination_connection, - run_id=preparation.pipeline_result.destination_project.run_id, - query_change_tracking=preparation.pipeline_result.destination_project.settings.query_change_tracking, + fingerprint_progress: CloneFingerprintProgressReporter = ( + create_clone_fingerprint_progress_reporter( + stream=invocation.progress_stream, use_color=invocation.use_color ) + ) + try: + with OperationLifecycle(operation_kind="clone", operation_name="clone_finalization"): + copy_clone_fingerprints( + result=result, + origin_model_entries=preparation.pipeline_result.origin_model_entries, + destination_model_entries=preparation.pipeline_result.destination_model_entries, + origin_seed_entries=preparation.pipeline_result.origin_seed_entries, + destination_seed_entries=preparation.pipeline_result.destination_seed_entries, + adapter=invocation.adapter, + destination_connection=connection_context.destination_connection, + run_id=preparation.pipeline_result.destination_project.run_id, + query_change_tracking=preparation.pipeline_result.destination_project.settings.query_change_tracking, + on_progress=fingerprint_progress, + ) + except BaseException: + fingerprint_progress.write_interrupted() + raise return CloneRunOutcome(result=result, elapsed=time.monotonic() - clone_start) diff --git a/src/sqlbuild/cli/commands/_helpers/clone/fingerprint_progress.py b/src/sqlbuild/cli/commands/_helpers/clone/fingerprint_progress.py new file mode 100644 index 0000000000..39b77edcf6 --- /dev/null +++ b/src/sqlbuild/cli/commands/_helpers/clone/fingerprint_progress.py @@ -0,0 +1,58 @@ +"""Clone fingerprint finalization progress reporting.""" + +from __future__ import annotations + +from typing import TextIO + +from sqlbuild.cli.commands._helpers.clone.output import ( + render_clone_fingerprint_interrupted_line, + render_clone_fingerprint_progress_line, +) +from sqlbuild.executor.clone.types import CloneFingerprintProgressReporter + + +class _CloneFingerprintProgressReporter: + def __init__(self, *, stream: TextIO, use_color: bool) -> None: + self._stream: TextIO = stream + self._use_color: bool = use_color + self._confirmed: int = 0 + self._total: int = 0 + self._pending_identities: tuple[str, ...] = () + + def __call__(self, *, completed: int, total: int, pending_identities: tuple[str, ...]) -> None: + self._confirmed = completed + self._total = total + self._pending_identities = pending_identities + if completed == 0: + return + self._stream.write( + render_clone_fingerprint_progress_line( + completed=completed, + total=total, + use_color=self._use_color, + ) + + "\n" + ) + self._stream.flush() + + def write_interrupted(self) -> None: + if self._total <= self._confirmed: + return + self._stream.write( + render_clone_fingerprint_interrupted_line( + completed=self._confirmed, + total=self._total, + pending_identities=self._pending_identities, + use_color=self._use_color, + ) + + "\n" + ) + self._stream.flush() + + +def create_clone_fingerprint_progress_reporter( + *, stream: TextIO, use_color: bool +) -> CloneFingerprintProgressReporter: + """Create a stateful reporter for one clone finalization phase.""" + + return _CloneFingerprintProgressReporter(stream=stream, use_color=use_color) diff --git a/src/sqlbuild/cli/commands/_helpers/clone/output.py b/src/sqlbuild/cli/commands/_helpers/clone/output.py index dc8aaedba1..aceb081685 100644 --- a/src/sqlbuild/cli/commands/_helpers/clone/output.py +++ b/src/sqlbuild/cli/commands/_helpers/clone/output.py @@ -51,6 +51,35 @@ def clone_relation_flow_text( ) +def render_clone_fingerprint_progress_line(*, completed: int, total: int, use_color: bool) -> str: + """Render confirmed clone fingerprint propagation progress.""" + + style: CliStyle = CliStyle(use_color=use_color) + width: int = len(str(total)) + return ( + f" {'fingerprint':<10}" + f"[{completed:>{width}}/{total}] propagated {style.status(status='OK')}" + ) + + +def render_clone_fingerprint_interrupted_line( + *, completed: int, total: int, pending_identities: tuple[str, ...], use_color: bool +) -> str: + """Render bounded context for interrupted clone fingerprint propagation.""" + + style: CliStyle = CliStyle(use_color=use_color) + displayed: tuple[str, ...] = pending_identities[:10] + remaining_count: int = len(pending_identities) - len(displayed) + pending_text: str = ", ".join(displayed) + if remaining_count: + pending_text = f"{pending_text} (+{remaining_count} more)" + return ( + f" {'fingerprint':<10}" + f"[{completed}/{total}] {style.status(status='FAIL')} " + f"not confirmed: {pending_text}" + ) + + def clone_relation_display_name(relation: str) -> str: """Normalize ordinary identifiers while preserving quoted identifier case.""" diff --git a/src/sqlbuild/cli/progress/classes/native_progress_projector.py b/src/sqlbuild/cli/progress/classes/native_progress_projector.py index 8ac13afb43..85c8ea0564 100644 --- a/src/sqlbuild/cli/progress/classes/native_progress_projector.py +++ b/src/sqlbuild/cli/progress/classes/native_progress_projector.py @@ -17,6 +17,7 @@ from sqlbuild.runtime.observability.models import LifecycleEvent _VISIBLE_OPERATION_LABELS: Mapping[str, str] = { + "clone_finalization": "Clone finalization", "discovery_declaration_parse": "Declaration parsing", "discovery_filesystem_walk": "Filesystem discovery", "discovery_project_assembly": "Project assembly", diff --git a/src/sqlbuild/compiler/fingerprints/_helpers/sql.py b/src/sqlbuild/compiler/fingerprints/_helpers/sql.py index 78765015c1..967c68c9cd 100644 --- a/src/sqlbuild/compiler/fingerprints/_helpers/sql.py +++ b/src/sqlbuild/compiler/fingerprints/_helpers/sql.py @@ -3,7 +3,7 @@ from __future__ import annotations import base64 -from collections.abc import Callable +from collections.abc import Callable, Sequence from sqlbuild.adapter.contract.types import FrameworkType from sqlbuild.compiler.fingerprints.constants import ( @@ -113,25 +113,31 @@ def build_insert_sql( ) -> str: """Build a complete INSERT statement for appending one fingerprint row.""" - run_id: str = fingerprint.run_id - definition_hash: str = fingerprint.definition_hash - version_hash: str = fingerprint.version_hash - schema_fingerprint: str = fingerprint.schema_fingerprint - ts: str = fingerprint.ts.isoformat() - qualified_name: str = build_qualified_table_name( + return build_insert_many_sql( database=database, schema=schema, + fingerprints=(fingerprint,), render_qualified_name=render_qualified_name, ) - encoded_definition: str = _encode_definition_storage(fingerprint.definition).replace("'", "''") - encoded_metadata_json: str = _encode_definition_storage(fingerprint.metadata_json).replace( - "'", "''" + + +def build_insert_many_sql( + *, + database: str | None, + schema: str, + fingerprints: Sequence[Fingerprint], + render_qualified_name: Callable[..., str | None], +) -> str: + """Build one append-only INSERT for a non-empty fingerprint batch.""" + + if not fingerprints: + raise FingerprintInputError("fingerprint insert requires at least one row") + qualified_name: str = build_qualified_table_name( + database=database, + schema=schema, + render_qualified_name=render_qualified_name, ) - node_type_literal: str = _required_string_literal(fingerprint.node_type) - node_name_literal: str = _required_string_literal(fingerprint.node_name) - target_database_literal: str = _optional_string_literal(fingerprint.target_database) - target_schema_literal: str = _optional_string_literal(fingerprint.target_schema) - target_name_literal: str = _optional_string_literal(fingerprint.target_name) + rows: str = ", ".join(_fingerprint_values(fingerprint) for fingerprint in fingerprints) return ( f"INSERT INTO {qualified_name} (" f"{COLUMN_NODE_TYPE}, " @@ -146,7 +152,27 @@ def build_insert_sql( f"{COLUMN_DEFINITION_B64}, " f"{COLUMN_METADATA_JSON_B64}, " f"{COLUMN_TIMESTAMP}" - f") VALUES (" + f") VALUES {rows}" + ) + + +def _fingerprint_values(fingerprint: Fingerprint) -> str: + run_id: str = fingerprint.run_id + definition_hash: str = fingerprint.definition_hash + version_hash: str = fingerprint.version_hash + schema_fingerprint: str = fingerprint.schema_fingerprint + ts: str = fingerprint.ts.isoformat() + encoded_definition: str = _encode_definition_storage(fingerprint.definition).replace("'", "''") + encoded_metadata_json: str = _encode_definition_storage(fingerprint.metadata_json).replace( + "'", "''" + ) + node_type_literal: str = _required_string_literal(fingerprint.node_type) + node_name_literal: str = _required_string_literal(fingerprint.node_name) + target_database_literal: str = _optional_string_literal(fingerprint.target_database) + target_schema_literal: str = _optional_string_literal(fingerprint.target_schema) + target_name_literal: str = _optional_string_literal(fingerprint.target_name) + return ( + "(" f"{node_type_literal}, " f"{node_name_literal}, " f"{target_database_literal}, " diff --git a/src/sqlbuild/compiler/fingerprints/_helpers/write_many.py b/src/sqlbuild/compiler/fingerprints/_helpers/write_many.py new file mode 100644 index 0000000000..2fdd597deb --- /dev/null +++ b/src/sqlbuild/compiler/fingerprints/_helpers/write_many.py @@ -0,0 +1,165 @@ +"""Bounded append-only fingerprint batch writes.""" + +from __future__ import annotations + +import logging +import time +from collections.abc import Callable, Sequence +from typing import Any + +from sqlbuild.adapter.contract.types import AdapterExecute, FrameworkType +from sqlbuild.compiler.fingerprints._helpers.sql import ( + build_create_table_sql, + build_insert_many_sql, + build_insert_sql, +) +from sqlbuild.compiler.fingerprints.constants import ( + FINGERPRINT_WRITE_ATTEMPTS, + FINGERPRINT_WRITE_BATCH_MAX_BYTES, + FINGERPRINT_WRITE_BATCH_SIZE, + FINGERPRINT_WRITE_RETRY_BASE_SECONDS, +) +from sqlbuild.compiler.fingerprints.exceptions import FingerprintInputError +from sqlbuild.compiler.fingerprints.models import Fingerprint +from sqlbuild.compiler.fingerprints.types import FingerprintWriteProgress + + +def write_fingerprints_impl( + *, + connection: Any, + execute: AdapterExecute[Any, Any], + database: str | None, + schema: str, + fingerprints: Sequence[Fingerprint], + render_qualified_name: Callable[..., str | None], + render_framework_type: Callable[[FrameworkType], str], + render_create_table_sql: Callable[..., str] | None, + render_create_index_sqls: Callable[..., tuple[str, ...]] | None, + on_progress: FingerprintWriteProgress | None, +) -> None: + if not fingerprints: + return + create_sql: str = ( + render_create_table_sql(database=database, schema=schema) + if render_create_table_sql is not None + else build_create_table_sql( + database=database, + schema=schema, + render_qualified_name=render_qualified_name, + render_framework_type=render_framework_type, + ) + ) + index_sqls: tuple[str, ...] = ( + render_create_index_sqls(database=database, schema=schema) + if render_create_index_sqls is not None + else () + ) + _ = _execute_with_retry( + connection=connection, + execute=execute, + sqls=(create_sql, *index_sqls), + fingerprints=fingerprints, + ) + batches: tuple[tuple[Fingerprint, ...], ...] = _fingerprint_batches( + database=database, + schema=schema, + fingerprints=fingerprints, + render_qualified_name=render_qualified_name, + batch_size=FINGERPRINT_WRITE_BATCH_SIZE, + batch_max_bytes=FINGERPRINT_WRITE_BATCH_MAX_BYTES, + ) + completed_count: int = 0 + batch: tuple[Fingerprint, ...] + for batch in batches: + insert_sql: str = build_insert_many_sql( + database=database, + schema=schema, + fingerprints=batch, + render_qualified_name=render_qualified_name, + ) + _ = _execute_with_retry( + connection=connection, + execute=execute, + sqls=(insert_sql,), + fingerprints=batch, + ) + completed_count += len(batch) + if on_progress is not None: + on_progress(completed=completed_count, total=len(fingerprints)) + + +def _fingerprint_batches( + *, + database: str | None, + schema: str, + fingerprints: Sequence[Fingerprint], + render_qualified_name: Callable[..., str | None], + batch_size: int, + batch_max_bytes: int, +) -> tuple[tuple[Fingerprint, ...], ...]: + if batch_size < 1: + raise FingerprintInputError("fingerprint batch size must be positive") + if batch_max_bytes < 1: + raise FingerprintInputError("fingerprint batch byte limit must be positive") + batches: list[tuple[Fingerprint, ...]] = [] + current_batch: list[Fingerprint] = [] + current_bytes: int = 0 + fingerprint: Fingerprint + for fingerprint in fingerprints: + single_row_bytes: int = len( + build_insert_sql( + database=database, + schema=schema, + fingerprint=fingerprint, + render_qualified_name=render_qualified_name, + ).encode("utf-8") + ) + if current_batch and ( + len(current_batch) >= batch_size or current_bytes + single_row_bytes > batch_max_bytes + ): + batches.append(tuple(current_batch)) + current_batch = [] + current_bytes = 0 + current_batch.append(fingerprint) + current_bytes += single_row_bytes + if current_batch: + batches.append(tuple(current_batch)) + return tuple(batches) + + +def _execute_with_retry( + *, + connection: Any, + execute: AdapterExecute[Any, Any], + sqls: Sequence[str], + fingerprints: Sequence[Fingerprint], +) -> None: + attempt: int + for attempt in range(FINGERPRINT_WRITE_ATTEMPTS): + try: + sql: str + for sql in sqls: + _ = execute(connection=connection, sql=sql) + return + except Exception as error: + if attempt + 1 == FINGERPRINT_WRITE_ATTEMPTS: + raise + _log_write_retry(fingerprints=fingerprints, attempt=attempt, error=error) + time.sleep(FINGERPRINT_WRITE_RETRY_BASE_SECONDS * (attempt + 1)) + + +def _log_write_retry( + *, fingerprints: Sequence[Fingerprint], attempt: int, error: Exception +) -> None: + first: Fingerprint = fingerprints[0] + logging.getLogger("sqlbuild.fingerprints").warning( + "fingerprint batch write attempt %s/%s failed for %s row(s), starting with %s '%s' " + "(likely concurrent-writer conflict); retrying: %s: %s", + attempt + 1, + FINGERPRINT_WRITE_ATTEMPTS, + len(fingerprints), + first.node_type, + first.node_name, + type(error).__name__, + error, + ) diff --git a/src/sqlbuild/compiler/fingerprints/constants.py b/src/sqlbuild/compiler/fingerprints/constants.py index 2fb473e3dc..2add2ac5c5 100644 --- a/src/sqlbuild/compiler/fingerprints/constants.py +++ b/src/sqlbuild/compiler/fingerprints/constants.py @@ -6,6 +6,8 @@ FINGERPRINT_WRITE_ATTEMPTS: int = 5 FINGERPRINT_WRITE_RETRY_BASE_SECONDS: float = 0.05 +FINGERPRINT_WRITE_BATCH_SIZE: int = 50 +FINGERPRINT_WRITE_BATCH_MAX_BYTES: int = 512_000 COLUMN_NODE_TYPE: str = "node_type" COLUMN_NODE_NAME: str = "node_name" diff --git a/src/sqlbuild/compiler/fingerprints/main/write_many.py b/src/sqlbuild/compiler/fingerprints/main/write_many.py new file mode 100644 index 0000000000..cac5b850c4 --- /dev/null +++ b/src/sqlbuild/compiler/fingerprints/main/write_many.py @@ -0,0 +1,40 @@ +"""Batched fingerprint write entrypoint.""" + +from __future__ import annotations + +from collections.abc import Callable, Sequence +from typing import Any + +from sqlbuild.adapter.contract.types import AdapterExecute, FrameworkType +from sqlbuild.compiler.fingerprints._helpers.write_many import write_fingerprints_impl +from sqlbuild.compiler.fingerprints.models import Fingerprint +from sqlbuild.compiler.fingerprints.types import FingerprintWriteProgress + + +def write_fingerprints( + *, + connection: Any, + execute: AdapterExecute[Any, Any], + database: str | None, + schema: str, + fingerprints: Sequence[Fingerprint], + render_qualified_name: Callable[..., str | None], + render_framework_type: Callable[[FrameworkType], str], + render_create_table_sql: Callable[..., str] | None = None, + render_create_index_sqls: Callable[..., tuple[str, ...]] | None = None, + on_progress: FingerprintWriteProgress | None = None, +) -> None: + """Append fingerprints using one table setup and bounded multi-row inserts.""" + + _ = write_fingerprints_impl( + connection=connection, + execute=execute, + database=database, + schema=schema, + fingerprints=fingerprints, + render_qualified_name=render_qualified_name, + render_framework_type=render_framework_type, + render_create_table_sql=render_create_table_sql, + render_create_index_sqls=render_create_index_sqls, + on_progress=on_progress, + ) diff --git a/src/sqlbuild/compiler/fingerprints/types.py b/src/sqlbuild/compiler/fingerprints/types.py new file mode 100644 index 0000000000..88c3256fcd --- /dev/null +++ b/src/sqlbuild/compiler/fingerprints/types.py @@ -0,0 +1,9 @@ +"""Fingerprint callable contracts.""" + +from __future__ import annotations + +from typing import Protocol + + +class FingerprintWriteProgress(Protocol): + def __call__(self, *, completed: int, total: int) -> None: ... diff --git a/src/sqlbuild/executor/clone/_helpers/fingerprinting.py b/src/sqlbuild/executor/clone/_helpers/fingerprinting.py index 163cf2f82c..8c45c66e3f 100644 --- a/src/sqlbuild/executor/clone/_helpers/fingerprinting.py +++ b/src/sqlbuild/executor/clone/_helpers/fingerprinting.py @@ -2,6 +2,7 @@ from __future__ import annotations +from collections import defaultdict from collections.abc import Sequence from dataclasses import replace from datetime import UTC, datetime @@ -16,12 +17,16 @@ NODE_TYPE_SEED, ) from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints -from sqlbuild.compiler.fingerprints.main.write import write_fingerprint +from sqlbuild.compiler.fingerprints.main.write_many import write_fingerprints from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet from sqlbuild.compiler.planner.models import ModelPlanEntry, SeedPlanEntry from sqlbuild.compiler.planner.types import MaterializationType from sqlbuild.executor.clone.models import CloneExecutionResult -from sqlbuild.executor.clone.types import CloneAction, CloneStatus +from sqlbuild.executor.clone.types import ( + CloneAction, + CloneFingerprintProgressCallback, + CloneStatus, +) def copy_clone_fingerprints( @@ -35,6 +40,7 @@ def copy_clone_fingerprints( destination_connection: Any, run_id: str, query_change_tracking: bool, + on_progress: CloneFingerprintProgressCallback | None = None, ) -> None: """Copy source fingerprints for successfully cloned/copied tables and seeds.""" @@ -49,7 +55,7 @@ def copy_clone_fingerprints( if not successful_names: return - _copy_entry_fingerprints( + copied_seed_fingerprints: tuple[Fingerprint, ...] = _collect_entry_fingerprints( node_type=NODE_TYPE_SEED, origin_entries=origin_seed_entries, destination_entries=destination_seed_entries, @@ -68,18 +74,27 @@ def copy_clone_fingerprints( for entry in destination_model_entries if entry.materialization_type != MaterializationType.VIEW ) - _copy_entry_fingerprints( - node_type=NODE_TYPE_MODEL, - origin_entries=table_origin_entries, - destination_entries=table_destination_entries, - successful_names=successful_names, + copied_fingerprints: tuple[Fingerprint, ...] = ( + copied_seed_fingerprints + + _collect_entry_fingerprints( + node_type=NODE_TYPE_MODEL, + origin_entries=table_origin_entries, + destination_entries=table_destination_entries, + successful_names=successful_names, + adapter=adapter, + connection=destination_connection, + run_id=run_id, + ) + ) + _write_collected_fingerprints( + fingerprints=copied_fingerprints, adapter=adapter, connection=destination_connection, - run_id=run_id, + on_progress=on_progress, ) -def _copy_entry_fingerprints( +def _collect_entry_fingerprints( *, node_type: str, origin_entries: Sequence[ModelPlanEntry | SeedPlanEntry], @@ -88,20 +103,28 @@ def _copy_entry_fingerprints( adapter: BaseAdapter, connection: Any, run_id: str, -) -> None: +) -> tuple[Fingerprint, ...]: origin_by_name: dict[str, ModelPlanEntry | SeedPlanEntry] = { entry.name: entry for entry in origin_entries } destination_by_name: dict[str, ModelPlanEntry | SeedPlanEntry] = { entry.name: entry for entry in destination_entries } + relevant_origin_entries: tuple[ModelPlanEntry | SeedPlanEntry, ...] = tuple( + entry + for entry in origin_entries + if entry.name in successful_names + and (destination_entry := destination_by_name.get(entry.name)) is not None + and destination_entry.destination.schema is not None + ) origin_fingerprint_sets: dict[tuple[str | None, str], FingerprintSet] = ( _read_origin_fingerprint_sets( - origin_entries=origin_entries, + origin_entries=relevant_origin_entries, adapter=adapter, connection=connection, ) ) + copied_fingerprints: list[Fingerprint] = [] name: str for name in sorted(successful_names): origin_entry: ModelPlanEntry | SeedPlanEntry | None = origin_by_name.get(name) @@ -115,25 +138,76 @@ def _copy_entry_fingerprints( ) if fingerprint is None or destination_entry.destination.schema is None: continue - copied_fingerprint: Fingerprint = replace( - fingerprint, - target_database=destination_entry.destination.database, - target_schema=destination_entry.destination.schema, - target_name=destination_entry.destination.name, - run_id=run_id, - ts=datetime.now(tz=UTC), + copied_fingerprints.append( + replace( + fingerprint, + target_database=destination_entry.destination.database, + target_schema=destination_entry.destination.schema, + target_name=destination_entry.destination.name, + run_id=run_id, + ts=datetime.now(tz=UTC), + ) ) - write_fingerprint( + return tuple(copied_fingerprints) + + +def _write_collected_fingerprints( + *, + fingerprints: tuple[Fingerprint, ...], + adapter: BaseAdapter, + connection: Any, + on_progress: CloneFingerprintProgressCallback | None, +) -> None: + grouped: dict[tuple[str | None, str], list[Fingerprint]] = defaultdict(list) + fingerprint: Fingerprint + for fingerprint in fingerprints: + if fingerprint.target_schema is not None: + grouped[(fingerprint.target_database, fingerprint.target_schema)].append(fingerprint) + ordered_groups: tuple[tuple[tuple[str | None, str], tuple[Fingerprint, ...]], ...] = tuple( + (key, tuple(grouped[key])) + for key in sorted(grouped, key=lambda item: (item[0] or "", item[1])) + ) + ordered_fingerprint_list: list[Fingerprint] = [] + ordered_group: tuple[tuple[str | None, str], tuple[Fingerprint, ...]] + for ordered_group in ordered_groups: + ordered_fingerprint_list.extend(ordered_group[1]) + ordered_fingerprints: tuple[Fingerprint, ...] = tuple(ordered_fingerprint_list) + pending_identities: tuple[str, ...] = tuple( + f"{fingerprint.node_type}:{fingerprint.node_name}" for fingerprint in ordered_fingerprints + ) + total_count: int = len(ordered_fingerprints) + completed_before_group: int = 0 + if on_progress is not None: + on_progress(completed=0, total=total_count, pending_identities=pending_identities) + namespace: tuple[str | None, str] + group: tuple[Fingerprint, ...] + for namespace, group in ordered_groups: + + def _on_group_progress( + *, completed: int, total: int, group_offset: int = completed_before_group + ) -> None: + del total + confirmed: int = group_offset + completed + if on_progress is not None: + on_progress( + completed=confirmed, + total=total_count, + pending_identities=pending_identities[confirmed:], + ) + + write_fingerprints( connection=connection, execute=adapter.execute, - database=destination_entry.destination.database, - schema=destination_entry.destination.schema, - fingerprint=copied_fingerprint, + database=namespace[0], + schema=namespace[1], + fingerprints=group, render_qualified_name=adapter.render_qualified_name, render_framework_type=adapter.render_framework_type, render_create_table_sql=adapter.render_create_fingerprint_table_sql, render_create_index_sqls=adapter.render_create_fingerprint_index_sqls, + on_progress=_on_group_progress, ) + completed_before_group += len(group) def _read_origin_fingerprint_sets( diff --git a/src/sqlbuild/executor/clone/main/fingerprinting.py b/src/sqlbuild/executor/clone/main/fingerprinting.py index de6dc28fd6..8de777f8f9 100644 --- a/src/sqlbuild/executor/clone/main/fingerprinting.py +++ b/src/sqlbuild/executor/clone/main/fingerprinting.py @@ -8,6 +8,7 @@ from sqlbuild.compiler.planner.models import ModelPlanEntry, SeedPlanEntry from sqlbuild.executor.clone._helpers.fingerprinting import copy_clone_fingerprints as _copy from sqlbuild.executor.clone.models import CloneExecutionResult +from sqlbuild.executor.clone.types import CloneFingerprintProgressCallback def copy_clone_fingerprints( @@ -21,6 +22,7 @@ def copy_clone_fingerprints( destination_connection: Any, run_id: str, query_change_tracking: bool, + on_progress: CloneFingerprintProgressCallback | None = None, ) -> None: _ = _copy( result=result, @@ -32,4 +34,5 @@ def copy_clone_fingerprints( destination_connection=destination_connection, run_id=run_id, query_change_tracking=query_change_tracking, + on_progress=on_progress, ) diff --git a/src/sqlbuild/executor/clone/types.py b/src/sqlbuild/executor/clone/types.py index c45506cdac..7f498ac274 100644 --- a/src/sqlbuild/executor/clone/types.py +++ b/src/sqlbuild/executor/clone/types.py @@ -13,6 +13,16 @@ class CloneItemCallback(Protocol): def __call__(self, *, index: int, total: int, item: CloneItemResult) -> None: ... +class CloneFingerprintProgressCallback(Protocol): + def __call__( + self, *, completed: int, total: int, pending_identities: tuple[str, ...] + ) -> None: ... + + +class CloneFingerprintProgressReporter(CloneFingerprintProgressCallback, Protocol): + def write_interrupted(self) -> None: ... + + class CloneStartCallback(Protocol): def __call__( self, *, origin_target_name: str, destination_target_name: str, total: int diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py index b6a578ce2c..4015e53c80 100644 --- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/_test_types.py @@ -35,6 +35,13 @@ class WriteCreatesTableTestCase: expected_table_exists: bool +@dataclass(frozen=True) +class BatchWriteTestCase: + description: str + row_count: int + expected_last_definition: str + + @dataclass(frozen=True) class LatestResolutionTestCase: description: str diff --git a/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py b/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py index 948eb36175..2259c049d9 100644 --- a/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py +++ b/tests/integration/src/sqlbuild/compiler/fingerprints/main/test_read_write.py @@ -12,8 +12,10 @@ from sqlbuild.compiler.fingerprints.constants import FINGERPRINT_TABLE_NAME from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints from sqlbuild.compiler.fingerprints.main.write import write_fingerprint +from sqlbuild.compiler.fingerprints.main.write_many import write_fingerprints from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet from tests.integration.src.sqlbuild.compiler.fingerprints.main._test_types import ( + BatchWriteTestCase, InvalidDefinitionStorageTestCase, LatestResolutionTestCase, OldFingerprintSchemaTestCase, @@ -294,6 +296,64 @@ def test_given_fingerprints_when_writing_and_reading_then_returns_expected( assert result.fingerprints_by_identity[identity_key].definition_hash == expected_hash +@pytest.mark.parametrize( + "test_case", + ( + BatchWriteTestCase( + description="writes every row across multiple batches", + row_count=101, + expected_last_definition="SELECT 100 AS order_id", + ), + ), + ids=lambda case: case.description, +) +def test_given_many_fingerprints_when_batch_writing_then_all_rows_are_readable( + test_case: BatchWriteTestCase, + connection: Any, + execute: Any, +) -> None: + fingerprints: tuple[Fingerprint, ...] = tuple( + Fingerprint( + node_type="model", + node_name=f"orders_{index}", + target_database=None, + target_schema="test_schema", + target_name=f"orders_{index}", + run_id="batch_run", + definition_hash=f"hash_{index}", + version_hash=f"version_{index}", + schema_fingerprint=f"schema_{index}", + definition=f"SELECT {index} AS order_id", + ts=datetime(2026, 1, 15, 12, 0, index % 60), + ) + for index in range(test_case.row_count) + ) + + write_fingerprints( + connection=connection, + execute=execute, + database=None, + schema="test_schema", + fingerprints=fingerprints, + render_qualified_name=RENDER_QUALIFIED_NAME, + render_framework_type=RENDER_FRAMEWORK_TYPE, + ) + result: FingerprintSet = read_latest_fingerprints( + connection=connection, + execute=execute, + table_exists=True, + database=None, + schema="test_schema", + render_qualified_name=RENDER_QUALIFIED_NAME, + render_read_latest_sql=RENDER_READ_LATEST_SQL, + ) + + assert len(result.fingerprints) == test_case.row_count + assert result.fingerprints[f"orders_{test_case.row_count - 1}"].definition == ( + test_case.expected_last_definition + ) + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/_test_types.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/_test_types.py index 72cfae182f..ea95a18383 100644 --- a/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/_test_types.py +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/_test_types.py @@ -41,6 +41,33 @@ class CloneConnectionLifecycleTestCase: expected_destination_config: dict[str, object] +@dataclass(frozen=True) +class RenderCloneFingerprintProgressTestCase: + description: str + completed: int + total: int + expected_line: str + + +@dataclass(frozen=True) +class RenderCloneFingerprintInterruptedTestCase: + description: str + completed: int + total: int + pending_identities: tuple[str, ...] + expected_fragments: tuple[str, ...] + unexpected_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class CloneFingerprintProgressReporterTestCase: + description: str + completed: int + total: int + pending_identities: tuple[str, ...] + expected_fragments: tuple[str, ...] + + @dataclass(frozen=True) class VirtualCloneLifecycleCase: description: str diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_fingerprint_progress.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_fingerprint_progress.py new file mode 100644 index 0000000000..fb959f4d0e --- /dev/null +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_fingerprint_progress.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from io import StringIO + +import pytest + +from sqlbuild.cli.commands._helpers.clone.fingerprint_progress import ( + create_clone_fingerprint_progress_reporter, +) +from sqlbuild.executor.clone.types import CloneFingerprintProgressReporter +from tests.unit.src.sqlbuild.cli.commands._helpers.clone._test_types import ( + CloneFingerprintProgressReporterTestCase, +) + + +@pytest.mark.parametrize( + "test_case", + ( + CloneFingerprintProgressReporterTestCase( + description="reports completed batch and unconfirmed identities after interruption", + completed=50, + total=52, + pending_identities=("model:orders_50", "model:orders_51"), + expected_fragments=( + "[50/52] propagated OK", + "[50/52] FAIL not confirmed: model:orders_50, model:orders_51", + ), + ), + ), + ids=lambda case: case.description, +) +def test_given_partial_progress_when_interrupted_then_reporter_names_unconfirmed_fingerprints( + test_case: CloneFingerprintProgressReporterTestCase, +) -> None: + stream: StringIO = StringIO() + reporter: CloneFingerprintProgressReporter = create_clone_fingerprint_progress_reporter( + stream=stream, + use_color=False, + ) + + reporter( + completed=0, + total=test_case.total, + pending_identities=("model:orders_0", *test_case.pending_identities), + ) + reporter( + completed=test_case.completed, + total=test_case.total, + pending_identities=test_case.pending_identities, + ) + reporter.write_interrupted() + + output: str = stream.getvalue() + fragment: str + for fragment in test_case.expected_fragments: + assert fragment in output diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_output.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_output.py index 5b93f761f2..896b49a3ac 100644 --- a/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_output.py +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/clone/test_output.py @@ -3,17 +3,73 @@ import pytest from sqlbuild.cli.commands._helpers.clone.output import ( + render_clone_fingerprint_interrupted_line, + render_clone_fingerprint_progress_line, render_clone_item_line, render_clone_output, ) from sqlbuild.executor.clone.models import CloneExecutionResult, CloneItemResult from sqlbuild.executor.clone.types import CloneAction, CloneStatus from tests.unit.src.sqlbuild.cli.commands._helpers.clone._test_types import ( + RenderCloneFingerprintInterruptedTestCase, + RenderCloneFingerprintProgressTestCase, RenderCloneItemLineTestCase, RenderCloneOutputTestCase, ) +@pytest.mark.parametrize( + "test_case", + ( + RenderCloneFingerprintProgressTestCase( + description="confirmed batch reports cumulative progress", + completed=50, + total=386, + expected_line=" fingerprint[ 50/386] propagated OK", + ), + ), + ids=lambda case: case.description, +) +def test_given_confirmed_fingerprint_batch_when_rendering_then_progress_is_explicit( + test_case: RenderCloneFingerprintProgressTestCase, +) -> None: + rendered: str = render_clone_fingerprint_progress_line( + completed=test_case.completed, + total=test_case.total, + use_color=False, + ) + + assert rendered == test_case.expected_line + + +@pytest.mark.parametrize( + "test_case", + ( + RenderCloneFingerprintInterruptedTestCase( + description="interruption bounds pending identity context", + completed=50, + total=62, + pending_identities=tuple(f"model:orders_{index}" for index in range(12)), + expected_fragments=("[50/62] FAIL", "model:orders_0", "model:orders_9", "(+2 more)"), + unexpected_fragments=("orders_10",), + ), + ), + ids=lambda case: case.description, +) +def test_given_interrupted_fingerprint_batch_when_rendering_then_pending_context_is_bounded( + test_case: RenderCloneFingerprintInterruptedTestCase, +) -> None: + rendered: str = render_clone_fingerprint_interrupted_line( + completed=test_case.completed, + total=test_case.total, + pending_identities=test_case.pending_identities, + use_color=False, + ) + + assert all(fragment in rendered for fragment in test_case.expected_fragments) + assert all(fragment not in rendered for fragment in test_case.unexpected_fragments) + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py b/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py index 5f19c8ad24..a1db1233a7 100644 --- a/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py +++ b/tests/unit/src/sqlbuild/cli/progress/classes/test_native_progress_projector.py @@ -621,6 +621,42 @@ def test_given_standalone_warehouse_operation_on_tty_when_projected_then_progres assert stream.getvalue() == test_case.expected_output +@pytest.mark.parametrize( + "test_case", + ( + CursorCleanupCase( + description="clone finalization phase is visible", + expected_output="Clone finalization START\nClone finalization OK (0.01s)\n", + ), + ), + ids=lambda case: case.description, +) +def test_given_clone_finalization_lifecycle_when_projected_then_phase_is_visible( + test_case: CursorCleanupCase, +) -> None: + stream: _FlushRecordingStream = _FlushRecordingStream(tty=False) + projector: NativeProgressProjector = NativeProgressProjector(stream=stream, use_color=False) + start: LifecycleEvent = lifecycle_event( + "operation_started", + operation_id="clone-finalization", + payload={ + "operation_kind": "clone", + "operation_name": "clone_finalization", + }, + ) + terminal: LifecycleEvent = replace( + start, + event_id="clone-finalization-terminal", + event_type="operation_completed", + payload={**start.payload, "duration_ms": 10.0}, + ) + + projector.consume(start) + projector.consume(terminal) + + assert stream.getvalue() == test_case.expected_output + + @pytest.mark.parametrize( "test_case", ( diff --git a/tests/unit/src/sqlbuild/compiler/fingerprints/main/_test_types.py b/tests/unit/src/sqlbuild/compiler/fingerprints/main/_test_types.py index bf4ab48128..58979b3228 100644 --- a/tests/unit/src/sqlbuild/compiler/fingerprints/main/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/fingerprints/main/_test_types.py @@ -57,6 +57,23 @@ class WriteFingerprintRetryExhaustionTestCase: expected_error_fragment: str +@dataclass(frozen=True) +class WriteFingerprintBatchTestCase: + description: str + fingerprint_count: int + definition_size: int + expected_rows_per_insert: tuple[int, ...] + expected_progress: tuple[tuple[int, int], ...] + + +@dataclass(frozen=True) +class WriteFingerprintPartialFailureTestCase: + description: str + fingerprint_count: int + expected_execute_count: int + expected_progress: tuple[tuple[int, int], ...] + + @dataclass(frozen=True) class NormalizeQuerySqlTestCase: description: str diff --git a/tests/unit/src/sqlbuild/compiler/fingerprints/main/test_write.py b/tests/unit/src/sqlbuild/compiler/fingerprints/main/test_write.py index 7940819f76..0cfaebc7b4 100644 --- a/tests/unit/src/sqlbuild/compiler/fingerprints/main/test_write.py +++ b/tests/unit/src/sqlbuild/compiler/fingerprints/main/test_write.py @@ -1,15 +1,23 @@ from __future__ import annotations +from dataclasses import replace +from unittest.mock import Mock + import pytest from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.compiler.fingerprints.constants import FINGERPRINT_WRITE_ATTEMPTS from sqlbuild.compiler.fingerprints.main.write import write_fingerprint +from sqlbuild.compiler.fingerprints.main.write_many import write_fingerprints +from sqlbuild.compiler.fingerprints.models import Fingerprint from tests.unit.src.sqlbuild.compiler.fingerprints.main._test_types import ( + WriteFingerprintBatchTestCase, + WriteFingerprintPartialFailureTestCase, WriteFingerprintRetryExhaustionTestCase, WriteFingerprintRetryTestCase, ) from tests.unit.src.sqlbuild.compiler.fingerprints.main.helpers import ( + FakeFingerprintWriteExecute, FlakyFingerprintWriteExecute, RecordingSleeper, build_write_test_fingerprint, @@ -17,6 +25,103 @@ ) +@pytest.mark.parametrize( + "test_case", + ( + WriteFingerprintBatchTestCase( + description="bounds batches by row count", + fingerprint_count=101, + definition_size=10, + expected_rows_per_insert=(50, 50, 1), + expected_progress=((50, 101), (100, 101), (101, 101)), + ), + WriteFingerprintBatchTestCase( + description="bounds batches by rendered byte size", + fingerprint_count=2, + definition_size=300_000, + expected_rows_per_insert=(1, 1), + expected_progress=((1, 2), (2, 2)), + ), + ), + ids=lambda case: case.description, +) +def test_given_fingerprints_when_writing_then_creates_once_and_inserts_bounded_batches( + test_case: WriteFingerprintBatchTestCase, +) -> None: + execute: FakeFingerprintWriteExecute = FakeFingerprintWriteExecute() + fingerprints: tuple[Fingerprint, ...] = tuple( + replace( + build_write_test_fingerprint(node_name=f"orders_{index}"), + definition="x" * test_case.definition_size, + ) + for index in range(test_case.fingerprint_count) + ) + progress: list[tuple[int, int]] = [] + + write_fingerprints( + connection=object(), + execute=execute, + database=None, + schema="analytics", + fingerprints=fingerprints, + render_qualified_name=render_qualified_name, + render_framework_type=DuckDbAdapter().render_framework_type, + on_progress=lambda completed, total: progress.append((completed, total)), + ) + + insert_statements: tuple[str, ...] = tuple(execute.executed_sql[1:]) + assert execute.executed_sql[0].startswith("CREATE TABLE") + assert tuple(sql.count("), (") + 1 for sql in insert_statements) == ( + test_case.expected_rows_per_insert + ) + assert tuple(progress) == test_case.expected_progress + + +@pytest.mark.parametrize( + "test_case", + ( + WriteFingerprintPartialFailureTestCase( + description="only confirms rows from completed batches", + fingerprint_count=51, + expected_execute_count=7, + expected_progress=((50, 51),), + ), + ), + ids=lambda case: case.description, +) +def test_given_later_batch_failure_when_writing_then_progress_only_confirms_completed_rows( + test_case: WriteFingerprintPartialFailureTestCase, + monkeypatch: pytest.MonkeyPatch, +) -> None: + execute: Mock = Mock( + side_effect=(None, None, *([RuntimeError("warehouse request interrupted")] * 5)) + ) + monkeypatch.setattr( + "sqlbuild.compiler.fingerprints._helpers.write_many.time.sleep", lambda _: None + ) + progress: list[tuple[int, int]] = [] + + with pytest.raises(RuntimeError, match="warehouse request interrupted"): + write_fingerprints( + connection=object(), + execute=execute, + database=None, + schema="analytics", + fingerprints=tuple( + build_write_test_fingerprint(node_name=f"orders_{index}") + for index in range(test_case.fingerprint_count) + ), + render_qualified_name=render_qualified_name, + render_framework_type=DuckDbAdapter().render_framework_type, + on_progress=lambda completed, total: progress.append((completed, total)), + ) + + assert execute.call_count == test_case.expected_execute_count + assert execute.call_args_list[0].kwargs["sql"].startswith("CREATE TABLE") + assert execute.call_args_list[1].kwargs["sql"].startswith("INSERT INTO") + assert tuple(progress) == test_case.expected_progress + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/helpers.py b/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/helpers.py index d8da5463dc..46b1b95b02 100644 --- a/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/helpers.py +++ b/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/helpers.py @@ -110,7 +110,7 @@ def patch_fingerprint_io( from sqlbuild.executor.clone._helpers import fingerprinting as fingerprinting_module monkeypatch.setattr(fingerprinting_module, "read_latest_fingerprints", read_latest) - monkeypatch.setattr(fingerprinting_module, "write_fingerprint", write) + monkeypatch.setattr(fingerprinting_module, "write_fingerprints", write) def build_fingerprint(node_type: str, node_name: str) -> Fingerprint: diff --git a/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/test_fingerprinting.py b/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/test_fingerprinting.py index d849b61b15..c4220e6bdd 100644 --- a/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/test_fingerprinting.py +++ b/tests/unit/src/sqlbuild/executor/clone/_helpers/fingerprinting/test_fingerprinting.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Callable from typing import Any, cast import pytest @@ -64,20 +65,26 @@ def read_latest(**kwargs: Any) -> FingerprintSet: }, ) - def write_fingerprint(**kwargs: Any) -> None: + def write_fingerprints(**kwargs: Any) -> None: used_connections.append(kwargs["connection"]) - fingerprint: Fingerprint = kwargs["fingerprint"] - written.append( - ( - fingerprint.node_type, - fingerprint.node_name, - fingerprint.target_schema, - fingerprint.target_name, - fingerprint.run_id, + fingerprint: Fingerprint + for fingerprint in kwargs["fingerprints"]: + written.append( + ( + fingerprint.node_type, + fingerprint.node_name, + fingerprint.target_schema, + fingerprint.target_name, + fingerprint.run_id, + ) ) + on_progress: Callable[..., None] = kwargs["on_progress"] + on_progress( + completed=len(kwargs["fingerprints"]), + total=len(kwargs["fingerprints"]), ) - patch_fingerprint_io(monkeypatch, read_latest=read_latest, write=write_fingerprint) + patch_fingerprint_io(monkeypatch, read_latest=read_latest, write=write_fingerprints) origin_model_entries: tuple[ModelPlanEntry, ...] = ( build_model_entry("orders", schema="prod", materialization=MaterializationType.TABLE), build_model_entry("orders_view", schema="prod", materialization=MaterializationType.VIEW),