latest_by_offset does not support DECIMAL

[krisajenkins]

Describe the bug
There is no latest_by_offset implementation for the DECIMAL type. This is a problem if, for example, you want the latest price of something.

To Reproduce

CREATE OR REPLACE STREAM stocks (
  stock_id VARCHAR KEY,
  description VARCHAR,
  price DECIMAL(10,2)
) WITH (
  KAFKA_TOPIC = 'stocks',
  VALUE_FORMAT = 'avro',
  PARTITIONS = 3
);

INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'CFLT', 'Confluent Inc.', 80.02 );
INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'CFLT', 'Confluent Inc.', 81.02 );
INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'NASDAQ', 'NASDAQ 100', 4003.24 );
INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'CFLT', 'Confluent Inc.', 92.02 );
INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'CFLT', 'Confluent Inc.', 83.02 );
INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'NASDAQ', 'NASDAQ 100', 4028.19 );
INSERT INTO stocks ( stock_id, description, price ) VALUES ( 'CFLT', 'Confluent Inc.', 81.02 );

SET 'auto.offset.reset' = 'earliest';
SET 'ksql.query.pull.table.scan.enabled' = 'true';

CREATE OR REPLACE TABLE current_stock_prices AS
  SELECT
    stock_id,
    latest_by_offset( price ) AS price
  FROM stocks
  GROUP BY stock_id;

SELECT *
FROM current_stock_prices;

Expected behavior
I was expecting a table of two prices, CFLT=81.02 and NASDAQ=4028.19.

Actual behaviour

 Message        
----------------
 Stream created 
----------------
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
Successfully changed local property 'ksql.query.pull.table.scan.enabled' to 'true'. Use the UNSET command to revert your change.
Could not determine output schema for query due to error: Function 'LATEST_BY_OFFSET' does not accept parameters (DECIMAL(10, 2)).
Valid alternatives are:
LATEST_BY_OFFSET(INT val, INT latestN, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(BIGINT val, INT latestN)
LATEST_BY_OFFSET(BOOLEAN val)
LATEST_BY_OFFSET(INT val, INT latestN)
LATEST_BY_OFFSET(BIGINT val, INT latestN, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(BOOLEAN val, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(DOUBLE val)
LATEST_BY_OFFSET(DOUBLE val, INT latestN)
LATEST_BY_OFFSET(VARCHAR val, INT latestN, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(INT val)
LATEST_BY_OFFSET(BIGINT val)
LATEST_BY_OFFSET(BOOLEAN val, INT latestN)
LATEST_BY_OFFSET(VARCHAR val)
LATEST_BY_OFFSET(VARCHAR val, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(BIGINT val, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(BOOLEAN val, INT latestN, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(INT val, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(DOUBLE val, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(DOUBLE val, INT latestN, BOOLEAN ignoreNulls)
LATEST_BY_OFFSET(VARCHAR val, INT latestN)
For detailed information on a function run: DESCRIBE FUNCTION <Function-Name>;
Statement: CREATE OR REPLACE TABLE CURRENT_STOCK_PRICES WITH (KAFKA_TOPIC='CURRENT_STOCK_PRICES', PARTITIONS=3, REPLICAS=1) AS SELECT
  STOCKS.STOCK_ID STOCK_ID,
  LATEST_BY_OFFSET(STOCKS.PRICE) PRICE
FROM STOCKS STOCKS
GROUP BY STOCKS.STOCK_ID
EMIT CHANGES;
Exception while preparing statement: CURRENT_STOCK_PRICES does not exist.
Statement: SELECT *
FROM current_stock_prices;
Caused by: CURRENT_STOCK_PRICES does not exist.

Additional context
This is running against ksql 7.0.0.

[krisajenkins]

As a workaround you can cast(price as double), but it loses precision so it's not perfect.

[dttouchdata]

Any update or workarounds ?

[krisajenkins]

@dttouchdata - one workaround is to use a source table to pick up the latest rows. I did this:

CREATE SOURCE TABLE latest_price (
  stock_id VARCHAR PRIMARY KEY,
  description VARCHAR,
  price DECIMAL(10,2)
) WITH (
  KAFKA_TOPIC = 'stocks',
  VALUE_FORMAT = 'avro'
);

SELECT * FROM latest_price;

...and that gave me the results I was expecting. Whether that will work for you depends on your use-case, but it's another option to explore.

[jnh5y]

#8878 should close this out!