[GitHub] [flink] godfreyhe commented on a diff in pull request #20388: [FLINK-28708][table-planner] Introduce planner rules to optimize dpp pattern

[GitBox <gi...@apache.org>]

godfreyhe commented on code in PR #20388:
URL: https://github.com/apache/flink/pull/20388#discussion_r934348621


##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+
+        if (acceptedFields.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether there are computes column on accepted fields. If there are, dynamic
+        // filtering cannot be generated.
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+        return factCalc == null || !computeColumnOnAcceptedFields(acceptedFieldIndices, factCalc);
+    }
+
+    /**
+     * Judge whether there are computes column on accepted fields. If there are, dynamic filtering
+     * cannot be generated.
+     */
+    private static boolean computeColumnOnAcceptedFields(
+            List<Integer> indexList, BatchPhysicalCalc factCalc) {
+        List<RexLocalRef> projectList = factCalc.getProgram().getProjectList();

Review Comment:
   The local ref should be expand, see RexProgram#expandLocalRef



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {

Review Comment:
   protected



##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalDynamicFilteringTableSourceScan.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flink.table.planner.plan.nodes.physical.batch
+
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode
+import org.apache.flink.table.planner.plan.nodes.exec.batch.BatchExecTableSourceScan
+import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSourceSpec
+import org.apache.flink.table.planner.plan.schema.TableSourceTable
+import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
+
+import com.google.common.collect.ImmutableList
+import org.apache.calcite.plan._
+import org.apache.calcite.rel.{RelNode, RelWriter}
+import org.apache.calcite.rel.hint.RelHint
+
+import java.util
+
+/**
+ * Batch physical RelNode to read data from an external source defined by a bounded
+ * [[org.apache.flink.table.connector.source.ScanTableSource]].
+ */
+class BatchPhysicalDynamicFilteringTableSourceScan(

Review Comment:
   please update the class doc



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {

Review Comment:
   please add more detailed comments
   
   rename DynamicFilteringRule to DynamicPartitionPruningRule, because this rule is only used for pruning partition



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+
+        if (acceptedFields.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether there are computes column on accepted fields. If there are, dynamic
+        // filtering cannot be generated.
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+        return factCalc == null || !computeColumnOnAcceptedFields(acceptedFieldIndices, factCalc);
+    }
+
+    /**
+     * Judge whether there are computes column on accepted fields. If there are, dynamic filtering
+     * cannot be generated.
+     */
+    private static boolean computeColumnOnAcceptedFields(
+            List<Integer> indexList, BatchPhysicalCalc factCalc) {
+        List<RexLocalRef> projectList = factCalc.getProgram().getProjectList();
+        for (int index : indexList) {
+            if (projectList.get(index).getIndex() != index) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean isDimSide(RelNode rel) {
+        DppDimSideFactors dimSideFactors = new DppDimSideFactors();
+        visitDimSide(rel, dimSideFactors);
+        return dimSideFactors.isDimSide();
+    }
+
+    /**
+     * Visit dim side to judge whether dim side has filter condition and whether dim side's source
+     * table scan is non partitioned scan.
+     */
+    private static void visitDimSide(RelNode rel, DppDimSideFactors dimSideFactors) {
+        if (rel instanceof TableScan) {
+            TableScan scan = (TableScan) rel;
+            TableSourceTable table = scan.getTable().unwrap(TableSourceTable.class);
+            if (table == null) {
+                return;
+            }
+            if (Arrays.stream(table.abilitySpecs())
+                    .anyMatch(spec -> spec instanceof FilterPushDownSpec)) {
+                dimSideFactors.hasFilter = true;
+            }
+
+            CatalogTable catalogTable = table.contextResolvedTable().getTable();
+            dimSideFactors.hasNonPartitionedScan = !catalogTable.isPartitioned();
+        } else if (rel instanceof HepRelVertex) {
+            visitDimSide(((HepRelVertex) rel).getCurrentRel(), dimSideFactors);
+        } else if (rel instanceof Exchange || rel instanceof Project) {
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Calc) {
+            if (((Calc) rel).getProgram().getCondition() != null) {
+                dimSideFactors.hasFilter = true;
+            }
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Filter && ((Filter) rel).getCondition() != null) {
+            dimSideFactors.hasFilter = true;
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else {
+            rel.getInputs().forEach(n -> visitDimSide(n, dimSideFactors));
+        }
+    }
+
+    private static class DppDimSideFactors {
+        private boolean hasFilter;
+        private boolean hasNonPartitionedScan;
+
+        public boolean isDimSide() {
+            return hasFilter && hasNonPartitionedScan;
+        }
+    }
+
+    protected BatchPhysicalDynamicFilteringTableSourceScan createDynamicFilteringTableSourceScan(
+            BatchPhysicalTableSourceScan factScan,
+            BatchPhysicalRel dimSide,
+            BatchPhysicalJoinBase join,
+            boolean factInLeft) {
+        JoinInfo joinInfo = join.analyzeCondition();
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+        List<Integer> dimJoinKeys = factInLeft ? joinInfo.rightKeys : joinInfo.leftKeys;
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+        if (acceptedFields.isEmpty()) {
+            return null;
+        }
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+
+        List<Integer> dynamicFilteringFieldIndices = new ArrayList<>();
+        for (int i = 0; i < joinInfo.leftKeys.size(); ++i) {
+            if (acceptedFieldIndices.contains(factJoinKeys.get(i))) {
+                dynamicFilteringFieldIndices.add(dimJoinKeys.get(i));
+            }
+        }
+        final BatchPhysicalDynamicFilteringDataCollector dppConnector =
+                createDynamicFilteringConnector(dimSide, dynamicFilteringFieldIndices);
+        return new BatchPhysicalDynamicFilteringTableSourceScan(
+                factScan.getCluster(),
+                factScan.getTraitSet(),
+                factScan.getHints(),
+                factScan.tableSourceTable(),
+                dppConnector);
+    }
+
+    private BatchPhysicalDynamicFilteringDataCollector createDynamicFilteringConnector(
+            RelNode dimSide, List<Integer> dynamicFilteringFieldIndices) {
+        final RelDataType outputType =
+                ((FlinkTypeFactory) dimSide.getCluster().getTypeFactory())
+                        .projectStructType(
+                                dimSide.getRowType(),
+                                dynamicFilteringFieldIndices.stream().mapToInt(i -> i).toArray());
+
+        return new BatchPhysicalDynamicFilteringDataCollector(
+                dimSide.getCluster(),
+                dimSide.getTraitSet(),
+                ignoreExchange(dimSide),
+                outputType,
+                dynamicFilteringFieldIndices.stream().mapToInt(i -> i).toArray());
+    }
+
+    private RelNode ignoreExchange(RelNode dimSide) {
+        if (dimSide instanceof Exchange) {
+            return dimSide.getInput(0);
+        } else {
+            return dimSide;
+        }
+    }
+
+    /** Simple dynamic filtering pattern with fact side in join right. */
+    protected static class DynamicFilteringFactInRightRule extends DynamicFilteringRule {
+
+        public DynamicFilteringFactInRightRule(RelRule.Config config) {
+            super(config);
+        }
+
+        /** Config. */
+        public interface Config extends RelRule.Config {
+            @Override
+            default DynamicFilteringFactInRightRule toRule() {
+                return new DynamicFilteringFactInRightRule(this);
+            }
+
+            default Config factInRight() {
+                return withOperandSupplier(
+                                b0 ->
+                                        b0.operand(BatchPhysicalJoinBase.class)
+                                                .inputs(
+                                                        l ->
+                                                                l.operand(BatchPhysicalRel.class)
+                                                                        .anyInputs(),
+                                                        r ->
+                                                                r.operand(
+                                                                                BatchPhysicalTableSourceScan
+                                                                                        .class)
+                                                                        .noInputs()))
+                        .as(Config.class);
+            }
+        }
+
+        @Override
+        public boolean matches(RelOptRuleCall call) {
+            final BatchPhysicalJoinBase join = call.rel(0);
+            final BatchPhysicalRel dimSide = call.rel(1);
+            final BatchPhysicalTableSourceScan factScan = call.rel(2);
+            return doMatches(join, dimSide, null, factScan, false);
+        }
+
+        @Override
+        public void onMatch(RelOptRuleCall call) {
+            final BatchPhysicalJoinBase join = call.rel(0);
+            final BatchPhysicalRel dimSide = call.rel(1);
+            final BatchPhysicalTableSourceScan factScan = call.rel(2);
+
+            final BatchPhysicalDynamicFilteringTableSourceScan newFactScan =
+                    createDynamicFilteringTableSourceScan(factScan, dimSide, join, false);
+            final Join newJoin = join.copy(join.getTraitSet(), Arrays.asList(dimSide, newFactScan));
+            call.transformTo(newJoin);
+        }
+    }
+
+    /** Simple dynamic filtering pattern with fact side in join left. */
+    protected static class DynamicFilteringFactInLeftRule extends DynamicFilteringRule {
+
+        public DynamicFilteringFactInLeftRule(RelRule.Config config) {
+            super(config);
+        }
+
+        /** Config. */
+        public interface Config extends RelRule.Config {
+            @Override
+            default DynamicFilteringFactInLeftRule toRule() {
+                return new DynamicFilteringFactInLeftRule(this);
+            }
+
+            default Config factInLeft() {

Review Comment:
   we can just define the Config instance just like JoinConditionPushRule#Config



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+
+        if (acceptedFields.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether there are computes column on accepted fields. If there are, dynamic
+        // filtering cannot be generated.
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+        return factCalc == null || !computeColumnOnAcceptedFields(acceptedFieldIndices, factCalc);
+    }
+
+    /**
+     * Judge whether there are computes column on accepted fields. If there are, dynamic filtering

Review Comment:
   accepted field should all be RexInputRef



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()

Review Comment:
   There may be a fact calc between fact source and join, and the position may be changed in calc. So we can not get `candidateFields` from factScan based on join keys. Do we have any tests to cover such case?



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(

Review Comment:
   please explain why we need so many rules ?



##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRuleTest.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+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.CatalogTableStatistics;
+import org.apache.flink.table.planner.factories.TestValuesCatalog;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test for rules that extend {@link DynamicFilteringRule} to create {@link
+ * org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan}.
+ */
+public class DynamicFilteringRuleTest extends TableTestBase {
+    protected BatchTableTestUtil util = batchTestUtil(TableConfig.getDefault());
+    TestValuesCatalog catalog = new TestValuesCatalog("testCatalog", "test_database", true);
+
+    @Before
+    public void setup() {
+        catalog.open();
+        util.tableEnv().registerCatalog("testCatalog", catalog);
+        util.tableEnv().useCatalog("testCatalog");
+        TableConfig tableConfig = util.tableEnv().getConfig();
+        tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED, true);
+
+        // partition fact table.
+        String ddl3 =
+                "CREATE TABLE test_database.fact_part (\n"
+                        + "  id BIGINT,\n"
+                        + "  name STRING,\n"
+                        + "  amount BIGINT,\n"
+                        + "  price BIGINT,\n"
+                        + "  fact_date_sk BIGINT\n"
+                        + ") PARTITIONED BY (fact_date_sk)\n"
+                        + "WITH (\n"
+                        + " 'connector' = 'values',\n"
+                        + " 'partition-list' = 'fact_date_sk:1990;fact_date_sk:1991;fact_date_sk:1992',\n"
+                        + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n"
+                        + " 'bounded' = 'true'\n"
+                        + ")";
+        util.tableEnv().executeSql(ddl3);
+
+        // dim table.
+        String ddl2 =
+                "CREATE TABLE test_database.dim (\n"
+                        + "  id BIGINT,\n"
+                        + "  name STRING,\n"
+                        + "  amount BIGINT,\n"
+                        + "  price BIGINT,\n"
+                        + "  dim_date_sk BIGINT\n"
+                        + ") WITH (\n"
+                        + " 'connector' = 'values',\n"
+                        + " 'bounded' = 'true'\n"
+                        + ")";
+        util.tableEnv().executeSql(ddl2);
+    }
+

Review Comment:
   add more tests:
   1.  fact table is a non partition table
   2. dim side without filter
   3. dim side with filter push down
   4. add full, anti join
   5. add more unsupported cases
   6. calc with more complex expression
   7. amount is a part of dynamic filter fields



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+
+        if (acceptedFields.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether there are computes column on accepted fields. If there are, dynamic
+        // filtering cannot be generated.
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+        return factCalc == null || !computeColumnOnAcceptedFields(acceptedFieldIndices, factCalc);
+    }
+
+    /**
+     * Judge whether there are computes column on accepted fields. If there are, dynamic filtering
+     * cannot be generated.
+     */
+    private static boolean computeColumnOnAcceptedFields(
+            List<Integer> indexList, BatchPhysicalCalc factCalc) {
+        List<RexLocalRef> projectList = factCalc.getProgram().getProjectList();
+        for (int index : indexList) {
+            if (projectList.get(index).getIndex() != index) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean isDimSide(RelNode rel) {
+        DppDimSideFactors dimSideFactors = new DppDimSideFactors();
+        visitDimSide(rel, dimSideFactors);
+        return dimSideFactors.isDimSide();
+    }
+
+    /**
+     * Visit dim side to judge whether dim side has filter condition and whether dim side's source
+     * table scan is non partitioned scan.
+     */
+    private static void visitDimSide(RelNode rel, DppDimSideFactors dimSideFactors) {
+        if (rel instanceof TableScan) {
+            TableScan scan = (TableScan) rel;
+            TableSourceTable table = scan.getTable().unwrap(TableSourceTable.class);
+            if (table == null) {
+                return;
+            }
+            if (Arrays.stream(table.abilitySpecs())
+                    .anyMatch(spec -> spec instanceof FilterPushDownSpec)) {
+                dimSideFactors.hasFilter = true;
+            }
+
+            CatalogTable catalogTable = table.contextResolvedTable().getTable();
+            dimSideFactors.hasNonPartitionedScan = !catalogTable.isPartitioned();
+        } else if (rel instanceof HepRelVertex) {
+            visitDimSide(((HepRelVertex) rel).getCurrentRel(), dimSideFactors);
+        } else if (rel instanceof Exchange || rel instanceof Project) {
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Calc) {
+            if (((Calc) rel).getProgram().getCondition() != null) {
+                dimSideFactors.hasFilter = true;
+            }
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Filter && ((Filter) rel).getCondition() != null) {
+            dimSideFactors.hasFilter = true;
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else {
+            rel.getInputs().forEach(n -> visitDimSide(n, dimSideFactors));
+        }
+    }
+
+    private static class DppDimSideFactors {
+        private boolean hasFilter;
+        private boolean hasNonPartitionedScan;
+
+        public boolean isDimSide() {
+            return hasFilter && hasNonPartitionedScan;
+        }
+    }
+
+    protected BatchPhysicalDynamicFilteringTableSourceScan createDynamicFilteringTableSourceScan(
+            BatchPhysicalTableSourceScan factScan,
+            BatchPhysicalRel dimSide,
+            BatchPhysicalJoinBase join,
+            boolean factInLeft) {
+        JoinInfo joinInfo = join.analyzeCondition();
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+        List<Integer> dimJoinKeys = factInLeft ? joinInfo.rightKeys : joinInfo.leftKeys;
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+        if (acceptedFields.isEmpty()) {
+            return null;
+        }
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+
+        List<Integer> dynamicFilteringFieldIndices = new ArrayList<>();
+        for (int i = 0; i < joinInfo.leftKeys.size(); ++i) {
+            if (acceptedFieldIndices.contains(factJoinKeys.get(i))) {
+                dynamicFilteringFieldIndices.add(dimJoinKeys.get(i));
+            }
+        }
+        final BatchPhysicalDynamicFilteringDataCollector dppConnector =

Review Comment:
   rename `dppConnector` to `dynamicFilteringDataCollector` ?



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+
+        if (acceptedFields.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether there are computes column on accepted fields. If there are, dynamic
+        // filtering cannot be generated.
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+        return factCalc == null || !computeColumnOnAcceptedFields(acceptedFieldIndices, factCalc);
+    }
+
+    /**
+     * Judge whether there are computes column on accepted fields. If there are, dynamic filtering
+     * cannot be generated.
+     */
+    private static boolean computeColumnOnAcceptedFields(
+            List<Integer> indexList, BatchPhysicalCalc factCalc) {
+        List<RexLocalRef> projectList = factCalc.getProgram().getProjectList();
+        for (int index : indexList) {
+            if (projectList.get(index).getIndex() != index) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean isDimSide(RelNode rel) {
+        DppDimSideFactors dimSideFactors = new DppDimSideFactors();
+        visitDimSide(rel, dimSideFactors);
+        return dimSideFactors.isDimSide();
+    }
+
+    /**
+     * Visit dim side to judge whether dim side has filter condition and whether dim side's source
+     * table scan is non partitioned scan.
+     */
+    private static void visitDimSide(RelNode rel, DppDimSideFactors dimSideFactors) {
+        if (rel instanceof TableScan) {
+            TableScan scan = (TableScan) rel;
+            TableSourceTable table = scan.getTable().unwrap(TableSourceTable.class);
+            if (table == null) {
+                return;
+            }
+            if (Arrays.stream(table.abilitySpecs())
+                    .anyMatch(spec -> spec instanceof FilterPushDownSpec)) {
+                dimSideFactors.hasFilter = true;
+            }
+
+            CatalogTable catalogTable = table.contextResolvedTable().getTable();
+            dimSideFactors.hasNonPartitionedScan = !catalogTable.isPartitioned();
+        } else if (rel instanceof HepRelVertex) {
+            visitDimSide(((HepRelVertex) rel).getCurrentRel(), dimSideFactors);
+        } else if (rel instanceof Exchange || rel instanceof Project) {
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Calc) {
+            if (((Calc) rel).getProgram().getCondition() != null) {
+                dimSideFactors.hasFilter = true;
+            }
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Filter && ((Filter) rel).getCondition() != null) {
+            dimSideFactors.hasFilter = true;
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else {
+            rel.getInputs().forEach(n -> visitDimSide(n, dimSideFactors));

Review Comment:
   will we support the complex pattern now ?



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRule.java:
##########
@@ -0,0 +1,824 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.plan.abilities.source.FilterPushDownSpec;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalCalc;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringDataCollector;
+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.BatchPhysicalJoinBase;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalRel;
+import org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalTableSourceScan;
+import org.apache.flink.table.planner.plan.schema.TableSourceTable;
+import org.apache.flink.table.planner.utils.ShortcutUtils;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.hep.HepRelVertex;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Exchange;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexLocalRef;
+import org.apache.calcite.tools.RuleSet;
+import org.apache.calcite.tools.RuleSets;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED;
+
+/** Rules for dynamic filtering. */
+public abstract class DynamicFilteringRule extends RelRule<RelRule.Config> {
+
+    public static final RuleSet DYNAMIC_FILTERING_RULES =
+            RuleSets.ofList(
+                    DynamicFilteringFactInRightRule.DynamicFilteringFactInRightRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRight")
+                            .as(DynamicFilteringFactInRightRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftRule.DynamicFilteringFactInLeftRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeft")
+                            .as(DynamicFilteringFactInLeftRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchange")
+                            .as(DynamicFilteringFactInRightWithExchangeRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchange")
+                            .as(DynamicFilteringFactInLeftWithExchangeRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithCalc")
+                            .as(DynamicFilteringFactInRightWithCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithCalc")
+                            .as(DynamicFilteringFactInLeftWithCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule(),
+                    DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInRightWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInRightWithExchangeAndCalcRule.Config.class)
+                            .factInRight()
+                            .toRule(),
+                    DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.EMPTY
+                            .withDescription("DynamicFilteringRule:factInLeftWithExchangeAndCalc")
+                            .as(DynamicFilteringFactInLeftWithExchangeAndCalcRule.Config.class)
+                            .factInLeft()
+                            .toRule());
+
+    public DynamicFilteringRule(RelRule.Config config) {
+        super(config);
+    }
+
+    public boolean doMatches(
+            BatchPhysicalJoinBase join,
+            BatchPhysicalRel dimSide,
+            @Nullable BatchPhysicalCalc factCalc,
+            BatchPhysicalTableSourceScan factScan,
+            boolean factInLeft) {
+        if (!ShortcutUtils.unwrapContext(join)
+                .getTableConfig()
+                .get(TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED)) {
+            return false;
+        }
+        if (factScan instanceof BatchPhysicalDynamicFilteringTableSourceScan) {
+            // rule applied
+            return false;
+        }
+
+        // Now dynamic filtering support left outer join, right outer join, inner join and semi
+        // join.
+        if (join.getJoinType() == JoinRelType.LEFT) {
+            if (factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() == JoinRelType.RIGHT) {
+            if (!factInLeft) {
+                return false;
+            }
+        } else if (join.getJoinType() != JoinRelType.INNER
+                && join.getJoinType() != JoinRelType.SEMI) {
+            return false;
+        }
+
+        JoinInfo joinInfo = join.analyzeCondition();
+        if (joinInfo.leftKeys.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether dimSide meets the conditions of dimSide.
+        if (!isDimSide(dimSide)) {
+            return false;
+        }
+
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        if (tableSourceTable == null) {
+            return false;
+        }
+        CatalogTable catalogTable = tableSourceTable.contextResolvedTable().getTable();
+        List<String> partitionKeys = catalogTable.getPartitionKeys();
+        if (partitionKeys.isEmpty()) {
+            return false;
+        }
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+        if (!(tableSource instanceof SupportsDynamicFiltering)) {
+            return false;
+        }
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+
+        List<String> candidateFields =
+                factJoinKeys.stream()
+                        .map(i -> factScan.getRowType().getFieldNames().get(i))
+                        .collect(Collectors.toList());
+        List<String> acceptedFields =
+                ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields);
+
+        if (acceptedFields.isEmpty()) {
+            return false;
+        }
+
+        // Judge whether there are computes column on accepted fields. If there are, dynamic
+        // filtering cannot be generated.
+        List<Integer> acceptedFieldIndices =
+                acceptedFields.stream()
+                        .map(f -> factScan.getRowType().getFieldNames().indexOf(f))
+                        .collect(Collectors.toList());
+        return factCalc == null || !computeColumnOnAcceptedFields(acceptedFieldIndices, factCalc);
+    }
+
+    /**
+     * Judge whether there are computes column on accepted fields. If there are, dynamic filtering
+     * cannot be generated.
+     */
+    private static boolean computeColumnOnAcceptedFields(
+            List<Integer> indexList, BatchPhysicalCalc factCalc) {
+        List<RexLocalRef> projectList = factCalc.getProgram().getProjectList();
+        for (int index : indexList) {
+            if (projectList.get(index).getIndex() != index) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static boolean isDimSide(RelNode rel) {
+        DppDimSideFactors dimSideFactors = new DppDimSideFactors();
+        visitDimSide(rel, dimSideFactors);
+        return dimSideFactors.isDimSide();
+    }
+
+    /**
+     * Visit dim side to judge whether dim side has filter condition and whether dim side's source
+     * table scan is non partitioned scan.
+     */
+    private static void visitDimSide(RelNode rel, DppDimSideFactors dimSideFactors) {
+        if (rel instanceof TableScan) {
+            TableScan scan = (TableScan) rel;
+            TableSourceTable table = scan.getTable().unwrap(TableSourceTable.class);
+            if (table == null) {
+                return;
+            }
+            if (Arrays.stream(table.abilitySpecs())
+                    .anyMatch(spec -> spec instanceof FilterPushDownSpec)) {
+                dimSideFactors.hasFilter = true;
+            }
+
+            CatalogTable catalogTable = table.contextResolvedTable().getTable();
+            dimSideFactors.hasNonPartitionedScan = !catalogTable.isPartitioned();
+        } else if (rel instanceof HepRelVertex) {
+            visitDimSide(((HepRelVertex) rel).getCurrentRel(), dimSideFactors);
+        } else if (rel instanceof Exchange || rel instanceof Project) {
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Calc) {
+            if (((Calc) rel).getProgram().getCondition() != null) {
+                dimSideFactors.hasFilter = true;
+            }
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else if (rel instanceof Filter && ((Filter) rel).getCondition() != null) {
+            dimSideFactors.hasFilter = true;
+            visitDimSide(rel.getInput(0), dimSideFactors);
+        } else {
+            rel.getInputs().forEach(n -> visitDimSide(n, dimSideFactors));
+        }
+    }
+
+    private static class DppDimSideFactors {
+        private boolean hasFilter;
+        private boolean hasNonPartitionedScan;
+
+        public boolean isDimSide() {
+            return hasFilter && hasNonPartitionedScan;
+        }
+    }
+
+    protected BatchPhysicalDynamicFilteringTableSourceScan createDynamicFilteringTableSourceScan(
+            BatchPhysicalTableSourceScan factScan,
+            BatchPhysicalRel dimSide,
+            BatchPhysicalJoinBase join,
+            boolean factInLeft) {
+        JoinInfo joinInfo = join.analyzeCondition();
+        TableSourceTable tableSourceTable = factScan.getTable().unwrap(TableSourceTable.class);
+        DynamicTableSource tableSource = tableSourceTable.tableSource();
+
+        List<Integer> factJoinKeys = factInLeft ? joinInfo.leftKeys : joinInfo.rightKeys;
+        List<Integer> dimJoinKeys = factInLeft ? joinInfo.rightKeys : joinInfo.leftKeys;
+        List<String> candidateFields =

Review Comment:
   ditto



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecDynamicFilteringDataCollector.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.nodes.exec.batch;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeConfig;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeContext;
+import org.apache.flink.table.planner.plan.nodes.exec.InputProperty;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Collections;
+import java.util.List;
+
+/** BatchExecDynamicFilteringCollector. */
+public class BatchExecDynamicFilteringDataCollector extends ExecNodeBase<Object>

Review Comment:
   If you do not test exec plan for dpp, this class is unnecessary.



##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRuleTest.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+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.CatalogTableStatistics;
+import org.apache.flink.table.planner.factories.TestValuesCatalog;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test for rules that extend {@link DynamicFilteringRule} to create {@link
+ * org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan}.
+ */
+public class DynamicFilteringRuleTest extends TableTestBase {
+    protected BatchTableTestUtil util = batchTestUtil(TableConfig.getDefault());
+    TestValuesCatalog catalog = new TestValuesCatalog("testCatalog", "test_database", true);
+
+    @Before
+    public void setup() {
+        catalog.open();
+        util.tableEnv().registerCatalog("testCatalog", catalog);
+        util.tableEnv().useCatalog("testCatalog");
+        TableConfig tableConfig = util.tableEnv().getConfig();
+        tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED, true);
+
+        // partition fact table.
+        String ddl3 =
+                "CREATE TABLE test_database.fact_part (\n"
+                        + "  id BIGINT,\n"
+                        + "  name STRING,\n"
+                        + "  amount BIGINT,\n"
+                        + "  price BIGINT,\n"
+                        + "  fact_date_sk BIGINT\n"
+                        + ") PARTITIONED BY (fact_date_sk)\n"
+                        + "WITH (\n"
+                        + " 'connector' = 'values',\n"
+                        + " 'partition-list' = 'fact_date_sk:1990;fact_date_sk:1991;fact_date_sk:1992',\n"
+                        + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n"
+                        + " 'bounded' = 'true'\n"
+                        + ")";
+        util.tableEnv().executeSql(ddl3);
+
+        // dim table.
+        String ddl2 =
+                "CREATE TABLE test_database.dim (\n"
+                        + "  id BIGINT,\n"
+                        + "  name STRING,\n"
+                        + "  amount BIGINT,\n"
+                        + "  price BIGINT,\n"
+                        + "  dim_date_sk BIGINT\n"
+                        + ") WITH (\n"
+                        + " 'connector' = 'values',\n"
+                        + " 'bounded' = 'true'\n"
+                        + ")";
+        util.tableEnv().executeSql(ddl2);
+    }
+
+    @Test
+    public void testDimTableFilteringFieldsNotInJoinKey() {
+        // fact_part.id not in dynamic-filtering-fields, so dynamic filtering will not succeed.
+        String query =
+                "Select * from dim, fact_part where fact_part.id = dim.id and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInRightRule() throws TableNotExistException {
+        CatalogTableStatistics tableStatistics = new CatalogTableStatistics(1, 1, 1, 1);

Review Comment:
   why we need statistics



##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRuleTest.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+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.CatalogTableStatistics;
+import org.apache.flink.table.planner.factories.TestValuesCatalog;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test for rules that extend {@link DynamicFilteringRule} to create {@link
+ * org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan}.
+ */
+public class DynamicFilteringRuleTest extends TableTestBase {
+    protected BatchTableTestUtil util = batchTestUtil(TableConfig.getDefault());
+    TestValuesCatalog catalog = new TestValuesCatalog("testCatalog", "test_database", true);
+
+    @Before
+    public void setup() {
+        catalog.open();
+        util.tableEnv().registerCatalog("testCatalog", catalog);
+        util.tableEnv().useCatalog("testCatalog");
+        TableConfig tableConfig = util.tableEnv().getConfig();
+        tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED, true);
+
+        // partition fact table.
+        String ddl3 =

Review Comment:
   ddl1 ?



##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala:
##########
@@ -430,13 +430,18 @@ object FlinkBatchRuleSets {
 
   /** RuleSet to optimize plans after batch exec execution. */
   val PHYSICAL_REWRITE: RuleSet = RuleSets.ofList(
-    EnforceLocalHashAggRule.INSTANCE,
-    EnforceLocalSortAggRule.INSTANCE,
-    PushLocalHashAggIntoScanRule.INSTANCE,
-    PushLocalHashAggWithCalcIntoScanRule.INSTANCE,
-    PushLocalSortAggIntoScanRule.INSTANCE,
-    PushLocalSortAggWithSortIntoScanRule.INSTANCE,
-    PushLocalSortAggWithCalcIntoScanRule.INSTANCE,
-    PushLocalSortAggWithSortAndCalcIntoScanRule.INSTANCE
+    (RuleSets

Review Comment:
   nit: RuleSets.ofList(



##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/processor/MultipleInputNodeCreationProcessor.java:
##########
@@ -222,7 +223,8 @@ private MultipleInputGroup canBeInSameGroupWithOutputs(ExecNodeWrapper wrapper)
     private boolean canBeRootOfMultipleInputGroup(ExecNodeWrapper wrapper) {
         // only a node with more than one input can be the root,
         // as one-input operator chaining are handled by operator chains
-        return wrapper.inputs.size() >= 2;
+        return wrapper.inputs.size() >= 2
+                && !(wrapper.execNode instanceof BatchExecDynamicFilteringDataCollector);

Review Comment:
   we do not need change this



##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java:
##########
@@ -1366,7 +1410,7 @@ public DynamicTableSource copy() {
      * scan source without lookup ability, e.g. testing temporal join changelog source.
      */
     private static class TestValuesScanLookupTableSource extends TestValuesScanTableSource
-            implements LookupTableSource {
+            implements LookupTableSource, SupportsDynamicFiltering {

Review Comment:
   Only scan table source need to implements SupportsDynamicFiltering



##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicFilteringRuleTest.java:
##########
@@ -0,0 +1,276 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.rules.physical.batch;
+
+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.CatalogTableStatistics;
+import org.apache.flink.table.planner.factories.TestValuesCatalog;
+import org.apache.flink.table.planner.utils.BatchTableTestUtil;
+import org.apache.flink.table.planner.utils.TableTestBase;
+
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test for rules that extend {@link DynamicFilteringRule} to create {@link
+ * org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalDynamicFilteringTableSourceScan}.
+ */
+public class DynamicFilteringRuleTest extends TableTestBase {
+    protected BatchTableTestUtil util = batchTestUtil(TableConfig.getDefault());
+    TestValuesCatalog catalog = new TestValuesCatalog("testCatalog", "test_database", true);
+
+    @Before
+    public void setup() {
+        catalog.open();
+        util.tableEnv().registerCatalog("testCatalog", catalog);
+        util.tableEnv().useCatalog("testCatalog");
+        TableConfig tableConfig = util.tableEnv().getConfig();
+        tableConfig.set(OptimizerConfigOptions.TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED, true);
+
+        // partition fact table.
+        String ddl3 =
+                "CREATE TABLE test_database.fact_part (\n"
+                        + "  id BIGINT,\n"
+                        + "  name STRING,\n"
+                        + "  amount BIGINT,\n"
+                        + "  price BIGINT,\n"
+                        + "  fact_date_sk BIGINT\n"
+                        + ") PARTITIONED BY (fact_date_sk)\n"
+                        + "WITH (\n"
+                        + " 'connector' = 'values',\n"
+                        + " 'partition-list' = 'fact_date_sk:1990;fact_date_sk:1991;fact_date_sk:1992',\n"
+                        + " 'dynamic-filtering-fields' = 'fact_date_sk;amount',\n"
+                        + " 'bounded' = 'true'\n"
+                        + ")";
+        util.tableEnv().executeSql(ddl3);
+
+        // dim table.
+        String ddl2 =
+                "CREATE TABLE test_database.dim (\n"
+                        + "  id BIGINT,\n"
+                        + "  name STRING,\n"
+                        + "  amount BIGINT,\n"
+                        + "  price BIGINT,\n"
+                        + "  dim_date_sk BIGINT\n"
+                        + ") WITH (\n"
+                        + " 'connector' = 'values',\n"
+                        + " 'bounded' = 'true'\n"
+                        + ")";
+        util.tableEnv().executeSql(ddl2);
+    }
+
+    @Test
+    public void testDimTableFilteringFieldsNotInJoinKey() {
+        // fact_part.id not in dynamic-filtering-fields, so dynamic filtering will not succeed.
+        String query =
+                "Select * from dim, fact_part where fact_part.id = dim.id and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInRightRule() throws TableNotExistException {
+        CatalogTableStatistics tableStatistics = new CatalogTableStatistics(1, 1, 1, 1);
+        catalog.alterTableStatistics(
+                new ObjectPath("test_database", "dim"), tableStatistics, false);
+        String query =
+                "Select * from dim, fact_part where fact_part.fact_date_sk = dim.dim_date_sk and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInLeftRule() throws TableNotExistException {
+        CatalogTableStatistics tableStatistics = new CatalogTableStatistics(1, 1, 1, 1);
+        catalog.alterTableStatistics(
+                new ObjectPath("test_database", "dim"), tableStatistics, false);
+        String query =
+                "Select * from fact_part, dim where fact_part.fact_date_sk = dim.dim_date_sk and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInRightWithExchangeRule() {
+        String query =
+                "Select * from dim, fact_part where fact_part.fact_date_sk = dim.dim_date_sk and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInLeftWithExchangeRule() {
+        String query =
+                "Select * from fact_part, dim where fact_part.fact_date_sk = dim.dim_date_sk and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInRightWithCalcRule() throws TableNotExistException {
+        CatalogTableStatistics tableStatistics = new CatalogTableStatistics(1, 1, 1, 1);
+        catalog.alterTableStatistics(
+                new ObjectPath("test_database", "dim"), tableStatistics, false);
+        String query =
+                "Select * from dim, fact_part where fact_part.fact_date_sk = dim.dim_date_sk"
+                        + " and fact_part.price > 200 and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInLeftWithCalcRule() throws TableNotExistException {
+        CatalogTableStatistics tableStatistics = new CatalogTableStatistics(1, 1, 1, 1);
+        catalog.alterTableStatistics(
+                new ObjectPath("test_database", "dim"), tableStatistics, false);
+        String query =
+                "Select * from fact_part, dim where fact_part.fact_date_sk = dim.dim_date_sk"
+                        + " and fact_part.price > 200 and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInRightWithExchangeAndCalcRule() {
+        String query =
+                "Select * from dim, fact_part where fact_part.fact_date_sk = dim.dim_date_sk"
+                        + " and fact_part.price > 200 and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringFactInLeftWithExchangeAndCalcRule() {
+        String query =
+                "Select * from fact_part, dim where fact_part.fact_date_sk = dim.dim_date_sk"
+                        + " and fact_part.price > 200 and dim.price < 500";
+        util.verifyRelPlan(query);
+    }
+
+    @Test
+    public void testDynamicFilteringRuleWithDynamicFilteringFieldInFactSide()
+            throws TableNotExistException {
+        CatalogTableStatistics tableStatistics = new CatalogTableStatistics(1, 1, 1, 1);
+        catalog.alterTableStatistics(
+                new ObjectPath("test_database", "dim"), tableStatistics, false);
+        // in this case. amount + 1 as amount will cause dynamic filtering produce wrong results. So
+        // dynamic filtering will not succeed.

Review Comment:
   `dim_date_sk ` can be pushed



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org