feat(snowflake): opt-in ACCESS_HISTORY lineage path (POC)

[ulixius9]

Summary

• New Snowflake lineage path that reads precomputed table and column lineage from ACCOUNT_USAGE.ACCESS_HISTORY instead of parsing every relevant query client-side. Targets large query windows where the current sqlglot/sqlfluff/sqlparse pipeline becomes a wall-clock bottleneck.
• Opt-in via connectionOptions.useAccessHistory: \"true\" on the Snowflake service connection. Default off — existing pipelines see zero change. A runtime probe demotes silently to the legacy parser path when ACCESS_HISTORY isn't readable (Standard Edition or missing IMPORTED PRIVILEGES ON DATABASE SNOWFLAKE).
• Single combined SQL: MAX_BY dedup on table edges, ARRAY_AGG(DISTINCT OBJECT_CONSTRUCT(...)) on column pairs per (downstream, upstream) edge, LEFT JOIN'd so the cursor streams one row per directed edge with column lineage already attached. Constant client memory regardless of edge count.
• COPY_HISTORY surfaced separately for external stage → Container lineage. External stages (s3://, azure://, gcs://, https://) are resolved against ingested Container entities via es_search_container_by_path; internal Snowflake stages (@~/, @%table/, @db.schema.stage/) are skipped silently.
• Stored-procedure body lineage path (StoredProcedureLineageMixin) is unchanged and continues to run regardless of the flag.
• POC scope: configured via existing connectionOptions Map<String,String> on the Snowflake connection — no JSON schema or generated-model changes. If validated, a follow-up PR will promote the key to a first-class field.

How to enable

Customer pipeline YAML:

serviceConnection:
  config:
    type: Snowflake
    ...
    connectionOptions:
      useAccessHistory: \"true\"
sourceConfig:
  config:
    queryLogDuration: 180  # set as wide as needed; ACCESS_HISTORY retains ~365 days

Files changed

• `ingestion/src/metadata/ingestion/source/database/snowflake/queries.py` — three new SQL constants: `SNOWFLAKE_ACCESS_HISTORY_PROBE`, `SNOWFLAKE_ACCESS_HISTORY_LINEAGE` (combined table+column), `SNOWFLAKE_COPY_HISTORY_LINEAGE`.
• `ingestion/src/metadata/ingestion/source/database/snowflake/connection.py` — `probe_access_history_available(engine, schema)` helper; failures logged at INFO (Standard Edition is a legitimate state).
• `ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py` — `init` reads and pops the option, runs probe, demotes on failure; `yield_query_lineage` dispatches to the new path or falls through to `super()`; new private yielders for ACCESS_HISTORY and COPY_HISTORY with LRU-cached entity resolution; `_parse_column_pairs` decodes the VARIANT array (handles both Python-list and JSON-string forms).
• `ingestion/tests/unit/topology/database/test_snowflake_access_history_lineage.py` — 25 unit tests covering SQL rendering, connectionOptions parsing and key pop, probe-failure demote, table edge yielding, column lineage attachment with single and multiple pairs, COPY edge resolution (resolved / unresolved / internal), legacy-parser-bypass regression, and `_parse_column_pairs` robustness.

Test plan

• `make py_format_check` clean (ruff lint + format)
• 25 new unit tests pass
• 51/51 Snowflake unit tests pass (no regression in existing `test_snowflake.py` or `test_snowflake_ordinal_position.py`)
• Manual run on an Enterprise Snowflake account with `useAccessHistory: "true"` and `queryLogDuration: 180`; record wall time and edge count
• Verify edge count ≥ legacy path on the same window
• Verify column lineage attaches end-to-end (e.g., `orders.amount → revenue.total_amount` on a known graph)
• Verify Standard Edition account: probe demotes silently and behavior is identical to today
• Verify pipeline without `useAccessHistory` set: behavior is bit-identical to today (default off)

Notes

• ACCESS_HISTORY has ~45-minute freshness — daily/weekly cron pipelines unaffected; not suitable for near-real-time lineage.
• `directSources` is NULL for INSERT…VALUES, CALL (stored procs), and certain dynamic SQL. Proc body lineage is preserved via the existing `StoredProcedureLineageMixin` join.
• COPY edges depend on the upstream Container being ingested via a Storage Service ingestion; unresolved external stages are logged at INFO so the operator can decide which storage services to ingest next.
• No incremental checkpointing wired in this POC; each run scans the full configured `queryLogDuration` window.

🤖 Generated with Claude Code

---

Summary by Gitar

• Enhanced lineage metadata:
  • Integrated QUERY_TEXT from ACCOUNT_USAGE.QUERY_HISTORY into the ACCESS_HISTORY lineage path to provide representative SQL context for edges.
  • Updated LineageDetails to include sqlQuery and added corresponding unit tests to verify SQL text attachment.
• Filtering capability:
  • Added _build_filter_condition_clause to inject sourceConfig.filterCondition directly into the ACCESS_HISTORY SQL query.
  • Updated SNOWFLAKE_ACCESS_HISTORY_LINEAGE query to support filtering at the source CTE level.
• Documentation:
  • Removed "POC" labels and references from log messages, docstrings, and unit test headers to reflect readiness.

_{This will update automatically on new commits.}

[gitar-bot]

💡 Edge Case: Quoted Snowflake identifiers silently dropped from lineage

_split_snowflake_fqn (line 439) rejects any FQN containing a double-quote character. Snowflake routinely quotes identifiers that contain spaces, mixed case, or special characters (e.g., "My DB"."My Schema"."My Table"). ACCESS_HISTORY returns quoted identifiers for such objects, so their lineage edges are silently skipped.

While documented as a POC limitation, this could lead to significant lineage gaps for customers with mixed-case or special-character naming conventions, with no visibility into what was missed (only the aggregate skip count is logged).

Strip quotes and split correctly, or at minimum log a DEBUG message per skipped FQN so operators can assess impact:

@staticmethod
def _split_snowflake_fqn(snowflake_fqn: str) -> Optional[Tuple[str, str, str]]:
    if not snowflake_fqn:
        return None
    # Strip surrounding quotes from each part
    parts = snowflake_fqn.split(".")
    # Quoted identifiers may contain dots — reassemble quoted segments
    # For POC: handle the simple case of "DB"."SCHEMA"."TABLE"
    stripped = [p.strip('"') for p in parts]
    if len(stripped) != 3:
        logger.debug(f"Skipping FQN with unexpected part count: {snowflake_fqn}")
        return None
    return stripped[0], stripped[1], stripped[2]

• Apply fix

_{Check the box to apply the fix or reply for a change | Was this helpful? React with 👍 / 👎}

[gitar-bot]

💡 Quality: Using private fqn._build instead of public API

fqn._build is a private helper (underscore-prefixed) that simply joins components with .. The public fqn.build function provides entity-type-aware FQN construction and is the standard across the codebase. Using the private function bypasses any future validation or normalization added to the public API.

_{Was this helpful? React with 👍 / 👎}

[gitar-bot]

Code Review 👍 Approved with suggestions 1 resolved / 3 findings

Introduces an opt-in Snowflake ACCESS_HISTORY lineage path for improved ingestion performance, including enhanced SQL context and filter support. Please address the silent dropping of quoted identifiers in FQN splitting and switch to the public FQN API instead of internal methods.

💡 Edge Case: Quoted Snowflake identifiers silently dropped from lineage

📄 ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:432-444

_split_snowflake_fqn (line 439) rejects any FQN containing a double-quote character. Snowflake routinely quotes identifiers that contain spaces, mixed case, or special characters (e.g., "My DB"."My Schema"."My Table"). ACCESS_HISTORY returns quoted identifiers for such objects, so their lineage edges are silently skipped.

While documented as a POC limitation, this could lead to significant lineage gaps for customers with mixed-case or special-character naming conventions, with no visibility into what was missed (only the aggregate skip count is logged).

Strip quotes and split correctly, or at minimum log a DEBUG message per skipped FQN so operators can assess impact

@staticmethod
def _split_snowflake_fqn(snowflake_fqn: str) -> Optional[Tuple[str, str, str]]:
    if not snowflake_fqn:
        return None
    # Strip surrounding quotes from each part
    parts = snowflake_fqn.split(".")
    # Quoted identifiers may contain dots — reassemble quoted segments
    # For POC: handle the simple case of "DB"."SCHEMA"."TABLE"
    stripped = [p.strip('"') for p in parts]
    if len(stripped) != 3:
        logger.debug(f"Skipping FQN with unexpected part count: {snowflake_fqn}")
        return None
    return stripped[0], stripped[1], stripped[2]

💡 Quality: Using private fqn._build instead of public API

📄 ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:376 📄 ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:406

fqn._build is a private helper (underscore-prefixed) that simply joins components with .. The public fqn.build function provides entity-type-aware FQN construction and is the standard across the codebase. Using the private function bypasses any future validation or normalization added to the public API.

✅ 1 resolved

✅ Bug: Pop of useAccessHistory happens after engine creation — ineffective

📄 ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:96-110
The docstring for _read_access_history_flag says "Popping it ensures the Snowflake driver never sees it in the URL." However, super().__init__() (line 102) creates the engine before _read_access_history_flag() runs (line 105). The engine URL is built via get_connection_url() which reads connectionOptions and appends them as URL query params. By the time .pop() executes, the key has already been baked into the engine's connection URL.

This means useAccessHistory=true is passed to Snowflake as a session parameter on every connection. Snowflake may log warnings or (in future versions) reject unknown session parameters, potentially breaking the connection.

Fix: pop the key before calling super().__init__(), or read/check it without popping (since by that point the engine already exists).

🤖 Prompt for agents

Code Review: Introduces an opt-in Snowflake ACCESS_HISTORY lineage path for improved ingestion performance, including enhanced SQL context and filter support. Please address the silent dropping of quoted identifiers in FQN splitting and switch to the public FQN API instead of internal methods.

1. 💡 Edge Case: Quoted Snowflake identifiers silently dropped from lineage
   Files: ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:432-444

   `_split_snowflake_fqn` (line 439) rejects any FQN containing a double-quote character. Snowflake routinely quotes identifiers that contain spaces, mixed case, or special characters (e.g., `"My DB"."My Schema"."My Table"`). ACCESS_HISTORY returns quoted identifiers for such objects, so their lineage edges are silently skipped.
   
   While documented as a POC limitation, this could lead to significant lineage gaps for customers with mixed-case or special-character naming conventions, with no visibility into what was missed (only the aggregate skip count is logged).

   Fix (Strip quotes and split correctly, or at minimum log a DEBUG message per skipped FQN so operators can assess impact):
   @staticmethod
   def _split_snowflake_fqn(snowflake_fqn: str) -> Optional[Tuple[str, str, str]]:
       if not snowflake_fqn:
           return None
       # Strip surrounding quotes from each part
       parts = snowflake_fqn.split(".")
       # Quoted identifiers may contain dots — reassemble quoted segments
       # For POC: handle the simple case of "DB"."SCHEMA"."TABLE"
       stripped = [p.strip('"') for p in parts]
       if len(stripped) != 3:
           logger.debug(f"Skipping FQN with unexpected part count: {snowflake_fqn}")
           return None
       return stripped[0], stripped[1], stripped[2]

2. 💡 Quality: Using private fqn._build instead of public API
   Files: ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:376, ingestion/src/metadata/ingestion/source/database/snowflake/lineage.py:406

   `fqn._build` is a private helper (underscore-prefixed) that simply joins components with `.`. The public `fqn.build` function provides entity-type-aware FQN construction and is the standard across the codebase. Using the private function bypasses any future validation or normalization added to the public API.

Options

Display: compact → Showing less information.

Comment with these commands to change:

Compact

gitar display:verbose

_{Was this helpful? React with 👍 / 👎 | Gitar}