[FLINK-34379] table.optimizer.dynamic-filtering.enabled lead to OutOfMemoryError

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.17.2, 1.18.1
• Fix Version/s: 1.17.3, 1.18.2, 1.20.0, 1.19.1
• Component/s: Table SQL / Planner
• Labels: pull-request-available

Environment

1.17.1

• Assignee: Jeyhun Karimov
• Reporter: zhu
• Created: 06/Feb/24 06:05
• Updated: 30/May/24 14:17
• Resolved: 30/May/24 14:16

Description

When using batch computing, I union all about 50 tables and then join other table. When compiling the execution plan, 
there throws OutOfMemoryError: Java heap space, which was no problem in  1.15.2. However, both 1.17.2 and 1.18.1 all throws same errors,This causes jobmanager to restart. Currently,it has been found that this is caused by table.optimizer.dynamic-filtering.enabled, which defaults is true,When I set table.optimizer.dynamic-filtering.enabled to false, it can be compiled and executed normally

code

TableEnvironment.create(EnvironmentSettings.newInstance()
.withConfiguration(configuration)
.inBatchMode().build())

sql=select att,filename,'table0' as mo_name from table0 UNION All select att,filename,'table1' as mo_name from table1 UNION All select att,filename,'table2' as mo_name from table2 UNION All select att,filename,'table3' as mo_name from table3 UNION All select att,filename,'table4' as mo_name from table4 UNION All select att,filename,'table5' as mo_name from table5 UNION All select att,filename,'table6' as mo_name from table6 UNION All select att,filename,'table7' as mo_name from table7 UNION All select att,filename,'table8' as mo_name from table8 UNION All select att,filename,'table9' as mo_name from table9 UNION All select att,filename,'table10' as mo_name from table10 UNION All select att,filename,'table11' as mo_name from table11 UNION All select att,filename,'table12' as mo_name from table12 UNION All select att,filename,'table13' as mo_name from table13 UNION All select att,filename,'table14' as mo_name from table14 UNION All select att,filename,'table15' as mo_name from table15 UNION All select att,filename,'table16' as mo_name from table16 UNION All select att,filename,'table17' as mo_name from table17 UNION All select att,filename,'table18' as mo_name from table18 UNION All select att,filename,'table19' as mo_name from table19 UNION All select att,filename,'table20' as mo_name from table20 UNION All select att,filename,'table21' as mo_name from table21 UNION All select att,filename,'table22' as mo_name from table22 UNION All select att,filename,'table23' as mo_name from table23 UNION All select att,filename,'table24' as mo_name from table24 UNION All select att,filename,'table25' as mo_name from table25 UNION All select att,filename,'table26' as mo_name from table26 UNION All select att,filename,'table27' as mo_name from table27 UNION All select att,filename,'table28' as mo_name from table28 UNION All select att,filename,'table29' as mo_name from table29 UNION All select att,filename,'table30' as mo_name from table30 UNION All select att,filename,'table31' as mo_name from table31 UNION All select att,filename,'table32' as mo_name from table32 UNION All select att,filename,'table33' as mo_name from table33 UNION All select att,filename,'table34' as mo_name from table34 UNION All select att,filename,'table35' as mo_name from table35 UNION All select att,filename,'table36' as mo_name from table36 UNION All select att,filename,'table37' as mo_name from table37 UNION All select att,filename,'table38' as mo_name from table38 UNION All select att,filename,'table39' as mo_name from table39 UNION All select att,filename,'table40' as mo_name from table40 UNION All select att,filename,'table41' as mo_name from table41 UNION All select att,filename,'table42' as mo_name from table42 UNION All select att,filename,'table43' as mo_name from table43 UNION All select att,filename,'table44' as mo_name from table44 UNION All select att,filename,'table45' as mo_name from table45 UNION All select att,filename,'table46' as mo_name from table46 UNION All select att,filename,'table47' as mo_name from table47 UNION All select att,filename,'table48' as mo_name from table48 UNION All select att,filename,'table49' as mo_name from table49 UNION All select att,filename,'table50' as mo_name from table50 UNION All select att,filename,'table51' as mo_name from table51 UNION All select att,filename,'table52' as mo_name from table52 UNION All select att,filename,'table53' as mo_name from table53；

Table allUnionTable = tEnv.sqlQuery(sql);
Table res =
allUnionTable.join(
allUnionTable
.groupBy(col("att"))
.select(col("att"), col("att").count().as(COUNT_NAME))
.filter(col(COUNT_NAME).isGreater(1))
.select(col(key).as("l_key")),
col(key).isEqual(col("l_key"))
);

res.printExplain(ExplainDetail.JSON_EXECUTION_PLAN);

 

 

Exception trace

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
    at java.util.Arrays.copyOf(Arrays.java:3181)
    at java.util.ArrayList.grow(ArrayList.java:267)
    at java.util.ArrayList.ensureExplicitCapacity(ArrayList.java:241)
    at java.util.ArrayList.ensureCapacityInternal(ArrayList.java:233)
    at java.util.ArrayList.add(ArrayList.java:464)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.setTables(DynamicPartitionPruningUtils.java:240)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:163)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:175)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)
    at org.apache.flink.table.planner.utils.DynamicPartitionPruningUtils$DppDimSideChecker.visitDimSide(DynamicPartitionPruningUtils.java:190)

 

 

Issue Links

• links to: GitHub Pull Request #24600
• links to: GitHub Pull Request #24788
• links to: GitHub Pull Request #24860