FlinkStreamSQL的insert时不支持union all函数

Commencez par SideSqlExec.java, en particulier joinFun, puis suivez l’appel depuis ExecuteProcessHelper.java indiqué dans la stack trace. Reproduisez la requête INSERT fournie avec son UNION ALL et examinez pourquoi le nom de la vue temporaire est créé plus d’une fois ; c’est terminé lorsque la requête n’échoue plus avec l’erreur de validation de table temporaire en double.

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)