FlinkStreamSQL的insert时不支持union all函数

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)