我正在使用时态表函数来连接两个类似的流,但遇到了此错误。集合类型和表达式类型之间的差异是proctime0的类型,一个具有NOT NULL
不同会怎么出现,又有什么办法解决这一点呢?
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]])
事实数据表:
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'
)
构建表:
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'
)
生成表函数的方式:
TemporalTableFunction table_rate = tEnv.from("currency_rates")
.createTemporalTableFunction("update_time", "currency");
tEnv.registerFunction("rates", table_rate);
联接逻辑:
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
1条答案
按热度按时间gzszwxb41#
请尝试使用tEnv.createTemporarySystemFunction(“rates”,table_rate)而不是使用.registerFunction()