Adding FLUSHING state to tasks and stages #4290

rohangarg · 2020-07-01T06:45:51Z

Adds a flushing state to tasks and stages.
This is useful in case of broadcast joins where if the build source stage is done processing and is just running to serving data from its output buffer to join tasks, its sub-stages can be shutdown by cancellation (since the new output produced by the sub stages will not be useful). The cancellation is helpful since the sub stages under build source can be complex and take significant memory (the output buffer size is configurable) and CPU (due to the processing) to get to a blocked state.

For tasks, flushing state signifies that there will be no new drivers, the existing drivers have finished and the output buffer of the task is at-least in a 'no-more-pages' state.
For stages, flushing state signifies that at-least one of its task is flushing and the non-flushing tasks are finished.
For both, once the state has started flushing it will never go back to 'RUNNING' state.

findepi · 2020-07-01T09:01:53Z

The cancellation is helpful since the sub stages under build source can be complex and take significant memory (the output buffer size is configurable) and CPU (due to the processing) to get to a blocked state.

If the build side completed building the lookup, the source is finished, otherwise the build side would not know the lookup is "full".
If there are computations lingering in the lower stages, this seems not specific to join build side.

To put differently, I do not understand the situation this is helping with.

rohangarg · 2020-07-01T09:34:54Z

If the build side completed building the lookup, the source is finished, otherwise the build side would not know the lookup is "full".
If there are computations lingering in the lower stages, this seems not specific to join build side.
To put differently, I do not understand the situation this is helping with.

consider a broadcast join scenario where the build side is very small (1KB) and is computed and stored in build source's output buffer. Being a broadcast buffer, it will keep running until it is told that no new tasks will be added in the join stage - since it would have to serve the new join tasks if created. There are scenarios where it will never be told that no new tasks are coming (that logic will be improved as well in future changes). Now since the build source is running, its sub stages will also keep on running until they get blocked (because the build source is not consuming their output). The time and memory taken to reach to blocking state can be significant.

sopel39

great job! left some comments. I think we need to improve testing for new FLUSHING state

presto-main/src/main/java/io/prestosql/execution/StageState.java

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

presto-main/src/main/java/io/prestosql/execution/StageState.java

@@ -112,4 +116,9 @@ public boolean canScheduleMoreTasks()
        }
        return true;
    }
+
+    public boolean isRunning()


Rename it to:

isRunningOrFlushing

sopel39 · 2020-07-06T14:25:41Z

presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java

@@ -524,6 +536,12 @@ else if (taskState == TaskState.FINISHED) {
        }
    }

+    private boolean isFlushing()


make it synchronized

sopel39 · 2020-07-06T14:31:15Z

presto-main/src/main/java/io/prestosql/execution/TaskState.java

@@ -64,4 +68,9 @@ public boolean isDone()
    {
        return doneState;
    }
+
+    public boolean isRunning()


rename to isRunningOrFlushing

sopel39 · 2020-07-06T14:31:39Z

presto-main/src/main/java/io/prestosql/execution/TaskState.java

@@ -30,6 +30,10 @@
     * Task is running.
     */
    RUNNING(false),
+    /**
+     * Task has finished executing and output is left to be consumed


add a sentence from PR description:

Flushing state signifies that there will be no new drivers, the existing drivers have finished and the output buffer of the task is at-least in a 'no-more-pages' state.

sopel39 · 2020-07-06T14:35:17Z

presto-main/src/main/java/io/prestosql/execution/MemoryRevokingScheduler.java

@@ -218,7 +218,7 @@ private boolean memoryRevokingNeeded(MemoryPool memoryPool)
    private long getMemoryAlreadyBeingRevoked(Collection<SqlTask> sqlTasks, MemoryPool memoryPool)
    {
        return sqlTasks.stream()
-                .filter(task -> task.getTaskStatus().getState() == TaskState.RUNNING)
+                .filter(task -> task.getTaskStatus().getState().isRunning())


We can keep == TaskState.RUNNING as FLUSHING tasks have no memory to revoke

sopel39 · 2020-07-06T14:37:46Z

presto-main/src/main/java/io/prestosql/execution/StageStateMachine.java

@@ -253,7 +259,7 @@ public BasicStageStats getBasicStageStats(Supplier<Iterable<TaskInfo>> taskInfos
        // information, the stage could finish, and the task states would
        // never be visible.
        StageState state = stageState.get();
-        boolean isScheduled = (state == RUNNING) || state.isDone();
+        boolean isScheduled = state.isRunning() || state.isDone();


Maybe we should have StageState#isScheduled method instead.

There is already a scheduled state for a stage - so having isScheduled could lead to ambiguity. Although, by that logic - the variable can be named differently (like begunExecution)

sopel39 · 2020-07-06T14:39:54Z

presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java

@@ -114,10 +114,10 @@ public void testEmptyQuery()
                createInitialEmptyOutputBuffers(PARTITIONED)
                        .withNoMoreBufferIds(),
                OptionalInt.empty());
-        assertEquals(taskInfo.getTaskStatus().getState(), TaskState.RUNNING);
+        assertTrue(taskInfo.getTaskStatus().getState().isRunning());


please test actual states (if possible) and don't use isRunning(). We need to have test where we validate FLUSHING state happens.

sopel39 · 2020-07-06T14:40:43Z

presto-main/src/test/java/io/prestosql/execution/TestSqlTaskExecution.java

@@ -177,7 +178,7 @@ public void testSimple(PipelineExecutionStrategy executionStrategy)

            //
            // test body
-            assertEquals(taskStateMachine.getState(), TaskState.RUNNING);
+            assertTrue(taskStateMachine.getState().isRunning());


please test actual states (if possible) and don't use isRunning(). We need to have test where we validate FLUSHING state happens

sopel39 · 2020-07-06T14:42:17Z

presto-main/src/test/java/io/prestosql/execution/TestSqlTaskExecution.java

@@ -279,6 +280,9 @@ public void testSimple(PipelineExecutionStrategy executionStrategy)

            outputBufferConsumer.abort(); // complete the task by calling abort on it
            TaskState taskState = taskStateMachine.getStateChange(TaskState.RUNNING).get(10, SECONDS);
+            if (taskState == TaskState.FLUSHING) {


We should incorporate check for FLUSHING state into test scenario

sopel39 · 2020-07-07T10:12:54Z

presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java

@@ -419,7 +420,7 @@ else if (partitioningHandle.equals(SCALED_WRITER_DISTRIBUTION)) {
        }
        Set<SqlStageExecution> childStages = childStagesBuilder.build();
        stage.addStateChangeListener(newState -> {
-            if (newState.isDone()) {
+            if (newState.isDone() || newState == FLUSHING) {


Please add a test for this. The test could be:

staring SELECT * FROM large_table LIMIT 1

NOT fetching results and waiting for SOURCE stage to be cancelled

rohangarg · 2020-07-08T15:18:08Z

@sopel39 : Thank you for the review! I have addressed all the comments w.r.t non-test code. Also, I have made the Task level tests more specific in terms of state asserting and incorporated exact testing for FLUSHING state. Further, I've made the FLUSHING transition more strict by only allowing it to happen from RUNNING state.
I'll also add more tests for Stage level flushing asap too.

dain

A couple of minor comments from me. I'll leave the final approval up to @sopel39

dain · 2020-07-13T22:17:15Z

presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java

@@ -524,6 +536,12 @@ else if (taskState == TaskState.FINISHED) {
        }
    }

+    private synchronized boolean isFlushing()
+    {
+        return flushingTasks.size() > 0


Add a comment explaining that to be flushing, there must be at least one flushing task, and all others must be flushing or finished.

also replace flushingTasks.size() > 0 with !flushingTasks.isEmpty()

dain · 2020-07-13T22:19:19Z

presto-main/src/main/java/io/prestosql/execution/TaskState.java

@@ -64,4 +70,9 @@ public boolean isDone()
    {
        return doneState;
    }
+
+    public boolean isRunningOrFlushing()


I'm not a fan of this method. In all uses, it seem to be used in an || with some other clause on the state. Instead, I would just inline this method.

rohangarg · 2020-07-15T17:20:42Z

@dain : Thanks for the review!

I have incorporated the current review comments and added a test for the case

Please add a test for this. The test could be:
SELECT * FROM large_table LIMIT 1
NOT fetching results and waiting for SOURCE stage to be cancelled

Will non query based UTs for SqlStageExecution be useful too? I tried to write it following the test present in TestSqlStageExecution in presto-main - but faced some difficulty with controlling state transitions for a stage. I can retry it if the tests seem to be needed.

sopel39

mostly lgtm % styling and testing comments.

good job!

presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java

presto-main/src/test/java/io/prestosql/execution/TestSqlTask.java

presto-main/src/main/java/io/prestosql/execution/scheduler/SqlQueryScheduler.java

sopel39 · 2020-07-28T11:19:47Z

presto-tests/src/test/java/io/prestosql/execution/TestSqlStageExecution.java

+    public void testFlushingState()
+            throws Exception
+    {
+        queryRunner = createQueryRunner(ImmutableMap.of("node-scheduler.include-coordinator", "false"));


Do we need distributed runner for this test?

No, we don't need a distributed runner for this test. Used it because TestQueryRunnerUtil provided a clean interface for waitForQueryState (without fetching results) - is any existing one with LocalQueryRunner too?

presto-tests/src/test/java/io/prestosql/execution/TestSqlStageExecution.java

sopel39 · 2020-07-28T11:25:17Z

tests are failing (I've seen some checkstyle issues)

presto-main/src/main/java/io/prestosql/execution/StageState.java

sopel39 · 2020-07-29T14:25:41Z

presto-main/src/main/java/io/prestosql/execution/SqlTaskExecution.java

@@ -641,6 +641,7 @@ private synchronized void checkTaskCompletion()

        // are there still pages in the output buffer
        if (!outputBuffer.isFinished()) {


@dain is this a valid transition for grouped execution or there should be additional conditions? I'm thinking that if there are more execution groups, the task should stay in RUNNING state.

sopel39 · 2020-07-29T15:31:21Z

presto-main/src/test/java/io/prestosql/execution/TestSqlTaskExecution.java

@@ -209,6 +212,7 @@ public void testSimple(PipelineExecutionStrategy executionStrategy)
                    // assert that task result is produced
                    outputBufferConsumer.consume(300 + 200, ASSERT_WAIT_TIMEOUT);
                    outputBufferConsumer.assertBufferComplete(ASSERT_WAIT_TIMEOUT);
+                    assertEquals(taskStateMachine.getStateChange(RUNNING).get(10, SECONDS), FLUSHING);


shouldn't that be moved above outputBufferConsumer.consume(300 + 200, ASSERT_WAIT_TIMEOUT); to prevent race condition?

Does outputBufferConsumer.assertBufferComplete(ASSERT_WAIT_TIMEOUT); consume all output?

Yes, assertBufferComplete makes sure that the buffer is complete. Added the assertions as discussed offline.

sopel39 · 2020-07-29T15:57:53Z

presto-tests/src/test/java/io/prestosql/execution/TestFlushingStageState.java

+        waitForQueryState(queryRunner, queryId, RUNNING);
+
+        QueryInfo queryInfo = queryRunner.getCoordinator().getFullQueryInfo(queryId);
+        while (queryInfo.getOutputStage().get().getState() != FLUSHING) {


please use assertEventually

fixed - also moved the creation of queryRunner to an unbounded setup. I think the test after setup should run within 30sec.

sopel39 · 2020-07-30T09:20:09Z

presto-main/src/test/java/io/prestosql/execution/TestSqlTaskManager.java

            assertEquals(results.getSerializedPages().size(), 1);
            assertEquals(results.getSerializedPages().get(0).getPositionCount(), 1);

            for (boolean moreResults = true; moreResults; moreResults = !results.isBufferComplete()) {
                results = sqlTaskManager.getTaskResults(taskId, OUT, results.getToken() + results.getSerializedPages().size(), DataSize.of(1, Unit.MEGABYTE)).get();
            }
-            assertEquals(results.isBufferComplete(), true);
+            assertTrue(results.isBufferComplete());


nit this should be separate commit

sopel39 · 2020-07-30T09:21:54Z

merged, thanks!

cla-bot bot added the cla-signed label Jul 1, 2020

rohangarg added the WIP label Jul 1, 2020

rohangarg requested review from sopel39 and dain July 1, 2020 06:46

rohangarg mentioned this pull request Jul 1, 2020

Relax task locking in SourcePartitionedScheduler #4294

Merged

sopel39 reviewed Jul 6, 2020

View reviewed changes

sopel39 reviewed Jul 7, 2020

View reviewed changes

rohangarg force-pushed the flushing_state branch from ff8cf5e to cfc220e Compare July 8, 2020 15:12

rohangarg removed the WIP label Jul 8, 2020

dain reviewed Jul 13, 2020

View reviewed changes

rohangarg force-pushed the flushing_state branch from cfc220e to ddc6f5e Compare July 15, 2020 17:09

rohangarg force-pushed the flushing_state branch from ddc6f5e to 581af4e Compare July 15, 2020 17:54

sopel39 reviewed Jul 28, 2020

View reviewed changes

presto-main/src/main/java/io/prestosql/execution/StageState.java Show resolved Hide resolved

rohangarg force-pushed the flushing_state branch from 581af4e to 875ed51 Compare July 29, 2020 13:09

sopel39 reviewed Jul 29, 2020

View reviewed changes

rohangarg added 2 commits July 29, 2020 22:01

Add FLUSHING state to tasks and stages

7fdbb52

Fix imports and cleanup in TestSqlTaskExecution

1304a49

rohangarg force-pushed the flushing_state branch from 875ed51 to 875693f Compare July 29, 2020 16:48

Cancel sub stages in FLUSHING state

d887993

rohangarg force-pushed the flushing_state branch from 875693f to d887993 Compare July 29, 2020 20:10

sopel39 approved these changes Jul 30, 2020

View reviewed changes

sopel39 merged commit 15ffd9c into trinodb:master Jul 30, 2020

sopel39 mentioned this pull request Jul 30, 2020

Release notes for 340 #4527

Closed

8 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Adding FLUSHING state to tasks and stages #4290

Adding FLUSHING state to tasks and stages #4290

rohangarg commented Jul 1, 2020

findepi commented Jul 1, 2020

rohangarg commented Jul 1, 2020

sopel39 left a comment

sopel39 Jul 6, 2020

sopel39 Jul 6, 2020

sopel39 Jul 6, 2020

sopel39 Jul 6, 2020

sopel39 Jul 6, 2020

sopel39 Jul 6, 2020

rohangarg Jul 8, 2020

sopel39 Jul 6, 2020 •

edited

Loading

sopel39 Jul 6, 2020 •

edited

Loading

sopel39 Jul 6, 2020

sopel39 Jul 7, 2020

rohangarg commented Jul 8, 2020 •

edited

Loading

dain left a comment

dain Jul 13, 2020

dain Jul 13, 2020

rohangarg commented Jul 15, 2020

sopel39 left a comment

sopel39 Jul 28, 2020

rohangarg Jul 28, 2020

sopel39 commented Jul 28, 2020

sopel39 Jul 29, 2020

sopel39 Jul 29, 2020

rohangarg Jul 29, 2020

sopel39 Jul 29, 2020

rohangarg Jul 29, 2020

sopel39 Jul 30, 2020

sopel39 commented Jul 30, 2020

		@@ -641,6 +641,7 @@ private synchronized void checkTaskCompletion()

		// are there still pages in the output buffer
		if (!outputBuffer.isFinished()) {

Adding FLUSHING state to tasks and stages #4290

Adding FLUSHING state to tasks and stages #4290

Conversation

rohangarg commented Jul 1, 2020

findepi commented Jul 1, 2020

rohangarg commented Jul 1, 2020

sopel39 left a 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

Choose a reason for hiding this comment

sopel39 Jul 6, 2020 • edited Loading

Choose a reason for hiding this comment

sopel39 Jul 6, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohangarg commented Jul 8, 2020 • edited Loading

dain left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rohangarg commented Jul 15, 2020

sopel39 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sopel39 commented Jul 28, 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

sopel39 commented Jul 30, 2020

sopel39 Jul 6, 2020 •

edited

Loading

sopel39 Jul 6, 2020 •

edited

Loading

rohangarg commented Jul 8, 2020 •

edited

Loading