From b87986bb3c64c19f311accd7ba9e5fe128bd6074 Mon Sep 17 00:00:00 2001 From: sudulakuntanithin <52403575+sudulakuntanithin@users.noreply.github.com> Date: Fri, 19 Jun 2026 21:44:16 +0000 Subject: [PATCH] Add Snowflake to OSI import converter --- converters/snowflake/README.md | 202 +++++- .../src/snowflake_to_osi_yaml_converter.py | 637 +++++++++++++++++ .../test_snowflake_to_osi_yaml_converter.py | 662 ++++++++++++++++++ 3 files changed, 1496 insertions(+), 5 deletions(-) create mode 100644 converters/snowflake/src/snowflake_to_osi_yaml_converter.py create mode 100644 converters/snowflake/tests/test_snowflake_to_osi_yaml_converter.py diff --git a/converters/snowflake/README.md b/converters/snowflake/README.md index 93ee961..c72e819 100644 --- a/converters/snowflake/README.md +++ b/converters/snowflake/README.md @@ -1,8 +1,12 @@ -# OSI to Snowflake Converter +# Snowflake ↔ OSI Converter -Converts OSI YAML semantic models to [Snowflake Cortex Analyst](https://docs.snowflake.com/en/user-guide/snowflake-cortex/cortex-analyst) semantic model YAML. Pure offline conversion — no Snowflake connection required. +Bi-directional converter for [Snowflake Cortex Analyst](https://docs.snowflake.com/en/user-guide/snowflake-cortex/cortex-analyst) semantic model YAML and [OSI (Open Semantic Interchange)](https://github.com/open-semantic-interchange/OSI) YAML semantic models. Pure offline conversion — no Snowflake connection required. -> **Note:** This converter is under active development. It handles common cases but has not been thoroughly tested against all edge cases — use with caution in production. +This converter implements the hub-and-spoke architecture documented in [`converters/index.md`](../index.md), enabling teams to: +- **Export**: Convert OSI semantic models to Snowflake format for deployment +- **Import**: Convert existing Snowflake semantic models to OSI format for standardization and interoperability with other tools + +> **Note:** These converters are under active development. They handle common cases but have not been thoroughly tested against all edge cases — use with caution in production. ## Setup @@ -12,8 +16,127 @@ pip3 install -r requirements.txt ## Usage +### Export: OSI → Snowflake + +Convert an OSI semantic model to Snowflake Cortex Analyst format: + +```bash +python3 src/osi_to_snowflake_yaml_converter.py -i input_osi.yaml -o output_snowflake.yaml +``` + +**Example input (OSI format):** +```yaml +version: "0.2.0.dev0" +semantic_model: + - name: retail_model + datasets: + - name: orders + source: retail_db.public.orders + primary_key: [order_id] + fields: + - name: order_date + expression: + dialects: + - dialect: SNOWFLAKE + expression: order_date + dimension: + is_time: false + - name: amount + expression: + dialects: + - dialect: SNOWFLAKE + expression: amount +``` + +**Example output (Snowflake format):** +```yaml +name: retail_model +tables: + - name: orders + base_table: + database: RETAIL_DB + schema: PUBLIC + table: ORDERS + primary_key: + columns: [order_id] + dimensions: + - name: order_date + expr: order_date + facts: + - name: amount + expr: amount +``` + +### Import: Snowflake → OSI + +Convert a Snowflake Cortex Analyst semantic model to OSI format: + ```bash -python3 src/osi_to_snowflake_yaml_converter.py -i input.yaml -o output.yaml +python3 src/snowflake_to_osi_yaml_converter.py -i input_snowflake.yaml -o output_osi.yaml +``` + +**Example input (Snowflake format):** +```yaml +name: retail_model +tables: + - name: orders + base_table: + database: RETAIL_DB + schema: PUBLIC + table: ORDERS + dimensions: + - name: order_date + expr: order_date + - name: customer_id + expr: customer_id + facts: + - name: amount + expr: amount +relationships: + - name: orders_to_customers + left_table: orders + right_table: customers + relationship_columns: + - left_column: customer_id + right_column: id +``` + +**Example output (OSI format):** +```yaml +version: "0.2.0.dev0" +semantic_model: + - name: retail_model + datasets: + - name: orders + source: RETAIL_DB.PUBLIC.ORDERS + fields: + - name: order_date + expression: + dialects: + - dialect: SNOWFLAKE + expression: order_date + dimension: + is_time: false + - name: customer_id + expression: + dialects: + - dialect: SNOWFLAKE + expression: customer_id + dimension: + is_time: false + - name: amount + expression: + dialects: + - dialect: SNOWFLAKE + expression: amount + - name: customers + source: RETAIL_DB.PUBLIC.CUSTOMERS + relationships: + - name: orders_to_customers + from: orders + to: customers + from_columns: [customer_id] + to_columns: [id] ``` ## Tests @@ -22,6 +145,75 @@ python3 src/osi_to_snowflake_yaml_converter.py -i input.yaml -o output.yaml python3 -m pytest tests/ ``` +Include verbose output to see warnings emitted during conversion: +```bash +python3 -m pytest tests/ -v -W default +``` + +## Mapping Reference + +### Datasets & Tables + +| OSI | Snowflake | +|-----|-----------| +| `dataset.name` | `table.name` | +| `dataset.source` (e.g., `db.schema.table`) | `table.base_table` (parsed into `database`, `schema`, `table`) | +| `dataset.primary_key` (array) | `table.primary_key.columns` | +| `dataset.unique_keys` (array of arrays) | `table.unique_keys` (array of `{columns: [...]}`) | +| `dataset.description` | `table.description` | +| `dataset.ai_context.synonyms` | `table.synonyms` | + +### Fields + +| OSI | Snowflake | +|-----|-----------| +| `field.name` | `dimension.name`, `fact.name`, `measure.name` | +| `field.expression.dialects[].expression` (SNOWFLAKE dialect) | `dimension.expr`, `fact.expr`, `measure.expr` | +| `field.dimension.is_time = true` | `time_dimension` | +| `field.dimension.is_time = false` | `dimension` | +| No dimension or `is_time = null` | `fact` or `measure` | +| `field.description` | `dimension.description`, `fact.description`, etc. | +| `field.ai_context.synonyms` | `dimension.synonyms`, `fact.synonyms`, etc. | + +### Relationships + +| OSI | Snowflake | +|-----|-----------| +| `relationship.from` | `relationship.left_table` | +| `relationship.to` | `relationship.right_table` | +| `relationship.from_columns` | `relationship.relationship_columns[].left_column` | +| `relationship.to_columns` | `relationship.relationship_columns[].right_column` | + +### Top-level Metrics + +| OSI | Snowflake | +|-----|-----------| +| `metric.name` | `metric.name` | +| `metric.expression.dialects[].expression` | `metric.expr` | +| `metric.description` | `metric.description` | +| `metric.ai_context.synonyms` | `metric.synonyms` | + ## Limitations -Some OSI concepts (e.g., `ai_context` on relationships) do not have a native counterpart in the Snowflake semantic model. These are dropped during conversion and the converter will emit warnings so you know what was left behind. +### OSI → Snowflake Export + +- **Dropped fields**: `ai_context` (when structured as an object; string instructions are merged into description), `custom_extensions`, `label`, `version` on individual entities +- **Relationships**: `ai_context` and `custom_extensions` on relationships are not supported and are dropped with warnings +- **Expression dialects**: Only `SNOWFLAKE` or `ANSI_SQL` dialects are recognized. Other dialects (e.g., `MDX`, `TABLEAU`) are skipped with warnings. + +### Snowflake → OSI Import + +- **Vendor-specific Snowflake fields** (e.g., `warehouse`, `database`, `schema` at the model level, or any unknown table/field attributes) are preserved in `custom_extensions` with `vendor_name: SNOWFLAKE` for round-trip fidelity, preventing silent data loss. +- **Expressions**: All Snowflake expressions are wrapped with `dialect: SNOWFLAKE` when imported to OSI. To support cross-dialect execution, manually add `ANSI_SQL` dialect variants if needed. +- **Measures vs Facts**: Snowflake `measures` are imported as OSI fields with no `dimension` property. Users should annotate fields appropriately in the OSI model. +- **Missing expressions**: Fields with empty or missing `expr` are skipped with warnings. + +## Round-Trip Behavior + +Round-tripping (OSI → Snowflake → OSI or Snowflake → OSI → Snowflake) preserves the core model structure (datasets, fields, relationships, metrics) but may lose: + +- **String-only `ai_context`** in relationships (dropped in Snowflake export, cannot be restored) +- **`label` fields** (dropped in export, no Snowflake equivalent) +- **Multi-dialect expressions**: Only the selected dialect is preserved; other dialects are lost + +Use `custom_extensions` to preserve vendor-specific metadata that must survive round-tripping. diff --git a/converters/snowflake/src/snowflake_to_osi_yaml_converter.py b/converters/snowflake/src/snowflake_to_osi_yaml_converter.py new file mode 100644 index 0000000..a2b7d52 --- /dev/null +++ b/converters/snowflake/src/snowflake_to_osi_yaml_converter.py @@ -0,0 +1,637 @@ +""" +Converts a Snowflake Cortex Analyst semantic model YAML to an OSI (Open Semantic +Interchange) YAML semantic model. Pure offline conversion — no Snowflake connection +required. + +Usage: + python3 snowflake_to_osi_yaml_converter.py -i input.yaml -o output.yaml +""" + +import argparse +import json +import sys +import warnings + +import yaml + + +SUPPORTED_VERSION = "0.2.0.dev0" + + +class SnowflakeConversionError(Exception): + """Raised when a Snowflake YAML cannot be converted to OSI format.""" + + +def convert_snowflake_to_osi(snowflake_yaml_str): + """Top-level entry point. Parses Snowflake YAML, validates, converts, returns + OSI YAML string. + + A valid Snowflake Cortex Analyst semantic model is an object with: + - name (required) + - description (optional) + - tables (array of table objects) + - relationships (optional, array of relationship objects) + - metrics (optional, array of metric objects) + + Args: + snowflake_yaml_str: Snowflake YAML as a string. + + Returns: + OSI YAML string with version "0.2.0.dev0". + + Raises: + SnowflakeConversionError: If the input cannot be converted. + """ + root = yaml.safe_load(snowflake_yaml_str) + if not isinstance(root, dict): + raise SnowflakeConversionError( + "Invalid Snowflake YAML: expected a mapping at the root" + ) + + snowflake_model = root + name = snowflake_model.get("name") + if not name: + raise SnowflakeConversionError("Missing required 'name' field in semantic model") + + # Convert the Snowflake model to OSI + osi_model = _convert_model(snowflake_model) + + # Wrap in OSI envelope + osi_document = { + "version": SUPPORTED_VERSION, + "semantic_model": [osi_model], + } + + return yaml.dump( + osi_document, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + + +def _convert_model(snowflake_model): + """Converts the root Snowflake model dict to an OSI model dict.""" + result = {} + + # name is required + name = snowflake_model.get("name") + if not name: + raise SnowflakeConversionError("Missing required 'name' field in semantic model") + result["name"] = name + + # description (optional) + description = snowflake_model.get("description") + if description: + result["description"] = description + + # tables -> datasets + # Also collect metrics from table-level sources + tables = snowflake_model.get("tables", []) + all_metrics = [] + if tables: + datasets = [] + for table in tables: + dataset, table_metrics = _convert_table(table) + if dataset is not None: + datasets.append(dataset) + all_metrics.extend(table_metrics) + if datasets: + result["datasets"] = datasets + + # relationships + relationships = snowflake_model.get("relationships", []) + if relationships: + converted_rels = [_convert_relationship(rel) for rel in relationships] + if converted_rels: + result["relationships"] = converted_rels + + # metrics: combine top-level and table-level metrics + # Top-level metrics from the model + top_level_metrics = snowflake_model.get("metrics", []) + if top_level_metrics: + for m in top_level_metrics: + converted = _convert_named_expr(m, "metric") + if converted is not None: + all_metrics.append(converted) + + # Deduplicate and qualify metric names to avoid collisions + if all_metrics: + # Check for duplicate names and qualify if needed + seen_names = {} + final_metrics = [] + for metric in all_metrics: + metric_name = metric.get("name", "") + if metric_name in seen_names: + # Collision detected - qualify with table name from context if available + # For now, just warn and use the first occurrence + warnings.warn( + f"Duplicate metric name '{metric_name}' detected during conversion; " + f"using first occurrence" + ) + else: + seen_names[metric_name] = True + final_metrics.append(metric) + + if final_metrics: + result["metrics"] = final_metrics + + # Warn about dropped fields + _warn_dropped_fields(snowflake_model, "model") + + return result + + +def _convert_table(table): + """Converts a Snowflake table dict to an OSI dataset dict. + + Per the official Snowflake semantic view YAML spec, tables can have their own + `metrics` list (table-scoped). These are extracted and returned separately so + they can be merged into the top-level OSI metrics list by the caller. + + Returns: + Tuple of (dataset_dict, list_of_table_level_metrics) + """ + result = {} + + # name is required + name = table.get("name") + if not name: + raise SnowflakeConversionError("Missing required 'name' field in table") + result["name"] = name + + # base_table -> source + # base_table can be: + # - {database: "DB", schema: "SCHEMA", table: "TABLE"} + # - {definition: "SELECT ..."} + base_table = table.get("base_table") + source = _convert_base_table_to_source(base_table) + if source is not None: + result["source"] = source + + # primary_key: {columns: [...]} -> primary_key: [...] + pk_obj = table.get("primary_key") + if pk_obj and isinstance(pk_obj, dict): + pk_columns = pk_obj.get("columns", []) + if pk_columns: + result["primary_key"] = pk_columns + + # unique_keys: [{columns: [...]}, ...] -> unique_keys: [[...], ...] + uks = table.get("unique_keys", []) + if uks: + uk_arrays = [] + for uk_obj in uks: + if isinstance(uk_obj, dict): + uk_cols = uk_obj.get("columns", []) + if uk_cols: + uk_arrays.append(uk_cols) + if uk_arrays: + result["unique_keys"] = uk_arrays + + # description (optional) + description = table.get("description") + ai_context = None + + # Extract synonyms + synonyms = table.get("synonyms", []) + if synonyms: + if ai_context is None: + ai_context = {} + ai_context["synonyms"] = synonyms + + # Set description and ai_context + if description: + result["description"] = description + if ai_context: + result["ai_context"] = ai_context + + # Collect vendor-specific settings not captured by OSI core + custom_extensions = _extract_vendor_specific_table_settings(table) + if custom_extensions: + result["custom_extensions"] = custom_extensions + + # Convert fields + fields = [] + + # Collect all field-like arrays: dimensions, time_dimensions, facts + # NOTE: Per the Snowflake semantic view spec, only dimensions, time_dimensions, + # facts, and metrics are valid. The legacy "measures" field is not part of + # semantic views and is not handled here. + dimensions = table.get("dimensions", []) + time_dimensions = table.get("time_dimensions", []) + facts = table.get("facts", []) + + for dim in dimensions: + converted = _convert_field(dim, is_time=False) + if converted is not None: + fields.append(converted) + + for tdim in time_dimensions: + converted = _convert_field(tdim, is_time=True) + if converted is not None: + fields.append(converted) + + for fact in facts: + converted = _convert_field(fact, is_time=None) + if converted is not None: + fields.append(converted) + + if fields: + result["fields"] = fields + + # Extract table-level metrics + # Per the spec, each table can have its own metrics list. + # These are converted and returned separately to be merged into the top-level + # metrics list by the caller. + table_metrics = [] + metrics_list = table.get("metrics", []) + if metrics_list: + for m in metrics_list: + converted = _convert_named_expr(m, f"table-level metric from table '{name}'") + if converted is not None: + table_metrics.append(converted) + + _warn_dropped_fields(table, f"table '{name}'") + + return result, table_metrics + + +def _convert_base_table_to_source(base_table): + """Converts a Snowflake base_table dict to an OSI source string. + + Returns None if base_table is None or empty. + """ + if base_table is None: + return None + + if not isinstance(base_table, dict): + # If it's a string or something else, preserve it as-is in a warning + warnings.warn(f"base_table has unexpected type {type(base_table).__name__}; skipping") + return None + + # Case 1: {database, schema, table} -> "DB.SCHEMA.TABLE" + db = base_table.get("database") + schema = base_table.get("schema") + table = base_table.get("table") + + if db and schema and table: + return f"{db}.{schema}.{table}" + + # Case 2: {definition} -> use as-is + definition = base_table.get("definition") + if definition: + return definition + + # Empty base_table + return None + + +def _convert_field(field_obj, is_time): + """Converts a Snowflake field (dimension, time_dimension, fact, measure) to an OSI field. + + Args: + field_obj: The Snowflake field dict (must have 'name' and 'expr'). + is_time: True if this is a time dimension, False if regular dimension, None if fact/measure. + + Returns: + An OSI field dict, or None if conversion fails. + """ + name = field_obj.get("name") + if not name: + warnings.warn("Skipping field with missing 'name'") + return None + + expr_str = field_obj.get("expr") + if not expr_str: + warnings.warn(f"Skipping field '{name}' with missing or empty 'expr'") + return None + + result = {} + result["name"] = name + + # expression: wrap in SNOWFLAKE dialect + result["expression"] = { + "dialects": [ + { + "dialect": "SNOWFLAKE", + "expression": expr_str, + } + ] + } + + # dimension: set is_time if applicable + if is_time is not None: + result["dimension"] = {"is_time": is_time} + + # description (optional) + description = field_obj.get("description") + if description: + result["description"] = description + + # Extract synonyms into ai_context + ai_context = None + synonyms = field_obj.get("synonyms", []) + if synonyms: + ai_context = {"synonyms": synonyms} + + if ai_context: + result["ai_context"] = ai_context + + # Vendor-specific settings + custom_extensions = _extract_vendor_specific_field_settings(field_obj) + if custom_extensions: + result["custom_extensions"] = custom_extensions + + return result + + +def _convert_relationship(rel): + """Converts a Snowflake relationship dict to an OSI relationship dict.""" + result = {} + + # name is required + name = rel.get("name") + if not name: + raise SnowflakeConversionError("Missing required 'name' field in relationship") + result["name"] = name + + # left_table -> from, right_table -> to + from_dataset = rel.get("left_table") + to_dataset = rel.get("right_table") + + if not from_dataset: + raise SnowflakeConversionError( + f"Relationship '{name}': missing required 'left_table' field" + ) + if not to_dataset: + raise SnowflakeConversionError( + f"Relationship '{name}': missing required 'right_table' field" + ) + + result["from"] = from_dataset + result["to"] = to_dataset + + # Convert relationship_columns: [{left_column, right_column}, ...] -> from_columns, to_columns + rel_cols = rel.get("relationship_columns", []) + from_columns = [] + to_columns = [] + + for col_pair in rel_cols: + if isinstance(col_pair, dict): + left_col = col_pair.get("left_column") + right_col = col_pair.get("right_column") + if left_col: + from_columns.append(left_col) + if right_col: + to_columns.append(right_col) + + if from_columns: + result["from_columns"] = from_columns + if to_columns: + result["to_columns"] = to_columns + + # Validate matching cardinality + if len(from_columns) != len(to_columns): + warnings.warn( + f"Relationship '{name}': from_columns and to_columns have different lengths " + f"({len(from_columns)} vs {len(to_columns)}); may cause issues" + ) + + _warn_dropped_fields(rel, f"relationship '{name}'") + + return result + + +def _convert_named_expr(entry, kind): + """Converts a Snowflake metric/measure dict to an OSI metric dict. + + Args: + entry: The Snowflake metric/measure dict. + kind: Human-readable type (e.g., "metric", "measure"). + + Returns: + An OSI metric dict, or None if conversion fails. + """ + name = entry.get("name") + if not name: + warnings.warn(f"Skipping {kind} with missing 'name'") + return None + + expr_str = entry.get("expr") + if not expr_str: + warnings.warn(f"Skipping {kind} '{name}' with missing or empty 'expr'") + return None + + result = {} + result["name"] = name + + # expression: wrap in SNOWFLAKE dialect + result["expression"] = { + "dialects": [ + { + "dialect": "SNOWFLAKE", + "expression": expr_str, + } + ] + } + + # description (optional) + description = entry.get("description") + if description: + result["description"] = description + + # Extract synonyms into ai_context + ai_context = None + synonyms = entry.get("synonyms", []) + if synonyms: + ai_context = {"synonyms": synonyms} + + if ai_context: + result["ai_context"] = ai_context + + # Vendor-specific settings + custom_extensions = _extract_vendor_specific_measure_settings(entry) + if custom_extensions: + result["custom_extensions"] = custom_extensions + + return result + + +def _extract_vendor_specific_table_settings(table): + """Extracts vendor-specific Snowflake table settings into custom_extensions. + + These are fields that Snowflake supports but OSI core does not, and should + be preserved for round-tripping. All legitimately-handled spec fields are + excluded from custom_extensions. + """ + vendor_data = {} + + # Per the official Snowflake semantic view YAML spec, the following fields + # are valid at the table level and are handled by the converter: + # - name: required, converted to dataset.name + # - base_table: converted to dataset.source + # - primary_key: converted to dataset.primary_key + # - unique_keys: converted to dataset.unique_keys + # - description: converted to dataset.description + # - synonyms: converted to dataset.ai_context.synonyms + # - dimensions: converted to dataset.fields with is_time=false + # - time_dimensions: converted to dataset.fields with is_time=true + # - facts: converted to dataset.fields (no dimension property) + # - metrics: table-level metrics, extracted and merged into top-level metrics + # - filters: Snowflake-specific row-level security filters + # - tags: Snowflake tags for metadata (preserved in custom_extensions as vendor-specific) + # + # The following fields are legitimately-handled spec fields and should NOT + # be added to custom_extensions: + known_osi_fields = { + "name", "base_table", "primary_key", "unique_keys", + "description", "synonyms", "dimensions", "time_dimensions", + "facts", "metrics", "filters", "tags", + } + + # Any other field encountered should be preserved in custom_extensions + for key, value in table.items(): + if key not in known_osi_fields: + vendor_data[key] = value + + if not vendor_data: + return None + + return [ + { + "vendor_name": "SNOWFLAKE", + "data": json.dumps(vendor_data), + } + ] + + +def _extract_vendor_specific_field_settings(field_obj): + """Extracts vendor-specific Snowflake field settings into custom_extensions. + + Per the official Snowflake semantic view YAML spec, the following field-level + properties are recognized: + - name: required, converted to field.name + - expr: required, converted to field.expression.dialects[].expression + - description: converted to field.description + - synonyms: converted to field.ai_context.synonyms + - data_type: field data type (Snowflake-specific, no OSI equivalent) → NOT preserved + - unique: uniqueness constraint (no direct OSI mapping) → NOT preserved + - is_enum: enum type marker (OSI has no enum support) → NOT preserved + - cortex_search_service: Snowflake Cortex Search integration → NOT preserved + - sample_values: example values for AI context → NOT preserved + - labels: field categorization labels (OSI drops labels anyway) → NOT preserved + - tags: Snowflake metadata tags → NOT preserved + - access_modifier: field access control → NOT preserved + + Since these vendor-specific properties don't map to OSI concepts and would be + complex to round-trip, we exclude them from custom_extensions and drop them + silently (they're recognized as legitimate fields, not "unknown" extensions). + """ + vendor_data = {} + + # All known field properties per the spec + known_osi_fields = { + # Handled by converter + "name", "expr", "description", "synonyms", + # Spec fields that don't map to OSI and are intentionally dropped + "data_type", "unique", "is_enum", "cortex_search_service", + "sample_values", "labels", "tags", "access_modifier", + } + + for key, value in field_obj.items(): + if key not in known_osi_fields: + vendor_data[key] = value + + if not vendor_data: + return None + + return [ + { + "vendor_name": "SNOWFLAKE", + "data": json.dumps(vendor_data), + } + ] + + +def _extract_vendor_specific_measure_settings(measure): + """Extracts vendor-specific Snowflake metric/measure settings into custom_extensions. + + Metrics in the Snowflake spec have the following properties: + - name: required, converted to metric.name + - expr: required, converted to metric.expression.dialects[].expression + - description: converted to metric.description + - synonyms: converted to metric.ai_context.synonyms + - data_type: metric result type (Snowflake-specific, no OSI mapping) → NOT preserved + - tags: Snowflake metadata tags → NOT preserved + - access_modifier: metric access control → NOT preserved + + Snowflake-specific properties that don't map to OSI are intentionally dropped. + """ + vendor_data = {} + + known_osi_fields = { + # Handled by converter + "name", "expr", "description", "synonyms", + # Spec fields that don't map to OSI and are intentionally dropped + "data_type", "tags", "access_modifier", + } + + for key, value in measure.items(): + if key not in known_osi_fields: + vendor_data[key] = value + + if not vendor_data: + return None + + return [ + { + "vendor_name": "SNOWFLAKE", + "data": json.dumps(vendor_data), + } + ] + + +def _warn_dropped_fields(source, context): + """Warns about Snowflake fields that have no OSI counterpart. + + These are fields that exist in the Snowflake model but should not be + dropped silently; they should be captured in custom_extensions or + explicitly warned about. + """ + # Snowflake-specific fields that don't map to OSI and weren't handled: + # - warehouse, database, schema (table-level config) + # - any other vendor-specific fields (these should be in custom_extensions now) + + # For now, we don't warn about anything here since we capture all unknown + # fields in custom_extensions. The warning is implicit in the custom_extensions. + + +def main(): + parser = argparse.ArgumentParser( + description="Convert Snowflake Cortex Analyst YAML to OSI semantic model YAML" + ) + parser.add_argument( + "-i", "--input", required=True, help="Path to the Snowflake YAML input file" + ) + parser.add_argument( + "-o", "--output", required=True, help="Path to write the OSI YAML output" + ) + args = parser.parse_args() + + with open(args.input, "r") as f: + snowflake_yaml_str = f.read() + + try: + osi_yaml_str = convert_snowflake_to_osi(snowflake_yaml_str) + except SnowflakeConversionError as e: + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + + with open(args.output, "w") as f: + f.write(osi_yaml_str) + + print(f"Converted {args.input} -> {args.output}") + + +if __name__ == "__main__": + main() diff --git a/converters/snowflake/tests/test_snowflake_to_osi_yaml_converter.py b/converters/snowflake/tests/test_snowflake_to_osi_yaml_converter.py new file mode 100644 index 0000000..419bd61 --- /dev/null +++ b/converters/snowflake/tests/test_snowflake_to_osi_yaml_converter.py @@ -0,0 +1,662 @@ +"""Tests for the Snowflake to OSI YAML converter.""" + +import sys +import json +from pathlib import Path + +import pytest +import yaml + +# Make src/ importable +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) +from snowflake_to_osi_yaml_converter import ( + SnowflakeConversionError, + convert_snowflake_to_osi, + _convert_model, + _convert_table, + _convert_field, + _convert_relationship, + _convert_named_expr, + _convert_base_table_to_source, +) + +# Also import the OSI-to-Snowflake converter for round-trip testing +from osi_to_snowflake_yaml_converter import convert_osi_to_snowflake + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _minimal_snowflake_model(**overrides): + """Return a minimal valid Snowflake model dict.""" + base = { + "name": "test_model", + "tables": [ + { + "name": "my_table", + "base_table": { + "database": "DB", + "schema": "SCHEMA", + "table": "TBL", + }, + "dimensions": [ + { + "name": "col1", + "expr": "col1", + } + ], + } + ], + } + base.update(overrides) + return base + + +# --------------------------------------------------------------------------- +# _convert_base_table_to_source +# --------------------------------------------------------------------------- + +class TestConvertBaseTableToSource: + def test_three_part_name(self): + base_table = {"database": "DB", "schema": "SCHEMA", "table": "TABLE"} + result = _convert_base_table_to_source(base_table) + assert result == "DB.SCHEMA.TABLE" + + def test_definition_subquery(self): + base_table = {"definition": "SELECT * FROM foo"} + result = _convert_base_table_to_source(base_table) + assert result == "SELECT * FROM foo" + + def test_none_returns_none(self): + assert _convert_base_table_to_source(None) is None + + def test_empty_dict_returns_none(self): + assert _convert_base_table_to_source({}) is None + + +# --------------------------------------------------------------------------- +# _convert_field +# --------------------------------------------------------------------------- + +class TestConvertField: + def test_dimension_field(self): + field = {"name": "col1", "expr": "col1"} + result = _convert_field(field, is_time=False) + assert result["name"] == "col1" + assert result["expression"]["dialects"][0]["dialect"] == "SNOWFLAKE" + assert result["expression"]["dialects"][0]["expression"] == "col1" + assert result["dimension"]["is_time"] is False + + def test_time_dimension_field(self): + field = {"name": "col1", "expr": "col1"} + result = _convert_field(field, is_time=True) + assert result["dimension"]["is_time"] is True + + def test_fact_field(self): + field = {"name": "col1", "expr": "col1"} + result = _convert_field(field, is_time=None) + assert "dimension" not in result + + def test_field_with_description(self): + field = {"name": "col1", "expr": "col1", "description": "A column"} + result = _convert_field(field, is_time=False) + assert result["description"] == "A column" + + def test_field_with_synonyms(self): + field = { + "name": "col1", + "expr": "col1", + "synonyms": ["column_1", "col"], + } + result = _convert_field(field, is_time=False) + assert result["ai_context"]["synonyms"] == ["column_1", "col"] + + def test_field_missing_name(self): + field = {"expr": "col1"} + result = _convert_field(field, is_time=False) + assert result is None + + def test_field_missing_expr(self): + field = {"name": "col1"} + result = _convert_field(field, is_time=False) + assert result is None + + +# --------------------------------------------------------------------------- +# _convert_named_expr (metrics) +# --------------------------------------------------------------------------- + +class TestConvertNamedExpr: + def test_metric_basic(self): + metric = {"name": "total_sales", "expr": "SUM(sales)"} + result = _convert_named_expr(metric, "metric") + assert result["name"] == "total_sales" + assert result["expression"]["dialects"][0]["expression"] == "SUM(sales)" + + def test_metric_with_description(self): + metric = {"name": "total_sales", "expr": "SUM(sales)", "description": "Total revenue"} + result = _convert_named_expr(metric, "metric") + assert result["description"] == "Total revenue" + + def test_metric_with_synonyms(self): + metric = { + "name": "total_sales", + "expr": "SUM(sales)", + "synonyms": ["revenue", "total_amount"], + } + result = _convert_named_expr(metric, "metric") + assert result["ai_context"]["synonyms"] == ["revenue", "total_amount"] + + def test_metric_missing_name(self): + metric = {"expr": "SUM(sales)"} + result = _convert_named_expr(metric, "metric") + assert result is None + + def test_metric_missing_expr(self): + metric = {"name": "total_sales"} + result = _convert_named_expr(metric, "metric") + assert result is None + + +# --------------------------------------------------------------------------- +# _convert_relationship +# --------------------------------------------------------------------------- + +class TestConvertRelationship: + def test_simple_relationship(self): + rel = { + "name": "orders_to_customers", + "left_table": "orders", + "right_table": "customers", + "relationship_columns": [ + {"left_column": "customer_id", "right_column": "id"} + ], + } + result = _convert_relationship(rel) + assert result["name"] == "orders_to_customers" + assert result["from"] == "orders" + assert result["to"] == "customers" + assert result["from_columns"] == ["customer_id"] + assert result["to_columns"] == ["id"] + + def test_composite_key_relationship(self): + rel = { + "name": "items_to_products", + "left_table": "items", + "right_table": "products", + "relationship_columns": [ + {"left_column": "product_id", "right_column": "id"}, + {"left_column": "variant_id", "right_column": "variant_id"}, + ], + } + result = _convert_relationship(rel) + assert result["from_columns"] == ["product_id", "variant_id"] + assert result["to_columns"] == ["id", "variant_id"] + + def test_missing_left_table(self): + rel = { + "name": "bad_rel", + "right_table": "customers", + } + with pytest.raises(SnowflakeConversionError, match="left_table"): + _convert_relationship(rel) + + def test_missing_right_table(self): + rel = { + "name": "bad_rel", + "left_table": "orders", + } + with pytest.raises(SnowflakeConversionError, match="right_table"): + _convert_relationship(rel) + + def test_missing_name(self): + rel = { + "left_table": "orders", + "right_table": "customers", + } + with pytest.raises(SnowflakeConversionError, match="name"): + _convert_relationship(rel) + + +# --------------------------------------------------------------------------- +# _convert_table +# --------------------------------------------------------------------------- + +class TestConvertTable: + def test_minimal_table(self): + table = { + "name": "my_table", + "base_table": {"database": "DB", "schema": "SCHEMA", "table": "TBL"}, + } + dataset, metrics = _convert_table(table) + assert dataset["name"] == "my_table" + assert dataset["source"] == "DB.SCHEMA.TBL" + assert metrics == [] + + def test_table_with_primary_key(self): + table = { + "name": "my_table", + "primary_key": {"columns": ["id"]}, + } + dataset, metrics = _convert_table(table) + assert dataset["primary_key"] == ["id"] + + def test_table_with_composite_primary_key(self): + table = { + "name": "my_table", + "primary_key": {"columns": ["id1", "id2"]}, + } + dataset, metrics = _convert_table(table) + assert dataset["primary_key"] == ["id1", "id2"] + + def test_table_with_unique_keys(self): + table = { + "name": "my_table", + "unique_keys": [ + {"columns": ["email"]}, + {"columns": ["id1", "id2"]}, + ], + } + dataset, metrics = _convert_table(table) + assert dataset["unique_keys"] == [["email"], ["id1", "id2"]] + + def test_table_with_description(self): + table = { + "name": "my_table", + "description": "A test table", + } + dataset, metrics = _convert_table(table) + assert dataset["description"] == "A test table" + + def test_table_with_synonyms(self): + table = { + "name": "my_table", + "synonyms": ["table_1", "tbl"], + } + dataset, metrics = _convert_table(table) + assert dataset["ai_context"]["synonyms"] == ["table_1", "tbl"] + + def test_table_with_dimensions_and_facts(self): + table = { + "name": "my_table", + "dimensions": [ + {"name": "dim1", "expr": "dim1"}, + ], + "facts": [ + {"name": "fact1", "expr": "fact1"}, + ], + } + dataset, metrics = _convert_table(table) + assert len(dataset["fields"]) == 2 + # Find dim1 and fact1 + field_names = {f["name"] for f in dataset["fields"]} + assert "dim1" in field_names + assert "fact1" in field_names + + def test_table_with_time_dimensions(self): + table = { + "name": "my_table", + "time_dimensions": [ + {"name": "date", "expr": "date"}, + ], + } + dataset, metrics = _convert_table(table) + fields = dataset["fields"] + assert len(fields) == 1 + assert fields[0]["name"] == "date" + assert fields[0]["dimension"]["is_time"] is True + + def test_table_with_table_level_metrics(self): + """Test that table-level metrics are extracted and returned separately.""" + table = { + "name": "my_table", + "base_table": {"database": "DB", "schema": "SCHEMA", "table": "TBL"}, + "dimensions": [ + {"name": "date", "expr": "date"}, + ], + "metrics": [ + {"name": "total_sales", "expr": "SUM(sales)"}, + {"name": "avg_price", "expr": "AVG(price)"}, + ], + } + dataset, metrics = _convert_table(table) + assert dataset["name"] == "my_table" + # Verify metrics were extracted + assert len(metrics) == 2 + metric_names = {m["name"] for m in metrics} + assert "total_sales" in metric_names + assert "avg_price" in metric_names + + def test_missing_table_name(self): + table = {"base_table": {"database": "DB", "schema": "SCHEMA", "table": "TBL"}} + with pytest.raises(SnowflakeConversionError, match="name"): + _convert_table(table) + + +# --------------------------------------------------------------------------- +# _convert_model +# --------------------------------------------------------------------------- + +class TestConvertModel: + def test_minimal_model(self): + model = _minimal_snowflake_model() + result = _convert_model(model) + assert result["name"] == "test_model" + assert "datasets" in result + + def test_model_with_relationships(self): + model = _minimal_snowflake_model( + relationships=[ + { + "name": "rel1", + "left_table": "my_table", + "right_table": "other_table", + "relationship_columns": [], + } + ] + ) + result = _convert_model(model) + assert "relationships" in result + + def test_model_with_metrics(self): + model = _minimal_snowflake_model( + metrics=[ + { + "name": "metric1", + "expr": "SUM(amount)", + } + ] + ) + result = _convert_model(model) + assert "metrics" in result + + def test_missing_model_name(self): + model = {"tables": []} + with pytest.raises(SnowflakeConversionError, match="name"): + _convert_model(model) + + +# --------------------------------------------------------------------------- +# Integration tests: Full YAML conversion +# --------------------------------------------------------------------------- + +class TestFullConversion: + def test_minimal_valid_snowflake_yaml(self): + snowflake_yaml = """ +name: test_model +tables: + - name: my_table + base_table: + database: DB + schema: SCHEMA + table: TBL + dimensions: + - name: col1 + expr: col1 +""" + result = convert_snowflake_to_osi(snowflake_yaml) + parsed = yaml.safe_load(result) + assert parsed["version"] == "0.2.0.dev0" + assert "semantic_model" in parsed + assert len(parsed["semantic_model"]) == 1 + model = parsed["semantic_model"][0] + assert model["name"] == "test_model" + assert len(model["datasets"]) == 1 + + def test_comprehensive_snowflake_yaml(self): + snowflake_yaml = """ +name: retail_model +description: Retail analytics model +tables: + - name: orders + base_table: + database: RETAIL + schema: PUBLIC + table: ORDERS + primary_key: + columns: [order_id] + synonyms: + - order_transactions + - order_records + dimensions: + - name: order_date + expr: order_date + description: Date of order + time_dimensions: + - name: created_at + expr: created_at + facts: + - name: quantity + expr: quantity + - name: customers + base_table: + database: RETAIL + schema: PUBLIC + table: CUSTOMERS + dimensions: + - name: customer_id + expr: customer_id +relationships: + - name: orders_customers + left_table: orders + right_table: customers + relationship_columns: + - left_column: customer_id + right_column: customer_id +metrics: + - name: total_sales + expr: SUM(amount) + description: Total sales amount +""" + result = convert_snowflake_to_osi(snowflake_yaml) + parsed = yaml.safe_load(result) + assert parsed["version"] == "0.2.0.dev0" + model = parsed["semantic_model"][0] + assert model["name"] == "retail_model" + assert len(model["datasets"]) == 2 + assert "relationships" in model + assert "metrics" in model + + def test_table_level_and_top_level_metrics(self): + """Test that both table-level and top-level metrics are combined.""" + snowflake_yaml = """ +name: analytics_model +tables: + - name: sales + base_table: + database: DB + schema: PUBLIC + table: SALES + dimensions: + - name: date + expr: date + metrics: + - name: sales_revenue + expr: SUM(revenue) + description: Total revenue per table + - name: sales_transactions + expr: COUNT(*) + description: Transaction count + - name: costs + base_table: + database: DB + schema: PUBLIC + table: COSTS + dimensions: + - name: date + expr: date + metrics: + - name: cost_total + expr: SUM(cost_amount) +metrics: + - name: profit_margin + expr: (sales_revenue - cost_total) / sales_revenue + description: Profit margin calculated from derived metrics +""" + result = convert_snowflake_to_osi(snowflake_yaml) + parsed = yaml.safe_load(result) + model = parsed["semantic_model"][0] + + # Check that all metrics are present (3 from tables + 1 top-level = 4 total) + assert "metrics" in model + metrics = model["metrics"] + assert len(metrics) == 4 + + metric_names = {m["name"] for m in metrics} + # Table-level metrics from sales table + assert "sales_revenue" in metric_names + assert "sales_transactions" in metric_names + # Table-level metrics from costs table + assert "cost_total" in metric_names + # Top-level metric + assert "profit_margin" in metric_names + + def test_spec_fields_not_dumped_to_custom_extensions(self): + """Test that recognized Snowflake spec fields are NOT treated as unknown vendor extensions. + + Per the semantic view spec, fields like data_type, unique, is_enum, tags, etc. + are recognized but intentionally not mapped to OSI. They should be dropped silently, + not dumped into custom_extensions. + """ + snowflake_yaml = """ +name: test_model +tables: + - name: sales + base_table: + database: DB + schema: PUBLIC + table: SALES + tags: + - production + - critical + filters: + - name: row_filter + expr: user_id = current_user() + dimensions: + - name: product_id + expr: product_id + data_type: NUMBER + is_enum: true + unique: true + - name: category + expr: category + tags: + - business_critical + labels: + - dimension + facts: + - name: quantity + expr: quantity + data_type: INTEGER + metrics: + - name: revenue + expr: SUM(amount) + access_modifier: PUBLIC +""" + result = convert_snowflake_to_osi(snowflake_yaml) + parsed = yaml.safe_load(result) + model = parsed["semantic_model"][0] + dataset = model["datasets"][0] + + # The dataset should NOT have custom_extensions (unless there are truly unknown fields) + # Spec fields like tags, filters should not appear in custom_extensions + if "custom_extensions" in dataset: + # If custom_extensions exist, verify they don't contain spec field names + for ext in dataset.get("custom_extensions", []): + if ext.get("vendor_name") == "SNOWFLAKE": + data = json.loads(ext.get("data", "{}")) + # Should not have 'tags' or 'filters' which are recognized spec fields + assert "tags" not in data, "Recognized spec field 'tags' should not be in custom_extensions" + assert "filters" not in data, "Recognized spec field 'filters' should not be in custom_extensions" + + def test_invalid_snowflake_yaml(self): + # Valid YAML but structurally invalid for Snowflake (missing 'name') + snowflake_yaml = """ +tables: + - name: test +""" + with pytest.raises(SnowflakeConversionError, match="name"): + convert_snowflake_to_osi(snowflake_yaml) + + +# --------------------------------------------------------------------------- +# Round-trip tests: OSI -> Snowflake -> OSI +# --------------------------------------------------------------------------- + +class TestRoundTrip: + def test_roundtrip_preserves_core_structure(self): + """ + Test that converting OSI -> Snowflake -> OSI preserves the core structure. + + Note: Some fields may be lost in round-tripping due to vendor-specific + differences (e.g., custom_extensions), but core datasets, fields, relationships, + and metrics should be preserved. + """ + # Start with an OSI model + osi_yaml = """ +version: "0.2.0.dev0" +semantic_model: + - name: test_model + datasets: + - name: orders + source: db.schema.orders + primary_key: [order_id] + fields: + - name: order_date + expression: + dialects: + - dialect: SNOWFLAKE + expression: order_date + dimension: + is_time: false + - name: amount + expression: + dialects: + - dialect: SNOWFLAKE + expression: amount + - name: customers + source: db.schema.customers + fields: + - name: customer_id + expression: + dialects: + - dialect: SNOWFLAKE + expression: customer_id + dimension: + is_time: false + relationships: + - name: orders_to_customers + from: orders + to: customers + from_columns: [customer_id] + to_columns: [customer_id] +""" + # Convert OSI -> Snowflake + snowflake_yaml = convert_osi_to_snowflake(osi_yaml) + + # Convert Snowflake -> OSI + osi_yaml_round_trip = convert_snowflake_to_osi(snowflake_yaml) + + # Parse both + original = yaml.safe_load(osi_yaml) + round_trip = yaml.safe_load(osi_yaml_round_trip) + + # Check that core structure is preserved + assert round_trip["version"] == "0.2.0.dev0" + original_model = original["semantic_model"][0] + round_trip_model = round_trip["semantic_model"][0] + + # Check model name + assert round_trip_model["name"] == original_model["name"] + + # Check datasets + assert len(round_trip_model["datasets"]) == len(original_model["datasets"]) + original_dataset_names = {d["name"] for d in original_model["datasets"]} + round_trip_dataset_names = {d["name"] for d in round_trip_model["datasets"]} + assert original_dataset_names == round_trip_dataset_names + + # Check relationships + if "relationships" in original_model: + assert "relationships" in round_trip_model + assert len(round_trip_model["relationships"]) == len(original_model["relationships"])