fix: batch clone fingerprint finalization

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)
 
 

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)

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."""
 

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",

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}, "