refactor: remove sqlglot dependency, migrate all SQL methods to sqlguard (Rust)

- Rewire 6 `server.py` dispatch branches (`sql.analyze`, `sql.translate`,
  `sql.optimize`, `sql.format`, `sql.fix`, `sql.rewrite`) to call `guard_*`
  functions directly instead of `sqlglot`-based Python modules
- Migrate `feedback_store.py` from `sqlglot` AST to regex-based fingerprinting
  and `guard_extract_metadata()`/`guard_complexity_score()` for cost prediction
- Update `cost_gate.py` to use `guard_lint()` instead of `analyze_sql()`
- Update `test_local.py` to use `guard_transpile()` instead of `translate_sql()`
- Delete 8 source files (~2,700 lines): analyzer, translator, formatter,
  optimizer, rewriter, fixer, schema_diff, confidence
- Delete 7 corresponding test files
- Remove `sqlglot>=25.0` from `pyproject.toml` dependencies
- Update tests to match `sqlguard` output format (`findings` key, `fixed` key,
  `total` complexity score)
- All 676 tests pass

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: kulvirgit

packages/altimate-engine/pyproject.toml

@@ -9,7 +9,6 @@ description = "Python engine for Altimate CLI - lineage, SQL execution, dbt inte
 requires-python = ">=3.10"
 dependencies = [
     "pydantic>=2.0",
-    "sqlglot>=25.0",
     "pyyaml>=6.0",
 ]
 

packages/altimate-engine/src/altimate_engine/ci/cost_gate.py

@@ -1,11 +1,11 @@
 """CI cost gate — scan changed SQL files for critical issues.
 
-Reads SQL files, runs existing analyze_sql(), and returns
+Reads SQL files, runs lint analysis, and returns
 pass/fail based on whether CRITICAL severity issues are found.
 
 Skips:
   - Jinja templates ({{ }}, {% %})
-  - sqlglot parse errors (likely Jinja or non-standard SQL)
+  - Parse errors (likely Jinja or non-standard SQL)
   - Non-SQL files
 """
 
@@ -15,7 +15,7 @@
 import re
 from typing import Any
 
-from altimate_engine.sql.analyzer import analyze_sql
+from altimate_engine.sql.guard import guard_lint
 
 
 # Jinja pattern: {{ ... }} or {% ... %} or {# ... #}
@@ -119,19 +119,19 @@ def scan_files(
         file_issues: list[dict[str, Any]] = []
 
         for stmt in statements:
-            # Run analyzer
-            analysis = analyze_sql(stmt, dialect)
-            if not analysis.get("success", True):
+            # Run lint analysis
+            lint_result = guard_lint(stmt)
+            if lint_result.get("error"):
                 # Parse error — skip this statement (likely incomplete SQL)
                 continue
 
-            for issue in analysis.get("issues", []):
-                severity = issue.get("severity", "warning")
+            for finding in lint_result.get("findings", lint_result.get("issues", [])):
+                severity = finding.get("severity", "warning")
                 file_issues.append({
-                    "type": issue.get("type", "UNKNOWN"),
+                    "type": finding.get("rule", finding.get("type", "UNKNOWN")),
                     "severity": severity,
-                    "message": issue.get("message", ""),
-                    "source": "analyzer",
+                    "message": finding.get("message", ""),
+                    "source": "lint",
                 })
                 total_issues += 1
                 if severity in ("error", "critical"):

packages/altimate-engine/src/altimate_engine/local/test_local.py

@@ -12,7 +12,7 @@ def test_sql_local(
 ) -> dict[str, Any]:
     """Execute SQL against a local DuckDB database for validation.
 
-    If target_dialect differs from DuckDB, auto-transpiles via sqlglot first.
+    If target_dialect differs from DuckDB, auto-transpiles first.
 
     Args:
         sql: The SQL to test.
@@ -37,11 +37,12 @@ def test_sql_local(
 
     if target_dialect and target_dialect.lower() not in ("duckdb", "duck"):
         try:
-            from altimate_engine.sql.translator import translate_sql
+            from altimate_engine.sql.guard import guard_transpile
 
-            result = translate_sql(sql, target_dialect, "duckdb")
-            if result.get("success") and result.get("translated_sql"):
-                test_sql = result["translated_sql"]
+            result = guard_transpile(sql, target_dialect, "duckdb")
+            translated = result.get("sql", result.get("translated_sql"))
+            if result.get("success") and translated:
+                test_sql = translated
                 transpiled = True
                 transpile_warnings = result.get("warnings", [])
         except Exception as e:

packages/altimate-engine/src/altimate_engine/server.py

@@ -67,6 +67,7 @@
     SqlFormatResult,
     SqlOptimizeParams,
     SqlOptimizeResult,
+    SqlOptimizeSuggestion,
     SqlPredictCostParams,
     SqlPredictCostResult,
     SqlRecordFeedbackParams,
@@ -114,16 +115,9 @@
     SqlGuardResult,
 )
 from altimate_engine.sql.executor import execute_sql
-from altimate_engine.sql.analyzer import analyze_sql
-from altimate_engine.sql.optimizer import optimize_sql
-from altimate_engine.sql.translator import translate_sql
-from altimate_engine.sql.formatter import format_sql
 from altimate_engine.sql.explainer import explain_sql
-from altimate_engine.sql.fixer import fix_sql
 from altimate_engine.sql.autocomplete import autocomplete_sql
 from altimate_engine.sql.diff import diff_sql
-from altimate_engine.sql.rewriter import rewrite_sql
-from altimate_engine.sql.schema_diff import diff_schema
 from altimate_engine.ci.cost_gate import scan_files
 from altimate_engine.schema.inspector import inspect_schema
 from altimate_engine.schema.pii_detector import detect_pii
@@ -132,7 +126,7 @@
 from altimate_engine.dbt.manifest import parse_manifest
 from altimate_engine.dbt.lineage import dbt_lineage
 from altimate_engine.connections import ConnectionRegistry
-# lineage.check now delegates to sqlguard via guard_column_lineage
+# lineage.check delegates to guard_column_lineage
 from altimate_engine.sql.feedback_store import FeedbackStore
 from altimate_engine.schema.cache import SchemaCache
 from altimate_engine.finops.query_history import get_query_history
@@ -317,21 +311,23 @@ def dispatch(request: JsonRpcRequest) -> JsonRpcResponse:
             result = inspect_schema(SchemaInspectParams(**params))
         elif method == "sql.analyze":
             params_obj = SqlAnalyzeParams(**params)
-            raw_result = analyze_sql(params_obj.sql, params_obj.dialect or "snowflake")
-            # Convert raw dict result to SqlAnalyzeResult model
             issues = []
-            for issue in raw_result.get("issues", []):
+
+            # Anti-pattern detection via sqlguard lint
+            lint_result = guard_lint(params_obj.sql, schema_context=params_obj.schema_context)
+            for issue in lint_result.get("findings", lint_result.get("issues", [])):
                 issues.append(
                     SqlAnalyzeIssue(
-                        type=issue["type"],
+                        type=issue.get("rule", issue.get("type", "LINT")),
                         severity=issue.get("severity", "warning"),
-                        message=issue["message"],
-                        recommendation=issue.get("recommendation", ""),
+                        message=issue.get("message", ""),
+                        recommendation=issue.get("suggestion", issue.get("recommendation", "")),
                         location=issue.get("location"),
                         confidence=issue.get("confidence", "high"),
                     )
                 )
-            # Append sqlguard semantic checks if available
+
+            # Semantic checks
             sem_result = guard_check_semantics(
                 params_obj.sql,
                 schema_context=params_obj.schema_context,
@@ -348,7 +344,8 @@ def dispatch(request: JsonRpcRequest) -> JsonRpcResponse:
                             confidence=si.get("confidence", "medium"),
                         )
                     )
-            # Append sqlguard safety scan if available
+
+            # Safety scan
             safety_result = guard_scan_safety(params_obj.sql)
             if safety_result.get("threats"):
                 for threat in safety_result["threats"]:
@@ -362,26 +359,60 @@ def dispatch(request: JsonRpcRequest) -> JsonRpcResponse:
                             confidence="high",
                         )
                     )
+
             result = SqlAnalyzeResult(
-                success=raw_result.get("success", True),
+                success=not lint_result.get("error"),
                 issues=issues,
                 issue_count=len(issues),
                 confidence=_compute_overall_confidence(issues),
-                confidence_factors=_get_confidence_factors(raw_result),
-                error=raw_result.get("error"),
+                confidence_factors=[] if not lint_result.get("error") else ["Lint parse failed"],
+                error=lint_result.get("error"),
             )
         elif method == "sql.translate":
             params_obj = SqlTranslateParams(**params)
-            result_dict = translate_sql(
-                params_obj.sql, params_obj.source_dialect, params_obj.target_dialect
+            raw = guard_transpile(params_obj.sql, params_obj.source_dialect, params_obj.target_dialect)
+            result = SqlTranslateResult(
+                success=raw.get("success", True),
+                translated_sql=raw.get("sql", raw.get("translated_sql")),
+                source_dialect=params_obj.source_dialect,
+                target_dialect=params_obj.target_dialect,
+                warnings=raw.get("warnings", []),
+                error=raw.get("error"),
             )
-            result = SqlTranslateResult(**result_dict)
         elif method == "sql.optimize":
             params_obj = SqlOptimizeParams(**params)
-            result_dict = optimize_sql(
-                params_obj.sql, params_obj.dialect, params_obj.schema_context
+            # Rewrite for optimization
+            rw = guard_rewrite_sql(params_obj.sql, schema_context=params_obj.schema_context)
+            # Lint for remaining issues
+            lint = guard_lint(params_obj.sql, schema_context=params_obj.schema_context)
+
+            suggestions = []
+            for r in rw.get("rewrites", []):
+                suggestions.append(
+                    SqlOptimizeSuggestion(
+                        type="REWRITE",
+                        description=r.get("explanation", "Optimization rewrite"),
+                        before=r.get("original_fragment", ""),
+                        after=r.get("rewritten_fragment", ""),
+                    )
+                )
+
+            anti_patterns = []
+            for issue in lint.get("findings", lint.get("issues", [])):
+                anti_patterns.append({
+                    "type": issue.get("rule", issue.get("type", "LINT")),
+                    "message": issue.get("message", ""),
+                    "suggestion": issue.get("suggestion", ""),
+                })
+
+            result = SqlOptimizeResult(
+                success=True,
+                original_sql=params_obj.sql,
+                optimized_sql=rw.get("rewritten_sql", params_obj.sql),
+                suggestions=suggestions,
+                anti_patterns=anti_patterns,
+                error=rw.get("error"),
             )
-            result = SqlOptimizeResult(**result_dict)
         elif method == "lineage.check":
             p = LineageCheckParams(**params)
             raw = guard_column_lineage(
@@ -457,46 +488,45 @@ def dispatch(request: JsonRpcRequest) -> JsonRpcResponse:
             result = SqlPredictCostResult(**prediction)
         elif method == "sql.format":
             fmt_params = SqlFormatParams(**params)
-            fmt_result = format_sql(
-                fmt_params.sql, fmt_params.dialect, fmt_params.indent
+            raw = guard_format_sql(fmt_params.sql, fmt_params.dialect)
+            result = SqlFormatResult(
+                success=raw.get("success", True),
+                formatted_sql=raw.get("formatted_sql", raw.get("sql")),
+                statement_count=raw.get("statement_count", 1),
+                error=raw.get("error"),
             )
-            result = SqlFormatResult(**fmt_result)
         elif method == "sql.explain":
             result = explain_sql(SqlExplainParams(**params))
         elif method == "sql.fix":
             fix_params = SqlFixParams(**params)
-            # Try sqlguard.fix first for Rust-powered fixing
             guard_result = guard_fix_sql(fix_params.sql)
-            if guard_result.get("success") and guard_result.get("fixed_sql"):
+            fixed = guard_result.get("fixed", guard_result.get("success", False))
+            fixed_sql = guard_result.get("fixed_sql")
+            if fixed and fixed_sql:
                 result = SqlFixResult(
                     success=True,
                     original_sql=fix_params.sql,
-                    fixed_sql=guard_result["fixed_sql"],
+                    fixed_sql=fixed_sql,
                     error_message=fix_params.error_message,
                     suggestions=[
                         SqlFixSuggestion(
                             type="SQLGUARD_FIX",
-                            message="Auto-fixed by sqlguard Rust engine",
+                            message="Auto-fixed by sqlguard engine",
                             confidence="high",
-                            fixed_sql=guard_result["fixed_sql"],
+                            fixed_sql=fixed_sql,
                         )
                     ],
                     suggestion_count=1,
                 )
             else:
-                # Fall back to sqlglot-based fixer
-                fix_result = fix_sql(
-                    fix_params.sql, fix_params.error_message, fix_params.dialect
-                )
                 result = SqlFixResult(
-                    success=fix_result["success"],
-                    original_sql=fix_result["original_sql"],
-                    fixed_sql=fix_result.get("fixed_sql"),
-                    error_message=fix_result["error_message"],
-                    suggestions=[
-                        SqlFixSuggestion(**s) for s in fix_result["suggestions"]
-                    ],
-                    suggestion_count=fix_result["suggestion_count"],
+                    success=False,
+                    original_sql=fix_params.sql,
+                    fixed_sql=fixed_sql,
+                    error_message=fix_params.error_message,
+                    suggestions=[],
+                    suggestion_count=0,
+                    error=guard_result.get("error", "Unable to auto-fix"),
                 )
         elif method == "sql.autocomplete":
             ac_params = SqlAutocompleteParams(**params)
@@ -627,7 +657,6 @@ def dispatch(request: JsonRpcRequest) -> JsonRpcResponse:
         # --- SQL rewrite ---
         elif method == "sql.rewrite":
             p = SqlRewriteParams(**params)
-            # Try sqlguard.rewrite first for Rust-powered rewriting
             guard_rw = guard_rewrite_sql(p.sql, schema_context=p.schema_context)
             if guard_rw.get("success") and guard_rw.get("rewritten_sql"):
                 rewrites = []
@@ -648,17 +677,12 @@ def dispatch(request: JsonRpcRequest) -> JsonRpcResponse:
                     rewrites_applied=rewrites,
                 )
             else:
-                # Fall back to sqlglot-based rewriter
-                raw = rewrite_sql(p.sql, p.dialect, p.schema_context)
                 result = SqlRewriteResult(
-                    success=raw["success"],
-                    original_sql=raw["original_sql"],
-                    rewritten_sql=raw.get("rewritten_sql"),
-                    rewrites_applied=[
-                        SqlRewriteRule(**r)
-                        for r in raw.get("rewrites_applied", [])
-                    ],
-                    error=raw.get("error"),
+                    success=False,
+                    original_sql=p.sql,
+                    rewritten_sql=None,
+                    rewrites_applied=[],
+                    error=guard_rw.get("error", "No rewrites applicable"),
                 )
         # --- sqlguard ---
         elif method == "sqlguard.validate":