This is an automated email from the ASF dual-hosted git repository. godfrey pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1a0f591a59b59f0c6ce71f5af9e0660293c33fc1 Author: zhengyunhong.zyh <[email protected]> AuthorDate: Tue Aug 23 11:52:53 2022 +0800 [FLINK-28993][table-planner] Refactor SupportsDynamicFiltering to provide getAcceptedFilterFields method to avoid modifing the DynamicTableSource object if calling applyDynamicFiltering method in join reorder rules This closes #20596 --- .../flink/connectors/hive/HiveTableSource.java | 66 ++++++++-------------- .../source/abilities/SupportsDynamicFiltering.java | 18 ++++-- .../batch/DynamicPartitionPruningRule.java | 52 +++-------------- .../utils/DynamicPartitionPruningUtils.java | 21 +++++++ .../planner/factories/TestValuesTableFactory.java | 21 ++----- 5 files changed, 74 insertions(+), 104 deletions(-) diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java index 2570c50b702..edb752ab7e0 100644 --- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java +++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java @@ -40,6 +40,7 @@ import org.apache.flink.orc.util.OrcFormatStatisticsReportUtil; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.TableSchema; import org.apache.flink.table.catalog.CatalogTable; import org.apache.flink.table.catalog.ObjectPath; @@ -76,7 +77,6 @@ import javax.annotation.Nullable; import java.time.LocalDateTime; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -86,8 +86,6 @@ import static org.apache.flink.connector.file.table.FileSystemConnectorOptions.P import static org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_CONSUME_START_OFFSET; import static org.apache.flink.connectors.hive.HiveOptions.STREAMING_SOURCE_ENABLE; import static org.apache.flink.connectors.hive.util.HivePartitionUtils.getAllPartitions; -import static org.apache.flink.util.Preconditions.checkArgument; -import static org.apache.flink.util.Preconditions.checkState; /** A TableSource implementation to read data from Hive tables. */ public class HiveTableSource @@ -252,47 +250,33 @@ public class HiveTableSource } @Override - public List<String> applyDynamicFiltering(List<String> candidateFilterFields) { - if (catalogTable.getPartitionKeys() != null - && catalogTable.getPartitionKeys().size() != 0) { - checkArgument( - !candidateFilterFields.isEmpty(), - "At least one field should be provided for dynamic filtering"); - - // only accept partition fields of supported types to do dynamic partition pruning - List<String> dynamicFilterPartitionKeys = new ArrayList<>(); - for (String field : candidateFilterFields) { - if (catalogTable.getPartitionKeys().contains(field) - && HiveSourceDynamicFileEnumerator.SUPPORTED_TYPES.contains( - catalogTable - .getSchema() - .getFieldDataType(field) - .map(DataType::getLogicalType) - .map(LogicalType::getTypeRoot) - .orElse(null))) { - dynamicFilterPartitionKeys.add(field); - } - } - if (dynamicFilterPartitionKeys.isEmpty()) { - LOG.warn( - "No dynamic filter field is accepted," - + " only partition fields can use for dynamic filtering."); + public List<String> listAcceptedFilterFields() { + List<String> acceptedFilterFields = new ArrayList<>(); + for (String partitionKey : catalogTable.getPartitionKeys()) { + // Only partition keys with supported types can be returned as accepted filter fields. + if (HiveSourceDynamicFileEnumerator.SUPPORTED_TYPES.contains( + catalogTable + .getSchema() + .getFieldDataType(partitionKey) + .map(DataType::getLogicalType) + .map(LogicalType::getTypeRoot) + .orElse(null))) { + acceptedFilterFields.add(partitionKey); } + } + + return acceptedFilterFields; + } - // sort before check to ensure the lists have same elements in same order - dynamicFilterPartitionKeys.sort(String::compareTo); - checkState( - this.dynamicFilterPartitionKeys == null - || this.dynamicFilterPartitionKeys.equals(dynamicFilterPartitionKeys), - "Dynamic filtering is applied twice but with different keys: %s != %s", - this.dynamicFilterPartitionKeys, - dynamicFilterPartitionKeys); - - this.dynamicFilterPartitionKeys = dynamicFilterPartitionKeys; - return dynamicFilterPartitionKeys; + @Override + public void applyDynamicFiltering(List<String> candidateFilterFields) { + if (catalogTable.isPartitioned()) { + dynamicFilterPartitionKeys = candidateFilterFields; } else { - LOG.warn("No dynamic filter field is accepted since the table is non-partitioned."); - return Collections.emptyList(); + throw new TableException( + String.format( + "Hive source table : %s is not a partition table, but try to apply dynamic filtering.", + catalogTable)); } } diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsDynamicFiltering.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsDynamicFiltering.java index ca84a9c90df..01a3fa5daf9 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsDynamicFiltering.java +++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsDynamicFiltering.java @@ -60,9 +60,19 @@ import java.util.List; public interface SupportsDynamicFiltering { /** - * Applies the candidate filter fields into the table source, and return the accepted fields. - * The data corresponding the filter fields will be provided in runtime, which can be used to - * filter the partitions or the input data. + * Return the filter fields this partition table source supported. This method is can tell the + * planner which fields can be used as dynamic filtering fields, the planner will pick some + * fields from the returned fields based on the query, and create dynamic filtering operator. */ - List<String> applyDynamicFiltering(List<String> candidateFilterFields); + List<String> listAcceptedFilterFields(); + + /** + * Applies the candidate filter fields into the table source. The data corresponding the filter + * fields will be provided in runtime, which can be used to filter the partitions or the input + * data. + * + * <p>NOTE: the candidate filter fields are always from the result of {@link + * #listAcceptedFilterFields()}. + */ + void applyDynamicFiltering(List<String> candidateFilterFields); } diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java index c56cca21ef9..b0767d27f05 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/DynamicPartitionPruningRule.java @@ -18,7 +18,6 @@ package org.apache.flink.table.planner.plan.rules.physical.batch; -import org.apache.flink.table.api.TableException; import org.apache.flink.table.api.config.OptimizerConfigOptions; import org.apache.flink.table.connector.source.DynamicTableSource; import org.apache.flink.table.connector.source.abilities.SupportsDynamicFiltering; @@ -139,29 +138,19 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config .collect(Collectors.toList()); } - List<String> acceptedFields = - ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(candidateFields); - - if (acceptedFields == null) { - return new ArrayList<>(); - } - - for (String field : acceptedFields) { - if (!candidateFields.contains(field)) { - throw new TableException( - String.format( - "Field: %s does not exist in the given fields: %s, " - + "please verify the applyDynamicFiltering method in connector: %s", - field, candidateFields, tableSource.asSummaryString())); - } - } + List<String> acceptedFilterFields = + DynamicPartitionPruningUtils.getSuitableDynamicFilteringFieldsInFactSide( + tableSource, candidateFields); + // Apply suitable accepted filter fields to source. + ((SupportsDynamicFiltering) tableSource).applyDynamicFiltering(acceptedFilterFields); if (factCalc == null) { - return acceptedFields.stream() + return acceptedFilterFields.stream() .map(f -> factScan.getRowType().getFieldNames().indexOf(f)) .collect(Collectors.toList()); } else { - return getAcceptedFieldsIndicesInCalc(acceptedFields, factJoinKeys, factCalc, factScan); + return getAcceptedFieldsIndicesInCalc( + acceptedFilterFields, factJoinKeys, factCalc, factScan); } } @@ -203,10 +192,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config List<Integer> acceptedFieldIndices = getAcceptedFieldIndices(factJoinKeys, factCalc, factScan, tableSource); - if (acceptedFieldIndices.isEmpty()) { - return null; - } - List<Integer> dynamicFilteringFieldIndices = new ArrayList<>(); for (int i = 0; i < joinInfo.leftKeys.size(); ++i) { if (acceptedFieldIndices.contains(factJoinKeys.get(i))) { @@ -293,9 +278,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, null, false); - if (newFactScan == null) { - return; - } final Join newJoin = join.copy(join.getTraitSet(), Arrays.asList(dimSide, newFactScan)); call.transformTo(newJoin); } @@ -348,9 +330,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, null, true); - if (newFactScan == null) { - return; - } final Join newJoin = join.copy(join.getTraitSet(), Arrays.asList(newFactScan, dimSide)); call.transformTo(newJoin); } @@ -409,9 +388,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, null, false); - if (newFactScan == null) { - return; - } final BatchPhysicalExchange newExchange = (BatchPhysicalExchange) exchange.copy( @@ -474,9 +450,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, null, true); - if (newFactScan == null) { - return; - } final BatchPhysicalExchange newExchange = (BatchPhysicalExchange) exchange.copy( @@ -601,9 +574,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, factCalc, true); - if (newFactScan == null) { - return; - } final BatchPhysicalCalc newCalc = (BatchPhysicalCalc) factCalc.copy( @@ -677,9 +647,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, factCalc, false); - if (newFactScan == null) { - return; - } final BatchPhysicalCalc newCalc = (BatchPhysicalCalc) factCalc.copy( @@ -757,9 +724,6 @@ public abstract class DynamicPartitionPruningRule extends RelRule<RelRule.Config final BatchPhysicalDynamicFilteringTableSourceScan newFactScan = createDynamicFilteringTableSourceScan(factScan, dimSide, join, factCalc, true); - if (newFactScan == null) { - return; - } final BatchPhysicalCalc newCalc = (BatchPhysicalCalc) factCalc.copy( diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java index c4167348ec6..ffad5054d5d 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java @@ -202,6 +202,27 @@ public class DynamicPartitionPruningUtils { } } + public static List<String> getSuitableDynamicFilteringFieldsInFactSide( + DynamicTableSource tableSource, List<String> candidateFields) { + List<String> acceptedFilterFields = + ((SupportsDynamicFiltering) tableSource).listAcceptedFilterFields(); + if (acceptedFilterFields == null || acceptedFilterFields.isEmpty()) { + return new ArrayList<>(); + } + + List<String> suitableFields = new ArrayList<>(); + // If candidateField not in acceptedFilterFields means dpp rule will not be matched, + // because we can not prune any partitions according to non-accepted filter fields + // provided by partition table source. + for (String candidateField : candidateFields) { + if (acceptedFilterFields.contains(candidateField)) { + suitableFields.add(candidateField); + } + } + + return suitableFields; + } + /** * Visit dim side to judge whether dim side has filter condition and whether dim side's source * table scan is non partitioned scan. diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java index 949b9f57e57..5254750a5f0 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java +++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java @@ -1302,22 +1302,13 @@ public final class TestValuesTableFactory } @Override - public List<String> applyDynamicFiltering(List<String> candidateFilterFields) { - if (dynamicFilteringFields != null && dynamicFilteringFields.size() != 0) { - checkArgument(!candidateFilterFields.isEmpty()); - acceptedPartitionFilterFields = new ArrayList<>(); - for (String field : candidateFilterFields) { - if (dynamicFilteringFields.contains(field)) { - acceptedPartitionFilterFields.add(field); - } - } + public List<String> listAcceptedFilterFields() { + return new ArrayList<>(dynamicFilteringFields); + } - return new ArrayList<>(acceptedPartitionFilterFields); - } else { - throw new UnsupportedOperationException( - "Should adding dynamic filtering fields by adding factor" - + " in with like: 'dynamic-filtering-fields' = 'a;b'."); - } + @Override + public void applyDynamicFiltering(List<String> candidateFilterFields) { + acceptedPartitionFilterFields = candidateFilterFields; } }
