[Python] Having problem to transform time attribute when transorming between stream and sql api

I have a simple Pyflink application where I need a primary key for temporal table joining. Since my source uses avro confluent-format and it has problems with primary keys I use transformation: SQL API -> Stream objects -> SQL API. And here I get a problem explaining it table has time attribute.

sensor_readings_ddl = f"""
CREATE TABLE sensor_readings (
kafka_key_id VARCHAR not null,
...
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
...
)

class TsExtractor(TimestampAssigner):
def extract_timestamp(self, element, record_timestamp):
return element.ts # Extracts timestamp from the `ts` field of the event

sensors_reading_stream = (
tenv.to_data_stream(sensor_readings_tab)
.assign_timestamps_and_watermarks(
WatermarkStrategy
.for_bounded_out_of_orderness(Duration.of_seconds(10)) # Define watermark strategy
.with_timestamp_assigner(TsExtractor()) # Use custom timestamp assigner
)
)

sensors_reading_schema = (Schema.new_builder().
column("kafka_key_id", DataTypes.STRING().not_null()).
...
column("ts", DataTypes.TIMESTAMP(3)).
primary_key("kafka_key_id").
watermark("ts", "ts - INTERVAL '5' SECOND").
build())
sensor_readings_view = tenv.from_data_stream(sensors_reading_stream, sensors_reading_schema)

sensor_readings_view_tab = tenv.create_temporary_view( "sensor_readings_view", sensors_reading_stream)

tumbling_w_sql = """
SELECT
sr.device_id,
das.metric_1,
das.metric_2,
TUMBLE_START(sr.ts, INTERVAL '30' SECONDS) AS window_start,
TUMBLE_END(sr.ts, INTERVAL '30' SECONDS) AS window_end,
SUM(sr.ampere_hour) AS charge_consumed
FROM sensor_readings_view FOR SYSTEM_TIME AS OF sr.ts AS sr
JOIN device_account_stats_view AS das ON sr.device_id = das.device_id
GROUP BY
TUMBLE(sr.ts, INTERVAL '30' SECONDS),
sr.device_id,
das.metric_1,
das.metric_2
"""


The error is:

pyflink.util.exceptions.TableException: >org.apache.flink.table.api.TableException: Window aggregate can only >be defined over a time attribute column, but TIMESTAMP(3) >encountered.
​

Help is appreciated.

Continue reading...