[SQL] Implement session windows #6778

SELECT * FROM TABLE(
  SESSION(TABLE orders, DESCRIPTOR(rowtime), DESCRIPTOR(product), INTERVAL '20' MINUTE));

WITH
-- LAG finds each row's predecessor within its key
lagged AS (
  SELECT *,
         LAG(rowtime) OVER (PARTITION BY product ORDER BY rowtime) AS prev
  FROM orders          
  WHERE rowtime IS NOT NULL
),
-- brk = 1 when the row starts a new session
breaks AS (
  SELECT rowtime, id, product, units,
         CASE WHEN prev IS NULL
                OR rowtime >= prev + INTERVAL '20' MINUTE
              THEN 1 ELSE 0 END AS brk
  FROM lagged
),
-- the running sum of breaks numbers each row's session within its key
sessionized AS (
  SELECT rowtime, id, product, units,
         SUM(brk) OVER (PARTITION BY product ORDER BY rowtime
                        RANGE UNBOUNDED PRECEDING) AS sid
  FROM breaks
),
-- one row per session holding its bounds
bounds AS (
  SELECT product, sid, MIN(rowtime) AS min_ts, MAX(rowtime) AS max_ts
  FROM sessionized
  GROUP BY product, sid
)
-- attach the bounds back to every row of the session
SELECT s.rowtime, s.id, s.product, s.units,
       b.min_ts                        AS window_start,
       b.max_ts + INTERVAL '20' MINUTE AS window_end
FROM sessionized s
JOIN bounds b
  ON  s.product IS NOT DISTINCT FROM b.product
  AND s.sid     IS NOT DISTINCT FROM b.sid;