Allow blocking page source until dynamic filters are ready #3414

rzeyde-varada · 2020-04-11T12:22:48Z

By default, we wait up to 1s for the build-side filters to be ready, to allow more efficient probe-side scan.
If the filters are not ready, we fallback to the previous behaviour (use current filters without blocking probe-side).

This feature can be configured using:

SET SESSION dynamic_filtering_probe_side_blocking_timeout='10s';

To disable, use:

SET SESSION dynamic_filtering_probe_side_blocking_timeout='0s';

rzeyde-varada · 2020-04-11T12:27:41Z

Using this feature, we can add (more) deterministic integration tests for dynamic filtering, e.g.:
https://github.com/prestosql/presto/blob/cb6ca89eca386cffb7c84933df5b22a786a03ee2/presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java#L95-L108

martint

Some initial comments

presto-main/src/main/java/io/prestosql/sql/planner/DynamicFilterSupplier.java

presto-main/src/main/java/io/prestosql/operator/ScanFilterAndProjectOperator.java

sopel39

I like the idea. However, I'm thinking that it should be the connector that decides whether to wait for dynamic filter and for how long. For example some connectors might be good producing data immediately since they have many splits to process (if few splits gets processed without dynamic filters it shouldn't be a big issue). Other connectors might have fixed and small amount of splits (e.g JDBC connectors) so it's preferred to wait for dynamic filter even much longer than 1 second global default. It could be huge difference if JDBC connector issues query with or without dynamic filter.
WDYT: @martint

presto-main/src/main/java/io/prestosql/sql/planner/DynamicFilter.java

sopel39 · 2020-05-12T09:09:36Z

To be landed after: #1072

rzeyde-varada · 2020-05-13T07:41:58Z

However, I'm thinking that it should be the connector that decides whether to wait for dynamic filter and for how long.

SGTM - I wasn't sure about adding a new connector API, so I've tried to add the minimal changes to the engine required by this feature.
I would be happy to adapt the PR to allow the connector to decide when to block and for how long.

sopel39 · 2020-05-13T08:53:58Z

I would be happy to adapt the PR to allow the connector to decide when to block and for how long

This requires making DynamicFilter part of an SPI. We can still fallback to the old method with Supplier dynamic filter.

rzeyde-varada · 2020-05-13T09:21:41Z

Rebased over latest master - to resolve a merge conflict.

rzeyde-varada · 2020-06-11T09:40:48Z

Rebased over latest master to resolve a merge conflict.

presto-main/src/main/java/io/prestosql/sql/planner/DynamicFilter.java

presto-hive/src/test/java/io/prestosql/plugin/hive/TestOrcPageSourceMemoryTracking.java

presto-main/src/main/java/io/prestosql/SystemSessionProperties.java

findepi · 2020-06-19T15:58:52Z

Other connectors might have fixed and small amount of splits (e.g JDBC connectors) so it's preferred to wait for dynamic filter even much longer than 1 second global default. It could be huge difference if JDBC connector issues query with or without dynamic filter.

I was not following, so this could have been answered already, please forgive me question in that case.

JDBC connectors to not consume the dynamic filter in any way. They just provide RecordCursor and let the engine handle the heavy lifting.
How would DF help a JDBC connector?

Second, are we positive this will not impact low latency queries? Ultimately, sleeping 1s seems to be heuristic and there seems to be a potential of a regression?

rzeyde-varada · 2020-07-05T12:02:03Z

Updated the PR to use DynamicFilter in ConnectorPageSourceProvider.
Please re-review :)

presto-spi/src/main/java/io/prestosql/spi/connector/DynamicFilter.java

presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java

sopel39 · 2020-07-06T13:37:49Z

presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java

+        Session session = Session.builder(getSession())
+                .setSystemProperty(ENABLE_DYNAMIC_FILTERING, "true")
+                .setSystemProperty(JOIN_DISTRIBUTION_TYPE, FeaturesConfig.JoinDistributionType.BROADCAST.name())
+                .setSystemProperty(PUSH_PARTIAL_AGGREGATION_THROUGH_JOIN, "true")


I have rewritten the test to use multi-join in order to verify the blocking behaviour:
https://github.com/prestosql/presto/pull/3414/files#diff-6102dfcf993241d1413b4fc63e4ca5a9R128
WDYT?

sopel39 · 2020-07-06T13:38:54Z

presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java

+                .setSystemProperty(PUSH_PARTIAL_AGGREGATION_THROUGH_JOIN, "true")
+                .build();
+        DistributedQueryRunner runner = (DistributedQueryRunner) getQueryRunner();
+        ResultWithQueryId<MaterializedResult> result = runner.executeWithQueryId(session, "SELECT MAX(quantity) FROM lineitem JOIN orders " +


add aggregation explicitly below join within query, e.g:

SELECT * FROM (SELECT MAX(quantity), orderkey FROM lineitem GROUP BY orderkey) JOIN orders on ...

Not sure I understand the suggestion - if I use the following query:

SELECT * FROM (SELECT MAX(quantity), orderkey FROM lineitem GROUP BY orderkey) l JOIN orders ON l.orderkey = orders.orderkey;

It results in a distributed join (due to a repartition operator between the TableScan of lineitem and MAX aggregation:

Fragment 1 [HASH] Output layout: [max, orderkey, custkey, orderstatus, totalprice, orderdate, orderpriority, clerk, shippriority, comment_10] Output partitioning: SINGLE [] Stage Execution Strategy: UNGROUPED_EXECUTION InnerJoin[("orderkey" = "orderkey_9")][$hashvalue_59, $hashvalue_60] │ Layout: [orderkey:bigint, max:double, custkey:bigint, orderstatus:varchar(1), totalprice:double, orderdate:date, orderpriority:varchar(15), clerk:varchar(15), shippriority:integer, comment_10:var │ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} │ Distribution: PARTITIONED │ dynamicFilterAssignments = {orderkey_9 -> df_283} ├─ Project[] │ │ Layout: [orderkey:bigint, max:double, $hashvalue_59:bigint] │ │ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} │ │ $hashvalue_59 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("orderkey"), 0)) │ └─ Aggregate[orderkey] │ │ Layout: [orderkey:bigint, max:double] │ │ Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?} │ │ max := max("quantity") │ └─ LocalExchange[HASH][$hashvalue] ("orderkey") │ │ Layout: [orderkey:bigint, quantity:double, $hashvalue:bigint] │ │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} │ └─ RemoteSource[2] │ Layout: [orderkey:bigint, quantity:double, $hashvalue_57:bigint] └─ LocalExchange[HASH][$hashvalue_60] ("orderkey_9") │ Layout: [orderkey_9:bigint, custkey:bigint, orderstatus:varchar(1), totalprice:double, orderdate:date, orderpriority:varchar(15), clerk:varchar(15), shippriority:integer, comment_10:varchar(79 │ Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?} └─ RemoteSource[3] Layout: [orderkey_9:bigint, custkey:bigint, orderstatus:varchar(1), totalprice:double, orderdate:date, orderpriority:varchar(15), clerk:varchar(15), shippriority:integer, comment_10:varchar

But then, local dynamic filtering doesn't take place - because it needs broadcast join...

sopel39 · 2020-07-06T13:42:14Z

presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSourceProvider.java

-                .map(page -> applyFilter(page, domains))
-                .collect(toList()));
+
+        return new FixedPageSource(pages)


Instead, add a wrapping page source, e.g:

DynamicFilteringPageSource

that would have FixedPageSource as a delegate field

Hope I did it correctly at https://github.com/prestosql/presto/pull/3414/files#diff-8b4a37d650086d97552712b506dfb6a8...

presto-main/src/main/java/io/prestosql/operator/TableScanOperator.java

presto-main/src/main/java/io/prestosql/split/PageSourceManager.java

presto-main/src/main/java/io/prestosql/sql/analyzer/FeaturesConfig.java

presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java

sopel39 · 2020-07-06T14:12:29Z

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java

    /**
     * May contains domains for dynamic filters for different table scans
     * (e.g. in case of co-located joins).
     */
    @GuardedBy("this")
    private Map<Symbol, Domain> dynamicFilterDomainsResult = new HashMap<>();

+    // Each future blocks until its dynamic filter is collected.
+    @GuardedBy("this")
+    private Map<DynamicFilterId, SettableFuture<Void>> futures = new HashMap<>();


This becomes very similar to DynamicFilterService from #4224.
Can we untangle these classes a bit?

LocalDynamicFiltersCollector should hold Map<DynamicFilterId, SettableFuture<Domain>> dynamicFilters

There should be method DynamicFilter LocalDynamicFiltersCollector#getDynamicFilter(List<DynamicFilters.Descriptor> dynamicFilters, Map<Symbol, ColumnHandle> columnHandles), similar as in DynamicFilterService

LocalDynamicFiltersCollector#addDynamicFilter(Map<Symbol, Domain> dynamicFilterDomains) would become addDynamicFilter(Map<DynamicFilterId, Domain> dynamicFilterDomains)

This would simplify LocalDynamicFilterConsumer as it wouldn't need to perform dynamic filter id -> symbol translation in io.prestosql.sql.planner.LocalDynamicFilterConsumer#convertTupleDomainForLocalFilters and some code could be reused between DynamicFilterService and LocalDynamicFiltersCollector

I did a bit different refactoring - allowing a bit more efficient implementation of DynamicFilter, by caching the resulting predicate and the blocked future:
https://github.com/prestosql/presto/pull/3414/files#diff-fc014adc6614290fe6f8a48d7b0f6739
WDYT?

rzeyde-varada · 2020-07-07T06:12:50Z

Many thanks for the review, will fix the issues and update the PR.

rzeyde-varada · 2020-07-14T11:53:30Z

Updated the PR to ae4a9e6 - please re-review :)

rzeyde-varada · 2020-08-13T10:35:41Z

Ping :)

sopel39

some comments until LocalDynamicFiltersCollector.java

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFilterConsumer.java

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java

rzeyde-varada · 2020-08-20T11:12:58Z

Many thanks for the review, will fix soon.

sopel39

some comments. Looks good overall.

presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java

presto-main/src/test/java/io/prestosql/sql/planner/TestLocalDynamicFiltersCollector.java

presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java

rzeyde-varada · 2020-08-22T18:45:56Z

I have addressed the comments above, and rebased the PR over the latest master (to resolve a merge conflict).

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java

presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java

sopel39

lgtm, minor comments + tests for large build side

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java

presto-memory/src/test/java/io/prestosql/plugin/memory/TestMemorySmoke.java

rzeyde-varada · 2020-08-25T16:26:42Z

Cherry-picked the commits from #4946, fixed the comments above and rebased.

sopel39 · 2020-08-25T18:57:22Z

there is a conflict. Please rebase. I will merge tomorrow then

It is enabled by default.

rzeyde-varada · 2020-08-25T19:00:13Z

Rebased over latest master to resolve the conflict - many thanks!

sopel39 · 2020-08-26T09:58:51Z

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java

+        // Iterate over dynamic filters that are collected (correspond to one of the futures), and required for filtering (correspond to one of the descriptors).
+        // It is possible that some dynamic filters are collected in a different stage - and will not available here.
+        // It is also possible that not all local dynamic filters are needed for this specific table scan.
+        Set<DynamicFilterId> filterIds = Sets.intersection(symbolsMap.keySet(), futures.keySet());


let's convert that into stream:

symbolsMap.keySet().stream() .filter(filterId -> futures.keySet()::contains) .map(...)

sopel39 · 2020-08-26T10:00:37Z

presto-main/src/main/java/io/prestosql/sql/planner/LocalExecutionPlanner.java

@@ -1998,6 +1999,13 @@ private PhysicalOperation createLookupJoin(
                Optional<Symbol> buildHashSymbol,
                LocalExecutionPlanContext context)
        {
+            // Register dynamic filters, allowing the scan operators to wait for the collection completion.
+            // Skip dynamic filters that are not used locally (e.g. in case of distributed joins).
+            Set<DynamicFilterId> localDynamicFilters = Sets.intersection(


Let's convert that into stream:

Set<DynamicFilterId> node.getDynamicFilters().keySet().stream() .filter(getConsumedDynamicFilterIds(probeNode)::contains) .collect(toImmutableSet())

sopel39 · 2020-08-26T10:01:45Z

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java

+        @GuardedBy("this")
+        private int futuresLeft;
+
+        public TableSpecificDynamicFilter(List<ListenableFuture<TupleDomain<ColumnHandle>>> predicateFutures)


make the constructor private

sopel39 · 2020-08-26T10:04:59Z

presto-memory/src/main/java/io/prestosql/plugin/memory/MemoryPageSourceProvider.java

+                return null;
+            }
+            Page page = delegate.getNextPage();
+            if (page != null) {


skip filtering when for predicate.isAll()

We allow the connector to block according to dynamic filter collection state. Some connectors would prefer to wait a bit, to benefit from more selective scanning (using the collected dynamic filters).

rzeyde-varada · 2020-08-26T10:53:02Z

Thanks!
Fixed the comments and force-pushed d5bb8c8.

sopel39 · 2020-08-26T12:29:08Z

merged, thanks!

rzeyde-varada · 2020-08-26T12:53:17Z

Much appreciated :)

cla-bot bot added the cla-signed label Apr 11, 2020

rzeyde-varada force-pushed the df-block-probe-side branch from cb6ca89 to b2b84a7 Compare April 11, 2020 21:07

martint reviewed Apr 16, 2020

View reviewed changes

rzeyde-varada force-pushed the df-block-probe-side branch from b2b84a7 to 9496899 Compare April 17, 2020 15:14

martint self-requested a review April 23, 2020 16:31

sopel39 reviewed May 12, 2020

View reviewed changes

presto-main/src/main/java/io/prestosql/sql/planner/DynamicFilter.java Outdated Show resolved Hide resolved

presto-main/src/main/java/io/prestosql/sql/planner/DynamicFilter.java Outdated Show resolved Hide resolved

rzeyde-varada force-pushed the df-block-probe-side branch from 9496899 to 17b1dd4 Compare May 13, 2020 09:19

sopel39 mentioned this pull request Jun 7, 2020

Fix join short circuit when probe is empty #3957

Open

rzeyde-varada force-pushed the df-block-probe-side branch from 17b1dd4 to 16cd01a Compare June 11, 2020 08:37

sopel39 reviewed Jun 19, 2020

View reviewed changes

rzeyde-varada force-pushed the df-block-probe-side branch 3 times, most recently from 0e5e831 to 0e587ab Compare July 5, 2020 07:44

sopel39 reviewed Jul 6, 2020

View reviewed changes

rzeyde-varada force-pushed the df-block-probe-side branch 3 times, most recently from c9e984b to ae4a9e6 Compare July 12, 2020 12:06

rzeyde-varada mentioned this pull request Jul 23, 2020

Allow dynamic filtering with comparison operators #2674

Merged

sopel39 reviewed Aug 19, 2020

View reviewed changes

presto-main/src/main/java/io/prestosql/sql/planner/LocalDynamicFiltersCollector.java Outdated Show resolved Hide resolved

sopel39 reviewed Aug 20, 2020

View reviewed changes

rzeyde-varada force-pushed the df-block-probe-side branch from ae4a9e6 to ba93c07 Compare August 22, 2020 17:58

sopel39 reviewed Aug 24, 2020

View reviewed changes

sopel39 approved these changes Aug 24, 2020

View reviewed changes

rzeyde-varada force-pushed the df-block-probe-side branch from ba93c07 to 6b8b4fc Compare August 24, 2020 19:39

sopel39 approved these changes Aug 25, 2020

View reviewed changes

rzeyde-varada force-pushed the df-block-probe-side branch from 6b8b4fc to 9547911 Compare August 25, 2020 15:37

rzeyde-varada force-pushed the df-block-probe-side branch from 9547911 to d9c4edb Compare August 25, 2020 16:26

Do not enable dynamic filtering explicitly in TestMemorySmoke

88691c7

It is enabled by default.

rzeyde-varada force-pushed the df-block-probe-side branch from d9c4edb to e23dae7 Compare August 25, 2020 18:59

sopel39 reviewed Aug 26, 2020

View reviewed changes

rzeyde-varada added 2 commits August 26, 2020 13:50

Allow blocking page source until dynamic filters are ready

572115e

We allow the connector to block according to dynamic filter collection state. Some connectors would prefer to wait a bit, to benefit from more selective scanning (using the collected dynamic filters).

Support lazy dynamic filtering at memory connector

d5bb8c8

rzeyde-varada force-pushed the df-block-probe-side branch from e23dae7 to d5bb8c8 Compare August 26, 2020 10:51

rzeyde-varada changed the title ~~Block probe-side scan until build-side dynamic filtering is ready~~ Allow blocking page source until dynamic filters are ready Aug 26, 2020

sopel39 merged commit 7fa21f6 into trinodb:master Aug 26, 2020

sopel39 mentioned this pull request Aug 26, 2020

Release notes for 341 #4755

Closed

9 tasks

rzeyde-varada deleted the df-block-probe-side branch August 26, 2020 12:53

martint added this to the 341 milestone Sep 9, 2020

rzeyde-varada mentioned this pull request Nov 12, 2021

Some joins never complete with dynamic filtering enabled #9917

Closed

Allow blocking page source until dynamic filters are ready #3414

Allow blocking page source until dynamic filters are ready #3414

Conversation

rzeyde-varada commented Apr 11, 2020

rzeyde-varada commented Apr 11, 2020

martint left a comment

Choose a reason for hiding this comment

sopel39 left a comment

Choose a reason for hiding this comment

sopel39 commented May 12, 2020

rzeyde-varada commented May 13, 2020

sopel39 commented May 13, 2020

rzeyde-varada commented May 13, 2020

rzeyde-varada commented Jun 11, 2020

findepi commented Jun 19, 2020

rzeyde-varada commented Jul 5, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rzeyde-varada commented Jul 7, 2020

rzeyde-varada commented Jul 14, 2020

rzeyde-varada commented Aug 13, 2020

sopel39 left a comment

Choose a reason for hiding this comment

rzeyde-varada commented Aug 20, 2020

sopel39 left a comment

Choose a reason for hiding this comment

rzeyde-varada commented Aug 22, 2020

sopel39 left a comment

Choose a reason for hiding this comment

rzeyde-varada commented Aug 25, 2020

sopel39 commented Aug 25, 2020

rzeyde-varada commented Aug 25, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rzeyde-varada commented Aug 26, 2020

sopel39 commented Aug 26, 2020

rzeyde-varada commented Aug 26, 2020