DTStack / DTStack/flinkStreamSQL

FlinkStreamSQL的insert时不支持union all函数

未关闭
#401 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
主要语言
Java
星标
2k
派生
913
PR 合并指标
30 天内没有已合并 PR

描述

insert into MyResult

select
r1.actionNum as actionNum,
r1.IMEI as IMEI,
r1.IMEIList as IMEIList,
r1.IMEIStatus as IMEIStatus,
r1.belongWarehouseId as belongWarehouseId,
r1.belongCustomerId as belongCustomerId,
r1.brand as brand,
r1.product as product,
r1.marketing_name as marketing_name,
r1.belongCustomer as belongCustomer,
r1.belongWarehouse as belongWarehouse
from
(
(
select
'0' as actionNum,
wcs.imei as IMEI,
wcs.imei_list as IMEIList,
'Available' as IMEIStatus,
wcs.warehouse_id as belongWarehouseId,
cw.enterprise_id as belongCustomerId,
sm.brand as brand,
concat(trim(sm.marketing_name), ' ', trim(sm.rom_ram), ' ', trim(sm.color)) as product,
sm.marketing_name as marketName,
'' as belongCustomer,
'' as belongWarehouse
from
t_channel_warehouse_current_stock as wcs
left join t_channel_warehouse as cw on wcs.warehouse_id=cw.id
left join t_enter_sap_mat as sm on wcs.material_id =sm.matnr
where
wcs.source_type in ('6')
and wcs.type in ('INSERT')
)
union all
(
select
'1' as actionNum,
wcs.imei as IMEI,
wcs.imei_list as IMEIList,
'Available' as IMEIStatus,
wcs.warehouse_id as belongWarehouseId,
cw.enterprise_id as belongCustomerId,
sm.brand as brand,
concat(trim(sm.marketing_name), ' ', trim(sm.rom_ram), ' ', trim(sm.color)) as product,
sm.marketing_name as marketName,
sm.rom_ram as rom_ram,
sm.color as color,
'' as belongCustomer,
'' as belongWarehouse
from
t_channel_warehouse_current_stock as wcs
left join t_channel_warehouse as cw on wcs.warehouse_id=cw.id
left join t_enter_sap_mat as sm on wcs.material_id =sm.matnr
where
wcs.source_type in ('6')
and wcs.status in ('1')
and wcs.type in ('UPDATE')
)
) r1

报错:

Exception in thread "main" org.apache.flink.table.api.ValidationException: Temporary table `default_catalog`.`default_database`.`t_channel_warehouse_current_stock_t_channel_warehouse_0` already exists
at org.apache.flink.table.catalog.CatalogManager.lambda$createTemporaryTable$10(CatalogManager.java:472)
at java.util.HashMap.compute(HashMap.java:1197)
at org.apache.flink.table.catalog.CatalogManager.createTemporaryTable(CatalogManager.java:470)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.createTemporaryView(TableEnvironmentImpl.java:282)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.createTemporaryView(TableEnvironmentImpl.java:269)
at com.dtstack.flink.sql.side.SideSqlExec.joinFun(SideSqlExec.java:554)
at com.dtstack.flink.sql.side.SideSqlExec.exec(SideSqlExec.java:175)
at com.dtstack.flink.sql.exec.ExecuteProcessHelper.sqlTranslation(ExecuteProcessHelper.java:232)
at com.dtstack.flink.sql.exec.ExecuteProcessHelper.getStreamExecution(ExecuteProcessHelper.java:170)
at com.dtstack.flink.sql.Main.main(Main.java:41)
at com.dtstack.flink.sql.launcher.LauncherMain.main(LauncherMain.java:140)

贡献指南

这个仓库没有索引到贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。