THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
Status
Current state: "Under Discussion"
Discussion thread:
JIRA: here (<- link to )
...
| Discussion thread | https://lists.apache.org/thread/v0b8pfh0o7rwtlok2mfs5s6q9w5vw8h6 | ||||||||
|---|---|---|---|---|---|---|---|---|---|
| Vote thread | |||||||||
| JIRA |
| ||||||||
| Release | 1.16 |
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
Currently, the Flink source operator does not have any input edge, and the splits are generated in SplitEnumerator for new source (FLIP-27). Based on the current mechanism, the dynamic partition pruning will happen in SplitEnumerator (such as: filesystem connector or hive connector). I in the future the source operator supports input, the input data could support dynamic filtering (such as: filter row-group for parquet based on statistics, or even filter records based on bloom filter).
In this FLIP we will introduce a mechanism for detecting dynamic partition pruning patterns in optimization phase and performing partition pruning at runtime by sending the dimension table results to the SplitEnumerator of fact table via existing coordinator existing coordinator mechanism.
In above example, the result from date_dim which d_year is 2000 will be send to join build side and the Coordinator, and the SplitEnumerator of store_returns will filter the partitions based on the filtered date_dim data, and send the real required splits to the source scan of store_returns.
...
- The parser parses a given query, convert to an AST (Abstract Syntax Tree) plan. The optimizer will detect with DPP pattern and transform the plan to via planner rules, and then the optimized physical plan will be converted to ExecNode plan. The ExecNode plan will be converted to StreamingGraph and then JobGraph.
- The client submits the generated JobGraph to the job manager.
- The JM schedules the job vertices for dim-source and DynamicFilteringDataCollector first. The filtered data from dim-source operator will be send both to DynamicFilteringDataCollector operator and the shuffle input of Join operator.
- DynamicFilteringDataCollector collects the input data, removes the irrelevant column data and duplicated records, and then sends the records (wrapped in DynamicFilteringEvent) to its OperatorCoordinator once it collects all input data in finish method. The coordinator delivers the event to the SourceCoordinators of the relating fact-source operator, then the SourceCoordinator delivers the DynamicFilteringEvent to DynamicFileSplitEnumerator.
- The
DynamicFileSplitEnumeratorfinds the the relevant partitions from the all partitions via the data from dim-source, and creates the target splits for fact-source. - The JM schedules the job vertices for fact-source and Placeholder operator.
- The fact-source gets the splits from
DynamicFileSplitEnumerator, reads the data, send the shuffle input of Join operator. - The join operator reads the input data and does join operation.
New public interfaces
...
@PublicEvolving
public interface SupportsDynamicFiltering {
/**
* applyReturns the candidate filter fields intothis thepartition table source, andsupported. returnThis themethod acceptedcan fields.tell Thethe
* data corresponding* theplanner filterwhich fields willcan be used as provideddynamic infiltering runtimefields, whichthe planner canwill bepick usedsome
to filter
* fields from the partitions returned fields based on the query, and thecreate dynamic inputfiltering dataoperator.
*/
List<String> applyDynamicFilteringlistAcceptedFilterFields(List<String> candidateFilterFields);
}
Since the split enumerator is executed in JM, the filtered partition data from dim-source operator should be send to the split enumerator via existing coordinator mechanism. So we introduce DynamicFilteringEvent to wrap the filtering data.
...
...
...
/**
...
...
...
* 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);
}
Since the split enumerator is executed in JM, the filtered partition data from dim-source operator should be send to the split enumerator via existing coordinator mechanism. So we introduce DynamicFilteringEvent to wrap the filtering dataDynamicFilteringData wraps the values without the schema, the value order is consistent with the argument value of SupportsDynamicFiltering#applyDynamicFiltering method.
@PublicEvolving
public class DynamicFilteringDataDynamicFilteringEvent implements SerializableSourceEvent {
private final TypeInformation<RowData>DynamicFilteringData typeInfodata;
privatepublic final RowType rowTypeDynamicFilteringEvent(DynamicFilteringData data) {;
private final List<byte[]> serializedData;
this.data private final boolean exceedThreshold= data;
private transient List<RowData> data;}
public RowTypeDynamicFilteringData getRowTypegetData() {
return rowTypedata;
}
public@Override
Optional<List<RowData>> getData() {
public if (exceedThresholdString toString() {
return "DynamicFilteringEvent{" + "data=" + data return Optional.empty()+ '}';
}
}
DynamicFilteringData wraps the values without the schema, the value order is consistent with the argument value of SupportsDynamicFiltering#applyDynamicFiltering method.
@PublicEvolving
public class DynamicFilteringData implements }Serializable {
private final if (data == null) {TypeInformation<RowData> typeInfo;
private final RowType rowType;
private final List<byte[]> serializedData;
private datafinal = new ArrayList<>()boolean exceedThreshold;
private transient List<RowData> data;
public TypeSerializer<RowData> serializer = typeInfo.createSerializer(new ExecutionConfig());RowType getRowType() {
return rowType;
for (byte[] bytes : serializedData) {
}
public Optional<List<RowData>> getData() {
if (exceedThreshold) {
try (ByteArrayInputStream bais = newreturn ByteArrayInputStreamOptional.empty(bytes);
}
if (data == null) {
DataInputViewStreamWrapper inView = new DataInputViewStreamWrapper(bais)) {
data = new ArrayList<>();
TypeSerializer<RowData> serializer RowData partition = serializer.deserialize(inViewtypeInfo.createSerializer(new ExecutionConfig());
for (byte[] bytes : serializedData) {
data.add(partition);
try (ByteArrayInputStream bais }= catchnew ByteArrayInputStream(Exception ebytes);
{
throw new TableException("Unable toDataInputViewStreamWrapper deserializeinView the value.", e);= new DataInputViewStreamWrapper(bais)) {
}
RowData partition = serializer.deserialize(inView);
}
}
return Optional.of(datadata.add(partition);
}
public} booleancatch contains(RowDataException rowe) {
// ...throw new TableException("Unable to deserialize the value.", e);
}
}
Currently, for FileSystem connector and Hive connector, the splits are created by FileEnumerator. However, existing FileEnumerators do not match the requirement, therefore we introduce a new FileEnumerator named DynamicFileEnumerator to support creating splits based on partition.
...
}
...
...
}...
...
...
...
...
}
...
...
return Optional.of(data);
}
public boolean
...
contains(RowData row) {
...
// ...
}
}
Currently, for FileSystem connector and Hive connector, the splits are created by FileEnumerator. However, existing FileEnumerators do not match the requirement, therefore we introduce a new FileEnumerator named DynamicFileEnumerator to support creating splits based on partition.
@PublicEvolving
public interface DynamicFileEnumerator extends FileEnumerator {
void setDynamicFilteringData(DynamicFilteringData data);
/** Provider for DynamicFileEnumerator. */
@FunctionalInterface
interface Provider extends FileEnumerator.Provider {
We introduce a config option to enable/disable this feature, default is true.
@Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
public static final ConfigOption<Boolean> TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED =
key("table.optimizer.dynamic-filtering.enabled")
.booleanType()
DynamicFileEnumerator .defaultValuecreate(true);
}
}
We introduce a config option to enable/disable this feature, default is true.
@Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
public static final .withDescription(
ConfigOption<Boolean> TABLE_OPTIMIZER_DYNAMIC_FILTERING_ENABLED =
key("table.optimizer.dynamic-filtering.enabled")
.booleanType()
"When it is true, the optimizer will try to push dynamic filtering into scan table source," .defaultValue(true)
.withDescription(
+ " the irrelevant partitions or input data will"When beit filteredis to reduce scan I/O in runtime.");
true, the optimizer will try to push dynamic filtering into scan table source,"
+ " the irrelevant partitions or input data will be filtered to reduce scan I/O in runtime.");
We can find whether a query successfully applied the DPP optimization via We can find whether a query successfully applied the DPP optimization via EXPLAIN result. The following snippets describes the explain result for the above query. (The optimized plan will contains DynamicFilteringDataCollector.)
...
== Optimized Physical Plan ==
HashJoin(joinType=[InnerJoin], where=[=(sr_returned_date_sk, d_date_sk)], select=[sr_returned_date_sk, sr_return_time_sk, sr_item_sk, d_date_sk, d_date_id, d_year], build=[right])
:- Exchange(distribution=[hash[sr_returned_date_sk]])
: +- DynamicFilteringTableSourceScan(table=[[test-catalog, default, store_returns]], fields=[sr_returned_date_sk, sr_return_time_sk, sr_item_sk])
: +- DynamicFilteringDataCollector(fields=[d_date_sk])
: +- Calc(select=[d_date_sk, d_date_id, d_year], where=[=(d_year, 12000)])
: +- TableSourceScan(table=[[test-catalog, default, date_dim]], fields=[d_date_sk, d_date_id, d_year])
+- Exchange(distribution=[hash[xd_date_sk]])
+- Calc(select=[d_date_sk, d_date_id, d_year], where=[=(d_year, 2000)])
+- TableSourceScan(table=[[test-catalog, default, date_dim]], fields=[d_date_sk, d_date_id, d_year])
...
- The optimization results are non-deterministic,
- Users can't visually see the data dependencies between them from the topology.
- Add a dependence operator (named Placeholder-Filter) which input is fact-source operator and DynamicFilteringDataCollector operator, this could ensure the DynamicFilteringDataCollector operator is scheduled before the fact-source operator . The dependency edge could be omitted if the fact source is chained with the join operator via the multiple-inputs tasks and there is already an edge between dim source and fact source.
...
The following graph describes the plan before transformation and plan after transformation for StreamGraph generated from the above query.
Connector changes
The following graph lists the new introduced classes (marked as green color) for Hive connector to support DPP. FileSystem collector has similar class structure, I won't repeat it here.
- DynamicFileSplitEnumerator will handle the SourceEvent from SourceCoordinator and get the PartitionData from SourceEvent if it's a DynamicFilteringEvent, create the DynamicFileEnumerator to get the relevant partitions based on PartitionData. All split requests will be add to buffer to wait for the DynamicFilteringEvent to be handled and then start to handle the awaiting requests. The pseudocode looks like:
...
DynamicFilteringDataCollector operator implementation
DynamicFilteringDataCollector operator collects the filtered dim source data, stores the relevant column data in memory buffer, the duplicated data will be removed. Once the memory threshold size (default is 32M) is exceeded, the data behind will be discarded, the
DynamicFilteringData will also be marked as `exceedThreshold=true`, which means no partition or input data can be pruned. This could avoid exceeding the akka's message size limit and avoid too much data being send to JM leading to OOM.
public class DynamicFilteringDataCollectorOperator extends AbstractStreamOperator<Object>
implements OneInputStreamOperator<RowData, Object> {
private final RowType dynamicFilteringFieldType;
private final List<Integer> dynamicFilteringFieldIndices;
private final long threshold;
private transient long currentSize;
/** Use Set instead of List to ignore the duplicated records. */
private transient Set<byte[]> buffer;
private transient TypeInformation<RowData> typeInfo;
private transient TypeSerializer<RowData> serializer;
private transient boolean eventSend;
private final OperatorEventGateway operatorEventGateway;
@Override
public void processElement(StreamRecord<RowData> element) throws Exception {
if (currentSize > threshold) {
return;
}
GenericRowData rowData = new GenericRowData(dynamicFilteringFieldIndices.size());
// fill the rowData from the given element based on dynamicFilteringFieldIndices
// serialize the rowData
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(baos);
serializer.serialize(rowData, wrapper);
boolean duplicated = !buffer.add(baos.toByteArray());
if (duplicated) {
return;
}
currentSize += baos.size();
if (exceedThreshold()) {
sendEvent();
}
}
private boolean exceedThreshold() {
return threshold > 0 && currentSize > threshold;
}
public void finish() throws Exception {
sendEvent();
}
private void sendEvent() {
if (eventSend) {
return;
}
final DynamicFilteringData dynamicFilteringData;
if (exceedThreshold()) {
dynamicFilteringData = new DynamicFilteringData(typeInfo, dynamicFilteringFieldType, new ArrayList<>(), true);
} else {
dynamicFilteringData = new DynamicFilteringData(typeInfo, dynamicFilteringFieldType, new ArrayList<>(buffer), false);
}
DynamicFilteringEvent event = new DynamicFilteringEvent(dynamicFilteringData);
operatorEventGateway.sendEventToCoordinator(new SourceEventWrapper(event));
this.eventSend = true;
}
}
Connector changes
The following graph lists the new introduced classes (marked as green color) for Hive connector to support DPP. FileSystem collector has similar class structure, I won't repeat it here.
- DynamicFileSplitEnumerator will try it best to do partition pruning based on received DynamicFilteringEvent. If the split requests arrives before SourceEvent, the split assigner will also assign a split based on all partitions. Once the DynamicFilteringEvent arrives, a new split assigner will be created, the only the relevant split will be assigned for the following split requests. From the "Scheduling deadlock" section we can know, the dim-side operators will be scheduled first, this could make sure the DynamicFilteringEvent arrives DynamicFileSplitEnumerator before split requests from fact-source operator for most cases. If the placeholder operator can't be added and fact-source can't be chained with its output operator (such as: operator chain is explicitly disabled), this best-effort strategy can avoid scheduling deadlock. The pseudocode looks like:
@Internal
public class DynamicFileSplitEnumerator<SplitT extends FileSourceSplit>
implements SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>>,
SupportsHandleExecutionAttemptSourceEvent {
private final SplitEnumeratorContext<SplitT> context;
private final DynamicFileEnumerator.Provider fileEnumeratorFactory;
private final FileSplitAssigner.Provider splitAssignerFactory;
private transient FileSplitAssigner splitAssigner;
private final Set<SplitT> assignedSplits;
@Override
public void handleSplitRequest(int subtask, @Nullable String hostname) {
if (!context.registeredReaders().containsKey(subtask)) {
// reader failed between sending the request and now. skip this request.
return;
}
if (splitAssigner == null) {
createSplitAssigner(null);
}
final Optional<FileSourceSplit> nextSplit = getNextSplit(hostname);
if (nextSplit.isPresent()) {
final FileSourceSplit split = nextSplit.get();
context.assignSplit((SplitT) split, subtask);
assignedSplits.add((SplitT) split);
} else {
context.signalNoMoreSplits(subtask);
}
}
private Optional<FileSourceSplit> getNextSplit(String hostname) {
do {
final Optional<FileSourceSplit> nextSplit = splitAssigner.getNext(hostname);
if (nextSplit.isPresent()) {
// ignore the split if it has been assigned
...
...
if (!assignedSplits.contains(nextSplit.get())) {
...
...
...
...
...
...
...
...
...
...
...
return ...
nextSplit;
...
...
...
...
...
...
...
...
...
...
...
...
}
...
...
...
...
...
...
}
...
else ...
{
...
...
...
...
...
...
...
...
...
...
...
...
return nextSplit;
...
...
...
...
...
...
...
...
}
...
}
...
while (
...
true);
...
}
@Override
public void
...
handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
if (sourceEvent instanceof
...
DynamicFilteringEvent) {
...
createSplitAssigner(((DynamicFilteringEvent) sourceEvent).getData());
...
...
...
...
...
}
}
...
@Override...
public void handleSourceEvent(int subtaskId, int attemptNumber, SourceEvent sourceEvent) {
...
if (sourceEvent instanceof DynamicFilteringEvent) {
...
...
LOG.info("Received DynamicFilteringEvent: {}", subtaskId);
...
createSplitAssigner(((DynamicFilteringEvent) sourceEvent).getData());
...
}
}
private void createSplitAssigner(@Nullable DynamicFilteringData dynamicFilteringData) {
...
DynamicFileEnumerator
...
fileEnumerator =
...
fileEnumeratorFactory.
...
create();
if (dynamicFilteringData != null) {
...
...
...
fileEnumerator.
...
setDynamicFilteringData(
...
dynamicFilteringData);
...
}
...
...
...
// create splits
...
splitAssigner = splitAssignerFactory.create(splits);
...
}
}
- DynamicFileEnumerator will do the partition filter of all partitions based on DynamicFilteringData, the pseudocode for HiveSourceDynamicFileEnumerator looks like:
public class HiveSourceDynamicFileEnumerator
...
implements ...
DynamicFileEnumerator {
...
private final List<String> dynamicPartitionKeys;
...
// For non-partition hive table, partitions only contains one partition which partitionValues is empty.
private
...
final ...
List<HiveTablePartition> allPartitions;
private final int threadNum;
...
...
private final ...
JobConf jobConf;
...
private transient List<HiveTablePartition>
...
finalPartitions;
public void
...
setDynamicFilteringData(
...
DynamicFilteringData data) {
...
...
try {
...
finalPartitions = new ArrayList<>(
...
);
...
Optional<List<RowData>> receivedDataOpt =
...
data.
...
getData(
...
);
...
if (!receivedDataOpt.isPresent()) {
...
...
...
...
//
...
the
...
DynamicFilteringData is too large
...
...
...
...
...
finalPartitions =
...
allPartitions;
...
...
...
...
- DynamicFileEnumerator will do the partition filter of all partitions based on PartitionData, the pseudocode looks like:
...
return;
}
...
...
...
...
...
RowType
...
rowType
...
= data.getRowType();
...
...
for ...
(...
HiveTablePartition partition : allPartitions) {
...
...
...
...
...
RowData rowData =
...
createRowData(
...
rowType, partition
...
);
...
...
...
if ...
(data.
...
contains(rowData))
...
{
finalPartitions.add(partition);
...
...
...
}
}
} catch
...
(Exception e) {
finalPartitions =
...
allPartitions;
}
}
private RowData
...
createRowData(RowType rowType, HiveTablePartition partition)
...
{
// convert the
...
partition to RowData based on the given row
...
type
}
...
@Override
...
public Collection<FileSourceSplit> enumerateSplits(Path[] paths, int minDesiredSplits) throws IOException {
...
return new ArrayList<>(createInputSplits(minDesiredSplits, finalPartitions
...
, threadNum, jobConf));
...
}
}Compatibility, Deprecation, and Migration Plan
...