You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by im...@apache.org on 2018/03/09 21:23:33 UTC
[2/2] asterixdb git commit: [ASTERIXDB-1952][TX][IDX] Filter logs pt.2
[ASTERIXDB-1952][TX][IDX] Filter logs pt.2
- user model changes: no
- storage format changes: yes
- interface changes: yes
Details:
- Add a log type specifically for filters
- Only log change when filter actually widens
- Stop logging of index + filter tuple during modification
- Redo index and filter tuples separately via their logs
Change-Id: Ie9e7795d9c8c212e8610dcb9bb5d26ec9fbbee8a
Reviewed-on: https://asterix-gerrit.ics.uci.edu/1857
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Ian Maxon <im...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/693844cc
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/693844cc
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/693844cc
Branch: refs/heads/master
Commit: 693844cc5cccaf78db6b1260f1656292558c8a70
Parents: 5cc4e7a
Author: Ian Maxon <im...@apache.org>
Authored: Thu Mar 8 17:45:42 2018 -0800
Committer: Ian Maxon <im...@apache.org>
Committed: Fri Mar 9 13:22:43 2018 -0800
----------------------------------------------------------------------
.../apache/asterix/app/nc/RecoveryManager.java | 121 +++++++++-------
.../asterix/common/transactions/ILogRecord.java | 104 +++++++-------
.../asterix/common/transactions/LogRecord.java | 143 ++++++++++++-------
.../asterix/common/transactions/LogType.java | 6 +-
.../TestLSMIndexOperationContext.java | 26 +++-
.../logging/RemoteLogsProcessor.java | 1 +
...tractIndexModificationOperationCallback.java | 75 ++++++----
...imaryIndexModificationOperationCallback.java | 8 +-
.../logging/LogManagerWithReplication.java | 1 +
.../hyracks/storage/am/btree/impls/BTree.java | 45 ++----
.../storage/am/btree/impls/BTreeOpContext.java | 19 ---
.../storage/am/btree/impls/DiskBTree.java | 11 --
.../IExtendedModificationOperationCallback.java | 37 +++++
.../impls/ExtendedIndexAccessParameters.java | 41 ++++++
.../common/impls/NoOpIndexAccessParameters.java | 3 +-
.../am/common/impls/NoOpOperationCallback.java | 9 +-
.../am/common/ophelpers/IndexOperation.java | 1 +
.../impls/ExternalBTreeWithBuddyOpContext.java | 3 +-
.../storage/am/lsm/btree/impls/LSMBTree.java | 14 +-
.../am/lsm/btree/impls/LSMBTreeOpContext.java | 15 +-
.../am/lsm/common/api/ILSMComponentFilter.java | 6 +-
.../common/api/ILSMComponentFilterManager.java | 4 +-
.../am/lsm/common/api/ILSMIndexAccessor.java | 7 +
.../common/api/ILSMIndexOperationContext.java | 11 +-
.../am/lsm/common/impls/AbstractLSMIndex.java | 13 +-
.../impls/AbstractLSMIndexOperationContext.java | 37 +++--
.../impls/ComponentReplacementContext.java | 23 ++-
.../am/lsm/common/impls/FilterBulkLoader.java | 3 +-
.../am/lsm/common/impls/LSMComponentFilter.java | 15 +-
.../common/impls/LSMComponentFilterManager.java | 10 +-
.../lsm/common/impls/LSMTreeIndexAccessor.java | 1 +
.../invertedindex/impls/LSMInvertedIndex.java | 11 +-
.../impls/LSMInvertedIndexAccessor.java | 5 +
.../impls/LSMInvertedIndexOpContext.java | 11 +-
.../inmemory/InMemoryInvertedIndex.java | 7 -
.../inmemory/InMemoryInvertedIndexAccessor.java | 14 --
.../PartitionedInMemoryInvertedIndex.java | 8 --
...artitionedInMemoryInvertedIndexAccessor.java | 6 -
.../am/lsm/rtree/impls/AbstractLSMRTree.java | 9 +-
.../storage/am/lsm/rtree/impls/LSMRTree.java | 8 +-
.../am/lsm/rtree/impls/LSMRTreeOpContext.java | 13 +-
.../impls/LSMRTreeWithAntiMatterTuples.java | 7 +-
.../hyracks/storage/am/rtree/impls/RTree.java | 32 +----
.../storage/am/rtree/impls/RTreeOpContext.java | 17 ---
...stractModificationOperationCallbackTest.java | 8 +-
.../am/common/TestOperationCallback.java | 8 +-
.../am/lsm/btree/LSMBTreeUpdateInPlaceTest.java | 8 +-
47 files changed, 576 insertions(+), 409 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
index 74277ce..4a2cf2d 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
@@ -77,6 +77,7 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentIdGenerator;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
import org.apache.hyracks.storage.common.IIndex;
@@ -221,6 +222,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
case LogType.FLUSH:
case LogType.WAIT:
case LogType.MARKER:
+ case LogType.FILTER:
break;
default:
throw new ACIDException("Unsupported LogType: " + logRecord.getLogType());
@@ -315,59 +317,64 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
foundWinner = true;
}
}
- if (foundWinner) {
- resourceId = logRecord.getResourceId();
- localResource = resourcesMap.get(resourceId);
- /*******************************************************************
- * [Notice]
- * -> Issue
- * Delete index may cause a problem during redo.
- * The index operation to be redone couldn't be redone because the corresponding index
- * may not exist in NC due to the possible index drop DDL operation.
- * -> Approach
- * Avoid the problem during redo.
- * More specifically, the problem will be detected when the localResource of
- * the corresponding index is retrieved, which will end up with 'null'.
- * If null is returned, then just go and process the next
- * log record.
- *******************************************************************/
- if (localResource == null) {
- LOGGER.log(Level.WARN, "resource was not found for resource id " + resourceId);
- logRecord = logReader.next();
- continue;
- }
- /*******************************************************************/
-
- //get index instance from IndexLifeCycleManager
- //if index is not registered into IndexLifeCycleManager,
- //create the index using LocalMetadata stored in LocalResourceRepository
- //get partition path in this node
- localResourceMetadata = (DatasetLocalResource) localResource.getResource();
- index = (ILSMIndex) datasetLifecycleManager.get(localResource.getPath());
- if (index == null) {
- //#. create index instance and register to indexLifeCycleManager
- index = (ILSMIndex) localResourceMetadata.createInstance(serviceCtx);
- datasetLifecycleManager.register(localResource.getPath(), index);
- datasetLifecycleManager.open(localResource.getPath());
- try {
- final DatasetResourceReference resourceReference =
- DatasetResourceReference.of(localResource);
- maxDiskLastLsn =
- indexCheckpointManagerProvider.get(resourceReference).getLowWatermark();
- } catch (HyracksDataException e) {
- datasetLifecycleManager.close(localResource.getPath());
- throw e;
- }
- //#. set resourceId and maxDiskLastLSN to the map
- resourceId2MaxLSNMap.put(resourceId, maxDiskLastLsn);
- } else {
- maxDiskLastLsn = resourceId2MaxLSNMap.get(resourceId);
- }
- // lsn @ maxDiskLastLsn is either a flush log or a master replica log
- if (lsn >= maxDiskLastLsn) {
- redo(logRecord, datasetLifecycleManager);
- redoCount++;
+ if (!foundWinner) {
+ break;
+ }
+ }
+ //fall through as FILTER is a subset of UPDATE
+ case LogType.FILTER:
+ if (partitions.contains(logRecord.getResourcePartition())) {
+ resourceId = logRecord.getResourceId();
+ localResource = resourcesMap.get(resourceId);
+ /*******************************************************************
+ * [Notice]
+ * -> Issue
+ * Delete index may cause a problem during redo.
+ * The index operation to be redone couldn't be redone because the corresponding index
+ * may not exist in NC due to the possible index drop DDL operation.
+ * -> Approach
+ * Avoid the problem during redo.
+ * More specifically, the problem will be detected when the localResource of
+ * the corresponding index is retrieved, which will end up with 'null'.
+ * If null is returned, then just go and process the next
+ * log record.
+ *******************************************************************/
+ if (localResource == null) {
+ LOGGER.log(Level.WARN, "resource was not found for resource id " + resourceId);
+ logRecord = logReader.next();
+ continue;
+ }
+ /*******************************************************************/
+
+ //get index instance from IndexLifeCycleManager
+ //if index is not registered into IndexLifeCycleManager,
+ //create the index using LocalMetadata stored in LocalResourceRepository
+ //get partition path in this node
+ localResourceMetadata = (DatasetLocalResource) localResource.getResource();
+ index = (ILSMIndex) datasetLifecycleManager.get(localResource.getPath());
+ if (index == null) {
+ //#. create index instance and register to indexLifeCycleManager
+ index = (ILSMIndex) localResourceMetadata.createInstance(serviceCtx);
+ datasetLifecycleManager.register(localResource.getPath(), index);
+ datasetLifecycleManager.open(localResource.getPath());
+ try {
+ final DatasetResourceReference resourceReference =
+ DatasetResourceReference.of(localResource);
+ maxDiskLastLsn =
+ indexCheckpointManagerProvider.get(resourceReference).getLowWatermark();
+ } catch (HyracksDataException e) {
+ datasetLifecycleManager.close(localResource.getPath());
+ throw e;
}
+ //#. set resourceId and maxDiskLastLSN to the map
+ resourceId2MaxLSNMap.put(resourceId, maxDiskLastLsn);
+ } else {
+ maxDiskLastLsn = resourceId2MaxLSNMap.get(resourceId);
+ }
+ // lsn @ maxDiskLastLsn is either a flush log or a master replica log
+ if (lsn >= maxDiskLastLsn) {
+ redo(logRecord, datasetLifecycleManager);
+ redoCount++;
}
}
break;
@@ -659,6 +666,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
throw new ACIDException("Unexpected LogType(" + logRecord.getLogType() + ") during abort.");
case LogType.ABORT:
case LogType.FLUSH:
+ case LogType.FILTER:
case LogType.WAIT:
case LogType.MARKER:
//ignore
@@ -741,6 +749,9 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
// undo, upsert the old value if found, otherwise, physical delete
undoUpsertOrDelete(indexAccessor, logRecord);
break;
+ case AbstractIndexModificationOperationCallback.FILTER_BYTE:
+ //do nothing, can't undo filters
+ break;
default:
throw new IllegalStateException("Unsupported OperationType: " + logRecord.getNewOp());
}
@@ -775,6 +786,9 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
long resourceId = logRecord.getResourceId();
ILSMIndex index = (ILSMIndex) datasetLifecycleManager.getIndex(datasetId, resourceId);
ILSMIndexAccessor indexAccessor = index.createAccessor(NoOpIndexAccessParameters.INSTANCE);
+ ILSMIndexOperationContext opCtx = indexAccessor.getOpContext();
+ opCtx.setFilterSkip(true);
+ opCtx.setRecovery(true);
if (logRecord.getNewOp() == AbstractIndexModificationOperationCallback.INSERT_BYTE) {
indexAccessor.forceInsert(logRecord.getNewValue());
} else if (logRecord.getNewOp() == AbstractIndexModificationOperationCallback.DELETE_BYTE) {
@@ -782,6 +796,9 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
} else if (logRecord.getNewOp() == AbstractIndexModificationOperationCallback.UPSERT_BYTE) {
// redo, upsert the new value
indexAccessor.forceUpsert(logRecord.getNewValue());
+ } else if (logRecord.getNewOp() == AbstractIndexModificationOperationCallback.FILTER_BYTE) {
+ opCtx.setFilterSkip(false);
+ indexAccessor.updateFilter(logRecord.getNewValue());
} else {
throw new IllegalStateException("Unsupported OperationType: " + logRecord.getNewOp());
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
index e58a6fa..04f9751 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/ILogRecord.java
@@ -50,121 +50,125 @@ public interface ILogRecord {
int FLUSHING_COMPONENT_MAXID_LEN = Long.BYTES;
int ALL_RECORD_HEADER_LEN = LOG_SOURCE_LEN + TYPE_LEN + TxnId.BYTES;
- int ENTITYCOMMIT_UPDATE_HEADER_LEN = RS_PARTITION_LEN + DatasetId.BYTES + PKHASH_LEN + PKSZ_LEN;
+ int ENTITY_RESOURCE_HEADER_LEN = RS_PARTITION_LEN + DatasetId.BYTES;
+ int ENTITY_VALUE_HEADER_LEN = PKHASH_LEN + PKSZ_LEN;
int UPDATE_LSN_HEADER = RSID_LEN + LOGRCD_SZ_LEN;
int UPDATE_BODY_HEADER = FLDCNT_LEN + NEWOP_LEN + NEWVALSZ_LEN;
int JOB_TERMINATE_LOG_SIZE = ALL_RECORD_HEADER_LEN + CHKSUM_LEN;
- int ENTITY_COMMIT_LOG_BASE_SIZE = ALL_RECORD_HEADER_LEN + ENTITYCOMMIT_UPDATE_HEADER_LEN + CHKSUM_LEN;
+ int ENTITY_COMMIT_LOG_BASE_SIZE =
+ ALL_RECORD_HEADER_LEN + ENTITY_RESOURCE_HEADER_LEN + ENTITY_VALUE_HEADER_LEN + CHKSUM_LEN;
int UPDATE_LOG_BASE_SIZE = ENTITY_COMMIT_LOG_BASE_SIZE + UPDATE_LSN_HEADER + UPDATE_BODY_HEADER;
+ int FILTER_LOG_BASE_SIZE =
+ ALL_RECORD_HEADER_LEN + ENTITY_RESOURCE_HEADER_LEN + UPDATE_BODY_HEADER + UPDATE_LSN_HEADER + CHKSUM_LEN;
int FLUSH_LOG_SIZE = ALL_RECORD_HEADER_LEN + DS_LEN + RS_PARTITION_LEN + FLUSHING_COMPONENT_MINID_LEN
+ FLUSHING_COMPONENT_MAXID_LEN + CHKSUM_LEN;
int WAIT_LOG_SIZE = ALL_RECORD_HEADER_LEN + CHKSUM_LEN;
int MARKER_BASE_LOG_SIZE =
ALL_RECORD_HEADER_LEN + CHKSUM_LEN + DS_LEN + RS_PARTITION_LEN + PRVLSN_LEN + LOGRCD_SZ_LEN;
- public RecordReadStatus readLogRecord(ByteBuffer buffer);
+ RecordReadStatus readLogRecord(ByteBuffer buffer);
- public void writeLogRecord(ByteBuffer buffer);
+ void writeLogRecord(ByteBuffer buffer);
- public ITransactionContext getTxnCtx();
+ ITransactionContext getTxnCtx();
- public void setTxnCtx(ITransactionContext txnCtx);
+ void setTxnCtx(ITransactionContext txnCtx);
- public boolean isFlushed();
+ boolean isFlushed();
- public void isFlushed(boolean isFlushed);
+ void isFlushed(boolean isFlushed);
- public byte getLogType();
+ byte getLogType();
- public void setLogType(byte logType);
+ void setLogType(byte logType);
long getTxnId();
void setTxnId(long jobId);
- public int getDatasetId();
+ int getDatasetId();
- public void setDatasetId(int datasetId);
+ void setDatasetId(int datasetId);
- public int getPKHashValue();
+ int getPKHashValue();
- public void setPKHashValue(int PKHashValue);
+ void setPKHashValue(int PKHashValue);
- public long getResourceId();
+ long getResourceId();
- public void setResourceId(long resourceId);
+ void setResourceId(long resourceId);
- public int getLogSize();
+ int getLogSize();
- public void setLogSize(int logSize);
+ void setLogSize(int logSize);
- public byte getNewOp();
+ byte getNewOp();
- public void setNewOp(byte newOp);
+ void setNewOp(byte newOp);
- public void setNewValueSize(int newValueSize);
+ void setNewValueSize(int newValueSize);
- public ITupleReference getNewValue();
+ ITupleReference getNewValue();
- public void setNewValue(ITupleReference newValue);
+ void setNewValue(ITupleReference newValue);
- public long getChecksum();
+ long getChecksum();
- public void setChecksum(long checksum);
+ void setChecksum(long checksum);
- public long getLSN();
+ long getLSN();
- public void setLSN(long LSN);
+ void setLSN(long LSN);
- public String getLogRecordForDisplay();
+ String getLogRecordForDisplay();
- public void computeAndSetLogSize();
+ void computeAndSetLogSize();
- public int getPKValueSize();
+ int getPKValueSize();
- public ITupleReference getPKValue();
+ ITupleReference getPKValue();
- public void setPKFields(int[] primaryKeyFields);
+ void setPKFields(int[] primaryKeyFields);
- public void computeAndSetPKValueSize();
+ void computeAndSetPKValueSize();
- public void setPKValue(ITupleReference PKValue);
+ void setPKValue(ITupleReference PKValue);
- public void readRemoteLog(ByteBuffer buffer);
+ void readRemoteLog(ByteBuffer buffer);
- public void setLogSource(byte logSource);
+ void setLogSource(byte logSource);
- public byte getLogSource();
+ byte getLogSource();
- public int getRemoteLogSize();
+ int getRemoteLogSize();
- public int getResourcePartition();
+ int getResourcePartition();
- public void setResourcePartition(int resourcePartition);
+ void setResourcePartition(int resourcePartition);
- public void setReplicated(boolean replicated);
+ void setReplicated(boolean replicated);
/**
* @return a flag indicating whether the log was replicated
*/
- public boolean isReplicated();
+ boolean isReplicated();
- public void writeRemoteLogRecord(ByteBuffer buffer);
+ void writeRemoteLogRecord(ByteBuffer buffer);
- public ITupleReference getOldValue();
+ ITupleReference getOldValue();
- public void setOldValue(ITupleReference tupleBefore);
+ void setOldValue(ITupleReference tupleBefore);
- public void setOldValueSize(int beforeSize);
+ void setOldValueSize(int beforeSize);
- public boolean isMarker();
+ boolean isMarker();
- public ByteBuffer getMarker();
+ ByteBuffer getMarker();
- public void logAppended(long lsn);
+ void logAppended(long lsn);
- public long getPreviousMarkerLSN();
+ long getPreviousMarkerLSN();
/**
* Sets flag indicating if this log should be replicated or not
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
index d85fd70..5fdb4e2 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogRecord.java
@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.zip.CRC32;
import org.apache.asterix.common.context.PrimaryIndexOperationTracker;
+import org.apache.commons.lang3.ArrayUtils;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.common.tuples.SimpleTupleReference;
import org.apache.hyracks.storage.am.common.tuples.SimpleTupleWriter;
@@ -36,18 +37,20 @@ import org.apache.hyracks.storage.am.common.tuples.SimpleTupleWriter;
* LogType(1)
* TxnId(8)
* ---------------------------
- * [Header2] (16 bytes + PKValueSize) : for entity_commit, upsert_entity_commit, and update log types
+ * [Header2] (8 bytes) : for entity_commit, upsert_entity_commit, filter and update log types
* DatasetId(4) //stored in dataset_dataset in Metadata Node
* ResourcePartition(4)
+ * ---------------------------
+ * [Header3] (8 bytes + PKValueSize) : for entity_commit, upsert_entity_commit, and update log types
* PKHashValue(4)
* PKValueSize(4)
* PKValue(PKValueSize)
* ---------------------------
- * [Header3] (12 bytes) : only for update log type
+ * [Header4] (12 bytes) : only for update, filter log type
* ResourceId(8) //stored in .metadata of the corresponding index in NC node
* LogRecordSize(4)
* ---------------------------
- * [Body] (9 bytes + NewValueSize) : only for update log type
+ * [Body] (9 bytes + NewValueSize) : only for update, filter log type
* FieldCnt(4)
* NewOp(1)
* NewValueSize(4)
@@ -57,7 +60,6 @@ import org.apache.hyracks.storage.am.common.tuples.SimpleTupleWriter;
* Checksum(8)
* ---------------------------
*/
-
public class LogRecord implements ILogRecord {
// ------------- fields in a log record (begin) ------------//
@@ -125,10 +127,12 @@ public class LogRecord implements ILogRecord {
buffer.putLong(txnId);
switch (logType) {
case LogType.ENTITY_COMMIT:
- writeEntityInfo(buffer);
+ writeEntityResource(buffer);
+ writeEntityValue(buffer);
break;
case LogType.UPDATE:
- writeEntityInfo(buffer);
+ writeEntityResource(buffer);
+ writeEntityValue(buffer);
buffer.putLong(resourceId);
buffer.putInt(logSize);
buffer.putInt(newValueFieldCount);
@@ -141,6 +145,15 @@ public class LogRecord implements ILogRecord {
writeTuple(buffer, oldValue, oldValueSize);
}
break;
+ case LogType.FILTER:
+ writeEntityResource(buffer);
+ buffer.putLong(resourceId);
+ buffer.putInt(logSize);
+ buffer.putInt(newValueFieldCount);
+ buffer.put(newOp);
+ buffer.putInt(newValueSize);
+ writeTuple(buffer, newValue, newValueSize);
+ break;
case LogType.FLUSH:
buffer.putInt(datasetId);
buffer.putInt(resourcePartition);
@@ -159,9 +172,7 @@ public class LogRecord implements ILogRecord {
}
}
- private void writeEntityInfo(ByteBuffer buffer) {
- buffer.putInt(resourcePartition);
- buffer.putInt(datasetId);
+ private void writeEntityValue(ByteBuffer buffer) {
buffer.putInt(PKHashValue);
if (PKValueSize <= 0) {
throw new IllegalStateException("Primary Key Size is less than or equal to 0");
@@ -170,6 +181,11 @@ public class LogRecord implements ILogRecord {
writePKValue(buffer);
}
+ private void writeEntityResource(ByteBuffer buffer) {
+ buffer.putInt(resourcePartition);
+ buffer.putInt(datasetId);
+ }
+
@Override
public void writeLogRecord(ByteBuffer buffer) {
int beginOffset = buffer.position();
@@ -264,51 +280,18 @@ public class LogRecord implements ILogRecord {
computeAndSetLogSize();
break;
case LogType.ENTITY_COMMIT:
- if (readEntityInfo(buffer)) {
+ if (readEntityResource(buffer) && readEntityValue(buffer)) {
computeAndSetLogSize();
} else {
return RecordReadStatus.TRUNCATED;
}
break;
case LogType.UPDATE:
- if (readEntityInfo(buffer)) {
- if (buffer.remaining() < UPDATE_LSN_HEADER + UPDATE_BODY_HEADER) {
- return RecordReadStatus.TRUNCATED;
- }
- resourceId = buffer.getLong();
- logSize = buffer.getInt();
- newValueFieldCount = buffer.getInt();
- newOp = buffer.get();
- newValueSize = buffer.getInt();
- if (buffer.remaining() < newValueSize) {
- if (logSize > buffer.capacity()) {
- return RecordReadStatus.LARGE_RECORD;
- }
- return RecordReadStatus.TRUNCATED;
- }
- newValue = readTuple(buffer, readNewValue, newValueFieldCount, newValueSize);
- if (logSize > getUpdateLogSizeWithoutOldValue()) {
- // Prev Image exists
- if (buffer.remaining() < Integer.BYTES) {
- return RecordReadStatus.TRUNCATED;
- }
- oldValueSize = buffer.getInt();
- if (buffer.remaining() < Integer.BYTES) {
- return RecordReadStatus.TRUNCATED;
- }
- oldValueFieldCount = buffer.getInt();
- if (buffer.remaining() < oldValueSize) {
- return RecordReadStatus.TRUNCATED;
- }
- oldValue = readTuple(buffer, readOldValue, oldValueFieldCount, oldValueSize);
- } else {
- oldValueSize = 0;
- oldValue = null;
- }
+ if (readEntityResource(buffer) && readEntityValue(buffer)) {
+ return readUpdateInfo(buffer);
} else {
return RecordReadStatus.TRUNCATED;
}
- break;
case LogType.MARKER:
if (buffer.remaining() < DS_LEN + RS_PARTITION_LEN + PRVLSN_LEN + LOGRCD_SZ_LEN) {
return RecordReadStatus.TRUNCATED;
@@ -331,19 +314,23 @@ public class LogRecord implements ILogRecord {
marker.position(lenRemaining);
marker.flip();
break;
+ case LogType.FILTER:
+ if (readEntityResource(buffer)) {
+ return readUpdateInfo(buffer);
+ } else {
+ return RecordReadStatus.TRUNCATED;
+ }
default:
break;
}
return RecordReadStatus.OK;
}
- private boolean readEntityInfo(ByteBuffer buffer) {
+ private boolean readEntityValue(ByteBuffer buffer) {
//attempt to read in the resourcePartition, dsid, PK hash and PK length
- if (buffer.remaining() < ENTITYCOMMIT_UPDATE_HEADER_LEN) {
+ if (buffer.remaining() < ENTITY_VALUE_HEADER_LEN) {
return false;
}
- resourcePartition = buffer.getInt();
- datasetId = buffer.getInt();
PKHashValue = buffer.getInt();
PKValueSize = buffer.getInt();
// attempt to read in the PK
@@ -357,6 +344,53 @@ public class LogRecord implements ILogRecord {
return true;
}
+ private boolean readEntityResource(ByteBuffer buffer) {
+ //attempt to read in the resourcePartition and dsid
+ if (buffer.remaining() < ENTITY_RESOURCE_HEADER_LEN) {
+ return false;
+ }
+ resourcePartition = buffer.getInt();
+ datasetId = buffer.getInt();
+ return true;
+ }
+
+ private RecordReadStatus readUpdateInfo(ByteBuffer buffer) {
+ if (buffer.remaining() < UPDATE_LSN_HEADER + UPDATE_BODY_HEADER) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ resourceId = buffer.getLong();
+ logSize = buffer.getInt();
+ newValueFieldCount = buffer.getInt();
+ newOp = buffer.get();
+ newValueSize = buffer.getInt();
+ if (buffer.remaining() < newValueSize) {
+ if (logSize > buffer.capacity()) {
+ return RecordReadStatus.LARGE_RECORD;
+ }
+ return RecordReadStatus.TRUNCATED;
+ }
+ newValue = readTuple(buffer, readNewValue, newValueFieldCount, newValueSize);
+ if (logSize > getUpdateLogSizeWithoutOldValue()) {
+ // Prev Image exists
+ if (buffer.remaining() < Integer.BYTES) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ oldValueSize = buffer.getInt();
+ if (buffer.remaining() < Integer.BYTES) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ oldValueFieldCount = buffer.getInt();
+ if (buffer.remaining() < oldValueSize) {
+ return RecordReadStatus.TRUNCATED;
+ }
+ oldValue = readTuple(buffer, readOldValue, oldValueFieldCount, oldValueSize);
+ } else {
+ oldValueSize = 0;
+ oldValue = null;
+ }
+ return RecordReadStatus.OK;
+ }
+
@Override
public void readRemoteLog(ByteBuffer buffer) {
//read common fields
@@ -403,6 +437,10 @@ public class LogRecord implements ILogRecord {
}
}
+ private int getFilterLogSize() {
+ return FILTER_LOG_BASE_SIZE + newValueSize;
+ }
+
private int getUpdateLogSizeWithoutOldValue() {
return UPDATE_LOG_BASE_SIZE + PKValueSize + newValueSize;
}
@@ -426,6 +464,9 @@ public class LogRecord implements ILogRecord {
case LogType.WAIT:
logSize = WAIT_LOG_SIZE;
break;
+ case LogType.FILTER:
+ logSize = getFilterLogSize();
+ break;
case LogType.MARKER:
setMarkerLogSize();
break;
@@ -499,8 +540,8 @@ public class LogRecord implements ILogRecord {
}
@Override
- public void setTxnId(long jobId) {
- this.txnId = jobId;
+ public void setTxnId(long txnId) {
+ this.txnId = txnId;
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
index 11c45ad..f02b0de 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
@@ -26,7 +26,8 @@ public class LogType {
public static final byte ABORT = 3;
public static final byte FLUSH = 4;
public static final byte WAIT = 6;
- public static final byte MARKER = 7;
+ public static final byte FILTER = 7;
+ public static final byte MARKER = 8;
private static final String STRING_UPDATE = "UPDATE";
private static final String STRING_JOB_COMMIT = "JOB_COMMIT";
@@ -34,6 +35,7 @@ public class LogType {
private static final String STRING_ABORT = "ABORT";
private static final String STRING_FLUSH = "FLUSH";
private static final String STRING_WAIT = "WAIT";
+ private static final String STRING_FILTER = "FILTER";
private static final String STRING_MARKER = "MARKER";
private static final String STRING_UNKNOWN_LOG_TYPE = "UNKNOWN_LOG_TYPE";
@@ -51,6 +53,8 @@ public class LogType {
return STRING_FLUSH;
case LogType.WAIT:
return STRING_WAIT;
+ case LogType.FILTER:
+ return STRING_FILTER;
case LogType.MARKER:
return STRING_MARKER;
default:
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/TestLSMIndexOperationContext.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/TestLSMIndexOperationContext.java b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/TestLSMIndexOperationContext.java
index 19536f6..9b749fa 100644
--- a/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/TestLSMIndexOperationContext.java
+++ b/asterixdb/asterix-common/src/test/java/org/apache/asterix/test/ioopcallbacks/TestLSMIndexOperationContext.java
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
@@ -45,6 +46,8 @@ public class TestLSMIndexOperationContext implements ILSMIndexOperationContext {
private IndexOperation op;
private LSMIOOperationType ioOperationType;
private ILSMDiskComponent newComponent;
+ private boolean filterSkip = false;
+ private boolean isRecovery = false;
public TestLSMIndexOperationContext(ILSMIndex index) {
this.index = index;
@@ -89,7 +92,7 @@ public class TestLSMIndexOperationContext implements ILSMIndexOperationContext {
}
@Override
- public IModificationOperationCallback getModificationCallback() {
+ public IExtendedModificationOperationCallback getModificationCallback() {
return NoOpOperationCallback.INSTANCE;
}
@@ -156,6 +159,27 @@ public class TestLSMIndexOperationContext implements ILSMIndexOperationContext {
}
@Override
+ public boolean isFilterSkipped() {
+ return filterSkip;
+ }
+
+ @Override
+ public void setFilterSkip(boolean skip) {
+ this.filterSkip = skip;
+ }
+
+ @Override
+ public boolean isRecovery() {
+ return isRecovery;
+ }
+
+ @Override
+ public void setRecovery(boolean recovery) {
+ this.isRecovery = recovery;
+
+ }
+
+ @Override
public LSMIOOperationType getIoOperationType() {
return ioOperationType;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogsProcessor.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogsProcessor.java b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogsProcessor.java
index 6189e37..b094d9e 100644
--- a/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogsProcessor.java
+++ b/asterixdb/asterix-replication/src/main/java/org/apache/asterix/replication/logging/RemoteLogsProcessor.java
@@ -53,6 +53,7 @@ public class RemoteLogsProcessor implements ILogRequester {
switch (reusableLog.getLogType()) {
case LogType.UPDATE:
case LogType.ENTITY_COMMIT:
+ case LogType.FILTER:
logManager.log(reusableLog);
break;
case LogType.JOB_COMMIT:
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
index 3da9e83..8746fba 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
@@ -29,20 +29,24 @@ import org.apache.asterix.common.transactions.LogRecord;
import org.apache.asterix.common.transactions.LogType;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.common.tuples.SimpleTupleWriter;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
public abstract class AbstractIndexModificationOperationCallback extends AbstractOperationCallback
- implements IModificationOperationCallback {
+ implements IExtendedModificationOperationCallback {
public static final byte INSERT_BYTE = 0x01;
public static final byte DELETE_BYTE = 0x02;
public static final byte UPSERT_BYTE = 0x03;
+ public static final byte FILTER_BYTE = 0x04;
public enum Operation {
INSERT(INSERT_BYTE),
DELETE(DELETE_BYTE),
- UPSERT(UPSERT_BYTE);
+ UPSERT(UPSERT_BYTE),
+ FILTER_MOD(FILTER_BYTE);
+
private byte value;
Operation(byte value) {
@@ -59,6 +63,8 @@ public abstract class AbstractIndexModificationOperationCallback extends Abstrac
return DELETE;
case INSERT:
return INSERT;
+ case FILTER_MOD:
+ return FILTER_MOD;
case UPSERT:
return UPSERT;
default:
@@ -71,7 +77,8 @@ public abstract class AbstractIndexModificationOperationCallback extends Abstrac
protected final byte resourceType;
protected final Operation indexOp;
protected final ITransactionSubsystem txnSubsystem;
- protected final ILogRecord logRecord;
+ protected final ILogRecord indexRecord;
+ protected final ILogRecord filterRecord;
protected AbstractIndexModificationOperationCallback(DatasetId datasetId, int[] primaryKeyFields,
ITransactionContext txnCtx, ILockManager lockManager, ITransactionSubsystem txnSubsystem, long resourceId,
@@ -80,35 +87,52 @@ public abstract class AbstractIndexModificationOperationCallback extends Abstrac
this.resourceType = resourceType;
this.indexOp = indexOp;
this.txnSubsystem = txnSubsystem;
- logRecord = new LogRecord();
- logRecord.setTxnCtx(txnCtx);
- logRecord.setLogType(LogType.UPDATE);
- logRecord.setTxnId(txnCtx.getTxnId().getId());
- logRecord.setDatasetId(datasetId.getId());
- logRecord.setResourceId(resourceId);
- logRecord.setResourcePartition(resourcePartition);
- logRecord.setNewOp(indexOp.value());
+ indexRecord = new LogRecord();
+ indexRecord.setTxnCtx(txnCtx);
+ indexRecord.setLogType(LogType.UPDATE);
+ indexRecord.setTxnId(txnCtx.getTxnId().getId());
+ indexRecord.setDatasetId(datasetId.getId());
+ indexRecord.setResourceId(resourceId);
+ indexRecord.setResourcePartition(resourcePartition);
+ indexRecord.setNewOp(indexOp.value());
+ filterRecord = new LogRecord();
+ filterRecord.setTxnCtx(txnCtx);
+ filterRecord.setLogType(LogType.FILTER);
+ filterRecord.setDatasetId(datasetId.getId());
+ filterRecord.setTxnId(txnCtx.getTxnId().getId());
+ filterRecord.setResourceId(resourceId);
+ filterRecord.setResourcePartition(resourcePartition);
+ filterRecord.setNewOp(Operation.FILTER_MOD.value());
}
protected void log(int PKHash, ITupleReference newValue, ITupleReference oldValue) throws ACIDException {
- logRecord.setPKHashValue(PKHash);
- logRecord.setPKFields(primaryKeyFields);
- logRecord.setPKValue(newValue);
- logRecord.computeAndSetPKValueSize();
+ indexRecord.setPKHashValue(PKHash);
+ indexRecord.setPKFields(primaryKeyFields);
+ indexRecord.setPKValue(newValue);
+ indexRecord.computeAndSetPKValueSize();
if (newValue != null) {
- logRecord.setNewValueSize(SimpleTupleWriter.INSTANCE.bytesRequired(newValue));
- logRecord.setNewValue(newValue);
+ indexRecord.setNewValueSize(SimpleTupleWriter.INSTANCE.bytesRequired(newValue));
+ indexRecord.setNewValue(newValue);
} else {
- logRecord.setNewValueSize(0);
+ indexRecord.setNewValueSize(0);
}
if (oldValue != null) {
- logRecord.setOldValueSize(SimpleTupleWriter.INSTANCE.bytesRequired(oldValue));
- logRecord.setOldValue(oldValue);
+ indexRecord.setOldValueSize(SimpleTupleWriter.INSTANCE.bytesRequired(oldValue));
+ indexRecord.setOldValue(oldValue);
} else {
- logRecord.setOldValueSize(0);
+ indexRecord.setOldValueSize(0);
+ }
+ indexRecord.computeAndSetLogSize();
+ txnSubsystem.getLogManager().log(indexRecord);
+ }
+
+ public void after(ITupleReference newValue) throws HyracksDataException {
+ if (newValue != null) {
+ filterRecord.setNewValueSize(SimpleTupleWriter.INSTANCE.bytesRequired(newValue));
+ filterRecord.setNewValue(newValue);
+ filterRecord.computeAndSetLogSize();
+ txnSubsystem.getLogManager().log(filterRecord);
}
- logRecord.computeAndSetLogSize();
- txnSubsystem.getLogManager().log(logRecord);
}
/**
@@ -116,9 +140,8 @@ public abstract class AbstractIndexModificationOperationCallback extends Abstrac
* a single operator to perform different operations per tuple
*
* @param op
- * @throws HyracksDataException
*/
- public void setOp(Operation op) throws HyracksDataException {
- logRecord.setNewOp(op.value());
+ public void setOp(Operation op) {
+ indexRecord.setNewOp(op.value());
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallback.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallback.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallback.java
index 2c8079d..3e41264 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallback.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/PrimaryIndexModificationOperationCallback.java
@@ -104,10 +104,10 @@ public class PrimaryIndexModificationOperationCallback extends AbstractIndexModi
}
private void logWait() throws ACIDException {
- logRecord.setLogType(LogType.WAIT);
- logRecord.computeAndSetLogSize();
- txnSubsystem.getLogManager().log(logRecord);
+ indexRecord.setLogType(LogType.WAIT);
+ indexRecord.computeAndSetLogSize();
+ txnSubsystem.getLogManager().log(indexRecord);
// set the log type back to UPDATE for normal updates
- logRecord.setLogType(LogType.UPDATE);
+ indexRecord.setLogType(LogType.UPDATE);
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
index a1aec1a..1e13883 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/logging/LogManagerWithReplication.java
@@ -48,6 +48,7 @@ public class LogManagerWithReplication extends LogManager {
case LogType.ENTITY_COMMIT:
case LogType.UPDATE:
case LogType.FLUSH:
+ case LogType.FILTER:
shouldReplicate = replicationStrategy.isMatch(logRecord.getDatasetId());
if (shouldReplicate && !replicatedTxn.contains(logRecord.getTxnId())) {
replicatedTxn.add(logRecord.getTxnId());
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
index 077a006..fb8770e 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTree.java
@@ -300,7 +300,7 @@ public class BTree extends AbstractTreeIndex {
}
// fall-through
case SUFFICIENT_CONTIGUOUS_SPACE: {
- foundModCallback(ctx, null, tuple);
+ ctx.getModificationCallback().found(null, tuple);
ctx.getLeafFrame().insert(tuple, targetTupleIndex);
ctx.getSplitKey().reset();
break;
@@ -308,7 +308,7 @@ public class BTree extends AbstractTreeIndex {
case SUFFICIENT_SPACE: {
int finalIndex = ctx.getLeafFrame().compact() ? ctx.getLeafFrame().findInsertTupleIndex(tuple)
: targetTupleIndex;
- foundModCallback(ctx, null, tuple);
+ ctx.getModificationCallback().found(null, tuple);
ctx.getLeafFrame().insert(tuple, finalIndex);
ctx.getSplitKey().reset();
break;
@@ -317,7 +317,7 @@ public class BTree extends AbstractTreeIndex {
// Try compressing the page first and see if there is space available.
if (ctx.getLeafFrame().compress()
&& ctx.getLeafFrame().hasSpaceInsert(tuple) == FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE) {
- foundModCallback(ctx, null, tuple);
+ ctx.getModificationCallback().found(null, tuple);
ctx.getLeafFrame().insert(tuple, ctx.getLeafFrame().findInsertTupleIndex(tuple));
ctx.getSplitKey().reset();
} else {
@@ -360,10 +360,10 @@ public class BTree extends AbstractTreeIndex {
// Perform an update (delete + insert) if the updateTupleIndex != -1
if (updateTupleIndex != -1) {
ITupleReference beforeTuple = ctx.getLeafFrame().getMatchingKeyTuple(tuple, updateTupleIndex);
- foundModCallback(ctx, beforeTuple, tuple);
+ ctx.getModificationCallback().found(beforeTuple, tuple);
ctx.getLeafFrame().delete(tuple, updateTupleIndex);
} else {
- foundModCallback(ctx, null, tuple);
+ ctx.getModificationCallback().found(null, tuple);
}
ctx.getLeafFrame().split(rightFrame, tuple, ctx.getSplitKey(), ctx, bufferCache);
@@ -398,7 +398,7 @@ public class BTree extends AbstractTreeIndex {
boolean restartOp = false;
switch (spaceStatus) {
case SUFFICIENT_INPLACE_SPACE: {
- foundModCallback(ctx, beforeTuple, tuple);
+ ctx.getModificationCallback().found(beforeTuple, tuple);
ctx.getLeafFrame().update(tuple, oldTupleIndex, true);
ctx.getSplitKey().reset();
break;
@@ -407,7 +407,7 @@ public class BTree extends AbstractTreeIndex {
// TODO: avoid repeated calculation of tuple size
// TODO: in-place update on expand
// Delete the old tuple, compact the frame, and insert the new tuple.
- foundModCallback(ctx, beforeTuple, tuple);
+ ctx.getModificationCallback().found(beforeTuple, tuple);
ctx.getLeafFrame().delete(tuple, oldTupleIndex);
ctx.getLeafFrame().compact();
ctx.getLeafFrame().ensureCapacity(bufferCache, tuple, ctx);
@@ -417,14 +417,14 @@ public class BTree extends AbstractTreeIndex {
break;
}
case SUFFICIENT_CONTIGUOUS_SPACE: {
- foundModCallback(ctx, beforeTuple, tuple);
+ ctx.getModificationCallback().found(beforeTuple, tuple);
ctx.getLeafFrame().update(tuple, oldTupleIndex, false);
ctx.getSplitKey().reset();
break;
}
case SUFFICIENT_SPACE: {
// Delete the old tuple, compact the frame, and insert the new tuple.
- foundModCallback(ctx, beforeTuple, tuple);
+ ctx.getModificationCallback().found(beforeTuple, tuple);
ctx.getLeafFrame().delete(tuple, oldTupleIndex);
ctx.getLeafFrame().compact();
int targetTupleIndex = ctx.getLeafFrame().findInsertTupleIndex(tuple);
@@ -759,12 +759,6 @@ public class BTree extends AbstractTreeIndex {
modificationCallback, searchCallback);
}
- private BTreeOpContext createOpContext(IIndexAccessor accessor, IModificationOperationCallback modificationCallback,
- ISearchOperationCallback searchCallback, int[] logTupleFields) {
- return new BTreeOpContext(accessor, leafFrameFactory, interiorFrameFactory, freePageManager, cmpFactories,
- modificationCallback, searchCallback, logTupleFields);
- }
-
@SuppressWarnings("rawtypes")
public String printTree(IBTreeLeafFrame leafFrame, IBTreeInteriorFrame interiorFrame,
ISerializerDeserializer[] keySerdes) throws Exception {
@@ -824,11 +818,6 @@ public class BTree extends AbstractTreeIndex {
return new BTreeAccessor(this, iap.getModificationCallback(), iap.getSearchOperationCallback());
}
- public BTreeAccessor createAccessor(IModificationOperationCallback modificationCallback,
- ISearchOperationCallback searchCallback, int[] logTupleFields) {
- return new BTreeAccessor(this, modificationCallback, searchCallback, logTupleFields);
- }
-
// TODO: Class should be private. But currently we need to expose the
// setOpContext() API to the LSM Tree for it to work correctly.
@@ -849,12 +838,6 @@ public class BTree extends AbstractTreeIndex {
this.ctx = btree.createOpContext(this, modificationCalback, searchCallback);
}
- public BTreeAccessor(BTree btree, IModificationOperationCallback modificationCalback,
- ISearchOperationCallback searchCallback, int[] logTupleFields) {
- this.btree = btree;
- this.ctx = btree.createOpContext(this, modificationCalback, searchCallback, logTupleFields);
- }
-
public void reset(BTree btree, IModificationOperationCallback modificationCallback,
ISearchOperationCallback searchCallback) {
this.btree = btree;
@@ -1251,14 +1234,4 @@ public class BTree extends AbstractTreeIndex {
public int getNumOfFilterFields() {
return 0;
}
-
- private void foundModCallback(BTreeOpContext ctx, ITupleReference before, ITupleReference after)
- throws HyracksDataException {
- if (ctx.getTupleWithNonIndexFields() == null) {
- ctx.getModificationCallback().found(before, after);
- } else {
- ctx.getModificationCallback().found(before, ctx.getTupleWithNonIndexFields());
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
index 96370fe..c082ad7 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/BTreeOpContext.java
@@ -39,7 +39,6 @@ import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.common.api.ITreeIndexMetadataFrame;
import org.apache.hyracks.storage.am.common.api.ITreeIndexTupleReference;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
-import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
import org.apache.hyracks.storage.common.IIndexAccessor;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
@@ -57,7 +56,6 @@ public class BTreeOpContext implements IIndexOperationContext, IExtraPageBlockHe
private final IBTreeInteriorFrame interiorFrame;
private final IPageManager freePageManager;
private final ITreeIndexMetadataFrame metaFrame;
- private PermutingTupleReference tupleWithNonIndexFields; // Optional, for filtered LSM Index transaction support
private ITreeIndexFrameFactory leafFrameFactory;
private IBTreeLeafFrame leafFrame;
private IndexOperation op;
@@ -115,15 +113,6 @@ public class BTreeOpContext implements IIndexOperationContext, IExtraPageBlockHe
this.interiorFrameTuple = getInteriorFrame().createTupleReference();
}
- public BTreeOpContext(IIndexAccessor accessor, ITreeIndexFrameFactory leafFrameFactory,
- ITreeIndexFrameFactory interiorFrameFactory, IPageManager freePageManager,
- IBinaryComparatorFactory[] cmpFactories, IModificationOperationCallback modificationCallback,
- ISearchOperationCallback searchCallback, int[] nonIndexFields) {
- this(accessor, leafFrameFactory, interiorFrameFactory, freePageManager, cmpFactories, modificationCallback,
- searchCallback);
- this.tupleWithNonIndexFields = new PermutingTupleReference(nonIndexFields);
- }
-
@Override
public void reset() {
if (pageLsns != null) {
@@ -378,14 +367,6 @@ public class BTreeOpContext implements IIndexOperationContext, IExtraPageBlockHe
this.leafFrameFactory = leafFrameFactory;
}
- public ITupleReference getTupleWithNonIndexFields() {
- return tupleWithNonIndexFields;
- }
-
- public void resetNonIndexFieldsTuple(ITupleReference newValue) {
- tupleWithNonIndexFields.reset(newValue);
- }
-
@Override
public void destroy() throws HyracksDataException {
if (destroyed) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
index eee43b5..523ed9b 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-btree/src/main/java/org/apache/hyracks/storage/am/btree/impls/DiskBTree.java
@@ -198,12 +198,6 @@ public class DiskBTree extends BTree {
return new DiskBTreeAccessor(this, iap.getModificationCallback(), iap.getSearchOperationCallback());
}
- @Override
- public DiskBTreeAccessor createAccessor(IModificationOperationCallback modificationCallback,
- ISearchOperationCallback searchCallback, int[] logTupleFields) {
- return new DiskBTreeAccessor(this, modificationCallback, searchCallback, logTupleFields);
- }
-
public class DiskBTreeAccessor extends BTreeAccessor {
public DiskBTreeAccessor(DiskBTree btree, IModificationOperationCallback modificationCalback,
@@ -211,11 +205,6 @@ public class DiskBTree extends BTree {
super(btree, modificationCalback, searchCallback);
}
- public DiskBTreeAccessor(DiskBTree btree, IModificationOperationCallback modificationCalback,
- ISearchOperationCallback searchCallback, int[] logTupleFields) {
- super(btree, modificationCalback, searchCallback, logTupleFields);
- }
-
@Override
public void insert(ITupleReference tuple) throws HyracksDataException {
throw new UnsupportedOperationException("Insert is not supported by DiskBTree. ");
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IExtendedModificationOperationCallback.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IExtendedModificationOperationCallback.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IExtendedModificationOperationCallback.java
new file mode 100644
index 0000000..7bb4e82
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/api/IExtendedModificationOperationCallback.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.common.api;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.common.IModificationOperationCallback;
+
+public interface IExtendedModificationOperationCallback extends IModificationOperationCallback {
+ /**
+ * Called after the action taken in found, to take action on a tuple that is not part of the index
+ * itself but is part of an ancillary structure that is updated alongside the index. An example would
+ * be a simple statistic on the index that records the minimum and maximum values.
+ *
+ * @param after
+ * The tuple to feed to the ancilliary structure
+ * @throws HyracksDataException
+ */
+
+ void after(ITupleReference after) throws HyracksDataException;
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/ExtendedIndexAccessParameters.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/ExtendedIndexAccessParameters.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/ExtendedIndexAccessParameters.java
new file mode 100644
index 0000000..dbefe74
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/ExtendedIndexAccessParameters.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.hyracks.storage.am.common.impls;
+
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
+import org.apache.hyracks.storage.common.IIndex;
+import org.apache.hyracks.storage.common.IIndexAccessParameters;
+import org.apache.hyracks.storage.common.ISearchOperationCallback;
+
+public class ExtendedIndexAccessParameters extends IndexAccessParameters implements IIndexAccessParameters {
+
+ protected final IExtendedModificationOperationCallback extendedModificationCallback;
+ // This map is used to put additional parameters to an index accessor.
+
+ public ExtendedIndexAccessParameters(IExtendedModificationOperationCallback extendedModificationCallback,
+ ISearchOperationCallback searchOperationCallback) {
+ super(extendedModificationCallback, searchOperationCallback);
+ this.extendedModificationCallback = extendedModificationCallback;
+ }
+
+ @Override
+ public IExtendedModificationOperationCallback getModificationCallback() {
+ return extendedModificationCallback;
+ }
+}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpIndexAccessParameters.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpIndexAccessParameters.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpIndexAccessParameters.java
index 2edea70..0068f4f 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpIndexAccessParameters.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpIndexAccessParameters.java
@@ -21,6 +21,7 @@ package org.apache.hyracks.storage.am.common.impls;
import java.util.Collections;
import java.util.Map;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.common.IIndexAccessParameters;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
@@ -34,7 +35,7 @@ public class NoOpIndexAccessParameters implements IIndexAccessParameters {
}
@Override
- public IModificationOperationCallback getModificationCallback() {
+ public IExtendedModificationOperationCallback getModificationCallback() {
return NoOpOperationCallback.INSTANCE;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpOperationCallback.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpOperationCallback.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpOperationCallback.java
index 15aba57..85417f2 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpOperationCallback.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/impls/NoOpOperationCallback.java
@@ -21,13 +21,15 @@ package org.apache.hyracks.storage.am.common.impls;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.ISearchOperationCallback;
/**
* Dummy operation callback that simply does nothing.
*/
-public enum NoOpOperationCallback implements IModificationOperationCallback, ISearchOperationCallback {
+public enum NoOpOperationCallback
+ implements IModificationOperationCallback, ISearchOperationCallback, IExtendedModificationOperationCallback {
INSTANCE;
@Override
@@ -59,4 +61,9 @@ public enum NoOpOperationCallback implements IModificationOperationCallback, ISe
public void complete(ITupleReference tuple) throws HyracksDataException {
// Do nothing.
}
+
+ @Override
+ public void after(ITupleReference tuple) throws HyracksDataException {
+ //Do nothing.
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/ophelpers/IndexOperation.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/ophelpers/IndexOperation.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/ophelpers/IndexOperation.java
index ff47d27..636e4f5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/ophelpers/IndexOperation.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/ophelpers/IndexOperation.java
@@ -26,6 +26,7 @@ public enum IndexOperation {
UPDATE,
UPSERT,
SEARCH,
+ FILTER_MOD,
DISKORDERSCAN,
PHYSICALDELETE,
NOOP,
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
index 1c74275..7f53ed5 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/ExternalBTreeWithBuddyOpContext.java
@@ -20,6 +20,7 @@ package org.apache.hyracks.storage.am.lsm.btree.impls;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
@@ -65,7 +66,7 @@ public class ExternalBTreeWithBuddyOpContext extends AbstractLSMIndexOperationCo
// This should never be needed for disk only indexes
@Override
- public IModificationOperationCallback getModificationCallback() {
+ public IExtendedModificationOperationCallback getModificationCallback() {
return null;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
index 4578eb3..41a11e6 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTree.java
@@ -32,11 +32,13 @@ import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.btree.impls.BTree;
import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeAccessor;
import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IIndexOperationContext;
import org.apache.hyracks.storage.am.common.api.IPageManager;
import org.apache.hyracks.storage.am.common.api.ITreeIndex;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
+import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.lsm.btree.tuples.LSMBTreeTupleReference;
import org.apache.hyracks.storage.am.lsm.common.api.AbstractLSMWithBloomFilterDiskComponent;
import org.apache.hyracks.storage.am.lsm.common.api.IComponentFilterHelper;
@@ -151,7 +153,6 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
if (ctx.getIndexTuple() != null) {
ctx.getIndexTuple().reset(tuple);
indexTuple = ctx.getIndexTuple();
- ctx.getCurrentMutableBTreeAccessor().getOpContext().resetNonIndexFieldsTuple(tuple);
} else {
indexTuple = tuple;
}
@@ -303,7 +304,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
List<ITupleReference> filterTuples = new ArrayList<>();
filterTuples.add(flushingComponent.getLSMComponentFilter().getMinTuple());
filterTuples.add(flushingComponent.getLSMComponentFilter().getMaxTuple());
- getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples);
+ getFilterManager().updateFilter(component.getLSMComponentFilter(), filterTuples,
+ NoOpOperationCallback.INSTANCE);
getFilterManager().writeFilter(component.getLSMComponentFilter(), component.getMetadataHolder());
}
// Write metadata from memory component to disk
@@ -353,7 +355,8 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMinTuple());
filterTuples.add(mergeOp.getMergingComponents().get(i).getLSMComponentFilter().getMaxTuple());
}
- getFilterManager().updateFilter(mergedComponent.getLSMComponentFilter(), filterTuples);
+ getFilterManager().updateFilter(mergedComponent.getLSMComponentFilter(), filterTuples,
+ NoOpOperationCallback.INSTANCE);
getFilterManager().writeFilter(mergedComponent.getLSMComponentFilter(),
mergedComponent.getMetadataHolder());
}
@@ -396,8 +399,9 @@ public class LSMBTree extends AbstractLSMIndex implements ITreeIndex {
int numBloomFilterKeyFields = hasBloomFilter
? ((LSMBTreeWithBloomFilterDiskComponentFactory) componentFactory).getBloomFilterKeyFields().length : 0;
return new LSMBTreeOpContext(this, memoryComponents, insertLeafFrameFactory, deleteLeafFrameFactory,
- iap.getModificationCallback(), iap.getSearchOperationCallback(), numBloomFilterKeyFields,
- getTreeFields(), getFilterFields(), getHarness(), getFilterCmpFactories(), tracer);
+ (IExtendedModificationOperationCallback) iap.getModificationCallback(),
+ iap.getSearchOperationCallback(), numBloomFilterKeyFields, getTreeFields(), getFilterFields(),
+ getHarness(), getFilterCmpFactories(), tracer);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
index 7969ba3..1cfc414 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-btree/src/main/java/org/apache/hyracks/storage/am/lsm/btree/impls/LSMBTreeOpContext.java
@@ -29,8 +29,10 @@ import org.apache.hyracks.storage.am.btree.impls.BTree;
import org.apache.hyracks.storage.am.btree.impls.BTreeOpContext;
import org.apache.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
+import org.apache.hyracks.storage.am.common.impls.NoOpIndexAccessParameters;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
@@ -69,7 +71,7 @@ public final class LSMBTreeOpContext extends AbstractLSMIndexOperationContext {
public LSMBTreeOpContext(ILSMIndex index, List<ILSMMemoryComponent> mutableComponents,
ITreeIndexFrameFactory insertLeafFrameFactory, ITreeIndexFrameFactory deleteLeafFrameFactory,
- IModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
+ IExtendedModificationOperationCallback modificationCallback, ISearchOperationCallback searchCallback,
int numBloomFilterKeyFields, int[] btreeFields, int[] filterFields, ILSMHarness lsmHarness,
IBinaryComparatorFactory[] filterCmpFactories, ITracer tracer) {
super(index, btreeFields, filterFields, filterCmpFactories, searchCallback, modificationCallback, tracer);
@@ -90,14 +92,9 @@ public final class LSMBTreeOpContext extends AbstractLSMIndexOperationContext {
for (int i = 0; i < mutableComponents.size(); i++) {
LSMBTreeMemoryComponent mutableComponent = (LSMBTreeMemoryComponent) mutableComponents.get(i);
mutableBTrees[i] = mutableComponent.getIndex();
- if (allFields != null) {
- mutableBTreeAccessors[i] = mutableBTrees[i].createAccessor(modificationCallback,
- NoOpOperationCallback.INSTANCE, allFields);
- } else {
- IIndexAccessParameters iap =
- new IndexAccessParameters(modificationCallback, NoOpOperationCallback.INSTANCE);
- mutableBTreeAccessors[i] = mutableBTrees[i].createAccessor(iap);
- }
+ IIndexAccessParameters iap =
+ new IndexAccessParameters(modificationCallback, NoOpOperationCallback.INSTANCE);
+ mutableBTreeAccessors[i] = mutableBTrees[i].createAccessor(iap);
mutableBTreeOpCtxs[i] = mutableBTreeAccessors[i].getOpContext();
}
this.insertLeafFrameFactory = insertLeafFrameFactory;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilter.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilter.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilter.java
index b92e0d2..34f8855 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilter.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilter.java
@@ -21,14 +21,16 @@ package org.apache.hyracks.storage.am.lsm.common.api;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.common.MultiComparator;
public interface ILSMComponentFilter {
- void update(ITupleReference tuple, MultiComparator cmp) throws HyracksDataException;
-
boolean satisfy(ITupleReference min, ITupleReference max, MultiComparator cmp) throws HyracksDataException;
+ void update(ITupleReference tuple, MultiComparator cmp, IExtendedModificationOperationCallback opCallback)
+ throws HyracksDataException;
+
ITupleReference getMinTuple();
ITupleReference getMaxTuple();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
index f310b4e..64b0c44 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMComponentFilterManager.java
@@ -22,11 +22,13 @@ import java.util.List;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.ITreeIndex;
public interface ILSMComponentFilterManager {
- void updateFilter(ILSMComponentFilter filter, List<ITupleReference> filterTuples) throws HyracksDataException;
+ void updateFilter(ILSMComponentFilter filter, List<ITupleReference> filterTuples,
+ IExtendedModificationOperationCallback opCallback) throws HyracksDataException;
boolean readFilter(ILSMComponentFilter filter, ITreeIndex index) throws HyracksDataException;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
index 0e1a5e4..61ef6cf 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
@@ -263,4 +263,11 @@ public interface ILSMIndexAccessor extends IIndexAccessor {
* @throws HyracksDataException
*/
void deleteComponents(Predicate<ILSMComponent> predicate) throws HyracksDataException;
+
+ /**
+ * Update the filter of an LSM index
+ * @param tuple
+ * @throws HyracksDataException
+ */
+ void updateFilter(ITupleReference tuple) throws HyracksDataException;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/693844cc/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
index 79b3262..b34b403 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexOperationContext.java
@@ -20,6 +20,7 @@ package org.apache.hyracks.storage.am.lsm.common.api;
import java.util.List;
+import org.apache.hyracks.storage.am.common.api.IExtendedModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.IIndexOperationContext;
import org.apache.hyracks.storage.am.common.tuples.PermutingTupleReference;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation.LSMIOOperationType;
@@ -35,7 +36,7 @@ public interface ILSMIndexOperationContext extends IIndexOperationContext {
ISearchOperationCallback getSearchOperationCallback();
- IModificationOperationCallback getModificationCallback();
+ IExtendedModificationOperationCallback getModificationCallback();
void setCurrentMutableComponentId(int currentMutableComponentId);
@@ -84,6 +85,14 @@ public interface ILSMIndexOperationContext extends IIndexOperationContext {
*/
boolean isTracingEnabled();
+ boolean isFilterSkipped();
+
+ void setFilterSkip(boolean skip);
+
+ boolean isRecovery();
+
+ void setRecovery(boolean recovery);
+
/**
* @return the IO Operation type associated with this context
*/