Data Lake: Show network activity for defined Sophos Process ID

This query will detail network activity for a defined Sophos Process ID

-- Data Lake show network activity for defined Sophos Process ID
-- VARIABLE $$sophos_pid$$,       SophosPID

WITH split_pids AS (
    SELECT
        x2.new_pid,
        x1.*
    FROM
        xdr_data AS x1
    CROSS JOIN
        UNNEST(SPLIT(x1.sophos_pids, ',')) AS x2(new_pid)
    WHERE
        x1.query_name IN ('sophos_ips_windows', 'sophos_urls_windows')
),

full_list AS (
    SELECT
        x1.meta_hostname AS ep_name,
        x1.query_name AS table_name,
        x2.time AS date_time,
        x2.username AS user_name,
        x2.parent_name AS parent_process_name,
        x2.name AS process_name,
        x2.cmdline AS cmd_line,
        x2.sophos_pid AS sophos_pid,
        x2.parent_sophos_pid AS parent_sophos_pid,
        x2.sha256,
        CASE CAST(x1.query_name = 'sophos_ips_windows' AS INT)
            WHEN 1 THEN x1.source_ip
            ELSE REGEXP_REPLACE(x1.source_ips, ',', CHR(10))
        END AS local_ip,
        x1.port AS local_port,
        CASE CAST(x1.query_name = 'sophos_ips_windows' AS INT)
            WHEN 1 THEN x1.destination_ip
            ELSE REGEXP_REPLACE(x1.destination_ips, ',', CHR(10))
        END AS remote_ip,
        x1.destination_port AS remote_port,
        x1.protocol AS protocol,
        x1.domain,
        x1.clean_urls,
        x2.path,
        x2.ml_score,
        x2.pua_score,
        x2.global_rep,
        x2.local_rep,
        x2.parent_path
    FROM
        xdr_data AS x2
    RIGHT JOIN split_pids AS x1 ON x2.query_name = 'running_processes_windows_sophos'
    WHERE
        x1.new_pid = x2.sophos_pid

    UNION ALL

    SELECT
        x1.meta_hostname AS ep_name,
        x1.query_name AS table_name,
        x2.time AS date_time,
        x2.username AS user_name,
        x2.parent_name AS parent_process_name,
        x2.name AS process_name,
        x2.cmdline AS cmd_line,
        x2.sophos_pid AS sophos_pid,
        x2.parent_sophos_pid AS parent_sophos_pid,
        x2.sha256,
        x1.local_address AS local_ip,
        x1.port AS local_port,
        x1.remote_address AS remote_ip,
        x1.remote_port AS remote_port,
        x1.protocol AS protocol,
        x1.domain,
        x1.clean_urls,
        x2.path,
        x2.ml_score,
        x2.pua_score,
        x2.global_rep,
        x2.local_rep,
        x2.parent_path
    FROM
        xdr_data AS x2
    RIGHT JOIN xdr_data AS x1 ON
        x2.query_name = 'running_processes_windows_sophos'
        AND x1.meta_hostname = x2.meta_hostname
        AND x1.name = x2.name
        AND x1.cmdline = x2.cmdline
        AND x1.pid = x2.pid
        AND x1.parent = x2.parent
    WHERE
        x1.query_name = 'open_sockets'

    UNION ALL

    SELECT
        x1.meta_hostname AS ep_name,
        x1.query_name AS table_name,
        x2.time AS date_time,
        x2.username AS user_name,
        x2.parent_name AS parent_process_name,
        x2.name AS process_name,
        x2.cmdline AS cmd_line,
        x2.sophos_pid AS sophos_pid,
        x2.parent_sophos_pid AS parent_sophos_pid,
        x2.sha256,
        x1.address AS local_ip,
        x1.port AS local_port,
        x1.remote_address AS remote_ip,
        x1.remote_port AS remote_port,
        x1.protocol AS protocol,
        x1.domain,
        x1.clean_urls,
        x2.path,
        x2.ml_score,
        x2.pua_score,
        x2.global_rep,
        x2.local_rep,
        x2.parent_path
    FROM
        xdr_data AS x2
    RIGHT JOIN xdr_data AS x1 ON
        x2.query_name = 'running_processes_windows_sophos'
        AND x1.meta_hostname = x2.meta_hostname
        AND x1.name = x2.name
        AND x1.pid = x2.pid
    WHERE
        x1.query_name = 'listening_ports'
        AND x1.address NOT IN ('::1', '0.0.0.0', '::')
)

SELECT
    ARRAY_JOIN(ARRAY_AGG(DISTINCT ep_name), CHR(10)) AS ep_list,
    COUNT(DISTINCT ep_name) AS ep_count,
    table_name,
    COUNT(DISTINCT sophos_pid) AS instances,
    process_name,
    path,
    cmd_line,
    DATE_FORMAT(FROM_UNIXTIME(MIN(date_time)), '%Y-%m-%dT%H:%i:%SZ') AS first_seen,
    DATE_FORMAT(FROM_UNIXTIME(MAX(date_time)), '%Y-%m-%dT%H:%i:%SZ') AS last_seen,
    user_name,
    parent_process_name,
    parent_path,
    ARRAY_JOIN(ARRAY_AGG(DISTINCT local_ip), CHR(10)) AS local_ip_list,
    local_port,
    ARRAY_JOIN(ARRAY_AGG(DISTINCT remote_ip), CHR(10)) AS remote_ip_list,
    remote_port,
    protocol,
    clean_urls,
    ARRAY_JOIN(ARRAY_AGG(DISTINCT sophos_pid), CHR(10)) AS sophos_pid_list,
    ARRAY_JOIN(ARRAY_AGG(DISTINCT parent_sophos_pid), CHR(10)) AS parent_sophos_pid_list,
    sha256,
    ml_score,
    pua_score,
    global_rep,
    local_rep
FROM full_list
WHERE LOWER(sophos_pid) LIKE LOWER('$$sophos_pid$$')
GROUP BY
    table_name,
    user_name,
    parent_process_name,
    process_name,
    cmd_line,
    local_port,
    remote_port,
    protocol,
    clean_urls,
    sha256,
    path,
    ml_score,
    pua_score,
    global_rep,
    local_rep,
    parent_path
ORDER BY last_seen DESC