Skip to content

Commit b0728f7

Browse files
eric-thomas-dagsterclaude
eric-thomas-dagster
and
claude
committed
fix(snowflake): child-task tolerance + SHOW for 3 remaining discovery loops
Two related changes per Eric's live-testing findings: 1. Child-task tolerance for EXECUTE TASK (the only-root-tasks rule): Snowflake errors with "non-root task" when you EXECUTE TASK on a child task — parents fire children via task-tree AFTER clauses. The _task_asset (workspace) and snowflake_task_execute_asset (single- entity) now wrap EXECUTE TASK in a try/except that catches the "non-root task" error, logs an info message ("...is a child task — skipping direct EXECUTE TASK (parent triggers it). Run the parent task asset or wait for its schedule..."), and continues to emit the materialization. SHOW TASKS metadata read still happens after. Without this, materializing a child task in dg dev fails the asset even though the parent run does exercise it via cascade. 2. SHOW ... IN SCHEMA for the 3 remaining INFORMATION_SCHEMA discovery loops that hadn't been swapped yet: - Stored procedures: INFORMATION_SCHEMA.PROCEDURES → SHOW PROCEDURES IN SCHEMA. Column renames: PROCEDURE_NAME → NAME, PROCEDURE_CATALOG dropped (use self.database), PROCEDURE_SCHEMA → SCHEMA_NAME, ARGUMENT_SIGNATURE → ARGUMENTS. - Stages: INFORMATION_SCHEMA.STAGES → SHOW STAGES IN SCHEMA. Column renames: STAGE_NAME → NAME, STAGE_CATALOG → DATABASE_NAME, STAGE_SCHEMA → SCHEMA_NAME, STAGE_URL → URL, STAGE_TYPE → TYPE. - External tables: INFORMATION_SCHEMA.TABLES (with table_type = 'EXTERNAL TABLE' filter) → SHOW EXTERNAL TABLES IN SCHEMA. Column renames: TABLE_NAME → NAME, TABLE_CATALOG → DATABASE_NAME, TABLE_SCHEMA → SCHEMA_NAME. Also surfaces richer fields (location, file_format_name, last_refreshed_on, owner). Combined with the v0.6.1 swaps for dynamic tables / tasks / pipes / MVs / alerts (materialize-time reads + DT sensor query), the full workspace component now uses SHOW everywhere a least-privilege role could hit "does not exist or not authorized" on INFORMATION_SCHEMA. Per Eric: "The component has the same shape for every workspace import (refresh/execute then read INFORMATION_SCHEMA). Each one is a future failure waiting to happen on either the action OR the metadata read. The SHOW alternative is privilege-light and exists for every entity type." L1 validation harness scorecard unchanged from v0.6.1: PASS: 562 WARN: 0 FAIL: 41 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 468bd0e commit b0728f7

4 files changed

Lines changed: 70 additions & 59 deletions

File tree

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

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -120,8 +120,19 @@ def snowflake_task_execute_asset(context: AssetExecutionContext) -> MaterializeR
120120
cursor = conn.cursor()
121121
try:
122122
fqn = f"{_self.database}.{_self.schema_name}.{_self.task_name}"
123-
context.log.info(f"EXECUTE TASK {fqn}")
124-
cursor.execute(f"EXECUTE TASK {fqn}")
123+
execute_query = f"EXECUTE TASK {fqn}"
124+
try:
125+
cursor.execute(execute_query)
126+
context.log.info(f"Executed Snowflake task: {_self.task_name}")
127+
except Exception as exc:
128+
if "non-root task" in str(exc).lower():
129+
context.log.info(
130+
f"{_self.task_name} is a child task — skipping direct EXECUTE TASK "
131+
f"(parent triggers it). Run the parent task asset (or wait for "
132+
f"its schedule) to actually exercise this asset."
133+
)
134+
else:
135+
raise
125136

126137
metadata: Dict[str, Any] = {
127138
"task_fqn": fqn,

‎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.1"
24+
__version__ = "0.6.2"
2525

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

‎integrations/snowflake_workspace/component.py‎

Lines changed: 55 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -459,8 +459,18 @@ def _task_asset(context: AssetExecutionContext):
459459
cursor = conn.cursor()
460460
try:
461461
execute_query = f"EXECUTE TASK {db_v}.{schema_v}.{task_name_v}"
462-
cursor.execute(execute_query)
463-
context.log.info(f"Executed Snowflake task: {task_name_v}")
462+
try:
463+
cursor.execute(execute_query)
464+
context.log.info(f"Executed Snowflake task: {task_name_v}")
465+
except Exception as exc:
466+
if "non-root task" in str(exc).lower():
467+
context.log.info(
468+
f"{task_name_v} is a child task — skipping direct EXECUTE TASK "
469+
f"(parent triggers it). Run the parent task asset (or wait for "
470+
f"its schedule) to actually exercise this asset."
471+
)
472+
else:
473+
raise
464474

465475
metadata = {
466476
"task_name": task_name_v,
@@ -542,21 +552,18 @@ def _task_asset(context: AssetExecutionContext):
542552
# Import Stored Procedures
543553
if self.import_stored_procedures:
544554
try:
545-
query = f"""
546-
SELECT
547-
procedure_name,
548-
procedure_schema,
549-
procedure_catalog,
550-
argument_signature,
551-
created
552-
FROM {self.database}.INFORMATION_SCHEMA.PROCEDURES
553-
WHERE procedure_schema = '{self.schema_name}'
554-
"""
555-
555+
# SHOW PROCEDURES (not INFORMATION_SCHEMA.PROCEDURES) —
556+
# SHOW only needs USAGE on the schema + any privilege on
557+
# the proc; INFORMATION_SCHEMA can be invisible to
558+
# least-privilege roles.
559+
query = f"SHOW PROCEDURES IN SCHEMA {self.database}.{self.schema_name}"
556560
procedures = self._execute_query(conn, query)
557561

558562
for proc in procedures:
559-
proc_name = proc['PROCEDURE_NAME']
563+
# SHOW PROCEDURES returns NAME (no PROCEDURE_NAME) and
564+
# has no CATALOG column — we already know the database
565+
# from self.database.
566+
proc_name = proc['NAME']
560567

561568
if not self._should_include_entity(proc_name):
562569
continue
@@ -571,9 +578,9 @@ def _task_asset(context: AssetExecutionContext):
571578
description=f"Snowflake stored procedure: {proc_name}",
572579
metadata={
573580
"snowflake_procedure_name": proc_name,
574-
"snowflake_database": proc['PROCEDURE_CATALOG'],
575-
"snowflake_schema": proc['PROCEDURE_SCHEMA'],
576-
"snowflake_signature": proc.get('ARGUMENT_SIGNATURE'),
581+
"snowflake_database": self.database,
582+
"snowflake_schema": proc.get('SCHEMA_NAME', self.schema_name),
583+
"snowflake_signature": proc.get('ARGUMENTS'),
577584
"entity_type": "stored_procedure",
578585
},
579586
))
@@ -605,8 +612,9 @@ def _procedure_asset(context: AssetExecutionContext):
605612
return _procedure_asset
606613

607614
assets_list.append(_make_proc_asset(
608-
proc_name, proc['PROCEDURE_CATALOG'],
609-
proc['PROCEDURE_SCHEMA'], _proc_kwargs, self,
615+
proc_name, self.database,
616+
proc.get('SCHEMA_NAME', self.schema_name),
617+
_proc_kwargs, self,
610618
))
611619

612620
except Exception as e:
@@ -894,23 +902,17 @@ def _snowpipe_asset(context: AssetExecutionContext):
894902
# Import Stages
895903
if self.import_stages:
896904
try:
897-
query = f"""
898-
SELECT
899-
stage_name,
900-
stage_schema,
901-
stage_catalog,
902-
stage_url,
903-
stage_type,
904-
stage_owner,
905-
created
906-
FROM {self.database}.INFORMATION_SCHEMA.STAGES
907-
WHERE stage_schema = '{self.schema_name}'
908-
"""
909-
905+
# SHOW STAGES (not INFORMATION_SCHEMA.STAGES) — SHOW only
906+
# needs USAGE on the schema; INFORMATION_SCHEMA can be
907+
# invisible to least-privilege roles.
908+
query = f"SHOW STAGES IN SCHEMA {self.database}.{self.schema_name}"
910909
stages = self._execute_query(conn, query)
911910

912911
for stage in stages:
913-
stage_name = stage['STAGE_NAME']
912+
# SHOW STAGES returns NAME (not STAGE_NAME), DATABASE_NAME,
913+
# SCHEMA_NAME, URL, TYPE, OWNER — different from
914+
# INFORMATION_SCHEMA.STAGES column names.
915+
stage_name = stage['NAME']
914916

915917
if not self._should_include_entity(stage_name):
916918
continue
@@ -925,10 +927,10 @@ def _snowpipe_asset(context: AssetExecutionContext):
925927
description=f"Snowflake stage: {stage_name}",
926928
metadata={
927929
"snowflake_stage_name": stage_name,
928-
"snowflake_database": stage['STAGE_CATALOG'],
929-
"snowflake_schema": stage['STAGE_SCHEMA'],
930-
"snowflake_url": stage.get('STAGE_URL'),
931-
"snowflake_type": stage.get('STAGE_TYPE'),
930+
"snowflake_database": stage.get('DATABASE_NAME', self.database),
931+
"snowflake_schema": stage.get('SCHEMA_NAME', self.schema_name),
932+
"snowflake_url": stage.get('URL'),
933+
"snowflake_type": stage.get('TYPE'),
932934
"entity_type": "stage",
933935
},
934936
))
@@ -955,7 +957,10 @@ def _stage_asset(context: AssetExecutionContext):
955957
return _stage_asset
956958

957959
assets_list.append(_make_stage_asset(
958-
stage_name, stage['STAGE_CATALOG'], stage['STAGE_SCHEMA'], _stage_kwargs, self,
960+
stage_name,
961+
stage.get('DATABASE_NAME', self.database),
962+
stage.get('SCHEMA_NAME', self.schema_name),
963+
_stage_kwargs, self,
959964
))
960965

961966
except Exception as e:
@@ -1052,23 +1057,18 @@ def _mv_asset(context: AssetExecutionContext):
10521057
# Import External Tables
10531058
if self.import_external_tables:
10541059
try:
1055-
query = f"""
1056-
SELECT
1057-
table_name,
1058-
table_schema,
1059-
table_catalog,
1060-
table_owner,
1061-
created,
1062-
last_altered
1063-
FROM {self.database}.INFORMATION_SCHEMA.TABLES
1064-
WHERE table_schema = '{self.schema_name}'
1065-
AND table_type = 'EXTERNAL TABLE'
1066-
"""
1067-
1060+
# SHOW EXTERNAL TABLES (not INFORMATION_SCHEMA.TABLES) —
1061+
# SHOW only needs USAGE on the schema; INFORMATION_SCHEMA
1062+
# can be invisible to least-privilege roles.
1063+
query = f"SHOW EXTERNAL TABLES IN SCHEMA {self.database}.{self.schema_name}"
10681064
ext_tables = self._execute_query(conn, query)
10691065

10701066
for ext_table in ext_tables:
1071-
table_name = ext_table['TABLE_NAME']
1067+
# SHOW EXTERNAL TABLES returns NAME (not TABLE_NAME),
1068+
# DATABASE_NAME (not TABLE_CATALOG), SCHEMA_NAME (not
1069+
# TABLE_SCHEMA), plus richer fields (location, owner,
1070+
# last_refreshed_on, file_format_name, etc.).
1071+
table_name = ext_table['NAME']
10721072

10731073
if not self._should_include_entity(table_name):
10741074
continue
@@ -1083,8 +1083,8 @@ def _mv_asset(context: AssetExecutionContext):
10831083
description=f"Snowflake external table: {table_name}",
10841084
metadata={
10851085
"snowflake_table_name": table_name,
1086-
"snowflake_database": ext_table['TABLE_CATALOG'],
1087-
"snowflake_schema": ext_table['TABLE_SCHEMA'],
1086+
"snowflake_database": ext_table.get('DATABASE_NAME', self.database),
1087+
"snowflake_schema": ext_table.get('SCHEMA_NAME', self.schema_name),
10881088
"entity_type": "external_table",
10891089
},
10901090
))
@@ -1135,7 +1135,7 @@ def _external_table_asset(context: AssetExecutionContext):
11351135
return _external_table_asset
11361136

11371137
assets_list.append(_make_external_table_asset(
1138-
table_name, ext_table['TABLE_CATALOG'], ext_table['TABLE_SCHEMA'], _ext_kwargs, self,
1138+
table_name, ext_table.get('DATABASE_NAME', self.database), ext_table.get('SCHEMA_NAME', self.schema_name), _ext_kwargs, self,
11391139
))
11401140

11411141
except Exception as e:

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
44

55
[project]
66
name = "dagster-community-components"
7-
version = "0.6.1"
7+
version = "0.6.2"
88
description = "Community-maintained Dagster components — ingestion, transforms, IO managers, sensors, sinks, resources, and more."
99
readme = "README.md"
1010
requires-python = ">=3.10"

0 commit comments

Comments
 (0)