diff --git a/.gitignore b/.gitignore index c18dd8d8..63e36ff7 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ -__pycache__/ +**/__pycache__/ +**/.venv/ diff --git a/converters/dbt/README.md b/converters/dbt/README.md new file mode 100644 index 00000000..ae31a011 --- /dev/null +++ b/converters/dbt/README.md @@ -0,0 +1,122 @@ +# osi-dbt + +Converts between dbt's [MetricFlow Semantic Interface](https://docs.getdbt.com/docs/build/about-metricflow) (MSI) and the [Open Semantic Interchange](https://github.com/open-semantic-interchange/OSI) (OSI) format. + +Both conversion directions are supported: + +- `msi-to-osi` — `semantic_manifest.json` (dbt output) → OSI YAML +- `osi-to-msi` — OSI YAML → `semantic_manifest.json` + +## Requirements + +- Python 3.11+ +- [uv](https://docs.astral.sh/uv/) (recommended) or pip + +## Installation + +```bash +pip install osi-dbt +``` + +Or with uv: + +```bash +uv add osi-dbt +``` + +## CLI usage + +### dbt → OSI + +Generate `semantic_manifest.json` from your dbt project first: + +```bash +dbt parse +# output: target/semantic_manifest.json +``` + +Then convert to OSI YAML: + +```bash +osi-dbt msi-to-osi -i target/semantic_manifest.json -o semantic_model.yaml +``` + +By default the OSI semantic model is named `semantic_model`. Override it with `--model-name`: + +```bash +osi-dbt msi-to-osi -i target/semantic_manifest.json -o semantic_model.yaml --model-name my_project +``` + +Conversion issues (e.g. dropped CONVERSION or PRIVATE metrics) are printed as warnings to stderr. The output file is still written. + +### OSI → dbt + +```bash +osi-dbt osi-to-msi -i semantic_model.yaml -o semantic_manifest.json +``` + +Produces a `semantic_manifest.json` that metricflow can load. + +### Help + +```bash +osi-dbt --help +osi-dbt msi-to-osi --help +osi-dbt osi-to-msi --help +``` + +## Python API + +```python +from osi_dbt import MSIToOSIConverter, OSIToMSIConverter +from metricflow_semantics.model.dbt_manifest_parser import parse_manifest_from_dbt_generated_manifest + +# dbt → OSI +manifest = parse_manifest_from_dbt_generated_manifest(Path("target/semantic_manifest.json").read_text()) +result = MSIToOSIConverter().convert(manifest, osi_model_name="my_project") + +for issue in result.issues: + print(f"[warning] {issue.issue_type.value}: {issue.element_name}") + +osi_yaml = result.output.to_osi_yaml() + +# OSI → dbt +import yaml +from osi import OSIDocument + +document = OSIDocument.model_validate(yaml.safe_load(Path("semantic_model.yaml").read_text())) +result = OSIToMSIConverter().convert(document) +manifest_json = result.output.model_dump_json(by_alias=True, exclude_none=True, indent=2) +``` + +### Conversion notes + +**MSI → OSI** is lossy in the following ways, each recorded as a `ConverterIssue` in the result: + +| Issue type | Reason | +|---|---| +| `CONVERSION_METRIC_DROPPED` | OSI has no conversion-funnel metric type | +| `PRIVATE_METRIC_DROPPED` | OSI has no visibility modifiers | +| `NATURAL_ENTITY_DROPPED` | OSI has no natural-key entity type | +| `CUMULATIVE_SEMANTICS_LOSS` | Window/grain semantics cannot be expressed in an OSI expression string; the base aggregation is preserved | + +**OSI → MSI** reconstructs a best-effort MSI manifest from OSI's simpler schema. Nothing is dropped, but OSI carries less structural information than MSI, so the converter makes the following choices: + +- Single aggregations (`SUM(col)`, `COUNT(DISTINCT col)`, etc.) → SIMPLE metric with `metric_aggregation_params` +- `(expr_a) / (expr_b)` → RATIO metric with auto-generated sub-metrics +- Anything else → SIMPLE metric with the raw expression stored verbatim +- Time dimensions always receive `TimeGranularity.DAY` (OSI carries no granularity field) + +## Development + +```bash +cd converters/dbt +uv sync +uv run pytest +``` + +Generate initial syrupy snapshots on first run: + +```bash +uv run pytest --snapshot-update +``` diff --git a/converters/dbt/pyproject.toml b/converters/dbt/pyproject.toml new file mode 100644 index 00000000..f1bb3b0e --- /dev/null +++ b/converters/dbt/pyproject.toml @@ -0,0 +1,34 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "osi-dbt" +version = "0.2.0.dev0" +description = "dbt (MetricFlow Semantic Interface) <> OSI converter" +requires-python = ">=3.11" +dependencies = [ + "osi-python>=0.2.0.dev0", + "metricflow>=0.200", + "sqlglot>=20.0", + "jinja2>=3.0", + "PyYAML>=6.0", +] + +[project.license] +text = "Apache-2.0" + +[project.scripts] +osi-dbt = "osi_dbt.cli:main" + +[tool.hatch.build.targets.wheel] +packages = ["src/osi_dbt"] + +[tool.uv] +dev-dependencies = [ + "pytest>=8.0", + "syrupy>=4.0", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/converters/dbt/src/osi_dbt/__init__.py b/converters/dbt/src/osi_dbt/__init__.py new file mode 100644 index 00000000..5cc35cbb --- /dev/null +++ b/converters/dbt/src/osi_dbt/__init__.py @@ -0,0 +1,11 @@ +from osi_dbt.converter_issues import ConverterIssue, ConverterIssueType, ConverterResult +from osi_dbt.msi_to_osi import MSIToOSIConverter +from osi_dbt.osi_to_msi import OSIToMSIConverter + +__all__ = [ + "ConverterIssue", + "ConverterIssueType", + "ConverterResult", + "MSIToOSIConverter", + "OSIToMSIConverter", +] diff --git a/converters/dbt/src/osi_dbt/cli.py b/converters/dbt/src/osi_dbt/cli.py new file mode 100644 index 00000000..5dd8a297 --- /dev/null +++ b/converters/dbt/src/osi_dbt/cli.py @@ -0,0 +1,91 @@ +"""CLI entry point for the osi-dbt converter. + +Usage: + osi-dbt msi-to-osi -i semantic_manifest.json -o output.yaml + osi-dbt osi-to-msi -i input.yaml -o semantic_manifest.json +""" + +import argparse +import json +import sys +from pathlib import Path + +import yaml + +from osi import OSIDocument +from osi_dbt.converter_issues import ConverterIssueType +from osi_dbt.msi_to_osi import MSIToOSIConverter +from osi_dbt.osi_to_msi import OSIToMSIConverter + +from metricflow_semantics.model.dbt_manifest_parser import parse_manifest_from_dbt_generated_manifest + +_ISSUE_REASON: dict[ConverterIssueType, str] = { + ConverterIssueType.CONVERSION_METRIC_DROPPED: "OSI has no conversion-funnel metric type", + ConverterIssueType.PRIVATE_METRIC_DROPPED: "OSI has no visibility modifiers", + ConverterIssueType.NATURAL_ENTITY_DROPPED: "OSI has no natural-key entity type", + ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS: "OSI expressions cannot represent window or grain semantics; the base aggregation was preserved", +} + +_DROPPED_ISSUE_TYPES = { + ConverterIssueType.CONVERSION_METRIC_DROPPED, + ConverterIssueType.PRIVATE_METRIC_DROPPED, + ConverterIssueType.NATURAL_ENTITY_DROPPED, +} + + +def _cmd_msi_to_osi(args: argparse.Namespace) -> None: + input_path = Path(args.input) + output_path = Path(args.output) + + manifest = parse_manifest_from_dbt_generated_manifest(input_path.read_text()) + result = MSIToOSIConverter().convert(manifest, osi_model_name=args.model_name) + + if result.issues: + for issue in result.issues: + verb = "was dropped" if issue.issue_type in _DROPPED_ISSUE_TYPES else "was converted with loss" + reason = _ISSUE_REASON[issue.issue_type] + print(f"[WARNING] {issue.issue_type.value}: {issue.element_name} {verb} during conversion because {reason}", file=sys.stderr) + + output_path.write_text(result.output.to_osi_yaml()) + print(f"Written to {output_path}", file=sys.stderr) + + +def _cmd_osi_to_msi(args: argparse.Namespace) -> None: + input_path = Path(args.input) + output_path = Path(args.output) + + raw = yaml.safe_load(input_path.read_text()) + document = OSIDocument.model_validate(raw) + result = OSIToMSIConverter().convert(document) + + output_path.write_text(result.output.model_dump_json(by_alias=True, exclude_none=True, indent=2)) + print(f"Written to {output_path}", file=sys.stderr) + + +def main() -> None: + parser = argparse.ArgumentParser( + prog="osi-dbt", + description="Convert between dbt semantic_manifest.json and OSI YAML.", + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + msi_to_osi = subparsers.add_parser("msi-to-osi", help="Convert semantic_manifest.json → OSI YAML") + msi_to_osi.add_argument("-i", "--input", required=True, metavar="FILE", help="Path to semantic_manifest.json") + msi_to_osi.add_argument("-o", "--output", required=True, metavar="FILE", help="Path for output OSI YAML") + msi_to_osi.add_argument( + "--model-name", default="semantic_model", metavar="NAME", help="OSI semantic model name (default: semantic_model)" + ) + + osi_to_msi = subparsers.add_parser("osi-to-msi", help="Convert OSI YAML → semantic_manifest.json") + osi_to_msi.add_argument("-i", "--input", required=True, metavar="FILE", help="Path to OSI YAML") + osi_to_msi.add_argument("-o", "--output", required=True, metavar="FILE", help="Path for output semantic_manifest.json") + + args = parser.parse_args() + if args.command == "msi-to-osi": + _cmd_msi_to_osi(args) + elif args.command == "osi-to-msi": + _cmd_osi_to_msi(args) + + +if __name__ == "__main__": + main() diff --git a/converters/dbt/src/osi_dbt/converter_issues.py b/converters/dbt/src/osi_dbt/converter_issues.py new file mode 100644 index 00000000..7c87862f --- /dev/null +++ b/converters/dbt/src/osi_dbt/converter_issues.py @@ -0,0 +1,31 @@ +from dataclasses import dataclass +from enum import Enum +from typing import Generic, List, TypeVar + + +class ConverterIssueType(Enum): + """Identifies the kind of information loss that occurred during conversion.""" + + CONVERSION_METRIC_DROPPED = "CONVERSION_METRIC_DROPPED" + PRIVATE_METRIC_DROPPED = "PRIVATE_METRIC_DROPPED" + NATURAL_ENTITY_DROPPED = "NATURAL_ENTITY_DROPPED" + CUMULATIVE_SEMANTICS_LOSS = "CUMULATIVE_SEMANTICS_LOSS" + + +@dataclass(frozen=True) +class ConverterIssue: + """Records a single instance of information loss during conversion.""" + + issue_type: ConverterIssueType + element_name: str + + +T = TypeVar("T") + + +@dataclass(frozen=True) +class ConverterResult(Generic[T]): + """Return value of a converter's convert() method, pairing the output with any conversion issues.""" + + output: T + issues: List[ConverterIssue] diff --git a/converters/dbt/src/osi_dbt/expression_utils.py b/converters/dbt/src/osi_dbt/expression_utils.py new file mode 100644 index 00000000..a9dfbd22 --- /dev/null +++ b/converters/dbt/src/osi_dbt/expression_utils.py @@ -0,0 +1,137 @@ +from typing import Optional, Tuple + +import sqlglot +import sqlglot.expressions as exp + +from metricflow_semantic_interfaces.type_enums import AggregationType + + +def _strip_qualifier(col: str) -> str: + """Strip a leading dataset qualifier, e.g. 'orders.amount' → 'amount'.""" + return col.rsplit(".", 1)[-1] if "." in col else col + + +def _col_name(node: exp.Expression) -> str: + """Return the bare (unqualified) column name from a sqlglot expression node.""" + if isinstance(node, exp.Column): + return node.name + rendered = node.sql() + return _strip_qualifier(rendered) + + +def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, Optional[float]]]: + """Parse a SQL aggregation expression using sqlglot. + + Returns ``(agg_type, bare_col, percentile)`` for recognised patterns, ``None`` otherwise. + ``percentile`` is only set for ``PERCENTILE`` aggregations; it is ``None`` for all others. + The returned column name has any dataset qualifier stripped. + """ + try: + tree = sqlglot.parse_one(expression.strip()) + except sqlglot.errors.ParseError: + return None + + # COUNT(DISTINCT col) + if isinstance(tree, exp.Count) and isinstance(tree.this, exp.Distinct): + cols = tree.this.expressions + if len(cols) == 1: + return AggregationType.COUNT_DISTINCT, _col_name(cols[0]), None + return None + + # COUNT(col) + if isinstance(tree, exp.Count): + return AggregationType.COUNT, _col_name(tree.this), None + + # SUM(CASE WHEN col THEN 1 ELSE 0 END) → SUM_BOOLEAN + if isinstance(tree, exp.Sum) and isinstance(tree.this, exp.Case): + case = tree.this + ifs = case.args.get("ifs", []) + default = case.args.get("default") + if ( + len(ifs) == 1 + and isinstance(default, exp.Literal) + and default.name == "0" + and isinstance(ifs[0].args.get("true"), exp.Literal) + and ifs[0].args["true"].name == "1" + ): + return AggregationType.SUM_BOOLEAN, ifs[0].this.sql(), None + return None + + # SUM(col) + if isinstance(tree, exp.Sum): + return AggregationType.SUM, _col_name(tree.this), None + + if isinstance(tree, exp.Avg): + return AggregationType.AVERAGE, _col_name(tree.this), None + + if isinstance(tree, exp.Min): + return AggregationType.MIN, _col_name(tree.this), None + + if isinstance(tree, exp.Max): + return AggregationType.MAX, _col_name(tree.this), None + + # PERCENTILE_CONT(p) WITHIN GROUP (ORDER BY col) + # sqlglot parses this as WithinGroup(this=PercentileCont(...), expression=Order(...)) + if isinstance(tree, exp.WithinGroup): + inner = tree.this + order = tree.args.get("expression") + if ( + isinstance(inner, (exp.PercentileCont, exp.PercentileDisc)) + and isinstance(order, exp.Order) + and order.expressions + ): + ordered = order.expressions[0] + col_node = ordered.this if isinstance(ordered, exp.Ordered) else ordered + col = _col_name(col_node) + try: + p = float(inner.this.name) + except (AttributeError, ValueError): + return None + if p == 0.5 and isinstance(inner, exp.PercentileCont): + return AggregationType.MEDIAN, col, None + return AggregationType.PERCENTILE, col, p + + return None + + +def _try_parse_ratio(expr_str: str) -> Optional[Tuple[str, str]]: + """Try to parse ``(expr_a) / (expr_b)`` using sqlglot, returning ``(num_expr, den_expr)`` or None.""" + try: + tree = sqlglot.parse_one(expr_str.strip()) + except sqlglot.errors.ParseError: + return None + + if not isinstance(tree, exp.Div): + return None + + num = tree.this + den = tree.expression + + # Unwrap outer parentheses if present + if isinstance(num, exp.Paren): + num = num.this + if isinstance(den, exp.Paren): + den = den.this + + return num.sql(), den.sql() + + +def _get_raw_inner_col(expression: str) -> Optional[str]: + """Extract the raw column reference from inside a simple aggregation, before stripping qualifiers.""" + try: + tree = sqlglot.parse_one(expression.strip()) + except sqlglot.errors.ParseError: + return None + + if not isinstance(tree, exp.AggFunc): + return None + + inner = tree.this + if inner is None: + return None + + # For COUNT(DISTINCT col), unwrap the Distinct node + if isinstance(inner, exp.Distinct) and inner.expressions: + return inner.expressions[0].sql() + + return inner.sql() diff --git a/converters/dbt/src/osi_dbt/filter_utils.py b/converters/dbt/src/osi_dbt/filter_utils.py new file mode 100644 index 00000000..a10642f5 --- /dev/null +++ b/converters/dbt/src/osi_dbt/filter_utils.py @@ -0,0 +1,139 @@ +from typing import List, Optional, Sequence + +import jinja2 + +from metricflow_semantic_interfaces.protocols.where_filter import WhereFilterIntersection + + +class _DimensionStub: + """Jinja sandbox stub for `{{ Dimension('entity__dim') }}`. + + Renders to the qualified column name, e.g. `order__status`. + Method chaining (`grain`, `date_part`) appends a `__` part. + """ + + def __init__(self, name: str, entity_path: Sequence[str] = ()) -> None: + self._col = "__".join(list(entity_path) + [name]) + self._suffix = "" + + def grain(self, time_granularity: str) -> "_DimensionStub": + self._suffix = f"__{time_granularity.lower()}" + return self + + def date_part(self, date_part_name: str) -> "_DimensionStub": + self._suffix = f"__{date_part_name.lower()}" + return self + + def descending(self, _is_descending: bool) -> "_DimensionStub": + return self + + def __str__(self) -> str: + return f"{self._col}{self._suffix}" + + +class _TimeDimensionStub: + """Jinja sandbox stub for `{{ TimeDimension('entity__dim', 'grain') }}`. + + Renders to `entity__dim` or `entity__dim__grain` when a granularity is provided. + """ + + def __init__( + self, + name: str, + time_granularity_name: Optional[str] = None, + entity_path: Sequence[str] = (), + **_kwargs: object, + ) -> None: + self._col = "__".join(list(entity_path) + [name]) + self._grain = time_granularity_name + + def grain(self, time_granularity: str) -> "_TimeDimensionStub": + self._grain = time_granularity + return self + + def date_part(self, date_part_name: str) -> "_TimeDimensionStub": + self._grain = date_part_name + return self + + def descending(self, _is_descending: bool) -> "_TimeDimensionStub": + return self + + def __str__(self) -> str: + if self._grain: + return f"{self._col}__{self._grain.lower()}" + return self._col + + +class _EntityStub: + """Jinja sandbox stub for `{{ Entity('name') }}`.""" + + def __init__(self, name: str, entity_path: Sequence[str] = ()) -> None: + self._col = "__".join(list(entity_path) + [name]) + + def descending(self, _is_descending: bool) -> "_EntityStub": + return self + + def __str__(self) -> str: + return self._col + + +class _MetricStub: + """Jinja sandbox stub for `{{ Metric('name') }}`.""" + + def __init__(self, name: str, group_by: Sequence[str] = ()) -> None: + self._name = name + + def descending(self, _is_descending: bool) -> "_MetricStub": + return self + + def __str__(self) -> str: + return self._name + + +def _render_filter_template(template: str) -> str: + """Render an MSI where-filter Jinja template to a plain SQL fragment. + + Jinja references such as `{{ Dimension('order__status') }}`, + `{{ TimeDimension('order__ds', 'day') }}`, `{{ Entity('user') }}`, + and `{{ Metric('revenue') }}` are resolved to their column-name + equivalents using lightweight stubs. The output is a best-effort SQL + string suitable for embedding in an OSI expression. + """ + return jinja2.Template(template, undefined=jinja2.StrictUndefined).render( + Dimension=_DimensionStub, + TimeDimension=_TimeDimensionStub, + Entity=_EntityStub, + Metric=_MetricStub, + ) + + +def _collect_filter_sql(*filters: Optional[WhereFilterIntersection]) -> Optional[str]: + """Render and merge MSI WhereFilterIntersection objects into a single SQL fragment. + + Jinja references (e.g. `{{ Dimension('order__status') }}`) are resolved + using lightweight stubs that produce MetricFlow-qualified column names such + as `order__status`. These are *not* fully resolved SQL column aliases — + resolving to actual table column names would require `WhereFilterSpecFactory` + and `ColumnAssociationResolver` from `metricflow_semantics`, which is out + of scope here. OSI consumers are expected to perform their own column + resolution against the source data. + """ + parts: List[str] = [] + for f in filters: + if f is None: + continue + for wf in f.where_filters: + rendered = _render_filter_template(wf.where_sql_template).strip() + if rendered: + parts.append(rendered) + return _merge_filter_sqls(*parts) + + +def _merge_filter_sqls(*parts: Optional[str]) -> Optional[str]: + """Join non-None SQL filter strings with AND, wrapping each in parens when multiple.""" + active = [p for p in parts if p] + if not active: + return None + if len(active) == 1: + return active[0] + return " AND ".join(f"({p})" for p in active) diff --git a/converters/dbt/src/osi_dbt/msi_to_osi.py b/converters/dbt/src/osi_dbt/msi_to_osi.py new file mode 100644 index 00000000..6a7dbec4 --- /dev/null +++ b/converters/dbt/src/osi_dbt/msi_to_osi.py @@ -0,0 +1,449 @@ +import re +from collections import defaultdict +from dataclasses import dataclass +from itertools import combinations +from typing import Dict, List, Optional, Sequence, Tuple + +from osi import ( + OSIDataset, + OSIDialect, + OSIDialectExpression, + OSIDimension, + OSIDocument, + OSIExpression, + OSIField, + OSIMetric, + OSIRelationship, + OSISemanticModel, +) +from osi_dbt.converter_issues import ConverterIssue, ConverterIssueType, ConverterResult +from osi_dbt.filter_utils import _collect_filter_sql, _merge_filter_sqls + +from metricflow_semantic_interfaces.enum_extension import assert_values_exhausted +from metricflow_semantic_interfaces.implementations.semantic_manifest import PydanticSemanticManifest +from metricflow_semantic_interfaces.protocols.dimension import Dimension +from metricflow_semantic_interfaces.protocols.entity import Entity +from metricflow_semantic_interfaces.protocols.measure import ( + Measure, + MeasureAggregationParameters, +) +from metricflow_semantic_interfaces.protocols.metric import Metric +from metricflow_semantic_interfaces.protocols.semantic_model import SemanticModel +from metricflow_semantic_interfaces.transformations.convert_count import ConvertCountMetricToSumRule +from metricflow_semantic_interfaces.transformations.semantic_manifest_transformer import ( + PydanticSemanticManifestTransformer, +) +from metricflow_semantic_interfaces.type_enums import ( + AggregationType, + DimensionType, + EntityType, + MetricType, +) + + +@dataclass(frozen=True) +class _EntityEntry: + dataset: str + col: str + entity_type: EntityType + + +@dataclass(frozen=True) +class _RelationshipDirection: + from_dataset: str + to_dataset: str + from_col: str + to_col: str + + +class MSIToOSIConverter: + """Converts an MSI SemanticManifest into an OSI Document.""" + + def __init__(self, dialect: OSIDialect = OSIDialect.ANSI_SQL) -> None: + self._dialect = dialect + + def convert( + self, manifest: PydanticSemanticManifest, osi_model_name: str = "semantic_model" + ) -> ConverterResult[OSIDocument]: + manifest = PydanticSemanticManifestTransformer.transform(manifest) + issues: List[ConverterIssue] = [] + + datasets = [self._convert_semantic_model(sm) for sm in manifest.semantic_models] + + entity_index, entity_issues = self._build_entity_index(manifest.semantic_models) + issues.extend(entity_issues) + relationships = self._build_relationships(entity_index) + + metric_index = self._build_metric_index(manifest.metrics) + expression_cache: Dict[Tuple[str, Optional[str]], str] = {} + + osi_metrics: List[OSIMetric] = [] + for metric in manifest.metrics: + if metric.type is MetricType.CONVERSION: + issues.append( + ConverterIssue(issue_type=ConverterIssueType.CONVERSION_METRIC_DROPPED, element_name=metric.name) + ) + continue + if metric.type_params.is_private: + issues.append( + ConverterIssue(issue_type=ConverterIssueType.PRIVATE_METRIC_DROPPED, element_name=metric.name) + ) + continue + if metric.type is MetricType.CUMULATIVE: + issues.append( + ConverterIssue(issue_type=ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS, element_name=metric.name) + ) + expr = self._resolve_metric_expression(metric, metric_index, expression_cache) + osi_metrics.append( + OSIMetric( + name=metric.name, + expression=self._make_expression(expr), + description=metric.description, + ) + ) + + return ConverterResult( + output=OSIDocument( + version="0.2.0.dev0", + dialects=[self._dialect], + semantic_model=[ + OSISemanticModel( + name=osi_model_name, + datasets=datasets, + relationships=relationships if relationships else None, + metrics=osi_metrics if osi_metrics else None, + ) + ], + ), + issues=issues, + ) + + def _convert_semantic_model(self, sm: SemanticModel) -> OSIDataset: + fields: List[OSIField] = [] + for entity in sm.entities: + fields.append(self._convert_entity(entity)) + for dim in sm.dimensions: + fields.append(self._convert_dimension(dim)) + for measure in sm.measures: + fields.append(self._convert_measure(measure)) + + primary_key, unique_keys = self._extract_keys(sm.entities) + + return OSIDataset( + name=sm.name, + source=sm.node_relation.relation_name, + primary_key=primary_key, + unique_keys=unique_keys if unique_keys else None, + description=sm.description, + fields=fields if fields else None, + ) + + def _convert_dimension(self, dim: Dimension) -> OSIField: + expr = dim.expr if dim.expr is not None else dim.name + is_time = dim.type is DimensionType.TIME + + return OSIField( + name=dim.name, + expression=self._make_expression(expr), + dimension=OSIDimension(is_time=is_time), + label=dim.label, + description=dim.description, + ) + + def _convert_entity(self, entity: Entity) -> OSIField: + expr = entity.expr if entity.expr is not None else entity.name + + return OSIField( + name=entity.name, + expression=self._make_expression(expr), + label=entity.label, + description=entity.description, + ) + + def _convert_measure(self, measure: Measure) -> OSIField: + expr = measure.expr if measure.expr is not None else measure.name + + return OSIField( + name=measure.name, + expression=self._make_expression(expr), + label=measure.label, + description=measure.description, + ) + + @staticmethod + def _extract_keys(entities: Sequence[Entity]) -> Tuple[Optional[List[str]], List[List[str]]]: + primary_key: Optional[List[str]] = None + unique_keys: List[List[str]] = [] + + for entity in entities: + col = entity.expr if entity.expr is not None else entity.name + if entity.type is EntityType.PRIMARY: + primary_key = [col] + elif entity.type is EntityType.UNIQUE: + unique_keys.append([col]) + + return primary_key, unique_keys + + @staticmethod + def _build_metric_index(metrics: Sequence[Metric]) -> Dict[str, Metric]: + """Map metric name to Metric for recursive resolution.""" + return {metric.name: metric for metric in metrics} + + @staticmethod + def _lookup_metric(metric_index: Dict[str, Metric], name: str, context: str) -> Metric: + """Look up a metric by name, raising a clear ValueError if not found.""" + try: + return metric_index[name] + except KeyError: + raise ValueError(f"Unknown metric referenced: context={context!r}, metric_name={name!r}") + + def _resolve_metric_expression( + self, + metric: Metric, + metric_index: Dict[str, Metric], + cache: Dict[Tuple[str, Optional[str]], str], + parent_filter: Optional[str] = None, + ) -> str: + """Recursively resolve a metric to a fully-inlined SQL expression string.""" + own_filter = _collect_filter_sql(metric.filter) + combined_filter = _merge_filter_sqls(parent_filter, own_filter) + + cache_key = (metric.name, combined_filter) + if cache_key in cache: + return cache[cache_key] + + if metric.type is MetricType.SIMPLE: + expr = self._resolve_simple(metric, combined_filter) + elif metric.type is MetricType.CUMULATIVE: + expr = self._resolve_cumulative(metric, metric_index, cache, combined_filter) + elif metric.type is MetricType.RATIO: + expr = self._resolve_ratio(metric, metric_index, cache, combined_filter) + elif metric.type is MetricType.DERIVED: + expr = self._resolve_derived(metric, metric_index, cache, combined_filter) + elif metric.type is MetricType.CONVERSION: + # CONVERSION metrics are skipped in convert(); this branch should never be reached. + raise RuntimeError(f"Unexpected CONVERSION metric in expression resolver: metric_name={metric.name!r}") + else: + assert_values_exhausted(metric.type) + + cache[cache_key] = expr + return expr + + def _resolve_simple( + self, + metric: Metric, + filter_sql: Optional[str] = None, + ) -> str: + """Resolve a SIMPLE metric using metric_aggregation_params (always set after transformation).""" + agg_params_obj = metric.type_params.metric_aggregation_params + if agg_params_obj is None: + raise ValueError( + f"SIMPLE metric has no metric_aggregation_params after transformation: metric_name={metric.name!r}" + ) + col = metric.type_params.expr if metric.type_params.expr is not None else metric.name + col = self._qualify_col(col, agg_params_obj.semantic_model) + return self._build_agg_expression(agg_params_obj.agg, col, agg_params_obj.agg_params, filter_sql) + + @staticmethod + def _qualify_col(col: str, semantic_model: str) -> str: + """Qualify col with semantic_model if it is an unqualified identifier or a COUNT-converted expr.""" + # Quoted identifiers (e.g. "my col", `my col`) are not handled — qualifying + # them correctly requires dialect-aware parsing and is deferred as a follow-up. + if re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", col): + return f"{semantic_model}.{col}" + m = ConvertCountMetricToSumRule.COUNT_CONVERSION_RE.match(col) + if m: + return f"CASE WHEN {semantic_model}.{m.group(1)} IS NOT NULL THEN 1 ELSE 0 END" + return col + + def _resolve_cumulative( + self, + metric: Metric, + metric_index: Dict[str, Metric], + cache: Dict[Tuple[str, Optional[str]], str], + filter_sql: Optional[str] = None, + ) -> str: + """Resolve a CUMULATIVE metric to its base aggregation expression. + + Window/grain semantics are not representable in an OSI expression string. + """ + cumulative_params = metric.type_params.cumulative_type_params + if cumulative_params is None or cumulative_params.metric is None: + raise ValueError( + f"CUMULATIVE metric has no sub-metric after transformation: metric_name={metric.name!r}" + ) + sub_input = cumulative_params.metric + sub_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(sub_input.filter)) + return self._resolve_metric_expression( + self._lookup_metric(metric_index, sub_input.name, f"CUMULATIVE metric '{metric.name}'"), + metric_index, + cache, + sub_filter, + ) + + def _resolve_ratio( + self, + metric: Metric, + metric_index: Dict[str, Metric], + cache: Dict[Tuple[str, Optional[str]], str], + filter_sql: Optional[str] = None, + ) -> str: + """Resolve a RATIO metric as (numerator) / (denominator), both fully inlined.""" + if metric.type_params.numerator is None or metric.type_params.denominator is None: + raise ValueError( + f"RATIO metric is missing numerator or denominator: metric_name={metric.name!r}" + ) + num_input = metric.type_params.numerator + den_input = metric.type_params.denominator + num_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(num_input.filter)) + den_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(den_input.filter)) + num_expr = self._resolve_metric_expression( + self._lookup_metric(metric_index, num_input.name, f"RATIO metric '{metric.name}' numerator"), + metric_index, + cache, + num_filter, + ) + den_expr = self._resolve_metric_expression( + self._lookup_metric(metric_index, den_input.name, f"RATIO metric '{metric.name}' denominator"), + metric_index, + cache, + den_filter, + ) + return f"({num_expr}) / ({den_expr})" + + def _resolve_derived( + self, + metric: Metric, + metric_index: Dict[str, Metric], + cache: Dict[Tuple[str, Optional[str]], str], + filter_sql: Optional[str] = None, + ) -> str: + """Resolve a DERIVED metric by substituting each input metric's expression into the expr string. + + Compound sub-expressions (DERIVED/RATIO) are wrapped in parentheses to preserve operator precedence. + """ + expr = metric.type_params.expr or "" + for input_metric in metric.type_params.metrics or []: + ref = input_metric.alias if input_metric.alias else input_metric.name + dep_metric = self._lookup_metric(metric_index, input_metric.name, f"DERIVED metric '{metric.name}'") + input_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(input_metric.filter)) + resolved = self._resolve_metric_expression(dep_metric, metric_index, cache, input_filter) + if dep_metric.type in (MetricType.DERIVED, MetricType.RATIO): + resolved = f"({resolved})" + expr = re.sub(rf"\b{re.escape(ref)}\b", resolved, expr) + return expr + + @staticmethod + def _build_entity_index( + semantic_models: Sequence[SemanticModel], + ) -> Tuple[Dict[str, List[_EntityEntry]], List[ConverterIssue]]: + """Map each entity name to the _EntityEntry objects that declare it.""" + index: Dict[str, List[_EntityEntry]] = defaultdict(list) + issues: List[ConverterIssue] = [] + for sm in semantic_models: + for entity in sm.entities: + if entity.type is EntityType.NATURAL: + issues.append( + ConverterIssue(issue_type=ConverterIssueType.NATURAL_ENTITY_DROPPED, element_name=entity.name) + ) + continue + col = entity.expr if entity.expr is not None else entity.name + index[entity.name].append(_EntityEntry(dataset=sm.name, col=col, entity_type=entity.type)) + return dict(index), issues + + @staticmethod + def _relationship_direction( + ds_a: str, col_a: str, type_a: EntityType, ds_b: str, col_b: str, type_b: EntityType + ) -> _RelationshipDirection: + """Return a _RelationshipDirection obeying OSI directionality. + + OSI spec: `from` is the many-side (FK holder), `to` is the one-side (PK holder). + FOREIGN entities are always the many-side; PRIMARY/UNIQUE are the one-side. + When both sides share the same cardinality tier, break ties alphabetically by dataset name. + """ + one_side_types = {EntityType.PRIMARY, EntityType.UNIQUE} + a_is_one_side = type_a in one_side_types + b_is_one_side = type_b in one_side_types + + if a_is_one_side and not b_is_one_side: + return _RelationshipDirection(from_dataset=ds_b, to_dataset=ds_a, from_col=col_b, to_col=col_a) + if b_is_one_side and not a_is_one_side: + return _RelationshipDirection(from_dataset=ds_a, to_dataset=ds_b, from_col=col_a, to_col=col_b) + # Same cardinality tier — use alphabetical order for determinism. + if ds_a <= ds_b: + return _RelationshipDirection(from_dataset=ds_a, to_dataset=ds_b, from_col=col_a, to_col=col_b) + return _RelationshipDirection(from_dataset=ds_b, to_dataset=ds_a, from_col=col_b, to_col=col_a) + + @staticmethod + def _build_relationships( + entity_index: Dict[str, List[_EntityEntry]], + ) -> List[OSIRelationship]: + """Resolve implicit MSI entity links into explicit OSI relationships. + + Every pair of datasets sharing an entity name is a valid join path. + """ + relationships: List[OSIRelationship] = [] + for entity_name, entries in entity_index.items(): + for entry_a, entry_b in combinations(entries, 2): + if entry_a.dataset == entry_b.dataset: + continue + direction = MSIToOSIConverter._relationship_direction( + entry_a.dataset, + entry_a.col, + entry_a.entity_type, + entry_b.dataset, + entry_b.col, + entry_b.entity_type, + ) + relationships.append( + OSIRelationship( + name=f"{direction.from_dataset}__{direction.to_dataset}__{entity_name}", + from_dataset=direction.from_dataset, + to=direction.to_dataset, + from_columns=[direction.from_col], + to_columns=[direction.to_col], + ) + ) + return relationships + + @staticmethod + def _build_agg_expression( + agg: AggregationType, + col: str, + agg_params: Optional[MeasureAggregationParameters], + filter_sql: Optional[str] = None, + ) -> str: + # Inject the filter as CASE WHEN inside the aggregation. NULL values + # produced by CASE WHEN are ignored by all standard SQL aggregate + # functions, preserving correct filtering semantics. + fc = f"CASE WHEN {filter_sql} THEN {col} END" if filter_sql else col + + if agg is AggregationType.SUM: + return f"SUM({fc})" + elif agg is AggregationType.MIN: + return f"MIN({fc})" + elif agg is AggregationType.MAX: + return f"MAX({fc})" + elif agg is AggregationType.COUNT: + return f"COUNT({fc})" + elif agg is AggregationType.COUNT_DISTINCT: + return f"COUNT(DISTINCT {fc})" + elif agg is AggregationType.AVERAGE: + return f"AVG({fc})" + elif agg is AggregationType.SUM_BOOLEAN: + # col is already a boolean condition; the filter becomes an extra AND term. + if filter_sql: + return f"SUM(CASE WHEN ({filter_sql}) AND ({col}) THEN 1 ELSE 0 END)" + return f"SUM(CASE WHEN {col} THEN 1 ELSE 0 END)" + elif agg is AggregationType.MEDIAN: + return f"PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY {fc})" + elif agg is AggregationType.PERCENTILE: + percentile = agg_params.percentile if agg_params and agg_params.percentile is not None else 0.5 + use_discrete = agg_params.use_discrete_percentile if agg_params else False + func = "PERCENTILE_DISC" if use_discrete else "PERCENTILE_CONT" + return f"{func}({percentile}) WITHIN GROUP (ORDER BY {fc})" + else: + assert_values_exhausted(agg) + + def _make_expression(self, expr: str) -> OSIExpression: + return OSIExpression(dialects=[OSIDialectExpression(dialect=self._dialect, expression=expr)]) diff --git a/converters/dbt/src/osi_dbt/osi_to_msi.py b/converters/dbt/src/osi_dbt/osi_to_msi.py new file mode 100644 index 00000000..5eb9e493 --- /dev/null +++ b/converters/dbt/src/osi_dbt/osi_to_msi.py @@ -0,0 +1,389 @@ +from dataclasses import dataclass +from typing import List, Optional, Set + +from osi import ( + OSIDataset, + OSIDialect, + OSIDocument, + OSIExpression, + OSIField, + OSISemanticModel, +) +from osi_dbt.converter_issues import ConverterResult +from osi_dbt.expression_utils import ( + _extract_agg_info, + _get_raw_inner_col, + _strip_qualifier, + _try_parse_ratio, +) + +from metricflow_semantic_interfaces.implementations.elements.dimension import ( + PydanticDimension, + PydanticDimensionTypeParams, +) +from metricflow_semantic_interfaces.implementations.elements.entity import PydanticEntity +from metricflow_semantic_interfaces.implementations.elements.measure import ( + PydanticMeasureAggregationParameters, +) +from metricflow_semantic_interfaces.implementations.metric import ( + PydanticMetric, + PydanticMetricAggregationParams, + PydanticMetricInput, + PydanticMetricTypeParams, +) +from metricflow_semantic_interfaces.implementations.project_configuration import ( + PydanticProjectConfiguration, +) +from metricflow_semantic_interfaces.implementations.semantic_manifest import ( + PydanticSemanticManifest, +) +from metricflow_semantic_interfaces.implementations.semantic_model import ( + PydanticNodeRelation, + PydanticSemanticModel, +) +from metricflow_semantic_interfaces.type_enums import ( + AggregationType, + DimensionType, + EntityType, + MetricType, + TimeGranularity, +) + + +@dataclass(frozen=True) +class _KeySets: + primary: Set[str] + unique: Set[str] + foreign: Set[str] + + +class OSIToMSIConverter: + """Converts an OSI Document into a PydanticSemanticManifest. + + The conversion is inherently lossy: OSI stores metrics as raw SQL expressions + and carries no metric-type metadata (SIMPLE / RATIO / CUMULATIVE / …). The + converter reconstructs a best-effort MSI manifest using the following rules: + + * Datasets → one PydanticSemanticModel each. + * Fields are classified as entities or dimensions using key and relationship + metadata. Aggregation info now lives directly on metrics (via + `metric_aggregation_params`), not on semantic model measures. + * Time dimensions always receive `TimeGranularity.DAY` — OSI has no + granularity field. + * Metric expressions are parsed with sqlglot: + - single-agg patterns (`SUM(col)`, `COUNT(DISTINCT col)`, …) → SIMPLE + metric with `metric_aggregation_params` (no measure reference needed) + - `(expr_a) / (expr_b)` → RATIO (with auto-generated sub-metrics) + - anything else → SIMPLE with the raw expression stored in `expr` + """ + + def __init__(self, dialect: OSIDialect = OSIDialect.ANSI_SQL) -> None: + self._dialect = dialect + + def convert(self, document: OSIDocument) -> ConverterResult[PydanticSemanticManifest]: + semantic_models: List[PydanticSemanticModel] = [] + metrics: List[PydanticMetric] = [] + + for osi_sm in document.semantic_model: + for dataset in osi_sm.datasets: + semantic_models.append(self._convert_dataset(dataset, osi_sm)) + metrics.extend(self._convert_metrics(osi_sm)) + + return ConverterResult( + output=PydanticSemanticManifest( + semantic_models=semantic_models, + metrics=metrics, + project_configuration=PydanticProjectConfiguration(), + ), + issues=[], + ) + + # ------------------------------------------------------------------ + # Dataset conversion + # ------------------------------------------------------------------ + + def _convert_dataset( + self, + dataset: OSIDataset, + osi_sm: OSISemanticModel, + ) -> PydanticSemanticModel: + key_sets = self._build_key_sets(dataset, osi_sm) + + entities: List[PydanticEntity] = [] + dimensions: List[PydanticDimension] = [] + + for field in dataset.fields or []: + expr = self._get_expression(field.expression) + self._classify_field( + field, + expr, + expr if expr != field.name else None, + key_sets.primary, + key_sets.unique, + key_sets.foreign, + entities, + dimensions, + ) + + return PydanticSemanticModel( + name=dataset.name, + node_relation=self._parse_source(dataset.source), + description=dataset.description, + entities=entities, + dimensions=dimensions, + measures=[], + ) + + @staticmethod + def _build_key_sets(dataset: OSIDataset, osi_sm: OSISemanticModel) -> _KeySets: + """Return a _KeySets with primary, unique, and foreign key column sets for a dataset.""" + return _KeySets( + primary=set(dataset.primary_key or []), + unique={col for keys in (dataset.unique_keys or []) for col in keys}, + foreign={ + col + for rel in (osi_sm.relationships or []) + if rel.from_dataset == dataset.name + for col in rel.from_columns + }, + ) + + def _classify_field( + self, + field: OSIField, + expr: str, + expr_or_none: Optional[str], + primary_key_cols: Set[str], + unique_key_cols: Set[str], + foreign_key_cols: Set[str], + entities: List[PydanticEntity], + dimensions: List[PydanticDimension], + ) -> None: + """Classify a single OSI field and append it to the appropriate list. + + Classification order (first match wins): + 1. primary_key → PRIMARY entity + 2. unique_keys → UNIQUE entity + 3. foreign key (from relationship) → FOREIGN entity + 4. dimension.is_time → TIME dimension (granularity defaults to DAY) + 5. fallback → CATEGORICAL dimension + + Aggregation info lives on metrics (`metric_aggregation_params`), not on + semantic model measures, so there is no measure classification step. + """ + if field.name in primary_key_cols: + entities.append( + PydanticEntity( + name=field.name, + type=EntityType.PRIMARY, + expr=expr_or_none, + description=field.description, + label=field.label, + role=None, + config=None, + ) + ) + return + if field.name in unique_key_cols: + entities.append( + PydanticEntity( + name=field.name, + type=EntityType.UNIQUE, + expr=expr_or_none, + description=field.description, + label=field.label, + role=None, + config=None, + ) + ) + return + if field.name in foreign_key_cols: + entities.append( + PydanticEntity( + name=field.name, + type=EntityType.FOREIGN, + expr=expr_or_none, + description=field.description, + label=field.label, + role=None, + config=None, + ) + ) + return + if field.dimension is not None and field.dimension.is_time: + # OSI carries no granularity metadata; default to DAY. + dimensions.append( + PydanticDimension( + name=field.name, + type=DimensionType.TIME, + type_params=PydanticDimensionTypeParams(time_granularity=TimeGranularity.DAY), + expr=expr_or_none, + description=field.description, + label=field.label, + config=None, + ) + ) + return + dimensions.append( + PydanticDimension( + name=field.name, + type=DimensionType.CATEGORICAL, + type_params=None, + expr=expr_or_none, + description=field.description, + label=field.label, + config=None, + ) + ) + + # ------------------------------------------------------------------ + # Metric conversion + # ------------------------------------------------------------------ + + def _convert_metrics(self, osi_sm: OSISemanticModel) -> List[PydanticMetric]: + metrics: List[PydanticMetric] = [] + for metric in osi_sm.metrics or []: + expr_str = self._get_expression(metric.expression) + metrics.extend(self._convert_metric(metric.name, expr_str, metric.description, osi_sm.datasets)) + return metrics + + def _convert_metric( + self, + name: str, + expr_str: str, + description: Optional[str], + datasets: List[OSIDataset], + ) -> List[PydanticMetric]: + """Return one or more PydanticMetric objects for the given OSI expression. + + Simple metrics use `metric_aggregation_params` to store aggregation type + and column expression directly — no intermediate measure is created. + + Multiple metrics are returned when a RATIO metric requires auto-generated + sub-metrics for its numerator and denominator. + """ + # --- SIMPLE: single aggregation --- + agg_result = _extract_agg_info(expr_str) + if agg_result is not None: + agg, col, percentile = agg_result + semantic_model_name = self._find_dataset_for_col(expr_str, col, datasets) + agg_params = PydanticMeasureAggregationParameters(percentile=percentile) if percentile is not None else None + return [ + PydanticMetric( + name=name, + description=description, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + expr=col, + metric_aggregation_params=PydanticMetricAggregationParams( + semantic_model=semantic_model_name, + agg=agg, + agg_params=agg_params, + agg_time_dimension=None, + non_additive_dimension=None, + ), + ), + filter=None, + metadata=None, + config=None, + ) + ] + + # --- RATIO: (num_expr) / (den_expr) --- + ratio_result = _try_parse_ratio(expr_str) + if ratio_result is not None: + num_expr, den_expr = ratio_result + num_name = f"{name}__numerator" + den_name = f"{name}__denominator" + num_metrics = self._convert_metric(num_name, num_expr, None, datasets) + den_metrics = self._convert_metric(den_name, den_expr, None, datasets) + ratio_metric = PydanticMetric( + name=name, + description=description, + type=MetricType.RATIO, + type_params=PydanticMetricTypeParams( + numerator=PydanticMetricInput(name=num_name, filter=None, alias=None), + denominator=PydanticMetricInput(name=den_name, filter=None, alias=None), + ), + filter=None, + metadata=None, + config=None, + ) + return [*num_metrics, *den_metrics, ratio_metric] + + # --- Fallback: complex expression that can't be decomposed --- + # Store the raw expression in `expr` with a best-guess aggregation type. + # The caller is responsible for reviewing and correcting these metrics. + fallback_dataset = datasets[0].name if datasets else "" + return [ + PydanticMetric( + name=name, + description=description, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + expr=expr_str, + metric_aggregation_params=PydanticMetricAggregationParams( + semantic_model=fallback_dataset, + agg=AggregationType.SUM, + agg_params=None, + agg_time_dimension=None, + non_additive_dimension=None, + ), + ), + filter=None, + metadata=None, + config=None, + ) + ] + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + @staticmethod + def _find_dataset_for_col( + raw_expr_str: str, + bare_col: str, + datasets: List[OSIDataset], + ) -> str: + """Determine which dataset a column belongs to for `metric_aggregation_params.semantic_model`. + + For qualified references like `SUM(orders.amount)` the qualifier is used directly. + For unqualified references the datasets are scanned for a matching field name. + Falls back to the first dataset's name if no match is found. + """ + # Check for a dataset qualifier in the raw expression (e.g. "orders.amount") + raw_inner = _get_raw_inner_col(raw_expr_str) + if raw_inner and "." in raw_inner: + ds_name, _ = raw_inner.rsplit(".", 1) + return ds_name + + # Scan datasets for a field whose name or expression matches the bare column + for dataset in datasets: + for field in dataset.fields or []: + if field.name == bare_col: + return dataset.name + field_expr = field.expression.dialects[0].expression if field.expression.dialects else "" + if _strip_qualifier(field_expr) == bare_col: + return dataset.name + + return datasets[0].name if datasets else "" + + def _get_expression(self, osi_expr: OSIExpression) -> str: + """Return the expression string for the preferred dialect (fallback: first available).""" + for dialect_expr in osi_expr.dialects: + if dialect_expr.dialect is self._dialect: + return dialect_expr.expression + return osi_expr.dialects[0].expression if osi_expr.dialects else "" + + @staticmethod + def _parse_source(source: str) -> PydanticNodeRelation: + """Parse `schema.table` or `db.schema.table` into a PydanticNodeRelation.""" + parts = source.split(".") + if len(parts) >= 3: + database, schema, alias = parts[0], parts[1], ".".join(parts[2:]) + return PydanticNodeRelation(alias=alias, schema_name=schema, database=database) + if len(parts) == 2: + schema, alias = parts + return PydanticNodeRelation(alias=alias, schema_name=schema) + return PydanticNodeRelation(alias=source, schema_name="") diff --git a/converters/dbt/tests/__init__.py b/converters/dbt/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/converters/dbt/tests/helpers.py b/converters/dbt/tests/helpers.py new file mode 100644 index 00000000..534b4587 --- /dev/null +++ b/converters/dbt/tests/helpers.py @@ -0,0 +1,212 @@ +"""Shared test helpers for OSI converter tests.""" + +from osi import ( + OSIDataset, + OSIDialect, + OSIDialectExpression, + OSIDimension, + OSIDocument, + OSIExpression, + OSIField, + OSIMetric, + OSIRelationship, + OSISemanticModel, +) +from metricflow_semantic_interfaces.implementations.elements.dimension import ( + PydanticDimension, + PydanticDimensionTypeParams, +) +from metricflow_semantic_interfaces.implementations.elements.entity import PydanticEntity +from metricflow_semantic_interfaces.implementations.elements.measure import PydanticMeasure +from metricflow_semantic_interfaces.implementations.filters.where_filter import ( + PydanticWhereFilter, + PydanticWhereFilterIntersection, +) +from metricflow_semantic_interfaces.implementations.metric import ( + PydanticMetric, + PydanticMetricInputMeasure, + PydanticMetricTypeParams, +) +from metricflow_semantic_interfaces.implementations.project_configuration import ( + PydanticProjectConfiguration, +) +from metricflow_semantic_interfaces.implementations.semantic_manifest import ( + PydanticSemanticManifest, +) +from metricflow_semantic_interfaces.test_utils import default_meta +from metricflow_semantic_interfaces.type_enums import ( + AggregationType, + DimensionType, + EntityType, + MetricType, + TimeGranularity, +) + +# --------------------------------------------------------------------------- +# MSI builders +# --------------------------------------------------------------------------- + + +def _manifest( + semantic_models: list | None = None, + metrics: list[PydanticMetric] | None = None, +) -> PydanticSemanticManifest: + return PydanticSemanticManifest( + semantic_models=semantic_models or [], + metrics=metrics or [], + project_configuration=PydanticProjectConfiguration(), + ) + + +def _simple_metric( + name: str, + measure_name: str, + description: str | None = None, +) -> PydanticMetric: + return PydanticMetric( + name=name, + description=description, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + measure=PydanticMetricInputMeasure(name=measure_name), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + + +def _dimension( + name: str, + dim_type: DimensionType = DimensionType.CATEGORICAL, + expr: str | None = None, + description: str | None = None, + label: str | None = None, + granularity: TimeGranularity | None = None, +) -> PydanticDimension: + type_params = PydanticDimensionTypeParams(time_granularity=granularity) if granularity else None + return PydanticDimension( + name=name, + type=dim_type, + expr=expr, + description=description, + label=label, + type_params=type_params, + metadata=default_meta(), + config=None, + ) + + +def _measure( + name: str, + agg: AggregationType = AggregationType.SUM, + expr: str | None = None, + description: str | None = None, + label: str | None = None, +) -> PydanticMeasure: + return PydanticMeasure( + name=name, + agg=agg, + expr=expr, + description=description, + label=label, + create_metric=None, + agg_params=None, + metadata=default_meta(), + ) + + +def _entity( + name: str, + entity_type: EntityType = EntityType.PRIMARY, + expr: str | None = None, +) -> PydanticEntity: + return PydanticEntity( + name=name, + type=entity_type, + expr=expr, + description=None, + role=None, + config=None, + ) + + +def _filter(sql: str) -> PydanticWhereFilterIntersection: + return PydanticWhereFilterIntersection(where_filters=[PydanticWhereFilter(where_sql_template=sql)]) + + +# --------------------------------------------------------------------------- +# OSI builders +# --------------------------------------------------------------------------- + + +def _osi_expr(expression: str, dialect: OSIDialect = OSIDialect.ANSI_SQL) -> OSIExpression: + return OSIExpression(dialects=[OSIDialectExpression(dialect=dialect, expression=expression)]) + + +def _osi_field( + name: str, + expression: str | None = None, + is_time: bool | None = None, + description: str | None = None, + label: str | None = None, +) -> OSIField: + return OSIField( + name=name, + expression=_osi_expr(expression if expression is not None else name), + dimension=OSIDimension(is_time=is_time) if is_time is not None else None, + description=description, + label=label, + ) + + +def _osi_dataset( + name: str, + source: str = "schema.table", + fields: list[OSIField] | None = None, + primary_key: list[str] | None = None, + unique_keys: list[list[str]] | None = None, + description: str | None = None, +) -> OSIDataset: + return OSIDataset( + name=name, + source=source, + fields=fields, + primary_key=primary_key, + unique_keys=unique_keys, + description=description, + ) + + +def _osi_metric(name: str, expression: str, description: str | None = None) -> OSIMetric: + return OSIMetric(name=name, expression=_osi_expr(expression), description=description) + + +def _osi_relationship( + name: str, from_dataset: str, to_dataset: str, from_columns: list[str], to_columns: list[str] +) -> OSIRelationship: + return OSIRelationship( + name=name, + from_dataset=from_dataset, + to=to_dataset, + from_columns=from_columns, + to_columns=to_columns, + ) + + +def _osi_doc( + datasets: list[OSIDataset] | None = None, + metrics: list[OSIMetric] | None = None, + relationships: list[OSIRelationship] | None = None, + model_name: str = "test", +) -> OSIDocument: + return OSIDocument( + semantic_model=[ + OSISemanticModel( + name=model_name, + datasets=datasets or [], + metrics=metrics if metrics else None, + relationships=relationships if relationships else None, + ) + ] + ) diff --git a/converters/dbt/tests/test_msi_to_osi.py b/converters/dbt/tests/test_msi_to_osi.py new file mode 100644 index 00000000..686106bb --- /dev/null +++ b/converters/dbt/tests/test_msi_to_osi.py @@ -0,0 +1,1160 @@ +import json +from typing import List, Optional + +import pytest +from syrupy.assertion import SnapshotAssertion + +from osi_dbt.converter_issues import ConverterIssueType +from osi_dbt.filter_utils import _render_filter_template +from osi import OSIDialect, OSIDocument +from osi_dbt.msi_to_osi import MSIToOSIConverter +from metricflow_semantic_interfaces.implementations.metric import ( + PydanticConversionTypeParams, + PydanticCumulativeTypeParams, + PydanticMetric, + PydanticMetricAggregationParams, + PydanticMetricInput, + PydanticMetricInputMeasure, + PydanticMetricTimeWindow, + PydanticMetricTypeParams, +) +from metricflow_semantic_interfaces.implementations.semantic_model import ( + PydanticNodeRelation, + PydanticSemanticModel, +) +from metricflow_semantic_interfaces.test_utils import default_meta, semantic_model_with_guaranteed_meta +from metricflow_semantic_interfaces.type_enums import ( + AggregationType, + DimensionType, + EntityType, + MetricType, + TimeGranularity, +) +from tests.helpers import ( + _dimension, + _entity, + _filter, + _manifest, + _measure, + _simple_metric, +) + +# --------------------------------------------------------------------------- +# Result navigation helpers +# --------------------------------------------------------------------------- + + +def _fields(result: OSIDocument, dataset_idx: int = 0) -> list: + """Return fields for a dataset, asserting they exist.""" + fields = result.semantic_model[0].datasets[dataset_idx].fields + assert fields is not None + return fields + + +def _field_expr(result: OSIDocument, field_idx: int = 0) -> str: + """Return the ANSI SQL expression for a field by index.""" + return _fields(result)[field_idx].expression.dialects[0].expression + + +def _osi_metrics(result: OSIDocument) -> list: + """Return OSI metrics for the first semantic model, asserting they exist.""" + metrics = result.semantic_model[0].metrics + assert metrics is not None + return metrics + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestBasicConversion: + def test_empty_manifest_produces_empty_datasets(self) -> None: + result = MSIToOSIConverter().convert(_manifest(), osi_model_name="test").output + + assert result.version == "0.2.0.dev0" + assert len(result.semantic_model) == 1 + assert result.semantic_model[0].name == "test" + assert result.semantic_model[0].datasets == [] + assert result.semantic_model[0].metrics is None + assert result.semantic_model[0].relationships is None + + def test_semantic_model_becomes_dataset(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + description="Order data", + node_relation=PydanticNodeRelation(schema_name="analytics", alias="orders_table"), + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + dataset = result.semantic_model[0].datasets[0] + assert dataset.name == "orders" + assert dataset.source == "analytics.orders_table" + assert dataset.description == "Order data" + + def test_source_includes_database_when_present(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + node_relation=PydanticNodeRelation(schema_name="analytics", alias="orders_table", database="prod"), + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert result.semantic_model[0].datasets[0].source == "prod.analytics.orders_table" + + def test_multiple_semantic_models_become_multiple_datasets(self) -> None: + sm_a = semantic_model_with_guaranteed_meta(name="orders") + sm_b = semantic_model_with_guaranteed_meta(name="users") + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm_a, sm_b])).output + + names = [ds.name for ds in result.semantic_model[0].datasets] + assert names == ["orders", "users"] + + +class TestDimensionConversion: + def test_categorical_dimension_has_is_time_false(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("status", dim_type=DimensionType.CATEGORICAL)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.name == "status" + assert field.dimension is not None + assert field.dimension.is_time is False + + def test_time_dimension_has_is_time_true(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("ds", dim_type=DimensionType.TIME, granularity=TimeGranularity.DAY)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.dimension is not None + assert field.dimension.is_time is True + + def test_dimension_with_expr_uses_expr(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("order_date", expr="DATE(created_at)")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert _field_expr(result) == "DATE(created_at)" + + def test_dimension_without_expr_falls_back_to_name(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("status")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert _field_expr(result) == "status" + + def test_dimension_description_and_label_carried_over(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("status", description="Order status", label="Status")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.description == "Order status" + assert field.label == "Status" + + +class TestMeasureConversion: + def test_measure_becomes_field_without_dimension_metadata(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.name == "revenue" + assert field.expression.dialects[0].expression == "amount" + assert field.dimension is None + + def test_measure_without_expr_falls_back_to_name(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("num_orders", agg=AggregationType.SUM)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert _field_expr(result) == "num_orders" + + def test_measure_description_and_label_carried_over(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", description="Total revenue", label="Revenue")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.description == "Total revenue" + assert field.label == "Revenue" + + +class TestEntityConversion: + def test_entity_becomes_field(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("order_id", entity_type=EntityType.PRIMARY)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.name == "order_id" + assert field.expression.dialects[0].expression == "order_id" + assert field.dimension is None + + def test_entity_with_expr_uses_expr(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("order_id", entity_type=EntityType.PRIMARY, expr="id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + field = _fields(result)[0] + assert field.name == "order_id" + assert field.expression.dialects[0].expression == "id" + + def test_foreign_entity_also_becomes_field(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("user_id", entity_type=EntityType.FOREIGN)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert _fields(result)[0].name == "user_id" + + +class TestEntityKeyExtraction: + @pytest.mark.parametrize( + "entity_type, name, expr, expected_pk, expected_uk", + [ + (EntityType.PRIMARY, "order_id", None, ["order_id"], None), + (EntityType.PRIMARY, "order_id", "id", ["id"], None), + (EntityType.UNIQUE, "email", None, None, [["email"]]), + (EntityType.FOREIGN, "user_id", None, None, None), + ], + ) + def test_key_extraction( + self, + entity_type: EntityType, + name: str, + expr: Optional[str], + expected_pk: Optional[List[str]], + expected_uk: Optional[List[List[str]]], + ) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity(name, entity_type=entity_type, expr=expr)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + dataset = result.semantic_model[0].datasets[0] + assert dataset.primary_key == expected_pk + assert dataset.unique_keys == expected_uk + + +class TestFieldOrdering: + def test_entities_then_dimensions_then_measures(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("order_id", entity_type=EntityType.PRIMARY)], + dimensions=[_dimension("status")], + measures=[_measure("revenue", expr="amount")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + fields = _fields(result) + assert fields[0].name == "order_id" + assert fields[1].name == "status" + assert fields[2].name == "revenue" + + +class TestDialectConfiguration: + def test_default_dialect_is_ansi_sql(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("status")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert result.dialects == [OSIDialect.ANSI_SQL] + assert _fields(result)[0].expression.dialects[0].dialect == OSIDialect.ANSI_SQL + + def test_configurable_dialect(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + dimensions=[_dimension("status")], + ) + result = MSIToOSIConverter(dialect=OSIDialect.SNOWFLAKE).convert(_manifest(semantic_models=[sm])).output + + assert result.dialects == [OSIDialect.SNOWFLAKE] + assert _fields(result)[0].expression.dialects[0].dialect == OSIDialect.SNOWFLAKE + + +class TestRelationshipConversion: + def test_shared_entity_name_produces_relationship(self) -> None: + listings = semantic_model_with_guaranteed_meta( + name="listings", + entities=[_entity("listing", entity_type=EntityType.PRIMARY, expr="listing_id")], + ) + bookings = semantic_model_with_guaranteed_meta( + name="bookings", + entities=[_entity("listing", entity_type=EntityType.FOREIGN, expr="listing_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[listings, bookings])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + assert len(rels) == 1 + rel = rels[0] + assert rel.from_dataset == "bookings" + assert rel.to == "listings" + assert rel.from_columns == ["listing_id"] + assert rel.to_columns == ["listing_id"] + + def test_same_type_entities_produce_relationship(self) -> None: + users_a = semantic_model_with_guaranteed_meta( + name="users_a", + entities=[_entity("user", entity_type=EntityType.PRIMARY, expr="user_id")], + ) + users_b = semantic_model_with_guaranteed_meta( + name="users_b", + entities=[_entity("user", entity_type=EntityType.PRIMARY, expr="uid")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[users_a, users_b])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + assert len(rels) == 1 + assert rels[0].from_columns == ["user_id"] + assert rels[0].to_columns == ["uid"] + + def test_single_dataset_with_entity_produces_no_relationship(self) -> None: + bookings = semantic_model_with_guaranteed_meta( + name="bookings", + entities=[_entity("listing", entity_type=EntityType.FOREIGN, expr="listing_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[bookings])).output + + assert result.semantic_model[0].relationships is None + + def test_same_dataset_entities_excluded(self) -> None: + orders = semantic_model_with_guaranteed_meta( + name="orders", + entities=[ + _entity("order", entity_type=EntityType.PRIMARY, expr="order_id"), + _entity("order", entity_type=EntityType.FOREIGN, expr="order_id"), + ], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[orders])).output + + assert result.semantic_model[0].relationships is None + + def test_three_datasets_produce_all_pairs(self, snapshot: SnapshotAssertion) -> None: + users_a = semantic_model_with_guaranteed_meta( + name="users_a", + entities=[_entity("user", entity_type=EntityType.PRIMARY, expr="user_id")], + ) + users_b = semantic_model_with_guaranteed_meta( + name="users_b", + entities=[_entity("user", entity_type=EntityType.UNIQUE, expr="user_id")], + ) + orders = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("user", entity_type=EntityType.FOREIGN, expr="user_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[users_a, users_b, orders])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + assert len(rels) == 3 + pairs = {(r.from_dataset, r.to) for r in rels} + assert pairs == {("users_a", "users_b"), ("orders", "users_a"), ("orders", "users_b")} + assert result.to_osi_yaml() == snapshot + + def test_columns_use_expr_when_present(self) -> None: + listings = semantic_model_with_guaranteed_meta( + name="listings", + entities=[_entity("listing", entity_type=EntityType.PRIMARY, expr="lid")], + ) + bookings = semantic_model_with_guaranteed_meta( + name="bookings", + entities=[_entity("listing", entity_type=EntityType.FOREIGN, expr="fk_lid")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[listings, bookings])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + rel = rels[0] + assert rel.from_columns == ["fk_lid"] + assert rel.to_columns == ["lid"] + + def test_columns_fall_back_to_name_without_expr(self) -> None: + listings = semantic_model_with_guaranteed_meta( + name="listings", + entities=[_entity("listing", entity_type=EntityType.PRIMARY)], + ) + bookings = semantic_model_with_guaranteed_meta( + name="bookings", + entities=[_entity("listing", entity_type=EntityType.FOREIGN)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[listings, bookings])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + assert rels[0].from_columns == ["listing"] + assert rels[0].to_columns == ["listing"] + + def test_primary_entity_shorthand_does_not_produce_relationship(self) -> None: + bookings = PydanticSemanticModel( + name="bookings", + description=None, + node_relation=PydanticNodeRelation(schema_name="schema", alias="table"), + primary_entity="booking", + entities=[], + metadata=default_meta(), + ) + orders = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("booking", entity_type=EntityType.FOREIGN, expr="booking_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[bookings, orders])).output + + assert result.semantic_model[0].relationships is None + + def test_relationship_name_format(self) -> None: + listings = semantic_model_with_guaranteed_meta( + name="listings", + entities=[_entity("listing", entity_type=EntityType.PRIMARY, expr="listing_id")], + ) + bookings = semantic_model_with_guaranteed_meta( + name="bookings", + entities=[_entity("listing", entity_type=EntityType.FOREIGN, expr="listing_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[listings, bookings])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + assert rels[0].name == "bookings__listings__listing" + + def test_natural_entity_excluded(self) -> None: + users = semantic_model_with_guaranteed_meta( + name="users", + entities=[_entity("user", entity_type=EntityType.NATURAL, expr="user_id")], + ) + orders = semantic_model_with_guaranteed_meta( + name="orders", + entities=[_entity("user", entity_type=EntityType.FOREIGN, expr="user_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[users, orders])).output + + assert result.semantic_model[0].relationships is None + + def test_direction_based_on_entity_type_not_manifest_order(self) -> None: + beta = semantic_model_with_guaranteed_meta( + name="beta", + entities=[_entity("shared", entity_type=EntityType.PRIMARY, expr="col_b")], + ) + alpha = semantic_model_with_guaranteed_meta( + name="alpha", + entities=[_entity("shared", entity_type=EntityType.FOREIGN, expr="col_a")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[beta, alpha])).output + + rels = result.semantic_model[0].relationships + assert rels is not None + assert rels[0].from_dataset == "alpha" + assert rels[0].to == "beta" + + +class TestMetricConversion: + # --- SIMPLE --- + + def test_simple_metric_resolves_through_measure(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + metric = _simple_metric("revenue", measure_name="revenue") + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + metrics = _osi_metrics(result) + assert len(metrics) == 1 + assert metrics[0].name == "revenue" + assert metrics[0].expression.dialects[0].expression == "SUM(orders.amount)" + + def test_simple_metric_description_carried_over(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + metric = _simple_metric("revenue", measure_name="revenue", description="Total revenue") + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + assert _osi_metrics(result)[0].description == "Total revenue" + + def test_simple_metric_with_metric_aggregation_params(self) -> None: + sm = semantic_model_with_guaranteed_meta(name="orders") + metric = PydanticMetric( + name="avg_price", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + expr="price", + metric_aggregation_params=PydanticMetricAggregationParams( + semantic_model="orders", + agg=AggregationType.AVERAGE, + agg_params=None, + agg_time_dimension=None, + non_additive_dimension=None, + ), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + assert _osi_metrics(result)[0].expression.dialects[0].expression == "AVG(orders.price)" + + # --- RATIO --- + + def test_ratio_metric_inlines_sub_expressions(self, snapshot: SnapshotAssertion) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("order_count", agg=AggregationType.COUNT, expr="order_id"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + order_count_m = _simple_metric("order_count", "order_count") + arpu = PydanticMetric( + name="arpu", + description=None, + type=MetricType.RATIO, + type_params=PydanticMetricTypeParams( + numerator=PydanticMetricInput(name="revenue"), + denominator=PydanticMetricInput(name="order_count"), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter() + .convert(_manifest(semantic_models=[sm], metrics=[revenue_m, order_count_m, arpu])) + .output + ) + + arpu_osi = next(m for m in _osi_metrics(result) if m.name == "arpu") + assert arpu_osi.expression.dialects[0].expression == ( + "(SUM(orders.amount)) / (SUM(CASE WHEN orders.order_id IS NOT NULL THEN 1 ELSE 0 END))" + ) + assert result.to_osi_yaml() == snapshot + + # --- DERIVED --- + + def test_derived_metric_inlines_sub_expressions(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("cost", agg=AggregationType.SUM, expr="cost_amount"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + cost_m = _simple_metric("cost", "cost") + profit = PydanticMetric( + name="profit", + description=None, + type=MetricType.DERIVED, + type_params=PydanticMetricTypeParams( + expr="revenue - cost", + metrics=[ + PydanticMetricInput(name="revenue"), + PydanticMetricInput(name="cost"), + ], + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, cost_m, profit])).output + ) + + profit_osi = next(m for m in _osi_metrics(result) if m.name == "profit") + assert profit_osi.expression.dialects[0].expression == "SUM(orders.amount) - SUM(orders.cost_amount)" + + def test_derived_metric_uses_alias_for_substitution(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("cost", agg=AggregationType.SUM, expr="cost_amount"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + cost_m = _simple_metric("cost", "cost") + profit = PydanticMetric( + name="profit", + description=None, + type=MetricType.DERIVED, + type_params=PydanticMetricTypeParams( + expr="r - c", + metrics=[ + PydanticMetricInput(name="revenue", alias="r"), + PydanticMetricInput(name="cost", alias="c"), + ], + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, cost_m, profit])).output + ) + + profit_osi = next(m for m in _osi_metrics(result) if m.name == "profit") + assert profit_osi.expression.dialects[0].expression == "SUM(orders.amount) - SUM(orders.cost_amount)" + + def test_derived_metric_nested(self, snapshot: SnapshotAssertion) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("cost", agg=AggregationType.SUM, expr="cost_amount"), + _measure("expenses", agg=AggregationType.SUM, expr="expense_amount"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + cost_m = _simple_metric("cost", "cost") + expenses_m = _simple_metric("expenses", "expenses") + gross_profit = PydanticMetric( + name="gross_profit", + description=None, + type=MetricType.DERIVED, + type_params=PydanticMetricTypeParams( + expr="revenue - cost", + metrics=[ + PydanticMetricInput(name="revenue"), + PydanticMetricInput(name="cost"), + ], + ), + filter=None, + metadata=default_meta(), + config=None, + ) + net_profit = PydanticMetric( + name="net_profit", + description=None, + type=MetricType.DERIVED, + type_params=PydanticMetricTypeParams( + expr="gross_profit - expenses", + metrics=[ + PydanticMetricInput(name="gross_profit"), + PydanticMetricInput(name="expenses"), + ], + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter() + .convert( + _manifest( + semantic_models=[sm], + metrics=[revenue_m, cost_m, expenses_m, gross_profit, net_profit], + ) + ) + .output + ) + + net_osi = next(m for m in _osi_metrics(result) if m.name == "net_profit") + assert ( + net_osi.expression.dialects[0].expression + == "(SUM(orders.amount) - SUM(orders.cost_amount)) - SUM(orders.expense_amount)" + ) + assert result.to_osi_yaml() == snapshot + + def test_derived_metric_ref_not_corrupted_by_prefix_match(self) -> None: + """Substituting 'revenue' must not corrupt 'revenue_adjusted' when it appears in the same expr.""" + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("revenue_adjusted", agg=AggregationType.SUM, expr="adjusted_amount"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + revenue_adjusted_m = _simple_metric("revenue_adjusted", "revenue_adjusted") + derived = PydanticMetric( + name="revenue_delta", + description=None, + type=MetricType.DERIVED, + type_params=PydanticMetricTypeParams( + expr="revenue - revenue_adjusted", + metrics=[ + PydanticMetricInput(name="revenue"), + PydanticMetricInput(name="revenue_adjusted"), + ], + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter() + .convert(_manifest(semantic_models=[sm], metrics=[revenue_m, revenue_adjusted_m, derived])) + .output + ) + + derived_osi = next(m for m in _osi_metrics(result) if m.name == "revenue_delta") + assert derived_osi.expression.dialects[0].expression == "SUM(orders.amount) - SUM(orders.adjusted_amount)" + + # --- CUMULATIVE --- + + def test_cumulative_metric_uses_base_aggregation(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + cumulative = PydanticMetric( + name="cumulative_revenue", + description=None, + type=MetricType.CUMULATIVE, + type_params=PydanticMetricTypeParams( + measure=PydanticMetricInputMeasure(name="revenue"), + cumulative_type_params=PydanticCumulativeTypeParams( + window=PydanticMetricTimeWindow(count=7, granularity="day"), + ), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[cumulative])).output + + metrics = _osi_metrics(result) + assert len(metrics) == 1 + assert metrics[0].name == "cumulative_revenue" + assert metrics[0].expression.dialects[0].expression == "SUM(orders.amount)" + + def test_cumulative_metric_via_sub_metric_reference(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + base = PydanticMetric( + name="total_revenue", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams(measure=PydanticMetricInputMeasure(name="revenue")), + filter=None, + metadata=default_meta(), + config=None, + ) + cumulative = PydanticMetric( + name="cumulative_revenue", + description=None, + type=MetricType.CUMULATIVE, + type_params=PydanticMetricTypeParams( + cumulative_type_params=PydanticCumulativeTypeParams( + metric=PydanticMetricInput(name="total_revenue"), + ), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[base, cumulative])).output + + cumulative_osi = next(m for m in _osi_metrics(result) if m.name == "cumulative_revenue") + assert cumulative_osi.expression.dialects[0].expression == "SUM(orders.amount)" + + # --- Edge cases --- + + def test_no_metrics_produces_no_osi_metrics(self) -> None: + sm = semantic_model_with_guaranteed_meta(name="orders") + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + + assert result.semantic_model[0].metrics is None + + def test_multiple_metrics_all_converted(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("order_count", agg=AggregationType.COUNT, expr="order_id"), + ], + ) + result = ( + MSIToOSIConverter() + .convert( + _manifest( + semantic_models=[sm], + metrics=[ + _simple_metric("revenue", "revenue"), + _simple_metric("order_count", "order_count"), + ], + ) + ) + .output + ) + + metrics = _osi_metrics(result) + assert len(metrics) == 2 + assert {m.name for m in metrics} == {"revenue", "order_count"} + + def test_conversion_metric_skipped(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("visits", agg=AggregationType.COUNT, expr="visit_id"), + _measure("purchases", agg=AggregationType.COUNT, expr="purchase_id"), + ], + ) + conversion = PydanticMetric( + name="purchase_rate", + description=None, + type=MetricType.CONVERSION, + type_params=PydanticMetricTypeParams( + conversion_type_params=PydanticConversionTypeParams( + base_measure=PydanticMetricInputMeasure(name="visits"), + conversion_measure=PydanticMetricInputMeasure(name="purchases"), + entity="user", + ), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[conversion])).output + + assert result.semantic_model[0].metrics is None + + +class TestConverterIssues: + def test_conversion_metric_emits_issue(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("visits", agg=AggregationType.COUNT, expr="visit_id"), + _measure("purchases", agg=AggregationType.COUNT, expr="purchase_id"), + ], + ) + conversion = PydanticMetric( + name="purchase_rate", + description=None, + type=MetricType.CONVERSION, + type_params=PydanticMetricTypeParams( + conversion_type_params=PydanticConversionTypeParams( + base_measure=PydanticMetricInputMeasure(name="visits"), + conversion_measure=PydanticMetricInputMeasure(name="purchases"), + entity="user", + ), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[conversion])) + + dropped = [i for i in result.issues if i.issue_type == ConverterIssueType.CONVERSION_METRIC_DROPPED] + assert len(dropped) == 1 + assert dropped[0].element_name == "purchase_rate" + + def test_private_metric_emits_issue(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + private_metric = PydanticMetric( + name="revenue_internal", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + measure=PydanticMetricInputMeasure(name="revenue"), + is_private=True, + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[private_metric])) + + assert len(result.issues) == 1 + assert result.issues[0].issue_type == ConverterIssueType.PRIVATE_METRIC_DROPPED + assert result.issues[0].element_name == "revenue_internal" + + def test_natural_entity_emits_issue(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="users", + entities=[_entity("user", entity_type=EntityType.NATURAL, expr="user_id")], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])) + + assert len(result.issues) == 1 + assert result.issues[0].issue_type == ConverterIssueType.NATURAL_ENTITY_DROPPED + assert result.issues[0].element_name == "user" + + def test_cumulative_metric_emits_issue(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + base = _simple_metric("revenue", "revenue") + cumulative = PydanticMetric( + name="cumulative_revenue", + description=None, + type=MetricType.CUMULATIVE, + type_params=PydanticMetricTypeParams( + measure=PydanticMetricInputMeasure(name="revenue"), + cumulative_type_params=PydanticCumulativeTypeParams( + window=PydanticMetricTimeWindow(count=7, granularity="day"), + ), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[base, cumulative])) + + cumulative_issues = [i for i in result.issues if i.issue_type == ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS] + assert len(cumulative_issues) == 1 + assert cumulative_issues[0].element_name == "cumulative_revenue" + + +class TestFilterRendering: + """Unit tests for the Jinja → SQL rendering of where-filter templates.""" + + def test_plain_sql_passthrough(self) -> None: + assert _render_filter_template("status = 'paid'") == "status = 'paid'" + + def test_dimension_reference(self) -> None: + assert _render_filter_template("{{ Dimension('order__status') }} = 'paid'") == "order__status = 'paid'" + + def test_dimension_with_grain(self) -> None: + result = _render_filter_template("{{ Dimension('order__ds').grain('day') }} >= '2023-01-01'") + assert result == "order__ds__day >= '2023-01-01'" + + def test_time_dimension_with_grain_arg(self) -> None: + result = _render_filter_template("{{ TimeDimension('order__ds', 'week') }} >= '2023-01-01'") + assert result == "order__ds__week >= '2023-01-01'" + + def test_time_dimension_without_grain(self) -> None: + assert _render_filter_template("{{ TimeDimension('metric_time') }} IS NOT NULL") == "metric_time IS NOT NULL" + + def test_entity_reference(self) -> None: + assert _render_filter_template("{{ Entity('user') }} != 'bot'") == "user != 'bot'" + + def test_metric_reference(self) -> None: + assert _render_filter_template("{{ Metric('revenue') }} > 0") == "revenue > 0" + + +class TestMetricFilterFlattening: + def test_metric_level_filter_inlines_case_when(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + metric = PydanticMetric( + name="paid_revenue", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams(measure=PydanticMetricInputMeasure(name="revenue")), + filter=_filter("status = 'paid'"), + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + assert ( + _osi_metrics(result)[0].expression.dialects[0].expression + == "SUM(CASE WHEN status = 'paid' THEN orders.amount END)" + ) + + def test_measure_level_filter_inlines_case_when(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + metric = PydanticMetric( + name="paid_revenue", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + measure=PydanticMetricInputMeasure(name="revenue", filter=_filter("status = 'paid'")), + ), + filter=None, + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + assert ( + _osi_metrics(result)[0].expression.dialects[0].expression + == "SUM(CASE WHEN status = 'paid' THEN orders.amount END)" + ) + + def test_metric_and_measure_filters_combined_with_and(self, snapshot: SnapshotAssertion) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + metric = PydanticMetric( + name="paid_intl_revenue", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams( + measure=PydanticMetricInputMeasure(name="revenue", filter=_filter("status = 'paid'")), + ), + filter=_filter("region = 'intl'"), + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + assert _osi_metrics(result)[0].expression.dialects[0].expression == ( + "SUM(CASE WHEN (status = 'paid') AND (region = 'intl') THEN orders.amount END)" + ) + assert result.to_osi_yaml() == snapshot + + def test_jinja_dimension_reference_rendered(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + metric = PydanticMetric( + name="us_revenue", + description=None, + type=MetricType.SIMPLE, + type_params=PydanticMetricTypeParams(measure=PydanticMetricInputMeasure(name="revenue")), + filter=_filter("{{ Dimension('order__country') }} = 'US'"), + metadata=default_meta(), + config=None, + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[metric])).output + + assert ( + _osi_metrics(result)[0].expression.dialects[0].expression + == "SUM(CASE WHEN order__country = 'US' THEN orders.amount END)" + ) + + def test_ratio_metric_filter_propagated_to_both_sides(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("order_count", agg=AggregationType.COUNT, expr="order_id"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + order_count_m = _simple_metric("order_count", "order_count") + arpu = PydanticMetric( + name="paid_arpu", + description=None, + type=MetricType.RATIO, + type_params=PydanticMetricTypeParams( + numerator=PydanticMetricInput(name="revenue"), + denominator=PydanticMetricInput(name="order_count"), + ), + filter=_filter("status = 'paid'"), + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter() + .convert(_manifest(semantic_models=[sm], metrics=[revenue_m, order_count_m, arpu])) + .output + ) + + paid_arpu = next(m for m in _osi_metrics(result) if m.name == "paid_arpu") + assert paid_arpu.expression.dialects[0].expression == ( + "(SUM(CASE WHEN status = 'paid' THEN orders.amount END))" + " / " + "(SUM(CASE WHEN status = 'paid' THEN CASE WHEN orders.order_id IS NOT NULL THEN 1 ELSE 0 END END))" + ) + + def test_derived_metric_filter_propagated_to_sub_expressions(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[ + _measure("revenue", agg=AggregationType.SUM, expr="amount"), + _measure("cost", agg=AggregationType.SUM, expr="cost_amount"), + ], + ) + revenue_m = _simple_metric("revenue", "revenue") + cost_m = _simple_metric("cost", "cost") + profit = PydanticMetric( + name="paid_profit", + description=None, + type=MetricType.DERIVED, + type_params=PydanticMetricTypeParams( + expr="revenue - cost", + metrics=[PydanticMetricInput(name="revenue"), PydanticMetricInput(name="cost")], + ), + filter=_filter("status = 'paid'"), + metadata=default_meta(), + config=None, + ) + result = ( + MSIToOSIConverter().convert(_manifest(semantic_models=[sm], metrics=[revenue_m, cost_m, profit])).output + ) + + paid_profit = next(m for m in _osi_metrics(result) if m.name == "paid_profit") + assert paid_profit.expression.dialects[0].expression == ( + "SUM(CASE WHEN status = 'paid' THEN orders.amount END)" + " - " + "SUM(CASE WHEN status = 'paid' THEN orders.cost_amount END)" + ) + + def test_no_filter_produces_plain_expression(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + measures=[_measure("revenue", agg=AggregationType.SUM, expr="amount")], + ) + result = ( + MSIToOSIConverter() + .convert(_manifest(semantic_models=[sm], metrics=[_simple_metric("revenue", "revenue")])) + .output + ) + + assert _osi_metrics(result)[0].expression.dialects[0].expression == "SUM(orders.amount)" + + +class TestOSIJsonSerialization: + def test_to_osi_json_produces_valid_json(self) -> None: + sm = semantic_model_with_guaranteed_meta( + name="orders", + description="Order data", + dimensions=[_dimension("ds", dim_type=DimensionType.TIME, granularity=TimeGranularity.DAY)], + measures=[_measure("revenue", expr="amount")], + entities=[_entity("order_id", entity_type=EntityType.PRIMARY)], + ) + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm]), osi_model_name="my_project").output + parsed = json.loads(result.to_osi_json()) + + assert parsed["version"] == "0.2.0.dev0" + assert len(parsed["semantic_model"]) == 1 + assert parsed["semantic_model"][0]["name"] == "my_project" + + def test_to_osi_json_excludes_none_fields(self) -> None: + sm = semantic_model_with_guaranteed_meta(name="orders") + result = MSIToOSIConverter().convert(_manifest(semantic_models=[sm])).output + parsed = json.loads(result.to_osi_json()) + + dataset = parsed["semantic_model"][0]["datasets"][0] + assert "primary_key" not in dataset + assert "unique_keys" not in dataset + assert "fields" not in dataset diff --git a/converters/dbt/tests/test_osi_to_msi.py b/converters/dbt/tests/test_osi_to_msi.py new file mode 100644 index 00000000..afb35893 --- /dev/null +++ b/converters/dbt/tests/test_osi_to_msi.py @@ -0,0 +1,370 @@ +"""Tests for OSIToMSIConverter.""" + +import pytest +from syrupy.assertion import SnapshotAssertion + +from osi_dbt.msi_to_osi import MSIToOSIConverter +from osi_dbt.osi_to_msi import OSIToMSIConverter +from metricflow_semantic_interfaces.type_enums import ( + AggregationType, + DimensionType, + MetricType, +) +from tests.helpers import ( + _osi_dataset, + _osi_doc, + _osi_field, + _osi_metric, + _osi_relationship, +) + + +class TestOSIToMSIBasicConversion: + def test_empty_document_produces_empty_manifest(self) -> None: + result = OSIToMSIConverter().convert(_osi_doc()).output + + assert result.semantic_models == [] + assert result.metrics == [] + + def test_single_dataset_becomes_semantic_model(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("orders", source="analytics.orders_table")]) + result = OSIToMSIConverter().convert(doc).output + + assert len(result.semantic_models) == 1 + sm = result.semantic_models[0] + assert sm.name == "orders" + assert sm.node_relation.schema_name == "analytics" + assert sm.node_relation.alias == "orders_table" + assert sm.node_relation.database is None + + def test_description_carried_over(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("orders", description="Order data")]) + result = OSIToMSIConverter().convert(doc).output + + assert result.semantic_models[0].description == "Order data" + + def test_source_two_parts(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("t", source="myschema.mytable")]) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert sm.node_relation.schema_name == "myschema" + assert sm.node_relation.alias == "mytable" + assert sm.node_relation.database is None + + def test_source_three_parts(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("t", source="mydb.myschema.mytable")]) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert sm.node_relation.database == "mydb" + assert sm.node_relation.schema_name == "myschema" + assert sm.node_relation.alias == "mytable" + + def test_source_bare_name(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("t", source="mytable")]) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert sm.node_relation.alias == "mytable" + assert sm.node_relation.schema_name == "" + + def test_multiple_datasets_become_multiple_models(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("orders"), _osi_dataset("users")]) + result = OSIToMSIConverter().convert(doc).output + + names = [sm.name for sm in result.semantic_models] + assert names == ["orders", "users"] + + +class TestOSIToMSIFieldClassification: + def test_primary_key_field_becomes_primary_entity(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset( + "orders", + fields=[_osi_field("order_id")], + primary_key=["order_id"], + ) + ] + ) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert len(sm.entities) == 1 + assert sm.entities[0].name == "order_id" + assert sm.entities[0].type.value == "primary" + + def test_unique_key_field_becomes_unique_entity(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset( + "users", + fields=[_osi_field("email")], + unique_keys=[["email"]], + ) + ] + ) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert len(sm.entities) == 1 + assert sm.entities[0].name == "email" + assert sm.entities[0].type.value == "unique" + + def test_relationship_from_column_becomes_foreign_entity(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset("orders", fields=[_osi_field("user_id")]), + _osi_dataset("users", primary_key=["user_id"]), + ], + relationships=[_osi_relationship("r", "orders", "users", ["user_id"], ["user_id"])], + ) + orders_sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert len(orders_sm.entities) == 1 + assert orders_sm.entities[0].name == "user_id" + assert orders_sm.entities[0].type.value == "foreign" + + @pytest.mark.parametrize( + "field_name, is_time, expected_type", + [ + ("created_at", True, DimensionType.TIME), + ("status", False, DimensionType.CATEGORICAL), + ("region", None, DimensionType.CATEGORICAL), + ], + ) + def test_field_becomes_dimension_by_is_time( + self, field_name: str, is_time: bool | None, expected_type: DimensionType + ) -> None: + doc = _osi_doc(datasets=[_osi_dataset("orders", fields=[_osi_field(field_name, is_time=is_time)])]) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert len(sm.dimensions) == 1 + assert sm.dimensions[0].name == field_name + assert sm.dimensions[0].type == expected_type + + def test_field_referenced_in_metric_stays_as_dimension(self) -> None: + """Fields referenced in metric expressions are no longer promoted to measures.""" + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])], + metrics=[_osi_metric("revenue", "SUM(amount)")], + ) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert len(sm.measures) == 0 + assert len(sm.dimensions) == 1 + assert sm.dimensions[0].name == "amount" + + def test_expr_different_from_name_is_preserved(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset( + "orders", + fields=[_osi_field("order_id", expression="id")], + primary_key=["order_id"], + ) + ] + ) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert sm.entities[0].expr == "id" + + def test_expr_same_as_name_is_stored_as_none(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset( + "orders", + fields=[_osi_field("order_id")], + primary_key=["order_id"], + ) + ] + ) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + assert sm.entities[0].expr is None + + def test_description_and_label_carried_over_to_dimension(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset( + "orders", + fields=[_osi_field("status", description="Order status", label="Status")], + ) + ] + ) + sm = OSIToMSIConverter().convert(doc).output.semantic_models[0] + + dim = sm.dimensions[0] + assert dim.description == "Order status" + assert dim.label == "Status" + + +class TestOSIToMSIMetricConversion: + def test_sum_expression_produces_simple_metric(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])], + metrics=[_osi_metric("revenue", "SUM(amount)")], + ) + result = OSIToMSIConverter().convert(doc).output + + assert len(result.metrics) == 1 + m = result.metrics[0] + assert m.name == "revenue" + assert m.type == MetricType.SIMPLE + assert m.type_params.measure is None + assert m.type_params.metric_aggregation_params is not None + assert m.type_params.metric_aggregation_params.agg == AggregationType.SUM + assert m.type_params.expr == "amount" + assert m.type_params.metric_aggregation_params.semantic_model == "orders" + + def test_count_distinct_expression(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("user_id")])], + metrics=[_osi_metric("unique_users", "COUNT(DISTINCT user_id)")], + ) + result = OSIToMSIConverter().convert(doc).output + + m = result.metrics[0] + assert m.type_params.measure is None + assert m.type_params.metric_aggregation_params is not None + assert m.type_params.metric_aggregation_params.agg == AggregationType.COUNT_DISTINCT + assert m.type_params.expr == "user_id" + sm = result.semantic_models[0] + assert len(sm.measures) == 0 + + def test_ratio_expression_produces_ratio_metric(self) -> None: + doc = _osi_doc( + datasets=[ + _osi_dataset( + "orders", + fields=[_osi_field("amount"), _osi_field("order_id")], + ) + ], + metrics=[_osi_metric("arpu", "(SUM(amount)) / (COUNT(order_id))")], + ) + result = OSIToMSIConverter().convert(doc).output + + ratio = next(m for m in result.metrics if m.type == MetricType.RATIO) + assert ratio.name == "arpu" + assert ratio.type_params.numerator is not None + assert ratio.type_params.denominator is not None + assert ratio.type_params.numerator.name == "arpu__numerator" + assert ratio.type_params.denominator.name == "arpu__denominator" + + def test_ratio_sub_metrics_are_simple(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount"), _osi_field("cnt")])], + metrics=[_osi_metric("ratio", "(SUM(amount)) / (COUNT(cnt))")], + ) + result = OSIToMSIConverter().convert(doc).output + + simple_metrics = [m for m in result.metrics if m.type == MetricType.SIMPLE] + assert len(simple_metrics) == 2 + names = {m.name for m in simple_metrics} + assert names == {"ratio__numerator", "ratio__denominator"} + + def test_complex_expression_falls_back_to_simple_with_raw_expr(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders")], + metrics=[_osi_metric("complex", "SUM(a) + SUM(b)")], + ) + result = OSIToMSIConverter().convert(doc).output + + assert len(result.metrics) == 1 + m = result.metrics[0] + assert m.type == MetricType.SIMPLE + assert m.type_params.measure is None + assert m.type_params.metric_aggregation_params is not None + assert m.type_params.expr == "SUM(a) + SUM(b)" + + def test_metric_description_carried_over(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])], + metrics=[_osi_metric("revenue", "SUM(amount)", description="Total revenue")], + ) + result = OSIToMSIConverter().convert(doc).output + + assert result.metrics[0].description == "Total revenue" + + def test_no_metrics_produces_empty_list(self) -> None: + doc = _osi_doc(datasets=[_osi_dataset("orders")]) + result = OSIToMSIConverter().convert(doc).output + + assert result.metrics == [] + + def test_dataset_qualified_column_reference(self) -> None: + """A metric referencing 'dataset.col' should resolve the semantic_model to that dataset.""" + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])], + metrics=[_osi_metric("revenue", "SUM(orders.amount)")], + ) + result = OSIToMSIConverter().convert(doc).output + + m = result.metrics[0] + assert m.type_params.metric_aggregation_params is not None + assert m.type_params.metric_aggregation_params.semantic_model == "orders" + assert m.type_params.expr == "amount" + + def test_percentile_cont_0_5_produces_median(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])], + metrics=[_osi_metric("median_amount", "PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY amount)")], + ) + result = OSIToMSIConverter().convert(doc).output + + m = result.metrics[0] + assert m.type_params.metric_aggregation_params is not None + assert m.type_params.metric_aggregation_params.agg == AggregationType.MEDIAN + assert m.type_params.metric_aggregation_params.agg_params is None + assert m.type_params.expr == "amount" + + def test_percentile_cont_non_median_carries_percentile_param(self) -> None: + doc = _osi_doc( + datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])], + metrics=[_osi_metric("p95_amount", "PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY amount)")], + ) + result = OSIToMSIConverter().convert(doc).output + + m = result.metrics[0] + assert m.type_params.metric_aggregation_params is not None + assert m.type_params.metric_aggregation_params.agg == AggregationType.PERCENTILE + assert m.type_params.metric_aggregation_params.agg_params is not None + assert m.type_params.metric_aggregation_params.agg_params.percentile == 0.95 + assert m.type_params.expr == "amount" + + +class TestOSIToMSIRoundTrip: + def test_osi_to_msi_to_osi_preserves_structure(self, snapshot: SnapshotAssertion) -> None: + """OSI → MSI → OSI preserves dataset names, fields, and metric expressions.""" + original = _osi_doc( + datasets=[ + _osi_dataset( + "orders", + source="analytics.orders", + fields=[ + _osi_field("order_id"), + _osi_field("status"), + _osi_field("created_at", is_time=True), + _osi_field("amount"), + ], + primary_key=["order_id"], + ) + ], + metrics=[_osi_metric("revenue", "SUM(orders.amount)")], + ) + + msi = OSIToMSIConverter().convert(original).output + assert msi.semantic_models[0].measures == [] + + osi_doc = MSIToOSIConverter().convert(msi).output + + dataset = osi_doc.semantic_model[0].datasets[0] + assert dataset.name == "orders" + + field_names = {f.name for f in dataset.fields or []} + assert "order_id" in field_names + assert "status" in field_names + assert "created_at" in field_names + assert "amount" in field_names + + metrics = osi_doc.semantic_model[0].metrics or [] + assert len(metrics) == 1 + assert metrics[0].name == "revenue" + assert metrics[0].expression.dialects[0].expression == "SUM(orders.amount)" + assert osi_doc.to_osi_yaml() == snapshot diff --git a/python/pyproject.toml b/python/pyproject.toml new file mode 100644 index 00000000..baf099bb --- /dev/null +++ b/python/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "osi-python" +version = "0.2.0.dev0" +description = "Python types for the Open Semantic Interchange (OSI) specification" +requires-python = ">=3.11" +dependencies = [ + "pydantic>=2.0", + "PyYAML>=6.0", +] + +[project.license] +text = "Apache-2.0" + +[tool.hatch.build.targets.wheel] +packages = ["src/osi"] + +[tool.uv] +dev-dependencies = [ + "pytest>=8.0", +] diff --git a/python/src/osi/__init__.py b/python/src/osi/__init__.py new file mode 100644 index 00000000..00970fd6 --- /dev/null +++ b/python/src/osi/__init__.py @@ -0,0 +1,33 @@ +from osi.models import ( + OSIAIContext, + OSIAIContextObject, + OSICustomExtension, + OSIDataset, + OSIDialect, + OSIDialectExpression, + OSIDimension, + OSIDocument, + OSIExpression, + OSIField, + OSIMetric, + OSIRelationship, + OSISemanticModel, + OSIVendor, +) + +__all__ = [ + "OSIAIContext", + "OSIAIContextObject", + "OSICustomExtension", + "OSIDataset", + "OSIDialect", + "OSIDialectExpression", + "OSIDimension", + "OSIDocument", + "OSIExpression", + "OSIField", + "OSIMetric", + "OSIRelationship", + "OSISemanticModel", + "OSIVendor", +] diff --git a/python/src/osi/models.py b/python/src/osi/models.py new file mode 100644 index 00000000..766b6e0f --- /dev/null +++ b/python/src/osi/models.py @@ -0,0 +1,163 @@ +from enum import Enum +from typing import Any, Optional, Union + +import yaml +from pydantic import BaseModel, ConfigDict, Field + + +class OSIDialect(str, Enum): + """Supported SQL and expression language dialects.""" + + ANSI_SQL = "ANSI_SQL" + SNOWFLAKE = "SNOWFLAKE" + MDX = "MDX" + MAQL = "MAQL" + TABLEAU = "TABLEAU" + DATABRICKS = "DATABRICKS" + + +class OSIVendor(str, Enum): + """Vendors with supported custom extensions.""" + + COMMON = "COMMON" + SNOWFLAKE = "SNOWFLAKE" + SALESFORCE = "SALESFORCE" + DBT = "DBT" + DATABRICKS = "DATABRICKS" + GOODDATA = "GOODDATA" + + +class OSIAIContextObject(BaseModel): + """Structured AI context with instructions, synonyms, and examples.""" + + model_config = ConfigDict(frozen=True, extra="allow") + + instructions: Optional[str] = None + synonyms: Optional[tuple[str, ...]] = None + examples: Optional[tuple[str, ...]] = None + + +OSIAIContext = Union[str, OSIAIContextObject] + + +class OSICustomExtension(BaseModel): + """Vendor-specific metadata as a serialized JSON string.""" + + model_config = ConfigDict(frozen=True) + + vendor_name: OSIVendor + data: str + + +class OSIDialectExpression(BaseModel): + """Expression in a specific dialect.""" + + model_config = ConfigDict(frozen=True) + + dialect: OSIDialect + expression: str + + +class OSIExpression(BaseModel): + """Expression definition with multi-dialect support.""" + + model_config = ConfigDict(frozen=True) + + dialects: list[OSIDialectExpression] + + +class OSIDimension(BaseModel): + """Dimension metadata on a field.""" + + model_config = ConfigDict(frozen=True) + + is_time: Optional[bool] = None + + +class OSIField(BaseModel): + """Row-level attribute for grouping, filtering, and metric expressions.""" + + model_config = ConfigDict(frozen=True) + + name: str + expression: OSIExpression + dimension: Optional[OSIDimension] = None + label: Optional[str] = None + description: Optional[str] = None + ai_context: Optional[OSIAIContext] = None + custom_extensions: Optional[list[OSICustomExtension]] = None + + +class OSIDataset(BaseModel): + """Logical dataset representing a business entity (fact or dimension table).""" + + model_config = ConfigDict(frozen=True) + + name: str + source: str + primary_key: Optional[list[str]] = None + unique_keys: Optional[list[list[str]]] = None + description: Optional[str] = None + ai_context: Optional[OSIAIContext] = None + fields: Optional[list[OSIField]] = None + custom_extensions: Optional[list[OSICustomExtension]] = None + + +class OSIRelationship(BaseModel): + """Foreign key relationship between datasets.""" + + model_config = ConfigDict(frozen=True, populate_by_name=True) + + name: str + from_dataset: str = Field(..., alias="from") + to: str + from_columns: list[str] + to_columns: list[str] + ai_context: Optional[OSIAIContext] = None + custom_extensions: Optional[list[OSICustomExtension]] = None + + +class OSIMetric(BaseModel): + """Quantitative measure defined on business data.""" + + model_config = ConfigDict(frozen=True) + + name: str + expression: OSIExpression + description: Optional[str] = None + ai_context: Optional[OSIAIContext] = None + custom_extensions: Optional[list[OSICustomExtension]] = None + + +class OSISemanticModel(BaseModel): + """Top-level container representing a complete semantic model.""" + + model_config = ConfigDict(frozen=True) + + name: str + description: Optional[str] = None + ai_context: Optional[OSIAIContext] = None + datasets: list[OSIDataset] + relationships: Optional[list[OSIRelationship]] = None + metrics: Optional[list[OSIMetric]] = None + custom_extensions: Optional[list[OSICustomExtension]] = None + + +class OSIDocument(BaseModel): + """Root OSI document.""" + + model_config = ConfigDict(frozen=True) + + version: str = "0.2.0.dev0" + dialects: Optional[list[OSIDialect]] = None + vendors: Optional[list[OSIVendor]] = None + semantic_model: list[OSISemanticModel] + + def to_osi_yaml(self, **kwargs: Any) -> str: + """Serialize to OSI-compliant YAML (uses field aliases and excludes None values).""" + data = self.model_dump(by_alias=True, exclude_none=True, mode="json", **kwargs) + return yaml.dump(data, default_flow_style=False, sort_keys=False, allow_unicode=True) + + def to_osi_json(self, **kwargs: Any) -> str: + """Serialize to OSI-compliant JSON (uses field aliases and excludes None values).""" + return self.model_dump_json(by_alias=True, exclude_none=True, **kwargs)