You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by am...@apache.org on 2017/11/21 05:32:31 UTC
[1/3] asterixdb git commit: [ASTERIXDB-2169][STO][TX] Unblock
modifications during full scan
Repository: asterixdb
Updated Branches:
refs/heads/master 76ecc4b19 -> e5a65429d
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
index b0abeb1..e1d5114 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
@@ -63,6 +63,7 @@ public class LSMHarness implements ILSMHarness {
private static final Logger LOGGER = Logger.getLogger(LSMHarness.class.getName());
protected final ILSMIndex lsmIndex;
+ protected final ComponentReplacementContext componentReplacementCtx;
protected final ILSMMergePolicy mergePolicy;
protected final ILSMOperationTracker opTracker;
protected final AtomicBoolean fullMergeIsRequested;
@@ -84,83 +85,97 @@ public class LSMHarness implements ILSMHarness {
if (replicationEnabled) {
this.componentsToBeReplicated = new ArrayList<>();
}
+ componentReplacementCtx = new ComponentReplacementContext(lsmIndex);
}
protected boolean getAndEnterComponents(ILSMIndexOperationContext ctx, LSMOperationType opType,
boolean isTryOperation) throws HyracksDataException {
- validateOperationEnterComponentsState(ctx);
- synchronized (opTracker) {
- while (true) {
- lsmIndex.getOperationalComponents(ctx);
- // Before entering the components, prune those corner cases that indeed should not proceed.
- switch (opType) {
- case FLUSH:
- // if the lsm index does not have memory components allocated, then nothing to flush
- if (!lsmIndex.isMemoryComponentsAllocated()) {
- return false;
- }
- ILSMMemoryComponent flushingComponent = (ILSMMemoryComponent) ctx.getComponentHolder().get(0);
- if (!flushingComponent.isModified()) {
- if (flushingComponent.getState() == ComponentState.READABLE_UNWRITABLE) {
- //The mutable component has not been modified by any writer. There is nothing to flush.
- //since the component is empty, set its state back to READABLE_WRITABLE only when it's
- //state has been set to READABLE_UNWRITABLE
- flushingComponent.setState(ComponentState.READABLE_WRITABLE);
- opTracker.notifyAll();
-
- // Call recycled only when we change it's state is reset back to READABLE_WRITABLE
- // Otherwise, if the component is in other state, e.g., INACTIVE, or
- // READABLE_UNWRITABLE_FLUSHING, it's not considered as being recycled here.
- lsmIndex.getIOOperationCallback().recycled(flushingComponent);
+ long before = 0L;
+ if (ctx.isTracingEnabled()) {
+ before = System.nanoTime();
+ }
+ try {
+ validateOperationEnterComponentsState(ctx);
+ synchronized (opTracker) {
+ while (true) {
+ lsmIndex.getOperationalComponents(ctx);
+ // Before entering the components, prune those corner cases that indeed should not proceed.
+ switch (opType) {
+ case FLUSH:
+ // if the lsm index does not have memory components allocated, then nothing to flush
+ if (!lsmIndex.isMemoryComponentsAllocated()) {
+ return false;
}
- return false;
- }
- if (flushingComponent.getWriterCount() > 0) {
- /*
- * This case is a case where even though FLUSH log was flushed to disk and scheduleFlush is triggered,
- * the current in-memory component (whose state was changed to READABLE_WRITABLE (RW)
- * from READABLE_UNWRITABLE(RU) before FLUSH log was written to log tail (which is memory buffer of log file)
- * and then the state was changed back to RW (as shown in the following scenario)) can have writers
- * based on the current code base/design.
- * Thus, the writer count of the component may be greater than 0.
- * if this happens, intead of throwing exception, scheduleFlush() deal with this situation by not flushing
- * the component.
- * Please see issue 884 for more detail information:
- * https://code.google.com/p/asterixdb/issues/detail?id=884&q=owner%3Akisskys%40gmail.com&colspec=ID%20Type%20Status%20Priority%20Milestone%20Owner%20Summary%20ETA%20Severity
- *
- */
- return false;
- }
- break;
- case MERGE:
- if (ctx.getComponentHolder().size() < 2
- && ctx.getOperation() != IndexOperation.DELETE_DISK_COMPONENTS) {
- // There is only a single component. There is nothing to merge.
- return false;
+ ILSMMemoryComponent flushingComponent =
+ (ILSMMemoryComponent) ctx.getComponentHolder().get(0);
+ if (!flushingComponent.isModified()) {
+ recycle(flushingComponent);
+ return false;
+ }
+ if (flushingComponent.getWriterCount() > 0) {
+ /*
+ * This case is a case where even though FLUSH log was flushed to disk
+ * and scheduleFlush is triggered, the current in-memory component (whose state was
+ * changed to READABLE_WRITABLE (RW) from READABLE_UNWRITABLE(RU) before FLUSH log
+ * was written to log tail (which is memory buffer of log file) and then the state was
+ * changed back to RW (as shown in the following scenario)) can have writers based on
+ * the current code base/design. Thus, the writer count of the component may be greater
+ * than 0. if this happens, intead of throwing exception, scheduleFlush() deal with
+ * this situation by not flushing the component.
+ * for more detailed information: ASTERIXDB-1027
+ */
+ return false;
+ }
+ break;
+ case MERGE:
+ if (ctx.getComponentHolder().size() < 2
+ && ctx.getOperation() != IndexOperation.DELETE_DISK_COMPONENTS) {
+ // There is only a single component. There is nothing to merge.
+ return false;
+ }
+ break;
+ default:
+ break;
+ }
+ if (enterComponents(ctx, opType)) {
+ return true;
+ } else if (isTryOperation) {
+ return false;
+ }
+ try {
+ // Flush and merge operations should never reach this wait call,
+ // because they are always try operations. If they fail to enter the components,
+ // then it means that there are an ongoing flush/merge operation on
+ // the same components, so they should not proceed.
+ if (opType == LSMOperationType.MODIFICATION) {
+ // before waiting, make sure the index is in a modifiable state to avoid waiting forever.
+ ensureIndexModifiable();
}
- break;
- default:
- break;
- }
- if (enterComponents(ctx, opType)) {
- return true;
- } else if (isTryOperation) {
- return false;
- }
- try {
- // Flush and merge operations should never reach this wait call, because they are always try operations.
- // If they fail to enter the components, then it means that there are an ongoing flush/merge operation on
- // the same components, so they should not proceed.
- if (opType == LSMOperationType.MODIFICATION) {
- // before waiting, make sure the index is in a modifiable state to avoid waiting forever.
- ensureIndexModifiable();
+ opTracker.wait();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw HyracksDataException.create(e);
}
- opTracker.wait();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw HyracksDataException.create(e);
}
}
+ } finally {
+ if (ctx.isTracingEnabled()) {
+ ctx.incrementEnterExitTime(System.nanoTime() - before);
+ }
+ }
+ }
+
+ private void recycle(ILSMMemoryComponent flushingComponent) throws HyracksDataException {
+ if (flushingComponent.getState() == ComponentState.READABLE_UNWRITABLE) {
+ //The mutable component has not been modified by any writer.
+ // There is nothing to flush. Since the component is empty, set its state back
+ // to READABLE_WRITABLE only when it's state has been set to READABLE_UNWRITABLE
+ flushingComponent.setState(ComponentState.READABLE_WRITABLE);
+ opTracker.notifyAll(); // NOSONAR: Always synchronized from caller
+ // Call recycled only when we change it's state is reset back to READABLE_WRITABLE
+ // Otherwise, if the component is in other state, e.g., INACTIVE, or
+ // READABLE_UNWRITABLE_FLUSHING, it's not considered as being recycled here.
+ lsmIndex.getIOOperationCallback().recycled(flushingComponent);
}
}
@@ -179,7 +194,7 @@ public class LSMHarness implements ILSMHarness {
numEntered++;
}
entranceSuccessful = numEntered == components.size();
- } catch (Throwable e) {
+ } catch (Throwable e) { // NOSONAR: Log and re-throw
if (LOGGER.isLoggable(Level.SEVERE)) {
LOGGER.log(Level.SEVERE, opType.name() + " failed to enter components on " + lsmIndex, e);
}
@@ -196,10 +211,12 @@ public class LSMHarness implements ILSMHarness {
i++;
numEntered--;
}
- return false;
}
- ctx.setAccessingComponents(true);
}
+ if (!entranceSuccessful) {
+ return false;
+ }
+ ctx.setAccessingComponents(true);
// Check if there is any action that is needed to be taken based on the operation type
switch (opType) {
case FLUSH:
@@ -209,7 +226,7 @@ public class LSMHarness implements ILSMHarness {
lsmIndex.changeMutableComponent();
// Notify all waiting threads whenever a flush has been scheduled since they will check
// again if they can grab and enter the mutable component.
- opTracker.notifyAll();
+ opTracker.notifyAll(); // NOSONAR: Always called from a synchronized block
break;
case MERGE:
lsmIndex.getIOOperationCallback().beforeOperation(LSMIOOperationType.MERGE);
@@ -221,8 +238,8 @@ public class LSMHarness implements ILSMHarness {
return true;
}
- private void exitComponents(ILSMIndexOperationContext ctx, LSMOperationType opType, ILSMDiskComponent newComponent,
- boolean failedOperation) throws HyracksDataException {
+ private void doExitComponents(ILSMIndexOperationContext ctx, LSMOperationType opType,
+ ILSMDiskComponent newComponent, boolean failedOperation) throws HyracksDataException {
/**
* FLUSH and MERGE operations should always exit the components
* to notify waiting threads.
@@ -249,75 +266,10 @@ public class LSMHarness implements ILSMHarness {
} else if (opType == LSMOperationType.MERGE) {
opTracker.notifyAll();
}
-
- int i = 0;
- // First check if there is any action that is needed to be taken based on the state of each component.
- for (ILSMComponent c : ctx.getComponentHolder()) {
- boolean isMutableComponent = i == 0 && c.getType() == LSMComponentType.MEMORY ? true : false;
- c.threadExit(opType, failedOperation, isMutableComponent);
- if (c.getType() == LSMComponentType.MEMORY) {
- switch (c.getState()) {
- case READABLE_UNWRITABLE:
- if (isMutableComponent && (opType == LSMOperationType.MODIFICATION
- || opType == LSMOperationType.FORCE_MODIFICATION)) {
- lsmIndex.changeFlushStatusForCurrentMutableCompoent(true);
- }
- break;
- case INACTIVE:
- tracer.instant(c.toString(), traceCategory, Scope.p, lsmIndex.toString());
- ((AbstractLSMMemoryComponent) c).reset();
- // Notify all waiting threads whenever the mutable component's state has changed to
- // inactive. This is important because even though we switched the mutable
- // components, it is possible that the component that we just switched to is still
- // busy flushing its data to disk. Thus, the notification that was issued upon
- // scheduling the flush is not enough.
- opTracker.notifyAll();
- break;
- default:
- break;
- }
- } else {
- switch (c.getState()) {
- case INACTIVE:
- lsmIndex.addInactiveDiskComponent((AbstractLSMDiskComponent) c);
- break;
- default:
- break;
- }
- }
- i++;
- }
+ exitOperationalComponents(ctx, opType, failedOperation);
ctx.setAccessingComponents(false);
- // Then, perform any action that is needed to be taken based on the operation type.
- switch (opType) {
- case FLUSH:
- // newComponent is null if the flush op. was not performed.
- if (!failedOperation && newComponent != null) {
- lsmIndex.addDiskComponent(newComponent);
- if (replicationEnabled) {
- componentsToBeReplicated.clear();
- componentsToBeReplicated.add(newComponent);
- triggerReplication(componentsToBeReplicated, false, opType);
- }
- mergePolicy.diskComponentAdded(lsmIndex, false);
- }
- break;
- case MERGE:
- // newComponent is null if the merge op. was not performed.
- if (!failedOperation && newComponent != null) {
- lsmIndex.subsumeMergedComponents(newComponent, ctx.getComponentHolder());
- if (replicationEnabled) {
- componentsToBeReplicated.clear();
- componentsToBeReplicated.add(newComponent);
- triggerReplication(componentsToBeReplicated, false, opType);
- }
- mergePolicy.diskComponentAdded(lsmIndex, fullMergeIsRequested.get());
- }
- break;
- default:
- break;
- }
- } catch (Throwable e) {
+ exitOperation(ctx, opType, newComponent, failedOperation);
+ } catch (Throwable e) { // NOSONAR: Log and re-throw
if (LOGGER.isLoggable(Level.SEVERE)) {
LOGGER.log(Level.SEVERE, e.getMessage(), e);
}
@@ -326,7 +278,8 @@ public class LSMHarness implements ILSMHarness {
if (failedOperation && (opType == LSMOperationType.MODIFICATION
|| opType == LSMOperationType.FORCE_MODIFICATION)) {
//When the operation failed, completeOperation() method must be called
- //in order to decrement active operation count which was incremented in beforeOperation() method.
+ //in order to decrement active operation count which was incremented
+ // in beforeOperation() method.
opTracker.completeOperation(lsmIndex, opType, ctx.getSearchOperationCallback(),
ctx.getModificationCallback());
} else {
@@ -344,9 +297,9 @@ public class LSMHarness implements ILSMHarness {
if (!inactiveDiskComponents.isEmpty()) {
for (ILSMDiskComponent inactiveComp : inactiveDiskComponents) {
if (inactiveComp.getFileReferenceCount() == 1) {
- if (inactiveDiskComponentsToBeDeleted == null) {
- inactiveDiskComponentsToBeDeleted = new LinkedList<>();
- }
+ inactiveDiskComponentsToBeDeleted =
+ inactiveDiskComponentsToBeDeleted == null ? new LinkedList<>()
+ : inactiveDiskComponentsToBeDeleted;
inactiveDiskComponentsToBeDeleted.add(inactiveComp);
}
}
@@ -370,15 +323,97 @@ public class LSMHarness implements ILSMHarness {
for (ILSMDiskComponent c : inactiveDiskComponentsToBeDeleted) {
c.deactivateAndDestroy();
}
- } catch (Throwable e) {
+ } catch (Throwable e) { // NOSONAR Log and re-throw
if (LOGGER.isLoggable(Level.WARNING)) {
LOGGER.log(Level.WARNING, "Failure scheduling replication or destroying merged component", e);
}
- throw e;
+ throw e; // NOSONAR: The last call in the finally clause
}
}
}
+ }
+ private void exitOperation(ILSMIndexOperationContext ctx, LSMOperationType opType, ILSMDiskComponent newComponent,
+ boolean failedOperation) throws HyracksDataException {
+ // Then, perform any action that is needed to be taken based on the operation type.
+ switch (opType) {
+ case FLUSH:
+ // newComponent is null if the flush op. was not performed.
+ if (!failedOperation && newComponent != null) {
+ lsmIndex.addDiskComponent(newComponent);
+ if (replicationEnabled) {
+ componentsToBeReplicated.clear();
+ componentsToBeReplicated.add(newComponent);
+ triggerReplication(componentsToBeReplicated, false, opType);
+ }
+ mergePolicy.diskComponentAdded(lsmIndex, false);
+ }
+ break;
+ case MERGE:
+ // newComponent is null if the merge op. was not performed.
+ if (!failedOperation && newComponent != null) {
+ lsmIndex.subsumeMergedComponents(newComponent, ctx.getComponentHolder());
+ if (replicationEnabled) {
+ componentsToBeReplicated.clear();
+ componentsToBeReplicated.add(newComponent);
+ triggerReplication(componentsToBeReplicated, false, opType);
+ }
+ mergePolicy.diskComponentAdded(lsmIndex, fullMergeIsRequested.get());
+ }
+ break;
+ default:
+ break;
+ }
+ }
+
+ private void exitOperationalComponents(ILSMIndexOperationContext ctx, LSMOperationType opType,
+ boolean failedOperation) throws HyracksDataException {
+ // First check if there is any action that is needed to be taken
+ // based on the state of each component.
+ for (int i = 0; i < ctx.getComponentHolder().size(); i++) {
+ ILSMComponent c = ctx.getComponentHolder().get(i);
+ boolean isMutableComponent = i == 0 && c.getType() == LSMComponentType.MEMORY ? true : false;
+ c.threadExit(opType, failedOperation, isMutableComponent);
+ if (c.getType() == LSMComponentType.MEMORY) {
+ switch (c.getState()) {
+ case READABLE_UNWRITABLE:
+ if (isMutableComponent && (opType == LSMOperationType.MODIFICATION
+ || opType == LSMOperationType.FORCE_MODIFICATION)) {
+ lsmIndex.changeFlushStatusForCurrentMutableCompoent(true);
+ }
+ break;
+ case INACTIVE:
+ tracer.instant(c.toString(), traceCategory, Scope.p, lsmIndex.toString());
+ ((AbstractLSMMemoryComponent) c).reset();
+ // Notify all waiting threads whenever the mutable component's state
+ // has changed to inactive. This is important because even though we switched
+ // the mutable components, it is possible that the component that we just
+ // switched to is still busy flushing its data to disk. Thus, the notification
+ // that was issued upon scheduling the flush is not enough.
+ opTracker.notifyAll(); // NOSONAR: Always called inside synchronized block
+ break;
+ default:
+ break;
+ }
+ } else if (c.getState() == ComponentState.INACTIVE) {
+ lsmIndex.addInactiveDiskComponent((AbstractLSMDiskComponent) c);
+ }
+ }
+ }
+
+ private void exitComponents(ILSMIndexOperationContext ctx, LSMOperationType opType, ILSMDiskComponent newComponent,
+ boolean failedOperation) throws HyracksDataException {
+ long before = 0L;
+ if (ctx.isTracingEnabled()) {
+ before = System.nanoTime();
+ }
+ try {
+ doExitComponents(ctx, opType, newComponent, failedOperation);
+ } finally {
+ if (ctx.isTracingEnabled()) {
+ ctx.incrementEnterExitTime(System.nanoTime() - before);
+ }
+ }
}
@Override
@@ -532,7 +567,7 @@ public class LSMHarness implements ILSMHarness {
newComponent = lsmIndex.flush(operation);
operation.getCallback().afterOperation(LSMIOOperationType.FLUSH, null, newComponent);
newComponent.markAsValid(lsmIndex.isDurable());
- } catch (Throwable e) {
+ } catch (Throwable e) { // NOSONAR Log and re-throw
failedOperation = true;
if (LOGGER.isLoggable(Level.SEVERE)) {
LOGGER.log(Level.SEVERE, "Flush failed on " + lsmIndex, e);
@@ -562,8 +597,9 @@ public class LSMHarness implements ILSMHarness {
throws HyracksDataException {
fullMergeIsRequested.set(true);
if (!getAndEnterComponents(ctx, LSMOperationType.MERGE, true)) {
- // If the merge cannot be scheduled because there is already an ongoing merge on subset/all of the components, then
- // whenever the current merge has finished, it will schedule the full merge again.
+ // If the merge cannot be scheduled because there is already an ongoing merge on
+ // subset/all of the components, then whenever the current merge has finished,
+ // it will schedule the full merge again.
callback.afterFinalize(LSMIOOperationType.MERGE, null);
return;
}
@@ -583,7 +619,7 @@ public class LSMHarness implements ILSMHarness {
newComponent = lsmIndex.merge(operation);
operation.getCallback().afterOperation(LSMIOOperationType.MERGE, ctx.getComponentHolder(), newComponent);
newComponent.markAsValid(lsmIndex.isDurable());
- } catch (Throwable e) {
+ } catch (Throwable e) { // NOSONAR: Log and re-throw
failedOperation = true;
if (LOGGER.isLoggable(Level.SEVERE)) {
LOGGER.log(Level.SEVERE, "Failed merge operation on " + lsmIndex, e);
@@ -707,6 +743,7 @@ public class LSMHarness implements ILSMHarness {
throw e;
} finally {
exit(ctx);
+ ctx.logPerformanceCounters(accessor.getTupleCount());
}
}
@@ -878,4 +915,36 @@ public class LSMHarness implements ILSMHarness {
public String toString() {
return getClass().getSimpleName() + ":" + lsmIndex;
}
+
+ @Override
+ public void replaceMemoryComponentsWithDiskComponents(ILSMIndexOperationContext ctx, int startIndex)
+ throws HyracksDataException {
+ synchronized (opTracker) {
+ componentReplacementCtx.reset();
+ for (int i = 0; i < ctx.getComponentHolder().size(); i++) {
+ if (i >= startIndex) {
+ ILSMComponent next = ctx.getComponentHolder().get(i);
+ if (next.getType() == LSMComponentType.MEMORY
+ && next.getState() == ComponentState.UNREADABLE_UNWRITABLE) {
+ componentReplacementCtx.getComponentHolder().add(next);
+ componentReplacementCtx.swapIndex(i);
+ }
+ }
+ }
+ if (componentReplacementCtx.getComponentHolder().isEmpty()) {
+ throw new IllegalStateException(
+ "replaceMemoryComponentsWithDiskComponents called with no potential components");
+ }
+ // before we exit, we should keep the replaced component ids
+ // we should also ensure that exact disk component replacement exist
+ if (componentReplacementCtx.proceed(lsmIndex.getDiskComponents())) {
+ // exit old component
+ exitComponents(componentReplacementCtx, LSMOperationType.SEARCH, null, false);
+ // enter new component
+ componentReplacementCtx.prepareToEnter();
+ enterComponents(componentReplacementCtx, LSMOperationType.SEARCH);
+ componentReplacementCtx.replace(ctx);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexDiskComponentBulkLoader.java
----------------------------------------------------------------------
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 000d5cf..7bc0660 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
@@ -62,7 +62,7 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader {
//then after operation should be called from harness as well
//https://issues.apache.org/jira/browse/ASTERIXDB-1764
lsmIndex.getIOOperationCallback().afterOperation(LSMIOOperationType.LOAD, null, component);
- lsmIndex.getLsmHarness().addBulkLoadedComponent(component);
+ lsmIndex.getHarness().addBulkLoadedComponent(component);
}
} finally {
lsmIndex.getIOOperationCallback().afterFinalize(LSMIOOperationType.LOAD, component);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
index a35d00e..1c28ef1 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexReplicationJob.java
@@ -45,7 +45,7 @@ public class LSMIndexReplicationJob extends AbstractReplicationJob implements IL
@Override
public void endReplication() throws HyracksDataException {
if (operationContext != null) {
- ((AbstractLSMIndex) (lsmIndex)).getLsmHarness().endReplication(operationContext);
+ ((AbstractLSMIndex) (lsmIndex)).getHarness().endReplication(operationContext);
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
index 724a909..17c681c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMIndexSearchCursor.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.PriorityQueue;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
@@ -36,9 +37,13 @@ import org.apache.hyracks.storage.common.MultiComparator;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
+ protected static final int SWITCH_COMPONENT_CYCLE = 100;
protected final ILSMIndexOperationContext opCtx;
protected final boolean returnDeletedTuples;
protected PriorityQueueElement outputElement;
+ protected final ArrayTupleBuilder[] switchComponentTupleBuilders;
+ protected final boolean[] switchRequest;
+ protected final PriorityQueueElement[] switchedElements;
protected IIndexCursor[] rangeCursors;
protected PriorityQueueElement[] pqes;
protected PriorityQueue<PriorityQueueElement> outputPriorityQueue;
@@ -47,6 +52,8 @@ public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
protected boolean needPushElementIntoQueue;
protected boolean includeMutableComponent;
protected ILSMHarness lsmHarness;
+ protected boolean switchPossible = true;
+ protected int hasNextCallCount = 0;
protected List<ILSMComponent> operationalComponents;
@@ -55,6 +62,9 @@ public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
this.returnDeletedTuples = returnDeletedTuples;
outputElement = null;
needPushElementIntoQueue = false;
+ switchComponentTupleBuilders = new ArrayTupleBuilder[opCtx.getIndex().getNumberOfAllMemoryComponents()];
+ switchRequest = new boolean[switchComponentTupleBuilders.length];
+ switchedElements = new PriorityQueueElement[switchComponentTupleBuilders.length];
}
public ILSMIndexOperationContext getOpCtx() {
@@ -98,9 +108,13 @@ public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
@Override
public void reset() throws HyracksDataException {
+ hasNextCallCount = 0;
+ switchPossible = true;
outputElement = null;
needPushElementIntoQueue = false;
-
+ for (int i = 0; i < switchRequest.length; i++) {
+ switchRequest[i] = false;
+ }
try {
if (outputPriorityQueue != null) {
outputPriorityQueue.clear();
@@ -121,6 +135,7 @@ public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
@Override
public boolean hasNext() throws HyracksDataException {
+ hasNextCallCount++;
checkPriorityQueue();
return !outputPriorityQueue.isEmpty();
}
@@ -196,7 +211,7 @@ public abstract class LSMIndexSearchCursor implements ITreeIndexCursor {
}
protected void checkPriorityQueue() throws HyracksDataException {
- while (!outputPriorityQueue.isEmpty() || (needPushElementIntoQueue == true)) {
+ while (!outputPriorityQueue.isEmpty() || needPushElementIntoQueue) {
if (!outputPriorityQueue.isEmpty()) {
PriorityQueueElement checkElement = outputPriorityQueue.peek();
// If there is no previous tuple or the previous tuple can be ignored
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
index 9e9f78a..030e4fd 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndex.java
@@ -237,7 +237,7 @@ public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex
deletedKeysBTreeAccessors,
((LSMInvertedIndexMemoryComponent) memoryComponents.get(currentMutableComponentId.get()))
.getBuddyIndex().getLeafFrameFactory(),
- ictx, includeMutableComponent, getLsmHarness(), operationalComponents);
+ ictx, includeMutableComponent, getHarness(), operationalComponents);
} else {
LSMInvertedIndexMemoryComponent mutableComponent =
(LSMInvertedIndexMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
@@ -246,7 +246,7 @@ public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex
initState = new LSMInvertedIndexRangeSearchCursorInitialState(tokensAndKeysCmp, keyCmp, keysOnlyTuple,
((LSMInvertedIndexMemoryComponent) memoryComponents.get(currentMutableComponentId.get()))
.getBuddyIndex().getLeafFrameFactory(),
- includeMutableComponent, getLsmHarness(), indexAccessors, deletedKeysBTreeAccessors, pred,
+ includeMutableComponent, getHarness(), indexAccessors, deletedKeysBTreeAccessors, pred,
operationalComponents);
}
return initState;
@@ -427,15 +427,15 @@ public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex
@Override
public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) throws HyracksDataException {
- return new LSMInvertedIndexAccessor(getLsmHarness(),
+ return new LSMInvertedIndexAccessor(getHarness(),
createOpContext(iap.getModificationCallback(), iap.getSearchOperationCallback()));
}
@Override
protected LSMInvertedIndexOpContext createOpContext(IModificationOperationCallback modificationCallback,
ISearchOperationCallback searchCallback) throws HyracksDataException {
- return new LSMInvertedIndexOpContext(memoryComponents, modificationCallback, searchCallback,
- invertedIndexFieldsForNonBulkLoadOps, filterFieldsForNonBulkLoadOps, getFilterCmpFactories());
+ return new LSMInvertedIndexOpContext(this, memoryComponents, modificationCallback, searchCallback,
+ invertedIndexFieldsForNonBulkLoadOps, filterFieldsForNonBulkLoadOps, getFilterCmpFactories(), tracer);
}
@Override
@@ -481,7 +481,7 @@ public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex
protected ILSMIOOperation createFlushOperation(AbstractLSMIndexOperationContext opCtx,
LSMComponentFileReferences componentFileRefs, ILSMIOOperationCallback callback)
throws HyracksDataException {
- return new LSMInvertedIndexFlushOperation(new LSMInvertedIndexAccessor(getLsmHarness(), opCtx),
+ return new LSMInvertedIndexFlushOperation(new LSMInvertedIndexAccessor(getHarness(), opCtx),
componentFileRefs.getInsertIndexFileReference(), componentFileRefs.getDeleteIndexFileReference(),
componentFileRefs.getBloomFilterFileReference(), callback, fileManager.getBaseDir().getAbsolutePath());
}
@@ -489,7 +489,7 @@ public class LSMInvertedIndex extends AbstractLSMIndex implements IInvertedIndex
@Override
protected ILSMIOOperation createMergeOperation(AbstractLSMIndexOperationContext opCtx,
LSMComponentFileReferences mergeFileRefs, ILSMIOOperationCallback callback) throws HyracksDataException {
- ILSMIndexAccessor accessor = new LSMInvertedIndexAccessor(getLsmHarness(), opCtx);
+ ILSMIndexAccessor accessor = new LSMInvertedIndexAccessor(getHarness(), opCtx);
IIndexCursor cursor = new LSMInvertedIndexRangeSearchCursor(opCtx);
return new LSMInvertedIndexMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(),
mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(), callback,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
index 12341da..55da252 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexOpContext.java
@@ -25,12 +25,14 @@ import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.invertedindex.api.IInvertedIndexAccessor;
import org.apache.hyracks.storage.common.IIndexAccessor;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
+import org.apache.hyracks.util.trace.ITracer;
public class LSMInvertedIndexOpContext extends AbstractLSMIndexOperationContext {
@@ -45,11 +47,12 @@ public class LSMInvertedIndexOpContext extends AbstractLSMIndexOperationContext
private IInvertedIndexAccessor currentMutableInvIndexAccessors;
private IIndexAccessor currentDeletedKeysBTreeAccessors;
- public LSMInvertedIndexOpContext(List<ILSMMemoryComponent> mutableComponents,
+ public LSMInvertedIndexOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
- int[] invertedIndexFields, int[] filterFields, IBinaryComparatorFactory[] filterComparatorFactories)
- throws HyracksDataException {
- super(invertedIndexFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback);
+ int[] invertedIndexFields, int[] filterFields, IBinaryComparatorFactory[] filterComparatorFactories,
+ ITracer tracer) throws HyracksDataException {
+ super(index, invertedIndexFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback,
+ tracer);
mutableInvIndexAccessors = new IInvertedIndexAccessor[mutableComponents.size()];
deletedKeysBTreeAccessors = new IIndexAccessor[mutableComponents.size()];
for (int i = 0; i < mutableComponents.size(); i++) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
index fda3775..623fc74 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/ondisk/OnDiskInvertedIndexOpContext.java
@@ -81,4 +81,5 @@ public class OnDiskInvertedIndexOpContext implements IIndexOperationContext {
public MultiComparator getPrefixSearchCmp() {
return prefixSearchCmp;
}
+
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
index 6918b19..9960590 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/dataflow/ExternalRTreeLocalResource.java
@@ -72,7 +72,7 @@ public class ExternalRTreeLocalResource extends LSMRTreeLocalResource {
bloomFilterFalsePositiveRate, mergePolicyFactory.createMergePolicy(mergePolicyProperties, ncServiceCtx),
opTrackerProvider.getOperationTracker(ncServiceCtx), ioSchedulerProvider.getIoScheduler(ncServiceCtx),
ioOpCallbackFactory, linearizeCmpFactory, buddyBTreeFields, durable, isPointMBR,
- metadataPageManagerFactory);
+ metadataPageManagerFactory, ncServiceCtx.getTracer());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
index 152b0a8..aac8905 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/AbstractLSMRTree.java
@@ -127,9 +127,9 @@ public abstract class AbstractLSMRTree extends AbstractLSMIndex implements ITree
ILinearizeComparatorFactory linearizer, int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray,
double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker,
ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable,
- boolean isPointMBR) {
+ boolean isPointMBR, ITracer tracer) {
super(ioManager, diskBufferCache, fileManager, bloomFilterFalsePositiveRate, mergePolicy, opTracker,
- ioScheduler, ioOpCallbackFactory, componentFactory, componentFactory, durable);
+ ioScheduler, ioOpCallbackFactory, componentFactory, componentFactory, durable, tracer);
this.rtreeInteriorFrameFactory = rtreeInteriorFrameFactory;
this.rtreeLeafFrameFactory = rtreeLeafFrameFactory;
this.btreeInteriorFrameFactory = btreeInteriorFrameFactory;
@@ -231,9 +231,9 @@ public abstract class AbstractLSMRTree extends AbstractLSMIndex implements ITree
@Override
protected LSMRTreeOpContext createOpContext(IModificationOperationCallback modCallback,
ISearchOperationCallback searchCallback) {
- return new LSMRTreeOpContext(memoryComponents, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
- btreeLeafFrameFactory, modCallback, searchCallback, getTreeFields(), getFilterFields(), getLsmHarness(),
- comparatorFields, linearizerArray, getFilterCmpFactories());
+ return new LSMRTreeOpContext(this, memoryComponents, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
+ btreeLeafFrameFactory, modCallback, searchCallback, getTreeFields(), getFilterFields(), getHarness(),
+ comparatorFields, linearizerArray, getFilterCmpFactories(), tracer);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/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 ca24b13..6f61935 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
@@ -61,6 +61,7 @@ import org.apache.hyracks.storage.common.IIndexCursor;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.ISearchPredicate;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
/**
* This is an lsm r-tree that does not have memory component and is modified
@@ -89,18 +90,18 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray, ILSMMergePolicy mergePolicy,
ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
ILSMIOOperationCallbackFactory ioOpCallbackFactory, int[] buddyBTreeFields, boolean durable,
- boolean isPointMBR) {
+ boolean isPointMBR, ITracer tracer) {
super(ioManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory, btreeInteriorFrameFactory,
btreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory, bloomFilterFalsePositiveRate,
rtreeCmpFactories, btreeCmpFactories, linearizer, comparatorFields, linearizerArray, mergePolicy,
- opTracker, ioScheduler, ioOpCallbackFactory, buddyBTreeFields, durable, isPointMBR);
+ opTracker, ioScheduler, ioOpCallbackFactory, buddyBTreeFields, durable, isPointMBR, tracer);
this.secondDiskComponents = new LinkedList<>();
this.fieldCount = fieldCount;
}
@Override
- public ExternalIndexHarness getLsmHarness() {
- return (ExternalIndexHarness) super.getLsmHarness();
+ public ExternalIndexHarness getHarness() {
+ return (ExternalIndexHarness) super.getHarness();
}
// The subsume merged components is overridden to account for:
@@ -207,7 +208,7 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
diskComponents.add(component);
secondDiskComponents.add(component);
}
- getLsmHarness().indexFirstTimeActivated();
+ getHarness().indexFirstTimeActivated();
} else {
// This index has been opened before or is brand new with no components
// components. It should also maintain the version pointer
@@ -349,7 +350,7 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
if (!isActive) {
throw new HyracksDataException("Failed to clear the index since it is not activated.");
}
- getLsmHarness().indexClear();
+ getHarness().indexClear();
for (ILSMDiskComponent c : diskComponents) {
c.deactivateAndDestroy();
@@ -532,7 +533,7 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
component.markAsValid(durable);
component.deactivate();
} else {
- getLsmHarness().addBulkLoadedComponent(component);
+ getHarness().addBulkLoadedComponent(component);
}
}
}
@@ -587,7 +588,7 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
LSMComponentFileReferences relMergeFileRefs =
getMergeFileReferences((ILSMDiskComponent) mergingComponents.get(0),
(ILSMDiskComponent) mergingComponents.get(mergingComponents.size() - 1));
- ILSMIndexAccessor accessor = new LSMRTreeAccessor(getLsmHarness(), rctx, buddyBTreeFields);
+ ILSMIndexAccessor accessor = new LSMRTreeAccessor(getHarness(), rctx, buddyBTreeFields);
// create the merge operation.
LSMRTreeMergeOperation mergeOp = new LSMRTreeMergeOperation(accessor, cursor,
relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getDeleteIndexFileReference(),
@@ -598,22 +599,22 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
@Override
public ILSMIndexAccessor createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
throws HyracksDataException {
- return new LSMRTreeAccessor(getLsmHarness(), createOpContext(searchCallback, targetIndexVersion),
+ return new LSMRTreeAccessor(getHarness(), createOpContext(searchCallback, targetIndexVersion),
buddyBTreeFields);
}
// This method creates the appropriate opContext for the targeted version
public ExternalRTreeOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
- return new ExternalRTreeOpContext(rtreeCmpFactories, btreeCmpFactories, searchCallback, targetVersion,
- getLsmHarness(), comparatorFields, linearizerArray, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
- btreeLeafFrameFactory);
+ return new ExternalRTreeOpContext(this, rtreeCmpFactories, btreeCmpFactories, searchCallback, targetVersion,
+ getHarness(), comparatorFields, linearizerArray, rtreeLeafFrameFactory, rtreeInteriorFrameFactory,
+ btreeLeafFrameFactory, tracer);
}
// The accessor for disk only indexes don't use modification callback and
// always carry the target index version with them
@Override
public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
- return new LSMRTreeAccessor(getLsmHarness(), createOpContext(iap.getSearchOperationCallback(), version),
+ return new LSMRTreeAccessor(getHarness(), createOpContext(iap.getSearchOperationCallback(), version),
buddyBTreeFields);
}
@@ -646,7 +647,7 @@ public class ExternalRTree extends LSMRTree implements ITwoPCIndex {
componentFileRefrences.getDeleteIndexFileReference(),
componentFileRefrences.getBloomFilterFileReference(), false);
}
- getLsmHarness().addTransactionComponents(component);
+ getHarness().addTransactionComponents(component);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
index 73be537..09237a3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/ExternalRTreeOpContext.java
@@ -21,9 +21,11 @@ package org.apache.hyracks.storage.am.lsm.rtree.impls;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
public class ExternalRTreeOpContext extends AbstractLSMIndexOperationContext {
private MultiComparator bTreeCmp;
@@ -31,12 +33,13 @@ public class ExternalRTreeOpContext extends AbstractLSMIndexOperationContext {
private final int targetIndexVersion;
private LSMRTreeCursorInitialState initialState;
- public ExternalRTreeOpContext(IBinaryComparatorFactory[] rtreeCmpFactories,
+ public ExternalRTreeOpContext(ILSMIndex index, IBinaryComparatorFactory[] rtreeCmpFactories,
IBinaryComparatorFactory[] btreeCmpFactories, ISearchOperationCallback searchCallback,
int targetIndexVersion, ILSMHarness lsmHarness, int[] comparatorFields,
IBinaryComparatorFactory[] linearizerArray, ITreeIndexFrameFactory rtreeLeafFrameFactory,
- ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory) {
- super(null, null, null, searchCallback, null);
+ ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory,
+ ITracer tracer) {
+ super(index, null, null, null, searchCallback, null, tracer);
this.targetIndexVersion = targetIndexVersion;
this.bTreeCmp = MultiComparator.create(btreeCmpFactories);
this.rTreeCmp = MultiComparator.create(rtreeCmpFactories);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/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 73cb206..9712b7c 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
@@ -63,6 +63,7 @@ import org.apache.hyracks.storage.common.IIndexAccessParameters;
import org.apache.hyracks.storage.common.IIndexCursor;
import org.apache.hyracks.storage.common.ISearchPredicate;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
public class LSMRTree extends AbstractLSMRTree {
protected final int[] buddyBTreeFields;
@@ -99,11 +100,11 @@ public class LSMRTree extends AbstractLSMRTree {
int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray, ILSMMergePolicy mergePolicy,
ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
ILSMIOOperationCallbackFactory ioOpCallbackFactory, int[] buddyBTreeFields, boolean durable,
- boolean isPointMBR) {
+ boolean isPointMBR, ITracer tracer) {
super(ioManager, rtreeInteriorFrameFactory, rtreeLeafFrameFactory, btreeInteriorFrameFactory,
btreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory, rtreeCmpFactories,
btreeCmpFactories, linearizer, comparatorFields, linearizerArray, bloomFilterFalsePositiveRate,
- mergePolicy, opTracker, ioScheduler, ioOpCallbackFactory, durable, isPointMBR);
+ mergePolicy, opTracker, ioScheduler, ioOpCallbackFactory, durable, isPointMBR, tracer);
this.buddyBTreeFields = buddyBTreeFields;
}
@@ -276,7 +277,7 @@ public class LSMRTree extends AbstractLSMRTree {
@Override
public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
- return new LSMRTreeAccessor(getLsmHarness(),
+ return new LSMRTreeAccessor(getHarness(),
createOpContext(iap.getModificationCallback(), iap.getSearchOperationCallback()), buddyBTreeFields);
}
@@ -321,7 +322,7 @@ public class LSMRTree extends AbstractLSMRTree {
protected ILSMIOOperation createFlushOperation(AbstractLSMIndexOperationContext opCtx,
LSMComponentFileReferences componentFileRefs, ILSMIOOperationCallback callback)
throws HyracksDataException {
- LSMRTreeAccessor accessor = new LSMRTreeAccessor(getLsmHarness(), opCtx, buddyBTreeFields);
+ LSMRTreeAccessor accessor = new LSMRTreeAccessor(getHarness(), opCtx, buddyBTreeFields);
return new LSMRTreeFlushOperation(accessor, componentFileRefs.getInsertIndexFileReference(),
componentFileRefs.getDeleteIndexFileReference(), componentFileRefs.getBloomFilterFileReference(),
callback, fileManager.getBaseDir().getAbsolutePath());
@@ -331,7 +332,7 @@ public class LSMRTree extends AbstractLSMRTree {
protected ILSMIOOperation createMergeOperation(AbstractLSMIndexOperationContext opCtx,
LSMComponentFileReferences mergeFileRefs, ILSMIOOperationCallback callback) throws HyracksDataException {
ITreeIndexCursor cursor = new LSMRTreeSortedCursor(opCtx, linearizer, buddyBTreeFields);
- ILSMIndexAccessor accessor = new LSMRTreeAccessor(getLsmHarness(), opCtx, buddyBTreeFields);
+ ILSMIndexAccessor accessor = new LSMRTreeAccessor(getHarness(), opCtx, buddyBTreeFields);
return new LSMRTreeMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(),
mergeFileRefs.getDeleteIndexFileReference(), mergeFileRefs.getBloomFilterFileReference(), callback,
fileManager.getBaseDir().getAbsolutePath());
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
index f54f8b9..50f1961 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/impls/LSMRTreeOpContext.java
@@ -29,6 +29,7 @@ import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.am.rtree.impls.RTree;
@@ -37,6 +38,7 @@ import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.ISearchPredicate;
import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
public final class LSMRTreeOpContext extends AbstractLSMIndexOperationContext {
@@ -50,12 +52,14 @@ public final class LSMRTreeOpContext extends AbstractLSMIndexOperationContext {
private BTreeOpContext currentBTreeOpContext;
private LSMRTreeCursorInitialState searchInitialState;
- public LSMRTreeOpContext(List<ILSMMemoryComponent> mutableComponents, ITreeIndexFrameFactory rtreeLeafFrameFactory,
- ITreeIndexFrameFactory rtreeInteriorFrameFactory, ITreeIndexFrameFactory btreeLeafFrameFactory,
- IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
- int[] rtreeFields, int[] filterFields, ILSMHarness lsmHarness, int[] comparatorFields,
- IBinaryComparatorFactory[] linearizerArray, IBinaryComparatorFactory[] filterComparatorFactories) {
- super(rtreeFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback);
+ public LSMRTreeOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
+ ITreeIndexFrameFactory rtreeLeafFrameFactory, ITreeIndexFrameFactory rtreeInteriorFrameFactory,
+ ITreeIndexFrameFactory btreeLeafFrameFactory, IModificationOperationCallback modificationCallback,
+ ISearchOperationCallback searchCallback, int[] rtreeFields, int[] filterFields, ILSMHarness lsmHarness,
+ int[] comparatorFields, IBinaryComparatorFactory[] linearizerArray,
+ IBinaryComparatorFactory[] filterComparatorFactories, ITracer tracer) {
+ super(index, rtreeFields, filterFields, filterComparatorFactories, searchCallback, modificationCallback,
+ tracer);
mutableRTreeAccessors = new RTree.RTreeAccessor[mutableComponents.size()];
mutableBTreeAccessors = new BTree.BTreeAccessor[mutableComponents.size()];
rtreeOpContexts = new RTreeOpContext[mutableComponents.size()];
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/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 a20eeb7..41e9b92 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
@@ -214,14 +214,14 @@ public class LSMRTreeWithAntiMatterTuples extends AbstractLSMRTree {
@Override
public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
LSMRTreeOpContext opCtx = createOpContext(iap.getModificationCallback(), iap.getSearchOperationCallback());
- return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
}
@Override
protected ILSMIOOperation createFlushOperation(AbstractLSMIndexOperationContext opCtx,
LSMComponentFileReferences componentFileRefs, ILSMIOOperationCallback callback)
throws HyracksDataException {
- ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
return new LSMRTreeFlushOperation(accessor, componentFileRefs.getInsertIndexFileReference(), null, null,
callback, fileManager.getBaseDir().getAbsolutePath());
}
@@ -235,7 +235,7 @@ public class LSMRTreeWithAntiMatterTuples extends AbstractLSMRTree {
returnDeletedTuples = true;
}
ITreeIndexCursor cursor = new LSMRTreeWithAntiMatterTuplesSearchCursor(opCtx, returnDeletedTuples);
- ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
return new LSMRTreeMergeOperation(accessor, cursor, mergeFileRefs.getInsertIndexFileReference(), null, null,
callback, fileManager.getBaseDir().getAbsolutePath());
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
index c9df11f..451b122 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-rtree/src/main/java/org/apache/hyracks/storage/am/lsm/rtree/utils/LSMRTreeUtils.java
@@ -72,6 +72,7 @@ import org.apache.hyracks.storage.am.rtree.linearize.ZCurveDoubleComparatorFacto
import org.apache.hyracks.storage.am.rtree.linearize.ZCurveIntComparatorFactory;
import org.apache.hyracks.storage.am.rtree.tuples.RTreeTypeAwareTupleWriterFactory;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
public class LSMRTreeUtils {
public static LSMRTree createLSMTree(IIOManager ioManager, List<IVirtualBufferCache> virtualBufferCaches,
@@ -243,7 +244,7 @@ public class LSMRTreeUtils {
ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
ILSMIOOperationCallbackFactory ioOpCallbackFactory, ILinearizeComparatorFactory linearizeCmpFactory,
int[] buddyBTreeFields, boolean durable, boolean isPointMBR,
- IMetadataPageManagerFactory freePageManagerFactory) throws HyracksDataException {
+ IMetadataPageManagerFactory freePageManagerFactory, ITracer tracer) throws HyracksDataException {
int keyFieldCount = rtreeCmpFactories.length;
int valueFieldCount = typeTraits.length - keyFieldCount;
@@ -290,7 +291,7 @@ public class LSMRTreeUtils {
btreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory, bloomFilterFalsePositiveRate,
typeTraits.length, rtreeCmpFactories, btreeCmpFactories, linearizeCmpFactory, comparatorFields,
linearizerArray, mergePolicy, opTracker, ioScheduler, ioOpCallbackFactory, buddyBTreeFields, durable,
- isPointMBR);
+ isPointMBR, tracer);
}
public static ILinearizeComparatorFactory proposeBestLinearizer(ITypeTraits[] typeTraits, int numKeyFields)
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.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/TestLsmBtree.java b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
index 70a29db..8782565 100644
--- a/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
+++ b/hyracks-fullstack/hyracks/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/org/apache/hyracks/storage/am/lsm/btree/impl/TestLsmBtree.java
@@ -60,6 +60,7 @@ public class TestLsmBtree extends LSMBTree {
private final List<ITestOpCallback> searchCallbacks = new ArrayList<>();
private final List<ITestOpCallback> flushCallbacks = new ArrayList<>();
private final List<ITestOpCallback> mergeCallbacks = new ArrayList<>();
+ private final List<ITestOpCallback> allocateComponentCallbacks = new ArrayList<>();
private volatile int numScheduledFlushes;
private volatile int numStartedFlushes;
@@ -171,7 +172,7 @@ public class TestLsmBtree extends LSMBTree {
@Override
public ILSMIndexAccessor createAccessor(AbstractLSMIndexOperationContext opCtx) {
- return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, ctx -> new TestLsmBtreeSearchCursor(ctx, this));
+ return new LSMTreeIndexAccessor(getHarness(), opCtx, ctx -> new TestLsmBtreeSearchCursor(ctx, this));
}
public int getNumScheduledFlushes() {
@@ -257,4 +258,26 @@ public class TestLsmBtree extends LSMBTree {
public Semaphore getSearchSemaphore() {
return searchSemaphore;
}
+
+ public void addAllocateCallback(ITestOpCallback callback) {
+ synchronized (allocateComponentCallbacks) {
+ allocateComponentCallbacks.add(callback);
+ }
+ }
+
+ public void clearAllocateCallbacks() {
+ synchronized (allocateComponentCallbacks) {
+ allocateComponentCallbacks.clear();
+ }
+ }
+
+ @Override
+ public void allocateMemoryComponents() throws HyracksDataException {
+ super.allocateMemoryComponents();
+ synchronized (allocateComponentCallbacks) {
+ for (ITestOpCallback callback : allocateComponentCallbacks) {
+ callback(callback, null);
+ }
+ }
+ }
}
[3/3] asterixdb git commit: [ASTERIXDB-2169][STO][TX] Unblock
modifications during full scan
Posted by am...@apache.org.
[ASTERIXDB-2169][STO][TX] Unblock modifications during full scan
- user model changes: no
- storage format changes: no
- interface changes: yes
- added ILSMHarness.replaceMemoryComponentsWithDiskComponents
details:
- During a long running query aka full scan, two things block
incoming modifications:
1) Memory component gets full, is flushed but can't be recycled
because of the search operation inside the component.
2) Read latches on the memory component not being released and
the memory component search cursor is not advancing.
The two cases are addressed in this change for the LSMBTree but
not yet addressed for other indexes.
The proposed solution for case (1) is to poll memory components
states every n records during the search operation. If a memory
component was found to have been flushed, its cursor is moved
to the corresponding disk component allowing the memory
component to be recycled.
The proposed solution for case (2) is to check memory component
cursor every n records. If the cursor has not advanced and the
component has writers, then the latches over the leaf page are
released, and the cursor re-do the operation entering from the
tree root.
- Added a test case.
- Added performance traces for enter and exit components.
Change-Id: I37ba52f6324ed1c5a78465c3a8cbcd351f1ed5bc
Reviewed-on: https://asterix-gerrit.ics.uci.edu/2166
Sonar-Qube: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Luo Chen <cl...@uci.edu>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/e5a65429
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/e5a65429
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/e5a65429
Branch: refs/heads/master
Commit: e5a65429d94c34b3079a3fe6bad206089c2c193c
Parents: 76ecc4b
Author: Abdullah Alamoudi <ba...@gmail.com>
Authored: Mon Nov 20 17:14:01 2017 -0800
Committer: abdullah alamoudi <ba...@gmail.com>
Committed: Mon Nov 20 21:32:04 2017 -0800
----------------------------------------------------------------------
.../app/bootstrap/TestNodeController.java | 58 +--
.../test/dataflow/ComponentRollbackTest.java | 94 +++--
.../asterix/test/dataflow/LogMarkerTest.java | 7 +-
.../dataflow/MultiPartitionLSMIndexTest.java | 196 ++++++++++
.../SearchCursorComponentSwitchTest.java | 266 +++++++++++++
.../asterix/test/dataflow/TestDataset.java | 3 +-
.../TestLsmBtreeIoOpCallbackFactory.java | 8 +-
.../asterix/test/logging/CheckpointingTest.java | 7 +-
.../asterix/test/storage/DeallocatableTest.java | 4 +-
.../asterix/test/storage/DiskIsFullTest.java | 8 +-
.../IndexDropOperatorNodePushableTest.java | 28 +-
.../AbstractLSMIOOperationCallback.java | 10 +-
...actOneInputOneOutputOneFramePushRuntime.java | 16 +-
.../meta/AlgebricksMetaOperatorDescriptor.java | 49 ++-
.../apache/hyracks/http/server/HttpServer.java | 10 +-
.../dataflow/ExternalBTreeLocalResource.java | 2 +-
.../ExternalBTreeWithBuddyLocalResource.java | 2 +-
.../am/lsm/btree/impls/ExternalBTree.java | 27 +-
.../lsm/btree/impls/ExternalBTreeOpContext.java | 8 +-
.../lsm/btree/impls/ExternalBTreeWithBuddy.java | 26 +-
.../impls/ExternalBTreeWithBuddyOpContext.java | 9 +-
.../storage/am/lsm/btree/impls/LSMBTree.java | 14 +-
.../am/lsm/btree/impls/LSMBTreeOpContext.java | 13 +-
.../btree/impls/LSMBTreeRangeSearchCursor.java | 152 +++++++-
.../am/lsm/btree/utils/LSMBTreeUtil.java | 9 +-
.../am/lsm/common/api/ILSMComponent.java | 5 +
.../storage/am/lsm/common/api/ILSMHarness.java | 13 +
.../storage/am/lsm/common/api/ILSMIndex.java | 5 +
.../common/api/ILSMIndexOperationContext.java | 26 ++
.../lsm/common/impls/AbstractLSMComponent.java | 7 +-
.../common/impls/AbstractLSMDiskComponent.java | 9 +-
.../am/lsm/common/impls/AbstractLSMIndex.java | 39 +-
.../impls/AbstractLSMIndexOperationContext.java | 42 ++-
.../impls/AbstractLSMMemoryComponent.java | 6 +
.../impls/ComponentReplacementContext.java | 215 +++++++++++
.../am/lsm/common/impls/LSMComponentId.java | 8 +-
.../storage/am/lsm/common/impls/LSMHarness.java | 373 +++++++++++--------
.../impls/LSMIndexDiskComponentBulkLoader.java | 2 +-
.../common/impls/LSMIndexReplicationJob.java | 2 +-
.../lsm/common/impls/LSMIndexSearchCursor.java | 19 +-
.../invertedindex/impls/LSMInvertedIndex.java | 14 +-
.../impls/LSMInvertedIndexOpContext.java | 11 +-
.../ondisk/OnDiskInvertedIndexOpContext.java | 1 +
.../dataflow/ExternalRTreeLocalResource.java | 2 +-
.../am/lsm/rtree/impls/AbstractLSMRTree.java | 10 +-
.../am/lsm/rtree/impls/ExternalRTree.java | 29 +-
.../lsm/rtree/impls/ExternalRTreeOpContext.java | 9 +-
.../storage/am/lsm/rtree/impls/LSMRTree.java | 11 +-
.../am/lsm/rtree/impls/LSMRTreeOpContext.java | 16 +-
.../impls/LSMRTreeWithAntiMatterTuples.java | 6 +-
.../am/lsm/rtree/utils/LSMRTreeUtils.java | 5 +-
.../storage/am/lsm/btree/impl/TestLsmBtree.java | 25 +-
52 files changed, 1514 insertions(+), 422 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
index 90b9a1e..505ef8f 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
@@ -76,6 +76,8 @@ import org.apache.hyracks.api.context.IHyracksJobletContext;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.ActivityId;
import org.apache.hyracks.api.dataflow.OperatorDescriptorId;
+import org.apache.hyracks.api.dataflow.TaskAttemptId;
+import org.apache.hyracks.api.dataflow.TaskId;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
@@ -121,11 +123,10 @@ public class TestNodeController {
public static final int DEFAULT_HYRACKS_CC_CLIENT_PORT = 1098;
public static final int DEFAULT_HYRACKS_CC_CLUSTER_PORT = 1099;
public static final int KB32 = 32768;
- public static final int PARTITION = 0;
public static final double BLOOM_FILTER_FALSE_POSITIVE_RATE = 0.01;
public static final TransactionSubsystemProvider TXN_SUBSYSTEM_PROVIDER = TransactionSubsystemProvider.INSTANCE;
// Mutables
- private long jobCounter = 0L;
+ private long jobCounter = 100L;
private final String testConfigFileName;
private final boolean runHDFS;
@@ -155,7 +156,11 @@ public class TestNodeController {
}
public TxnId getTxnJobId(IHyracksTaskContext ctx) {
- return new TxnId(ctx.getJobletContext().getJobId().getId());
+ return getTxnJobId(ctx.getJobletContext().getJobId());
+ }
+
+ public TxnId getTxnJobId(JobId jobId) {
+ return new TxnId(jobId.getId());
}
public Pair<LSMInsertDeleteOperatorNodePushable, CommitRuntime> getInsertPipeline(IHyracksTaskContext ctx,
@@ -177,12 +182,14 @@ public class TestNodeController {
IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
- LSMInsertDeleteOperatorNodePushable insertOp = new LSMInsertDeleteOperatorNodePushable(ctx, PARTITION,
- primaryIndexInfo.primaryIndexInsertFieldsPermutations,
- recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0), op,
- true, indexHelperFactory, modOpCallbackFactory, null);
- CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(),
- primaryIndexInfo.primaryKeyIndexes, false, true, PARTITION, true);
+ LSMInsertDeleteOperatorNodePushable insertOp =
+ new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
+ primaryIndexInfo.primaryIndexInsertFieldsPermutations,
+ recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
+ op, true, indexHelperFactory, modOpCallbackFactory, null);
+ CommitRuntime commitOp =
+ new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), primaryIndexInfo.primaryKeyIndexes,
+ false, true, ctx.getTaskAttemptId().getTaskId().getPartition(), true);
insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc);
commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
return Pair.of(insertOp, commitOp);
@@ -203,7 +210,8 @@ public class TestNodeController {
null, null, true, true, indexDataflowHelperFactory, false, false, null,
NoOpOperationCallbackFactory.INSTANCE, filterFields, filterFields, false);
BTreeSearchOperatorNodePushable searchOp =
- searchOpDesc.createPushRuntime(ctx, primaryIndexInfo.getSearchRecordDescriptorProvider(), PARTITION, 1);
+ searchOpDesc.createPushRuntime(ctx, primaryIndexInfo.getSearchRecordDescriptorProvider(),
+ ctx.getTaskAttemptId().getTaskId().getPartition(), 1);
emptyTupleOp.setOutputFrameWriter(0, searchOp,
primaryIndexInfo.getSearchRecordDescriptorProvider().getInputRecordDescriptor(null, 0));
searchOp.setOutputFrameWriter(0, countOp, primaryIndexInfo.rDesc);
@@ -236,7 +244,7 @@ public class TestNodeController {
public PrimaryIndexInfo createPrimaryIndex(Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType,
ARecordType metaType, int[] filterFields, IStorageComponentProvider storageComponentProvider,
- int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators)
+ int[] primaryKeyIndexes, List<Integer> primaryKeyIndicators, int partition)
throws AlgebricksException, HyracksDataException, RemoteException, ACIDException {
MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
org.apache.hyracks.algebricks.common.utils.Pair<ILSMMergePolicyFactory, Map<String, String>> mergePolicy =
@@ -254,8 +262,8 @@ public class TestNodeController {
IndexBuilderFactory indexBuilderFactory =
new IndexBuilderFactory(storageComponentProvider.getStorageManager(),
primaryIndexInfo.getFileSplitProvider(), resourceFactory, !dataset.isTemp());
- IHyracksTaskContext ctx = createTestContext(false);
- IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, 0);
+ IHyracksTaskContext ctx = createTestContext(newJobId(), partition, false);
+ IIndexBuilder indexBuilder = indexBuilderFactory.create(ctx, partition);
indexBuilder.build();
} finally {
mdProvider.getLocks().unlock();
@@ -292,12 +300,12 @@ public class TestNodeController {
return primaryIndexTypeTraits;
}
- public IHyracksTaskContext createTestContext(boolean withMessaging) throws HyracksDataException {
+ public IHyracksTaskContext createTestContext(JobId jobId, int partition, boolean withMessaging)
+ throws HyracksDataException {
IHyracksTaskContext ctx = TestUtils.create(KB32);
if (withMessaging) {
TaskUtil.put(HyracksConstants.KEY_MESSAGE, new VSizeFrame(ctx), ctx);
}
- JobId jobId = newJobId();
IHyracksJobletContext jobletCtx = Mockito.mock(IHyracksJobletContext.class);
JobEventListenerFactory factory = new JobEventListenerFactory(new TxnId(jobId.getId()), true);
Mockito.when(jobletCtx.getJobletEventListenerFactory()).thenReturn(factory);
@@ -306,6 +314,9 @@ public class TestNodeController {
ctx = Mockito.spy(ctx);
Mockito.when(ctx.getJobletContext()).thenReturn(jobletCtx);
Mockito.when(ctx.getIoManager()).thenReturn(ExecutionTestUtil.integrationUtil.ncs[0].getIoManager());
+ TaskAttemptId taskId =
+ new TaskAttemptId(new TaskId(new ActivityId(new OperatorDescriptorId(0), 0), partition), 0);
+ Mockito.when(ctx.getTaskAttemptId()).thenReturn(taskId);
return ctx;
}
@@ -377,7 +388,7 @@ public class TestNodeController {
(CcApplicationContext) ExecutionTestUtil.integrationUtil.cc.getApplicationContext();
FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset,
index.getIndexName(), nodes);
- fileSplitProvider = new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1));
+ fileSplitProvider = new ConstantFileSplitProvider(splits);
}
public Index getIndex() {
@@ -448,12 +459,13 @@ public class TestNodeController {
IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
IIndexDataflowHelperFactory indexHelperFactory = new IndexDataflowHelperFactory(
storageComponentProvider.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
- LSMPrimaryUpsertOperatorNodePushable insertOp = new LSMPrimaryUpsertOperatorNodePushable(ctx, PARTITION,
- indexHelperFactory, primaryIndexInfo.primaryIndexInsertFieldsPermutations,
- recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
- modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, recordType, -1,
- frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory() : frameOpCallbackFactory,
- MissingWriterFactory.INSTANCE, hasSecondaries);
+ LSMPrimaryUpsertOperatorNodePushable insertOp =
+ new LSMPrimaryUpsertOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
+ indexHelperFactory, primaryIndexInfo.primaryIndexInsertFieldsPermutations,
+ recordDescProvider.getInputRecordDescriptor(new ActivityId(new OperatorDescriptorId(0), 0), 0),
+ modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, recordType, -1,
+ frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory() : frameOpCallbackFactory,
+ MissingWriterFactory.INSTANCE, hasSecondaries);
RecordDescriptor upsertOutRecDesc = getUpsertOutRecDesc(primaryIndexInfo.rDesc, dataset,
filterFields == null ? 0 : filterFields.length, recordType, metaType);
// fix pk fields
@@ -463,7 +475,7 @@ public class TestNodeController {
pkFieldsInCommitOp[i] = diff + i;
}
CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(ctx), dataset.getDatasetId(), pkFieldsInCommitOp,
- false, true, PARTITION, true);
+ false, true, ctx.getTaskAttemptId().getTaskId().getPartition(), true);
insertOp.setOutputFrameWriter(0, commitOp, upsertOutRecDesc);
commitOp.setInputRecordDescriptor(0, upsertOutRecDesc);
return Pair.of(insertOp, commitOp);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
index b09e000..1ac1aa6 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ComponentRollbackTest.java
@@ -58,6 +58,7 @@ import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.test.CountAnswer;
import org.apache.hyracks.api.test.FrameWriterTestUtils;
import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation;
@@ -133,20 +134,21 @@ public class ComponentRollbackTest {
public void createIndex() throws Exception {
List<List<String>> partitioningKeys = new ArrayList<>();
partitioningKeys.add(Collections.singletonList("key"));
+ int partition = 0;
dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
partitioningKeys, null, null, null, false, null, false),
null, DatasetType.INTERNAL, DATASET_ID, 0);
PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
- storageManager, KEY_INDEXES, KEY_INDICATORS_LIST);
+ storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, partition);
IndexDataflowHelperFactory iHelperFactory =
new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
- ctx = nc.createTestContext(false);
- indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
+ JobId jobId = nc.newJobId();
+ ctx = nc.createTestContext(jobId, partition, false);
+ indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), partition);
indexDataflowHelper.open();
lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance();
indexDataflowHelper.close();
- nc.newJobId();
txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
@@ -158,7 +160,7 @@ public class ComponentRollbackTest {
indexDataflowHelper.destroy();
}
- private void allowAllOps(TestLsmBtree lsmBtree) {
+ static void allowAllOps(TestLsmBtree lsmBtree) {
lsmBtree.addModifyCallback(sem -> sem.release());
lsmBtree.addFlushCallback(sem -> sem.release());
lsmBtree.addSearchCallback(sem -> sem.release());
@@ -197,19 +199,17 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
// rollback a memory component
lsmAccessor.deleteComponents(memoryComponentsPredicate);
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
// rollback the last disk component
lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata());
DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
lsmAccessor.deleteComponents(pred);
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
} catch (Throwable e) {
e.printStackTrace();
Assert.fail(e.getMessage());
@@ -247,12 +247,11 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
// rollback a memory component
lsmAccessor.deleteComponents(memoryComponentsPredicate);
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
// insert again
nc.newJobId();
@@ -270,14 +269,13 @@ public class ComponentRollbackTest {
}
insertOp.close();
nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// rollback the last disk component
lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata());
DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
lsmAccessor.deleteComponents(pred);
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
} catch (Throwable e) {
e.printStackTrace();
Assert.fail(e.getMessage());
@@ -317,12 +315,11 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
// wait till firstSearcher enter the components
firstSearcher.waitUntilEntered();
// now that we enetered, we will rollback
ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
// rollback a memory component
lsmAccessor.deleteComponents(
c -> (c instanceof ILSMMemoryComponent && ((ILSMMemoryComponent) c).isModified()));
@@ -331,24 +328,23 @@ public class ComponentRollbackTest {
lsmBtree.allowSearch(1);
Assert.assertTrue(firstSearcher.result());
// search now and ensure
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
// rollback the last disk component
// re-block searches
lsmBtree.clearSearchCallbacks();
- Searcher secondSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree,
+ Searcher secondSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree,
TOTAL_NUM_OF_RECORDS - RECORDS_PER_COMPONENT);
// wait till firstSearcher enter the components
secondSearcher.waitUntilEntered();
lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
long lsn = AbstractLSMIOOperationCallback.getTreeIndexLSN(diskComponents.get(0).getMetadata());
DiskComponentLsnPredicate pred = new DiskComponentLsnPredicate(lsn);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
lsmAccessor.deleteComponents(pred);
// now that the rollback has completed, we will unblock the search
lsmBtree.addSearchCallback(sem -> sem.release());
lsmBtree.allowSearch(1);
Assert.assertTrue(secondSearcher.result());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS - (2 * RECORDS_PER_COMPONENT));
} catch (Throwable e) {
e.printStackTrace();
Assert.fail(e.getMessage());
@@ -386,7 +382,7 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// disable flushes
lsmBtree.clearFlushCallbacks();
Flusher firstFlusher = new Flusher(lsmBtree);
@@ -395,7 +391,7 @@ public class ComponentRollbackTest {
// now that we enetered, we will rollback. This will not proceed since it is waiting for the flush to complete
Rollerback rollerback = new Rollerback(lsmBtree, memoryComponentsPredicate);
// now that the rollback has completed, we will search
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
//unblock the flush
lsmBtree.allowFlush(1);
// ensure rollback completed
@@ -403,7 +399,7 @@ public class ComponentRollbackTest {
// ensure current mem component is not modified
Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
// search now and ensure
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
} catch (Throwable e) {
e.printStackTrace();
Assert.fail(e.getMessage());
@@ -442,7 +438,7 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// Now, we will start a full merge
Merger merger = new Merger(lsmBtree);
ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -459,7 +455,7 @@ public class ComponentRollbackTest {
Rollerback rollerback = new Rollerback(lsmBtree, new DiskComponentLsnPredicate(lsn));
// rollback is now waiting for the merge to complete
// we will search
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
//unblock the merge
lsmBtree.allowMerge(1);
// ensure rollback completes
@@ -467,7 +463,7 @@ public class ComponentRollbackTest {
// ensure current mem component is not modified
Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
// search now and ensure that we rolled back the merged component
- searchAndAssertCount(nc, ctx, dataset, storageManager,
+ searchAndAssertCount(nc, 0, dataset, storageManager,
TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT));
} catch (Throwable e) {
e.printStackTrace();
@@ -506,7 +502,7 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// disable flushes
// disable searches
lsmBtree.clearFlushCallbacks();
@@ -514,7 +510,7 @@ public class ComponentRollbackTest {
Flusher firstFlusher = new Flusher(lsmBtree);
dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
firstFlusher.waitUntilCount(1);
- Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
// wait till firstSearcher enter the components
firstSearcher.waitUntilEntered();
// now that we enetered, we will rollback rollback a memory component
@@ -528,7 +524,7 @@ public class ComponentRollbackTest {
rollerback.complete();
Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
// search now and ensure the rollback was no op since it waits for ongoing flushes
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
} catch (Throwable e) {
e.printStackTrace();
Assert.fail(e.getMessage());
@@ -566,7 +562,7 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// disable flushes
// disable searches
lsmBtree.clearFlushCallbacks();
@@ -574,7 +570,7 @@ public class ComponentRollbackTest {
dsLifecycleMgr.flushDataset(dataset.getDatasetId(), true);
firstFlusher.waitUntilCount(1);
lsmBtree.clearSearchCallbacks();
- Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
// wait till firstSearcher enter the components
firstSearcher.waitUntilEntered();
// now that we enetered, we will rollback
@@ -589,7 +585,7 @@ public class ComponentRollbackTest {
rollerback.complete();
Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
// search now and ensure
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
} catch (Throwable e) {
e.printStackTrace();
Assert.fail(e.getMessage());
@@ -628,7 +624,7 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// Now, we will start a merge
Merger merger = new Merger(lsmBtree);
ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -643,7 +639,7 @@ public class ComponentRollbackTest {
merger.waitUntilCount(1);
// we will block search
lsmBtree.clearSearchCallbacks();
- Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
// wait till firstSearcher enter the components
firstSearcher.waitUntilEntered();
// now that we enetered, we will rollback
@@ -657,7 +653,7 @@ public class ComponentRollbackTest {
Assert.assertTrue(firstSearcher.result());
rollerback.complete();
// now that the rollback has completed, we will search
- searchAndAssertCount(nc, ctx, dataset, storageManager,
+ searchAndAssertCount(nc, 0, dataset, storageManager,
TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT));
// ensure current mem component is not modified
Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
@@ -699,7 +695,7 @@ public class ComponentRollbackTest {
List<ILSMDiskComponent> diskComponents = lsmBtree.getDiskComponents();
Assert.assertEquals(9, diskComponents.size());
Assert.assertTrue(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
// Now, we will start a merge
Merger merger = new Merger(lsmBtree);
ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
@@ -714,7 +710,7 @@ public class ComponentRollbackTest {
merger.waitUntilCount(1);
// we will block search
lsmBtree.clearSearchCallbacks();
- Searcher firstSearcher = new Searcher(nc, ctx, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ Searcher firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
// wait till firstSearcher enter the components
firstSearcher.waitUntilEntered();
// now that we enetered, we will rollback
@@ -724,11 +720,11 @@ public class ComponentRollbackTest {
lsmBtree.allowSearch(1);
Assert.assertTrue(firstSearcher.result());
// even though rollback has been called, it is still waiting for the merge to complete
- searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
//unblock the merge
lsmBtree.allowMerge(1);
rollerBack.complete();
- searchAndAssertCount(nc, ctx, dataset, storageManager,
+ searchAndAssertCount(nc, 0, dataset, storageManager,
TOTAL_NUM_OF_RECORDS - ((numMergedComponents + 1/*memory component*/) * RECORDS_PER_COMPONENT));
// ensure current mem component is not modified
Assert.assertFalse(memComponents.get(lsmBtree.getCurrentMemoryComponentIndex()).isModified());
@@ -748,7 +744,6 @@ public class ComponentRollbackTest {
@Override
public void run() {
ILSMIndexAccessor lsmAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- dsLifecycleMgr.getComponentIdGenerator(DATASET_ID).refresh();
try {
lsmAccessor.deleteComponents(predicate);
} catch (HyracksDataException e) {
@@ -768,13 +763,13 @@ public class ComponentRollbackTest {
}
}
- private class Searcher {
+ static class Searcher {
private final ExecutorService executor = Executors.newSingleThreadExecutor();
private Future<Boolean> task;
private volatile boolean entered = false;
- public Searcher(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset,
- StorageComponentProvider storageManager, TestLsmBtree lsmBtree, int numOfRecords) {
+ public Searcher(TestNodeController nc, int partition, Dataset dataset, StorageComponentProvider storageManager,
+ TestLsmBtree lsmBtree, int numOfRecords) {
lsmBtree.addSearchCallback(sem -> {
synchronized (Searcher.this) {
entered = true;
@@ -784,7 +779,7 @@ public class ComponentRollbackTest {
Callable<Boolean> callable = new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
- searchAndAssertCount(nc, ctx, dataset, storageManager, numOfRecords);
+ searchAndAssertCount(nc, partition, dataset, storageManager, numOfRecords);
return true;
}
};
@@ -840,7 +835,7 @@ public class ComponentRollbackTest {
}
}
- private class DiskComponentLsnPredicate implements Predicate<ILSMComponent> {
+ private static class DiskComponentLsnPredicate implements Predicate<ILSMComponent> {
private final long lsn;
public DiskComponentLsnPredicate(long lsn) {
@@ -860,10 +855,11 @@ public class ComponentRollbackTest {
}
}
- private void searchAndAssertCount(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset,
+ static void searchAndAssertCount(TestNodeController nc, int partition, Dataset dataset,
StorageComponentProvider storageManager, int numOfRecords)
throws HyracksDataException, AlgebricksException {
- nc.newJobId();
+ JobId jobId = nc.newJobId();
+ IHyracksTaskContext ctx = nc.createTestContext(jobId, partition, false);
TestTupleCounterFrameWriter countOp = create(nc.getSearchOutputDesc(KEY_TYPES, RECORD_TYPE, META_TYPE),
Collections.emptyList(), Collections.emptyList(), false);
IPushRuntime emptyTupleOp = nc.getFullScanPipeline(countOp, ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
index 1907ed6..aa08c2d 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
@@ -50,6 +50,7 @@ import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.test.CountAnswer;
import org.apache.hyracks.api.test.FrameWriterTestUtils;
import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation;
@@ -117,9 +118,9 @@ public class LogMarkerTest {
null, DatasetType.INTERNAL, DATASET_ID, 0);
try {
PrimaryIndexInfo indexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
- storageManager, KEY_INDEXES, KEY_INDICATORS_LIST);
- IHyracksTaskContext ctx = nc.createTestContext(true);
- nc.newJobId();
+ storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
+ JobId jobId = nc.newJobId();
+ IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
new file mode 100644
index 0000000..ec9c2f6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/MultiPartitionLSMIndexTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.asterix.test.dataflow;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.asterix.app.bootstrap.TestNodeController;
+import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo;
+import org.apache.asterix.app.data.gen.TupleGenerator;
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.nc.NCAppRuntimeContext;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
+import org.apache.asterix.external.util.DataflowUtils;
+import org.apache.asterix.file.StorageComponentProvider;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.entities.Index;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.test.common.TestHelper;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class MultiPartitionLSMIndexTest {
+ private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 };
+ private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" },
+ new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false);
+ private static final GenerationFunction[] RECORD_GEN_FUNCTION =
+ { GenerationFunction.DETERMINISTIC, GenerationFunction.DETERMINISTIC };
+ private static final boolean[] UNIQUE_RECORD_FIELDS = { true, false };
+ private static final ARecordType META_TYPE = null;
+ private static final GenerationFunction[] META_GEN_FUNCTION = null;
+ private static final boolean[] UNIQUE_META_FIELDS = null;
+ private static final int[] KEY_INDEXES = { 0 };
+ private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR };
+ private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR });
+ private static final int TOTAL_NUM_OF_RECORDS = 10000;
+ private static final int RECORDS_PER_COMPONENT = 1000;
+ private static final int DATASET_ID = 101;
+ private static final String DATAVERSE_NAME = "TestDV";
+ private static final String DATASET_NAME = "TestDS";
+ private static final String DATA_TYPE_NAME = "DUMMY";
+ private static final String NODE_GROUP_NAME = "DEFAULT";
+ private static final StorageComponentProvider storageManager = new StorageComponentProvider();
+ private static final int NUM_PARTITIONS = 2;
+ private static TestNodeController nc;
+ private static NCAppRuntimeContext ncAppCtx;
+ private static IDatasetLifecycleManager dsLifecycleMgr;
+ private static Dataset dataset;
+ private static ITransactionContext txnCtx;
+ private static TestLsmBtree[] primarylsmBtrees;
+ private static IHyracksTaskContext[] taskCtxs;
+ private static IIndexDataflowHelper[] indexDataflowHelpers;
+ private static LSMInsertDeleteOperatorNodePushable[] insertOps;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ System.out.println("SetUp: ");
+ TestHelper.deleteExistingInstanceFiles();
+ nc = new TestNodeController(null, false);
+ nc.init();
+ ncAppCtx = nc.getAppRuntimeContext();
+ dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ System.out.println("TearDown");
+ nc.deInit();
+ TestHelper.deleteExistingInstanceFiles();
+ }
+
+ @Before
+ public void createIndex() throws Exception {
+ List<List<String>> partitioningKeys = new ArrayList<>();
+ partitioningKeys.add(Collections.singletonList("key"));
+ dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
+ NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+ partitioningKeys, null, null, null, false, null, false),
+ null, DatasetType.INTERNAL, DATASET_ID, 0);
+ taskCtxs = new IHyracksTaskContext[NUM_PARTITIONS];
+ indexDataflowHelpers = new IIndexDataflowHelper[NUM_PARTITIONS];
+ primarylsmBtrees = new TestLsmBtree[NUM_PARTITIONS];
+ insertOps = new LSMInsertDeleteOperatorNodePushable[NUM_PARTITIONS];
+ JobId jobId = nc.newJobId();
+ txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(jobId),
+ new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
+ for (int i = 0; i < taskCtxs.length; i++) {
+ taskCtxs[i] = nc.createTestContext(jobId, i, false);
+ PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
+ storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, i);
+ IndexDataflowHelperFactory iHelperFactory =
+ new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
+ indexDataflowHelpers[i] = iHelperFactory.create(taskCtxs[i].getJobletContext().getServiceContext(), i);
+ indexDataflowHelpers[i].open();
+ primarylsmBtrees[i] = (TestLsmBtree) indexDataflowHelpers[i].getIndexInstance();
+ indexDataflowHelpers[i].close();
+ insertOps[i] = nc.getInsertPipeline(taskCtxs[i], dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
+ KEY_INDEXES, KEY_INDICATORS_LIST, storageManager).getLeft();
+ }
+ }
+
+ @After
+ public void destroyIndex() throws Exception {
+ for (IIndexDataflowHelper indexDataflowHelper : indexDataflowHelpers) {
+ indexDataflowHelper.destroy();
+ }
+ }
+
+ @Test
+ public void testFlushOneFullOneEmpty() {
+ try {
+ // allow all operations
+ for (int i = 0; i < NUM_PARTITIONS; i++) {
+ ComponentRollbackTest.allowAllOps(primarylsmBtrees[i]);
+ }
+
+ insertOps[0].open();
+ TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+ RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+ VSizeFrame frame = new VSizeFrame(taskCtxs[0]);
+ FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+ int numFlushes = 0;
+ for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
+ // flush every RECORDS_PER_COMPONENT records
+ if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOps[0], true);
+ }
+ dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+ numFlushes++;
+ }
+ ITupleReference tuple = tupleGenerator.next();
+ DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOps[0]);
+ }
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOps[0], true);
+ }
+ insertOps[0].close();
+ dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+ numFlushes++;
+ nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+ // search now and ensure partition 0 has all the records
+ ComponentRollbackTest.searchAndAssertCount(nc, 0, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ // and that partition 1 has no records
+ ComponentRollbackTest.searchAndAssertCount(nc, 1, dataset, storageManager, 0);
+ // and that partition 0 has numFlushes disk components
+ Assert.assertEquals(numFlushes, primarylsmBtrees[0].getDiskComponents().size());
+ // and that partition 1 has no disk components
+ Assert.assertEquals(0, primarylsmBtrees[1].getDiskComponents().size());
+ } catch (Throwable e) {
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
new file mode 100644
index 0000000..77d3795
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/SearchCursorComponentSwitchTest.java
@@ -0,0 +1,266 @@
+/*
+ * 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.asterix.test.dataflow;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.asterix.app.bootstrap.TestNodeController;
+import org.apache.asterix.app.bootstrap.TestNodeController.PrimaryIndexInfo;
+import org.apache.asterix.app.data.gen.TestTupleCounterFrameWriter;
+import org.apache.asterix.app.data.gen.TupleGenerator;
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.app.nc.NCAppRuntimeContext;
+import org.apache.asterix.common.api.IDatasetLifecycleManager;
+import org.apache.asterix.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.common.transactions.TransactionOptions;
+import org.apache.asterix.external.util.DataflowUtils;
+import org.apache.asterix.file.StorageComponentProvider;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.entities.Index;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails;
+import org.apache.asterix.metadata.entities.InternalDatasetDetails.PartitioningStrategy;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.BuiltinType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.test.common.TestHelper;
+import org.apache.asterix.test.dataflow.ComponentRollbackTest.Searcher;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.lsm.btree.impl.TestLsmBtree;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+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.NoMergePolicyFactory;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class SearchCursorComponentSwitchTest {
+ private static final IAType[] KEY_TYPES = { BuiltinType.AINT32 };
+ private static final ARecordType RECORD_TYPE = new ARecordType("TestRecordType", new String[] { "key", "value" },
+ new IAType[] { BuiltinType.AINT32, BuiltinType.AINT64 }, false);
+ private static final GenerationFunction[] RECORD_GEN_FUNCTION =
+ { GenerationFunction.DETERMINISTIC, GenerationFunction.DETERMINISTIC };
+ private static final boolean[] UNIQUE_RECORD_FIELDS = { true, false };
+ private static final ARecordType META_TYPE = null;
+ private static final GenerationFunction[] META_GEN_FUNCTION = null;
+ private static final boolean[] UNIQUE_META_FIELDS = null;
+ private static final int[] KEY_INDEXES = { 0 };
+ private static final int[] KEY_INDICATORS = { Index.RECORD_INDICATOR };
+ private static final List<Integer> KEY_INDICATORS_LIST = Arrays.asList(new Integer[] { Index.RECORD_INDICATOR });
+ private static final int TOTAL_NUM_OF_RECORDS = 2000;
+ private static final int RECORDS_PER_COMPONENT = 1000;
+ private static final int DATASET_ID = 101;
+ private static final String DATAVERSE_NAME = "TestDV";
+ private static final String DATASET_NAME = "TestDS";
+ private static final String DATA_TYPE_NAME = "DUMMY";
+ private static final String NODE_GROUP_NAME = "DEFAULT";
+ private static final StorageComponentProvider storageManager = new StorageComponentProvider();
+ private static TestNodeController nc;
+ private static TestLsmBtree lsmBtree;
+ private static NCAppRuntimeContext ncAppCtx;
+ private static IDatasetLifecycleManager dsLifecycleMgr;
+ private static Dataset dataset;
+ private static IHyracksTaskContext ctx;
+ private static IIndexDataflowHelper indexDataflowHelper;
+ private static ITransactionContext txnCtx;
+ private static LSMInsertDeleteOperatorNodePushable insertOp;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ System.out.println("SetUp: ");
+ TestHelper.deleteExistingInstanceFiles();
+ nc = new TestNodeController(null, false);
+ nc.init();
+ ncAppCtx = nc.getAppRuntimeContext();
+ dsLifecycleMgr = ncAppCtx.getDatasetLifecycleManager();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ System.out.println("TearDown");
+ nc.deInit();
+ TestHelper.deleteExistingInstanceFiles();
+ }
+
+ @Before
+ public void createIndex() throws Exception {
+ List<List<String>> partitioningKeys = new ArrayList<>();
+ partitioningKeys.add(Collections.singletonList("key"));
+ dataset = new TestDataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
+ NoMergePolicyFactory.NAME, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+ partitioningKeys, null, null, null, false, null, false),
+ null, DatasetType.INTERNAL, DATASET_ID, 0);
+ PrimaryIndexInfo primaryIndexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null,
+ storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
+ IndexDataflowHelperFactory iHelperFactory =
+ new IndexDataflowHelperFactory(nc.getStorageManager(), primaryIndexInfo.getFileSplitProvider());
+ JobId jobId = nc.newJobId();
+ ctx = nc.createTestContext(jobId, 0, false);
+ indexDataflowHelper = iHelperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
+ indexDataflowHelper.open();
+ lsmBtree = (TestLsmBtree) indexDataflowHelper.getIndexInstance();
+ indexDataflowHelper.close();
+ txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
+ new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
+ insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, KEY_INDEXES,
+ KEY_INDICATORS_LIST, storageManager).getLeft();
+ }
+
+ @After
+ public void destroyIndex() throws Exception {
+ indexDataflowHelper.destroy();
+ }
+
+ void unblockSearch(TestLsmBtree lsmBtree) {
+ lsmBtree.addSearchCallback(sem -> sem.release());
+ lsmBtree.allowSearch(1);
+ }
+
+ @Test
+ public void testCursorSwitchSucceed() {
+ try {
+ // allow all operations
+ ComponentRollbackTest.allowAllOps(lsmBtree);
+ // except search
+ lsmBtree.clearSearchCallbacks();
+ insertOp.open();
+ TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+ RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+ VSizeFrame frame = new VSizeFrame(ctx);
+ FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+ Searcher firstSearcher = null;
+ for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
+ // flush every RECORDS_PER_COMPONENT records
+ if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOp, true);
+ }
+ dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+ }
+ ITupleReference tuple = tupleGenerator.next();
+ DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
+ }
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOp, true);
+ }
+ insertOp.close();
+ // start the search
+ firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ // wait till firstSearcher enter the components
+ firstSearcher.waitUntilEntered();
+ dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+ nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+ // unblock the search
+ unblockSearch(lsmBtree);
+ // ensure the search got the correct number
+ Assert.assertTrue(firstSearcher.result());
+ // search now and ensure
+ searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ } catch (Throwable e) {
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void testCursorSwitchFails() {
+ try {
+ // allow all operations
+ ComponentRollbackTest.allowAllOps(lsmBtree);
+ // except search
+ lsmBtree.clearSearchCallbacks();
+ insertOp.open();
+ TupleGenerator tupleGenerator = new TupleGenerator(RECORD_TYPE, META_TYPE, KEY_INDEXES, KEY_INDICATORS,
+ RECORD_GEN_FUNCTION, UNIQUE_RECORD_FIELDS, META_GEN_FUNCTION, UNIQUE_META_FIELDS);
+ VSizeFrame frame = new VSizeFrame(ctx);
+ FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+ Searcher firstSearcher = null;
+ for (int j = 0; j < TOTAL_NUM_OF_RECORDS; j++) {
+ // flush every RECORDS_PER_COMPONENT records
+ if (j % RECORDS_PER_COMPONENT == (RECORDS_PER_COMPONENT - 1) && j + 1 != TOTAL_NUM_OF_RECORDS) {
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOp, true);
+ }
+ dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+ }
+ ITupleReference tuple = tupleGenerator.next();
+ DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
+ }
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOp, true);
+ }
+ insertOp.close();
+ // start the search
+ firstSearcher = new Searcher(nc, 0, dataset, storageManager, lsmBtree, TOTAL_NUM_OF_RECORDS);
+ // wait till firstSearcher enter the components
+ firstSearcher.waitUntilEntered();
+ dsLifecycleMgr.flushDataset(dataset.getDatasetId(), false);
+ nc.getTransactionManager().commitTransaction(txnCtx.getTxnId());
+ // merge all components
+ ILSMIndexAccessor mergeAccessor = lsmBtree.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ List<ILSMDiskComponent> mergedComponents = new ArrayList<>(lsmBtree.getDiskComponents());
+ BlockingIOOperationCallbackWrapper ioCallback =
+ new BlockingIOOperationCallbackWrapper(lsmBtree.getIOOperationCallback());
+ mergeAccessor.scheduleMerge(ioCallback, mergedComponents);
+ ioCallback.waitForIO();
+ // unblock the search
+ unblockSearch(lsmBtree);
+ // ensure the search got the correct number
+ Assert.assertTrue(firstSearcher.result());
+ // search now and ensure
+ searchAndAssertCount(nc, ctx, dataset, storageManager, TOTAL_NUM_OF_RECORDS);
+ } catch (Throwable e) {
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ private void searchAndAssertCount(TestNodeController nc, IHyracksTaskContext ctx, Dataset dataset,
+ StorageComponentProvider storageManager, int numOfRecords)
+ throws HyracksDataException, AlgebricksException {
+ nc.newJobId();
+ TestTupleCounterFrameWriter countOp =
+ ComponentRollbackTest.create(nc.getSearchOutputDesc(KEY_TYPES, RECORD_TYPE, META_TYPE),
+ Collections.emptyList(), Collections.emptyList(), false);
+ IPushRuntime emptyTupleOp = nc.getFullScanPipeline(countOp, ctx, dataset, KEY_TYPES, RECORD_TYPE, META_TYPE,
+ new NoMergePolicyFactory(), null, null, KEY_INDEXES, KEY_INDICATORS_LIST, storageManager);
+ emptyTupleOp.open();
+ emptyTupleOp.close();
+ Assert.assertEquals(numOfRecords, countOp.getCount());
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
index e0502de..0c4983a 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestDataset.java
@@ -21,7 +21,6 @@ package org.apache.asterix.test.dataflow;
import java.util.Map;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
-import org.apache.asterix.common.context.DatasetLSMComponentIdGeneratorFactory;
import org.apache.asterix.metadata.IDatasetDetails;
import org.apache.asterix.metadata.declared.MetadataProvider;
import org.apache.asterix.metadata.entities.Dataset;
@@ -63,6 +62,6 @@ public class TestDataset extends Dataset {
@Override
public ILSMIOOperationCallbackFactory getIoOperationCallbackFactory(Index index) throws AlgebricksException {
- return new TestLsmBtreeIoOpCallbackFactory(new DatasetLSMComponentIdGeneratorFactory(getDatasetId()));
+ return new TestLsmBtreeIoOpCallbackFactory(getComponentIdGeneratorFactory());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
index fa37c20..44967e3 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestLsmBtreeIoOpCallbackFactory.java
@@ -18,8 +18,9 @@
*/
package org.apache.asterix.test.dataflow;
-import org.apache.asterix.common.ioopcallbacks.AbstractLSMIndexIOOperationCallbackFactory;
import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallback;
+import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGeneratorFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
@@ -28,7 +29,7 @@ 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.impls.EmptyComponent;
-public class TestLsmBtreeIoOpCallbackFactory extends AbstractLSMIndexIOOperationCallbackFactory {
+public class TestLsmBtreeIoOpCallbackFactory extends LSMBTreeIOOperationCallbackFactory {
private static final long serialVersionUID = 1L;
@@ -97,7 +98,8 @@ public class TestLsmBtreeIoOpCallbackFactory extends AbstractLSMIndexIOOperation
}
@Override
- public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) {
+ public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent)
+ throws HyracksDataException {
super.afterFinalize(opType, newComponent);
synchronized (TestLsmBtreeIoOpCallbackFactory.this) {
if (newComponent != null) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
index 1cc24d5..1be1d26 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/logging/CheckpointingTest.java
@@ -54,6 +54,7 @@ import org.apache.asterix.transaction.management.service.logging.LogManager;
import org.apache.asterix.transaction.management.service.recovery.AbstractCheckpointManager;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
@@ -126,9 +127,9 @@ public class CheckpointingTest {
null, DatasetType.INTERNAL, DATASET_ID, 0);
try {
nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES,
- KEY_INDICATOR_LIST);
- IHyracksTaskContext ctx = nc.createTestContext(false);
- nc.newJobId();
+ KEY_INDICATOR_LIST, 0);
+ JobId jobId = nc.newJobId();
+ IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false);
ITransactionContext txnCtx = nc.getTransactionManager().beginTransaction(nc.getTxnJobId(ctx),
new TransactionOptions(ITransactionManager.AtomicityLevel.ENTITY_LEVEL));
// Prepare insert operation
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
index 5ee0e9f..cfd251b 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DeallocatableTest.java
@@ -28,6 +28,7 @@ import org.apache.asterix.test.runtime.LangExecutionUtil;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.ConnectorDescriptorId;
import org.apache.hyracks.api.dataflow.TaskAttemptId;
+import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.api.partitions.PartitionId;
import org.apache.hyracks.comm.channels.NetworkOutputChannel;
import org.apache.hyracks.control.nc.NodeControllerService;
@@ -58,7 +59,8 @@ public class DeallocatableTest {
final NodeControllerService ncs =
(NodeControllerService) nc.getAppRuntimeContext().getServiceContext().getControllerService();
final TaskAttemptId taId = Mockito.mock(TaskAttemptId.class);
- final IHyracksTaskContext ctx = nc.createTestContext(true);
+ JobId jobId = nc.newJobId();
+ final IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
final ConnectorDescriptorId codId = new ConnectorDescriptorId(1);
final PartitionId pid = new PartitionId(ctx.getJobletContext().getJobId(), codId, 1, 1);
final ChannelControlBlock ccb = ncs.getNetworkManager()
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
index cb83d56..8e8b3e9 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/DiskIsFullTest.java
@@ -33,7 +33,6 @@ import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
import org.apache.asterix.common.config.DatasetConfig.DatasetType;
import org.apache.asterix.common.dataflow.LSMInsertDeleteOperatorNodePushable;
import org.apache.asterix.common.exceptions.ExceptionUtils;
-import org.apache.asterix.common.transactions.ITransactionContext;
import org.apache.asterix.external.util.DataflowUtils;
import org.apache.asterix.file.StorageComponentProvider;
import org.apache.asterix.metadata.entities.Dataset;
@@ -49,6 +48,7 @@ import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
@@ -125,9 +125,9 @@ public class DiskIsFullTest {
null, DatasetType.INTERNAL, DATASET_ID, 0);
try {
nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES,
- KEY_INDICATOR_LIST);
- IHyracksTaskContext ctx = nc.createTestContext(false);
- nc.newJobId();
+ KEY_INDICATOR_LIST, 0);
+ JobId jobId = nc.newJobId();
+ IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, false);
// Prepare insert operation
LSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
RECORD_TYPE, META_TYPE, null, KEY_INDEXES, KEY_INDICATOR_LIST, storageManager).getLeft();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
index 7ced87d..cc6c0f7 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/storage/IndexDropOperatorNodePushableTest.java
@@ -18,8 +18,6 @@
*/
package org.apache.asterix.test.storage;
-import static org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -51,9 +49,11 @@ import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileSplit;
+import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider;
import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelperFactory;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorDescriptor.DropOption;
import org.apache.hyracks.storage.am.common.dataflow.IndexDropOperatorNodePushable;
import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
import org.junit.After;
@@ -106,15 +106,15 @@ public class IndexDropOperatorNodePushableTest {
Dataset dataset = new Dataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME, NODE_GROUP_NAME,
NoMergePolicyFactory.NAME, null,
new InternalDatasetDetails(null, InternalDatasetDetails.PartitioningStrategy.HASH, partitioningKeys,
- null, null, null, false, null, false), null, DatasetConfig.DatasetType.INTERNAL, DATASET_ID,
- 0);
+ null, null, null, false, null, false),
+ null, DatasetConfig.DatasetType.INTERNAL, DATASET_ID, 0);
// create dataset
- TestNodeController.PrimaryIndexInfo indexInfo =
- nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, null, storageManager, KEY_INDEXES,
- KEY_INDICATORS_LIST);
+ TestNodeController.PrimaryIndexInfo indexInfo = nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE,
+ META_TYPE, null, storageManager, KEY_INDEXES, KEY_INDICATORS_LIST, 0);
IndexDataflowHelperFactory helperFactory =
new IndexDataflowHelperFactory(nc.getStorageManager(), indexInfo.getFileSplitProvider());
- IHyracksTaskContext ctx = nc.createTestContext(true);
+ JobId jobId = nc.newJobId();
+ IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
IIndexDataflowHelper dataflowHelper = helperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
dropInUse(ctx, helperFactory, dataflowHelper);
dropInUseWithWait(ctx, helperFactory, dataflowHelper);
@@ -144,21 +144,21 @@ public class IndexDropOperatorNodePushableTest {
testExecutor.executeSqlppUpdateOrDdl("CREATE DATASET " + datasetName + "(KeyType) PRIMARY KEY id;", format);
testExecutor.executeSqlppUpdateOrDdl("CREATE INDEX " + indexName + " on " + datasetName + "(foo)", format);
final MetadataTransactionContext mdTxn = MetadataManager.INSTANCE.beginTransaction();
- ICcApplicationContext appCtx =
- (ICcApplicationContext) ExecutionTestUtil.integrationUtil.getClusterControllerService()
- .getApplicationContext();
+ ICcApplicationContext appCtx = (ICcApplicationContext) ExecutionTestUtil.integrationUtil
+ .getClusterControllerService().getApplicationContext();
MetadataProvider metadataProver = new MetadataProvider(appCtx, null);
metadataProver.setMetadataTxnContext(mdTxn);
final String defaultDv = MetadataBuiltinEntities.DEFAULT_DATAVERSE.getDataverseName();
final Dataset dataset = MetadataManager.INSTANCE.getDataset(mdTxn, defaultDv, datasetName);
MetadataManager.INSTANCE.commitTransaction(mdTxn);
- FileSplit[] splits = SplitsAndConstraintsUtil
- .getIndexSplits(appCtx.getClusterStateManager(), dataset, indexName, Arrays.asList("asterix_nc1"));
+ FileSplit[] splits = SplitsAndConstraintsUtil.getIndexSplits(appCtx.getClusterStateManager(), dataset,
+ indexName, Arrays.asList("asterix_nc1"));
final ConstantFileSplitProvider constantFileSplitProvider =
new ConstantFileSplitProvider(Arrays.copyOfRange(splits, 0, 1));
IndexDataflowHelperFactory helperFactory =
new IndexDataflowHelperFactory(nc.getStorageManager(), constantFileSplitProvider);
- IHyracksTaskContext ctx = nc.createTestContext(true);
+ JobId jobId = nc.newJobId();
+ IHyracksTaskContext ctx = nc.createTestContext(jobId, 0, true);
IIndexDataflowHelper dataflowHelper = helperFactory.create(ctx.getJobletContext().getServiceContext(), 0);
dropInUse(ctx, helperFactory, dataflowHelper);
} finally {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
index e445fe4..c33e2d1 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/ioopcallbacks/AbstractLSMIOOperationCallback.java
@@ -34,6 +34,7 @@ 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.ILSMMemoryComponent;
import org.apache.hyracks.storage.am.lsm.common.impls.DiskComponentMetadata;
+import org.apache.hyracks.storage.am.lsm.common.impls.EmptyComponent;
import org.apache.hyracks.storage.am.lsm.common.util.ComponentUtils;
import org.apache.hyracks.storage.am.lsm.common.util.LSMComponentIdUtils;
@@ -88,7 +89,7 @@ public abstract class AbstractLSMIOOperationCallback implements ILSMIOOperationC
}
@Override
- public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) {
+ public void afterFinalize(LSMIOOperationType opType, ILSMDiskComponent newComponent) throws HyracksDataException {
// The operation was complete and the next I/O operation for the LSM index didn't start yet
if (opType == LSMIOOperationType.FLUSH && newComponent != null) {
synchronized (this) {
@@ -100,6 +101,13 @@ public abstract class AbstractLSMIOOperationCallback implements ILSMIOOperationC
}
readIndex = (readIndex + 1) % mutableLastLSNs.length;
}
+ if (newComponent == EmptyComponent.INSTANCE) {
+ // This component was just deleted, we refresh the component id, when it gets recycled, it will get
+ // the new id from the component id generator.
+ // It is assumed that the component delete caller will ensure that corresponding components in secondary
+ // indexes are deleted as well
+ idGenerator.refresh();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
index 2d8eaed..a7468a7 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/base/AbstractOneInputOneOutputOneFramePushRuntime.java
@@ -51,13 +51,25 @@ public abstract class AbstractOneInputOneOutputOneFramePushRuntime extends Abstr
@Override
public void close() throws HyracksDataException {
+ HyracksDataException closeException = null;
try {
flushIfNotFailed();
} catch (Exception e) {
+ closeException = HyracksDataException.create(e);
writer.fail();
- throw e;
} finally {
- writer.close();
+ try {
+ writer.close();
+ } catch (Exception e) {
+ if (closeException == null) {
+ closeException = HyracksDataException.create(e);
+ } else {
+ closeException.addSuppressed(e);
+ }
+ }
+ }
+ if (closeException != null) {
+ throw closeException;
}
}
[2/3] asterixdb git commit: [ASTERIXDB-2169][STO][TX] Unblock
modifications during full scan
Posted by am...@apache.org.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
index f6ebf19..25a4aea 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
@@ -78,33 +78,48 @@ public class AlgebricksMetaOperatorDescriptor extends AbstractSingleActivityOper
public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
if (inputArity == 0) {
- return createSourceInputPushRuntime(ctx);
+ return new SourcePushRuntime(ctx);
} else {
return createOneInputOneOutputPushRuntime(ctx, recordDescProvider);
}
}
- private IOperatorNodePushable createSourceInputPushRuntime(final IHyracksTaskContext ctx) {
- return new AbstractUnaryOutputSourceOperatorNodePushable() {
+ private class SourcePushRuntime extends AbstractUnaryOutputSourceOperatorNodePushable {
+ private final IHyracksTaskContext ctx;
- @Override
- public void initialize() throws HyracksDataException {
- IFrameWriter startOfPipeline;
- RecordDescriptor pipelineOutputRecordDescriptor =
- outputArity > 0 ? AlgebricksMetaOperatorDescriptor.this.outRecDescs[0] : null;
- PipelineAssembler pa =
- new PipelineAssembler(pipeline, inputArity, outputArity, null, pipelineOutputRecordDescriptor);
- startOfPipeline = pa.assemblePipeline(writer, ctx);
+ public SourcePushRuntime(IHyracksTaskContext ctx) {
+ this.ctx = ctx;
+ }
+
+ @Override
+ public void initialize() throws HyracksDataException {
+ IFrameWriter startOfPipeline;
+ RecordDescriptor pipelineOutputRecordDescriptor =
+ outputArity > 0 ? AlgebricksMetaOperatorDescriptor.this.outRecDescs[0] : null;
+ PipelineAssembler pa =
+ new PipelineAssembler(pipeline, inputArity, outputArity, null, pipelineOutputRecordDescriptor);
+ startOfPipeline = pa.assemblePipeline(writer, ctx);
+ HyracksDataException exception = null;
+ try {
+ startOfPipeline.open();
+ } catch (Exception e) {
+ startOfPipeline.fail();
+ exception = HyracksDataException.create(e);
+ } finally {
try {
- startOfPipeline.open();
- } catch (Exception e) {
- startOfPipeline.fail();
- throw e;
- } finally {
startOfPipeline.close();
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = HyracksDataException.create(e);
+ } else {
+ exception.addSuppressed(e);
+ }
}
}
- };
+ if (exception != null) {
+ throw exception;
+ }
+ }
}
private IOperatorNodePushable createOneInputOneOutputPushRuntime(final IHyracksTaskContext ctx,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java b/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
index d64858b..71ddbc0 100644
--- a/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
+++ b/hyracks-fullstack/hyracks/hyracks-http/src/main/java/org/apache/hyracks/http/server/HttpServer.java
@@ -49,8 +49,8 @@ public class HttpServer {
// Constants
private static final int LOW_WRITE_BUFFER_WATER_MARK = 8 * 1024;
private static final int HIGH_WRITE_BUFFER_WATER_MARK = 32 * 1024;
- protected static final WriteBufferWaterMark WRITE_BUFFER_WATER_MARK = new WriteBufferWaterMark(
- LOW_WRITE_BUFFER_WATER_MARK, HIGH_WRITE_BUFFER_WATER_MARK);
+ protected static final WriteBufferWaterMark WRITE_BUFFER_WATER_MARK =
+ new WriteBufferWaterMark(LOW_WRITE_BUFFER_WATER_MARK, HIGH_WRITE_BUFFER_WATER_MARK);
protected static final int RECEIVE_BUFFER_SIZE = 4096;
protected static final int DEFAULT_NUM_EXECUTOR_THREADS = 16;
protected static final int DEFAULT_REQUEST_QUEUE_SIZE = 256;
@@ -92,8 +92,8 @@ public class HttpServer {
long directMemoryBudget = numExecutorThreads * (long) HIGH_WRITE_BUFFER_WATER_MARK
+ numExecutorThreads * HttpServerInitializer.RESPONSE_CHUNK_SIZE;
LOGGER.log(Level.INFO, "The output direct memory budget for this server is " + directMemoryBudget + " bytes");
- long inputBudgetEstimate = (long) HttpServerInitializer.MAX_REQUEST_INITIAL_LINE_LENGTH
- * (requestQueueSize + numExecutorThreads);
+ long inputBudgetEstimate =
+ (long) HttpServerInitializer.MAX_REQUEST_INITIAL_LINE_LENGTH * (requestQueueSize + numExecutorThreads);
inputBudgetEstimate = inputBudgetEstimate * 2;
LOGGER.log(Level.INFO,
"The \"estimated\" input direct memory budget for this server is " + inputBudgetEstimate + " bytes");
@@ -111,7 +111,7 @@ public class HttpServer {
doStart();
setStarted();
} catch (Throwable e) { // NOSONAR
- LOGGER.log(Level.SEVERE, "Failure starting an Http Server", e);
+ LOGGER.log(Level.SEVERE, "Failure starting an Http Server with port: " + port, e);
setFailed(e);
throw e;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
index a859f68..673bd3b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeLocalResource.java
@@ -62,6 +62,6 @@ public class ExternalBTreeLocalResource extends LSMBTreeLocalResource {
cmpFactories, bloomFilterKeyFields, bloomFilterFalsePositiveRate,
mergePolicyFactory.createMergePolicy(mergePolicyProperties, serviceCtx),
opTrackerProvider.getOperationTracker(serviceCtx), ioSchedulerProvider.getIoScheduler(serviceCtx),
- ioOpCallbackFactory, durable, metadataPageManagerFactory);
+ ioOpCallbackFactory, durable, metadataPageManagerFactory, serviceCtx.getTracer());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
index 9422253..7e44c63 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/dataflow/ExternalBTreeWithBuddyLocalResource.java
@@ -65,6 +65,6 @@ public class ExternalBTreeWithBuddyLocalResource extends LSMBTreeLocalResource {
typeTraits, cmpFactories, bloomFilterFalsePositiveRate,
mergePolicyFactory.createMergePolicy(mergePolicyProperties, serviceCtx),
opTrackerProvider.getOperationTracker(serviceCtx), ioSchedulerProvider.getIoScheduler(serviceCtx),
- ioOpCallbackFactory, bloomFilterKeyFields, durable, metadataPageManagerFactory);
+ ioOpCallbackFactory, bloomFilterKeyFields, durable, metadataPageManagerFactory, serviceCtx.getTracer());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
index 7dc5939..5b6ff9e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTree.java
@@ -62,6 +62,7 @@ import org.apache.hyracks.storage.common.IIndexCursor;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.ISearchPredicate;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
/**
* This is an lsm b-tree that does not have memory component and is modified
@@ -94,18 +95,18 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
ILSMDiskComponentFactory bulkLoadComponentFactory, ILSMDiskComponentFactory transactionComponentFactory,
double bloomFilterFalsePositiveRate, IBinaryComparatorFactory[] cmpFactories, ILSMMergePolicy mergePolicy,
ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
- ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable) {
+ ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable, ITracer tracer) {
super(ioManager, insertLeafFrameFactory, deleteLeafFrameFactory, bufferCache, fileManager, componentFactory,
bulkLoadComponentFactory, bloomFilterFalsePositiveRate, cmpFactories, mergePolicy, opTracker,
- ioScheduler, ioOpCallbackFactory, false, durable);
+ ioScheduler, ioOpCallbackFactory, false, durable, tracer);
this.transactionComponentFactory = transactionComponentFactory;
this.secondDiskComponents = new LinkedList<>();
this.interiorFrameFactory = interiorFrameFactory;
}
@Override
- public ExternalIndexHarness getLsmHarness() {
- return (ExternalIndexHarness) super.getLsmHarness();
+ public ExternalIndexHarness getHarness() {
+ return (ExternalIndexHarness) super.getHarness();
}
// The subsume merged components is overridden to account for:
@@ -162,9 +163,9 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
// This method creates the appropriate opContext for the targeted version
public ExternalBTreeOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
- return new ExternalBTreeOpContext(insertLeafFrameFactory, deleteLeafFrameFactory, searchCallback,
+ return new ExternalBTreeOpContext(this, insertLeafFrameFactory, deleteLeafFrameFactory, searchCallback,
((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length,
- cmpFactories, targetVersion, getLsmHarness());
+ cmpFactories, targetVersion, getHarness(), tracer);
}
// The only reason to override the following method is that it uses a different context object
@@ -194,7 +195,7 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
FileReference lastFile = lastBTree.getFileReference();
LSMComponentFileReferences relMergeFileRefs =
fileManager.getRelMergeFileReference(firstFile.getFile().getName(), lastFile.getFile().getName());
- ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
ioScheduler.scheduleOperation(new LSMBTreeMergeOperation(accessor, cursor,
relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getBloomFilterFileReference(),
callback, fileManager.getBaseDir().getAbsolutePath()));
@@ -259,7 +260,7 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
diskComponents.add(component);
secondDiskComponents.add(component);
}
- getLsmHarness().indexFirstTimeActivated();
+ getHarness().indexFirstTimeActivated();
} else {
// This index has been opened before
for (ILSMDiskComponent c : diskComponents) {
@@ -309,7 +310,7 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
if (!isActive) {
throw new HyracksDataException("Failed to clear the index since it is not activated.");
}
- getLsmHarness().indexClear();
+ getHarness().indexClear();
for (ILSMDiskComponent c : diskComponents) {
c.deactivateAndDestroy();
@@ -451,7 +452,7 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
component.markAsValid(durable);
component.deactivate();
} else {
- getLsmHarness().addBulkLoadedComponent(component);
+ getHarness().addBulkLoadedComponent(component);
}
}
}
@@ -490,14 +491,14 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
@Override
public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) {
ExternalBTreeOpContext opCtx = createOpContext(iap.getSearchOperationCallback(), version);
- return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
}
@Override
public ILSMIndexAccessor createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
throws HyracksDataException {
ExternalBTreeOpContext opCtx = createOpContext(searchCallback, targetIndexVersion);
- return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
}
@Override
@@ -553,7 +554,7 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
component = createDiskComponent(componentFactory, componentFileReferences.getInsertIndexFileReference(),
null, componentFileReferences.getBloomFilterFileReference(), false);
}
- getLsmHarness().addTransactionComponents(component);
+ getHarness().addTransactionComponents(component);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
index f94c38a..b0c31ad 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeOpContext.java
@@ -22,9 +22,11 @@ import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.storage.am.btree.api.IBTreeLeafFrame;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
public class ExternalBTreeOpContext extends AbstractLSMIndexOperationContext {
private IBTreeLeafFrame insertLeafFrame;
@@ -34,11 +36,11 @@ public class ExternalBTreeOpContext extends AbstractLSMIndexOperationContext {
private final int targetIndexVersion;
private LSMBTreeCursorInitialState searchInitialState;
- public ExternalBTreeOpContext(ITreeIndexFrameFactory insertLeafFrameFactory,
+ public ExternalBTreeOpContext(ILSMIndex index, ITreeIndexFrameFactory insertLeafFrameFactory,
ITreeIndexFrameFactory deleteLeafFrameFactory, ISearchOperationCallback searchCallback,
int numBloomFilterKeyFields, IBinaryComparatorFactory[] cmpFactories, int targetIndexVersion,
- ILSMHarness lsmHarness) {
- super(null, null, null, searchCallback, null);
+ ILSMHarness lsmHarness, ITracer tracer) {
+ super(index, null, null, null, searchCallback, null, tracer);
if (cmpFactories != null) {
this.cmp = MultiComparator.create(cmpFactories);
} else {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
index c495b69..c6ce339 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddy.java
@@ -66,6 +66,7 @@ import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.ISearchPredicate;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
+import org.apache.hyracks.util.trace.ITracer;
public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeIndex, ITwoPCIndex {
@@ -89,9 +90,10 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
ILSMDiskComponentFactory bulkLoadComponentFactory, double bloomFilterFalsePositiveRate,
ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
ILSMIOOperationCallbackFactory ioOpCallbackFactory, IBinaryComparatorFactory[] btreeCmpFactories,
- IBinaryComparatorFactory[] buddyBtreeCmpFactories, int[] buddyBTreeFields, boolean durable) {
+ IBinaryComparatorFactory[] buddyBtreeCmpFactories, int[] buddyBTreeFields, boolean durable,
+ ITracer tracer) {
super(ioManager, diskBufferCache, fileManager, bloomFilterFalsePositiveRate, mergePolicy, opTracker,
- ioScheduler, ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable);
+ ioScheduler, ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable, tracer);
this.btreeCmpFactories = btreeCmpFactories;
this.buddyBtreeCmpFactories = buddyBtreeCmpFactories;
this.buddyBTreeFields = buddyBTreeFields;
@@ -125,7 +127,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
diskComponents.add(component);
secondDiskComponents.add(component);
}
- ((ExternalIndexHarness) getLsmHarness()).indexFirstTimeActivated();
+ ((ExternalIndexHarness) getHarness()).indexFirstTimeActivated();
} else {
// This index has been opened before or is brand new with no
// components. It should also maintain the version pointer
@@ -147,7 +149,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
if (!isActive) {
throw new HyracksDataException("Failed to clear the index since it is not activated.");
}
- ((ExternalIndexHarness) getLsmHarness()).indexClear();
+ ((ExternalIndexHarness) getHarness()).indexClear();
for (ILSMDiskComponent c : diskComponents) {
c.deactivateAndDestroy();
// Remove from second list to avoid destroying twice
@@ -182,7 +184,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
@Override
public ILSMIndexAccessor createAccessor(IIndexAccessParameters iap) throws HyracksDataException {
- return new LSMTreeIndexAccessor(getLsmHarness(), createOpContext(iap.getSearchOperationCallback(), version),
+ return new LSMTreeIndexAccessor(getHarness(), createOpContext(iap.getSearchOperationCallback(), version),
ctx -> new LSMBTreeWithBuddySearchCursor(ctx, buddyBTreeFields));
}
@@ -273,7 +275,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
List<ILSMComponent> mergingComponents = ctx.getComponentHolder();
ITreeIndexCursor cursor = new LSMBTreeWithBuddySortedCursor(bctx, buddyBTreeFields);
LSMComponentFileReferences relMergeFileRefs = getMergeTargetFileName(mergingComponents);
- ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getLsmHarness(), bctx,
+ ILSMIndexAccessor accessor = new LSMTreeIndexAccessor(getHarness(), bctx,
opCtx -> new LSMBTreeWithBuddySearchCursor(opCtx, buddyBTreeFields));
// Since we have two lists of components, to tell whether we need to
@@ -296,9 +298,9 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
// This method creates the appropriate opContext for the targeted version
public ExternalBTreeWithBuddyOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
- return new ExternalBTreeWithBuddyOpContext(btreeCmpFactories, buddyBtreeCmpFactories, searchCallback,
- targetVersion, getLsmHarness(), btreeInteriorFrameFactory, btreeLeafFrameFactory,
- buddyBtreeLeafFrameFactory);
+ return new ExternalBTreeWithBuddyOpContext(this, btreeCmpFactories, buddyBtreeCmpFactories, searchCallback,
+ targetVersion, getHarness(), btreeInteriorFrameFactory, btreeLeafFrameFactory,
+ buddyBtreeLeafFrameFactory, tracer);
}
@Override
@@ -528,7 +530,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
component.markAsValid(durable);
component.deactivate();
} else {
- getLsmHarness().addBulkLoadedComponent(component);
+ getHarness().addBulkLoadedComponent(component);
}
}
}
@@ -564,7 +566,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
@Override
public ILSMIndexAccessor createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
throws HyracksDataException {
- return new LSMTreeIndexAccessor(getLsmHarness(), createOpContext(searchCallback, targetIndexVersion),
+ return new LSMTreeIndexAccessor(getHarness(), createOpContext(searchCallback, targetIndexVersion),
ctx -> new LSMBTreeWithBuddySearchCursor(ctx, buddyBTreeFields));
}
@@ -609,7 +611,7 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
componentFileRefrences.getDeleteIndexFileReference(),
componentFileRefrences.getBloomFilterFileReference(), false);
}
- ((ExternalIndexHarness) getLsmHarness()).addTransactionComponents(component);
+ ((ExternalIndexHarness) getHarness()).addTransactionComponents(component);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
index d5cd2e2..85d4ab2 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
@@ -22,10 +22,12 @@ import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
public class ExternalBTreeWithBuddyOpContext extends AbstractLSMIndexOperationContext {
private MultiComparator bTreeCmp;
@@ -33,11 +35,12 @@ public class ExternalBTreeWithBuddyOpContext extends AbstractLSMIndexOperationCo
private final int targetIndexVersion;
private LSMBTreeWithBuddyCursorInitialState searchInitialState;
- public ExternalBTreeWithBuddyOpContext(IBinaryComparatorFactory[] btreeCmpFactories,
+ public ExternalBTreeWithBuddyOpContext(ILSMIndex index, IBinaryComparatorFactory[] btreeCmpFactories,
IBinaryComparatorFactory[] buddyBtreeCmpFactories, ISearchOperationCallback searchCallback,
int targetIndexVersion, ILSMHarness lsmHarness, ITreeIndexFrameFactory btreeInteriorFrameFactory,
- ITreeIndexFrameFactory btreeLeafFrameFactory, ITreeIndexFrameFactory buddyBtreeLeafFrameFactory) {
- super(null, null, null, searchCallback, null);
+ ITreeIndexFrameFactory btreeLeafFrameFactory, ITreeIndexFrameFactory buddyBtreeLeafFrameFactory,
+ ITracer tracer) {
+ super(index, null, null, null, searchCallback, null, tracer);
this.targetIndexVersion = targetIndexVersion;
this.bTreeCmp = MultiComparator.create(btreeCmpFactories);
this.buddyBTreeCmp = MultiComparator.create(buddyBtreeCmpFactories);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
index e69aae1..6929530 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
@@ -125,9 +125,10 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
ILSMDiskComponentFactory componentFactory, ILSMDiskComponentFactory bulkLoadComponentFactory,
double bloomFilterFalsePositiveRate, IBinaryComparatorFactory[] cmpFactories, ILSMMergePolicy mergePolicy,
ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
- ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean needKeyDupCheck, boolean durable) {
+ ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean needKeyDupCheck, boolean durable,
+ ITracer tracer) {
super(ioManager, bufferCache, fileManager, bloomFilterFalsePositiveRate, mergePolicy, opTracker, ioScheduler,
- ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable);
+ ioOpCallbackFactory, componentFactory, bulkLoadComponentFactory, durable, tracer);
this.insertLeafFrameFactory = insertLeafFrameFactory;
this.deleteLeafFrameFactory = deleteLeafFrameFactory;
this.cmpFactories = cmpFactories;
@@ -364,10 +365,11 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
public LSMBTreeOpContext createOpContext(IModificationOperationCallback modificationCallback,
ISearchOperationCallback searchCallback) {
int numBloomFilterKeyFields = hasBloomFilter
- ? ((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length : 0;
- return new LSMBTreeOpContext(memoryComponents, insertLeafFrameFactory, deleteLeafFrameFactory,
+ ? ((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length
+ : 0;
+ return new LSMBTreeOpContext(this, memoryComponents, insertLeafFrameFactory, deleteLeafFrameFactory,
modificationCallback, searchCallback, numBloomFilterKeyFields, getTreeFields(), getFilterFields(),
- getLsmHarness(), getFilterCmpFactories());
+ getHarness(), getFilterCmpFactories(), tracer);
}
@Override
@@ -376,7 +378,7 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
}
public ILSMIndexAccessor createAccessor(AbstractLSMIndexOperationContext opCtx) {
- return new LSMTreeIndexAccessor(getLsmHarness(), opCtx, cursorFactory);
+ return new LSMTreeIndexAccessor(getHarness(), opCtx, cursorFactory);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
index 951ec81..9cab94e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
@@ -32,12 +32,14 @@ import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMemoryComponent;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndexOperationContext;
import org.apache.hyracks.storage.common.IIndexAccessParameters;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
public final class LSMBTreeOpContext extends AbstractLSMIndexOperationContext {
@@ -62,11 +64,12 @@ public final class LSMBTreeOpContext extends AbstractLSMIndexOperationContext {
private BTree.BTreeAccessor currentMutableBTreeAccessor;
private BTreeOpContext currentMutableBTreeOpCtx;
- public LSMBTreeOpContext(List<ILSMMemoryComponent> mutableComponents, ITreeIndexFrameFactory insertLeafFrameFactory,
- ITreeIndexFrameFactory deleteLeafFrameFactory, IModificationOperationCallback modificationCallback,
- ISearchOperationCallback searchCallback, int numBloomFilterKeyFields, int[] btreeFields, int[] filterFields,
- ILSMHarness lsmHarness, IBinaryComparatorFactory[] filterCmpFactories) {
- super(btreeFields, filterFields, filterCmpFactories, searchCallback, modificationCallback);
+ public LSMBTreeOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
+ ITreeIndexFrameFactory insertLeafFrameFactory, ITreeIndexFrameFactory deleteLeafFrameFactory,
+ IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
+ int numBloomFilterKeyFields, int[] btreeFields, int[] filterFields, ILSMHarness lsmHarness,
+ IBinaryComparatorFactory[] filterCmpFactories, ITracer tracer) {
+ super(index, btreeFields, filterFields, filterCmpFactories, searchCallback, modificationCallback, tracer);
LSMBTreeMemoryComponent c = (LSMBTreeMemoryComponent) mutableComponents.get(0);
IBinaryComparatorFactory cmpFactories[] = c.getIndex().getComparatorFactories();
if (cmpFactories[0] != null) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
index 6cc8fbb..3e14fb9 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeRangeSearchCursor.java
@@ -34,8 +34,10 @@ import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.ComponentState;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent.LSMComponentType;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMMemoryComponent;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMIndexSearchCursor;
import org.apache.hyracks.storage.common.ICursorInitialState;
import org.apache.hyracks.storage.common.IIndexCursor;
@@ -47,10 +49,11 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
private final RangePredicate reusablePred;
private ISearchOperationCallback searchCallback;
- private RangePredicate predicate;
+
private BTreeAccessor[] btreeAccessors;
private ArrayTupleBuilder tupleBuilder;
private boolean canCallProceed = true;
+ private int tupleFromMemoryComponentCount = 0;
public LSMBTreeRangeSearchCursor(ILSMIndexOperationContext opCtx) {
this(opCtx, false);
@@ -73,6 +76,9 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
outputElement = outputPriorityQueue.poll();
needPushElementIntoQueue = true;
canCallProceed = false;
+ if (outputElement.getCursorIndex() == 0) {
+ tupleFromMemoryComponentCount++;
+ }
}
/**
@@ -87,6 +93,15 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
*/
@Override
protected void checkPriorityQueue() throws HyracksDataException {
+ // Every SWITCH_COMPONENT_CYCLE calls, check if memory components need to be swapped with disk components
+ // We should do this regardless of the value of includeMutableComponent. This is because if the cursor
+ // of the memory component has gone past the end of the in memory component, then the includeMutableComponent
+ // will be set to false. Still, when that happens, we want to exit the memory component to allow it to be
+ // recycled and used for modifications.
+ if (hasNextCallCount >= SWITCH_COMPONENT_CYCLE) {
+ replaceMemoryComponentWithDiskComponentIfNeeded();
+ hasNextCallCount = 0;
+ }
while (!outputPriorityQueue.isEmpty() || needPushElementIntoQueue) {
if (!outputPriorityQueue.isEmpty()) {
PriorityQueueElement queueHead = outputPriorityQueue.peek();
@@ -97,7 +112,7 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
if (!searchCallback.proceed(queueHead.getTuple())) {
// In case proceed() fails and there is an in-memory component,
// we can't simply use this element since there might be a change.
- PriorityQueueElement mutableElement = removeMutable(outputPriorityQueue);
+ PriorityQueueElement mutableElement = remove(outputPriorityQueue, 0);
if (mutableElement != null) {
// Copies the current queue head
if (tupleBuilder == null) {
@@ -166,10 +181,7 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
} else {
// If the previous tuple and the head tuple are different
// the info of previous tuple is useless
- if (needPushElementIntoQueue == true) {
- pushIntoQueueFromCursorAndReplaceThisElement(outputElement);
- needPushElementIntoQueue = false;
- }
+ pushOutputElementIntoQueueIfNeeded();
canCallProceed = true;
outputElement = null;
}
@@ -185,16 +197,128 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
}
- private PriorityQueueElement removeMutable(PriorityQueue<PriorityQueueElement> outputPriorityQueue) {
- // Scans the PQ for the mutable component's element and delete it
- // since it can be changed.
- // (i.e. we can't ensure that the element is the most current one.)
+ private void pushOutputElementIntoQueueIfNeeded() throws HyracksDataException {
+ if (needPushElementIntoQueue) {
+ pushIntoQueueFromCursorAndReplaceThisElement(outputElement);
+ needPushElementIntoQueue = false;
+ }
+ }
+
+ private void replaceMemoryComponentWithDiskComponentIfNeeded() throws HyracksDataException {
+ int replaceFrom = replaceFrom();
+ if (replaceFrom < 0) {
+ // no switch is needed, check if we need to re-do the search on the memory component.
+ // searches and modifications compete on the pages of the memory component
+ // if the cursor on the memory component is not advancing, we re-do the operation in order
+ // to release the latches and allow modifications to proceed
+ redoMemoryComponentSearchIfNeeded();
+ return;
+ }
+ opCtx.getIndex().getHarness().replaceMemoryComponentsWithDiskComponents(getOpCtx(), replaceFrom);
+ // redo the search on the new component
+ for (int i = replaceFrom; i < switchRequest.length; i++) {
+ if (switchRequest[i] && switchedElements[i] != null) {
+ copyTuple.reset(switchComponentTupleBuilders[i].getFieldEndOffsets(),
+ switchComponentTupleBuilders[i].getByteArray());
+ reusablePred.setLowKey(copyTuple, true);
+ rangeCursors[i].reset();
+ ILSMComponent component = operationalComponents.get(i);
+ BTree btree = (BTree) component.getIndex();
+ if (i == 0 && component.getType() != LSMComponentType.MEMORY) {
+ includeMutableComponent = false;
+ }
+ btreeAccessors[i].reset(btree, NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+ btreeAccessors[i].search(rangeCursors[i], reusablePred);
+ pushIntoQueueFromCursorAndReplaceThisElement(switchedElements[i]);
+ }
+ switchRequest[i] = false;
+ // any failed switch makes further switches pointless
+ switchPossible = switchPossible && operationalComponents.get(i).getType() == LSMComponentType.DISK;
+ }
+ }
+
+ private int replaceFrom() throws HyracksDataException {
+ int replaceFrom = -1;
+ if (!switchPossible) {
+ return replaceFrom;
+ }
+ for (int i = 0; i < operationalComponents.size(); i++) {
+ ILSMComponent next = operationalComponents.get(i);
+ if (next.getType() == LSMComponentType.DISK) {
+ if (i == 0) {
+ // if the first component is a disk component, then switch is not possible
+ switchPossible = false;
+ }
+ break;
+ } else if (next.getState() == ComponentState.UNREADABLE_UNWRITABLE) {
+ // if the component is UNREADABLE_UNWRITABLE, then it means that the flush has been completed while
+ // the search cursor is inside the component, a switch candidate
+ if (replaceFrom < 0) {
+ replaceFrom = i;
+ }
+ // we return the outputElement to the priority queue if it came from this component
+ if (outputElement != null && outputElement.getCursorIndex() == i) {
+ pushIntoQueueFromCursorAndReplaceThisElement(outputElement);
+ needPushElementIntoQueue = false;
+ outputElement = null;
+ canCallProceed = true;
+ }
+ PriorityQueueElement element = remove(outputPriorityQueue, i);
+ // if this cursor is still active (has an element)
+ // then we copy the search key to restart the operation after
+ // replacing the component
+ if (element != null) {
+ if (switchComponentTupleBuilders[i] == null) {
+ switchComponentTupleBuilders[i] = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+ }
+ TupleUtils.copyTuple(switchComponentTupleBuilders[i], element.getTuple(), cmp.getKeyFieldCount());
+ }
+ rangeCursors[i].reset();
+ rangeCursors[i].close();
+ switchRequest[i] = true;
+ switchedElements[i] = element;
+ }
+ }
+ return replaceFrom;
+ }
+
+ private void redoMemoryComponentSearchIfNeeded() throws HyracksDataException {
+ if (!includeMutableComponent) {
+ return;
+ }
+ // if the last n records, none were from memory and there are writers inside the component,
+ // we need to re-do the search so the cursor doesn't block modifications due to latches over page
+ if (tupleFromMemoryComponentCount == 0
+ && ((AbstractLSMMemoryComponent) operationalComponents.get(0)).getWriterCount() > 0) {
+ // When we reach here, we know that the mutable component element is not the outputElement
+ // since if it was the output element, the tupleFromMemoryComponentCount would be at least 1
+ PriorityQueueElement mutableElement = remove(outputPriorityQueue, 0);
+ if (mutableElement != null) {
+ // if the element is null, then there is nothing to do since no latches are held
+ if (tupleBuilder == null) {
+ tupleBuilder = new ArrayTupleBuilder(cmp.getKeyFieldCount());
+ }
+ TupleUtils.copyTuple(tupleBuilder, mutableElement.getTuple(), cmp.getKeyFieldCount());
+ copyTuple.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
+ // Unlatches/unpins the leaf page of the index.
+ rangeCursors[0].reset();
+ // Re-traverses the index.
+ reusablePred.setLowKey(copyTuple, true);
+ btreeAccessors[0].search(rangeCursors[0], reusablePred);
+ includeMutableComponent = pushIntoQueueFromCursorAndReplaceThisElement(mutableElement);
+ }
+ }
+ tupleFromMemoryComponentCount = 0;
+ }
+
+ private PriorityQueueElement remove(PriorityQueue<PriorityQueueElement> outputPriorityQueue, int cursorIndex) {
+ // Scans the PQ for the component's element and delete it
Iterator<PriorityQueueElement> it = outputPriorityQueue.iterator();
while (it.hasNext()) {
- PriorityQueueElement mutableElement = it.next();
- if (mutableElement.getCursorIndex() == 0) {
+ PriorityQueueElement e = it.next();
+ if (e.getCursorIndex() == cursorIndex) {
it.remove();
- return mutableElement;
+ return e;
}
}
return null;
@@ -207,7 +331,7 @@ public class LSMBTreeRangeSearchCursor extends LSMIndexSearchCursor {
operationalComponents = lsmInitialState.getOperationalComponents();
lsmHarness = lsmInitialState.getLSMHarness();
searchCallback = lsmInitialState.getSearchOperationCallback();
- predicate = (RangePredicate) lsmInitialState.getSearchPredicate();
+ RangePredicate predicate = (RangePredicate) lsmInitialState.getSearchPredicate();
reusablePred.setLowKeyComparator(cmp);
reusablePred.setHighKey(predicate.getHighKey(), predicate.isHighKeyInclusive());
reusablePred.setHighKeyComparator(predicate.getHighKeyComparator());
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
index a0b1905..08e5af0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/utils/LSMBTreeUtil.java
@@ -132,7 +132,7 @@ public class LSMBTreeUtil {
int[] bloomFilterKeyFields, double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy,
ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler,
ILSMIOOperationCallbackFactory ioOpCallbackFactory, boolean durable,
- IMetadataPageManagerFactory freePageManagerFactory) {
+ IMetadataPageManagerFactory freePageManagerFactory, ITracer tracer) {
LSMBTreeTupleWriterFactory insertTupleWriterFactory =
new LSMBTreeTupleWriterFactory(typeTraits, cmpFactories.length, false, false);
LSMBTreeTupleWriterFactory deleteTupleWriterFactory =
@@ -176,14 +176,15 @@ public class LSMBTreeUtil {
return new ExternalBTree(ioManager, interiorFrameFactory, insertLeafFrameFactory, deleteLeafFrameFactory,
diskBufferCache, fileNameManager, componentFactory, bulkLoadComponentFactory,
transactionComponentFactory, bloomFilterFalsePositiveRate, cmpFactories, mergePolicy, opTracker,
- ioScheduler, ioOpCallbackFactory, durable);
+ ioScheduler, ioOpCallbackFactory, durable, tracer);
}
public static ExternalBTreeWithBuddy createExternalBTreeWithBuddy(IIOManager ioManager, FileReference file,
IBufferCache diskBufferCache, ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker,
ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackFactory ioOpCallbackFactory,
- int[] buddyBTreeFields, boolean durable, IMetadataPageManagerFactory freePageManagerFactory) {
+ int[] buddyBTreeFields, boolean durable, IMetadataPageManagerFactory freePageManagerFactory,
+ ITracer tracer) {
ITypeTraits[] buddyBtreeTypeTraits = new ITypeTraits[buddyBTreeFields.length];
IBinaryComparatorFactory[] buddyBtreeCmpFactories = new IBinaryComparatorFactory[buddyBTreeFields.length];
for (int i = 0; i < buddyBtreeTypeTraits.length; i++) {
@@ -232,6 +233,6 @@ public class LSMBTreeUtil {
return new ExternalBTreeWithBuddy(ioManager, interiorFrameFactory, insertLeafFrameFactory,
buddyBtreeLeafFrameFactory, diskBufferCache, fileNameManager, componentFactory,
bulkLoadComponentFactory, bloomFilterFalsePositiveRate, mergePolicy, opTracker, ioScheduler,
- ioOpCallbackFactory, cmpFactories, buddyBtreeCmpFactories, buddyBTreeFields, durable);
+ ioOpCallbackFactory, cmpFactories, buddyBtreeCmpFactories, buddyBTreeFields, durable, tracer);
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
index ab8e899..fc55ce5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponent.java
@@ -129,6 +129,11 @@ public interface ILSMComponent {
IIndex getIndex();
/**
+ * @return the {@link ILSMIndex} this component belong to
+ */
+ ILSMIndex getLsmIndex();
+
+ /**
*
* @return id of the component
* @throws HyracksDataException
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
index 89c8cb9..b32dd0f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
@@ -253,4 +253,17 @@ public interface ILSMHarness {
*/
void deleteComponents(ILSMIndexOperationContext ctx, Predicate<ILSMComponent> predicate)
throws HyracksDataException;
+
+ /**
+ * Replace the memory components in this operation context with their corresponding disk
+ * components if possible
+ *
+ * @param ctx
+ * the operation context
+ * @param startIndex
+ * the index of the first component to switch
+ * @throws HyracksDataException
+ */
+ void replaceMemoryComponentsWithDiskComponents(ILSMIndexOperationContext ctx, int startIndex)
+ throws HyracksDataException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
index 52f9f06..62493f4 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndex.java
@@ -143,4 +143,9 @@ public interface ILSMIndex extends IIndex {
* @return The number of all memory components (active and inactive)
*/
int getNumberOfAllMemoryComponents();
+
+ /**
+ * @return the {@link ILSMHarness} of the index
+ */
+ ILSMHarness getHarness();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
index 5b0378a..ec9124d 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
@@ -56,4 +56,30 @@ public interface ILSMIndexOperationContext extends IIndexOperationContext {
PermutingTupleReference getFilterTuple();
MultiComparator getFilterCmp();
+
+ /**
+ * @return the {@link ILSMIndex} of the component
+ */
+ ILSMIndex getIndex();
+
+ /**
+ * Performance tracing method. Logs the accumulated counters for number of tuples
+ *
+ * @param tupleCount
+ * the number of tuples represented by the counters
+ */
+ void logPerformanceCounters(int tupleCount);
+
+ /**
+ * Increment the time taken for entering and exiting components
+ *
+ * @param increment
+ * the time increment in nanoseconds
+ */
+ void incrementEnterExitTime(long increment);
+
+ /**
+ * @return true if performance tracing is enabled, false otherwise
+ */
+ boolean isTracingEnabled();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
index ac124ba..84d2fe5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMComponent.java
@@ -28,7 +28,7 @@ public abstract class AbstractLSMComponent implements ILSMComponent {
protected final ILSMComponentFilter filter;
protected final AbstractLSMIndex lsmIndex;
// Mutables
- protected ComponentState state;
+ protected volatile ComponentState state;
protected int readerCount;
public AbstractLSMComponent(AbstractLSMIndex lsmIndex, ILSMComponentFilter filter) {
@@ -46,4 +46,9 @@ public abstract class AbstractLSMComponent implements ILSMComponent {
public ILSMComponentFilter getLSMComponentFilter() {
return filter;
}
+
+ @Override
+ public final AbstractLSMIndex getLsmIndex() {
+ return lsmIndex;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
index b664102..bb27236 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMDiskComponent.java
@@ -18,6 +18,7 @@
*/
package org.apache.hyracks.storage.am.lsm.common.impls;
+import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -49,11 +50,6 @@ public abstract class AbstractLSMDiskComponent extends AbstractLSMComponent impl
}
@Override
- public AbstractLSMIndex getLsmIndex() {
- return lsmIndex;
- }
-
- @Override
public boolean threadEnter(LSMOperationType opType, boolean isMutableComponent) {
if (state == ComponentState.INACTIVE) {
throw new IllegalStateException("Trying to enter an inactive disk component");
@@ -148,6 +144,9 @@ public abstract class AbstractLSMDiskComponent extends AbstractLSMComponent impl
@Override
public void markAsValid(boolean persist) throws HyracksDataException {
ComponentUtils.markAsValid(getMetadataHolder(), persist);
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.log(Level.INFO, "Marked as valid component with id: " + getId());
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
index b0cc318..dc808ad 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndex.java
@@ -27,6 +27,8 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.ErrorCode;
@@ -71,6 +73,7 @@ import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.util.trace.ITracer;
public abstract class AbstractLSMIndex implements ILSMIndex {
+ private static final Logger LOGGER = Logger.getLogger(AbstractLSMIndex.class.getName());
protected final ILSMHarness lsmHarness;
protected final IIOManager ioManager;
protected final ILSMIOOperationScheduler ioScheduler;
@@ -142,7 +145,7 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy, ILSMOperationTracker opTracker,
ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallbackFactory ioOpCallbackFactory,
ILSMDiskComponentFactory componentFactory, ILSMDiskComponentFactory bulkLoadComponentFactory,
- boolean durable) {
+ boolean durable, ITracer tracer) {
this.ioManager = ioManager;
this.diskBufferCache = diskBufferCache;
this.fileManager = fileManager;
@@ -152,6 +155,7 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
this.componentFactory = componentFactory;
this.bulkLoadComponentFactory = bulkLoadComponentFactory;
this.durable = durable;
+ this.tracer = tracer;
lsmHarness = new ExternalIndexHarness(this, mergePolicy, opTracker, diskBufferCache.isReplicationEnabled());
isActive = false;
diskComponents = new LinkedList<>();
@@ -303,12 +307,12 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
operationalComponents.add(memoryComponents.get(cmc));
break;
case INSERT:
- addOperationalMutableComponents(operationalComponents);
+ addOperationalMutableComponents(operationalComponents, true);
operationalComponents.addAll(immutableComponents);
break;
case SEARCH:
if (memoryComponentsAllocated) {
- addOperationalMutableComponents(operationalComponents);
+ addOperationalMutableComponents(operationalComponents, false);
}
if (filterManager != null) {
for (ILSMComponent c : immutableComponents) {
@@ -375,18 +379,23 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
ioScheduler.scheduleOperation(TracedIOOperation.wrap(mergeOp, tracer));
}
- private void addOperationalMutableComponents(List<ILSMComponent> operationalComponents) {
+ private void addOperationalMutableComponents(List<ILSMComponent> operationalComponents, boolean modification) {
int cmc = currentMutableComponentId.get();
int numMutableComponents = memoryComponents.size();
for (int i = 0; i < numMutableComponents - 1; i++) {
ILSMMemoryComponent c = memoryComponents.get((cmc + i + 1) % numMutableComponents);
if (c.isReadable()) {
- // Make sure newest components are added first
+ // Make sure newest components are added first if readable
operationalComponents.add(0, c);
}
}
- // The current mutable component is always added
- operationalComponents.add(0, memoryComponents.get(cmc));
+ // The current mutable component is added if modification operation or if readable
+ // This ensures that activation of new component only happens in case of modifications
+ // and allow for controlling that without stopping search operations
+ ILSMMemoryComponent c = memoryComponents.get(cmc);
+ if (modification || c.isReadable()) {
+ operationalComponents.add(0, c);
+ }
}
@Override
@@ -421,7 +430,7 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
}
@Override
- public final synchronized void allocateMemoryComponents() throws HyracksDataException {
+ public synchronized void allocateMemoryComponents() throws HyracksDataException {
if (!isActive) {
throw HyracksDataException.create(ErrorCode.CANNOT_ALLOCATE_MEMORY_FOR_INACTIVE_INDEX);
}
@@ -652,7 +661,8 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
return filterManager;
}
- public ILSMHarness getLsmHarness() {
+ @Override
+ public ILSMHarness getHarness() {
return lsmHarness;
}
@@ -681,8 +691,15 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
public final ILSMDiskComponent flush(ILSMIOOperation operation) throws HyracksDataException {
ILSMIndexAccessor accessor = operation.getAccessor();
ILSMIndexOperationContext opCtx = accessor.getOpContext();
- return opCtx.getOperation() == IndexOperation.DELETE_MEMORY_COMPONENT ? EmptyComponent.INSTANCE
- : doFlush(operation);
+ if (opCtx.getOperation() == IndexOperation.DELETE_MEMORY_COMPONENT) {
+ return EmptyComponent.INSTANCE;
+ } else {
+ if (LOGGER.isLoggable(Level.INFO)) {
+ FlushOperation flushOp = (FlushOperation) operation;
+ LOGGER.log(Level.INFO, "Flushing component with id: " + flushOp.getFlushingComponent().getId());
+ }
+ return doFlush(operation);
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
index 065d465..0818b08 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMIndexOperationContext.java
@@ -27,14 +27,18 @@ import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
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.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
import org.apache.hyracks.storage.common.ISearchPredicate;
import org.apache.hyracks.storage.common.MultiComparator;
+import org.apache.hyracks.util.trace.ITracer;
+import org.apache.hyracks.util.trace.ITracer.Scope;
public abstract class AbstractLSMIndexOperationContext implements ILSMIndexOperationContext {
+ protected final ILSMIndex index;
protected final PermutingTupleReference indexTuple;
protected final MultiComparator filterCmp;
protected final PermutingTupleReference filterTuple;
@@ -47,10 +51,14 @@ public abstract class AbstractLSMIndexOperationContext implements ILSMIndexOpera
protected IndexOperation op;
protected boolean accessingComponents = false;
protected ISearchPredicate searchPredicate;
+ protected final ITracer tracer;
+ protected final long traceCategory;
+ private long enterExitTime = 0L;
- public AbstractLSMIndexOperationContext(int[] treeFields, int[] filterFields,
+ public AbstractLSMIndexOperationContext(ILSMIndex index, int[] treeFields, int[] filterFields,
IBinaryComparatorFactory[] filterCmpFactories, ISearchOperationCallback searchCallback,
- IModificationOperationCallback modificationCallback) {
+ IModificationOperationCallback modificationCallback, ITracer tracer) {
+ this.index = index;
this.searchCallback = searchCallback;
this.modificationCallback = modificationCallback;
this.componentHolder = new LinkedList<>();
@@ -73,6 +81,8 @@ public abstract class AbstractLSMIndexOperationContext implements ILSMIndexOpera
filterTuple = null;
allFields = null;
}
+ this.tracer = tracer;
+ this.traceCategory = tracer.getRegistry().get("op-ctx");
}
@Override
@@ -153,4 +163,32 @@ public abstract class AbstractLSMIndexOperationContext implements ILSMIndexOpera
public ISearchPredicate getSearchPredicate() {
return searchPredicate;
}
+
+ @Override
+ public final boolean isTracingEnabled() {
+ return tracer.isEnabled(traceCategory);
+ }
+
+ @Override
+ public void logPerformanceCounters(int tupleCount) {
+ if (isTracingEnabled()) {
+ tracer.instant("store-counters", traceCategory, Scope.t,
+ "{\"count\":" + tupleCount + ",\"enter-exit-duration-ns\":" + enterExitTime + "}");
+ resetCounters();
+ }
+ }
+
+ public void resetCounters() {
+ enterExitTime = 0L;
+ }
+
+ @Override
+ public void incrementEnterExitTime(long increment) {
+ enterExitTime += increment;
+ }
+
+ @Override
+ public ILSMIndex getIndex() {
+ return index;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
index 6a186dc..17dadcb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractLSMMemoryComponent.java
@@ -19,6 +19,8 @@
package org.apache.hyracks.storage.am.lsm.common.impls;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
@@ -32,6 +34,7 @@ import org.apache.hyracks.storage.common.buffercache.IBufferCache;
public abstract class AbstractLSMMemoryComponent extends AbstractLSMComponent implements ILSMMemoryComponent {
+ private static final Logger LOGGER = Logger.getLogger(AbstractLSMMemoryComponent.class.getName());
private final IVirtualBufferCache vbc;
private final AtomicBoolean isModified;
private int writerCount;
@@ -277,6 +280,9 @@ public abstract class AbstractLSMMemoryComponent extends AbstractLSMComponent im
throw new IllegalStateException(
"LSM memory component receives illegal id. Old id " + this.componentId + ", new id " + componentId);
}
+ if (LOGGER.isLoggable(Level.INFO)) {
+ LOGGER.log(Level.INFO, "Component Id was reset from " + this.componentId + " to " + componentId);
+ }
this.componentId = componentId;
LSMComponentIdUtils.persist(this.componentId, metadata);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java
new file mode 100644
index 0000000..98c1560
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ComponentReplacementContext.java
@@ -0,0 +1,215 @@
+/*
+ * 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.common.impls;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
+import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
+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.ILSMComponentId;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+import org.apache.hyracks.storage.common.ISearchOperationCallback;
+import org.apache.hyracks.storage.common.ISearchPredicate;
+import org.apache.hyracks.storage.common.MultiComparator;
+
+public class ComponentReplacementContext implements ILSMIndexOperationContext {
+ private static final Logger LOGGER = Logger.getLogger(ComponentReplacementContext.class.getName());
+ private final List<ILSMComponent> components;
+ private final List<ILSMComponent> diskComponents;
+ private final List<ILSMComponentId> replacedComponentIds;
+ private final int[] swapIndexes;
+ private int count = 0;
+ boolean accessingComponent = true;
+
+ public ComponentReplacementContext(ILSMIndex lsmIndex) {
+ components = new ArrayList<>(lsmIndex.getNumberOfAllMemoryComponents());
+ replacedComponentIds = new ArrayList<>(lsmIndex.getNumberOfAllMemoryComponents());
+ swapIndexes = new int[lsmIndex.getNumberOfAllMemoryComponents()];
+ diskComponents = new ArrayList<>(lsmIndex.getNumberOfAllMemoryComponents());
+ }
+
+ @Override
+ public void setOperation(IndexOperation newOp) throws HyracksDataException {
+ // Do nothing
+ }
+
+ @Override
+ public IndexOperation getOperation() {
+ return IndexOperation.SEARCH;
+ }
+
+ @Override
+ public void reset() {
+ accessingComponent = true;
+ components.clear();
+ diskComponents.clear();
+ replacedComponentIds.clear();
+ count = 0;
+ }
+
+ @Override
+ public List<ILSMComponent> getComponentHolder() {
+ return components;
+ }
+
+ @Override
+ public List<ILSMDiskComponent> getComponentsToBeMerged() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public ISearchOperationCallback getSearchOperationCallback() {
+ return null;
+ }
+
+ @Override
+ public IModificationOperationCallback getModificationCallback() {
+ return null;
+ }
+
+ @Override
+ public void setCurrentMutableComponentId(int currentMutableComponentId) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void setSearchPredicate(ISearchPredicate searchPredicate) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ISearchPredicate getSearchPredicate() {
+ return null;
+ }
+
+ @Override
+ public List<ILSMDiskComponent> getComponentsToBeReplicated() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isAccessingComponents() {
+ return accessingComponent;
+ }
+
+ @Override
+ public void setAccessingComponents(boolean accessingComponents) {
+ // Ignore since this is only used for component replacement
+ }
+
+ @Override
+ public PermutingTupleReference getIndexTuple() {
+ return null;
+ }
+
+ @Override
+ public PermutingTupleReference getFilterTuple() {
+ return null;
+ }
+
+ @Override
+ public MultiComparator getFilterCmp() {
+ return null;
+ }
+
+ @Override
+ public void logPerformanceCounters(int tupleCount) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void incrementEnterExitTime(long increment) {
+ // Ignore since this is only used for component replacement
+ }
+
+ public boolean proceed(List<ILSMDiskComponent> allDiskComponents) throws HyracksDataException {
+ for (int i = 0; i < components.size(); i++) {
+ replacedComponentIds.add(components.get(i).getId());
+ // ensure that disk component exists
+ boolean found = false;
+ LOGGER.log(Level.INFO, "Looking for a component with the id: " + replacedComponentIds.get(i));
+ for (int j = 0; j < allDiskComponents.size(); j++) {
+ ILSMDiskComponent dc = allDiskComponents.get(j);
+ ILSMComponentId diskComponentId = dc.getId();
+ LOGGER.log(Level.INFO, "Next disk component id: " + diskComponentId);
+ if (diskComponentId.equals(replacedComponentIds.get(i))) {
+ found = true;
+ diskComponents.add(dc);
+ break;
+ }
+ }
+ if (!found) {
+ // component has been merged?
+ LOGGER.log(Level.WARNING, "Memory Component with id = " + replacedComponentIds.get(i)
+ + " was flushed and merged before search cursor replaces it");
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public void swapIndex(int i) {
+ swapIndexes[count] = i;
+ count++;
+ }
+
+ public void prepareToEnter() {
+ components.clear();
+ components.addAll(diskComponents);
+ accessingComponent = false;
+ }
+
+ public void replace(ILSMIndexOperationContext ctx) {
+ // Called after exit and enter has been completed
+ try {
+ for (int i = 0; i < count; i++) {
+ ILSMComponent removed = ctx.getComponentHolder().remove(swapIndexes[i]);
+ if (removed.getType() == LSMComponentType.MEMORY) {
+ LOGGER.log(Level.INFO, "Removed a memory component from the search operation");
+ } else {
+ throw new IllegalStateException("Disk components can't be removed from the search operation");
+ }
+ ctx.getComponentHolder().add(swapIndexes[i], diskComponents.get(i));
+ }
+ } catch (Exception e) {
+ LOGGER.log(Level.WARNING, "Failure replacing memory components with disk components", e);
+ throw e;
+ }
+ }
+
+ @Override
+ public ILSMIndex getIndex() {
+ return null;
+ }
+
+ @Override
+ public boolean isTracingEnabled() {
+ return false;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/e5a65429/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
index dd86f65..442af56 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMComponentId.java
@@ -78,13 +78,7 @@ public class LSMComponentId implements ILSMComponentId {
return false;
}
LSMComponentId other = (LSMComponentId) obj;
- if (maxId != other.maxId) {
- return false;
- }
- if (minId != other.minId) {
- return false;
- }
- return true;
+ return maxId == other.maxId && minId == other.minId;
}
@Override