You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2023/03/28 06:43:52 UTC
Change in asterixdb[master]: WIP: BTree multi partitions modification
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17452 )
Change subject: WIP: BTree multi partitions modification
......................................................................
WIP: BTree multi partitions modification
Change-Id: I08da28f2a26fcaf581c2256312455fe541fae5ea
---
M hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.java
M hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorDescriptor.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IFrameTupleProcessor.java
M asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorDescriptor.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorNodePushable.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMTreeIndexAccessor.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorNodePushable.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/impls/LSMHarness.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorDescriptor.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorNodePushable.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorNodePushable.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMTreeInsertDeleteOperatorDescriptor.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorDescriptor.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor.java
M hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/ILSMHarness.java
24 files changed, 605 insertions(+), 232 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/52/17452/1
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
index f6caf5d..fbc01b4 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/bootstrap/TestNodeController.java
@@ -256,7 +256,7 @@
LSMPrimaryInsertOperatorNodePushable insertOp = new LSMPrimaryInsertOperatorNodePushable(ctx,
ctx.getTaskAttemptId().getTaskId().getPartition(), indexHelperFactory, pkIndexHelperFactory,
primaryIndexInfo.primaryIndexInsertFieldsPermutations, recordDesc, modOpCallbackFactory,
- searchOpCallbackFactory, primaryKeyIndexes.length, filterFields, null);
+ searchOpCallbackFactory, primaryKeyIndexes.length, filterFields, null, null, null);
// For now, this assumes a single secondary index. recordDesc is always <pk-record-meta>
// for the index, we will have to create an assign operator that extract the sk
// then the secondary LSMInsertDeleteOperatorNodePushable
@@ -307,7 +307,7 @@
LSMInsertDeleteOperatorNodePushable secondaryInsertOp = new LSMInsertDeleteOperatorNodePushable(ctx,
ctx.getTaskAttemptId().getTaskId().getPartition(), secondaryIndexInfo.insertFieldsPermutations,
secondaryIndexInfo.rDesc, IndexOperation.INSERT, false, secondaryIndexHelperFactory,
- secondaryModCallbackFactory, null, null);
+ secondaryModCallbackFactory, null, null, null, null);
assignOp.setOutputFrameWriter(0, secondaryInsertOp, secondaryIndexInfo.rDesc);
IPushRuntime commitOp =
@@ -356,7 +356,7 @@
LSMInsertDeleteOperatorNodePushable deleteOp =
new LSMInsertDeleteOperatorNodePushable(ctx, ctx.getTaskAttemptId().getTaskId().getPartition(),
primaryIndexInfo.primaryIndexInsertFieldsPermutations, recordDesc, IndexOperation.DELETE,
- true, indexHelperFactory, modOpCallbackFactory, null, null);
+ true, indexHelperFactory, modOpCallbackFactory, null, null, null, null);
// For now, this assumes a single secondary index. recordDesc is always <pk-record-meta>
// for the index, we will have to create an assign operator that extract the sk
// then the secondary LSMInsertDeleteOperatorNodePushable
@@ -407,7 +407,7 @@
LSMInsertDeleteOperatorNodePushable secondaryInsertOp = new LSMInsertDeleteOperatorNodePushable(ctx,
ctx.getTaskAttemptId().getTaskId().getPartition(), secondaryIndexInfo.insertFieldsPermutations,
secondaryIndexInfo.rDesc, IndexOperation.DELETE, false, secondaryIndexHelperFactory,
- secondaryModCallbackFactory, null, null);
+ secondaryModCallbackFactory, null, null, null, null);
assignOp.setOutputFrameWriter(0, secondaryInsertOp, secondaryIndexInfo.rDesc);
IPushRuntime commitOp =
@@ -813,7 +813,7 @@
modificationCallbackFactory, searchCallbackFactory, keyIndexes.length, 0, recordType, -1,
frameOpCallbackFactory == null ? dataset.getFrameOpCallbackFactory(mdProvider)
: frameOpCallbackFactory,
- MissingWriterFactory.INSTANCE, hasSecondaries, NoOpTupleProjectorFactory.INSTANCE);
+ MissingWriterFactory.INSTANCE, hasSecondaries, NoOpTupleProjectorFactory.INSTANCE, null, null);
RecordDescriptor upsertOutRecDesc = getUpsertOutRecDesc(primaryIndexInfo.rDesc, dataset,
filterFields == null ? 0 : filterFields.length, recordType, metaType);
// fix pk fields
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
index ab0732b..8f3126d 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMInsertDeleteOperatorNodePushable.java
@@ -25,25 +25,31 @@
import org.apache.asterix.common.transactions.PrimaryIndexLogMarkerCallback;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.exceptions.SourceLocation;
+import org.apache.hyracks.api.util.ExceptionUtils;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
import org.apache.hyracks.dataflow.common.data.accessors.FrameTupleReference;
import org.apache.hyracks.dataflow.common.utils.TaskUtil;
+import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.api.ITupleFilterFactory;
import org.apache.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
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.util.ResourceReleaseUtils;
+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.dataflow.LSMIndexInsertUpdateDeleteOperatorNodePushable;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
import org.apache.hyracks.storage.common.IIndexAccessParameters;
+import org.apache.hyracks.storage.common.LocalResource;
public class LSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUpdateDeleteOperatorNodePushable {
@@ -51,7 +57,8 @@
protected final boolean isPrimary;
protected final SourceLocation sourceLoc;
// This class has both lsmIndex and index (in super class) pointing to the same object
- private AbstractLSMIndex lsmIndex;
+ // private AbstractLSMIndex lsmIndex;
+ private AbstractLSMIndex[] lsmIndexes;
protected int i = 0;
/**
@@ -71,9 +78,10 @@
public LSMInsertDeleteOperatorNodePushable(IHyracksTaskContext ctx, int partition, int[] fieldPermutation,
RecordDescriptor inputRecDesc, IndexOperation op, boolean isPrimary,
IIndexDataflowHelperFactory indexHelperFactory, IModificationOperationCallbackFactory modCallbackFactory,
- ITupleFilterFactory tupleFilterFactory, SourceLocation sourceLoc) throws HyracksDataException {
+ ITupleFilterFactory tupleFilterFactory, SourceLocation sourceLoc,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) throws HyracksDataException {
super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, op, modCallbackFactory,
- tupleFilterFactory);
+ tupleFilterFactory, tuplePartitionerFactory, map);
this.isPrimary = isPrimary;
this.sourceLoc = sourceLoc;
}
@@ -87,25 +95,41 @@
accessor = new FrameTupleAccessor(inputRecDesc);
writeBuffer = new VSizeFrame(ctx);
appender = new FrameTupleAppender(writeBuffer);
- indexHelper.open();
- lsmIndex = (AbstractLSMIndex) indexHelper.getIndexInstance();
+ // indexHelper.open();
+ // lsmIndex = (AbstractLSMIndex) indexHelper.getIndexInstance();
try {
+ INcApplicationContext runtimeCtx =
+ (INcApplicationContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
+
+ for (int i = 0; i < indexHelpers.length; i++) {
+ IIndexDataflowHelper indexHelper = indexHelpers[i];
+ indexHelper.open();
+ indexes[i] = indexHelper.getIndexInstance();
+ LocalResource resource = indexHelper.getResource();
+ modCallbacks[i] = modOpCallbackFactory.createModificationOperationCallback(resource, ctx, this);
+ IIndexAccessParameters iap = new IndexAccessParameters(modCallbacks[i], NoOpOperationCallback.INSTANCE);
+ indexAccessors[i] = indexes[i].createAccessor(iap);
+ LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) indexes[i],
+ runtimeCtx.getTransactionSubsystem().getLogManager());
+ }
+
if (isPrimary && ctx.getSharedObject() != null) {
- PrimaryIndexLogMarkerCallback callback = new PrimaryIndexLogMarkerCallback(lsmIndex);
+ PrimaryIndexLogMarkerCallback callback = new PrimaryIndexLogMarkerCallback((ILSMIndex) indexes[0]);
TaskUtil.put(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
}
writer.open();
- modCallback =
- modOpCallbackFactory.createModificationOperationCallback(indexHelper.getResource(), ctx, this);
- IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
- indexAccessor = lsmIndex.createAccessor(iap);
+ writerOpen = true;
+ // modCallback =
+ // modOpCallbackFactory.createModificationOperationCallback(indexHelper.getResource(), ctx, this);
+ // IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
+ // indexAccessor = lsmIndex.createAccessor(iap);
if (tupleFilterFactory != null) {
tupleFilter = tupleFilterFactory.createTupleFilter(ctx);
frameTuple = new FrameTupleReference();
}
- INcApplicationContext runtimeCtx =
- (INcApplicationContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
- LSMIndexUtil.checkAndSetFirstLSN(lsmIndex, runtimeCtx.getTransactionSubsystem().getLogManager());
+ // INcApplicationContext runtimeCtx =
+ // (INcApplicationContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
+ // LSMIndexUtil.checkAndSetFirstLSN(lsmIndex, runtimeCtx.getTransactionSubsystem().getLogManager());
} catch (Throwable th) {
throw HyracksDataException.create(th);
}
@@ -114,7 +138,6 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
accessor.reset(buffer);
- ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessor;
int tupleCount = accessor.getTupleCount();
try {
for (; i < tupleCount; i++, currentTupleIdx++) {
@@ -125,6 +148,9 @@
}
}
tuple.reset(accessor, i);
+ int storagePartition = tuplePartitioner.partition(accessor, i);
+ int tupleIndex = storagePartitionId2Index.get(storagePartition);
+ ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessors[tupleIndex];
switch (op) {
case INSERT:
if (i == 0 && isPrimary) {
@@ -188,18 +214,32 @@
@Override
public void close() throws HyracksDataException {
- if (lsmIndex != null) {
+ Throwable failure = null;
+ for (IIndexDataflowHelper indexHelper : indexHelpers) {
+ failure = ResourceReleaseUtils.close(indexHelper, failure);
+ }
+ if (writerOpen) {
try {
- indexHelper.close();
- } finally {
writer.close();
+ } catch (Throwable th) {
+ failure = ExceptionUtils.suppress(failure, th);
}
}
+ if (failure != null) {
+ throw HyracksDataException.create(failure);
+ }
+ // if (lsmIndex != null) {
+ // try {
+ // indexHelper.close();
+ // } finally {
+ // writer.close();
+ // }
+ // }
}
@Override
public void fail() throws HyracksDataException {
- if (lsmIndex != null) {
+ if (writerOpen) {
writer.fail();
}
}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMTreeInsertDeleteOperatorDescriptor.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMTreeInsertDeleteOperatorDescriptor.java
index 2226ca0..bd8a5c9 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMTreeInsertDeleteOperatorDescriptor.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/dataflow/LSMTreeInsertDeleteOperatorDescriptor.java
@@ -21,6 +21,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
@@ -38,8 +39,10 @@
public LSMTreeInsertDeleteOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc,
int[] fieldPermutation, IndexOperation op, IIndexDataflowHelperFactory indexHelperFactory,
ITupleFilterFactory tupleFilterFactory, boolean isPrimary,
- IModificationOperationCallbackFactory modCallbackFactory) {
- super(spec, outRecDesc, indexHelperFactory, fieldPermutation, op, modCallbackFactory, tupleFilterFactory);
+ IModificationOperationCallbackFactory modCallbackFactory, ITuplePartitionerFactory tuplePartitionerFactory,
+ int[][] map) {
+ super(spec, outRecDesc, indexHelperFactory, fieldPermutation, op, modCallbackFactory, tupleFilterFactory,
+ tuplePartitionerFactory, map);
this.isPrimary = isPrimary;
}
@@ -48,7 +51,7 @@
IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
RecordDescriptor inputRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
return new LSMInsertDeleteOperatorNodePushable(ctx, partition, fieldPermutation, inputRecDesc, op, isPrimary,
- indexHelperFactory, modCallbackFactory, tupleFilterFactory, sourceLoc);
+ indexHelperFactory, modCallbackFactory, tupleFilterFactory, sourceLoc, tuplePartitionerFactory, map);
}
public boolean isPrimary() {
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
index 3692a8d..4883298 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/declared/MetadataProvider.java
@@ -1060,9 +1060,11 @@
int[] fieldPermutation = new int[numKeys + 1 + numFilterFields
+ (additionalNonFilteringFields == null ? 0 : additionalNonFilteringFields.size())];
int[] bloomFilterKeyFields = new int[numKeys];
+ int[] pkHashFields = new int[numKeys];
int i = 0;
for (LogicalVariable varKey : keys) {
int idx = propagatedSchema.findVariable(varKey);
+ pkHashFields[i] = idx;
fieldPermutation[i] = idx;
bloomFilterKeyFields[i] = i;
i++;
@@ -1097,6 +1099,19 @@
.getModificationCallbackFactory(storageComponentProvider, primaryIndex, indexOp, primaryKeyFields);
IIndexDataflowHelperFactory idfh =
new IndexDataflowHelperFactory(storageComponentProvider.getStorageManager(), splitsAndConstraint.first);
+
+ int numPartitions;
+ if (splitsAndConstraint.second
+ .getPartitionConstraintType() == AlgebricksPartitionConstraint.PartitionConstraintType.COUNT) {
+ numPartitions = ((AlgebricksCountPartitionConstraint) splitsAndConstraint.second).getCount();
+ } else {
+ numPartitions = ((AlgebricksAbsolutePartitionConstraint) splitsAndConstraint.second).getLocations().length;
+ }
+ int[][] partitionsMap = getPartitionsMap(numPartitions);
+ ITuplePartitionerFactory tuplePartitionerFactory = null;
+ IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(this);
+ tuplePartitionerFactory = new FieldHashPartitionerFactory(pkHashFields, pkHashFunFactories, numPartitions);
+
IOperatorDescriptor op;
if (bulkload) {
long numElementsHint = getCardinalityPerPartitionHint(dataset);
@@ -1119,11 +1134,12 @@
primaryKeySplitsAndConstraint.first);
}
op = createLSMPrimaryInsertOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, idfh, pkidfh,
- modificationCallbackFactory, searchCallbackFactory, numKeys, filterFields);
+ modificationCallbackFactory, searchCallbackFactory, numKeys, filterFields,
+ tuplePartitionerFactory, partitionsMap);
} else {
op = createLSMTreeInsertDeleteOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, indexOp, idfh,
- null, true, modificationCallbackFactory);
+ null, true, modificationCallbackFactory, tuplePartitionerFactory, partitionsMap);
}
}
return new Pair<>(op, splitsAndConstraint.second);
@@ -1132,18 +1148,21 @@
protected LSMPrimaryInsertOperatorDescriptor createLSMPrimaryInsertOperatorDescriptor(JobSpecification spec,
RecordDescriptor inputRecordDesc, int[] fieldPermutation, IIndexDataflowHelperFactory idfh,
IIndexDataflowHelperFactory pkidfh, IModificationOperationCallbackFactory modificationCallbackFactory,
- ISearchOperationCallbackFactory searchCallbackFactory, int numKeys, int[] filterFields) {
+ ISearchOperationCallbackFactory searchCallbackFactory, int numKeys, int[] filterFields,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] partitionsMap) {
// this can be used by extensions to pick up their own operators
return new LSMPrimaryInsertOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, idfh, pkidfh,
- modificationCallbackFactory, searchCallbackFactory, numKeys, filterFields);
+ modificationCallbackFactory, searchCallbackFactory, numKeys, filterFields, tuplePartitionerFactory,
+ partitionsMap);
}
protected LSMTreeInsertDeleteOperatorDescriptor createLSMTreeInsertDeleteOperatorDescriptor(
IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc, int[] fieldPermutation, IndexOperation op,
IIndexDataflowHelperFactory indexHelperFactory, ITupleFilterFactory tupleFilterFactory, boolean isPrimary,
- IModificationOperationCallbackFactory modCallbackFactory) {
+ IModificationOperationCallbackFactory modCallbackFactory, ITuplePartitionerFactory tuplePartitionerFactory,
+ int[][] partitionsMap) {
return new LSMTreeInsertDeleteOperatorDescriptor(spec, outRecDesc, fieldPermutation, op, indexHelperFactory,
- tupleFilterFactory, isPrimary, modCallbackFactory);
+ tupleFilterFactory, isPrimary, modCallbackFactory, tuplePartitionerFactory, partitionsMap);
}
private Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getIndexInsertOrDeleteOrUpsertRuntime(
@@ -1234,6 +1253,7 @@
// generate field permutations
int[] fieldPermutation = new int[numKeys + numFilterFields];
int[] modificationCallbackPrimaryKeyFields = new int[primaryKeys.size()];
+ int[] pkFields = new int[primaryKeys.size()];
int i = 0;
int j = 0;
for (LogicalVariable varKey : secondaryKeys) {
@@ -1244,6 +1264,7 @@
for (LogicalVariable varKey : primaryKeys) {
int idx = propagatedSchema.findVariable(varKey);
fieldPermutation[i] = idx;
+ pkFields[j] = idx;
modificationCallbackPrimaryKeyFields[j] = i;
i++;
j++;
@@ -1286,6 +1307,20 @@
storageComponentProvider, secondaryIndex, indexOp, modificationCallbackPrimaryKeyFields);
IIndexDataflowHelperFactory idfh = new IndexDataflowHelperFactory(
storageComponentProvider.getStorageManager(), splitsAndConstraint.first);
+
+ int numPartitions;
+ if (splitsAndConstraint.second
+ .getPartitionConstraintType() == AlgebricksPartitionConstraint.PartitionConstraintType.COUNT) {
+ numPartitions = ((AlgebricksCountPartitionConstraint) splitsAndConstraint.second).getCount();
+ } else {
+ numPartitions =
+ ((AlgebricksAbsolutePartitionConstraint) splitsAndConstraint.second).getLocations().length;
+ }
+ int[][] partitionsMap = getPartitionsMap(numPartitions);
+ ITuplePartitionerFactory tuplePartitionerFactory = null;
+ IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(this);
+ tuplePartitionerFactory = new FieldHashPartitionerFactory(pkFields, pkHashFunFactories, numPartitions);
+
IOperatorDescriptor op;
if (bulkload) {
long numElementsHint = getCardinalityPerPartitionHint(dataset);
@@ -1296,10 +1331,10 @@
int operationFieldIndex = propagatedSchema.findVariable(operationVar);
op = new LSMSecondaryUpsertOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, idfh,
filterFactory, prevFilterFactory, modificationCallbackFactory, operationFieldIndex,
- BinaryIntegerInspector.FACTORY, prevFieldPermutation);
+ BinaryIntegerInspector.FACTORY, prevFieldPermutation, tuplePartitionerFactory, partitionsMap);
} else {
op = new LSMTreeInsertDeleteOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, indexOp, idfh,
- filterFactory, false, modificationCallbackFactory);
+ filterFactory, false, modificationCallbackFactory, tuplePartitionerFactory, partitionsMap);
}
return new Pair<>(op, splitsAndConstraint.second);
} catch (Exception e) {
@@ -1320,11 +1355,13 @@
// Generate field permutations (this only includes primary keys and filter fields).
int[] fieldPermutation = new int[numPrimaryKeys + numFilterFields];
int[] modificationCallbackPrimaryKeyFields = new int[primaryKeys.size()];
+ int[] pkFields = new int[primaryKeys.size()];
int i = 0;
int j = 0;
for (LogicalVariable varKey : primaryKeys) {
int idx = propagatedSchema.findVariable(varKey);
fieldPermutation[i] = idx;
+ pkFields[j] = idx;
modificationCallbackPrimaryKeyFields[j] = i;
i++;
j++;
@@ -1345,15 +1382,31 @@
storageComponentProvider, secondaryIndex, indexOp, modificationCallbackPrimaryKeyFields);
IIndexDataflowHelperFactory idfh = new IndexDataflowHelperFactory(
storageComponentProvider.getStorageManager(), splitsAndConstraint.first);
+
+ int numPartitions;
+ if (splitsAndConstraint.second
+ .getPartitionConstraintType() == AlgebricksPartitionConstraint.PartitionConstraintType.COUNT) {
+ numPartitions = ((AlgebricksCountPartitionConstraint) splitsAndConstraint.second).getCount();
+ } else {
+ numPartitions =
+ ((AlgebricksAbsolutePartitionConstraint) splitsAndConstraint.second).getLocations().length;
+ }
+ int[][] partitionsMap = getPartitionsMap(numPartitions);
+ ITuplePartitionerFactory tuplePartitionerFactory = null;
+ IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(this);
+ tuplePartitionerFactory = new FieldHashPartitionerFactory(pkFields, pkHashFunFactories, numPartitions);
+
IOperatorDescriptor op;
if (indexOp == IndexOperation.UPSERT) {
int operationFieldIndex = propagatedSchema.findVariable(operationVar);
op = new LSMSecondaryUpsertWithNestedPlanOperatorDescriptor(spec, inputRecordDesc, fieldPermutation,
idfh, modificationCallbackFactory, operationFieldIndex, BinaryIntegerInspector.FACTORY,
- secondaryKeysPipelines.get(0), secondaryKeysPipelines.get(1));
+ secondaryKeysPipelines.get(0), secondaryKeysPipelines.get(1), tuplePartitionerFactory,
+ partitionsMap);
} else {
op = new LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor(spec, inputRecordDesc,
- fieldPermutation, indexOp, idfh, modificationCallbackFactory, secondaryKeysPipelines.get(0));
+ fieldPermutation, indexOp, idfh, modificationCallbackFactory, secondaryKeysPipelines.get(0),
+ tuplePartitionerFactory, partitionsMap);
}
return new Pair<>(op, splitsAndConstraint.second);
} catch (Exception e) {
@@ -1390,6 +1443,7 @@
int numFilterFields = DatasetUtil.getFilterField(dataset) == null ? 0 : 1;
int[] fieldPermutation = new int[numKeys + numFilterFields];
int[] modificationCallbackPrimaryKeyFields = new int[primaryKeys.size()];
+ int[] pkFields = new int[primaryKeys.size()];
int i = 0;
int j = 0;
@@ -1401,6 +1455,7 @@
for (LogicalVariable varKey : primaryKeys) {
int idx = propagatedSchema.findVariable(varKey);
fieldPermutation[i] = idx;
+ pkFields[j] = idx;
modificationCallbackPrimaryKeyFields[j] = i;
i++;
j++;
@@ -1441,6 +1496,19 @@
storageComponentProvider, secondaryIndex, indexOp, modificationCallbackPrimaryKeyFields);
IIndexDataflowHelperFactory indexDataflowHelperFactory =
new IndexDataflowHelperFactory(storageComponentProvider.getStorageManager(), splitsAndConstraint.first);
+
+ int numPartitions;
+ if (splitsAndConstraint.second
+ .getPartitionConstraintType() == AlgebricksPartitionConstraint.PartitionConstraintType.COUNT) {
+ numPartitions = ((AlgebricksCountPartitionConstraint) splitsAndConstraint.second).getCount();
+ } else {
+ numPartitions = ((AlgebricksAbsolutePartitionConstraint) splitsAndConstraint.second).getLocations().length;
+ }
+ int[][] partitionsMap = getPartitionsMap(numPartitions);
+ ITuplePartitionerFactory tuplePartitionerFactory = null;
+ IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(this);
+ tuplePartitionerFactory = new FieldHashPartitionerFactory(pkFields, pkHashFunFactories, numPartitions);
+
IOperatorDescriptor op;
if (bulkload) {
long numElementsHint = getCardinalityPerPartitionHint(dataset);
@@ -1451,10 +1519,12 @@
int operationFieldIndex = propagatedSchema.findVariable(operationVar);
op = new LSMSecondaryUpsertOperatorDescriptor(spec, recordDesc, fieldPermutation,
indexDataflowHelperFactory, filterFactory, prevFilterFactory, modificationCallbackFactory,
- operationFieldIndex, BinaryIntegerInspector.FACTORY, prevFieldPermutation);
+ operationFieldIndex, BinaryIntegerInspector.FACTORY, prevFieldPermutation, tuplePartitionerFactory,
+ partitionsMap);
} else {
op = new LSMTreeInsertDeleteOperatorDescriptor(spec, recordDesc, fieldPermutation, indexOp,
- indexDataflowHelperFactory, filterFactory, false, modificationCallbackFactory);
+ indexDataflowHelperFactory, filterFactory, false, modificationCallbackFactory,
+ tuplePartitionerFactory, partitionsMap);
}
return new Pair<>(op, splitsAndConstraint.second);
}
@@ -1492,6 +1562,7 @@
// generate field permutations
int[] fieldPermutation = new int[numKeys + numFilterFields];
int[] modificationCallbackPrimaryKeyFields = new int[primaryKeys.size()];
+ int[] pkFields = new int[primaryKeys.size()];
int i = 0;
int j = 0;
@@ -1505,6 +1576,7 @@
for (LogicalVariable varKey : primaryKeys) {
int idx = propagatedSchema.findVariable(varKey);
fieldPermutation[i] = idx;
+ pkFields[j] = idx;
modificationCallbackPrimaryKeyFields[j] = i;
i++;
j++;
@@ -1550,6 +1622,20 @@
storageComponentProvider, secondaryIndex, indexOp, modificationCallbackPrimaryKeyFields);
IIndexDataflowHelperFactory indexDataFlowFactory = new IndexDataflowHelperFactory(
storageComponentProvider.getStorageManager(), splitsAndConstraint.first);
+
+ int numPartitions;
+ if (splitsAndConstraint.second
+ .getPartitionConstraintType() == AlgebricksPartitionConstraint.PartitionConstraintType.COUNT) {
+ numPartitions = ((AlgebricksCountPartitionConstraint) splitsAndConstraint.second).getCount();
+ } else {
+ numPartitions =
+ ((AlgebricksAbsolutePartitionConstraint) splitsAndConstraint.second).getLocations().length;
+ }
+ int[][] partitionsMap = getPartitionsMap(numPartitions);
+ ITuplePartitionerFactory tuplePartitionerFactory = null;
+ IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(this);
+ tuplePartitionerFactory = new FieldHashPartitionerFactory(pkFields, pkHashFunFactories, numPartitions);
+
IOperatorDescriptor op;
if (bulkload) {
long numElementsHint = getCardinalityPerPartitionHint(dataset);
@@ -1560,10 +1646,11 @@
int upsertOperationFieldIndex = propagatedSchema.findVariable(operationVar);
op = new LSMSecondaryUpsertOperatorDescriptor(spec, recordDesc, fieldPermutation, indexDataFlowFactory,
filterFactory, prevFilterFactory, modificationCallbackFactory, upsertOperationFieldIndex,
- BinaryIntegerInspector.FACTORY, prevFieldPermutation);
+ BinaryIntegerInspector.FACTORY, prevFieldPermutation, tuplePartitionerFactory, partitionsMap);
} else {
op = new LSMTreeInsertDeleteOperatorDescriptor(spec, recordDesc, fieldPermutation, indexOp,
- indexDataFlowFactory, filterFactory, false, modificationCallbackFactory);
+ indexDataFlowFactory, filterFactory, false, modificationCallbackFactory,
+ tuplePartitionerFactory, partitionsMap);
}
return new Pair<>(op, splitsAndConstraint.second);
} catch (Exception e) {
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
index b5dd8bf..79b5766 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/utils/DatasetUtil.java
@@ -66,6 +66,8 @@
import org.apache.asterix.runtime.projection.DataProjectionFiltrationInfo;
import org.apache.asterix.runtime.utils.RuntimeUtils;
import org.apache.asterix.transaction.management.opcallbacks.PrimaryIndexInstantSearchOperationCallbackFactory;
+import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
+import org.apache.hyracks.algebricks.common.constraints.AlgebricksCountPartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraintHelper;
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
@@ -73,8 +75,10 @@
import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory;
import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -83,6 +87,7 @@
import org.apache.hyracks.data.std.util.ArrayBackedValueStorage;
import org.apache.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import org.apache.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import org.apache.hyracks.dataflow.common.data.partition.FieldHashPartitionerFactory;
import org.apache.hyracks.dataflow.std.file.IFileSplitProvider;
import org.apache.hyracks.dataflow.std.misc.ConstantTupleSourceOperatorDescriptor;
import org.apache.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
@@ -439,8 +444,10 @@
// prepare callback
int[] primaryKeyFields = new int[numKeys];
+ int[] pkHash = new int[numKeys];
for (int i = 0; i < numKeys; i++) {
primaryKeyFields[i] = i;
+ pkHash[i] = fieldPermutation[i];
}
boolean hasSecondaries =
metadataProvider.getDatasetIndexes(dataset.getDataverseName(), dataset.getDatasetName()).size() > 1;
@@ -506,13 +513,33 @@
ITupleProjectorFactory projectorFactory = IndexUtil.createUpsertTupleProjectorFactory(
dataset.getDatasetFormatInfo(), requestedType, itemType, metaItemType, numKeys);
+ int numPartitions;
+ if (splitsAndConstraint.second
+ .getPartitionConstraintType() == AlgebricksPartitionConstraint.PartitionConstraintType.COUNT) {
+ numPartitions = ((AlgebricksCountPartitionConstraint) splitsAndConstraint.second).getCount();
+ } else {
+ numPartitions = ((AlgebricksAbsolutePartitionConstraint) splitsAndConstraint.second).getLocations().length;
+ }
+ int[][] partitionsMap = getPartitionsMap(numPartitions);
+ ITuplePartitionerFactory tuplePartitionerFactory = null;
+ IBinaryHashFunctionFactory[] pkHashFunFactories = dataset.getPrimaryHashFunctionFactories(metadataProvider);
+ tuplePartitionerFactory = new FieldHashPartitionerFactory(pkHash, pkHashFunFactories, numPartitions);
+
op = new LSMPrimaryUpsertOperatorDescriptor(spec, outputRecordDesc, fieldPermutation, idfh,
missingWriterFactory, modificationCallbackFactory, searchCallbackFactory,
dataset.getFrameOpCallbackFactory(metadataProvider), numKeys, filterSourceIndicator, filterItemType,
- fieldIdx, hasSecondaries, projectorFactory);
+ fieldIdx, hasSecondaries, projectorFactory, tuplePartitionerFactory, partitionsMap);
return new Pair<>(op, splitsAndConstraint.second);
}
+ private static int[][] getPartitionsMap(int numPartitions) {
+ int[][] map = new int[numPartitions][1];
+ for (int i = 0; i < numPartitions; i++) {
+ map[i] = new int[] { i };
+ }
+ return map;
+ }
+
/**
* Returns a type that contains indexed fields for columnar datasets.
* The type is used retrieve the previous record with only the indexed fields -- minimizing the
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorDescriptor.java
index d639f3d..a48dc03 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorDescriptor.java
@@ -22,6 +22,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
@@ -43,9 +44,10 @@
int[] fieldPermutation, IIndexDataflowHelperFactory indexHelperFactory,
IIndexDataflowHelperFactory keyIndexHelperFactory,
IModificationOperationCallbackFactory modificationOpCallbackFactory,
- ISearchOperationCallbackFactory searchOpCallbackFactory, int numOfPrimaryKeys, int[] filterFields) {
+ ISearchOperationCallbackFactory searchOpCallbackFactory, int numOfPrimaryKeys, int[] filterFields,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) {
super(spec, outRecDesc, fieldPermutation, IndexOperation.UPSERT, indexHelperFactory, null, true,
- modificationOpCallbackFactory);
+ modificationOpCallbackFactory, tuplePartitionerFactory, map);
this.keyIndexHelperFactory = keyIndexHelperFactory;
this.searchOpCallbackFactory = searchOpCallbackFactory;
this.numOfPrimaryKeys = numOfPrimaryKeys;
@@ -58,6 +60,6 @@
RecordDescriptor intputRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
return new LSMPrimaryInsertOperatorNodePushable(ctx, partition, indexHelperFactory, keyIndexHelperFactory,
fieldPermutation, intputRecDesc, modCallbackFactory, searchOpCallbackFactory, numOfPrimaryKeys,
- filterFields, sourceLoc);
+ filterFields, sourceLoc, tuplePartitionerFactory, map);
}
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorNodePushable.java
index eadb614..c629d33 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryInsertOperatorNodePushable.java
@@ -28,6 +28,7 @@
import org.apache.asterix.transaction.management.opcallbacks.LockThenSearchOperationCallback;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -52,6 +53,7 @@
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.lsm.common.api.IFrameOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.IFrameTupleProcessor;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.am.lsm.common.dataflow.LSMIndexInsertUpdateDeleteOperatorNodePushable;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
@@ -60,38 +62,62 @@
import org.apache.hyracks.storage.common.IIndexCursor;
import org.apache.hyracks.storage.common.MultiComparator;
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.ints.IntSet;
+
public class LSMPrimaryInsertOperatorNodePushable extends LSMIndexInsertUpdateDeleteOperatorNodePushable {
- private final IIndexDataflowHelper keyIndexHelper;
+ // private final IIndexDataflowHelper keyIndexHelper;
+ private final IIndexDataflowHelper[] keyIndexHelpers;
private MultiComparator keySearchCmp;
private RangePredicate searchPred;
- private IIndexCursor cursor;
- private LockThenSearchOperationCallback searchCallback;
+ // private IIndexCursor cursor;
+ private final IIndexCursor[] cursors;
+ // private LockThenSearchOperationCallback searchCallback;
+ private final LockThenSearchOperationCallback[] searchCallbacks;
private final ISearchOperationCallbackFactory searchCallbackFactory;
private final IFrameTupleProcessor processor;
- private LSMTreeIndexAccessor lsmAccessor;
- private LSMTreeIndexAccessor lsmAccessorForKeyIndex;
- private LSMTreeIndexAccessor lsmAccessorForUniqunessCheck;
+ // private LSMTreeIndexAccessor lsmAccessor;
+ // private LSMTreeIndexAccessor lsmAccessorForKeyIndex;
+ private final LSMTreeIndexAccessor[] lsmAccessorForKeyIndexes;
+ // private LSMTreeIndexAccessor lsmAccessorForUniqunessCheck;
+ private final LSMTreeIndexAccessor[] lsmAccessorForUniqunessChecks;
- private final IFrameOperationCallback frameOpCallback;
+ // private final IFrameOperationCallback frameOpCallback;
+ private final IFrameOperationCallback[] frameOpCallbacks;
private boolean flushedPartialTuples;
private int currentTupleIdx;
private int lastFlushedTupleIdx;
private final PermutingFrameTupleReference keyTuple;
+ private final Int2ObjectMap<IntSet> partition2TuplesMap = new Int2ObjectOpenHashMap<>();
public LSMPrimaryInsertOperatorNodePushable(IHyracksTaskContext ctx, int partition,
IIndexDataflowHelperFactory indexHelperFactory, IIndexDataflowHelperFactory keyIndexHelperFactory,
int[] fieldPermutation, RecordDescriptor inputRecDesc,
IModificationOperationCallbackFactory modCallbackFactory,
ISearchOperationCallbackFactory searchCallbackFactory, int numOfPrimaryKeys, int[] filterFields,
- SourceLocation sourceLoc) throws HyracksDataException {
+ SourceLocation sourceLoc, ITuplePartitionerFactory tuplePartitionerFactory, int[][] map)
+ throws HyracksDataException {
super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, IndexOperation.UPSERT,
- modCallbackFactory, null);
- if (keyIndexHelperFactory != null) {
- this.keyIndexHelper = keyIndexHelperFactory.create(ctx.getJobletContext().getServiceContext(), partition);
- } else {
- this.keyIndexHelper = null;
+ modCallbackFactory, null, tuplePartitionerFactory, map);
+ this.frameOpCallbacks = new IFrameOperationCallback[partitions.length];
+ this.searchCallbacks = new LockThenSearchOperationCallback[partitions.length];
+ this.cursors = new IIndexCursor[partitions.length];
+ this.lsmAccessorForUniqunessChecks = new LSMTreeIndexAccessor[partitions.length];
+ this.lsmAccessorForKeyIndexes = new LSMTreeIndexAccessor[partitions.length];
+ this.keyIndexHelpers = new IIndexDataflowHelper[partitions.length];
+ for (int i = 0; i < partitions.length; i++) {
+ if (keyIndexHelperFactory != null) {
+ this.keyIndexHelpers[i] =
+ keyIndexHelperFactory.create(ctx.getJobletContext().getServiceContext(), partitions[i]);
+ } else {
+ this.keyIndexHelpers[i] = null;
+ }
+ frameOpCallbacks[i] = NoOpFrameOperationCallbackFactory.INSTANCE.createFrameOperationCallback(ctx,
+ (ILSMIndexAccessor) indexAccessors[i]);
}
this.searchCallbackFactory = searchCallbackFactory;
int numFilterFieds = filterFields != null ? filterFields.length : 0;
@@ -106,7 +132,6 @@
}
keyTuple = new PermutingFrameTupleReference(searchKeyPermutations);
processor = createTupleProcessor(sourceLoc);
- frameOpCallback = NoOpFrameOperationCallbackFactory.INSTANCE.createFrameOperationCallback(ctx, lsmAccessor);
}
protected void beforeModification(ITupleReference tuple) {
@@ -117,7 +142,8 @@
protected IFrameTupleProcessor createTupleProcessor(SourceLocation sourceLoc) {
return new IFrameTupleProcessor() {
@Override
- public void process(ITupleReference tuple, int index) throws HyracksDataException {
+ public void process(FrameTupleAccessor accessor, ITupleReference tuple, int index, int partition)
+ throws HyracksDataException {
if (index < currentTupleIdx) {
// already processed; skip
return;
@@ -126,21 +152,21 @@
searchPred.reset(keyTuple, keyTuple, true, true, keySearchCmp, keySearchCmp);
boolean duplicate = false;
- lsmAccessorForUniqunessCheck.search(cursor, searchPred);
+ lsmAccessorForUniqunessChecks[partition].search(cursors[partition], searchPred);
try {
- if (cursor.hasNext()) {
+ if (cursors[partition].hasNext()) {
// duplicate, skip
- searchCallback.release();
+ searchCallbacks[partition].release();
duplicate = true;
}
} finally {
- cursor.close();
+ cursors[partition].close();
}
if (!duplicate) {
beforeModification(tuple);
- lsmAccessor.forceUpsert(tuple);
- if (lsmAccessorForKeyIndex != null) {
- lsmAccessorForKeyIndex.forceUpsert(keyTuple);
+ ((ILSMIndexAccessor) indexAccessors[partition]).forceUpsert(tuple);
+ if (lsmAccessorForKeyIndexes[partition] != null) {
+ lsmAccessorForKeyIndexes[partition].forceUpsert(keyTuple);
}
} else {
// we should flush previous inserted records so that these transactions can commit
@@ -154,17 +180,17 @@
}
@Override
- public void start() throws HyracksDataException {
- lsmAccessor.getCtx().setOperation(IndexOperation.UPSERT);
+ public void start(int partition) throws HyracksDataException {
+ ((LSMTreeIndexAccessor) indexAccessors[partition]).getCtx().setOperation(IndexOperation.UPSERT);
}
@Override
- public void finish() throws HyracksDataException {
- lsmAccessor.getCtx().setOperation(IndexOperation.UPSERT);
+ public void finish(int partition) throws HyracksDataException {
+ ((LSMTreeIndexAccessor) indexAccessors[partition]).getCtx().setOperation(IndexOperation.UPSERT);
}
@Override
- public void fail(Throwable th) {
+ public void fail(Throwable th, int partition) {
// no op
}
};
@@ -177,48 +203,57 @@
flushedPartialTuples = false;
accessor = new FrameTupleAccessor(inputRecDesc);
writeBuffer = new VSizeFrame(ctx);
- indexHelper.open();
- index = indexHelper.getIndexInstance();
- IIndex indexForUniquessCheck;
- if (keyIndexHelper != null) {
- keyIndexHelper.open();
- indexForUniquessCheck = keyIndexHelper.getIndexInstance();
- } else {
- indexForUniquessCheck = index;
- }
try {
- if (ctx.getSharedObject() != null) {
- PrimaryIndexLogMarkerCallback callback = new PrimaryIndexLogMarkerCallback((AbstractLSMIndex) index);
- TaskUtil.put(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
- }
- frameOpCallback.open();
- writer.open();
- keySearchCmp =
- BTreeUtils.getSearchMultiComparator(((ITreeIndex) index).getComparatorFactories(), frameTuple);
- searchPred = new RangePredicate(frameTuple, frameTuple, true, true, keySearchCmp, keySearchCmp, null, null);
- appender = new FrameTupleAppender(new VSizeFrame(ctx), true);
- modCallback =
- modOpCallbackFactory.createModificationOperationCallback(indexHelper.getResource(), ctx, this);
- searchCallback = (LockThenSearchOperationCallback) searchCallbackFactory
- .createSearchOperationCallback(indexHelper.getResource().getId(), ctx, this);
- IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
- indexAccessor = index.createAccessor(iap);
- lsmAccessor = (LSMTreeIndexAccessor) indexAccessor;
- if (keyIndexHelper != null) {
- lsmAccessorForKeyIndex = (LSMTreeIndexAccessor) indexForUniquessCheck.createAccessor(iap);
- }
-
- IIndexAccessParameters iapForUniquenessCheck =
- new IndexAccessParameters(NoOpOperationCallback.INSTANCE, searchCallback);
- lsmAccessorForUniqunessCheck =
- (LSMTreeIndexAccessor) indexForUniquessCheck.createAccessor(iapForUniquenessCheck);
-
- cursor = lsmAccessorForUniqunessCheck.createSearchCursor(false);
- frameTuple = new FrameTupleReference();
INcApplicationContext appCtx =
(INcApplicationContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
- LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) index,
- appCtx.getTransactionSubsystem().getLogManager());
+ writer.open();
+ writerOpen = true;
+
+ for (int i = 0; i < partitions.length; i++) {
+ IIndexDataflowHelper indexHelper = indexHelpers[i];
+ indexHelper.open();
+ indexes[i] = indexHelper.getIndexInstance();
+ IIndex index = indexes[i];
+ IIndexDataflowHelper keyIndexHelper = keyIndexHelpers[i];
+ IIndex indexForUniquessCheck;
+ if (keyIndexHelper != null) {
+ keyIndexHelper.open();
+ indexForUniquessCheck = keyIndexHelper.getIndexInstance();
+ } else {
+ indexForUniquessCheck = index;
+ }
+ if (ctx.getSharedObject() != null) {
+ PrimaryIndexLogMarkerCallback callback =
+ new PrimaryIndexLogMarkerCallback((AbstractLSMIndex) index);
+ TaskUtil.put(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
+ }
+ frameOpCallbacks[i].open();
+
+ modCallbacks[i] =
+ modOpCallbackFactory.createModificationOperationCallback(indexHelper.getResource(), ctx, this);
+ searchCallbacks[i] = (LockThenSearchOperationCallback) searchCallbackFactory
+ .createSearchOperationCallback(indexHelper.getResource().getId(), ctx, this);
+ IIndexAccessParameters iap = new IndexAccessParameters(modCallbacks[i], NoOpOperationCallback.INSTANCE);
+ indexAccessors[i] = index.createAccessor(iap);
+ // lsmAccessor = (LSMTreeIndexAccessor) indexAccessors[i];
+ if (keyIndexHelper != null) {
+ lsmAccessorForKeyIndexes[i] = (LSMTreeIndexAccessor) indexForUniquessCheck.createAccessor(iap);
+ }
+
+ IIndexAccessParameters iapForUniquenessCheck =
+ new IndexAccessParameters(NoOpOperationCallback.INSTANCE, searchCallbacks[i]);
+ lsmAccessorForUniqunessChecks[i] =
+ (LSMTreeIndexAccessor) indexForUniquessCheck.createAccessor(iapForUniquenessCheck);
+
+ cursors[i] = lsmAccessorForUniqunessChecks[i].createSearchCursor(false);
+ LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) index,
+ appCtx.getTransactionSubsystem().getLogManager());
+ }
+ keySearchCmp =
+ BTreeUtils.getSearchMultiComparator(((ITreeIndex) indexes[0]).getComparatorFactories(), frameTuple);
+ searchPred = new RangePredicate(frameTuple, frameTuple, true, true, keySearchCmp, keySearchCmp, null, null);
+ appender = new FrameTupleAppender(new VSizeFrame(ctx), true);
+ frameTuple = new FrameTupleReference();
} catch (Throwable e) { // NOSONAR: Re-thrown
throw HyracksDataException.create(e);
}
@@ -227,7 +262,20 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
accessor.reset(buffer);
- lsmAccessor.batchOperate(accessor, tuple, processor, frameOpCallback);
+ partition2TuplesMap.clear();
+ int itemCount = accessor.getTupleCount();
+ for (int i = 0; i < itemCount; i++) {
+ int storagePartition = tuplePartitioner.partition(accessor, i);
+ int pIdx = storagePartitionId2Index.get(storagePartition);
+ IntSet tupleIndexes = partition2TuplesMap.computeIfAbsent(pIdx, k -> new IntOpenHashSet());
+ tupleIndexes.add(i);
+ }
+ for (Int2ObjectMap.Entry<IntSet> p2tuplesMapEntry : partition2TuplesMap.int2ObjectEntrySet()) {
+ int pIdx = p2tuplesMapEntry.getIntKey();
+ LSMTreeIndexAccessor lsmAccessor = (LSMTreeIndexAccessor) indexAccessors[pIdx];
+ IFrameOperationCallback frameOpCallback = frameOpCallbacks[pIdx];
+ lsmAccessor.batchOperate(accessor, tuple, processor, frameOpCallback, p2tuplesMapEntry.getValue(), pIdx);
+ }
writeBuffer.ensureFrameSize(buffer.capacity());
if (flushedPartialTuples) {
@@ -260,10 +308,10 @@
@Override
public void close() throws HyracksDataException {
- Throwable failure = CleanupUtils.destroy(null, cursor);
+ Throwable failure = CleanupUtils.destroy(null, cursors);
failure = CleanupUtils.close(writer, failure);
- failure = CleanupUtils.close(indexHelper, failure);
- failure = CleanupUtils.close(keyIndexHelper, failure);
+ failure = CleanupUtils.close(indexHelpers, failure);
+ failure = CleanupUtils.close(keyIndexHelpers, failure);
if (failure != null) {
throw HyracksDataException.create(failure);
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorDescriptor.java
index 3f1cd15..2c7db62 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMPrimaryUpsertOperatorDescriptor.java
@@ -24,6 +24,7 @@
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
@@ -46,6 +47,8 @@
protected final IMissingWriterFactory missingWriterFactory;
protected final boolean hasSecondaries;
private final ITupleProjectorFactory projectorFactory;
+ private final ITuplePartitionerFactory partitionerFactory;
+ private final int[][] map;
public LSMPrimaryUpsertOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc,
int[] fieldPermutation, IIndexDataflowHelperFactory indexHelperFactory,
@@ -54,9 +57,9 @@
ISearchOperationCallbackFactory searchOpCallbackFactory,
IFrameOperationCallbackFactory frameOpCallbackFactory, int numPrimaryKeys, Integer filterSourceIndicator,
ARecordType filterItemType, int filterIndex, boolean hasSecondaries,
- ITupleProjectorFactory projectorFactory) {
+ ITupleProjectorFactory projectorFactory, ITuplePartitionerFactory partitionerFactory, int[][] map) {
super(spec, outRecDesc, fieldPermutation, IndexOperation.UPSERT, indexHelperFactory, null, true,
- modificationOpCallbackFactory);
+ modificationOpCallbackFactory, partitionerFactory, map);
this.frameOpCallbackFactory = frameOpCallbackFactory;
this.searchOpCallbackFactory = searchOpCallbackFactory;
this.numPrimaryKeys = numPrimaryKeys;
@@ -66,6 +69,8 @@
this.filterIndex = filterIndex;
this.hasSecondaries = hasSecondaries;
this.projectorFactory = projectorFactory;
+ this.partitionerFactory = partitionerFactory;
+ this.map = map;
}
@Override
@@ -75,6 +80,6 @@
return new LSMPrimaryUpsertOperatorNodePushable(ctx, partition, indexHelperFactory, fieldPermutation,
intputRecDesc, modCallbackFactory, searchOpCallbackFactory, numPrimaryKeys, filterSourceIndicator,
filterItemType, filterIndex, frameOpCallbackFactory, missingWriterFactory, hasSecondaries,
- projectorFactory);
+ projectorFactory, partitionerFactory, map);
}
}
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 5172cde..ebee165 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
@@ -42,6 +42,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IMissingWriter;
import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.util.CleanupUtils;
@@ -57,6 +58,7 @@
import org.apache.hyracks.dataflow.common.utils.TaskUtil;
import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
import org.apache.hyracks.storage.am.btree.util.BTreeUtils;
+import org.apache.hyracks.storage.am.common.api.IIndexDataflowHelper;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.api.ISearchOperationCallbackFactory;
import org.apache.hyracks.storage.am.common.api.ITreeIndex;
@@ -66,10 +68,10 @@
import org.apache.hyracks.storage.am.lsm.common.api.IFrameOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.IFrameOperationCallbackFactory;
import org.apache.hyracks.storage.am.lsm.common.api.IFrameTupleProcessor;
-import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessor;
import org.apache.hyracks.storage.am.lsm.common.dataflow.LSMIndexInsertUpdateDeleteOperatorNodePushable;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
+import org.apache.hyracks.storage.common.IIndex;
import org.apache.hyracks.storage.common.IIndexAccessParameters;
import org.apache.hyracks.storage.common.IIndexCursor;
import org.apache.hyracks.storage.common.MultiComparator;
@@ -81,6 +83,11 @@
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import it.unimi.dsi.fastutil.ints.Int2ObjectMap;
+import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.ints.IntSet;
+
public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDeleteOperatorNodePushable {
public static final AInt8 UPSERT_NEW = new AInt8((byte) 0);
@@ -96,7 +103,8 @@
protected ArrayTupleBuilder tb;
private DataOutput dos;
protected RangePredicate searchPred;
- protected IIndexCursor cursor;
+ // protected IIndexCursor cursor;
+ protected final IIndexCursor[] cursors;
protected ITupleReference prevTuple;
protected final int numOfPrimaryKeys;
protected boolean isFiltered = false;
@@ -110,27 +118,33 @@
private final boolean hasMeta;
private final int filterFieldIndex;
private final int metaFieldIndex;
- protected LockThenSearchOperationCallback searchCallback;
- protected IFrameOperationCallback frameOpCallback;
+ // protected LockThenSearchOperationCallback searchCallback;
+ protected final LockThenSearchOperationCallback[] searchCallbacks;
+ // protected IFrameOperationCallback frameOpCallback;
+ protected final IFrameOperationCallback[] frameOpCallbacks;
private final IFrameOperationCallbackFactory frameOpCallbackFactory;
- protected AbstractIndexModificationOperationCallback abstractModCallback;
+ // protected AbstractIndexModificationOperationCallback abstractModCallback;
private final ISearchOperationCallbackFactory searchCallbackFactory;
private final IFrameTupleProcessor processor;
- protected LSMTreeIndexAccessor lsmAccessor;
+ // protected LSMTreeIndexAccessor lsmAccessor;
private final ITracer tracer;
private final long traceCategory;
private final ITupleProjector tupleProjector;
private long lastRecordInTimeStamp = 0L;
+ private final Int2ObjectMap<IntSet> partition2TuplesMap = new Int2ObjectOpenHashMap<>();
public LSMPrimaryUpsertOperatorNodePushable(IHyracksTaskContext ctx, int partition,
IIndexDataflowHelperFactory indexHelperFactory, int[] fieldPermutation, RecordDescriptor inputRecDesc,
IModificationOperationCallbackFactory modCallbackFactory,
ISearchOperationCallbackFactory searchCallbackFactory, int numOfPrimaryKeys, Integer filterSourceIndicator,
ARecordType filterItemType, int filterFieldIndex, IFrameOperationCallbackFactory frameOpCallbackFactory,
- IMissingWriterFactory missingWriterFactory, boolean hasSecondaries, ITupleProjectorFactory projectorFactory)
- throws HyracksDataException {
+ IMissingWriterFactory missingWriterFactory, boolean hasSecondaries, ITupleProjectorFactory projectorFactory,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) throws HyracksDataException {
super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, IndexOperation.UPSERT,
- modCallbackFactory, null);
+ modCallbackFactory, null, tuplePartitionerFactory, map);
+ this.frameOpCallbacks = new IFrameOperationCallback[partitions.length];
+ this.searchCallbacks = new LockThenSearchOperationCallback[partitions.length];
+ this.cursors = new IIndexCursor[partitions.length];
this.key = new PermutingFrameTupleReference();
this.searchCallbackFactory = searchCallbackFactory;
this.numOfPrimaryKeys = numOfPrimaryKeys;
@@ -166,9 +180,15 @@
protected IFrameTupleProcessor createTupleProcessor(final boolean hasSecondaries) {
return new IFrameTupleProcessor() {
@Override
- public void process(ITupleReference tuple, int index) throws HyracksDataException {
+ public void process(FrameTupleAccessor accessor, ITupleReference tuple, int index, int partition)
+ throws HyracksDataException {
try {
tb.reset();
+ LSMTreeIndexAccessor lsmAccessor = (LSMTreeIndexAccessor) indexAccessors[partition];
+ IIndexCursor cursor = cursors[partition];
+ LockThenSearchOperationCallback searchCallback = searchCallbacks[partition];
+ AbstractIndexModificationOperationCallback abstractModCallback =
+ (AbstractIndexModificationOperationCallback) modCallbacks[partition];
boolean recordWasInserted = false;
boolean recordWasDeleted = false;
boolean isDelete = isDeleteOperation(tuple, numOfPrimaryKeys);
@@ -213,26 +233,26 @@
// need to update the filter of the new component with the previous value
lsmAccessor.updateFilter(prevTuple);
}
- writeOutput(index, recordWasInserted, recordWasDeleted);
+ writeOutput(index, recordWasInserted, recordWasDeleted, searchCallback);
} catch (Exception e) {
throw HyracksDataException.create(e);
}
}
@Override
- public void start() throws HyracksDataException {
- lsmAccessor.getCtx().setOperation(IndexOperation.UPSERT);
+ public void start(int partition) throws HyracksDataException {
+ ((LSMTreeIndexAccessor) indexAccessors[partition]).getCtx().setOperation(IndexOperation.UPSERT);
}
@Override
- public void finish() throws HyracksDataException {
- lsmAccessor.getCtx().setOperation(IndexOperation.UPSERT);
+ public void finish(int partition) throws HyracksDataException {
+ ((LSMTreeIndexAccessor) indexAccessors[partition]).getCtx().setOperation(IndexOperation.UPSERT);
}
@Override
- public void fail(Throwable th) {
+ public void fail(Throwable th, int partition) {
// We must fail before we exit the components
- frameOpCallback.fail(th);
+ frameOpCallbacks[partition].fail(th);
}
};
}
@@ -247,13 +267,8 @@
accessor = new FrameTupleAccessor(inputRecDesc);
writeBuffer = new VSizeFrame(ctx);
writer.open();
- indexHelper.open();
- index = indexHelper.getIndexInstance();
+ writerOpen = true;
try {
- if (ctx.getSharedObject() != null) {
- PrimaryIndexLogMarkerCallback callback = new PrimaryIndexLogMarkerCallback((AbstractLSMIndex) index);
- TaskUtil.put(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
- }
missingTupleBuilder = new ArrayTupleBuilder(1);
DataOutput out = missingTupleBuilder.getDataOutput();
try {
@@ -262,31 +277,52 @@
throw HyracksDataException.create(e);
}
missingTupleBuilder.addFieldEndOffset();
- searchPred = createSearchPredicate();
tb = new ArrayTupleBuilder(recordDesc.getFieldCount());
dos = tb.getDataOutput();
appender = new FrameTupleAppender(new VSizeFrame(ctx), true);
- modCallback =
- modOpCallbackFactory.createModificationOperationCallback(indexHelper.getResource(), ctx, this);
- abstractModCallback = (AbstractIndexModificationOperationCallback) modCallback;
- searchCallback = (LockThenSearchOperationCallback) searchCallbackFactory
- .createSearchOperationCallback(indexHelper.getResource().getId(), ctx, this);
- IIndexAccessParameters iap = new IndexAccessParameters(abstractModCallback, searchCallback);
- iap.getParameters().put(HyracksConstants.TUPLE_PROJECTOR, tupleProjector);
- indexAccessor = index.createAccessor(iap);
- lsmAccessor = (LSMTreeIndexAccessor) indexAccessor;
- cursor = indexAccessor.createSearchCursor(false);
- frameTuple = new FrameTupleReference();
INcApplicationContext appCtx =
(INcApplicationContext) ctx.getJobletContext().getServiceContext().getApplicationContext();
- LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) index,
- appCtx.getTransactionSubsystem().getLogManager());
- frameOpCallback = new IFrameOperationCallback() {
+ for (int i = 0; i < indexHelpers.length; i++) {
+ IIndexDataflowHelper indexHelper = indexHelpers[i];
+ indexHelper.open();
+ indexes[i] = indexHelper.getIndexInstance();
+ //TODO(fix)
+ if (ctx.getSharedObject() != null) {
+ PrimaryIndexLogMarkerCallback callback =
+ new PrimaryIndexLogMarkerCallback((AbstractLSMIndex) indexes[i]);
+ TaskUtil.put(ILogMarkerCallback.KEY_MARKER_CALLBACK, callback, ctx);
+ }
+ modCallbacks[i] =
+ modOpCallbackFactory.createModificationOperationCallback(indexHelper.getResource(), ctx, this);
+ searchCallbacks[i] = (LockThenSearchOperationCallback) searchCallbackFactory
+ .createSearchOperationCallback(indexHelper.getResource().getId(), ctx, this);
+ IIndexAccessParameters iap = new IndexAccessParameters(
+ (AbstractIndexModificationOperationCallback) modCallbacks[i], searchCallbacks[i]);
+ iap.getParameters().put(HyracksConstants.TUPLE_PROJECTOR, tupleProjector);
+ indexAccessors[i] = indexes[i].createAccessor(iap);
+ LSMTreeIndexAccessor indexAccessor = (LSMTreeIndexAccessor) indexAccessors[i];
+ cursors[i] = indexAccessor.createSearchCursor(false);
+ LSMIndexUtil.checkAndSetFirstLSN((AbstractLSMIndex) indexes[i],
+ appCtx.getTransactionSubsystem().getLogManager());
+ }
+ searchPred = createSearchPredicate(indexes[0]);
+ frameTuple = new FrameTupleReference();
+ createFrameOpCallbacks();
+ } catch (Throwable e) { // NOSONAR: Re-thrown
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ private void createFrameOpCallbacks() throws HyracksDataException {
+ for (int i = 0; i < partitions.length; i++) {
+ LSMTreeIndexAccessor lsmAccessor = (LSMTreeIndexAccessor) indexAccessors[i];
+ frameOpCallbacks[i] = new IFrameOperationCallback() {
final IFrameOperationCallback callback =
- frameOpCallbackFactory.createFrameOperationCallback(ctx, (ILSMIndexAccessor) indexAccessor);
+ frameOpCallbackFactory.createFrameOperationCallback(ctx, lsmAccessor);
@Override
public void frameCompleted() throws HyracksDataException {
+ //TODO: mixed-frame vs frame-per-storage-partition
appender.write(writer, true);
callback.frameCompleted();
}
@@ -306,9 +342,7 @@
callback.open();
}
};
- frameOpCallback.open();
- } catch (Throwable e) { // NOSONAR: Re-thrown
- throw HyracksDataException.create(e);
+ frameOpCallbacks[i].open();
}
}
@@ -317,7 +351,8 @@
searchPred.reset(key, key, true, true, keySearchCmp, keySearchCmp);
}
- protected void writeOutput(int tupleIndex, boolean recordWasInserted, boolean recordWasDeleted) throws IOException {
+ protected void writeOutput(int tupleIndex, boolean recordWasInserted, boolean recordWasDeleted,
+ LockThenSearchOperationCallback searchCallback) throws IOException {
if (recordWasInserted || recordWasDeleted) {
frameTuple.reset(accessor, tupleIndex);
for (int i = 0; i < frameTuple.getFieldCount(); i++) {
@@ -346,8 +381,24 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
accessor.reset(buffer);
+ partition2TuplesMap.clear();
int itemCount = accessor.getTupleCount();
- lsmAccessor.batchOperate(accessor, tuple, processor, frameOpCallback);
+ for (int i = 0; i < itemCount; i++) {
+ int storagePartition = tuplePartitioner.partition(accessor, i);
+ int pIdx = storagePartitionId2Index.get(storagePartition);
+ IntSet tupleIndexes = partition2TuplesMap.computeIfAbsent(pIdx, k -> new IntOpenHashSet());
+ tupleIndexes.add(i);
+ }
+ // to ensure all partitions will be processed at least once, add partitions with missing tuples
+ for (int partition : storagePartitionId2Index.values()) {
+ partition2TuplesMap.computeIfAbsent(partition, k -> new IntOpenHashSet());
+ }
+ for (Int2ObjectMap.Entry<IntSet> p2tuplesMapEntry : partition2TuplesMap.int2ObjectEntrySet()) {
+ int pIdx = p2tuplesMapEntry.getIntKey();
+ LSMTreeIndexAccessor lsmAccessor = (LSMTreeIndexAccessor) indexAccessors[pIdx];
+ IFrameOperationCallback frameOpCallback = frameOpCallbacks[pIdx];
+ lsmAccessor.batchOperate(accessor, tuple, processor, frameOpCallback, p2tuplesMapEntry.getValue(), pIdx);
+ }
if (itemCount > 0) {
lastRecordInTimeStamp = System.currentTimeMillis();
}
@@ -441,7 +492,7 @@
}
}
- private RangePredicate createSearchPredicate() {
+ private RangePredicate createSearchPredicate(IIndex index) {
keySearchCmp = BTreeUtils.getSearchMultiComparator(((ITreeIndex) index).getComparatorFactories(), key);
return new RangePredicate(key, key, true, true, keySearchCmp, keySearchCmp, null, null);
}
@@ -449,10 +500,10 @@
@Override
public void close() throws HyracksDataException {
traceLastRecordIn();
- Throwable failure = CleanupUtils.close(frameOpCallback, null);
- failure = CleanupUtils.destroy(failure, cursor);
+ Throwable failure = CleanupUtils.close(frameOpCallbacks, null);
+ failure = CleanupUtils.destroy(failure, cursors);
failure = CleanupUtils.close(writer, failure);
- failure = CleanupUtils.close(indexHelper, failure);
+ failure = CleanupUtils.close(indexHelpers, failure);
if (failure != null) {
throw HyracksDataException.create(failure);
}
@@ -461,11 +512,11 @@
@SuppressWarnings({ "squid:S1181", "squid:S1166" })
private void traceLastRecordIn() {
try {
- if (tracer.isEnabled(traceCategory) && lastRecordInTimeStamp > 0 && indexHelper != null
- && indexHelper.getIndexInstance() != null) {
+ if (tracer.isEnabled(traceCategory) && lastRecordInTimeStamp > 0 && indexHelpers[0] != null
+ && indexHelpers[0].getIndexInstance() != null) {
tracer.instant("UpsertClose", traceCategory, Scope.t,
() -> "{\"last-record-in\":\"" + DATE_FORMAT.get().format(new Date(lastRecordInTimeStamp))
- + "\", \"index\":" + indexHelper.getIndexInstance().toString() + "}");
+ + "\", \"index\":" + indexHelpers[0].getIndexInstance().toString() + "}");
}
} catch (Throwable traceFailure) {
try {
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor.java
index ae03be2..b7129c4 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor.java
@@ -24,6 +24,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.JobSpecification;
@@ -39,8 +40,10 @@
public LSMSecondaryInsertDeleteWithNestedPlanOperatorDescriptor(JobSpecification spec, RecordDescriptor outRecDesc,
int[] fieldPermutation, IndexOperation op, IIndexDataflowHelperFactory indexHelperFactory,
- IModificationOperationCallbackFactory modCallbackFactory, List<AlgebricksPipeline> secondaryKeysPipeline) {
- super(spec, outRecDesc, indexHelperFactory, fieldPermutation, op, modCallbackFactory, null);
+ IModificationOperationCallbackFactory modCallbackFactory, List<AlgebricksPipeline> secondaryKeysPipeline,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) {
+ super(spec, outRecDesc, indexHelperFactory, fieldPermutation, op, modCallbackFactory, null,
+ tuplePartitionerFactory, map);
this.secondaryKeysPipeline = secondaryKeysPipeline;
}
@@ -49,6 +52,7 @@
IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
RecordDescriptor inputRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
return new LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable(ctx, partition, fieldPermutation,
- inputRecDesc, op, indexHelperFactory, modCallbackFactory, secondaryKeysPipeline);
+ inputRecDesc, op, indexHelperFactory, modCallbackFactory, secondaryKeysPipeline,
+ tuplePartitionerFactory, map);
}
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable.java
index 5712991..efc333a 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable.java
@@ -29,6 +29,7 @@
import org.apache.hyracks.algebricks.runtime.operators.std.NestedTupleSourceRuntimeFactory.NestedTupleSourceRuntime;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
@@ -49,8 +50,10 @@
public LSMSecondaryInsertDeleteWithNestedPlanOperatorNodePushable(IHyracksTaskContext ctx, int partition,
int[] fieldPermutation, RecordDescriptor inputRecDesc, IndexOperation op,
IIndexDataflowHelperFactory indexHelperFactory, IModificationOperationCallbackFactory modCallbackFactory,
- List<AlgebricksPipeline> secondaryKeysPipeline) throws HyracksDataException {
- super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, op, modCallbackFactory, null);
+ List<AlgebricksPipeline> secondaryKeysPipeline, ITuplePartitionerFactory tuplePartitionerFactory,
+ int[][] map) throws HyracksDataException {
+ super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, op, modCallbackFactory, null,
+ tuplePartitionerFactory, map);
this.numberOfPrimaryKeyAndFilterFields = fieldPermutation.length;
// Build our pipeline.
@@ -138,7 +141,7 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- ILSMIndexAccessor workingLSMAccessor = (ILSMIndexAccessor) indexAccessor;
+ // ILSMIndexAccessor workingLSMAccessor = (ILSMIndexAccessor) indexAccessors[0];
endOfPipelineTupleAccessor.reset(buffer);
int nTuple = endOfPipelineTupleAccessor.getTupleCount();
@@ -152,6 +155,9 @@
// Add the primary keys and filter fields.
endTupleReference.addTuple(tuple);
+ int storagePartition = tuplePartitioner.partition(accessor, t);
+ int partition = storagePartitionId2Index.get(storagePartition);
+ ILSMIndexAccessor workingLSMAccessor = (ILSMIndexAccessor) indexAccessors[partition];
// Pass the tuple to our accessor. There are only two operations: insert or delete.
if (op.equals(IndexOperation.INSERT)) {
try {
@@ -161,7 +167,6 @@
throw e;
}
}
-
} else {
try {
workingLSMAccessor.forceDelete(endTupleReference);
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorDescriptor.java
index 9de0827..c0fa313 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorDescriptor.java
@@ -23,6 +23,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
@@ -43,9 +44,10 @@
int[] fieldPermutation, IIndexDataflowHelperFactory indexHelperFactory,
ITupleFilterFactory tupleFilterFactory, ITupleFilterFactory prevTupleFilterFactory,
IModificationOperationCallbackFactory modificationOpCallbackFactory, int operationFieldIndex,
- IBinaryIntegerInspectorFactory operationInspectorFactory, int[] prevValuePermutation) {
+ IBinaryIntegerInspectorFactory operationInspectorFactory, int[] prevValuePermutation,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) {
super(spec, outRecDesc, fieldPermutation, IndexOperation.UPSERT, indexHelperFactory, tupleFilterFactory, false,
- modificationOpCallbackFactory);
+ modificationOpCallbackFactory, tuplePartitionerFactory, map);
this.prevValuePermutation = prevValuePermutation;
this.operationFieldIndex = operationFieldIndex;
this.operationInspectorFactory = operationInspectorFactory;
@@ -58,6 +60,6 @@
RecordDescriptor intputRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
return new LSMSecondaryUpsertOperatorNodePushable(ctx, partition, indexHelperFactory, modCallbackFactory,
tupleFilterFactory, prevTupleFilterFactory, fieldPermutation, intputRecDesc, operationFieldIndex,
- operationInspectorFactory, prevValuePermutation);
+ operationInspectorFactory, prevValuePermutation, tuplePartitionerFactory, map);
}
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorNodePushable.java
index 955d5aa..9960320 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertOperatorNodePushable.java
@@ -27,6 +27,7 @@
import org.apache.hyracks.algebricks.data.IBinaryIntegerInspector;
import org.apache.hyracks.algebricks.data.IBinaryIntegerInspectorFactory;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.util.FrameUtils;
@@ -63,16 +64,16 @@
protected final int operationFieldIndex;
protected final IBinaryIntegerInspector operationInspector;
- protected AbstractIndexModificationOperationCallback abstractModCallback;
+ // protected AbstractIndexModificationOperationCallback abstractModCallback;
public LSMSecondaryUpsertOperatorNodePushable(IHyracksTaskContext ctx, int partition,
IIndexDataflowHelperFactory indexHelperFactory, IModificationOperationCallbackFactory modCallbackFactory,
ITupleFilterFactory tupleFilterFactory, ITupleFilterFactory prevTupleFilterFactory, int[] fieldPermutation,
RecordDescriptor inputRecDesc, int operationFieldIndex,
- IBinaryIntegerInspectorFactory operationInspectorFactory, int[] prevTuplePermutation)
- throws HyracksDataException {
+ IBinaryIntegerInspectorFactory operationInspectorFactory, int[] prevTuplePermutation,
+ ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) throws HyracksDataException {
super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, IndexOperation.UPSERT,
- modCallbackFactory, tupleFilterFactory);
+ modCallbackFactory, tupleFilterFactory, tuplePartitionerFactory, map);
this.prevTuple.setFieldPermutation(prevTuplePermutation);
this.operationFieldIndex = operationFieldIndex;
this.operationInspector = operationInspectorFactory.createBinaryIntegerInspector(ctx);
@@ -85,7 +86,7 @@
public void open() throws HyracksDataException {
super.open();
frameTuple = new FrameTupleReference();
- abstractModCallback = (AbstractIndexModificationOperationCallback) modCallback;
+ // abstractModCallback = (AbstractIndexModificationOperationCallback) modCallbacks[0];
if (prevTupleFilterFactory != null) {
prevTupleFilter = prevTupleFilterFactory.createTupleFilter(ctx);
}
@@ -94,12 +95,17 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
accessor.reset(buffer);
- ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessor;
+ // ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessors[0];
int tupleCount = accessor.getTupleCount();
boolean tupleFilterIsNull = tupleFilter == null;
boolean prevTupleFilterIsNull = prevTupleFilter == null;
for (int i = 0; i < tupleCount; i++) {
try {
+ int storagePartition = tuplePartitioner.partition(accessor, i);
+ int partition = storagePartitionId2Index.get(storagePartition);
+ ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessors[partition];
+ AbstractIndexModificationOperationCallback abstractModCallback =
+ (AbstractIndexModificationOperationCallback) modCallbacks[partition];
frameTuple.reset(accessor, i);
int operation = operationInspector.getIntegerValue(frameTuple.getFieldData(operationFieldIndex),
frameTuple.getFieldStart(operationFieldIndex), frameTuple.getFieldLength(operationFieldIndex));
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorDescriptor.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorDescriptor.java
index 41bd0fb..393adc8 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorDescriptor.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorDescriptor.java
@@ -25,6 +25,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.JobSpecification;
@@ -41,9 +42,10 @@
int[] fieldPermutation, IIndexDataflowHelperFactory indexHelperFactory,
IModificationOperationCallbackFactory modCallbackFactory, int operationFieldIndex,
IBinaryIntegerInspectorFactory operationInspectorFactory, List<AlgebricksPipeline> secondaryKeysPipeline,
- List<AlgebricksPipeline> prevSecondaryKeysPipeline) {
+ List<AlgebricksPipeline> prevSecondaryKeysPipeline, ITuplePartitionerFactory tuplePartitionerFactory,
+ int[][] map) {
super(spec, outRecDesc, fieldPermutation, indexHelperFactory, null, null, modCallbackFactory,
- operationFieldIndex, operationInspectorFactory, null);
+ operationFieldIndex, operationInspectorFactory, null, tuplePartitionerFactory, map);
this.secondaryKeysPipeline = secondaryKeysPipeline;
this.prevSecondaryKeysPipeline = prevSecondaryKeysPipeline;
}
@@ -54,6 +56,6 @@
RecordDescriptor inputRecDesc = recordDescProvider.getInputRecordDescriptor(getActivityId(), 0);
return new LSMSecondaryUpsertWithNestedPlanOperatorNodePushable(ctx, partition, indexHelperFactory,
modCallbackFactory, fieldPermutation, inputRecDesc, operationFieldIndex, operationInspectorFactory,
- secondaryKeysPipeline, prevSecondaryKeysPipeline);
+ secondaryKeysPipeline, prevSecondaryKeysPipeline, tuplePartitionerFactory, map);
}
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorNodePushable.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorNodePushable.java
index 08fd566..94d8b2b 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorNodePushable.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/operators/LSMSecondaryUpsertWithNestedPlanOperatorNodePushable.java
@@ -31,6 +31,7 @@
import org.apache.hyracks.algebricks.runtime.operators.std.NestedTupleSourceRuntimeFactory.NestedTupleSourceRuntime;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
@@ -50,9 +51,10 @@
IIndexDataflowHelperFactory indexHelperFactory, IModificationOperationCallbackFactory modCallbackFactory,
int[] fieldPermutation, RecordDescriptor inputRecDesc, int operationFieldIndex,
IBinaryIntegerInspectorFactory operationInspectorFactory, List<AlgebricksPipeline> secondaryKeysPipeline,
- List<AlgebricksPipeline> prevSecondaryKeysPipeline) throws HyracksDataException {
+ List<AlgebricksPipeline> prevSecondaryKeysPipeline, ITuplePartitionerFactory tuplePartitionerFactory,
+ int[][] map) throws HyracksDataException {
super(ctx, partition, indexHelperFactory, modCallbackFactory, null, null, fieldPermutation, inputRecDesc,
- operationFieldIndex, operationInspectorFactory, null);
+ operationFieldIndex, operationInspectorFactory, null, tuplePartitionerFactory, map);
this.numberOfPrimaryKeyAndFilterFields = fieldPermutation.length;
this.startOfNewKeyPipelines = buildStartOfPipelines(secondaryKeysPipeline, inputRecDesc, false);
this.startOfPrevKeyPipelines = buildStartOfPipelines(prevSecondaryKeysPipeline, inputRecDesc, true);
@@ -94,7 +96,7 @@
public void open() throws HyracksDataException {
super.open();
frameTuple = new FrameTupleReference();
- abstractModCallback = (AbstractIndexModificationOperationCallback) modCallback;
+ // abstractModCallback = (AbstractIndexModificationOperationCallback) modCallbacks[0];
}
@Override
@@ -168,7 +170,7 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- ILSMIndexAccessor workingLSMAccessor = (ILSMIndexAccessor) indexAccessor;
+ // ILSMIndexAccessor workingLSMAccessor = (ILSMIndexAccessor) indexAccessors[0];
endOfPipelineTupleAccessor.reset(buffer);
int nTuple = endOfPipelineTupleAccessor.getTupleCount();
@@ -187,6 +189,11 @@
// Add the primary keys and filter fields.
endTupleReference.addTuple(tuple);
+ int storagePartition = tuplePartitioner.partition(accessor, t);
+ int partition = storagePartitionId2Index.get(storagePartition);
+ ILSMIndexAccessor workingLSMAccessor = (ILSMIndexAccessor) indexAccessors[partition];
+ AbstractIndexModificationOperationCallback abstractModCallback =
+ (AbstractIndexModificationOperationCallback) modCallbacks[partition];
// Finally, pass the tuple to our accessor. There are only two operations: insert or delete.
if (this.isInsert) {
abstractModCallback.setOp(AbstractIndexModificationOperationCallback.Operation.INSERT);
diff --git a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
index 220311e..a50248f 100644
--- a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
+++ b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/CleanupUtils.java
@@ -102,6 +102,15 @@
}
}
+ public static Throwable close(AutoCloseable[] closables, Throwable root) {
+ if (closables != null) {
+ for (AutoCloseable closable : closables) {
+ root = close(closable, root);
+ }
+ }
+ return root;
+ }
+
/**
* Close the AutoCloseable and suppress any Throwable thrown by the close call.
* This method must NEVER throw any Throwable
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
index 9ff9249..425e049 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/IndexInsertUpdateDeleteOperatorNodePushable.java
@@ -23,6 +23,8 @@
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitioner;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -38,53 +40,81 @@
import org.apache.hyracks.storage.am.common.impls.IndexAccessParameters;
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.util.ResourceReleaseUtils;
import org.apache.hyracks.storage.common.IIndex;
import org.apache.hyracks.storage.common.IIndexAccessParameters;
import org.apache.hyracks.storage.common.IIndexAccessor;
import org.apache.hyracks.storage.common.IModificationOperationCallback;
import org.apache.hyracks.storage.common.LocalResource;
+import it.unimi.dsi.fastutil.ints.Int2IntMap;
+import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
+
public class IndexInsertUpdateDeleteOperatorNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
protected final IHyracksTaskContext ctx;
- protected final IIndexDataflowHelper indexHelper;
+ // protected final IIndexDataflowHelper indexHelper;
+ protected final IIndexDataflowHelper[] indexHelpers;
protected final RecordDescriptor inputRecDesc;
protected final IndexOperation op;
protected final PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
protected FrameTupleAccessor accessor;
protected FrameTupleReference frameTuple;
protected IFrame writeBuffer;
- protected IIndexAccessor indexAccessor;
+ // protected IIndexAccessor indexAccessor;
+ protected final IIndexAccessor[] indexAccessors;
protected ITupleFilter tupleFilter;
- protected IModificationOperationCallback modCallback;
- protected IIndex index;
+ // protected IModificationOperationCallback modCallback;
+ protected final IModificationOperationCallback[] modCallbacks;
+ // protected IIndex index;
+ protected final IIndex[] indexes;
protected final IModificationOperationCallbackFactory modOpCallbackFactory;
protected final ITupleFilterFactory tupleFilterFactory;
+ protected final ITuplePartitioner tuplePartitioner;
+ protected final int[] partitions;
+ protected final Int2IntMap storagePartitionId2Index = new Int2IntOpenHashMap();
+ protected boolean writerOpen;
public IndexInsertUpdateDeleteOperatorNodePushable(IHyracksTaskContext ctx, int partition,
IIndexDataflowHelperFactory indexHelperFactory, int[] fieldPermutation, RecordDescriptor inputRecDesc,
IndexOperation op, IModificationOperationCallbackFactory modOpCallbackFactory,
- ITupleFilterFactory tupleFilterFactory) throws HyracksDataException {
+ ITupleFilterFactory tupleFilterFactory, ITuplePartitionerFactory tuplePartitionerFactory, int[][] map)
+ throws HyracksDataException {
this.ctx = ctx;
- this.indexHelper = indexHelperFactory.create(ctx.getJobletContext().getServiceContext(), partition);
+ this.partitions = map != null ? map[partition] : new int[] { partition };
+ for (int i = 0; i < partitions.length; i++) {
+ storagePartitionId2Index.put(partitions[i], i);
+ }
+ this.indexHelpers = new IIndexDataflowHelper[partitions.length];
+ this.indexes = new IIndex[partitions.length];
+ this.indexAccessors = new IIndexAccessor[partitions.length];
+ this.modCallbacks = new IModificationOperationCallback[partitions.length];
+ for (int i = 0; i < partitions.length; i++) {
+ indexHelpers[i] = indexHelperFactory.create(ctx.getJobletContext().getServiceContext(), partitions[i]);
+ }
this.modOpCallbackFactory = modOpCallbackFactory;
this.tupleFilterFactory = tupleFilterFactory;
this.inputRecDesc = inputRecDesc;
this.op = op;
this.tuple.setFieldPermutation(fieldPermutation);
+ this.tuplePartitioner = tuplePartitionerFactory.createPartitioner(ctx);
}
@Override
public void open() throws HyracksDataException {
accessor = new FrameTupleAccessor(inputRecDesc);
writeBuffer = new VSizeFrame(ctx);
- indexHelper.open();
- index = indexHelper.getIndexInstance();
try {
+ for (int i = 0; i < indexHelpers.length; i++) {
+ IIndexDataflowHelper indexHelper = indexHelpers[i];
+ indexHelper.open();
+ indexes[i] = indexHelper.getIndexInstance();
+ LocalResource resource = indexHelper.getResource();
+ modCallbacks[i] = modOpCallbackFactory.createModificationOperationCallback(resource, ctx, this);
+ IIndexAccessParameters iap = new IndexAccessParameters(modCallbacks[i], NoOpOperationCallback.INSTANCE);
+ indexAccessors[i] = indexes[i].createAccessor(iap);
+ }
writer.open();
- LocalResource resource = indexHelper.getResource();
- modCallback = modOpCallbackFactory.createModificationOperationCallback(resource, ctx, this);
- IIndexAccessParameters iap = new IndexAccessParameters(modCallback, NoOpOperationCallback.INSTANCE);
- indexAccessor = index.createAccessor(iap);
+ writerOpen = true;
if (tupleFilterFactory != null) {
tupleFilter = tupleFilterFactory.createTupleFilter(ctx);
frameTuple = new FrameTupleReference();
@@ -108,6 +138,9 @@
}
tuple.reset(accessor, i);
+ int storagePartition = tuplePartitioner.partition(accessor, i);
+ int tupleIndex = storagePartitionId2Index.get(storagePartition);
+ IIndexAccessor indexAccessor = indexAccessors[tupleIndex];
switch (op) {
case INSERT: {
try {
@@ -158,18 +191,24 @@
@Override
public void close() throws HyracksDataException {
- if (index != null) {
- try {
+ Throwable failure = null;
+ try {
+ if (writerOpen) {
writer.close();
- } finally {
- indexHelper.close();
}
+ } finally {
+ for (IIndexDataflowHelper indexHelper : indexHelpers) {
+ failure = ResourceReleaseUtils.close(indexHelper, failure);
+ }
+ }
+ if (failure != null) {
+ throw HyracksDataException.create(failure);
}
}
@Override
public void fail() throws HyracksDataException {
- if (index != null) {
+ if (writerOpen) {
writer.fail();
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexInsertUpdateDeleteOperatorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
index f5cab5c..2cf33f6 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-common/src/main/java/org/apache/hyracks/storage/am/common/dataflow/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
@@ -57,6 +57,6 @@
IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
return new IndexInsertUpdateDeleteOperatorNodePushable(ctx, partition, indexHelperFactory, fieldPermutation,
recordDescProvider.getInputRecordDescriptor(getActivityId(), 0), op, modificationOpCallbackFactory,
- tupleFilterFactory);
+ tupleFilterFactory, null, null);
}
}
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IFrameTupleProcessor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IFrameTupleProcessor.java
index b6192c1..14ce449 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IFrameTupleProcessor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/api/IFrameTupleProcessor.java
@@ -19,6 +19,7 @@
package org.apache.hyracks.storage.am.lsm.common.api;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
public interface IFrameTupleProcessor {
@@ -26,7 +27,7 @@
/**
* Called once per batch before starting the batch process
*/
- void start() throws HyracksDataException;
+ void start(int partition) throws HyracksDataException;
/**
* process the frame tuple with the frame index
@@ -37,17 +38,19 @@
* the index of the tuple in the frame
* @throws HyracksDataException
*/
- void process(ITupleReference tuple, int index) throws HyracksDataException;
+ void process(FrameTupleAccessor accessor, ITupleReference tuple, int index, int partition)
+ throws HyracksDataException;
/**
* Called once per batch before ending the batch process
*/
- void finish() throws HyracksDataException;
+ void finish(int partition) throws HyracksDataException;
/**
* Called when a failure is encountered processing a frame
*
* @param th
+ * @param partition
*/
- void fail(Throwable th);
+ void fail(Throwable th, int partition);
}
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 9e8c568..f059171 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
@@ -19,6 +19,7 @@
package org.apache.hyracks.storage.am.lsm.common.api;
import java.util.List;
+import java.util.Set;
import java.util.function.Predicate;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -238,7 +239,8 @@
* @throws HyracksDataException
*/
void batchOperate(ILSMIndexOperationContext ctx, FrameTupleAccessor accessor, FrameTupleReference tuple,
- IFrameTupleProcessor processor, IFrameOperationCallback frameOpCallback) throws HyracksDataException;
+ IFrameTupleProcessor processor, IFrameOperationCallback frameOpCallback, Set<Integer> tuples, int partition)
+ throws HyracksDataException;
/**
* Rollback components that match the passed predicate
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
index 02744ee..f779112 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMIndexInsertUpdateDeleteOperatorNodePushable.java
@@ -21,6 +21,7 @@
import java.nio.ByteBuffer;
import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.dataflow.common.comm.io.FrameTupleAppender;
@@ -47,15 +48,15 @@
public LSMIndexInsertUpdateDeleteOperatorNodePushable(IHyracksTaskContext ctx, int partition,
IIndexDataflowHelperFactory indexHelperFactory, int[] fieldPermutation, RecordDescriptor inputRecDesc,
IndexOperation op, IModificationOperationCallbackFactory modCallbackFactory,
- ITupleFilterFactory tupleFilterFactory) throws HyracksDataException {
+ ITupleFilterFactory tupleFilterFactory, ITuplePartitionerFactory tuplePartitionerFactory, int[][] map)
+ throws HyracksDataException {
super(ctx, partition, indexHelperFactory, fieldPermutation, inputRecDesc, op, modCallbackFactory,
- tupleFilterFactory);
+ tupleFilterFactory, tuplePartitionerFactory, map);
}
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
accessor.reset(buffer);
- ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessor;
int nextFlushTupleIndex = 0;
int tupleCount = accessor.getTupleCount();
for (int i = 0; i < tupleCount; i++) {
@@ -68,6 +69,9 @@
}
tuple.reset(accessor, i);
+ int storagePartition = tuplePartitioner.partition(accessor, i);
+ int tupleIndex = storagePartitionId2Index.get(storagePartition);
+ ILSMIndexAccessor lsmAccessor = (ILSMIndexAccessor) indexAccessors[tupleIndex];
switch (op) {
case INSERT: {
if (!lsmAccessor.tryInsert(tuple)) {
diff --git a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.java
index fb884f7..9dcc8ea 100644
--- a/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.java
+++ b/hyracks-fullstack/hyracks/hyracks-storage-am-lsm-common/src/main/java/org/apache/hyracks/storage/am/lsm/common/dataflow/LSMTreeIndexInsertUpdateDeleteOperatorDescriptor.java
@@ -22,6 +22,7 @@
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.IOperatorNodePushable;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import org.apache.hyracks.api.dataflow.value.ITuplePartitionerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
@@ -40,17 +41,21 @@
protected final IIndexDataflowHelperFactory indexHelperFactory;
protected final IModificationOperationCallbackFactory modCallbackFactory;
protected final ITupleFilterFactory tupleFilterFactory;
+ protected final ITuplePartitionerFactory tuplePartitionerFactory;
+ protected final int[][] map;
public LSMTreeIndexInsertUpdateDeleteOperatorDescriptor(IOperatorDescriptorRegistry spec,
RecordDescriptor outRecDesc, IIndexDataflowHelperFactory indexHelperFactory, int[] fieldPermutation,
IndexOperation op, IModificationOperationCallbackFactory modCallbackFactory,
- ITupleFilterFactory tupleFilterFactory) {
+ ITupleFilterFactory tupleFilterFactory, ITuplePartitionerFactory tuplePartitionerFactory, int[][] map) {
super(spec, 1, 1);
this.indexHelperFactory = indexHelperFactory;
this.modCallbackFactory = modCallbackFactory;
this.tupleFilterFactory = tupleFilterFactory;
this.fieldPermutation = fieldPermutation;
this.op = op;
+ this.tuplePartitionerFactory = tuplePartitionerFactory;
+ this.map = map;
this.outRecDescs[0] = outRecDesc;
}
@@ -59,6 +64,6 @@
IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) throws HyracksDataException {
return new LSMIndexInsertUpdateDeleteOperatorNodePushable(ctx, partition, indexHelperFactory, fieldPermutation,
recordDescProvider.getInputRecordDescriptor(getActivityId(), 0), op, modCallbackFactory,
- tupleFilterFactory);
+ tupleFilterFactory, tuplePartitionerFactory, map);
}
}
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 950a8e5..a7b838d 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
@@ -21,6 +21,7 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate;
@@ -711,18 +712,19 @@
@Override
public void batchOperate(ILSMIndexOperationContext ctx, FrameTupleAccessor accessor, FrameTupleReference tuple,
- IFrameTupleProcessor processor, IFrameOperationCallback frameOpCallback) throws HyracksDataException {
- processor.start();
+ IFrameTupleProcessor processor, IFrameOperationCallback frameOpCallback, Set<Integer> tuples, int partition)
+ throws HyracksDataException {
+ processor.start(partition);
enter(ctx);
try {
try {
- processFrame(accessor, tuple, processor);
+ processFrame(accessor, tuple, processor, tuples, partition);
frameOpCallback.frameCompleted();
} catch (Throwable th) {
- processor.fail(th);
+ processor.fail(th, partition);
throw th;
} finally {
- processor.finish();
+ processor.finish(partition);
}
} catch (HyracksDataException e) {
LOGGER.warn("Failed to process frame", e);
@@ -860,13 +862,18 @@
}
private static void processFrame(FrameTupleAccessor accessor, FrameTupleReference tuple,
- IFrameTupleProcessor processor) throws HyracksDataException {
+ IFrameTupleProcessor processor, Set<Integer> tuples, int partition) throws HyracksDataException {
int tupleCount = accessor.getTupleCount();
- int i = 0;
- while (i < tupleCount) {
- tuple.reset(accessor, i);
- processor.process(tuple, i);
- i++;
+ for (int i = 0; i < tupleCount; i++) {
+ if (tuples != null) {
+ if (!tuples.contains(i)) {
+ continue;
+ }
+ tuple.reset(accessor, i);
+ } else {
+ tuple.reset(accessor, i);
+ }
+ processor.process(accessor, tuple, i, partition);
}
}
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 8412b8c..8873844 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
@@ -20,6 +20,7 @@
package org.apache.hyracks.storage.am.lsm.common.impls;
import java.util.List;
+import java.util.Set;
import java.util.function.Predicate;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -210,8 +211,13 @@
}
public void batchOperate(FrameTupleAccessor accessor, FrameTupleReference tuple, IFrameTupleProcessor processor,
- IFrameOperationCallback frameOpCallback) throws HyracksDataException {
- lsmHarness.batchOperate(ctx, accessor, tuple, processor, frameOpCallback);
+ IFrameOperationCallback frameOpCallback, int partition) throws HyracksDataException {
+ lsmHarness.batchOperate(ctx, accessor, tuple, processor, frameOpCallback, null, partition);
+ }
+
+ public void batchOperate(FrameTupleAccessor accessor, FrameTupleReference tuple, IFrameTupleProcessor processor,
+ IFrameOperationCallback frameOpCallback, Set<Integer> tuples, int partition) throws HyracksDataException {
+ lsmHarness.batchOperate(ctx, accessor, tuple, processor, frameOpCallback, tuples, partition);
}
@Override
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17452
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I08da28f2a26fcaf581c2256312455fe541fae5ea
Gerrit-Change-Number: 17452
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>
Gerrit-MessageType: newchange