Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 27 additions & 6 deletions src/sqlbuild/compiler/lineage/_helpers/rich_columns.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@ def build_rich_project_column_lineage(
return None

schema: dict[str, dict[str, str]] = _build_schema_mapping(project)
polyglot_tables: dict[str, dict[str, object]] = _polyglot_schema_tables(schema)
model_results: dict[str, ModelColumnLineage] = {}
collapsed_edges: list[ColumnLineageEdge] = []

Expand All @@ -94,6 +95,7 @@ def build_rich_project_column_lineage(
result: ModelColumnLineage | None = _build_polyglot_model_column_lineage(
model=model,
schema=schema,
polyglot_tables=polyglot_tables,
dialect=dialect,
)
if result is None:
Expand Down Expand Up @@ -123,6 +125,7 @@ def _build_polyglot_model_column_lineage(
*,
model: CompiledModel,
schema: dict[str, dict[str, str]],
polyglot_tables: dict[str, dict[str, object]],
dialect: str | None,
) -> ModelColumnLineage | None:
normalized_sql: str
Expand All @@ -137,7 +140,10 @@ def _build_polyglot_model_column_lineage(
normalized_sql,
{
"dialect": _polyglot_dialect(dialect),
"schema": _polyglot_schema(schema),
"schema": _referenced_polyglot_schema(
polyglot_tables=polyglot_tables,
physical_resources=physical_resources,
),
},
)
except polyglot_module.PolyglotError as error:
Expand Down Expand Up @@ -292,14 +298,29 @@ def _star_expanded_columns(analysis: dict[str, Any]) -> frozenset[str]:
return frozenset(expanded)


def _polyglot_schema(schema: dict[str, dict[str, str]]) -> dict[str, object]:
tables: list[dict[str, object]] = []
for table_name, columns in sorted(schema.items()):
def _polyglot_schema_tables(schema: dict[str, dict[str, str]]) -> dict[str, dict[str, object]]:
tables: dict[str, dict[str, object]] = {}
for table_name, columns in schema.items():
table_columns: list[dict[str, str]] = []
for column_name, column_type in sorted(columns.items()):
table_columns.append({"name": column_name, "type": column_type or "UNKNOWN"})
tables.append({"name": table_name, "columns": table_columns})
return {"tables": tables}
tables[table_name] = {"name": table_name, "columns": table_columns}
return tables


def _referenced_polyglot_schema(
*,
polyglot_tables: dict[str, dict[str, object]],
physical_resources: tuple[PhysicalResource, ...],
) -> dict[str, object]:
referenced: set[str] = {resource.physical_name for resource in physical_resources}
return {
"tables": [
polyglot_tables[table_name]
for table_name in sorted(referenced)
if table_name in polyglot_tables
]
}


def _polyglot_dialect(dialect: str | TypeDialect | None) -> PolyglotAnalysisDialect:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,9 @@ fans the last layer back in. Every lattice model is reachable along exponentiall
compile guards do not enable the lattice, so their fingerprints and budgets are unchanged.

After one compile and warm lineage and scope caches, the guard runs `sqb lineage` downstream from
the hub and upstream from the rollup as text and JSON, `sqb dag --json` and `sqb scope --json`,
each in a fresh process. Every command
the hub and upstream from the rollup as text and JSON, rich column traces upstream from
`shared_orders_rollup.amount` and downstream from `shared_orders_hub.amount`, `sqb dag --json` and
`sqb scope --json`, each in a fresh process. Every command
has a wall-time ceiling, a peak-RSS ceiling and an output-line bound linear in the model count, so
a renderer or graph walk that re-expands shared nodes per path fails on size before it can hide
behind a loose timing budget. It runs in the 3,000-model fresh-process compile job.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,35 @@
expected_max_wall_seconds=3.0,
expected_max_rss_bytes=128 * _MIB,
),
InspectionCommandPerformanceGuardTestCase(
description="models_3000_lineage_column_text_upstream_from_fan_in_rollup",
sqb_args=("lineage", f"{SHARED_DIAMOND_ROLLUP}.amount", "--direction", "upstream"),
expected_fragments=(
f"Column trace {SHARED_DIAMOND_ROLLUP}.amount upstream",
"shared_orders_l15_s00.amount (cast)",
),
expected_max_output_lines=_MODEL_COUNT,
expected_max_wall_seconds=8.0,
expected_max_rss_bytes=512 * _MIB,
),
InspectionCommandPerformanceGuardTestCase(
description="models_3000_lineage_column_json_downstream_from_fan_out_hub",
sqb_args=(
"lineage",
f"{SHARED_DIAMOND_HUB}.amount",
"--direction",
"downstream",
"--format",
"json",
),
expected_fragments=(
'"direction": "downstream"',
f'"resource_name": "{SHARED_DIAMOND_ROLLUP}"',
),
expected_max_output_lines=20 * _MODEL_COUNT,
expected_max_wall_seconds=8.0,
expected_max_rss_bytes=512 * _MIB,
),
InspectionCommandPerformanceGuardTestCase(
description="models_3000_dag_json",
sqb_args=("dag", "--json"),
Expand Down
10 changes: 10 additions & 0 deletions tests/unit/src/sqlbuild/compiler/lineage/_test_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,16 @@ class SqlAnalysisDisabledLineageTestCase:
expected_result_is_none: bool


@dataclass(frozen=True)
class RichLineageAnalysisScalingTestCase:
description: str
chain_length: int
unrelated_model_counts: tuple[int, ...]
expected_schema_builds: int
expected_analysis_calls: int
expected_schema_tables_per_call: tuple[tuple[str, ...], ...]


@dataclass(frozen=True)
class ExpectedCountTestCase:
description: str
Expand Down
27 changes: 27 additions & 0 deletions tests/unit/src/sqlbuild/compiler/lineage/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,3 +90,30 @@ def make_compiled_seed(*, name: str, columns: tuple[str, ...]) -> CompiledSeed:

def edge_label(source_name: str, source_column: str, target_name: str, target_column: str) -> str:
return f"{source_name}.{source_column}->{target_name}.{target_column}"


def make_chain_models(
*, chain_length: int, unrelated_model_count: int
) -> tuple[CompiledModel, ...]:
root: CompiledModel = make_compiled_model(
name="orders_0",
query_sql="SELECT 1 AS order_id, 10 AS amount",
inferred_columns=("order_id", "amount"),
)
chain: tuple[CompiledModel, ...] = tuple(
make_compiled_model(
name=f"orders_{index}",
query_sql=f'SELECT order_id, amount + 1 AS amount FROM __ref("orders_{index - 1}")',
inferred_columns=("order_id", "amount"),
)
for index in range(1, chain_length)
)
unrelated: tuple[CompiledModel, ...] = tuple(
make_compiled_model(
name=f"inventory_{index}",
query_sql="SELECT 1 AS product_id, 5 AS quantity",
inferred_columns=("product_id", "quantity"),
)
for index in range(unrelated_model_count)
)
return (root, *chain, *unrelated)
81 changes: 81 additions & 0 deletions tests/unit/src/sqlbuild/compiler/lineage/test_columns.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

from dataclasses import replace
from typing import Any

import pytest

Expand All @@ -11,21 +12,26 @@
CompiledProject,
)
from sqlbuild.compiler.compile.types import CompiledResourceType
from sqlbuild.compiler.lineage._helpers import rich_columns
from sqlbuild.compiler.lineage._helpers.columns import _build_schema_mapping
from sqlbuild.compiler.lineage.main.columns import build_project_column_lineage
from sqlbuild.compiler.lineage.models import ColumnLineage, ModelColumnLineage, ProjectColumnLineage
from sqlbuild.compiler.lineage.types import (
ColumnLineageConfidence,
ColumnLineageMode,
ColumnTransformKind,
)
from sqlbuild.compiler.sql_analysis.main.import_polyglot_sql import import_polyglot_sql
from tests.unit.src.sqlbuild.compiler.lineage._test_types import (
ColumnLineageAnalyzerTestCase,
ExpectedCountTestCase,
ProjectLineageGraphTestCase,
RichLineageAnalysisScalingTestCase,
SqlAnalysisDisabledLineageTestCase,
)
from tests.unit.src.sqlbuild.compiler.lineage.helpers import (
edge_label,
make_chain_models,
make_compiled_model,
make_compiled_project,
make_compiled_seed,
Expand Down Expand Up @@ -653,3 +659,78 @@ def test_given_sql_analysis_disabled_when_building_column_lineage_then_returns_n
result: ProjectColumnLineage | None = build_project_column_lineage(project=project)

assert (result is None) is test_case.expected_result_is_none


@pytest.mark.parametrize(
"test_case",
[
RichLineageAnalysisScalingTestCase(
description="builds the schema once and analyses each traced model against its refs",
chain_length=4,
unrelated_model_counts=(0, 250),
expected_schema_builds=1,
expected_analysis_calls=4,
expected_schema_tables_per_call=(
(),
("__sqlbuild_model__orders_0",),
("__sqlbuild_model__orders_1",),
("__sqlbuild_model__orders_2",),
),
)
],
ids=lambda case: case.description,
)
def test_given_unrelated_models_when_tracing_rich_lineage_then_analysis_work_stays_linear(
test_case: RichLineageAnalysisScalingTestCase,
monkeypatch: pytest.MonkeyPatch,
) -> None:
polyglot_module: Any = import_polyglot_sql()
analyze_query: Any = polyglot_module.analyze_query
schema_builds: list[CompiledProject] = []
analysed_tables: list[tuple[str, ...]] = []

def counting_schema_mapping(project: CompiledProject) -> dict[str, dict[str, str]]:
schema_builds.append(project)
return _build_schema_mapping(project)

def recording_analyze_query(sql: str, options: dict[str, Any]) -> object:
analysed_tables.append(tuple(table["name"] for table in options["schema"]["tables"]))
return analyze_query(sql, options)

monkeypatch.setattr(rich_columns, "_build_schema_mapping", counting_schema_mapping)
monkeypatch.setattr(polyglot_module, "analyze_query", recording_analyze_query)
chain_names: frozenset[str] = frozenset(
f"orders_{index}" for index in range(test_case.chain_length)
)
edges_by_project_size: list[tuple[str, ...]] = []
for unrelated_model_count in test_case.unrelated_model_counts:
schema_builds.clear()
analysed_tables.clear()
project: CompiledProject = make_compiled_project(
models=make_chain_models(
chain_length=test_case.chain_length,
unrelated_model_count=unrelated_model_count,
)
)

result: ProjectColumnLineage | None = build_project_column_lineage(
project=project, model_names=chain_names
)

assert result is not None
assert len(schema_builds) == test_case.expected_schema_builds
assert len(analysed_tables) == test_case.expected_analysis_calls
assert tuple(analysed_tables) == test_case.expected_schema_tables_per_call
edges_by_project_size.append(
tuple(
edge_label(
edge.source.resource_name,
edge.source.column_name,
edge.target.resource_name,
edge.target.column_name,
)
for edge in result.edges
)
)
assert len(set(edges_by_project_size)) == 1
assert "orders_2.amount->orders_3.amount" in edges_by_project_size[0]
Loading