[WIP] ajay opec #21135

ajaygeorge · 2023-10-12T23:32:50Z

Prepare for next development iteration - 0.284-edge19-SNAPSHOT
Prepare for next development iteration - 0.284-edge19.1-SNAPSHOT
Prepare for next development iteration - 0.284-edge19.2-SNAPSHOT
Add split retry feature
Worker Isolation
Add Thrift serializer for LongSet
Fix worker isolation
Fix multiple issues with retry
Fix query runner
Fix stuck query on intermediate workers
Recovering opec worker
Schedule queries with LocalExchange to intermediate nodes
reduce no_more_retry_delay to 5 sec
setNoMorePages for shutting down task's outputbuffer
Handle FixedSourcePartitionedScheduler for splitting leaf and intermediate
Logging, metrics and handle race conditions
Handle race condition Add byte[] backed tuple and block implementations #2
Change the test to make the pageCount randomly generated
Add complex case for graceful shutdown
Fix the corner case where the outputBuffer has not received noMoreBuffer from the coordinator
Fix compilation errors
Use the latest retried logic
Add a new GRACEFUL_FAILED stsate for task
Attempt to fix TOO_MANY_REQUEST_FAILED in the verifier results
Using callback to decide whether to unblock the SqlStageExecution
Revert the cacnellation conditions
Add SplitRetrySourcePartitionedScheduler to support noMoreRetry
Commented out line for checking coordinator count
Add debug log
Fix noMoreSplits error on the intermediate stage

Description

Motivation and Context

Impact

Test Plan

Contributor checklist

Please make sure your submission complies with our development, formatting, commit message, and attribution guidelines.
PR description addresses the issue accurately and concisely. If the change is non-trivial, a GitHub Issue is referenced.
Documented new properties (with its default value), SQL syntax, functions, or other functionality.
If release notes are required, they follow the release notes guidelines.
Adequate tests were added if applicable.
CI passed.

Release Notes

Please follow release notes guidelines and fill in the release notes below.

== RELEASE NOTES ==

General Changes
* ...
* ...

Hive Changes
* ...
* ...

If release note is NOT required, use:

== NO RELEASE NOTE ==

…diate

Add metric for retry and unit test for graceful shutdown

…fer from the coordinator

ajaygeorge · 2023-10-12T23:59:11Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskShutdownManager.java

+    public void handleShutdown(TaskId taskId)
+    {
+        String errorMessage = String.format("killing pending task %s due to host being shutting down", taskId);
+        taskStateMachine.graceful_failed(new HostShuttingDownException(errorMessage, System.nanoTime()));


System.nanoTime is to measure elapsed time and not absolute time.

nit : graceful_failed method name

ajaygeorge · 2023-10-13T00:11:44Z

presto-main/src/test/java/com/facebook/presto/execution/MockRemoteTaskFactory.java

            taskStateMachine.cancel();
        }

+        public void graceful_failed()


no one calls this

ajaygeorge · 2023-10-13T00:14:19Z

presto-main/src/main/java/com/facebook/presto/execution/RemoteTask.java

    void start();

-    void addSplits(Multimap<PlanNodeId, Split> splits);
+    boolean addSplits(Multimap<PlanNodeId, Split> splits);


nit : add javadoc to say what boolean return value means

ajaygeorge · 2023-10-13T00:17:13Z

presto-main/src/main/java/com/facebook/presto/server/remotetask/HttpRemoteTask.java

        return pendingSourceSplitCount;
    }

+    public synchronized void setIsRetried()


why synchronized. ?

ajaygeorge · 2023-10-13T00:21:55Z

presto-main/src/main/java/com/facebook/presto/execution/SqlStageExecution.java

+
+        if (failedTasks.isEmpty()) {
+            checkState(finishedTasks.isEmpty());
+            List<RemoteTask> idleRunningHttpRemoteTasks = getAllTasks().stream()


AGPQN idleRunningHttpRemoteTasks sounds like an oxymoron

ajaygeorge · 2023-10-13T00:22:12Z

presto-main/src/main/java/com/facebook/presto/execution/SqlStageExecution.java

+                    .filter(task -> task.getTaskStatus().getState() == TaskState.RUNNING)
+                    .filter(task -> task.isTaskIdling())


combine both filters into one

ajaygeorge · 2023-10-13T19:15:16Z

presto-main/src/main/java/com/facebook/presto/Session.java

    private final AccessControlContext context;
    private final Optional<Tracer> tracer;
    private final WarningCollector warningCollector;
+    private final ConcurrentHashMap<PlanFragmentId, Pair<NodePoolType, String>> fragmentToPoolTypeMapping = new ConcurrentHashMap<>();


nit. code to interfaces. ConcurrentMap

ajaygeorge · 2023-10-13T19:20:41Z

presto-main/src/main/java/com/facebook/presto/execution/scheduler/SectionExecutionFactory.java

+        ConcurrentHashMap<PlanFragmentId, Session.Pair<NodePoolType, String>> fragmentToPoolTypeMapping = session.getFragmentToPoolTypeMapping();
+        Session.Pair<NodePoolType, String> nodePoolTypeStringPair = fragmentToPoolTypeMapping.get(plan.getFragment().getId());
+        if (nodePoolTypeStringPair != null && nodePoolTypeStringPair.getKey() != workerPoolType) {
+            log.error("Error in pool type evaluation, plan =%s,hasLocalExchangeAtRoot = %s, isLeaf=%s, plan in map =%s", plan.getFragment().getJsonRepresentation().orElse(null), hasLocalExchangeAtRoot, plan.getFragment().isLeaf(), nodePoolTypeStringPair.getValue());
+            throw new RuntimeException("Error in pool type evaluation");
+        }
+        else {
+            fragmentToPoolTypeMapping.put(plan.getFragment().getId(), new Session.Pair<>(workerPoolType, plan.getFragment().getJsonRepresentation().orElse("")));


AGPQN when do we enter into the erorr situation.
Also why do we need the entire plan fragment json stuffed into the session

ajaygeorge · 2023-10-14T01:18:16Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

+                if (taskSplit.getScheduledSplit() != null) {
+                    log.warn("Adding split %s to pending split tracker for task %s", taskSplit.getScheduledSplit().getSequenceId(), taskHandle.getTaskId());
+                    gracefulShutdownSplitTracker.getPendingSplits().computeIfAbsent(taskHandle.getTaskId(), k -> ConcurrentHashMap.newKeySet()).add(taskSplit.getScheduledSplit().getSequenceId());
+                }


why is gracefulShutdownSplitTracker needed for non shutdown cases?

ajaygeorge · 2023-10-16T17:19:37Z

presto-spi/src/main/java/com/facebook/presto/spi/StandardErrorCode.java

    QUERY_PLANNING_TIMEOUT(0x0001_0027, INTERNAL_ERROR),
    NATIVE_EXECUTION_TASK_ERROR(0x0001_0028, INTERNAL_ERROR),
    UNSUPPORTED_ANALYZER_TYPE(0x0001_0029, INTERNAL_ERROR),
+    REMOTE_HOST_GONE_INTERMEDIATE(0x0001_0029, INTERNAL_ERROR, true),


not used anywhere

ajaygeorge · 2023-10-16T17:25:20Z

presto-main/src/main/java/com/facebook/presto/execution/SqlTaskManager.java

        return tasks.getUnchecked(taskId).abort();
    }

+    public void failTask(TaskId taskId)


only called for testing

ajaygeorge · 2023-10-16T17:36:19Z

presto-main/src/main/java/com/facebook/presto/execution/buffer/PartitionedOutputBuffer.java

+    {
+        for (ClientBuffer partition : partitions) {
+            if (!partition.isEmptyPages()) {
+                return false;
+            }
+        }
+        return true;


nit this logic is duplicated in multiple output buffer implementations.

ajaygeorge · 2023-10-16T21:11:44Z

presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java

                .withFilter(GenerateTraceTokenRequestFilter.class)
                .withConfigDefaults(config -> {
-                    config.setRequestTimeout(new Duration(10, SECONDS));
+                    config.setRequestTimeout(new Duration(30, SECONDS));


random change?

ajaygeorge · 2023-10-16T23:43:44Z

presto-main/src/main/java/com/facebook/presto/execution/SqlQueryExecution.java

            AtomicReference<SqlQuerySchedulerInterface> queryScheduler = this.queryScheduler;
            stateMachine.addStateChangeListener(state -> {
-                if (!state.isDone()) {
+                if (!state.isDone() || state == QueryState.FINISHED) {


AGPQN not Done or finished seems interesting.

ajaygeorge · 2023-10-18T21:54:17Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskHandle.java

+    public void gracefulShutdown()
+    {
+        isShuttingDown.set(true);
+    }


nit calling it graceful shutdown implies there is some action to be done.
maybe just call it as markTaskAsShutdown

ajaygeorge · 2023-10-18T22:02:59Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

+    private final TimeStat outputBufferEmptyWaitTime = new TimeStat(NANOSECONDS);
+    private final TimeStat waitForRunningSplitTime = new TimeStat(NANOSECONDS);
    private volatile boolean closed;
+    private final ExecutorService taskShutdownExecutor = newCachedThreadPool(daemonThreadsNamed("task-shutdown-%s"));


instead of a cached thread pool which starts from no threads , it might be better to go with a thread pool of min size = p90 number of tasks on a dying node since thread creation takes time.

ajaygeorge · 2023-10-18T22:59:58Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

+                                    builderWithOutputBufferInfo("init", shuttingdownNode, outputBuffer)
+                                            .build());
+
+                            while (taskHandle.getRunningLeafSplits() > 0 || taskHandle.getRunningIntermediateSplits() > 0) {


AGPQN : why are we checking for taskHandle.getRunningIntermediateSplits()

ajaygeorge · 2023-10-26T17:00:37Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

+    public static final String OB_WAIT_OVER = "obWait-over";
+    public static final String OB_WAIT = "obWait";
+    public static final String SPLIT_WAIT = "splitWait";
+    public static final String SPLIT_WAIT_OVER = "splitWait-over";


if we are planning to keep it in prod, might be good to add ordinals prefixed so that it is easier to understand the sequence of events

ajaygeorge · 2023-10-26T17:31:59Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

+                                catch (Exception ex) {
+                                    log.error(ex, "GracefulShutdown got interrupted while waiting for split completion for task %s", taskId);
+                                }


If the intend is to catch Interrupted Exception, we need to specifically catch that and set the interrupt flag back.

ajaygeorge · 2023-10-31T15:57:02Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

+    {
+        isGracefulShutdownStarted.set(true);
+        long shutdownStartTime = System.nanoTime();
+        //TODO throw error for new task creation instead of looping here.


AGPQN : Are we creating tasks here?

ajaygeorge · 2023-10-31T16:13:11Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskExecutor.java

            Duration splitConcurrencyAdjustFrequency,
            OptionalInt maxDriversPerTask)
+    {
+        return addTask(taskId, utilizationSupplier, initialSplitConcurrency, splitConcurrencyAdjustFrequency, maxDriversPerTask, Optional.empty(), Optional.empty());


nit: get rid of this wrapper since the above method is only called in 1 place

ajaygeorge · 2023-10-31T16:45:19Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskHandle.java

+        if (!hostShutDownListener.isPresent()) {
+            return;
+        }
+        hostShutDownListener.get().handleShutdown(taskId);


nit : rewrite in functional style as

hostShutDownListener.ifPresent(taskShutDownListener -> taskShutDownListener.handleShutdown(taskId));

ajaygeorge · 2023-10-31T16:45:58Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskHandle.java

+        if (!hostShutDownListener.isPresent()) {
+            return;
+        }
+        hostShutDownListener.get().addStats(taskShutdownStats);


same as above

hostShutDownListener.ifPresent(taskShutDownListener -> taskShutDownListener.addStats(taskShutdownStats));

ajaygeorge · 2023-10-31T16:57:34Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskHandle.java

-        builder.addAll(runningLeafSplits);
-        builder.addAll(queuedLeafSplits);
+        //To avoid queued split marked as completed splits to pollute the retryable splits
+        if (isShuttingDown.get()) {


AGPQN : This check is a bit confusing. Can we call this method destroy say if we want to do speculative execution and cancel/destroy the work which is independent of the notion of a shutdown.

ajaygeorge · 2023-10-31T17:14:50Z

presto-main/src/main/java/com/facebook/presto/execution/executor/TaskHandle.java

+            queuedLeafSplits.add(split);
+        }
+        else {
+            checkState(!split.isSplitAlreadyStarted(), "Split we are avoiding to queue was already started!");


nit avoiding -> adding

presto-release-bot and others added 30 commits September 15, 2023 17:10

Prepare for next development iteration - 0.284-edge19-SNAPSHOT

3d446bc

Prepare for next development iteration - 0.284-edge19.1-SNAPSHOT

8e5c638

Prepare for next development iteration - 0.284-edge19.2-SNAPSHOT

c99a5e3

Add split retry feature

95ea2a7

Worker Isolation

e38076c

Add Thrift serializer for LongSet

a8ba20d

Fix worker isolation

04d56d8

Fix multiple issues with retry

5b6131c

Fix query runner

dbe80a7

Fix stuck query on intermediate workers

3b946d6

Recovering opec worker

ec3d939

Schedule queries with LocalExchange to intermediate nodes

a48684e

reduce no_more_retry_delay to 5 sec

77c2e9f

setNoMorePages for shutting down task's outputbuffer

d2bebee

Handle FixedSourcePartitionedScheduler for splitting leaf and interme…

c2d623c

…diate

Logging, metrics and handle race conditions

8efc854

Add metric for retry and unit test for graceful shutdown

Handle race condition #2

ba66981

Change the test to make the pageCount randomly generated

c8d14ae

Add complex case for graceful shutdown

445c0a0

Fix the corner case where the outputBuffer has not received noMoreBuf…

0f16ad0

…fer from the coordinator

Fix compilation errors

270d8e7

Use the latest retried logic

6a85dd8

Add a new GRACEFUL_FAILED stsate for task

bc3bdd4

Attempt to fix TOO_MANY_REQUEST_FAILED in the verifier results

49d90ee

Using callback to decide whether to unblock the SqlStageExecution

79fa396

Revert the cacnellation conditions

0fe719d

Add SplitRetrySourcePartitionedScheduler to support noMoreRetry

5bcd2bb

Commented out line for checking coordinator count

b44813b

Add debug log

11f19f5

Fix noMoreSplits error on the intermediate stage

9dfd827

ajaygeorge changed the title ~~ajay opec~~ [WIP] ajay opec Oct 12, 2023

ajaygeorge commented Oct 12, 2023

View reviewed changes

ajaygeorge commented Oct 13, 2023

View reviewed changes

ajaygeorge commented Oct 14, 2023

View reviewed changes

ajaygeorge commented Oct 16, 2023

View reviewed changes

ajaygeorge commented Oct 18, 2023

View reviewed changes

ajaygeorge commented Oct 26, 2023

View reviewed changes

ajaygeorge commented Oct 31, 2023

View reviewed changes

		.filter(task -> task.getTaskStatus().getState() == TaskState.RUNNING)
		.filter(task -> task.isTaskIdling())

[WIP] ajay opec #21135

Are you sure you want to change the base?

[WIP] ajay opec #21135

Uh oh!

Conversation

ajaygeorge commented Oct 12, 2023

Description

Motivation and Context

Impact

Test Plan

Contributor checklist

Release Notes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ajaygeorge Oct 13, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ajaygeorge Oct 14, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ajaygeorge Oct 18, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ajaygeorge Oct 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ajaygeorge Oct 31, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

ajaygeorge Oct 13, 2023 •

edited

Loading

ajaygeorge Oct 14, 2023 •

edited

Loading

ajaygeorge Oct 18, 2023 •

edited

Loading

ajaygeorge Oct 31, 2023 •

edited

Loading

ajaygeorge Oct 31, 2023 •

edited

Loading