diff --git a/src/sqlbuild/cli/output/_helpers/plan_text.py b/src/sqlbuild/cli/output/_helpers/plan_text.py index 7e4c627355..bba68b6b80 100644 --- a/src/sqlbuild/cli/output/_helpers/plan_text.py +++ b/src/sqlbuild/cli/output/_helpers/plan_text.py @@ -90,10 +90,14 @@ } -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} " @@ -101,6 +105,15 @@ def _format_retention(*, lines: list[str], plan: PlanOutput) -> list[str]: ) 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( diff --git a/src/sqlbuild/compiler/planner/_helpers/changes/detect.py b/src/sqlbuild/compiler/planner/_helpers/changes/detect.py index e21637a2ed..57cda287d5 100644 --- a/src/sqlbuild/compiler/planner/_helpers/changes/detect.py +++ b/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") diff --git a/src/sqlbuild/compiler/planner/_helpers/changes/schema.py b/src/sqlbuild/compiler/planner/_helpers/changes/schema.py index 90b1d831dc..c3b405e6c4 100644 --- a/src/sqlbuild/compiler/planner/_helpers/changes/schema.py +++ b/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( diff --git a/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py b/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py index db562a44bd..45c5b2af52 100644 --- a/src/sqlbuild/compiler/planner/_helpers/warehouse/snapshot.py +++ b/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(), ) diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py index 69618db180..121bfdb6d6 100644 --- a/src/sqlbuild/compiler/planner/models.py +++ b/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) diff --git a/src/sqlbuild/executor/build/_helpers/retention.py b/src/sqlbuild/executor/build/_helpers/retention.py index dd7c737121..81bbaa22a0 100644 --- a/src/sqlbuild/executor/build/_helpers/retention.py +++ b/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, diff --git a/src/sqlbuild/executor/build/classes/build_scheduler.py b/src/sqlbuild/executor/build/classes/build_scheduler.py index eb74f7d97c..fd0f9e1d57 100644 --- a/src/sqlbuild/executor/build/classes/build_scheduler.py +++ b/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 ( diff --git a/src/sqlbuild/executor/build/main/_execute.py b/src/sqlbuild/executor/build/main/_execute.py index b70358600c..4aed32c523 100644 --- a/src/sqlbuild/executor/build/main/_execute.py +++ b/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, diff --git a/tests/integration/src/sqlbuild/adapters/snowflake/test_table_type_conversion.py b/tests/integration/src/sqlbuild/adapters/snowflake/test_table_type_conversion.py index 29789ff58a..daed05fd6e 100644 --- a/tests/integration/src/sqlbuild/adapters/snowflake/test_table_type_conversion.py +++ b/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, diff --git a/tests/integration/src/sqlbuild/compiler/planner/_helpers/test_warehouse_snapshot.py b/tests/integration/src/sqlbuild/compiler/planner/_helpers/test_warehouse_snapshot.py index 78a10f97d7..94fa2037e0 100644 --- a/tests/integration/src/sqlbuild/compiler/planner/_helpers/test_warehouse_snapshot.py +++ b/tests/integration/src/sqlbuild/compiler/planner/_helpers/test_warehouse_snapshot.py @@ -266,6 +266,7 @@ def test_given_warehouse_state_when_gathering_snapshot_then_returns_expected( assert ( frozenset(snapshot.fingerprints.seeds.keys()) == test_case.expected_seed_fingerprint_names ) + assert snapshot.column_dialect == adapter.sql_analysis_dialect() @pytest.mark.parametrize( diff --git a/tests/integration/src/sqlbuild/executor/build/_test_types.py b/tests/integration/src/sqlbuild/executor/build/_test_types.py index 424042b415..8e9ab894a8 100644 --- a/tests/integration/src/sqlbuild/executor/build/_test_types.py +++ b/tests/integration/src/sqlbuild/executor/build/_test_types.py @@ -183,3 +183,13 @@ class SnapshotHardDeleteExecutionTestCase: expected_initial_rows: tuple[tuple[object, ...], ...] expected_deleted_rows: tuple[tuple[object, ...], ...] expected_query: str + + +@dataclass(frozen=True) +class TableTypeRoutingTestCase: + description: str + conversion_error: str | None + expected_exit_code: int + expected_conversions: tuple[tuple[str, int], ...] + expected_order_rows: int + expected_error_fragment: str | None = None diff --git a/tests/integration/src/sqlbuild/executor/build/helpers.py b/tests/integration/src/sqlbuild/executor/build/helpers.py index 1b9515964d..8d2fa49301 100644 --- a/tests/integration/src/sqlbuild/executor/build/helpers.py +++ b/tests/integration/src/sqlbuild/executor/build/helpers.py @@ -2,17 +2,22 @@ from __future__ import annotations +import json +from collections.abc import Callable from pathlib import Path from textwrap import dedent from typing import Any +import duckdb + from sqlbuild.adapter.contract.types import TablePromotionMode from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter +from sqlbuild.cli.commands.main.entrypoint.entry import main from sqlbuild.compiler.discovery.main.discover import discover_project_inputs from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs from sqlbuild.compiler.pipeline.main.compile import run_compile_pipeline from sqlbuild.compiler.pipeline.models import CompilePipelineOptions, CompilePipelineResult -from sqlbuild.compiler.planner.models import PlanOutput +from sqlbuild.compiler.planner.models import PlanOutput, TableTypePlanEntry from sqlbuild.executor.build.main._execute import execute_build_plan from sqlbuild.executor.build.models import ( BuildExecutionResult, @@ -430,3 +435,72 @@ def verify_warehouse_state( f"WHERE table_name = '{name}'" + f" AND table_schema = '{schema}'" * bool(schema) ) assert cursor.fetchone() is None, f"Relation {relation} should not exist but was found" + + +def drift_for_models( + *, model_names: tuple[str, ...] +) -> Callable[..., tuple[TableTypePlanEntry, ...]]: + """Return a planner stand-in reporting permanent-desired drift for the named models.""" + + def plan_table_types( + *, runtime: Any, warehouse: Any, scope: Any + ) -> tuple[TableTypePlanEntry, ...]: + del warehouse, scope + models_by_name: dict[str, Any] = {model.name: model for model in runtime.project.models} + return tuple( + TableTypePlanEntry( + model_name=name, + destination=models_by_name[name].destination, + copy_name=f"__sqb_type_swap__{name}", + desired_type="permanent", + actual_type="transient", + source="target", + downgrade=False, + downgrade_policy="require_confirmation", + ) + for name in model_names + ) + + return plan_table_types + + +def recording_table_type_conversion(*, conversions: list[tuple[str, int]]) -> Callable[..., None]: + """Record each converted model with its row count at conversion time.""" + + def record(*, entry: TableTypePlanEntry, adapter: object, connection: Any) -> None: + del adapter + rows: list[tuple[int]] = connection.execute( + f"SELECT COUNT(*) FROM {entry.model_name}" + ).fetchall() + conversions.append((entry.model_name, rows[0][0])) + + return record + + +def failing_table_type_conversion( + *, conversions: list[tuple[str, int]], error: str +) -> Callable[..., None]: + """Record each attempted conversion, then fail it with the given message.""" + + record: Callable[..., None] = recording_table_type_conversion(conversions=conversions) + + def fail(*, entry: TableTypePlanEntry, adapter: object, connection: Any) -> None: + record(entry=entry, adapter=adapter, connection=connection) + raise RuntimeError(error) + + return fail + + +def build_project_json(*, project_dir: Path, capsys: Any) -> tuple[int, dict[str, Any]]: + """Run one real CLI build with JSON output.""" + + exit_code: int = main(["--project-dir", str(project_dir), "build", "--json"]) + return exit_code, json.loads(capsys.readouterr().out) + + +def duckdb_row_count(*, database: Path, table: str) -> int: + """Count rows in one table of a DuckDB project database.""" + + with duckdb.connect(str(database)) as connection: + rows: list[tuple[int]] = connection.execute(f"SELECT COUNT(*) FROM {table}").fetchall() + return rows[0][0] diff --git a/tests/integration/src/sqlbuild/executor/build/test_table_type_routing.py b/tests/integration/src/sqlbuild/executor/build/test_table_type_routing.py new file mode 100644 index 0000000000..8a2d05cb8d --- /dev/null +++ b/tests/integration/src/sqlbuild/executor/build/test_table_type_routing.py @@ -0,0 +1,153 @@ +"""Integration tests for per-model table-type conversion routing during real builds.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import duckdb +import pytest + +import sqlbuild.compiler.planner.main.execution.execution as planner_execution_module +import sqlbuild.executor.build.classes.build_scheduler as build_scheduler_module +from tests.integration.src.sqlbuild.executor.build._test_types import TableTypeRoutingTestCase +from tests.integration.src.sqlbuild.executor.build.helpers import ( + build_project_json, + drift_for_models, + duckdb_row_count, + failing_table_type_conversion, + recording_table_type_conversion, + write_build_project_files, +) + +_PROJECT_FILES: dict[str, str] = { + "sqlbuild_project.toml": ( + 'name = "demo"\nadapter = "duckdb"\n\n[connection]\ndatabase = "demo.duckdb"\n' + ), + "models/customer_totals.sql": ( + "MODEL (materialized table);\n\n" + "SELECT customer_id, SUM(amount) AS total FROM raw_orders GROUP BY customer_id" + ), + "models/orders.sql": ( + "MODEL (\n" + " materialized incremental,\n" + " incremental_strategy append,\n" + " cursor updated_at,\n" + " cursor_type timestamp,\n" + " cursor_grain second,\n" + ");\n\n" + "SELECT id, customer_id, amount, updated_at FROM raw_orders" + ), +} +_SETUP_SQL: str = ( + "CREATE TABLE raw_orders AS SELECT * FROM (VALUES " + "(1, 10, 5, TIMESTAMP '2024-01-01'), (2, 10, 7, TIMESTAMP '2024-01-02')) " + "AS t(id, customer_id, amount, updated_at)" +) +_NEW_ORDER_SQL: str = "INSERT INTO raw_orders VALUES (3, 11, 4, TIMESTAMP '2024-01-03')" + + +@pytest.mark.parametrize( + "test_case", + [ + TableTypeRoutingTestCase( + description="rebuilt table skips conversion and carried-over table converts first", + conversion_error=None, + expected_exit_code=0, + expected_conversions=(("orders", 2),), + expected_order_rows=5, + ) + ], + ids=lambda case: case.description, +) +def test_given_table_type_drift_when_building_then_only_carried_over_tables_convert_before_build( + test_case: TableTypeRoutingTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + write_build_project_files(project_dir=tmp_path, project_files=_PROJECT_FILES) + with duckdb.connect(str(tmp_path / "demo.duckdb")) as setup: + setup.execute(_SETUP_SQL) + initial_exit, _ = build_project_json(project_dir=tmp_path, capsys=capsys) + with duckdb.connect(str(tmp_path / "demo.duckdb")) as setup: + setup.execute(_NEW_ORDER_SQL) + conversions: list[tuple[str, int]] = [] + monkeypatch.setattr( + planner_execution_module, + "plan_table_types", + drift_for_models(model_names=("customer_totals", "orders")), + ) + monkeypatch.setattr( + build_scheduler_module, + "apply_table_type_conversion", + recording_table_type_conversion(conversions=conversions), + ) + + exit_code, _ = build_project_json(project_dir=tmp_path, capsys=capsys) + + assert initial_exit == 0 + assert exit_code == test_case.expected_exit_code + assert tuple(conversions) == test_case.expected_conversions + assert ( + duckdb_row_count(database=tmp_path / "demo.duckdb", table="orders") + == test_case.expected_order_rows + ) + + +@pytest.mark.parametrize( + "test_case", + [ + TableTypeRoutingTestCase( + description="failed conversion fails its model before any data change", + conversion_error="conversion copy was not created with the desired type", + expected_exit_code=1, + expected_conversions=(("orders", 2),), + expected_order_rows=2, + ) + ], + ids=lambda case: case.description, +) +def test_given_failing_table_type_conversion_when_building_then_model_fails_without_dml( + test_case: TableTypeRoutingTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + write_build_project_files(project_dir=tmp_path, project_files=_PROJECT_FILES) + with duckdb.connect(str(tmp_path / "demo.duckdb")) as setup: + setup.execute(_SETUP_SQL) + initial_exit, _ = build_project_json(project_dir=tmp_path, capsys=capsys) + with duckdb.connect(str(tmp_path / "demo.duckdb")) as setup: + setup.execute(_NEW_ORDER_SQL) + conversions: list[tuple[str, int]] = [] + monkeypatch.setattr( + planner_execution_module, + "plan_table_types", + drift_for_models(model_names=("customer_totals", "orders")), + ) + monkeypatch.setattr( + build_scheduler_module, + "apply_table_type_conversion", + failing_table_type_conversion( + conversions=conversions, error=str(test_case.conversion_error) + ), + ) + + exit_code, payload = build_project_json(project_dir=tmp_path, capsys=capsys) + serialized: str = json.dumps(payload) + statuses: dict[str, str] = {item["name"]: item["status"] for item in payload["assets"]} + + assert initial_exit == 0 + assert exit_code == test_case.expected_exit_code + assert tuple(conversions) == test_case.expected_conversions + assert str(test_case.conversion_error) in serialized + assert statuses == {"customer_totals": "success", "orders": "failed"} + assert ( + duckdb_row_count(database=tmp_path / "demo.duckdb", table="orders") + == test_case.expected_order_rows + ) + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/src/sqlbuild/cli/output/main/plan/helpers.py b/tests/unit/src/sqlbuild/cli/output/main/plan/helpers.py index 6a92910963..739ad650e8 100644 --- a/tests/unit/src/sqlbuild/cli/output/main/plan/helpers.py +++ b/tests/unit/src/sqlbuild/cli/output/main/plan/helpers.py @@ -23,6 +23,7 @@ SchemaFinding, SeedPlanEntry, SourceLoadPlanEntry, + TableTypePlanEntry, ) from sqlbuild.compiler.planner.types import ( BackfillAction, @@ -130,6 +131,7 @@ def build_plan_output( warnings: tuple[PlanWarning, ...] = (), provider_usages: tuple[PlanProviderUsage, ...] = (), metadata: dict[str, object] | None = None, + table_type_entries: tuple[TableTypePlanEntry, ...] = (), ) -> PlanOutput: """Build a minimal PlanOutput for formatter tests.""" @@ -146,6 +148,27 @@ def build_plan_output( warnings=warnings, provider_usages=provider_usages, metadata=metadata or {}, + table_type_entries=table_type_entries, + ) + + +def build_table_type_entry(*, model_name: str) -> TableTypePlanEntry: + """Build one permanent-desired Snowflake table-type entry for formatter tests.""" + + return TableTypePlanEntry( + model_name=model_name, + destination=CompiledRelationLocation( + database="analytics", + schema="marts", + name=model_name, + qualified_name=f"analytics.marts.{model_name}", + ), + copy_name=f"{model_name}__sqlbuild_table_type", + desired_type="permanent", + actual_type="transient", + source="target", + downgrade=False, + downgrade_policy="warn", ) diff --git a/tests/unit/src/sqlbuild/cli/output/main/plan/test_plan_text.py b/tests/unit/src/sqlbuild/cli/output/main/plan/test_plan_text.py index d947c3472b..c15779d1a8 100644 --- a/tests/unit/src/sqlbuild/cli/output/main/plan/test_plan_text.py +++ b/tests/unit/src/sqlbuild/cli/output/main/plan/test_plan_text.py @@ -43,6 +43,7 @@ build_schema_finding, build_seed_entry, build_source_load_entry, + build_table_type_entry, build_warning, ) @@ -1350,6 +1351,46 @@ ), unexpected_fragments=("used by 3 selected Python surfaces", "parameter"), ), + FormatPlanTestCase( + description="table-type conversions and warnings are capped like other sections", + plan_output=build_plan_output( + table_type_entries=tuple( + build_table_type_entry(model_name=f"orders_{index:02d}") for index in range(25) + ), + warnings=tuple( + build_warning( + model_name=f"customers_{index:02d}", + message="enforced column status type mismatch", + ) + for index in range(25) + ), + ), + expected_fragments=( + "orders_19 desired=permanent actual=transient source=target", + "customers_19", + "Warnings (25)", + "... and 5 more (use --verbose to show all)", + ), + unexpected_fragments=("orders_20 ", "customers_20"), + ), + FormatPlanTestCase( + description="verbose output shows every table-type conversion and warning", + plan_output=build_plan_output( + table_type_entries=tuple( + build_table_type_entry(model_name=f"orders_{index:02d}") for index in range(25) + ), + warnings=tuple( + build_warning( + model_name=f"customers_{index:02d}", + message="enforced column status type mismatch", + ) + for index in range(25) + ), + ), + display_options=DisplayOptions(max_entries_per_section=None), + expected_fragments=("orders_24 desired=permanent", "customers_24"), + unexpected_fragments=("use --verbose",), + ), ], ids=lambda case: case.description, ) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py index 42d945f677..b38cfcd1f9 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py @@ -57,6 +57,7 @@ class DetectSchemaChangesTestCase: expected_findings: tuple[SchemaFinding, ...] inferred_schema_complete: bool = True dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = () + dialect: str | None = None @dataclass(frozen=True) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py index c883deafb9..6ef31435d4 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py @@ -290,6 +290,75 @@ ), expected_findings=(), ), + DetectSchemaChangesTestCase( + description="snowflake type aliases match warehouse canonical types", + yml_columns=( + ColumnInfo(name="order_id", type="INTEGER"), + ColumnInfo(name="customer_name", type="VARCHAR"), + ColumnInfo(name="notes", type="TEXT"), + ColumnInfo(name="ordered_at", type="TIMESTAMP"), + ColumnInfo(name="amount", type="NUMERIC(10, 2)"), + ), + inferred_columns=None, + warehouse_columns=( + ColumnInfo(name="order_id", type="NUMBER(38,0)"), + ColumnInfo(name="customer_name", type="VARCHAR(16777216)"), + ColumnInfo(name="notes", type="VARCHAR(16777216)"), + ColumnInfo(name="ordered_at", type="TIMESTAMP_NTZ"), + ColumnInfo(name="amount", type="NUMBER(10,2)"), + ), + type_enforcement=True, + dialect="snowflake", + expected_findings=(), + ), + DetectSchemaChangesTestCase( + description="snowflake still reports a real varchar length change", + yml_columns=(ColumnInfo(name="customer_name", type="VARCHAR(20)"),), + inferred_columns=None, + warehouse_columns=(ColumnInfo(name="customer_name", type="VARCHAR(16777216)"),), + type_enforcement=True, + dialect="snowflake", + expected_findings=( + SchemaFinding( + kind=SchemaChangeKind.COLUMN_TYPE_CHANGED, + column_name="customer_name", + source=SchemaColumnSource.YML, + expected_type="VARCHAR(20)", + actual_type="VARCHAR(16777216)", + ), + ), + ), + DetectSchemaChangesTestCase( + description="snowflake matches uppercase declared and inferred names case-insensitively", + yml_columns=(ColumnInfo(name="ORDER_DATE", type="DATE"),), + inferred_columns=( + InferredColumn(name="ORDER_DATE", type="DATE"), + InferredColumn(name="STATUS", type="VARCHAR"), + ), + warehouse_columns=( + ColumnInfo(name="order_date", type="DATE"), + ColumnInfo(name="status", type="VARCHAR(16777216)"), + ), + type_enforcement=True, + dialect="snowflake", + expected_findings=(), + ), + DetectSchemaChangesTestCase( + description="without a dialect aliases and name case still differ", + yml_columns=(ColumnInfo(name="ORDER_ID", type="INTEGER"),), + inferred_columns=None, + warehouse_columns=(ColumnInfo(name="ORDER_ID", type="NUMBER(38,0)"),), + type_enforcement=True, + expected_findings=( + SchemaFinding( + kind=SchemaChangeKind.COLUMN_TYPE_CHANGED, + column_name="ORDER_ID", + source=SchemaColumnSource.YML, + expected_type="INTEGER", + actual_type="NUMBER(38,0)", + ), + ), + ), ], ids=lambda case: case.description, ) @@ -303,6 +372,7 @@ def test_given_columns_when_detecting_schema_changes_then_returns_expected_findi type_enforcement=test_case.type_enforcement, inferred_schema_complete=test_case.inferred_schema_complete, dynamic_columns=test_case.dynamic_columns, + dialect=test_case.dialect, ) assert result == test_case.expected_findings diff --git a/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py b/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py index dff65fa1dd..1ba7f16451 100644 --- a/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/executor/build/_helpers/_test_types.py @@ -4,7 +4,7 @@ from sqlbuild.adapter.contract.models import RelationInfo from sqlbuild.adapter.contract.types import RetentionChangePhase from sqlbuild.compiler.discovery.models import DiscoveredLoaderFunction -from sqlbuild.compiler.planner.types import RetentionPlanPhase +from sqlbuild.compiler.planner.types import MaterializationType, PlanAction, RetentionPlanPhase from sqlbuild.executor.build.models import BuildExecutionResult from sqlbuild.executor.build.types import BuildStatus from sqlbuild.executor.scheduling.types import ExecutionStatus @@ -156,3 +156,12 @@ class BatchSummaryTestCase: batch_size: str | None = None expected_absent_fragments: tuple[str, ...] = field(default_factory=tuple) expected_none: bool = False + + +@dataclass(frozen=True) +class MaterializationRecreatesRelationTestCase: + description: str + materialization_type: MaterializationType + action: PlanAction + incremental_mode: str | None + expected_recreates: bool diff --git a/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py b/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py index 54c46c6f2c..8dc8e77c72 100644 --- a/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py +++ b/tests/unit/src/sqlbuild/executor/build/_helpers/test_retention.py @@ -1,5 +1,6 @@ from __future__ import annotations +from dataclasses import replace from unittest.mock import Mock, call import pytest @@ -12,15 +13,24 @@ ) from sqlbuild.adapter.contract.types import RetentionChangePhase, RetentionScope from sqlbuild.compiler.compile.models import CompiledRelationLocation -from sqlbuild.compiler.planner.models import PlanOutput, RetentionPlanEntry, TableTypePlanEntry +from sqlbuild.compiler.planner.models import ( + ModelPlanEntry, + PlanOutput, + RetentionPlanEntry, + TableTypePlanEntry, +) from sqlbuild.compiler.planner.types import ( + IncrementalMode, + MaterializationType, + PlanAction, RetentionDirection, RetentionPlanPhase, ) from sqlbuild.errors.contracts.exceptions import ExecutorInputError from sqlbuild.executor.build._helpers.retention import ( apply_retention_phase, - apply_table_type_conversions, + apply_table_type_conversion, + materialization_recreates_relation, reconcile_model_retention, ) from sqlbuild.observability import EventDispatcher, LifecycleEvent, dispatcher_scope @@ -28,9 +38,11 @@ BuildModelRetentionReconciliationTestCase, BuildRetentionPhaseTestCase, LifecycleProgressTestCase, + MaterializationRecreatesRelationTestCase, TableTypeConversionErrorTestCase, TableTypeConversionTestCase, ) +from tests.unit.src.sqlbuild.executor.build._helpers.helpers import build_model_plan_entry _TARGET: RelationInfo = RelationInfo( database="warehouse", @@ -270,9 +282,7 @@ def test_given_recoverable_table_type_state_when_converting_then_uses_inspection downgrade_policy="require_confirmation", ) - apply_table_type_conversions( - plan=PlanOutput(table_type_entries=(entry,)), adapter=adapter, connection=connection - ) + apply_table_type_conversion(entry=entry, adapter=adapter, connection=connection) assert tuple(item.kwargs["sql"] for item in adapter.execute.call_args_list) == ( test_case.expected_statements @@ -324,9 +334,7 @@ def test_given_unknown_live_table_type_when_converting_then_fails_closed( ) with pytest.raises(ExecutorInputError, match="metadata is unknown"): - apply_table_type_conversions( - plan=PlanOutput(table_type_entries=(entry,)), adapter=adapter, connection=object() - ) + apply_table_type_conversion(entry=entry, adapter=adapter, connection=object()) assert tuple(item.kwargs["sql"] for item in adapter.execute.call_args_list) == ( test_case.expected_statements @@ -392,9 +400,7 @@ def test_given_unrecoverable_table_type_state_when_converting_then_fails_before_ ) with pytest.raises(ExecutorInputError, match=test_case.expected_error_fragment): - apply_table_type_conversions( - plan=PlanOutput(table_type_entries=(entry,)), adapter=adapter, connection=object() - ) + apply_table_type_conversion(entry=entry, adapter=adapter, connection=object()) assert tuple(item.kwargs["sql"] for item in adapter.execute.call_args_list) == ( test_case.expected_statements @@ -505,15 +511,67 @@ def test_given_table_type_conversion_when_ddl_blocks_then_start_is_already_dispa ) with dispatcher_scope(dispatcher): - apply_table_type_conversions( - plan=PlanOutput(table_type_entries=(entry,)), - adapter=adapter, - connection=object(), - ) + apply_table_type_conversion(entry=entry, adapter=adapter, connection=object()) assert tuple(event.event_type for event in events) == test_case.expected_event_types assert barrier_events[0].event_type == "operation_started" +@pytest.mark.parametrize( + "test_case", + [ + MaterializationRecreatesRelationTestCase( + description="table rebuild recreates the relation", + materialization_type=MaterializationType.TABLE, + action=PlanAction.CREATE_TABLE, + incremental_mode=None, + expected_recreates=True, + ), + MaterializationRecreatesRelationTestCase( + description="incremental full rebuild recreates the relation", + materialization_type=MaterializationType.INCREMENTAL, + action=PlanAction.CREATE_TABLE, + incremental_mode=IncrementalMode.FULL, + expected_recreates=True, + ), + MaterializationRecreatesRelationTestCase( + description="microbatch full refresh may keep the relation without batches", + materialization_type=MaterializationType.INCREMENTAL, + action=PlanAction.CREATE_TABLE, + incremental_mode=IncrementalMode.MICROBATCH, + expected_recreates=False, + ), + MaterializationRecreatesRelationTestCase( + description="incremental run keeps the relation", + materialization_type=MaterializationType.INCREMENTAL, + action=PlanAction.INCREMENTAL_MERGE, + incremental_mode=IncrementalMode.FULL, + expected_recreates=False, + ), + MaterializationRecreatesRelationTestCase( + description="planned skip keeps the relation", + materialization_type=MaterializationType.TABLE, + action=PlanAction.SKIP, + incremental_mode=None, + expected_recreates=False, + ), + ], + ids=lambda case: case.description, +) +def test_given_model_action_when_checking_recreation_then_matches_materialization_contract( + test_case: MaterializationRecreatesRelationTestCase, +) -> None: + entry: ModelPlanEntry = replace( + build_model_plan_entry( + name="orders", + materialization_type=test_case.materialization_type, + action=test_case.action, + ), + incremental_mode=test_case.incremental_mode, + ) + + assert materialization_recreates_relation(entry) is test_case.expected_recreates + + if __name__ == "__main__": pytest.main([__file__, "-vv"])