fix(snowflake): SHOW ... LIKE instead of INFORMATION_SCHEMA for least-privilege roles

Eric's testing surfaced that least-privilege Snowflake roles (e.g.
DAGSTER_RUNNER with USAGE on the database but not full ACCOUNTADMIN)
can hit "does not exist or not authorized" errors on
INFORMATION_SCHEMA views even when SHOW commands work fine on the
same objects. This is a documented Snowflake quirk affecting new-feature
INFORMATION_SCHEMA views specifically.

Fix: in the materialize-time metadata-read blocks (after the actual
work — REFRESH / EXECUTE / ALTER — has succeeded), swap
INFORMATION_SCHEMA queries to SHOW ... LIKE '<name>' IN SCHEMA
<db>.<schema>. Wrap each in try/except so the action still wins even
if the SHOW also fails.

Updated in snowflake_workspace/component.py:
  - _dynamic_table_asset: INFORMATION_SCHEMA.DYNAMIC_TABLES → SHOW
    DYNAMIC TABLES LIKE (refresh_mode, scheduling_state, target_lag,
    rows, bytes)
  - _task_asset: added SHOW TASKS LIKE (state, schedule, warehouse,
    owner) + TASK_HISTORY wrapped in try/except
  - _snowpipe_asset: added SHOW PIPES LIKE (owner, notification_channel,
    definition) + SYSTEM$PIPE_STATUS wrapped + COPY_HISTORY wrapped
  - _mv_asset: INFORMATION_SCHEMA.TABLES → SHOW MATERIALIZED VIEWS LIKE
    (rows, bytes, cluster_by, behind_by, invalid, owner)
  - _external_table_asset: INFORMATION_SCHEMA.TABLES → SHOW EXTERNAL
    TABLES LIKE (rows, location, file_format_name, last_refreshed_on)
  - _alert_asset: added SHOW ALERTS LIKE (state, schedule) +
    ALERT_HISTORY wrapped in try/except
  - observation sensor: DYNAMIC_TABLES → SHOW DYNAMIC TABLES LIKE

Updated in Wave 3 single-entity components:
  - snowflake_dynamic_table_refresh_asset: same DYNAMIC_TABLES → SHOW
    pattern
  - snowflake_task_execute_asset: added SHOW TASKS LIKE + TASK_HISTORY
    wrapped in try/except
  - snowflake_task_completion_sensor: improved skip_reason error
    message to point at MONITOR grant as the likely fix

Rationale per Eric's diagnosis: "SHOW only requires USAGE on the
schema + any privilege on the DT, which is what least-privilege roles
typically have. INFORMATION_SCHEMA.DYNAMIC_TABLES, by contrast, can
be invisible to scoped roles on some Snowflake accounts (account-level
policy) even with USAGE on the database — Snowflake reports the view
as 'does not exist or not authorized'."

The action (REFRESH / EXECUTE / ALTER) is what the customer actually
wants done; metadata enrichment is best-effort. Even if both SHOW and
INFORMATION_SCHEMA reads fail, the asset still materializes with the
basic metadata (name, database, schema) and a clear log warning about
the missing enrichment.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: eric-thomas-dagster

assets/data-warehouse/snowflake_dynamic_table_refresh_asset/component.py

@@ -120,34 +120,41 @@ def snowflake_dynamic_table_refresh_asset(context: AssetExecutionContext) -> Mat
                 context.log.info(f"ALTER DYNAMIC TABLE {fqn} REFRESH")
                 cursor.execute(f"ALTER DYNAMIC TABLE {fqn} REFRESH")
 
-                # Pull current state from INFORMATION_SCHEMA.DYNAMIC_TABLES.
-                cursor.execute(f"""
-                    SELECT name, scheduling_state, last_refresh_status,
-                           target_lag, refresh_mode,
-                           last_successful_refresh_time
-                    FROM {_self.database}.INFORMATION_SCHEMA.DYNAMIC_TABLES
-                    WHERE schema_name = '{_self.schema_name}'
-                      AND name = '{_self.dynamic_table_name}'
-                """)
-                row = cursor.fetchone()
-
                 metadata: Dict[str, Any] = {
                     "dynamic_table_fqn": fqn,
                     "dynamic_table_name": _self.dynamic_table_name,
                     "database": _self.database,
                     "schema": _self.schema_name,
                 }
-                if row:
-                    columns = [c[0] for c in cursor.description]
-                    rd = dict(zip(columns, row))
-                    metadata.update({
-                        "scheduling_state": rd.get("SCHEDULING_STATE"),
-                        "last_refresh_status": rd.get("LAST_REFRESH_STATUS"),
-                        "target_lag": rd.get("TARGET_LAG"),
-                        "refresh_mode": rd.get("REFRESH_MODE"),
-                        "last_successful_refresh_time": str(rd.get("LAST_SUCCESSFUL_REFRESH_TIME"))
-                            if rd.get("LAST_SUCCESSFUL_REFRESH_TIME") else None,
-                    })
+
+                # SHOW DYNAMIC TABLES (not INFORMATION_SCHEMA.DYNAMIC_TABLES) —
+                # INFORMATION_SCHEMA can be invisible to least-privilege roles
+                # (e.g. DAGSTER_RUNNER) even with USAGE on the database. SHOW
+                # only requires USAGE on the schema + any privilege on the DT.
+                # Wrap in try/except so refresh still wins even if SHOW fails.
+                try:
+                    cursor.execute(
+                        f"SHOW DYNAMIC TABLES LIKE '{_self.dynamic_table_name}' "
+                        f"IN SCHEMA {_self.database}.{_self.schema_name}"
+                    )
+                    row = cursor.fetchone()
+                    if row:
+                        columns = [c[0].lower() for c in cursor.description]
+                        rd = dict(zip(columns, row))
+                        metadata.update({
+                            "scheduling_state": rd.get("scheduling_state"),
+                            "last_refresh_status": rd.get("last_refresh_state"),
+                            "target_lag": rd.get("target_lag"),
+                            "refresh_mode": rd.get("refresh_mode"),
+                            "rows": rd.get("rows"),
+                            "bytes": rd.get("bytes"),
+                            "owner": rd.get("owner"),
+                        })
+                except Exception as exc:
+                    context.log.warning(
+                        f"Could not read DT metadata for {_self.dynamic_table_name}: {exc}. "
+                        f"Refresh succeeded; emitting asset without enriched metadata."
+                    )
                 return MaterializeResult(metadata=metadata)
             finally:
                 cursor.close()

assets/data-warehouse/snowflake_task_execute_asset/component.py

@@ -123,38 +123,67 @@ def snowflake_task_execute_asset(context: AssetExecutionContext) -> MaterializeR
                 context.log.info(f"EXECUTE TASK {fqn}")
                 cursor.execute(f"EXECUTE TASK {fqn}")
 
-                # Pull the most recent history entry so the materialization
-                # has the run-id / state / scheduled_time on it.
-                cursor.execute(f"""
-                    SELECT query_id, state, scheduled_time, query_start_time,
-                           completed_time, error_code, error_message
-                    FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
-                        TASK_NAME => '{_self.task_name}',
-                        SCHEDULED_TIME_RANGE_START =>
-                            DATEADD('minute', -5, CURRENT_TIMESTAMP())
-                    ))
-                    ORDER BY scheduled_time DESC
-                    LIMIT 1
-                """)
-                row = cursor.fetchone()
-
                 metadata: Dict[str, Any] = {
                     "task_fqn": fqn,
                     "task_name": _self.task_name,
                     "database": _self.database,
                     "schema": _self.schema_name,
                 }
-                if row:
-                    columns = [c[0] for c in cursor.description]
-                    rd = dict(zip(columns, row))
-                    metadata.update({
-                        "query_id": rd.get("QUERY_ID"),
-                        "state": rd.get("STATE"),
-                        "scheduled_time": str(rd.get("SCHEDULED_TIME")) if rd.get("SCHEDULED_TIME") else None,
-                        "completed_time": str(rd.get("COMPLETED_TIME")) if rd.get("COMPLETED_TIME") else None,
-                        "error_code": rd.get("ERROR_CODE"),
-                        "error_message": rd.get("ERROR_MESSAGE"),
-                    })
+
+                # SHOW TASKS — works for least-privilege roles where
+                # INFORMATION_SCHEMA may be invisible. Surfaces schedule + state.
+                try:
+                    cursor.execute(
+                        f"SHOW TASKS LIKE '{_self.task_name}' "
+                        f"IN SCHEMA {_self.database}.{_self.schema_name}"
+                    )
+                    info = cursor.fetchone()
+                    if info:
+                        columns = [c[0].lower() for c in cursor.description]
+                        rd = dict(zip(columns, info))
+                        metadata.update({
+                            "task_state": rd.get("state"),
+                            "task_schedule": rd.get("schedule"),
+                            "warehouse": rd.get("warehouse"),
+                            "owner": rd.get("owner"),
+                        })
+                except Exception as exc:
+                    context.log.warning(
+                        f"Could not read task metadata for {_self.task_name}: {exc}. "
+                        f"Execute succeeded; emitting asset without enriched metadata."
+                    )
+
+                # TASK_HISTORY is a table function — wrap in try/except so
+                # EXECUTE TASK still wins even if the role can't read history.
+                try:
+                    cursor.execute(f"""
+                        SELECT query_id, state, scheduled_time, query_start_time,
+                               completed_time, error_code, error_message
+                        FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
+                            TASK_NAME => '{_self.task_name}',
+                            SCHEDULED_TIME_RANGE_START =>
+                                DATEADD('minute', -5, CURRENT_TIMESTAMP())
+                        ))
+                        ORDER BY scheduled_time DESC
+                        LIMIT 1
+                    """)
+                    row = cursor.fetchone()
+                    if row:
+                        columns = [c[0] for c in cursor.description]
+                        rd = dict(zip(columns, row))
+                        metadata.update({
+                            "query_id": rd.get("QUERY_ID"),
+                            "state": rd.get("STATE"),
+                            "scheduled_time": str(rd.get("SCHEDULED_TIME")) if rd.get("SCHEDULED_TIME") else None,
+                            "completed_time": str(rd.get("COMPLETED_TIME")) if rd.get("COMPLETED_TIME") else None,
+                            "error_code": rd.get("ERROR_CODE"),
+                            "error_message": rd.get("ERROR_MESSAGE"),
+                        })
+                except Exception as exc:
+                    context.log.warning(
+                        f"Could not read TASK_HISTORY for {_self.task_name}: {exc}. "
+                        f"Execute succeeded; emitting asset without history metadata."
+                    )
                 return MaterializeResult(metadata=metadata)
             finally:
                 cursor.close()