Feat/snowflake ddl sql import solve #789 and #885 schema.description and properties.description on snowflake and postgresql

CHANGELOG.md

@@ -10,9 +10,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - Escape single quotes in string values for SodaCL checks (#1090)
 - Fixed catalog export SpecView not having a tags property for the index.html template (#1059)
+- Fix SQL importer type mappings: binary types, datetime/time, uuid now map to correct ODCS logicalType and format (#790)
 
 ### Added
 - Support additional PyArrow types in Parquet importer (#1091)
+- Populate `logicalTypeOptions.format` for SQL import from binary and uuid types (#790)
+- Snowflake DDL import with tags, descriptions, and template variable handling (#790)
 
 ## [0.11.6] - 2026-03-17
 

datacontract/imports/dbml_importer.py

@@ -97,7 +97,7 @@ def import_table_fields(table, references) -> List[SchemaProperty]:
         description = field.note.text if field.note else None
         is_primary_key = field.pk
         is_unique = field.unique
-        logical_type = map_type_from_sql(field.type)
+        logical_type, format = map_type_from_sql(field.type)
 
         ref = get_reference(field, references)
 
@@ -109,6 +109,7 @@ def import_table_fields(table, references) -> List[SchemaProperty]:
             name=field_name,
             logical_type=logical_type if logical_type else "string",
             physical_type=field.type,
+            format=format,
             description=description,
             required=required if required else None,
             primary_key=is_primary_key if is_primary_key else None,

datacontract/imports/odcs_helper.py

@@ -34,6 +34,7 @@ def create_schema_object(
     description: str = None,
     business_name: str = None,
     properties: List[SchemaProperty] = None,
+    tags: List[str] = None,
 ) -> SchemaObject:
     """Create a SchemaObject (equivalent to DCS Model)."""
     schema = SchemaObject(
@@ -48,6 +49,8 @@ def create_schema_object(
         schema.businessName = business_name
     if properties:
         schema.properties = properties
+    if tags:
+        schema.tags = tags
     return schema
 
 

datacontract/imports/sql_importer.py

@@ -1,5 +1,6 @@
 import logging
 import os
+import re
 
 import sqlglot
 from open_data_contract_standard.model import OpenDataContractStandard
@@ -18,17 +19,17 @@
 
 class SqlImporter(Importer):
     def import_source(self, source: str, import_args: dict) -> OpenDataContractStandard:
-        return import_sql(self.import_format, source, import_args)
+        return import_sql(source, import_args)
 
 
-def import_sql(format: str, source: str, import_args: dict = None) -> OpenDataContractStandard:
+def import_sql(source: str, import_args: dict = None) -> OpenDataContractStandard:
     sql = read_file(source)
     dialect = to_dialect(import_args)
 
     try:
         parsed = sqlglot.parse_one(sql=sql, read=dialect)
     except Exception as e:
-        logging.error(f"Error parsing SQL: {str(e)}")
+        logging.error(f"Error sqlglot SQL: {str(e)}")
         raise DataContractException(
             type="import",
             name=f"Reading source from {source}",
@@ -57,12 +58,13 @@ def import_sql(format: str, source: str, import_args: dict = None) -> OpenDataCo
 
             col_name = column.this.name
             col_type = to_col_type(column, dialect)
-            logical_type = map_type_from_sql(col_type)
+            logical_type, format = map_type_from_sql(col_type)
             col_description = get_description(column)
             max_length = get_max_length(column)
             precision, scale = get_precision_scale(column)
             is_primary_key = get_primary_key(column)
             is_required = column.find(sqlglot.exp.NotNullColumnConstraint) is not None or None
+            tags = get_tags(column)
 
             prop = create_property(
                 name=col_name,
@@ -72,19 +74,36 @@ def import_sql(format: str, source: str, import_args: dict = None) -> OpenDataCo
                 max_length=max_length,
                 precision=precision,
                 scale=scale,
+                format=format,
                 primary_key=is_primary_key,
                 primary_key_position=primary_key_position if is_primary_key else None,
                 required=is_required if is_required else None,
+                tags=tags,
             )
 
             if is_primary_key:
                 primary_key_position += 1
 
             properties.append(prop)
 
+        table_comment_property = parsed.find(sqlglot.expressions.SchemaCommentProperty)
+
+        table_description = None
+        if table_comment_property:
+            table_description = table_comment_property.this.this
+
+        table_tags = None
+        table_props = parsed.find(sqlglot.expressions.Properties)
+        if table_props:
+            tags = table_props.find(sqlglot.expressions.Tags)
+            if tags:
+                table_tags = [str(t) for t in tags.expressions]
+
         schema_obj = create_schema_object(
             name=table_name,
             physical_type="table",
+            description=table_description,
+            tags=table_tags,
             properties=properties,
         )
         odcs.schema_.append(schema_obj)
@@ -112,27 +131,9 @@ def to_dialect(import_args: dict) -> Dialects | None:
         return Dialects.TSQL
     if dialect.upper() in Dialects.__members__:
         return Dialects[dialect.upper()]
-    if dialect == "sqlserver":
-        return Dialects.TSQL
     return None
 
 
-def to_physical_type_key(dialect: Dialects | str | None) -> str:
-    dialect_map = {
-        Dialects.TSQL: "sqlserverType",
-        Dialects.POSTGRES: "postgresType",
-        Dialects.BIGQUERY: "bigqueryType",
-        Dialects.SNOWFLAKE: "snowflakeType",
-        Dialects.REDSHIFT: "redshiftType",
-        Dialects.ORACLE: "oracleType",
-        Dialects.MYSQL: "mysqlType",
-        Dialects.DATABRICKS: "databricksType",
-    }
-    if isinstance(dialect, str):
-        dialect = Dialects[dialect.upper()] if dialect.upper() in Dialects.__members__ else None
-    return dialect_map.get(dialect, "physicalType")
-
-
 def to_server_type(source, dialect: Dialects | None) -> str | None:
     if dialect is None:
         return None
@@ -166,10 +167,22 @@ def to_col_type_normalized(column):
 
 def get_description(column: sqlglot.expressions.ColumnDef) -> str | None:
     if column.comments is None:
-        return None
+        description = column.find(sqlglot.expressions.CommentColumnConstraint)
+        if description:
+            return description.this.this
+        else:
+            return None
     return " ".join(comment.strip() for comment in column.comments)
 
 
+def get_tags(column: sqlglot.expressions.ColumnDef) -> list[str] | None:
+    tags = column.find(sqlglot.expressions.Tags)
+    if tags:
+        return [str(t) for t in tags.expressions]
+    else:
+        return None
+
+
 def get_max_length(column: sqlglot.expressions.ColumnDef) -> int | None:
     col_type = to_col_type_normalized(column)
     if col_type is None:
@@ -212,79 +225,97 @@ def get_precision_scale(column):
     return None, None
 
 
-def map_type_from_sql(sql_type: str) -> str | None:
-    """Map SQL type to ODCS logical type."""
+def map_type_from_sql(sql_type: str) -> tuple[str, str | None]:
+    """Map SQL type to ODCS logical type and optional format.
+
+    Returns (logicalType, format).
+    The format corresponds to ODCS logicalTypeOptions.format (e.g. "binary", "uuid").
+    """
     if sql_type is None:
-        return None
+        return ("string", None)
 
     sql_type_normed = sql_type.lower().strip()
 
     if sql_type_normed.startswith("varchar"):
-        return "string"
+        return ("string", None)
     elif sql_type_normed.startswith("char"):
-        return "string"
+        return ("string", None)
     elif sql_type_normed.startswith("string"):
-        return "string"
+        return ("string", None)
     elif sql_type_normed.startswith("nchar"):
-        return "string"
+        return ("string", None)
     elif sql_type_normed.startswith("text"):
-        return "string"
+        return ("string", None)
     elif sql_type_normed.startswith("nvarchar"):
-        return "string"
+        return ("string", None)
     elif sql_type_normed.startswith("ntext"):
-        return "string"
-    elif sql_type_normed.startswith("int") and not sql_type_normed.startswith("interval"):
-        return "integer"
-    elif sql_type_normed.startswith("bigint"):
-        return "integer"
-    elif sql_type_normed.startswith("tinyint"):
-        return "integer"
-    elif sql_type_normed.startswith("smallint"):
-        return "integer"
+        return ("string", None)
+    elif sql_type_normed.endswith("int") and not sql_type_normed.endswith("point"):
+        return ("integer", None)
+    elif sql_type_normed.endswith("integer"):
+        return ("integer", None)
     elif sql_type_normed.startswith("float"):
-        return "number"
+        return ("number", None)
     elif sql_type_normed.startswith("double"):
-        return "number"
-    elif sql_type_normed.startswith("decimal"):
-        return "number"
+        return ("number", None)
+    elif sql_type_normed == "real":
+        return ("number", None)
+    elif sql_type_normed.startswith("number"):
+        return ("number", None)
     elif sql_type_normed.startswith("numeric"):
-        return "number"
+        return ("number", None)
+    elif sql_type_normed.startswith("decimal"):
+        return ("number", None)
+    elif sql_type_normed.startswith("money"):
+        return ("number", None)
     elif sql_type_normed.startswith("bool"):
-        return "boolean"
+        return ("boolean", None)
     elif sql_type_normed.startswith("bit"):
-        return "boolean"
+        return ("boolean", None)
     elif sql_type_normed.startswith("binary"):
-        return "array"
+        return ("string", "binary")
     elif sql_type_normed.startswith("varbinary"):
-        return "array"
+        return ("string", "binary")
     elif sql_type_normed.startswith("raw"):
-        return "array"
-    elif sql_type_normed == "blob" or sql_type_normed == "bfile":
-        return "array"
+        return ("string", "binary")
+    elif sql_type_normed == "blob":
+        return ("string", "binary")
+    elif sql_type_normed == "bfile":
+        return ("string", "binary")
+    elif sql_type_normed.startswith("bytea"):
+        return ("string", "binary")
+    elif sql_type_normed == "image":
+        return ("string", "binary")
     elif sql_type_normed == "date":
-        return "date"
+        return ("date", None)
     elif sql_type_normed == "time":
-        return "string"
+        return ("time", None)
     elif sql_type_normed.startswith("timestamp"):
-        return "date"
-    elif sql_type_normed == "datetime" or sql_type_normed == "datetime2":
-        return "date"
+        return ("timestamp", None)
     elif sql_type_normed == "smalldatetime":
-        return "date"
-    elif sql_type_normed == "datetimeoffset":
-        return "date"
+        return ("timestamp", None)
+    elif sql_type_normed.startswith("datetime"):  # tsql datetime2, datetimeoffset
+        return ("timestamp", None)
     elif sql_type_normed == "uniqueidentifier":  # tsql
-        return "string"
+        return ("string", "uuid")
     elif sql_type_normed == "json":
-        return "object"
+        return ("object", None)
     elif sql_type_normed == "xml":  # tsql
-        return "string"
-    elif sql_type_normed.startswith("number"):
-        return "number"
+        return ("string", None)
     elif sql_type_normed == "clob" or sql_type_normed == "nclob":
-        return "string"
+        return ("string", None)
     else:
-        return "object"
+        return ("object", None)
+
+
+def remove_variable_tokens(sql_script: str) -> str:
+    """Replace templating placeholders with bare variable names so sqlglot can parse the SQL."""
+    variable_pattern = re.compile(
+        r"\$\((\w+)\)"  # $(var) — sqlcmd (T-SQL)
+        r"|\$\{(\w+)\}"  # ${var} — Liquibase
+        r"|\{\{(\w+)\}\}"  # {{var}} — Jinja / dbt
+    )
+    return variable_pattern.sub(lambda m: m.group(1) or m.group(2) or m.group(3), sql_script)
 
 
 def read_file(path):
@@ -298,4 +329,5 @@ def read_file(path):
         )
     with open(path, "r") as file:
         file_content = file.read()
-    return file_content
+
+    return remove_variable_tokens(file_content)

datacontract/imports/unity_importer.py

@@ -152,14 +152,15 @@ def import_table_fields(columns: List[ColumnInfo]) -> List[SchemaProperty]:
 def _to_property(column: ColumnInfo) -> SchemaProperty:
     """Convert a Unity ColumnInfo to an ODCS SchemaProperty."""
     sql_type = str(column.type_text) if column.type_text else "string"
-    logical_type = map_type_from_sql(sql_type)
+    logical_type, format = map_type_from_sql(sql_type)
     required = column.nullable is None or not column.nullable
 
     return create_property(
         name=column.name,
         logical_type=logical_type if logical_type else "string",
         physical_type=sql_type,
         description=column.comment,
+        format=format,
         required=required if required else None,
         custom_properties={"databricksType": sql_type} if sql_type else None,
     )

tests/fixtures/databricks-unity/import/datacontract.yaml

@@ -53,7 +53,7 @@ schema:
     customProperties:
     - property: databricksType
       value: timestamp
-    logicalType: date
+    logicalType: timestamp
   - name: is_active
     physicalType: boolean
     customProperties:

tests/fixtures/dbml/import/datacontract.yaml

@@ -26,7 +26,7 @@ schema:
     physicalType: timestamp
     description: The business timestamp in UTC when the order was successfully registered
       in the source system and the payment was successful.
-    logicalType: date
+    logicalType: timestamp
     required: true
   - name: order_total
     physicalType: record
@@ -46,7 +46,7 @@ schema:
   - name: processed_timestamp
     physicalType: timestamp
     description: The timestamp when the record was processed by the data platform.
-    logicalType: date
+    logicalType: timestamp
     required: true
 - name: line_items
   physicalType: table

tests/fixtures/dbml/import/datacontract_table_filtered.yaml

@@ -26,7 +26,7 @@ schema:
     physicalType: timestamp
     description: The business timestamp in UTC when the order was successfully registered
       in the source system and the payment was successful.
-    logicalType: date
+    logicalType: timestamp
     required: true
   - name: order_total
     physicalType: record
@@ -46,5 +46,5 @@ schema:
   - name: processed_timestamp
     physicalType: timestamp
     description: The timestamp when the record was processed by the data platform.
-    logicalType: date
+    logicalType: timestamp
     required: true