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;

Condition 1: Aggregation phase strategy allows two-phase

table.optimizer.agg-phase-strategy is set to AUTO (default) or TWO_PHASE. Setting it to ONE_PHASE disables two-phase optimization.

Condition 2: Window uses event time

The window must use event time (rowtime). Processing-time windows are not supported.

Condition 3: Window type is not SESSION

TUMBLE, HOP, and CUMULATE window types are supported. SESSION windows do not support two-phase optimization.

Condition 4: All aggregate functions support partial merge

All aggregate functions must support the merge operation. Built-in functions such as SUM, COUNT, MIN, MAX, and AVG are supported. Custom UDAFs must implement the merge() method.

Condition 5: Input stream is insert-only and the window can be converted to TVF form

The following conditions must all be met:

• The input stream is insert-only.

• table.exec.emit.early-fire.enabled is set to false (default).

• table.exec.emit.late-fire.enabled is set to false (default).

• For HOP windows, the window must be aligned (window size is divisible by the slide interval).

Condition 6: Data distribution does not already satisfy partitioning requirements

The input data distribution does not already meet the partitioning requirements for aggregation. If data is already distributed by the partition key, the optimizer determines that no additional pre-aggregation is needed and does not generate a Local Aggregate node.

Parameter reference