Function 'LATEST_BY_OFFSET' does not accept parameters (TIMESTAMP)

[rmoff]

tl;dr

Please make LATEST_BY_OFFSET support TIMESTAMP type.

Why?

STRINGTOTIMESTAMP is marked as deprecated in favour of PARSE_TIMESTAMP, but if I use PARSE_TIMESTAMP I get a TIMESTAMP type (as expected) which then does not work with existing LATEST_BY_OFFSET statements which did using the previous BIGINT type that STRINGTOTIMESTAMP emits.

>CREATE STREAM CARPARK_EVENTS WITH (VALUE_FORMAT='PROTOBUF') AS
>SELECT PARSE_TIMESTAMP(DATE + ' ' + TIME ,'yyyy-MM-dd HH:mm','Europe/London' ) AS TS,
>       *,
>       (CAST((CAPACITY - EMPTY_PLACES) AS DOUBLE) /
>        CAST(CAPACITY AS DOUBLE)) * 100 AS PCT_FULL,
>       STRUCT("lat" := LATITUDE, "lon":= LONGITUDE) AS "location",
>       'v2/Council/CarParkCurrent' as SOURCE
>  FROM CARPARK_SRC
>  EMIT CHANGES;

>CREATE TABLE CARPARK AS
>SELECT NAME,
>       FORMAT_TIMESTAMP(LATEST_BY_OFFSET(TS),
>                         'yyyy-MM-dd HH:mm:ss','Europe/London') AS LATEST_TS,
>       LATEST_BY_OFFSET(CAPACITY)      AS CAPACITY,
>       LATEST_BY_OFFSET(EMPTY_PLACES)  AS CURRENT_EMPTY_PLACES,
>       LATEST_BY_OFFSET(PCT_FULL)      AS CURRENT_PCT_FULL,
>       AVG(EMPTY_PLACES)               AS AVG_EMPTY_PLACES,
>       LATEST_BY_OFFSET(STATUS)        AS CURRENT_STATUS,
>       LATEST_BY_OFFSET(LATITUDE)      AS LATITUDE,
>       LATEST_BY_OFFSET(LONGITUDE)     AS LONGITUDE,
>       LATEST_BY_OFFSET(DIRECTIONSURL) AS DIRECTIONSURL
>    FROM CARPARK_EVENTS
>    GROUP BY NAME;
Successfully changed local property 'auto.offset.reset' from 'earliest' to 'earliest'.
Could not determine output schema for query due to error: Function 'LATEST_BY_OFFSET' does not accept parameters (TIMESTAMP).

Instead I have to revert to previous code

ksql> CREATE STREAM CARPARK_EVENTS WITH (VALUE_FORMAT='PROTOBUF') AS
>SELECT STRINGTOTIMESTAMP(DATE + ' ' + TIME ,'yyyy-MM-dd HH:mm','Europe/London' ) AS TS,
>       *,
>       (CAST((CAPACITY - EMPTY_PLACES) AS DOUBLE) /
>        CAST(CAPACITY AS DOUBLE)) * 100 AS PCT_FULL,
>       STRUCT("lat" := LATITUDE, "lon":= LONGITUDE) AS "location",
>       'v2/Council/CarParkCurrent' as SOURCE
>  FROM CARPARK_SRC
>  EMIT CHANGES;

>CREATE TABLE CARPARK AS
>SELECT NAME,
>       TIMESTAMPTOSTRING(LATEST_BY_OFFSET(TS),
>                         'yyyy-MM-dd HH:mm:ss','Europe/London') AS LATEST_TS,
>       LATEST_BY_OFFSET(CAPACITY)      AS CAPACITY,
>       LATEST_BY_OFFSET(EMPTY_PLACES)  AS CURRENT_EMPTY_PLACES,
>       LATEST_BY_OFFSET(PCT_FULL)      AS CURRENT_PCT_FULL,
>       AVG(EMPTY_PLACES)               AS AVG_EMPTY_PLACES,
>       LATEST_BY_OFFSET(STATUS)        AS CURRENT_STATUS,
>       LATEST_BY_OFFSET(LATITUDE)      AS LATITUDE,
>       LATEST_BY_OFFSET(LONGITUDE)     AS LONGITUDE,
>       LATEST_BY_OFFSET(DIRECTIONSURL) AS DIRECTIONSURL
>    FROM CARPARK_EVENTS2
>    GROUP BY NAME;
Successfully changed local property 'auto.offset.reset' from 'earliest' to 'earliest'.

 Message
--------------------------------------
 Created query with ID CTAS_CARPARK_7
--------------------------------------

[jzaralim]

Duplicate of #7787