Gitiles
Code Review Sign In
asterix-gerrit.ics.uci.edu / asterixdb / 3a4415cd7437563c1ccf20b2b0d62d26058ed6dc / . / diff_file
blob: e0e67149c8d811a6a6331ed68ea4a57f48412246 [file] [log] [blame]
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionManager.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionManager.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionManager.java (working copy)
@@ -103,12 +103,14 @@
//for entity-level commit
if (PKHashVal != -1) {
transactionProvider.getLockManager().unlock(datasetId, PKHashVal, txnContext, true);
+ /*****************************
try {
//decrease the transaction reference count on index
txnContext.decreaseActiveTransactionCountOnIndexes();
} catch (HyracksDataException e) {
throw new ACIDException("failed to complete index operation", e);
}
+ *****************************/
return;
}
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionContext.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionContext.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionContext.java (working copy)
@@ -19,6 +19,7 @@
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
import edu.uci.ics.asterix.transaction.management.opcallbacks.AbstractOperationCallback;
@@ -169,5 +170,14 @@
closeable.close(this);
}
}
+
+ @Override
+ public int hashCode() {
+ return jobId.getId();
+ }
+ @Override
+ public boolean equals(Object o) {
+ return (o == this);
+ }
}
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/locking/LockManager.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/locking/LockManager.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/locking/LockManager.java (working copy)
@@ -567,7 +567,7 @@
if (commitFlag) {
if (txnContext.getTransactionType().equals(TransactionContext.TransactionType.READ_WRITE)) {
try {
- txnSubsystem.getLogManager().log(LogType.COMMIT, txnContext, datasetId.getId(),
+ txnSubsystem.getLogManager().log(LogType.ENTITY_COMMIT, txnContext, datasetId.getId(),
entityHashValue, -1, (byte) 0, 0, null, null, logicalLogLocator);
} catch (ACIDException e) {
try {
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/FileBasedBuffer.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/FileBasedBuffer.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/FileBasedBuffer.java (working copy)
@@ -75,6 +75,7 @@
buffer.position(0);
buffer.limit(size);
fileChannel.write(buffer);
+ fileChannel.force(false);
erase();
}
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManager.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManager.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManager.java (working copy)
@@ -1,5 +1,5 @@
/*
- * Copyright 2009-2010 by The Regents of the University of California
+ * Copyright 2009-2012 by The Regents of the University of California
* Licensed 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 from
@@ -21,7 +21,12 @@
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -30,22 +35,25 @@
import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
import edu.uci.ics.asterix.transaction.management.service.logging.IndexLogger.ReusableLogContentObject;
+import edu.uci.ics.asterix.transaction.management.service.logging.LogManager.PageOwnershipStatus;
+import edu.uci.ics.asterix.transaction.management.service.logging.LogManager.PageState;
import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionContext;
import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionManagementConstants;
import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionSubsystem;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
public class LogManager implements ILogManager {
public static final boolean IS_DEBUG_MODE = false;//true
private static final Logger LOGGER = Logger.getLogger(LogManager.class.getName());
- private TransactionSubsystem provider;
+ private final TransactionSubsystem provider;
private LogManagerProperties logManagerProperties;
+ private LogPageFlushThread logPageFlusher;
/*
* the array of log pages. The number of log pages is configurable. Pages
* taken together form an in-memory log buffer.
*/
-
private IFileBasedBuffer[] logPages;
private ILogRecordHelper logRecordHelper;
@@ -54,6 +62,7 @@
* Number of log pages that constitute the in-memory log buffer.
*/
private int numLogPages;
+
/*
* Initially all pages have an owner count of 1 that is the LogManager. When
* a transaction requests to write in a log page, the owner count is
@@ -62,12 +71,11 @@
* (covering the whole log record). When the content has been put, the log
* manager computes the checksum and puts it after the content. At this
* point, the ownership count is decremented as the transaction is done with
- * using the page. When a page is full, the log manager decrements the count
- * by one indicating that it has released its ownership of the log page.
- * There could be other transaction(s) still owning the page (that is they
- * could still be mid-way putting the log content). When the ownership count
- * eventually reaches zero, the thread responsible for flushing the log page
- * is notified and the page is flushed to disk.
+ * using the page. When a page is requested to be flushed, logPageFlusher
+ * set the count to 0(LOG_FLUSHER: meaning that the page is being flushed)
+ * only if the count is 1(LOG_WRITER: meaning that there is no other
+ * transactions who own the page to write logs.) After flushing the page,
+ * logPageFlusher set this count to 1.
*/
private AtomicInteger[] logPageOwnerCount;
@@ -78,18 +86,16 @@
/*
* LogPageStatus: A page is either ACTIVE or INACTIVE. The status for each
- * page is maintained in a map called logPageStatus. A page is ACTIVE when
- * the LogManager can allocate space in the page for writing a log record.
- * Initially all pages are ACTIVE. As transactions fill up space by writing
- * log records, a page may not have sufficient space left for serving a
- * request by a transaction. When this happens, the page is marked INACTIVE.
- * An INACTIVE page with no owners ( logPageOwnerCount.get(<pageIndex>) ==
- * 0) indicates that the page must be flushed to disk before any other log
- * record is written on the page.F
+ * page is maintained in logPageStatus. A page is ACTIVE when the LogManager
+ * can allocate space in the page for writing a log record. Initially all
+ * pages are ACTIVE. As transactions fill up space by writing log records,
+ * a page may not have sufficient space left for serving a request by a
+ * transaction. When this happens, the page is flushed to disk by calling
+ * logPageFlusher.requestFlush(). In the requestFlush(), after groupCommitWaitTime,
+ * the page status is set to INACTIVE. Then, there is no more writer on the
+ * page(meaning the corresponding logPageOwnerCount is 1), the page is flushed
+ * by the logPageFlusher and the status is reset to ACTIVE by the logPageFlusher.
*/
-
- // private Map<Integer, Integer> logPageStatus = new
- // ConcurrentHashMap<Integer, Integer>();
private AtomicInteger[] logPageStatus;
static class PageState {
@@ -98,41 +104,8 @@
}
private AtomicLong lastFlushedLsn = new AtomicLong(-1);
- private AtomicInteger lastFlushedPage = new AtomicInteger(-1);
/*
- * pendingFlushRequests is a map with key as Integer denoting the page
- * index. When a (transaction) thread discovers the need to flush a page, it
- * puts its Thread object into the corresponding value that is a
- * LinkedBlockingQueue. The LogManager has a LogFlusher thread that scans
- * this map in order of page index (and circling around). The flusher thread
- * needs to flush pages in order and waits for a thread to deposit an object
- * in the blocking queue corresponding to the next page in order. A request
- * to flush a page is conveyed to the flush thread by simply depositing an
- * object in to corresponding blocking queue. It is blocking in the sense
- * that the flusher thread will continue to wait for an object to arrive in
- * the queue. The object itself is ignored by the fliusher and just acts as
- * a signal/event that a page needs to be flushed.
- */
-
- private LinkedBlockingQueue[] pendingFlushRequests;
-
- /*
- * ICommitResolver is an interface that provides an API that can answer a
- * simple boolean - Given the commit requests so far, should a page be
- * flushed. The implementation of the interface contains the logic (or you
- * can say the policy) for commit. It could be group commit in which case
- * the commit resolver may not return a true indicating that it wishes to
- * delay flushing of the page.
- */
- private ICommitResolver commitResolver;
-
- /*
- * An object that keeps track of the submitted commit requests.
- */
- private CommitRequestStatistics commitRequestStatistics;
-
- /*
* When the transaction eco-system comes to life, the log manager positions
* itself to the end of the last written log. the startingLsn represent the
* lsn value of the next log record to be written after a system (re)start.
@@ -146,16 +119,10 @@
*/
private AtomicLong lsn = new AtomicLong(0);
- /*
- * A map that tracks the flush requests submitted for each page. The
- * requests for a page are cleared when the page is flushed.
- */
- public LinkedBlockingQueue<Thread> getPendingFlushRequests(int pageIndex) {
- return pendingFlushRequests[pageIndex];
- }
+ private List<HashMap<TransactionContext, Integer>> activeTxnCountMaps;
- public void addFlushRequest(int pageIndex) {
- pendingFlushRequests[pageIndex].add(pendingFlushRequests);
+ public void addFlushRequest(int pageIndex, long lsn, boolean isSynchronous) {
+ logPageFlusher.requestFlush(pageIndex, lsn, isSynchronous);
}
public AtomicLong getLastFlushedLsn() {
@@ -233,19 +200,12 @@
numLogPages = logManagerProperties.getNumLogPages();
logPageOwnerCount = new AtomicInteger[numLogPages];
logPageStatus = new AtomicInteger[numLogPages];
- pendingFlushRequests = new LinkedBlockingQueue[numLogPages];
- if (logManagerProperties.getGroupCommitWaitPeriod() > 0) { // configure
- // the
- // Commit
- // Resolver
- commitResolver = new GroupCommitResolver(); // Group Commit is
- // enabled
- commitRequestStatistics = new CommitRequestStatistics(numLogPages);
- } else {
- commitResolver = new BasicCommitResolver(); // the basic commit
- // resolver
+
+ activeTxnCountMaps = new ArrayList<HashMap<TransactionContext, Integer>>(numLogPages);
+ for (int i = 0; i < numLogPages; i++) {
+ activeTxnCountMaps.add(new HashMap<TransactionContext, Integer>());
}
- this.commitResolver.init(this); // initialize the commit resolver
+
logPages = new FileBasedBuffer[numLogPages];
/*
@@ -264,7 +224,6 @@
for (int i = 0; i < numLogPages; i++) {
logPageOwnerCount[i] = new AtomicInteger(PageOwnershipStatus.LOG_WRITER);
logPageStatus[i] = new AtomicInteger(PageState.ACTIVE);
- pendingFlushRequests[i] = new LinkedBlockingQueue<Thread>();
}
/*
@@ -278,9 +237,9 @@
* daemon thread so that it does not stop the JVM from exiting when all
* other threads are done with their work.
*/
- LogPageFlushThread logFlusher = new LogPageFlushThread(this);
- logFlusher.setDaemon(true);
- logFlusher.start();
+ logPageFlusher = new LogPageFlushThread(this);
+ logPageFlusher.setDaemon(true);
+ logPageFlusher.start();
}
public int getLogPageIndex(long lsnValue) {
@@ -312,7 +271,7 @@
*/
private void waitUntillPageIsAvailableForWritingLog(int pageIndex) throws ACIDException {
if (logPageStatus[pageIndex].get() == PageState.ACTIVE
- && getLogPageOwnershipCount(pageIndex).get() >= PageOwnershipStatus.LOG_WRITER) {
+ && logPageOwnerCount[pageIndex].get() >= PageOwnershipStatus.LOG_WRITER) {
return;
}
try {
@@ -338,47 +297,40 @@
*/
private long getLsn(int entrySize, byte logType) throws ACIDException {
long pageSize = logManagerProperties.getLogPageSize();
- boolean requiresFlushing = logType == LogType.COMMIT;
+
while (true) {
boolean forwardPage = false;
- boolean shouldFlushPage = false;
long old = lsn.get();
- int pageIndex = getLogPageIndex(old); // get the log page
- // corresponding to the
- // current lsn value
+
+ //get the log page corresponding to the current lsn value
+ int pageIndex = getLogPageIndex(old);
long retVal = old;
- long next = old + entrySize; // the lsn value for the next request,
- // if the current request is served.
+
+ // the lsn value for the next request if the current request is served.
+ long next = old + entrySize;
int prevPage = -1;
- if ((next - 1) / pageSize != old / pageSize // check if the log
- // record will cross
- // page boundaries, a
- // case that is not
- // allowed.
- || (next % pageSize == 0)) {
+
+ // check if the log record will cross page boundaries, a case that is not allowed.
+ if ((next - 1) / pageSize != old / pageSize || (next % pageSize == 0)) {
+
if ((old != 0 && old % pageSize == 0)) {
- retVal = old; // On second thought, this shall never be the
- // case as it means that the lsn is
- // currently at the beginning of a page and
- // we still need to forward the page which
- // means that the entrySize exceeds a log
- // page size. If this is the case, an
- // exception is thrown before calling this
- // API.
- // would remove this case.
+ // On second thought, this shall never be the case as it means that the lsn is
+ // currently at the beginning of a page and we still need to forward the page which
+ // means that the entrySize exceeds a log page size. If this is the case, an
+ // exception is thrown before calling this API. would remove this case.
+ retVal = old;
} else {
- retVal = ((old / pageSize) + 1) * pageSize; // set the lsn
- // to point to
- // the beginning
- // of the next
- // page.
+ // set the lsn to point to the beginning of the next page.
+ retVal = ((old / pageSize) + 1) * pageSize;
}
+
next = retVal;
- forwardPage = true; // as the log record shall cross log page
- // boundary, we must re-assign the lsn (so
- // that the log record begins on a different
- // location.
+
+ // as the log record shall cross log page boundary, we must re-assign the lsn so
+ // that the log record begins on a different location.
+ forwardPage = true;
+
prevPage = pageIndex;
pageIndex = getNextPageInSequence(pageIndex);
}
@@ -397,109 +349,51 @@
*/
waitUntillPageIsAvailableForWritingLog(pageIndex);
- if (!forwardPage && requiresFlushing) {
- shouldFlushPage = commitResolver.shouldCommitPage(pageIndex, this, commitRequestStatistics);
- if (shouldFlushPage) {
- next = ((next / pageSize) + 1) * pageSize; /*
- * next
- * represents the
- * next value of
- * lsn after this
- * log record has
- * been written.
- * If the page
- * needs to be
- * flushed, then
- * we do not give
- * any more LSNs
- * from this
- * page.
- */
- }
- }
- if (!lsn.compareAndSet(old, next)) { // Atomic call -> returns true
- // only when the value
- // represented by lsn is same as
- // "old". The value is updated
- // to "next".
+ if (!lsn.compareAndSet(old, next)) {
+ // Atomic call -> returns true only when the value represented by lsn is same as
+ // "old". The value is updated to "next".
continue;
}
if (forwardPage) {
- //TODO
- //this is not safe since the incoming thread may reach the same page slot with this page
- //(differ by the log buffer size)
- logPageStatus[prevPage].set(PageState.INACTIVE); // mark
- // previous
- // page
- // inactive
+ addFlushRequest(prevPage, old, false);
- /*
- * decrement on the behalf of the log manager. if there are no
- * more owners (count == 0) the page must be marked as a
- * candidate to be flushed.
- */
- int pageDirtyCount = getLogPageOwnershipCount(prevPage).decrementAndGet();
- if (pageDirtyCount == 0) {
- addFlushRequest(prevPage);
- }
-
- /*
- * The transaction thread that discovers the need to forward a
- * page is made to re-acquire a lsn.
- */
+ // The transaction thread that discovers the need to forward a
+ // page is made to re-acquire a lsn.
continue;
+
} else {
- /*
- * the transaction thread has been given a space in a log page,
- * but is made to wait until the page is available.
- */
+ // the transaction thread has been given a space in a log page,
+ // but is made to wait until the page is available.
+ // (Is this needed? when does this wait happen?)
waitUntillPageIsAvailableForWritingLog(pageIndex);
- /*
- * increment the counter as the transaction thread now holds a
- * space in the log page and hence is an owner.
- */
+
+ // increment the counter as the transaction thread now holds a
+ // space in the log page and hence is an owner.
logPageOwnerCount[pageIndex].incrementAndGet();
- }
- if (requiresFlushing) {
- if (!shouldFlushPage) {
- /*
- * the log record requires the page to be flushed but under
- * the commit policy, the flush task has been deferred. The
- * transaction thread submits its request to flush the page.
- */
- commitRequestStatistics.registerCommitRequest(pageIndex);
- } else {
- /*
- * the flush request was approved by the commit resolver.
- * Thus the page is marked INACTIVE as no more logs will be
- * written on this page. The log manager needs to release
- * its ownership. Note that transaction threads may still
- * continue to be owners of the log page till they fill up
- * the space allocated to them.
- */
- logPageStatus[pageIndex].set(PageState.INACTIVE);
- logPageOwnerCount[pageIndex].decrementAndGet(); // on
- // the
- // behalf
- // of
- // log
- // manager
+
+ // Before the count is incremented, if the flusher flushed the allocated page,
+ // then retry to get new LSN. Otherwise, the log with allocated lsn will be lost.
+ if (lastFlushedLsn.get() >= retVal) {
+ logPageOwnerCount[pageIndex].decrementAndGet();
+ continue;
}
}
+
return retVal;
}
}
@Override
- public void log(byte logType, TransactionContext context, int datasetId, int PKHashValue, long resourceId,
+ public void log(byte logType, TransactionContext txnCtx, int datasetId, int PKHashValue, long resourceId,
byte resourceMgrId, int logContentSize, ReusableLogContentObject reusableLogContentObject, ILogger logger,
LogicalLogLocator logicalLogLocator) throws ACIDException {
- /*
- * logLocator is a re-usable object that is appropriately set in each
- * invocation. If the reference is null, the log manager must throw an
- * exception
- */
+
+ HashMap<TransactionContext, Integer> map = null;
+ int activeTxnCount;
+
+ // logLocator is a re-usable object that is appropriately set in each invocation.
+ // If the reference is null, the log manager must throw an exception.
if (logicalLogLocator == null) {
throw new ACIDException(
" you need to pass in a non-null logLocator, if you dont have it, then pass in a dummy so that the +"
@@ -519,20 +413,19 @@
// all constraints checked and we are good to go and acquire a lsn.
long previousLSN = -1;
- long currentLSN; // the will be set to the location (a long value)
- // where the log record needs to be placed.
- /*
- * The logs written by a transaction need to be linked to each other for
- * a successful rollback/recovery. However there could be multiple
- * threads operating concurrently that are part of a common transaction.
- * These threads need to synchronize and record the lsn corresponding to
- * the last log record written by (any thread of) the transaction.
- */
- synchronized (context) {
- previousLSN = context.getLastLogLocator().getLsn();
+ // the will be set to the location (a long value) where the log record needs to be placed.
+ long currentLSN;
+
+ // The logs written by a transaction need to be linked to each other for
+ // a successful rollback/recovery. However there could be multiple
+ // threads operating concurrently that are part of a common transaction.
+ // These threads need to synchronize and record the lsn corresponding to
+ // the last log record written by (any thread of) the transaction.
+ synchronized (txnCtx) {
+ previousLSN = txnCtx.getLastLogLocator().getLsn();
currentLSN = getLsn(totalLogSize, logType);
- context.setLastLSN(currentLSN);
+ txnCtx.setLastLSN(currentLSN);
if (IS_DEBUG_MODE) {
System.out.println("--------------> LSN(" + currentLSN + ") is allocated");
}
@@ -547,48 +440,37 @@
* performed correctly that is ownership is released.
*/
- boolean decremented = false; // indicates if the transaction thread
- // has release ownership of the
- // page.
- boolean addedFlushRequest = false; // indicates if the transaction
- // thread has submitted a flush
- // request.
+ // indicates if the transaction thread has release ownership of the page.
+ boolean decremented = false;
int pageIndex = (int) getLogPageIndex(currentLSN);
- /*
- * the lsn has been obtained for the log record. need to set the
- * LogLocator instance accordingly.
- */
-
+ // the lsn has been obtained for the log record. need to set the
+ // LogLocator instance accordingly.
try {
-
logicalLogLocator.setBuffer(logPages[pageIndex]);
int pageOffset = getLogPageOffset(currentLSN);
logicalLogLocator.setMemoryOffset(pageOffset);
- /*
- * write the log header.
- */
- logRecordHelper.writeLogHeader(logicalLogLocator, logType, context, datasetId, PKHashValue, previousLSN,
+ // write the log header.
+ logRecordHelper.writeLogHeader(logicalLogLocator, logType, txnCtx, datasetId, PKHashValue, previousLSN,
resourceId, resourceMgrId, logContentSize);
// increment the offset so that the transaction can fill up the
// content in the correct region of the allocated space.
logicalLogLocator.increaseMemoryOffset(logRecordHelper.getLogHeaderSize(logType));
- // a COMMIT log record does not have any content
- // and hence the logger (responsible for putting the log content) is
- // not invoked.
+ // a COMMIT log record does not have any content and hence
+ // the logger (responsible for putting the log content) is not invoked.
if (logContentSize != 0) {
- logger.preLog(context, reusableLogContentObject);
+ logger.preLog(txnCtx, reusableLogContentObject);
}
if (logContentSize != 0) {
// call the logger implementation and ask to fill in the log
// record content at the allocated space.
- logger.log(context, logicalLogLocator, logContentSize, reusableLogContentObject);
- logger.postLog(context, reusableLogContentObject);
+ logger.log(txnCtx, logicalLogLocator, logContentSize, reusableLogContentObject);
+ logger.postLog(txnCtx, reusableLogContentObject);
if (IS_DEBUG_MODE) {
logicalLogLocator.setMemoryOffset(logicalLogLocator.getMemoryOffset()
- logRecordHelper.getLogHeaderSize(logType));
@@ -597,10 +479,8 @@
}
}
- /*
- * The log record has been written. For integrity checks, compute
- * the checksum and put it at the end of the log record.
- */
+ // The log record has been written. For integrity checks, compute
+ // the checksum and put it at the end of the log record.
int startPosChecksum = logicalLogLocator.getMemoryOffset() - logRecordHelper.getLogHeaderSize(logType);
int length = totalLogSize - logRecordHelper.getLogChecksumSize();
long checksum = DataUtil.getChecksum(logPages[pageIndex], startPosChecksum, length);
@@ -611,46 +491,31 @@
System.out.println("--------------> LSN(" + currentLSN + ") is written");
}
- /*
- * release the ownership as the log record has been placed in
- * created space.
- */
- int pageDirtyCount = logPageOwnerCount[pageIndex].decrementAndGet();
+ // release the ownership as the log record has been placed in created space.
+ logPageOwnerCount[pageIndex].decrementAndGet();
// indicating that the transaction thread has released ownership
decremented = true;
- /*
- * If the transaction thread happens to be the last owner of the log
- * page the page must by marked as a candidate to be flushed.
- */
- if (pageDirtyCount == 0 && logPageStatus[pageIndex].get() == PageState.INACTIVE) {
- addFlushRequest(pageIndex);
- addedFlushRequest = true;
- }
-
- /*
- * If the log type is commit, a flush request is registered, if the
- * log record has not reached the disk. It may be possible that this
- * thread does not get CPU cycles and in-between the log record has
- * been flushed to disk because the containing log page filled up.
- */
- if (logType == LogType.COMMIT) {
- synchronized (logPages[pageIndex]) {
- while (getLastFlushedLsn().get() < currentLSN) {
- logPages[pageIndex].wait();
- }
+ if (logType == LogType.ENTITY_COMMIT) {
+ map = activeTxnCountMaps.get(pageIndex);
+ if (map.containsKey(txnCtx)) {
+ activeTxnCount = (Integer) map.get(txnCtx);
+ activeTxnCount++;
+ map.put(txnCtx, activeTxnCount);
+ } else {
+ map.put(txnCtx, 1);
}
+ addFlushRequest(pageIndex, currentLSN, false);
+ } else if (logType == LogType.COMMIT) {
+ addFlushRequest(pageIndex, currentLSN, true);
}
} catch (Exception e) {
e.printStackTrace();
- throw new ACIDException(context, "Thread: " + Thread.currentThread().getName()
+ throw new ACIDException(txnCtx, "Thread: " + Thread.currentThread().getName()
+ " logger encountered exception", e);
} finally {
- /*
- * If an exception was encountered and we did not release ownership
- */
if (!decremented) {
logPageOwnerCount[pageIndex].decrementAndGet();
}
@@ -667,9 +532,6 @@
logPages[pageIndex].reset(filePath, LogUtil.getFileOffset(this, nextWritePosition),
logManagerProperties.getLogPageSize());
-
- //TODO Check if this is necessary
- //Arrays.fill(logPages[pageIndex].getArray(), (byte) 0);
}
@Override
@@ -747,16 +609,13 @@
//minimize memory allocation overhead. current code allocates the log page size per reading a log record.
byte[] pageContent = new byte[logManagerProperties.getLogPageSize()];
- // take a lock on the log page so that the page is not flushed to
- // disk interim
+
+ // take a lock on the log page so that the page is not flushed to disk interim
synchronized (logPages[pageIndex]) {
- if (lsnValue > getLastFlushedLsn().get()) { // need to check
- // again
- // (this
- // thread may have got
- // de-scheduled and must
- // refresh!)
+ // need to check again (this thread may have got de-scheduled and must refresh!)
+ if (lsnValue > getLastFlushedLsn().get()) {
+
// get the log record length
logPages[pageIndex].getBytes(pageContent, 0, pageContent.length);
byte logType = pageContent[pageOffset + 4];
@@ -765,9 +624,7 @@
int logRecordSize = logHeaderSize + logBodySize + logRecordHelper.getLogChecksumSize();
logRecord = new byte[logRecordSize];
- /*
- * copy the log record content
- */
+ // copy the log record content
System.arraycopy(pageContent, pageOffset, logRecord, 0, logRecordSize);
MemBasedBuffer memBuffer = new MemBasedBuffer(logRecord);
if (logicalLogLocator == null) {
@@ -790,9 +647,7 @@
}
}
- /*
- * the log record is residing on the disk, read it from there.
- */
+ // the log record is residing on the disk, read it from there.
readDiskLog(lsnValue, logicalLogLocator);
}
@@ -860,30 +715,40 @@
return logPageOwnerCount[pageIndex];
}
- public ICommitResolver getCommitResolver() {
- return commitResolver;
- }
-
- public CommitRequestStatistics getCommitRequestStatistics() {
- return commitRequestStatistics;
- }
-
public IFileBasedBuffer[] getLogPages() {
return logPages;
}
- public int getLastFlushedPage() {
- return lastFlushedPage.get();
- }
-
- public void setLastFlushedPage(int lastFlushedPage) {
- this.lastFlushedPage.set(lastFlushedPage);
- }
-
@Override
public TransactionSubsystem getTransactionSubsystem() {
return provider;
}
+
+ public void decrementActiveTxnCountOnIndexes(int pageIndex) throws HyracksDataException {
+ TransactionContext ctx = null;
+ int count = 0;
+ int i = 0;
+
+ HashMap<TransactionContext, Integer> map = activeTxnCountMaps.get(pageIndex);
+ Set<Map.Entry<TransactionContext, Integer>> entrySet = map.entrySet();
+ if (entrySet != null) {
+ for (Map.Entry<TransactionContext, Integer> entry : entrySet) {
+ if (entry != null) {
+ if (entry.getValue() != null) {
+ count = entry.getValue();
+ }
+ if (count > 0) {
+ ctx = entry.getKey();
+ for (i = 0; i < count; i++) {
+ ctx.decreaseActiveTransactionCountOnIndexes();
+ }
+ }
+ }
+ }
+ }
+
+ map.clear();
+ }
}
/*
@@ -895,36 +760,82 @@
class LogPageFlushThread extends Thread {
private LogManager logManager;
+ /*
+ * pendingFlushRequests is a map with key as Integer denoting the page
+ * index. When a (transaction) thread discovers the need to flush a page, it
+ * puts its Thread object into the corresponding value that is a
+ * LinkedBlockingQueue. The LogManager has a LogFlusher thread that scans
+ * this map in order of page index (and circling around). The flusher thread
+ * needs to flush pages in order and waits for a thread to deposit an object
+ * in the blocking queue corresponding to the next page in order. A request
+ * to flush a page is conveyed to the flush thread by simply depositing an
+ * object in to corresponding blocking queue. It is blocking in the sense
+ * that the flusher thread will continue to wait for an object to arrive in
+ * the queue. The object itself is ignored by the fliusher and just acts as
+ * a signal/event that a page needs to be flushed.
+ */
+ private final LinkedBlockingQueue<Object>[] flushRequestQueue;
+ private final Object[] flushRequests;
+ private int lastFlushedPageIndex;
+ private final long groupCommitWaitPeriod;
public LogPageFlushThread(LogManager logManager) {
this.logManager = logManager;
setName("Flusher");
+ int numLogPages = logManager.getLogManagerProperties().getNumLogPages();
+ this.flushRequestQueue = new LinkedBlockingQueue[numLogPages];
+ this.flushRequests = new Object[numLogPages];
+ for (int i = 0; i < numLogPages; i++) {
+ flushRequestQueue[i] = new LinkedBlockingQueue<Object>(1);
+ flushRequests[i] = new Object();
+ }
+ this.lastFlushedPageIndex = -1;
+ groupCommitWaitPeriod = logManager.getLogManagerProperties().getGroupCommitWaitPeriod();
}
+ public void requestFlush(int pageIndex, long lsn, boolean isSynchronous) {
+ synchronized (logManager.getLogPage(pageIndex)) {
+ //return if flushedLSN >= lsn
+ if (logManager.getLastFlushedLsn().get() >= lsn) {
+ return;
+ }
+
+ //put a new request to the queue only if the request on the page is not in the queue.
+ flushRequestQueue[pageIndex].offer(flushRequests[pageIndex]);
+
+ //return if the request is asynchronous
+ if (!isSynchronous) {
+ return;
+ }
+
+ //wait until there is flush.
+ boolean isNotified = false;
+ while (!isNotified) {
+ try {
+ logManager.getLogPage(pageIndex).wait();
+ isNotified = true;
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
@Override
public void run() {
while (true) {
try {
- int pageToFlush = logManager.getNextPageInSequence(logManager.getLastFlushedPage());
+ int pageToFlush = logManager.getNextPageInSequence(lastFlushedPageIndex);
- /*
- * A wait call on the linkedBLockingQueue. The flusher thread is
- * notified when an object is added to the queue. Please note
- * that each page has an associated blocking queue.
- */
- logManager.getPendingFlushRequests(pageToFlush).take();
+ // A wait call on the linkedBLockingQueue. The flusher thread is
+ // notified when an object is added to the queue. Please note
+ // that each page has an associated blocking queue.
+ flushRequestQueue[pageToFlush].take();
- /*
- * The LogFlusher was waiting for a page to be marked as a
- * candidate for flushing. Now that has happened. The thread
- * shall proceed to take a lock on the log page
- */
- synchronized (logManager.getLogPages()[pageToFlush]) {
+ synchronized (logManager.getLogPage(pageToFlush)) {
- /*
- * lock the internal state of the log manager and create a
- * log file if necessary.
- */
+ // lock the internal state of the log manager and create a
+ // log file if necessary.
int prevLogFileId = logManager.getLogFileId(logManager.getLastFlushedLsn().get());
int nextLogFileId = logManager.getLogFileId(logManager.getLastFlushedLsn().get()
+ logManager.getLogManagerProperties().getLogPageSize());
@@ -936,198 +847,60 @@
logManager.getLogManagerProperties().getLogPageSize());
}
- logManager.getLogPage(pageToFlush).flush(); // put the
- // content to
- // disk, the
- // thread still
- // has a lock on
- // the log page
+ //#. sleep during the groupCommitWaitTime
+ sleep(groupCommitWaitPeriod);
- /*
- * acquire lock on the log manager as we need to update the
- * internal bookkeeping data.
- */
+ //#. set the logPageStatus to INACTIVE in order to prevent other txns from writing on this page.
+ logManager.getLogPageStatus(pageToFlush).set(PageState.INACTIVE);
- // increment the last flushed lsn.
- long lastFlushedLsn = logManager.incrementLastFlushedLsn(logManager.getLogManagerProperties()
- .getLogPageSize());
+ //#. need to wait until the logPageOwnerCount reaches 1 (LOG_WRITER)
+ // meaning every one has finished writing logs on this page.
+ while (logManager.getLogPageOwnershipCount(pageToFlush).get() != PageOwnershipStatus.LOG_WRITER) {
+ sleep(0);
+ }
- /*
- * the log manager gains back ownership of the page. this is
- * reflected by incrementing the owner count of the page.
- * recall that when the page is begin flushed the owner
- * count is actually 0 Value of zero implicitly indicates
- * that the page is operated upon by the log flusher thread.
- */
- logManager.getLogPageOwnershipCount(pageToFlush).incrementAndGet();
+ //#. set the logPageOwnerCount to 0 (LOG_FLUSHER)
+ // meaning it is flushing.
+ logManager.getLogPageOwnershipCount(pageToFlush).set(PageOwnershipStatus.LOG_FLUSHER);
- /*
- * get the number of log buffers that have been written so
- * far. A log buffer = number of log pages * size of a log
- * page
- */
- int numCycles = (int) lastFlushedLsn / logManager.getLogManagerProperties().getLogBufferSize();
- if (lastFlushedLsn % logManager.getLogManagerProperties().getLogBufferSize() == 0) {
- numCycles--;
- }
+ // put the content to disk (the thread still has a lock on the log page)
+ logManager.getLogPage(pageToFlush).flush();
- /*
- * Map the log page to a new region in the log file.
- */
+ // increment the last flushed lsn and lastFlushedPage
+ logManager.incrementLastFlushedLsn(logManager.getLogManagerProperties().getLogPageSize());
+ lastFlushedPageIndex = pageToFlush;
+ // decrement activeTxnCountOnIndexes
+ logManager.decrementActiveTxnCountOnIndexes(pageToFlush);
+
+ // reset the count to 1
+ logManager.getLogPageOwnershipCount(pageToFlush).set(PageOwnershipStatus.LOG_WRITER);
+
+ // Map the log page to a new region in the log file.
long nextWritePosition = logManager.getLogPages()[pageToFlush].getNextWritePosition()
+ logManager.getLogManagerProperties().getLogBufferSize();
- /*
- * long nextPos = (numCycles + 1)
- * logManager.getLogManagerProperties() .getLogBufferSize()
- * + pageToFlush logManager.getLogManagerProperties()
- * .getLogPageSize();
- */
logManager.resetLogPage(nextWritePosition, pageToFlush);
// mark the page as ACTIVE
logManager.getLogPageStatus(pageToFlush).set(LogManager.PageState.ACTIVE);
- // notify all waiting (transaction) threads.
- // Transaction thread may be waiting for the page to be
- // available or may have a commit log record on the page
- // that got flushed.
- logManager.getLogPages()[pageToFlush].notifyAll();
- logManager.setLastFlushedPage(pageToFlush);
+ //#. checks the queue whether there is another flush request on the same log buffer
+ // If there is another request, then simply remove it.
+ if (flushRequestQueue[pageToFlush].peek() != null) {
+ flushRequestQueue[pageToFlush].take();
+ }
+ // notify all waiting (transaction) threads.
+ logManager.getLogPage(pageToFlush).notifyAll();
}
} catch (IOException ioe) {
ioe.printStackTrace();
throw new Error(" exception in flushing log page", ioe);
} catch (InterruptedException e) {
e.printStackTrace();
- break; // must break from the loop as the exception indicates
- // some thing horrendous has happened elsewhere
+ break;
}
}
}
-}
-
-/*
- * TODO: By default the commit policy is to commit at each request and not have
- * a group commit. The following code needs to change to support group commit.
- * The code for group commit has not been tested thoroughly and is under
- * development.
- */
-class BasicCommitResolver implements ICommitResolver {
-
- public boolean shouldCommitPage(int pageIndex, LogManager logManager,
- CommitRequestStatistics commitRequestStatistics) {
- return true;
- }
-
- public void init(LogManager logManager) {
- }
-}
-
-class GroupCommitResolver implements ICommitResolver {
-
- public boolean shouldCommitPage(int pageIndex, LogManager logManager,
- CommitRequestStatistics commitRequestStatistics) {
- long maxCommitWait = logManager.getLogManagerProperties().getGroupCommitWaitPeriod();
- long timestamp = commitRequestStatistics.getPageLevelLastCommitRequestTimestamp(pageIndex);
- if (timestamp == -1) {
- if (maxCommitWait == 0) {
- return true;
- } else {
- timestamp = System.currentTimeMillis();
- }
- }
- long currenTime = System.currentTimeMillis();
- if (currenTime - timestamp > maxCommitWait) {
- return true;
- }
- return false;
- }
-
- public void init(LogManager logManager) {
- GroupCommitHandlerThread groupCommitHandler = new GroupCommitHandlerThread(logManager);
- groupCommitHandler.setDaemon(true);
- groupCommitHandler.start();
- }
-
- class GroupCommitHandlerThread extends Thread {
-
- private LogManager logManager;
-
- public GroupCommitHandlerThread(LogManager logManager) {
- this.logManager = logManager;
- setName("Group Commit Handler");
- }
-
- @Override
- public void run() {
- int pageIndex = -1;
- while (true) {
- pageIndex = logManager.getNextPageInSequence(pageIndex);
- long lastCommitRequeestTimestamp = logManager.getCommitRequestStatistics()
- .getPageLevelLastCommitRequestTimestamp(pageIndex);
- if (lastCommitRequeestTimestamp != -1
- && System.currentTimeMillis() - lastCommitRequeestTimestamp > logManager
- .getLogManagerProperties().getGroupCommitWaitPeriod()) {
- int dirtyCount = logManager.getLogPageOwnershipCount(pageIndex).decrementAndGet();
- if (dirtyCount == 0) {
- try {
- logManager.getLogPageStatus(pageIndex).set(LogManager.PageState.INACTIVE);
- logManager.getPendingFlushRequests(pageIndex).put(Thread.currentThread());
- } catch (InterruptedException e) {
- e.printStackTrace();
- break;
- }
- logManager.getCommitRequestStatistics().committedPage(pageIndex);
- }
- }
- }
- }
- }
-
-}
-
-interface ICommitResolver {
- public boolean shouldCommitPage(int pageIndex, LogManager logManager,
- CommitRequestStatistics commitRequestStatistics);
-
- public void init(LogManager logManager);
-}
-
-/**
- * Represents a collection of all commit requests by transactions for each log
- * page. The requests are accumulated until the commit policy triggers a flush
- * of the corresponding log page. Upon a flush of a page, all commit requests
- * for the page are cleared.
- */
-class CommitRequestStatistics {
-
- AtomicInteger[] pageLevelCommitRequestCount;
- AtomicLong[] pageLevelLastCommitRequestTimestamp;
-
- public CommitRequestStatistics(int numPages) {
- pageLevelCommitRequestCount = new AtomicInteger[numPages];
- pageLevelLastCommitRequestTimestamp = new AtomicLong[numPages];
- for (int i = 0; i < numPages; i++) {
- pageLevelCommitRequestCount[i] = new AtomicInteger(0);
- pageLevelLastCommitRequestTimestamp[i] = new AtomicLong(-1L);
- }
- }
-
- public void registerCommitRequest(int pageIndex) {
- pageLevelCommitRequestCount[pageIndex].incrementAndGet();
- pageLevelLastCommitRequestTimestamp[pageIndex].set(System.currentTimeMillis());
- }
-
- public long getPageLevelLastCommitRequestTimestamp(int pageIndex) {
- return pageLevelLastCommitRequestTimestamp[pageIndex].get();
- }
-
- public void committedPage(int pageIndex) {
- pageLevelCommitRequestCount[pageIndex].set(0);
- pageLevelLastCommitRequestTimestamp[pageIndex].set(-1L);
- }
-
-}
+}
\ No newline at end of file
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogRecordHelper.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogRecordHelper.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogRecordHelper.java (working copy)
@@ -152,6 +152,9 @@
case LogType.UPDATE:
logTypeDisplay = "UPDATE";
break;
+ case LogType.ENTITY_COMMIT:
+ logTypeDisplay = "ENTITY_COMMIT";
+ break;
}
builder.append(" LSN : ").append(logicalLogLocator.getLsn());
builder.append(" Log Type : ").append(logTypeDisplay);
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogType.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogType.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogType.java (working copy)
@@ -18,5 +18,6 @@
public static final byte UPDATE = 0;
public static final byte COMMIT = 1;
+ public static final byte ENTITY_COMMIT = 2;
}
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManagerProperties.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManagerProperties.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManagerProperties.java (working copy)
@@ -1,5 +1,5 @@
/*
- * Copyright 2009-2010 by The Regents of the University of California
+ * Copyright 2009-2012 by The Regents of the University of California
* Licensed 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 from
@@ -41,7 +41,7 @@
private int logPageSize = 128 * 1024; // 128 KB
private int numLogPages = 8; // number of log pages in the log buffer.
- private long groupCommitWaitPeriod = 0; // time in milliseconds for which a
+ private long groupCommitWaitPeriod = 1; // time in milliseconds for which a
// commit record will wait before
// the housing page is marked for
// flushing.
Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/recovery/RecoveryManager.java
===================================================================
--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/recovery/RecoveryManager.java (revision 1194)
+++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/recovery/RecoveryManager.java (working copy)
@@ -184,6 +184,7 @@
break;
case LogType.COMMIT:
+ case LogType.ENTITY_COMMIT:
tempKeyTxnId.setTxnId(logRecordHelper.getJobId(currentLogLocator),
logRecordHelper.getDatasetId(currentLogLocator),
logRecordHelper.getPKHashValue(currentLogLocator));
@@ -218,6 +219,7 @@
IIndex index = null;
LocalResource localResource = null;
ILocalResourceMetadata localResourceMetadata = null;
+ List<Long> resourceIdList = new ArrayList<Long>();
//#. get indexLifeCycleManager
IAsterixAppRuntimeContextProvider appRuntimeContext = txnSubsystem.getAsterixAppRuntimeContextProvider();
@@ -272,6 +274,8 @@
index = localResourceMetadata.createIndexInstance(appRuntimeContext,
localResource.getResourceName(), localResource.getPartition());
indexLifecycleManager.register(resourceId, index);
+ indexLifecycleManager.open(resourceId);
+ resourceIdList.add(resourceId);
}
/***************************************************/
@@ -300,6 +304,7 @@
break;
case LogType.COMMIT:
+ case LogType.ENTITY_COMMIT:
//do nothing
break;
@@ -308,6 +313,11 @@
}
}
+ //close all indexes
+ for (long r : resourceIdList) {
+ indexLifecycleManager.close(r);
+ }
+
JobIdFactory.initJobId(maxJobId);
}
@@ -539,6 +549,7 @@
break;
case LogType.COMMIT:
+ case LogType.ENTITY_COMMIT:
undoLSNSet = loserTxnTable.get(tempKeyTxnId);
if (undoLSNSet != null) {
loserTxnTable.remove(tempKeyTxnId);
Index: asterix-app/src/test/java/edu/uci/ics/asterix/test/metadata/MetadataTest.java
===================================================================
--- asterix-app/src/test/java/edu/uci/ics/asterix/test/metadata/MetadataTest.java (revision 1194)
+++ asterix-app/src/test/java/edu/uci/ics/asterix/test/metadata/MetadataTest.java (working copy)
@@ -42,6 +42,16 @@
List<CompilationUnit> cUnits = tcCtx.getTestCase().getCompilationUnit();
for (CompilationUnit cUnit : cUnits) {
File testFile = tcCtx.getTestFile(cUnit);
+
+ /*****************
+ if (!testFile.getAbsolutePath().contains("meta09.aql")) {
+ System.out.println(testFile.getAbsolutePath());
+ continue;
+ }
+ System.out.println(testFile.getAbsolutePath());
+ *****************/
+
+
File expectedResultFile = tcCtx.getExpectedResultFile(cUnit);
File actualFile = new File(PATH_ACTUAL + File.separator
+ tcCtx.getTestCase().getFilePath().replace(File.separator, "_") + "_" + cUnit.getName() + ".adm");
Index: asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java
===================================================================
--- asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java (revision 1194)
+++ asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java (working copy)
@@ -95,9 +95,10 @@
File testFile = tcCtx.getTestFile(cUnit);
/*************** to avoid run failure cases ****************
- if (!testFile.getAbsolutePath().contains("index-selection/")) {
+ if (!testFile.getAbsolutePath().contains("query-issue205.aql")) {
continue;
}
+ System.out.println(testFile.getAbsolutePath());
************************************************************/
File expectedResultFile = tcCtx.getExpectedResultFile(cUnit);
Index: diff_file
===================================================================
--- diff_file (revision 1194)
+++ diff_file (working copy)
@@ -1,2098 +1,1252 @@
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DataverseTupleTranslator.java
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionManager.java
===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DataverseTupleTranslator.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DataverseTupleTranslator.java (working copy)
-@@ -25,8 +25,11 @@
- import edu.uci.ics.asterix.metadata.bootstrap.MetadataPrimaryIndexes;
- import edu.uci.ics.asterix.metadata.bootstrap.MetadataRecordTypes;
- import edu.uci.ics.asterix.metadata.entities.Dataverse;
-+import edu.uci.ics.asterix.om.base.AInt32;
-+import edu.uci.ics.asterix.om.base.AMutableInt32;
- import edu.uci.ics.asterix.om.base.ARecord;
- import edu.uci.ics.asterix.om.base.AString;
-+import edu.uci.ics.asterix.om.types.BuiltinType;
- import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
- import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionManager.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionManager.java (working copy)
+@@ -103,12 +103,14 @@
+ //for entity-level commit
+ if (PKHashVal != -1) {
+ transactionProvider.getLockManager().unlock(datasetId, PKHashVal, txnContext, true);
++ /*****************************
+ try {
+ //decrease the transaction reference count on index
+ txnContext.decreaseActiveTransactionCountOnIndexes();
+ } catch (HyracksDataException e) {
+ throw new ACIDException("failed to complete index operation", e);
+ }
++ *****************************/
+ return;
+ }
-@@ -40,12 +43,18 @@
- // Payload field containing serialized Dataverse.
- public static final int DATAVERSE_PAYLOAD_TUPLE_FIELD_INDEX = 1;
-
-+ private AMutableInt32 aInt32;
-+ protected ISerializerDeserializer<AInt32> aInt32Serde;
-+
- @SuppressWarnings("unchecked")
- private ISerializerDeserializer<ARecord> recordSerDes = AqlSerializerDeserializerProvider.INSTANCE
- .getSerializerDeserializer(MetadataRecordTypes.DATAVERSE_RECORDTYPE);
-
-+ @SuppressWarnings("unchecked")
- public DataverseTupleTranslator(boolean getTuple) {
- super(getTuple, MetadataPrimaryIndexes.DATAVERSE_DATASET.getFieldCount());
-+ aInt32 = new AMutableInt32(-1);
-+ aInt32Serde = AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT32);
- }
-
- @Override
-@@ -57,7 +66,8 @@
- DataInput in = new DataInputStream(stream);
- ARecord dataverseRecord = recordSerDes.deserialize(in);
- return new Dataverse(((AString) dataverseRecord.getValueByPos(0)).getStringValue(),
-- ((AString) dataverseRecord.getValueByPos(1)).getStringValue());
-+ ((AString) dataverseRecord.getValueByPos(1)).getStringValue(),
-+ ((AInt32) dataverseRecord.getValueByPos(2)).getIntegerValue());
- }
-
- @Override
-@@ -88,6 +98,12 @@
- stringSerde.serialize(aString, fieldValue.getDataOutput());
- recordBuilder.addField(MetadataRecordTypes.DATAVERSE_ARECORD_TIMESTAMP_FIELD_INDEX, fieldValue);
-
-+ // write field 3
-+ fieldValue.reset();
-+ aInt32.setValue(instance.getPendingOp());
-+ aInt32Serde.serialize(aInt32, fieldValue.getDataOutput());
-+ recordBuilder.addField(MetadataRecordTypes.DATAVERSE_ARECORD_PENDINGOP_FIELD_INDEX, fieldValue);
-+
- recordBuilder.write(tupleBuilder.getDataOutput(), true);
- tupleBuilder.addFieldEndOffset();
-
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionContext.java
===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/DatasetTupleTranslator.java (working copy)
-@@ -77,9 +77,9 @@
- protected ISerializerDeserializer<AInt32> aInt32Serde;
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionContext.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/transaction/TransactionContext.java (working copy)
+@@ -19,6 +19,7 @@
+ import java.util.HashSet;
+ import java.util.List;
+ import java.util.Set;
++import java.util.concurrent.atomic.AtomicInteger;
- @SuppressWarnings("unchecked")
-- public DatasetTupleTranslator(boolean getTuple) {
-+ public DatasetTupleTranslator(boolean getTuple) {
- super(getTuple, MetadataPrimaryIndexes.DATASET_DATASET.getFieldCount());
-- aInt32 = new AMutableInt32(-1);
-+ aInt32 = new AMutableInt32(-1);
- aInt32Serde = AqlSerializerDeserializerProvider.INSTANCE.getSerializerDeserializer(BuiltinType.AINT32);
- }
-
-@@ -104,8 +104,10 @@
- .getValueByPos(MetadataRecordTypes.DATASET_ARECORD_DATATYPENAME_FIELD_INDEX)).getStringValue();
- DatasetType datasetType = DatasetType.valueOf(((AString) datasetRecord.getValueByPos(3)).getStringValue());
- IDatasetDetails datasetDetails = null;
-- int datasetId = ((AInt32) datasetRecord
-+ int datasetId = ((AInt32) datasetRecord
- .getValueByPos(MetadataRecordTypes.DATASET_ARECORD_DATASETID_FIELD_INDEX)).getIntegerValue();
-+ int pendingOp = ((AInt32) datasetRecord
-+ .getValueByPos(MetadataRecordTypes.DATASET_ARECORD_PENDINGOP_FIELD_INDEX)).getIntegerValue();
- switch (datasetType) {
- case FEED:
- case INTERNAL: {
-@@ -197,7 +199,7 @@
- }
- datasetDetails = new ExternalDatasetDetails(adapter, properties);
+ import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
+ import edu.uci.ics.asterix.transaction.management.opcallbacks.AbstractOperationCallback;
+@@ -169,5 +170,14 @@
+ closeable.close(this);
}
-- return new Dataset(dataverseName, datasetName, typeName, datasetDetails, datasetType, datasetId);
-+ return new Dataset(dataverseName, datasetName, typeName, datasetDetails, datasetType, datasetId, pendingOp);
}
++
++ @Override
++ public int hashCode() {
++ return jobId.getId();
++ }
- @Override
-@@ -248,13 +250,19 @@
- aString.setValue(Calendar.getInstance().getTime().toString());
- stringSerde.serialize(aString, fieldValue.getDataOutput());
- recordBuilder.addField(MetadataRecordTypes.DATASET_ARECORD_TIMESTAMP_FIELD_INDEX, fieldValue);
--
-+
- // write field 8
- fieldValue.reset();
- aInt32.setValue(dataset.getDatasetId());
- aInt32Serde.serialize(aInt32, fieldValue.getDataOutput());
- recordBuilder.addField(MetadataRecordTypes.DATASET_ARECORD_DATASETID_FIELD_INDEX, fieldValue);
--
-+
-+ // write field 9
-+ fieldValue.reset();
-+ aInt32.setValue(dataset.getPendingOp());
-+ aInt32Serde.serialize(aInt32, fieldValue.getDataOutput());
-+ recordBuilder.addField(MetadataRecordTypes.DATASET_ARECORD_PENDINGOP_FIELD_INDEX, fieldValue);
-+
- // write record
- recordBuilder.write(tupleBuilder.getDataOutput(), true);
- tupleBuilder.addFieldEndOffset();
-@@ -290,13 +298,15 @@
- fieldValue.reset();
- aString.setValue(name);
- stringSerde.serialize(aString, fieldValue.getDataOutput());
-- propertyRecordBuilder.addField(MetadataRecordTypes.DATASOURCE_ADAPTER_PROPERTIES_ARECORD_NAME_FIELD_INDEX, fieldValue);
-+ propertyRecordBuilder.addField(MetadataRecordTypes.DATASOURCE_ADAPTER_PROPERTIES_ARECORD_NAME_FIELD_INDEX,
-+ fieldValue);
-
- // write field 1
- fieldValue.reset();
- aString.setValue(value);
- stringSerde.serialize(aString, fieldValue.getDataOutput());
-- propertyRecordBuilder.addField(MetadataRecordTypes.DATASOURCE_ADAPTER_PROPERTIES_ARECORD_VALUE_FIELD_INDEX, fieldValue);
-+ propertyRecordBuilder.addField(MetadataRecordTypes.DATASOURCE_ADAPTER_PROPERTIES_ARECORD_VALUE_FIELD_INDEX,
-+ fieldValue);
-
- propertyRecordBuilder.write(out, true);
- }
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/IndexTupleTranslator.java
++ @Override
++ public boolean equals(Object o) {
++ return (o == this);
++ }
+ }
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/locking/LockManager.java
===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/IndexTupleTranslator.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entitytupletranslators/IndexTupleTranslator.java (working copy)
-@@ -96,13 +96,15 @@
- }
- Boolean isPrimaryIndex = ((ABoolean) rec.getValueByPos(MetadataRecordTypes.INDEX_ARECORD_ISPRIMARY_FIELD_INDEX))
- .getBoolean();
-+ int pendingOp = ((AInt32) rec.getValueByPos(MetadataRecordTypes.INDEX_ARECORD_PENDINGOP_FIELD_INDEX))
-+ .getIntegerValue();
- // Check if there is a gram length as well.
- int gramLength = -1;
- int gramLenPos = rec.getType().findFieldPosition(GRAM_LENGTH_FIELD_NAME);
- if (gramLenPos >= 0) {
- gramLength = ((AInt32) rec.getValueByPos(gramLenPos)).getIntegerValue();
- }
-- return new Index(dvName, dsName, indexName, indexStructure, searchKey, gramLength, isPrimaryIndex);
-+ return new Index(dvName, dsName, indexName, indexStructure, searchKey, gramLength, isPrimaryIndex, pendingOp);
- }
-
- @Override
-@@ -174,7 +176,12 @@
- stringSerde.serialize(aString, fieldValue.getDataOutput());
- recordBuilder.addField(MetadataRecordTypes.INDEX_ARECORD_TIMESTAMP_FIELD_INDEX, fieldValue);
-
-- // write optional field 7
-+ // write field 7
-+ fieldValue.reset();
-+ intSerde.serialize(new AInt32(instance.getPendingOp()), fieldValue.getDataOutput());
-+ recordBuilder.addField(MetadataRecordTypes.INDEX_ARECORD_PENDINGOP_FIELD_INDEX, fieldValue);
-+
-+ // write optional field 8
- if (instance.getGramLength() > 0) {
- fieldValue.reset();
- nameValue.reset();
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/locking/LockManager.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/locking/LockManager.java (working copy)
+@@ -567,7 +567,7 @@
+ if (commitFlag) {
+ if (txnContext.getTransactionType().equals(TransactionContext.TransactionType.READ_WRITE)) {
+ try {
+- txnSubsystem.getLogManager().log(LogType.COMMIT, txnContext, datasetId.getId(),
++ txnSubsystem.getLogManager().log(LogType.ENTITY_COMMIT, txnContext, datasetId.getId(),
+ entityHashValue, -1, (byte) 0, 0, null, null, logicalLogLocator);
+ } catch (ACIDException e) {
+ try {
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/FileBasedBuffer.java
===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java (working copy)
-@@ -129,7 +129,7 @@
-
- public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, String> {
- private static Logger LOGGER = Logger.getLogger(AqlMetadataProvider.class.getName());
-- private final MetadataTransactionContext mdTxnCtx;
-+ private MetadataTransactionContext mdTxnCtx;
- private boolean isWriteTransaction;
- private Map<String, String[]> stores;
- private Map<String, String> config;
-@@ -156,8 +156,7 @@
- return config;
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/FileBasedBuffer.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/FileBasedBuffer.java (working copy)
+@@ -75,6 +75,7 @@
+ buffer.position(0);
+ buffer.limit(size);
+ fileChannel.write(buffer);
++ fileChannel.force(false);
+ erase();
}
-- public AqlMetadataProvider(MetadataTransactionContext mdTxnCtx, Dataverse defaultDataverse) {
-- this.mdTxnCtx = mdTxnCtx;
-+ public AqlMetadataProvider(Dataverse defaultDataverse) {
- this.defaultDataverse = defaultDataverse;
- this.stores = AsterixProperties.INSTANCE.getStores();
- }
-@@ -181,6 +180,10 @@
- public void setWriterFactory(IAWriterFactory writerFactory) {
- this.writerFactory = writerFactory;
- }
-+
-+ public void setMetadataTxnContext(MetadataTransactionContext mdTxnCtx) {
-+ this.mdTxnCtx = mdTxnCtx;
-+ }
-
- public MetadataTransactionContext getMetadataTxnContext() {
- return mdTxnCtx;
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Dataset.java
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManager.java
===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Dataset.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Dataset.java (working copy)
-@@ -35,15 +35,18 @@
- private final DatasetType datasetType;
- private IDatasetDetails datasetDetails;
- private final int datasetId;
-+ // Type of pending operations with respect to atomic DDL operation
-+ private final int pendingOp;
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManager.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManager.java (working copy)
+@@ -1,5 +1,5 @@
+ /*
+- * Copyright 2009-2010 by The Regents of the University of California
++ * Copyright 2009-2012 by The Regents of the University of California
+ * Licensed 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 from
+@@ -21,7 +21,12 @@
+ import java.io.RandomAccessFile;
+ import java.nio.ByteBuffer;
+ import java.nio.channels.FileChannel;
++import java.util.ArrayList;
++import java.util.HashMap;
++import java.util.List;
++import java.util.Map;
+ import java.util.Properties;
++import java.util.Set;
+ import java.util.concurrent.LinkedBlockingQueue;
+ import java.util.concurrent.atomic.AtomicInteger;
+ import java.util.concurrent.atomic.AtomicLong;
+@@ -30,22 +35,25 @@
- public Dataset(String dataverseName, String datasetName, String itemTypeName, IDatasetDetails datasetDetails,
-- DatasetType datasetType, int datasetId) {
-+ DatasetType datasetType, int datasetId, int pendingOp) {
- this.dataverseName = dataverseName;
- this.datasetName = datasetName;
- this.itemTypeName = itemTypeName;
- this.datasetType = datasetType;
- this.datasetDetails = datasetDetails;
- this.datasetId = datasetId;
-+ this.pendingOp = pendingOp;
- }
+ import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
+ import edu.uci.ics.asterix.transaction.management.service.logging.IndexLogger.ReusableLogContentObject;
++import edu.uci.ics.asterix.transaction.management.service.logging.LogManager.PageOwnershipStatus;
++import edu.uci.ics.asterix.transaction.management.service.logging.LogManager.PageState;
+ import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionContext;
+ import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionManagementConstants;
+ import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionSubsystem;
++import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
- public String getDataverseName() {
-@@ -73,6 +76,10 @@
- public int getDatasetId() {
- return datasetId;
- }
-+
-+ public int getPendingOp() {
-+ return pendingOp;
-+ }
+ public class LogManager implements ILogManager {
- @Override
- public Object addToCache(MetadataCache cache) {
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Index.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Index.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Index.java (working copy)
-@@ -45,9 +45,11 @@
- private final boolean isPrimaryIndex;
- // Specific to NGRAM indexes.
- private final int gramLength;
-+ // Type of pending operations with respect to atomic DDL operation
-+ private final int pendingOp;
+ public static final boolean IS_DEBUG_MODE = false;//true
+ private static final Logger LOGGER = Logger.getLogger(LogManager.class.getName());
+- private TransactionSubsystem provider;
++ private final TransactionSubsystem provider;
+ private LogManagerProperties logManagerProperties;
++ private LogPageFlushThread logPageFlusher;
- public Index(String dataverseName, String datasetName, String indexName, IndexType indexType,
-- List<String> keyFieldNames, int gramLength, boolean isPrimaryIndex) {
-+ List<String> keyFieldNames, int gramLength, boolean isPrimaryIndex, int pendingOp) {
- this.dataverseName = dataverseName;
- this.datasetName = datasetName;
- this.indexName = indexName;
-@@ -55,10 +57,11 @@
- this.keyFieldNames = keyFieldNames;
- this.gramLength = gramLength;
- this.isPrimaryIndex = isPrimaryIndex;
-+ this.pendingOp = pendingOp;
- }
+ /*
+ * the array of log pages. The number of log pages is configurable. Pages
+ * taken together form an in-memory log buffer.
+ */
+-
+ private IFileBasedBuffer[] logPages;
- public Index(String dataverseName, String datasetName, String indexName, IndexType indexType,
-- List<String> keyFieldNames, boolean isPrimaryIndex) {
-+ List<String> keyFieldNames, boolean isPrimaryIndex, int pendingOp) {
- this.dataverseName = dataverseName;
- this.datasetName = datasetName;
- this.indexName = indexName;
-@@ -66,6 +69,7 @@
- this.keyFieldNames = keyFieldNames;
- this.gramLength = -1;
- this.isPrimaryIndex = isPrimaryIndex;
-+ this.pendingOp = pendingOp;
- }
+ private ILogRecordHelper logRecordHelper;
+@@ -54,6 +62,7 @@
+ * Number of log pages that constitute the in-memory log buffer.
+ */
+ private int numLogPages;
++
+ /*
+ * Initially all pages have an owner count of 1 that is the LogManager. When
+ * a transaction requests to write in a log page, the owner count is
+@@ -62,12 +71,11 @@
+ * (covering the whole log record). When the content has been put, the log
+ * manager computes the checksum and puts it after the content. At this
+ * point, the ownership count is decremented as the transaction is done with
+- * using the page. When a page is full, the log manager decrements the count
+- * by one indicating that it has released its ownership of the log page.
+- * There could be other transaction(s) still owning the page (that is they
+- * could still be mid-way putting the log content). When the ownership count
+- * eventually reaches zero, the thread responsible for flushing the log page
+- * is notified and the page is flushed to disk.
++ * using the page. When a page is requested to be flushed, logPageFlusher
++ * set the count to 0(LOG_FLUSHER: meaning that the page is being flushed)
++ * only if the count is 1(LOG_WRITER: meaning that there is no other
++ * transactions who own the page to write logs.) After flushing the page,
++ * logPageFlusher set this count to 1.
+ */
+ private AtomicInteger[] logPageOwnerCount;
- public String getDataverseName() {
-@@ -95,6 +99,10 @@
- public boolean isPrimaryIndex() {
- return isPrimaryIndex;
- }
-+
-+ public int getPendingOp() {
-+ return pendingOp;
-+ }
+@@ -78,18 +86,16 @@
- public boolean isSecondaryIndex() {
- return !isPrimaryIndex();
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Dataverse.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Dataverse.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/entities/Dataverse.java (working copy)
-@@ -27,10 +27,12 @@
- // Enforced to be unique within an Asterix cluster..
- private final String dataverseName;
- private final String dataFormat;
-+ private final int pendingOp;
+ /*
+ * LogPageStatus: A page is either ACTIVE or INACTIVE. The status for each
+- * page is maintained in a map called logPageStatus. A page is ACTIVE when
+- * the LogManager can allocate space in the page for writing a log record.
+- * Initially all pages are ACTIVE. As transactions fill up space by writing
+- * log records, a page may not have sufficient space left for serving a
+- * request by a transaction. When this happens, the page is marked INACTIVE.
+- * An INACTIVE page with no owners ( logPageOwnerCount.get(<pageIndex>) ==
+- * 0) indicates that the page must be flushed to disk before any other log
+- * record is written on the page.F
++ * page is maintained in logPageStatus. A page is ACTIVE when the LogManager
++ * can allocate space in the page for writing a log record. Initially all
++ * pages are ACTIVE. As transactions fill up space by writing log records,
++ * a page may not have sufficient space left for serving a request by a
++ * transaction. When this happens, the page is flushed to disk by calling
++ * logPageFlusher.requestFlush(). In the requestFlush(), after groupCommitWaitTime,
++ * the page status is set to INACTIVE. Then, there is no more writer on the
++ * page(meaning the corresponding logPageOwnerCount is 1), the page is flushed
++ * by the logPageFlusher and the status is reset to ACTIVE by the logPageFlusher.
+ */
+-
+- // private Map<Integer, Integer> logPageStatus = new
+- // ConcurrentHashMap<Integer, Integer>();
+ private AtomicInteger[] logPageStatus;
-- public Dataverse(String dataverseName, String format) {
-+ public Dataverse(String dataverseName, String format, int pendingOp) {
- this.dataverseName = dataverseName;
- this.dataFormat = format;
-+ this.pendingOp = pendingOp;
+ static class PageState {
+@@ -98,41 +104,8 @@
}
- public String getDataverseName() {
-@@ -40,6 +42,10 @@
- public String getDataFormat() {
- return dataFormat;
- }
-+
-+ public int getPendingOp() {
-+ return pendingOp;
-+ }
+ private AtomicLong lastFlushedLsn = new AtomicLong(-1);
+- private AtomicInteger lastFlushedPage = new AtomicInteger(-1);
- @Override
- public Object addToCache(MetadataCache cache) {
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java (working copy)
-@@ -25,6 +25,7 @@
- import edu.uci.ics.asterix.common.exceptions.AsterixException;
- import edu.uci.ics.asterix.common.functions.FunctionSignature;
- import edu.uci.ics.asterix.formats.nontagged.AqlSerializerDeserializerProvider;
-+import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
- import edu.uci.ics.asterix.metadata.api.IMetadataIndex;
- import edu.uci.ics.asterix.metadata.api.IMetadataNode;
- import edu.uci.ics.asterix.metadata.api.IValueExtractor;
-@@ -160,7 +161,7 @@
- // Add the primary index for the dataset.
- InternalDatasetDetails id = (InternalDatasetDetails) dataset.getDatasetDetails();
- Index primaryIndex = new Index(dataset.getDataverseName(), dataset.getDatasetName(),
-- dataset.getDatasetName(), IndexType.BTREE, id.getPrimaryKey(), true);
-+ dataset.getDatasetName(), IndexType.BTREE, id.getPrimaryKey(), true, dataset.getPendingOp());
- addIndex(jobId, primaryIndex);
- ITupleReference nodeGroupTuple = createTuple(id.getNodeGroupName(), dataset.getDataverseName(),
- dataset.getDatasetName());
-@@ -260,7 +261,7 @@
- IIndexAccessor indexAccessor = indexInstance.createAccessor(NoOpOperationCallback.INSTANCE,
- NoOpOperationCallback.INSTANCE);
- TransactionContext txnCtx = transactionProvider.getTransactionManager().getTransactionContext(jobId);
-- transactionProvider.getLockManager().lock(index.getDatasetId(), -1, LockMode.X, txnCtx);
-+ //transactionProvider.getLockManager().lock(index.getDatasetId(), -1, LockMode.X, txnCtx);
- // TODO: fix exceptions once new BTree exception model is in hyracks.
- indexAccessor.insert(tuple);
- //TODO: extract the key from the tuple and get the PKHashValue from the key.
-@@ -536,7 +537,7 @@
- // The transaction with txnId will have an S lock on the
- // resource. Note that lock converters have a higher priority than
- // regular waiters in the LockManager.
-- transactionProvider.getLockManager().lock(index.getDatasetId(), -1, LockMode.X, txnCtx);
-+ //transactionProvider.getLockManager().lock(index.getDatasetId(), -1, LockMode.X, txnCtx);
- indexAccessor.delete(tuple);
- //TODO: extract the key from the tuple and get the PKHashValue from the key.
- //check how to get the oldValue.
-@@ -803,7 +804,9 @@
- private <ResultType> void searchIndex(JobId jobId, IMetadataIndex index, ITupleReference searchKey,
- IValueExtractor<ResultType> valueExtractor, List<ResultType> results) throws Exception {
- TransactionContext txnCtx = transactionProvider.getTransactionManager().getTransactionContext(jobId);
-- transactionProvider.getLockManager().lock(index.getDatasetId(), -1, LockMode.S, txnCtx);
-+ //#. currently lock is not needed to access any metadata
-+ // since the non-compatible concurrent access is always protected by the latch in the MetadataManager.
-+ //transactionProvider.getLockManager().lock(index.getDatasetId(), -1, LockMode.S, txnCtx);
- IBinaryComparatorFactory[] comparatorFactories = index.getKeyBinaryComparatorFactory();
- long resourceID = index.getResourceID();
- IIndex indexInstance = indexLifecycleManager.getIndex(resourceID);
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataEntity.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataEntity.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataEntity.java (working copy)
-@@ -20,6 +20,11 @@
- import edu.uci.ics.asterix.metadata.MetadataCache;
-
- public interface IMetadataEntity extends Serializable {
-+
-+ public static final int PENDING_NO_OP = 0;
-+ public static final int PENDING_ADD_OP = 1;
-+ public static final int PENDING_DROP_OP = 2;
-+
- Object addToCache(MetadataCache cache);
-
- Object dropFromCache(MetadataCache cache);
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataManager.java (working copy)
-@@ -17,6 +17,8 @@
-
- import java.rmi.RemoteException;
- import java.util.List;
-+import java.util.concurrent.locks.ReadWriteLock;
-+import java.util.concurrent.locks.ReentrantReadWriteLock;
-
- import edu.uci.ics.asterix.common.functions.FunctionSignature;
- import edu.uci.ics.asterix.metadata.api.IAsterixStateProxy;
-@@ -79,11 +81,10 @@
- public class MetadataManager implements IMetadataManager {
- // Set in init().
- public static MetadataManager INSTANCE;
+ /*
+- * pendingFlushRequests is a map with key as Integer denoting the page
+- * index. When a (transaction) thread discovers the need to flush a page, it
+- * puts its Thread object into the corresponding value that is a
+- * LinkedBlockingQueue. The LogManager has a LogFlusher thread that scans
+- * this map in order of page index (and circling around). The flusher thread
+- * needs to flush pages in order and waits for a thread to deposit an object
+- * in the blocking queue corresponding to the next page in order. A request
+- * to flush a page is conveyed to the flush thread by simply depositing an
+- * object in to corresponding blocking queue. It is blocking in the sense
+- * that the flusher thread will continue to wait for an object to arrive in
+- * the queue. The object itself is ignored by the fliusher and just acts as
+- * a signal/event that a page needs to be flushed.
+- */
-
- private final MetadataCache cache = new MetadataCache();
- private IAsterixStateProxy proxy;
- private IMetadataNode metadataNode;
+- private LinkedBlockingQueue[] pendingFlushRequests;
-
-+
- public MetadataManager(IAsterixStateProxy proxy) {
- if (proxy == null) {
- throw new Error("Null proxy given to MetadataManager.");
-@@ -206,11 +207,14 @@
+- /*
+- * ICommitResolver is an interface that provides an API that can answer a
+- * simple boolean - Given the commit requests so far, should a page be
+- * flushed. The implementation of the interface contains the logic (or you
+- * can say the policy) for commit. It could be group commit in which case
+- * the commit resolver may not return a true indicating that it wishes to
+- * delay flushing of the page.
+- */
+- private ICommitResolver commitResolver;
+-
+- /*
+- * An object that keeps track of the submitted commit requests.
+- */
+- private CommitRequestStatistics commitRequestStatistics;
+-
+- /*
+ * When the transaction eco-system comes to life, the log manager positions
+ * itself to the end of the last written log. the startingLsn represent the
+ * lsn value of the next log record to be written after a system (re)start.
+@@ -146,16 +119,10 @@
+ */
+ private AtomicLong lsn = new AtomicLong(0);
- @Override
- public void addDataset(MetadataTransactionContext ctx, Dataset dataset) throws MetadataException {
-+ // add dataset into metadataNode
- try {
- metadataNode.addDataset(ctx.getJobId(), dataset);
- } catch (RemoteException e) {
- throw new MetadataException(e);
- }
-+
-+ // reflect the dataset into the cache
- ctx.addDataset(dataset);
+- /*
+- * A map that tracks the flush requests submitted for each page. The
+- * requests for a page are cleared when the page is flushed.
+- */
+- public LinkedBlockingQueue<Thread> getPendingFlushRequests(int pageIndex) {
+- return pendingFlushRequests[pageIndex];
+- }
++ private List<HashMap<TransactionContext, Integer>> activeTxnCountMaps;
+
+- public void addFlushRequest(int pageIndex) {
+- pendingFlushRequests[pageIndex].add(pendingFlushRequests);
++ public void addFlushRequest(int pageIndex, long lsn, boolean isSynchronous) {
++ logPageFlusher.requestFlush(pageIndex, lsn, isSynchronous);
}
-@@ -585,4 +589,5 @@
+ public AtomicLong getLastFlushedLsn() {
+@@ -233,19 +200,12 @@
+ numLogPages = logManagerProperties.getNumLogPages();
+ logPageOwnerCount = new AtomicInteger[numLogPages];
+ logPageStatus = new AtomicInteger[numLogPages];
+- pendingFlushRequests = new LinkedBlockingQueue[numLogPages];
+- if (logManagerProperties.getGroupCommitWaitPeriod() > 0) { // configure
+- // the
+- // Commit
+- // Resolver
+- commitResolver = new GroupCommitResolver(); // Group Commit is
+- // enabled
+- commitRequestStatistics = new CommitRequestStatistics(numLogPages);
+- } else {
+- commitResolver = new BasicCommitResolver(); // the basic commit
+- // resolver
++
++ activeTxnCountMaps = new ArrayList<HashMap<TransactionContext, Integer>>(numLogPages);
++ for (int i = 0; i < numLogPages; i++) {
++ activeTxnCountMaps.add(new HashMap<TransactionContext, Integer>());
}
- return adapter;
- }
+- this.commitResolver.init(this); // initialize the commit resolver
+
- }
-\ No newline at end of file
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataTransactionContext.java (working copy)
-@@ -19,6 +19,7 @@
+ logPages = new FileBasedBuffer[numLogPages];
- import edu.uci.ics.asterix.common.functions.FunctionSignature;
- import edu.uci.ics.asterix.external.dataset.adapter.AdapterIdentifier;
-+import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
- import edu.uci.ics.asterix.metadata.entities.Dataset;
- import edu.uci.ics.asterix.metadata.entities.DatasourceAdapter;
- import edu.uci.ics.asterix.metadata.entities.Datatype;
-@@ -104,19 +105,19 @@
- }
+ /*
+@@ -264,7 +224,6 @@
+ for (int i = 0; i < numLogPages; i++) {
+ logPageOwnerCount[i] = new AtomicInteger(PageOwnershipStatus.LOG_WRITER);
+ logPageStatus[i] = new AtomicInteger(PageState.ACTIVE);
+- pendingFlushRequests[i] = new LinkedBlockingQueue<Thread>();
+ }
- public void dropDataset(String dataverseName, String datasetName) {
-- Dataset dataset = new Dataset(dataverseName, datasetName, null, null, null, -1);
-+ Dataset dataset = new Dataset(dataverseName, datasetName, null, null, null, -1, IMetadataEntity.PENDING_NO_OP);
- droppedCache.addDatasetIfNotExists(dataset);
- logAndApply(new MetadataLogicalOperation(dataset, false));
+ /*
+@@ -278,9 +237,9 @@
+ * daemon thread so that it does not stop the JVM from exiting when all
+ * other threads are done with their work.
+ */
+- LogPageFlushThread logFlusher = new LogPageFlushThread(this);
+- logFlusher.setDaemon(true);
+- logFlusher.start();
++ logPageFlusher = new LogPageFlushThread(this);
++ logPageFlusher.setDaemon(true);
++ logPageFlusher.start();
}
- public void dropIndex(String dataverseName, String datasetName, String indexName) {
-- Index index = new Index(dataverseName, datasetName, indexName, null, null, false);
-+ Index index = new Index(dataverseName, datasetName, indexName, null, null, false, IMetadataEntity.PENDING_NO_OP);
- droppedCache.addIndexIfNotExists(index);
- logAndApply(new MetadataLogicalOperation(index, false));
- }
-
- public void dropDataverse(String dataverseName) {
-- Dataverse dataverse = new Dataverse(dataverseName, null);
-+ Dataverse dataverse = new Dataverse(dataverseName, null, IMetadataEntity.PENDING_NO_OP);
- droppedCache.addDataverseIfNotExists(dataverse);
- logAndApply(new MetadataLogicalOperation(dataverse, false));
- }
-@@ -162,7 +163,7 @@
+ public int getLogPageIndex(long lsnValue) {
+@@ -312,7 +271,7 @@
+ */
+ private void waitUntillPageIsAvailableForWritingLog(int pageIndex) throws ACIDException {
+ if (logPageStatus[pageIndex].get() == PageState.ACTIVE
+- && getLogPageOwnershipCount(pageIndex).get() >= PageOwnershipStatus.LOG_WRITER) {
++ && logPageOwnerCount[pageIndex].get() >= PageOwnershipStatus.LOG_WRITER) {
+ return;
}
- return droppedCache.getDataset(dataverseName, datasetName) != null;
- }
--
+ try {
+@@ -338,47 +297,40 @@
+ */
+ private long getLsn(int entrySize, byte logType) throws ACIDException {
+ long pageSize = logManagerProperties.getLogPageSize();
+- boolean requiresFlushing = logType == LogType.COMMIT;
+
- public boolean indexIsDropped(String dataverseName, String datasetName, String indexName) {
- if (droppedCache.getDataverse(dataverseName) != null) {
- return true;
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataRecordTypes.java (working copy)
-@@ -80,10 +80,11 @@
- public static final int DATAVERSE_ARECORD_NAME_FIELD_INDEX = 0;
- public static final int DATAVERSE_ARECORD_FORMAT_FIELD_INDEX = 1;
- public static final int DATAVERSE_ARECORD_TIMESTAMP_FIELD_INDEX = 2;
-+ public static final int DATAVERSE_ARECORD_PENDINGOP_FIELD_INDEX = 3;
+ while (true) {
+ boolean forwardPage = false;
+- boolean shouldFlushPage = false;
+ long old = lsn.get();
+- int pageIndex = getLogPageIndex(old); // get the log page
+- // corresponding to the
+- // current lsn value
++
++ //get the log page corresponding to the current lsn value
++ int pageIndex = getLogPageIndex(old);
+ long retVal = old;
+- long next = old + entrySize; // the lsn value for the next request,
+- // if the current request is served.
++
++ // the lsn value for the next request if the current request is served.
++ long next = old + entrySize;
+ int prevPage = -1;
+- if ((next - 1) / pageSize != old / pageSize // check if the log
+- // record will cross
+- // page boundaries, a
+- // case that is not
+- // allowed.
+- || (next % pageSize == 0)) {
++
++ // check if the log record will cross page boundaries, a case that is not allowed.
++ if ((next - 1) / pageSize != old / pageSize || (next % pageSize == 0)) {
++
+ if ((old != 0 && old % pageSize == 0)) {
+- retVal = old; // On second thought, this shall never be the
+- // case as it means that the lsn is
+- // currently at the beginning of a page and
+- // we still need to forward the page which
+- // means that the entrySize exceeds a log
+- // page size. If this is the case, an
+- // exception is thrown before calling this
+- // API.
+- // would remove this case.
++ // On second thought, this shall never be the case as it means that the lsn is
++ // currently at the beginning of a page and we still need to forward the page which
++ // means that the entrySize exceeds a log page size. If this is the case, an
++ // exception is thrown before calling this API. would remove this case.
++ retVal = old;
- private static final ARecordType createDataverseRecordType() {
-- return new ARecordType("DataverseRecordType", new String[] { "DataverseName", "DataFormat", "Timestamp" },
-- new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING }, true);
-+ return new ARecordType("DataverseRecordType", new String[] { "DataverseName", "DataFormat", "Timestamp", "PendingOp" },
-+ new IAType[] { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.AINT32 }, true);
- }
+ } else {
+- retVal = ((old / pageSize) + 1) * pageSize; // set the lsn
+- // to point to
+- // the beginning
+- // of the next
+- // page.
++ // set the lsn to point to the beginning of the next page.
++ retVal = ((old / pageSize) + 1) * pageSize;
+ }
++
+ next = retVal;
+- forwardPage = true; // as the log record shall cross log page
+- // boundary, we must re-assign the lsn (so
+- // that the log record begins on a different
+- // location.
++
++ // as the log record shall cross log page boundary, we must re-assign the lsn so
++ // that the log record begins on a different location.
++ forwardPage = true;
++
+ prevPage = pageIndex;
+ pageIndex = getNextPageInSequence(pageIndex);
+ }
+@@ -397,109 +349,51 @@
+ */
+ waitUntillPageIsAvailableForWritingLog(pageIndex);
- // Helper constants for accessing fields in an ARecord of anonymous type
-@@ -158,10 +159,11 @@
- public static final int DATASET_ARECORD_FEEDDETAILS_FIELD_INDEX = 6;
- public static final int DATASET_ARECORD_TIMESTAMP_FIELD_INDEX = 7;
- public static final int DATASET_ARECORD_DATASETID_FIELD_INDEX = 8;
-+ public static final int DATASET_ARECORD_PENDINGOP_FIELD_INDEX = 9;
+- if (!forwardPage && requiresFlushing) {
+- shouldFlushPage = commitResolver.shouldCommitPage(pageIndex, this, commitRequestStatistics);
+- if (shouldFlushPage) {
+- next = ((next / pageSize) + 1) * pageSize; /*
+- * next
+- * represents the
+- * next value of
+- * lsn after this
+- * log record has
+- * been written.
+- * If the page
+- * needs to be
+- * flushed, then
+- * we do not give
+- * any more LSNs
+- * from this
+- * page.
+- */
+- }
+- }
+- if (!lsn.compareAndSet(old, next)) { // Atomic call -> returns true
+- // only when the value
+- // represented by lsn is same as
+- // "old". The value is updated
+- // to "next".
++ if (!lsn.compareAndSet(old, next)) {
++ // Atomic call -> returns true only when the value represented by lsn is same as
++ // "old". The value is updated to "next".
+ continue;
+ }
- private static final ARecordType createDatasetRecordType() {
- String[] fieldNames = { "DataverseName", "DatasetName", "DataTypeName", "DatasetType", "InternalDetails",
-- "ExternalDetails", "FeedDetails", "Timestamp", "DatasetId" };
-+ "ExternalDetails", "FeedDetails", "Timestamp", "DatasetId", "PendingOp" };
+ if (forwardPage) {
+- //TODO
+- //this is not safe since the incoming thread may reach the same page slot with this page
+- //(differ by the log buffer size)
+- logPageStatus[prevPage].set(PageState.INACTIVE); // mark
+- // previous
+- // page
+- // inactive
++ addFlushRequest(prevPage, old, false);
- List<IAType> internalRecordUnionList = new ArrayList<IAType>();
- internalRecordUnionList.add(BuiltinType.ANULL);
-@@ -179,7 +181,8 @@
- AUnionType feedRecordUnion = new AUnionType(feedRecordUnionList, null);
-
- IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING,
-- internalRecordUnion, externalRecordUnion, feedRecordUnion, BuiltinType.ASTRING, BuiltinType.AINT32 };
-+ internalRecordUnion, externalRecordUnion, feedRecordUnion, BuiltinType.ASTRING, BuiltinType.AINT32,
-+ BuiltinType.AINT32 };
- return new ARecordType("DatasetRecordType", fieldNames, fieldTypes, true);
+- /*
+- * decrement on the behalf of the log manager. if there are no
+- * more owners (count == 0) the page must be marked as a
+- * candidate to be flushed.
+- */
+- int pageDirtyCount = getLogPageOwnershipCount(prevPage).decrementAndGet();
+- if (pageDirtyCount == 0) {
+- addFlushRequest(prevPage);
+- }
+-
+- /*
+- * The transaction thread that discovers the need to forward a
+- * page is made to re-acquire a lsn.
+- */
++ // The transaction thread that discovers the need to forward a
++ // page is made to re-acquire a lsn.
+ continue;
++
+ } else {
+- /*
+- * the transaction thread has been given a space in a log page,
+- * but is made to wait until the page is available.
+- */
++ // the transaction thread has been given a space in a log page,
++ // but is made to wait until the page is available.
++ // (Is this needed? when does this wait happen?)
+ waitUntillPageIsAvailableForWritingLog(pageIndex);
+- /*
+- * increment the counter as the transaction thread now holds a
+- * space in the log page and hence is an owner.
+- */
++
++ // increment the counter as the transaction thread now holds a
++ // space in the log page and hence is an owner.
+ logPageOwnerCount[pageIndex].incrementAndGet();
+- }
+- if (requiresFlushing) {
+- if (!shouldFlushPage) {
+- /*
+- * the log record requires the page to be flushed but under
+- * the commit policy, the flush task has been deferred. The
+- * transaction thread submits its request to flush the page.
+- */
+- commitRequestStatistics.registerCommitRequest(pageIndex);
+- } else {
+- /*
+- * the flush request was approved by the commit resolver.
+- * Thus the page is marked INACTIVE as no more logs will be
+- * written on this page. The log manager needs to release
+- * its ownership. Note that transaction threads may still
+- * continue to be owners of the log page till they fill up
+- * the space allocated to them.
+- */
+- logPageStatus[pageIndex].set(PageState.INACTIVE);
+- logPageOwnerCount[pageIndex].decrementAndGet(); // on
+- // the
+- // behalf
+- // of
+- // log
+- // manager
++
++ // Before the count is incremented, if the flusher flushed the allocated page,
++ // then retry to get new LSN. Otherwise, the log with allocated lsn will be lost.
++ if (lastFlushedLsn.get() >= retVal) {
++ logPageOwnerCount[pageIndex].decrementAndGet();
++ continue;
+ }
+ }
++
+ return retVal;
+ }
}
-@@ -264,13 +267,14 @@
- public static final int INDEX_ARECORD_SEARCHKEY_FIELD_INDEX = 4;
- public static final int INDEX_ARECORD_ISPRIMARY_FIELD_INDEX = 5;
- public static final int INDEX_ARECORD_TIMESTAMP_FIELD_INDEX = 6;
-+ public static final int INDEX_ARECORD_PENDINGOP_FIELD_INDEX = 7;
+ @Override
+- public void log(byte logType, TransactionContext context, int datasetId, int PKHashValue, long resourceId,
++ public void log(byte logType, TransactionContext txnCtx, int datasetId, int PKHashValue, long resourceId,
+ byte resourceMgrId, int logContentSize, ReusableLogContentObject reusableLogContentObject, ILogger logger,
+ LogicalLogLocator logicalLogLocator) throws ACIDException {
+- /*
+- * logLocator is a re-usable object that is appropriately set in each
+- * invocation. If the reference is null, the log manager must throw an
+- * exception
+- */
++
++ HashMap<TransactionContext, Integer> map = null;
++ int activeTxnCount;
++
++ // logLocator is a re-usable object that is appropriately set in each invocation.
++ // If the reference is null, the log manager must throw an exception.
+ if (logicalLogLocator == null) {
+ throw new ACIDException(
+ " you need to pass in a non-null logLocator, if you dont have it, then pass in a dummy so that the +"
+@@ -519,20 +413,19 @@
- private static final ARecordType createIndexRecordType() {
- AOrderedListType olType = new AOrderedListType(BuiltinType.ASTRING, null);
- String[] fieldNames = { "DataverseName", "DatasetName", "IndexName", "IndexStructure", "SearchKey",
-- "IsPrimary", "Timestamp" };
-+ "IsPrimary", "Timestamp", "PendingOp" };
- IAType[] fieldTypes = { BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING, BuiltinType.ASTRING,
-- olType, BuiltinType.ABOOLEAN, BuiltinType.ASTRING };
-+ olType, BuiltinType.ABOOLEAN, BuiltinType.ASTRING, BuiltinType.AINT32 };
- return new ARecordType("IndexRecordType", fieldNames, fieldTypes, true);
- };
+ // all constraints checked and we are good to go and acquire a lsn.
+ long previousLSN = -1;
+- long currentLSN; // the will be set to the location (a long value)
+- // where the log record needs to be placed.
-Index: asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
-===================================================================
---- asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java (revision 1061)
-+++ asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java (working copy)
-@@ -31,6 +31,7 @@
- import edu.uci.ics.asterix.metadata.IDatasetDetails;
- import edu.uci.ics.asterix.metadata.MetadataManager;
- import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
-+import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
- import edu.uci.ics.asterix.metadata.api.IMetadataIndex;
- import edu.uci.ics.asterix.metadata.entities.AsterixBuiltinTypeMap;
- import edu.uci.ics.asterix.metadata.entities.Dataset;
-@@ -226,7 +227,7 @@
- public static void insertInitialDataverses(MetadataTransactionContext mdTxnCtx) throws Exception {
- String dataverseName = MetadataPrimaryIndexes.DATAVERSE_DATASET.getDataverseName();
- String dataFormat = NonTaggedDataFormat.NON_TAGGED_DATA_FORMAT;
-- MetadataManager.INSTANCE.addDataverse(mdTxnCtx, new Dataverse(dataverseName, dataFormat));
-+ MetadataManager.INSTANCE.addDataverse(mdTxnCtx, new Dataverse(dataverseName, dataFormat, IMetadataEntity.PENDING_NO_OP));
- }
+- /*
+- * The logs written by a transaction need to be linked to each other for
+- * a successful rollback/recovery. However there could be multiple
+- * threads operating concurrently that are part of a common transaction.
+- * These threads need to synchronize and record the lsn corresponding to
+- * the last log record written by (any thread of) the transaction.
+- */
+- synchronized (context) {
+- previousLSN = context.getLastLogLocator().getLsn();
++ // the will be set to the location (a long value) where the log record needs to be placed.
++ long currentLSN;
++
++ // The logs written by a transaction need to be linked to each other for
++ // a successful rollback/recovery. However there could be multiple
++ // threads operating concurrently that are part of a common transaction.
++ // These threads need to synchronize and record the lsn corresponding to
++ // the last log record written by (any thread of) the transaction.
++ synchronized (txnCtx) {
++ previousLSN = txnCtx.getLastLogLocator().getLsn();
+ currentLSN = getLsn(totalLogSize, logType);
+- context.setLastLSN(currentLSN);
++ txnCtx.setLastLSN(currentLSN);
+ if (IS_DEBUG_MODE) {
+ System.out.println("--------------> LSN(" + currentLSN + ") is allocated");
+ }
+@@ -547,48 +440,37 @@
+ * performed correctly that is ownership is released.
+ */
- public static void insertInitialDatasets(MetadataTransactionContext mdTxnCtx) throws Exception {
-@@ -236,7 +237,7 @@
- primaryIndexes[i].getNodeGroupName());
- MetadataManager.INSTANCE.addDataset(mdTxnCtx, new Dataset(primaryIndexes[i].getDataverseName(),
- primaryIndexes[i].getIndexedDatasetName(), primaryIndexes[i].getPayloadRecordType().getTypeName(),
-- id, DatasetType.INTERNAL, primaryIndexes[i].getDatasetId().getId()));
-+ id, DatasetType.INTERNAL, primaryIndexes[i].getDatasetId().getId(), IMetadataEntity.PENDING_NO_OP));
- }
- }
+- boolean decremented = false; // indicates if the transaction thread
+- // has release ownership of the
+- // page.
+- boolean addedFlushRequest = false; // indicates if the transaction
+- // thread has submitted a flush
+- // request.
++ // indicates if the transaction thread has release ownership of the page.
++ boolean decremented = false;
-@@ -267,7 +268,7 @@
- for (int i = 0; i < secondaryIndexes.length; i++) {
- MetadataManager.INSTANCE.addIndex(mdTxnCtx, new Index(secondaryIndexes[i].getDataverseName(),
- secondaryIndexes[i].getIndexedDatasetName(), secondaryIndexes[i].getIndexName(), IndexType.BTREE,
-- secondaryIndexes[i].getPartitioningExpr(), false));
-+ secondaryIndexes[i].getPartitioningExpr(), false, IMetadataEntity.PENDING_NO_OP));
- }
- }
+ int pageIndex = (int) getLogPageIndex(currentLSN);
-Index: asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java
-===================================================================
---- asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java (revision 1061)
-+++ asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java (working copy)
-@@ -95,11 +95,11 @@
- File testFile = tcCtx.getTestFile(cUnit);
-
- /*************** to avoid run failure cases ****************
-- if (!testFile.getAbsolutePath().contains("runtimets/queries/failure/")) {
-+ if (!testFile.getAbsolutePath().contains("index-selection/")) {
- continue;
- }
- ************************************************************/
+- /*
+- * the lsn has been obtained for the log record. need to set the
+- * LogLocator instance accordingly.
+- */
-
-+
- File expectedResultFile = tcCtx.getExpectedResultFile(cUnit);
- File actualFile = new File(PATH_ACTUAL + File.separator
- + tcCtx.getTestCase().getFilePath().replace(File.separator, "_") + "_" + cUnit.getName() + ".adm");
-Index: asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
-===================================================================
---- asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java (revision 1061)
-+++ asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java (working copy)
-@@ -90,7 +90,7 @@
++ // the lsn has been obtained for the log record. need to set the
++ // LogLocator instance accordingly.
+ try {
+-
+ logicalLogLocator.setBuffer(logPages[pageIndex]);
+ int pageOffset = getLogPageOffset(currentLSN);
+ logicalLogLocator.setMemoryOffset(pageOffset);
- private static Logger LOGGER = Logger.getLogger(DatasetOperations.class.getName());
+- /*
+- * write the log header.
+- */
+- logRecordHelper.writeLogHeader(logicalLogLocator, logType, context, datasetId, PKHashValue, previousLSN,
++ // write the log header.
++ logRecordHelper.writeLogHeader(logicalLogLocator, logType, txnCtx, datasetId, PKHashValue, previousLSN,
+ resourceId, resourceMgrId, logContentSize);
-- public static JobSpecification[] createDropDatasetJobSpec(CompiledDatasetDropStatement datasetDropStmt,
-+ public static JobSpecification createDropDatasetJobSpec(CompiledDatasetDropStatement datasetDropStmt,
- AqlMetadataProvider metadataProvider) throws AlgebricksException, HyracksDataException, RemoteException,
- ACIDException, AsterixException {
+ // increment the offset so that the transaction can fill up the
+ // content in the correct region of the allocated space.
+ logicalLogLocator.increaseMemoryOffset(logRecordHelper.getLogHeaderSize(logType));
-@@ -111,67 +111,10 @@
- throw new AlgebricksException("DROP DATASET: No metadata for dataset " + datasetName);
- }
- if (dataset.getDatasetType() == DatasetType.EXTERNAL) {
-- return new JobSpecification[0];
-+ return new JobSpecification();
- }
--
-- List<Index> datasetIndexes = metadataProvider.getDatasetIndexes(dataset.getDataverseName(),
-- dataset.getDatasetName());
-- int numSecondaryIndexes = 0;
-- for (Index index : datasetIndexes) {
-- if (index.isSecondaryIndex()) {
-- numSecondaryIndexes++;
-- }
-- }
-- JobSpecification[] specs;
-- if (numSecondaryIndexes > 0) {
-- specs = new JobSpecification[numSecondaryIndexes + 1];
-- int i = 0;
-- // First, drop secondary indexes.
-- for (Index index : datasetIndexes) {
-- if (index.isSecondaryIndex()) {
-- specs[i] = new JobSpecification();
-- Pair<IFileSplitProvider, AlgebricksPartitionConstraint> idxSplitsAndConstraint = metadataProvider
-- .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(dataset.getDataverseName(),
-- datasetName, index.getIndexName());
-- IIndexDataflowHelperFactory dfhFactory;
-- switch (index.getIndexType()) {
-- case BTREE:
-- dfhFactory = new LSMBTreeDataflowHelperFactory(
-- AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER,
-- AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER,
-- AsterixRuntimeComponentsProvider.LSMBTREE_PROVIDER);
-- break;
-- case RTREE:
-- dfhFactory = new LSMRTreeDataflowHelperFactory(
-- new IPrimitiveValueProviderFactory[] { null }, RTreePolicyType.RTREE,
-- new IBinaryComparatorFactory[] { null },
-- AsterixRuntimeComponentsProvider.LSMRTREE_PROVIDER,
-- AsterixRuntimeComponentsProvider.LSMRTREE_PROVIDER,
-- AsterixRuntimeComponentsProvider.LSMRTREE_PROVIDER, null);
-- break;
-- case NGRAM_INVIX:
-- case WORD_INVIX:
-- dfhFactory = new LSMInvertedIndexDataflowHelperFactory(
-- AsterixRuntimeComponentsProvider.LSMINVERTEDINDEX_PROVIDER,
-- AsterixRuntimeComponentsProvider.LSMINVERTEDINDEX_PROVIDER,
-- AsterixRuntimeComponentsProvider.LSMINVERTEDINDEX_PROVIDER);
-- break;
-- default:
-- throw new AsterixException("Unknown index type provided.");
-- }
-- IndexDropOperatorDescriptor secondaryBtreeDrop = new IndexDropOperatorDescriptor(specs[i],
-- AsterixRuntimeComponentsProvider.NOINDEX_PROVIDER,
-- AsterixRuntimeComponentsProvider.NOINDEX_PROVIDER, idxSplitsAndConstraint.first, dfhFactory);
-- AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(specs[i], secondaryBtreeDrop,
-- idxSplitsAndConstraint.second);
-- i++;
-- }
-- }
-- } else {
-- specs = new JobSpecification[1];
-- }
-+
- JobSpecification specPrimary = new JobSpecification();
-- specs[specs.length - 1] = specPrimary;
+- // a COMMIT log record does not have any content
+- // and hence the logger (responsible for putting the log content) is
+- // not invoked.
++ // a COMMIT log record does not have any content and hence
++ // the logger (responsible for putting the log content) is not invoked.
+ if (logContentSize != 0) {
+- logger.preLog(context, reusableLogContentObject);
++ logger.preLog(txnCtx, reusableLogContentObject);
+ }
- Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = metadataProvider
- .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(dataset.getDataverseName(), datasetName,
-@@ -187,7 +130,7 @@
+ if (logContentSize != 0) {
+ // call the logger implementation and ask to fill in the log
+ // record content at the allocated space.
+- logger.log(context, logicalLogLocator, logContentSize, reusableLogContentObject);
+- logger.postLog(context, reusableLogContentObject);
++ logger.log(txnCtx, logicalLogLocator, logContentSize, reusableLogContentObject);
++ logger.postLog(txnCtx, reusableLogContentObject);
+ if (IS_DEBUG_MODE) {
+ logicalLogLocator.setMemoryOffset(logicalLogLocator.getMemoryOffset()
+ - logRecordHelper.getLogHeaderSize(logType));
+@@ -597,10 +479,8 @@
+ }
+ }
- specPrimary.addRoot(primaryBtreeDrop);
+- /*
+- * The log record has been written. For integrity checks, compute
+- * the checksum and put it at the end of the log record.
+- */
++ // The log record has been written. For integrity checks, compute
++ // the checksum and put it at the end of the log record.
+ int startPosChecksum = logicalLogLocator.getMemoryOffset() - logRecordHelper.getLogHeaderSize(logType);
+ int length = totalLogSize - logRecordHelper.getLogChecksumSize();
+ long checksum = DataUtil.getChecksum(logPages[pageIndex], startPosChecksum, length);
+@@ -611,46 +491,31 @@
+ System.out.println("--------------> LSN(" + currentLSN + ") is written");
+ }
-- return specs;
-+ return specPrimary;
- }
+- /*
+- * release the ownership as the log record has been placed in
+- * created space.
+- */
+- int pageDirtyCount = logPageOwnerCount[pageIndex].decrementAndGet();
++ // release the ownership as the log record has been placed in created space.
++ logPageOwnerCount[pageIndex].decrementAndGet();
- public static JobSpecification createDatasetJobSpec(Dataverse dataverse, String datasetName,
-Index: asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java
-===================================================================
---- asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java (revision 1061)
-+++ asterix-app/src/main/java/edu/uci/ics/asterix/aql/translator/AqlTranslator.java (working copy)
-@@ -21,6 +21,8 @@
- import java.util.HashMap;
- import java.util.List;
- import java.util.Map;
-+import java.util.concurrent.locks.ReadWriteLock;
-+import java.util.concurrent.locks.ReentrantReadWriteLock;
+ // indicating that the transaction thread has released ownership
+ decremented = true;
- import org.json.JSONException;
+- /*
+- * If the transaction thread happens to be the last owner of the log
+- * page the page must by marked as a candidate to be flushed.
+- */
+- if (pageDirtyCount == 0 && logPageStatus[pageIndex].get() == PageState.INACTIVE) {
+- addFlushRequest(pageIndex);
+- addedFlushRequest = true;
+- }
+-
+- /*
+- * If the log type is commit, a flush request is registered, if the
+- * log record has not reached the disk. It may be possible that this
+- * thread does not get CPU cycles and in-between the log record has
+- * been flushed to disk because the containing log page filled up.
+- */
+- if (logType == LogType.COMMIT) {
+- synchronized (logPages[pageIndex]) {
+- while (getLastFlushedLsn().get() < currentLSN) {
+- logPages[pageIndex].wait();
+- }
++ if (logType == LogType.ENTITY_COMMIT) {
++ map = activeTxnCountMaps.get(pageIndex);
++ if (map.containsKey(txnCtx)) {
++ activeTxnCount = (Integer) map.get(txnCtx);
++ activeTxnCount++;
++ map.put(txnCtx, activeTxnCount);
++ } else {
++ map.put(txnCtx, 1);
+ }
++ addFlushRequest(pageIndex, currentLSN, false);
++ } else if (logType == LogType.COMMIT) {
++ addFlushRequest(pageIndex, currentLSN, true);
+ }
-@@ -68,6 +70,7 @@
- import edu.uci.ics.asterix.metadata.MetadataException;
- import edu.uci.ics.asterix.metadata.MetadataManager;
- import edu.uci.ics.asterix.metadata.MetadataTransactionContext;
-+import edu.uci.ics.asterix.metadata.api.IMetadataEntity;
- import edu.uci.ics.asterix.metadata.declared.AqlMetadataProvider;
- import edu.uci.ics.asterix.metadata.entities.Dataset;
- import edu.uci.ics.asterix.metadata.entities.Datatype;
-@@ -112,6 +115,7 @@
- private final PrintWriter out;
- private final SessionConfig sessionConfig;
- private final DisplayFormat pdf;
-+ private final ReadWriteLock cacheLatch;
- private Dataverse activeDefaultDataverse;
- private List<FunctionDecl> declaredFunctions;
+ } catch (Exception e) {
+ e.printStackTrace();
+- throw new ACIDException(context, "Thread: " + Thread.currentThread().getName()
++ throw new ACIDException(txnCtx, "Thread: " + Thread.currentThread().getName()
+ + " logger encountered exception", e);
+ } finally {
+- /*
+- * If an exception was encountered and we did not release ownership
+- */
+ if (!decremented) {
+ logPageOwnerCount[pageIndex].decrementAndGet();
+ }
+@@ -667,9 +532,6 @@
-@@ -121,6 +125,7 @@
- this.out = out;
- this.sessionConfig = pc;
- this.pdf = pdf;
-+ this.cacheLatch = new ReentrantReadWriteLock(true);
- declaredFunctions = getDeclaredFunctions(aqlStatements);
+ logPages[pageIndex].reset(filePath, LogUtil.getFileOffset(this, nextWritePosition),
+ logManagerProperties.getLogPageSize());
+-
+- //TODO Check if this is necessary
+- //Arrays.fill(logPages[pageIndex].getArray(), (byte) 0);
}
-@@ -143,8 +148,7 @@
+ @Override
+@@ -747,16 +609,13 @@
+ //minimize memory allocation overhead. current code allocates the log page size per reading a log record.
- for (Statement stmt : aqlStatements) {
- validateOperation(activeDefaultDataverse, stmt);
-- MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-- AqlMetadataProvider metadataProvider = new AqlMetadataProvider(mdTxnCtx, activeDefaultDataverse);
-+ AqlMetadataProvider metadataProvider = new AqlMetadataProvider(activeDefaultDataverse);
- metadataProvider.setWriterFactory(writerFactory);
- metadataProvider.setOutputFile(outputFile);
- metadataProvider.setConfig(config);
-@@ -253,15 +257,9 @@
- }
+ byte[] pageContent = new byte[logManagerProperties.getLogPageSize()];
+- // take a lock on the log page so that the page is not flushed to
+- // disk interim
++
++ // take a lock on the log page so that the page is not flushed to disk interim
+ synchronized (logPages[pageIndex]) {
+- if (lsnValue > getLastFlushedLsn().get()) { // need to check
+- // again
+- // (this
+- // thread may have got
+- // de-scheduled and must
+- // refresh!)
- }
-- MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
- } catch (Exception e) {
-- MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
- throw new AlgebricksException(e);
++ // need to check again (this thread may have got de-scheduled and must refresh!)
++ if (lsnValue > getLastFlushedLsn().get()) {
++
+ // get the log record length
+ logPages[pageIndex].getBytes(pageContent, 0, pageContent.length);
+ byte logType = pageContent[pageOffset + 4];
+@@ -765,9 +624,7 @@
+ int logRecordSize = logHeaderSize + logBodySize + logRecordHelper.getLogChecksumSize();
+ logRecord = new byte[logRecordSize];
+
+- /*
+- * copy the log record content
+- */
++ // copy the log record content
+ System.arraycopy(pageContent, pageOffset, logRecord, 0, logRecordSize);
+ MemBasedBuffer memBuffer = new MemBasedBuffer(logRecord);
+ if (logicalLogLocator == null) {
+@@ -790,9 +647,7 @@
}
-- // Following jobs are run under a separate transaction, that is committed/aborted by the JobEventListener
-- for (JobSpecification jobspec : jobsToExecute) {
-- runJob(hcc, jobspec);
-- }
}
- return executionResult;
+
+- /*
+- * the log record is residing on the disk, read it from there.
+- */
++ // the log record is residing on the disk, read it from there.
+ readDiskLog(lsnValue, logicalLogLocator);
}
-@@ -289,398 +287,802 @@
- private Dataverse handleUseDataverseStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws MetadataException, RemoteException, ACIDException {
-- DataverseDecl dvd = (DataverseDecl) stmt;
-- String dvName = dvd.getDataverseName().getValue();
-- Dataverse dv = MetadataManager.INSTANCE.getDataverse(metadataProvider.getMetadataTxnContext(), dvName);
-- if (dv == null) {
-- throw new MetadataException("Unknown dataverse " + dvName);
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
-+
-+ try {
-+ DataverseDecl dvd = (DataverseDecl) stmt;
-+ String dvName = dvd.getDataverseName().getValue();
-+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(metadataProvider.getMetadataTxnContext(), dvName);
-+ if (dv == null) {
-+ throw new MetadataException("Unknown dataverse " + dvName);
-+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ return dv;
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new MetadataException(e);
-+ } finally {
-+ releaseReadLatch();
- }
-- return dv;
+@@ -860,30 +715,40 @@
+ return logPageOwnerCount[pageIndex];
}
- private void handleCreateDataverseStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws MetadataException, AlgebricksException, RemoteException,
- ACIDException {
-- CreateDataverseStatement stmtCreateDataverse = (CreateDataverseStatement) stmt;
-- String dvName = stmtCreateDataverse.getDataverseName().getValue();
-- Dataverse dv = MetadataManager.INSTANCE.getDataverse(metadataProvider.getMetadataTxnContext(), dvName);
-- if (dv != null && !stmtCreateDataverse.getIfNotExists()) {
-- throw new AlgebricksException("A dataverse with this name " + dvName + " already exists.");
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ CreateDataverseStatement stmtCreateDataverse = (CreateDataverseStatement) stmt;
-+ String dvName = stmtCreateDataverse.getDataverseName().getValue();
-+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(metadataProvider.getMetadataTxnContext(), dvName);
-+ if (dv != null && !stmtCreateDataverse.getIfNotExists()) {
-+ throw new AlgebricksException("A dataverse with this name " + dvName + " already exists.");
-+ }
-+ MetadataManager.INSTANCE.addDataverse(metadataProvider.getMetadataTxnContext(), new Dataverse(dvName,
-+ stmtCreateDataverse.getFormat(), IMetadataEntity.PENDING_NO_OP));
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
-- MetadataManager.INSTANCE.addDataverse(metadataProvider.getMetadataTxnContext(), new Dataverse(dvName,
-- stmtCreateDataverse.getFormat()));
+- public ICommitResolver getCommitResolver() {
+- return commitResolver;
+- }
+-
+- public CommitRequestStatistics getCommitRequestStatistics() {
+- return commitRequestStatistics;
+- }
+-
+ public IFileBasedBuffer[] getLogPages() {
+ return logPages;
}
- private void handleCreateDatasetStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws AsterixException, Exception {
-- DatasetDecl dd = (DatasetDecl) stmt;
-- String dataverseName = dd.getDataverse() != null ? dd.getDataverse().getValue()
-- : activeDefaultDataverse != null ? activeDefaultDataverse.getDataverseName() : null;
-- if (dataverseName == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-- }
-- String datasetName = dd.getName().getValue();
-- DatasetType dsType = dd.getDatasetType();
-- String itemTypeName = dd.getItemTypeName().getValue();
-
-- IDatasetDetails datasetDetails = null;
-- Dataset ds = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(), dataverseName,
-- datasetName);
-- if (ds != null) {
-- if (dd.getIfNotExists()) {
-- return;
-- } else {
-- throw new AlgebricksException("A dataset with this name " + datasetName + " already exists.");
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
+- public int getLastFlushedPage() {
+- return lastFlushedPage.get();
+- }
+-
+- public void setLastFlushedPage(int lastFlushedPage) {
+- this.lastFlushedPage.set(lastFlushedPage);
+- }
+-
+ @Override
+ public TransactionSubsystem getTransactionSubsystem() {
+ return provider;
+ }
+
-+ try {
-+ DatasetDecl dd = (DatasetDecl) stmt;
-+ String dataverseName = dd.getDataverse() != null ? dd.getDataverse().getValue()
-+ : activeDefaultDataverse != null ? activeDefaultDataverse.getDataverseName() : null;
-+ if (dataverseName == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
- }
-- }
-- Datatype dt = MetadataManager.INSTANCE.getDatatype(metadataProvider.getMetadataTxnContext(), dataverseName,
-- itemTypeName);
-- if (dt == null) {
-- throw new AlgebricksException(": type " + itemTypeName + " could not be found.");
-- }
-- switch (dd.getDatasetType()) {
-- case INTERNAL: {
-- IAType itemType = dt.getDatatype();
-- if (itemType.getTypeTag() != ATypeTag.RECORD) {
-- throw new AlgebricksException("Can only partition ARecord's.");
-+ String datasetName = dd.getName().getValue();
-+ DatasetType dsType = dd.getDatasetType();
-+ String itemTypeName = dd.getItemTypeName().getValue();
++ public void decrementActiveTxnCountOnIndexes(int pageIndex) throws HyracksDataException {
++ TransactionContext ctx = null;
++ int count = 0;
++ int i = 0;
+
-+ IDatasetDetails datasetDetails = null;
-+ Dataset ds = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(), dataverseName,
-+ datasetName);
-+ if (ds != null) {
-+ if (dd.getIfNotExists()) {
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ return;
-+ } else {
-+ throw new AlgebricksException("A dataset with this name " + datasetName + " already exists.");
- }
-- List<String> partitioningExprs = ((InternalDetailsDecl) dd.getDatasetDetailsDecl())
-- .getPartitioningExprs();
-- String ngName = ((InternalDetailsDecl) dd.getDatasetDetailsDecl()).getNodegroupName().getValue();
-- datasetDetails = new InternalDatasetDetails(InternalDatasetDetails.FileStructure.BTREE,
-- InternalDatasetDetails.PartitioningStrategy.HASH, partitioningExprs, partitioningExprs, ngName);
-- break;
- }
-- case EXTERNAL: {
-- String adapter = ((ExternalDetailsDecl) dd.getDatasetDetailsDecl()).getAdapter();
-- Map<String, String> properties = ((ExternalDetailsDecl) dd.getDatasetDetailsDecl()).getProperties();
-- datasetDetails = new ExternalDatasetDetails(adapter, properties);
-- break;
-+ Datatype dt = MetadataManager.INSTANCE.getDatatype(metadataProvider.getMetadataTxnContext(), dataverseName,
-+ itemTypeName);
-+ if (dt == null) {
-+ throw new AlgebricksException(": type " + itemTypeName + " could not be found.");
- }
-- case FEED: {
-- IAType itemType = dt.getDatatype();
-- if (itemType.getTypeTag() != ATypeTag.RECORD) {
-- throw new AlgebricksException("Can only partition ARecord's.");
-+ switch (dd.getDatasetType()) {
-+ case INTERNAL: {
-+ IAType itemType = dt.getDatatype();
-+ if (itemType.getTypeTag() != ATypeTag.RECORD) {
-+ throw new AlgebricksException("Can only partition ARecord's.");
++ HashMap<TransactionContext, Integer> map = activeTxnCountMaps.get(pageIndex);
++ Set<Map.Entry<TransactionContext, Integer>> entrySet = map.entrySet();
++ if (entrySet != null) {
++ for (Map.Entry<TransactionContext, Integer> entry : entrySet) {
++ if (entry != null) {
++ if (entry.getValue() != null) {
++ count = entry.getValue();
+ }
-+ List<String> partitioningExprs = ((InternalDetailsDecl) dd.getDatasetDetailsDecl())
-+ .getPartitioningExprs();
-+ String ngName = ((InternalDetailsDecl) dd.getDatasetDetailsDecl()).getNodegroupName().getValue();
-+ datasetDetails = new InternalDatasetDetails(InternalDatasetDetails.FileStructure.BTREE,
-+ InternalDatasetDetails.PartitioningStrategy.HASH, partitioningExprs, partitioningExprs,
-+ ngName);
-+ break;
- }
-- List<String> partitioningExprs = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getPartitioningExprs();
-- String ngName = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getNodegroupName().getValue();
-- String adapter = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getAdapterFactoryClassname();
-- Map<String, String> configuration = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getConfiguration();
-- FunctionSignature signature = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getFunctionSignature();
-- datasetDetails = new FeedDatasetDetails(InternalDatasetDetails.FileStructure.BTREE,
-- InternalDatasetDetails.PartitioningStrategy.HASH, partitioningExprs, partitioningExprs, ngName,
-- adapter, configuration, signature, FeedDatasetDetails.FeedState.INACTIVE.toString());
-- break;
-+ case EXTERNAL: {
-+ String adapter = ((ExternalDetailsDecl) dd.getDatasetDetailsDecl()).getAdapter();
-+ Map<String, String> properties = ((ExternalDetailsDecl) dd.getDatasetDetailsDecl()).getProperties();
-+ datasetDetails = new ExternalDatasetDetails(adapter, properties);
-+ break;
-+ }
-+ case FEED: {
-+ IAType itemType = dt.getDatatype();
-+ if (itemType.getTypeTag() != ATypeTag.RECORD) {
-+ throw new AlgebricksException("Can only partition ARecord's.");
++ if (count > 0) {
++ ctx = entry.getKey();
++ for (i = 0; i < count; i++) {
++ ctx.decreaseActiveTransactionCountOnIndexes();
++ }
+ }
-+ List<String> partitioningExprs = ((FeedDetailsDecl) dd.getDatasetDetailsDecl())
-+ .getPartitioningExprs();
-+ String ngName = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getNodegroupName().getValue();
-+ String adapter = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getAdapterFactoryClassname();
-+ Map<String, String> configuration = ((FeedDetailsDecl) dd.getDatasetDetailsDecl())
-+ .getConfiguration();
-+ FunctionSignature signature = ((FeedDetailsDecl) dd.getDatasetDetailsDecl()).getFunctionSignature();
-+ datasetDetails = new FeedDatasetDetails(InternalDatasetDetails.FileStructure.BTREE,
-+ InternalDatasetDetails.PartitioningStrategy.HASH, partitioningExprs, partitioningExprs,
-+ ngName, adapter, configuration, signature, FeedDatasetDetails.FeedState.INACTIVE.toString());
-+ break;
+ }
- }
-+
-+ //#. add a new dataset with PendingAddOp
-+ Dataset dataset = new Dataset(dataverseName, datasetName, itemTypeName, datasetDetails, dsType,
-+ DatasetIdFactory.generateDatasetId(), IMetadataEntity.PENDING_ADD_OP);
-+ MetadataManager.INSTANCE.addDataset(metadataProvider.getMetadataTxnContext(), dataset);
-+
-+ if (dd.getDatasetType() == DatasetType.INTERNAL || dd.getDatasetType() == DatasetType.FEED) {
-+ Dataverse dataverse = MetadataManager.INSTANCE.getDataverse(metadataProvider.getMetadataTxnContext(),
-+ dataverseName);
-+ JobSpecification jobSpec = DatasetOperations.createDatasetJobSpec(dataverse, datasetName,
-+ metadataProvider);
-+
-+ //#. make metadataTxn commit before calling runJob.
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ //#. runJob
-+ runJob(hcc, jobSpec);
-+
-+ //#. begin new metadataTxn
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
+ }
++ }
+
-+ //#. add a new dataset with PendingNoOp after deleting the dataset with PendingAddOp
-+ MetadataManager.INSTANCE.dropDataset(metadataProvider.getMetadataTxnContext(), dataverseName, datasetName);
-+ MetadataManager.INSTANCE.addDataset(metadataProvider.getMetadataTxnContext(), new Dataset(dataverseName,
-+ datasetName, itemTypeName, datasetDetails, dsType, dataset.getDatasetId(),
-+ IMetadataEntity.PENDING_NO_OP));
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
-- MetadataManager.INSTANCE.addDataset(metadataProvider.getMetadataTxnContext(), new Dataset(dataverseName,
-- datasetName, itemTypeName, datasetDetails, dsType, DatasetIdFactory.generateDatasetId()));
-- if (dd.getDatasetType() == DatasetType.INTERNAL || dd.getDatasetType() == DatasetType.FEED) {
-- Dataverse dataverse = MetadataManager.INSTANCE.getDataverse(metadataProvider.getMetadataTxnContext(),
-- dataverseName);
-- runJob(hcc, DatasetOperations.createDatasetJobSpec(dataverse, datasetName, metadataProvider));
-- }
- }
++ map.clear();
++ }
+ }
- private void handleCreateIndexStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- CreateIndexStatement stmtCreateIndex = (CreateIndexStatement) stmt;
-- String dataverseName = stmtCreateIndex.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : stmtCreateIndex.getDataverseName().getValue();
-- if (dataverseName == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-- }
-- String datasetName = stmtCreateIndex.getDatasetName().getValue();
-- Dataset ds = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(), dataverseName,
-- datasetName);
-- if (ds == null) {
-- throw new AlgebricksException("There is no dataset with this name " + datasetName + " in dataverse "
-- + dataverseName);
-- }
-- String indexName = stmtCreateIndex.getIndexName().getValue();
-- Index idx = MetadataManager.INSTANCE.getIndex(metadataProvider.getMetadataTxnContext(), dataverseName,
-- datasetName, indexName);
-- if (idx != null) {
-- if (!stmtCreateIndex.getIfNotExists()) {
-- throw new AlgebricksException("An index with this name " + indexName + " already exists.");
-- } else {
-- stmtCreateIndex.setNeedToCreate(false);
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ CreateIndexStatement stmtCreateIndex = (CreateIndexStatement) stmt;
-+ String dataverseName = stmtCreateIndex.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : stmtCreateIndex.getDataverseName().getValue();
-+ if (dataverseName == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
- }
-- } else {
-+ String datasetName = stmtCreateIndex.getDatasetName().getValue();
-+
-+ Dataset ds = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(), dataverseName,
-+ datasetName);
-+ if (ds == null) {
-+ throw new AlgebricksException("There is no dataset with this name " + datasetName + " in dataverse "
-+ + dataverseName);
-+ }
-+
-+ String indexName = stmtCreateIndex.getIndexName().getValue();
-+ Index idx = MetadataManager.INSTANCE.getIndex(metadataProvider.getMetadataTxnContext(), dataverseName,
-+ datasetName, indexName);
-+
-+ if (idx != null) {
-+ if (!stmtCreateIndex.getIfNotExists()) {
-+ throw new AlgebricksException("An index with this name " + indexName + " already exists.");
-+ } else {
-+ stmtCreateIndex.setNeedToCreate(false);
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ return;
-+ }
-+ }
-+
-+ //#. add a new index with PendingAddOp
- Index index = new Index(dataverseName, datasetName, indexName, stmtCreateIndex.getIndexType(),
-- stmtCreateIndex.getFieldExprs(), stmtCreateIndex.getGramLength(), false);
-+ stmtCreateIndex.getFieldExprs(), stmtCreateIndex.getGramLength(), false,
-+ IMetadataEntity.PENDING_ADD_OP);
- MetadataManager.INSTANCE.addIndex(metadataProvider.getMetadataTxnContext(), index);
-- runCreateIndexJob(hcc, stmtCreateIndex, metadataProvider);
+ /*
+@@ -895,36 +760,82 @@
+ class LogPageFlushThread extends Thread {
-+ //#. create the index artifact in NC.
- CompiledCreateIndexStatement cis = new CompiledCreateIndexStatement(index.getIndexName(), dataverseName,
- index.getDatasetName(), index.getKeyFieldNames(), index.getGramLength(), index.getIndexType());
-- JobSpecification loadIndexJobSpec = IndexOperations
-- .buildSecondaryIndexLoadingJobSpec(cis, metadataProvider);
-- runJob(hcc, loadIndexJobSpec);
-+ JobSpecification spec = IndexOperations.buildSecondaryIndexCreationJobSpec(cis, metadataProvider);
-+ if (spec == null) {
-+ throw new AsterixException("Failed to create job spec for creating index '"
-+ + stmtCreateIndex.getDatasetName() + "." + stmtCreateIndex.getIndexName() + "'");
-+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ runJob(hcc, spec);
-+
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+
-+ //#. load data into the index in NC.
-+ cis = new CompiledCreateIndexStatement(index.getIndexName(), dataverseName, index.getDatasetName(),
-+ index.getKeyFieldNames(), index.getGramLength(), index.getIndexType());
-+ spec = IndexOperations.buildSecondaryIndexLoadingJobSpec(cis, metadataProvider);
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ runJob(hcc, spec);
-+
-+ //#. begin new metadataTxn
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+
-+ //#. add another new index with PendingNoOp after deleting the index with PendingAddOp
-+ MetadataManager.INSTANCE.dropIndex(metadataProvider.getMetadataTxnContext(), dataverseName, datasetName,
-+ indexName);
-+ index = new Index(dataverseName, datasetName, indexName, stmtCreateIndex.getIndexType(),
-+ stmtCreateIndex.getFieldExprs(), stmtCreateIndex.getGramLength(), false,
-+ IMetadataEntity.PENDING_NO_OP);
-+ MetadataManager.INSTANCE.addIndex(metadataProvider.getMetadataTxnContext(), index);
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
- }
+ private LogManager logManager;
++ /*
++ * pendingFlushRequests is a map with key as Integer denoting the page
++ * index. When a (transaction) thread discovers the need to flush a page, it
++ * puts its Thread object into the corresponding value that is a
++ * LinkedBlockingQueue. The LogManager has a LogFlusher thread that scans
++ * this map in order of page index (and circling around). The flusher thread
++ * needs to flush pages in order and waits for a thread to deposit an object
++ * in the blocking queue corresponding to the next page in order. A request
++ * to flush a page is conveyed to the flush thread by simply depositing an
++ * object in to corresponding blocking queue. It is blocking in the sense
++ * that the flusher thread will continue to wait for an object to arrive in
++ * the queue. The object itself is ignored by the fliusher and just acts as
++ * a signal/event that a page needs to be flushed.
++ */
++ private final LinkedBlockingQueue<Object>[] flushRequestQueue;
++ private final Object[] flushRequests;
++ private int lastFlushedPageIndex;
++ private final long groupCommitWaitPeriod;
- private void handleCreateTypeStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws AlgebricksException, RemoteException, ACIDException,
- MetadataException {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- TypeDecl stmtCreateType = (TypeDecl) stmt;
-- String dataverseName = stmtCreateType.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : stmtCreateType.getDataverseName().getValue();
-- if (dataverseName == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-- }
-- String typeName = stmtCreateType.getIdent().getValue();
-- Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dataverseName);
-- if (dv == null) {
-- throw new AlgebricksException("Unknonw dataverse " + dataverseName);
-- }
-- Datatype dt = MetadataManager.INSTANCE.getDatatype(mdTxnCtx, dataverseName, typeName);
-- if (dt != null) {
-- if (!stmtCreateType.getIfNotExists())
-- throw new AlgebricksException("A datatype with this name " + typeName + " already exists.");
-- } else {
-- if (builtinTypeMap.get(typeName) != null) {
-- throw new AlgebricksException("Cannot redefine builtin type " + typeName + ".");
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ TypeDecl stmtCreateType = (TypeDecl) stmt;
-+ String dataverseName = stmtCreateType.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : stmtCreateType.getDataverseName().getValue();
-+ if (dataverseName == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
-+ }
-+ String typeName = stmtCreateType.getIdent().getValue();
-+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dataverseName);
-+ if (dv == null) {
-+ throw new AlgebricksException("Unknonw dataverse " + dataverseName);
-+ }
-+ Datatype dt = MetadataManager.INSTANCE.getDatatype(mdTxnCtx, dataverseName, typeName);
-+ if (dt != null) {
-+ if (!stmtCreateType.getIfNotExists()) {
-+ throw new AlgebricksException("A datatype with this name " + typeName + " already exists.");
-+ }
- } else {
-- Map<TypeSignature, IAType> typeMap = TypeTranslator.computeTypes(mdTxnCtx, (TypeDecl) stmt,
-- dataverseName);
-- TypeSignature typeSignature = new TypeSignature(dataverseName, typeName);
-- IAType type = typeMap.get(typeSignature);
-- MetadataManager.INSTANCE.addDatatype(mdTxnCtx, new Datatype(dataverseName, typeName, type, false));
-+ if (builtinTypeMap.get(typeName) != null) {
-+ throw new AlgebricksException("Cannot redefine builtin type " + typeName + ".");
-+ } else {
-+ Map<TypeSignature, IAType> typeMap = TypeTranslator.computeTypes(mdTxnCtx, (TypeDecl) stmt,
-+ dataverseName);
-+ TypeSignature typeSignature = new TypeSignature(dataverseName, typeName);
-+ IAType type = typeMap.get(typeSignature);
-+ MetadataManager.INSTANCE.addDatatype(mdTxnCtx, new Datatype(dataverseName, typeName, type, false));
-+ }
- }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
+ public LogPageFlushThread(LogManager logManager) {
+ this.logManager = logManager;
+ setName("Flusher");
++ int numLogPages = logManager.getLogManagerProperties().getNumLogPages();
++ this.flushRequestQueue = new LinkedBlockingQueue[numLogPages];
++ this.flushRequests = new Object[numLogPages];
++ for (int i = 0; i < numLogPages; i++) {
++ flushRequestQueue[i] = new LinkedBlockingQueue<Object>(1);
++ flushRequests[i] = new Object();
++ }
++ this.lastFlushedPageIndex = -1;
++ groupCommitWaitPeriod = logManager.getLogManagerProperties().getGroupCommitWaitPeriod();
}
- private void handleDataverseDropStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- DataverseDropStatement stmtDelete = (DataverseDropStatement) stmt;
-- String dvName = stmtDelete.getDataverseName().getValue();
-
-- Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dvName);
-- if (dv == null) {
-- if (!stmtDelete.getIfExists()) {
-- throw new AlgebricksException("There is no dataverse with this name " + dvName + ".");
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ DataverseDropStatement stmtDelete = (DataverseDropStatement) stmt;
-+ String dvName = stmtDelete.getDataverseName().getValue();
-+
-+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dvName);
-+ if (dv == null) {
-+ if (!stmtDelete.getIfExists()) {
-+ throw new AlgebricksException("There is no dataverse with this name " + dvName + ".");
-+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
++ public void requestFlush(int pageIndex, long lsn, boolean isSynchronous) {
++ synchronized (logManager.getLogPage(pageIndex)) {
++ //return if flushedLSN >= lsn
++ if (logManager.getLastFlushedLsn().get() >= lsn) {
+ return;
- }
-- } else {
-+
-+ //#. prepare jobs which will drop corresponding datasets with indexes.
- List<Dataset> datasets = MetadataManager.INSTANCE.getDataverseDatasets(mdTxnCtx, dvName);
- for (int j = 0; j < datasets.size(); j++) {
- String datasetName = datasets.get(j).getDatasetName();
- DatasetType dsType = datasets.get(j).getDatasetType();
- if (dsType == DatasetType.INTERNAL || dsType == DatasetType.FEED) {
-+
- List<Index> indexes = MetadataManager.INSTANCE.getDatasetIndexes(mdTxnCtx, dvName, datasetName);
- for (int k = 0; k < indexes.size(); k++) {
- if (indexes.get(k).isSecondaryIndex()) {
-- compileIndexDropStatement(hcc, dvName, datasetName, indexes.get(k).getIndexName(),
-- metadataProvider);
-+ CompiledIndexDropStatement cds = new CompiledIndexDropStatement(dvName, datasetName,
-+ indexes.get(k).getIndexName());
-+ jobsToExecute.add(IndexOperations.buildDropSecondaryIndexJobSpec(cds, metadataProvider));
- }
- }
-+
-+ CompiledDatasetDropStatement cds = new CompiledDatasetDropStatement(dvName, datasetName);
-+ jobsToExecute.add(DatasetOperations.createDropDatasetJobSpec(cds, metadataProvider));
- }
-- compileDatasetDropStatement(hcc, dvName, datasetName, metadataProvider);
- }
-
-+ //#. mark PendingDropOp on the dataverse record by
-+ // first, deleting the dataverse record from the DATAVERSE_DATASET
-+ // second, inserting the dataverse record with the PendingDropOp value into the DATAVERSE_DATASET
- MetadataManager.INSTANCE.dropDataverse(mdTxnCtx, dvName);
-+ MetadataManager.INSTANCE.addDataverse(mdTxnCtx, new Dataverse(dvName, dv.getDataFormat(),
-+ IMetadataEntity.PENDING_DROP_OP));
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ for (JobSpecification jobSpec : jobsToExecute) {
-+ runJob(hcc, jobSpec);
+ }
+
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
++ //put a new request to the queue only if the request on the page is not in the queue.
++ flushRequestQueue[pageIndex].offer(flushRequests[pageIndex]);
+
-+ //#. finally, delete the dataverse.
-+ MetadataManager.INSTANCE.dropDataverse(mdTxnCtx, dvName);
- if (activeDefaultDataverse != null && activeDefaultDataverse.getDataverseName() == dvName) {
- activeDefaultDataverse = null;
- }
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
++ //return if the request is asynchronous
++ if (!isSynchronous) {
++ return;
+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
- }
-
- private void handleDatasetDropStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- DropStatement stmtDelete = (DropStatement) stmt;
-- String dataverseName = stmtDelete.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : stmtDelete.getDataverseName().getValue();
-- if (dataverseName == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-- }
-- String datasetName = stmtDelete.getDatasetName().getValue();
-- Dataset ds = MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverseName, datasetName);
-- if (ds == null) {
-- if (!stmtDelete.getIfExists())
-- throw new AlgebricksException("There is no dataset with this name " + datasetName + " in dataverse "
-- + dataverseName + ".");
-- } else {
+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ DropStatement stmtDelete = (DropStatement) stmt;
-+ String dataverseName = stmtDelete.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : stmtDelete.getDataverseName().getValue();
-+ if (dataverseName == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
-+ }
-+ String datasetName = stmtDelete.getDatasetName().getValue();
-+
-+ Dataset ds = MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverseName, datasetName);
-+ if (ds == null) {
-+ if (!stmtDelete.getIfExists()) {
-+ throw new AlgebricksException("There is no dataset with this name " + datasetName
-+ + " in dataverse " + dataverseName + ".");
++ //wait until there is flush.
++ boolean isNotified = false;
++ while (!isNotified) {
++ try {
++ logManager.getLogPage(pageIndex).wait();
++ isNotified = true;
++ } catch (InterruptedException e) {
++ e.printStackTrace();
+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ return;
+ }
++ }
++ }
+
- if (ds.getDatasetType() == DatasetType.INTERNAL || ds.getDatasetType() == DatasetType.FEED) {
-+
-+ //#. prepare jobs to drop the datatset and the indexes in NC
- List<Index> indexes = MetadataManager.INSTANCE.getDatasetIndexes(mdTxnCtx, dataverseName, datasetName);
- for (int j = 0; j < indexes.size(); j++) {
-- if (indexes.get(j).isPrimaryIndex()) {
-- compileIndexDropStatement(hcc, dataverseName, datasetName, indexes.get(j).getIndexName(),
-- metadataProvider);
-+ if (indexes.get(j).isSecondaryIndex()) {
-+ CompiledIndexDropStatement cds = new CompiledIndexDropStatement(dataverseName, datasetName,
-+ indexes.get(j).getIndexName());
-+ jobsToExecute.add(IndexOperations.buildDropSecondaryIndexJobSpec(cds, metadataProvider));
+ @Override
+ public void run() {
+ while (true) {
+ try {
+- int pageToFlush = logManager.getNextPageInSequence(logManager.getLastFlushedPage());
++ int pageToFlush = logManager.getNextPageInSequence(lastFlushedPageIndex);
+
+- /*
+- * A wait call on the linkedBLockingQueue. The flusher thread is
+- * notified when an object is added to the queue. Please note
+- * that each page has an associated blocking queue.
+- */
+- logManager.getPendingFlushRequests(pageToFlush).take();
++ // A wait call on the linkedBLockingQueue. The flusher thread is
++ // notified when an object is added to the queue. Please note
++ // that each page has an associated blocking queue.
++ flushRequestQueue[pageToFlush].take();
+
+- /*
+- * The LogFlusher was waiting for a page to be marked as a
+- * candidate for flushing. Now that has happened. The thread
+- * shall proceed to take a lock on the log page
+- */
+- synchronized (logManager.getLogPages()[pageToFlush]) {
++ synchronized (logManager.getLogPage(pageToFlush)) {
+
+- /*
+- * lock the internal state of the log manager and create a
+- * log file if necessary.
+- */
++ // lock the internal state of the log manager and create a
++ // log file if necessary.
+ int prevLogFileId = logManager.getLogFileId(logManager.getLastFlushedLsn().get());
+ int nextLogFileId = logManager.getLogFileId(logManager.getLastFlushedLsn().get()
+ + logManager.getLogManagerProperties().getLogPageSize());
+@@ -936,198 +847,60 @@
+ logManager.getLogManagerProperties().getLogPageSize());
}
- }
-+ CompiledDatasetDropStatement cds = new CompiledDatasetDropStatement(dataverseName, datasetName);
-+ jobsToExecute.add(DatasetOperations.createDropDatasetJobSpec(cds, metadataProvider));
-+
-+ //#. mark the existing dataset as PendingDropOp
-+ MetadataManager.INSTANCE.dropDataset(mdTxnCtx, dataverseName, datasetName);
-+ MetadataManager.INSTANCE.addDataset(
-+ mdTxnCtx,
-+ new Dataset(dataverseName, datasetName, ds.getItemTypeName(), ds.getDatasetDetails(), ds
-+ .getDatasetType(), ds.getDatasetId(), IMetadataEntity.PENDING_DROP_OP));
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ //#. run the jobs
-+ for (JobSpecification jobSpec : jobsToExecute) {
-+ runJob(hcc, jobSpec);
-+ }
-+
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
- }
-- compileDatasetDropStatement(hcc, dataverseName, datasetName, metadataProvider);
-+
-+ //#. finally, delete the dataset.
-+ MetadataManager.INSTANCE.dropDataset(mdTxnCtx, dataverseName, datasetName);
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
- }
- private void handleIndexDropStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- IndexDropStatement stmtIndexDrop = (IndexDropStatement) stmt;
-- String datasetName = stmtIndexDrop.getDatasetName().getValue();
-- String dataverseName = stmtIndexDrop.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : stmtIndexDrop.getDataverseName().getValue();
-- if (dataverseName == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ IndexDropStatement stmtIndexDrop = (IndexDropStatement) stmt;
-+ String datasetName = stmtIndexDrop.getDatasetName().getValue();
-+ String dataverseName = stmtIndexDrop.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : stmtIndexDrop.getDataverseName().getValue();
-+ if (dataverseName == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
-+ }
-+
-+ Dataset ds = MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverseName, datasetName);
-+ if (ds == null) {
-+ throw new AlgebricksException("There is no dataset with this name " + datasetName + " in dataverse "
-+ + dataverseName);
-+ }
-+
-+ if (ds.getDatasetType() == DatasetType.INTERNAL || ds.getDatasetType() == DatasetType.FEED) {
-+ String indexName = stmtIndexDrop.getIndexName().getValue();
-+ Index index = MetadataManager.INSTANCE.getIndex(mdTxnCtx, dataverseName, datasetName, indexName);
-+ if (index == null) {
-+ if (!stmtIndexDrop.getIfExists()) {
-+ throw new AlgebricksException("There is no index with this name " + indexName + ".");
-+ }
-+ } else {
-+ //#. prepare a job to drop the index in NC.
-+ CompiledIndexDropStatement cds = new CompiledIndexDropStatement(dataverseName, datasetName,
-+ indexName);
-+ jobsToExecute.add(IndexOperations.buildDropSecondaryIndexJobSpec(cds, metadataProvider));
-+
-+ //#. mark PendingDropOp on the existing index
-+ MetadataManager.INSTANCE.dropIndex(mdTxnCtx, dataverseName, datasetName, indexName);
-+ MetadataManager.INSTANCE.addIndex(
-+ mdTxnCtx,
-+ new Index(dataverseName, datasetName, indexName, index.getIndexType(), index
-+ .getKeyFieldNames(), index.isPrimaryIndex(), IMetadataEntity.PENDING_DROP_OP));
-+
-+ //#. commit the existing transaction before calling runJob.
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ for (JobSpecification jobSpec : jobsToExecute) {
-+ runJob(hcc, jobSpec);
-+ }
-+
-+ //#. begin a new transaction
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+
-+ //#. finally, delete the existing index
-+ MetadataManager.INSTANCE.dropIndex(mdTxnCtx, dataverseName, datasetName, indexName);
-+ }
-+ } else {
-+ throw new AlgebricksException(datasetName
-+ + " is an external dataset. Indexes are not maintained for external datasets.");
-+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+
-+ } finally {
-+ releaseWriteLatch();
- }
-- Dataset ds = MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverseName, datasetName);
-- if (ds == null)
-- throw new AlgebricksException("There is no dataset with this name " + datasetName + " in dataverse "
-- + dataverseName);
-- if (ds.getDatasetType() == DatasetType.INTERNAL || ds.getDatasetType() == DatasetType.FEED) {
-- String indexName = stmtIndexDrop.getIndexName().getValue();
-- Index idx = MetadataManager.INSTANCE.getIndex(mdTxnCtx, dataverseName, datasetName, indexName);
-- if (idx == null) {
-- if (!stmtIndexDrop.getIfExists())
-- throw new AlgebricksException("There is no index with this name " + indexName + ".");
-- } else
-- compileIndexDropStatement(hcc, dataverseName, datasetName, indexName, metadataProvider);
-- } else {
-- throw new AlgebricksException(datasetName
-- + " is an external dataset. Indexes are not maintained for external datasets.");
-- }
- }
+- logManager.getLogPage(pageToFlush).flush(); // put the
+- // content to
+- // disk, the
+- // thread still
+- // has a lock on
+- // the log page
++ //#. sleep during the groupCommitWaitTime
++ sleep(groupCommitWaitPeriod);
- private void handleTypeDropStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws AlgebricksException, MetadataException, RemoteException,
- ACIDException {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- TypeDropStatement stmtTypeDrop = (TypeDropStatement) stmt;
-- String dataverseName = stmtTypeDrop.getDataverseName() == null ? (activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName()) : stmtTypeDrop.getDataverseName().getValue();
-- if (dataverseName == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ TypeDropStatement stmtTypeDrop = (TypeDropStatement) stmt;
-+ String dataverseName = stmtTypeDrop.getDataverseName() == null ? (activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName()) : stmtTypeDrop.getDataverseName().getValue();
-+ if (dataverseName == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
-+ }
-+ String typeName = stmtTypeDrop.getTypeName().getValue();
-+ Datatype dt = MetadataManager.INSTANCE.getDatatype(mdTxnCtx, dataverseName, typeName);
-+ if (dt == null) {
-+ if (!stmtTypeDrop.getIfExists())
-+ throw new AlgebricksException("There is no datatype with this name " + typeName + ".");
-+ } else {
-+ MetadataManager.INSTANCE.dropDatatype(mdTxnCtx, dataverseName, typeName);
-+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
-- String typeName = stmtTypeDrop.getTypeName().getValue();
-- Datatype dt = MetadataManager.INSTANCE.getDatatype(mdTxnCtx, dataverseName, typeName);
-- if (dt == null) {
-- if (!stmtTypeDrop.getIfExists())
-- throw new AlgebricksException("There is no datatype with this name " + typeName + ".");
-- } else {
-- MetadataManager.INSTANCE.dropDatatype(mdTxnCtx, dataverseName, typeName);
-- }
- }
+- /*
+- * acquire lock on the log manager as we need to update the
+- * internal bookkeeping data.
+- */
++ //#. set the logPageStatus to INACTIVE in order to prevent other txns from writing on this page.
++ logManager.getLogPageStatus(pageToFlush).set(PageState.INACTIVE);
- private void handleNodegroupDropStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws MetadataException, AlgebricksException, RemoteException,
- ACIDException {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- NodeGroupDropStatement stmtDelete = (NodeGroupDropStatement) stmt;
-- String nodegroupName = stmtDelete.getNodeGroupName().getValue();
-- NodeGroup ng = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, nodegroupName);
-- if (ng == null) {
-- if (!stmtDelete.getIfExists())
-- throw new AlgebricksException("There is no nodegroup with this name " + nodegroupName + ".");
-- } else {
-- MetadataManager.INSTANCE.dropNodegroup(mdTxnCtx, nodegroupName);
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ NodeGroupDropStatement stmtDelete = (NodeGroupDropStatement) stmt;
-+ String nodegroupName = stmtDelete.getNodeGroupName().getValue();
-+ NodeGroup ng = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, nodegroupName);
-+ if (ng == null) {
-+ if (!stmtDelete.getIfExists())
-+ throw new AlgebricksException("There is no nodegroup with this name " + nodegroupName + ".");
-+ } else {
-+ MetadataManager.INSTANCE.dropNodegroup(mdTxnCtx, nodegroupName);
-+ }
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
- }
+- // increment the last flushed lsn.
+- long lastFlushedLsn = logManager.incrementLastFlushedLsn(logManager.getLogManagerProperties()
+- .getLogPageSize());
++ //#. need to wait until the logPageOwnerCount reaches 1 (LOG_WRITER)
++ // meaning every one has finished writing logs on this page.
++ while (logManager.getLogPageOwnershipCount(pageToFlush).get() != PageOwnershipStatus.LOG_WRITER) {
++ sleep(0);
++ }
- private void handleCreateFunctionStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws AlgebricksException, MetadataException, RemoteException,
- ACIDException {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- CreateFunctionStatement cfs = (CreateFunctionStatement) stmt;
-- String dataverse = cfs.getSignature().getNamespace() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : cfs.getSignature().getNamespace();
-- if (dataverse == null) {
-- throw new AlgebricksException(" dataverse not specified ");
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ CreateFunctionStatement cfs = (CreateFunctionStatement) stmt;
-+ String dataverse = cfs.getSignature().getNamespace() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : cfs.getSignature().getNamespace();
-+ if (dataverse == null) {
-+ throw new AlgebricksException(" dataverse not specified ");
-+ }
-+ Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dataverse);
-+ if (dv == null) {
-+ throw new AlgebricksException("There is no dataverse with this name " + dataverse + ".");
-+ }
-+ Function function = new Function(dataverse, cfs.getaAterixFunction().getName(), cfs.getaAterixFunction()
-+ .getArity(), cfs.getParamList(), Function.RETURNTYPE_VOID, cfs.getFunctionBody(),
-+ Function.LANGUAGE_AQL, FunctionKind.SCALAR.toString());
-+ MetadataManager.INSTANCE.addFunction(mdTxnCtx, function);
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
-- Dataverse dv = MetadataManager.INSTANCE.getDataverse(mdTxnCtx, dataverse);
-- if (dv == null) {
-- throw new AlgebricksException("There is no dataverse with this name " + dataverse + ".");
-- }
-- Function function = new Function(dataverse, cfs.getaAterixFunction().getName(), cfs.getaAterixFunction()
-- .getArity(), cfs.getParamList(), Function.RETURNTYPE_VOID, cfs.getFunctionBody(),
-- Function.LANGUAGE_AQL, FunctionKind.SCALAR.toString());
-- MetadataManager.INSTANCE.addFunction(mdTxnCtx, function);
- }
+- /*
+- * the log manager gains back ownership of the page. this is
+- * reflected by incrementing the owner count of the page.
+- * recall that when the page is begin flushed the owner
+- * count is actually 0 Value of zero implicitly indicates
+- * that the page is operated upon by the log flusher thread.
+- */
+- logManager.getLogPageOwnershipCount(pageToFlush).incrementAndGet();
++ //#. set the logPageOwnerCount to 0 (LOG_FLUSHER)
++ // meaning it is flushing.
++ logManager.getLogPageOwnershipCount(pageToFlush).set(PageOwnershipStatus.LOG_FLUSHER);
- private void handleFunctionDropStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws MetadataException, RemoteException, ACIDException,
- AlgebricksException {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- FunctionDropStatement stmtDropFunction = (FunctionDropStatement) stmt;
-- FunctionSignature signature = stmtDropFunction.getFunctionSignature();
-- Function function = MetadataManager.INSTANCE.getFunction(mdTxnCtx, signature);
-- if (function == null) {
-- if (!stmtDropFunction.getIfExists())
-- throw new AlgebricksException("Unknonw function " + signature);
-- } else {
-- MetadataManager.INSTANCE.dropFunction(mdTxnCtx, signature);
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ FunctionDropStatement stmtDropFunction = (FunctionDropStatement) stmt;
-+ FunctionSignature signature = stmtDropFunction.getFunctionSignature();
-+ Function function = MetadataManager.INSTANCE.getFunction(mdTxnCtx, signature);
-+ if (function == null) {
-+ if (!stmtDropFunction.getIfExists())
-+ throw new AlgebricksException("Unknonw function " + signature);
-+ } else {
-+ MetadataManager.INSTANCE.dropFunction(mdTxnCtx, signature);
-+ }
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
- }
- }
+- /*
+- * get the number of log buffers that have been written so
+- * far. A log buffer = number of log pages * size of a log
+- * page
+- */
+- int numCycles = (int) lastFlushedLsn / logManager.getLogManagerProperties().getLogBufferSize();
+- if (lastFlushedLsn % logManager.getLogManagerProperties().getLogBufferSize() == 0) {
+- numCycles--;
+- }
++ // put the content to disk (the thread still has a lock on the log page)
++ logManager.getLogPage(pageToFlush).flush();
- private void handleLoadFromFileStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- LoadFromFileStatement loadStmt = (LoadFromFileStatement) stmt;
-- String dataverseName = loadStmt.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : loadStmt.getDataverseName().getValue();
-- CompiledLoadFromFileStatement cls = new CompiledLoadFromFileStatement(dataverseName, loadStmt.getDatasetName()
-- .getValue(), loadStmt.getAdapter(), loadStmt.getProperties(), loadStmt.dataIsAlreadySorted());
+- /*
+- * Map the log page to a new region in the log file.
+- */
++ // increment the last flushed lsn and lastFlushedPage
++ logManager.incrementLastFlushedLsn(logManager.getLogManagerProperties().getLogPageSize());
++ lastFlushedPageIndex = pageToFlush;
-- IDataFormat format = getDataFormat(metadataProvider.getMetadataTxnContext(), dataverseName);
-- Job job = DatasetOperations.createLoadDatasetJobSpec(metadataProvider, cls, format);
-- jobsToExecute.add(job.getJobSpec());
-- // Also load the dataset's secondary indexes.
-- List<Index> datasetIndexes = MetadataManager.INSTANCE.getDatasetIndexes(mdTxnCtx, dataverseName, loadStmt
-- .getDatasetName().getValue());
-- for (Index index : datasetIndexes) {
-- if (!index.isSecondaryIndex()) {
-- continue;
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
++ // decrement activeTxnCountOnIndexes
++ logManager.decrementActiveTxnCountOnIndexes(pageToFlush);
+
-+ try {
-+ LoadFromFileStatement loadStmt = (LoadFromFileStatement) stmt;
-+ String dataverseName = loadStmt.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : loadStmt.getDataverseName().getValue();
-+ CompiledLoadFromFileStatement cls = new CompiledLoadFromFileStatement(dataverseName, loadStmt
-+ .getDatasetName().getValue(), loadStmt.getAdapter(), loadStmt.getProperties(),
-+ loadStmt.dataIsAlreadySorted());
++ // reset the count to 1
++ logManager.getLogPageOwnershipCount(pageToFlush).set(PageOwnershipStatus.LOG_WRITER);
+
-+ IDataFormat format = getDataFormat(metadataProvider.getMetadataTxnContext(), dataverseName);
-+ Job job = DatasetOperations.createLoadDatasetJobSpec(metadataProvider, cls, format);
-+ jobsToExecute.add(job.getJobSpec());
-+ // Also load the dataset's secondary indexes.
-+ List<Index> datasetIndexes = MetadataManager.INSTANCE.getDatasetIndexes(mdTxnCtx, dataverseName, loadStmt
-+ .getDatasetName().getValue());
-+ for (Index index : datasetIndexes) {
-+ if (!index.isSecondaryIndex()) {
-+ continue;
-+ }
-+ // Create CompiledCreateIndexStatement from metadata entity 'index'.
-+ CompiledCreateIndexStatement cis = new CompiledCreateIndexStatement(index.getIndexName(),
-+ dataverseName, index.getDatasetName(), index.getKeyFieldNames(), index.getGramLength(),
-+ index.getIndexType());
-+ jobsToExecute.add(IndexOperations.buildSecondaryIndexLoadingJobSpec(cis, metadataProvider));
- }
-- // Create CompiledCreateIndexStatement from metadata entity 'index'.
-- CompiledCreateIndexStatement cis = new CompiledCreateIndexStatement(index.getIndexName(), dataverseName,
-- index.getDatasetName(), index.getKeyFieldNames(), index.getGramLength(), index.getIndexType());
-- jobsToExecute.add(IndexOperations.buildSecondaryIndexLoadingJobSpec(cis, metadataProvider));
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ for (JobSpecification jobspec : jobsToExecute) {
-+ runJob(hcc, jobspec);
-+ }
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
- }
- }
++ // Map the log page to a new region in the log file.
+ long nextWritePosition = logManager.getLogPages()[pageToFlush].getNextWritePosition()
+ + logManager.getLogManagerProperties().getLogBufferSize();
- private void handleWriteFromQueryResultStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- metadataProvider.setWriteTransaction(true);
-- WriteFromQueryResultStatement st1 = (WriteFromQueryResultStatement) stmt;
-- String dataverseName = st1.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : st1.getDataverseName().getValue();
-- CompiledWriteFromQueryResultStatement clfrqs = new CompiledWriteFromQueryResultStatement(dataverseName, st1
-- .getDatasetName().getValue(), st1.getQuery(), st1.getVarCounter());
+- /*
+- * long nextPos = (numCycles + 1)
+- * logManager.getLogManagerProperties() .getLogBufferSize()
+- * + pageToFlush logManager.getLogManagerProperties()
+- * .getLogPageSize();
+- */
+ logManager.resetLogPage(nextWritePosition, pageToFlush);
-- Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, clfrqs.getQuery(), clfrqs);
-- if (compiled.first != null) {
-- jobsToExecute.add(compiled.first);
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
-+
-+ try {
-+ metadataProvider.setWriteTransaction(true);
-+ WriteFromQueryResultStatement st1 = (WriteFromQueryResultStatement) stmt;
-+ String dataverseName = st1.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : st1.getDataverseName().getValue();
-+ CompiledWriteFromQueryResultStatement clfrqs = new CompiledWriteFromQueryResultStatement(dataverseName, st1
-+ .getDatasetName().getValue(), st1.getQuery(), st1.getVarCounter());
-+
-+ Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, clfrqs.getQuery(),
-+ clfrqs);
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+ if (compiled.first != null) {
-+ runJob(hcc, compiled.first);
-+ }
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
- }
- }
+ // mark the page as ACTIVE
+ logManager.getLogPageStatus(pageToFlush).set(LogManager.PageState.ACTIVE);
- private void handleInsertStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- metadataProvider.setWriteTransaction(true);
-- InsertStatement stmtInsert = (InsertStatement) stmt;
-- String dataverseName = stmtInsert.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : stmtInsert.getDataverseName().getValue();
-- CompiledInsertStatement clfrqs = new CompiledInsertStatement(dataverseName, stmtInsert.getDatasetName()
-- .getValue(), stmtInsert.getQuery(), stmtInsert.getVarCounter());
-- Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, clfrqs.getQuery(), clfrqs);
-- if (compiled.first != null) {
-- jobsToExecute.add(compiled.first);
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
-+
-+ try {
-+ metadataProvider.setWriteTransaction(true);
-+ InsertStatement stmtInsert = (InsertStatement) stmt;
-+ String dataverseName = stmtInsert.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : stmtInsert.getDataverseName().getValue();
-+ CompiledInsertStatement clfrqs = new CompiledInsertStatement(dataverseName, stmtInsert.getDatasetName()
-+ .getValue(), stmtInsert.getQuery(), stmtInsert.getVarCounter());
-+ Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, clfrqs.getQuery(),
-+ clfrqs);
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ if (compiled.first != null) {
-+ runJob(hcc, compiled.first);
-+ }
-+
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
- }
- }
+- // notify all waiting (transaction) threads.
+- // Transaction thread may be waiting for the page to be
+- // available or may have a commit log record on the page
+- // that got flushed.
+- logManager.getLogPages()[pageToFlush].notifyAll();
+- logManager.setLastFlushedPage(pageToFlush);
++ //#. checks the queue whether there is another flush request on the same log buffer
++ // If there is another request, then simply remove it.
++ if (flushRequestQueue[pageToFlush].peek() != null) {
++ flushRequestQueue[pageToFlush].take();
++ }
- private void handleDeleteStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- metadataProvider.setWriteTransaction(true);
-- DeleteStatement stmtDelete = (DeleteStatement) stmt;
-- String dataverseName = stmtDelete.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : stmtDelete.getDataverseName().getValue();
-- CompiledDeleteStatement clfrqs = new CompiledDeleteStatement(stmtDelete.getVariableExpr(), dataverseName,
-- stmtDelete.getDatasetName().getValue(), stmtDelete.getCondition(), stmtDelete.getDieClause(),
-- stmtDelete.getVarCounter(), metadataProvider);
-- Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, clfrqs.getQuery(), clfrqs);
-- if (compiled.first != null) {
-- jobsToExecute.add(compiled.first);
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
-+
-+ try {
-+ metadataProvider.setWriteTransaction(true);
-+ DeleteStatement stmtDelete = (DeleteStatement) stmt;
-+ String dataverseName = stmtDelete.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : stmtDelete.getDataverseName().getValue();
-+ CompiledDeleteStatement clfrqs = new CompiledDeleteStatement(stmtDelete.getVariableExpr(), dataverseName,
-+ stmtDelete.getDatasetName().getValue(), stmtDelete.getCondition(), stmtDelete.getDieClause(),
-+ stmtDelete.getVarCounter(), metadataProvider);
-+ Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, clfrqs.getQuery(),
-+ clfrqs);
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ if (compiled.first != null) {
-+ runJob(hcc, compiled.first);
-+ }
-+
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
++ // notify all waiting (transaction) threads.
++ logManager.getLogPage(pageToFlush).notifyAll();
+ }
+ } catch (IOException ioe) {
+ ioe.printStackTrace();
+ throw new Error(" exception in flushing log page", ioe);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+- break; // must break from the loop as the exception indicates
+- // some thing horrendous has happened elsewhere
++ break;
+ }
}
}
+-}
+-
+-/*
+- * TODO: By default the commit policy is to commit at each request and not have
+- * a group commit. The following code needs to change to support group commit.
+- * The code for group commit has not been tested thoroughly and is under
+- * development.
+- */
+-class BasicCommitResolver implements ICommitResolver {
+-
+- public boolean shouldCommitPage(int pageIndex, LogManager logManager,
+- CommitRequestStatistics commitRequestStatistics) {
+- return true;
+- }
+-
+- public void init(LogManager logManager) {
+- }
+-}
+-
+-class GroupCommitResolver implements ICommitResolver {
+-
+- public boolean shouldCommitPage(int pageIndex, LogManager logManager,
+- CommitRequestStatistics commitRequestStatistics) {
+- long maxCommitWait = logManager.getLogManagerProperties().getGroupCommitWaitPeriod();
+- long timestamp = commitRequestStatistics.getPageLevelLastCommitRequestTimestamp(pageIndex);
+- if (timestamp == -1) {
+- if (maxCommitWait == 0) {
+- return true;
+- } else {
+- timestamp = System.currentTimeMillis();
+- }
+- }
+- long currenTime = System.currentTimeMillis();
+- if (currenTime - timestamp > maxCommitWait) {
+- return true;
+- }
+- return false;
+- }
+-
+- public void init(LogManager logManager) {
+- GroupCommitHandlerThread groupCommitHandler = new GroupCommitHandlerThread(logManager);
+- groupCommitHandler.setDaemon(true);
+- groupCommitHandler.start();
+- }
+-
+- class GroupCommitHandlerThread extends Thread {
+-
+- private LogManager logManager;
+-
+- public GroupCommitHandlerThread(LogManager logManager) {
+- this.logManager = logManager;
+- setName("Group Commit Handler");
+- }
+-
+- @Override
+- public void run() {
+- int pageIndex = -1;
+- while (true) {
+- pageIndex = logManager.getNextPageInSequence(pageIndex);
+- long lastCommitRequeestTimestamp = logManager.getCommitRequestStatistics()
+- .getPageLevelLastCommitRequestTimestamp(pageIndex);
+- if (lastCommitRequeestTimestamp != -1
+- && System.currentTimeMillis() - lastCommitRequeestTimestamp > logManager
+- .getLogManagerProperties().getGroupCommitWaitPeriod()) {
+- int dirtyCount = logManager.getLogPageOwnershipCount(pageIndex).decrementAndGet();
+- if (dirtyCount == 0) {
+- try {
+- logManager.getLogPageStatus(pageIndex).set(LogManager.PageState.INACTIVE);
+- logManager.getPendingFlushRequests(pageIndex).put(Thread.currentThread());
+- } catch (InterruptedException e) {
+- e.printStackTrace();
+- break;
+- }
+- logManager.getCommitRequestStatistics().committedPage(pageIndex);
+- }
+- }
+- }
+- }
+- }
+-
+-}
+-
+-interface ICommitResolver {
+- public boolean shouldCommitPage(int pageIndex, LogManager logManager,
+- CommitRequestStatistics commitRequestStatistics);
+-
+- public void init(LogManager logManager);
+-}
+-
+-/**
+- * Represents a collection of all commit requests by transactions for each log
+- * page. The requests are accumulated until the commit policy triggers a flush
+- * of the corresponding log page. Upon a flush of a page, all commit requests
+- * for the page are cleared.
+- */
+-class CommitRequestStatistics {
+-
+- AtomicInteger[] pageLevelCommitRequestCount;
+- AtomicLong[] pageLevelLastCommitRequestTimestamp;
+-
+- public CommitRequestStatistics(int numPages) {
+- pageLevelCommitRequestCount = new AtomicInteger[numPages];
+- pageLevelLastCommitRequestTimestamp = new AtomicLong[numPages];
+- for (int i = 0; i < numPages; i++) {
+- pageLevelCommitRequestCount[i] = new AtomicInteger(0);
+- pageLevelLastCommitRequestTimestamp[i] = new AtomicLong(-1L);
+- }
+- }
+-
+- public void registerCommitRequest(int pageIndex) {
+- pageLevelCommitRequestCount[pageIndex].incrementAndGet();
+- pageLevelLastCommitRequestTimestamp[pageIndex].set(System.currentTimeMillis());
+- }
+-
+- public long getPageLevelLastCommitRequestTimestamp(int pageIndex) {
+- return pageLevelLastCommitRequestTimestamp[pageIndex].get();
+- }
+-
+- public void committedPage(int pageIndex) {
+- pageLevelCommitRequestCount[pageIndex].set(0);
+- pageLevelLastCommitRequestTimestamp[pageIndex].set(-1L);
+- }
+-
+-}
++}
+\ No newline at end of file
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogRecordHelper.java
+===================================================================
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogRecordHelper.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogRecordHelper.java (working copy)
+@@ -152,6 +152,9 @@
+ case LogType.UPDATE:
+ logTypeDisplay = "UPDATE";
+ break;
++ case LogType.ENTITY_COMMIT:
++ logTypeDisplay = "ENTITY_COMMIT";
++ break;
+ }
+ builder.append(" LSN : ").append(logicalLogLocator.getLsn());
+ builder.append(" Log Type : ").append(logTypeDisplay);
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogType.java
+===================================================================
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogType.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogType.java (working copy)
+@@ -18,5 +18,6 @@
-@@ -704,46 +1106,109 @@
+ public static final byte UPDATE = 0;
+ public static final byte COMMIT = 1;
++ public static final byte ENTITY_COMMIT = 2;
- private void handleBeginFeedStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- BeginFeedStatement bfs = (BeginFeedStatement) stmt;
-- String dataverseName = bfs.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : bfs.getDatasetName().getValue();
+ }
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManagerProperties.java
+===================================================================
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManagerProperties.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/LogManagerProperties.java (working copy)
+@@ -1,5 +1,5 @@
+ /*
+- * Copyright 2009-2010 by The Regents of the University of California
++ * Copyright 2009-2012 by The Regents of the University of California
+ * Licensed 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 from
+@@ -41,7 +41,7 @@
+ private int logPageSize = 128 * 1024; // 128 KB
+ private int numLogPages = 8; // number of log pages in the log buffer.
-- CompiledBeginFeedStatement cbfs = new CompiledBeginFeedStatement(dataverseName,
-- bfs.getDatasetName().getValue(), bfs.getQuery(), bfs.getVarCounter());
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
+- private long groupCommitWaitPeriod = 0; // time in milliseconds for which a
++ private long groupCommitWaitPeriod = 1; // time in milliseconds for which a
+ // commit record will wait before
+ // the housing page is marked for
+ // flushing.
+Index: asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/recovery/RecoveryManager.java
+===================================================================
+--- asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/recovery/RecoveryManager.java (revision 1194)
++++ asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/recovery/RecoveryManager.java (working copy)
+@@ -184,6 +184,7 @@
+ break;
-- Dataset dataset;
-- dataset = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(), dataverseName, bfs
-- .getDatasetName().getValue());
-- IDatasetDetails datasetDetails = dataset.getDatasetDetails();
-- if (datasetDetails.getDatasetType() != DatasetType.FEED) {
-- throw new IllegalArgumentException("Dataset " + bfs.getDatasetName().getValue() + " is not a feed dataset");
-+ try {
-+ BeginFeedStatement bfs = (BeginFeedStatement) stmt;
-+ String dataverseName = bfs.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : bfs.getDatasetName().getValue();
-+
-+ CompiledBeginFeedStatement cbfs = new CompiledBeginFeedStatement(dataverseName, bfs.getDatasetName()
-+ .getValue(), bfs.getQuery(), bfs.getVarCounter());
-+
-+ Dataset dataset;
-+ dataset = MetadataManager.INSTANCE.getDataset(metadataProvider.getMetadataTxnContext(), dataverseName, bfs
-+ .getDatasetName().getValue());
-+ IDatasetDetails datasetDetails = dataset.getDatasetDetails();
-+ if (datasetDetails.getDatasetType() != DatasetType.FEED) {
-+ throw new IllegalArgumentException("Dataset " + bfs.getDatasetName().getValue()
-+ + " is not a feed dataset");
-+ }
-+ bfs.initialize(metadataProvider.getMetadataTxnContext(), dataset);
-+ cbfs.setQuery(bfs.getQuery());
-+ Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, bfs.getQuery(), cbfs);
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ if (compiled.first != null) {
-+ runJob(hcc, compiled.first);
-+ }
-+
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
- }
-- bfs.initialize(metadataProvider.getMetadataTxnContext(), dataset);
-- cbfs.setQuery(bfs.getQuery());
-- Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, bfs.getQuery(), cbfs);
-- if (compiled.first != null) {
-- jobsToExecute.add(compiled.first);
-- }
- }
+ case LogType.COMMIT:
++ case LogType.ENTITY_COMMIT:
+ tempKeyTxnId.setTxnId(logRecordHelper.getJobId(currentLogLocator),
+ logRecordHelper.getDatasetId(currentLogLocator),
+ logRecordHelper.getPKHashValue(currentLogLocator));
+@@ -218,6 +219,7 @@
+ IIndex index = null;
+ LocalResource localResource = null;
+ ILocalResourceMetadata localResourceMetadata = null;
++ List<Long> resourceIdList = new ArrayList<Long>();
- private void handleControlFeedStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- IHyracksClientConnection hcc, List<JobSpecification> jobsToExecute) throws Exception {
-- ControlFeedStatement cfs = (ControlFeedStatement) stmt;
-- String dataverseName = cfs.getDataverseName() == null ? activeDefaultDataverse == null ? null
-- : activeDefaultDataverse.getDataverseName() : cfs.getDatasetName().getValue();
-- CompiledControlFeedStatement clcfs = new CompiledControlFeedStatement(cfs.getOperationType(), dataverseName,
-- cfs.getDatasetName().getValue(), cfs.getAlterAdapterConfParams());
-- jobsToExecute.add(FeedOperations.buildControlFeedJobSpec(clcfs, metadataProvider));
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
-+
-+ try {
-+ ControlFeedStatement cfs = (ControlFeedStatement) stmt;
-+ String dataverseName = cfs.getDataverseName() == null ? activeDefaultDataverse == null ? null
-+ : activeDefaultDataverse.getDataverseName() : cfs.getDatasetName().getValue();
-+ CompiledControlFeedStatement clcfs = new CompiledControlFeedStatement(cfs.getOperationType(),
-+ dataverseName, cfs.getDatasetName().getValue(), cfs.getAlterAdapterConfParams());
-+ JobSpecification jobSpec = FeedOperations.buildControlFeedJobSpec(clcfs, metadataProvider);
-+
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ runJob(hcc, jobSpec);
-+
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
-+ }
- }
+ //#. get indexLifeCycleManager
+ IAsterixAppRuntimeContextProvider appRuntimeContext = txnSubsystem.getAsterixAppRuntimeContextProvider();
+@@ -272,6 +274,8 @@
+ index = localResourceMetadata.createIndexInstance(appRuntimeContext,
+ localResource.getResourceName(), localResource.getPartition());
+ indexLifecycleManager.register(resourceId, index);
++ indexLifecycleManager.open(resourceId);
++ resourceIdList.add(resourceId);
+ }
- private QueryResult handleQuery(AqlMetadataProvider metadataProvider, Query query, IHyracksClientConnection hcc,
- List<JobSpecification> jobsToExecute) throws Exception {
-- Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, query, null);
-- if (compiled.first != null) {
-- GlobalConfig.ASTERIX_LOGGER.info(compiled.first.toJSON().toString(1));
-- jobsToExecute.add(compiled.first);
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ boolean bActiveTxn = true;
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireReadLatch();
-+
-+ try {
-+ Pair<JobSpecification, FileSplit> compiled = rewriteCompileQuery(metadataProvider, query, null);
-+
-+ QueryResult queryResult = new QueryResult(query, compiled.second.getLocalFile().getFile().getAbsolutePath());
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ bActiveTxn = false;
-+
-+ if (compiled.first != null) {
-+ GlobalConfig.ASTERIX_LOGGER.info(compiled.first.toJSON().toString(1));
-+ runJob(hcc, compiled.first);
-+ }
-+
-+ return queryResult;
-+ } catch (Exception e) {
-+ if (bActiveTxn) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ }
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseReadLatch();
- }
-- return new QueryResult(query, compiled.second.getLocalFile().getFile().getAbsolutePath());
- }
+ /***************************************************/
+@@ -300,6 +304,7 @@
+ break;
- private void runCreateIndexJob(IHyracksClientConnection hcc, CreateIndexStatement stmtCreateIndex,
-@@ -768,20 +1233,32 @@
- private void handleCreateNodeGroupStatement(AqlMetadataProvider metadataProvider, Statement stmt,
- List<JobSpecification> jobsToExecute) throws MetadataException, AlgebricksException, RemoteException,
- ACIDException {
-- MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-- NodegroupDecl stmtCreateNodegroup = (NodegroupDecl) stmt;
-- String ngName = stmtCreateNodegroup.getNodegroupName().getValue();
-- NodeGroup ng = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, ngName);
-- if (ng != null) {
-- if (!stmtCreateNodegroup.getIfNotExists())
-- throw new AlgebricksException("A nodegroup with this name " + ngName + " already exists.");
-- } else {
-- List<Identifier> ncIdentifiers = stmtCreateNodegroup.getNodeControllerNames();
-- List<String> ncNames = new ArrayList<String>(ncIdentifiers.size());
-- for (Identifier id : ncIdentifiers) {
-- ncNames.add(id.getValue());
-+
-+ MetadataTransactionContext mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ acquireWriteLatch();
-+
-+ try {
-+ NodegroupDecl stmtCreateNodegroup = (NodegroupDecl) stmt;
-+ String ngName = stmtCreateNodegroup.getNodegroupName().getValue();
-+ NodeGroup ng = MetadataManager.INSTANCE.getNodegroup(mdTxnCtx, ngName);
-+ if (ng != null) {
-+ if (!stmtCreateNodegroup.getIfNotExists())
-+ throw new AlgebricksException("A nodegroup with this name " + ngName + " already exists.");
-+ } else {
-+ List<Identifier> ncIdentifiers = stmtCreateNodegroup.getNodeControllerNames();
-+ List<String> ncNames = new ArrayList<String>(ncIdentifiers.size());
-+ for (Identifier id : ncIdentifiers) {
-+ ncNames.add(id.getValue());
-+ }
-+ MetadataManager.INSTANCE.addNodegroup(mdTxnCtx, new NodeGroup(ngName, ncNames));
+ case LogType.COMMIT:
++ case LogType.ENTITY_COMMIT:
+ //do nothing
+ break;
+
+@@ -308,6 +313,11 @@
}
-- MetadataManager.INSTANCE.addNodegroup(mdTxnCtx, new NodeGroup(ngName, ncNames));
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+ } catch (Exception e) {
-+ MetadataManager.INSTANCE.abortTransaction(mdTxnCtx);
-+ throw new AlgebricksException(e);
-+ } finally {
-+ releaseWriteLatch();
}
- }
-
-@@ -791,10 +1268,37 @@
-
- private void compileIndexDropStatement(IHyracksClientConnection hcc, String dataverseName, String datasetName,
- String indexName, AqlMetadataProvider metadataProvider) throws Exception {
-+ MetadataTransactionContext mdTxnCtx = metadataProvider.getMetadataTxnContext();
-+ Index index = MetadataManager.INSTANCE.getIndex(mdTxnCtx, dataverseName, datasetName, indexName);
-+
-+ //#. mark PendingDropOp on the existing index
-+ MetadataManager.INSTANCE.dropIndex(mdTxnCtx, dataverseName, datasetName, indexName);
-+ MetadataManager.INSTANCE.addIndex(
-+ mdTxnCtx,
-+ new Index(dataverseName, datasetName, indexName, index.getIndexType(), index.getKeyFieldNames(), index
-+ .isPrimaryIndex(), IMetadataEntity.PENDING_DROP_OP));
- CompiledIndexDropStatement cds = new CompiledIndexDropStatement(dataverseName, datasetName, indexName);
-- runJob(hcc, IndexOperations.buildDropSecondaryIndexJobSpec(cds, metadataProvider));
-- MetadataManager.INSTANCE.dropIndex(metadataProvider.getMetadataTxnContext(), dataverseName, datasetName,
-- indexName);
-+ JobSpecification jobSpec = IndexOperations.buildDropSecondaryIndexJobSpec(cds, metadataProvider);
-+
-+ //#. commit the existing transaction before calling runJob.
-+ // the caller should begin the transaction before calling this function.
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+
-+ try {
-+ runJob(hcc, jobSpec);
-+ } catch (Exception e) {
-+ //need to create the mdTxnCtx to be aborted by caller properly
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ throw e;
+
++ //close all indexes
++ for (long r : resourceIdList) {
++ indexLifecycleManager.close(r);
+ }
-+
-+ //#. begin a new transaction
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+
-+ //#. finally, delete the existing index
-+ MetadataManager.INSTANCE.dropIndex(mdTxnCtx, dataverseName, datasetName, indexName);
++
+ JobIdFactory.initJobId(maxJobId);
}
- private void compileDatasetDropStatement(IHyracksClientConnection hcc, String dataverseName, String datasetName,
-@@ -803,10 +1307,32 @@
- CompiledDatasetDropStatement cds = new CompiledDatasetDropStatement(dataverseName, datasetName);
- Dataset ds = MetadataManager.INSTANCE.getDataset(mdTxnCtx, dataverseName, datasetName);
- if (ds.getDatasetType() == DatasetType.INTERNAL || ds.getDatasetType() == DatasetType.FEED) {
-- JobSpecification[] jobSpecs = DatasetOperations.createDropDatasetJobSpec(cds, metadataProvider);
-- for (JobSpecification spec : jobSpecs)
-- runJob(hcc, spec);
-+ JobSpecification jobSpec = DatasetOperations.createDropDatasetJobSpec(cds, metadataProvider);
-+
-+ //#. mark PendingDropOp on the existing dataset
-+ MetadataManager.INSTANCE.dropDataset(mdTxnCtx, dataverseName, datasetName);
-+ MetadataManager.INSTANCE.addDataset(mdTxnCtx, new Dataset(dataverseName, datasetName, ds.getItemTypeName(),
-+ ds.getDatasetDetails(), ds.getDatasetType(), ds.getDatasetId(), IMetadataEntity.PENDING_DROP_OP));
-+
-+ //#. commit the transaction
-+ MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
-+
-+ //#. run the job
-+ try {
-+ runJob(hcc, jobSpec);
-+ } catch (Exception e) {
-+ //need to create the mdTxnCtx to be aborted by caller properly
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
-+ throw e;
+@@ -539,6 +549,7 @@
+ break;
+
+ case LogType.COMMIT:
++ case LogType.ENTITY_COMMIT:
+ undoLSNSet = loserTxnTable.get(tempKeyTxnId);
+ if (undoLSNSet != null) {
+ loserTxnTable.remove(tempKeyTxnId);
+Index: asterix-app/src/test/java/edu/uci/ics/asterix/test/metadata/MetadataTest.java
+===================================================================
+--- asterix-app/src/test/java/edu/uci/ics/asterix/test/metadata/MetadataTest.java (revision 1194)
++++ asterix-app/src/test/java/edu/uci/ics/asterix/test/metadata/MetadataTest.java (working copy)
+@@ -42,6 +42,16 @@
+ List<CompilationUnit> cUnits = tcCtx.getTestCase().getCompilationUnit();
+ for (CompilationUnit cUnit : cUnits) {
+ File testFile = tcCtx.getTestFile(cUnit);
++
++ /*****************
++ if (!testFile.getAbsolutePath().contains("meta09.aql")) {
++ System.out.println(testFile.getAbsolutePath());
++ continue;
+ }
-+
-+ //#. start a new transaction
-+ mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();
-+ metadataProvider.setMetadataTxnContext(mdTxnCtx);
- }
-+
-+ //#. finally, delete the existing dataset.
- MetadataManager.INSTANCE.dropDataset(mdTxnCtx, dataverseName, datasetName);
- }
++ System.out.println(testFile.getAbsolutePath());
++ *****************/
++
++
+ File expectedResultFile = tcCtx.getExpectedResultFile(cUnit);
+ File actualFile = new File(PATH_ACTUAL + File.separator
+ + tcCtx.getTestCase().getFilePath().replace(File.separator, "_") + "_" + cUnit.getName() + ".adm");
+Index: asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java
+===================================================================
+--- asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java (revision 1194)
++++ asterix-app/src/test/java/edu/uci/ics/asterix/test/runtime/ExecutionTest.java (working copy)
+@@ -95,9 +95,10 @@
+ File testFile = tcCtx.getTestFile(cUnit);
-@@ -831,4 +1357,20 @@
- }
- return format;
- }
-+
-+ private void acquireWriteLatch() {
-+ cacheLatch.writeLock().lock();
-+ }
-+
-+ private void releaseWriteLatch() {
-+ cacheLatch.writeLock().unlock();
-+ }
-+
-+ private void acquireReadLatch() {
-+ cacheLatch.readLock().lock();
-+ }
-+
-+ private void releaseReadLatch() {
-+ cacheLatch.readLock().unlock();
-+ }
- }
+ /*************** to avoid run failure cases ****************
+- if (!testFile.getAbsolutePath().contains("index-selection/")) {
++ if (!testFile.getAbsolutePath().contains("query-issue205.aql")) {
+ continue;
+ }
++ System.out.println(testFile.getAbsolutePath());
+ ************************************************************/
+
+ File expectedResultFile = tcCtx.getExpectedResultFile(cUnit);