diff --git a/.gitignore b/.gitignore index c18dd8d..b267296 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ __pycache__/ +.pytest_cache/ +.ruff_cache/ diff --git a/ROADMAP.md b/ROADMAP.md index 60af966..5a8bcbe 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -386,6 +386,7 @@ Broad ecosystem adoption depends on practical tools that let teams validate thei - [JSON Schema (osi-schema.json)](core-spec/osi-schema.json) — schema for structural validation - [Validation Script (validate.py)](validation/validate.py) — validates OSI YAML against JSON Schema, unique names, references, and SQL syntax - [Snowflake Converter](converters/snowflake/) — OSI → Snowflake Cortex Analyst YAML converter +- [Databricks Converter](converters/databricks/) — bidirectional OSI ↔ Databricks Unity Catalog Metric View YAML converter **Related Issues & PRs:** diff --git a/converters/databricks/README.md b/converters/databricks/README.md new file mode 100644 index 0000000..022abd9 --- /dev/null +++ b/converters/databricks/README.md @@ -0,0 +1,98 @@ +# OSI <-> Databricks Converter + +Bidirectional converter between OSI YAML semantic models and [Databricks Unity +Catalog Metric View](https://docs.databricks.com/aws/en/metric-views/) YAML. +Pure offline conversion — no Databricks workspace connection required. + +## Setup + +```bash +pip3 install -r requirements.txt +``` + +## Usage + +```bash +# OSI -> Databricks UC Metric View +python3 src/osi_to_databricks_metric_view.py -i input.yaml -o output.yaml + +# Databricks UC Metric View -> OSI +python3 src/databricks_metric_view_to_osi.py -i metric_view.yaml -o osi.yaml +``` + +## Tests + +```bash +python3 -m pytest tests/ -q +python3 -m pyflakes src/ tests/ +python3 -m ruff check src/ tests/ +``` + +## Mapping summary + +| OSI | Databricks UC Metric View | +|---|---| +| `version: 0.1.1` | `version: 1.1` (current UC metric view spec) | +| `semantic_model[].description` + `ai_context` | top-level `comment` | +| `dataset.source` (`db.schema.table` or subquery) | `source` (3-part name or inline SQL) | +| primary fact dataset | `source` of the metric view | +| other datasets reachable via `relationships` | `joins[]` entries (`source` + `sql_on`) | +| `dataset.fields[]` | `dimensions[]` (qualified by dataset name) | +| `metrics[]` | `measures[]` | +| `expression.dialects[DATABRICKS]` (else `ANSI_SQL`) | `expr` | +| `relationships[].from_columns`/`to_columns` | `sql_on:` boolean built from positional pairs | +| `custom_extensions[DATABRICKS]` | applied to top-level `filter`, `comment`, etc. | + +## Picking the primary dataset + +A single OSI semantic model has N datasets and M relationships. A Databricks +metric view has one `source` plus joined tables. The converter picks the +"primary" dataset using this priority: + +1. `custom_extensions[vendor_name=DATABRICKS]` with `{"primary_dataset": "..."}`. +2. The dataset most often on the `from` side of relationships (typically the + fact table). +3. The first dataset declared. + +Datasets unreachable from the primary via the relationship graph are emitted +as a warning and excluded from the metric view; the OSI model should be split +or the user should provide an explicit `primary_dataset` hint. + +## Expression qualification + +Bare single-identifier expressions (e.g. `expression: customer_id`) are +auto-qualified with their dataset name so they resolve unambiguously after +joins (`customer.customer_id`). + +Multi-token expressions (operators, function calls, multi-column +references) are emitted **verbatim** because string-prepending only +qualifies the first identifier, leaving subsequent column references +ambiguous after joins. For computed fields on a non-primary dataset, +provide a `DATABRICKS` dialect entry that's already table-qualified: + +```yaml +fields: + - name: customer_full_name + expression: + dialects: + - dialect: ANSI_SQL + expression: c_first_name || ' ' || c_last_name + - dialect: DATABRICKS + expression: customer.c_first_name || ' ' || customer.c_last_name +``` + +The converter prefers the `DATABRICKS` dialect when present and emits a +warning when a non-primary multi-token expression has only `ANSI_SQL`. + +## Limitations + +- Round-trip from a UC metric view that references columns through more than + one hop of joins may flatten dataset attribution to the immediate join + source. The converter parses simple `. = .` `sql_on` + clauses; complex boolean expressions are preserved verbatim in + `custom_extensions[DATABRICKS].raw_joins` and re-emitted on the inverse + conversion. +- OSI `dimension.is_time` does not have a dedicated UC counterpart and is + preserved through `custom_extensions[DATABRICKS]` for round-trip fidelity. +- AI context fields (`ai_context.instructions`, `synonyms`) on relationships + are dropped on export with a warning. diff --git a/converters/databricks/requirements.txt b/converters/databricks/requirements.txt new file mode 100644 index 0000000..8d5fd98 --- /dev/null +++ b/converters/databricks/requirements.txt @@ -0,0 +1 @@ +PyYAML>=5.0 diff --git a/converters/databricks/src/databricks_metric_view_to_osi.py b/converters/databricks/src/databricks_metric_view_to_osi.py new file mode 100644 index 0000000..33bf046 --- /dev/null +++ b/converters/databricks/src/databricks_metric_view_to_osi.py @@ -0,0 +1,349 @@ +""" +Convert a Databricks Unity Catalog Metric View YAML to an OSI semantic model +YAML. Pure offline conversion — no Databricks workspace connection required. + +Usage: + python3 databricks_metric_view_to_osi.py -i metric_view.yaml -o osi.yaml +""" + +import argparse +import json +import re +import sys +import warnings + +import yaml + + +OSI_VERSION = "0.1.1" +DATABRICKS_DIALECT = "DATABRICKS" +DATABRICKS_VENDOR = "DATABRICKS" + +# Pattern for "." in dimension expressions and join `sql_on` +# clauses. Quoted identifiers are not handled (matches the OSI Snowflake +# converter's basic-only stance on quoting). +_QUALIFIED_COL_RE = re.compile(r"^\s*([A-Za-z_][\w]*)\.([A-Za-z_][\w]*)\s*$") +_EQUALITY_RE = re.compile( + r"\s*([A-Za-z_][\w]*)\.([A-Za-z_][\w]*)\s*=\s*" + r"([A-Za-z_][\w]*)\.([A-Za-z_][\w]*)\s*" +) + + +class DatabricksConversionError(Exception): + """Raised when a UC Metric View YAML cannot be converted to OSI.""" + + +def convert_databricks_to_osi(metric_view_yaml_str, model_name=None): + """Parse a UC Metric View YAML and return an OSI YAML string. + + Args: + metric_view_yaml_str: UC Metric View YAML. + model_name: Optional explicit name for the resulting semantic model. + Defaults to the source's table component or "metric_view_model". + + Returns: + OSI YAML as a string with the standard envelope. + """ + root = yaml.safe_load(metric_view_yaml_str) + if not isinstance(root, dict): + raise DatabricksConversionError( + "Invalid metric view YAML: expected a mapping at the root" + ) + + source = root.get("source") + if not source: + raise DatabricksConversionError( + "Invalid metric view YAML: missing 'source'" + ) + + primary_name = _name_from_source(source) + if not model_name: + model_name = f"{primary_name}_model" + + by_join_name = {} # join name -> dataset name (= table component) + datasets = [] + + primary_fields = [] + primary_dataset = { + "name": primary_name, + "source": source, + "fields": primary_fields, + } + datasets.append(primary_dataset) + table_to_dataset = {primary_name: primary_dataset} + + relationships = [] + join_warnings = [] + + for join in root.get("joins") or []: + if not isinstance(join, dict): + continue + j_name = join.get("name") or _name_from_source(join.get("source") or "") + j_source = join.get("source") + if not j_source: + continue + ds_name = _name_from_source(j_source) + if not ds_name: + continue + # Avoid collision with primary + unique_ds_name = ds_name + suffix = 1 + while unique_ds_name in table_to_dataset: + suffix += 1 + unique_ds_name = f"{ds_name}_{suffix}" + joined_ds = { + "name": unique_ds_name, + "source": j_source, + "fields": [], + } + datasets.append(joined_ds) + table_to_dataset[unique_ds_name] = joined_ds + by_join_name[j_name] = unique_ds_name + + rel = _parse_join_clause(join, primary_name, unique_ds_name) + if rel is not None: + relationships.append(rel) + else: + join_warnings.append(j_name) + + if join_warnings: + warnings.warn( + f"Could not parse `sql_on` for joins {join_warnings}; " + f"the raw clause is preserved in custom_extensions[DATABRICKS]" + ) + + for dim in root.get("dimensions") or []: + if not isinstance(dim, dict): + continue + target_ds = primary_dataset + bare_expr = dim.get("expr") + m = _QUALIFIED_COL_RE.match(str(bare_expr or "")) + if m: + table, _ = m.groups() + if table in table_to_dataset: + target_ds = table_to_dataset[table] + else: + # Try locating via join_name -> dataset_name mapping + ds_name = by_join_name.get(table) + if ds_name: + target_ds = table_to_dataset[ds_name] + target_ds["fields"].append(_dim_to_field(dim)) + + metrics = [_measure_to_metric(m) for m in (root.get("measures") or []) if isinstance(m, dict)] + + osi_model = {"name": model_name} + + description = root.get("comment") or root.get("description") + if description: + osi_model["description"] = description + + osi_model["datasets"] = datasets + if relationships: + osi_model["relationships"] = relationships + if metrics: + osi_model["metrics"] = metrics + + # Preserve unmapped UC fields in a DATABRICKS extension for round-trip. + preserved = _build_databricks_extension(root, primary_name, join_warnings) + if preserved: + osi_model["custom_extensions"] = [ + {"vendor_name": DATABRICKS_VENDOR, "data": json.dumps(preserved)} + ] + + envelope = {"version": OSI_VERSION, "semantic_model": [osi_model]} + return yaml.dump( + envelope, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + + +def _name_from_source(source): + """Return a dataset name from a UC source string. Uses last '.' segment + for 3-part names; for SQL queries falls back to a generic name. + """ + if not source: + return "" + s = str(source).strip() + upper = s.upper() + if upper.startswith(("SELECT ", "SELECT\n", "SELECT\t", + "WITH ", "WITH\n", "WITH\t")): + return "metric_view_source" + parts = s.split(".") + return parts[-1].strip() if parts else s + + +def _dim_to_field(dim): + """Convert one UC dimension dict to an OSI field dict.""" + name = dim.get("name") + if not name: + raise DatabricksConversionError("UC dimension missing required 'name'") + + expr = dim.get("expr") + if not expr: + raise DatabricksConversionError( + f"UC dimension '{name}' is missing 'expr'" + ) + + field = { + "name": name, + "expression": { + "dialects": [ + {"dialect": DATABRICKS_DIALECT, "expression": str(expr)} + ] + }, + } + description = dim.get("comment") or dim.get("description") + if description: + field["description"] = description + return field + + +def _measure_to_metric(measure): + """Convert one UC measure dict to an OSI metric dict.""" + name = measure.get("name") + if not name: + raise DatabricksConversionError("UC measure missing required 'name'") + + expr = measure.get("expr") + if not expr: + raise DatabricksConversionError( + f"UC measure '{name}' is missing 'expr'" + ) + + metric = { + "name": name, + "expression": { + "dialects": [ + {"dialect": DATABRICKS_DIALECT, "expression": str(expr)} + ] + }, + } + description = measure.get("comment") or measure.get("description") + if description: + metric["description"] = description + return metric + + +def _parse_join_clause(join, primary_name, joined_name): + """Parse a UC join's `sql_on` or `using` clause into an OSI relationship. + + Supports: + - `sql_on: ".= .[AND ...]"`, where one side is the + primary table and the other is the joined table. + - `using: [col1, col2]` — same column on both sides. + + Returns an OSI relationship dict, or None if the clause is too complex. + """ + name = join.get("name") or f"join_{joined_name}" + + using = join.get("using") + if using and isinstance(using, list): + return { + "name": name, + "from": primary_name, + "to": joined_name, + "from_columns": list(using), + "to_columns": list(using), + } + + sql_on = join.get("sql_on") or join.get("on") + if not sql_on: + return None + + parts = re.split(r"\s+AND\s+", str(sql_on), flags=re.IGNORECASE) + from_cols = [] + to_cols = [] + for p in parts: + m = _EQUALITY_RE.match(p) + if not m: + return None + l_table, l_col, r_table, r_col = m.groups() + if l_table == primary_name and r_table == joined_name: + from_cols.append(l_col) + to_cols.append(r_col) + elif r_table == primary_name and l_table == joined_name: + from_cols.append(r_col) + to_cols.append(l_col) + else: + return None + + if not from_cols: + return None + + return { + "name": name, + "from": primary_name, + "to": joined_name, + "from_columns": from_cols, + "to_columns": to_cols, + } + + +def _build_databricks_extension(root, primary_name, unparsed_join_names): + """Capture metric-view-level fields that have no OSI core counterpart. + + Stored in a `custom_extensions[vendor_name=DATABRICKS]` entry so a + subsequent OSI -> UC export can re-emit them. + """ + preserved = {"primary_dataset": primary_name} + + if root.get("filter"): + preserved["filter"] = root["filter"] + + mv_version = root.get("version") + if mv_version is not None: + preserved["metric_view_version"] = mv_version + + raw_joins = [] + for j in root.get("joins") or []: + if not isinstance(j, dict): + continue + if (j.get("name") or "") in unparsed_join_names: + raw_joins.append(j) + if raw_joins: + preserved["raw_joins"] = raw_joins + + if len(preserved) <= 1: # Only primary_dataset means nothing extra to keep + return preserved if preserved else None + return preserved + + +def main(): + parser = argparse.ArgumentParser( + description=( + "Convert Databricks Unity Catalog Metric View YAML to OSI YAML" + ) + ) + parser.add_argument( + "-i", "--input", required=True, + help="Path to the Databricks UC Metric View YAML input file", + ) + parser.add_argument( + "-o", "--output", required=True, + help="Path to write the OSI YAML output", + ) + parser.add_argument( + "--model-name", + help="Override the name of the resulting OSI semantic model", + ) + args = parser.parse_args() + + with open(args.input, "r") as f: + mv_yaml_str = f.read() + + try: + out_str = convert_databricks_to_osi(mv_yaml_str, model_name=args.model_name) + except DatabricksConversionError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + + with open(args.output, "w") as f: + f.write(out_str) + + print(f"Converted {args.input} -> {args.output}") + + +if __name__ == "__main__": + main() diff --git a/converters/databricks/src/osi_to_databricks_metric_view.py b/converters/databricks/src/osi_to_databricks_metric_view.py new file mode 100644 index 0000000..fbfb3fb --- /dev/null +++ b/converters/databricks/src/osi_to_databricks_metric_view.py @@ -0,0 +1,495 @@ +""" +Convert an OSI (Open Semantic Interchange) YAML semantic model to a Databricks +Unity Catalog Metric View YAML. Pure offline conversion — no Databricks +workspace connection required. + +Usage: + python3 osi_to_databricks_metric_view.py -i input.yaml -o output.yaml +""" + +import argparse +import json +import re +import sys +import warnings +from collections import Counter, deque + +import yaml + + +SUPPORTED_OSI_VERSION = "0.1.1" +METRIC_VIEW_VERSION = "1.1" +DATABRICKS_DIALECT = "DATABRICKS" +ANSI_DIALECT = "ANSI_SQL" +DATABRICKS_VENDOR = "DATABRICKS" + +# Bare single-identifier expressions can be safely auto-qualified with +# their dataset name. Anything else (operators, function calls, string +# literals, multi-column references) is left verbatim — auto-qualifying +# such expressions is unsafe because only the first identifier would be +# prefixed, leaving subsequent column references unqualified and ambiguous +# after joins. +_BARE_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") + + +class OsiConversionError(Exception): + """Raised when an OSI YAML cannot be converted to a UC Metric View.""" + + +def convert_osi_to_databricks(osi_yaml_str): + """Parse an OSI YAML string and return the equivalent UC Metric View YAML. + + Args: + osi_yaml_str: OSI YAML as a string with the standard envelope:: + + version: "0.1.1" + semantic_model: + - name: ... + + Returns: + Databricks UC Metric View YAML as a string. + + Raises: + OsiConversionError: If the input cannot be converted. + """ + root = yaml.safe_load(osi_yaml_str) + if not isinstance(root, dict): + raise OsiConversionError("Invalid OSI YAML: expected a mapping at the root") + + version_str = str(root.get("version", "")) + if version_str != SUPPORTED_OSI_VERSION: + raise OsiConversionError( + f"Unsupported OSI specification version '{version_str}'. " + f"Supported: {SUPPORTED_OSI_VERSION}" + ) + + semantic_model = root.get("semantic_model") + if not isinstance(semantic_model, list) or not semantic_model: + raise OsiConversionError( + "Invalid OSI YAML: 'semantic_model' must be a non-empty list" + ) + + if len(semantic_model) > 1: + warnings.warn( + f"OSI YAML contains {len(semantic_model)} semantic models; " + f"only the first will be converted" + ) + + osi = semantic_model[0] + if not isinstance(osi, dict): + raise OsiConversionError( + "Invalid OSI YAML: 'semantic_model' entries must be mappings" + ) + + metric_view = _convert_model(osi) + return yaml.dump( + metric_view, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + + +def _convert_model(osi): + name = osi.get("name") + if not name: + raise OsiConversionError("Missing required 'name' field in semantic model") + + datasets = osi.get("datasets") or [] + if not datasets: + raise OsiConversionError( + f"Semantic model '{name}' has no datasets to convert" + ) + + relationships = osi.get("relationships") or [] + metrics = osi.get("metrics") or [] + + by_name = { + ds["name"]: ds + for ds in datasets + if isinstance(ds, dict) and ds.get("name") + } + + primary = _pick_primary_dataset(osi, datasets, relationships) + primary_name = primary.get("name") + + primary_source = _format_source(primary.get("source")) + if primary_source is None: + raise OsiConversionError( + f"Primary dataset '{primary_name}' is missing a 'source' value" + ) + + result = { + "version": METRIC_VIEW_VERSION, + "source": primary_source, + } + + description = _combined_description(osi) + if description: + result["comment"] = description + + # Apply Databricks custom_extensions hints onto the metric view top-level. + db_ext = _extract_databricks_extension(osi) + if db_ext.get("filter"): + result["filter"] = db_ext["filter"] + + reachable = _datasets_reachable_from(primary_name, relationships, by_name) + unreachable = [n for n in by_name if n not in reachable] + if unreachable: + warnings.warn( + f"Datasets {unreachable} are not reachable from primary " + f"'{primary_name}' via relationships; their fields will not " + f"appear in the metric view" + ) + + joins = _emit_joins(primary_name, by_name, relationships) + # Restore raw joins that the inverse converter could not parse back into + # OSI relationships, preserved in custom_extensions[DATABRICKS].raw_joins. + raw_joins = db_ext.get("raw_joins") + if isinstance(raw_joins, list): + for j in raw_joins: + if isinstance(j, dict): + joins.append(j) + if joins: + result["joins"] = joins + + dimensions = _emit_dimensions(primary_name, by_name, reachable) + if dimensions: + result["dimensions"] = dimensions + + measures = _emit_measures(metrics) + if measures: + result["measures"] = measures + + return result + + +def _combined_description(node): + """Merge OSI description and string ai_context. Returns string or None.""" + desc = node.get("description") + ai = node.get("ai_context") + if isinstance(ai, str) and ai: + return f"{desc}\n{ai}" if desc else ai + return desc + + +def _extract_databricks_extension(osi): + """Return the parsed DATABRICKS custom_extension data dict, or {}.""" + for ext in osi.get("custom_extensions") or []: + if not isinstance(ext, dict): + continue + if (ext.get("vendor_name") or "").upper() != DATABRICKS_VENDOR: + continue + raw = ext.get("data") + if not raw: + continue + try: + parsed = json.loads(raw) if isinstance(raw, str) else raw + except json.JSONDecodeError: + warnings.warn( + "Could not parse Databricks custom_extension data as JSON; " + "ignoring" + ) + return {} + if isinstance(parsed, dict): + return parsed + return {} + + +def _pick_primary_dataset(osi, datasets, relationships): + """Pick the primary fact dataset for the metric view's `source`. + + Priority: explicit DATABRICKS extension hint, then most-`from` count, + then first declared. + """ + db_ext = _extract_databricks_extension(osi) + hint = db_ext.get("primary_dataset") + if hint: + for ds in datasets: + if isinstance(ds, dict) and ds.get("name") == hint: + return ds + raise OsiConversionError( + f"primary_dataset hint '{hint}' is not a defined dataset" + ) + + counts = Counter( + r.get("from") for r in relationships + if isinstance(r, dict) and r.get("from") + ) + if counts: + winner_name, _ = counts.most_common(1)[0] + for ds in datasets: + if isinstance(ds, dict) and ds.get("name") == winner_name: + return ds + + return datasets[0] + + +def _datasets_reachable_from(start, relationships, by_name): + """Undirected BFS over the relationship graph from `start`. + + Returns a set of dataset names reachable via any relationship hop. + Datasets not in `by_name` are not added. + """ + reachable = {start} + queue = deque([start]) + while queue: + current = queue.popleft() + for r in relationships: + if not isinstance(r, dict): + continue + f, t = r.get("from"), r.get("to") + if f == current and t and t in by_name and t not in reachable: + reachable.add(t) + queue.append(t) + elif t == current and f and f in by_name and f not in reachable: + reachable.add(f) + queue.append(f) + return reachable + + +def _emit_joins(primary_name, by_name, relationships): + """Emit UC join entries by traversing the relationship graph from primary. + + Each new dataset reached produces one join with `source` and `sql_on`. + Edges where `from_columns`/`to_columns` are mismatched in length raise. + """ + joins = [] + visited = {primary_name} + queue = deque([primary_name]) + + while queue: + current = queue.popleft() + for r in relationships: + if not isinstance(r, dict): + continue + f, t = r.get("from"), r.get("to") + from_cols = r.get("from_columns") or [] + to_cols = r.get("to_columns") or [] + + if f == current and t in by_name and t not in visited: + left, right = f, t + left_cols, right_cols = from_cols, to_cols + other = t + elif t == current and f in by_name and f not in visited: + left, right = t, f + left_cols, right_cols = to_cols, from_cols + other = f + else: + continue + + if len(left_cols) != len(right_cols) or not left_cols: + raise OsiConversionError( + f"Relationship '{r.get('name')}': from_columns and " + f"to_columns must have the same non-zero length " + f"(got {len(from_cols)} and {len(to_cols)})" + ) + + sql_on = " AND ".join( + f"{left}.{lc} = {right}.{rc}" + for lc, rc in zip(left_cols, right_cols) + ) + join_name = r.get("name") or f"join_{other}" + joins.append({ + "name": join_name, + "source": _format_source(by_name[other].get("source")), + "sql_on": sql_on, + }) + visited.add(other) + queue.append(other) + + return joins + + +def _emit_dimensions(primary_name, by_name, reachable): + """Emit one dimension per OSI field across all reachable datasets. + + Bare column expressions are qualified with their dataset name so UC can + resolve them after the joins. Name collisions across datasets are + disambiguated by prefixing with the dataset name. + """ + dimensions = [] + seen_names = set() + + # Process primary first, then other reachable datasets in declaration order + ordered = [primary_name] + [n for n in by_name if n != primary_name and n in reachable] + + for ds_name in ordered: + ds = by_name.get(ds_name) + if not ds: + continue + is_primary = ds_name == primary_name + for field in ds.get("fields") or []: + if not isinstance(field, dict): + continue + converted = _convert_field_to_dim(field, ds_name, is_primary) + if converted is None: + continue + if converted["name"] in seen_names: + converted["name"] = f"{ds_name}_{converted['name']}" + if converted["name"] in seen_names: + warnings.warn( + f"Skipping duplicate dimension name " + f"'{converted['name']}' from dataset '{ds_name}'" + ) + continue + dimensions.append(converted) + seen_names.add(converted["name"]) + + return dimensions + + +def _convert_field_to_dim(field, ds_name, is_primary): + """Convert one OSI field dict to a UC dimension dict, or None to skip. + + Bare single-identifier expressions are auto-qualified with the dataset + name (`.`) so they resolve unambiguously after joins. + Multi-token expressions (operators, function calls, multi-column + references) are emitted verbatim — auto-qualifying them is unsafe + because only the first identifier would be prefixed. For computed + fields on a non-primary dataset, callers should provide a DATABRICKS + dialect entry that's already qualified; otherwise a warning is emitted. + """ + name = field.get("name") + if not name: + raise OsiConversionError(f"Missing required 'name' in field of '{ds_name}'") + + expr = _extract_expression(field.get("expression"), name) + if expr is None: + return None + + if _BARE_IDENT_RE.match(expr): + qualified = f"{ds_name}.{expr}" + else: + qualified = expr + if not is_primary and f"{ds_name}." not in expr: + warnings.warn( + f"Field '{ds_name}.{name}' has a multi-token expression " + f"that the converter cannot auto-qualify for a UC metric " + f"view; column references may be ambiguous after joins. " + f"Provide a DATABRICKS dialect with table-qualified " + f"references to silence this warning." + ) + + result = {"name": name, "expr": qualified} + + description = _combined_description(field) + if description: + result["comment"] = description + + return result + + +def _emit_measures(metrics): + """Convert OSI top-level metrics to UC measure dicts.""" + measures = [] + for m in metrics: + if not isinstance(m, dict): + continue + name = m.get("name") + if not name: + raise OsiConversionError("Missing required 'name' in metric") + + expr = _extract_expression(m.get("expression"), name) + if expr is None: + continue + + result = {"name": name, "expr": expr} + description = _combined_description(m) + if description: + result["comment"] = description + measures.append(result) + return measures + + +def _extract_expression(expression, field_name): + """Pick the best dialect for Databricks. Returns expression string or None. + + DATABRICKS dialect is preferred; falls back to ANSI_SQL. Returns None + (and emits a warning) if neither is present, signalling the field/metric + should be skipped. Raises if the dialects list is missing entirely. + """ + if not isinstance(expression, dict): + raise OsiConversionError( + f"Missing or malformed expression for field/metric '{field_name}'" + ) + + dialects = expression.get("dialects") + if not dialects: + raise OsiConversionError( + f"Missing dialects for field/metric '{field_name}'" + ) + + db_expr = None + ansi_expr = None + for d in dialects: + if not isinstance(d, dict): + continue + dialect_name = (d.get("dialect") or "").upper() + if dialect_name == DATABRICKS_DIALECT: + db_expr = d.get("expression") + elif dialect_name == ANSI_DIALECT: + ansi_expr = d.get("expression") + + if db_expr is not None: + return db_expr + if ansi_expr is not None: + return ansi_expr + + dialect_names = [ + d.get("dialect", "") for d in dialects if isinstance(d, dict) + ] + warnings.warn( + f"Skipping field/metric '{field_name}': no Databricks-compatible " + f"expression (has dialects: {', '.join(dialect_names)}; requires " + f"DATABRICKS or ANSI_SQL)" + ) + return None + + +def _format_source(source): + """Pass through OSI source string. Subqueries and 3-part names are kept + verbatim — Databricks Unity Catalog uses `catalog.schema.table` and + accepts inline SQL queries. + """ + if source is None: + return None + s = str(source).strip() + if not s: + return None + return s + + +def main(): + parser = argparse.ArgumentParser( + description=( + "Convert OSI YAML semantic model to Databricks Unity Catalog " + "Metric View YAML" + ) + ) + parser.add_argument( + "-i", "--input", required=True, help="Path to the OSI YAML input file" + ) + parser.add_argument( + "-o", "--output", required=True, + help="Path to write the Databricks UC Metric View YAML output", + ) + args = parser.parse_args() + + with open(args.input, "r") as f: + osi_yaml_str = f.read() + + try: + out_str = convert_osi_to_databricks(osi_yaml_str) + except OsiConversionError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + + with open(args.output, "w") as f: + f.write(out_str) + + print(f"Converted {args.input} -> {args.output}") + + +if __name__ == "__main__": + main() diff --git a/converters/databricks/tests/conftest.py b/converters/databricks/tests/conftest.py new file mode 100644 index 0000000..094565e --- /dev/null +++ b/converters/databricks/tests/conftest.py @@ -0,0 +1,8 @@ +"""Test configuration: make `src/` importable from tests.""" + +import sys +from pathlib import Path + +_SRC = Path(__file__).resolve().parent.parent / "src" +if str(_SRC) not in sys.path: + sys.path.insert(0, str(_SRC)) diff --git a/converters/databricks/tests/test_databricks_to_osi.py b/converters/databricks/tests/test_databricks_to_osi.py new file mode 100644 index 0000000..bcc49fa --- /dev/null +++ b/converters/databricks/tests/test_databricks_to_osi.py @@ -0,0 +1,181 @@ +"""Tests for the Databricks UC Metric View -> OSI converter.""" + +import json +import warnings + +import pytest +import yaml + +from databricks_metric_view_to_osi import ( + DatabricksConversionError, + _name_from_source, + _parse_join_clause, + convert_databricks_to_osi, +) + + +def _basic_metric_view(): + return yaml.dump({ + "version": "0.1", + "source": "main.sales.orders", + "comment": "Sales metric view", + "joins": [ + { + "name": "orders_to_customers", + "source": "main.sales.customers", + "sql_on": "orders.customer_id = customers.id", + }, + ], + "dimensions": [ + {"name": "order_id", "expr": "orders.order_id"}, + {"name": "customer_email", "expr": "customers.email"}, + ], + "measures": [ + {"name": "total_revenue", "expr": "SUM(orders.amount)", + "comment": "Total order revenue"}, + ], + }) + + +def test_round_trip_top_level_shape(): + out = yaml.safe_load(convert_databricks_to_osi(_basic_metric_view())) + assert out["version"] == "0.1.1" + model = out["semantic_model"][0] + assert model["name"] == "orders_model" + assert model["description"] == "Sales metric view" + + ds_names = [d["name"] for d in model["datasets"]] + assert ds_names == ["orders", "customers"] + + rels = model["relationships"] + assert len(rels) == 1 + assert rels[0]["from_columns"] == ["customer_id"] + assert rels[0]["to_columns"] == ["id"] + + metric_names = [m["name"] for m in model["metrics"]] + assert metric_names == ["total_revenue"] + + +def test_dimensions_are_attributed_to_table_in_expression(): + out = yaml.safe_load(convert_databricks_to_osi(_basic_metric_view())) + model = out["semantic_model"][0] + by_ds = {d["name"]: d for d in model["datasets"]} + assert [f["name"] for f in by_ds["orders"]["fields"]] == ["order_id"] + assert [f["name"] for f in by_ds["customers"]["fields"]] == ["customer_email"] + + +def test_explicit_model_name_used(): + out = yaml.safe_load(convert_databricks_to_osi( + _basic_metric_view(), model_name="my_model" + )) + assert out["semantic_model"][0]["name"] == "my_model" + + +def test_missing_source_raises(): + with pytest.raises(DatabricksConversionError, match="missing 'source'"): + convert_databricks_to_osi(yaml.dump({"version": "0.1"})) + + +def test_root_must_be_mapping(): + with pytest.raises(DatabricksConversionError, match="mapping at the root"): + convert_databricks_to_osi("- nope") + + +def test_using_clause_produces_same_columns_on_both_sides(): + mv = yaml.dump({ + "version": "0.1", + "source": "main.sales.orders", + "joins": [ + {"name": "j", "source": "main.sales.tax_rates", "using": ["region"]} + ], + "dimensions": [{"name": "k", "expr": "orders.k"}], + "measures": [{"name": "m", "expr": "SUM(orders.x)"}], + }) + out = yaml.safe_load(convert_databricks_to_osi(mv)) + rels = out["semantic_model"][0]["relationships"] + assert rels[0]["from_columns"] == ["region"] + assert rels[0]["to_columns"] == ["region"] + + +def test_unparseable_sql_on_falls_back_to_extension_warning(): + mv = yaml.dump({ + "version": "0.1", + "source": "main.sales.orders", + "joins": [{ + "name": "complex_join", + "source": "main.sales.customers", + "sql_on": "DATEDIFF(orders.dt, customers.dt) < 30", + }], + "dimensions": [{"name": "k", "expr": "orders.k"}], + "measures": [{"name": "m", "expr": "SUM(orders.x)"}], + }) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + out = yaml.safe_load(convert_databricks_to_osi(mv)) + assert any("Could not parse `sql_on`" in str(w.message) for w in caught) + model = out["semantic_model"][0] + # No relationship was inferred for the complex join + assert "relationships" not in model or not model.get("relationships") + # The raw join is preserved in the DATABRICKS extension + ext = model["custom_extensions"][0] + data = json.loads(ext["data"]) + assert any(j["name"] == "complex_join" for j in data["raw_joins"]) + + +def test_filter_preserved_in_databricks_extension(): + mv = yaml.dump({ + "version": "0.1", + "source": "main.sales.orders", + "filter": "orders.status = 'COMPLETED'", + "dimensions": [{"name": "k", "expr": "orders.k"}], + "measures": [{"name": "m", "expr": "SUM(orders.x)"}], + }) + out = yaml.safe_load(convert_databricks_to_osi(mv)) + ext = out["semantic_model"][0]["custom_extensions"][0] + data = json.loads(ext["data"]) + assert data["filter"] == "orders.status = 'COMPLETED'" + assert data["primary_dataset"] == "orders" + + +def test_subquery_source_yields_generic_name(): + assert _name_from_source("SELECT 1 AS x") == "metric_view_source" + assert _name_from_source("WITH c AS (SELECT 1) SELECT * FROM c") == ( + "metric_view_source" + ) + + +def test_three_part_source_uses_last_segment(): + assert _name_from_source("main.sales.orders") == "orders" + assert _name_from_source("orders") == "orders" + + +def test_parse_join_clause_handles_swapped_table_order(): + rel = _parse_join_clause( + {"name": "j", "sql_on": "customers.id = orders.customer_id"}, + primary_name="orders", + joined_name="customers", + ) + assert rel["from_columns"] == ["customer_id"] + assert rel["to_columns"] == ["id"] + + +def test_parse_join_returns_none_for_unrelated_tables(): + assert _parse_join_clause( + {"name": "j", "sql_on": "x.id = y.id"}, + primary_name="orders", + joined_name="customers", + ) is None + + +def test_dimension_with_unqualified_expression_attached_to_primary(): + mv = yaml.dump({ + "version": "0.1", + "source": "main.sales.orders", + "dimensions": [{"name": "k", "expr": "UPPER(unqualified)"}], + "measures": [{"name": "m", "expr": "SUM(unqualified)"}], + }) + out = yaml.safe_load(convert_databricks_to_osi(mv)) + model = out["semantic_model"][0] + primary = model["datasets"][0] + assert primary["name"] == "orders" + assert [f["name"] for f in primary["fields"]] == ["k"] diff --git a/converters/databricks/tests/test_osi_to_databricks.py b/converters/databricks/tests/test_osi_to_databricks.py new file mode 100644 index 0000000..1cdda75 --- /dev/null +++ b/converters/databricks/tests/test_osi_to_databricks.py @@ -0,0 +1,435 @@ +"""Tests for the OSI -> Databricks UC Metric View converter.""" + +import warnings + +import pytest +import yaml + +from osi_to_databricks_metric_view import ( + OsiConversionError, + _datasets_reachable_from, + _emit_dimensions, + _emit_joins, + _emit_measures, + _extract_databricks_extension, + _extract_expression, + _format_source, + _pick_primary_dataset, + convert_osi_to_databricks, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _wrap(model_dict): + return yaml.dump( + {"version": "0.1.1", "semantic_model": [model_dict]}, + default_flow_style=False, + ) + + +def _field(name, dialect="ANSI_SQL", expression=None, is_time=None): + expr = expression if expression is not None else name + field = { + "name": name, + "expression": {"dialects": [{"dialect": dialect, "expression": expr}]}, + } + if is_time is not None: + field["dimension"] = {"is_time": is_time} + return field + + +def _metric(name, expr, dialect="ANSI_SQL"): + return { + "name": name, + "expression": {"dialects": [{"dialect": dialect, "expression": expr}]}, + } + + +def _minimal_two_table_model(): + return { + "name": "sales", + "description": "Sales analytics", + "datasets": [ + { + "name": "orders", + "source": "main.sales.orders", + "fields": [ + _field("order_id"), + _field("customer_id"), + _field("amount"), + ], + }, + { + "name": "customers", + "source": "main.sales.customers", + "fields": [ + _field("id"), + _field("email"), + ], + }, + ], + "relationships": [ + { + "name": "orders_to_customers", + "from": "orders", + "to": "customers", + "from_columns": ["customer_id"], + "to_columns": ["id"], + } + ], + "metrics": [ + _metric("total_revenue", "SUM(orders.amount)"), + ], + } + + +# --------------------------------------------------------------------------- +# Top-level conversion +# --------------------------------------------------------------------------- + +def test_minimal_model_top_level_shape(): + out = yaml.safe_load(convert_osi_to_databricks(_wrap(_minimal_two_table_model()))) + + assert out["version"] == "1.1" + assert out["source"] == "main.sales.orders" + assert out["comment"] == "Sales analytics" + + join_names = [j["name"] for j in out["joins"]] + assert join_names == ["orders_to_customers"] + assert out["joins"][0]["source"] == "main.sales.customers" + assert out["joins"][0]["sql_on"] == "orders.customer_id = customers.id" + + dim_names = [d["name"] for d in out["dimensions"]] + assert dim_names == ["order_id", "customer_id", "amount", "id", "email"] + + measure_names = [m["name"] for m in out["measures"]] + assert measure_names == ["total_revenue"] + assert out["measures"][0]["expr"] == "SUM(orders.amount)" + + +def test_dimension_expressions_are_table_qualified(): + out = yaml.safe_load(convert_osi_to_databricks(_wrap(_minimal_two_table_model()))) + by_name = {d["name"]: d for d in out["dimensions"]} + assert by_name["order_id"]["expr"] == "orders.order_id" + assert by_name["email"]["expr"] == "customers.email" + + +def test_unsupported_osi_version_rejected(): + with pytest.raises(OsiConversionError, match="Unsupported OSI"): + convert_osi_to_databricks(yaml.dump( + {"version": "0.0.9", "semantic_model": [{"name": "x"}]} + )) + + +def test_missing_semantic_model_rejected(): + with pytest.raises(OsiConversionError, match="non-empty list"): + convert_osi_to_databricks(yaml.dump({"version": "0.1.1"})) + + +def test_root_must_be_mapping(): + with pytest.raises(OsiConversionError, match="mapping at the root"): + convert_osi_to_databricks("- just_a_list_item") + + +def test_multiple_models_warns_and_uses_first(): + payload = { + "version": "0.1.1", + "semantic_model": [ + _minimal_two_table_model(), + {**_minimal_two_table_model(), "name": "ignored"}, + ], + } + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + out = yaml.safe_load(convert_osi_to_databricks(yaml.dump(payload))) + assert any("only the first will be converted" in str(w.message) for w in caught) + assert out["source"] == "main.sales.orders" + + +# --------------------------------------------------------------------------- +# Dialect selection +# --------------------------------------------------------------------------- + +def test_databricks_dialect_preferred_over_ansi(): + expr = { + "dialects": [ + {"dialect": "ANSI_SQL", "expression": "LOWER(email)"}, + {"dialect": "DATABRICKS", "expression": "lower(email)"}, + ] + } + assert _extract_expression(expr, "f") == "lower(email)" + + +def test_ansi_fallback_when_no_databricks_dialect(): + expr = {"dialects": [{"dialect": "ANSI_SQL", "expression": "id"}]} + assert _extract_expression(expr, "f") == "id" + + +def test_no_compatible_dialect_warns_and_skips(): + expr = {"dialects": [{"dialect": "MAQL", "expression": "..."}]} + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + result = _extract_expression(expr, "skip_me") + assert result is None + assert any("Skipping field/metric 'skip_me'" in str(w.message) for w in caught) + + +def test_missing_dialects_raises(): + with pytest.raises(OsiConversionError, match="Missing dialects"): + _extract_expression({}, "f") + + +def test_missing_expression_raises(): + with pytest.raises(OsiConversionError, match="malformed expression"): + _extract_expression(None, "f") + + +# --------------------------------------------------------------------------- +# Dataset and relationship handling +# --------------------------------------------------------------------------- + +def test_no_datasets_raises(): + with pytest.raises(OsiConversionError, match="no datasets"): + convert_osi_to_databricks(_wrap({"name": "empty"})) + + +def test_primary_picked_by_most_from_count(): + model = _minimal_two_table_model() + primary = _pick_primary_dataset(model, model["datasets"], model["relationships"]) + assert primary["name"] == "orders" + + +def test_primary_overridden_by_databricks_extension_hint(): + model = _minimal_two_table_model() + model["custom_extensions"] = [ + {"vendor_name": "DATABRICKS", "data": '{"primary_dataset": "customers"}'} + ] + primary = _pick_primary_dataset(model, model["datasets"], model["relationships"]) + assert primary["name"] == "customers" + + +def test_invalid_primary_hint_raises(): + model = _minimal_two_table_model() + model["custom_extensions"] = [ + {"vendor_name": "DATABRICKS", "data": '{"primary_dataset": "ghost"}'} + ] + with pytest.raises(OsiConversionError, match="primary_dataset hint"): + _pick_primary_dataset(model, model["datasets"], model["relationships"]) + + +def test_composite_relationship_emits_multi_column_sql_on(): + model = _minimal_two_table_model() + model["relationships"][0] = { + "name": "order_lines_to_products", + "from": "orders", + "to": "customers", + "from_columns": ["customer_id", "region"], + "to_columns": ["id", "region"], + } + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + assert out["joins"][0]["sql_on"] == ( + "orders.customer_id = customers.id AND orders.region = customers.region" + ) + + +def test_relationship_column_count_mismatch_raises(): + model = _minimal_two_table_model() + model["relationships"][0]["to_columns"] = ["id", "extra"] + with pytest.raises(OsiConversionError, match="must have the same"): + convert_osi_to_databricks(_wrap(model)) + + +def test_unreachable_dataset_warns_and_skips(): + model = _minimal_two_table_model() + model["datasets"].append( + {"name": "orphan", "source": "main.misc.orphan", "fields": [_field("k")]} + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + assert any("not reachable from primary" in str(w.message) for w in caught) + assert all(d["name"] != "k" for d in out["dimensions"]) + + +def test_datasets_reachable_traverses_both_directions(): + by_name = {"a": {}, "b": {}, "c": {}} + rels = [ + {"from": "a", "to": "b", "from_columns": ["x"], "to_columns": ["y"]}, + {"from": "c", "to": "b", "from_columns": ["x"], "to_columns": ["y"]}, + ] + assert _datasets_reachable_from("a", rels, by_name) == {"a", "b", "c"} + + +# --------------------------------------------------------------------------- +# Field / metric mapping +# --------------------------------------------------------------------------- + +def test_metric_with_no_compatible_dialect_skipped_with_warning(): + model = _minimal_two_table_model() + model["metrics"].append(_metric("untranslatable", "...", dialect="MAQL")) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + measure_names = [m["name"] for m in out["measures"]] + assert "untranslatable" not in measure_names + assert any("untranslatable" in str(w.message) for w in caught) + + +def test_dimension_name_collision_disambiguated(): + model = _minimal_two_table_model() + # Add a duplicate "id" field to orders so it collides with customers.id + model["datasets"][0]["fields"].append(_field("id")) + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + names = [d["name"] for d in out["dimensions"]] + # Primary's "id" wins; customers' "id" gets prefixed + assert "id" in names + assert "customers_id" in names + + +def test_description_and_string_ai_context_merged_into_comment(): + model = _minimal_two_table_model() + model["ai_context"] = "Prefer the orders fact for revenue queries." + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + assert "Sales analytics" in out["comment"] + assert "orders fact" in out["comment"] + + +def test_format_source_passes_through_subqueries_and_three_part(): + assert _format_source("main.sales.orders") == "main.sales.orders" + assert _format_source("SELECT * FROM main.sales.orders") == ( + "SELECT * FROM main.sales.orders" + ) + assert _format_source(None) is None + assert _format_source(" ") is None + + +def test_extract_databricks_extension_handles_invalid_json(): + osi = {"custom_extensions": [{"vendor_name": "DATABRICKS", "data": "{not json"}]} + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + result = _extract_databricks_extension(osi) + assert result == {} + assert any("could not parse" in str(w.message).lower() for w in caught) + + +# --------------------------------------------------------------------------- +# Direct unit calls into emit helpers +# --------------------------------------------------------------------------- + +def test_emit_joins_skips_self_loop_and_visited(): + by_name = {"a": {"source": "a"}, "b": {"source": "b"}} + rels = [{ + "name": "a_b", + "from": "a", "to": "b", + "from_columns": ["x"], "to_columns": ["y"], + }] + joins = _emit_joins("a", by_name, rels) + assert len(joins) == 1 + assert joins[0]["sql_on"] == "a.x = b.y" + + +def test_emit_dimensions_includes_unqualified_when_expr_already_qualified(): + by_name = { + "a": { + "name": "a", + "fields": [_field("col1", expression="a.col1")], + } + } + dims = _emit_dimensions("a", by_name, {"a"}) + assert dims == [{"name": "col1", "expr": "a.col1"}] + + +# --------------------------------------------------------------------------- +# Multi-token expression handling (Bug 1 regression coverage) +# --------------------------------------------------------------------------- + +def test_multi_token_expression_emitted_verbatim_on_primary(): + """A computed field on the primary dataset is emitted as-is — the + metric view's `source` makes bare references unambiguous, so no + auto-qualification is required. + """ + model = _minimal_two_table_model() + model["datasets"][0]["fields"].append( + _field("upper_id", expression="UPPER(order_id)") + ) + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + by_name = {d["name"]: d for d in out["dimensions"]} + # Function call expression must NOT have a stray qualifier prepended. + assert by_name["upper_id"]["expr"] == "UPPER(order_id)" + + +def test_multi_token_expression_on_join_warns_when_unqualified(): + """A computed field on a *non-primary* dataset with a multi-token + expression that doesn't already mention the dataset name should warn + the user to provide a DATABRICKS dialect with explicit qualifiers. + """ + model = _minimal_two_table_model() + model["datasets"][1]["fields"].append( + _field("full_name", expression="first_name || ' ' || last_name") + ) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + by_name = {d["name"]: d for d in out["dimensions"]} + # Critically, we do NOT emit `customers.first_name || ' ' || last_name` + # — that would only qualify the first column and leave `last_name` + # ambiguous after joins. The expression is left verbatim instead. + assert by_name["full_name"]["expr"] == "first_name || ' ' || last_name" + assert any("multi-token expression" in str(w.message) for w in caught) + + +def test_databricks_dialect_with_qualified_expression_silences_warning(): + """If the OSI author supplies a DATABRICKS dialect entry that already + contains the dataset-qualified references, the converter prefers it + and does not warn. + """ + model = _minimal_two_table_model() + model["datasets"][1]["fields"].append({ + "name": "full_name", + "expression": {"dialects": [ + {"dialect": "ANSI_SQL", + "expression": "first_name || ' ' || last_name"}, + {"dialect": "DATABRICKS", + "expression": "customers.first_name || ' ' || customers.last_name"}, + ]}, + }) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + by_name = {d["name"]: d for d in out["dimensions"]} + assert by_name["full_name"]["expr"] == ( + "customers.first_name || ' ' || customers.last_name" + ) + assert not any("multi-token expression" in str(w.message) for w in caught) + + +def test_raw_joins_from_databricks_extension_round_tripped(): + """An OSI model carrying a DATABRICKS custom_extension with raw_joins + (preserved from a previous UC -> OSI conversion) re-emits those joins + verbatim alongside the relationship-derived ones. + """ + model = _minimal_two_table_model() + model["custom_extensions"] = [{ + "vendor_name": "DATABRICKS", + "data": ( + '{"raw_joins": [{"name": "complex_join", ' + '"source": "main.sales.tax_rates", ' + '"sql_on": "DATEDIFF(orders.dt, tax_rates.dt) < 30"}]}' + ), + }] + out = yaml.safe_load(convert_osi_to_databricks(_wrap(model))) + join_names = [j["name"] for j in out["joins"]] + assert "orders_to_customers" in join_names + assert "complex_join" in join_names + + +def test_emit_measures_skips_metric_with_only_unsupported_dialects(): + metrics = [_metric("ok", "SUM(x)"), _metric("bad", "...", dialect="MAQL")] + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + measures = _emit_measures(metrics) + names = [m["name"] for m in measures] + assert names == ["ok"] diff --git a/converters/databricks/tests/test_roundtrip.py b/converters/databricks/tests/test_roundtrip.py new file mode 100644 index 0000000..ea81676 --- /dev/null +++ b/converters/databricks/tests/test_roundtrip.py @@ -0,0 +1,120 @@ +"""Round-trip tests: OSI -> Databricks UC Metric View -> OSI. + +The point of round-trips is to confirm the converter preserves the structure +that both formats can express. Lossy fields (e.g. ai_context.synonyms, +non-DATABRICKS dialects, dimension.is_time) are not expected to survive a +round trip and are explicitly preserved out of band where the spec allows it. +""" + +from pathlib import Path + +import pytest +import yaml + +from databricks_metric_view_to_osi import convert_databricks_to_osi +from osi_to_databricks_metric_view import convert_osi_to_databricks + + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent.parent +TPCDS_PATH = REPO_ROOT / "examples" / "tpcds_semantic_model.yaml" + + +def _wrap(model): + return yaml.dump( + {"version": "0.1.1", "semantic_model": [model]}, + default_flow_style=False, + ) + + +def _two_table_model(): + return { + "name": "sales", + "datasets": [ + { + "name": "orders", + "source": "main.sales.orders", + "fields": [ + {"name": "order_id", "expression": {"dialects": [ + {"dialect": "ANSI_SQL", "expression": "order_id"} + ]}}, + {"name": "customer_id", "expression": {"dialects": [ + {"dialect": "ANSI_SQL", "expression": "customer_id"} + ]}}, + ], + }, + { + "name": "customers", + "source": "main.sales.customers", + "fields": [ + {"name": "id", "expression": {"dialects": [ + {"dialect": "ANSI_SQL", "expression": "id"} + ]}}, + {"name": "email", "expression": {"dialects": [ + {"dialect": "ANSI_SQL", "expression": "email"} + ]}}, + ], + }, + ], + "relationships": [{ + "name": "orders_to_customers", + "from": "orders", + "to": "customers", + "from_columns": ["customer_id"], + "to_columns": ["id"], + }], + "metrics": [{ + "name": "total_revenue", + "expression": {"dialects": [ + {"dialect": "ANSI_SQL", "expression": "SUM(orders.amount)"} + ]}, + }], + } + + +def test_round_trip_preserves_datasets_and_relationships(): + osi_in = _wrap(_two_table_model()) + mv_yaml = convert_osi_to_databricks(osi_in) + osi_out = yaml.safe_load(convert_databricks_to_osi(mv_yaml)) + + model = osi_out["semantic_model"][0] + ds_names = sorted(d["name"] for d in model["datasets"]) + assert ds_names == ["customers", "orders"] + + rels = model["relationships"] + assert len(rels) == 1 + assert rels[0]["from"] == "orders" + assert rels[0]["to"] == "customers" + assert rels[0]["from_columns"] == ["customer_id"] + assert rels[0]["to_columns"] == ["id"] + + +def test_round_trip_preserves_metric_names_and_expressions(): + osi_in = _wrap(_two_table_model()) + mv_yaml = convert_osi_to_databricks(osi_in) + osi_out = yaml.safe_load(convert_databricks_to_osi(mv_yaml)) + metrics = osi_out["semantic_model"][0]["metrics"] + assert [m["name"] for m in metrics] == ["total_revenue"] + expr = metrics[0]["expression"]["dialects"][0] + assert expr["expression"] == "SUM(orders.amount)" + + +@pytest.mark.skipif(not TPCDS_PATH.exists(), reason="tpcds example not found") +def test_tpcds_example_converts_without_error(): + osi_yaml = TPCDS_PATH.read_text() + mv = yaml.safe_load(convert_osi_to_databricks(osi_yaml)) + + # Primary should be store_sales — the table on the `from` side of all 4 + # relationships in the TPC-DS example. + assert mv["source"] == "tpcds.public.store_sales" + + # All four joins should be present, one per relationship in the OSI model + join_names = sorted(j["name"] for j in mv["joins"]) + assert join_names == [ + "store_sales_to_customer", + "store_sales_to_date", + "store_sales_to_item", + "store_sales_to_store", + ] + + measure_names = {m["name"] for m in mv["measures"]} + assert {"total_sales", "total_profit", "customer_lifetime_value"} <= measure_names