Ian Maxon has uploaded a new change for review.
https://asterix-gerrit.ics.uci.edu/1857
Change subject: !WIP![ASTERIXDB-1952][TX][IDX]Log incoming filter vals
......................................................................
!WIP![ASTERIXDB-1952][TX][IDX]Log incoming filter vals
Change-Id: Ie9e7795d9c8c212e8610dcb9bb5d26ec9fbbee8a
---
M
asterixdb/asterix-app/src/main/java/org/apache/asterix/app/nc/RecoveryManager.java
M
asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
M
asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorNodePushable.java
M
asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/opcallbacks/AbstractIndexModificationOperationCallback.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/ophelpers/IndexOperation.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMIndexAccessor.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
M
hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
10 files changed, 58 insertions(+), 2 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/57/1857/2
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 275b055..4da856f 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
@@ -704,6 +704,8 @@
} else {
indexAccessor.forceUpsert(logRecord.getOldValue());
}
+ } else if (logRecord.getNewOp() ==
AbstractIndexModificationOperationCallback.FILTER_BYTE) {
+ // filters are not able to be undone, they are widen-only.
} else {
throw new IllegalStateException("Unsupported OperationType: "
+ logRecord.getNewOp());
}
@@ -726,6 +728,9 @@
} else if (logRecord.getNewOp() ==
AbstractIndexModificationOperationCallback.UPSERT_BYTE) {
// redo, upsert the new value
indexAccessor.forceUpsert(logRecord.getNewValue());
+ } else if (logRecord.getNewOp() ==
AbstractIndexModificationOperationCallback.FILTER_BYTE){
+ indexAccessor.updateFilter(logRecord.getNewValue());
+
} else {
throw new IllegalStateException("Unsupported OperationType: "
+ logRecord.getNewOp());
}
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..7a64791 100644
---
a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
+++
b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/LogType.java
@@ -27,6 +27,7 @@
public static final byte FLUSH = 4;
public static final byte WAIT = 6;
public static final byte MARKER = 7;
+ public static final byte FILTER = 8;
private static final String STRING_UPDATE = "UPDATE";
private static final String STRING_JOB_COMMIT = "JOB_COMMIT";
@@ -35,6 +36,7 @@
private static final String STRING_FLUSH = "FLUSH";
private static final String STRING_WAIT = "WAIT";
private static final String STRING_MARKER = "MARKER";
+ private static final String STRING_FILTER = "FILTER";
private static final String STRING_UNKNOWN_LOG_TYPE = "UNKNOWN_LOG_TYPE";
public static String toString(byte logType) {
@@ -53,6 +55,8 @@
return STRING_WAIT;
case LogType.MARKER:
return STRING_MARKER;
+ case LogType.FILTER:
+ return STRING_FILTER;
default:
return STRING_UNKNOWN_LOG_TYPE;
}
diff --git
a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorNodePushable.java
b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorNodePushable.java
index 1f41c82..2023a84 100644
---
a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorNodePushable.java
+++
b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorNodePushable.java
@@ -164,7 +164,8 @@
}
if (isFiltered && prevTuple != null) {
// need to update the filter of the new component with
the previous value
- lsmAccessor.updateFilter(prevTuple);
+ abstractModCallback.setOp(Operation.FILTER_MOD);
+ lsmAccessor.updateFilter(prevTuple, true);
}
writeOutput(index, recordWasInserted, recordWasDeleted);
} catch (Exception e) {
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 1acc235..f507790 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
@@ -38,11 +38,13 @@
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) {
@@ -61,6 +63,8 @@
return INSERT;
case UPSERT:
return UPSERT;
+ case FILTER_MOD:
+ return FILTER_MOD;
default:
throw new IllegalArgumentException();
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 71a3f71..d47c5b4 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
@@ -25,6 +25,7 @@
DELETE,
UPDATE,
UPSERT,
+ FILTER_MOD,
SEARCH,
DISKORDERSCAN,
PHYSICALDELETE,
diff --git
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
index c0a3f2d..4b601bc 100644
---
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
+++
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
@@ -226,6 +226,14 @@
void updateFilter(ILSMIndexOperationContext ctx, ITupleReference tuple)
throws HyracksDataException;
/**
+ * Update the filter with the value in the passed tuple, also give the
tuple to the modification callback
+ *
+ * @param ctx
+ * @throws HyracksDataException
+ */
+ void updateFilter(ILSMIndexOperationContext ctx, ITupleReference tuple,
boolean callback) throws HyracksDataException;
+
+ /**
* Perform batch operation on all tuples in the passed frame tuple accessor
*
* @param ctx
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 1042df2..fb38fe2 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
@@ -245,4 +245,11 @@
* If the BufferCache throws while un/pinning or un/latching.
*/
void scanDiskComponents(IIndexCursor cursor) throws HyracksDataException;
+
+ /**
+ * Update the filter of an LSM index
+ * @param tuple
+ * @throws HyracksDataException
+ */
+ void updateFilter(ITupleReference tuple) throws HyracksDataException;
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
index 50eac67..bfeca94 100644
---
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
+++
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
@@ -618,6 +618,21 @@
lsmIndex.updateFilter(ctx, tuple);
}
+ @Override
+ public void updateFilter(ILSMIndexOperationContext ctx, ITupleReference
tuple, boolean callback)
+ throws HyracksDataException {
+ if (!callback) {
+ updateFilter(ctx, tuple);
+ return;
+ }
+
+ if (!lsmIndex.isMemoryComponentsAllocated()) {
+ lsmIndex.allocateMemoryComponents();
+ }
+ ctx.getModificationCallback().found(null, tuple);
+ lsmIndex.updateFilter(ctx, tuple);
+ }
+
private void enter(ILSMIndexOperationContext ctx) throws
HyracksDataException {
if (!lsmIndex.isMemoryComponentsAllocated()) {
lsmIndex.allocateMemoryComponents();
diff --git
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
index b9714a6..fa743d4 100644
---
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
+++
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
@@ -204,11 +204,17 @@
return cursorFactory.create(ctx);
}
+ @Override
public void updateFilter(ITupleReference tuple) throws
HyracksDataException {
ctx.setOperation(IndexOperation.UPSERT);
lsmHarness.updateFilter(ctx, tuple);
}
+ public void updateFilter(ITupleReference tuple, boolean callback) throws
HyracksDataException {
+ ctx.setOperation(IndexOperation.UPSERT);
+ lsmHarness.updateFilter(ctx, tuple, true);
+ }
+
public void batchOperate(FrameTupleAccessor accessor, FrameTupleReference
tuple, IFrameTupleProcessor processor,
IFrameOperationCallback frameOpCallback) throws
HyracksDataException {
lsmHarness.batchOperate(ctx, accessor, tuple, processor,
frameOpCallback);
diff --git
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
index 7751116..fd4cb31 100644
---
a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
+++
b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-invertedindex/src/main/java/org/apache/hyracks/storage/am/lsm/invertedindex/impls/LSMInvertedIndexAccessor.java
@@ -200,6 +200,11 @@
}
@Override
+ public void updateFilter(ITupleReference tuple) throws
HyracksDataException {
+ lsmHarness.updateFilter(ctx,tuple);
+ }
+
+ @Override
public void forceUpsert(ITupleReference tuple) throws HyracksDataException
{
throw new UnsupportedOperationException("Upsert not supported by lsm
inverted index.");
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/1857
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie9e7795d9c8c212e8610dcb9bb5d26ec9fbbee8a
Gerrit-PatchSet: 2
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: Ian Maxon <[email protected]>
Gerrit-Reviewer: Jenkins <[email protected]>
