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 2016/07/22 19:37:00 UTC
[1/4] asterixdb git commit: Add Test NodeController,
Test Data Generator, and Marker Logs
Repository: asterixdb
Updated Branches:
refs/heads/master 973a0d34e -> 7575785ac
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetaDataPageManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetaDataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetaDataPageManager.java
index 2550ab4..3982a3a 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetaDataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IMetaDataPageManager.java
@@ -31,12 +31,15 @@ public interface IMetaDataPageManager {
*/
/**
* Open an index file's metadata
- * @param fileId The file which to open the metadata of
+ *
+ * @param fileId
+ * The file which to open the metadata of
*/
public void open(int fileId);
/**
* Close an index file's metadata.
+ *
* @throws HyracksDataException
*/
@@ -44,7 +47,9 @@ public interface IMetaDataPageManager {
/**
* Get the location of a free page to use for index operations
- * @param metaFrame A metadata frame to use to wrap the raw page
+ *
+ * @param metaFrame
+ * A metadata frame to use to wrap the raw page
* @return A page location, or -1 if no free page could be found or allocated
* @throws HyracksDataException
*/
@@ -53,7 +58,9 @@ public interface IMetaDataPageManager {
/**
* Get the location of a block of free pages to use for index operations
- * @param metaFrame A metadata frame to use to wrap the raw page
+ *
+ * @param metaFrame
+ * A metadata frame to use to wrap the raw page
* @return The starting page location, or -1 if a block of free pages could be found or allocated
* @throws HyracksDataException
*/
@@ -62,8 +69,11 @@ public interface IMetaDataPageManager {
/**
* Add a page back to the pool of free pages within an index file
- * @param metaFrame A metadata frame to use to wrap the raw page
- * @param freePage The page which to return to the free space
+ *
+ * @param metaFrame
+ * A metadata frame to use to wrap the raw page
+ * @param freePage
+ * The page which to return to the free space
* @throws HyracksDataException
*/
@@ -74,7 +84,9 @@ public interface IMetaDataPageManager {
/**
* Gets the highest page offset according to the metadata
- * @param metaFrame A metadata frame to use to wrap the raw page
+ *
+ * @param metaFrame
+ * A metadata frame to use to wrap the raw page
* @return The locaiton of the highest offset page
* @throws HyracksDataException
*/
@@ -83,8 +95,11 @@ public interface IMetaDataPageManager {
/**
* Initializes the index metadata
- * @param metaFrame A metadata farme to use to wrap the raw page
- * @param currentMaxPage The highest page offset to consider valid
+ *
+ * @param metaFrame
+ * A metadata farme to use to wrap the raw page
+ * @param currentMaxPage
+ * The highest page offset to consider valid
* @throws HyracksDataException
*/
@@ -105,6 +120,7 @@ public interface IMetaDataPageManager {
/**
* Determines where the metadata page is located in an index file
+ *
* @return The locaiton of the metadata page, or -1 if the file appears to be corrupt
* @throws HyracksDataException
*/
@@ -113,7 +129,9 @@ public interface IMetaDataPageManager {
/**
* Initializes the metadata manager on an open index file
- * @param metaFrame A metadata frame used to wrap the raw page
+ *
+ * @param metaFrame
+ * A metadata frame used to wrap the raw page
* @throws HyracksDataException
*/
@@ -121,6 +139,7 @@ public interface IMetaDataPageManager {
/**
* Locate the filter page in an index file
+ *
* @return The offset of the filter page if it exists, or less than zero if no filter page exists yet
* @throws HyracksDataException
*/
@@ -135,7 +154,9 @@ public interface IMetaDataPageManager {
/**
* Set the cached page to manage for filter data
- * @param page The page to manage
+ *
+ * @param page
+ * The page to manage
*/
void setFilterPage(ICachedPage page);
@@ -149,4 +170,6 @@ public interface IMetaDataPageManager {
* @throws HyracksDataException
*/
long getLSNOffset() throws HyracksDataException;
+
+ public long getLastMarkerLSN() throws HyracksDataException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
index e33b949..bdfa9e1 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/ITreeIndexMetaDataFrame.java
@@ -62,4 +62,9 @@ public interface ITreeIndexMetaDataFrame {
public long getLSN();
public void setLSN(long lsn);
+
+ // Special placeholder for LSN information of a marker log. used for rollback information
+ public long getLastMarkerLSN();
+
+ public void setLastMarkerLSN(long lsn);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
index 4f9e6c4..44778a2 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexSearchOperatorNodePushable.java
@@ -199,7 +199,9 @@ public abstract class IndexSearchOperatorNodePushable extends AbstractUnaryInput
if (index != null) {
// if index == null, then the index open was not successful
try {
- appender.write(writer, true);
+ if (appender.getTupleCount() > 0) {
+ appender.write(writer, true);
+ }
} catch (Throwable th) {
closeException = new HyracksDataException(th);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
index 16fdecd..38b43c5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/frames/LIFOMetaDataFrame.java
@@ -34,38 +34,43 @@ public class LIFOMetaDataFrame implements ITreeIndexMetaDataFrame {
// Arbitrarily chosen magic integer.
protected static final int MAGIC_VALID_INT = 0x5bd1e995;
- protected static final int tupleCountOff = 0; //0
- protected static final int freeSpaceOff = tupleCountOff + 4; //4
- protected static final int maxPageOff = freeSpaceOff + 4; //8
- protected static final int levelOff = maxPageOff + 12; //20
- protected static final int nextPageOff = levelOff + 1; // 21
- protected static final int validOff = nextPageOff + 4; // 25
+ protected static final int TUPLE_COUNT_OFFSET = 0; //0
+ protected static final int FREE_SPACE_OFFSET = TUPLE_COUNT_OFFSET + 4; //4
+ protected static final int MAX_PAGE_OFFSET = FREE_SPACE_OFFSET + 4; //8
+ protected static final int LEVEL_OFFSET = MAX_PAGE_OFFSET + 12; //20
+ protected static final int NEXT_PAGE_OFFSET = LEVEL_OFFSET + 1; // 21
+ protected static final int VALID_OFFSET = NEXT_PAGE_OFFSET + 4; // 25
// The additionalFilteringPageOff is used only for LSM indexes.
// We store the page id that will be used to store the information of the the filter that is associated with a disk component.
// It is only set in the first meta page other meta pages (i.e., with level -2) have junk in the max page field.
- private static final int additionalFilteringPageOff = validOff + 4; // 29
- public static final int lsnOff = additionalFilteringPageOff + 4; // 33
+ private static final int ADDITIONAL_FILTERING_PAGE_OFFSET = VALID_OFFSET + 4; // 29
+ public static final int LSN_OFFSET = ADDITIONAL_FILTERING_PAGE_OFFSET + 4; // 33
+ private static final int LAST_MARKER_LSN_OFFSET = LSN_OFFSET + 8; // 41
+ private static final int HEADER_END_OFFSET = LAST_MARKER_LSN_OFFSET + 8;
protected ICachedPage page = null;
protected ByteBuffer buf = null;
+ @Override
public int getMaxPage() {
- return buf.getInt(maxPageOff);
+ return buf.getInt(MAX_PAGE_OFFSET);
}
+ @Override
public void setMaxPage(int maxPage) {
- buf.putInt(maxPageOff, maxPage);
+ buf.putInt(MAX_PAGE_OFFSET, maxPage);
}
+ @Override
public int getFreePage() {
- int tupleCount = buf.getInt(tupleCountOff);
+ int tupleCount = buf.getInt(TUPLE_COUNT_OFFSET);
if (tupleCount > 0) {
// return the last page from the linked list of free pages
// TODO: this is a dumb policy, but good enough for now
- int lastPageOff = buf.getInt(freeSpaceOff) - 4;
- buf.putInt(freeSpaceOff, lastPageOff);
- buf.putInt(tupleCountOff, tupleCount - 1);
+ int lastPageOff = buf.getInt(FREE_SPACE_OFFSET) - 4;
+ buf.putInt(FREE_SPACE_OFFSET, lastPageOff);
+ buf.putInt(TUPLE_COUNT_OFFSET, tupleCount - 1);
return buf.getInt(lastPageOff);
} else {
return -1;
@@ -75,26 +80,28 @@ public class LIFOMetaDataFrame implements ITreeIndexMetaDataFrame {
// must be checked before adding free page
// user of this class is responsible for getting a free page as a new meta
// page, latching it, etc. if there is no space on this page
+ @Override
public boolean hasSpace() {
- return buf.getInt(freeSpaceOff) + 4 < buf.capacity();
+ return buf.getInt(FREE_SPACE_OFFSET) + 4 < buf.capacity();
}
// no bounds checking is done, there must be free space
+ @Override
public void addFreePage(int freePage) {
- int freeSpace = buf.getInt(freeSpaceOff);
+ int freeSpace = buf.getInt(FREE_SPACE_OFFSET);
buf.putInt(freeSpace, freePage);
- buf.putInt(freeSpaceOff, freeSpace + 4);
- buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
+ buf.putInt(FREE_SPACE_OFFSET, freeSpace + 4);
+ buf.putInt(TUPLE_COUNT_OFFSET, buf.getInt(TUPLE_COUNT_OFFSET) + 1);
}
@Override
public byte getLevel() {
- return buf.get(levelOff);
+ return buf.get(LEVEL_OFFSET);
}
@Override
public void setLevel(byte level) {
- buf.put(levelOff, level);
+ buf.put(LEVEL_OFFSET, level);
}
@Override
@@ -110,56 +117,67 @@ public class LIFOMetaDataFrame implements ITreeIndexMetaDataFrame {
@Override
public void initBuffer(byte level) {
- buf.putInt(tupleCountOff, 0);
- buf.putInt(freeSpaceOff, lsnOff + 8);
- buf.putInt(maxPageOff, 0);
- buf.put(levelOff, level);
- buf.putInt(nextPageOff, -1);
- buf.putInt(additionalFilteringPageOff, -1);
+ buf.putInt(TUPLE_COUNT_OFFSET, 0);
+ buf.putInt(FREE_SPACE_OFFSET, HEADER_END_OFFSET);
+ buf.putInt(MAX_PAGE_OFFSET, 0);
+ buf.put(LEVEL_OFFSET, level);
+ buf.putInt(NEXT_PAGE_OFFSET, -1);
+ buf.putInt(ADDITIONAL_FILTERING_PAGE_OFFSET, -1);
+ buf.putLong(LAST_MARKER_LSN_OFFSET, -1L);
setValid(false);
}
@Override
public int getNextPage() {
- return buf.getInt(nextPageOff);
+ return buf.getInt(NEXT_PAGE_OFFSET);
}
@Override
public void setNextPage(int nextPage) {
- buf.putInt(nextPageOff, nextPage);
+ buf.putInt(NEXT_PAGE_OFFSET, nextPage);
}
@Override
public boolean isValid() {
- return buf.getInt(validOff) == MAGIC_VALID_INT;
+ return buf.getInt(VALID_OFFSET) == MAGIC_VALID_INT;
}
@Override
public void setValid(boolean isValid) {
if (isValid) {
- buf.putInt(validOff, MAGIC_VALID_INT);
+ buf.putInt(VALID_OFFSET, MAGIC_VALID_INT);
} else {
- buf.putInt(validOff, 0);
+ buf.putInt(VALID_OFFSET, 0);
}
}
@Override
public long getLSN() {
- return buf.getLong(lsnOff);
+ return buf.getLong(LSN_OFFSET);
}
@Override
public void setLSN(long lsn) {
- buf.putLong(lsnOff, lsn);
+ buf.putLong(LSN_OFFSET, lsn);
+ }
+
+ @Override
+ public long getLastMarkerLSN() {
+ return buf.getLong(LAST_MARKER_LSN_OFFSET);
+ }
+
+ @Override
+ public void setLastMarkerLSN(long lsn) {
+ buf.putLong(LAST_MARKER_LSN_OFFSET, lsn);
}
@Override
public int getLSMComponentFilterPageId() {
- return buf.getInt(additionalFilteringPageOff);
+ return buf.getInt(ADDITIONAL_FILTERING_PAGE_OFFSET);
}
@Override
public void setLSMComponentFilterPageId(int filterPage) {
- buf.putInt(additionalFilteringPageOff, filterPage);
+ buf.putInt(ADDITIONAL_FILTERING_PAGE_OFFSET, filterPage);
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
index 468654a..4d56a80 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/freepage/LinkedMetaDataPageManager.java
@@ -361,8 +361,9 @@ public class LinkedMetaDataPageManager implements IMetaDataPageManager {
*/
@Override
public int getFirstMetadataPage() throws HyracksDataException {
- if (headPage != IBufferCache.INVALID_PAGEID)
+ if (headPage != IBufferCache.INVALID_PAGEID) {
return headPage;
+ }
ITreeIndexMetaDataFrame metaFrame = metaDataFrameFactory.createFrame();
@@ -472,8 +473,29 @@ public class LinkedMetaDataPageManager implements IMetaDataPageManager {
public long getLSNOffset() throws HyracksDataException {
int metadataPageNum = getFirstMetadataPage();
if (metadataPageNum != IBufferCache.INVALID_PAGEID) {
- return (metadataPageNum * bufferCache.getPageSize()) + LIFOMetaDataFrame.lsnOff;
+ return (metadataPageNum * (long) bufferCache.getPageSize()) + LIFOMetaDataFrame.LSN_OFFSET;
}
return IMetaDataPageManager.INVALID_LSN_OFFSET;
}
+
+ @Override
+ public long getLastMarkerLSN() throws HyracksDataException {
+ ICachedPage metaNode;
+ if (!appendOnly || (appendOnly && confiscatedMetaNode == null)) {
+ metaNode = bufferCache.pin(BufferedFileHandle.getDiskPageId(fileId, getFirstMetadataPage()), false);
+ } else {
+ metaNode = confiscatedMetaNode;
+ }
+ ITreeIndexMetaDataFrame metaFrame = metaDataFrameFactory.createFrame();
+ metaNode.acquireReadLatch();
+ try {
+ metaFrame.setPage(metaNode);
+ return metaFrame.getLastMarkerLSN();
+ } finally {
+ metaNode.releaseReadLatch();
+ if (!appendOnly || (appendOnly && confiscatedMetaNode == null)) {
+ bufferCache.unpin(metaNode);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 9e75360..6d673a9 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
@@ -129,8 +129,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
++i;
}
componentFactory = new LSMBTreeDiskComponentFactory(diskBTreeFactory, bloomFilterFactory, filterFactory);
- bulkLoadComponentFactory = new LSMBTreeDiskComponentFactory(bulkLoadBTreeFactory, bloomFilterFactory,
- filterFactory);
+ bulkLoadComponentFactory =
+ new LSMBTreeDiskComponentFactory(bulkLoadBTreeFactory, bloomFilterFactory, filterFactory);
this.needKeyDupCheck = needKeyDupCheck;
this.btreeFields = btreeFields;
this.hasBloomFilter = needKeyDupCheck;
@@ -184,9 +184,9 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
for (LSMComponentFileReferences lsmComonentFileReference : validFileReferences) {
LSMBTreeDiskComponent component;
try {
- component = createDiskComponent(componentFactory,
- lsmComonentFileReference.getInsertIndexFileReference(),
- lsmComonentFileReference.getBloomFilterFileReference(), false);
+ component =
+ createDiskComponent(componentFactory, lsmComonentFileReference.getInsertIndexFileReference(),
+ lsmComonentFileReference.getBloomFilterFileReference(), false);
} catch (IndexException e) {
throw new HyracksDataException(e);
}
@@ -424,7 +424,7 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
ILSMIndexAccessorInternal flushAccessor = new LSMBTreeAccessor(lsmHarness, opCtx);
ioScheduler.scheduleOperation(new LSMBTreeFlushOperation(flushAccessor, flushingComponent,
componentFileRefs.getInsertIndexFileReference(), componentFileRefs.getBloomFilterFileReference(),
- callback, fileManager.getBaseDir()));
+ callback, fileManager.getBaseDir(), flushingComponent.getMostRecentMarkerLSN()));
}
@Override
@@ -489,7 +489,7 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
filterManager.updateFilterInfo(component.getLSMComponentFilter(), filterTuples);
filterManager.writeFilterInfo(component.getLSMComponentFilter(), component.getBTree());
}
-
+ component.setMostRecentMarkerLSN(flushOp.getPrevMarkerLSN());
bulkLoader.end();
return component;
@@ -511,8 +511,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
BTree lastBTree = ((LSMBTreeDiskComponent) mergingComponents.get(mergingComponents.size() - 1)).getBTree();
FileReference firstFile = firstBTree.getFileReference();
FileReference lastFile = lastBTree.getFileReference();
- LSMComponentFileReferences relMergeFileRefs = fileManager
- .getRelMergeFileReference(firstFile.getFile().getName(), lastFile.getFile().getName());
+ LSMComponentFileReferences relMergeFileRefs =
+ fileManager.getRelMergeFileReference(firstFile.getFile().getName(), lastFile.getFile().getName());
ILSMIndexAccessorInternal accessor = new LSMBTreeAccessor(lsmHarness, opCtx);
ioScheduler.scheduleOperation(new LSMBTreeMergeOperation(accessor, mergingComponents, cursor,
relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getBloomFilterFileReference(),
@@ -542,8 +542,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
LSMBTreeDiskComponent mergedComponent = createDiskComponent(componentFactory, mergeOp.getBTreeMergeTarget(),
mergeOp.getBloomFilterMergeTarget(), true);
- IIndexBulkLoader bulkLoader = mergedComponent.getBTree().createBulkLoader(1.0f, false, numElements, false,
- true);
+ IIndexBulkLoader bulkLoader =
+ mergedComponent.getBTree().createBulkLoader(1.0f, false, numElements, false, true);
IIndexBulkLoader builder = null;
if (hasBloomFilter) {
builder = mergedComponent.getBloomFilter().createBuilder(numElements, bloomFilterSpec.getNumHashes(),
@@ -574,6 +574,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
filterManager.writeFilterInfo(mergedComponent.getLSMComponentFilter(), mergedComponent.getBTree());
}
+ mergedComponent
+ .setMostRecentMarkerLSN(mergedComponents.get(mergedComponents.size() - 1).getMostRecentMarkerLSN());
bulkLoader.end();
return mergedComponent;
@@ -581,7 +583,7 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
protected LSMBTreeDiskComponent createDiskComponent(LSMBTreeDiskComponentFactory factory,
FileReference btreeFileRef, FileReference bloomFilterFileRef, boolean createComponent)
- throws HyracksDataException, IndexException {
+ throws HyracksDataException, IndexException {
// Create new BTree instance.
LSMBTreeDiskComponent component = (LSMBTreeDiskComponent) factory
.createLSMComponentInstance(new LSMComponentFileReferences(btreeFileRef, null, bloomFilterFileRef));
@@ -595,6 +597,11 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
if (component.getLSMComponentFilter() != null && !createComponent) {
filterManager.readFilterInfo(component.getLSMComponentFilter(), component.getBTree());
}
+
+ if (!createComponent) {
+ component.readMostRecentMarkerLSN(component.getBTree());
+ }
+
return component;
}
@@ -651,8 +658,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
if (hasBloomFilter) {
int maxBucketsPerElement = BloomCalculations.maxBucketsPerElement(numElementsHint);
- BloomFilterSpecification bloomFilterSpec = BloomCalculations.computeBloomSpec(maxBucketsPerElement,
- bloomFilterFalsePositiveRate);
+ BloomFilterSpecification bloomFilterSpec =
+ BloomCalculations.computeBloomSpec(maxBucketsPerElement, bloomFilterFalsePositiveRate);
builder = ((LSMBTreeDiskComponent) component).getBloomFilter().createBuilder(numElementsHint,
bloomFilterSpec.getNumHashes(), bloomFilterSpec.getNumBucketsPerElements());
} else {
@@ -727,7 +734,7 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
filterManager.writeFilterInfo(component.getLSMComponentFilter(),
((LSMBTreeDiskComponent) component).getBTree());
}
-
+ component.setMostRecentMarkerLSN(-1L);
bulkLoader.end();
if (isEmptyComponent) {
@@ -792,36 +799,36 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
@Override
public ITreeIndexFrameFactory getInteriorFrameFactory() {
- LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) memoryComponents
- .get(currentMutableComponentId.get());
+ LSMBTreeMemoryComponent mutableComponent =
+ (LSMBTreeMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
return mutableComponent.getBTree().getInteriorFrameFactory();
}
@Override
public int getFieldCount() {
- LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) memoryComponents
- .get(currentMutableComponentId.get());
+ LSMBTreeMemoryComponent mutableComponent =
+ (LSMBTreeMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
return mutableComponent.getBTree().getFieldCount();
}
@Override
public int getFileId() {
- LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) memoryComponents
- .get(currentMutableComponentId.get());
+ LSMBTreeMemoryComponent mutableComponent =
+ (LSMBTreeMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
return mutableComponent.getBTree().getFileId();
}
@Override
public IMetaDataPageManager getMetaManager() {
- LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) memoryComponents
- .get(currentMutableComponentId.get());
+ LSMBTreeMemoryComponent mutableComponent =
+ (LSMBTreeMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
return mutableComponent.getBTree().getMetaManager();
}
@Override
public ITreeIndexFrameFactory getLeafFrameFactory() {
- LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) memoryComponents
- .get(currentMutableComponentId.get());
+ LSMBTreeMemoryComponent mutableComponent =
+ (LSMBTreeMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
return mutableComponent.getBTree().getLeafFrameFactory();
}
@@ -838,8 +845,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
@Override
public int getRootPageId() {
- LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) memoryComponents
- .get(currentMutableComponentId.get());
+ LSMBTreeMemoryComponent mutableComponent =
+ (LSMBTreeMemoryComponent) memoryComponents.get(currentMutableComponentId.get());
return mutableComponent.getBTree().getRootPageId();
}
@@ -892,11 +899,21 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
if (memoryComponentsAllocated) {
return;
}
+ long markerLSN = -1L;
+ if (!diskComponents.isEmpty()) {
+ markerLSN = diskComponents.get(diskComponents.size() - 1).getMostRecentMarkerLSN();
+ } else {
+ // Needed in case a marker was added before any record
+ if (memoryComponents != null && !memoryComponents.isEmpty()) {
+ markerLSN = memoryComponents.get(0).getMostRecentMarkerLSN();
+ }
+ }
for (ILSMComponent c : memoryComponents) {
LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) c;
((IVirtualBufferCache) mutableComponent.getBTree().getBufferCache()).open();
mutableComponent.getBTree().create();
mutableComponent.getBTree().activate();
+ mutableComponent.setMostRecentMarkerLSN(markerLSN);
}
memoryComponentsAllocated = true;
}
@@ -946,4 +963,14 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
memoryComponentsAllocated = false;
}
}
+
+ public synchronized long getMostRecentMarkerLSN() throws HyracksDataException {
+ if (!isPrimaryIndex()) {
+ throw new HyracksDataException("Markers are only supported for primary indexes");
+ }
+ LSMBTreeOpContext opCtx = createOpContext(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
+ opCtx.setOperation(IndexOperation.SEARCH);
+ getOperationalComponents(opCtx);
+ return !opCtx.getComponentHolder().isEmpty() ? opCtx.getComponentHolder().get(0).getMostRecentMarkerLSN() : -1L;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
index c43590b..424dfd3 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeDiskComponent.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.storage.am.lsm.btree.impls;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilter;
import org.apache.hyracks.storage.am.btree.impls.BTree;
+import org.apache.hyracks.storage.am.common.api.IMetaDataPageManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFilter;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractDiskLSMComponent;
@@ -62,4 +63,9 @@ public class LSMBTreeDiskComponent extends AbstractDiskLSMComponent {
public int getFileReferenceCount() {
return btree.getBufferCache().getFileReferenceCount(btree.getFileId());
}
+
+ public void readMostRecentMarkerLSN(BTree treeIndex) throws HyracksDataException {
+ IMetaDataPageManager treeMetaManager = treeIndex.getMetaManager();
+ mostRecentMarkerLSN = treeMetaManager.getLastMarkerLSN();
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 a30527d..f82a1b0 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
@@ -39,16 +39,18 @@ public class LSMBTreeFlushOperation implements ILSMIOOperation, Comparable<LSMBT
private final FileReference bloomFilterFlushTarget;
private final ILSMIOOperationCallback callback;
private final String indexIdentifier;
+ private final long prevMarkerLSN;
public LSMBTreeFlushOperation(ILSMIndexAccessorInternal accessor, ILSMComponent flushingComponent,
FileReference btreeFlushTarget, FileReference bloomFilterFlushTarget, ILSMIOOperationCallback callback,
- String indexIdentifier) {
+ String indexIdentifier, long prevMarkerLSN) {
this.accessor = accessor;
this.flushingComponent = flushingComponent;
this.btreeFlushTarget = btreeFlushTarget;
this.bloomFilterFlushTarget = bloomFilterFlushTarget;
this.callback = callback;
this.indexIdentifier = indexIdentifier;
+ this.prevMarkerLSN = prevMarkerLSN;
}
@Override
@@ -107,4 +109,8 @@ public class LSMBTreeFlushOperation implements ILSMIOOperation, Comparable<LSMBT
public int compareTo(LSMBTreeFlushOperation o) {
return btreeFlushTarget.getFile().getName().compareTo(o.getBTreeFlushTarget().getFile().getName());
}
+
+ public long getPrevMarkerLSN() {
+ return prevMarkerLSN;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeMemoryComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeMemoryComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeMemoryComponent.java
index dbf47a1..1a103f7 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeMemoryComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeMemoryComponent.java
@@ -29,11 +29,16 @@ public class LSMBTreeMemoryComponent extends AbstractMemoryLSMComponent {
private final BTree btree;
- public LSMBTreeMemoryComponent(BTree btree, IVirtualBufferCache vbc, boolean isActive, ILSMComponentFilter filter) {
- super(vbc, isActive, filter);
+ public LSMBTreeMemoryComponent(BTree btree, IVirtualBufferCache vbc, boolean isActive, ILSMComponentFilter filter,
+ long mostRecentMarkerLSN) {
+ super(vbc, isActive, filter, mostRecentMarkerLSN);
this.btree = btree;
}
+ public LSMBTreeMemoryComponent(BTree btree, IVirtualBufferCache vbc, boolean isActive, ILSMComponentFilter filter) {
+ this(btree, vbc, isActive, filter, -1L);
+ }
+
public BTree getBTree() {
return btree;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 b847876..a888dd5 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
@@ -46,4 +46,8 @@ public interface ILSMComponent {
public ComponentState getState();
public ILSMComponentFilter getLSMComponentFilter();
+
+ public void setMostRecentMarkerLSN(long lsn);
+
+ public long getMostRecentMarkerLSN();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualMetaDataPageManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualMetaDataPageManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualMetaDataPageManager.java
index c83cbf2..158c68f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualMetaDataPageManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/freepage/VirtualMetaDataPageManager.java
@@ -67,10 +67,12 @@ public class VirtualMetaDataPageManager implements IVirtualMetaDataPageManager {
return NullMetadataFrameFactory.INSTANCE;
}
+ @Override
public int getCapacity() {
return capacity - 2;
}
+ @Override
public void reset() {
currentPageId.set(1);
}
@@ -157,10 +159,12 @@ public class VirtualMetaDataPageManager implements IVirtualMetaDataPageManager {
// Method doesn't make sense for this free page manager.
}
+ @Override
public void setFilterPage(ICachedPage page) {
// Method doesn't make sense for this free page manager.
}
+ @Override
public ICachedPage getFilterPage() {
return null;
}
@@ -174,4 +178,10 @@ public class VirtualMetaDataPageManager implements IVirtualMetaDataPageManager {
public long getLSNOffset() throws HyracksDataException {
return IMetaDataPageManager.INVALID_LSN_OFFSET;
}
+
+ @Override
+ public long getLastMarkerLSN() throws HyracksDataException {
+ // Method doesn't make sense for this free page manager.
+ return -1L;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 b2c55dc..c678878 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
@@ -26,12 +26,18 @@ public abstract class AbstractLSMComponent implements ILSMComponent {
protected ComponentState state;
protected int readerCount;
protected final ILSMComponentFilter filter;
+ protected long mostRecentMarkerLSN;
- public AbstractLSMComponent(ILSMComponentFilter filter) {
+ public AbstractLSMComponent(ILSMComponentFilter filter, long mostRecentMarkerLSN) {
this.filter = filter;
+ this.mostRecentMarkerLSN = mostRecentMarkerLSN;
readerCount = 0;
}
+ public AbstractLSMComponent(ILSMComponentFilter filter) {
+ this(filter, -1L);
+ }
+
public AbstractLSMComponent() {
this(null);
}
@@ -45,4 +51,14 @@ public abstract class AbstractLSMComponent implements ILSMComponent {
public ILSMComponentFilter getLSMComponentFilter() {
return filter;
}
+
+ @Override
+ public long getMostRecentMarkerLSN() {
+ return mostRecentMarkerLSN;
+ }
+
+ @Override
+ public void setMostRecentMarkerLSN(long lsn) {
+ this.mostRecentMarkerLSN = lsn;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 440ad31..cb3a2db 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
@@ -275,8 +275,8 @@ public abstract class AbstractLSMIndex implements ILSMIndexInternal {
}
//create replication job and submit it
- LSMIndexReplicationJob job = new LSMIndexReplicationJob(this, ctx, componentFiles, operation, executionType,
- opType);
+ LSMIndexReplicationJob job =
+ new LSMIndexReplicationJob(this, ctx, componentFiles, operation, executionType, opType);
try {
diskBufferCache.getIOReplicationManager().submitJob(job);
} catch (IOException e) {
@@ -296,4 +296,8 @@ public abstract class AbstractLSMIndex implements ILSMIndexInternal {
public boolean isDurable() {
return durable;
}
+
+ public ILSMComponent getCurrentMemoryComponent() {
+ return memoryComponents.get(currentMutableComponentId.get());
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractMemoryLSMComponent.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractMemoryLSMComponent.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractMemoryLSMComponent.java
index 4b8fc9b..500996f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractMemoryLSMComponent.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/AbstractMemoryLSMComponent.java
@@ -32,8 +32,9 @@ public abstract class AbstractMemoryLSMComponent extends AbstractLSMComponent {
private int writerCount;
private boolean requestedToBeActive;
- public AbstractMemoryLSMComponent(IVirtualBufferCache vbc, boolean isActive, ILSMComponentFilter filter) {
- super(filter);
+ public AbstractMemoryLSMComponent(IVirtualBufferCache vbc, boolean isActive, ILSMComponentFilter filter,
+ long mostRecentMarkerLSN) {
+ super(filter, mostRecentMarkerLSN);
this.vbc = vbc;
writerCount = 0;
if (isActive) {
@@ -44,6 +45,10 @@ public abstract class AbstractMemoryLSMComponent extends AbstractLSMComponent {
isModified = new AtomicBoolean();
}
+ public AbstractMemoryLSMComponent(IVirtualBufferCache vbc, boolean isActive, ILSMComponentFilter filter) {
+ this(vbc, isActive, filter, -1L);
+ }
+
public AbstractMemoryLSMComponent(IVirtualBufferCache vbc, boolean isActive) {
this(vbc, isActive, null);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 8ddab88..896d513 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
@@ -172,7 +172,11 @@ public class LSMHarness implements ILSMHarness {
lsmIndex.getIOOperationCallback().beforeOperation(LSMOperationType.FLUSH);
// Changing the flush status should *always* precede changing the mutable component.
lsmIndex.changeFlushStatusForCurrentMutableCompoent(false);
+ // Flushing! => carry over the marker lsn to the next component
+ long mostRecentMarkerLSN =
+ ((AbstractLSMIndex) lsmIndex).getCurrentMemoryComponent().getMostRecentMarkerLSN();
lsmIndex.changeMutableComponent();
+ ((AbstractLSMIndex) lsmIndex).getCurrentMemoryComponent().setMostRecentMarkerLSN(mostRecentMarkerLSN);
// 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();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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 3cca96e..92155e4 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
@@ -133,7 +133,8 @@ public abstract class LSMTreeIndexAccessor implements ILSMIndexAccessorInternal
}
@Override
- public void scheduleReplication(List<ILSMComponent> lsmComponents, boolean bulkload, LSMOperationType opType) throws HyracksDataException {
+ public void scheduleReplication(List<ILSMComponent> lsmComponents, boolean bulkload, LSMOperationType opType)
+ throws HyracksDataException {
ctx.setOperation(IndexOperation.REPLICATE);
ctx.getComponentsToBeReplicated().clear();
ctx.getComponentsToBeReplicated().addAll(lsmComponents);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
index f2ec64b..aa0a7bd 100644
--- a/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-test-support/src/main/java/org/apache/hyracks/test/support/TestTaskContext.java
@@ -42,7 +42,6 @@ public class TestTaskContext implements IHyracksTaskContext {
private final TestJobletContext jobletContext;
private final TaskAttemptId taskId;
private WorkspaceFileFactory fileFactory;
-
private Map<Object, IStateObject> stateObjectMap = new HashMap<>();
private Object sharedObject;
@@ -149,12 +148,12 @@ public class TestTaskContext implements IHyracksTaskContext {
}
@Override
- public Object getSharedObject() {
- return sharedObject;
+ public void setSharedObject(Object object) {
+ sharedObject = object;
}
@Override
- public void setSharedObject(Object sharedObject) {
- this.sharedObject = sharedObject;
+ public Object getSharedObject() {
+ return sharedObject;
}
}
[2/4] asterixdb git commit: Add Test NodeController,
Test Data Generator, and Marker Logs
Posted by am...@apache.org.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedWithMetaDataFlowController.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedWithMetaDataFlowController.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedWithMetaDataFlowController.java
index 44aac60..790289a 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedWithMetaDataFlowController.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedWithMetaDataFlowController.java
@@ -32,8 +32,8 @@ public class FeedWithMetaDataFlowController<T> extends FeedRecordDataFlowControl
public FeedWithMetaDataFlowController(IHyracksTaskContext ctx, FeedTupleForwarder tupleForwarder,
FeedLogManager feedLogManager, int numOfOutputFields, IRecordWithMetadataParser<T> dataParser,
- IRecordReader<T> recordReader) throws HyracksDataException {
- super(ctx, tupleForwarder, feedLogManager, numOfOutputFields, dataParser, recordReader);
+ IRecordReader<T> recordReader, boolean sendMarker) throws HyracksDataException {
+ super(ctx, tupleForwarder, feedLogManager, numOfOutputFields, dataParser, recordReader, sendMarker);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
index 2e1c83f..6159908 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FeedRuntimeInputHandler.java
@@ -72,11 +72,12 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
IFrameWriter writer, FeedPolicyAccessor fpa, FrameTupleAccessor fta, ConcurrentFramePool framePool)
throws HyracksDataException {
this.writer = writer;
+
this.spiller =
- new FrameSpiller(ctx,
+ fpa.spillToDiskOnCongestion() ? new FrameSpiller(ctx,
connectionId.getFeedId() + "_" + connectionId.getDatasetName() + "_"
+ runtimeId.getFeedRuntimeType() + "_" + runtimeId.getPartition(),
- fpa.getMaxSpillOnDisk());
+ fpa.getMaxSpillOnDisk()) : null;
this.exceptionHandler = new FeedExceptionHandler(ctx, fta);
this.fpa = fpa;
this.framePool = framePool;
@@ -122,7 +123,9 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
LOGGER.log(Level.WARNING, th.getMessage(), th);
}
try {
- spiller.close();
+ if (spiller != null) {
+ spiller.close();
+ }
} catch (Throwable th) {
LOGGER.log(Level.WARNING, th.getMessage(), th);
}
@@ -459,34 +462,34 @@ public class FeedRuntimeInputHandler extends AbstractUnaryInputUnaryOutputOperat
frame = inbox.poll();
if (frame == null) {
// Memory queue is empty. Check spill
- frame = spiller.next();
- while (frame != null) {
- if (consume(frame) != null) {
- // We don't release the frame since this is a spill frame that we didn't get from memory
- // manager
- return;
- }
+ if (spiller != null) {
frame = spiller.next();
+ while (frame != null) {
+ if (consume(frame) != null) {
+ // We don't release the frame since this is a spill frame that we didn't get from memory
+ // manager
+ return;
+ }
+ frame = spiller.next();
+ }
}
writer.flush();
// At this point. We consumed all memory and spilled
// We can't assume the next will be in memory. what if there is 0 memory?
synchronized (mutex) {
frame = inbox.poll();
- if (frame == null) {
- // Nothing in memory
- if (spiller.switchToMemory()) {
- if (poisoned) {
- break;
- }
- if (DEBUG) {
- LOGGER.info("Consumer is going to sleep");
- }
- // Nothing in disk
- mutex.wait();
- if (DEBUG) {
- LOGGER.info("Consumer is waking up");
- }
+ // Nothing in memory
+ if (frame == null && (spiller == null || spiller.switchToMemory())) {
+ if (poisoned) {
+ break;
+ }
+ if (DEBUG) {
+ LOGGER.info("Consumer is going to sleep");
+ }
+ // Nothing in disk
+ mutex.wait();
+ if (DEBUG) {
+ LOGGER.info("Consumer is waking up");
}
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FrameSpiller.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FrameSpiller.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FrameSpiller.java
index a2f19bb..09e03a3 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FrameSpiller.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/dataflow/FrameSpiller.java
@@ -28,10 +28,10 @@ import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.ArrayDeque;
+import org.apache.asterix.common.utils.StoragePathUtil;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.log4j.Level;
import org.apache.log4j.Logger;
/**
@@ -49,28 +49,30 @@ public class FrameSpiller {
private final String fileNamePrefix;
private final ArrayDeque<File> files = new ArrayDeque<>();
private final VSizeFrame frame;
- private final int budget; // Max current frames in disk allowed
- private BufferedOutputStream bos; // Current output stream
- private BufferedInputStream bis; // Current input stream
- private File currentWriteFile; // Current write file
- private File currentReadFile; // Current read file
- private int currentWriteCount = 0; // Current file write count
- private int currentReadCount = 0; // Current file read count
- private int totalWriteCount = 0; // Total frames spilled
- private int totalReadCount = 0; // Total frames read
- private int fileCount = 0; // How many spill files?
+ private final int budget; // Max current frames in disk allowed
+ private BufferedOutputStream bos; // Current output stream
+ private BufferedInputStream bis; // Current input stream
+ private File currentWriteFile; // Current write file
+ private File currentReadFile; // Current read file
+ private int currentWriteCount = 0; // Current file write count
+ private int currentReadCount = 0; // Current file read count
+ private int totalWriteCount = 0; // Total frames spilled
+ private int totalReadCount = 0; // Total frames read
+ private int fileCount = 0; // How many spill files?
public FrameSpiller(IHyracksTaskContext ctx, String fileNamePrefix, long budgetInBytes)
throws HyracksDataException {
this.frame = new VSizeFrame(ctx);
this.fileNamePrefix = fileNamePrefix;
- this.budget = (int) (budgetInBytes / ctx.getInitialFrameSize());
-
+ this.budget = (int) Math.min(budgetInBytes / ctx.getInitialFrameSize(), Integer.MAX_VALUE);
+ if (budget <= 0) {
+ throw new HyracksDataException("Invalid budget " + budgetInBytes + ". Budget must be larger than 0");
+ }
}
public void open() throws HyracksDataException {
try {
- this.currentWriteFile = createFile();
+ this.currentWriteFile = StoragePathUtil.createFile(fileNamePrefix, fileCount++);
this.currentReadFile = currentWriteFile;
this.bos = new BufferedOutputStream(new FileOutputStream(currentWriteFile));
this.bis = new BufferedInputStream(new FileInputStream(currentReadFile));
@@ -135,7 +137,7 @@ public class FrameSpiller {
}
public double usedBudget() {
- return ((double) (totalWriteCount - totalReadCount) / (double) budget);
+ return (double) (totalWriteCount - totalReadCount) / (double) budget;
}
public synchronized boolean spill(ByteBuffer frame) throws HyracksDataException {
@@ -150,7 +152,7 @@ public class FrameSpiller {
if (currentWriteCount >= FRAMES_PER_FILE) {
bos.close();
currentWriteCount = 0;
- currentWriteFile = createFile();
+ currentWriteFile = StoragePathUtil.createFile(fileNamePrefix, fileCount++);
files.add(currentWriteFile);
bos = new BufferedOutputStream(new FileOutputStream(currentWriteFile));
}
@@ -161,26 +163,6 @@ public class FrameSpiller {
}
}
- private File createFile() throws HyracksDataException {
- try {
- String fileName = fileNamePrefix + "_" + fileCount++;
- File file = new File(fileName);
- if (!file.exists()) {
- boolean success = file.createNewFile();
- if (!success) {
- throw new HyracksDataException("Unable to create spill file " + fileName);
- } else {
- if (LOGGER.isEnabledFor(Level.INFO)) {
- LOGGER.info("Created spill file " + file.getAbsolutePath());
- }
- }
- }
- return file;
- } catch (Throwable th) {
- throw new HyracksDataException(th);
- }
- }
-
public synchronized void close() {
// Do proper cleanup
if (bos != null) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/runtime/IngestionRuntime.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/runtime/IngestionRuntime.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/runtime/IngestionRuntime.java
index 8ee3e2b..9661890 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/runtime/IngestionRuntime.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/feed/runtime/IngestionRuntime.java
@@ -28,6 +28,8 @@ import org.apache.asterix.external.feed.dataflow.FeedFrameCollector;
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.util.HyracksConstants;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
public class IngestionRuntime extends SubscribableRuntime {
@@ -48,8 +50,9 @@ public class IngestionRuntime extends SubscribableRuntime {
dWriter.subscribe(collector);
subscribers.add(collectionRuntime);
if (numSubscribers == 0) {
- ctx.setSharedObject(new VSizeFrame(ctx));
- collectionRuntime.getCtx().setSharedObject(ctx.getSharedObject());
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, new VSizeFrame(ctx), ctx);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE,
+ TaskUtils.<VSizeFrame> get(HyracksConstants.KEY_MESSAGE, ctx), collectionRuntime.getCtx());
adapterRuntimeManager.start();
}
numSubscribers++;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
index 54e17ef..37a42a7 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaComputeNodePushable.java
@@ -40,7 +40,9 @@ import org.apache.hyracks.api.dataflow.IActivity;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
/*
@@ -112,7 +114,7 @@ public class FeedMetaComputeNodePushable extends AbstractUnaryInputUnaryOutputOp
this.feedManager = (ActiveManager) ((IAsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext()
.getApplicationObject()).getFeedManager();
this.message = new VSizeFrame(ctx);
- ctx.setSharedObject(message);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, message, ctx);
this.opDesc = feedMetaOperatorDescriptor;
this.recordDescProvider = recordDescProvider;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
index 6f679f7..95bebad 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/operators/FeedMetaStoreNodePushable.java
@@ -41,7 +41,9 @@ import org.apache.hyracks.api.dataflow.IActivity;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
public class FeedMetaStoreNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
@@ -107,7 +109,7 @@ public class FeedMetaStoreNodePushable extends AbstractUnaryInputUnaryOutputOper
.getApplicationObject()).getFeedManager();
this.targetId = targetId;
this.message = new VSizeFrame(ctx);
- ctx.setSharedObject(message);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, message, ctx);
this.recordDescProvider = recordDescProvider;
this.opDesc = feedMetaOperatorDescriptor;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
index 4ad08b3..98cb4b0 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/provider/DataflowControllerProvider.java
@@ -59,7 +59,7 @@ public class DataflowControllerProvider {
public static IDataFlowController getDataflowController(ARecordType recordType, IHyracksTaskContext ctx,
int partition, IExternalDataSourceFactory dataSourceFactory, IDataParserFactory dataParserFactory,
Map<String, String> configuration, boolean indexingOp, boolean isFeed, FeedLogManager feedLogManager)
- throws HyracksDataException {
+ throws HyracksDataException {
try {
switch (dataSourceFactory.getDataSourceType()) {
case RECORDS:
@@ -67,6 +67,7 @@ public class DataflowControllerProvider {
IRecordReader<?> recordReader = recordReaderFactory.createRecordReader(ctx, partition);
IRecordDataParserFactory<?> recordParserFactory = (IRecordDataParserFactory<?>) dataParserFactory;
IRecordDataParser<?> dataParser = recordParserFactory.createRecordParser(ctx);
+ boolean sendMarker = ExternalDataUtils.isSendMarker(configuration);
if (indexingOp) {
return new IndexingDataFlowController(ctx,
DataflowUtils.getTupleForwarder(configuration, feedLogManager), dataParser,
@@ -80,18 +81,19 @@ public class DataflowControllerProvider {
if (isChangeFeed) {
int numOfKeys = ExternalDataUtils.getNumberOfKeys(configuration);
return new ChangeFeedWithMetaDataFlowController(ctx, tupleForwarder, feedLogManager,
- numOfKeys + 2, (IRecordWithMetadataParser) dataParser, recordReader);
+ numOfKeys + 2, (IRecordWithMetadataParser) dataParser, recordReader,
+ sendMarker);
} else {
return new FeedWithMetaDataFlowController(ctx, tupleForwarder, feedLogManager, 2,
- (IRecordWithMetadataParser) dataParser, recordReader);
+ (IRecordWithMetadataParser) dataParser, recordReader, sendMarker);
}
} else if (isChangeFeed) {
int numOfKeys = ExternalDataUtils.getNumberOfKeys(configuration);
return new ChangeFeedDataFlowController(ctx, tupleForwarder, feedLogManager, numOfKeys + 1,
- (IRecordWithPKDataParser) dataParser, recordReader);
+ (IRecordWithPKDataParser) dataParser, recordReader, sendMarker);
} else {
return new FeedRecordDataFlowController(ctx, tupleForwarder, feedLogManager, 1, dataParser,
- recordReader);
+ recordReader, sendMarker);
}
} else {
return new RecordDataFlowController(ctx,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/DataflowUtils.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/DataflowUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/DataflowUtils.java
index ad945f2..ed811ad 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/DataflowUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/DataflowUtils.java
@@ -30,6 +30,7 @@ import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
public class DataflowUtils {
public static void addTupleToFrame(FrameTupleAppender appender, ArrayTupleBuilder tb, IFrameWriter writer)
@@ -67,4 +68,14 @@ public class DataflowUtils {
throw new HyracksDataException("Unknown tuple forward policy");
}
}
+
+ public static void addTupleToFrame(FrameTupleAppender appender, ITupleReference tuple, IFrameWriter writer)
+ throws HyracksDataException {
+ if (!appender.append(tuple)) {
+ appender.write(writer, true);
+ if (!appender.append(tuple)) {
+ throw new HyracksDataException("Tuple is too large for a frame");
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index 55dee04..e251f32 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -46,6 +46,8 @@ public class ExternalDataConstants {
public static final String KEY_FILESYSTEM = "fs";
// specifies the address of the HDFS name node
public static final String KEY_HADOOP_FILESYSTEM_URI = "fs.defaultFS";
+ // specifies whether a feed sends progress markers or not
+ public static final String KEY_SEND_MARKER = "send-marker";
// specifies the class implementation of the accessed instance of HDFS
public static final String KEY_HADOOP_FILESYSTEM_CLASS = "fs.hdfs.impl";
public static final String KEY_HADOOP_INPUT_DIR = "mapred.input.dir";
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
index 19781f9..23cd39c 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
@@ -332,4 +332,8 @@ public class ExternalDataUtils {
}
return intIndicators;
}
+
+ public static boolean isSendMarker(Map<String, String> configuration) {
+ return Boolean.parseBoolean(configuration.get(ExternalDataConstants.KEY_SEND_MARKER));
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
index 6b7eb31..6e1b9e8 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/FeedUtils.java
@@ -64,6 +64,9 @@ public class FeedUtils {
SPILL, // Memory budget has been consumed. Now we're writing to disk
DISCARD // Memory and Disk space budgets have been consumed. Now we're discarding
}
+
+ private FeedUtils() {
+ }
private static String prepareDataverseFeedName(String dataverseName, String feedName) {
return dataverseName + File.separator + feedName;
@@ -87,7 +90,7 @@ public class FeedUtils {
throw new AsterixException("Can't create file splits for adapter with count partitioning constraints");
}
String[] locations = ((AlgebricksAbsolutePartitionConstraint) partitionConstraints).getLocations();
- List<FileSplit> splits = new ArrayList<FileSplit>();
+ List<FileSplit> splits = new ArrayList<>();
for (String nd : locations) {
splits.add(splitsForAdapter(dataverseName, feedName, nd,
AsterixClusterProperties.INSTANCE.getNodePartitions(nd)[0]));
@@ -120,6 +123,7 @@ public class FeedUtils {
int offset = fta.getTupleStartOffset(tc);
int len = fta.getTupleLength(tc);
int newSize = FrameHelper.calcAlignedFrameSizeToStore(1, len, message.getMinSize());
+ message.reset();
message.ensureFrameSize(newSize);
message.getBuffer().clear();
message.getBuffer().put(input.array(), offset, len);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateDataverseStatement.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateDataverseStatement.java b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateDataverseStatement.java
index 820ae5f..7a5b722 100644
--- a/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateDataverseStatement.java
+++ b/asterixdb/asterix-lang-common/src/main/java/org/apache/asterix/lang/common/statement/CreateDataverseStatement.java
@@ -22,6 +22,7 @@ import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.lang.common.base.Statement;
import org.apache.asterix.lang.common.struct.Identifier;
import org.apache.asterix.lang.common.visitor.base.ILangVisitor;
+import org.apache.asterix.runtime.formats.NonTaggedDataFormat;
public class CreateDataverseStatement extends Statement {
@@ -31,11 +32,7 @@ public class CreateDataverseStatement extends Statement {
public CreateDataverseStatement(Identifier dataverseName, String format, boolean ifNotExists) {
this.dataverseName = dataverseName;
- if (format == null) {
- this.format = "org.apache.asterix.runtime.formats.NonTaggedDataFormat";
- } else {
- this.format = format;
- }
+ this.format = (format == null) ? NonTaggedDataFormat.class.getName() : format;
this.ifNotExists = ifNotExists;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AInt64SerializerDeserializer.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AInt64SerializerDeserializer.java b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AInt64SerializerDeserializer.java
index 8738c97..000339f 100644
--- a/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AInt64SerializerDeserializer.java
+++ b/asterixdb/asterix-om/src/main/java/org/apache/asterix/dataflow/data/nontagged/serde/AInt64SerializerDeserializer.java
@@ -58,7 +58,7 @@ public class AInt64SerializerDeserializer implements ISerializerDeserializer<AIn
return (((long) (bytes[offset] & 0xff)) << 56) + (((long) (bytes[offset + 1] & 0xff)) << 48)
+ (((long) (bytes[offset + 2] & 0xff)) << 40) + (((long) (bytes[offset + 3] & 0xff)) << 32)
+ (((long) (bytes[offset + 4] & 0xff)) << 24) + (((long) (bytes[offset + 5] & 0xff)) << 16)
- + (((long) (bytes[offset + 6] & 0xff)) << 8) + (((long) (bytes[offset + 7] & 0xff)) << 0);
+ + (((long) (bytes[offset + 6] & 0xff)) << 8) + (bytes[offset + 7] & 0xff);
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
index cabfc77..d247490 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/management/ReplicationChannel.java
@@ -119,8 +119,9 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
lsmComponentLSNMappingService = new LSMComponentsSyncService();
replicationNotifier = new ReplicationNotifier();
replicationThreads = Executors.newCachedThreadPool(appContext.getThreadFactory());
- Map<String, ClusterPartition[]> nodePartitions = ((IAsterixPropertiesProvider) asterixAppRuntimeContextProvider
- .getAppContext()).getMetadataProperties().getNodePartitions();
+ Map<String, ClusterPartition[]> nodePartitions =
+ ((IAsterixPropertiesProvider) asterixAppRuntimeContextProvider.getAppContext()).getMetadataProperties()
+ .getNodePartitions();
Set<String> nodeReplicationClients = replicationProperties.getNodeReplicationClients(nodeId);
List<Integer> clientsPartitions = new ArrayList<>();
for (String clientId : nodeReplicationClients) {
@@ -141,8 +142,8 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
try {
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(true);
- InetSocketAddress replicationChannelAddress = new InetSocketAddress(InetAddress.getByName(nodeIP),
- dataPort);
+ InetSocketAddress replicationChannelAddress =
+ new InetSocketAddress(InetAddress.getByName(nodeIP), dataPort);
serverSocketChannel.socket().bind(replicationChannelAddress);
lsmComponentLSNMappingService.start();
replicationNotifier.start();
@@ -169,8 +170,9 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
if (remainingFile == 0) {
if (lsmCompProp.getOpType() == LSMOperationType.FLUSH && lsmCompProp.getReplicaLSN() != null
&& replicaUniqueLSN2RemoteMapping.containsKey(lsmCompProp.getNodeUniqueLSN())) {
- int remainingIndexes = replicaUniqueLSN2RemoteMapping
- .get(lsmCompProp.getNodeUniqueLSN()).numOfFlushedIndexes.decrementAndGet();
+ int remainingIndexes =
+ replicaUniqueLSN2RemoteMapping.get(lsmCompProp.getNodeUniqueLSN()).numOfFlushedIndexes
+ .decrementAndGet();
if (remainingIndexes == 0) {
/**
* Note: there is a chance that this will never be removed because some
@@ -216,8 +218,8 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
public void run() {
Thread.currentThread().setName("Replication Thread");
try {
- ReplicationRequestType replicationFunction = ReplicationProtocol.getRequestType(socketChannel,
- inBuffer);
+ ReplicationRequestType replicationFunction =
+ ReplicationProtocol.getRequestType(socketChannel, inBuffer);
while (replicationFunction != ReplicationRequestType.GOODBYE) {
switch (replicationFunction) {
case REPLICATE_LOG:
@@ -281,8 +283,8 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
Set<Integer> datasetsToForceFlush = new HashSet<>();
for (IndexInfo iInfo : openIndexesInfo) {
if (requestedIndexesToBeFlushed.contains(iInfo.getResourceId())) {
- AbstractLSMIOOperationCallback ioCallback = (AbstractLSMIOOperationCallback) iInfo.getIndex()
- .getIOOperationCallback();
+ AbstractLSMIOOperationCallback ioCallback =
+ (AbstractLSMIOOperationCallback) iInfo.getIndex().getIOOperationCallback();
//if an index has a pending flush, then the request to flush it will succeed.
if (ioCallback.hasPendingFlush()) {
//remove index to indicate that it will be flushed
@@ -373,8 +375,9 @@ public class ReplicationChannel extends Thread implements IReplicationChannel {
List<String> filesList;
Set<String> replicaIds = request.getReplicaIds();
Set<String> requesterExistingFiles = request.getExistingFiles();
- Map<String, ClusterPartition[]> nodePartitions = ((IAsterixPropertiesProvider) appContextProvider
- .getAppContext()).getMetadataProperties().getNodePartitions();
+ Map<String, ClusterPartition[]> nodePartitions =
+ ((IAsterixPropertiesProvider) appContextProvider.getAppContext()).getMetadataProperties()
+ .getNodePartitions();
for (String replicaId : replicaIds) {
//get replica partitions
ClusterPartition[] replicaPatitions = nodePartitions.get(replicaId);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
index 857f1e2..afd6019 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/AsterixLSMPrimaryUpsertOperatorNodePushable.java
@@ -25,6 +25,8 @@ import java.nio.ByteBuffer;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.dataflow.AsterixLSMIndexUtil;
import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.common.transactions.ILogMarkerCallback;
+import org.apache.asterix.common.transactions.PrimaryIndexLogMarkerCallback;
import org.apache.asterix.om.pointables.nonvisitor.ARecordPointable;
import org.apache.asterix.om.types.ARecordType;
import org.apache.asterix.om.types.ATypeTag;
@@ -41,6 +43,7 @@ import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
import org.apache.hyracks.storage.am.btree.util.BTreeUtils;
import org.apache.hyracks.storage.am.common.api.IIndexCursor;
@@ -114,8 +117,11 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
writer.open();
indexHelper.open();
index = indexHelper.getIndexInstance();
-
try {
+ if (ctx.getSharedObject() != null) {
+ PrimaryIndexLogMarkerCallback callback = new PrimaryIndexLogMarkerCallback((AbstractLSMIndex) index);
+ TaskUtils.putInSharedMap(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
+ }
missingTupleBuilder = new ArrayTupleBuilder(1);
DataOutput out = missingTupleBuilder.getDataOutput();
try {
@@ -135,8 +141,8 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
.createSearchOperationCallback(indexHelper.getResourceID(), ctx, this));
cursor = indexAccessor.createSearchCursor(false);
frameTuple = new FrameTupleReference();
- IAsterixAppRuntimeContext runtimeCtx = (IAsterixAppRuntimeContext) ctx.getJobletContext()
- .getApplicationContext().getApplicationObject();
+ IAsterixAppRuntimeContext runtimeCtx =
+ (IAsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject();
AsterixLSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) index,
runtimeCtx.getTransactionSubsystem().getLogManager());
} catch (Exception e) {
@@ -241,10 +247,7 @@ public class AsterixLSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertU
writeOutput(i, recordWasInserted);
i++;
}
- if (tupleCount > 0) {
- // All tuples has to move forward to maintain the correctness of the transaction pipeline
- appender.write(writer, true);
- }
+ appender.write(writer, true);
} catch (IndexException | IOException | AsterixException e) {
throw new HyracksDataException(e);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
index 94d2a8c..9a66aa5 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManager.java
@@ -80,8 +80,8 @@ public class LogManager implements ILogManager, ILifeCycleComponent {
public LogManager(TransactionSubsystem txnSubsystem) {
this.txnSubsystem = txnSubsystem;
- logManagerProperties = new LogManagerProperties(this.txnSubsystem.getTransactionProperties(),
- this.txnSubsystem.getId());
+ logManagerProperties =
+ new LogManagerProperties(this.txnSubsystem.getTransactionProperties(), this.txnSubsystem.getId());
logFileSize = logManagerProperties.getLogPartitionSize();
logPageSize = logManagerProperties.getLogPageSize();
numLogPages = logManagerProperties.getNumLogPages();
@@ -172,6 +172,9 @@ public class LogManager implements ILogManager, ILifeCycleComponent {
if (logRecord.getLogType() == LogType.FLUSH) {
logRecord.setLSN(appendLSN.get());
}
+ if (logRecord.isMarker()) {
+ logRecord.logAppended(appendLSN.get());
+ }
appendLSN.addAndGet(logRecord.getLogSize());
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
index afb926b..0183b29 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
@@ -92,7 +92,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
private final long SHARP_CHECKPOINT_LSN = -1;
private final boolean replicationEnabled;
public static final long NON_SHARP_CHECKPOINT_TARGET_LSN = -1;
- private final static String RECOVERY_FILES_DIR_NAME = "recovery_temp";
+ private static final String RECOVERY_FILES_DIR_NAME = "recovery_temp";
private Map<Integer, JobEntityCommits> jobId2WinnerEntitiesMap = null;
private final long cachedEntityCommitsPerJobSize;
private final PersistentLocalResourceRepository localResourceRepository;
@@ -108,8 +108,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
this.txnSubsystem = txnSubsystem;
logMgr = (LogManager) txnSubsystem.getLogManager();
checkpointHistory = txnSubsystem.getTransactionProperties().getCheckpointHistory();
- IAsterixPropertiesProvider propertiesProvider = (IAsterixPropertiesProvider) txnSubsystem
- .getAsterixAppRuntimeContextProvider().getAppContext();
+ IAsterixPropertiesProvider propertiesProvider =
+ (IAsterixPropertiesProvider) txnSubsystem.getAsterixAppRuntimeContextProvider().getAppContext();
replicationEnabled = propertiesProvider.getReplicationProperties().isReplicationEnabled();
localResourceRepository = (PersistentLocalResourceRepository) txnSubsystem.getAsterixAppRuntimeContextProvider()
.getLocalResourceRepository();
@@ -271,6 +271,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
break;
case LogType.FLUSH:
case LogType.WAIT:
+ case LogType.MARKER:
break;
default:
throw new ACIDException("Unsupported LogType: " + logRecord.getLogType());
@@ -361,10 +362,10 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
//if index is not registered into IndexLifeCycleManager,
//create the index using LocalMetadata stored in LocalResourceRepository
//get partition path in this node
- String partitionIODevicePath = localResourceRepository
- .getPartitionPath(localResource.getPartition());
- String resourceAbsolutePath = partitionIODevicePath + File.separator
- + localResource.getResourceName();
+ String partitionIODevicePath =
+ localResourceRepository.getPartitionPath(localResource.getPartition());
+ String resourceAbsolutePath =
+ partitionIODevicePath + File.separator + localResource.getResourceName();
localResource.setResourcePath(resourceAbsolutePath);
index = (ILSMIndex) datasetLifecycleManager.getIndex(resourceAbsolutePath);
if (index == null) {
@@ -379,8 +380,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
//#. get maxDiskLastLSN
ILSMIndex lsmIndex = index;
try {
- maxDiskLastLsn = ((AbstractLSMIOOperationCallback) lsmIndex
- .getIOOperationCallback())
+ maxDiskLastLsn =
+ ((AbstractLSMIOOperationCallback) lsmIndex.getIOOperationCallback())
.getComponentLSN(lsmIndex.getImmutableComponents());
} catch (HyracksDataException e) {
datasetLifecycleManager.close(resourceAbsolutePath);
@@ -405,6 +406,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
case LogType.ABORT:
case LogType.FLUSH:
case LogType.UPSERT_ENTITY_COMMIT:
+ case LogType.MARKER:
//do nothing
break;
default:
@@ -443,8 +445,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
//right after the new checkpoint file is written.
File[] prevCheckpointFiles = getPreviousCheckpointFiles();
- IDatasetLifecycleManager datasetLifecycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
- .getDatasetLifecycleManager();
+ IDatasetLifecycleManager datasetLifecycleManager =
+ txnSubsystem.getAsterixAppRuntimeContextProvider().getDatasetLifecycleManager();
//flush all in-memory components if it is the sharp checkpoint
if (isSharpCheckpoint) {
datasetLifecycleManager.flushAllDatasets();
@@ -467,8 +469,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
Set<Integer> deadReplicasPartitions = new HashSet<>();
//get partitions of the dead replicas that are not active on this node
for (String deadReplicaId : deadReplicaIds) {
- ClusterPartition[] nodePartitons = metadataProperties.getNodePartitions()
- .get(deadReplicaId);
+ ClusterPartition[] nodePartitons =
+ metadataProperties.getNodePartitions().get(deadReplicaId);
for (ClusterPartition partition : nodePartitons) {
if (!localResourceRepository.getActivePartitions()
.contains(partition.getPartitionId())) {
@@ -492,8 +494,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
datasetLifecycleManager.scheduleAsyncFlushForLaggingDatasets(nonSharpCheckpointTargetLSN);
if (replicationEnabled) {
//request remote replicas to flush lagging indexes
- IReplicationManager replicationManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
- .getAppContext().getReplicationManager();
+ IReplicationManager replicationManager =
+ txnSubsystem.getAsterixAppRuntimeContextProvider().getAppContext().getReplicationManager();
try {
replicationManager.requestFlushLaggingReplicaIndexes(nonSharpCheckpointTargetLSN);
} catch (IOException e) {
@@ -564,16 +566,16 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
@Override
public long getLocalMinFirstLSN() throws HyracksDataException {
- IDatasetLifecycleManager datasetLifecycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
- .getDatasetLifecycleManager();
+ IDatasetLifecycleManager datasetLifecycleManager =
+ txnSubsystem.getAsterixAppRuntimeContextProvider().getDatasetLifecycleManager();
List<IIndex> openIndexList = datasetLifecycleManager.getOpenIndexes();
long firstLSN;
//the min first lsn can only be the current append or smaller
long minFirstLSN = logMgr.getAppendLSN();
if (openIndexList.size() > 0) {
for (IIndex index : openIndexList) {
- AbstractLSMIOOperationCallback ioCallback = (AbstractLSMIOOperationCallback) ((ILSMIndex) index)
- .getIOOperationCallback();
+ AbstractLSMIOOperationCallback ioCallback =
+ (AbstractLSMIOOperationCallback) ((ILSMIndex) index).getIOOperationCallback();
if (!((AbstractLSMIndex) index).isCurrentMutableComponentEmpty() || ioCallback.hasPendingFlush()) {
firstLSN = ioCallback.getFirstLSN();
@@ -585,8 +587,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
}
private long getRemoteMinFirstLSN() {
- IReplicaResourcesManager remoteResourcesManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
- .getAppContext().getReplicaResourcesManager();
+ IReplicaResourcesManager remoteResourcesManager =
+ txnSubsystem.getAsterixAppRuntimeContextProvider().getAppContext().getReplicaResourcesManager();
long minRemoteLSN = remoteResourcesManager.getPartitionsMinLSN(localResourceRepository.getInactivePartitions());
return minRemoteLSN;
}
@@ -783,6 +785,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
case LogType.ABORT:
case LogType.FLUSH:
case LogType.WAIT:
+ case LogType.MARKER:
//ignore
break;
default:
@@ -798,8 +801,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
//undo loserTxn's effect
LOGGER.log(Level.INFO, "undoing loser transaction's effect");
- IDatasetLifecycleManager datasetLifecycleManager = txnSubsystem.getAsterixAppRuntimeContextProvider()
- .getDatasetLifecycleManager();
+ IDatasetLifecycleManager datasetLifecycleManager =
+ txnSubsystem.getAsterixAppRuntimeContextProvider().getDatasetLifecycleManager();
//TODO sort loser entities by smallest LSN to undo in one pass.
Iterator<Entry<TxnId, List<Long>>> iter = jobLoserEntity2LSNsMap.entrySet().iterator();
int undoCount = 0;
@@ -855,10 +858,10 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
private static void undo(ILogRecord logRecord, IDatasetLifecycleManager datasetLifecycleManager) {
try {
- ILSMIndex index = (ILSMIndex) datasetLifecycleManager.getIndex(logRecord.getDatasetId(),
- logRecord.getResourceId());
- ILSMIndexAccessor indexAccessor = index.createAccessor(NoOpOperationCallback.INSTANCE,
- NoOpOperationCallback.INSTANCE);
+ ILSMIndex index =
+ (ILSMIndex) datasetLifecycleManager.getIndex(logRecord.getDatasetId(), logRecord.getResourceId());
+ ILSMIndexAccessor indexAccessor =
+ index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
if (logRecord.getNewOp() == IndexOperation.INSERT.ordinal()) {
indexAccessor.forceDelete(logRecord.getNewValue());
} else if (logRecord.getNewOp() == IndexOperation.DELETE.ordinal()) {
@@ -873,10 +876,10 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
private static void redo(ILogRecord logRecord, IDatasetLifecycleManager datasetLifecycleManager) {
try {
- ILSMIndex index = (ILSMIndex) datasetLifecycleManager.getIndex(logRecord.getDatasetId(),
- logRecord.getResourceId());
- ILSMIndexAccessor indexAccessor = index.createAccessor(NoOpOperationCallback.INSTANCE,
- NoOpOperationCallback.INSTANCE);
+ ILSMIndex index =
+ (ILSMIndex) datasetLifecycleManager.getIndex(logRecord.getDatasetId(), logRecord.getResourceId());
+ ILSMIndexAccessor indexAccessor =
+ index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
if (logRecord.getNewOp() == IndexOperation.INSERT.ordinal()) {
indexAccessor.forceInsert(logRecord.getNewValue());
} else if (logRecord.getNewOp() == IndexOperation.DELETE.ordinal()) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
index d94f933..19d7afb 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/comm/IFrameWriter.java
@@ -64,6 +64,7 @@ public interface IFrameWriter {
/**
* Provide data to the stream of this {@link IFrameWriter}.
+ *
* @param buffer
* - Buffer containing data.
* @throws HyracksDataException
@@ -72,21 +73,24 @@ public interface IFrameWriter {
/**
* request the frame to push its content forward and flush its consumers
+ *
* @throws HyracksDataException
*/
public default void flush() throws HyracksDataException {
- throw new HyracksDataException("flush() is not supported in this IFrameWriter");
+ // No Op
}
/**
* Indicate that a failure was encountered and the current stream is to be
* aborted.
+ *
* @throws HyracksDataException
*/
public void fail() throws HyracksDataException;
/**
* Close this {@link IFrameWriter} and give up all resources.
+ *
* @throws HyracksDataException
*/
public void close() throws HyracksDataException;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
index 3781489..4eb3ebf 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/context/IHyracksTaskContext.java
@@ -45,7 +45,7 @@ public interface IHyracksTaskContext
public void sendApplicationMessageToCC(byte[] message, DeploymentId deploymendId) throws Exception;
- public void setSharedObject(Object sharedObject);
+ public void setSharedObject(Object object);
public Object getSharedObject();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/HyracksConstants.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/HyracksConstants.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/HyracksConstants.java
new file mode 100644
index 0000000..8d55235
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/HyracksConstants.java
@@ -0,0 +1,26 @@
+/*
+ * 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.api.util;
+
+public class HyracksConstants {
+ public static final String KEY_MESSAGE = "HYX:MSG";
+
+ private HyracksConstants() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index 2f8def1..43cac74 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -260,7 +260,8 @@ public class NodeControllerService implements IControllerService {
init();
datasetNetworkManager.start();
- IIPCHandle ccIPCHandle = ipc.getHandle(new InetSocketAddress(ncConfig.ccHost, ncConfig.ccPort), ncConfig.retries);
+ IIPCHandle ccIPCHandle =
+ ipc.getHandle(new InetSocketAddress(ncConfig.ccHost, ncConfig.ccPort), ncConfig.retries);
this.ccs = new ClusterControllerRemoteProxy(ccIPCHandle);
HeartbeatSchema.GarbageCollectorInfo[] gcInfos = new HeartbeatSchema.GarbageCollectorInfo[gcMXBeans.size()];
for (int i = 0; i < gcInfos.length; ++i) {
@@ -270,12 +271,11 @@ public class NodeControllerService implements IControllerService {
// Use "public" versions of network addresses and ports
NetworkAddress datasetAddress = datasetNetworkManager.getPublicNetworkAddress();
NetworkAddress netAddress = netManager.getPublicNetworkAddress();
- ccs.registerNode(new NodeRegistration(ipc.getSocketAddress(), id, ncConfig, netAddress,
- datasetAddress, osMXBean.getName(), osMXBean.getArch(), osMXBean
- .getVersion(), osMXBean.getAvailableProcessors(), runtimeMXBean.getVmName(), runtimeMXBean
- .getVmVersion(), runtimeMXBean.getVmVendor(), runtimeMXBean.getClassPath(), runtimeMXBean
- .getLibraryPath(), runtimeMXBean.getBootClassPath(), runtimeMXBean.getInputArguments(),
- runtimeMXBean.getSystemProperties(), hbSchema));
+ ccs.registerNode(new NodeRegistration(ipc.getSocketAddress(), id, ncConfig, netAddress, datasetAddress,
+ osMXBean.getName(), osMXBean.getArch(), osMXBean.getVersion(), osMXBean.getAvailableProcessors(),
+ runtimeMXBean.getVmName(), runtimeMXBean.getVmVersion(), runtimeMXBean.getVmVendor(),
+ runtimeMXBean.getClassPath(), runtimeMXBean.getLibraryPath(), runtimeMXBean.getBootClassPath(),
+ runtimeMXBean.getInputArguments(), runtimeMXBean.getSystemProperties(), hbSchema));
synchronized (this) {
while (registrationPending) {
@@ -490,7 +490,8 @@ public class NodeControllerService implements IControllerService {
CCNCFunctions.Function fn = (CCNCFunctions.Function) payload;
switch (fn.getFunctionId()) {
case SEND_APPLICATION_MESSAGE: {
- CCNCFunctions.SendApplicationMessageFunction amf = (CCNCFunctions.SendApplicationMessageFunction) fn;
+ CCNCFunctions.SendApplicationMessageFunction amf =
+ (CCNCFunctions.SendApplicationMessageFunction) fn;
queue.schedule(new ApplicationMessageWork(NodeControllerService.this, amf.getMessage(),
amf.getDeploymentId(), amf.getNodeId()));
return;
@@ -515,7 +516,8 @@ public class NodeControllerService implements IControllerService {
}
case REPORT_PARTITION_AVAILABILITY: {
- CCNCFunctions.ReportPartitionAvailabilityFunction rpaf = (CCNCFunctions.ReportPartitionAvailabilityFunction) fn;
+ CCNCFunctions.ReportPartitionAvailabilityFunction rpaf =
+ (CCNCFunctions.ReportPartitionAvailabilityFunction) fn;
queue.schedule(new ReportPartitionAvailabilityWork(NodeControllerService.this,
rpaf.getPartitionId(), rpaf.getNetworkAddress()));
return;
@@ -528,7 +530,8 @@ public class NodeControllerService implements IControllerService {
}
case GET_NODE_CONTROLLERS_INFO_RESPONSE: {
- CCNCFunctions.GetNodeControllersInfoResponseFunction gncirf = (CCNCFunctions.GetNodeControllersInfoResponseFunction) fn;
+ CCNCFunctions.GetNodeControllersInfoResponseFunction gncirf =
+ (CCNCFunctions.GetNodeControllersInfoResponseFunction) fn;
setNodeControllersInfo(gncirf.getNodeControllerInfos());
return;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
index 134154c..f463bfa 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/Task.java
@@ -394,8 +394,8 @@ public class Task implements IHyracksTaskContext, ICounterContext, Runnable {
}
@Override
- public void setSharedObject(Object sharedObject) {
- this.sharedObject = sharedObject;
+ public void setSharedObject(Object object) {
+ this.sharedObject = object;
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
index 7d12296..7d90a8b 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/partitions/MaterializingPipelinedPartition.java
@@ -106,20 +106,20 @@ public class MaterializingPipelinedPartition implements IFrameWriter, IPartition
ByteBuffer buffer = ctx.allocateFrame();
boolean fail = false;
boolean done = false;
- boolean flush = false;
while (!fail && !done) {
synchronized (MaterializingPipelinedPartition.this) {
- if (flushRequest) {
- flushRequest = false;
- flush = true;
- }
- while (offset >= size && !eos && !failed && !flush) {
+ while (offset >= size && !eos && !failed) {
+ if (flushRequest) {
+ flushRequest = false;
+ writer.flush();
+ }
try {
MaterializingPipelinedPartition.this.wait();
} catch (InterruptedException e) {
throw new HyracksDataException(e);
}
}
+ flushRequest = false;
fail = failed;
done = eos && offset >= size;
}
@@ -134,10 +134,6 @@ public class MaterializingPipelinedPartition implements IFrameWriter, IPartition
offset += readLen;
buffer.flip();
writer.nextFrame(buffer);
- if (flush) {
- writer.flush();
- flush = false;
- }
}
}
}
@@ -213,4 +209,4 @@ public class MaterializingPipelinedPartition implements IFrameWriter, IPartition
flushRequest = true;
notifyAll();
}
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
index e7131d5..57f8072 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
@@ -89,9 +89,7 @@ public class AbstractFrameAppender implements IFrameAppender {
@Override
public void write(IFrameWriter outWriter, boolean clearFrame) throws HyracksDataException {
getBuffer().clear();
- if (getTupleCount() > 0) {
- outWriter.nextFrame(getBuffer());
- }
+ outWriter.nextFrame(getBuffer());
if (clearFrame) {
frame.reset();
reset(getBuffer(), true);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
index ef11b5b..3ef8b28 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
@@ -45,13 +45,13 @@ public class FrameTupleAppender extends AbstractFrameAppender implements IFrameT
* append fieldSlots and bytes to the current frame
*/
@Override
- public boolean append(int[] fieldSlots, byte[] bytes, int offset, int length) throws HyracksDataException {
- if (canHoldNewTuple(fieldSlots.length, length)) {
- for (int i = 0; i < fieldSlots.length; ++i) {
- IntSerDeUtils.putInt(array, tupleDataEndOffset + i * 4, fieldSlots[i]);
+ public boolean append(int[] fieldEndOffsets, byte[] bytes, int offset, int length) throws HyracksDataException {
+ if (canHoldNewTuple(fieldEndOffsets.length, length)) {
+ for (int i = 0; i < fieldEndOffsets.length; ++i) {
+ IntSerDeUtils.putInt(array, tupleDataEndOffset + i * 4, fieldEndOffsets[i]);
}
- System.arraycopy(bytes, offset, array, tupleDataEndOffset + fieldSlots.length * 4, length);
- tupleDataEndOffset += fieldSlots.length * 4 + length;
+ System.arraycopy(bytes, offset, array, tupleDataEndOffset + fieldEndOffsets.length * 4, length);
+ tupleDataEndOffset += fieldEndOffsets.length * 4 + length;
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1), tupleDataEndOffset);
++tupleCount;
@@ -63,19 +63,24 @@ public class FrameTupleAppender extends AbstractFrameAppender implements IFrameT
}
public boolean append(ITupleReference tuple) throws HyracksDataException {
- int tupleSize = 0;
+ int length = 0;
for (int i = 0; i < tuple.getFieldCount(); i++) {
- tupleSize += tuple.getFieldLength(i);
+ length += tuple.getFieldLength(i);
}
- if (canHoldNewTuple(tuple.getFieldCount(), tupleSize)) {
- int offset = 0;
+
+ if (canHoldNewTuple(tuple.getFieldCount(), length)) {
+ length = 0;
+ for (int i = 0; i < tuple.getFieldCount(); ++i) {
+ length += tuple.getFieldLength(i);
+ IntSerDeUtils.putInt(array, tupleDataEndOffset + i * 4, length);
+ }
+ length = 0;
for (int i = 0; i < tuple.getFieldCount(); ++i) {
- IntSerDeUtils.putInt(array, tupleDataEndOffset + i * 4, offset);
System.arraycopy(tuple.getFieldData(i), tuple.getFieldStart(i), array,
- tupleDataEndOffset + tuple.getFieldCount() * 4, tuple.getFieldLength(i));
- offset += tuple.getFieldLength(i);
+ tupleDataEndOffset + tuple.getFieldCount() * 4 + length, tuple.getFieldLength(i));
+ length += tuple.getFieldLength(i);
}
- tupleDataEndOffset += tuple.getFieldCount() * 4 + tupleSize;
+ tupleDataEndOffset += tuple.getFieldCount() * 4 + length;
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1), tupleDataEndOffset);
++tupleCount;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/MessagingFrameTupleAppender.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/MessagingFrameTupleAppender.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/MessagingFrameTupleAppender.java
index cae659d..7f518cd 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/MessagingFrameTupleAppender.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/io/MessagingFrameTupleAppender.java
@@ -26,7 +26,9 @@ import org.apache.hyracks.api.comm.IFrameWriter;
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.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.util.IntSerDeUtils;
/**
@@ -39,7 +41,9 @@ public class MessagingFrameTupleAppender extends FrameTupleAppender {
private static final int NULL_MESSAGE_SIZE = 1;
public static final byte NULL_FEED_MESSAGE = 0x01;
public static final byte ACK_REQ_FEED_MESSAGE = 0x02;
- public static final byte SNAPSHOT_MESSAGE = 0x03;
+ public static final byte MARKER_MESSAGE = 0x03;
+ private boolean initialized = false;
+ private VSizeFrame message;
public MessagingFrameTupleAppender(IHyracksTaskContext ctx) {
this.ctx = ctx;
@@ -59,8 +63,8 @@ public class MessagingFrameTupleAppender extends FrameTupleAppender {
case ACK_REQ_FEED_MESSAGE:
aString.append("Ack Request, ");
break;
- case SNAPSHOT_MESSAGE:
- aString.append("Snapshot, ");
+ case MARKER_MESSAGE:
+ aString.append("Marker, ");
break;
default:
aString.append("Unknown, ");
@@ -78,8 +82,8 @@ public class MessagingFrameTupleAppender extends FrameTupleAppender {
return NULL_FEED_MESSAGE;
case ACK_REQ_FEED_MESSAGE:
return ACK_REQ_FEED_MESSAGE;
- case SNAPSHOT_MESSAGE:
- return SNAPSHOT_MESSAGE;
+ case MARKER_MESSAGE:
+ return MARKER_MESSAGE;
default:
throw new HyracksDataException("Unknown message type");
}
@@ -101,24 +105,35 @@ public class MessagingFrameTupleAppender extends FrameTupleAppender {
@Override
public void write(IFrameWriter outWriter, boolean clearFrame) throws HyracksDataException {
+ if (!initialized) {
+ message = TaskUtils.<VSizeFrame> get(HyracksConstants.KEY_MESSAGE, ctx);
+ initialized = true;
+ }
// If message fits, we append it, otherwise, we append a null message, then send a message only
// frame with the message
- ByteBuffer message = ((VSizeFrame) ctx.getSharedObject()).getBuffer();
- int messageSize = message.limit() - message.position();
- if (hasEnoughSpace(1, messageSize)) {
- appendMessage(message);
- forward(outWriter);
- } else {
+ if (message == null) {
if (tupleCount > 0) {
appendNullMessage();
forward(outWriter);
}
- if (!hasEnoughSpace(1, messageSize)) {
- frame.ensureFrameSize(FrameHelper.calcAlignedFrameSizeToStore(1, messageSize, frame.getMinSize()));
- reset(frame.getBuffer(), true);
+ } else {
+ ByteBuffer buffer = message.getBuffer();
+ int messageSize = buffer.limit() - buffer.position();
+ if (hasEnoughSpace(1, messageSize)) {
+ appendMessage(buffer);
+ forward(outWriter);
+ } else {
+ if (tupleCount > 0) {
+ appendNullMessage();
+ forward(outWriter);
+ }
+ if (!hasEnoughSpace(1, messageSize)) {
+ frame.ensureFrameSize(FrameHelper.calcAlignedFrameSizeToStore(1, messageSize, frame.getMinSize()));
+ reset(frame.getBuffer(), true);
+ }
+ appendMessage(buffer);
+ forward(outWriter);
}
- appendMessage(message);
- forward(outWriter);
}
}
@@ -130,8 +145,9 @@ public class MessagingFrameTupleAppender extends FrameTupleAppender {
}
private void appendMessage(ByteBuffer message) {
- System.arraycopy(message.array(), message.position(), array, tupleDataEndOffset, message.limit());
- tupleDataEndOffset += message.limit();
+ int messageLength = message.limit() - message.position();
+ System.arraycopy(message.array(), message.position(), array, tupleDataEndOffset, messageLength);
+ tupleDataEndOffset += messageLength;
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1), tupleDataEndOffset);
++tupleCount;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/TaskUtils.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/TaskUtils.java b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/TaskUtils.java
new file mode 100644
index 0000000..4f27d79
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/util/TaskUtils.java
@@ -0,0 +1,77 @@
+/*
+ * 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.dataflow.common.util;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+
+/**
+ * A Utility class for facilitating common operations used with a hyracks task
+ */
+public class TaskUtils {
+ private TaskUtils() {
+ }
+
+ /**
+ * get the shared object of a task as a Map<String,Object>
+ *
+ * @param ctx
+ * the task context
+ * @param create
+ * @return the task shared map
+ */
+ @SuppressWarnings("unchecked")
+ public static Map<String, Object> getSharedMap(IHyracksTaskContext ctx, boolean create) {
+ if (ctx.getSharedObject() != null) {
+ return (Map<String, Object>) ctx.getSharedObject();
+ } else if (create) {
+ Map<String, Object> taskMap = new HashMap<>();
+ ctx.setSharedObject(taskMap);
+ return taskMap;
+ } else {
+ return null;
+ }
+ }
+
+ /**
+ * put the key value pair in a map task object
+ *
+ * @param key
+ * @param ctx
+ * @param object
+ */
+ public static void putInSharedMap(String key, Object object, IHyracksTaskContext ctx) {
+ TaskUtils.getSharedMap(ctx, true).put(key, object);
+ }
+
+ /**
+ * get a <T> object from the shared map of the task
+ *
+ * @param key
+ * @param ctx
+ * @return the value associated with the key casted as T
+ */
+ @SuppressWarnings("unchecked")
+ public static <T> T get(String key, IHyracksTaskContext ctx) {
+ Map<String, Object> sharedMap = TaskUtils.getSharedMap(ctx, false);
+ return sharedMap == null ? null : (T) sharedMap.get(key);
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
index 30ee3c0..b4e51be 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/preclustered/PreclusteredGroupWriter.java
@@ -65,8 +65,8 @@ public class PreclusteredGroupWriter implements IFrameWriter {
RecordDescriptor outRecordDesc, IFrameWriter writer) throws HyracksDataException {
this.groupFields = groupFields;
this.comparators = comparators;
- this.aggregator = aggregatorFactory.createAggregator(ctx, inRecordDesc, outRecordDesc, groupFields, groupFields,
- writer);
+ this.aggregator =
+ aggregatorFactory.createAggregator(ctx, inRecordDesc, outRecordDesc, groupFields, groupFields, writer);
this.aggregateState = aggregator.createAggregateStates();
copyFrame = new VSizeFrame(ctx);
inFrameAccessor = new FrameTupleAccessor(inRecordDesc);
@@ -91,29 +91,32 @@ public class PreclusteredGroupWriter implements IFrameWriter {
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
inFrameAccessor.reset(buffer);
int nTuples = inFrameAccessor.getTupleCount();
- for (int i = 0; i < nTuples; ++i) {
- if (first) {
+ if (nTuples != 0) {
+ for (int i = 0; i < nTuples; ++i) {
+ if (first) {
- tupleBuilder.reset();
- for (int j = 0; j < groupFields.length; j++) {
- tupleBuilder.addField(inFrameAccessor, i, groupFields[j]);
- }
- aggregator.init(tupleBuilder, inFrameAccessor, i, aggregateState);
+ tupleBuilder.reset();
+ for (int j = 0; j < groupFields.length; j++) {
+ tupleBuilder.addField(inFrameAccessor, i, groupFields[j]);
+ }
+ aggregator.init(tupleBuilder, inFrameAccessor, i, aggregateState);
- first = false;
+ first = false;
- } else {
- if (i == 0) {
- switchGroupIfRequired(copyFrameAccessor, copyFrameAccessor.getTupleCount() - 1, inFrameAccessor, i);
} else {
- switchGroupIfRequired(inFrameAccessor, i - 1, inFrameAccessor, i);
- }
+ if (i == 0) {
+ switchGroupIfRequired(copyFrameAccessor, copyFrameAccessor.getTupleCount() - 1, inFrameAccessor,
+ i);
+ } else {
+ switchGroupIfRequired(inFrameAccessor, i - 1, inFrameAccessor, i);
+ }
+ }
}
+ copyFrame.ensureFrameSize(buffer.capacity());
+ FrameUtils.copyAndFlip(buffer, copyFrame.getBuffer());
+ copyFrameAccessor.reset(copyFrame.getBuffer());
}
- copyFrame.ensureFrameSize(buffer.capacity());
- FrameUtils.copyAndFlip(buffer, copyFrame.getBuffer());
- copyFrameAccessor.reset(copyFrame.getBuffer());
}
private void switchGroupIfRequired(FrameTupleAccessor prevTupleAccessor, int prevTupleIndex,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/hyracks-fullstack/hyracks/hyracks-maven-plugins/hyracks-virtualcluster-maven-plugin/pom.xml
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-maven-plugins/hyracks-virtualcluster-maven-plugin/pom.xml b/hyracks-fullstack/hyracks/hyracks-maven-plugins/hyracks-virtualcluster-maven-plugin/pom.xml
index 7267cb7..7784199 100644
--- a/hyracks-fullstack/hyracks/hyracks-maven-plugins/hyracks-virtualcluster-maven-plugin/pom.xml
+++ b/hyracks-fullstack/hyracks/hyracks-maven-plugins/hyracks-virtualcluster-maven-plugin/pom.xml
@@ -33,4 +33,35 @@
<root.dir>${basedir}/../../..</root.dir>
</properties>
+ <build>
+ <pluginManagement>
+ <plugins>
+ <!--This plugin's configuration is used to store Eclipse m2e settings only. It has no influence on the Maven build itself.-->
+ <plugin>
+ <groupId>org.eclipse.m2e</groupId>
+ <artifactId>lifecycle-mapping</artifactId>
+ <version>1.0.0</version>
+ <configuration>
+ <lifecycleMappingMetadata>
+ <pluginExecutions>
+ <pluginExecution>
+ <pluginExecutionFilter>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-plugin-plugin</artifactId>
+ <versionRange>[3.3,)</versionRange>
+ <goals>
+ <goal>descriptor</goal>
+ </goals>
+ </pluginExecutionFilter>
+ <action>
+ <ignore></ignore>
+ </action>
+ </pluginExecution>
+ </pluginExecutions>
+ </lifecycleMappingMetadata>
+ </configuration>
+ </plugin>
+ </plugins>
+ </pluginManagement>
+ </build>
</project>
[4/4] asterixdb git commit: Add Test NodeController,
Test Data Generator, and Marker Logs
Posted by am...@apache.org.
Add Test NodeController, Test Data Generator, and Marker Logs
This test enable creating a node controller for unit test purposes.
The Node controller is identical to the regular node controller
except that it doesn't communicate with a cluster controller at all.
In this change, Test Data Generator is introduced which should
facilitate writing unit test cases which requires data generation.
The change also includes enabling feeds to send progress data. progress
information can then be sent through the pipeline and persisted in the
transaction logs and primary index component. A Unit test case has
been created to test adding progress markers to logs and index
components and then reading them.
The last part of this change is the addition of marker logs and their
callbacks. They enable components to create arbitrary logs and get a
callback when they are written to the transaction logs. Initial set of
unit tests were added for marker logs.
Change-Id: I3b9aa8de758b7d26ca34868b16e5ce693e0c0243
Reviewed-on: https://asterix-gerrit.ics.uci.edu/962
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Michael Blow <mb...@apache.org>
Reviewed-by: abdullah alamoudi <ba...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/7575785a
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/7575785a
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/7575785a
Branch: refs/heads/master
Commit: 7575785accf54f0a1bf0857c9add6365f44d386e
Parents: 973a0d3
Author: Abdullah Alamoudi <ba...@gmail.com>
Authored: Fri Jul 22 20:52:47 2016 +0300
Committer: abdullah alamoudi <ba...@gmail.com>
Committed: Fri Jul 22 12:35:47 2016 -0700
----------------------------------------------------------------------
.../operators/physical/CommitRuntime.java | 37 +-
.../app/bootstrap/TestNodeController.java | 481 +++++++++++++++++++
.../data/gen/ABooleanFieldValueGenerator.java | 94 ++++
.../data/gen/ADoubleFieldValueGenerator.java | 153 ++++++
.../app/data/gen/AInt32FieldValueGenerator.java | 152 ++++++
.../app/data/gen/AInt64FieldValueGenerator.java | 152 ++++++
.../app/data/gen/ARecordValueGenerator.java | 119 +++++
.../data/gen/AStringFieldValueGenerator.java | 159 ++++++
.../data/gen/IAsterixFieldValueGenerator.java | 48 ++
.../data/gen/TestTupleCounterFrameWriter.java | 51 ++
.../asterix/app/data/gen/TupleGenerator.java | 126 +++++
.../asterix/test/common/TestTupleGenerator.java | 29 --
.../asterix/test/common/TestTupleReference.java | 74 +++
.../ConnectorDescriptorWithMessagingTest.java | 18 +-
.../asterix/test/dataflow/LogMarkerTest.java | 211 ++++++++
.../dataflow/TestRecordDescriptorFactory.java | 28 --
asterixdb/asterix-common/pom.xml | 5 +
.../context/PrimaryIndexOperationTracker.java | 10 +-
...erixLSMInsertDeleteOperatorNodePushable.java | 17 +-
.../common/transactions/ILogMarkerCallback.java | 45 ++
.../asterix/common/transactions/ILogRecord.java | 49 +-
.../asterix/common/transactions/LogRecord.java | 239 +++++----
.../asterix/common/transactions/LogType.java | 9 +-
.../PrimaryIndexLogMarkerCallback.java | 51 ++
.../asterix/common/utils/AsterixConstants.java | 30 ++
.../apache/asterix/common/utils/FrameStack.java | 149 ++++++
.../asterix/common/utils/StoragePathUtil.java | 39 ++
.../asterix/common/utils/TransactionUtil.java | 16 +
.../asterix/event/util/AsterixConstants.java | 25 -
.../asterix/event/util/PatternCreator.java | 17 +-
.../asterix/external/api/IFeedMarker.java | 35 ++
.../asterix/external/api/IRecordConverter.java | 1 +
.../asterix/external/api/IRecordReader.java | 4 +
.../dataflow/ChangeFeedDataFlowController.java | 4 +-
.../ChangeFeedWithMetaDataFlowController.java | 6 +-
.../dataflow/FeedRecordDataFlowController.java | 99 +++-
.../external/dataflow/FeedTupleForwarder.java | 4 +-
.../FeedWithMetaDataFlowController.java | 4 +-
.../feed/dataflow/FeedRuntimeInputHandler.java | 51 +-
.../external/feed/dataflow/FrameSpiller.java | 54 +--
.../external/feed/runtime/IngestionRuntime.java | 7 +-
.../operators/FeedMetaComputeNodePushable.java | 4 +-
.../operators/FeedMetaStoreNodePushable.java | 4 +-
.../provider/DataflowControllerProvider.java | 12 +-
.../asterix/external/util/DataflowUtils.java | 11 +
.../external/util/ExternalDataConstants.java | 2 +
.../external/util/ExternalDataUtils.java | 4 +
.../apache/asterix/external/util/FeedUtils.java | 6 +-
.../statement/CreateDataverseStatement.java | 7 +-
.../serde/AInt64SerializerDeserializer.java | 2 +-
.../management/ReplicationChannel.java | 27 +-
...rixLSMPrimaryUpsertOperatorNodePushable.java | 17 +-
.../management/service/logging/LogManager.java | 7 +-
.../service/recovery/RecoveryManager.java | 65 +--
.../apache/hyracks/api/comm/IFrameWriter.java | 6 +-
.../api/context/IHyracksTaskContext.java | 2 +-
.../hyracks/api/util/HyracksConstants.java | 26 +
.../control/nc/NodeControllerService.java | 23 +-
.../org/apache/hyracks/control/nc/Task.java | 4 +-
.../MaterializingPipelinedPartition.java | 18 +-
.../common/comm/io/AbstractFrameAppender.java | 4 +-
.../common/comm/io/FrameTupleAppender.java | 33 +-
.../common/io/MessagingFrameTupleAppender.java | 52 +-
.../hyracks/dataflow/common/util/TaskUtils.java | 77 +++
.../preclustered/PreclusteredGroupWriter.java | 39 +-
.../hyracks-virtualcluster-maven-plugin/pom.xml | 31 ++
.../am/common/api/IMetaDataPageManager.java | 43 +-
.../am/common/api/ITreeIndexMetaDataFrame.java | 5 +
.../IndexSearchOperatorNodePushable.java | 4 +-
.../am/common/frames/LIFOMetaDataFrame.java | 90 ++--
.../freepage/LinkedMetaDataPageManager.java | 26 +-
.../storage/am/lsm/btree/impls/LSMBTree.java | 81 ++--
.../lsm/btree/impls/LSMBTreeDiskComponent.java | 6 +
.../lsm/btree/impls/LSMBTreeFlushOperation.java | 8 +-
.../btree/impls/LSMBTreeMemoryComponent.java | 9 +-
.../am/lsm/common/api/ILSMComponent.java | 4 +
.../freepage/VirtualMetaDataPageManager.java | 10 +
.../lsm/common/impls/AbstractLSMComponent.java | 18 +-
.../am/lsm/common/impls/AbstractLSMIndex.java | 8 +-
.../impls/AbstractMemoryLSMComponent.java | 9 +-
.../storage/am/lsm/common/impls/LSMHarness.java | 4 +
.../lsm/common/impls/LSMTreeIndexAccessor.java | 3 +-
.../hyracks/test/support/TestTaskContext.java | 9 +-
83 files changed, 3194 insertions(+), 528 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
index 0f31935..c6c71f6 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/algebra/operators/physical/CommitRuntime.java
@@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.transactions.ILogManager;
+import org.apache.asterix.common.transactions.ILogMarkerCallback;
import org.apache.asterix.common.transactions.ITransactionContext;
import org.apache.asterix.common.transactions.ITransactionManager;
import org.apache.asterix.common.transactions.JobId;
@@ -32,12 +33,16 @@ import org.apache.asterix.common.transactions.LogType;
import org.apache.asterix.common.utils.TransactionUtil;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
import org.apache.hyracks.api.comm.IFrameWriter;
+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.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.dataflow.common.io.MessagingFrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.storage.am.bloomfilter.impls.MurmurHash128Bit;
public class CommitRuntime implements IPushRuntime {
@@ -52,17 +57,18 @@ public class CommitRuntime implements IPushRuntime {
protected final boolean isTemporaryDatasetWriteJob;
protected final boolean isWriteTransaction;
protected final long[] longHashes;
- protected final LogRecord logRecord;
protected final FrameTupleReference frameTupleReference;
-
+ protected final IHyracksTaskContext ctx;
+ protected final int resourcePartition;
protected ITransactionContext transactionContext;
+ protected LogRecord logRecord;
protected FrameTupleAccessor frameTupleAccessor;
- protected final int resourcePartition;
public CommitRuntime(IHyracksTaskContext ctx, JobId jobId, int datasetId, int[] primaryKeyFields,
boolean isTemporaryDatasetWriteJob, boolean isWriteTransaction, int resourcePartition) {
- IAsterixAppRuntimeContext runtimeCtx = (IAsterixAppRuntimeContext) ctx.getJobletContext()
- .getApplicationContext().getApplicationObject();
+ this.ctx = ctx;
+ IAsterixAppRuntimeContext runtimeCtx =
+ (IAsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject();
this.transactionManager = runtimeCtx.getTransactionSubsystem().getTransactionManager();
this.logMgr = runtimeCtx.getTransactionSubsystem().getLogManager();
this.jobId = jobId;
@@ -73,7 +79,6 @@ public class CommitRuntime implements IPushRuntime {
this.isWriteTransaction = isWriteTransaction;
this.resourcePartition = resourcePartition;
longHashes = new long[2];
- logRecord = new LogRecord();
}
@Override
@@ -81,6 +86,9 @@ public class CommitRuntime implements IPushRuntime {
try {
transactionContext = transactionManager.getTransactionContext(jobId, false);
transactionContext.setWriteTxn(isWriteTransaction);
+ ILogMarkerCallback callback =
+ TaskUtils.<ILogMarkerCallback> get(ILogMarkerCallback.KEY_MARKER_CALLBACK, ctx);
+ logRecord = new LogRecord(callback);
} catch (ACIDException e) {
throw new HyracksDataException(e);
}
@@ -113,6 +121,23 @@ public class CommitRuntime implements IPushRuntime {
}
}
}
+ VSizeFrame message = TaskUtils.<VSizeFrame> get(HyracksConstants.KEY_MESSAGE, ctx);
+ if (message != null
+ && MessagingFrameTupleAppender.getMessageType(message) == MessagingFrameTupleAppender.MARKER_MESSAGE) {
+ try {
+ formMarkerLogRecords(message.getBuffer());
+ logMgr.log(logRecord);
+ } catch (ACIDException e) {
+ throw new HyracksDataException(e);
+ }
+ message.reset();
+ message.getBuffer().put(MessagingFrameTupleAppender.NULL_FEED_MESSAGE);
+ message.getBuffer().flip();
+ }
+ }
+
+ private void formMarkerLogRecords(ByteBuffer marker) {
+ TransactionUtil.formMarkerLogRecord(logRecord, transactionContext, datasetId, resourcePartition, marker);
}
protected void formLogRecord(ByteBuffer buffer, int t) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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
new file mode 100644
index 0000000..5c3aefe
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
@@ -0,0 +1,481 @@
+/*
+ * 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.app.bootstrap;
+
+import java.io.File;
+import java.util.Map;
+import java.util.logging.Logger;
+
+import org.apache.asterix.algebra.operators.physical.CommitRuntime;
+import org.apache.asterix.api.common.AsterixAppRuntimeContext;
+import org.apache.asterix.api.common.AsterixHyracksIntegrationUtil;
+import org.apache.asterix.app.external.TestLibrarian;
+import org.apache.asterix.common.api.ILocalResourceMetadata;
+import org.apache.asterix.common.config.AsterixTransactionProperties;
+import org.apache.asterix.common.context.AsterixVirtualBufferCacheProvider;
+import org.apache.asterix.common.context.DatasetLifecycleManager;
+import org.apache.asterix.common.context.TransactionSubsystemProvider;
+import org.apache.asterix.common.dataflow.AsterixLSMInsertDeleteOperatorNodePushable;
+import org.apache.asterix.common.dataflow.AsterixLSMTreeInsertDeleteOperatorDescriptor;
+import org.apache.asterix.common.exceptions.ACIDException;
+import org.apache.asterix.common.exceptions.AsterixException;
+import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory;
+import org.apache.asterix.common.transactions.IRecoveryManager.ResourceType;
+import org.apache.asterix.common.transactions.ITransactionManager;
+import org.apache.asterix.formats.nontagged.AqlBinaryComparatorFactoryProvider;
+import org.apache.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
+import org.apache.asterix.formats.nontagged.AqlTypeTraitProvider;
+import org.apache.asterix.metadata.entities.Dataset;
+import org.apache.asterix.metadata.utils.DatasetUtils;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.runtime.formats.NonTaggedDataFormat;
+import org.apache.asterix.test.runtime.ExecutionTestUtil;
+import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexModificationOperationCallbackFactory;
+import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexOperationTrackerProvider;
+import org.apache.asterix.transaction.management.resource.LSMBTreeLocalResourceMetadata;
+import org.apache.asterix.transaction.management.resource.PersistentLocalResourceFactoryProvider;
+import org.apache.asterix.transaction.management.service.logging.LogReader;
+import org.apache.asterix.transaction.management.service.transaction.AsterixRuntimeComponentsProvider;
+import org.apache.asterix.transaction.management.service.transaction.TransactionSubsystem;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
+import org.apache.hyracks.algebricks.runtime.operators.std.EmptyTupleSourceRuntimeFactory;
+import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksJobletContext;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.exceptions.HyracksException;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.JobSpecification;
+import org.apache.hyracks.api.util.HyracksConstants;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
+import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider;
+import org.apache.hyracks.dataflow.std.file.FileSplit;
+import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
+import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorNodePushable;
+import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
+import org.apache.hyracks.storage.am.common.dataflow.IndexDataflowHelper;
+import org.apache.hyracks.storage.am.common.dataflow.TreeIndexCreateOperatorDescriptor;
+import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallbackFactory;
+import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
+import org.apache.hyracks.storage.am.lsm.btree.dataflow.LSMBTreeDataflowHelper;
+import org.apache.hyracks.storage.am.lsm.btree.dataflow.LSMBTreeDataflowHelperFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
+import org.apache.hyracks.storage.common.file.ILocalResourceFactoryProvider;
+import org.apache.hyracks.storage.common.file.LocalResource;
+import org.apache.hyracks.test.support.TestUtils;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+public class TestNodeController {
+ protected static final Logger LOGGER = Logger.getLogger(TestNodeController.class.getName());
+
+ protected static final String PATH_ACTUAL = "unittest" + File.separator;
+ protected static final String PATH_BASE =
+ StringUtils.join(new String[] { "src", "test", "resources", "nodetests" }, File.separator);
+
+ protected static final String TEST_CONFIG_FILE_NAME = "asterix-build-configuration.xml";
+ protected static AsterixTransactionProperties txnProperties;
+ private static final boolean cleanupOnStart = true;
+ private static final boolean cleanupOnStop = true;
+
+ // Constants
+ 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 = new TransactionSubsystemProvider();
+ // Mutables
+ private JobId jobId;
+ private long jobCounter = 0L;
+ private IHyracksJobletContext jobletCtx;
+
+ public TestNodeController() throws AsterixException, HyracksException, ACIDException {
+ }
+
+ public void init() throws Exception {
+ try {
+ File outdir = new File(PATH_ACTUAL);
+ outdir.mkdirs();
+ // remove library directory
+ TestLibrarian.removeLibraryDir();
+ ExecutionTestUtil.setUp(cleanupOnStart);
+ } catch (Throwable th) {
+ th.printStackTrace();
+ throw th;
+ }
+ jobletCtx = Mockito.mock(IHyracksJobletContext.class);
+ Mockito.when(jobletCtx.getApplicationContext())
+ .thenReturn(AsterixHyracksIntegrationUtil.ncs[0].getApplicationContext());
+ Mockito.when(jobletCtx.getJobId()).thenAnswer(new Answer<JobId>() {
+ @Override
+ public JobId answer(InvocationOnMock invocation) throws Throwable {
+ return jobId;
+ }
+ });
+ }
+
+ public void deInit() throws Exception {
+ TestLibrarian.removeLibraryDir();
+ ExecutionTestUtil.tearDown(cleanupOnStop);
+ }
+
+ public org.apache.asterix.common.transactions.JobId getTxnJobId() {
+ return new org.apache.asterix.common.transactions.JobId((int) jobId.getId());
+ }
+
+ public AsterixLSMInsertDeleteOperatorNodePushable getInsertPipeline(IHyracksTaskContext ctx, Dataset dataset,
+ IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType,
+ ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties, int[] filterFields)
+ throws AlgebricksException {
+ PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
+ mergePolicyFactory, mergePolicyProperties, filterFields);
+ IndexOperation op = IndexOperation.INSERT;
+ IModificationOperationCallbackFactory modOpCallbackFactory =
+ new PrimaryIndexModificationOperationCallbackFactory(getTxnJobId(), dataset.getDatasetId(),
+ primaryIndexInfo.primaryKeyIndexes, TXN_SUBSYSTEM_PROVIDER, op, ResourceType.LSM_BTREE, true);
+ AsterixLSMTreeInsertDeleteOperatorDescriptor indexOpDesc =
+ getInsertOpratorDesc(primaryIndexInfo, modOpCallbackFactory);
+ LSMBTreeDataflowHelperFactory dataflowHelperFactory =
+ getPrimaryIndexDataflowHelperFactory(ctx, primaryIndexInfo);
+ Mockito.when(indexOpDesc.getIndexDataflowHelperFactory()).thenReturn(dataflowHelperFactory);
+ IRecordDescriptorProvider recordDescProvider = primaryIndexInfo.getInsertRecordDescriptorProvider();
+ AsterixLSMInsertDeleteOperatorNodePushable insertOp =
+ new AsterixLSMInsertDeleteOperatorNodePushable(indexOpDesc, ctx, PARTITION,
+ primaryIndexInfo.primaryIndexInsertFieldsPermutations, recordDescProvider, op, true);
+ CommitRuntime commitOp = new CommitRuntime(ctx, getTxnJobId(), dataset.getDatasetId(),
+ primaryIndexInfo.primaryKeyIndexes, false, true, PARTITION);
+ insertOp.setOutputFrameWriter(0, commitOp, primaryIndexInfo.rDesc);
+ commitOp.setInputRecordDescriptor(0, primaryIndexInfo.rDesc);
+ return insertOp;
+ }
+
+ public IPushRuntime getFullScanPipeline(IFrameWriter countOp, IHyracksTaskContext ctx, Dataset dataset,
+ IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType,
+ NoMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties, int[] filterFields)
+ throws HyracksDataException, AlgebricksException {
+ IPushRuntime emptyTupleOp = new EmptyTupleSourceRuntimeFactory().createPushRuntime(ctx);
+ JobSpecification spec = new JobSpecification();
+ PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
+ mergePolicyFactory, mergePolicyProperties, filterFields);
+ LSMBTreeDataflowHelperFactory indexDataflowHelperFactory =
+ getPrimaryIndexDataflowHelperFactory(ctx, primaryIndexInfo);
+ BTreeSearchOperatorDescriptor searchOpDesc = new BTreeSearchOperatorDescriptor(spec, primaryIndexInfo.rDesc,
+ AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER, AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER,
+ primaryIndexInfo.fileSplitProvider, primaryIndexInfo.primaryIndexTypeTraits,
+ primaryIndexInfo.primaryIndexComparatorFactories, primaryIndexInfo.primaryIndexBloomFilterKeyFields,
+ primaryIndexInfo.primaryKeyIndexes, primaryIndexInfo.primaryKeyIndexes, true, true,
+ indexDataflowHelperFactory, false, false, null, NoOpOperationCallbackFactory.INSTANCE, filterFields,
+ filterFields);
+ BTreeSearchOperatorNodePushable searchOp = new BTreeSearchOperatorNodePushable(searchOpDesc, ctx, 0,
+ primaryIndexInfo.getSearchRecordDescriptorProvider(), /*primaryIndexInfo.primaryKeyIndexes*/null,
+ /*primaryIndexInfo.primaryKeyIndexes*/null, true, true, filterFields, filterFields);
+ emptyTupleOp.setFrameWriter(0, searchOp,
+ primaryIndexInfo.getSearchRecordDescriptorProvider().getInputRecordDescriptor(null, 0));
+ searchOp.setOutputFrameWriter(0, countOp, primaryIndexInfo.rDesc);
+ return emptyTupleOp;
+ }
+
+ public LogReader getTransactionLogReader(boolean isRecoveryMode) {
+ return (LogReader) getTransactionSubsystem().getLogManager().getLogReader(isRecoveryMode);
+ }
+
+ public JobId newJobId() {
+ jobId = new JobId(jobCounter++);
+ return jobId;
+ }
+
+ public AsterixLSMTreeInsertDeleteOperatorDescriptor getInsertOpratorDesc(PrimaryIndexInfo primaryIndexInfo,
+ IModificationOperationCallbackFactory modOpCallbackFactory) {
+ AsterixLSMTreeInsertDeleteOperatorDescriptor indexOpDesc =
+ Mockito.mock(AsterixLSMTreeInsertDeleteOperatorDescriptor.class);
+ Mockito.when(indexOpDesc.getLifecycleManagerProvider())
+ .thenReturn(AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER);
+ Mockito.when(indexOpDesc.getStorageManager()).thenReturn(AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER);
+ Mockito.when(indexOpDesc.getFileSplitProvider()).thenReturn(primaryIndexInfo.fileSplitProvider);
+ Mockito.when(indexOpDesc.getLocalResourceFactoryProvider())
+ .thenReturn(primaryIndexInfo.localResourceFactoryProvider);
+ Mockito.when(indexOpDesc.getTreeIndexTypeTraits()).thenReturn(primaryIndexInfo.primaryIndexTypeTraits);
+ Mockito.when(indexOpDesc.getTreeIndexComparatorFactories())
+ .thenReturn(primaryIndexInfo.primaryIndexComparatorFactories);
+ Mockito.when(indexOpDesc.getTreeIndexBloomFilterKeyFields())
+ .thenReturn(primaryIndexInfo.primaryIndexBloomFilterKeyFields);
+ Mockito.when(indexOpDesc.getModificationOpCallbackFactory()).thenReturn(modOpCallbackFactory);
+ return indexOpDesc;
+ }
+
+ public TreeIndexCreateOperatorDescriptor getIndexCreateOpDesc(PrimaryIndexInfo primaryIndexInfo) {
+ TreeIndexCreateOperatorDescriptor indexOpDesc = Mockito.mock(TreeIndexCreateOperatorDescriptor.class);
+ Mockito.when(indexOpDesc.getLifecycleManagerProvider())
+ .thenReturn(AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER);
+ Mockito.when(indexOpDesc.getStorageManager()).thenReturn(AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER);
+ Mockito.when(indexOpDesc.getFileSplitProvider()).thenReturn(primaryIndexInfo.fileSplitProvider);
+ Mockito.when(indexOpDesc.getLocalResourceFactoryProvider())
+ .thenReturn(primaryIndexInfo.localResourceFactoryProvider);
+ Mockito.when(indexOpDesc.getTreeIndexTypeTraits()).thenReturn(primaryIndexInfo.primaryIndexTypeTraits);
+ Mockito.when(indexOpDesc.getTreeIndexComparatorFactories())
+ .thenReturn(primaryIndexInfo.primaryIndexComparatorFactories);
+ Mockito.when(indexOpDesc.getTreeIndexBloomFilterKeyFields())
+ .thenReturn(primaryIndexInfo.primaryIndexBloomFilterKeyFields);
+ return indexOpDesc;
+ }
+
+ public ConstantFileSplitProvider getFileSplitProvider(Dataset dataset) {
+ FileSplit fileSplit = new FileSplit(AsterixHyracksIntegrationUtil.ncs[0].getId(),
+ dataset.getDataverseName() + File.separator + dataset.getDatasetName());
+ return new ConstantFileSplitProvider(new FileSplit[] { fileSplit });
+ }
+
+ public ILocalResourceFactoryProvider getPrimaryIndexLocalResourceMetadataProvider(Dataset dataset,
+ ITypeTraits[] primaryIndexTypeTraits, IBinaryComparatorFactory[] primaryIndexComparatorFactories,
+ int[] primaryIndexBloomFilterKeyFields, ILSMMergePolicyFactory mergePolicyFactory,
+ Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
+ IBinaryComparatorFactory[] filterCmpFactories, int[] btreeFields, int[] filterFields) {
+ ILocalResourceMetadata localResourceMetadata =
+ new LSMBTreeLocalResourceMetadata(primaryIndexTypeTraits, primaryIndexComparatorFactories,
+ primaryIndexBloomFilterKeyFields, true, dataset.getDatasetId(), mergePolicyFactory,
+ mergePolicyProperties, filterTypeTraits, filterCmpFactories, btreeFields, filterFields);
+ ILocalResourceFactoryProvider localResourceFactoryProvider =
+ new PersistentLocalResourceFactoryProvider(localResourceMetadata, LocalResource.LSMBTreeResource);
+ return localResourceFactoryProvider;
+ }
+
+ public LSMBTreeDataflowHelper getPrimaryIndexDataflowHelper(IHyracksTaskContext ctx,
+ PrimaryIndexInfo primaryIndexInfo, TreeIndexCreateOperatorDescriptor indexOpDesc)
+ throws AlgebricksException {
+ LSMBTreeDataflowHelperFactory dataflowHelperFactory = new LSMBTreeDataflowHelperFactory(
+ new AsterixVirtualBufferCacheProvider(primaryIndexInfo.dataset.getDatasetId()),
+ primaryIndexInfo.mergePolicyFactory, primaryIndexInfo.mergePolicyProperties,
+ new PrimaryIndexOperationTrackerProvider(primaryIndexInfo.dataset.getDatasetId()),
+ AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER, LSMBTreeIOOperationCallbackFactory.INSTANCE,
+ BLOOM_FILTER_FALSE_POSITIVE_RATE, true, primaryIndexInfo.filterTypeTraits,
+ primaryIndexInfo.filterCmpFactories, primaryIndexInfo.btreeFields, primaryIndexInfo.filterFields, true);
+ IndexDataflowHelper dataflowHelper =
+ dataflowHelperFactory.createIndexDataflowHelper(indexOpDesc, ctx, PARTITION);
+ return (LSMBTreeDataflowHelper) dataflowHelper;
+ }
+
+ public LSMBTreeDataflowHelperFactory getPrimaryIndexDataflowHelperFactory(IHyracksTaskContext ctx,
+ PrimaryIndexInfo primaryIndexInfo) throws AlgebricksException {
+ return new LSMBTreeDataflowHelperFactory(
+ new AsterixVirtualBufferCacheProvider(primaryIndexInfo.dataset.getDatasetId()),
+ primaryIndexInfo.mergePolicyFactory, primaryIndexInfo.mergePolicyProperties,
+ new PrimaryIndexOperationTrackerProvider(primaryIndexInfo.dataset.getDatasetId()),
+ AsterixRuntimeComponentsProvider.RUNTIME_PROVIDER, LSMBTreeIOOperationCallbackFactory.INSTANCE,
+ BLOOM_FILTER_FALSE_POSITIVE_RATE, true, primaryIndexInfo.filterTypeTraits,
+ primaryIndexInfo.filterCmpFactories, primaryIndexInfo.btreeFields, primaryIndexInfo.filterFields, true);
+ }
+
+ public LSMBTreeDataflowHelper getPrimaryIndexDataflowHelper(Dataset dataset, IAType[] primaryKeyTypes,
+ ARecordType recordType, ARecordType metaType, ILSMMergePolicyFactory mergePolicyFactory,
+ Map<String, String> mergePolicyProperties, int[] filterFields)
+ throws AlgebricksException, HyracksDataException {
+ PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
+ mergePolicyFactory, mergePolicyProperties, filterFields);
+ TreeIndexCreateOperatorDescriptor indexOpDesc = getIndexCreateOpDesc(primaryIndexInfo);
+ return getPrimaryIndexDataflowHelper(createTestContext(), primaryIndexInfo, indexOpDesc);
+ }
+
+ public void createPrimaryIndex(Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType,
+ ARecordType metaType, ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
+ int[] filterFields) throws AlgebricksException, HyracksDataException {
+ PrimaryIndexInfo primaryIndexInfo = new PrimaryIndexInfo(dataset, primaryKeyTypes, recordType, metaType,
+ mergePolicyFactory, mergePolicyProperties, filterFields);
+ TreeIndexCreateOperatorDescriptor indexOpDesc = getIndexCreateOpDesc(primaryIndexInfo);
+ LSMBTreeDataflowHelper dataflowHelper =
+ getPrimaryIndexDataflowHelper(createTestContext(), primaryIndexInfo, indexOpDesc);
+ dataflowHelper.create();
+ }
+
+ private int[] createPrimaryIndexBloomFilterFields(int length) {
+ int[] primaryIndexBloomFilterKeyFields = new int[length];
+ for (int j = 0; j < length; ++j) {
+ primaryIndexBloomFilterKeyFields[j] = j;
+ }
+ return primaryIndexBloomFilterKeyFields;
+ }
+
+ private IBinaryComparatorFactory[] createPrimaryIndexComparatorFactories(IAType[] primaryKeyTypes) {
+ IBinaryComparatorFactory[] primaryIndexComparatorFactories =
+ new IBinaryComparatorFactory[primaryKeyTypes.length];
+ for (int j = 0; j < primaryKeyTypes.length; ++j) {
+ primaryIndexComparatorFactories[j] =
+ AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(primaryKeyTypes[j], true);
+ }
+ return primaryIndexComparatorFactories;
+ }
+
+ private ISerializerDeserializer<?>[] createPrimaryIndexSerdes(int primaryIndexNumOfTupleFields,
+ IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType) {
+ int i = 0;
+ ISerializerDeserializer<?>[] primaryIndexSerdes = new ISerializerDeserializer<?>[primaryIndexNumOfTupleFields];
+ for (; i < primaryKeyTypes.length; i++) {
+ primaryIndexSerdes[i] =
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(primaryKeyTypes[i]);
+ }
+ primaryIndexSerdes[i++] = AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(recordType);
+ if (metaType != null) {
+ primaryIndexSerdes[i] = AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(metaType);
+ }
+ return primaryIndexSerdes;
+ }
+
+ private ITypeTraits[] createPrimaryIndexTypeTraits(int primaryIndexNumOfTupleFields, IAType[] primaryKeyTypes,
+ ARecordType recordType, ARecordType metaType) {
+ ITypeTraits[] primaryIndexTypeTraits = new ITypeTraits[primaryIndexNumOfTupleFields];
+ int i = 0;
+ for (; i < primaryKeyTypes.length; i++) {
+ primaryIndexTypeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(primaryKeyTypes[i]);
+ }
+ primaryIndexTypeTraits[i++] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(recordType);
+ if (metaType != null) {
+ primaryIndexTypeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(metaType);
+ }
+ return primaryIndexTypeTraits;
+ }
+
+ public IHyracksTaskContext createTestContext() throws HyracksDataException {
+ IHyracksTaskContext ctx = TestUtils.create(KB32);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, new VSizeFrame(ctx), ctx);
+ ctx = Mockito.spy(ctx);
+ Mockito.when(ctx.getJobletContext()).thenReturn(jobletCtx);
+ Mockito.when(ctx.getIOManager())
+ .thenReturn(AsterixHyracksIntegrationUtil.ncs[0].getRootContext().getIOManager());
+ return ctx;
+ }
+
+ public TransactionSubsystem getTransactionSubsystem() {
+ return (TransactionSubsystem) ((AsterixAppRuntimeContext) AsterixHyracksIntegrationUtil.ncs[0]
+ .getApplicationContext().getApplicationObject()).getTransactionSubsystem();
+ }
+
+ public ITransactionManager getTransactionManager() {
+ return getTransactionSubsystem().getTransactionManager();
+ }
+
+ public AsterixAppRuntimeContext getAppRuntimeContext() {
+ return (AsterixAppRuntimeContext) AsterixHyracksIntegrationUtil.ncs[0].getApplicationContext()
+ .getApplicationObject();
+ }
+
+ public DatasetLifecycleManager getDatasetLifecycleManager() {
+ return (DatasetLifecycleManager) getAppRuntimeContext().getDatasetLifecycleManager();
+ }
+
+ @SuppressWarnings("unused")
+ private class PrimaryIndexInfo {
+ private Dataset dataset;
+ private IAType[] primaryKeyTypes;
+ private ARecordType recordType;
+ private ARecordType metaType;
+ private ILSMMergePolicyFactory mergePolicyFactory;
+ private Map<String, String> mergePolicyProperties;
+ private int[] filterFields;
+ private int primaryIndexNumOfTupleFields;
+ private IBinaryComparatorFactory[] primaryIndexComparatorFactories;
+ private ITypeTraits[] primaryIndexTypeTraits;
+ private ISerializerDeserializer<?>[] primaryIndexSerdes;
+ private int[] primaryIndexBloomFilterKeyFields;
+ private ITypeTraits[] filterTypeTraits;
+ private IBinaryComparatorFactory[] filterCmpFactories;
+ private int[] btreeFields;
+ private ILocalResourceFactoryProvider localResourceFactoryProvider;
+ private ConstantFileSplitProvider fileSplitProvider;
+ private RecordDescriptor rDesc;
+ private int[] primaryIndexInsertFieldsPermutations;
+ private int[] primaryKeyIndexes;
+
+ public PrimaryIndexInfo(Dataset dataset, IAType[] primaryKeyTypes, ARecordType recordType, ARecordType metaType,
+ ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
+ int[] filterFields) throws AlgebricksException {
+ this.dataset = dataset;
+ this.primaryKeyTypes = primaryKeyTypes;
+ this.recordType = recordType;
+ this.metaType = metaType;
+ this.mergePolicyFactory = mergePolicyFactory;
+ this.mergePolicyProperties = mergePolicyProperties;
+ this.filterFields = filterFields;
+ primaryIndexNumOfTupleFields = primaryKeyTypes.length + (1 + ((metaType == null) ? 0 : 1));
+ primaryIndexTypeTraits =
+ createPrimaryIndexTypeTraits(primaryIndexNumOfTupleFields, primaryKeyTypes, recordType, metaType);
+ primaryIndexComparatorFactories = createPrimaryIndexComparatorFactories(primaryKeyTypes);
+ primaryIndexBloomFilterKeyFields = createPrimaryIndexBloomFilterFields(primaryKeyTypes.length);
+ filterTypeTraits = DatasetUtils.computeFilterTypeTraits(dataset, recordType);
+ filterCmpFactories = DatasetUtils.computeFilterBinaryComparatorFactories(dataset, recordType,
+ NonTaggedDataFormat.INSTANCE.getBinaryComparatorFactoryProvider());
+ btreeFields = DatasetUtils.createBTreeFieldsWhenThereisAFilter(dataset);
+ localResourceFactoryProvider = getPrimaryIndexLocalResourceMetadataProvider(dataset, primaryIndexTypeTraits,
+ primaryIndexComparatorFactories, primaryIndexBloomFilterKeyFields, mergePolicyFactory,
+ mergePolicyProperties, filterTypeTraits, filterCmpFactories, btreeFields, filterFields);
+ fileSplitProvider = getFileSplitProvider(dataset);
+ primaryIndexSerdes =
+ createPrimaryIndexSerdes(primaryIndexNumOfTupleFields, primaryKeyTypes, recordType, metaType);
+ rDesc = new RecordDescriptor(primaryIndexSerdes, primaryIndexTypeTraits);
+ primaryIndexInsertFieldsPermutations = new int[primaryIndexNumOfTupleFields];
+ for (int i = 0; i < primaryIndexNumOfTupleFields; i++) {
+ primaryIndexInsertFieldsPermutations[i] = i;
+ }
+ primaryKeyIndexes = new int[primaryKeyTypes.length];
+ for (int i = 0; i < primaryKeyIndexes.length; i++) {
+ primaryKeyIndexes[i] = i;
+ }
+ }
+
+ public IRecordDescriptorProvider getInsertRecordDescriptorProvider() {
+ IRecordDescriptorProvider rDescProvider = Mockito.mock(IRecordDescriptorProvider.class);
+ Mockito.when(rDescProvider.getInputRecordDescriptor(Mockito.any(), Mockito.anyInt())).thenReturn(rDesc);
+ return rDescProvider;
+ }
+
+ public IRecordDescriptorProvider getSearchRecordDescriptorProvider() {
+ ITypeTraits[] primaryKeyTypeTraits = new ITypeTraits[primaryKeyTypes.length];
+ ISerializerDeserializer<?>[] primaryKeySerdes = new ISerializerDeserializer<?>[primaryKeyTypes.length];
+ for (int i = 0; i < primaryKeyTypes.length; i++) {
+ primaryKeyTypeTraits[i] = AqlTypeTraitProvider.INSTANCE.getTypeTrait(primaryKeyTypes[i]);
+ primaryKeySerdes[i] =
+ AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(primaryKeyTypes[i]);
+ }
+ RecordDescriptor searcgRecDesc = new RecordDescriptor(primaryKeySerdes, primaryKeyTypeTraits);
+ IRecordDescriptorProvider rDescProvider = Mockito.mock(IRecordDescriptorProvider.class);
+ Mockito.when(rDescProvider.getInputRecordDescriptor(Mockito.any(), Mockito.anyInt()))
+ .thenReturn(searcgRecDesc);
+ return rDescProvider;
+ }
+ }
+
+ public RecordDescriptor getSearchOutputDesc(IAType[] keyTypes, ARecordType recordType, ARecordType metaType) {
+ int primaryIndexNumOfTupleFields = keyTypes.length + (1 + ((metaType == null) ? 0 : 1));
+ ITypeTraits[] primaryIndexTypeTraits =
+ createPrimaryIndexTypeTraits(primaryIndexNumOfTupleFields, keyTypes, recordType, metaType);
+ ISerializerDeserializer<?>[] primaryIndexSerdes =
+ createPrimaryIndexSerdes(primaryIndexNumOfTupleFields, keyTypes, recordType, metaType);
+ return new RecordDescriptor(primaryIndexSerdes, primaryIndexTypeTraits);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
new file mode 100644
index 0000000..2eba473
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ABooleanFieldValueGenerator.java
@@ -0,0 +1,94 @@
+/*
+ * 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.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Random;
+
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.om.types.ATypeTag;
+
+public class ABooleanFieldValueGenerator implements IAsterixFieldValueGenerator<Boolean> {
+ private final GenerationFunction generationFunction;
+ private final boolean tagged;
+ private final Random rand = new Random();
+ private boolean value;
+
+ public ABooleanFieldValueGenerator(GenerationFunction generationFunction, boolean tagged) {
+ this.generationFunction = generationFunction;
+ this.tagged = tagged;
+ switch (generationFunction) {
+ case DECREASING:
+ value = true;
+ break;
+ case DETERMINISTIC:
+ value = false;
+ break;
+ case INCREASING:
+ value = false;
+ break;
+ case RANDOM:
+ value = rand.nextBoolean();
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void next(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG);
+ }
+ generate();
+ out.writeBoolean(value);
+ }
+
+ private void generate() {
+ switch (generationFunction) {
+ case DETERMINISTIC:
+ value = !value;
+ break;
+ case RANDOM:
+ value = rand.nextBoolean();
+ break;
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public Boolean next() throws IOException {
+ generate();
+ return value;
+ }
+
+ @Override
+ public void get(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_BOOLEAN_TYPE_TAG);
+ }
+ out.writeBoolean(value);
+ }
+
+ @Override
+ public Boolean get() throws IOException {
+ return value;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
new file mode 100644
index 0000000..e698676
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ADoubleFieldValueGenerator.java
@@ -0,0 +1,153 @@
+/*
+* 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.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.om.types.ATypeTag;
+
+public class ADoubleFieldValueGenerator implements IAsterixFieldValueGenerator<Double> {
+ private static final double START = 1000000000.0;
+ private static final int BATCH_SIZE = 1000;
+ private static final double INCREMENT = 0.1;
+ private final GenerationFunction generationFunction;
+ private final boolean unique;
+ private final boolean tagged;
+ private final Random rand = new Random();
+ private double value;
+ private int cycle;
+ private List<Double> uniques;
+ private Iterator<Double> iterator;
+
+ public ADoubleFieldValueGenerator(GenerationFunction generationFunction, boolean unique, boolean tagged) {
+ this.generationFunction = generationFunction;
+ this.unique = unique;
+ this.tagged = tagged;
+ switch (generationFunction) {
+ case DECREASING:
+ value = Integer.MAX_VALUE;
+ break;
+ case DETERMINISTIC:
+ value = START;
+ break;
+ case INCREASING:
+ value = 0;
+ break;
+ case RANDOM:
+ if (unique) {
+ double lowerBound = START;
+ double upperBound = lowerBound + (BATCH_SIZE * INCREMENT);
+ uniques = new ArrayList<>();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound += INCREMENT;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ }
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void next(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_DOUBLE_TYPE_TAG);
+ }
+ generate();
+ out.writeDouble(value);
+ }
+
+ private void generate() {
+ switch (generationFunction) {
+ case DECREASING:
+ value -= INCREMENT;
+ case DETERMINISTIC:
+ if (value >= START) {
+ cycle++;
+ value = START - (cycle * INCREMENT);
+ } else {
+ value = START + (cycle * INCREMENT);
+ }
+ break;
+ case INCREASING:
+ value += INCREMENT;
+ break;
+ case RANDOM:
+ if (unique) {
+ if (iterator.hasNext()) {
+ value = iterator.next();
+ } else {
+ // generate next patch
+ cycle++;
+ double lowerBound;
+ if (cycle % 2 == 0) {
+ // even
+ lowerBound = START + ((cycle / 2) * (BATCH_SIZE * INCREMENT));
+ } else {
+ // odd
+ lowerBound = START - ((cycle / 2 + 1) * (BATCH_SIZE * INCREMENT));
+ }
+ double upperBound = lowerBound + (BATCH_SIZE * INCREMENT);
+ uniques.clear();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound += INCREMENT;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ value = iterator.next();
+ }
+ } else {
+ value = rand.nextDouble();
+ }
+ break;
+ default:
+ break;
+
+ }
+ }
+
+ @Override
+ public Double next() throws IOException {
+ generate();
+ return value;
+ }
+
+ @Override
+ public void get(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_DOUBLE_TYPE_TAG);
+ }
+ out.writeDouble(value);
+ }
+
+ @Override
+ public Double get() throws IOException {
+ return value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
new file mode 100644
index 0000000..7c6556b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt32FieldValueGenerator.java
@@ -0,0 +1,152 @@
+/*
+* 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.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.om.types.ATypeTag;
+
+public class AInt32FieldValueGenerator implements IAsterixFieldValueGenerator<Integer> {
+ private static final int START = 1000000000;
+ private static final int BATCH_SIZE = 1000;
+ private final GenerationFunction generationFunction;
+ private final boolean unique;
+ private final boolean tagged;
+ private final Random rand = new Random();
+ private int value;
+ private int cycle;
+ private List<Integer> uniques;
+ private Iterator<Integer> iterator;
+
+ public AInt32FieldValueGenerator(GenerationFunction generationFunction, boolean unique, boolean tagged) {
+ this.generationFunction = generationFunction;
+ this.unique = unique;
+ this.tagged = tagged;
+ switch (generationFunction) {
+ case DECREASING:
+ value = Integer.MAX_VALUE;
+ break;
+ case DETERMINISTIC:
+ value = START;
+ break;
+ case INCREASING:
+ value = 0;
+ break;
+ case RANDOM:
+ if (unique) {
+ int lowerBound = START;
+ int upperBound = lowerBound + BATCH_SIZE;
+ uniques = new ArrayList<>();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound++;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ }
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void next(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_INT32_TYPE_TAG);
+ }
+ generate();
+ out.writeInt(value);
+ }
+
+ private void generate() {
+ switch (generationFunction) {
+ case DECREASING:
+ value--;
+ case DETERMINISTIC:
+ if (value >= START) {
+ cycle++;
+ value = START - cycle;
+ } else {
+ value = START + cycle;
+ }
+ break;
+ case INCREASING:
+ value++;
+ break;
+ case RANDOM:
+ if (unique) {
+ if (iterator.hasNext()) {
+ value = iterator.next();
+ } else {
+ // generate next patch
+ cycle++;
+ int lowerBound;
+ if (cycle % 2 == 0) {
+ // even
+ lowerBound = START + ((cycle / 2) * BATCH_SIZE);
+ } else {
+ // odd
+ lowerBound = START - ((cycle / 2 + 1) * BATCH_SIZE);
+ }
+ int upperBound = lowerBound + BATCH_SIZE;
+ uniques.clear();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound++;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ value = iterator.next();
+ }
+ } else {
+ value = rand.nextInt();
+ }
+ break;
+ default:
+ break;
+
+ }
+ }
+
+ @Override
+ public Integer next() throws IOException {
+ generate();
+ return value;
+ }
+
+ @Override
+ public void get(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_INT32_TYPE_TAG);
+ }
+ out.writeInt(value);
+ }
+
+ @Override
+ public Integer get() throws IOException {
+ return value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
new file mode 100644
index 0000000..2a2496e
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AInt64FieldValueGenerator.java
@@ -0,0 +1,152 @@
+/*
+* 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.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.om.types.ATypeTag;
+
+public class AInt64FieldValueGenerator implements IAsterixFieldValueGenerator<Long> {
+ private static final long START = 4500000000000000000L;
+ private static final long BATCH_SIZE = 1000L;
+ private final GenerationFunction generationFunction;
+ private final boolean unique;
+ private final boolean tagged;
+ private final Random rand = new Random();
+ private long value;
+ private int cycle;
+ private List<Long> uniques;
+ private Iterator<Long> iterator;
+
+ public AInt64FieldValueGenerator(GenerationFunction generationFunction, boolean unique, boolean tagged) {
+ this.generationFunction = generationFunction;
+ this.unique = unique;
+ this.tagged = tagged;
+ switch (generationFunction) {
+ case DECREASING:
+ value = Long.MAX_VALUE;
+ break;
+ case DETERMINISTIC:
+ value = START;
+ break;
+ case INCREASING:
+ value = 0L;
+ break;
+ case RANDOM:
+ if (unique) {
+ long lowerBound = START;
+ long upperBound = lowerBound + BATCH_SIZE;
+ uniques = new ArrayList<>();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound++;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ }
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void next(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_INT64_TYPE_TAG);
+ }
+ generate();
+ out.writeLong(value);
+ }
+
+ private void generate() {
+ switch (generationFunction) {
+ case DECREASING:
+ value--;
+ case DETERMINISTIC:
+ if (value >= START) {
+ cycle++;
+ value = START - cycle;
+ } else {
+ value = START + cycle;
+ }
+ break;
+ case INCREASING:
+ value++;
+ break;
+ case RANDOM:
+ if (unique) {
+ if (iterator.hasNext()) {
+ value = iterator.next();
+ } else {
+ // generate next patch
+ cycle++;
+ long lowerBound;
+ if (cycle % 2 == 0) {
+ // even
+ lowerBound = START + ((cycle / 2) * BATCH_SIZE);
+ } else {
+ // odd
+ lowerBound = START - ((cycle / 2 + 1) * BATCH_SIZE);
+ }
+ long upperBound = lowerBound + BATCH_SIZE;
+ uniques.clear();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound++;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ value = iterator.next();
+ }
+ } else {
+ value = rand.nextLong();
+ }
+ break;
+ default:
+ break;
+
+ }
+ }
+
+ @Override
+ public Long next() throws IOException {
+ generate();
+ return value;
+ }
+
+ @Override
+ public void get(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_INT64_TYPE_TAG);
+ }
+ out.writeLong(value);
+ }
+
+ @Override
+ public Long get() throws IOException {
+ return value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
new file mode 100644
index 0000000..df717e9
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/ARecordValueGenerator.java
@@ -0,0 +1,119 @@
+/*
+* 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.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.builders.RecordBuilder;
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.test.common.TestTupleReference;
+import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public class ARecordValueGenerator implements IAsterixFieldValueGenerator<ITupleReference> {
+ private final IAsterixFieldValueGenerator<?>[] generators;
+ private final boolean tagged;
+ private final ARecordType recordType;
+ private final RecordBuilder recBuilder;
+ private final ArrayBackedValueStorage fieldValueBuffer;
+ private final TestTupleReference tuple;
+
+ public ARecordValueGenerator(GenerationFunction[] generationFunctions, ARecordType recordType, boolean[] uniques,
+ boolean tagged) {
+ this.tagged = tagged;
+ this.recordType = recordType;
+ tuple = new TestTupleReference(1);
+ fieldValueBuffer = new ArrayBackedValueStorage();
+ recBuilder = new RecordBuilder();
+ recBuilder.reset(recordType);
+ recBuilder.init();
+ generators = new IAsterixFieldValueGenerator<?>[recordType.getFieldTypes().length];
+ for (int i = 0; i < recordType.getFieldTypes().length; i++) {
+ ATypeTag tag = recordType.getFieldTypes()[i].getTypeTag();
+ switch (tag) {
+ case BOOLEAN:
+ generators[i] = new ABooleanFieldValueGenerator(generationFunctions[i], true);
+ break;
+ case DOUBLE:
+ generators[i] = new ADoubleFieldValueGenerator(generationFunctions[i], uniques[i], true);
+ break;
+ case INT32:
+ generators[i] = new AInt32FieldValueGenerator(generationFunctions[i], uniques[i], true);
+ break;
+ case INT64:
+ generators[i] = new AInt64FieldValueGenerator(generationFunctions[i], uniques[i], true);
+ break;
+ case STRING:
+ generators[i] = new AStringFieldValueGenerator(generationFunctions[i], uniques[i], true);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported type " + tag);
+ }
+ }
+ }
+
+ @Override
+ public void next(DataOutput out) throws IOException {
+ recBuilder.reset(recordType);
+ recBuilder.init();
+ for (int i = 0; i < generators.length; i++) {
+ fieldValueBuffer.reset();
+ generators[i].next(fieldValueBuffer.getDataOutput());
+ recBuilder.addField(i, fieldValueBuffer);
+ }
+ recBuilder.write(out, tagged);
+ }
+
+ @Override
+ public ITupleReference next() throws IOException {
+ tuple.reset();
+ next(tuple.getFields()[0].getDataOutput());
+ return tuple;
+ }
+
+ @Override
+ public void get(DataOutput out) throws IOException {
+ recBuilder.reset(recordType);
+ recBuilder.init();
+ for (int i = 0; i < generators.length; i++) {
+ fieldValueBuffer.reset();
+ generators[i].get(fieldValueBuffer.getDataOutput());
+ recBuilder.addField(i, fieldValueBuffer);
+ }
+ recBuilder.write(out, tagged);
+ }
+
+ @Override
+ public ITupleReference get() throws IOException {
+ tuple.reset();
+ get(tuple.getFields()[0].getDataOutput());
+ return tuple;
+ }
+
+ public void get(int i, DataOutput out) throws IOException {
+ generators[i].get(out);
+ }
+
+ public Object get(int i) throws IOException {
+ return generators[i].get();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
new file mode 100644
index 0000000..5ee6d40
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/AStringFieldValueGenerator.java
@@ -0,0 +1,159 @@
+/*
+ * 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.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Random;
+
+import org.apache.asterix.app.data.gen.TupleGenerator.GenerationFunction;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
+import org.apache.hyracks.util.string.UTF8StringReader;
+import org.apache.hyracks.util.string.UTF8StringWriter;
+
+public class AStringFieldValueGenerator implements IAsterixFieldValueGenerator<String> {
+ private static final String PREFIX = "A String Value #";
+ private static final int START = 1000000000;
+ private static final int BATCH_SIZE = 1000;
+ private final GenerationFunction generationFunction;
+ private final boolean unique;
+ private final boolean tagged;
+ private final Random rand = new Random();
+ private int value;
+ private int cycle;
+ private List<Integer> uniques;
+ private Iterator<Integer> iterator;
+ private String aString;
+ private UTF8StringSerializerDeserializer stringSerde =
+ new UTF8StringSerializerDeserializer(new UTF8StringWriter(), new UTF8StringReader());
+
+ public AStringFieldValueGenerator(GenerationFunction generationFunction, boolean unique, boolean tagged) {
+ this.generationFunction = generationFunction;
+ this.unique = unique;
+ this.tagged = tagged;
+ switch (generationFunction) {
+ case DECREASING:
+ value = Integer.MAX_VALUE;
+ break;
+ case DETERMINISTIC:
+ value = START;
+ break;
+ case INCREASING:
+ value = 0;
+ break;
+ case RANDOM:
+ if (unique) {
+ int lowerBound = START;
+ int upperBound = lowerBound + BATCH_SIZE;
+ uniques = new ArrayList<>();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound++;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ }
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void next(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_STRING_TYPE_TAG);
+ }
+ generate();
+ stringSerde.serialize(aString, out);
+ }
+
+ private void generate() {
+ switch (generationFunction) {
+ case DECREASING:
+ value--;
+ case DETERMINISTIC:
+ if (value >= START) {
+ cycle++;
+ value = START - cycle;
+ } else {
+ value = START + cycle;
+ }
+ break;
+ case INCREASING:
+ value++;
+ break;
+ case RANDOM:
+ if (unique) {
+ if (iterator.hasNext()) {
+ value = iterator.next();
+ } else {
+ // generate next patch
+ cycle++;
+ int lowerBound;
+ if (cycle % 2 == 0) {
+ // even
+ lowerBound = START + ((cycle / 2) * BATCH_SIZE);
+ } else {
+ // odd
+ lowerBound = START - ((cycle / 2 + 1) * BATCH_SIZE);
+ }
+ int upperBound = lowerBound + BATCH_SIZE;
+ uniques.clear();
+ while (lowerBound < upperBound) {
+ uniques.add(lowerBound);
+ lowerBound++;
+ }
+ Collections.shuffle(uniques);
+ iterator = uniques.iterator();
+ value = iterator.next();
+ }
+ } else {
+ value = rand.nextInt();
+ }
+ break;
+ default:
+ break;
+ }
+ aString = PREFIX + String.format("%08d", value);
+ }
+
+ @Override
+ public String next() throws IOException {
+ generate();
+ return aString;
+ }
+
+ @Override
+ public void get(DataOutput out) throws IOException {
+ if (tagged) {
+ out.writeByte(ATypeTag.SERIALIZED_STRING_TYPE_TAG);
+ }
+ stringSerde.serialize(aString, out);
+ }
+
+ @Override
+ public String get() throws IOException {
+ return aString;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/IAsterixFieldValueGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/IAsterixFieldValueGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/IAsterixFieldValueGenerator.java
new file mode 100644
index 0000000..17bdcf8
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/IAsterixFieldValueGenerator.java
@@ -0,0 +1,48 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.apache.asterix.app.data.gen;
+
+import java.io.DataOutput;
+import java.io.IOException;
+
+public interface IAsterixFieldValueGenerator<T> {
+ /**
+ * @param out
+ * @throws IOException
+ */
+ public void next(DataOutput out) throws IOException;
+
+ /**
+ * @return
+ * @throws IOException
+ */
+ public T next() throws IOException;
+
+ /**
+ * @param out
+ * @throws IOException
+ */
+ public void get(DataOutput out) throws IOException;
+
+ /**
+ * @return
+ * @throws IOException
+ */
+ public T get() throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TestTupleCounterFrameWriter.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TestTupleCounterFrameWriter.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TestTupleCounterFrameWriter.java
new file mode 100644
index 0000000..ee3de51
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TestTupleCounterFrameWriter.java
@@ -0,0 +1,51 @@
+/*
+ * 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.app.data.gen;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.test.CountAnswer;
+import org.apache.hyracks.api.test.TestFrameWriter;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+
+public class TestTupleCounterFrameWriter extends TestFrameWriter {
+
+ private final FrameTupleAccessor accessor;
+ private int count = 0;
+
+ public TestTupleCounterFrameWriter(RecordDescriptor recordDescriptor, CountAnswer openAnswer,
+ CountAnswer nextAnswer, CountAnswer flushAnswer, CountAnswer failAnswer, CountAnswer closeAnswer,
+ boolean deepCopyInputFrames) {
+ super(openAnswer, nextAnswer, flushAnswer, failAnswer, closeAnswer, deepCopyInputFrames);
+ accessor = new FrameTupleAccessor(recordDescriptor);
+ }
+
+ @Override
+ public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+ super.nextFrame(buffer);
+ accessor.reset(buffer);
+ count += accessor.getTupleCount();
+ }
+
+ public int getCount() {
+ return count;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java
new file mode 100644
index 0000000..98c57a0
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/data/gen/TupleGenerator.java
@@ -0,0 +1,126 @@
+/*
+* 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.app.data.gen;
+
+import java.io.IOException;
+
+import org.apache.asterix.om.types.ARecordType;
+import org.apache.asterix.om.types.ATypeTag;
+import org.apache.asterix.om.types.IAType;
+import org.apache.asterix.test.common.TestTupleReference;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public class TupleGenerator {
+
+ private final int[] keyIndexes;
+ private final int[] keyIndicators;
+ private final ARecordValueGenerator recordGenerator;
+ private final ARecordValueGenerator metaGenerator;
+ private final TestTupleReference tuple;
+
+ public enum GenerationFunction {
+ RANDOM,
+ DETERMINISTIC,
+ INCREASING,
+ DECREASING
+ }
+
+ /**
+ * @param recordType
+ * @param metaType
+ * @param key
+ * @param keyIndexes
+ * @param keyIndicators
+ * @param recordGeneration
+ * @param uniqueRecordFields
+ * @param metaGeneration
+ * @param uniqueMetaFields
+ */
+ public TupleGenerator(ARecordType recordType, ARecordType metaType, int[] keyIndexes, int[] keyIndicators,
+ GenerationFunction[] recordGeneration, boolean[] uniqueRecordFields, GenerationFunction[] metaGeneration,
+ boolean[] uniqueMetaFields) {
+ this.keyIndexes = keyIndexes;
+ this.keyIndicators = keyIndicators;
+ for (IAType field : recordType.getFieldTypes()) {
+ validate(field);
+ }
+ recordGenerator = new ARecordValueGenerator(recordGeneration, recordType, uniqueRecordFields, true);
+ if (metaType != null) {
+ for (IAType field : metaType.getFieldTypes()) {
+ validate(field);
+ }
+ metaGenerator = new ARecordValueGenerator(metaGeneration, metaType, uniqueMetaFields, true);
+ } else {
+ metaGenerator = null;
+ }
+ int numOfFields = keyIndexes.length + 1 + ((metaType != null) ? 1 : 0);
+ tuple = new TestTupleReference(numOfFields);
+ boolean atLeastOneKeyFieldIsNotRandomAndNotBoolean = false;
+ for (int i = 0; i < keyIndexes.length; i++) {
+ if (keyIndicators[i] < 0 || keyIndicators[i] > 1) {
+ throw new IllegalArgumentException("key field indicator must be either 0 or 1");
+ }
+ atLeastOneKeyFieldIsNotRandomAndNotBoolean = atLeastOneKeyFieldIsNotRandomAndNotBoolean
+ || validateKey(keyIndexes[i], keyIndicators[i] == 0 ? recordType : metaType,
+ keyIndicators[i] == 0 ? uniqueRecordFields[i] : uniqueMetaFields[i]);
+ }
+ if (!atLeastOneKeyFieldIsNotRandomAndNotBoolean) {
+ throw new IllegalArgumentException("at least one key field must be unique and not boolean");
+ }
+ if (keyIndexes.length != keyIndicators.length) {
+ throw new IllegalArgumentException("number of key indexes must equals number of key indicators");
+ }
+ }
+
+ private boolean validateKey(int i, ARecordType type, boolean unique) {
+ if (type.getFieldNames().length <= i) {
+ throw new IllegalArgumentException("key index must be less than number of fields");
+ }
+ return unique && type.getFieldTypes()[i].getTypeTag() != ATypeTag.BOOLEAN;
+ }
+
+ public ITupleReference next() throws IOException {
+ tuple.reset();
+ recordGenerator.next(tuple.getFields()[keyIndexes.length].getDataOutput());
+ if (metaGenerator != null) {
+ recordGenerator.next(tuple.getFields()[keyIndexes.length + 1].getDataOutput());
+ }
+ for (int i = 0; i < keyIndexes.length; i++) {
+ if (keyIndicators[i] == 0) {
+ recordGenerator.get(keyIndexes[i], tuple.getFields()[i].getDataOutput());
+ } else {
+ metaGenerator.get(keyIndexes[i], tuple.getFields()[i].getDataOutput());
+ }
+ }
+ return tuple;
+ }
+
+ private void validate(IAType field) {
+ switch (field.getTypeTag()) {
+ case BOOLEAN:
+ case DOUBLE:
+ case INT32:
+ case INT64:
+ case STRING:
+ break;
+ default:
+ throw new IllegalArgumentException("Generating data of type " + field + " is not supported");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleGenerator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleGenerator.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleGenerator.java
index e267cc7..54bdb1b 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleGenerator.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleGenerator.java
@@ -96,33 +96,4 @@ public class TestTupleGenerator {
}
return tuple;
}
-
- private class TestTupleReference implements ITupleReference {
- private final GrowableArray[] fields;
-
- private TestTupleReference(GrowableArray[] fields) {
- this.fields = fields;
- }
-
- @Override
- public int getFieldCount() {
- return fields.length;
- }
-
- @Override
- public byte[] getFieldData(int fIdx) {
-
- return fields[fIdx].getByteArray();
- }
-
- @Override
- public int getFieldStart(int fIdx) {
- return 0;
- }
-
- @Override
- public int getFieldLength(int fIdx) {
- return fields[fIdx].getLength();
- }
- }
}
[3/4] asterixdb git commit: Add Test NodeController,
Test Data Generator, and Marker Logs
Posted by am...@apache.org.
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleReference.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleReference.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleReference.java
new file mode 100644
index 0000000..b676dfd
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/common/TestTupleReference.java
@@ -0,0 +1,74 @@
+/*
+ * 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.common;
+
+import org.apache.hyracks.data.std.util.GrowableArray;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+
+public class TestTupleReference implements ITupleReference {
+ private GrowableArray[] fields;
+ private int[] offsets;
+
+ public TestTupleReference(GrowableArray[] fields) {
+ this.fields = fields;
+ offsets = new int[fields.length];
+ }
+
+ public TestTupleReference(int numfields) {
+ this.fields = new GrowableArray[numfields];
+ for (int i = 0; i < numfields; i++) {
+ fields[i] = new GrowableArray();
+ }
+ offsets = new int[fields.length];
+ }
+
+ public GrowableArray[] getFields() {
+ return fields;
+ }
+
+ public void setFields(GrowableArray[] fields) {
+ this.fields = fields;
+ }
+
+ @Override
+ public int getFieldCount() {
+ return fields.length;
+ }
+
+ @Override
+ public byte[] getFieldData(int fIdx) {
+ return fields[fIdx].getByteArray();
+ }
+
+ @Override
+ public int getFieldStart(int fIdx) {
+ return offsets[fIdx];
+ }
+
+ @Override
+ public int getFieldLength(int fIdx) {
+ return fields[fIdx].getLength();
+ }
+
+ public void reset() {
+ for (GrowableArray field : fields) {
+ field.reset();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ConnectorDescriptorWithMessagingTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ConnectorDescriptorWithMessagingTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ConnectorDescriptorWithMessagingTest.java
index 2f712cc..a253ac0 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ConnectorDescriptorWithMessagingTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/ConnectorDescriptorWithMessagingTest.java
@@ -36,6 +36,7 @@ import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IConnectorDescriptorRegistry;
import org.apache.hyracks.api.test.TestFrameWriter;
+import org.apache.hyracks.api.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
@@ -44,6 +45,7 @@ import org.apache.hyracks.dataflow.common.data.marshalling.DoubleSerializerDeser
import org.apache.hyracks.dataflow.common.data.marshalling.Integer64SerializerDeserializer;
import org.apache.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import org.apache.hyracks.dataflow.common.io.MessagingFrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.dataflow.std.connectors.MToNPartitioningWithMessageConnectorDescriptor;
import org.apache.hyracks.dataflow.std.connectors.PartitionWithMessageDataWriter;
import org.apache.hyracks.test.support.TestUtils;
@@ -70,7 +72,7 @@ public class ConnectorDescriptorWithMessagingTest {
IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
VSizeFrame message = new VSizeFrame(ctx);
VSizeFrame tempBuffer = new VSizeFrame(ctx);
- ctx.setSharedObject(message);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, message, ctx);
message.getBuffer().clear();
message.getBuffer().put(MessagingFrameTupleAppender.NULL_FEED_MESSAGE);
message.getBuffer().flip();
@@ -144,8 +146,8 @@ public class ConnectorDescriptorWithMessagingTest {
IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
VSizeFrame message = new VSizeFrame(ctx);
VSizeFrame tempBuffer = new VSizeFrame(ctx);
- ctx.setSharedObject(message);
- writeRandomMessage(message, MessagingFrameTupleAppender.SNAPSHOT_MESSAGE, DEFAULT_FRAME_SIZE + 1);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, message, ctx);
+ writeRandomMessage(message, MessagingFrameTupleAppender.MARKER_MESSAGE, DEFAULT_FRAME_SIZE + 1);
ISerializerDeserializer<?>[] serdes = new ISerializerDeserializer<?>[] {
Integer64SerializerDeserializer.INSTANCE, DoubleSerializerDeserializer.INSTANCE,
BooleanSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer() };
@@ -165,7 +167,7 @@ public class ConnectorDescriptorWithMessagingTest {
fta.reset(writer.getLastFrame());
Assert.assertEquals(fta.getTupleCount(), 1);
FeedUtils.processFeedMessage(writer.getLastFrame(), tempBuffer, fta);
- Assert.assertEquals(MessagingFrameTupleAppender.SNAPSHOT_MESSAGE,
+ Assert.assertEquals(MessagingFrameTupleAppender.MARKER_MESSAGE,
MessagingFrameTupleAppender.getMessageType(tempBuffer));
}
message.getBuffer().clear();
@@ -228,9 +230,9 @@ public class ConnectorDescriptorWithMessagingTest {
IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
VSizeFrame message = new VSizeFrame(ctx);
VSizeFrame tempBuffer = new VSizeFrame(ctx);
- ctx.setSharedObject(message);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, message, ctx);
message.getBuffer().clear();
- writeRandomMessage(message, MessagingFrameTupleAppender.SNAPSHOT_MESSAGE, DEFAULT_FRAME_SIZE);
+ writeRandomMessage(message, MessagingFrameTupleAppender.MARKER_MESSAGE, DEFAULT_FRAME_SIZE);
ISerializerDeserializer<?>[] serdes = new ISerializerDeserializer<?>[] {
Integer64SerializerDeserializer.INSTANCE, DoubleSerializerDeserializer.INSTANCE,
BooleanSerializerDeserializer.INSTANCE, new UTF8StringSerializerDeserializer() };
@@ -264,7 +266,7 @@ public class ConnectorDescriptorWithMessagingTest {
fta.reset(writer.getLastFrame());
Assert.assertEquals(fta.getTupleCount(), 1);
FeedUtils.processFeedMessage(writer.getLastFrame(), tempBuffer, fta);
- Assert.assertEquals(MessagingFrameTupleAppender.SNAPSHOT_MESSAGE,
+ Assert.assertEquals(MessagingFrameTupleAppender.MARKER_MESSAGE,
MessagingFrameTupleAppender.getMessageType(tempBuffer));
}
partitioner.close();
@@ -286,7 +288,7 @@ public class ConnectorDescriptorWithMessagingTest {
IHyracksTaskContext ctx = TestUtils.create(DEFAULT_FRAME_SIZE);
VSizeFrame message = new VSizeFrame(ctx);
VSizeFrame tempBuffer = new VSizeFrame(ctx);
- ctx.setSharedObject(message);
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, message, ctx);
message.getBuffer().clear();
message.getBuffer().put(MessagingFrameTupleAppender.ACK_REQ_FEED_MESSAGE);
message.getBuffer().flip();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/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
new file mode 100644
index 0000000..a0ef31e
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/LogMarkerTest.java
@@ -0,0 +1,211 @@
+/*
+ * 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.io.File;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.asterix.app.bootstrap.TestNodeController;
+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.common.config.DatasetConfig.DatasetType;
+import org.apache.asterix.common.config.GlobalConfig;
+import org.apache.asterix.common.dataflow.AsterixLSMInsertDeleteOperatorNodePushable;
+import org.apache.asterix.common.transactions.DatasetId;
+import org.apache.asterix.common.transactions.ILogRecord;
+import org.apache.asterix.common.transactions.ITransactionContext;
+import org.apache.asterix.external.util.DataflowUtils;
+import org.apache.asterix.metadata.entities.Dataset;
+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.transaction.management.service.logging.LogReader;
+import org.apache.commons.io.FileUtils;
+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.test.CountAnswer;
+import org.apache.hyracks.api.test.FrameWriterTestUtils;
+import org.apache.hyracks.api.test.FrameWriterTestUtils.FrameWriterOperation;
+import org.apache.hyracks.api.util.HyracksConstants;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.dataflow.common.io.MessagingFrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
+import org.apache.hyracks.storage.am.lsm.btree.dataflow.LSMBTreeDataflowHelper;
+import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTree;
+import org.apache.hyracks.storage.am.lsm.common.impls.NoMergePolicyFactory;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class LogMarkerTest {
+
+ private static final String TEST_CONFIG_FILE_NAME = "asterix-build-configuration.xml";
+ 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 = { 0 };
+ private static final int NUM_OF_RECORDS = 100000;
+ private static final int SNAPSHOT_SIZE = 1000;
+ private static final int DATASET_ID = 101;
+ private static final String SPILL_AREA = "target" + File.separator + "spill_area";
+ 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";
+
+ @Before
+ public void setUp() throws Exception {
+ System.setProperty(GlobalConfig.CONFIG_FILE_PROPERTY, TEST_CONFIG_FILE_NAME);
+ System.out.println("SetUp: ");
+ File f = new File(System.getProperty("user.dir") + File.separator + "target" + File.separator + "txnLogDir");
+ FileUtils.deleteQuietly(f);
+ System.out.println("Dir " + f.getName() + " deleted");
+ f = new File(System.getProperty("user.dir") + File.separator + "target" + File.separator + "IODevice");
+ FileUtils.deleteQuietly(f);
+ System.out.println("Dir " + f.getName() + " deleted");
+ f = new File(System.getProperty("user.dir") + File.separator + SPILL_AREA);
+ FileUtils.deleteQuietly(f);
+ System.out.println("Dir " + f.getName() + " deleted");
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ System.out.println("TearDown");
+ File f = new File(System.getProperty("user.dir") + File.separator + "target" + File.separator + "txnLogDir");
+ FileUtils.deleteQuietly(f);
+ System.out.println("Dir " + f.getName() + " deleted");
+ f = new File(System.getProperty("user.dir") + File.separator + "target" + File.separator + "IODevice");
+ FileUtils.deleteQuietly(f);
+ System.out.println("Dir " + f.getName() + " deleted");
+ f = new File(System.getProperty("user.dir") + File.separator + SPILL_AREA);
+ FileUtils.deleteQuietly(f);
+ System.out.println("Dir " + f.getName() + " deleted");
+ }
+
+ @Test
+ public void testInsertWithSnapshot() {
+ try {
+ TestNodeController nc = new TestNodeController();
+ nc.init();
+ Dataset dataset = new Dataset(DATAVERSE_NAME, DATASET_NAME, DATAVERSE_NAME, DATA_TYPE_NAME,
+ NODE_GROUP_NAME, null, null, new InternalDatasetDetails(null, PartitioningStrategy.HASH,
+ Collections.emptyList(), null, null, null, false, null, false),
+ null, DatasetType.INTERNAL, DATASET_ID, 0);
+ try {
+ nc.createPrimaryIndex(dataset, KEY_TYPES, RECORD_TYPE, META_TYPE, new NoMergePolicyFactory(), null,
+ null);
+ IHyracksTaskContext ctx = nc.createTestContext();
+ nc.newJobId();
+ ITransactionContext txnCtx = nc.getTransactionManager().getTransactionContext(nc.getTxnJobId(), true);
+ AsterixLSMInsertDeleteOperatorNodePushable insertOp = nc.getInsertPipeline(ctx, dataset, KEY_TYPES,
+ RECORD_TYPE, META_TYPE, new NoMergePolicyFactory(), null, null);
+ 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);
+ VSizeFrame marker = new VSizeFrame(ctx);
+ FrameTupleAppender tupleAppender = new FrameTupleAppender(frame);
+ long markerId = 0L;
+ for (int j = 0; j < NUM_OF_RECORDS; j++) {
+ if (j % SNAPSHOT_SIZE == 0) {
+ marker.reset();
+ marker.getBuffer().put(MessagingFrameTupleAppender.MARKER_MESSAGE);
+ marker.getBuffer().putLong(markerId);
+ marker.getBuffer().flip();
+ markerId++;
+ TaskUtils.putInSharedMap(HyracksConstants.KEY_MESSAGE, marker, ctx);
+ tupleAppender.flush(insertOp);
+ }
+ ITupleReference tuple = tupleGenerator.next();
+ DataflowUtils.addTupleToFrame(tupleAppender, tuple, insertOp);
+ }
+ if (tupleAppender.getTupleCount() > 0) {
+ tupleAppender.write(insertOp, true);
+ }
+ insertOp.close();
+ nc.getTransactionManager().completedTransaction(txnCtx, new DatasetId(-1), -1, true);
+ LSMBTreeDataflowHelper dataflowHelper = nc.getPrimaryIndexDataflowHelper(dataset, KEY_TYPES,
+ RECORD_TYPE, META_TYPE, new NoMergePolicyFactory(), null, null);
+ dataflowHelper.open();
+ LSMBTree btree = (LSMBTree) dataflowHelper.getIndexInstance();
+ long lsn = btree.getMostRecentMarkerLSN();
+ int numOfMarkers = 0;
+ LogReader logReader = (LogReader) nc.getTransactionSubsystem().getLogManager().getLogReader(false);
+ long expectedMarkerId = markerId - 1;
+ while (lsn >= 0) {
+ numOfMarkers++;
+ ILogRecord logRecord = logReader.read(lsn);
+ lsn = logRecord.getPreviousMarkerLSN();
+ long logMarkerId = logRecord.getMarker().getLong();
+ Assert.assertEquals(expectedMarkerId, logMarkerId);
+ expectedMarkerId--;
+ }
+ logReader.close();
+ dataflowHelper.close();
+ Assert.assertEquals(markerId, numOfMarkers);
+ nc.newJobId();
+ 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, new NoMergePolicyFactory(), null, null);
+ emptyTupleOp.open();
+ emptyTupleOp.close();
+ Assert.assertEquals(NUM_OF_RECORDS, countOp.getCount());
+ } finally {
+ nc.deInit();
+ }
+ } catch (Throwable e) {
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+
+ }
+
+ public static TestTupleCounterFrameWriter create(RecordDescriptor recordDescriptor,
+ Collection<FrameWriterOperation> exceptionThrowingOperations,
+ Collection<FrameWriterOperation> errorThrowingOperations, boolean deepCopyInputFrames) {
+ CountAnswer openAnswer = FrameWriterTestUtils.createAnswer(FrameWriterOperation.Open,
+ exceptionThrowingOperations, errorThrowingOperations);
+ CountAnswer nextAnswer = FrameWriterTestUtils.createAnswer(FrameWriterOperation.NextFrame,
+ exceptionThrowingOperations, errorThrowingOperations);
+ CountAnswer flushAnswer = FrameWriterTestUtils.createAnswer(FrameWriterOperation.Flush,
+ exceptionThrowingOperations, errorThrowingOperations);
+ CountAnswer failAnswer = FrameWriterTestUtils.createAnswer(FrameWriterOperation.Fail,
+ exceptionThrowingOperations, errorThrowingOperations);
+ CountAnswer closeAnswer = FrameWriterTestUtils.createAnswer(FrameWriterOperation.Close,
+ exceptionThrowingOperations, errorThrowingOperations);
+ return new TestTupleCounterFrameWriter(recordDescriptor, openAnswer, nextAnswer, flushAnswer, failAnswer,
+ closeAnswer, deepCopyInputFrames);
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestRecordDescriptorFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestRecordDescriptorFactory.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestRecordDescriptorFactory.java
deleted file mode 100644
index 536bf3a..0000000
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/dataflow/TestRecordDescriptorFactory.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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 org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-
-public class TestRecordDescriptorFactory {
- public RecordDescriptor createRecordDescriptor(ISerializerDeserializer<?>... serdes) {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/pom.xml
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/pom.xml b/asterixdb/asterix-common/pom.xml
index fc1c221..f6c0c99 100644
--- a/asterixdb/asterix-common/pom.xml
+++ b/asterixdb/asterix-common/pom.xml
@@ -252,6 +252,11 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <version>1.2.17</version>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
index b3eb281..71c30d5 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/context/PrimaryIndexOperationTracker.java
@@ -79,7 +79,7 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker {
@Override
public synchronized void completeOperation(ILSMIndex index, LSMOperationType opType,
ISearchOperationCallback searchCallback, IModificationOperationCallback modificationCallback)
- throws HyracksDataException {
+ throws HyracksDataException {
if (opType == LSMOperationType.MODIFICATION || opType == LSMOperationType.FORCE_MODIFICATION) {
decrementNumActiveOperations(modificationCallback);
if (numActiveOperations.get() == 0) {
@@ -148,12 +148,12 @@ public class PrimaryIndexOperationTracker extends BaseOperationTracker {
for (ILSMIndex lsmIndex : dsInfo.getDatasetIndexes()) {
//get resource
- ILSMIndexAccessor accessor = lsmIndex.createAccessor(NoOpOperationCallback.INSTANCE,
- NoOpOperationCallback.INSTANCE);
+ ILSMIndexAccessor accessor =
+ lsmIndex.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
//update resource lsn
- AbstractLSMIOOperationCallback ioOpCallback = (AbstractLSMIOOperationCallback) lsmIndex
- .getIOOperationCallback();
+ AbstractLSMIOOperationCallback ioOpCallback =
+ (AbstractLSMIOOperationCallback) lsmIndex.getIOOperationCallback();
ioOpCallback.updateLastLSN(logRecord.getLSN());
//schedule flush after update
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/AsterixLSMInsertDeleteOperatorNodePushable.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/AsterixLSMInsertDeleteOperatorNodePushable.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/AsterixLSMInsertDeleteOperatorNodePushable.java
index 9a76b40..cf66d30 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/AsterixLSMInsertDeleteOperatorNodePushable.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/AsterixLSMInsertDeleteOperatorNodePushable.java
@@ -22,6 +22,8 @@ import java.nio.ByteBuffer;
import org.apache.asterix.common.api.IAsterixAppRuntimeContext;
import org.apache.asterix.common.exceptions.FrameDataException;
+import org.apache.asterix.common.transactions.ILogMarkerCallback;
+import org.apache.asterix.common.transactions.PrimaryIndexLogMarkerCallback;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
@@ -31,6 +33,7 @@ import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.hyracks.storage.am.common.api.ITupleFilterFactory;
import org.apache.hyracks.storage.am.common.dataflow.IIndexOperatorDescriptor;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
@@ -41,7 +44,9 @@ import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
public class AsterixLSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUpdateDeleteOperatorNodePushable {
+ public static final String KEY_INDEX = "Index";
private final boolean isPrimary;
+ // This class has both lsmIndex and index (in super class) pointing to the same object
private AbstractLSMIndex lsmIndex;
private int i = 0;
@@ -59,10 +64,6 @@ public class AsterixLSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUp
private int currentTupleIdx;
private int lastFlushedTupleIdx;
- public boolean isPrimary() {
- return isPrimary;
- }
-
public AsterixLSMInsertDeleteOperatorNodePushable(IIndexOperatorDescriptor opDesc, IHyracksTaskContext ctx,
int partition, int[] fieldPermutation, IRecordDescriptorProvider recordDescProvider, IndexOperation op,
boolean isPrimary) {
@@ -79,6 +80,10 @@ public class AsterixLSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUp
indexHelper.open();
lsmIndex = (AbstractLSMIndex) indexHelper.getIndexInstance();
try {
+ if (isPrimary && ctx.getSharedObject() != null) {
+ PrimaryIndexLogMarkerCallback callback = new PrimaryIndexLogMarkerCallback(lsmIndex);
+ TaskUtils.putInSharedMap(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
+ }
writer.open();
modCallback = opDesc.getModificationOpCallbackFactory().createModificationOperationCallback(
indexHelper.getResourcePath(), indexHelper.getResourceID(), indexHelper.getResourcePartition(),
@@ -185,4 +190,8 @@ public class AsterixLSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUp
writer.fail();
}
}
+
+ public boolean isPrimary() {
+ return isPrimary;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogMarkerCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogMarkerCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogMarkerCallback.java
new file mode 100644
index 0000000..11d649b
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogMarkerCallback.java
@@ -0,0 +1,45 @@
+/*
+ * 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.common.transactions;
+
+import java.nio.ByteBuffer;
+
+/**
+ * This interface provide callback mechanism for adding marker logs to the transaction log file
+ */
+public interface ILogMarkerCallback {
+
+ String KEY_MARKER_CALLBACK = "MARKER_CALLBACK";
+
+ /**
+ * Called before writing the marker log allowing addition of specific information to the log record
+ *
+ * @param buffer:
+ * the log buffer to write to
+ */
+ void before(ByteBuffer buffer);
+
+ /**
+ * Called after the log's been appended to the log tail passing the position of the log used for random access
+ *
+ * @param lsn
+ */
+ void after(long lsn);
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
index cd05ba0..29af931 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
@@ -32,11 +32,38 @@ public interface ILogRecord {
LARGE_RECORD
}
- public static final int JOB_TERMINATE_LOG_SIZE = 14; //JOB_COMMIT or ABORT log type
- public static final int ENTITY_COMMIT_LOG_BASE_SIZE = 30;
- public static final int UPDATE_LOG_BASE_SIZE = 51;
- public static final int FLUSH_LOG_SIZE = 18;
- public static final int WAIT_LOG_SIZE = 14;
+ public static final int CHKSUM_LEN = Long.BYTES;
+ public static final int FLDCNT_LEN = Integer.BYTES;
+ public static final int DS_LEN = Integer.BYTES;
+ public static final int LOG_SOURCE_LEN = Byte.BYTES;
+ public static final int LOGRCD_SZ_LEN = Integer.BYTES;
+ public static final int NEWOP_LEN = Byte.BYTES;
+ public static final int NEWVALSZ_LEN = Integer.BYTES;
+ public static final int PKHASH_LEN = Integer.BYTES;
+ public static final int PKSZ_LEN = Integer.BYTES;
+ public static final int PRVLSN_LEN = Long.BYTES;
+ public static final int RS_PARTITION_LEN = Integer.BYTES;
+ public static final int RSID_LEN = Long.BYTES;
+ public static final int SEQ_NUM_LEN = Long.BYTES;
+ public static final int TYPE_LEN = Byte.BYTES;
+ public static final int UUID_LEN = Long.BYTES;
+ public static final int VBUCKET_ID_LEN = Short.BYTES;
+
+ public static final int ALL_RECORD_HEADER_LEN = LOG_SOURCE_LEN + TYPE_LEN + JobId.BYTES;
+ public static final int ENTITYCOMMIT_UPDATE_HEADER_LEN = RS_PARTITION_LEN + DatasetId.BYTES + PKHASH_LEN + PKSZ_LEN;
+ public static final int UPDATE_LSN_HEADER = RSID_LEN + LOGRCD_SZ_LEN;
+ public static final int UPDATE_BODY_HEADER = FLDCNT_LEN + NEWOP_LEN + NEWVALSZ_LEN;
+ // What are these fields? vvvvv
+ public static final int REMOTE_FLUSH_LOG_EXTRA_FIELDS_LEN = Long.BYTES + Integer.BYTES + Integer.BYTES;
+
+ // How are the following computed?
+ public static final int JOB_TERMINATE_LOG_SIZE = ALL_RECORD_HEADER_LEN + CHKSUM_LEN;
+ public static final int ENTITY_COMMIT_LOG_BASE_SIZE = 30; // ALL_RECORD_HEADER_LEN + CHKSUM_LEN +?
+ public static final int UPDATE_LOG_BASE_SIZE = 51; // ALL_RECORD_HEADER_LEN + CHKSUM_LEN +?
+ public static final int FLUSH_LOG_SIZE = 18; // ALL_RECORD_HEADER_LEN + CHKSUM_LEN +?
+ public static final int WAIT_LOG_SIZE = ALL_RECORD_HEADER_LEN + CHKSUM_LEN;
+ public static final int MARKER_BASE_LOG_SIZE =
+ ALL_RECORD_HEADER_LEN + CHKSUM_LEN + DS_LEN + RS_PARTITION_LEN + PRVLSN_LEN + LOGRCD_SZ_LEN;
public RecordReadStatus readLogRecord(ByteBuffer buffer);
@@ -135,7 +162,15 @@ public interface ILogRecord {
public ITupleReference getOldValue();
- public void setOldValue(ITupleReference oldValue);
+ public void setOldValue(ITupleReference tupleBefore);
- public void setOldValueSize(int oldValueSize);
+ public void setOldValueSize(int beforeSize);
+
+ public boolean isMarker();
+
+ public ByteBuffer getMarker();
+
+ public void logAppended(long lsn);
+
+ public long getPreviousMarkerLSN();
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
index 23fdd0f..306b888 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
@@ -71,24 +71,6 @@ import org.apache.hyracks.storage.am.common.tuples.SimpleTupleWriter;
public class LogRecord implements ILogRecord {
- private static final int LOG_SOURCE_LEN = Byte.BYTES;
- private static final int TYPE_LEN = Byte.BYTES;
- public static final int PKHASH_LEN = Integer.BYTES;
- public static final int PKSZ_LEN = Integer.BYTES;
- private static final int RS_PARTITION_LEN = Integer.BYTES;
- private static final int RSID_LEN = Long.BYTES;
- private static final int LOGRCD_SZ_LEN = Integer.BYTES;
- private static final int FLDCNT_LEN = Integer.BYTES;
- private static final int NEWOP_LEN = Byte.BYTES;
- private static final int NEWVALSZ_LEN = Integer.BYTES;
- private static final int CHKSUM_LEN = Long.BYTES;
-
- private static final int ALL_RECORD_HEADER_LEN = LOG_SOURCE_LEN + TYPE_LEN + JobId.BYTES;
- private static final int ENTITYCOMMIT_UPDATE_HEADER_LEN = RS_PARTITION_LEN + DatasetId.BYTES + PKHASH_LEN
- + PKSZ_LEN;
- private static final int UPDATE_LSN_HEADER = RSID_LEN + LOGRCD_SZ_LEN;
- private static final int UPDATE_BODY_HEADER = FLDCNT_LEN + NEWOP_LEN + NEWVALSZ_LEN;
-
// ------------- fields in a log record (begin) ------------//
private byte logSource;
private byte logType;
@@ -108,8 +90,10 @@ public class LogRecord implements ILogRecord {
private ITupleReference oldValue;
private int oldValueFieldCount;
private long checksum;
+ private long prevMarkerLSN;
+ private ByteBuffer marker;
// ------------- fields in a log record (end) --------------//
-
+ private final ILogMarkerCallback callback; // A callback for log mark operations
private int PKFieldCnt;
private ITransactionContext txnCtx;
private long LSN;
@@ -129,7 +113,8 @@ public class LogRecord implements ILogRecord {
private String nodeId;
private boolean replicated = false;
- public LogRecord() {
+ public LogRecord(ILogMarkerCallback callback) {
+ this.callback = callback;
isFlushed = new AtomicBoolean(false);
readPKValue = new PrimaryKeyTupleReference();
readNewValue = SimpleTupleWriter.INSTANCE.createTupleReference();
@@ -138,49 +123,70 @@ public class LogRecord implements ILogRecord {
logSource = LogSource.LOCAL;
}
- private void writeLogRecordCommonFields(ByteBuffer buffer) {
+ public LogRecord() {
+ this(null);
+ }
+
+ private void doWriteLogRecord(ByteBuffer buffer) {
buffer.put(logSource);
buffer.put(logType);
buffer.putInt(jobId);
- if (logType == LogType.UPDATE || logType == LogType.ENTITY_COMMIT || logType == LogType.UPSERT_ENTITY_COMMIT) {
- buffer.putInt(resourcePartition);
- buffer.putInt(datasetId);
- buffer.putInt(PKHashValue);
- if (PKValueSize <= 0) {
- throw new IllegalStateException("Primary Key Size is less than or equal to 0");
- }
- buffer.putInt(PKValueSize);
- writePKValue(buffer);
- }
- if (logType == LogType.UPDATE) {
- buffer.putLong(resourceId);
- buffer.putInt(logSize);
- buffer.putInt(newValueFieldCount);
- buffer.put(newOp);
- buffer.putInt(newValueSize);
- writeTuple(buffer, newValue, newValueSize);
- if (oldValueSize > 0) {
- buffer.putInt(oldValueSize);
- buffer.putInt(oldValueFieldCount);
- writeTuple(buffer, oldValue, oldValueSize);
- }
+ switch (logType) {
+ case LogType.ENTITY_COMMIT:
+ case LogType.UPSERT_ENTITY_COMMIT:
+ writeEntityInfo(buffer);
+ break;
+ case LogType.UPDATE:
+ writeEntityInfo(buffer);
+ buffer.putLong(resourceId);
+ buffer.putInt(logSize);
+ buffer.putInt(newValueFieldCount);
+ buffer.put(newOp);
+ buffer.putInt(newValueSize);
+ writeTuple(buffer, newValue, newValueSize);
+ if (oldValueSize > 0) {
+ buffer.putInt(oldValueSize);
+ buffer.putInt(oldValueFieldCount);
+ writeTuple(buffer, oldValue, oldValueSize);
+ }
+ break;
+ case LogType.FLUSH:
+ buffer.putInt(datasetId);
+ break;
+ case LogType.MARKER:
+ buffer.putInt(datasetId);
+ buffer.putInt(resourcePartition);
+ callback.before(buffer);
+ buffer.putInt(logSize);
+ buffer.put(marker);
+ break;
+ default:
+ // Do nothing
}
- if (logType == LogType.FLUSH) {
- buffer.putInt(datasetId);
+ }
+
+ private void writeEntityInfo(ByteBuffer buffer) {
+ buffer.putInt(resourcePartition);
+ buffer.putInt(datasetId);
+ buffer.putInt(PKHashValue);
+ if (PKValueSize <= 0) {
+ throw new IllegalStateException("Primary Key Size is less than or equal to 0");
}
+ buffer.putInt(PKValueSize);
+ writePKValue(buffer);
}
@Override
public void writeLogRecord(ByteBuffer buffer) {
int beginOffset = buffer.position();
- writeLogRecordCommonFields(buffer);
+ doWriteLogRecord(buffer);
checksum = generateChecksum(buffer, beginOffset, logSize - CHKSUM_LEN);
buffer.putLong(checksum);
}
@Override
public void writeRemoteLogRecord(ByteBuffer buffer) {
- writeLogRecordCommonFields(buffer);
+ doWriteLogRecord(buffer);
if (logType == LogType.FLUSH) {
buffer.putLong(LSN);
buffer.putInt(numOfFlushedIndexes);
@@ -222,7 +228,7 @@ public class LogRecord implements ILogRecord {
int beginOffset = buffer.position();
//read common fields
- RecordReadStatus status = readLogCommonFields(buffer);
+ RecordReadStatus status = doReadLogRecord(buffer);
if (status != RecordReadStatus.OK) {
buffer.position(beginOffset);
return status;
@@ -241,7 +247,7 @@ public class LogRecord implements ILogRecord {
return RecordReadStatus.OK;
}
- private RecordReadStatus readLogCommonFields(ByteBuffer buffer) {
+ private RecordReadStatus doReadLogRecord(ByteBuffer buffer) {
//first we need the logtype and Job ID, if the buffer isn't that big, then no dice.
if (buffer.remaining() < ALL_RECORD_HEADER_LEN) {
return RecordReadStatus.TRUNCATED;
@@ -255,64 +261,88 @@ public class LogRecord implements ILogRecord {
return RecordReadStatus.TRUNCATED;
}
datasetId = buffer.getInt();
- resourceId = 0L;
+ resourceId = 0l;
+ // fall throuh
+ case LogType.WAIT:
+ computeAndSetLogSize();
break;
- case LogType.ABORT:
case LogType.JOB_COMMIT:
+ case LogType.ABORT:
datasetId = -1;
PKHashValue = -1;
+ computeAndSetLogSize();
break;
case LogType.ENTITY_COMMIT:
case LogType.UPSERT_ENTITY_COMMIT:
- if (!readEntityInfo(buffer)) {
+ if (readEntityInfo(buffer)) {
+ computeAndSetLogSize();
+ } else {
return RecordReadStatus.TRUNCATED;
}
break;
case LogType.UPDATE:
- if (!readEntityInfo(buffer)) {
+ if (readEntityInfo(buffer)) {
+ if (buffer.remaining() < UPDATE_LSN_HEADER + UPDATE_BODY_HEADER) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ resourceId = buffer.getLong();
+ logSize = buffer.getInt();
+ newValueFieldCount = buffer.getInt();
+ newOp = buffer.get();
+ newValueSize = buffer.getInt();
+ if (buffer.remaining() < newValueSize) {
+ if (logSize > buffer.capacity()) {
+ return RecordReadStatus.LARGE_RECORD;
+ }
+ return RecordReadStatus.TRUNCATED;
+ }
+ newValue = readTuple(buffer, readNewValue, newValueFieldCount, newValueSize);
+ if (logSize > getUpdateLogSizeWithoutOldValue()) {
+ // Prev Image exists
+ if (buffer.remaining() < Integer.BYTES) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ oldValueSize = buffer.getInt();
+ if (buffer.remaining() < Integer.BYTES) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ oldValueFieldCount = buffer.getInt();
+ if (buffer.remaining() < oldValueSize) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ oldValue = readTuple(buffer, readOldValue, oldValueFieldCount, oldValueSize);
+ } else {
+ oldValueSize = 0;
+ }
+ } else {
return RecordReadStatus.TRUNCATED;
}
- if (buffer.remaining() < UPDATE_LSN_HEADER + UPDATE_BODY_HEADER) {
+ break;
+ case LogType.MARKER:
+ if (buffer.remaining() < DS_LEN + RS_PARTITION_LEN + PRVLSN_LEN + LOGRCD_SZ_LEN) {
return RecordReadStatus.TRUNCATED;
}
- resourceId = buffer.getLong();
+ datasetId = buffer.getInt();
+ resourcePartition = buffer.getInt();
+ prevMarkerLSN = buffer.getLong();
logSize = buffer.getInt();
- newValueFieldCount = buffer.getInt();
- newOp = buffer.get();
- newValueSize = buffer.getInt();
- return readEntity(buffer);
+ int lenRemaining = logSize - MARKER_BASE_LOG_SIZE;
+ if (buffer.remaining() < lenRemaining) {
+ return RecordReadStatus.TRUNCATED;
+ }
+
+ if (marker == null || marker.capacity() < lenRemaining) {
+ // TODO(amoudi): account for memory allocation
+ marker = ByteBuffer.allocate(lenRemaining + Short.BYTES);
+ }
+ marker.clear();
+ buffer.get(marker.array(), 0, lenRemaining);
+ marker.position(lenRemaining);
+ marker.flip();
+ break;
default:
break;
}
- computeAndSetLogSize();
- return RecordReadStatus.OK;
- }
-
- private RecordReadStatus readEntity(ByteBuffer buffer) {
- if (buffer.remaining() < newValueSize) {
- if (logSize > buffer.capacity()) {
- return RecordReadStatus.LARGE_RECORD;
- }
- return RecordReadStatus.TRUNCATED;
- }
- newValue = readTuple(buffer, readNewValue, newValueFieldCount, newValueSize);
- if (logSize > getUpdateLogSizeWithoutOldValue()) {
- // Prev Image exists
- if (buffer.remaining() < Integer.BYTES) {
- return RecordReadStatus.TRUNCATED;
- }
- oldValueSize = buffer.getInt();
- if (buffer.remaining() < Integer.BYTES) {
- return RecordReadStatus.TRUNCATED;
- }
- oldValueFieldCount = buffer.getInt();
- if (buffer.remaining() < oldValueSize) {
- return RecordReadStatus.TRUNCATED;
- }
- oldValue = readTuple(buffer, readOldValue, oldValueFieldCount, oldValueSize);
- } else {
- oldValueSize = 0;
- }
return RecordReadStatus.OK;
}
@@ -339,7 +369,7 @@ public class LogRecord implements ILogRecord {
@Override
public void readRemoteLog(ByteBuffer buffer) {
//read common fields
- readLogCommonFields(buffer);
+ doReadLogRecord(buffer);
if (logType == LogType.FLUSH) {
LSN = buffer.getLong();
@@ -412,11 +442,18 @@ public class LogRecord implements ILogRecord {
case LogType.WAIT:
logSize = WAIT_LOG_SIZE;
break;
+ case LogType.MARKER:
+ setMarkerLogSize();
+ break;
default:
throw new IllegalStateException("Unsupported Log Type");
}
}
+ private void setMarkerLogSize() {
+ logSize = MARKER_BASE_LOG_SIZE + marker.remaining();
+ }
+
@Override
public String getLogRecordForDisplay() {
StringBuilder builder = new StringBuilder();
@@ -688,4 +725,28 @@ public class LogRecord implements ILogRecord {
public void setOldValueSize(int oldValueSize) {
this.oldValueSize = oldValueSize;
}
+
+ public void setMarker(ByteBuffer marker) {
+ this.marker = marker;
+ }
+
+ @Override
+ public boolean isMarker() {
+ return logType == LogType.MARKER;
+ }
+
+ @Override
+ public void logAppended(long lsn) {
+ callback.after(lsn);
+ }
+
+ @Override
+ public long getPreviousMarkerLSN() {
+ return prevMarkerLSN;
+ }
+
+ @Override
+ public ByteBuffer getMarker() {
+ return marker;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
index 714b8f7..269e4b9 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
@@ -27,6 +27,7 @@ public class LogType {
public static final byte FLUSH = 4;
public static final byte UPSERT_ENTITY_COMMIT = 5;
public static final byte WAIT = 6;
+ public static final byte MARKER = 7;
private static final String STRING_UPDATE = "UPDATE";
private static final String STRING_JOB_COMMIT = "JOB_COMMIT";
@@ -35,8 +36,8 @@ public class LogType {
private static final String STRING_FLUSH = "FLUSH";
private static final String STRING_UPSERT_ENTITY_COMMIT = "UPSERT_ENTITY_COMMIT";
private static final String STRING_WAIT = "WAIT";
-
- private static final String STRING_INVALID_LOG_TYPE = "INVALID_LOG_TYPE";
+ private static final String STRING_MARKER = "MARKER";
+ private static final String STRING_UNKNOWN_LOG_TYPE = "UNKNOWN_LOG_TYPE";
public static String toString(byte logType) {
switch (logType) {
@@ -54,8 +55,10 @@ public class LogType {
return STRING_UPSERT_ENTITY_COMMIT;
case LogType.WAIT:
return STRING_WAIT;
+ case LogType.MARKER:
+ return STRING_MARKER;
default:
- return STRING_INVALID_LOG_TYPE;
+ return STRING_UNKNOWN_LOG_TYPE;
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/PrimaryIndexLogMarkerCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/PrimaryIndexLogMarkerCallback.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/PrimaryIndexLogMarkerCallback.java
new file mode 100644
index 0000000..7dae65f
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/PrimaryIndexLogMarkerCallback.java
@@ -0,0 +1,51 @@
+/*
+ * 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.common.transactions;
+
+import java.nio.ByteBuffer;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
+
+/**
+ * A basic callback used to write marker to transaction logs
+ */
+public class PrimaryIndexLogMarkerCallback implements ILogMarkerCallback {
+
+ private AbstractLSMIndex index;
+
+ /**
+ * @param index:
+ * a pointer to the primary index used to store marker log info
+ * @throws HyracksDataException
+ */
+ public PrimaryIndexLogMarkerCallback(AbstractLSMIndex index) throws HyracksDataException {
+ this.index = index;
+ }
+
+ @Override
+ public void before(ByteBuffer buffer) {
+ buffer.putLong(index.getCurrentMemoryComponent().getMostRecentMarkerLSN());
+ }
+
+ @Override
+ public void after(long lsn) {
+ index.getCurrentMemoryComponent().setMostRecentMarkerLSN(lsn);
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixConstants.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixConstants.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixConstants.java
new file mode 100644
index 0000000..4bca216
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/AsterixConstants.java
@@ -0,0 +1,30 @@
+/*
+ * 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.common.utils;
+
+/**
+ * A static class that stores asterix constants
+ */
+public class AsterixConstants {
+ public static final String ASTERIX_ROOT_METADATA_DIR = "asterix_root_metadata";
+
+ private AsterixConstants() {
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/FrameStack.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/FrameStack.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/FrameStack.java
new file mode 100644
index 0000000..8c83687
--- /dev/null
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/FrameStack.java
@@ -0,0 +1,149 @@
+/*
+ * 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.common.utils;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.util.ArrayDeque;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+
+/**
+ * Not thread safe stack that is used to store fixed size buffers in memory
+ * Once memory is consumed, it uses disk to store buffers
+ */
+public class FrameStack implements Closeable {
+ private static final AtomicInteger stackIdGenerator = new AtomicInteger(0);
+ private static final String STACK_FILE_NAME = "stack";
+ private final int stackId;
+ private final int frameSize;
+ private final int numOfMemoryFrames;
+ private final ArrayDeque<ByteBuffer> fullBuffers;
+ private final ArrayDeque<ByteBuffer> emptyBuffers;
+ private int totalWriteCount = 0;
+ private int totalReadCount = 0;
+ private final File file;
+ private final RandomAccessFile iostream;
+ private final byte[] frame;
+
+ /**
+ * Create a hybrid of memory and disk stack of byte buffers
+ *
+ * @param dir
+ * @param frameSize
+ * @param numOfMemoryFrames
+ * @throws HyracksDataException
+ * @throws FileNotFoundException
+ */
+ public FrameStack(String dir, int frameSize, int numOfMemoryFrames)
+ throws HyracksDataException, FileNotFoundException {
+ this.stackId = stackIdGenerator.getAndIncrement();
+ this.frameSize = frameSize;
+ this.numOfMemoryFrames = numOfMemoryFrames;
+ this.fullBuffers = numOfMemoryFrames <= 0 ? null : new ArrayDeque<>();
+ this.emptyBuffers = numOfMemoryFrames <= 0 ? null : new ArrayDeque<>();
+ this.file = StoragePathUtil.createFile(
+ ((dir == null) ? "" : (dir.endsWith(File.separator) ? dir : (dir + File.separator))) + STACK_FILE_NAME,
+ stackId);
+ this.iostream = new RandomAccessFile(file, "rw");
+ this.frame = new byte[frameSize];
+ }
+
+ /**
+ * @return the number of remaining frames to be read in the stack
+ */
+ public int remaining() {
+ return totalWriteCount - totalReadCount;
+ }
+
+ /**
+ * copy content of buffer into the stack
+ *
+ * @param buffer
+ * @throws IOException
+ */
+ public synchronized void push(ByteBuffer buffer) throws IOException {
+ int diff = totalWriteCount - totalReadCount;
+ if (diff < numOfMemoryFrames) {
+ ByteBuffer aBuffer = allocate();
+ aBuffer.put(buffer.array());
+ aBuffer.flip();
+ fullBuffers.push(aBuffer);
+ } else {
+ long position = (long) (diff - numOfMemoryFrames) * frameSize;
+ if (position != iostream.getFilePointer()) {
+ iostream.seek(position);
+ }
+ iostream.write(buffer.array());
+ }
+ totalWriteCount++;
+ }
+
+ private ByteBuffer allocate() {
+ ByteBuffer aBuffer = emptyBuffers.poll();
+ if (aBuffer == null) {
+ aBuffer = ByteBuffer.allocate(frameSize);
+ }
+ aBuffer.clear();
+ return aBuffer;
+ }
+
+ /**
+ * Free a frame off of the stack and copy it into dest
+ *
+ * @param dest
+ * @throws IOException
+ */
+ public synchronized void pop(ByteBuffer dest) throws IOException {
+ dest.clear();
+ int diff = totalWriteCount - totalReadCount - 1;
+ if (diff >= 0) {
+ if (diff < numOfMemoryFrames) {
+ totalReadCount++;
+ ByteBuffer aBuffer = fullBuffers.pop();
+ emptyBuffers.push(aBuffer);
+ dest.put(aBuffer.array());
+ } else {
+ long position = (long) (diff - numOfMemoryFrames) * frameSize;
+ iostream.seek(position);
+ iostream.readFully(frame);
+ dest.put(frame);
+ }
+ }
+ dest.flip();
+ }
+
+ /**
+ * Closing this stack will result in the data being deleted
+ *
+ * @throws IOException
+ */
+ @Override
+ public void close() throws IOException {
+ iostream.close();
+ Files.delete(file.toPath());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
index 78b06fb..615e8af 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/StoragePathUtil.java
@@ -24,12 +24,16 @@ import org.apache.asterix.common.cluster.ClusterPartition;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.utils.Pair;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.std.file.ConstantFileSplitProvider;
import org.apache.hyracks.dataflow.std.file.FileSplit;
import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
+import org.apache.log4j.Level;
+import org.apache.log4j.Logger;
public class StoragePathUtil {
+ private static final Logger LOGGER = Logger.getLogger(StoragePathUtil.class.getName());
public static final String PARTITION_DIR_PREFIX = "partition_";
public static final String TEMP_DATASETS_STORAGE_FOLDER = "temp";
public static final String DATASET_INDEX_NAME_SEPARATOR = "_idx_";
@@ -70,4 +74,39 @@ public class StoragePathUtil {
public static int getPartitionNumFromName(String name) {
return Integer.parseInt(name.substring(PARTITION_DIR_PREFIX.length()));
}
+
+ /**
+ * Create a file
+ * Note: this method is not thread safe. It is the responsibility of the caller to ensure no path conflict when
+ * creating files simultaneously
+ *
+ * @param name
+ * @param count
+ * @return
+ * @throws HyracksDataException
+ */
+ public static File createFile(String name, int count) throws HyracksDataException {
+ try {
+ String fileName = name + "_" + count;
+ File file = new File(fileName);
+ if (file.getParentFile() != null) {
+ file.getParentFile().mkdirs();
+ }
+ if (!file.exists()) {
+ boolean success = file.createNewFile();
+ if (!success) {
+ throw new HyracksDataException("Unable to create spill file " + fileName);
+ } else {
+ if (LOGGER.isEnabledFor(Level.INFO)) {
+ LOGGER.info("Created spill file " + file.getAbsolutePath());
+ }
+ }
+ } else {
+ throw new HyracksDataException("spill file " + fileName + " already exists");
+ }
+ return file;
+ } catch (Exception e) {
+ throw new HyracksDataException(e);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/TransactionUtil.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/TransactionUtil.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/TransactionUtil.java
index 1d5b15e..2878d5a 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/TransactionUtil.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/TransactionUtil.java
@@ -18,9 +18,12 @@
*/
package org.apache.asterix.common.utils;
+import java.nio.ByteBuffer;
+
import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
import org.apache.asterix.common.transactions.ITransactionContext;
import org.apache.asterix.common.transactions.LogRecord;
+import org.apache.asterix.common.transactions.LogSource;
import org.apache.asterix.common.transactions.LogType;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
@@ -64,4 +67,17 @@ public class TransactionUtil {
logRecord.computeAndSetPKValueSize();
logRecord.computeAndSetLogSize();
}
+
+ public static void formMarkerLogRecord(LogRecord logRecord, ITransactionContext txnCtx, int datasetId,
+ int resourcePartition, ByteBuffer marker) {
+ logRecord.setTxnCtx(txnCtx);
+ logRecord.setLogSource(LogSource.LOCAL);
+ logRecord.setLogType(LogType.MARKER);
+ logRecord.setJobId(txnCtx.getJobId().getId());
+ logRecord.setDatasetId(datasetId);
+ logRecord.setResourcePartition(resourcePartition);
+ marker.get(); // read the first byte since it is not part of the marker object
+ logRecord.setMarker(marker);
+ logRecord.computeAndSetLogSize();
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/AsterixConstants.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/AsterixConstants.java b/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/AsterixConstants.java
deleted file mode 100644
index 11a2510..0000000
--- a/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/AsterixConstants.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * 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.event.util;
-
-public class AsterixConstants {
-
- public static String ASTERIX_ROOT_METADATA_DIR = "asterix_root_metadata";
-
-}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/PatternCreator.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/PatternCreator.java b/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/PatternCreator.java
index b19a722..1780a51 100644
--- a/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/PatternCreator.java
+++ b/asterixdb/asterix-events/src/main/java/org/apache/asterix/event/util/PatternCreator.java
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.asterix.common.utils.AsterixConstants;
import org.apache.asterix.event.driver.EventDriver;
import org.apache.asterix.event.error.VerificationUtil;
import org.apache.asterix.event.model.AsterixInstance;
@@ -72,8 +73,8 @@ public class PatternCreator {
for (Node node : cluster.getNode()) {
if (copyHyracksToNC) {
- Pattern copyHyracksForNC = createCopyHyracksPattern(asterixInstanceName, cluster, node.getClusterIp(),
- destDir);
+ Pattern copyHyracksForNC =
+ createCopyHyracksPattern(asterixInstanceName, cluster, node.getClusterIp(), destDir);
ps.add(copyHyracksForNC);
}
}
@@ -389,8 +390,8 @@ public class PatternCreator {
Nodeid nodeid = new Nodeid(new Value(null, EventDriver.CLIENT_NODE.getId()));
String username = cluster.getUsername() != null ? cluster.getUsername() : System.getProperty("user.name");
String workingDir = cluster.getWorkingDir().getDir();
- String destDir = workingDir + File.separator + "library" + File.separator + dataverse + File.separator
- + libraryName;
+ String destDir =
+ workingDir + File.separator + "library" + File.separator + dataverse + File.separator + libraryName;
String fileToTransfer = new File(libraryPath).getAbsolutePath();
Iterator<Node> installTargets = cluster.getNode().iterator();
@@ -434,8 +435,8 @@ public class PatternCreator {
patternList.add(p);
Iterator<Node> uninstallTargets = cluster.getNode().iterator();
- String libDir = workingDir + File.separator + "library" + File.separator + dataverse + File.separator
- + libraryName;
+ String libDir =
+ workingDir + File.separator + "library" + File.separator + dataverse + File.separator + libraryName;
Node uninstallNode = uninstallTargets.next();
nodeid = new Nodeid(new Value(null, uninstallNode.getId()));
event = new Event("file_delete", nodeid, libDir);
@@ -606,8 +607,8 @@ public class PatternCreator {
String username = cluster.getUsername() == null ? System.getProperty("user.name") : cluster.getUsername();
String srcHost = cluster.getMasterNode().getClientIp();
Nodeid nodeid = new Nodeid(new Value(null, EventDriver.CLIENT_NODE.getId()));
- String srcDir = cluster.getMasterNode().getLogDir() == null ? cluster.getLogDir()
- : cluster.getMasterNode().getLogDir();
+ String srcDir =
+ cluster.getMasterNode().getLogDir() == null ? cluster.getLogDir() : cluster.getMasterNode().getLogDir();
String destDir = outputDir + File.separator + "cc";
String pargs = username + " " + srcHost + " " + srcDir + " " + destDir;
Event event = new Event("directory_copy", nodeid, pargs);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IFeedMarker.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IFeedMarker.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IFeedMarker.java
new file mode 100644
index 0000000..487b47d
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IFeedMarker.java
@@ -0,0 +1,35 @@
+/*
+ * 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.external.api;
+
+import org.apache.hyracks.api.comm.VSizeFrame;
+
+public interface IFeedMarker {
+
+ /**
+ * Mark the frame with a mark denoting the progress of the feed
+ * The mark will be eventually written to the transaction log
+ *
+ * @param mark
+ * a frame to write the progress mark in
+ * @return
+ */
+ public boolean mark(VSizeFrame mark);
+
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
index 0f5ada4..9d9ff28 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordConverter.java
@@ -20,6 +20,7 @@ package org.apache.asterix.external.api;
import java.io.IOException;
+@FunctionalInterface
public interface IRecordConverter<I, O> {
public O convert(IRawRecord<? extends I> input) throws IOException;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
index 9cce1c9..08ffe18 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/api/IRecordReader.java
@@ -73,4 +73,8 @@ public interface IRecordReader<T> extends Closeable {
* gives the record reader a chance to recover from IO errors during feed intake
*/
public boolean handleException(Throwable th);
+
+ public default IFeedMarker getProgressReporter() {
+ return null;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedDataFlowController.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedDataFlowController.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedDataFlowController.java
index a301ac9..7806489 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedDataFlowController.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedDataFlowController.java
@@ -34,9 +34,9 @@ public class ChangeFeedDataFlowController<T> extends FeedRecordDataFlowControlle
public ChangeFeedDataFlowController(final IHyracksTaskContext ctx, final FeedTupleForwarder tupleForwarder,
final FeedLogManager feedLogManager, final int numOfOutputFields,
- final IRecordWithPKDataParser<T> dataParser, final IRecordReader<T> recordReader)
+ final IRecordWithPKDataParser<T> dataParser, final IRecordReader<T> recordReader, boolean sendMarker)
throws HyracksDataException {
- super(ctx, tupleForwarder, feedLogManager, numOfOutputFields, dataParser, recordReader);
+ super(ctx, tupleForwarder, feedLogManager, numOfOutputFields, dataParser, recordReader, sendMarker);
this.dataParser = dataParser;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedWithMetaDataFlowController.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedWithMetaDataFlowController.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedWithMetaDataFlowController.java
index b47d278..7d65c52 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedWithMetaDataFlowController.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/ChangeFeedWithMetaDataFlowController.java
@@ -32,9 +32,9 @@ public class ChangeFeedWithMetaDataFlowController<T> extends FeedWithMetaDataFlo
public ChangeFeedWithMetaDataFlowController(final IHyracksTaskContext ctx, final FeedTupleForwarder tupleForwarder,
final FeedLogManager feedLogManager, final int numOfOutputFields,
- final IRecordWithMetadataParser<T> dataParser, final IRecordReader<T> recordReader)
- throws HyracksDataException {
- super(ctx, tupleForwarder, feedLogManager, numOfOutputFields, dataParser, recordReader);
+ final IRecordWithMetadataParser<T> dataParser, final IRecordReader<T> recordReader, boolean sendMarker)
+ throws HyracksDataException {
+ super(ctx, tupleForwarder, feedLogManager, numOfOutputFields, dataParser, recordReader, sendMarker);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedRecordDataFlowController.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedRecordDataFlowController.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedRecordDataFlowController.java
index 87daffa..be9056b 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedRecordDataFlowController.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedRecordDataFlowController.java
@@ -19,10 +19,15 @@
package org.apache.asterix.external.dataflow;
import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.Nonnull;
+import org.apache.asterix.external.api.IFeedMarker;
import org.apache.asterix.external.api.IRawRecord;
import org.apache.asterix.external.api.IRecordDataParser;
import org.apache.asterix.external.api.IRecordReader;
@@ -30,9 +35,13 @@ import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataExceptionUtils;
import org.apache.asterix.external.util.FeedLogManager;
import org.apache.hyracks.api.comm.IFrameWriter;
+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.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
+import org.apache.hyracks.dataflow.common.io.MessagingFrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
import org.apache.log4j.Logger;
public class FeedRecordDataFlowController<T> extends AbstractFeedDataFlowController {
@@ -40,38 +49,52 @@ public class FeedRecordDataFlowController<T> extends AbstractFeedDataFlowControl
protected final IRecordDataParser<T> dataParser;
protected final IRecordReader<? extends T> recordReader;
protected final AtomicBoolean closed = new AtomicBoolean(false);
- protected final long interval = 1000;
+ protected static final long INTERVAL = 1000;
+ protected final Object mutex = new Object();
+ protected final boolean sendMarker;
protected boolean failed = false;
public FeedRecordDataFlowController(IHyracksTaskContext ctx, FeedTupleForwarder tupleForwarder,
@Nonnull FeedLogManager feedLogManager, int numOfOutputFields, @Nonnull IRecordDataParser<T> dataParser,
- @Nonnull IRecordReader<T> recordReader) throws HyracksDataException {
+ @Nonnull IRecordReader<T> recordReader, boolean sendMarker) throws HyracksDataException {
super(ctx, tupleForwarder, feedLogManager, numOfOutputFields);
this.dataParser = dataParser;
this.recordReader = recordReader;
+ this.sendMarker = sendMarker;
recordReader.setFeedLogManager(feedLogManager);
recordReader.setController(this);
}
@Override
public void start(IFrameWriter writer) throws HyracksDataException {
+ ExecutorService executorService = sendMarker ? Executors.newSingleThreadExecutor() : null;
+ Future<?> result = null;
+ if (sendMarker) {
+ DataflowMarker dataflowMarker = new DataflowMarker(recordReader.getProgressReporter(),
+ TaskUtils.<VSizeFrame> get(HyracksConstants.KEY_MESSAGE, ctx));
+ result = executorService.submit(dataflowMarker);
+ }
HyracksDataException hde = null;
try {
failed = false;
tupleForwarder.initialize(ctx, writer);
while (recordReader.hasNext()) {
- IRawRecord<? extends T> record = recordReader.next();
- if (record == null) {
- flush();
- Thread.sleep(interval);
- continue;
+ // synchronized on mutex before we call next() so we don't a marker before its record
+ synchronized (mutex) {
+ IRawRecord<? extends T> record = recordReader.next();
+ if (record == null) {
+ flush();
+ wait(INTERVAL);
+ continue;
+ }
+ tb.reset();
+ parseAndForward(record);
}
- tb.reset();
- parseAndForward(record);
}
} catch (InterruptedException e) {
//TODO: Find out what could cause an interrupted exception beside termination of a job/feed
- LOGGER.warn("Feed has been interrupted. Closing the feed");
+ LOGGER.warn("Feed has been interrupted. Closing the feed", e);
+ Thread.currentThread().interrupt();
} catch (Exception e) {
failed = true;
tupleForwarder.flush();
@@ -90,10 +113,13 @@ public class FeedRecordDataFlowController<T> extends AbstractFeedDataFlowControl
hde = ExternalDataExceptionUtils.suppressIntoHyracksDataException(hde, th);
} finally {
closeSignal();
- if (hde != null) {
- throw hde;
+ if (sendMarker && result != null) {
+ result.cancel(true);
}
}
+ if (hde != null) {
+ throw hde;
+ }
}
private void parseAndForward(IRawRecord<? extends T> record) throws IOException {
@@ -170,4 +196,53 @@ public class FeedRecordDataFlowController<T> extends AbstractFeedDataFlowControl
// This is not a parser record. most likely, this error happened in the record reader.
return recordReader.handleException(th);
}
+
+ private class DataflowMarker implements Runnable {
+ private final IFeedMarker marker;
+ private final VSizeFrame mark;
+ private volatile boolean stopped = false;
+
+ public DataflowMarker(IFeedMarker marker, VSizeFrame mark) {
+ this.marker = marker;
+ this.mark = mark;
+ }
+
+ public synchronized void stop() {
+ stopped = true;
+ notify();
+ }
+
+ @Override
+ public void run() {
+ try {
+ while (true) {
+ synchronized (this) {
+ if (!stopped) {
+ // TODO (amoudi): find a better reactive way to do this
+ // sleep for two seconds
+ wait(TimeUnit.SECONDS.toMillis(2));
+ } else {
+ break;
+ }
+ }
+ synchronized (mutex) {
+ if (marker.mark(mark)) {
+ // broadcast
+ tupleForwarder.flush();
+ // clear
+ mark.getBuffer().clear();
+ mark.getBuffer().put(MessagingFrameTupleAppender.NULL_FEED_MESSAGE);
+ mark.getBuffer().flip();
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ LOGGER.warn("Marker stopped", e);
+ Thread.currentThread().interrupt();
+ return;
+ } catch (Exception e) {
+ LOGGER.warn("Marker stopped", e);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/7575785a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedTupleForwarder.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedTupleForwarder.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedTupleForwarder.java
index f1eb870..36c6c2f 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedTupleForwarder.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/dataflow/FeedTupleForwarder.java
@@ -30,10 +30,12 @@ import org.apache.hyracks.api.comm.IFrameWriter;
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.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
import org.apache.hyracks.dataflow.common.io.MessagingFrameTupleAppender;
+import org.apache.hyracks.dataflow.common.util.TaskUtils;
public class FeedTupleForwarder implements ITupleForwarder {
@@ -59,7 +61,7 @@ public class FeedTupleForwarder implements ITupleForwarder {
this.writer = writer;
this.appender = new FrameTupleAppender(frame);
// Set null feed message
- VSizeFrame message = (VSizeFrame) ctx.getSharedObject();
+ VSizeFrame message = TaskUtils.<VSizeFrame> get(HyracksConstants.KEY_MESSAGE, ctx);
// a null message
message.getBuffer().put(MessagingFrameTupleAppender.NULL_FEED_MESSAGE);
message.getBuffer().flip();