FlinkStreamSQL的insert时不支持union all函数

Comienza en SideSqlExec.java, especialmente en joinFun, y luego sigue la llamada desde ExecuteProcessHelper.java que se muestra en el stack trace. Reproduce la consulta INSERT proporcionada con su UNION ALL e inspecciona por qué el nombre de la vista temporal se crea más de una vez; se considera terminado cuando la consulta ya no falla con el error de validación de tabla temporal duplicada.

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)