fix: faster Snowflake table-type conversion and accurate schema diffs

src/sqlbuild/cli/output/_helpers/plan_text.py

@@ -90,17 +90,30 @@
 }
 
 
-def _format_retention(*, lines: list[str], plan: PlanOutput) -> list[str]:
+def _format_retention(
+    *, lines: list[str], plan: PlanOutput, display_options: DisplayOptions
+) -> list[str]:
     if plan.table_type_entries:
         lines.append("Table type conversions")
-        for table_type_entry in plan.table_type_entries:
+        for table_type_entry in visible_entries(
+            entries=plan.table_type_entries, options=display_options
+        ):
             actual_type: str = table_type_entry.actual_type or "unknown"
             lines.append(
                 f"  {table_type_entry.model_name} desired={table_type_entry.desired_type} "
                 f"actual={actual_type} source={table_type_entry.source}"
             )
             if table_type_entry.irreversible_warning is not None:
                 lines.append(f"    WARNING: {table_type_entry.irreversible_warning}")
+        lines = append_overflow_line(
+            lines=lines,
+            total_count=len(plan.table_type_entries),
+            visible_count=len(
+                visible_entries(entries=plan.table_type_entries, options=display_options)
+            ),
+            indent="  ",
+            options=display_options,
+        )
     if plan.retention_entries:
         lines.append("Retention")
     for entry in plan.retention_entries:
@@ -159,8 +172,9 @@ def format_plan(
             lines=lines,
             plan=plan,
             include_direct_freshness_diagnostics=(include_direct_freshness_diagnostics),
+            display_options=resolved_display_options,
         )
-        lines = _format_retention(lines=lines, plan=plan)
+        lines = _format_retention(lines=lines, plan=plan, display_options=resolved_display_options)
         result: str = "\n".join(lines)
         return result if use_color else _strip_ansi(result)
 
@@ -211,7 +225,7 @@ def format_plan(
         display_options=resolved_display_options,
         skipped_header_style=style.muted,
     )
-    lines = _format_retention(lines=lines, plan=plan)
+    lines = _format_retention(lines=lines, plan=plan, display_options=resolved_display_options)
 
     lines = _format_python_plan_entries(
         lines=lines,
@@ -329,6 +343,7 @@ def format_plan(
         lines=lines,
         plan=plan,
         include_direct_freshness_diagnostics=include_direct_freshness_diagnostics,
+        display_options=resolved_display_options,
     )
 
     output: str = "\n".join(lines)
@@ -1420,6 +1435,7 @@ def _format_warnings(
     lines: list[str],
     plan: PlanOutput,
     include_direct_freshness_diagnostics: bool,
+    display_options: DisplayOptions,
 ) -> list[str]:
     """Append the warnings section."""
 
@@ -1443,10 +1459,13 @@ def _format_warnings(
     style: CliStyle = CliStyle(use_color=True)
     lines.append("")
     lines.append(style.warning_strong(f"Warnings ({len(warning_entries)})"))
+    shown_warnings: Sequence[PlanWarning] = visible_entries(
+        entries=warning_entries, options=display_options
+    )
     warning: PlanWarning
     warning_index: int
-    for warning_index, warning in enumerate(warning_entries):
-        connector: str = tree_connector(style=style, last=warning_index == len(warning_entries) - 1)
+    for warning_index, warning in enumerate(shown_warnings):
+        connector: str = tree_connector(style=style, last=warning_index == len(shown_warnings) - 1)
         message_lines: list[str] = warning.message.split("\n")
         if warning.model_name is not None:
             lines.append(f"{connector} {style.object_name(warning.model_name)}")
@@ -1462,7 +1481,13 @@ def _format_warnings(
             continuation: str
             for continuation in message_lines[1:]:
                 lines.append(f"    {style.warning(continuation)}")
-    return lines
+    return append_overflow_line(
+        lines=lines,
+        total_count=len(warning_entries),
+        visible_count=len(shown_warnings),
+        indent="  ",
+        options=display_options,
+    )
 
 
 def _format_virtual_metadata(

src/sqlbuild/compiler/planner/_helpers/changes/detect.py

@@ -240,6 +240,7 @@ def detect_model_changes(
             dynamic_columns=(
                 model.schema_entry.dynamic_columns if model.schema_entry is not None else ()
             ),
+            dialect=snapshot.column_dialect,
         )
         if schema_findings:
             raw_policy = get_config_str(values=model.config.values, key="replay_on_change")

src/sqlbuild/compiler/planner/_helpers/changes/schema.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from dataclasses import replace
+
 from sqlbuild.adapter.contract.models import ColumnInfo
 from sqlbuild.adapter.type_system.main.types_equal import types_equal
 from sqlbuild.compiler.compile.models import InferredColumn
@@ -10,6 +12,8 @@
 from sqlbuild.spec.contracts.main.matching_dynamic_families import matching_dynamic_families
 from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily
 
+_CASE_FOLDING_DIALECT: str = "snowflake"
+
 
 def detect_schema_changes(
     *,
@@ -19,16 +23,28 @@ def detect_schema_changes(
     type_enforcement: bool,
     inferred_schema_complete: bool,
     dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = (),
+    dialect: str | None = None,
 ) -> tuple[SchemaFinding, ...]:
     """Compare yml and inferred columns against warehouse columns and return findings."""
 
+    if dialect == _CASE_FOLDING_DIALECT:
+        yml_columns = tuple(replace(col, name=col.name.lower()) for col in yml_columns)
+        if inferred_columns is not None:
+            inferred_columns = tuple(
+                replace(col, name=col.name.lower()) for col in inferred_columns
+            )
+        warehouse_columns = tuple(replace(col, name=col.name.lower()) for col in warehouse_columns)
     warehouse_map: dict[str, str] = {col.name: col.type for col in warehouse_columns}
     findings: list[SchemaFinding] = []
 
     seen_names: set[str] = set()
 
     if type_enforcement:
-        findings.extend(_compare_yml_columns(yml_columns=yml_columns, warehouse_map=warehouse_map))
+        findings.extend(
+            _compare_yml_columns(
+                yml_columns=yml_columns, warehouse_map=warehouse_map, dialect=dialect
+            )
+        )
         col: ColumnInfo
         for col in yml_columns:
             seen_names.add(col.name)
@@ -39,6 +55,7 @@ def detect_schema_changes(
                     inferred_columns=inferred_columns,
                     warehouse_map=warehouse_map,
                     seen_names=seen_names,
+                    dialect=dialect,
                 )
             )
             inferred_col: InferredColumn
@@ -51,6 +68,7 @@ def detect_schema_changes(
                     inferred_columns=inferred_columns,
                     warehouse_map=warehouse_map,
                     seen_names=seen_names,
+                    dialect=dialect,
                 )
             )
             inferred_col_ne: InferredColumn
@@ -62,6 +80,7 @@ def detect_schema_changes(
                 yml_columns=yml_columns,
                 warehouse_map=warehouse_map,
                 seen_names=seen_names,
+                dialect=dialect,
             )
         )
         col_ne: ColumnInfo
@@ -79,7 +98,7 @@ def detect_schema_changes(
             continue
         family: SchemaDynamicColumnFamily = matching[0]
         seen_names.add(column_name)
-        if not types_equal(left=family.type, right=column_type, dialect=None):
+        if not types_equal(left=family.type, right=column_type, dialect=dialect):
             findings.append(
                 SchemaFinding(
                     kind=SchemaChangeKind.COLUMN_TYPE_CHANGED,
@@ -113,6 +132,7 @@ def _compare_yml_columns(
     *,
     yml_columns: tuple[ColumnInfo, ...],
     warehouse_map: dict[str, str],
+    dialect: str | None,
 ) -> list[SchemaFinding]:
     """Compare yml-declared columns against warehouse state."""
 
@@ -128,7 +148,7 @@ def _compare_yml_columns(
                     expected_type=col.type,
                 )
             )
-        elif not types_equal(left=warehouse_map[col.name], right=col.type, dialect=None):
+        elif not types_equal(left=warehouse_map[col.name], right=col.type, dialect=dialect):
             findings.append(
                 SchemaFinding(
                     kind=SchemaChangeKind.COLUMN_TYPE_CHANGED,
@@ -146,6 +166,7 @@ def _compare_yml_columns_non_enforced(
     yml_columns: tuple[ColumnInfo, ...],
     warehouse_map: dict[str, str],
     seen_names: set[str],
+    dialect: str | None,
 ) -> list[SchemaFinding]:
     """Compare yml columns against warehouse when type enforcement is off."""
 
@@ -163,7 +184,7 @@ def _compare_yml_columns_non_enforced(
                     expected_type=col.type,
                 )
             )
-        elif not types_equal(left=warehouse_map[col.name], right=col.type, dialect=None):
+        elif not types_equal(left=warehouse_map[col.name], right=col.type, dialect=dialect):
             findings.append(
                 SchemaFinding(
                     kind=SchemaChangeKind.COLUMN_TYPE_CHANGED,
@@ -181,6 +202,7 @@ def _compare_inferred_columns(
     inferred_columns: tuple[InferredColumn, ...],
     warehouse_map: dict[str, str],
     seen_names: set[str],
+    dialect: str | None,
 ) -> list[SchemaFinding]:
     """Compare sql_analysis-inferred columns against warehouse state, skipping yml-covered names."""
 
@@ -199,7 +221,7 @@ def _compare_inferred_columns(
                 )
             )
         elif col.type is not None and not types_equal(
-            left=warehouse_map[col.name], right=col.type, dialect=None
+            left=warehouse_map[col.name], right=col.type, dialect=dialect
         ):
             findings.append(
                 SchemaFinding(

src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py

@@ -275,6 +275,7 @@ def gather_warehouse_snapshot(
         fingerprints=fingerprints,
         cursor_snapshots=cursor_snapshots,
         source_freshness_state_schemas=freshness_state_schemas,
+        column_dialect=adapter.sql_analysis_dialect(),
     )
 
 

src/sqlbuild/compiler/planner/models.py

@@ -659,6 +659,7 @@ class WarehouseSnapshot:
     fingerprints: WarehouseFingerprints = field(default_factory=WarehouseFingerprints)
     cursor_snapshots: dict[str, ModelCursorSnapshot] = field(default_factory=dict)
     source_freshness_state_schemas: frozenset[str] = field(default_factory=frozenset)
+    column_dialect: str | None = None
 
 
 @dataclass(frozen=True)

src/sqlbuild/executor/build/_helpers/retention.py

@@ -10,8 +10,13 @@
 from sqlbuild.adapter.relations.main.resolve_qualified_name_parts import (
     resolve_qualified_name_parts,
 )
-from sqlbuild.compiler.planner.models import PlanOutput, RetentionPlanEntry, TableTypePlanEntry
-from sqlbuild.compiler.planner.types import RetentionPlanPhase
+from sqlbuild.compiler.planner.models import (
+    ModelPlanEntry,
+    PlanOutput,
+    RetentionPlanEntry,
+    TableTypePlanEntry,
+)
+from sqlbuild.compiler.planner.types import IncrementalMode, PlanAction, RetentionPlanPhase
 from sqlbuild.errors.contracts.exceptions import ExecutorInputError
 from sqlbuild.runtime.observability.classes.operation_lifecycle import (
     OperationAttributes,
@@ -23,28 +28,20 @@
 from sqlbuild.spec.contracts.types import TableType
 
 
-def apply_table_type_conversions(
-    *, plan: PlanOutput, adapter: BaseAdapter, connection: Any
-) -> None:
-    """Recover by inspection: clean desired targets or recreate and swap undesired targets."""
+def materialization_recreates_relation(entry: ModelPlanEntry) -> bool:
+    """Return whether the build always recreates the relation with the planned table type."""
 
-    _apply_table_type_entries(
-        entries=plan.table_type_entries, adapter=adapter, connection=connection
+    return (
+        entry.action == PlanAction.CREATE_TABLE
+        and entry.incremental_mode != IncrementalMode.MICROBATCH
     )
 
 
-def _apply_table_type_entries(
-    *, entries: tuple[TableTypePlanEntry, ...], adapter: BaseAdapter, connection: Any
-) -> None:
-    if not entries:
-        return
-    _apply_table_type_entry(entry=entries[0], adapter=adapter, connection=connection)
-    _apply_table_type_entries(entries=entries[1:], adapter=adapter, connection=connection)
-
-
-def _apply_table_type_entry(
+def apply_table_type_conversion(
     *, entry: TableTypePlanEntry, adapter: BaseAdapter, connection: Any
 ) -> None:
+    """Recover by inspection: clean a desired target or recreate and swap an undesired one."""
+
     destination: str = resolve_qualified_name_parts(
         adapter=adapter,
         database=entry.destination.database,

src/sqlbuild/executor/build/classes/build_scheduler.py

@@ -30,6 +30,7 @@
     PlanOutput,
     SeedPlanEntry,
     SqlTestPlanEntry,
+    TableTypePlanEntry,
 )
 from sqlbuild.compiler.planner.types import (
     IncrementalMode,
@@ -45,7 +46,11 @@
 from sqlbuild.executor.build._helpers.blocking import downstream_blocked_keys
 from sqlbuild.executor.build._helpers.end_audits import run_end_audits
 from sqlbuild.executor.build._helpers.indexes import build_execution_indexes
-from sqlbuild.executor.build._helpers.retention import reconcile_model_retention
+from sqlbuild.executor.build._helpers.retention import (
+    apply_table_type_conversion,
+    materialization_recreates_relation,
+    reconcile_model_retention,
+)
 from sqlbuild.executor.build._helpers.scheduler import (
     _build_worker_failure_completion,
     _build_worker_success_completion,
@@ -158,6 +163,9 @@ def __init__(
         self._callbacks: BuildCallbacks = callbacks
         self._plan: PlanOutput = plan
         self._indexes: BuildIndexes = build_execution_indexes(plan)
+        self._table_type_entries: dict[str, TableTypePlanEntry] = {
+            entry.model_name: entry for entry in plan.table_type_entries
+        }
         self._adapter: BaseAdapter = adapter
         self._connection_config: dict[str, object] = connection_config
         self._connections: tuple[Any, ...] = connections
@@ -1010,6 +1018,15 @@ def _execute_model_node(
             sqlbuild_kind=model_entry.materialization_type,
         ):
             try:
+                table_type_entry: TableTypePlanEntry | None = self._table_type_entries.get(
+                    model_entry.name
+                )
+                if table_type_entry is not None and not materialization_recreates_relation(
+                    model_entry
+                ):
+                    apply_table_type_conversion(
+                        entry=table_type_entry, adapter=self._adapter, connection=connection
+                    )
                 if self._before_model_materialize is not None:
                     self._before_model_materialize(entry=model_entry, connection=connection)
                 if (

src/sqlbuild/executor/build/main/_execute.py

@@ -15,7 +15,6 @@
 from sqlbuild.executor.auditing.models import AuditExecutionResult
 from sqlbuild.executor.build._helpers.retention import (
     apply_retention_phase,
-    apply_table_type_conversions,
     reconcile_retention_after_build,
 )
 from sqlbuild.executor.build.classes.build_scheduler import BuildScheduler
@@ -70,7 +69,6 @@ def execute_build_plan(
         schema_prepared=schema_prepared,
     )
 
-    _ = apply_table_type_conversions(plan=plan, adapter=adapter, connection=scheduler_connection)
     _ = apply_retention_phase(
         plan=plan,
         adapter=adapter,

tests/integration/src/sqlbuild/adapters/snowflake/test_table_type_conversion.py

@@ -9,8 +9,8 @@
 from sqlbuild.adapter.contract.models import RelationInfo
 from sqlbuild.compiler.compile.models import CompiledRelationLocation
 from sqlbuild.compiler.planner._helpers.planning.retention import table_type_copy_name
-from sqlbuild.compiler.planner.models import PlanOutput, TableTypePlanEntry
-from sqlbuild.executor.build._helpers.retention import apply_table_type_conversions
+from sqlbuild.compiler.planner.models import TableTypePlanEntry
+from sqlbuild.executor.build._helpers.retention import apply_table_type_conversion
 from sqlbuild.spec.contracts.types import TableType
 from tests.integration.src.sqlbuild.adapters.snowflake._test_types import (
     SnowflakeTableTypeConversionTestCase,
@@ -99,11 +99,7 @@ def test_given_existing_snowflake_table_when_applying_table_type_then_metadata_a
     )
     adapter.statement_recorder.events.clear()
 
-    apply_table_type_conversions(
-        plan=PlanOutput(table_type_entries=(entry,)),
-        adapter=adapter,
-        connection=connection,
-    )
+    apply_table_type_conversion(entry=entry, adapter=adapter, connection=connection)
     conversion_statement_count: int = len(adapter.statement_recorder.snapshot())
     relations: tuple[RelationInfo, ...] = adapter.list_relations(
         connection=connection,