请教下大家Flink CDC，flinksql的这个代码要如何修改才能实现动态传参？

请教下大家Flink CDC，flinksql的这个代码要如何修改才能实现动态传参formatted-date这个参数的动态传参?实现临时表 rltest的formatted_date字段动态输出； Table rltest = tableEnv.sqlQuery(
"SELECT " +
", " +
"DATE_FORMAT(create_date, 'yyyy-MM-dd') as formatted_date " +
"FROM temp_table"
);
tableEnv.createTemporaryView("newb_temp_view", rltest);
String formattedDate ="SELECT formatted_date FROM " + rltest ; // 从 rltest 检索
tableEnv.createTemporaryView("selected_create_date",tableEnv.sqlQuery(formattedDate));
Table drrt = tableEnv.sqlQuery(
"SELECT bo.station_id , " +
"ss.station_name, " +
"COUNT(bo.id)/2 AS switch_count, " +
"ROUND(SUM(bov.switch_electric_quantity), 3)/2 AS total_switch_electric_quantity, " +
"ROUND(SUM(bo.pay_amount), 3)/2 AS total_pay_amount, " +
"'(SELECT formatted_date FROM selected_create_date)' AS create_date " +
"FROM border bo " +
"INNER JOIN \n" +
" bs_order_vehicle bov ON bo.id = bov.order_id " +
"INNER JOIN \n" +
" sys_station ss ON bo.station_id = ss.id " +
"WHERE \n" +
" bo.station_id IN (SELECT station_id FROM selected_station_id)" +
" AND DATE_FORMAT(bo.create_date, 'yyyy-MM-dd') IN (SELECT formatted_date FROM selected_create_date) " +
" AND bo.pay_type <> 0 " +
" AND bo.pay_type IS NOT NULL " +
" AND bo.order_status = 3 " +
" AND bo.order_type = 0 " +
" AND bo.del_flag = 0 " +
" GROUP BY bo.station_id, ss.station_name "
);
tableEnv.createTemporaryView("temp_table4", drrt);
Table drt = tableEnv.sqlQuery("SELECT FROM temp_table4");
tableEnv.toChangelogStream(drt).print("SQL");
env.execute();