Temporal Table function: Cannot add expression of different type to set

Question

I am useing the temporal table funciton to join two stream like this, but got this error. The diff between set type and expression type is the type of proctime0, one with NOT NULL

How will different appears, and any ways to solve this?

Exception in thread "main" java.lang.AssertionError: Cannot add expression of different type to set:
set type is RecordType(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" order_id, DECIMAL(32, 2) price, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, TIMESTAMP(3) order_time, TIMESTAMP_LTZ(3) *PROCTIME* NOT NULL proctime, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency0, BIGINT conversion_rate, TIMESTAMP(3) update_time, TIMESTAMP_LTZ(3) *PROCTIME* proctime0) NOT NULL
expression type is RecordType(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" order_id, DECIMAL(32, 2) price, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency, TIMESTAMP(3) order_time, TIMESTAMP_LTZ(3) *PROCTIME* NOT NULL proctime, VARCHAR(2147483647) CHARACTER SET "UTF-16LE" currency0, BIGINT conversion_rate, TIMESTAMP(3) update_time, TIMESTAMP_LTZ(3) *PROCTIME* NOT NULL proctime0) NOT NULL
set is rel#61:LogicalCorrelate.NONE.any.None: 0.[NONE].[NONE](left=HepRelVertex#59,right=HepRelVertex#60,correlation=$cor0,joinType=inner,requiredColumns={4})
expression is LogicalJoin(condition=[__TEMPORAL_JOIN_CONDITION($4, $7, __TEMPORAL_JOIN_CONDITION_PRIMARY_KEY($5))], joinType=[inner])
  LogicalProject(order_id=[$0], price=[$1], currency=[$2], order_time=[$3], proctime=[PROCTIME()])
    LogicalTableScan(table=[[default_catalog, default_database, orders]])
  LogicalProject(currency=[$0], conversion_rate=[$1], update_time=[$2], proctime=[PROCTIME()])
    LogicalTableScan(table=[[default_catalog, default_database, currency_rates]])

Fact Table:

CREATE TABLE `orders` (
    order_id    STRING,
    price       DECIMAL(32,2),
    currency    STRING,
    order_time  TIMESTAMP(3),
    proctime as PROCTIME()
 ) WITH (
    'properties.bootstrap.servers' = '127.0.0.1:9092',
    'properties.group.id' = 'test',
    'scan.topic-partition-discovery.interval' = '10000',
    'connector' = 'kafka',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset', 
    'topic' = 'test1'
  )

Build Table:

CREATE TABLE `currency_rates` (
    currency    STRING,
    conversion_rate BIGINT,
    update_time  TIMESTAMP(3),
    proctime as PROCTIME()
 ) WITH (
    'properties.bootstrap.servers' = '127.0.0.1:9092',
    'properties.group.id' = 'test',
    'scan.topic-partition-discovery.interval' = '10000',
    'connector' = 'kafka',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset', 
    'topic' = 'test3'
  )

The way to generate table function:

TemporalTableFunction table_rate = tEnv.from("currency_rates")
.createTemporalTableFunction("update_time", "currency");

tEnv.registerFunction("rates", table_rate);

Join logic:

SELECT
    order_id,
    price,
    s.currency,
    conversion_rate,
    order_time
 FROM orders AS o,  
 LATERAL TABLE (rates(o.proctime)) AS s
 WHERE o.currency = s.currency

Temporal Table function: Cannot add expression of different type to set

0 Answers