From 1c929ef6cc8e5b16dbbc4703e1e10bc7f2d2691c Mon Sep 17 00:00:00 2001 From: Sandy Gao Date: Mon, 14 Jun 2021 12:44:32 -0400 Subject: [PATCH] Various improvements - Recognize more errors that are resumable - Handle error when trying to save query snapshot result - Update snapshot documentation to remove "spill-to-disk" limitation - Remove "memory" catalog from default docker image --- docker/default/etc/catalog/memory.properties | 14 -------------- hetu-docs/en/admin/reliable-execution.md | 1 - hetu-docs/zh/admin/reliable-execution.md | 1 - .../prestosql/execution/SqlStageExecution.java | 17 ++++++++++++----- .../operator/HttpPageBufferClient.java | 2 +- .../snapshot/MultiInputSnapshotState.java | 1 - .../snapshot/QuerySnapshotManager.java | 12 ++++++++---- .../snapshot/TestTaskSnapshotManager.java | 5 +++++ 8 files changed, 26 insertions(+), 27 deletions(-) delete mode 100644 docker/default/etc/catalog/memory.properties diff --git a/docker/default/etc/catalog/memory.properties b/docker/default/etc/catalog/memory.properties deleted file mode 100644 index f6b1351ca..000000000 --- a/docker/default/etc/catalog/memory.properties +++ /dev/null @@ -1,14 +0,0 @@ -# Copyright (C) 2020. Huawei Technologies Co., Ltd. All rights reserved. -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -connector.name=memory diff --git a/hetu-docs/en/admin/reliable-execution.md b/hetu-docs/en/admin/reliable-execution.md index 9443477bd..9aca993ba 100644 --- a/hetu-docs/en/admin/reliable-execution.md +++ b/hetu-docs/en/admin/reliable-execution.md @@ -30,7 +30,6 @@ To be able to resume execution from a previously saved snapshot, there must be a - **Interaction with other features**: distributed snapshot does not yet work with the following features: - Reuse exchange, i.e. `optimizer.reuse-table-scan` - Reuse common table expression (CTE), i.e. `optimizer.cte-reuse-enabled` - - Spill, i.e. `experimental.spill-enabled` When a query that does not meet the above requirements is submitted with distributed snapshot enabled, the query will be executed as if the distributed snapshot feature is _not_ turned on. diff --git a/hetu-docs/zh/admin/reliable-execution.md b/hetu-docs/zh/admin/reliable-execution.md index c4d38bdb8..88124ba30 100644 --- a/hetu-docs/zh/admin/reliable-execution.md +++ b/hetu-docs/zh/admin/reliable-execution.md @@ -30,7 +30,6 @@ - **与其他功能的交互**：分布式快照目前无法与以下功能一起使用： - 重用交换，即`optimizer.reuse-table-scan` - 重用公用表表达式（CTE），即`optimizer.cte-reuse-enabled` - - 溢出，即`experimental.spill-enabled` 在启用分布式快照的情况下提交不满足上述要求的查询时，查询按未启用分布式快照功能的场景执行。 diff --git a/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java b/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java index b52b8cde7..23bb4cfff 100644 --- a/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java +++ b/presto-main/src/main/java/io/prestosql/execution/SqlStageExecution.java @@ -20,6 +20,7 @@ import com.google.common.collect.ImmutableMultimap; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Multimap; import com.google.common.collect.Sets; +import io.airlift.http.client.HttpStatus; import io.airlift.log.Logger; import io.airlift.units.Duration; import io.prestosql.Session; @@ -78,6 +79,7 @@ import static io.prestosql.SystemSessionProperties.isSnapshotEnabled; import static io.prestosql.failuredetector.FailureDetector.State.GONE; import static io.prestosql.operator.ExchangeOperator.REMOTE_CONNECTOR_ID; import static io.prestosql.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR; +import static io.prestosql.spi.StandardErrorCode.PAGE_TRANSPORT_ERROR; import static io.prestosql.spi.StandardErrorCode.REMOTE_HOST_GONE; import static java.util.Objects.requireNonNull; @@ -593,11 +595,16 @@ public final class SqlStageExecution .map(this::rewriteTransportFailure) .map(ExecutionFailureInfo::toException) .orElse(new PrestoException(GENERIC_INTERNAL_ERROR, "A task failed for an unknown reason")); - // Snapshot: if remote task failed because they received 5xx from other tasks, then we treat it as resumable. - if (failure.getMessage() != null && failure.getMessage().contains("to be 200, but was 5")) { - log.debug(failure, "Task %s on node %s failed but is resumable. Triggering rescheduling.", taskStatus.getTaskId(), taskStatus.getNodeId()); - stateMachine.transitionToResumableFailure(); - return; + // Snapshot: if remote task failed because they received unexpecte response (5xx or missing/wrong header), then we treat it as resumable. + if (isSnapshotEnabled && failure instanceof Failure && ((Failure) failure).getErrorCode().equals(PAGE_TRANSPORT_ERROR.toErrorCode())) { + String message = failure.getMessage(); + // message ends with "(response status code %d)" + int responseCode = Integer.parseInt(message.substring(message.lastIndexOf(' ') + 1, message.lastIndexOf(')'))); + if (responseCode >= 500 || responseCode == HttpStatus.OK.code()) { + log.debug(failure, "Task %s on node %s failed but is resumable. Triggering rescheduling.", taskStatus.getTaskId(), taskStatus.getNodeId()); + stateMachine.transitionToResumableFailure(); + return; + } } stateMachine.transitionToFailed(failure); } diff --git a/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java b/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java index f025ee202..eae33edef 100644 --- a/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java +++ b/presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java @@ -638,7 +638,7 @@ public final class HttpPageBufferClient return createEmptyPagesResponse(0, 0, false); } } - throw new PageTransportErrorException(format("Error fetching %s: %s", request.getUri().toASCIIString(), e.getMessage()), e); + throw new PageTransportErrorException(format("Error fetching %s: %s (response status code %d)", request.getUri().toASCIIString(), e.getMessage(), response.getStatusCode()), e); } } diff --git a/presto-main/src/main/java/io/prestosql/snapshot/MultiInputSnapshotState.java b/presto-main/src/main/java/io/prestosql/snapshot/MultiInputSnapshotState.java index ce213d51c..81f3fefba 100644 --- a/presto-main/src/main/java/io/prestosql/snapshot/MultiInputSnapshotState.java +++ b/presto-main/src/main/java/io/prestosql/snapshot/MultiInputSnapshotState.java @@ -383,7 +383,6 @@ public class MultiInputSnapshotState // First time a marker is received for a snapshot. Store operator state and make marker available. snapshot = new SnapshotState(marker); pendingMarkers.add(marker); - // No need to take snapshot if not all input channels are known. The snapshot won't be complete. try { snapshot.states.add(restorable.capture(pagesSerde)); } diff --git a/presto-main/src/main/java/io/prestosql/snapshot/QuerySnapshotManager.java b/presto-main/src/main/java/io/prestosql/snapshot/QuerySnapshotManager.java index b2260a421..fad997f30 100644 --- a/presto-main/src/main/java/io/prestosql/snapshot/QuerySnapshotManager.java +++ b/presto-main/src/main/java/io/prestosql/snapshot/QuerySnapshotManager.java @@ -193,7 +193,14 @@ public class QuerySnapshotManager } } - saveQuerySnapshotResult(); + try { + saveQuerySnapshotResult(); + } + catch (Exception e) { + LOG.warn(e, "Failed to save query snapshot state for %s: %s", queryId, e.getMessage()); + invalidateAllSnapshots(); + result = OptionalLong.empty(); + } } if (result.isPresent()) { @@ -211,7 +218,6 @@ public class QuerySnapshotManager for (Long snapshotId : captureResults.keySet()) { captureResults.put(snapshotId, SnapshotResult.NA); } - saveQuerySnapshotResult(); } } @@ -415,8 +421,6 @@ public class QuerySnapshotManager SnapshotResult oldResult = captureResults.put(snapshotId, snapshotResult); if (snapshotResult != oldResult && snapshotResult.isDone()) { LOG.debug("Finished capturing snapshot %d for query %s. Result is %s.", snapshotId, queryId.getId(), snapshotResult); - // Store snapshot information for this query in filesystem, so it can be accessed by tasks, e.g. during backtrack state loading. - saveQuerySnapshotResult(); } } } diff --git a/presto-main/src/test/java/io/prestosql/snapshot/TestTaskSnapshotManager.java b/presto-main/src/test/java/io/prestosql/snapshot/TestTaskSnapshotManager.java index a3dbd8240..bac093126 100644 --- a/presto-main/src/test/java/io/prestosql/snapshot/TestTaskSnapshotManager.java +++ b/presto-main/src/test/java/io/prestosql/snapshot/TestTaskSnapshotManager.java @@ -121,6 +121,7 @@ public class TestTaskSnapshotManager QuerySnapshotManager querySnapshotManager = new QuerySnapshotManager(queryId, snapshotUtils, TEST_SNAPSHOT_SESSION); querySnapshotManager.addNewTask(taskId); querySnapshotManager.updateQueryCapture(taskId, Collections.singletonMap(1L, SnapshotResult.SUCCESSFUL)); + querySnapshotManager.getResumeSnapshotId(); SnapshotStateId newStateId = stateId.withSnapshotId(2); Optional loadedState = snapshotManager.loadState(newStateId); @@ -187,13 +188,17 @@ public class TestTaskSnapshotManager // Try2: Previous snapshots are setup, so load should be successful querySnapshotManager.updateQueryCapture(taskId, Collections.singletonMap(2L, SnapshotResult.SUCCESSFUL)); + querySnapshotManager.getResumeSnapshotId(); assertTrue(snapshotManager.loadFile(id4load, targetPath)); String output = Files.readAllLines(targetPath).get(0); Assert.assertEquals(output, fileContent); // Try3: Previous snapshot failed + querySnapshotManager.updateQueryRestore(taskId, Optional.of(new RestoreResult(2, SnapshotResult.SUCCESSFUL))); querySnapshotManager.updateQueryCapture(taskId, Collections.singletonMap(2L, SnapshotResult.FAILED)); + querySnapshotManager.updateQueryCapture(taskId, Collections.singletonMap(3L, SnapshotResult.SUCCESSFUL)); + querySnapshotManager.getResumeSnapshotId(); assertFalse(snapshotManager.loadFile(id4load, targetPath)); }