[Python] Simple Flink operation that looks to be impossible in Table API

table_env.execute_sql(
f"""
CREATE TEMPORARY VIEW cod_last_pre_window AS
SELECT DISTINCT
name,
FIRST_VALUE(CASE WHEN cod_out IS NOT NULL THEN ts END) OVER w AS cod_out_last_ts,
FIRST_VALUE(CASE WHEN cod_in IS NOT NULL THEN ts END) OVER w AS cod_in_last_ts
FROM cod_late_data_cleaned
WHERE data_arrival_ts < (recalculated_ts - INTERVAL '{MINUTES_BETWEEN_DAGSTER_TRIGGER}' MINUTE)
WINDOW w AS (
PARTITION BY name
ORDER BY ts DESC
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
)
"""
)

table_env.execute_sql(
f"""
CREATE TEMPORARY VIEW cod_last_values AS
SELECT
name,
LAST_VALUE(cod_in) OVER w AS cod_in_last,
LAST_VALUE(cod_out) OVER w AS cod_out_last,
ts,
recalculated_ts,
data_arrival_ts
FROM cod_late_data_cleaned
WHERE data_arrival_ts >= (recalculated_ts - INTERVAL '{MINUTES_BETWEEN_DAGSTER_TRIGGER}' MINUTE)
OR ts IN (SELECT cod_out_last_ts FROM cod_last_pre_window WHERE cod_last_pre_window.name = cod_late_data_cleaned.name)
OR ts IN (SELECT cod_in_last_ts FROM cod_last_pre_window WHERE cod_last_pre_window.name = cod_late_data_cleaned.name)
WINDOW w AS (
PARTITION BY name
ORDER BY ts
ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
)
"""
)


I have tried many ways of implementing the following functionality, but I feel like it might be impossible in Apache Flink?

The calculation: Get the part of the data that matches the expression: data_arrival_ts >= (recalculated_ts - INTERVAL '{MINUTES_BETWEEN_DAGSTER_TRIGGER}' MINUTE), combined with the data containing the last cod_in & cod_out values before the start of this window.

Then for this complete window (last values before window + window) calculate last values.

Is this impossible? I thought my approach above would work, because all you feed into cod_last_values is a subset of the original watermarked set, yet flink will say the timestamps are not equal (one is rowtime watermark & other not (casting does not seem to fix this?)

Bit new to flink, don't know the exact limits yet, any help is appreciated

The code above just uses a subset of the watermarked input data for the last_value calculation, but even this approach gives me a TIMESTAMP(3) -> TIMESTAMP(3) ROWTIME error.

(The calculation works & is very simple without including the last cod_values before the start of the window, this seems to be the impossible part)

Continue reading...