1.10streamSQLkafka关联mysql维表，出现的问题

我的sql
`INSERT INTO dwd_business_flow
SELECT
SUBSTRING(CAST(TIMESTAMPADD(DAY,0,LOCALTIMESTAMP) as varchar),1,10) AS stat_date,
CASE WHEN t2.fee_type = 'WAGES' THEN '010'
WHEN t2.fee_type = 'RENT' THEN '011'
WHEN t2.fee_type = 'DEBIT' THEN '012'
WHEN t2.fee_type = 'MEAL' THEN '013'
WHEN t2.fee_type = 'REIMBURSE' THEN '015'
WHEN t2.fee_type = 'WELFARE' THEN '014'
WHEN t2.fee_type = 'RENTBICYLE' THEN '016'
WHEN t2.fee_type = 'RENTCLOTHES'THEN '017'
WHEN t2.fee_type = 'RENTMOBILE' THEN '019'
WHEN t2.fee_type = 'RENTBETTRY' THEN '018'
WHEN t2.fee_type = 'RENTCAR' THEN '020'
WHEN t2.fee_type = 'LABOUR' THEN '021'
WHEN t2.fee_type = 'YEARENDREWARD' THEN '022'
WHEN t2.fee_type = 'OTHER' THEN '023'
WHEN t2.fee_type = 'SERVICE' THEN '024'
ELSE '010' END AS business_type_code, -- 业务类型编号
t2.fee_name AS business_type_name, -- 业务类型名称
t1.order_id AS order_code, -- 订单号
cast(t1.order_detail_no as varchar) AS child_order_code, -- 子订单号
t1.deal_apply_time AS order_create_time, -- 订单创建时间
t1.busi_complete_data AS order_pay_time, -- 订单支付时间
t1.sal_should_pay AS pay_amount, -- 应发金额
t1.sal_real_pay AS pay_amount_real, -- 实发金额
'2' AS trade_state, -- 交易状态(1-未完成，2-已完成)
t3.user_code AS user_code, -- 用户编号
t1.user_name, -- 用户名
t1.id_number AS identity_no, -- 身份证号码
t1.mobile, -- 手机号码
t2.corp_no AS company_code, -- 企业编号
t3.company_name AS company_name, -- 企业名称
'1' AS pay_way, -- 支付渠道(1-零钱支付 2-积分支付 3-支付宝支付 4-微信支付)
'2' AS trade_type, -- 交易类型(1-个人费用，2-因公费用)
'2' AS trade_flow_type, -- 流水类型(1-支出，2-收入)
t1.user_info_id AS employee_code, -- 员工主键
t2.relea_month AS sal8_real_month, -- 实际薪酬年月
t1.wallet_acc_id AS trade_account_code, -- 账户编码
case when t4.acct_type = 'DIRECT_RELEASE' then '直发账户'
when t4.acct_type = 'WALLET_CUP_NEW' then '好易联账户'
when t4.acct_type = 'WALLET_ZSY' then '中顺易账户'
when t4.acct_type = 'WALLET_LIMIT' then '额度账户'
when t4.acct_type = 'WALLET_FL_PDS' then '华夏PDS账户'
else '全部账户' end AS trade_account_name,
t1.remark AS remark_,
t3.zt_user_code AS user_open_id, -- 用户id
t3.zt_employee_code AS staff_open_id, -- 员工id
t3.zt_company_code AS company_open_id, -- 企业id
t3.company_short_name AS company_short_name
FROM (select
t1.enable_status,
t1.busi_status,
t1.busi_complete_data,
t1.order_id,
t1.wallet_acc_id,
t1.id_number,
t1.user_info_id,
t1.order_detail_no,
t1.deal_apply_time,
t1.busi_complete_data,
t1.sal_should_pay,
t1.sal_real_pay,
t1.user_name,
t1.mobile,
t1.remark
from to_adw_sal_orders_detail t1
WHERE t1.enable_status = '1'
and t1.busi_status = '1'
and t1.busi_complete_data >= SUBSTRING(CAST(TIMESTAMPADD(DAY,0,LOCALTIMESTAMP) as varchar),1,10)
) t1
JOIN to_adw_sal_orders t2 ON t1.order_id = t2.busi_proc_id
LEFT JOIN (select g.acct_type,g.acct_number from to_adw_sal_orders_detail e left join to_acct_person_account_info g on e.wallet_acc_id = g.acct_number where g.enable_ = '1') t4 ON t1.wallet_acc_id = t4.acct_number
LEFT JOIN (select
g.client_id,
g.identity_no,
g.employee_code,
g.user_code,
g.company_name,
g.zt_user_code,
g.zt_employee_code,
g.zt_company_code,
g.company_short_name
from to_adw_sal_orders_detail t1 left join dim_zt_user_info g on t1.id_number = g.identity_no where g.client_id = '1ec595e8ba7211e99c7e7cd30ad3a6a8') t3 ON t1.id_number = t3.identity_no and t1.user_info_id = t3.employee_code`

报错：
`Exception in thread "main" org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: org.apache.calcite.sql.SqlJoin cannot be cast to org.apache.calcite.sql.SqlBasicCall
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:335)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205)
at org.apache.flink.client.program.OptimizerPlanEnvironment.getPipeline(OptimizerPlanEnvironment.java:79)
at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:101)
at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:56)
at org.apache.flink.client.program.PackagedProgramUtils.createJobGraph(PackagedProgramUtils.java:85)
at com.dtstack.flink.sql.launcher.utils.JobGraphBuildUtil.buildJobGraph(JobGraphBuildUtil.java:80)
at com.dtstack.flink.sql.launcher.executor.YarnJobClusterExecutor.exec(YarnJobClusterExecutor.java:65)
at com.dtstack.flink.sql.launcher.LauncherMain.main(LauncherMain.java:144)
Caused by: java.lang.ClassCastException: org.apache.calcite.sql.SqlJoin cannot be cast to org.apache.calcite.sql.SqlBasicCall
at com.dtstack.flink.sql.side.JoinNodeDealer.dealJoinNode(JoinNodeDealer.java:198)
at com.dtstack.flink.sql.side.SideSQLParser.parseSql(SideSQLParser.java:164)
at com.dtstack.flink.sql.side.SideSQLParser.parseSql(SideSQLParser.java:144)
at com.dtstack.flink.sql.side.SideSQLParser.parseSql(SideSQLParser.java:137)
at com.dtstack.flink.sql.side.SideSQLParser.getExeQueue(SideSQLParser.java:72)
at com.dtstack.flink.sql.side.SideSqlExec.exec(SideSqlExec.java:122)
at com.dtstack.flink.sql.exec.ExecuteProcessHelper.sqlTranslation(ExecuteProcessHelper.java:223)
at com.dtstack.flink.sql.exec.ExecuteProcessHelper.getStreamExecution(ExecuteProcessHelper.java:165)
at com.dtstack.flink.sql.Main.main(Main.java:42)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:321)
... 8 more`，