Ensure dynamic filters are always delivered to coordinator #13695
Conversation
|
Is this a regression? Should this be marked as |
This part is definitely a regression as we weren't relying on |
I guess if we had JDBC DF in Master, then we would spot that. |
|
It shouldn't matter for pipelined execution, as with pipelined execution dynamic filters are reported before task is finished. For fault tolerant execution there's a small possibility that the execution might get stuck if the underlying connector waits indefinitely for dynamic filters. Is this the case for JDBC based connectors? |
JDBC will wait until timeout. However, waitinf until timeout for every query (even if build side completed quickly) would introduce regressions |
|
@sopel39 The likelihood of it happening is rather very small. As long as a query is capable to unblock on timeout it shouldn't be a big issue. |
| } | ||
|
|
||
| @GuardedBy("this") | ||
| private PerPipelineStatus per(int pipelineId) |
There was a problem hiding this comment.
Earlier updates to different pipeline status could not be done in parallel, now they can be (which appears to be a good thing), was this level of synchronisation unnecessary ?
There was a problem hiding this comment.
It looks like so, unless there was a necessity when the grouped execution was still there
core/trino-main/src/main/java/io/trino/execution/SqlTaskExecution.java
Outdated
Show resolved
Hide resolved
core/trino-main/src/main/java/io/trino/execution/SqlTaskExecution.java
Outdated
Show resolved
Hide resolved
core/trino-main/src/main/java/io/trino/execution/SqlTaskExecution.java
Outdated
Show resolved
Hide resolved
Remove DriverSplitRunnerFactory#Status
Currently dynamic filters are delivered via DynamicFilterFetcher. When a TaskStatus contains an updated dynamic filter version the DynamicFilterFetcher is responsible for fetching the new dynamic filters from a worker. However tasks were getting transitioned to FINISHED before the LocalDynamicFilterConsumer#setPartitionCount is set and dynamic filters updated. When task is transitioned to FINISHED the TaskStatusFetcher no longer tries to update TaskStatus so the coordinator is not able to learn that there is dynamic filters to be fetched.
Otherwise there is a chance that the task may finish before task holder is updated resulting in incomplete final TaskStatus being created. Since the TaskStatus is used to signal DynamicFilterFetcher to get dynamic filters the coordinator may never learn that there are dynamic filters to be retrieved.
arhimondr
force-pushed
the
fix-df-task-race
branch
from
e323ce1 to
2856bc4
Compare
|
@raunaqmorarka Updated |
Description
In certain scenarios dynamic filters were not getting delivered to coordinator as expected. This was causing problems with tests on CI (#13507).
The tests are configured to wait for dynamic filters indefinitely for consistency reasons. If filters are not delivered the query will never finish.
Currently dynamic filters are delivered via
DynamicFilterFetcher. When aTaskStatuscontains an updated dynamic filter version theDynamicFilterFetcheris responsible for fetching the new dynamic filters from a worker.In certain cases the dynamic filter version was never updated:
LocalDynamicFilterConsumer#setPartitionCountis set (addressed by theTransition task to FINISHED after noMoreOperators is setcommit).Update task holder before starting task executioncommit)Fix
Core engine
N/ARelated issues, pull requests, and links
#13507
#12152
Documentation
(X) No documentation is needed.
( ) Sufficient documentation is included in this PR.
( ) Documentation PR is available with #prnumber.
( ) Documentation issue #issuenumber is filed, and can be handled later.
Release notes
(X) No release notes entries required.
( ) Release notes entries required with the following suggested text: