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 2018/05/21 03:35:22 UTC
[2/5] asterixdb git commit: [NO ISSUE][STO] Misc Storage Fixes and
Improvements
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 cfffbe1..a504f7e 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
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -395,16 +396,16 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
// For initial load
@Override
- public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint)
- throws HyracksDataException {
- return new LSMTwoPCBTreeBulkLoader(fillLevel, verifyInput, numElementsHint, false);
+ public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
+ Map<String, Object> parameters) throws HyracksDataException {
+ return new LSMTwoPCBTreeBulkLoader(fillLevel, verifyInput, numElementsHint, false, parameters);
}
// For transaction bulk load <- could consolidate with the above method ->
@Override
- public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint)
- throws HyracksDataException {
- return new LSMTwoPCBTreeBulkLoader(fillLevel, verifyInput, numElementsHint, true);
+ public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
+ Map<String, Object> parameters) throws HyracksDataException {
+ return new LSMTwoPCBTreeBulkLoader(fillLevel, verifyInput, numElementsHint, true, parameters);
}
// The bulk loader used for both initial loading and transaction
@@ -412,19 +413,35 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
public class LSMTwoPCBTreeBulkLoader implements IIndexBulkLoader, ITwoPCIndexBulkLoader {
private final ILSMDiskComponent component;
private final ILSMDiskComponentBulkLoader componentBulkLoader;
+ private final LoadOperation loadOp;
private final boolean isTransaction;
public LSMTwoPCBTreeBulkLoader(float fillFactor, boolean verifyInput, long numElementsHint,
- boolean isTransaction) throws HyracksDataException {
+ boolean isTransaction, Map<String, Object> parameters) throws HyracksDataException {
this.isTransaction = isTransaction;
// Create the appropriate target
+ LSMComponentFileReferences componentFileRefs;
if (isTransaction) {
- component = createTransactionTarget();
+ try {
+ componentFileRefs = fileManager.getNewTransactionFileReference();
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ component = createDiskComponent(transactionComponentFactory,
+ componentFileRefs.getInsertIndexFileReference(), null,
+ componentFileRefs.getBloomFilterFileReference(), true);
} else {
- component = createBulkLoadTarget();
+ componentFileRefs = fileManager.getRelFlushFileReference();
+ component =
+ createDiskComponent(bulkLoadComponentFactory, componentFileRefs.getInsertIndexFileReference(),
+ componentFileRefs.getDeleteIndexFileReference(),
+ componentFileRefs.getBloomFilterFileReference(), true);
}
- LoadOperation loadOp = new LoadOperation(ioOpCallback, getIndexIdentifier());
+ loadOp = new LoadOperation(componentFileRefs, ioOpCallback, getIndexIdentifier(), parameters);
+ loadOp.setNewComponent(component);
+ ioOpCallback.scheduled(loadOp);
+ ioOpCallback.beforeOperation(loadOp);
componentBulkLoader =
component.createBulkLoader(loadOp, fillFactor, verifyInput, numElementsHint, false, true, true);
}
@@ -438,16 +455,23 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
@Override
public void end() throws HyracksDataException {
- componentBulkLoader.end();
- if (component.getComponentSize() > 0) {
- if (isTransaction) {
- // Since this is a transaction component, validate and
- // deactivate. it could later be added or deleted
- component.markAsValid(durable);
- component.deactivate();
- } else {
- getHarness().addBulkLoadedComponent(component);
+ try {
+ ioOpCallback.afterOperation(loadOp);
+ componentBulkLoader.end();
+ if (component.getComponentSize() > 0) {
+ if (isTransaction) {
+ // Since this is a transaction component, validate and
+ // deactivate. it could later be added or deleted
+ component.markAsValid(durable);
+ ioOpCallback.afterFinalize(loadOp);
+ component.deactivate();
+ } else {
+ ioOpCallback.afterFinalize(loadOp);
+ getHarness().addBulkLoadedComponent(component);
+ }
}
+ } finally {
+ ioOpCallback.completed(loadOp);
}
}
@@ -459,25 +483,12 @@ public class ExternalBTree extends LSMBTree implements ITwoPCIndex {
}
@Override
- public void abort() {
+ public void abort() throws HyracksDataException {
try {
componentBulkLoader.abort();
- } catch (Exception e) {
- // Do nothing
- }
- }
-
- // This method is used to create a target for a bulk modify operation. This
- // component must then be either committed or deleted
- private ILSMDiskComponent createTransactionTarget() throws HyracksDataException {
- LSMComponentFileReferences componentFileRefs;
- try {
- componentFileRefs = fileManager.getNewTransactionFileReference();
- } catch (IOException e) {
- throw HyracksDataException.create(e);
+ } finally {
+ ioOpCallback.completed(loadOp);
}
- return createDiskComponent(transactionComponentFactory, componentFileRefs.getInsertIndexFileReference(),
- null, componentFileRefs.getBloomFilterFileReference(), true);
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 3e762c1..b727a39 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
@@ -21,6 +21,7 @@ package org.apache.hyracks.storage.am.lsm.btree.impls;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.ErrorCode;
@@ -216,16 +217,16 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
// For initial load
@Override
- public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint)
- throws HyracksDataException {
- return new LSMTwoPCBTreeWithBuddyBulkLoader(fillLevel, verifyInput, 0, false);
+ public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
+ Map<String, Object> parameters) throws HyracksDataException {
+ return new LSMTwoPCBTreeWithBuddyBulkLoader(fillLevel, verifyInput, 0, false, parameters);
}
// For transaction bulk load <- could consolidate with the above method ->
@Override
- public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint)
- throws HyracksDataException {
- return new LSMTwoPCBTreeWithBuddyBulkLoader(fillLevel, verifyInput, numElementsHint, true);
+ public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
+ Map<String, Object> parameters) throws HyracksDataException {
+ return new LSMTwoPCBTreeWithBuddyBulkLoader(fillLevel, verifyInput, numElementsHint, true, parameters);
}
@Override
@@ -494,20 +495,37 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
// modifications
public class LSMTwoPCBTreeWithBuddyBulkLoader implements IIndexBulkLoader, ITwoPCIndexBulkLoader {
private final ILSMDiskComponent component;
+ private final LoadOperation loadOp;
private final ILSMDiskComponentBulkLoader componentBulkLoader;
private final boolean isTransaction;
public LSMTwoPCBTreeWithBuddyBulkLoader(float fillFactor, boolean verifyInput, long numElementsHint,
- boolean isTransaction) throws HyracksDataException {
+ boolean isTransaction, Map<String, Object> parameters) throws HyracksDataException {
this.isTransaction = isTransaction;
// Create the appropriate target
+ LSMComponentFileReferences componentFileRefs;
if (isTransaction) {
- component = createTransactionTarget();
+ try {
+ componentFileRefs = fileManager.getNewTransactionFileReference();
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ component =
+ createDiskComponent(bulkLoadComponentFactory, componentFileRefs.getInsertIndexFileReference(),
+ componentFileRefs.getDeleteIndexFileReference(),
+ componentFileRefs.getBloomFilterFileReference(), true);
} else {
- component = createBulkLoadTarget();
+ componentFileRefs = fileManager.getRelFlushFileReference();
+ component =
+ createDiskComponent(bulkLoadComponentFactory, componentFileRefs.getInsertIndexFileReference(),
+ componentFileRefs.getDeleteIndexFileReference(),
+ componentFileRefs.getBloomFilterFileReference(), true);
}
- LoadOperation loadOp = new LoadOperation(ioOpCallback, getIndexIdentifier());
+ loadOp = new LoadOperation(componentFileRefs, ioOpCallback, getIndexIdentifier(), parameters);
+ loadOp.setNewComponent(component);
+ ioOpCallback.scheduled(loadOp);
+ ioOpCallback.beforeOperation(loadOp);
componentBulkLoader =
component.createBulkLoader(loadOp, fillFactor, verifyInput, numElementsHint, false, true, false);
}
@@ -519,16 +537,23 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
@Override
public void end() throws HyracksDataException {
- componentBulkLoader.end();
- if (component.getComponentSize() > 0) {
- if (isTransaction) {
- // Since this is a transaction component, validate and
- // deactivate. it could later be added or deleted
- component.markAsValid(durable);
- component.deactivate();
- } else {
- getHarness().addBulkLoadedComponent(component);
+ try {
+ ioOpCallback.afterOperation(loadOp);
+ componentBulkLoader.end();
+ if (component.getComponentSize() > 0) {
+ if (isTransaction) {
+ // Since this is a transaction component, validate and
+ // deactivate. it could later be added or deleted
+ component.markAsValid(durable);
+ ioOpCallback.afterFinalize(loadOp);
+ component.deactivate();
+ } else {
+ ioOpCallback.afterFinalize(loadOp);
+ getHarness().addBulkLoadedComponent(component);
+ }
}
+ } finally {
+ ioOpCallback.completed(loadOp);
}
}
@@ -538,25 +563,16 @@ public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeInd
}
@Override
- public void abort() {
- try {
- componentBulkLoader.abort();
- } catch (Exception e) {
- }
- }
-
- // This method is used to create a target for a bulk modify operation. This
- // component must then eventually be either committed or deleted
- private ILSMDiskComponent createTransactionTarget() throws HyracksDataException {
- LSMComponentFileReferences componentFileRefs;
+ public void abort() throws HyracksDataException {
try {
- componentFileRefs = fileManager.getNewTransactionFileReference();
- } catch (IOException e) {
- throw HyracksDataException.create(e);
+ try {
+ componentBulkLoader.abort();
+ } finally {
+ ioOpCallback.afterFinalize(loadOp);
+ }
+ } finally {
+ ioOpCallback.completed(loadOp);
}
- return createDiskComponent(bulkLoadComponentFactory, componentFileRefs.getInsertIndexFileReference(),
- componentFileRefs.getDeleteIndexFileReference(), componentFileRefs.getBloomFilterFileReference(),
- true);
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java
index 7acc59f..2ad4fee 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeFlushOperation.java
@@ -26,11 +26,13 @@ import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences
public class LSMBTreeFlushOperation extends FlushOperation {
private final FileReference bloomFilterFlushTarget;
+ private final LSMComponentFileReferences fileReferences;
public LSMBTreeFlushOperation(ILSMIndexAccessor accessor, FileReference flushTarget,
FileReference bloomFilterFlushTarget, ILSMIOOperationCallback callback, String indexIdentifier) {
super(accessor, flushTarget, callback, indexIdentifier);
this.bloomFilterFlushTarget = bloomFilterFlushTarget;
+ fileReferences = new LSMComponentFileReferences(target, null, bloomFilterFlushTarget);
}
public FileReference getBloomFilterTarget() {
@@ -39,6 +41,6 @@ public class LSMBTreeFlushOperation extends FlushOperation {
@Override
public LSMComponentFileReferences getComponentFiles() {
- return new LSMComponentFileReferences(target, null, bloomFilterFlushTarget);
+ return fileReferences;
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 1bfea53..fef59e7 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
@@ -148,4 +148,9 @@ public interface ILSMComponent {
* @throws HyracksDataException
*/
void schedule(LSMIOOperationType ioOperationType) throws HyracksDataException;
+
+ /**
+ * @return the number of readers inside a component
+ */
+ int getReaderCount();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentId.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentId.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentId.java
index 5662862..c3835eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentId.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentId.java
@@ -38,4 +38,14 @@ public interface ILSMComponentId {
boolean missing();
IdCompareResult compareTo(ILSMComponentId id);
+
+ /**
+ * @return the min Id
+ */
+ long getMinId();
+
+ /**
+ * @return the max Id
+ */
+ long getMaxId();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 8a041ec..c4a0352 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
@@ -169,13 +169,11 @@ public interface ILSMHarness {
* the operation context
* @param diskComponents
* the disk component to be replicated
- * @param bulkload
- * true if the components were bulk loaded, false otherwise
* @param opType
* The operation type
* @throws HyracksDataException
*/
- void scheduleReplication(ILSMIndexOperationContext ctx, List<ILSMDiskComponent> diskComponents, boolean bulkload,
+ void scheduleReplication(ILSMIndexOperationContext ctx, List<ILSMDiskComponent> diskComponents,
LSMOperationType opType) throws HyracksDataException;
/**
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperation.java
index 88dcc5a..3245455 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIOOperation.java
@@ -18,6 +18,7 @@
*/
package org.apache.hyracks.storage.am.lsm.common.api;
+import java.util.Map;
import java.util.concurrent.Callable;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -135,4 +136,16 @@ public interface ILSMIOOperation extends Callable<LSMIOOperationStatus> {
* @throws InterruptedException
*/
void sync() throws InterruptedException;
+
+ /**
+ * Add a listener for operation complete event
+ *
+ * @param listener
+ */
+ void addCompleteListener(IoOperationCompleteListener listener);
+
+ /**
+ * Get parameters passed when calling this IO operation
+ */
+ Map<String, Object> getParameters();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 1374524..5cb05a7 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
@@ -20,6 +20,7 @@
package org.apache.hyracks.storage.am.lsm.common.api;
import java.util.List;
+import java.util.Map;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation;
@@ -28,6 +29,7 @@ import org.apache.hyracks.storage.am.common.api.IIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMHarness;
import org.apache.hyracks.storage.common.IIndex;
import org.apache.hyracks.storage.common.IIndexAccessParameters;
+import org.apache.hyracks.storage.common.IIndexBulkLoader;
import org.apache.hyracks.storage.common.IIndexCursor;
import org.apache.hyracks.storage.common.ISearchPredicate;
@@ -128,7 +130,7 @@ public interface ILSMIndex extends IIndex {
boolean isCurrentMutableComponentEmpty() throws HyracksDataException;
- void scheduleReplication(ILSMIndexOperationContext ctx, List<ILSMDiskComponent> diskComponents, boolean bulkload,
+ void scheduleReplication(ILSMIndexOperationContext ctx, List<ILSMDiskComponent> diskComponents,
ReplicationOperation operation, LSMOperationType opType) throws HyracksDataException;
boolean isMemoryComponentsAllocated();
@@ -185,4 +187,28 @@ public interface ILSMIndex extends IIndex {
*/
void cleanUpFilesForFailedOperation(ILSMIOOperation operation);
+ /**
+ * @return the absolute path of the index
+ */
+ String getIndexIdentifier();
+
+ /**
+ * Create a bulk loader
+ *
+ * @param fillFactor
+ * @param verifyInput
+ * @param numElementsHint
+ * @param checkIfEmptyIndex
+ * @param parameters
+ * @return
+ * @throws HyracksDataException
+ */
+ IIndexBulkLoader createBulkLoader(float fillFactor, boolean verifyInput, long numElementsHint,
+ boolean checkIfEmptyIndex, Map<String, Object> parameters) throws HyracksDataException;
+
+ /**
+ * Reset the current memory component id to 0.
+ */
+ void resetCurrentComponentIndex();
+
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
index 34844d6..42d3ab7 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
@@ -186,13 +186,11 @@ public interface ILSMIndexAccessor extends IIndexAccessor {
*
* @param diskComponents
* the components to be replicated
- * @param bulkload
- * true if the components were bulkloaded, false otherwise
* @param opType
* the operation type
* @throws HyracksDataException
*/
- void scheduleReplication(List<ILSMDiskComponent> diskComponents, boolean bulkload, LSMOperationType opType)
+ void scheduleReplication(List<ILSMDiskComponent> diskComponents, LSMOperationType opType)
throws HyracksDataException;
/**
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ITwoPCIndex.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ITwoPCIndex.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ITwoPCIndex.java
index fc1d5f4..74a5d60 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ITwoPCIndex.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ITwoPCIndex.java
@@ -19,6 +19,7 @@
package org.apache.hyracks.storage.am.lsm.common.api;
import java.util.List;
+import java.util.Map;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.common.IIndexBulkLoader;
@@ -30,8 +31,8 @@ public interface ITwoPCIndex {
* This function is used to create a BulkLoader for a transaction that is capable of insertions and deletions
* and the bulk loaded component is hidden from the index
*/
- public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint)
- throws HyracksDataException;
+ public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
+ Map<String, Object> parameters) throws HyracksDataException;
/**
* This function is used to commit the previous transaction if it was resulted in creating any components
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IoOperationCompleteListener.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IoOperationCompleteListener.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IoOperationCompleteListener.java
new file mode 100644
index 0000000..e5ba81e
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IoOperationCompleteListener.java
@@ -0,0 +1,29 @@
+/*
+ * 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.api;
+
+public interface IoOperationCompleteListener {
+
+ /**
+ * Called when an IO operation completes
+ *
+ * @param operation
+ */
+ void completed(ILSMIOOperation operation);
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/LSMOperationType.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/LSMOperationType.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/LSMOperationType.java
index 63d2697..e200bfd 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/LSMOperationType.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/LSMOperationType.java
@@ -22,6 +22,7 @@ public enum LSMOperationType {
SEARCH,
MODIFICATION,
FORCE_MODIFICATION,
+ LOAD,
FLUSH,
MERGE,
REPLICATE,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractIoOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractIoOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractIoOperation.java
index b3252d9..fc9a362 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractIoOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractIoOperation.java
@@ -18,6 +18,10 @@
*/
package org.apache.hyracks.storage.am.lsm.common.impls;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IODeviceHandle;
import org.apache.hyracks.api.util.ExceptionUtils;
@@ -25,6 +29,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.common.api.IoOperationCompleteListener;
public abstract class AbstractIoOperation implements ILSMIOOperation {
@@ -36,6 +41,7 @@ public abstract class AbstractIoOperation implements ILSMIOOperation {
private LSMIOOperationStatus status = LSMIOOperationStatus.SUCCESS;
private ILSMDiskComponent newComponent;
private boolean completed = false;
+ private List<IoOperationCompleteListener> completeListeners;
public AbstractIoOperation(ILSMIndexAccessor accessor, FileReference target, ILSMIOOperationCallback callback,
String indexIdentifier) {
@@ -77,6 +83,7 @@ public abstract class AbstractIoOperation implements ILSMIOOperation {
@Override
public void setFailure(Throwable failure) {
+ status = LSMIOOperationStatus.FAILURE;
this.failure = ExceptionUtils.suppress(this.failure, failure);
}
@@ -107,6 +114,12 @@ public abstract class AbstractIoOperation implements ILSMIOOperation {
}
callback.completed(this);
completed = true;
+ if (completeListeners != null) {
+ for (IoOperationCompleteListener listener : completeListeners) {
+ listener.completed(this);
+ }
+ completeListeners = null;
+ }
notifyAll();
}
@@ -116,4 +129,21 @@ public abstract class AbstractIoOperation implements ILSMIOOperation {
wait();
}
}
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return accessor.getOpContext().getParameters();
+ }
+
+ @Override
+ public synchronized void addCompleteListener(IoOperationCompleteListener listener) {
+ if (completed) {
+ listener.completed(this);
+ } else {
+ if (completeListeners == null) {
+ completeListeners = new LinkedList<>();
+ }
+ completeListeners.add(listener);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 84d2fe5..574a371 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
@@ -51,4 +51,9 @@ public abstract class AbstractLSMComponent implements ILSMComponent {
public final AbstractLSMIndex getLsmIndex() {
return lsmIndex;
}
+
+ @Override
+ public int getReaderCount() {
+ return readerCount;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 a999b25..c9fb328 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
@@ -385,7 +385,8 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
List<ILSMDiskComponent> mergingComponents = ctx.getComponentsToBeMerged();
// Merge operation can fail if another merge is already scheduled on those components
// This should be guarded against by the merge policy but we still protect against here
- if (isDeactivating || mergingComponents.size() < 2 && ctx.getOperation() != IndexOperation.DELETE_COMPONENTS) {
+ if (isDeactivating
+ || (mergingComponents.size() < 2 && ctx.getOperation() != IndexOperation.DELETE_COMPONENTS)) {
return NoOpIoOperation.INSTANCE;
}
for (int i = 0; i < mergingComponents.size(); i++) {
@@ -398,7 +399,7 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
mergeCtx.setOperation(ctx.getOperation());
mergeCtx.getComponentHolder().addAll(mergingComponents);
propagateMap(ctx, mergeCtx);
- mergingComponents.stream().map(ILSMDiskComponent.class::cast).forEach(mergeCtx.getComponentsToBeMerged()::add);
+ mergingComponents.stream().forEach(mergeCtx.getComponentsToBeMerged()::add);
ILSMDiskComponent firstComponent = mergingComponents.get(0);
ILSMDiskComponent lastComponent = mergingComponents.get(mergingComponents.size() - 1);
LSMComponentFileReferences mergeFileRefs = getMergeFileReferences(firstComponent, lastComponent);
@@ -494,16 +495,27 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
@Override
public final IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
boolean checkIfEmptyIndex) throws HyracksDataException {
+ return createBulkLoader(fillLevel, verifyInput, numElementsHint, checkIfEmptyIndex, null);
+ }
+
+ @Override
+ public IIndexBulkLoader createBulkLoader(float fillFactor, boolean verifyInput, long numElementsHint,
+ boolean checkIfEmptyIndex, Map<String, Object> parameters) throws HyracksDataException {
if (checkIfEmptyIndex && !isEmptyIndex()) {
throw HyracksDataException.create(ErrorCode.LOAD_NON_EMPTY_INDEX);
}
- return createBulkLoader(fillLevel, verifyInput, numElementsHint);
+ return createBulkLoader(fillFactor, verifyInput, numElementsHint, parameters);
}
- public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint)
- throws HyracksDataException {
+ public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
+ Map<String, Object> parameters) throws HyracksDataException {
AbstractLSMIndexOperationContext opCtx = createOpContext(NoOpIndexAccessParameters.INSTANCE);
- LoadOperation loadOp = new LoadOperation(ioOpCallback, getIndexIdentifier());
+ opCtx.setParameters(parameters);
+ LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
+ LoadOperation loadOp = new LoadOperation(componentFileRefs, ioOpCallback, getIndexIdentifier(), parameters);
+ loadOp.setNewComponent(createDiskComponent(bulkLoadComponentFactory,
+ componentFileRefs.getInsertIndexFileReference(), componentFileRefs.getDeleteIndexFileReference(),
+ componentFileRefs.getBloomFilterFileReference(), true));
ioOpCallback.scheduled(loadOp);
opCtx.setIoOperation(loadOp);
return new LSMIndexDiskComponentBulkLoader(this, opCtx, fillLevel, verifyInput, numElementsHint);
@@ -679,7 +691,7 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
@Override
public void scheduleReplication(ILSMIndexOperationContext ctx, List<ILSMDiskComponent> lsmComponents,
- boolean bulkload, ReplicationOperation operation, LSMOperationType opType) throws HyracksDataException {
+ ReplicationOperation operation, LSMOperationType opType) throws HyracksDataException {
//get set of files to be replicated for this component
Set<String> componentFiles = new HashSet<>();
@@ -689,7 +701,7 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
}
ReplicationExecutionType executionType;
- if (bulkload) {
+ if (opType == LSMOperationType.LOAD) {
executionType = ReplicationExecutionType.SYNC;
} else {
executionType = ReplicationExecutionType.ASYNC;
@@ -796,6 +808,30 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
}
@Override
+ public void resetCurrentComponentIndex() {
+ synchronized (lsmHarness.getOperationTracker()) {
+ // validate no reader in any of the memory components and that all of them are INVALID
+ for (ILSMMemoryComponent c : memoryComponents) {
+ if (c.getReaderCount() > 0) {
+ throw new IllegalStateException(
+ "Attempt to reset current component index while readers are inside the components. " + c);
+ }
+ if (c.getState() != ComponentState.INACTIVE) {
+ throw new IllegalStateException(
+ "Attempt to reset current component index while a component is not INACTIVE. " + c);
+ }
+ }
+ currentMutableComponentId.set(0);
+ memoryComponents.get(0);
+ try {
+ memoryComponents.get(0).resetId(null, true);
+ } catch (HyracksDataException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+ }
+
+ @Override
public final ILSMDiskComponent flush(ILSMIOOperation operation) throws HyracksDataException {
ILSMIndexAccessor accessor = operation.getAccessor();
ILSMIndexOperationContext opCtx = accessor.getOpContext();
@@ -840,7 +876,8 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
}
}
- protected String getIndexIdentifier() {
+ @Override
+ public String getIndexIdentifier() {
return fileManager.getBaseDir().getAbsolutePath();
}
@@ -864,5 +901,4 @@ public abstract class AbstractLSMIndex implements ILSMIndex {
protected abstract ILSMDiskComponent doFlush(ILSMIOOperation operation) throws HyracksDataException;
protected abstract ILSMDiskComponent doMerge(ILSMIOOperation operation) throws HyracksDataException;
-
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 9ee1ff8..ab00dff 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
@@ -339,7 +339,9 @@ public abstract class AbstractLSMMemoryComponent extends AbstractLSMComponent im
LOGGER.log(Level.INFO, "Component Id was reset from " + this.componentId + " to " + componentId);
}
this.componentId = componentId;
- LSMComponentIdUtils.persist(this.componentId, metadata);
+ if (componentId != null) {
+ LSMComponentIdUtils.persist(this.componentId, metadata);
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
index 232b42c..6e0606a 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ChainedLSMDiskComponentBulkLoader.java
@@ -25,8 +25,9 @@ import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponentBulkLoader;
-import org.apache.hyracks.util.annotations.CriticalPath;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationStatus;
+import org.apache.hyracks.util.annotations.CriticalPath;
/**
* Class encapsulates a chain of operations, happening during an LSM disk component bulkload
@@ -51,6 +52,7 @@ public class ChainedLSMDiskComponentBulkLoader implements ILSMDiskComponentBulkL
bulkloaderChain.add(bulkloader);
}
+ @SuppressWarnings("squid:S1181")
@Override
@CriticalPath
public void add(ITupleReference tuple) throws HyracksDataException {
@@ -60,7 +62,8 @@ public class ChainedLSMDiskComponentBulkLoader implements ILSMDiskComponentBulkL
for (int i = 0; i < bulkloadersCount; i++) {
t = bulkloaderChain.get(i).add(t);
}
- } catch (Exception e) {
+ } catch (Throwable e) {
+ operation.setFailure(e);
cleanupArtifacts();
throw e;
}
@@ -69,6 +72,7 @@ public class ChainedLSMDiskComponentBulkLoader implements ILSMDiskComponentBulkL
}
}
+ @SuppressWarnings("squid:S1181")
@Override
@CriticalPath
public void delete(ITupleReference tuple) throws HyracksDataException {
@@ -78,7 +82,8 @@ public class ChainedLSMDiskComponentBulkLoader implements ILSMDiskComponentBulkL
for (int i = 0; i < bulkloadersCount; i++) {
t = bulkloaderChain.get(i).delete(t);
}
- } catch (Exception e) {
+ } catch (Throwable e) {
+ operation.setFailure(e);
cleanupArtifacts();
throw e;
}
@@ -112,6 +117,7 @@ public class ChainedLSMDiskComponentBulkLoader implements ILSMDiskComponentBulkL
@Override
public void abort() throws HyracksDataException {
+ operation.setStatus(LSMIOOperationStatus.FAILURE);
for (IChainedComponentBulkLoader lsmOperation : bulkloaderChain) {
lsmOperation.abort();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
index ced04e7..4c2ddb6 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/EmptyComponent.java
@@ -86,7 +86,7 @@ public class EmptyComponent implements ILSMDiskComponent {
@Override
public ILSMComponentId getId() {
- return LSMComponentId.MISSING_COMPONENT_ID;
+ return LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID;
}
@Override
@@ -156,4 +156,9 @@ public class EmptyComponent implements ILSMDiskComponent {
public void schedule(LSMIOOperationType ioOperationType) throws HyracksDataException {
// Do nothing
}
+
+ @Override
+ public int getReaderCount() {
+ return 0;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ExternalIndexHarness.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ExternalIndexHarness.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ExternalIndexHarness.java
index 1ca91b3..ab70ba1 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ExternalIndexHarness.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/ExternalIndexHarness.java
@@ -122,7 +122,7 @@ public class ExternalIndexHarness extends LSMHarness {
if (replicationEnabled) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add((ILSMDiskComponent) c);
- lsmIndex.scheduleReplication(null, componentsToBeReplicated, false,
+ lsmIndex.scheduleReplication(null, componentsToBeReplicated,
ReplicationOperation.DELETE, opType);
}
((ILSMDiskComponent) c).deactivateAndDestroy();
@@ -142,7 +142,7 @@ public class ExternalIndexHarness extends LSMHarness {
if (replicationEnabled) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add(newComponent);
- triggerReplication(componentsToBeReplicated, false, opType);
+ triggerReplication(componentsToBeReplicated, opType);
}
mergePolicy.diskComponentAdded(lsmIndex, fullMergeIsRequested.get());
}
@@ -200,7 +200,7 @@ public class ExternalIndexHarness extends LSMHarness {
if (replicationEnabled) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add(c);
- triggerReplication(componentsToBeReplicated, true, LSMOperationType.MERGE);
+ triggerReplication(componentsToBeReplicated, LSMOperationType.LOAD);
}
// Enter the component
enterComponent(c);
@@ -292,7 +292,7 @@ public class ExternalIndexHarness extends LSMHarness {
if (replicationEnabled) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add(diskComponent);
- lsmIndex.scheduleReplication(null, componentsToBeReplicated, false, ReplicationOperation.DELETE, null);
+ lsmIndex.scheduleReplication(null, componentsToBeReplicated, ReplicationOperation.DELETE, null);
}
diskComponent.deactivateAndDestroy();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/FlushOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/FlushOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/FlushOperation.java
index b61355f..b2a2e48 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/FlushOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/FlushOperation.java
@@ -23,11 +23,10 @@ import java.util.Objects;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
-public abstract class FlushOperation extends AbstractIoOperation implements Comparable<ILSMIOOperation> {
+public abstract class FlushOperation extends AbstractIoOperation {
public FlushOperation(ILSMIndexAccessor accessor, FileReference target, ILSMIOOperationCallback callback,
String indexIdentifier) {
@@ -65,14 +64,6 @@ public abstract class FlushOperation extends AbstractIoOperation implements Comp
}
@Override
- public int compareTo(ILSMIOOperation o) {
- if (o instanceof FlushOperation) {
- return target.getFile().getName().compareTo(((FlushOperation) o).getTarget().getFile().getName());
- }
- return -1;
- }
-
- @Override
public boolean equals(Object o) {
return (o instanceof FlushOperation)
&& Objects.equals(target.getFile().getName(), ((FlushOperation) o).target.getFile().getName());
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 442af56..c7990a6 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
@@ -25,11 +25,11 @@ public class LSMComponentId implements ILSMComponentId {
public static final long NOT_FOUND = -1;
- // Use to handle legacy datasets which do not have the component Id
- public static final ILSMComponentId MISSING_COMPONENT_ID = new LSMComponentId(NOT_FOUND, NOT_FOUND);
+ // Used to represent an empty index with no components
+ public static final LSMComponentId EMPTY_INDEX_LAST_COMPONENT_ID = new LSMComponentId(NOT_FOUND, NOT_FOUND);
// A default component id used for bulk loaded component
- public static final ILSMComponentId DEFAULT_COMPONENT_ID = new LSMComponentId(0, 0);
+ public static final LSMComponentId DEFAULT_COMPONENT_ID = new LSMComponentId(0, 0);
private long minId;
@@ -46,10 +46,12 @@ public class LSMComponentId implements ILSMComponentId {
this.maxId = maxId;
}
+ @Override
public long getMinId() {
return this.minId;
}
+ @Override
public long getMaxId() {
return this.maxId;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 53e6a57..7a87a13 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
@@ -247,7 +247,7 @@ public class LSMHarness implements ILSMHarness {
try {
//schedule a replication job to delete these inactive disk components from replicas
if (replicationEnabled) {
- lsmIndex.scheduleReplication(null, inactiveDiskComponentsToBeDeleted, false,
+ lsmIndex.scheduleReplication(null, inactiveDiskComponentsToBeDeleted,
ReplicationOperation.DELETE, opType);
}
for (ILSMDiskComponent c : inactiveDiskComponentsToBeDeleted) {
@@ -271,10 +271,12 @@ public class LSMHarness implements ILSMHarness {
// newComponent is null if the flush op. was not performed.
if (!failedOperation && newComponent != null) {
lsmIndex.addDiskComponent(newComponent);
+ // TODO: The following should also replicate component Id
+ // even if empty component
if (replicationEnabled && newComponent != EmptyComponent.INSTANCE) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add(newComponent);
- triggerReplication(componentsToBeReplicated, false, opType);
+ triggerReplication(componentsToBeReplicated, opType);
}
mergePolicy.diskComponentAdded(lsmIndex, false);
}
@@ -286,7 +288,7 @@ public class LSMHarness implements ILSMHarness {
if (replicationEnabled && newComponent != EmptyComponent.INSTANCE) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add(newComponent);
- triggerReplication(componentsToBeReplicated, false, opType);
+ triggerReplication(componentsToBeReplicated, opType);
}
mergePolicy.diskComponentAdded(lsmIndex, fullMergeIsRequested.get());
}
@@ -619,7 +621,7 @@ public class LSMHarness implements ILSMHarness {
if (replicationEnabled) {
componentsToBeReplicated.clear();
componentsToBeReplicated.add(c);
- triggerReplication(componentsToBeReplicated, true, LSMOperationType.MERGE);
+ triggerReplication(componentsToBeReplicated, LSMOperationType.LOAD);
}
mergePolicy.diskComponentAdded(lsmIndex, false);
}
@@ -630,20 +632,20 @@ public class LSMHarness implements ILSMHarness {
return opTracker;
}
- protected void triggerReplication(List<ILSMDiskComponent> lsmComponents, boolean bulkload, LSMOperationType opType)
+ protected void triggerReplication(List<ILSMDiskComponent> lsmComponents, LSMOperationType opType)
throws HyracksDataException {
ILSMIndexAccessor accessor = lsmIndex.createAccessor(NoOpIndexAccessParameters.INSTANCE);
- accessor.scheduleReplication(lsmComponents, bulkload, opType);
+ accessor.scheduleReplication(lsmComponents, opType);
}
@Override
public void scheduleReplication(ILSMIndexOperationContext ctx, List<ILSMDiskComponent> lsmComponents,
- boolean bulkload, LSMOperationType opType) throws HyracksDataException {
+ LSMOperationType opType) throws HyracksDataException {
//enter the LSM components to be replicated to prevent them from being deleted until they are replicated
if (!getAndEnterComponents(ctx, LSMOperationType.REPLICATE, false)) {
return;
}
- lsmIndex.scheduleReplication(ctx, lsmComponents, bulkload, ReplicationOperation.REPLICATE, opType);
+ lsmIndex.scheduleReplication(ctx, lsmComponents, ReplicationOperation.REPLICATE, opType);
}
@Override
@@ -740,7 +742,6 @@ public class LSMHarness implements ILSMHarness {
ILSMIOOperation ioOperation = null;
synchronized (opTracker) {
waitForFlushesAndMerges();
- ensureNoFailedFlush();
// We always start with the memory component
ILSMMemoryComponent memComponent = lsmIndex.getCurrentMemoryComponent();
deleteMemoryComponent = predicate.test(memComponent);
@@ -769,9 +770,6 @@ public class LSMHarness implements ILSMHarness {
List<ILSMDiskComponent> toBeDeleted;
synchronized (opTracker) {
waitForFlushesAndMerges();
- // Ensure that current memory component is empty and that no failed flushes happened so far
- // This is a workaround until ASTERIXDB-2106 is fixed
- ensureNoFailedFlush();
List<ILSMDiskComponent> diskComponents = lsmIndex.getDiskComponents();
for (ILSMDiskComponent component : diskComponents) {
if (predicate.test(component)) {
@@ -804,21 +802,6 @@ public class LSMHarness implements ILSMHarness {
}
}
- /**
- * This can only be called in the steady state where:
- * 1. no scheduled flushes
- * 2. no incoming data
- *
- * @throws HyracksDataException
- */
- private void ensureNoFailedFlush() throws HyracksDataException {
- for (ILSMMemoryComponent memoryComponent : lsmIndex.getMemoryComponents()) {
- if (memoryComponent.getState() == ComponentState.READABLE_UNWRITABLE) {
- throw HyracksDataException.create(ErrorCode.A_FLUSH_OPERATION_HAS_FAILED);
- }
- }
- }
-
private void waitForFlushesAndMerges() throws HyracksDataException {
while (flushingOrMerging()) {
try {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/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 416106d..10074f9 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
@@ -35,33 +35,41 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader {
boolean verifyInput, long numElementsHint) throws HyracksDataException {
this.lsmIndex = lsmIndex;
this.opCtx = opCtx;
- opCtx.getIoOperation().setNewComponent(lsmIndex.createBulkLoadTarget());
this.componentBulkLoader = opCtx.getIoOperation().getNewComponent().createBulkLoader(opCtx.getIoOperation(),
fillFactor, verifyInput, numElementsHint, false, true, true);
- lsmIndex.getIOOperationCallback().beforeOperation(opCtx.getIoOperation());
}
public ILSMDiskComponent getComponent() {
return opCtx.getIoOperation().getNewComponent();
}
+ @SuppressWarnings("squid:S1181")
@Override
public void add(ITupleReference tuple) throws HyracksDataException {
- componentBulkLoader.add(tuple);
+ try {
+ componentBulkLoader.add(tuple);
+ } catch (Throwable th) {
+ opCtx.getIoOperation().setFailure(th);
+ throw th;
+ }
}
+ @SuppressWarnings("squid:S1181")
public void delete(ITupleReference tuple) throws HyracksDataException {
- componentBulkLoader.delete(tuple);
+ try {
+ componentBulkLoader.delete(tuple);
+ } catch (Throwable th) {
+ opCtx.getIoOperation().setFailure(th);
+ throw th;
+ }
}
@Override
public void end() throws HyracksDataException {
try {
try {
+ lsmIndex.getIOOperationCallback().afterOperation(opCtx.getIoOperation());
componentBulkLoader.end();
- if (opCtx.getIoOperation().getNewComponent().getComponentSize() > 0) {
- lsmIndex.getIOOperationCallback().afterOperation(opCtx.getIoOperation());
- }
} catch (Throwable th) { // NOSONAR Must not call afterFinalize without setting failure
opCtx.getIoOperation().setStatus(LSMIOOperationStatus.FAILURE);
opCtx.getIoOperation().setFailure(th);
@@ -69,7 +77,8 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader {
} finally {
lsmIndex.getIOOperationCallback().afterFinalize(opCtx.getIoOperation());
}
- if (opCtx.getIoOperation().getNewComponent().getComponentSize() > 0) {
+ if (opCtx.getIoOperation().getStatus() == LSMIOOperationStatus.SUCCESS
+ && opCtx.getIoOperation().getNewComponent().getComponentSize() > 0) {
lsmIndex.getHarness().addBulkLoadedComponent(opCtx.getIoOperation().getNewComponent());
}
} finally {
@@ -80,7 +89,6 @@ public class LSMIndexDiskComponentBulkLoader implements IIndexBulkLoader {
@Override
public void abort() throws HyracksDataException {
opCtx.getIoOperation().setStatus(LSMIOOperationStatus.FAILURE);
- opCtx.getIoOperation().setNewComponent(null);
try {
try {
componentBulkLoader.abort();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
index f2c8d35..8412b8c 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
@@ -148,12 +148,12 @@ public class LSMTreeIndexAccessor implements ILSMIndexAccessor {
}
@Override
- public void scheduleReplication(List<ILSMDiskComponent> lsmComponents, boolean bulkload, LSMOperationType opType)
+ public void scheduleReplication(List<ILSMDiskComponent> lsmComponents, LSMOperationType opType)
throws HyracksDataException {
ctx.setOperation(IndexOperation.REPLICATE);
ctx.getComponentsToBeReplicated().clear();
ctx.getComponentsToBeReplicated().addAll(lsmComponents);
- lsmHarness.scheduleReplication(ctx, lsmComponents, bulkload, opType);
+ lsmHarness.scheduleReplication(ctx, lsmComponents, opType);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LoadOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LoadOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LoadOperation.java
index 3d63a6a..21c52d0 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LoadOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LoadOperation.java
@@ -18,13 +18,21 @@
*/
package org.apache.hyracks.storage.am.lsm.common.impls;
+import java.util.Map;
+
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
public class LoadOperation extends AbstractIoOperation {
- public LoadOperation(ILSMIOOperationCallback callback, String indexIdentifier) {
- super(null, null, callback, indexIdentifier);
+ private final LSMComponentFileReferences fileReferences;
+ private final Map<String, Object> parameters;
+
+ public LoadOperation(LSMComponentFileReferences fileReferences, ILSMIOOperationCallback callback,
+ String indexIdentifier, Map<String, Object> parameters) {
+ super(null, fileReferences.getInsertIndexFileReference(), callback, indexIdentifier);
+ this.fileReferences = fileReferences;
+ this.parameters = parameters;
}
@Override
@@ -39,11 +47,16 @@ public class LoadOperation extends AbstractIoOperation {
@Override
public LSMComponentFileReferences getComponentFiles() {
- return null;
+ return fileReferences;
}
@Override
public void sync() {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return parameters;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoOpIoOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoOpIoOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoOpIoOperation.java
index 6d7e7ec..f57c4ef 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoOpIoOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/NoOpIoOperation.java
@@ -19,6 +19,8 @@
package org.apache.hyracks.storage.am.lsm.common.impls;
+import java.util.Map;
+
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IODeviceHandle;
@@ -26,6 +28,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.common.api.IoOperationCompleteListener;
public class NoOpIoOperation implements ILSMIOOperation {
public static final NoOpIoOperation INSTANCE = new NoOpIoOperation();
@@ -113,4 +116,14 @@ public class NoOpIoOperation implements ILSMIOOperation {
// No Op
}
+ @Override
+ public void addCompleteListener(IoOperationCompleteListener listener) {
+ listener.completed(this);
+ }
+
+ @Override
+ public Map<String, Object> getParameters() {
+ return null;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/TracedIOOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/TracedIOOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/TracedIOOperation.java
index 5ada349..f1172f3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/TracedIOOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/TracedIOOperation.java
@@ -19,6 +19,8 @@
package org.apache.hyracks.storage.am.lsm.common.impls;
+import java.util.Map;
+
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.api.io.IODeviceHandle;
@@ -26,6 +28,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.common.api.IoOperationCompleteListener;
import org.apache.hyracks.util.trace.ITracer;
import org.apache.hyracks.util.trace.ITracer.Scope;
import org.apache.hyracks.util.trace.TraceUtils;
@@ -54,10 +57,7 @@ class TracedIOOperation implements ILSMIOOperation {
if (tracer.isEnabled(traceCategory)) {
tracer.instant("schedule-" + ioOpName, traceCategory, Scope.p,
"{\"path\": \"" + ioOp.getTarget().getRelativePath() + "\"}");
- }
- if (tracer.isEnabled(traceCategory)) {
- return ioOp instanceof Comparable ? new ComparableTracedIOOperation(ioOp, tracer, traceCategory)
- : new TracedIOOperation(ioOp, tracer, traceCategory);
+ return new TracedIOOperation(ioOp, tracer, traceCategory);
}
return ioOp;
}
@@ -152,33 +152,14 @@ class TracedIOOperation implements ILSMIOOperation {
public void sync() throws InterruptedException {
ioOp.sync();
}
-}
-
-class ComparableTracedIOOperation extends TracedIOOperation implements Comparable<ILSMIOOperation> {
-
- protected ComparableTracedIOOperation(ILSMIOOperation ioOp, ITracer trace, long traceCategory) {
- super(ioOp, trace, traceCategory);
- }
@Override
- public int hashCode() {
- return this.ioOp.hashCode();
+ public void addCompleteListener(IoOperationCompleteListener listener) {
+ ioOp.addCompleteListener(listener);
}
@Override
- public boolean equals(Object other) {
- return other instanceof ILSMIOOperation && compareTo((ILSMIOOperation) other) == 0;
+ public Map<String, Object> getParameters() {
+ return ioOp.getParameters();
}
-
- @Override
- public int compareTo(ILSMIOOperation other) {
- final ILSMIOOperation myIoOp = this.ioOp;
- if (myIoOp instanceof Comparable && other instanceof ComparableTracedIOOperation) {
- return ((Comparable) myIoOp).compareTo(((ComparableTracedIOOperation) other).getIoOp());
- }
- LOGGER.warn("Comparing ioOps of type " + myIoOp.getClass().getSimpleName() + " and "
- + other.getClass().getSimpleName() + " in " + getClass().getSimpleName());
- return Integer.signum(hashCode() - other.hashCode());
- }
-
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/LSMComponentIdUtils.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/LSMComponentIdUtils.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/LSMComponentIdUtils.java
index 6d4b0a7..1779527 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/LSMComponentIdUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/util/LSMComponentIdUtils.java
@@ -25,8 +25,11 @@ import org.apache.hyracks.storage.am.common.freepage.MutableArrayValueReference;
import org.apache.hyracks.storage.am.lsm.common.api.IComponentMetadata;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
public class LSMComponentIdUtils {
+ private static final Logger LOGGER = LogManager.getLogger();
private static final MutableArrayValueReference COMPONENT_ID_MIN_KEY =
new MutableArrayValueReference("Component_Id_Min".getBytes());
@@ -43,7 +46,9 @@ public class LSMComponentIdUtils {
long minId = ComponentUtils.getLong(metadata, COMPONENT_ID_MIN_KEY, LSMComponentId.NOT_FOUND, buffer);
long maxId = ComponentUtils.getLong(metadata, COMPONENT_ID_MAX_KEY, LSMComponentId.NOT_FOUND, buffer);
if (minId == LSMComponentId.NOT_FOUND || maxId == LSMComponentId.NOT_FOUND) {
- return LSMComponentId.MISSING_COMPONENT_ID;
+ LOGGER.warn("Invalid component id {} was persisted to a component metadata",
+ LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID);
+ return LSMComponentId.EMPTY_INDEX_LAST_COMPONENT_ID;
} else {
return new LSMComponentId(minId, maxId);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/10a3f21d/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
index a6471f8..8e39b62 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
@@ -104,12 +104,12 @@ public class LSMInvertedIndexAccessor implements ILSMIndexAccessor, IInvertedInd
}
@Override
- public void scheduleReplication(List<ILSMDiskComponent> lsmComponents, boolean bulkload, LSMOperationType opType)
+ public void scheduleReplication(List<ILSMDiskComponent> lsmComponents, LSMOperationType opType)
throws HyracksDataException {
ctx.setOperation(IndexOperation.REPLICATE);
ctx.getComponentsToBeReplicated().clear();
ctx.getComponentsToBeReplicated().addAll(lsmComponents);
- lsmHarness.scheduleReplication(ctx, lsmComponents, bulkload, opType);
+ lsmHarness.scheduleReplication(ctx, lsmComponents, opType);
}
@Override