feat(ingest/snowflake): snowflake enhancements to support tasks,snowpipe and stages (#16888)

Author: alokr-dhub

metadata-ingestion/docs/sources/snowflake/snowflake_post.md

@@ -163,6 +163,43 @@ semantic_view_pattern:
 - Requires `REFERENCES` or `SELECT` privileges on semantic views (they are treated as views in Snowflake's permission model)
 - The semantic view definition (SQL DDL) is extracted when available through the `GET_DDL` function
 
+#### Stages, Tasks, and Pipes
+
+DataHub supports ingestion of Snowflake Stages, Tasks, and Snowpipe objects. All three features are disabled by default and can be enabled independently.
+
+##### Stages (`include_stages: true`)
+
+Stages are ingested as containers nested under their parent schema. Internal stages additionally emit a placeholder dataset representing the staged data, which is used for pipe lineage resolution. External stages (S3, GCS, Azure) resolve their URLs to the corresponding cloud platform dataset URN.
+
+```yaml
+include_stages: true
+stage_pattern:
+  allow:
+    - "MY_DB.MY_SCHEMA.*"
+```
+
+##### Tasks (`include_tasks: true`)
+
+Tasks are ingested as DataJob entities grouped under a per-schema DataFlow. Predecessor dependencies between tasks are captured as `inputDatajobs` on the DataJobInputOutput aspect, preserving the DAG structure.
+
+```yaml
+include_tasks: true
+task_pattern:
+  allow:
+    - "MY_DB.MY_SCHEMA.*"
+```
+
+##### Pipes (`include_pipes: true`)
+
+Snowpipe objects are ingested as DataJob entities with lineage derived from parsing the `COPY INTO` statement. The pipe's source stage resolves to an upstream dataset (internal placeholder or external cloud URN) and the target table resolves to a downstream dataset. Enabling pipes automatically scans stages for lineage resolution, even if `include_stages` is false.
+
+```yaml
+include_pipes: true
+pipe_pattern:
+  allow:
+    - "MY_DB.MY_SCHEMA.*"
+```
+
 ### Limitations
 
 Module behavior is constrained by source APIs, permissions, and metadata exposed by the platform. Refer to capability notes for unsupported or conditional features.

metadata-ingestion/docs/sources/snowflake/snowflake_pre.md

@@ -54,6 +54,14 @@ grant imported privileges on database snowflake to role datahub_role;
 // Optional - required if extracting Streamlit Apps
 grant usage on all streamlits in database "<your-database>" to role datahub_role;
 grant usage on future streamlits in database "<your-database>" to role datahub_role;
+
+// Optional - required if extracting Stages, Tasks, or Pipes
+grant usage on all stages in database "<your-database>" to role datahub_role;
+grant usage on future stages in database "<your-database>" to role datahub_role;
+grant monitor on all tasks in database "<your-database>" to role datahub_role;
+grant monitor on future tasks in database "<your-database>" to role datahub_role;
+grant monitor on all pipes in database "<your-database>" to role datahub_role;
+grant monitor on future pipes in database "<your-database>" to role datahub_role;
 ```
 
 The details of each granted privilege can be viewed in the [Snowflake docs](https://docs.snowflake.com/en/user-guide/security-access-control-privileges.html). A summary of each privilege and why it is required for this connector:
@@ -71,6 +79,9 @@ grant usage on schema "<your-database>"."<your-schema>" to role datahub_role;
 
 - `select` on `streams` is required for stream definitions to be available. This does not allow selecting the data (not required) unless the underlying dataset has select access as well.
 - `usage` on `streamlit` is required to show streamlits in a database. See the schema-level `usage` example above.
+- `usage` on `stages` is required to list stages via `SHOW STAGES`. Only needed if `include_stages: true` or `include_pipes: true`.
+- `monitor` on `tasks` is required to list tasks via `SHOW TASKS`. Only needed if `include_tasks: true`.
+- `monitor` on `pipes` is required to list pipes via `SHOW PIPES`. Only needed if `include_pipes: true`.
 
 This represents the bare minimum privileges required to extract databases, schemas, views, and tables from Snowflake.
 

metadata-ingestion/src/datahub/ingestion/source/common/subtypes.py

@@ -40,6 +40,7 @@ class DatasetSubTypes(StrEnum):
     GOOGLE_SHEETS = "Google Sheets"
     GOOGLE_SHEETS_NAMED_RANGE = "Google Sheets Named Range"
     SEMANTIC_MODEL = "Semantic Model"
+    SNOWFLAKE_STAGE_DATA = "Snowflake Stage Data"
 
     # TODO: Create separate entity...
     NOTEBOOK = "Notebook"
@@ -74,6 +75,7 @@ class DatasetContainerSubTypes(StrEnum):
     FABRIC_LAKEHOUSE = "Fabric Lakehouse"
     FABRIC_WAREHOUSE = "Fabric Warehouse"
     FABRIC_SCHEMA = "Fabric Schema"
+    SNOWFLAKE_STAGE = "Snowflake Stage"
     # Pinecone
     PINECONE_INDEX = "Pinecone Index"
     PINECONE_NAMESPACE = "Pinecone Namespace"
@@ -102,6 +104,8 @@ class FlowContainerSubTypes(StrEnum):
     MSSQL_JOB = "Job"
     PROCEDURE_CONTAINER = "Procedures Container"
     ADF_DATA_FACTORY = "Data Factory"
+    SNOWFLAKE_TASK_GROUP = "Snowflake Task Group"
+    SNOWFLAKE_PIPE_GROUP = "Snowflake Pipe Group"
 
 
 class JobContainerSubTypes(StrEnum):
@@ -220,6 +224,10 @@ class DataJobSubTypes(StrEnum):
     FABRIC_DATA_LAKE_ANALYTICS = "Data Lake Analytics"
     FABRIC_AZURE_ML_EXECUTE_PIPELINE = "Azure ML Execute Pipeline"
 
+    # Snowflake
+    SNOWFLAKE_TASK = "Snowflake Task"
+    SNOWFLAKE_PIPE = "Snowflake Pipe"
+
 
 def create_source_capability_modifier_enum():
     all_values: Dict[str, Any] = {}

metadata-ingestion/src/datahub/ingestion/source/snowflake/constants.py

@@ -60,6 +60,9 @@ class SnowflakeObjectDomain(StrEnum):
     PROCEDURE = "procedure"
     DYNAMIC_TABLE = "dynamic table"
     STREAMLIT = "streamlit"
+    STAGE = "stage"
+    TASK = "task"
+    PIPE = "pipe"
 
 
 GENERIC_PERMISSION_ERROR_KEY = "permission-error"

metadata-ingestion/src/datahub/ingestion/source/snowflake/snowflake_config.py

@@ -189,6 +189,24 @@ class SnowflakeFilterConfig(SQLFilterConfig):
         " use the regex 'Analytics.public.sales.*'",
     )
 
+    stage_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns for stages to filter in ingestion. "
+        "Specify regex to match the entire stage name in database.schema.stage format.",
+    )
+
+    task_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns for tasks to filter in ingestion. "
+        "Specify regex to match the entire task name in database.schema.task format.",
+    )
+
+    pipe_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="Regex patterns for pipes to filter in ingestion. "
+        "Specify regex to match the entire pipe name in database.schema.pipe format.",
+    )
+
     match_fully_qualified_names: bool = Field(
         default=False,
         description="Whether `schema_pattern` is matched against fully qualified schema name `<catalog>.<schema>`.",
@@ -438,6 +456,21 @@ class SnowflakeV2Config(
         description="Configuration for semantic views ingestion.",
     )
 
+    include_stages: bool = Field(
+        default=False,
+        description="If enabled, Snowflake Stages will be ingested as containers with associated metadata.",
+    )
+
+    include_tasks: bool = Field(
+        default=False,
+        description="If enabled, Snowflake Tasks will be ingested as DataJobs with DAG dependencies and SQL lineage.",
+    )
+
+    include_pipes: bool = Field(
+        default=False,
+        description="If enabled, Snowflake Snowpipe objects will be ingested as DataJobs with COPY INTO lineage.",
+    )
+
     structured_property_pattern: AllowDenyPattern = Field(
         default=AllowDenyPattern.allow_all(),
         description=(