You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by wa...@apache.org on 2023/12/14 09:26:37 UTC
(flink) 07/11: [FLINK-32844][table-planner] Does not apply runtime filter and DPP on same key
This is an automated email from the ASF dual-hosted git repository.
wanglijie pushed a commit to branch debug_q58
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3e48f9b007f45bb5686cf6db2f48f04baef6c1c7
Author: Lijie Wang <wa...@gmail.com>
AuthorDate: Tue Aug 15 14:08:27 2023 +0800
[FLINK-32844][table-planner] Does not apply runtime filter and DPP on same key
This closes #23215
---
.../plan/optimize/program/FlinkRuntimeFilterProgram.java | 13 +++++++++++++
.../table/planner/utils/DynamicPartitionPruningUtils.java | 3 ++-
.../BatchPhysicalDynamicFilteringTableSourceScan.scala | 9 ++++++---
3 files changed, 21 insertions(+), 4 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java
index f2753dbd7a7..1e8101fdf50 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/program/FlinkRuntimeFilterProgram.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.planner.plan.optimize.program;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.planner.plan.nodes.FlinkConventions;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalExchange;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalGroupAggregateBase;
import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalHashJoin;
@@ -52,9 +53,11 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
@@ -485,6 +488,16 @@ public class FlinkRuntimeFilterProgram implements FlinkOptimizeProgram<BatchOpti
input, probeIndices, buildSideInfo, filterHasBenefit));
}
return union.copy(union.getTraitSet(), newInputs, union.all);
+ } else if (rel instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+ BatchPhysicalDynamicFilteringTableSourceScan tableScan =
+ (BatchPhysicalDynamicFilteringTableSourceScan) rel;
+ final Set<Integer> dynamicFilteringIndices =
+ new HashSet<>(tableScan.dynamicFilteringIndices());
+ if (dynamicFilteringIndices.containsAll(probeIndices)) {
+ // do nothing, return current probe side directly. Because the fields have already
+ // filtered by DPP.
+ return rel;
+ }
} else {
// the above cases can cover all cases of TPC-DS test
// we may find more cases later
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 90f7b40bc0b..d6cde62b6ee 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
@@ -346,7 +346,8 @@ public class DynamicPartitionPruningUtils {
scan.getTraitSet(),
scan.getHints(),
tableSourceTable,
- dynamicFilteringDataCollector);
+ dynamicFilteringDataCollector,
+ acceptedFieldIndices);
} else if (rel instanceof Exchange || rel instanceof Filter) {
return rel.copy(
rel.getTraitSet(),
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalDynamicFilteringTableSourceScan.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalDynamicFilteringTableSourceScan.scala
index 90fbc030845..1826e3b0c04 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalDynamicFilteringTableSourceScan.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalDynamicFilteringTableSourceScan.scala
@@ -40,7 +40,8 @@ class BatchPhysicalDynamicFilteringTableSourceScan(
traitSet: RelTraitSet,
hints: util.List[RelHint],
tableSourceTable: TableSourceTable,
- var input: RelNode) // var for updating
+ var input: RelNode, // var for updating
+ val dynamicFilteringIndices: util.List[Integer])
extends BatchPhysicalTableSourceScan(cluster, traitSet, hints, tableSourceTable) {
override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
@@ -49,7 +50,8 @@ class BatchPhysicalDynamicFilteringTableSourceScan(
traitSet,
getHints,
tableSourceTable,
- inputs.get(0))
+ inputs.get(0),
+ dynamicFilteringIndices)
}
override def copy(
@@ -60,7 +62,8 @@ class BatchPhysicalDynamicFilteringTableSourceScan(
traitSet,
getHints,
tableSourceTable,
- input)
+ input,
+ dynamicFilteringIndices)
}
override def replaceInput(ordinalInParent: Int, rel: RelNode): Unit = {