Realtime Compute for Apache Flink:Window aggregation

SESSION window TVF syntax: VVR 11.x vs VVR 8.x

The SESSION window TVF syntax differs between VVR versions. Upgrade to VVR 11.1 or later to use the full-featured syntax.

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

SESSION(TABLE data, DESCRIPTOR(timecol), gap)

-- The Bid table schema (used in all examples below)
> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

-- VVR 11.x: partition field declared explicitly in SESSION()
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

-- VVR 8.x: partition field inferred from GROUP BY
> SELECT window_start, window_end, item, SUM(price) AS total_price
  FROM TABLE(
      SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
  GROUP BY item, window_start, window_end;

When window TVF aggregation falls back to non-TVF mode

When a query includes a window TVF but does not meet the conditions for the TVF and the aggregation to be merged, the system falls back to a non-TVF execution plan.

The window TVF and the aggregation statement cannot be merged when any of the following conditions is met:

1. Filtering or computation on window time fields. window_start, window_end, or window_time is filtered or modified before aggregation.

   -- Filtering on window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
       WHERE window_start >= TIMESTAMP '2020-04-15 08:06:00.000')
       GROUP BY item, window_start, window_end;
   
   -- Arithmetic on window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, window_start + (INTERVAL '1' SECOND) AS window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
       GROUP BY item, window_start, window_end;
   
   -- Type casting on window_start
   > SELECT window_start, window_end, item, SUM(price) AS total_price
       FROM
       (SELECT item, price, CAST(window_start AS varchar) AS window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)))
       GROUP BY item, window_start, window_end;

2. A window TVF is used with a user-defined table-valued function (UDTF).

   > SELECT window_start, window_end, category, SUM(price) AS total_price
       FROM
       (SELECT category, price, window_start, window_end FROM
       TABLE(
       SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)),
       LATERAL TABLE(category_udtf(item)) AS T(category))
       GROUP BY category, window_start, window_end;

3. The `GROUP BY` clause is missing `window_start` or `window_end`.

   > SELECT window_start, item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY item, window_start;

4. A Python user-defined aggregate function (UDAF) is used.

5. `GROUPING SETS`, `CUBE`, or `ROLLUP` groups separately by `window_start` or `window_end`.

   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY GROUPING SETS((item), (window_start), (window_end));
   
   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY CUBE (item, window_start, window_end);
   
   > SELECT item, SUM(price) AS total_price
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY ROLLUP (item, window_start, window_end);

6. An aggregate function is applied to `window_start`, `window_end`, or `window_time`.

   > SELECT window_start, window_end, item, SUM(price) AS total_price, MAX(window_end) AS max_end
     FROM TABLE(
         SESSION(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))
     GROUP BY item, window_start, window_end;