Merge pull request #6 from thisisqubika/DC-308-phase-1-metadata-extraction-validation

Dc 308 phase 1 metadata extraction validation

Author: ssolarteserna

pyproject.toml

@@ -22,4 +22,6 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry.scripts]
 migration-accelerator = "migration_accelerator_package.main:main"
 snowpark-reader = "migration_accelerator_package.snowpark:main"
+snowflake-validator = "migration_accelerator_package.ingestion_validation:main"
+
 

resources/jobs.yml

@@ -14,6 +14,15 @@ resources:
             entry_point: snowpark-reader
             package_name: migration_accelerator_package
           environment_key: "serverless_wheel"
+        
+        - task_key: snowflake_ingestion_validation_task
+          depends_on:
+            - task_key: snowflake_ingestion_task
+          python_wheel_task:
+            entry_point: snowflake-validator
+            package_name: migration_accelerator_package
+          environment_key: "serverless_wheel"
+              
 
       environments:
         - environment_key: "serverless_wheel"

src/migration_accelerator_package/artifact_validators.py

@@ -0,0 +1,191 @@
+"""
+Artifact Validator Class 
+Provides a clean interface for validating completeness and correctness of the ingested types of Snowflake artifacts.
+"""
+import json
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any
+from snowflake.snowpark import Session
+from migration_accelerator_package.constants import ArtifactType, ArtifactFileName
+from databricks.sdk.runtime import *
+
+def normalize_column(col: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Normalize column metadata for comparison.
+    Only keeps the essential fields needed for correctness validation.
+    """
+    col = {k.lower(): v for k, v in col.items()}
+
+    return {
+        "column_name": col.get("column_name"),
+        "data_type": col.get("data_type"),
+        "is_nullable": col.get("is_nullable"),       
+    }
+
+
+class MetadataValidator:
+    """
+    Validates completeness and correctness of extracted Snowflake metadata.
+    """
+    def __init__(self, session: Session, volume_path: str):
+        self.session = session
+        self.volume_path = volume_path
+
+    def _load_extracted(self, filename: str) -> Dict[str, Any]:
+        path = f"{self.volume_path}/{filename}"
+        raw = dbutils.fs.head(path, 50_000_000)
+        return json.loads(raw)
+
+
+
+    def load_all_artifacts(self) -> Dict[str, Dict[str, Any]]:
+        extracted = {}
+        for artifact_type in ArtifactType:
+            file_enum = ArtifactFileName[artifact_type.name]
+            filename = file_enum.value
+            extracted[artifact_type.value] = self._load_extracted(filename)
+        return extracted
+
+    def count_snowflake_artifacts(self, artifact_type: ArtifactType, db: str, schema: str) -> int:
+        if artifact_type == ArtifactType.TABLES:
+            query = f"""
+                SELECT COUNT(*) FROM {db}.information_schema.tables
+                WHERE table_schema = '{schema}'
+                AND table_type = 'BASE TABLE'
+            """
+        elif artifact_type == ArtifactType.VIEWS:
+            query = f"""
+                SELECT COUNT(*)
+                FROM {db}.information_schema.views
+                WHERE table_schema = '{schema}'
+            """
+        elif artifact_type == ArtifactType.PROCEDURES:
+            query = f"""
+                SELECT COUNT(*)
+                FROM {db}.information_schema.procedures
+                WHERE procedure_schema = '{schema}'
+            """
+        elif artifact_type == ArtifactType.FUNCTIONS:
+            query = f"""
+                SELECT COUNT(*)
+                FROM {db}.information_schema.functions
+                WHERE function_schema = '{schema}'
+            """
+        elif artifact_type == ArtifactType.SEQUENCES:
+            query = f"""
+                SELECT COUNT(*)
+                FROM {db}.information_schema.sequences
+                WHERE sequence_schema = '{schema}'
+            """
+        elif artifact_type == ArtifactType.STAGES:
+            query = f"SHOW STAGES IN SCHEMA {db}.{schema}"
+            return len(self.session.sql(query).collect())
+        elif artifact_type == ArtifactType.FILE_FORMATS:
+            query = f"SHOW FILE FORMATS IN SCHEMA {db}.{schema}"
+            return len(self.session.sql(query).collect())
+        elif artifact_type == ArtifactType.TASKS:
+            query = f"SHOW TASKS IN SCHEMA {db}.{schema}"
+            return len(self.session.sql(query).collect())
+        elif artifact_type == ArtifactType.STREAMS:
+            query = f"SHOW STREAMS IN SCHEMA {db}.{schema}"
+            return len(self.session.sql(query).collect())
+        elif artifact_type == ArtifactType.PIPES:
+            query = f"SHOW PIPES IN SCHEMA {db}.{schema}"
+            return len(self.session.sql(query).collect())
+        else:
+            return 0
+
+        return self.session.sql(query).collect()[0][0]
+    
+    def validate_completeness(self, extracted: Dict[str, Any], db: str, schema: str):
+        completeness = {}
+
+        for artifact_type in ArtifactType:
+            snowflake_count = self.count_snowflake_artifacts(artifact_type, db, schema)
+            extracted_count = len(extracted[artifact_type.value][artifact_type.value])
+
+            coverage = (extracted_count / snowflake_count) if snowflake_count > 0 else 1.0
+
+            completeness[artifact_type.value] = {
+                "snowflake": snowflake_count,
+                "extracted": extracted_count,
+                "coverage_pct": round(coverage * 100, 2),
+                "perfect_match": extracted_count == snowflake_count
+            }
+
+        return completeness
+
+    def validate_table_definition(self, db, schema, extracted_table: Dict[str, Any]) -> Dict[str, Any]:
+        table_name = extracted_table["table_name"]
+
+        query = f"""
+        SELECT column_name, data_type, is_nullable,
+            character_maximum_length, numeric_precision, numeric_scale,
+            column_default, comment
+        FROM {db}.information_schema.columns
+        WHERE table_schema = '{schema}'
+        AND table_name = '{table_name}'
+        ORDER BY ordinal_position
+        """
+
+        # Normalize Snowflake columns
+        sf_columns_raw = [dict(row.as_dict()) for row in self.session.sql(query).collect()]
+        sf_columns = [normalize_column(col) for col in sf_columns_raw]
+
+        # Normalize extracted columns
+        extracted_columns_raw = extracted_table.get("columns", [])
+        extracted_columns = [normalize_column(col) for col in extracted_columns_raw]
+
+        # Compute correctness statistics
+        total = max(len(sf_columns), len(extracted_columns), 1)
+        matches = sum(1 for sf, ex in zip(sf_columns, extracted_columns) if sf == ex)
+        correctness_pct = round((matches / total) * 100, 2)
+
+        return {
+            "table": table_name,
+            "snowflake_column_count": len(sf_columns),
+            "extracted_column_count": len(extracted_columns),
+            "matches": matches,
+            "total_columns": total,
+            "correctness_pct": correctness_pct,
+            "columns_match_exactly": sf_columns == extracted_columns,
+            "snowflake": sf_columns,
+            "extracted": extracted_columns
+        }
+
+
+
+    def validate_view_definition(self, db, schema, extracted_view: Dict[str, Any]) -> Dict[str, Any]:
+        view_name = extracted_view["view_name"]
+
+        query = f"""
+        SELECT view_definition
+        FROM {db}.information_schema.views
+        WHERE table_schema = '{schema}'
+        AND table_name = '{view_name}'
+        """
+        result = self.session.sql(query).collect()
+        sf_def = result[0]["VIEW_DEFINITION"] if result else ""
+
+        extracted_def = extracted_view.get("view_definition", "")
+
+        # Normalize whitespace for fair comparison
+        def normalize(s):
+            return " ".join(s.lower().strip().split())
+
+        sf_norm = normalize(sf_def)
+        ex_norm = normalize(extracted_def)
+
+        match = sf_norm == ex_norm
+
+        return {
+            "view": view_name,
+            "match": match,
+            "snowflake_definition": sf_def,
+            "extracted_definition": extracted_def
+        }
+
+
+
+    
+

src/migration_accelerator_package/ingestion_validation.py

@@ -0,0 +1,79 @@
+"""
+Entry point for Snowflake metadata validation.
+Runs as a Databricks wheel task using 'snowflake-validator'.
+"""
+
+import json
+from snowflake.snowpark import Session
+from databricks.sdk.runtime import dbutils
+
+from migration_accelerator_package.snowpark_utils import (
+    build_snowflake_connection_params,
+    get_uc_volume_path,
+)
+
+from migration_accelerator_package.artifact_validators import MetadataValidator
+from migration_accelerator_package.constants import SnowflakeConfig
+
+
+def main():
+    print("=" * 80)
+    print(" SNOWFLAKE METADATA VALIDATION ")
+    print("=" * 80)
+
+    connection_parameters = build_snowflake_connection_params()
+    session = Session.builder.configs(connection_parameters).create()
+
+    db = SnowflakeConfig.SNOWFLAKE_DATABASE.value
+    schema = SnowflakeConfig.SNOWFLAKE_SCHEMA.value
+
+    volume_path = get_uc_volume_path()
+    print(f"UC Volume Path: {volume_path}")
+
+
+    validator = MetadataValidator(session, volume_path)
+
+    print("Loading extracted metadata...")
+    extracted = validator.load_all_artifacts()
+    print("✓ Loaded all JSON files")
+
+    print("Running completeness validation...")
+    completeness_report = validator.validate_completeness(extracted, db, schema)
+    print("✓ Completeness check done")
+
+    print("Running correctness checks...")
+
+    sample_tables = extracted["tables"]["tables"][:5]
+    table_results = [
+        validator.validate_table_definition(db, schema, t)
+        for t in sample_tables
+    ]
+
+    sample_views = extracted["views"]["views"][:5]
+    view_results = [
+        validator.validate_view_definition(db, schema, v)
+        for v in sample_views
+    ]
+
+    report = {
+        "database": db,
+        "schema": schema,
+        "completeness": completeness_report,
+        "correctness": {
+            "tables": table_results,
+            "views": view_results,
+        }
+    }
+
+    print("✓ Validation complete")
+    print(json.dumps(report, indent=2))
+
+    output_path = f"{volume_path}/validation_report.json"
+    dbutils.fs.put(output_path, json.dumps(report, indent=2), overwrite=True)
+    print(f"✓ Validation report saved to {output_path}")
+
+    session.close()
+
+
+if __name__ == "__main__":
+    main()

src/migration_accelerator_package/snowpark_utils.py

@@ -0,0 +1,38 @@
+"""
+Utility functions shared across ingestion + validation entrypoints.
+"""
+
+import os
+from databricks.sdk.runtime import dbutils
+from migration_accelerator_package.constants import SnowflakeConfig, UnityCatalogConfig
+
+
+def get_secret(secret_name: str):
+    """Retrieve secrets from Databricks secret scope or fallback to env variables."""
+    try:
+        return dbutils.secrets.get("migration-accelerator", secret_name)
+    except Exception:
+        return os.getenv(secret_name, "")
+
+
+def build_snowflake_connection_params():
+    """Return Snowflake connection parameters used by all wheel entrypoints."""
+    return {
+        "account": get_secret("SNOWFLAKE_ACCOUNT"),
+        "user": get_secret("SNOWFLAKE_USER"),
+        "password": get_secret("SNOWFLAKE_PASSWORD"),
+        "role": SnowflakeConfig.SNOWFLAKE_ROLE.value,
+        "warehouse": SnowflakeConfig.SNOWFLAKE_WAREHOUSE.value,
+        "database": SnowflakeConfig.SNOWFLAKE_DATABASE.value,
+        "schema": SnowflakeConfig.SNOWFLAKE_SCHEMA.value,
+    }
+
+
+def get_uc_volume_path() -> str:
+    """Return the base UC volume path where JSON artifacts live."""
+    return (
+        f"/Volumes/"
+        f"{UnityCatalogConfig.CATALOG.value}/"
+        f"{UnityCatalogConfig.SCHEMA.value}/"
+        f"{UnityCatalogConfig.RAW_VOLUME.value}"
+    )