!932 [I3FHGK] Improve snapshot stability
Merge pull request !932 from sandygao/improve-snapshot-stability
This commit is contained in:
Gitee
commit
df01ea5e1c
|
|
@ -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
|
||||
|
|
@ -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.
|
||||
|
||||
|
|
|
|||
|
|
@ -30,7 +30,6 @@
|
|||
- **与其他功能的交互**：分布式快照目前无法与以下功能一起使用：
|
||||
- 重用交换，即`optimizer.reuse-table-scan`
|
||||
- 重用公用表表达式（CTE），即`optimizer.cte-reuse-enabled`
|
||||
- 溢出，即`experimental.spill-enabled`
|
||||
|
||||
在启用分布式快照的情况下提交不满足上述要求的查询时，查询按未启用分布式快照功能的场景执行。
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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));
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Object> 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));
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue