subquery.sql

CREATE TYPE user_composite_type AS (tenant_id bigint, user_id bigint);



CREATE FUNCTION user_composite_type_equal(user_composite_type, 
user_composite_type) 
returns boolean AS 'select $1.tenant_id = $2.tenant_id AND $1.user_id = $2.user_id;' language sql immutable
returns NULL ON NULL input; 

CREATE OR REPLACE FUNCTION btree_comparison(v1 user_composite_type, v2 user_composite_type) returns integer AS 
$$ 
BEGIN 
  if v1.tenant_id = v2.tenant_id AND v1.user_id = v2.user_id THEN 
    return 0; 
  end if;

    if  v1.tenant_id > v2.tenant_id THEN 
        RETURN 1;
    end if;

    if v1.tenant_id < v2.tenant_id THEN 
        RETURN -1;
    end if;

    RETURN 0;
END;
$$ language 'plpgsql' immutable 
returns NULL ON NULL input;

CREATE operator = ( leftarg = user_composite_type, rightarg = user_composite_type, PROCEDURE = user_composite_type_equal, commutator = =, RESTRICT = eqsel, JOIN = eqjoinsel, merges, hashes );CREATE FUNCTION user_composite_type_hash(user_composite_type)
returns int AS 'SELECT hashtext( ($1.tenant_id + $1.user_id)::text);' language sql immutable 
returns NULL ON NULL input; 



create operator class user_composite_type_btree_op_class DEFAULT FOR type user_composite_type using btree AS operator 3 = (user_composite_type, user_composite_type), FUNCTION 1 btree_comparison(user_composite_type, user_composite_type);
create operator class user_composite_type_hash_op_class DEFAULT FOR type user_composite_type using hash AS operator 1 = (user_composite_type, user_composite_type), FUNCTION 1 user_composite_type_hash(user_composite_type);

SET citus.shard_count TO 4;

CREATE TABLE events (
    composite_id user_composite_type,
    event_id bigint,
    event_type character varying(255),
    event_time bigint
);
SELECT create_distributed_table('events', 'composite_id');

\COPY events FROM STDIN WITH CSV
"(1,1001)",20001,click,1472807012
"(1,1001)",20002,submit,1472807015
"(1,1001)",20003,pay,1472807020
"(1,1002)",20010,click,1472807022
"(1,1002)",20011,click,1472807023
"(1,1002)",20012,submit,1472807025
"(1,1002)",20013,pay,1472807030
"(1,1003)",20014,click,1472807032
"(1,1003)",20015,click,1472807033
"(1,1003)",20016,click,1472807034
"(1,1003)",20017,submit,1472807035
\.

CREATE TABLE users (
    composite_id user_composite_type,
    lastseen bigint
);
SELECT create_distributed_table('users', 'composite_id');

\COPY users FROM STDIN WITH CSV
"(1,1001)",1472807115
"(1,1002)",1472807215
"(1,1003)",1472807315
\.


set citus.subquery_pushdown TO on;       

SELECT
    avg(array_length(events, 1)) AS event_average,
    hasdone
FROM
    (SELECT
        subquery_1.tenant_id,
        subquery_1.user_id,
        array_agg(event ORDER BY event_time) AS events,
        COALESCE(hasdone, 'Has not done paying') AS hasdone
    FROM
    (
        (SELECT
            (users.composite_id).tenant_id,
            (users.composite_id).user_id,
            (users.composite_id),
            'action=>1'AS event,
            events.event_time
        FROM
            users,
            events
        WHERE
            (users.composite_id) = (events.composite_id) AND
            users.composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
            users.composite_id <= '(1, 9223372036854775807)'::user_composite_type AND
            event_type = 'click')
        UNION
        (SELECT
            (users.composite_id).tenant_id,
            (users.composite_id).user_id,
            (users.composite_id),
            'action=>2'AS event,
            events.event_time
        FROM
            users,
            events
        WHERE
            (users.composite_id) = (events.composite_id) AND
            users.composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
            users.composite_id <= '(1, 9223372036854775807)'::user_composite_type AND
            event_type = 'submit')
    ) AS subquery_1
    LEFT JOIN
    (SELECT
        composite_id,
        (composite_id).tenant_id,
        (composite_id).user_id,
        'Has done paying'::TEXT AS hasdone
    FROM
        events
    WHERE
        events.composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
        events.composite_id <= '(1, 9223372036854775807)'::user_composite_type AND
        event_type = 'pay') AS subquery_2
    ON
        subquery_1.composite_id = subquery_2.composite_id 
    GROUP BY
        subquery_1.tenant_id,
        subquery_1.user_id,
        hasdone) AS subquery_top
GROUP BY
    hasdone;


SELECT
    avg(array_length(events, 1)) AS event_average,
    hasdone
FROM
    (SELECT
        subquery_1.tenant_id,
        subquery_1.user_id,
        array_agg(event ORDER BY event_time) AS events,
        COALESCE(hasdone, 'Has not done paying') AS hasdone
    FROM
    (
        (SELECT
            (users.composite_id).tenant_id,
            (users.composite_id).user_id,
            (users.composite_id), 
            'action=>1'AS event,
            events.event_time
        FROM
            users,
            events
        WHERE
            (users.composite_id) = (events.composite_id) AND
            users.composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
            users.composite_id <= '(1, 9223372036854775807)'::user_composite_type AND
            event_type = 'click')
        UNION
        (SELECT
            (users.composite_id).tenant_id,
            (users.composite_id).user_id,
            (users.composite_id), 
            'action=>2'AS event,
            events.event_time
        FROM
            users,
            events
        WHERE
            (users.composite_id) = (events.composite_id) AND
            users.composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
            users.composite_id <= '(1, 9223372036854775807)'::user_composite_type AND
            event_type = 'submit')
    ) AS subquery_1
    LEFT JOIN
    (SELECT
         composite_id,
        (composite_id).tenant_id,
        (composite_id).user_id,
        'Has done paying'::TEXT AS hasdone
    FROM
        events
    WHERE
        events.composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
        events.composite_id <= '(1, 9223372036854775807)'::user_composite_type AND
        event_type = 'pay') AS subquery_2
    ON
        subquery_1.composite_id = subquery_2.composite_id
    GROUP BY
        subquery_1.tenant_id,
        subquery_1.user_id,
        hasdone) AS subquery_top
GROUP BY
    hasdone;


-- Lateral join subquery pushdown
SELECT
    tenant_id,
    user_id,
    user_lastseen,
    event_array
FROM
    (SELECT
        tenant_id,
        user_id,
        max(lastseen) as user_lastseen,
        array_agg(event_type ORDER BY event_time) AS event_array
    FROM
        (SELECT
            (composite_id).tenant_id,
            (composite_id).user_id,
            composite_id,
            lastseen
        FROM
            users
        WHERE
            composite_id >= '(1, -9223372036854775808)'::user_composite_type AND
            composite_id <= '(1, 9223372036854775807)'::user_composite_type
        ORDER BY
            lastseen DESC
        LIMIT
            10
        ) AS subquery_top
        LEFT JOIN LATERAL
            (SELECT
                event_type,
                event_time
            FROM
                events
            WHERE
                (composite_id) = subquery_top.composite_id
            ORDER BY
                event_time DESC
            LIMIT
                99) AS subquery_lateral
        ON
            true
        GROUP BY
            tenant_id,
            user_id
    ) AS shard_union
ORDER BY
    user_lastseen DESC
LIMIT
    10;