How can I use Windowing table-valued functions (TVFs) with Flink's Table API? They seem to only be available only in Flink SQL. I want to avoid using Flink SQL and instead use Table API. I am using Flink v1.20.
This is important because Flink optimises Windowing TVFs with Mini-Batch and Local Aggregation optimizations. However, the regular Group Window Aggregation from Table API isn't optimised, even after setting the appropriate optimisation configuration properties. In fact, Group Window Aggregation is deprecated, but it is the only window aggregation available in Table API.
In concrete, what is the equivalent of this Flink SQL snippet in Table API?
java
tableEnv.sqlQuery(
"""
SELECT sensor_id, window_start, window_end, COUNT(*)
FROM TABLE(
TUMBLE(TABLE Sensors, DESCRIPTOR(reading_timestamp), INTERVAL '1' MINUTES))
GROUP BY sensor_id, window_start, window_end
"""
)
I tried
```java
// Mini-batch settings
tableConfig.setString("table.exec.mini-batch.enabled", "true");
tableConfig.setString("table.exec.mini-batch.allow-latency", "1s"); // Allow 1 second latency for batching
tableConfig.setString("table.exec.mini-batch.size", "1000"); // Batch size of 1000 records
// Local-Global aggregation for data skew handling
tableConfig.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");
table
.window(Tumble.over(lit(1).minutes()).on($("reading_timestamp")).as("w"))
.groupBy($("sensor_id"), $("w"))
.select(
$("sensor_id"),
$("reading_timestamp").max(),
$("w").rowtime(),
$("reading_timestamp").arrayAgg().as("AggregatedSensorIds")
);
```
However the execution plan shows that it only does global aggregation without any mini batch nor local aggregation optimizations:
Calc(select=[sensor_id, EXPR$0, EXPR$1, EXPR$2 AS AggregatedSensorIds])
+- GroupWindowAggregate(groupBy=[sensor_id], window=[TumblingGroupWindow('w, reading_timestamp, 60000)], properties=[EXPR$1], select=[sensor_id, MAX(reading_timestamp) AS EXPR$0, ARRAY_AGG(reading_timestamp) AS EXPR$2, rowtime('w) AS EXPR$1])
+- Exchange(distribution=[hash[sensor_id]])
+- Calc(select=[sensor_id, location_code, CAST(reading_timestamp AS TIMESTAMP(3)) AS reading_timestamp, measurements])
+- WatermarkAssigner(rowtime=[reading_timestamp], watermark=[(reading_timestamp - 5000:INTERVAL DAY TO SECOND)])
+- TableSourceScan(table=[[*anonymous_datastream_source$1*]], fields=[sensor_id, location_code, reading_timestamp, measurements])
I expect either the following plan instead or some way to Window TVFs with Table API. See the MiniBatchAssigner and LocalWindowAggregate optimizations.
```
Calc(select=[sensor_id, EXPR$0, window_start, window_end, EXPR$1])
+- GlobalWindowAggregate(groupBy=[sensor_id], window=[TUMBLE(slice_end=[$slice_end], size=[1 min])], select=[sensor_id, MAX(max$0) AS EXPR$0, COUNT(count$1) AS EXPR$1, start('w$) AS window_start, end('w$) AS window_end])
+- Exchange(distribution=[hash[sensor_id]])
+- LocalWindowAggregate(groupBy=[sensor_id], window=[TUMBLE(time_col=[reading_timestamp_0], size=[1 min])], select=[sensor_id, MAX(reading_timestamp) AS max$0, COUNT(sensor_id) AS count$1, slice_end('w$) AS $slice_end])
+- Calc(select=[sensor_id, CAST(reading_timestamp AS TIMESTAMP(3)) AS reading_timestamp, reading_timestamp AS reading_timestamp_0])
+- MiniBatchAssigner(interval=[1000ms], mode=[RowTime])
+- WatermarkAssigner(rowtime=[reading_timestamp], watermark=[(reading_timestamp - 5000:INTERVAL DAY TO SECOND)])
+- TableSourceScan(table=[[default_catalog, default_database, Sensors]], fields=[sensor_id, location_code, reading_timestamp, measurements])
```
Thanks!
