Murtadha Hubail has uploaded a new change for review.
https://asterix-gerrit.ics.uci.edu/2970
Change subject: [NO ISSUE][STO] Ensure Files From Failed Bulkload Are Deleted
......................................................................
[NO ISSUE][STO] Ensure Files From Failed Bulkload Are Deleted
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- If a bulkload is interrupted at the end stage (i.e. after all
tuples have been added but before the component is added to
the index's disk components), then the bulkload operation will
fail but the files from the failed operation are not deleted.
This change ensures that if a bulkload fails in such stage,
the files of the generated component are deleted to avoid
file is already mapped exception if the bulkload is attempted
again.
- Ensure disk component is only destroyed once in case of
bulkload cleanup.
- Stop attempting to write a checkpoint early if the failure was
due to the thread being interrupted.
Change-Id: I7f63054ac72a9482e779e49eb4da658a08fe7e9d
---
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
3 files changed, 34 insertions(+), 15 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/70/2970/1
diff --git
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
index 3c62d99..8df50e6 100644
---
a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
+++
b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/IndexCheckpointManager.java
@@ -196,6 +196,9 @@
// ensure it was written correctly by reading it
read(checkpointPath);
return;
+ } catch (ClosedByInterruptException e) {
+ LOGGER.info("interrupted while writing checkponint at {}",
checkpointPath);
+ throw HyracksDataException.create(e);
} catch (IOException e) {
if (i == MAX_CHECKPOINT_WRITE_ATTEMPTS) {
throw HyracksDataException.create(e);
diff --git
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
index ab59b59..3fa45c9 100644
---
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
+++
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
@@ -101,8 +101,8 @@
for (int i = 0; i < bulkloadersCount; i++) {
bulkloaderChain.get(i).cleanupArtifacts();;
}
+ diskComponent.deactivateAndDestroy();
}
- diskComponent.deactivateAndDestroy();
}
@Override
diff --git
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
index 3a43ba7..c739ad0 100644
---
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
+++
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
@@ -22,6 +22,7 @@
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import
org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponentBulkLoader;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
import
org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
import org.apache.hyracks.storage.common.IIndexBulkLoader;
@@ -31,6 +32,7 @@
private final AbstractLSMIndex lsmIndex;
private final ILSMDiskComponentBulkLoader componentBulkLoader;
private final ILSMIndexOperationContext opCtx;
+ private boolean failed = false;
public LSMIndexDiskComponentBulkLoader(AbstractLSMIndex lsmIndex,
ILSMIndexOperationContext opCtx, float fillFactor,
boolean verifyInput, long numElementsHint) throws
HyracksDataException {
@@ -68,20 +70,10 @@
@Override
public void end() throws HyracksDataException {
try {
- try {
-
lsmIndex.getIOOperationCallback().afterOperation(opCtx.getIoOperation());
- componentBulkLoader.end();
- } catch (Throwable th) { // NOSONAR Must not call afterFinalize
without setting failure
- opCtx.getIoOperation().setStatus(LSMIOOperationStatus.FAILURE);
- opCtx.getIoOperation().setFailure(th);
- throw th;
- } finally {
-
lsmIndex.getIOOperationCallback().afterFinalize(opCtx.getIoOperation());
- }
- if (opCtx.getIoOperation().getStatus() ==
LSMIOOperationStatus.SUCCESS
- &&
opCtx.getIoOperation().getNewComponent().getComponentSize() > 0) {
-
lsmIndex.getHarness().addBulkLoadedComponent(opCtx.getIoOperation());
- }
+ presistComponentToDisk();
+ } catch (Throwable th) { // NOSONAR must cleanup in case of any failure
+ fail(th);
+ throw th;
} finally {
lsmIndex.getIOOperationCallback().completed(opCtx.getIoOperation());
}
@@ -116,4 +108,28 @@
return opCtx.getIoOperation().getFailure();
}
+ private void presistComponentToDisk() throws HyracksDataException {
+ try {
+
lsmIndex.getIOOperationCallback().afterOperation(opCtx.getIoOperation());
+ componentBulkLoader.end();
+ } catch (Throwable th) { // NOSONAR Must not call afterFinalize
without setting failure
+ fail(th);
+ throw th;
+ } finally {
+
lsmIndex.getIOOperationCallback().afterFinalize(opCtx.getIoOperation());
+ }
+ if (opCtx.getIoOperation().getStatus() == LSMIOOperationStatus.SUCCESS
+ && opCtx.getIoOperation().getNewComponent().getComponentSize()
> 0) {
+
lsmIndex.getHarness().addBulkLoadedComponent(opCtx.getIoOperation());
+ }
+ }
+
+ private void fail(Throwable th) {
+ if (!failed) {
+ failed = true;
+ final ILSMIOOperation loadOp = opCtx.getIoOperation();
+ loadOp.setFailure(th);
+ lsmIndex.cleanUpFilesForFailedOperation(loadOp);
+ }
+ }
}
\ No newline at end of file
--
To view, visit https://asterix-gerrit.ics.uci.edu/2970
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I7f63054ac72a9482e779e49eb4da658a08fe7e9d
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: stabilization-f69489
Gerrit-Owner: Murtadha Hubail <[email protected]>
