Fix multi-join dynamic filtering

[rzeyde-varada]

It seems that dynamic filters may get "stuck" in join's node filter (when they can't be pushed down to scan nodes), e.g. see "lower" join node in the following plan:

> EXPLAIN SELECT k1 FROM t0, t1, t2 WHERE (k0 = k1) AND (k0 = k2) AND (v0 + v1 = v2)

Output[k1]
│   Layout: [k0:integer]
│   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
│   k1 := k0
└─ RemoteExchange[GATHER]
   │   Layout: [k0:integer]
   │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
   └─ InnerJoin[("k0" = "k2") AND ("expr_3" = "v2")][$hashvalue_7, $hashvalue_8]
      │   Layout: [k0:integer]
      │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
      │   Distribution: REPLICATED
      │   dynamicFilterAssignments = {k2 -> 421}
      ├─ Project[]
      │  │   Layout: [expr_3:real, k0:integer, $hashvalue_7:bigint]
      │  │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
      │  │   $hashvalue_7 := "combine_hash"("combine_hash"(bigint '0', COALESCE("$operator$hash_code"("k0"), 0)), COALESCE("$operator$hash_code"("expr_3"), 0))
      │  └─ Project[]
      │     │   Layout: [expr_3:real, k0:integer]
      │     │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
      │     │   expr_3 := ("v0" + "v1")
      │     └─ InnerJoin[("k0" = "k1") AND (("$internal$dynamic_filter_function"(("v0" + "v1"), 'EQUAL', '123') AND "$internal$dynamic_filter_function"(("v0" + "v1"), 'EQUAL', '422')) AND "$internal$dynamic_filter_function"(("v0" + "v1"), 'EQUAL', '254'))][$hashvalue, $hashvalue_4]
      │        │   Layout: [k0:integer, v0:real, v1:real]
      │        │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
      │        │   Distribution: REPLICATED
      │        │   dynamicFilterAssignments = {k1 -> 424}
      │        ├─ ScanFilterProject[table = memory:0, filterPredicate = ("@$internal$dynamic_filter_function|scalar|boolean|integer|varchar|varchar@$internal$dynamic_filter_function<t>(t,varchar,varchar):boolean"("k0", 'EQUAL', '421') AND "@$internal$dynamic_filter_function|scalar|boolean|integer|varchar|varchar@$internal$dynamic_filter_function<t>(t,varchar,varchar):boolean"("k0", 'EQUAL', '424'))]
      │        │      Layout: [k0:integer, v0:real, $hashvalue:bigint]
      │        │      Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}
      │        │      $hashvalue := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("k0"), 0))
      │        │      k0 := 0
      │        │      v0 := 1
      │        └─ LocalExchange[HASH][$hashvalue_4] ("k1")
      │           │   Layout: [k1:integer, v1:real, $hashvalue_4:bigint]
      │           │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
      │           └─ RemoteExchange[REPLICATE]
      │              │   Layout: [k1:integer, v1:real, $hashvalue_5:bigint]
      │              │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
      │              └─ ScanFilterProject[table = memory:1, filterPredicate = "@$internal$dynamic_filter_function|scalar|boolean|integer|varchar|varchar@$internal$dynamic_filter_function<t>(t,varchar,varchar):boolean"("k1", 'EQUAL', '421')]
      │                     Layout: [k1:integer, v1:real, $hashvalue_6:bigint]
      │                     Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}
      │                     $hashvalue_6 := "combine_hash"(bigint '0', COALESCE("$operator$hash_code"("k1"), 0))
      │                     k1 := 0
      │                     v1 := 1
      └─ LocalExchange[HASH][$hashvalue_8] ("k2", "v2")
         │   Layout: [k2:integer, v2:real, $hashvalue_8:bigint]
         │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
         └─ RemoteExchange[REPLICATE]
            │   Layout: [k2:integer, v2:real, $hashvalue_9:bigint]
            │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
            └─ ScanProject[table = memory:2]
                   Layout: [k2:integer, v2:real, $hashvalue_10:bigint]
                   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}/{rows: ? (?), cpu: ?, memory: 0B, network: 0B}
                   $hashvalue_10 := "combine_hash"("combine_hash"(bigint '0', COALESCE("$operator$hash_code"("k2"), 0)), COALESCE("$operator$hash_code"("v2"), 0))
                   k2 := 0
                   v2 := 1

In this case, Presto fails during the execution - since applying dynamic filters is implemented only for ScanFilterAndProject operators (and not LookupJoin operators):

[ERROR] Tests run: 17, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 47.153 s <<< FAILURE! - in io.prestosql.plugin.memory.TestMemorySmoke
[ERROR] testJoinDynamicFilteringMultiJoin(io.prestosql.plugin.memory.TestMemorySmoke)  Time elapsed: 1.321 s  <<< FAILURE!
java.lang.AssertionError: Execution of 'actual' query failed: SELECT k0, k1, k2 FROM t0, t1, t2 WHERE (k0 = k1) AND (k0 = k2) AND (v0 + v1 = v2)
	at org.testng.Assert.fail(Assert.java:83)
	at io.prestosql.testing.QueryAssertions.assertQuery(QueryAssertions.java:147)
	at io.prestosql.testing.QueryAssertions.assertQuery(QueryAssertions.java:103)
	at io.prestosql.testing.AbstractTestQueryFramework.assertQuery(AbstractTestQueryFramework.java:135)
	at io.prestosql.plugin.memory.TestMemorySmoke.testJoinDynamicFilteringMultiJoin(TestMemorySmoke.java:156)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:104)
	at org.testng.internal.Invoker.invokeMethod(Invoker.java:645)
	at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:851)
	at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1177)
	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:129)
	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:112)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.RuntimeException: java.lang.UnsupportedOperationException
	at io.prestosql.testing.AbstractTestingPrestoClient.execute(AbstractTestingPrestoClient.java:114)
	at io.prestosql.testing.DistributedQueryRunner.execute(DistributedQueryRunner.java:400)
	at io.prestosql.testing.QueryAssertions.assertQuery(QueryAssertions.java:144)
	... 16 more
Caused by: java.lang.UnsupportedOperationException
	at io.prestosql.sql.DynamicFilters$Function.dynamicFilter(DynamicFilters.java:206)
	at io.prestosql.$gen.JoinFilterFunction_20200128_223248_88.filter(Unknown Source)
	at io.prestosql.operator.StandardJoinFilterFunction.filter(StandardJoinFilterFunction.java:49)
	at io.prestosql.operator.JoinHash.isJoinPositionEligible(JoinHash.java:118)
	at io.prestosql.operator.PartitionedLookupSource.isJoinPositionEligible(PartitionedLookupSource.java:169)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.joinCurrentPosition(LookupJoinOperator.java:616)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.processProbe(LookupJoinOperator.java:547)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.lambda$processProbe$3(LookupJoinOperator.java:480)
	at io.prestosql.operator.PartitionedLookupSourceFactory$SpillAwareLookupSourceProvider.withLease(PartitionedLookupSourceFactory.java:437)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.processProbe(LookupJoinOperator.java:477)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.getOutput(LookupJoinOperator.java:392)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.process(LookupJoinOperator.java:688)
	at io.prestosql.operator.LookupJoinOperator$JoinProcessor.process(LookupJoinOperator.java:139)
	at io.prestosql.operator.WorkProcessorUtils$3.process(WorkProcessorUtils.java:319)
	at io.prestosql.operator.WorkProcessorUtils$ProcessWorkProcessor.process(WorkProcessorUtils.java:372)
	at io.prestosql.operator.WorkProcessorOperatorAdapter.getOutput(WorkProcessorOperatorAdapter.java:90)
	at io.prestosql.operator.Driver.processInternal(Driver.java:379)
	at io.prestosql.operator.Driver.lambda$processFor$8(Driver.java:283)
	at io.prestosql.operator.Driver.tryWithLock(Driver.java:675)
	at io.prestosql.operator.Driver.processFor(Driver.java:276)
	at io.prestosql.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:1075)
	at io.prestosql.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:163)
	at io.prestosql.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:484)
	at io.prestosql.$gen.Presto_null__testversion____20200128_223220_4.run(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)

IIUC, the fix may be to ignore dynamic filters that are left at the join nodes - since they should not affect correctness.

[sopel39]

lgtm % @raunaqmorarka comments

[rzeyde-varada]

Rebased and force-pushed (to re-run which has failed on seemingly unrelated test).

[sopel39]

merged, thanks!