[ https://issues.apache.org/jira/browse/FLINK-27922?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
ASF GitHub Bot updated FLINK-27922: ----------------------------------- Labels: pull-request-available (was: ) > Flink SQL unique key lost when set table.exec.mini-batch.enabled=true > --------------------------------------------------------------------- > > Key: FLINK-27922 > URL: https://issues.apache.org/jira/browse/FLINK-27922 > Project: Flink > Issue Type: Bug > Components: Table SQL / Planner > Affects Versions: 1.12.2, 1.15.0 > Environment: Flink1.12.2 > Reporter: zhangbin > Priority: Major > Labels: pull-request-available > > Flink SQL table has primary keys, but when set table.exec.mini-batch.enabled > =true, the unique key is lost. > {code:java} > @Test > public void testJoinUniqueKey() throws Exception { > StreamExecutionEnvironment env = > StreamExecutionEnvironment.getExecutionEnvironment(); > StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); > Configuration configuration = tableEnv.getConfig().getConfiguration(); > configuration.setString("table.exec.mini-batch.enabled", "true"); > configuration.setString("table.exec.mini-batch.allow-latency", "5 s"); > configuration.setString("table.exec.mini-batch.size", "5000"); > StatementSet statementSet = tableEnv.createStatementSet(); > tableEnv.executeSql("CREATE TABLE `t_apply_sku_test`(`dt` > BIGINT,`refund_apply_id` BIGINT,`base_sku_id` BIGINT,`order_id` > BIGINT,`user_id` BIGINT,`poi_id` BIGINT,`refund_type` > BIGINT,`apply_refund_reason_code` BIGINT,`apply_refund_reason_desc` > VARCHAR,`apply_refund_review_status` BIGINT,`apply_refund_review_status_desc` > VARCHAR,`apply_refund_reject_reason` VARCHAR,`apply_is_refunded` > INTEGER,`apply_pic_url` VARCHAR,`remark` VARCHAR,`refund_apply_originator` > BIGINT,`second_reason_code` BIGINT,`second_reason` > VARCHAR,`refund_target_account` BIGINT,`after_service_id` > BIGINT,`receipt_status` BIGINT,`group_header_goods_status` > INTEGER,`apply_operator_mis_name` VARCHAR,`refund_apply_time` > BIGINT,`update_time` BIGINT,`base_sku_name` VARCHAR,`apply_refund_num` > BIGINT,`view_qty` DECIMAL(38,18),`refund_scale_type` > BIGINT,`refund_scale_type_desc` VARCHAR,`refund_scale` > DECIMAL(38,18),`apply_refund_amt` DECIMAL(38,18),`refund_scale_user_real_pay` > DECIMAL(38,18),`refund_red_packet_price` DECIMAL(38,18),`load_time` > VARCHAR,`take_rate_type` BIGINT,`platform_rate` > DECIMAL(38,18),`order_sku_type` INTEGER,`second_reason_aggregated_code` > INTEGER,`second_reason_aggregated` VARCHAR,`compensation_amount` > DECIMAL(38,18),`aftersale_type` INTEGER,`group_header_parallel_status` > INTEGER,`grid_parallel_status` INTEGER) WITH ('connector'='blackhole')"); > tableEnv.executeSql("CREATE TABLE `t_name`(`id` BIGINT,`after_service_id` > BIGINT PRIMARY KEY NOT ENFORCED,`order_id` BIGINT,`user_id` BIGINT,`poi_id` > BIGINT,`city_id` BIGINT,`refund_type` INTEGER,`first_reason_code` > INTEGER,`first_reason` VARCHAR,`second_reason_code` INTEGER,`second_reason` > VARCHAR,`pic_url` VARCHAR,`remark` VARCHAR,`refund_price` > INTEGER,`refund_red_packet_price` INTEGER,`refund_total_price` > INTEGER,`refund_promotion_price` INTEGER,`refund_coupon_price` > INTEGER,`refund_other_price` INTEGER,`user_receipt_status` > INTEGER,`collect_status` INTEGER,`refund_target_account` INTEGER,`status` > INTEGER,`flow_instance_id` BIGINT,`create_time` BIGINT,`modify_time` BIGINT) > WITH ('connector'='datagen')"); > tableEnv.executeSql("CREATE TABLE `t_item_name`(`id` > BIGINT,`refund_fwd_item_id` BIGINT,`after_service_id` BIGINT,`order_id` > BIGINT,`order_item_id` BIGINT,`stack_id` BIGINT,`sku_id` BIGINT,`sku_name` > VARCHAR,`supplier_id` BIGINT,`refund_quantity` INTEGER,`item_sku_type` > INTEGER,`refund_scale_type` INTEGER,`refund_scale` INTEGER,`accurate_refund` > INTEGER,`refund_price` INTEGER,`refund_red_packet_price` > INTEGER,`refund_price_info` VARCHAR,`refund_total_price` > INTEGER,`refund_promotion_price` INTEGER,`refund_coupon_price` > INTEGER,`refund_other_price` INTEGER,`extend_info` VARCHAR,`create_time` > BIGINT,`modify_time` BIGINT) WITH ('connector'='datagen')"); > tableEnv.executeSql("CREATE TABLE `t_progress_name`(`id` > BIGINT,`after_service_id` BIGINT PRIMARY KEY NOT ENFORCED,`order_id` > BIGINT,`progress_node` VARCHAR,`progress_node_status` INTEGER,`operator` > VARCHAR,`parallel` INTEGER,`flow_element_id` VARCHAR,`extend_info` > VARCHAR,`create_time` BIGINT,`modify_time` BIGINT) WITH > ('connector'='datagen')"); > tableEnv.executeSql("CREATE TABLE `t_attr_name`(`id` > BIGINT,`after_service_id` BIGINT PRIMARY KEY NOT ENFORCED,`order_id` > BIGINT,`name` VARCHAR,`value` VARCHAR,`create_time` BIGINT,`modify_time` > BIGINT) WITH ('connector'='datagen')"); > tableEnv.executeSql("CREATE TABLE `t_apply_status_name`(`id` > BIGINT,`after_service_id` BIGINT PRIMARY KEY NOT ENFORCED,`order_id` > BIGINT,`apply_status` INTEGER,`ai_audit_status` > INTEGER,`group_header_confirm_status` INTEGER,`group_header_retrieve_status` > INTEGER,`driver_retrieve_status` INTEGER,`group_header_parallel_status` > INTEGER,`grid_parallel_status` INTEGER,`create_time` BIGINT,`modify_time` > BIGINT) WITH ('connector'='datagen')"); > tableEnv.executeSql("CREATE VIEW `v_refund_fwd_item_sku` AS SELECT > `after_service_id` AS `refund_apply_id`, `order_id`, `sku_id` AS > `base_sku_id`, SUM(`if`(`refund_quantity` IS NULL, 0, `refund_quantity`)) AS > `apply_refund_num`, SUM(`if`(`refund_quantity` IS NULL, 0, > CAST(`refund_quantity` AS DOUBLE))) AS `view_qty`, SUM(`if`(`refund_price` IS > NULL, 0, `refund_price`)) AS `apply_refund_amt`, SUM(`if`(`refund_price` IS > NULL, 0, `refund_price`)) AS `refund_scale_user_real_pay`, > SUM(`if`(`refund_red_packet_price` IS NULL, 0, `refund_red_packet_price`)) AS > `refund_red_packet_price`\n" > + "FROM `t_item_name`\n" > + "GROUP BY `after_service_id`, `order_id`, `sku_id`"); > statementSet.addInsertSql("INSERT INTO `t_apply_sku_test` SELECT > CAST(`FROM_UNIXTIME`(`ord`.`modify_time` / 1000, 'yyyyMMdd') AS BIGINT) AS > `dt`, `sku`.`refund_apply_id`,`sku`.`base_sku_id`, `sku`.`order_id`, > `ord`.`user_id`, `ord`.`poi_id`, CAST(`ord`.`refund_type` AS BIGINT) AS > `refund_type`, CAST(`ord`.`first_reason_code` AS BIGINT) AS > `apply_refund_reason_code`, `ord`.`first_reason` AS > `apply_refund_reason_desc`, CAST(`stat`.`apply_status` AS BIGINT) AS > `apply_refund_review_status`, CASE WHEN `stat`.`apply_status` = 0 OR > `stat`.`apply_status` = 13 THEN 'a' WHEN `stat`.`apply_status` = 1 THEN 'b' > WHEN `stat`.`apply_status` = 2 THEN 'c' WHEN `stat`.`apply_status` = 3 THEN > 'd' WHEN `stat`.`apply_status` = 4 THEN 'e' WHEN `stat`.`apply_status` = 5 > THEN 'f' WHEN `stat`.`apply_status` = 6 THEN 'g' ELSE 'x' END AS > `apply_refund_review_status_desc`, `if`(`progress`.`progress_node_status` = > 30, `extend_info`, '') AS `apply_refund_reject_reason`, CASE WHEN > `progress`.`progress_node` = 'refund.node' THEN 1 ELSE 0 END AS > `apply_is_refunded`, `ord`.`pic_url` AS `apply_pic_url`, `ord`.`remark`, > CAST(NULL AS BIGINT) AS `refund_apply_originator`, > CAST(`ord`.`second_reason_code` AS BIGINT) AS `second_reason_code`, > `ord`.`second_reason`, CAST(`ord`.`refund_target_account` AS BIGINT) AS > `refund_target_account`, `ord`.`after_service_id`, > CAST(`ord`.`user_receipt_status` AS BIGINT) AS `receipt_status`, CASE WHEN > `attr`.`name` = 'fwd_ext' AND `attr`.`value` = '1' THEN 1 WHEN `attr`.`name` > = 'fwd_ext' AND `attr`.`value` = '2' THEN 2 ELSE 0 END AS > `group_header_goods_status`, `progress`.`operator` AS > `apply_operator_mis_name`, `ord`.`create_time` AS `refund_apply_time`, > `ord`.`modify_time` AS `update_time`, CAST(NULL AS VARCHAR) AS > `base_sku_name`, CAST(`sku`.`apply_refund_num` AS BIGINT) AS > `apply_refund_num`, `sku`.`view_qty` AS `view_qty`, CAST(NULL AS BIGINT) AS > `refund_scale_type`, CAST(NULL AS VARCHAR) AS `refund_scale_type_desc`, > CAST(NULL AS DECIMAL) AS `refund_scale`, `sku`.`apply_refund_amt` AS > `apply_refund_amt`, `sku`.`refund_scale_user_real_pay` AS > `refund_scale_user_real_pay`, `sku`.`refund_red_packet_price` AS > `refund_red_packet_price`, CAST(LOCALTIMESTAMP AS VARCHAR) AS `load_time`, > CAST(NULL AS BIGINT) AS `take_rate_type`, CAST(NULL AS DECIMAL) AS > `platform_rate`, 1 AS `order_sku_type`, CAST(NULL AS INTEGER) AS > `second_reason_aggregated_code`, CAST(NULL AS VARCHAR) AS > `second_reason_aggregated`, CAST(NULL AS DECIMAL) AS `compensation_amount`, 1 > AS `aftersale_type`, CASE WHEN `progress`.`progress_node` = > 'group.header.audit.node' AND `parallel` = 1 AND `progress_node_status` = 10 > THEN 1 WHEN `progress`.`progress_node` = 'group.header.audit.node' AND > `parallel` = 1 AND `progress_node_status` = 20 THEN 2 WHEN > `progress`.`progress_node` = 'group.header.audit.node' AND `parallel` = 1 AND > `progress_node_status` = 30 THEN 3 WHEN `progress`.`progress_node` = > 'group.header.audit.node' AND `parallel` = 1 AND `progress_node_status` = 40 > THEN 4 WHEN `progress`.`progress_node` = 'group.header.audit.node' AND > `parallel` = 1 AND `progress_node_status` = 50 THEN 5 ELSE 0 END AS > `group_header_parallel_status`, CASE WHEN `progress`.`progress_node` = > 'grid.audit.node' AND `parallel` = 1 AND `progress_node_status` = 10 THEN 1 > WHEN `progress`.`progress_node` = 'grid.audit.node' AND `parallel` = 1 AND > `progress_node_status` = 20 THEN 2 WHEN `progress`.`progress_node` = > 'grid.audit.node' AND `parallel` = 1 AND `progress_node_status` = 30 THEN 3 > WHEN `progress`.`progress_node` = 'grid.audit.node' AND `parallel` = 1 AND > `progress_node_status` = 40 THEN 4 WHEN `progress`.`progress_node` = > 'grid.audit.node' AND `parallel` = 1 AND `progress_node_status` = 50 THEN 5 > ELSE 0 END AS `grid_parallel_status`\n" > + "FROM `t_name` AS `ord`\n" > + "LEFT JOIN `t_progress_name` AS `progress` ON > `ord`.`after_service_id` = `progress`.`after_service_id`\n" > + "LEFT JOIN `t_attr_name` AS `attr` ON `ord`.`after_service_id` > = `attr`.`after_service_id`\n" > + "LEFT JOIN `t_apply_status_name` AS `stat` ON > `ord`.`after_service_id` = `stat`.`after_service_id`\n" > + "LEFT JOIN `v_refund_fwd_item_sku` AS `sku` ON > `stat`.`after_service_id` = `sku`.`refund_apply_id`"); > System.out.println(statementSet.explain()); > }{code} > == Optimized Logical Plan == > {code:java} > Sink(table=[default_catalog.default_database.t_apply_sku_test], fields=[dt, > refund_apply_id, base_sku_id, order_id, user_id, poi_id, refund_type, > apply_refund_reason_code, apply_refund_reason_desc, > apply_refund_review_status, apply_refund_review_status_desc, > apply_refund_reject_reason, apply_is_refunded, apply_pic_url, remark, > refund_apply_originator, second_reason_code, second_reason, > refund_target_account, after_service_id, receipt_status, > group_header_goods_status, apply_operator_mis_name, refund_apply_time, > update_time, base_sku_name, apply_refund_num, view_qty, refund_scale_type, > refund_scale_type_desc, refund_scale, apply_refund_amt, > refund_scale_user_real_pay, refund_red_packet_price, load_time, > take_rate_type, platform_rate, order_sku_type, second_reason_aggregated_code, > second_reason_aggregated, compensation_amount, aftersale_type, > group_header_parallel_status, grid_parallel_status]) > +- Calc(select=[CAST(((modify_time / 1000) FROM_UNIXTIME > _UTF-16LE'yyyyMMdd')) AS dt, refund_apply_id, base_sku_id, order_id, user_id, > poi_id, CAST(refund_type) AS refund_type, CAST(first_reason_code) AS > apply_refund_reason_code, first_reason AS apply_refund_reason_desc, > CAST(apply_status) AS apply_refund_review_status, CAST(((apply_status SEARCH > Sarg[0, 13]) CASE _UTF-16LE'a' CASE (apply_status = 1) CASE _UTF-16LE'b' CASE > (apply_status = 2) CASE _UTF-16LE'c' CASE (apply_status = 3) CASE > _UTF-16LE'd' CASE (apply_status = 4) CASE _UTF-16LE'e' CASE (apply_status = > 5) CASE _UTF-16LE'f' CASE (apply_status = 6) CASE _UTF-16LE'g' CASE > _UTF-16LE'x')) AS apply_refund_review_status_desc, ((progress_node_status = > 30) IF extend_info IF _UTF-16LE'') AS apply_refund_reject_reason, > CAST(((progress_node = _UTF-16LE'refund.node':VARCHAR(2147483647) CHARACTER > SET "UTF-16LE") CASE 1 CASE 0)) AS apply_is_refunded, pic_url AS > apply_pic_url, remark, null:BIGINT AS refund_apply_originator, > CAST(second_reason_code) AS second_reason_code, second_reason, > CAST(refund_target_account) AS refund_target_account, CAST(after_service_id) > AS after_service_id, CAST(user_receipt_status) AS receipt_status, > CAST((((name = _UTF-16LE'fwd_ext':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (value = _UTF-16LE'1':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE")) CASE 1 CASE ((name = _UTF-16LE'fwd_ext':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE") AND (value = _UTF-16LE'2':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE")) CASE 2 CASE 0)) AS group_header_goods_status, > operator AS apply_operator_mis_name, create_time AS refund_apply_time, > modify_time AS update_time, null:VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > AS base_sku_name, CAST(apply_refund_num) AS apply_refund_num, CAST(view_qty) > AS view_qty, null:BIGINT AS refund_scale_type, null:VARCHAR(2147483647) > CHARACTER SET "UTF-16LE" AS refund_scale_type_desc, null:DECIMAL(38, 18) AS > refund_scale, CAST(apply_refund_amt) AS apply_refund_amt, > CAST(refund_scale_user_real_pay) AS refund_scale_user_real_pay, > CAST(refund_red_packet_price) AS refund_red_packet_price, CAST(CAST(())) AS > load_time, null:BIGINT AS take_rate_type, null:DECIMAL(38, 18) AS > platform_rate, 1 AS order_sku_type, null:INTEGER AS > second_reason_aggregated_code, null:VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" AS second_reason_aggregated, null:DECIMAL(38, 18) AS > compensation_amount, 1 AS aftersale_type, CAST((((progress_node = > _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 10)) CASE 1 CASE > ((progress_node = _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 20)) > CASE 2 CASE ((progress_node = > _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 30)) CASE 3 CASE > ((progress_node = _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 40)) > CASE 4 CASE ((progress_node = > _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 50)) CASE 5 CASE > 0)) AS group_header_parallel_status, CAST((((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 10)) CASE 1 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 20)) CASE 2 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 30)) CASE 3 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 40)) CASE 4 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 50)) CASE 5 CASE 0)) AS > grid_parallel_status]) > +- Join(joinType=[LeftOuterJoin], where=[(after_service_id0 = > refund_apply_id)], select=[after_service_id, user_id, poi_id, refund_type, > first_reason_code, first_reason, second_reason_code, second_reason, pic_url, > remark, user_receipt_status, refund_target_account, create_time, modify_time, > progress_node, progress_node_status, operator, parallel, extend_info, name, > value, after_service_id0, apply_status, refund_apply_id, order_id, > base_sku_id, apply_refund_num, view_qty, apply_refund_amt, > refund_scale_user_real_pay, refund_red_packet_price], > leftInputSpec=[NoUniqueKey], rightInputSpec=[HasUniqueKey]) > :- Exchange(distribution=[hash[after_service_id0]]) > : +- Join(joinType=[LeftOuterJoin], where=[(after_service_id = > after_service_id0)], select=[after_service_id, user_id, poi_id, refund_type, > first_reason_code, first_reason, second_reason_code, second_reason, pic_url, > remark, user_receipt_status, refund_target_account, create_time, modify_time, > progress_node, progress_node_status, operator, parallel, extend_info, name, > value, after_service_id0, apply_status], leftInputSpec=[NoUniqueKey], > rightInputSpec=[NoUniqueKey]) > : :- Exchange(distribution=[hash[after_service_id]]) > : : +- Calc(select=[after_service_id, user_id, poi_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time, progress_node, progress_node_status, operator, > parallel, extend_info, name, value]) > : : +- Join(joinType=[LeftOuterJoin], where=[(after_service_id > = after_service_id0)], select=[after_service_id, user_id, poi_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time, progress_node, progress_node_status, operator, > parallel, extend_info, after_service_id0, name, value], > leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey]) > : : :- Exchange(distribution=[hash[after_service_id]]) > : : : +- Calc(select=[after_service_id, user_id, poi_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time, progress_node, progress_node_status, operator, > parallel, extend_info]) > : : : +- Join(joinType=[LeftOuterJoin], > where=[(after_service_id = after_service_id0)], select=[after_service_id, > user_id, poi_id, refund_type, first_reason_code, first_reason, > second_reason_code, second_reason, pic_url, remark, user_receipt_status, > refund_target_account, create_time, modify_time, after_service_id0, > progress_node, progress_node_status, operator, parallel, extend_info], > leftInputSpec=[NoUniqueKey], rightInputSpec=[NoUniqueKey]) > : : : :- > Exchange(distribution=[hash[after_service_id]]) > : : : : +- Calc(select=[after_service_id, user_id, > poi_id, refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time]) > : : : : +- DropUpdateBefore > : : : : +- > MiniBatchAssigner(interval=[5000ms], mode=[ProcTime]) > : : : : +- > TableSourceScan(table=[[default_catalog, default_database, t_name]], > fields=[id, after_service_id, order_id, user_id, poi_id, city_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, refund_price, refund_red_packet_price, > refund_total_price, refund_promotion_price, refund_coupon_price, > refund_other_price, user_receipt_status, collect_status, > refund_target_account, status, flow_instance_id, create_time, modify_time]) > : : : +- > Exchange(distribution=[hash[after_service_id]]) > : : : +- Calc(select=[after_service_id, > progress_node, progress_node_status, operator, parallel, extend_info]) > : : : +- DropUpdateBefore > : : : +- > MiniBatchAssigner(interval=[5000ms], mode=[ProcTime]) > : : : +- > TableSourceScan(table=[[default_catalog, default_database, t_progress_name]], > fields=[id, after_service_id, order_id, progress_node, progress_node_status, > operator, parallel, flow_element_id, extend_info, create_time, modify_time]) > : : +- Exchange(distribution=[hash[after_service_id]]) > : : +- Calc(select=[after_service_id, name, value]) > : : +- DropUpdateBefore > : : +- MiniBatchAssigner(interval=[5000ms], > mode=[ProcTime]) > : : +- TableSourceScan(table=[[default_catalog, > default_database, t_attr_name]], fields=[id, after_service_id, order_id, > name, value, create_time, modify_time]) > : +- Exchange(distribution=[hash[after_service_id]]) > : +- Calc(select=[after_service_id, apply_status]) > : +- DropUpdateBefore > : +- MiniBatchAssigner(interval=[5000ms], mode=[ProcTime]) > : +- TableSourceScan(table=[[default_catalog, > default_database, t_apply_status_name]], fields=[id, after_service_id, > order_id, apply_status, ai_audit_status, group_header_confirm_status, > group_header_retrieve_status, driver_retrieve_status, > group_header_parallel_status, grid_parallel_status, create_time, modify_time]) > +- Exchange(distribution=[hash[refund_apply_id]]) > +- Calc(select=[refund_apply_id, order_id, base_sku_id, > apply_refund_num, view_qty, refund_scale_user_real_pay AS apply_refund_amt, > refund_scale_user_real_pay, refund_red_packet_price]) > +- GlobalGroupAggregate(groupBy=[refund_apply_id, order_id, > base_sku_id], select=[refund_apply_id, order_id, base_sku_id, > SUM_RETRACT((sum$0, count$1)) AS apply_refund_num, SUM_RETRACT((sum$2, > count$3)) AS view_qty, SUM_RETRACT((sum$4, count$5)) AS > refund_scale_user_real_pay, SUM_RETRACT((sum$6, count$7)) AS > refund_red_packet_price]) > +- Exchange(distribution=[hash[refund_apply_id, order_id, > base_sku_id]]) > +- LocalGroupAggregate(groupBy=[refund_apply_id, order_id, > base_sku_id], select=[refund_apply_id, order_id, base_sku_id, > SUM_RETRACT($f3) AS (sum$0, count$1), SUM_RETRACT($f4) AS (sum$2, count$3), > SUM_RETRACT($f5) AS (sum$4, count$5), SUM_RETRACT($f6) AS (sum$6, count$7), > COUNT_RETRACT(*) AS count1$8]) > +- Calc(select=[after_service_id AS refund_apply_id, > order_id, sku_id AS base_sku_id, (refund_quantity IS NULL IF 0 IF > refund_quantity) AS $f3, (refund_quantity IS NULL IF 0 IF > CAST(refund_quantity)) AS $f4, (refund_price IS NULL IF 0 IF refund_price) AS > $f5, (refund_red_packet_price IS NULL IF 0 IF refund_red_packet_price) AS > $f6]) > +- MiniBatchAssigner(interval=[5000ms], > mode=[ProcTime]) > +- TableSourceScan(table=[[default_catalog, > default_database, t_item_name]], fields=[id, refund_fwd_item_id, > after_service_id, order_id, order_item_id, stack_id, sku_id, sku_name, > supplier_id, refund_quantity, item_sku_type, refund_scale_type, refund_scale, > accurate_refund, refund_price, refund_red_packet_price, refund_price_info, > refund_total_price, refund_promotion_price, refund_coupon_price, > refund_other_price, extend_info, create_time, modify_time]) {code} > when set table.exec.mini-batch.enabled=false > == Optimized Logical Plan == > {code:java} > Sink(table=[default_catalog.default_database.t_apply_sku_test], fields=[dt, > refund_apply_id, base_sku_id, order_id, user_id, poi_id, refund_type, > apply_refund_reason_code, apply_refund_reason_desc, > apply_refund_review_status, apply_refund_review_status_desc, > apply_refund_reject_reason, apply_is_refunded, apply_pic_url, remark, > refund_apply_originator, second_reason_code, second_reason, > refund_target_account, after_service_id, receipt_status, > group_header_goods_status, apply_operator_mis_name, refund_apply_time, > update_time, base_sku_name, apply_refund_num, view_qty, refund_scale_type, > refund_scale_type_desc, refund_scale, apply_refund_amt, > refund_scale_user_real_pay, refund_red_packet_price, load_time, > take_rate_type, platform_rate, order_sku_type, second_reason_aggregated_code, > second_reason_aggregated, compensation_amount, aftersale_type, > group_header_parallel_status, grid_parallel_status]) > +- Calc(select=[CAST(((modify_time / 1000) FROM_UNIXTIME > _UTF-16LE'yyyyMMdd')) AS dt, refund_apply_id, base_sku_id, order_id, user_id, > poi_id, CAST(refund_type) AS refund_type, CAST(first_reason_code) AS > apply_refund_reason_code, first_reason AS apply_refund_reason_desc, > CAST(apply_status) AS apply_refund_review_status, CAST(((apply_status SEARCH > Sarg[0, 13]) CASE _UTF-16LE'a' CASE (apply_status = 1) CASE _UTF-16LE'b' CASE > (apply_status = 2) CASE _UTF-16LE'c' CASE (apply_status = 3) CASE > _UTF-16LE'd' CASE (apply_status = 4) CASE _UTF-16LE'e' CASE (apply_status = > 5) CASE _UTF-16LE'f' CASE (apply_status = 6) CASE _UTF-16LE'g' CASE > _UTF-16LE'x')) AS apply_refund_review_status_desc, ((progress_node_status = > 30) IF extend_info IF _UTF-16LE'') AS apply_refund_reject_reason, > CAST(((progress_node = _UTF-16LE'refund.node':VARCHAR(2147483647) CHARACTER > SET "UTF-16LE") CASE 1 CASE 0)) AS apply_is_refunded, pic_url AS > apply_pic_url, remark, null:BIGINT AS refund_apply_originator, > CAST(second_reason_code) AS second_reason_code, second_reason, > CAST(refund_target_account) AS refund_target_account, CAST(after_service_id) > AS after_service_id, CAST(user_receipt_status) AS receipt_status, > CAST((((name = _UTF-16LE'fwd_ext':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (value = _UTF-16LE'1':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE")) CASE 1 CASE ((name = _UTF-16LE'fwd_ext':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE") AND (value = _UTF-16LE'2':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE")) CASE 2 CASE 0)) AS group_header_goods_status, > operator AS apply_operator_mis_name, create_time AS refund_apply_time, > modify_time AS update_time, null:VARCHAR(2147483647) CHARACTER SET "UTF-16LE" > AS base_sku_name, CAST(apply_refund_num) AS apply_refund_num, CAST(view_qty) > AS view_qty, null:BIGINT AS refund_scale_type, null:VARCHAR(2147483647) > CHARACTER SET "UTF-16LE" AS refund_scale_type_desc, null:DECIMAL(38, 18) AS > refund_scale, CAST(apply_refund_amt) AS apply_refund_amt, > CAST(refund_scale_user_real_pay) AS refund_scale_user_real_pay, > CAST(refund_red_packet_price) AS refund_red_packet_price, CAST(CAST(())) AS > load_time, null:BIGINT AS take_rate_type, null:DECIMAL(38, 18) AS > platform_rate, 1 AS order_sku_type, null:INTEGER AS > second_reason_aggregated_code, null:VARCHAR(2147483647) CHARACTER SET > "UTF-16LE" AS second_reason_aggregated, null:DECIMAL(38, 18) AS > compensation_amount, 1 AS aftersale_type, CAST((((progress_node = > _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 10)) CASE 1 CASE > ((progress_node = _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 20)) > CASE 2 CASE ((progress_node = > _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 30)) CASE 3 CASE > ((progress_node = _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) > CHARACTER SET "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 40)) > CASE 4 CASE ((progress_node = > _UTF-16LE'group.header.audit.node':VARCHAR(2147483647) CHARACTER SET > "UTF-16LE") AND (parallel = 1) AND (progress_node_status = 50)) CASE 5 CASE > 0)) AS group_header_parallel_status, CAST((((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 10)) CASE 1 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 20)) CASE 2 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 30)) CASE 3 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 40)) CASE 4 CASE ((progress_node = > _UTF-16LE'grid.audit.node':VARCHAR(2147483647) CHARACTER SET "UTF-16LE") AND > (parallel = 1) AND (progress_node_status = 50)) CASE 5 CASE 0)) AS > grid_parallel_status]) > +- Join(joinType=[LeftOuterJoin], where=[(after_service_id0 = > refund_apply_id)], select=[after_service_id, user_id, poi_id, refund_type, > first_reason_code, first_reason, second_reason_code, second_reason, pic_url, > remark, user_receipt_status, refund_target_account, create_time, modify_time, > progress_node, progress_node_status, operator, parallel, extend_info, name, > value, after_service_id0, apply_status, refund_apply_id, order_id, > base_sku_id, apply_refund_num, view_qty, apply_refund_amt, > refund_scale_user_real_pay, refund_red_packet_price], > leftInputSpec=[HasUniqueKey], rightInputSpec=[HasUniqueKey]) > :- Exchange(distribution=[hash[after_service_id0]]) > : +- Join(joinType=[LeftOuterJoin], where=[(after_service_id = > after_service_id0)], select=[after_service_id, user_id, poi_id, refund_type, > first_reason_code, first_reason, second_reason_code, second_reason, pic_url, > remark, user_receipt_status, refund_target_account, create_time, modify_time, > progress_node, progress_node_status, operator, parallel, extend_info, name, > value, after_service_id0, apply_status], > leftInputSpec=[JoinKeyContainsUniqueKey], > rightInputSpec=[JoinKeyContainsUniqueKey]) > : :- Exchange(distribution=[hash[after_service_id]]) > : : +- Calc(select=[after_service_id, user_id, poi_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time, progress_node, progress_node_status, operator, > parallel, extend_info, name, value]) > : : +- Join(joinType=[LeftOuterJoin], where=[(after_service_id > = after_service_id0)], select=[after_service_id, user_id, poi_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time, progress_node, progress_node_status, operator, > parallel, extend_info, after_service_id0, name, value], > leftInputSpec=[JoinKeyContainsUniqueKey], > rightInputSpec=[JoinKeyContainsUniqueKey]) > : : :- Exchange(distribution=[hash[after_service_id]]) > : : : +- Calc(select=[after_service_id, user_id, poi_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time, progress_node, progress_node_status, operator, > parallel, extend_info]) > : : : +- Join(joinType=[LeftOuterJoin], > where=[(after_service_id = after_service_id0)], select=[after_service_id, > user_id, poi_id, refund_type, first_reason_code, first_reason, > second_reason_code, second_reason, pic_url, remark, user_receipt_status, > refund_target_account, create_time, modify_time, after_service_id0, > progress_node, progress_node_status, operator, parallel, extend_info], > leftInputSpec=[JoinKeyContainsUniqueKey], > rightInputSpec=[JoinKeyContainsUniqueKey]) > : : : :- > Exchange(distribution=[hash[after_service_id]]) > : : : : +- Calc(select=[after_service_id, user_id, > poi_id, refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, user_receipt_status, refund_target_account, > create_time, modify_time]) > : : : : +- DropUpdateBefore > : : : : +- > TableSourceScan(table=[[default_catalog, default_database, t_name]], > fields=[id, after_service_id, order_id, user_id, poi_id, city_id, > refund_type, first_reason_code, first_reason, second_reason_code, > second_reason, pic_url, remark, refund_price, refund_red_packet_price, > refund_total_price, refund_promotion_price, refund_coupon_price, > refund_other_price, user_receipt_status, collect_status, > refund_target_account, status, flow_instance_id, create_time, modify_time]) > : : : +- > Exchange(distribution=[hash[after_service_id]]) > : : : +- Calc(select=[after_service_id, > progress_node, progress_node_status, operator, parallel, extend_info]) > : : : +- DropUpdateBefore > : : : +- > TableSourceScan(table=[[default_catalog, default_database, t_progress_name]], > fields=[id, after_service_id, order_id, progress_node, progress_node_status, > operator, parallel, flow_element_id, extend_info, create_time, modify_time]) > : : +- Exchange(distribution=[hash[after_service_id]]) > : : +- Calc(select=[after_service_id, name, value]) > : : +- DropUpdateBefore > : : +- TableSourceScan(table=[[default_catalog, > default_database, t_attr_name]], fields=[id, after_service_id, order_id, > name, value, create_time, modify_time]) > : +- Exchange(distribution=[hash[after_service_id]]) > : +- Calc(select=[after_service_id, apply_status]) > : +- DropUpdateBefore > : +- TableSourceScan(table=[[default_catalog, > default_database, t_apply_status_name]], fields=[id, after_service_id, > order_id, apply_status, ai_audit_status, group_header_confirm_status, > group_header_retrieve_status, driver_retrieve_status, > group_header_parallel_status, grid_parallel_status, create_time, modify_time]) > +- Exchange(distribution=[hash[refund_apply_id]]) > +- Calc(select=[refund_apply_id, order_id, base_sku_id, > apply_refund_num, view_qty, refund_scale_user_real_pay AS apply_refund_amt, > refund_scale_user_real_pay, refund_red_packet_price]) > +- GroupAggregate(groupBy=[refund_apply_id, order_id, > base_sku_id], select=[refund_apply_id, order_id, base_sku_id, > SUM_RETRACT($f3) AS apply_refund_num, SUM_RETRACT($f4) AS view_qty, > SUM_RETRACT($f5) AS refund_scale_user_real_pay, SUM_RETRACT($f6) AS > refund_red_packet_price]) > +- Exchange(distribution=[hash[refund_apply_id, order_id, > base_sku_id]]) > +- Calc(select=[after_service_id AS refund_apply_id, > order_id, sku_id AS base_sku_id, (refund_quantity IS NULL IF 0 IF > refund_quantity) AS $f3, (refund_quantity IS NULL IF 0 IF > CAST(refund_quantity)) AS $f4, (refund_price IS NULL IF 0 IF refund_price) AS > $f5, (refund_red_packet_price IS NULL IF 0 IF refund_red_packet_price) AS > $f6]) > +- TableSourceScan(table=[[default_catalog, > default_database, t_item_name]], fields=[id, refund_fwd_item_id, > after_service_id, order_id, order_item_id, stack_id, sku_id, sku_name, > supplier_id, refund_quantity, item_sku_type, refund_scale_type, refund_scale, > accurate_refund, refund_price, refund_red_packet_price, refund_price_info, > refund_total_price, refund_promotion_price, refund_coupon_price, > refund_other_price, extend_info, create_time, modify_time]) {code} > -- This message was sent by Atlassian Jira (v8.20.7#820007)