You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mh...@apache.org on 2018/04/30 20:50:37 UTC
[2/8] asterixdb git commit: [NO ISSUE][STO] Add consistency to flush
lifecycle
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
index 7b12250..f3162c9 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTree.java
@@ -40,8 +40,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponentFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
@@ -50,7 +48,6 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import org.apache.hyracks.storage.am.lsm.common.api.ITwoPCIndex;
-import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.common.impls.ExternalIndexHarness;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
import org.apache.hyracks.storage.am.rtree.impls.SearchPredicate;
@@ -323,12 +320,6 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
if (!isActive) {
throw new HyracksDataException("Failed to deactivate the index since it is already deactivated.");
}
- if (flushOnExit) {
- AbstractLSMIndexOperationContext opCtx = createOpContext(NoOpOperationCallback.INSTANCE, version);
- opCtx.setIoOperationType(LSMIOOperationType.FLUSH);
- ioOpCallback.afterFinalize(opCtx);
- }
-
for (ILSMDiskComponent c : diskComponents) {
c.deactivateAndPurge();
}
@@ -383,16 +374,13 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
version = 0;
}
- // Not supported
@Override
public void modify(IIndexOperationContext ictx, ITupleReference tuple) throws HyracksDataException {
throw new UnsupportedOperationException("tuple modify not supported in LSM-Disk-Only-RTree");
}
- // Not supported
@Override
- public void scheduleFlush(ILSMIndexOperationContext ctx, ILSMIOOperationCallback callback)
- throws HyracksDataException {
+ public ILSMIOOperation createFlushOperation(ILSMIndexOperationContext ctx) throws HyracksDataException {
throw new UnsupportedOperationException("flush not supported in LSM-Disk-Only-RTree");
}
@@ -576,8 +564,7 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
// The only change the the schedule merge is the method used to create the
// opCtx. first line <- in schedule merge, we->
@Override
- public void scheduleMerge(ILSMIndexOperationContext ctx, ILSMIOOperationCallback callback)
- throws HyracksDataException {
+ public ILSMIOOperation createMergeOperation(ILSMIndexOperationContext ctx) throws HyracksDataException {
ILSMIndexOperationContext rctx = createOpContext(NoOpOperationCallback.INSTANCE, -1);
rctx.setOperation(IndexOperation.MERGE);
List<ILSMComponent> mergingComponents = ctx.getComponentHolder();
@@ -587,10 +574,12 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
(ILSMDiskComponent) mergingComponents.get(mergingComponents.size() - 1));
ILSMIndexAccessor accessor = new LSMRTreeAccessor(getHarness(), rctx, buddyBTreeFields);
// create the merge operation.
- LSMRTreeMergeOperation mergeOp = new LSMRTreeMergeOperation(accessor, cursor,
- relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getDeleteIndexFileReference(),
- relMergeFileRefs.getBloomFilterFileReference(), callback, fileManager.getBaseDir().getAbsolutePath());
- ioScheduler.scheduleOperation(mergeOp);
+ LSMRTreeMergeOperation mergeOp =
+ new LSMRTreeMergeOperation(accessor, cursor, relMergeFileRefs.getInsertIndexFileReference(),
+ relMergeFileRefs.getDeleteIndexFileReference(), relMergeFileRefs.getBloomFilterFileReference(),
+ ioOpCallback, fileManager.getBaseDir().getAbsolutePath());
+ ioOpCallback.scheduled(mergeOp);
+ return mergeOp;
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
index fae6e1a..4566d32 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTree.java
@@ -45,7 +45,6 @@ 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.ILSMDiskComponentFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
@@ -135,8 +134,8 @@ public class LSMRTree extends AbstractLSMRTree {
rTreeTupleSorter.sort();
component = createDiskComponent(componentFactory, flushOp.getTarget(), flushOp.getBTreeTarget(),
flushOp.getBloomFilterTarget(), true);
- componentBulkLoader = component.createBulkLoader(LSMIOOperationType.FLUSH, 1.0f, false,
- numBTreeTuples.longValue(), false, false, false);
+ componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, numBTreeTuples.longValue(),
+ false, false, false);
flushLoadRTree(isEmpty, rTreeTupleSorter, componentBulkLoader);
// scan the memory BTree and bulk load delete tuples
flushLoadBtree(memBTreeAccessor, componentBulkLoader, btreeNullPredicate);
@@ -332,13 +331,13 @@ public class LSMRTree extends AbstractLSMRTree {
numElements += ((LSMRTreeDiskComponent) mergeOp.getMergingComponents().get(i)).getBloomFilter()
.getNumElements();
}
- componentBulkLoader = mergedComponent.createBulkLoader(LSMIOOperationType.MERGE, 1.0f, false,
- numElements, false, false, false);
+ componentBulkLoader =
+ mergedComponent.createBulkLoader(mergeOp, 1.0f, false, numElements, false, false, false);
mergeLoadBTree(opCtx, rtreeSearchPred, componentBulkLoader);
} else {
//no buddy-btree needed
- componentBulkLoader = mergedComponent.createBulkLoader(LSMIOOperationType.MERGE, 1.0f, false, 0L,
- false, false, false);
+ componentBulkLoader =
+ mergedComponent.createBulkLoader(mergeOp, 1.0f, false, 0L, false, false, false);
}
//search old rtree components
while (cursor.hasNext()) {
@@ -425,7 +424,7 @@ public class LSMRTree extends AbstractLSMRTree {
LSMRTreeAccessor accessor = new LSMRTreeAccessor(getHarness(), opCtx, buddyBTreeFields);
return new LSMRTreeFlushOperation(accessor, componentFileRefs.getInsertIndexFileReference(),
componentFileRefs.getDeleteIndexFileReference(), componentFileRefs.getBloomFilterFileReference(),
- callback, fileManager.getBaseDir().getAbsolutePath());
+ callback, getIndexIdentifier());
}
@Override
@@ -435,6 +434,6 @@ public class LSMRTree extends AbstractLSMRTree {
ILSMIndexAccessor accessor = new LSMRTreeAccessor(getHarness(), opCtx, buddyBTreeFields);
return new LSMRTreeMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(),
mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(), callback,
- fileManager.getBaseDir().getAbsolutePath());
+ getIndexIdentifier());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMemoryComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMemoryComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMemoryComponent.java
index 2e1ba68..ef7b815 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMemoryComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeMemoryComponent.java
@@ -21,9 +21,9 @@ package org.apache.hyracks.storage.am.lsm.rtree.impls;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.btree.impls.BTree;
-import org.apache.hyracks.storage.am.lsm.common.api.AbstractLSMWithBuddyMemoryComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMWithBuddyMemoryComponent;
import org.apache.hyracks.storage.am.rtree.impls.RTree;
public class LSMRTreeMemoryComponent extends AbstractLSMWithBuddyMemoryComponent {
@@ -32,8 +32,8 @@ public class LSMRTreeMemoryComponent extends AbstractLSMWithBuddyMemoryComponent
private final BTree btree;
public LSMRTreeMemoryComponent(AbstractLSMRTree lsmIndex, RTree rtree, BTree btree, IVirtualBufferCache vbc,
- boolean isActive, ILSMComponentFilter filter) {
- super(lsmIndex, vbc, isActive, filter);
+ ILSMComponentFilter filter) {
+ super(lsmIndex, vbc, filter);
this.rtree = rtree;
this.btree = btree;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
index f4e919a..2757595 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeWithAntiMatterTuples.java
@@ -40,7 +40,6 @@ 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.ILSMDiskComponentFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
@@ -110,8 +109,7 @@ public class LSMRTreeWithAntiMatterTuples extends AbstractLSMRTree {
try {
memRTreeAccessor.search(rtreeScanCursor, rtreeNullPredicate);
component = createDiskComponent(componentFactory, flushOp.getTarget(), null, null, true);
- componentBulkLoader =
- component.createBulkLoader(LSMIOOperationType.FLUSH, 1.0f, false, 0L, false, false, false);
+ componentBulkLoader = component.createBulkLoader(operation, 1.0f, false, 0L, false, false, false);
// Since the LSM-RTree is used as a secondary assumption, the
// primary key will be the last comparator in the BTree comparators
rTreeTupleSorter = new TreeTupleSorter(flushingComponent.getIndex().getFileId(), linearizerArray,
@@ -237,7 +235,7 @@ public class LSMRTreeWithAntiMatterTuples extends AbstractLSMRTree {
ILSMDiskComponent component = createDiskComponent(componentFactory, mergeOp.getTarget(), null, null, true);
ILSMDiskComponentBulkLoader componentBulkLoader =
- component.createBulkLoader(LSMIOOperationType.MERGE, 1.0f, false, 0L, false, false, false);
+ component.createBulkLoader(operation, 1.0f, false, 0L, false, false, false);
try {
while (cursor.hasNext()) {
cursor.next();
@@ -275,7 +273,7 @@ public class LSMRTreeWithAntiMatterTuples extends AbstractLSMRTree {
throws HyracksDataException {
ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
return new LSMRTreeFlushOperation(accessor, componentFileRefs.getInsertIndexFileReference(), null, null,
- callback, fileManager.getBaseDir().getAbsolutePath());
+ callback, getIndexIdentifier());
}
@Override
@@ -290,6 +288,6 @@ public class LSMRTreeWithAntiMatterTuples extends AbstractLSMRTree {
new LSMRTreeWithAntiMatterTuplesSearchCursor(opCtx, returnDeletedTuples);
ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
return new LSMRTreeMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(), null, null,
- callback, fileManager.getBaseDir().getAbsolutePath());
+ callback, getIndexIdentifier());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/IIndexBulkLoader.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/IIndexBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/IIndexBulkLoader.java
index fc54903..d6c954e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/IIndexBulkLoader.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/IIndexBulkLoader.java
@@ -35,7 +35,7 @@ public interface IIndexBulkLoader {
public void add(ITupleReference tuple) throws HyracksDataException;
/**
- * Finalize the bulk loading operation in the given context.
+ * Finalize the bulk loading operation in the given context and release all resources.
*
* @throws HyracksDataException
* If the BufferCache throws while un/pinning or un/latching.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
index 1443bbc..9f82f02 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/BufferCache.java
@@ -1451,4 +1451,23 @@ public class BufferCache implements IBufferCacheInternal, ILifeCycleComponent {
return multiplier;
}
}
+
+ @Override
+ public void closeFileIfOpen(FileReference fileRef) {
+ synchronized (fileInfoMap) {
+ if (fileMapManager.isMapped(fileRef)) {
+ int fileId;
+ try {
+ fileId = fileMapManager.lookupFileId(fileRef);
+ } catch (HyracksDataException e) {
+ throw new IllegalStateException(e);
+ }
+ BufferedFileHandle fInfo = fileInfoMap.get(fileId);
+ if (fInfo != null && fInfo.getReferenceCount() > 0) {
+ fInfo.decReferenceCount();
+ }
+ }
+ }
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
index 1cbe404..9e92f21 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/DebugBufferCache.java
@@ -246,4 +246,10 @@ public class DebugBufferCache implements IBufferCache {
deleteFileCount.incrementAndGet();
bufferCache.deleteFile(file);
}
+
+ @Override
+ public void closeFileIfOpen(FileReference fileRef) {
+ bufferCache.closeFileIfOpen(fileRef);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
index 8dccc4a..21d3677 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-common/src/main/java/org/apache/hyracks/storage/common/buffercache/IBufferCache.java
@@ -269,4 +269,13 @@ public interface IBufferCache {
*/
void resizePage(ICachedPage page, int multiplier, IExtraPageBlockHelper extraPageBlockHelper)
throws HyracksDataException;
+
+ /**
+ * Close the file if open.
+ *
+ * @param fileRef
+ * @throws HyracksDataException
+ */
+ void closeFileIfOpen(FileReference fileRef);
+
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallback.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallback.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallback.java
new file mode 100644
index 0000000..cb791c5
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallback.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.hyracks.storage.am.lsm.btree;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.btree.impl.ITestOpCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
+
+public class EncapsulatingIoCallback implements ILSMIOOperationCallback {
+
+ private final ILSMIOOperationCallback encapsulated;
+ private final ITestOpCallback<ILSMIOOperation> scheduledCallback;
+ private final ITestOpCallback<ILSMIOOperation> beforeOperationCallback;
+ private final ITestOpCallback<ILSMIOOperation> afterOperationCallback;
+ private final ITestOpCallback<ILSMIOOperation> afterFinalizeCallback;
+ private final ITestOpCallback<ILSMIOOperation> completedCallback;
+
+ public EncapsulatingIoCallback(ILSMIOOperationCallback inner, ITestOpCallback<ILSMIOOperation> scheduledCallback,
+ ITestOpCallback<ILSMIOOperation> beforeOperationCallback,
+ ITestOpCallback<ILSMIOOperation> afterOperationCallback,
+ ITestOpCallback<ILSMIOOperation> afterFinalizeCallback,
+ ITestOpCallback<ILSMIOOperation> completedCallback) {
+ this.encapsulated = inner;
+ this.scheduledCallback = scheduledCallback;
+ this.beforeOperationCallback = beforeOperationCallback;
+ this.afterOperationCallback = afterOperationCallback;
+ this.afterFinalizeCallback = afterFinalizeCallback;
+ this.completedCallback = completedCallback;
+ }
+
+ @Override
+ public void scheduled(ILSMIOOperation operation) throws HyracksDataException {
+ scheduledCallback.before(operation);
+ encapsulated.scheduled(operation);
+ scheduledCallback.after(operation);
+ }
+
+ @Override
+ public void beforeOperation(ILSMIOOperation operation) throws HyracksDataException {
+ beforeOperationCallback.before(operation);
+ encapsulated.beforeOperation(operation);
+ beforeOperationCallback.after(operation);
+ }
+
+ @Override
+ public void afterOperation(ILSMIOOperation operation) throws HyracksDataException {
+ afterOperationCallback.before(operation);
+ encapsulated.afterOperation(operation);
+ afterOperationCallback.after(operation);
+ }
+
+ @Override
+ public void afterFinalize(ILSMIOOperation operation) throws HyracksDataException {
+ afterFinalizeCallback.before(operation);
+ encapsulated.afterFinalize(operation);
+ afterFinalizeCallback.after(operation);
+ }
+
+ @Override
+ public void completed(ILSMIOOperation operation) {
+ try {
+ completedCallback.before(operation);
+ } catch (HyracksDataException e) {
+ throw new IllegalStateException(e);
+ }
+ encapsulated.completed(operation);
+ try {
+ completedCallback.after(operation);
+ } catch (HyracksDataException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public void recycled(ILSMMemoryComponent component) throws HyracksDataException {
+ encapsulated.recycled(component);
+ }
+
+ @Override
+ public void allocated(ILSMMemoryComponent component) throws HyracksDataException {
+ encapsulated.allocated(component);
+ }
+
+ public ILSMIOOperationCallback getEncapsulated() {
+ return encapsulated;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallbackFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallbackFactory.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallbackFactory.java
new file mode 100644
index 0000000..bb10236
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/EncapsulatingIoCallbackFactory.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.hyracks.storage.am.lsm.btree;
+
+import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.btree.impl.ITestOpCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.common.IResource;
+
+public class EncapsulatingIoCallbackFactory implements ILSMIOOperationCallbackFactory {
+
+ private static final long serialVersionUID = 1L;
+ private final ILSMIOOperationCallbackFactory encapsulated;
+ private final ITestOpCallback<ILSMIOOperation> scheduledCallback;
+ private final ITestOpCallback<ILSMIOOperation> beforeOperationCallback;
+ private final ITestOpCallback<ILSMIOOperation> afterOperationCallback;
+ private final ITestOpCallback<ILSMIOOperation> afterFinalizeCallback;
+ private final ITestOpCallback<ILSMIOOperation> completedCallback;
+
+ public EncapsulatingIoCallbackFactory(ILSMIOOperationCallbackFactory factory,
+ ITestOpCallback<ILSMIOOperation> scheduledCallback,
+ ITestOpCallback<ILSMIOOperation> beforeOperationCallback,
+ ITestOpCallback<ILSMIOOperation> afterOperationCallback,
+ ITestOpCallback<ILSMIOOperation> afterFinalizeCallback,
+ ITestOpCallback<ILSMIOOperation> completedCallback) {
+ encapsulated = factory;
+ this.scheduledCallback = scheduledCallback;
+ this.beforeOperationCallback = beforeOperationCallback;
+ this.afterOperationCallback = afterOperationCallback;
+ this.afterFinalizeCallback = afterFinalizeCallback;
+ this.completedCallback = completedCallback;
+ }
+
+ @Override
+ public void initialize(INCServiceContext ncCtx, IResource resource) {
+ encapsulated.initialize(ncCtx, resource);
+ }
+
+ @Override
+ public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
+ ILSMIOOperationCallback inner = encapsulated.createIoOpCallback(index);
+ return new EncapsulatingIoCallback(inner, scheduledCallback, beforeOperationCallback, afterOperationCallback,
+ afterFinalizeCallback, completedCallback);
+ }
+
+ @Override
+ public int getCurrentMemoryComponentIndex() throws HyracksDataException {
+ return encapsulated.getCurrentMemoryComponentIndex();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeComponentLifecycleTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeComponentLifecycleTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeComponentLifecycleTest.java
new file mode 100644
index 0000000..21f3877
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeComponentLifecycleTest.java
@@ -0,0 +1,470 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.hyracks.storage.am.lsm.btree;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.SingleThreadEventProcessor;
+import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import org.apache.hyracks.storage.am.btree.OrderedIndexTestContext;
+import org.apache.hyracks.storage.am.btree.OrderedIndexTestUtils;
+import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
+import org.apache.hyracks.storage.am.lsm.btree.impl.CountingIoOperationCallback;
+import org.apache.hyracks.storage.am.lsm.btree.impl.CountingIoOperationCallbackFactory;
+import org.apache.hyracks.storage.am.lsm.btree.impl.ITestOpCallback;
+import org.apache.hyracks.storage.am.lsm.btree.impl.NoOpTestCallback;
+import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestContext;
+import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.LSMComponentType;
+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.ILSMIOOperationCallbackFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.IIoOperationFailedCallback;
+import org.apache.hyracks.storage.am.lsm.common.impls.AsynchronousScheduler;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LSMBTreeComponentLifecycleTest {
+ private static final Logger LOGGER = LogManager.getLogger();
+ private final LSMBTreeTestHarness harness = new LSMBTreeTestHarness();
+ private final OrderedIndexTestUtils testUtils = new OrderedIndexTestUtils();
+ private final ISerializerDeserializer[] fieldSerdes =
+ { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+ private final int numKeys = 1;
+ private static final int numTuplesToInsert = 100;
+
+ @Before
+ public void setUp() throws HyracksDataException {
+ harness.setUp();
+ }
+
+ @After
+ public void tearDown() throws HyracksDataException {
+ harness.tearDown();
+ }
+
+ private OrderedIndexTestContext createTestContext(ISerializerDeserializer[] fieldSerdes, int numKeys,
+ ILSMIOOperationScheduler scheduler, ILSMIOOperationCallbackFactory ioCallbackFactory) throws Exception {
+ return LSMBTreeTestContext.create(harness.getIOManager(), harness.getVirtualBufferCaches(),
+ harness.getFileReference(), harness.getDiskBufferCache(), fieldSerdes, numKeys,
+ harness.getBoomFilterFalsePositiveRate(), harness.getMergePolicy(), harness.getOperationTracker(),
+ scheduler, ioCallbackFactory, harness.getMetadataPageManagerFactory(), false, true, false);
+ }
+
+ private OrderedIndexTestContext createTestContext(ISerializerDeserializer[] fieldSerdes, int numKeys)
+ throws Exception {
+ return createTestContext(fieldSerdes, numKeys, harness.getIOScheduler(),
+ harness.getIOOperationCallbackFactory());
+ }
+
+ @Test
+ public void testFlushUnallocatedIndex() throws Exception {
+ OrderedIndexTestContext ctx = createTestContext(fieldSerdes, numKeys);
+ ILSMIndex index = (ILSMIndex) ctx.getIndex();
+ index.create();
+ index.activate();
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(1), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(0, index.getDiskComponents().size());
+ CountingIoOperationCallback ioCallback = (CountingIoOperationCallback) index.getIOOperationCallback();
+ // assert equal before, after, after were called
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // insert into the index
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ // flush
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(1, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // insert more
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ // flush
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(1), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(2, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ ctx.getIndex().deactivate();
+ ctx.getIndex().destroy();
+ }
+
+ @Test
+ public void testFlushUnallocatedIndexStartFromSecondComponent() throws Exception {
+ CountingIoOperationCallbackFactory.STARTING_INDEX = 1;
+ try {
+ testFlushUnallocatedIndex();
+ } finally {
+ CountingIoOperationCallbackFactory.STARTING_INDEX = 0;
+ }
+ }
+
+ @Test
+ public void testNormalFlushOperation() throws Exception {
+ OrderedIndexTestContext ctx = createTestContext(fieldSerdes, numKeys);
+ ILSMIndex index = (ILSMIndex) ctx.getIndex();
+ index.create();
+ index.activate();
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(1), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(1, index.getDiskComponents().size());
+ CountingIoOperationCallback ioCallback = (CountingIoOperationCallback) index.getIOOperationCallback();
+ // assert equal before, after, after were called
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // insert into the index
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ // flush
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(2, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // insert more
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ // flush
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(1), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(3, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ ctx.getIndex().deactivate();
+ ctx.getIndex().destroy();
+ }
+
+ @Test
+ public void testNormalFlushOperationStartFromSecondComponent() throws Exception {
+ CountingIoOperationCallbackFactory.STARTING_INDEX = 1;
+ try {
+ testNormalFlushOperation();
+ } finally {
+ CountingIoOperationCallbackFactory.STARTING_INDEX = 0;
+ }
+ }
+
+ @Test
+ public void testFlushUnModifiedComponent() throws Exception {
+ OrderedIndexTestContext ctx = createTestContext(fieldSerdes, numKeys);
+ ILSMIndex index = (ILSMIndex) ctx.getIndex();
+ index.create();
+ index.activate();
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(1), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(1, index.getDiskComponents().size());
+ CountingIoOperationCallback ioCallback = (CountingIoOperationCallback) index.getIOOperationCallback();
+ // assert equal before, after, finalize were called
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // flush, there was no insert before
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(1, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // insert more
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ // flush
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(1), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(2, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // insert more
+ testUtils.insertIntTuples(ctx, numTuplesToInsert, harness.getRandom());
+ // flush
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(3, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ ctx.getIndex().deactivate();
+ ctx.getIndex().destroy();
+ }
+
+ @Test
+ public void testFlushUnModifiedComponentStartFromSecondComponent() throws Exception {
+ CountingIoOperationCallbackFactory.STARTING_INDEX = 1;
+ try {
+ testFlushUnModifiedComponent();
+ } finally {
+ CountingIoOperationCallbackFactory.STARTING_INDEX = 0;
+ }
+ }
+
+ public int getExpectedMemoryComponentIndex(int expectedIndex) {
+ return (CountingIoOperationCallbackFactory.STARTING_INDEX + expectedIndex) % 2;
+ }
+
+ @Test
+ public void testScheduleMoreFlushesThanComponents() throws Exception {
+ final AtomicInteger counter = new AtomicInteger();
+ final Semaphore flushSemaphore = new Semaphore(0);
+ OrderedIndexTestContext ctx = createTestContext(fieldSerdes, numKeys, new AsynchronousScheduler(
+ r -> new Thread(r, "LsmIoThread-" + counter.getAndIncrement()), new IIoOperationFailedCallback() {
+ @Override
+ public void schedulerFailed(ILSMIOOperationScheduler scheduler, Throwable failure) {
+ LOGGER.log(Level.ERROR, "Scheduler failed", failure);
+ }
+
+ @Override
+ public void operationFailed(ILSMIOOperation operation, Throwable failure) {
+ LOGGER.log(Level.ERROR, "Operation {} failed", operation, failure);
+ }
+ }), new EncapsulatingIoCallbackFactory(harness.getIOOperationCallbackFactory(), NoOpTestCallback.get(),
+ NoOpTestCallback.get(), new ITestOpCallback<ILSMIOOperation>() {
+ @Override
+ public void before(ILSMIOOperation t) throws HyracksDataException {
+ try {
+ flushSemaphore.acquire();
+ } catch (InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ @Override
+ public void after(ILSMIOOperation t) throws HyracksDataException {
+ }
+ }, NoOpTestCallback.get(), NoOpTestCallback.get()));
+ ILSMIndex index = (ILSMIndex) ctx.getIndex();
+ index.create();
+ index.activate();
+ Assert.assertEquals(getExpectedMemoryComponentIndex(0), index.getCurrentMemoryComponentIndex());
+ int numMemoryComponents = index.getNumberOfAllMemoryComponents();
+ // create a flusher that will schedule 13 flushes.
+ // wait for all flushes to be scheduled.
+ // create an inserter that will insert some records.
+ // one by one allow flushes until one flush remains, and ensure no record went in.
+ // allow the last flush, then wait for the inserts to succeed, and ensure they went to
+ // the expected memory component
+ final int numFlushes = 13;
+ User firstUser = new User("FirstUser");
+ User secondUser = new User("SecondUser");
+ Request flushRequest = new Request(Request.Statement.FLUSH, ctx, numFlushes);
+ firstUser.add(flushRequest);
+ firstUser.step();
+ // wait until all flushes have been scheduled.. Not yet performed
+ flushRequest.await(1);
+ // create an inserter and allow it to go all the way
+ Request insertRequest = new Request(Request.Statement.INSERT, ctx, 1);
+ secondUser.add(insertRequest);
+ secondUser.step();
+ secondUser.step();
+ ILSMIndexAccessor accessor = index.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ ILSMIndexOperationContext opCtx = accessor.getOpContext();
+ assertCorrectSearchComponents(opCtx, index, 0);
+ // Allow one flush at a time and ensure that inserter didn't succeed
+ for (int i = 0; i < numFlushes - 1; i++) {
+ flushSemaphore.release();
+ firstUser.step();
+ flushRequest.await(2 + i);
+ Assert.assertEquals(0, insertRequest.getSteps());
+ // also ensure that you get the correct components when searching
+ assertCorrectSearchComponents(opCtx, index, i + 1);
+ }
+ // Allow last flush to proceed
+ flushSemaphore.release();
+ // wait for the insert to complete
+ insertRequest.await();
+ firstUser.step();
+ firstUser.step();
+ flushRequest.await();
+ firstUser.stop();
+ secondUser.stop();
+
+ int expectedMemoryComponent = numFlushes % numMemoryComponents;
+ Assert.assertEquals(getExpectedMemoryComponentIndex(expectedMemoryComponent),
+ index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(0, index.getDiskComponents().size());
+
+ EncapsulatingIoCallback encapsulating = (EncapsulatingIoCallback) index.getIOOperationCallback();
+ CountingIoOperationCallback ioCallback = (CountingIoOperationCallback) encapsulating.getEncapsulated();
+ // assert equal before, after, finalize were called
+ Assert.assertEquals(numFlushes, ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // flush, there was no insert before
+ flushSemaphore.release();
+ flush(ctx);
+ Assert.assertEquals(getExpectedMemoryComponentIndex((expectedMemoryComponent + 1) % numMemoryComponents),
+ index.getCurrentMemoryComponentIndex());
+ Assert.assertEquals(1, index.getDiskComponents().size());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterOperationCount());
+ Assert.assertEquals(ioCallback.getBeforeOperationCount(), ioCallback.getAfterFinalizeCount());
+ // deactivate will cause a flush
+ flushSemaphore.release();
+ ctx.getIndex().deactivate();
+ ctx.getIndex().destroy();
+ }
+
+ private void assertCorrectSearchComponents(ILSMIndexOperationContext opCtx, ILSMIndex index,
+ int numSuccesfullyCompletedFlushes) throws HyracksDataException {
+ opCtx.reset();
+ opCtx.setOperation(IndexOperation.SEARCH);
+ index.getOperationalComponents(opCtx);
+ List<ILSMMemoryComponent> memComponents = index.getMemoryComponents();
+ int first = numSuccesfullyCompletedFlushes % memComponents.size();
+ Assert.assertEquals(memComponents.get(first), getFirstMemoryComponent(opCtx));
+ }
+
+ private ILSMComponent getFirstMemoryComponent(ILSMIndexOperationContext opCtx) {
+ List<ILSMComponent> components = opCtx.getComponentHolder();
+ // backward
+ for (int i = components.size() - 1; i >= 0; i--) {
+ ILSMComponent next = components.get(i);
+ if (next.getType() == LSMComponentType.MEMORY) {
+ return next;
+ }
+ }
+ return null;
+ }
+
+ private void flush(OrderedIndexTestContext ctx) throws HyracksDataException, InterruptedException {
+ ILSMIOOperation flush = scheduleFlush(ctx);
+ flush.sync();
+ if (flush.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flush.getFailure());
+ }
+ }
+
+ private ILSMIOOperation scheduleFlush(OrderedIndexTestContext ctx)
+ throws HyracksDataException, InterruptedException {
+ ILSMIndexAccessor accessor =
+ (ILSMIndexAccessor) ctx.getIndex().createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ return accessor.scheduleFlush();
+ }
+
+ private static class Request {
+ private enum Statement {
+ FLUSH,
+ INSERT
+ }
+
+ private final Statement statement;
+ private final OrderedIndexTestContext ctx;
+ private final int repeats;
+ private boolean done = false;
+ private int step = 0;
+
+ public Request(Statement statement, OrderedIndexTestContext ctx, int repeats) {
+ this.statement = statement;
+ this.ctx = ctx;
+ this.repeats = repeats;
+ }
+
+ Statement statement() {
+ return statement;
+ }
+
+ synchronized void complete() {
+ done = true;
+ notifyAll();
+ }
+
+ synchronized void await() throws InterruptedException {
+ while (!done) {
+ wait();
+ }
+ }
+
+ synchronized void step() {
+ step++;
+ notifyAll();
+ }
+
+ synchronized int getSteps() {
+ return step;
+ }
+
+ synchronized void await(int step) throws InterruptedException {
+ while (this.step < step) {
+ wait();
+ }
+ }
+ }
+
+ private class User extends SingleThreadEventProcessor<Request> {
+
+ private Semaphore step = new Semaphore(0);
+
+ public User(String username) {
+ super(username);
+ }
+
+ public void step() {
+ step.release();
+ }
+
+ @Override
+ protected void handle(Request req) throws Exception {
+ try {
+ step.acquire();
+ switch (req.statement()) {
+ case FLUSH:
+ List<ILSMIOOperation> flushes = new ArrayList<>(req.repeats);
+ for (int i = 0; i < req.repeats; i++) {
+ flushes.add(scheduleFlush(req.ctx));
+ }
+ req.step();
+ for (ILSMIOOperation op : flushes) {
+ step.acquire();
+ op.sync();
+ if (op.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(op.getFailure());
+ }
+ req.step(); // report after completion of each flush
+ }
+ break;
+ case INSERT:
+ testUtils.insertIntTuples(req.ctx, numTuplesToInsert, harness.getRandom());
+ break;
+ default:
+ break;
+ }
+ req.step();
+ step.acquire();
+ } finally {
+ req.step();
+ req.complete();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFileManagerTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFileManagerTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFileManagerTest.java
index 9d768aa..03b221e 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFileManagerTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFileManagerTest.java
@@ -29,7 +29,6 @@ import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTree;
import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTreeWithBloomFilterDiskComponent;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestContext;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.junit.After;
import org.junit.Assert;
@@ -71,8 +70,8 @@ public class LSMBTreeFileManagerTest {
ILSMIndexAccessor accessor = (ILSMIndexAccessor) ctx.getIndexAccessor();
accessor.insert(tuple);
- // Flush to generate a disk component
- accessor.scheduleFlush(((ILSMIndex) ctx.getIndex()).getIOOperationCallback());
+ // Flush to generate a disk component. This uses synchronous scheduler
+ accessor.scheduleFlush();
// Make sure the disk component was generated
LSMBTree btree = (LSMBTree) ctx.getIndex();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFilterMergeTestDriver.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFilterMergeTestDriver.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFilterMergeTestDriver.java
index 815c8bc..1b9a7a5 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFilterMergeTestDriver.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeFilterMergeTestDriver.java
@@ -40,9 +40,9 @@ import org.apache.hyracks.storage.am.config.AccessMethodTestsConfig;
import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTree;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+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.ILSMIndexAccessor;
-import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper;
-import org.apache.hyracks.storage.am.lsm.common.impls.StubIOOperationCallback;
import org.junit.Assert;
/**
@@ -113,13 +113,14 @@ public abstract class LSMBTreeFilterMergeTestDriver extends OrderedIndexTestDriv
TreeIndexTestUtils.compareFilterTuples(obsMinMax.getRight(), minMax.getRight(), comp));
}
- StubIOOperationCallback stub = new StubIOOperationCallback();
- BlockingIOOperationCallbackWrapper waiter = new BlockingIOOperationCallbackWrapper(stub);
- accessor.scheduleFlush(waiter);
- waiter.waitForIO();
+ ILSMIOOperation flush = accessor.scheduleFlush();
+ flush.sync();
+ if (flush.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flush.getFailure());
+ }
if (minMax != null) {
Pair<ITupleReference, ITupleReference> obsMinMax =
- filterToMinMax(stub.getLastNewComponent().getLSMComponentFilter());
+ filterToMinMax(flush.getNewComponent().getLSMComponentFilter());
Assert.assertEquals(0,
TreeIndexTestUtils.compareFilterTuples(obsMinMax.getLeft(), minMax.getLeft(), comp));
Assert.assertEquals(0,
@@ -143,8 +144,7 @@ public abstract class LSMBTreeFilterMergeTestDriver extends OrderedIndexTestDriv
expectedMergeMinMax.setRight(componentMinMax.getRight());
}
}
- accessor.scheduleMerge(((LSMBTree) ctx.getIndex()).getIOOperationCallback(),
- ((LSMBTree) ctx.getIndex()).getDiskComponents());
+ accessor.scheduleMerge(((LSMBTree) ctx.getIndex()).getDiskComponents());
flushedComponents = ((LSMBTree) ctx.getIndex()).getDiskComponents();
Pair<ITupleReference, ITupleReference> mergedMinMax =
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeFailTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeFailTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeFailTest.java
index 475ab9c..a8639c3 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeFailTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeFailTest.java
@@ -35,6 +35,7 @@ import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTreeRangeSearchCursor;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestContext;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
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.ILSMIOOperationScheduler;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
@@ -76,32 +77,24 @@ public class LSMBTreeMergeFailTest {
ITupleReference tuple1 = TupleUtils.createIntegerTuple(1, 1, 1);
accessor.insert(tuple1);
// flush component
- accessor.scheduleFlush(btree.getIOOperationCallback());
+ accessor.scheduleFlush();
ITupleReference tuple2 = TupleUtils.createIntegerTuple(2, 2, 2);
accessor.insert(tuple2);
// flush component
- accessor.scheduleFlush(btree.getIOOperationCallback());
+ accessor.scheduleFlush();
ITupleReference tuple3 = TupleUtils.createIntegerTuple(3, 3, 3);
accessor.insert(tuple3);
// flush component
- accessor.scheduleFlush(btree.getIOOperationCallback());
-
+ accessor.scheduleFlush();
scheduler.modify = true;
-
- boolean exceptionThrown = false;
- try {
- accessor.scheduleMerge(btree.getIOOperationCallback(), btree.getDiskComponents());
- } catch (HyracksDataException e) {
- exceptionThrown = true;
- }
- Assert.assertTrue(exceptionThrown);
-
+ ILSMIOOperation merge = accessor.scheduleMerge(btree.getDiskComponents());
+ merge.sync();
+ Assert.assertEquals(LSMIOOperationStatus.FAILURE, merge.getStatus());
scheduler.modify = false;
- accessor.scheduleMerge(btree.getIOOperationCallback(), btree.getDiskComponents());
+ accessor.scheduleMerge(btree.getDiskComponents());
Assert.assertEquals(1, btree.getDiskComponents().size());
-
btree.deactivate();
btree.destroy();
}
@@ -120,14 +113,18 @@ public class LSMBTreeMergeFailTest {
@Override
public void scheduleOperation(ILSMIOOperation operation) throws HyracksDataException {
- if (modify) {
- try {
- modifyOperation(operation);
- } catch (Exception e) {
- throw HyracksDataException.create(e);
+ try {
+ if (modify) {
+ try {
+ modifyOperation(operation);
+ } catch (Exception e) {
+ throw HyracksDataException.create(e);
+ }
}
+ operation.call();
+ } finally {
+ operation.complete();
}
- operation.call();
}
private void modifyOperation(ILSMIOOperation operation) throws Exception {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeTestDriver.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeTestDriver.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeTestDriver.java
index 87ddf33..7c59671 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeTestDriver.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeMergeTestDriver.java
@@ -74,8 +74,7 @@ public abstract class LSMBTreeMergeTestDriver extends OrderedIndexTestDriver {
}
ILSMIndexAccessor accessor = (ILSMIndexAccessor) ctx.getIndexAccessor();
- accessor.scheduleMerge(((LSMBTree) ctx.getIndex()).getIOOperationCallback(),
- ((LSMBTree) ctx.getIndex()).getDiskComponents());
+ accessor.scheduleMerge(((LSMBTree) ctx.getIndex()).getDiskComponents());
orderedIndexTestUtils.checkPointSearches(ctx);
orderedIndexTestUtils.checkScan(ctx);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeModificationOperationCallbackTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeModificationOperationCallbackTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeModificationOperationCallbackTest.java
index 9f17efa..4fafb38 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeModificationOperationCallbackTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeModificationOperationCallbackTest.java
@@ -19,6 +19,7 @@
package org.apache.hyracks.storage.am.lsm.btree;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.utils.SerdeUtils;
import org.apache.hyracks.dataflow.common.utils.TupleUtils;
import org.apache.hyracks.storage.am.btree.AbstractModificationOperationCallbackTest;
@@ -26,9 +27,9 @@ import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
import org.apache.hyracks.storage.am.lsm.btree.utils.LSMBTreeUtil;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+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.ILSMIndexAccessor;
-import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper;
import org.apache.hyracks.storage.am.lsm.common.impls.NoOpOperationTrackerFactory;
import org.apache.hyracks.util.trace.ITracer;
import org.junit.Test;
@@ -72,8 +73,6 @@ public class LSMBTreeModificationOperationCallbackTest extends AbstractModificat
IndexAccessParameters actx = new IndexAccessParameters(cb, NoOpOperationCallback.INSTANCE);
ILSMIndexAccessor accessor = (ILSMIndexAccessor) index.createAccessor(actx);
- BlockingIOOperationCallbackWrapper ioOpCallback =
- new BlockingIOOperationCallbackWrapper(((ILSMIndex) index).getIOOperationCallback());
for (int j = 0; j < 2; j++) {
isFoundNull = true;
for (int i = 0; i < NUM_TUPLES; i++) {
@@ -82,8 +81,11 @@ public class LSMBTreeModificationOperationCallbackTest extends AbstractModificat
}
if (j == 1) {
- accessor.scheduleFlush(ioOpCallback);
- ioOpCallback.waitForIO();
+ ILSMIOOperation flush = accessor.scheduleFlush();
+ flush.sync();
+ if (flush.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flush.getFailure());
+ }
isFoundNull = true;
} else {
isFoundNull = false;
@@ -95,8 +97,7 @@ public class LSMBTreeModificationOperationCallbackTest extends AbstractModificat
}
if (j == 1) {
- accessor.scheduleFlush(ioOpCallback);
- ioOpCallback.waitForIO();
+ accessor.scheduleFlush().sync();
isFoundNull = true;
} else {
isFoundNull = false;
@@ -106,9 +107,7 @@ public class LSMBTreeModificationOperationCallbackTest extends AbstractModificat
TupleUtils.createIntegerTuple(builder, tuple, i);
accessor.delete(tuple);
}
-
- accessor.scheduleFlush(ioOpCallback);
- ioOpCallback.waitForIO();
+ accessor.scheduleFlush().sync();
}
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeScanDiskComponentsTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeScanDiskComponentsTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeScanDiskComponentsTest.java
index 48812b4..2574c4d 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeScanDiskComponentsTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeScanDiskComponentsTest.java
@@ -44,8 +44,6 @@ import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTree;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestContext;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.common.IIndexCursor;
import org.junit.After;
@@ -113,21 +111,20 @@ public class LSMBTreeScanDiskComponentsTest extends OrderedIndexTestDriver {
protected void test(OrderedIndexTestContext ctx, ISerializerDeserializer[] fieldSerdes)
throws HyracksDataException {
ILSMIndexAccessor accessor = (ILSMIndexAccessor) ctx.getIndexAccessor();
- ILSMIOOperationCallback ioCallback = ((ILSMIndex) ctx.getIndex()).getIOOperationCallback();
//component 2 contains 1 and 2
upsertTuple(ctx, fieldSerdes, getValue(1, fieldSerdes));
upsertTuple(ctx, fieldSerdes, getValue(2, fieldSerdes));
- accessor.scheduleFlush(ioCallback);
+ accessor.scheduleFlush();
//component 1 contains 1 and -2
upsertTuple(ctx, fieldSerdes, getValue(1, fieldSerdes));
deleteTuple(ctx, fieldSerdes, getValue(2, fieldSerdes));
- accessor.scheduleFlush(ioCallback);
+ accessor.scheduleFlush();
//component 0 contains 2 and 3
upsertTuple(ctx, fieldSerdes, getValue(3, fieldSerdes));
upsertTuple(ctx, fieldSerdes, getValue(2, fieldSerdes));
- accessor.scheduleFlush(ioCallback);
+ accessor.scheduleFlush();
LSMBTree btree = (LSMBTree) ctx.getIndex();
Assert.assertEquals("Check disk components", 3, btree.getDiskComponents().size());
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceScanDiskComponentsTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceScanDiskComponentsTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceScanDiskComponentsTest.java
index 3a48160..afcb3c1 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceScanDiskComponentsTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceScanDiskComponentsTest.java
@@ -52,8 +52,6 @@ import org.apache.hyracks.storage.am.lsm.btree.tuples.LSMBTreeTupleReference;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestContext;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.common.IIndexCursor;
import org.junit.After;
@@ -352,11 +350,10 @@ public class LSMBTreeUpdateInPlaceScanDiskComponentsTest extends OrderedIndexTes
throws HyracksDataException {
ILSMIndexAccessor accessor = (ILSMIndexAccessor) ctx.getIndexAccessor();
- ILSMIOOperationCallback ioCallback = ((ILSMIndex) ctx.getIndex()).getIOOperationCallback();
op1.performOperation(ctx, AccessMethodTestsConfig.BTREE_NUM_TUPLES_TO_INSERT);
op2.performOperation(ctx,
AccessMethodTestsConfig.BTREE_NUM_TUPLES_TO_INSERT / AccessMethodTestsConfig.BTREE_NUM_INSERT_ROUNDS);
- accessor.scheduleFlush(ioCallback);
+ accessor.scheduleFlush();
LSMBTree btree = (LSMBTree) ctx.getIndex();
Assert.assertEquals("Check disk components", 1, btree.getDiskComponents().size());
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceTest.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceTest.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceTest.java
index fc852cd..0914541 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceTest.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/LSMBTreeUpdateInPlaceTest.java
@@ -35,9 +35,9 @@ import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.config.AccessMethodTestsConfig;
import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeTestHarness;
import org.apache.hyracks.storage.am.lsm.btree.utils.LSMBTreeUtil;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+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.ILSMIndexAccessor;
-import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper;
import org.apache.hyracks.storage.am.lsm.common.impls.NoOpOperationTrackerFactory;
import org.apache.hyracks.storage.common.IIndexAccessor;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
@@ -99,8 +99,6 @@ public class LSMBTreeUpdateInPlaceTest extends AbstractOperationCallbackTest {
private void test(IndexModification op1, IndexModification op2) throws Exception {
ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) accessor;
- BlockingIOOperationCallbackWrapper ioOpCallback =
- new BlockingIOOperationCallbackWrapper(((ILSMIndex) index).getIOOperationCallback());
for (int j = 0; j < 2; j++) {
index.clear();
isFoundNull = true;
@@ -111,8 +109,11 @@ public class LSMBTreeUpdateInPlaceTest extends AbstractOperationCallbackTest {
}
if (j == 1) {
- lsmAccessor.scheduleFlush(ioOpCallback);
- ioOpCallback.waitForIO();
+ ILSMIOOperation flush = lsmAccessor.scheduleFlush();
+ flush.sync();
+ if (flush.getStatus() == LSMIOOperationStatus.FAILURE) {
+ throw HyracksDataException.create(flush.getFailure());
+ }
isFoundNull = true;
isUpdated = false;
} else {
@@ -126,8 +127,7 @@ public class LSMBTreeUpdateInPlaceTest extends AbstractOperationCallbackTest {
}
if (j == 1) {
- lsmAccessor.scheduleFlush(ioOpCallback);
- ioOpCallback.waitForIO();
+ lsmAccessor.scheduleFlush().sync();
} else {
isFoundNull = false;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/AllowTestOpCallback.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/AllowTestOpCallback.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/AllowTestOpCallback.java
index 19a9872..2e43198 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/AllowTestOpCallback.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/AllowTestOpCallback.java
@@ -35,7 +35,7 @@ public class AllowTestOpCallback implements ITestOpCallback<Semaphore> {
}
@Override
- public void after() throws HyracksDataException {
+ public void after(Semaphore t) throws HyracksDataException {
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallback.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallback.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallback.java
new file mode 100644
index 0000000..84b530a
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallback.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.hyracks.storage.am.lsm.btree.impl;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
+
+public class CountingIoOperationCallback implements ILSMIOOperationCallback {
+ private int beforeOperation;
+ private int afterOperation;
+ private int afterFinalize;
+ private int recycled;
+ private int allocated;
+ private int beforeSchedule;
+ private int destroy;
+
+ public CountingIoOperationCallback() {
+ }
+
+ public int getAfterFinalizeCount() {
+ return afterFinalize;
+ }
+
+ @Override
+ public void recycled(ILSMMemoryComponent component) throws HyracksDataException {
+ recycled++;
+ }
+
+ public int getRecycledCount() {
+ return recycled;
+ }
+
+ @Override
+ public void allocated(ILSMMemoryComponent component) throws HyracksDataException {
+ allocated++;
+ }
+
+ public int getAllocatedCount() {
+ return allocated;
+ }
+
+ @Override
+ public void scheduled(ILSMIOOperation operation) throws HyracksDataException {
+ beforeSchedule++;
+ }
+
+ public int getBeforeScheduleCount() {
+ return beforeSchedule;
+ }
+
+ @Override
+ public void beforeOperation(ILSMIOOperation operation) throws HyracksDataException {
+ beforeOperation++;
+ }
+
+ public int getBeforeOperationCount() {
+ return beforeOperation;
+ }
+
+ @Override
+ public void afterOperation(ILSMIOOperation operation) throws HyracksDataException {
+ afterOperation++;
+ }
+
+ public int getAfterOperationCount() {
+ return afterOperation;
+ }
+
+ @Override
+ public void afterFinalize(ILSMIOOperation operation) throws HyracksDataException {
+ afterFinalize++;
+ }
+
+ @Override
+ public void completed(ILSMIOOperation operation) {
+ destroy++;
+ }
+
+ public int getDestroyCount() {
+ return destroy;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallbackFactory.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallbackFactory.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallbackFactory.java
new file mode 100644
index 0000000..b64b9b1
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/CountingIoOperationCallbackFactory.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+package org.apache.hyracks.storage.am.lsm.btree.impl;
+
+import org.apache.hyracks.api.application.INCServiceContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallbackFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.common.IResource;
+
+public class CountingIoOperationCallbackFactory implements ILSMIOOperationCallbackFactory {
+
+ private static final long serialVersionUID = 1L;
+ public static int STARTING_INDEX = 0;
+
+ @Override
+ public void initialize(INCServiceContext ncCtx, IResource resource) {
+ // No op
+ }
+
+ @Override
+ public ILSMIOOperationCallback createIoOpCallback(ILSMIndex index) throws HyracksDataException {
+ return new CountingIoOperationCallback();
+ }
+
+ @Override
+ public int getCurrentMemoryComponentIndex() throws HyracksDataException {
+ return STARTING_INDEX;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/4ed0eac5/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
index e888238..fa3097d 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/ITestOpCallback.java
@@ -23,5 +23,5 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
public interface ITestOpCallback<T> {
void before(T t) throws HyracksDataException;
- void after() throws HyracksDataException;
+ void after(T t) throws HyracksDataException;
}