This is an automated email from the ASF dual-hosted git repository. godfrey pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1ed1deb351bd7a9d4fbe46dd9e9cf40b08b97fd9 Author: zhengyunhong.zyh <[email protected]> AuthorDate: Tue Aug 23 11:55:39 2022 +0800 [FLINK-28993][table-planner] Fix adjusting join cost for dpp query pattern error This closes #20596 --- .../utils/DynamicPartitionPruningUtils.java | 9 +- .../planner/plan/metadata/FlinkRelMdRowCount.scala | 4 +- .../batch/DynamicPartitionPruningRuleTest.java | 133 ++++++++++++++-- .../batch/DynamicPartitionPruningRuleTest.xml | 171 ++++++++++++++++----- 4 files changed, 256 insertions(+), 61 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java index ffad5054d5d..d569772b2e1 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java @@ -171,7 +171,14 @@ public class DynamicPartitionPruningUtils { joinKeys.stream() .map(i -> scan.getRowType().getFieldNames().get(i)) .collect(Collectors.toList()); - factSideFactors.isSuitableFactScanSource = !candidateFields.isEmpty(); + if (candidateFields.isEmpty()) { + factSideFactors.isSuitableFactScanSource = false; + return; + } + + factSideFactors.isSuitableFactScanSource = + !getSuitableDynamicFilteringFieldsInFactSide(tableSource, candidateFields) + .isEmpty(); } else if (rel instanceof HepRelVertex) { visitFactSide(((HepRelVertex) rel).getCurrentRel(), factSideFactors, joinKeys); } else if (rel instanceof Exchange || rel instanceof Filter) { diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala index cfb3ee08cdd..b20868b1835 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/metadata/FlinkRelMdRowCount.scala @@ -369,14 +369,14 @@ class FlinkRelMdRowCount private extends MetadataHandler[BuiltInMetadata.RowCoun } else { leftRowCount * selectivityOfNonEquiPred } - return outputRowCount + return outputRowCount * dynamicPartitionPruningFactor } // if joinCondition has no ndv stats and no uniqueKeys stats, // rowCount = (leftRowCount + rightRowCount) * join condition selectivity val crossJoin = copyJoinWithNewCondition(join, rexBuilder.makeLiteral(true)) val selectivity = fmq.getSelectivity(crossJoin, condition) - (leftRowCount + rightRowCount) * selectivity + (leftRowCount + rightRowCount) * selectivity * dynamicPartitionPruningFactor } private def copyJoinWithNewCondition(join: Join, newCondition: RexNode): Join = { diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java index 51efc661996..0a7f0c7bc9d 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.java @@ -22,6 +22,9 @@ import org.apache.flink.table.api.TableConfig; import org.apache.flink.table.api.config.OptimizerConfigOptions; import org.apache.flink.table.catalog.ObjectPath; import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.stats.CatalogColumnStatistics; +import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataBase; +import org.apache.flink.table.catalog.stats.CatalogColumnStatisticsDataLong; import org.apache.flink.table.catalog.stats.CatalogTableStatistics; import org.apache.flink.table.planner.factories.TestValuesCatalog; import org.apache.flink.table.planner.utils.BatchTableTestUtil; @@ -30,6 +33,12 @@ import org.apache.flink.table.planner.utils.TableTestBase; import org.junit.Before; import org.junit.Test; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + /** * Test for rules that extend {@link DynamicPartitionPruningRule} to create {@link * org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan}. @@ -428,12 +437,12 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { // --------------------------dpp factor test --------------------------------------------- @Test - public void testDPPFactorToReorderTable() { + public void testDPPFactorToReorderTableWithoutStats() { // While there are several joins, and fact table not adjacent to dim table directly. dynamic // partition pruning factor will try best to reorder join relations to make fact table // adjacent to dim table. - String ddl1 = - "CREATE TABLE test_database.sales (\n" + String ddl = + "CREATE TABLE test_database.item (\n" + " id BIGINT,\n" + " amount BIGINT,\n" + " price BIGINT\n" @@ -441,9 +450,23 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { + " 'connector' = 'values',\n" + " 'bounded' = 'true'\n" + ")"; - util.tableEnv().executeSql(ddl1); + util.tableEnv().executeSql(ddl); + TableConfig tableConfig = util.tableEnv().getConfig(); + // Join reorder need open. + tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true); + + String query = + "Select * from fact_part, item, dim" + + " where fact_part.fact_date_sk = dim.dim_date_sk" + + " and fact_part.id = item.id" + + " and dim.id = item.id " + + " and dim.price < 500 and dim.price > 300"; + util.verifyRelPlan(query); + } - String ddl2 = + @Test + public void testDPPFactorToReorderTableWithStats() throws TableNotExistException { + String ddl = "CREATE TABLE test_database.item (\n" + " id BIGINT,\n" + " amount BIGINT,\n" @@ -452,26 +475,53 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { + " 'connector' = 'values',\n" + " 'bounded' = 'true'\n" + ")"; - util.tableEnv().executeSql(ddl2); + util.tableEnv().executeSql(ddl); TableConfig tableConfig = util.tableEnv().getConfig(); // Join reorder need open. tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true); + // Alter table stats and column stats. + CatalogTableStatistics tableStatistics = new CatalogTableStatistics(10, 10, 10, 10); + catalog.alterTableStatistics( + new ObjectPath("test_database", "dim"), tableStatistics, false); + catalog.alterTableColumnStatistics( + new ObjectPath("test_database", "dim"), + createJoinKeyColumnStats( + Arrays.asList("dim_date_sk", "id", "price"), 10L, 1000L, 5L, 10L), + false); + + // table item have same stats with table dim, but item not meets dpp pattern. + tableStatistics = new CatalogTableStatistics(10, 10, 10, 10); + catalog.alterTableStatistics( + new ObjectPath("test_database", "item"), tableStatistics, false); + catalog.alterTableColumnStatistics( + new ObjectPath("test_database", "item"), + createJoinKeyColumnStats(Collections.singletonList("id"), 10L, 1000L, 5L, 10L), + false); + + tableStatistics = new CatalogTableStatistics(10000, 10000, 10000, 10000); + catalog.alterTableStatistics( + new ObjectPath("test_database", "fact_part"), tableStatistics, false); + catalog.alterTableColumnStatistics( + new ObjectPath("test_database", "fact_part"), + createJoinKeyColumnStats( + Arrays.asList("fact_date_sk", "id"), 100L, 1000000L, 8500L, 9800L), + false); + String query = - "Select * from fact_part, item, sales, dim" + "Select * from fact_part, item, dim" + " where fact_part.fact_date_sk = dim.dim_date_sk" + " and fact_part.id = item.id" - + " and fact_part.id = sales.id" + " and dim.id = item.id " + " and dim.price < 500 and dim.price > 300"; util.verifyRelPlan(query); } @Test - public void testDPPFactorWithDimSideJoinKeyChanged() { - // If partition keys changed in dim side. DPP factor will not works. - String ddl1 = - "CREATE TABLE test_database.sales (\n" + public void testDPPFactorWithFactSideJoinKeyChanged() { + // If partition keys changed in fact side. DPP factor will not work. + String ddl = + "CREATE TABLE test_database.item (\n" + " id BIGINT,\n" + " amount BIGINT,\n" + " price BIGINT\n" @@ -479,9 +529,23 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { + " 'connector' = 'values',\n" + " 'bounded' = 'true'\n" + ")"; - util.tableEnv().executeSql(ddl1); + util.tableEnv().executeSql(ddl); + TableConfig tableConfig = util.tableEnv().getConfig(); + // Join reorder need open. + tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true); - String ddl2 = + String query = + "Select * from (select fact_date_sk + 1 as fact_date_sk, id from fact_part) fact_part1 join item on " + + "fact_part1.id = item.id" + + " join dim on fact_part1.fact_date_sk = dim.dim_date_sk" + + " where dim.price < 500 and dim.price > 300"; + util.verifyRelPlan(query); + } + + @Test + public void testDPPFactorWithDimSideJoinKeyChanged() { + // Although partition keys changed in dim side. DPP factor will work. + String ddl = "CREATE TABLE test_database.item (\n" + " id BIGINT,\n" + " amount BIGINT,\n" @@ -490,17 +554,54 @@ public class DynamicPartitionPruningRuleTest extends TableTestBase { + " 'connector' = 'values',\n" + " 'bounded' = 'true'\n" + ")"; - util.tableEnv().executeSql(ddl2); + util.tableEnv().executeSql(ddl); TableConfig tableConfig = util.tableEnv().getConfig(); // Join reorder need open. tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true); String query = "Select * from fact_part join item on fact_part.id = item.id" - + " join sales on fact_part.id = sales.id" + " join (select dim_date_sk + 1 as dim_date_sk, price from dim) dim1" + " on fact_part.fact_date_sk = dim1.dim_date_sk" + " where dim1.price < 500 and dim1.price > 300"; util.verifyRelPlan(query); } + + @Test + public void testDPPFactorWithJoinKeysNotIncludePartitionKeys() { + // If join keys of partition table join with dim table not include partition keys, dpp + // factor will not be adjusted and dpp will not succeed. + String ddl = + "CREATE TABLE test_database.item (\n" + + " id BIGINT,\n" + + " amount BIGINT,\n" + + " price BIGINT\n" + + ") WITH (\n" + + " 'connector' = 'values',\n" + + " 'bounded' = 'true'\n" + + ")"; + util.tableEnv().executeSql(ddl); + TableConfig tableConfig = util.tableEnv().getConfig(); + // Join reorder need open. + tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_JOIN_REORDER_ENABLED, true); + + String query = + "Select * from fact_part, item, dim" + + " where fact_part.id = dim.id" + + " and fact_part.id = item.id" + + " and dim.id = item.id " + + " and dim.price < 500 and dim.price > 300"; + util.verifyRelPlan(query); + } + + private CatalogColumnStatistics createJoinKeyColumnStats( + List<String> columnNames, Long min, Long max, Long ndv, Long nullCount) { + CatalogColumnStatisticsDataLong longColStats = + new CatalogColumnStatisticsDataLong(min, max, ndv, nullCount); + Map<String, CatalogColumnStatisticsDataBase> colStatsMap = new HashMap<>(1); + for (String columnName : columnNames) { + colStatsMap.put(columnName, longColStats); + } + return new CatalogColumnStatistics(colStatsMap); + } } diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml index b1951732632..b9d9dfc726b 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRuleTest.xml @@ -237,39 +237,66 @@ HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)], select=[id, ]]> </Resource> </TestCase> - <TestCase name="testDPPFactorToReorderTable"> + <TestCase name="testDPPFactorToReorderTableWithoutStats"> <Resource name="sql"> - <![CDATA[Select * from fact_part, item, sales, dim where fact_part.fact_date_sk = dim.dim_date_sk and fact_part.id = item.id and fact_part.id = sales.id and dim.id = item.id and dim.price < 500 and dim.price > 300]]> + <![CDATA[Select * from fact_part, item, dim where fact_part.fact_date_sk = dim.dim_date_sk and fact_part.id = item.id and dim.id = item.id and dim.price < 500 and dim.price > 300]]> </Resource> <Resource name="ast"> <![CDATA[ -LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4], id0=[$5], amount0=[$6], price0=[$7], id1=[$8], amount1=[$9], price1=[$10], id2=[$11], male=[$12], amount2=[$13], price2=[$14], dim_date_sk=[$15]) -+- LogicalFilter(condition=[AND(=($4, $15), =($0, $5), =($0, $8), =($11, $5), <($14, 500), >($14, 300))]) +LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4], id0=[$5], amount0=[$6], price0=[$7], id1=[$8], male=[$9], amount1=[$10], price1=[$11], dim_date_sk=[$12]) ++- LogicalFilter(condition=[AND(=($4, $12), =($0, $5), =($8, $5), <($11, 500), >($11, 300))]) +- LogicalJoin(condition=[true], joinType=[inner]) :- LogicalJoin(condition=[true], joinType=[inner]) - : :- LogicalJoin(condition=[true], joinType=[inner]) - : : :- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) - : : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) - : +- LogicalTableScan(table=[[testCatalog, test_database, sales]]) + : :- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) + : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) +- LogicalTableScan(table=[[testCatalog, test_database, dim]]) ]]> </Resource> <Resource name="optimized rel plan"> <![CDATA[ -Calc(select=[id00 AS id, name, amount00 AS amount, price00 AS price, fact_date_sk, id AS id0, amount AS amount0, price AS price0, id0 AS id1, amount0 AS amount1, price0 AS price1, id000 AS id2, male, amount000 AS amount2, price000 AS price2, dim_date_sk]) -+- HashJoin(joinType=[InnerJoin], where=[=(id, id00)], select=[id, amount, price, id0, amount0, price0, id00, name, amount00, price00, fact_date_sk, id000, male, amount000, price000, dim_date_sk], build=[right]) +Calc(select=[id0 AS id, name, amount0 AS amount, price0 AS price, fact_date_sk, id AS id0, amount AS amount0, price AS price0, id00 AS id1, male, amount00 AS amount1, price00 AS price1, dim_date_sk]) ++- HashJoin(joinType=[InnerJoin], where=[=(id, id0)], select=[id, amount, price, id0, name, amount0, price0, fact_date_sk, id00, male, amount00, price00, dim_date_sk], build=[right]) :- Exchange(distribution=[hash[id]]) : +- TableSourceScan(table=[[testCatalog, test_database, item]], fields=[id, amount, price]) - +- HashJoin(joinType=[InnerJoin], where=[=(id, id0)], select=[id, amount, price, id0, name, amount0, price0, fact_date_sk, id00, male, amount00, price00, dim_date_sk], build=[right]) + +- HashJoin(joinType=[InnerJoin], where=[AND(=(fact_date_sk, dim_date_sk), =(id0, id))], select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0, dim_date_sk], build=[right]) :- Exchange(distribution=[hash[id]]) - : +- TableSourceScan(table=[[testCatalog, test_database, sales]], fields=[id, amount, price]) - +- HashJoin(joinType=[InnerJoin], where=[AND(=(fact_date_sk, dim_date_sk), =(id0, id))], select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0, dim_date_sk], build=[right]) - :- Exchange(distribution=[hash[id]]) - : +- DynamicFilteringTableSourceScan(table=[[testCatalog, test_database, fact_part]], fields=[id, name, amount, price, fact_date_sk]) - : +- DynamicFilteringDataCollector(fields=[dim_date_sk]) - : +- Calc(select=[id, male, amount, price, dim_date_sk], where=[SEARCH(price, Sarg[(300..500)])]) - : +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) - +- Exchange(distribution=[hash[id]]) + : +- DynamicFilteringTableSourceScan(table=[[testCatalog, test_database, fact_part]], fields=[id, name, amount, price, fact_date_sk]) + : +- DynamicFilteringDataCollector(fields=[dim_date_sk]) + : +- Calc(select=[id, male, amount, price, dim_date_sk], where=[SEARCH(price, Sarg[(300..500)])]) + : +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) + +- Exchange(distribution=[hash[id]]) + +- Calc(select=[id, male, amount, price, dim_date_sk], where=[SEARCH(price, Sarg[(300..500)])]) + +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) +]]> + </Resource> + </TestCase> + <TestCase name="testDPPFactorToReorderTableWithStats"> + <Resource name="sql"> + <![CDATA[Select * from fact_part, item, dim where fact_part.fact_date_sk = dim.dim_date_sk and fact_part.id = item.id and dim.id = item.id and dim.price < 500 and dim.price > 300]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4], id0=[$5], amount0=[$6], price0=[$7], id1=[$8], male=[$9], amount1=[$10], price1=[$11], dim_date_sk=[$12]) ++- LogicalFilter(condition=[AND(=($4, $12), =($0, $5), =($8, $5), <($11, 500), >($11, 300))]) + +- LogicalJoin(condition=[true], joinType=[inner]) + :- LogicalJoin(condition=[true], joinType=[inner]) + : :- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) + : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) + +- LogicalTableScan(table=[[testCatalog, test_database, dim]]) +]]> + </Resource> + <Resource name="optimized rel plan"> + <![CDATA[ +Calc(select=[id0 AS id, name, amount0 AS amount, price0 AS price, fact_date_sk, id AS id0, amount AS amount0, price AS price0, id00 AS id1, male, amount00 AS amount1, price00 AS price1, dim_date_sk]) ++- NestedLoopJoin(joinType=[InnerJoin], where=[=(id, id0)], select=[id, amount, price, id0, name, amount0, price0, fact_date_sk, id00, male, amount00, price00, dim_date_sk], build=[right]) + :- TableSourceScan(table=[[testCatalog, test_database, item]], fields=[id, amount, price]) + +- Exchange(distribution=[broadcast]) + +- NestedLoopJoin(joinType=[InnerJoin], where=[AND(=(fact_date_sk, dim_date_sk), =(id0, id))], select=[id, name, amount, price, fact_date_sk, id0, male, amount0, price0, dim_date_sk], build=[right]) + :- DynamicFilteringTableSourceScan(table=[[testCatalog, test_database, fact_part]], fields=[id, name, amount, price, fact_date_sk]) + : +- DynamicFilteringDataCollector(fields=[dim_date_sk]) + : +- Calc(select=[id, male, amount, price, dim_date_sk], where=[SEARCH(price, Sarg[(300..500)])]) + : +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) + +- Exchange(distribution=[broadcast]) +- Calc(select=[id, male, amount, price, dim_date_sk], where=[SEARCH(price, Sarg[(300..500)])]) +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) ]]> @@ -277,38 +304,98 @@ Calc(select=[id00 AS id, name, amount00 AS amount, price00 AS price, fact_date_s </TestCase> <TestCase name="testDPPFactorWithDimSideJoinKeyChanged"> <Resource name="sql"> - <![CDATA[Select * from fact_part join item on fact_part.id = item.id join sales on fact_part.id = sales.id join (select dim_date_sk + 1 as dim_date_sk, price from dim) dim1 on fact_part.fact_date_sk = dim1.dim_date_sk where dim1.price < 500 and dim1.price > 300]]> + <![CDATA[Select * from fact_part join item on fact_part.id = item.id join (select dim_date_sk + 1 as dim_date_sk, price from dim) dim1 on fact_part.fact_date_sk = dim1.dim_date_sk where dim1.price < 500 and dim1.price > 300]]> </Resource> <Resource name="ast"> <![CDATA[ -LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4], id0=[$5], amount0=[$6], price0=[$7], id1=[$8], amount1=[$9], price1=[$10], dim_date_sk=[$11], price2=[$12]) -+- LogicalFilter(condition=[AND(<($12, 500), >($12, 300))]) - +- LogicalJoin(condition=[=($4, $11)], joinType=[inner]) - :- LogicalJoin(condition=[=($0, $8)], joinType=[inner]) - : :- LogicalJoin(condition=[=($0, $5)], joinType=[inner]) - : : :- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) - : : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) - : +- LogicalTableScan(table=[[testCatalog, test_database, sales]]) +LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4], id0=[$5], amount0=[$6], price0=[$7], dim_date_sk=[$8], price1=[$9]) ++- LogicalFilter(condition=[AND(<($9, 500), >($9, 300))]) + +- LogicalJoin(condition=[=($4, $8)], joinType=[inner]) + :- LogicalJoin(condition=[=($0, $5)], joinType=[inner]) + : :- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) + : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) +- LogicalProject(dim_date_sk=[+($4, 1)], price=[$3]) +- LogicalTableScan(table=[[testCatalog, test_database, dim]]) ]]> </Resource> <Resource name="optimized rel plan"> <![CDATA[ -Calc(select=[id, name, amount, price0 AS price, fact_date_sk, id0, amount0, price00 AS price0, id00 AS id1, amount00 AS amount1, price000 AS price1, dim_date_sk, price AS price2]) -+- HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)], select=[dim_date_sk, price, id, name, amount, price0, fact_date_sk, id0, amount0, price00, id00, amount00, price000], build=[left]) - :- Exchange(distribution=[hash[dim_date_sk]]) - : +- Calc(select=[+(dim_date_sk, 1) AS dim_date_sk, price], where=[SEARCH(price, Sarg[(300..500)])]) - : +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[], project=[price, dim_date_sk], metadata=[]]], fields=[price, dim_date_sk]) - +- Exchange(distribution=[hash[fact_date_sk]]) - +- HashJoin(joinType=[InnerJoin], where=[=(id, id0)], select=[id, name, amount, price, fact_date_sk, id0, amount0, price0, id00, amount00, price00], build=[right]) - :- Exchange(distribution=[hash[id]]) - : +- TableSourceScan(table=[[testCatalog, test_database, fact_part]], fields=[id, name, amount, price, fact_date_sk]) - +- HashJoin(joinType=[InnerJoin], where=[=(id0, id)], select=[id, amount, price, id0, amount0, price0], build=[right]) - :- Exchange(distribution=[hash[id]]) - : +- TableSourceScan(table=[[testCatalog, test_database, item]], fields=[id, amount, price]) - +- Exchange(distribution=[hash[id]]) - +- TableSourceScan(table=[[testCatalog, test_database, sales]], fields=[id, amount, price]) +Calc(select=[id0 AS id, name, amount0 AS amount, price0 AS price, fact_date_sk, id AS id0, amount AS amount0, price AS price0, dim_date_sk, price00 AS price1]) ++- HashJoin(joinType=[InnerJoin], where=[=(id0, id)], select=[id, amount, price, id0, name, amount0, price0, fact_date_sk, dim_date_sk, price00], build=[right]) + :- Exchange(distribution=[hash[id]]) + : +- TableSourceScan(table=[[testCatalog, test_database, item]], fields=[id, amount, price]) + +- Exchange(distribution=[hash[id]]) + +- HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)], select=[id, name, amount, price, fact_date_sk, dim_date_sk, price0], build=[right]) + :- Exchange(distribution=[hash[fact_date_sk]]) + : +- DynamicFilteringTableSourceScan(table=[[testCatalog, test_database, fact_part]], fields=[id, name, amount, price, fact_date_sk]) + : +- DynamicFilteringDataCollector(fields=[dim_date_sk]) + : +- Calc(select=[+(dim_date_sk, 1) AS dim_date_sk, price], where=[SEARCH(price, Sarg[(300..500)])]) + : +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[], project=[price, dim_date_sk], metadata=[]]], fields=[price, dim_date_sk]) + +- Exchange(distribution=[hash[dim_date_sk]]) + +- Calc(select=[+(dim_date_sk, 1) AS dim_date_sk, price], where=[SEARCH(price, Sarg[(300..500)])]) + +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[], project=[price, dim_date_sk], metadata=[]]], fields=[price, dim_date_sk]) +]]> + </Resource> + </TestCase> + <TestCase name="testDPPFactorWithFactSideJoinKeyChanged"> + <Resource name="sql"> + <![CDATA[Select * from (select fact_date_sk + 1 as fact_date_sk, id from fact_part) fact_part1 join item on fact_part1.id = item.id join dim on fact_part1.fact_date_sk = dim.dim_date_sk where dim.price < 500 and dim.price > 300]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(fact_date_sk=[$0], id=[$1], id0=[$2], amount=[$3], price=[$4], id1=[$5], male=[$6], amount0=[$7], price0=[$8], dim_date_sk=[$9]) ++- LogicalFilter(condition=[AND(<($8, 500), >($8, 300))]) + +- LogicalJoin(condition=[=($0, $9)], joinType=[inner]) + :- LogicalJoin(condition=[=($1, $2)], joinType=[inner]) + : :- LogicalProject(fact_date_sk=[+($4, 1)], id=[$0]) + : : +- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) + : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) + +- LogicalTableScan(table=[[testCatalog, test_database, dim]]) +]]> + </Resource> + <Resource name="optimized rel plan"> + <![CDATA[ +Calc(select=[fact_date_sk, id0 AS id, id AS id0, amount, price, id00 AS id1, male, amount0, price0, dim_date_sk]) ++- HashJoin(joinType=[InnerJoin], where=[=(id0, id)], select=[id, amount, price, fact_date_sk, id0, id00, male, amount0, price0, dim_date_sk], build=[right]) + :- Exchange(distribution=[hash[id]]) + : +- TableSourceScan(table=[[testCatalog, test_database, item]], fields=[id, amount, price]) + +- Exchange(distribution=[hash[id]]) + +- HashJoin(joinType=[InnerJoin], where=[=(fact_date_sk, dim_date_sk)], select=[fact_date_sk, id, id0, male, amount, price, dim_date_sk], build=[right]) + :- Exchange(distribution=[hash[fact_date_sk]]) + : +- Calc(select=[+(fact_date_sk, 1) AS fact_date_sk, id]) + : +- TableSourceScan(table=[[testCatalog, test_database, fact_part, project=[fact_date_sk, id], metadata=[]]], fields=[fact_date_sk, id]) + +- Exchange(distribution=[hash[dim_date_sk]]) + +- Calc(select=[id, male, amount, price, dim_date_sk], where=[SEARCH(price, Sarg[(300..500)])]) + +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) +]]> + </Resource> + </TestCase> + <TestCase name="testDPPFactorWithJoinKeysNotIncludePartitionKeys"> + <Resource name="sql"> + <![CDATA[Select * from fact_part, item, dim where fact_part.id = dim.id and fact_part.id = item.id and dim.id = item.id and dim.price < 500 and dim.price > 300]]> + </Resource> + <Resource name="ast"> + <![CDATA[ +LogicalProject(id=[$0], name=[$1], amount=[$2], price=[$3], fact_date_sk=[$4], id0=[$5], amount0=[$6], price0=[$7], id1=[$8], male=[$9], amount1=[$10], price1=[$11], dim_date_sk=[$12]) ++- LogicalFilter(condition=[AND(=($0, $8), =($0, $5), =($8, $5), <($11, 500), >($11, 300))]) + +- LogicalJoin(condition=[true], joinType=[inner]) + :- LogicalJoin(condition=[true], joinType=[inner]) + : :- LogicalTableScan(table=[[testCatalog, test_database, fact_part]]) + : +- LogicalTableScan(table=[[testCatalog, test_database, item]]) + +- LogicalTableScan(table=[[testCatalog, test_database, dim]]) +]]> + </Resource> + <Resource name="optimized rel plan"> + <![CDATA[ +HashJoin(joinType=[InnerJoin], where=[=(id, id0)], select=[id, name, amount, price, fact_date_sk, id0, amount0, price0, id00, male, amount00, price00, dim_date_sk], build=[right]) +:- Exchange(distribution=[hash[id]]) +: +- TableSourceScan(table=[[testCatalog, test_database, fact_part]], fields=[id, name, amount, price, fact_date_sk]) ++- HashJoin(joinType=[InnerJoin], where=[=(id0, id)], select=[id, amount, price, id0, male, amount0, price0, dim_date_sk], build=[right]) + :- Exchange(distribution=[hash[id]]) + : +- TableSourceScan(table=[[testCatalog, test_database, item]], fields=[id, amount, price]) + +- Exchange(distribution=[hash[id]]) + +- Calc(select=[id, male, amount, price, dim_date_sk], where=[AND(SEARCH(price, Sarg[(300..500)]), IS NOT NULL(id))]) + +- TableSourceScan(table=[[testCatalog, test_database, dim, filter=[]]], fields=[id, male, amount, price, dim_date_sk]) ]]> </Resource> </TestCase>
