Skip to content

Commit 671c150

Browse files
eric-thomas-dagsterclaude
eric-thomas-dagster
and
claude
committed
feat(snowflake): plottable numeric metadata across workspace + Wave 3
Dagster auto-charts MetadataValue.int/float fields on each asset's Plots tab over time. Wiring numeric metadata into every Snowflake materialization turns each demo run into a per-asset time series of duration, rows, bytes, credits, partition pruning. New module-level _emit_query_perf(cursor, query_id) helper added to: - integrations/snowflake_workspace/component.py - assets/data-warehouse/snowflake_task_execute_asset/component.py - assets/data-warehouse/snowflake_dynamic_table_refresh_asset/component.py - assets/data-warehouse/snowflake_stored_procedure_call_asset/component.py Helper queries TABLE(INFORMATION_SCHEMA.QUERY_HISTORY_BY_QUERY_ID(...)) for cursor.sfqid (the just-executed query) and returns 7 plottable fields under a stable snowflake/* namespace: - snowflake/query_duration_ms (MetadataValue.int) - snowflake/rows_produced (MetadataValue.int) - snowflake/bytes_scanned (MetadataValue.int) - snowflake/bytes_spilled_local (MetadataValue.int) - snowflake/credits_used (MetadataValue.float) - snowflake/partitions_scanned (MetadataValue.int) - snowflake/partitions_total (MetadataValue.int) Returns {} on any failure so callers can blindly metadata.update(...) without risking the materialization itself. Wired after EXECUTE TASK / CALL / ALTER DYNAMIC TABLE REFRESH / ALTER MV RESUME / ALTER EXTERNAL TABLE REFRESH / ALTER PIPE REFRESH across: Workspace component (multi-entity): - _task_asset: perf after EXECUTE TASK - _make_proc_asset: perf after CALL - _dynamic_table_asset: perf after REFRESH + rows/bytes as MetadataValue.int - _mv_asset: perf after RESUME + rows/bytes as int - _external_table_asset: perf after REFRESH + rows as int - _snowpipe_asset: perf after REFRESH + parses SYSTEM$PIPE_STATUS JSON for pending_file_count + execution_state, plus recent_loads count - _stream_asset: rewritten as proper ObserveResult with snowflake/has_data (0/1 int) + snowflake/pending_rows (int) - _stage_asset: rewritten as proper ObserveResult with snowflake/file_count (int) + snowflake/total_bytes (int) Wave 3 single-entity: - snowflake_task_execute_asset: perf after EXECUTE TASK - snowflake_dynamic_table_refresh_asset: perf after REFRESH + rows/bytes as int - snowflake_stored_procedure_call_asset: perf after CALL The result: every materialized Snowflake asset gets a Performance time series in Dagster's catalog without any customer config. Demo-impressive: bytes scanned trending up over a week, query duration spiking before a refactor, partition pruning ratio, credits-per-run, etc. L1 validation harness scorecard unchanged: PASS: 562 WARN: 0 FAIL: 41 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 1ced5bf commit 671c150

6 files changed

Lines changed: 287 additions & 33 deletions

File tree

‎assets/data-warehouse/snowflake_dynamic_table_refresh_asset/component.py‎

Lines changed: 42 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,47 @@
1616
ComponentLoadContext,
1717
Definitions,
1818
MaterializeResult,
19+
MetadataValue,
1920
Model,
2021
Resolvable,
2122
asset,
2223
)
2324
from pydantic import Field
2425

2526

27+
def _emit_query_perf(cursor, query_id) -> dict:
28+
"""Per-query Snowflake perf metrics as plottable MetadataValue.int/float fields.
29+
30+
Dagster auto-plots numeric metadata on each asset's Plots tab; calling
31+
this after every ``cursor.execute()`` adds a time-series of duration,
32+
rows produced, bytes scanned, credits used, partition pruning.
33+
"""
34+
if not query_id:
35+
return {}
36+
try:
37+
cursor.execute(
38+
"SELECT total_elapsed_time, rows_produced, bytes_scanned, "
39+
" bytes_spilled_to_local_storage, "
40+
" credits_used_cloud_services, "
41+
" partitions_scanned, partitions_total "
42+
f"FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY_BY_QUERY_ID('{query_id}'))"
43+
)
44+
row = cursor.fetchone()
45+
if not row:
46+
return {}
47+
return {
48+
"snowflake/query_duration_ms": MetadataValue.int(int(row[0] or 0)),
49+
"snowflake/rows_produced": MetadataValue.int(int(row[1] or 0)),
50+
"snowflake/bytes_scanned": MetadataValue.int(int(row[2] or 0)),
51+
"snowflake/bytes_spilled_local": MetadataValue.int(int(row[3] or 0)),
52+
"snowflake/credits_used": MetadataValue.float(float(row[4] or 0.0)),
53+
"snowflake/partitions_scanned": MetadataValue.int(int(row[5] or 0)),
54+
"snowflake/partitions_total": MetadataValue.int(int(row[6] or 0)),
55+
}
56+
except Exception:
57+
return {}
58+
59+
2660
class SnowflakeDynamicTableRefreshAssetComponent(Component, Model, Resolvable):
2761
"""Materialize by ALTER DYNAMIC TABLE ... REFRESH on a named dynamic table.
2862
@@ -119,13 +153,16 @@ def snowflake_dynamic_table_refresh_asset(context: AssetExecutionContext) -> Mat
119153
try:
120154
context.log.info(f"ALTER DYNAMIC TABLE {fqn} REFRESH")
121155
cursor.execute(f"ALTER DYNAMIC TABLE {fqn} REFRESH")
156+
refresh_sfqid = cursor.sfqid
122157

123158
metadata: Dict[str, Any] = {
124159
"dynamic_table_fqn": fqn,
125160
"dynamic_table_name": _self.dynamic_table_name,
126161
"database": _self.database,
127162
"schema": _self.schema_name,
128163
}
164+
# Per-run numeric perf trace (auto-plots on Plots tab).
165+
metadata.update(_emit_query_perf(cursor, refresh_sfqid))
129166

130167
# SHOW DYNAMIC TABLES (not INFORMATION_SCHEMA.DYNAMIC_TABLES) —
131168
# INFORMATION_SCHEMA can be invisible to least-privilege roles
@@ -146,10 +183,13 @@ def snowflake_dynamic_table_refresh_asset(context: AssetExecutionContext) -> Mat
146183
"last_refresh_status": rd.get("last_refresh_state"),
147184
"target_lag": rd.get("target_lag"),
148185
"refresh_mode": rd.get("refresh_mode"),
149-
"rows": rd.get("rows"),
150-
"bytes": rd.get("bytes"),
151186
"owner": rd.get("owner"),
152187
})
188+
# Numeric fields → plottable.
189+
if rd.get("rows") is not None:
190+
metadata["snowflake/rows"] = MetadataValue.int(int(rd["rows"]))
191+
if rd.get("bytes") is not None:
192+
metadata["snowflake/bytes"] = MetadataValue.int(int(rd["bytes"]))
153193
except Exception as exc:
154194
context.log.warning(
155195
f"Could not read DT metadata for {_self.dynamic_table_name}: {exc}. "

‎assets/data-warehouse/snowflake_stored_procedure_call_asset/component.py‎

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,13 +13,47 @@
1313
ComponentLoadContext,
1414
Definitions,
1515
MaterializeResult,
16+
MetadataValue,
1617
Model,
1718
Resolvable,
1819
asset,
1920
)
2021
from pydantic import Field
2122

2223

24+
def _emit_query_perf(cursor, query_id) -> dict:
25+
"""Per-query Snowflake perf metrics as plottable MetadataValue.int/float fields.
26+
27+
Dagster auto-plots numeric metadata on each asset's Plots tab; calling
28+
this after every ``cursor.execute()`` adds a time-series of duration,
29+
rows produced, bytes scanned, credits used, partition pruning.
30+
"""
31+
if not query_id:
32+
return {}
33+
try:
34+
cursor.execute(
35+
"SELECT total_elapsed_time, rows_produced, bytes_scanned, "
36+
" bytes_spilled_to_local_storage, "
37+
" credits_used_cloud_services, "
38+
" partitions_scanned, partitions_total "
39+
f"FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY_BY_QUERY_ID('{query_id}'))"
40+
)
41+
row = cursor.fetchone()
42+
if not row:
43+
return {}
44+
return {
45+
"snowflake/query_duration_ms": MetadataValue.int(int(row[0] or 0)),
46+
"snowflake/rows_produced": MetadataValue.int(int(row[1] or 0)),
47+
"snowflake/bytes_scanned": MetadataValue.int(int(row[2] or 0)),
48+
"snowflake/bytes_spilled_local": MetadataValue.int(int(row[3] or 0)),
49+
"snowflake/credits_used": MetadataValue.float(float(row[4] or 0.0)),
50+
"snowflake/partitions_scanned": MetadataValue.int(int(row[5] or 0)),
51+
"snowflake/partitions_total": MetadataValue.int(int(row[6] or 0)),
52+
}
53+
except Exception:
54+
return {}
55+
56+
2357
class SnowflakeStoredProcedureCallAssetComponent(Component, Model, Resolvable):
2458
"""Materialize by CALL on a named Snowflake stored procedure.
2559
@@ -141,15 +175,19 @@ def snowflake_stored_procedure_call_asset(context: AssetExecutionContext) -> Mat
141175
try:
142176
context.log.info(call_sql)
143177
cursor.execute(call_sql)
178+
call_sfqid = cursor.sfqid
144179
result = cursor.fetchone()
145-
return MaterializeResult(metadata={
180+
metadata = {
146181
"procedure_fqn": fqn,
147182
"procedure_name": _self.procedure_name,
148183
"database": _self.database,
149184
"schema": _self.schema_name,
150185
"call_sql": call_sql,
151186
"return_value": str(result[0]) if result else None,
152-
})
187+
}
188+
# Per-run numeric perf trace (auto-plots on Plots tab).
189+
metadata.update(_emit_query_perf(cursor, call_sfqid))
190+
return MaterializeResult(metadata=metadata)
153191
finally:
154192
cursor.close()
155193
conn.close()

‎assets/data-warehouse/snowflake_task_execute_asset/component.py‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,50 @@
1818
ComponentLoadContext,
1919
Definitions,
2020
MaterializeResult,
21+
MetadataValue,
2122
Model,
2223
Resolvable,
2324
asset,
2425
)
2526
from pydantic import Field
2627

2728

29+
def _emit_query_perf(cursor, query_id) -> dict:
30+
"""Pull plottable per-query perf metrics from QUERY_HISTORY for ``query_id``.
31+
32+
Dagster auto-plots ``MetadataValue.int`` / ``MetadataValue.float`` on
33+
each asset's Plots tab — so calling this after ``cursor.execute()``
34+
turns every materialization into a time-series of duration / rows /
35+
bytes / credits / partition pruning. ``cursor.sfqid`` exposed by
36+
snowflake-connector returns the last query id. Returns ``{}`` on any
37+
failure so callers can blindly ``metadata.update(...)``.
38+
"""
39+
if not query_id:
40+
return {}
41+
try:
42+
cursor.execute(
43+
"SELECT total_elapsed_time, rows_produced, bytes_scanned, "
44+
" bytes_spilled_to_local_storage, "
45+
" credits_used_cloud_services, "
46+
" partitions_scanned, partitions_total "
47+
f"FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY_BY_QUERY_ID('{query_id}'))"
48+
)
49+
row = cursor.fetchone()
50+
if not row:
51+
return {}
52+
return {
53+
"snowflake/query_duration_ms": MetadataValue.int(int(row[0] or 0)),
54+
"snowflake/rows_produced": MetadataValue.int(int(row[1] or 0)),
55+
"snowflake/bytes_scanned": MetadataValue.int(int(row[2] or 0)),
56+
"snowflake/bytes_spilled_local": MetadataValue.int(int(row[3] or 0)),
57+
"snowflake/credits_used": MetadataValue.float(float(row[4] or 0.0)),
58+
"snowflake/partitions_scanned": MetadataValue.int(int(row[5] or 0)),
59+
"snowflake/partitions_total": MetadataValue.int(int(row[6] or 0)),
60+
}
61+
except Exception:
62+
return {}
63+
64+
2865
class SnowflakeTaskExecuteAssetComponent(Component, Model, Resolvable):
2966
"""Materialize by EXECUTE TASK on a named Snowflake task.
3067
@@ -121,8 +158,10 @@ def snowflake_task_execute_asset(context: AssetExecutionContext) -> MaterializeR
121158
try:
122159
fqn = f"{_self.database}.{_self.schema_name}.{_self.task_name}"
123160
execute_query = f"EXECUTE TASK {fqn}"
161+
exec_sfqid = None
124162
try:
125163
cursor.execute(execute_query)
164+
exec_sfqid = cursor.sfqid
126165
context.log.info(f"Executed Snowflake task: {_self.task_name}")
127166
except Exception as exc:
128167
if "non-root task" in str(exc).lower():
@@ -140,6 +179,8 @@ def snowflake_task_execute_asset(context: AssetExecutionContext) -> MaterializeR
140179
"database": _self.database,
141180
"schema": _self.schema_name,
142181
}
182+
# Per-run numeric perf trace (auto-plots on Plots tab).
183+
metadata.update(_emit_query_perf(cursor, exec_sfqid))
143184

144185
# SHOW TASKS — works for least-privilege roles where
145186
# INFORMATION_SCHEMA may be invisible. Surfaces schedule + state.

‎dagster_community_components/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
import importlib.util
2222
from pathlib import Path
2323

24-
__version__ = "0.6.3"
24+
__version__ = "0.6.4"
2525

2626
_PACKAGE_ROOT = Path(__file__).resolve().parent
2727

0 commit comments

Comments
 (0)