Snowflake Query Profile SpotApp

Create or replace table <database_name>.<schema_name>.query_operator_stats_cache
(
QUERY_ID     VARCHAR(16777216),
QUERY_TEXT     VARCHAR(16777216),
DATABASE_NAME     VARCHAR(16777216),
SCHEMA_NAME     VARCHAR(16777216),
WAREHOUSE_NAME     VARCHAR(16777216),
WAREHOUSE_SIZE     VARCHAR(16777216),
EXECUTION_STATUS     VARCHAR(16777216),
CLIENT_APPLICATION_ID     VARCHAR(16777216),
APPLICATION_NAME     VARCHAR(16777216),
START_TIME     TIMESTAMP_LTZ(6),
END_TIME     TIMESTAMP_LTZ(6),
USER_NAME     VARCHAR(16777216),
STEP_ID     NUMBER(38,0),
OPERATOR_ID     NUMBER(38,0),
PARENT_OPERATOR_ID     NUMBER(38,0),
OPERATOR_TYPE     VARCHAR(16777216),
OPERATOR_STATISTICS     VARIANT,
EXECUTION_TIME_BREAKDOWN     VARIANT,
OPERATOR_ATTRIBUTES     VARIANT,
TABLENAME     VARCHAR(16777216),
TABLE_NCOLS     NUMBER(9,0),
OPERATOR_EXECUTION_TIME     FLOAT,
LOCAL_DISK_IO     FLOAT,
PROCESSING     FLOAT,
SYNCHRONIZATION     FLOAT,
BYTES_SCANNED     NUMBER(38,0),
PERCENTAGE_SCANNED_FROM_CACHE     FLOAT,
NETWORK_BYTES     NUMBER(38,0),
OUTPUT_ROWS     VARIANT,
INPUT_ROWS     VARIANT,
ROW_MULTIPLE     FLOAT,
BYTES_SPILLED_LOCAL_STORAGE     VARIANT,
BYTES_SPILLED_REMOTE_STORAGE     VARIANT,
PARTITIONS_SCANNED     VARIANT,
PARTITIONS_TOTAL     VARIANT,
PARTITION_SCAN_RATIO     FLOAT,
EXPLODING_JOIN     VARCHAR(16777216),
UNION_WITHOUT_ALL     VARCHAR(16777216),
QUERIES_TOO_LARGE_MEMORY     VARCHAR(16777216),
INEFFICIENT_PRUNING_FLAG     VARCHAR(16777216),
ELAPSED_SECONDS     NUMBER(38,6),
CLUSTERING_KEY     VARCHAR(16777216));

create or replace task query_session_task
  warehouse = <warehouse_names>
  SCHEDULE = '60 minute'
  AS
  create or replace table <database_name>.<schema_name>.query_sessions
    as
    select     QUERY_ID
    FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY q
    INNER JOIN SNOWFLAKE.ACCOUNT_USAGE.SESSIONS s
    ON q.SESSION_ID = s.SESSION_ID
            where bytes_scanned > 1024 * 1024
	and total_elapsed_time / 1000 > 120
            and execution_status = 'SUCCESS'
            and end_time >= CURRENT_DATE - INTERVAL '13 DAY'
            AND query_type = 'SELECT'
            AND warehouse_name IS NOT NULL
            order by start_time desc;

CREATE or replace TASK query_stats_task
  warehouse = <warehouse_names>
  SCHEDULE = '65 minute'
  AS
  execute immediate $$
  declare
    query_id string;
    query_tag string;
    qid int;
    session_id int;
    c1 cursor for
      select query_id
      from <database_name>.<schema_name>.query_sessions
      where query_id not in (select query_id from <database_name>.<schema_name>.query_operator_stats_cache);
  begin
    open c1;
    for record in c1 do
      fetch c1 into query_id;
      insert into <database_name>.<schema_name>.query_operator_stats_cache
        with query_stats as (
            select
            q_h.QUERY_ID,
    	    q_h.QUERY_TEXT,
   	        q_h.DATABASE_NAME,
    	    q_h.SCHEMA_NAME,
    	    q_h.WAREHOUSE_NAME,
   	        q_h.WAREHOUSE_SIZE,
    	    q_h.EXECUTION_STATUS,
            q_h.SESSION_ID,
            q_h.START_TIME,
            q_h.END_TIME,
            q_h.USER_NAME,
            q_p.STEP_ID,
            q_p.OPERATOR_ID,
            q_p.PARENT_OPERATORS,
            q_p.OPERATOR_TYPE,
            q_p.OPERATOR_STATISTICS,
            q_p.EXECUTION_TIME_BREAKDOWN,
            q_p.OPERATOR_ATTRIBUTES,
            q_p.operator_attributes:table_name::string tablename,
            array_size(q_p.operator_attributes:columns) table_ncols,
            q_p.EXECUTION_TIME_BREAKDOWN:overall_percentage::float as OPERATOR_EXECUTION_TIME,
            q_p.execution_time_breakdown:local_disk_io::float local_disk_io,
            q_p.execution_time_breakdown:processing::float processing,
            q_p.execution_time_breakdown:synchronization::float synchronization,
            q_p.operator_statistics:io:bytes_scanned::int bytes_scanned,
            q_p.operator_statistics:io:percentage_scanned_from_cache::float percentage_scanned_from_cache,
            q_p.operator_statistics:network:network_bytes::int network_bytes,
            q_p.OPERATOR_STATISTICS:output_rows as output_rows,
            q_p.OPERATOR_STATISTICS:input_rows as input_rows,
            div0(q_p.operator_statistics:output_rows , operator_statistics:input_rows) as row_multiple,
            q_p.OPERATOR_STATISTICS:spilling:bytes_spilled_local_storage as bytes_spilled_local_storage,
            q_p.OPERATOR_STATISTICS:spilling:bytes_spilled_remote_storage as bytes_spilled_remote_storage,
            q_p.OPERATOR_STATISTICS:pruning:partitions_scanned as partitions_scanned,
            q_p.OPERATOR_STATISTICS:pruning:partitions_total as partitions_total,        div0(q_p.OPERATOR_STATISTICS:pruning:partitions_scanned,OPERATOR_STATISTICS:pruning:partitions_total) as partition_scan_ratio,
            case when div0(q_p.operator_statistics:output_rows , operator_statistics:input_rows) > 1 then 1 else 0 end as EXPLODING_JOIN,
            case when OPERATOR_TYPE = 'UnionAll' and lag(OPERATOR_TYPE) over (ORDER BY OPERATOR_ID) = 'Aggregate' THEN 1 ELSE 0 END AS UNION_WITHOUT_ALL,
            CASE WHEN q_p.OPERATOR_STATISTICS:spilling:bytes_spilled_local_storage>0 OR q_p.OPERATOR_STATISTICS:spilling:bytes_spilled_remote_storage>0 THEN 1 ELSE 0 END AS QUERIES_TOO_LARGE_MEMORY,
            CASE WHEN div0(q_p.OPERATOR_STATISTICS:pruning:partitions_scanned,OPERATOR_STATISTICS:pruning:partitions_total) >= .8 AND q_p.OPERATOR_STATISTICS:pruning:partitions_total >= 20000 THEN 1 ELSE 0 END AS INEFFICIENT_PRUNING_FLAG,
            q_h.total_elapsed_time / 1000 as elapsed_seconds
            from "SNOWFLAKE"."ACCOUNT_USAGE"."QUERY_HISTORY" as q_h,
            table(get_query_operator_stats(:query_id)) as q_p
            where q_h.QUERY_ID=(:query_id) order by STEP_ID, OPERATOR_ID )
            select
            QUERY_ID,
            QUERY_TEXT,
            DATABASE_NAME,
            SCHEMA_NAME,
            WAREHOUSE_NAME,
            WAREHOUSE_SIZE,
            EXECUTION_STATUS,
            CLIENT_APPLICATION_ID,
            PARSE_JSON(CLIENT_ENVIRONMENT):APPLICATION::string as APPLICATION_NAME,
            START_TIME,
            END_TIME,
            query_stats.USER_NAME,
            STEP_ID,
            OPERATOR_ID,
            cast(PARENT_OPERATORS[0] as number(38,0)) as PARENT_OPERATOR_ID,
            OPERATOR_TYPE,
            OPERATOR_STATISTICS,
            EXECUTION_TIME_BREAKDOWN,
            OPERATOR_ATTRIBUTES,
            tablename,
            table_ncols,
            OPERATOR_EXECUTION_TIME,
            local_disk_io,
            processing,
            synchronization,
            bytes_scanned,
            percentage_scanned_from_cache,
            network_bytes,
            output_rows,
            input_rows,
            row_multiple,
            bytes_spilled_local_storage,
            bytes_spilled_remote_storage,
            partitions_scanned,
            partitions_total,
            partition_scan_ratio,
            EXPLODING_JOIN,
            UNION_WITHOUT_ALL,
            QUERIES_TOO_LARGE_MEMORY,
            INEFFICIENT_PRUNING_FLAG,
            elapsed_seconds,
            CLUSTERING_KEY
            from query_stats
            inner join "SNOWFLAKE"."ACCOUNT_USAGE".SESSIONS s on query_stats.SESSION_ID = s.SESSION_ID
            left join
            "SNOWFLAKE"."INFORMATION_SCHEMA"."TABLES" t
            on query_stats.tablename= t.TABLE_CATALOG || '.' || t.TABLE_SCHEMA || '.' || t.TABLE_NAME
            order by STEP_ID, OPERATOR_ID;
    end for;
    return query_id;
  end
  $$;