You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Guillermo Ortiz Fernández <gu...@gmail.com> on 2025/03/12 07:25:40 UTC
Re: FlinkSQL LAG, strange behavior?
Is it possible to add an extra field in the ORDER BY, or for multiple
records with the same value in the ORDER BY to return their own individual
value?
I have checked the LAG code, and the problem is that when multiple records
have the exact same value in the ORDER BY, they are processed as a single
block, resulting in a "curious" behavior.
El jue, 20 feb 2025 a las 15:56, Guillermo Ortiz Fernández (<
guillermo.ortiz.fdez@gmail.com>) escribió:
> And another option I tried it's:
>
> WITH ranked AS (
> select *
> FROM (
> SELECT
> msisdn,
> eventTimestamp,
> zoneIds,
> ts,
> ROW_NUMBER() OVER (PARTITION BY msisdn,ts ORDER BY ts) AS rownum
> FROM example)
> WHERE rownum = 1
> )
> SELECT
> msisdn,
> eventTimestamp,
> ARRAY_REMOVE(IFNULL(zoneIds, ARRAY[-1]), -1) AS zoneIds,
> ARRAY_REMOVE(IFNULL(LAG(zoneIds, 1) OVER (PARTITION BY msisdn ORDER BY
> ts), ARRAY[-1]), -1) AS prev_zoneIds,
> ts
> FROM ranked;
>
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.table.api.TableException: StreamPhysicalOverAggregate
> doesn't support consuming update and delete changes which is produced by
> node Deduplicate(keep=[FirstRow], key=[msisdn, ts], order=[ROWTIME])
>
> El jue, 20 feb 2025 a las 12:00, Guillermo Ortiz Fernández (<
> guillermo.ortiz.fdez@gmail.com>) escribió:
>
>> Another option would be add an extra field like:
>>
>> ARRAY_REMOVE(IFNULL(LAG(zoneIds, 1) OVER (PARTITION BY msisdn ORDER
>> BY *ts, row_number*), ARRAY[-1]), -1) AS prev_zoneIds,
>>
>> But it isn't possible either.
>>
>> El jue, 20 feb 2025 a las 11:48, Guillermo Ortiz Fernández (<
>> guillermo.ortiz.fdez@gmail.com>) escribió:
>>
>>> I tried to create a additional field based on ROW_NUMBER but it can't be
>>> used in LAG function.
>>>
>>> The idea was:
>>> WITH ranked AS (
>>> SELECT
>>> msisdn,
>>> eventTimestamp,
>>> zoneIds,
>>> ts,
>>> TO_TIMESTAMP_LTZ(ROW_NUMBER() OVER (PARTITION BY msisdn ORDER BY
>>> ts), 3) AS ts_ranked
>>> FROM example_2
>>> )
>>> SELECT
>>> msisdn,
>>> eventTimestamp,
>>> ARRAY_REMOVE(IFNULL(zoneIds, ARRAY[-1]), -1) AS zoneIds,
>>> ARRAY_REMOVE(IFNULL(LAG(zoneIds, 1) OVER (PARTITION BY msisdn ORDER
>>> BY ts_ranked), ARRAY[-1]), -1) AS prev_zoneIds,
>>> ts
>>> FROM ranked;
>>>
>>> RANKED subselect:
>>> msisdn eventTimestamp zoneIds
>>> ts row_num ts_ranked
>>> 673944959 1739996380000 [1]
>>> 2025-02-19 21:19:40.000 1 1970-01-01 01:00:00.001
>>> 673944959 1739996380000 [1]
>>> 2025-02-19 21:19:40.000 2 1970-01-01 01:00:00.002
>>>
>>> *But [ERROR] Could not execute SQL statement. Reason:*
>>> *org.apache.flink.table.api.TableException: OVER windows' ordering in
>>> stream mode must be defined on a time attribute. *
>>>
>>> I guess just could use "ts" field because if it's the temporal field?
>>>
>>>
>>> El jue, 20 feb 2025 a las 10:41, Alessio Bernesco Làvore (<
>>> alessio.bernesco@gmail.com>) escribió:
>>>
>>>> Hello,
>>>> don't know if it's strictly related, but using a TIMESTAMP_LTZ vs.
>>>> TIMESTAMP ts i had some not understandable behaviors.
>>>>
>>>> Changing the ts to TIMESTAMP (not LTZ) the identical query works as
>>>> expected.
>>>> So, first all, we changed all the parsing to TIMESTAMP to exclude this
>>>> kind of problems.
>>>>
>>>> If someone is interested i can provide the exact query to reproduce the
>>>> problem we figured out.
>>>>
>>>> Greetings,
>>>> Alessio
>>>>
>>>> On Thu, Feb 20, 2025 at 10:31 AM Guillermo Ortiz Fernández <
>>>> guillermo.ortiz.fdez@gmail.com> wrote:
>>>>
>>>>> I have created a table that reads from a Kafka topic. What I want to
>>>>> do is order the data by eventTime and add a new field that represents
>>>>> the previous value using the LAG function.
>>>>>
>>>>> The problem arises when two records have exactly the same eventTime,
>>>>> which produces a "strange" behavior.
>>>>>
>>>>>
>>>>> CREATE TABLE example (
>>>>> eventTimestamp BIGINT NOT NULL,
>>>>> msisdn INT NOT NULL,
>>>>> zoneIds ARRAY<INT NOT NULL> NOT NULL,
>>>>> ts AS TO_TIMESTAMP_LTZ(eventTimestamp, 3),
>>>>> `kafka_offset` BIGINT METADATA FROM 'offset' VIRTUAL,
>>>>> WATERMARK FOR ts AS ts
>>>>> ) WITH (
>>>>> 'connector' = 'kafka',
>>>>> 'topic' = 'example-offset',
>>>>> 'properties.bootstrap.servers' = 'xxxx',
>>>>> 'properties.auto.offset.reset' = 'latest',
>>>>> 'scan.startup.mode' = 'latest-offset',
>>>>> 'key.format' = 'raw',
>>>>> 'key.fields' = 'msisdn',
>>>>> 'value.format' = 'avro',
>>>>> 'value.fields-include' = 'ALL',
>>>>> 'scan.watermark.idle-timeout' = '1000',
>>>>>
>>>>> );
>>>>>
>>>>>
>>>>> INSERT INTO example (eventTimestamp, msisdn, zoneIds)VALUES
>>>>> (1739996380000, 673944959, ARRAY[1]),
>>>>> (1739996380000, 673944959, ARRAY[1]);
>>>>>
>>>>>
>>>>> SELECT
>>>>> msisdn,
>>>>> eventTimestamp,
>>>>> ARRAY_REMOVE(IFNULL(zoneIds, ARRAY[-1]), -1) AS zoneIds,
>>>>> ARRAY_REMOVE(IFNULL(LAG(zoneIds, 1) OVER (PARTITION BY msisdn ORDER BY ts, kafka_offset), ARRAY[-1]), -1) AS prev_zoneIds,
>>>>> tsFROM example;
>>>>>
>>>>> *Actual Result:*
>>>>> msisdneventTimestampzoneIdsprev_zoneIdsts
>>>>> 673944959 1739996380000 [1] [1] 2025-02-19 21:19:40.000
>>>>> 673944959 1739996380000 [1] [1] 2025-02-19 21:19:40.000*Expected
>>>>> Result:*
>>>>> msisdneventTimestampzoneIdsprev_zoneIdsts
>>>>> 673944959 1739996380000 [1] [ ] 2025-02-19 21:19:40.000
>>>>> 673944959 1739996380000 [1] [1] 2025-02-19 21:19:40.000
>>>>> ------------------------------
>>>>> *Is this behavior normal?*
>>>>>
>>>>> I am trying to achieve the expected behavior by including the metadata
>>>>> of the offset in the example table and adding it to the OVER clause
>>>>> in the LAG function. However, it seems that Flink does not allow
>>>>> ordering by more than one column:
>>>>>
>>>>>
>>>>> ARRAY_REMOVE(IFNULL(LAG(zoneIds, 1) OVER (PARTITION BY msisdn ORDER BY ts, kafka_offset), ARRAY[-1]), -1) AS prev_zoneIds,
>>>>>
>>>>> Results in:
>>>>>
>>>>> [ERROR] Could not execute SQL statement. Reason:
>>>>> org.apache.flink.table.api.TableException: The window can only be ordered by a single time column.
>>>>>
>>>>> ------------------------------
>>>>>
>>>>> Would you happen to know how to achieve the expected result?
>>>>>
>>>>