Dynamic filter pushdown for selective broadcast joins

[mbasmanova]

This commit introduces a mechanism to push down dynamically generated filters from one operator to another operator upstream. The new mechanism is used to push down a filter on the join keys from a highly-selective broadcast (or simply colocated) hash join into probe-side table scan. This allows the ORC reader to skip whole files and/or row groups based on the extra filter.

The workflow is:

• HashProbe operator waits for the HashBulid operator to load all the data and build a hash table.
• HashProbe operator checks if join keys in the hash table are low-cardinality, e.g. table_->hashMode() != BaseHashTable::HashMode::kHash. If not, no push-down will occur.
• HashProbe operator asks the Driver if it can push down filters on the join key columns. If not, no push-down will occur. It is possible that Driver can push down filter for some join keys, but not all, in which case HashProbe will produce filters only for these keys.
• HashProbe starts processing the input and monitor the selectivity of individual join keys, e.g. how many rows build-side VectorHasher is dropping for each key.
• Join keys whose corresponding built-side VectorHashers drop at least 1/3 of rows (after seeing at least 10K rows of input) are used to generated filters.
• Driver loop periodically checks if an operator has generated dynamic filters and, if so, pushes them down.
• In this change, only TableScan accepts dynamic filters. Since Exchange operator doesn't accept dynamic filter, filter pushdown doesn't happen for non-broadcast joins.

Individual changes that make up the new mechanism are:

• Extend connector interface to add DataSource::addDynamicFilter() method to add dynamically generated filter.
• Update Hive connector to add dynamic filters to the scanSpec_ before processing the next batch of rows. If a column already has a filter, dynamic filter is merged into an existing filter using Filter::mergeWith().
• The new Driver::canPushdownFilters() method returns whether it is possible to pushdown filters generated by a given operator for the specified columns. To push down a filter, there should be an upstream operator that accepts dynamic filters, e.g. Operator::canAddDynamicFilter() returns true, and the filter column should not be identity-projected from that operator all the way to the filter producing operator.
• Modified driver loop to check if an operator produced some dynamic filters and, if so, push them down.
• Modified HashProbe operator to generate dynamic filters for join keys which have highly-selective VectorHashers on the build side. This logic applies only to inner joins with array-based lookup on the build side.
• Modified TableScan operator to pass dynamic filters to the connector.

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[oerling]

You might prefer an API contract that has the filters take effect on the next batch, not the next split. This is for the case where the filter replaces the whole operator, you need to know when it takes effect.

[mbasmanova]

You might prefer an API contract that has the filters take effect on the next batch, not the next split. This is for the case where the filter replaces the whole operator, you need to know when it takes effect.

I was thinking about that. What would it take to allow changing ScanSpec mid-split?

[oerling]

You would need to keep a pointer to the ScanSpec of the current data source and you could then install the filter right there. Then you would call the predicate ordering function of ScanSpec. The next read of the top level struct would probably have the new filter as first. This could work, it would find its place soon enough. Would have to step through the mechanics but I don’t think there is anything any particular difficulty.

[mbasmanova]

@oerling Orri, thank you for reviewing. I appreciate the feedback. I updated the PR to apply the pushed down filter to the next batch vs. next split. With this we can make a follow-up PR to turn the join into a no-op under certain conditions after pushing filter down. Would you take another look?

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[oerling]

When filters are installed, it is not obvious if cached filtering results with dictionaries get reset. This will be somewhat rare but could happen if many joins add filters to the same dictionary encoded column or if there already is a non-join filter on the dict. If there was no filter before pushdown, it may be that the dictionary is loaded but the filter cache is not inited, which could crash. The ScanSpec does not reference the reader. But it can have a version number that the dictionary readers will then check. Just a ‘changed’ flag would do.

[oerling]

Maybe in the documentation one could say colocated instead of broadcast since this also applies too copartitioned cases.

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[facebook-github-bot]

@mbasmanova has imported this pull request. If you are a Facebook employee, you can view this diff on Phabricator.

[mbasmanova]

@oerling Orri, thanks for bring up the issue with filter caches. I added SelectiveColumnReader::resetFilterCaches() and DwrfRowReader::columnReader() methods and updated HiveDataSource::addDynamicFilter to use these:

  auto columnReader =
      dynamic_cast<SelectiveColumnReader*>(rowReader_->columnReader());
  columnReader->resetFilterCaches();

[facebook-github-bot]

@mbasmanova merged this pull request in a6b6dca.