You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by xi...@apache.org on 2020/08/21 00:10:39 UTC
[asterixdb] branch master updated: [NO ISSUE] Adding more extension
APIs
This is an automated email from the ASF dual-hosted git repository.
xikui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/asterixdb.git
The following commit(s) were added to refs/heads/master by this push:
new 44f2abd [NO ISSUE] Adding more extension APIs
44f2abd is described below
commit 44f2abde4998310c1ab61d3ddf4369f76fe70a3c
Author: Xikui Wang <xk...@gmail.com>
AuthorDate: Thu Aug 20 14:55:20 2020 -0700
[NO ISSUE] Adding more extension APIs
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
1. Allow extensions to override LangExpressionToPlanTranslator to
tranlsate insert/upsert/delete.
2. Reuse the current statement executor in feed if possible to make sure
extensions' translator can be used properly.
3. Add new methods in MetadataProvider so that extensions could override
them to use customized LSM runtime operators.
4. Add new operational interface in LSM runtime to allow extensions to
modify data before persistence.
Change-Id: Iada557f15af9de6fbdb6f6d4e41c0266c1d3fbff
Reviewed-on: https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/7643
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Xikui Wang <xk...@gmail.com>
---
.../translator/LangExpressionToPlanTranslator.java | 14 +++---
.../asterix/app/active/FeedEventsListener.java | 10 ++--
.../asterix/app/translator/QueryTranslator.java | 8 ++--
.../org/apache/asterix/utils/FeedOperations.java | 9 ++--
.../asterix/test/active/ActiveStatsTest.java | 2 +-
.../test/active/DummyFeedEventsListener.java | 6 ++-
.../LSMInsertDeleteOperatorNodePushable.java | 12 ++---
.../LSMTreeInsertDeleteOperatorDescriptor.java | 2 +-
.../metadata/declared/MetadataProvider.java | 33 +++++++++++--
.../LSMPrimaryInsertOperatorDescriptor.java | 8 ++--
.../LSMPrimaryInsertOperatorNodePushable.java | 15 ++++--
.../LSMPrimaryUpsertOperatorDescriptor.java | 14 +++---
.../LSMPrimaryUpsertOperatorNodePushable.java | 56 +++++++++++++---------
13 files changed, 122 insertions(+), 67 deletions(-)
diff --git a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java
index eb4bfd4..4bb1adf 100644
--- a/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java
+++ b/asterixdb/asterix-algebra/src/main/java/org/apache/asterix/translator/LangExpressionToPlanTranslator.java
@@ -443,7 +443,7 @@ abstract class LangExpressionToPlanTranslator
return plan;
}
- private ILogicalOperator translateDelete(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
+ protected ILogicalOperator translateDelete(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
List<Mutable<ILogicalExpression>> varRefsForLoading,
List<Mutable<ILogicalExpression>> additionalFilteringExpressions, ILogicalOperator assign,
ICompiledDmlStatement stmt) throws AlgebricksException {
@@ -464,7 +464,7 @@ abstract class LangExpressionToPlanTranslator
return leafOperator;
}
- private ILogicalOperator translateUpsert(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
+ protected ILogicalOperator translateUpsert(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
List<Mutable<ILogicalExpression>> varRefsForLoading,
List<Mutable<ILogicalExpression>> additionalFilteringExpressions, ILogicalOperator assign,
List<String> additionalFilteringField, LogicalVariable unnestVar, ILogicalOperator topOp,
@@ -579,7 +579,7 @@ abstract class LangExpressionToPlanTranslator
return processReturningExpression(rootOperator, upsertOp, compiledUpsert, resultMetadata);
}
- private ILogicalOperator translateInsert(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
+ protected ILogicalOperator translateInsert(DatasetDataSource targetDatasource, Mutable<ILogicalExpression> varRef,
List<Mutable<ILogicalExpression>> varRefsForLoading,
List<Mutable<ILogicalExpression>> additionalFilteringExpressions, ILogicalOperator assign,
ICompiledDmlStatement stmt, IResultMetadata resultMetadata) throws AlgebricksException {
@@ -609,7 +609,7 @@ abstract class LangExpressionToPlanTranslator
// Stitches the translated operators for the returning expression into the query
// plan.
- private ILogicalOperator processReturningExpression(ILogicalOperator inputOperator,
+ protected ILogicalOperator processReturningExpression(ILogicalOperator inputOperator,
InsertDeleteUpsertOperator insertOp, CompiledInsertStatement compiledInsert, IResultMetadata resultMetadata)
throws AlgebricksException {
Expression returnExpression = compiledInsert.getReturnExpression();
@@ -651,7 +651,7 @@ abstract class LangExpressionToPlanTranslator
return distResultOperator;
}
- private DatasetDataSource validateDatasetInfo(MetadataProvider metadataProvider, DataverseName dataverseName,
+ protected DatasetDataSource validateDatasetInfo(MetadataProvider metadataProvider, DataverseName dataverseName,
String datasetName, SourceLocation sourceLoc) throws AlgebricksException {
Dataset dataset = metadataProvider.findDataset(dataverseName, datasetName);
if (dataset == null) {
@@ -671,7 +671,7 @@ abstract class LangExpressionToPlanTranslator
dataset.getDatasetDetails(), domain);
}
- private FileSplit getDefaultOutputFileLocation(ICcApplicationContext appCtx) throws AlgebricksException {
+ protected FileSplit getDefaultOutputFileLocation(ICcApplicationContext appCtx) throws AlgebricksException {
String outputDir = System.getProperty("java.io.tmpDir");
String filePath =
outputDir + System.getProperty("file.separator") + OUTPUT_FILE_PREFIX + outputFileID.incrementAndGet();
@@ -1760,7 +1760,7 @@ abstract class LangExpressionToPlanTranslator
* the query plan.
* @throws CompilationException
*/
- private void eliminateSharedOperatorReferenceForPlan(ILogicalPlan plan) throws CompilationException {
+ protected void eliminateSharedOperatorReferenceForPlan(ILogicalPlan plan) throws CompilationException {
for (Mutable<ILogicalOperator> opRef : plan.getRoots()) {
Set<Mutable<ILogicalOperator>> opRefSet = new HashSet<>();
eliminateSharedOperatorReference(opRef, opRefSet);
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java
index d4a52fb..c4db4eb 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/active/FeedEventsListener.java
@@ -26,6 +26,7 @@ import org.apache.asterix.active.ActivityState;
import org.apache.asterix.active.EntityId;
import org.apache.asterix.active.IActiveEntityEventSubscriber;
import org.apache.asterix.active.IRetryPolicyFactory;
+import org.apache.asterix.algebra.base.ILangExtension;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.common.exceptions.ErrorCode;
import org.apache.asterix.common.exceptions.RuntimeDataException;
@@ -50,14 +51,17 @@ public class FeedEventsListener extends ActiveEntityEventsListener {
private final Feed feed;
private final List<FeedConnection> feedConnections;
+ private final ILangExtension.Language translatorLang;
public FeedEventsListener(IStatementExecutor statementExecutor, ICcApplicationContext appCtx,
IHyracksClientConnection hcc, EntityId entityId, List<Dataset> datasets,
AlgebricksAbsolutePartitionConstraint locations, String runtimeName, IRetryPolicyFactory retryPolicyFactory,
- Feed feed, final List<FeedConnection> feedConnections) throws HyracksDataException {
+ Feed feed, final List<FeedConnection> feedConnections, ILangExtension.Language translatorLang)
+ throws HyracksDataException {
super(statementExecutor, appCtx, hcc, entityId, datasets, locations, runtimeName, retryPolicyFactory);
this.feed = feed;
this.feedConnections = feedConnections;
+ this.translatorLang = translatorLang;
}
@Override
@@ -94,8 +98,8 @@ public class FeedEventsListener extends ActiveEntityEventsListener {
@Override
protected JobId compileAndStartJob(MetadataProvider mdProvider) throws HyracksDataException {
try {
- Pair<JobSpecification, AlgebricksAbsolutePartitionConstraint> jobInfo =
- FeedOperations.buildStartFeedJob(mdProvider, feed, feedConnections, statementExecutor, hcc);
+ Pair<JobSpecification, AlgebricksAbsolutePartitionConstraint> jobInfo = FeedOperations
+ .buildStartFeedJob(mdProvider, feed, feedConnections, statementExecutor, hcc, translatorLang);
JobSpecification feedJob = jobInfo.getLeft();
feedJob.setProperty(ActiveNotificationHandler.ACTIVE_ENTITY_PROPERTY_NAME, entityId);
// TODO(Yingyi): currently we do not check IFrameWriter protocol violations for Feed jobs.
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 78a82d6..f8e0cee 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -1986,7 +1986,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
declaredFunctions.add(fds);
}
- protected void handleCreateFunctionStatement(MetadataProvider metadataProvider, Statement stmt,
+ public void handleCreateFunctionStatement(MetadataProvider metadataProvider, Statement stmt,
IStatementRewriter stmtRewriter) throws Exception {
CreateFunctionStatement cfs = (CreateFunctionStatement) stmt;
FunctionSignature signature = cfs.getFunctionSignature();
@@ -2985,7 +2985,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
}
}
- private void handleStartFeedStatement(MetadataProvider metadataProvider, Statement stmt,
+ protected void handleStartFeedStatement(MetadataProvider metadataProvider, Statement stmt,
IHyracksClientConnection hcc) throws Exception {
StartFeedStatement sfs = (StartFeedStatement) stmt;
SourceLocation sourceLoc = sfs.getSourceLocation();
@@ -3024,7 +3024,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
}
listener = new FeedEventsListener(this, metadataProvider.getApplicationContext(), hcc, entityId,
datasets, null, FeedIntakeOperatorNodePushable.class.getSimpleName(),
- NoRetryPolicyFactory.INSTANCE, feed, feedConnections);
+ NoRetryPolicyFactory.INSTANCE, feed, feedConnections, compilationProvider.getLanguage());
}
MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
committed = true;
@@ -3039,7 +3039,7 @@ public class QueryTranslator extends AbstractLangTranslator implements IStatemen
}
}
- private void handleStopFeedStatement(MetadataProvider metadataProvider, Statement stmt) throws Exception {
+ protected void handleStopFeedStatement(MetadataProvider metadataProvider, Statement stmt) throws Exception {
StopFeedStatement sfst = (StopFeedStatement) stmt;
SourceLocation sourceLoc = sfst.getSourceLocation();
DataverseName dataverseName = getActiveDataverseName(sfst.getDataverseName());
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
index 20a97ed..68234b7 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/utils/FeedOperations.java
@@ -29,6 +29,7 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeSet;
+import org.apache.asterix.algebra.base.ILangExtension;
import org.apache.asterix.app.result.ResponsePrinter;
import org.apache.asterix.app.translator.DefaultStatementExecutorFactory;
import org.apache.asterix.app.translator.QueryTranslator;
@@ -450,7 +451,8 @@ public class FeedOperations {
public static Pair<JobSpecification, AlgebricksAbsolutePartitionConstraint> buildStartFeedJob(
MetadataProvider metadataProvider, Feed feed, List<FeedConnection> feedConnections,
- IStatementExecutor statementExecutor, IHyracksClientConnection hcc) throws Exception {
+ IStatementExecutor statementExecutor, IHyracksClientConnection hcc, ILangExtension.Language translatorLang)
+ throws Exception {
FeedPolicyAccessor fpa = new FeedPolicyAccessor(new HashMap<>());
Pair<JobSpecification, ITypedAdapterFactory> intakeInfo = buildFeedIntakeJobSpec(feed, metadataProvider, fpa);
List<JobSpecification> jobsList = new ArrayList<>();
@@ -465,8 +467,9 @@ public class FeedOperations {
metadataProvider.getConfig().put(FeedActivityDetails.COLLECT_LOCATIONS,
StringUtils.join(ingestionLocations, ','));
// TODO: Once we deprecated AQL, this extra queryTranslator can be removed.
- IStatementExecutor translator =
- getSQLPPTranslator(metadataProvider, ((QueryTranslator) statementExecutor).getSessionOutput());
+ IStatementExecutor translator = translatorLang == ILangExtension.Language.AQL
+ ? getSQLPPTranslator(metadataProvider, ((QueryTranslator) statementExecutor).getSessionOutput())
+ : statementExecutor;
// Add connection job
for (FeedConnection feedConnection : feedConnections) {
JobSpecification connectionJob =
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java
index 963ba7c..b23b3cf 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/ActiveStatsTest.java
@@ -107,7 +107,7 @@ public class ActiveStatsTest {
// Add event listener
ActiveEntityEventsListener eventsListener = new DummyFeedEventsListener(statementExecutor, appCtx, null,
entityId, datasetList, partitionConstraint, FeedIntakeOperatorNodePushable.class.getSimpleName(),
- NoRetryPolicyFactory.INSTANCE, null, Collections.emptyList());
+ NoRetryPolicyFactory.INSTANCE, null, Collections.emptyList(), Language.SQLPP);
// Register mock runtime
NCAppRuntimeContext nc1AppCtx =
(NCAppRuntimeContext) ExecutionTestUtil.integrationUtil.ncs[0].getApplicationContext();
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
index 88f1332..884d209 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/active/DummyFeedEventsListener.java
@@ -26,6 +26,7 @@ import org.apache.asterix.active.ActivityState;
import org.apache.asterix.active.EntityId;
import org.apache.asterix.active.IActiveEntityEventSubscriber;
import org.apache.asterix.active.IRetryPolicyFactory;
+import org.apache.asterix.algebra.base.ILangExtension;
import org.apache.asterix.app.active.FeedEventsListener;
import org.apache.asterix.common.dataflow.ICcApplicationContext;
import org.apache.asterix.external.feed.watch.WaitForStateSubscriber;
@@ -43,9 +44,10 @@ public class DummyFeedEventsListener extends FeedEventsListener {
public DummyFeedEventsListener(IStatementExecutor statementExecutor, ICcApplicationContext appCtx,
IHyracksClientConnection hcc, EntityId entityId, List<Dataset> datasets,
AlgebricksAbsolutePartitionConstraint locations, String runtimeName, IRetryPolicyFactory retryPolicyFactory,
- Feed feed, List<FeedConnection> feedConnections) throws HyracksDataException {
+ Feed feed, List<FeedConnection> feedConnections, ILangExtension.Language translatorLang)
+ throws HyracksDataException {
super(statementExecutor, appCtx, hcc, entityId, datasets, locations, runtimeName, retryPolicyFactory, feed,
- feedConnections);
+ feedConnections, translatorLang);
}
@Override
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 5e4b5a1..2ad2f7a 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
@@ -48,11 +48,11 @@ import org.apache.hyracks.storage.common.IIndexAccessParameters;
public class LSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUpdateDeleteOperatorNodePushable {
public static final String KEY_INDEX = "Index";
- private final boolean isPrimary;
- private final SourceLocation sourceLoc;
+ 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 int i = 0;
+ protected int i = 0;
/**
* The following three variables are used to keep track of the information regarding flushing partial frame such as
@@ -64,9 +64,9 @@ public class LSMInsertDeleteOperatorNodePushable extends LSMIndexInsertUpdateDel
* ==> captured in currentTupleIdx variable
* These variables are reset for each frame, i.e., whenever nextFrame() is called, these variables are reset.
*/
- private boolean flushedPartialTuples;
- private int currentTupleIdx;
- private int lastFlushedTupleIdx;
+ protected boolean flushedPartialTuples;
+ protected int currentTupleIdx;
+ protected int lastFlushedTupleIdx;
public LSMInsertDeleteOperatorNodePushable(IHyracksTaskContext ctx, int partition, int[] fieldPermutation,
RecordDescriptor inputRecDesc, IndexOperation op, boolean isPrimary,
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 45661e4..2226ca0 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
@@ -33,7 +33,7 @@ import org.apache.hyracks.storage.am.lsm.common.dataflow.LSMTreeIndexInsertUpdat
public class LSMTreeInsertDeleteOperatorDescriptor extends LSMTreeIndexInsertUpdateDeleteOperatorDescriptor {
private static final long serialVersionUID = 1L;
- private final boolean isPrimary;
+ protected final boolean isPrimary;
public LSMTreeInsertDeleteOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc,
int[] fieldPermutation, IndexOperation op, IIndexDataflowHelperFactory indexHelperFactory,
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 2933756..8a16014 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
@@ -131,12 +131,14 @@ import org.apache.hyracks.algebricks.runtime.operators.std.SinkWriterRuntimeFact
import org.apache.hyracks.api.dataflow.IOperatorDescriptor;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.IMissingWriterFactory;
import org.apache.hyracks.api.dataflow.value.IResultSerializerFactory;
import org.apache.hyracks.api.dataflow.value.ISerializerDeserializer;
import org.apache.hyracks.api.dataflow.value.ITypeTraits;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.IWarningCollector;
import org.apache.hyracks.api.io.FileSplit;
+import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
import org.apache.hyracks.api.job.JobSpecification;
import org.apache.hyracks.api.result.IResultMetadata;
import org.apache.hyracks.api.result.ResultSetId;
@@ -975,10 +977,18 @@ public class MetadataProvider implements IMetadataProvider<DataSourceId, String>
fieldPermutation[i++] = idx;
}
}
- return DatasetUtil.createPrimaryIndexUpsertOp(spec, this, dataset, recordDesc, fieldPermutation,
+ return createPrimaryIndexUpsertOp(spec, this, dataset, recordDesc, fieldPermutation,
context.getMissingWriterFactory());
}
+ protected Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> createPrimaryIndexUpsertOp(JobSpecification spec,
+ MetadataProvider metadataProvider, Dataset dataset, RecordDescriptor inputRecordDesc,
+ int[] fieldPermutation, IMissingWriterFactory missingWriterFactory) throws AlgebricksException {
+ // this can be used by extensions to pick up their own operators
+ return DatasetUtil.createPrimaryIndexUpsertOp(spec, this, dataset, inputRecordDesc, fieldPermutation,
+ missingWriterFactory);
+ }
+
public Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> buildExternalDatasetDataScannerRuntime(
JobSpecification jobSpec, IAType itemType, ITypedAdapterFactory adapterFactory) throws AlgebricksException {
if (itemType.getTypeTag() != ATypeTag.OBJECT) {
@@ -1120,17 +1130,34 @@ public class MetadataProvider implements IMetadataProvider<DataSourceId, String>
pkidfh = new IndexDataflowHelperFactory(storageComponentProvider.getStorageManager(),
primaryKeySplitsAndConstraint.first);
}
- op = new LSMPrimaryInsertOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, idfh, pkidfh,
+ op = createLSMPrimaryInsertOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, idfh, pkidfh,
modificationCallbackFactory, searchCallbackFactory, numKeys, filterFields);
} else {
- op = new LSMTreeInsertDeleteOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, indexOp, idfh,
+ op = createLSMTreeInsertDeleteOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, indexOp, idfh,
null, true, modificationCallbackFactory);
}
}
return new Pair<>(op, splitsAndConstraint.second);
}
+ protected LSMPrimaryInsertOperatorDescriptor createLSMPrimaryInsertOperatorDescriptor(JobSpecification spec,
+ RecordDescriptor inputRecordDesc, int[] fieldPermutation, IIndexDataflowHelperFactory idfh,
+ IIndexDataflowHelperFactory pkidfh, IModificationOperationCallbackFactory modificationCallbackFactory,
+ ISearchOperationCallbackFactory searchCallbackFactory, int numKeys, int[] filterFields) {
+ // this can be used by extensions to pick up their own operators
+ return new LSMPrimaryInsertOperatorDescriptor(spec, inputRecordDesc, fieldPermutation, idfh, pkidfh,
+ modificationCallbackFactory, searchCallbackFactory, numKeys, filterFields);
+ }
+
+ protected LSMTreeInsertDeleteOperatorDescriptor createLSMTreeInsertDeleteOperatorDescriptor(
+ IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc, int[] fieldPermutation, IndexOperation op,
+ IIndexDataflowHelperFactory indexHelperFactory, ITupleFilterFactory tupleFilterFactory, boolean isPrimary,
+ IModificationOperationCallbackFactory modCallbackFactory) {
+ return new LSMTreeInsertDeleteOperatorDescriptor(spec, outRecDesc, fieldPermutation, op, indexHelperFactory,
+ tupleFilterFactory, isPrimary, modCallbackFactory);
+ }
+
private Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> getIndexInsertOrDeleteOrUpsertRuntime(
IndexOperation indexOp, IDataSourceIndex<String, DataSourceId> dataSourceIndex,
IOperatorSchema propagatedSchema, IOperatorSchema[] inputSchemas, IVariableTypeEnvironment typeEnv,
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 42b8f29..d639f3d 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
@@ -34,10 +34,10 @@ public class LSMPrimaryInsertOperatorDescriptor extends LSMTreeInsertDeleteOpera
private static final long serialVersionUID = 1L;
- private final IIndexDataflowHelperFactory keyIndexHelperFactory;
- private final ISearchOperationCallbackFactory searchOpCallbackFactory;
- private final int numOfPrimaryKeys;
- private final int[] filterFields;
+ protected final IIndexDataflowHelperFactory keyIndexHelperFactory;
+ protected final ISearchOperationCallbackFactory searchOpCallbackFactory;
+ protected final int numOfPrimaryKeys;
+ protected final int[] filterFields;
public LSMPrimaryInsertOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc,
int[] fieldPermutation, IIndexDataflowHelperFactory indexHelperFactory,
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 0bb42d4..a886161 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
@@ -105,7 +105,17 @@ public class LSMPrimaryInsertOperatorNodePushable extends LSMIndexInsertUpdateDe
}
}
keyTuple = new PermutingFrameTupleReference(searchKeyPermutations);
- processor = new IFrameTupleProcessor() {
+ processor = createTupleProcessor(sourceLoc);
+ frameOpCallback = NoOpFrameOperationCallbackFactory.INSTANCE.createFrameOperationCallback(ctx, lsmAccessor);
+ }
+
+ protected void beforeModification(ITupleReference tuple) {
+ // this is used for extensions to modify tuples before persistence
+ // do nothing in the master branch
+ }
+
+ protected IFrameTupleProcessor createTupleProcessor(SourceLocation sourceLoc) {
+ return new IFrameTupleProcessor() {
@Override
public void process(ITupleReference tuple, int index) throws HyracksDataException {
if (index < currentTupleIdx) {
@@ -127,6 +137,7 @@ public class LSMPrimaryInsertOperatorNodePushable extends LSMIndexInsertUpdateDe
cursor.close();
}
if (!duplicate) {
+ beforeModification(tuple);
lsmAccessor.forceUpsert(tuple);
if (lsmAccessorForKeyIndex != null) {
lsmAccessorForKeyIndex.forceUpsert(keyTuple);
@@ -157,8 +168,6 @@ public class LSMPrimaryInsertOperatorNodePushable extends LSMIndexInsertUpdateDe
// no op
}
};
-
- frameOpCallback = NoOpFrameOperationCallbackFactory.INSTANCE.createFrameOperationCallback(ctx, lsmAccessor);
}
@Override
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 7587ca6..43f54f2 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
@@ -36,13 +36,13 @@ import org.apache.hyracks.storage.am.lsm.common.api.IFrameOperationCallbackFacto
public class LSMPrimaryUpsertOperatorDescriptor extends LSMTreeInsertDeleteOperatorDescriptor {
private static final long serialVersionUID = 1L;
- private final IFrameOperationCallbackFactory frameOpCallbackFactory;
- private final ARecordType recordType;
- private final int filterIndex;
- private ISearchOperationCallbackFactory searchOpCallbackFactory;
- private final int numPrimaryKeys;
- private final IMissingWriterFactory missingWriterFactory;
- private final boolean hasSecondaries;
+ protected final IFrameOperationCallbackFactory frameOpCallbackFactory;
+ protected final ARecordType recordType;
+ protected final int filterIndex;
+ protected ISearchOperationCallbackFactory searchOpCallbackFactory;
+ protected final int numPrimaryKeys;
+ protected final IMissingWriterFactory missingWriterFactory;
+ protected final boolean hasSecondaries;
public LSMPrimaryUpsertOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor outRecDesc,
int[] fieldPermutation, IIndexDataflowHelperFactory indexHelperFactory,
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 3e43e02..316665d 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
@@ -83,17 +83,17 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
private static final Logger LOGGER = LogManager.getLogger();
private static final ThreadLocal<DateFormat> DATE_FORMAT =
ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS"));
- private final PermutingFrameTupleReference key;
+ protected final PermutingFrameTupleReference key;
private MultiComparator keySearchCmp;
private ArrayTupleBuilder missingTupleBuilder;
private final IMissingWriter missingWriter;
- private ArrayTupleBuilder tb;
+ protected ArrayTupleBuilder tb;
private DataOutput dos;
- private RangePredicate searchPred;
- private IIndexCursor cursor;
- private ITupleReference prevTuple;
- private final int numOfPrimaryKeys;
- boolean isFiltered = false;
+ protected RangePredicate searchPred;
+ protected IIndexCursor cursor;
+ protected ITupleReference prevTuple;
+ protected final int numOfPrimaryKeys;
+ protected boolean isFiltered = false;
private final ArrayTupleReference prevTupleWithFilter = new ArrayTupleReference();
private ArrayTupleBuilder prevRecWithPKWithFilterValue;
private ARecordType recordType;
@@ -103,13 +103,13 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
private final boolean hasMeta;
private final int filterFieldIndex;
private final int metaFieldIndex;
- private LockThenSearchOperationCallback searchCallback;
- private IFrameOperationCallback frameOpCallback;
+ protected LockThenSearchOperationCallback searchCallback;
+ protected IFrameOperationCallback frameOpCallback;
private final IFrameOperationCallbackFactory frameOpCallbackFactory;
- private AbstractIndexModificationOperationCallback abstractModCallback;
+ protected AbstractIndexModificationOperationCallback abstractModCallback;
private final ISearchOperationCallbackFactory searchCallbackFactory;
private final IFrameTupleProcessor processor;
- private LSMTreeIndexAccessor lsmAccessor;
+ protected LSMTreeIndexAccessor lsmAccessor;
private final ITracer tracer;
private final long traceCategory;
private long lastRecordInTimeStamp = 0L;
@@ -144,7 +144,18 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
this.prevRecWithPKWithFilterValue = new ArrayTupleBuilder(fieldPermutation.length + (hasMeta ? 1 : 0));
this.prevDos = prevRecWithPKWithFilterValue.getDataOutput();
}
- processor = new IFrameTupleProcessor() {
+ processor = createTupleProcessor(hasSecondaries);
+ tracer = ctx.getJobletContext().getServiceContext().getTracer();
+ traceCategory = tracer.getRegistry().get(TraceUtils.LATENCY);
+ }
+
+ protected void beforeModification(ITupleReference tuple) {
+ // this is used for extensions to modify tuples before persistence
+ // do nothing in the master branch
+ }
+
+ protected IFrameTupleProcessor createTupleProcessor(final boolean hasSecondaries) {
+ return new IFrameTupleProcessor() {
@Override
public void process(ITupleReference tuple, int index) throws HyracksDataException {
try {
@@ -176,6 +187,7 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
appendUpsertIndicator(!isDelete);
appendPreviousTupleAsMissing();
}
+ beforeModification(tuple);
if (isDelete && prevTuple != null) {
// Only delete if it is a delete and not upsert
// And previous tuple with the same key was found
@@ -213,8 +225,6 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
frameOpCallback.fail(th);
}
};
- tracer = ctx.getJobletContext().getServiceContext().getTracer();
- traceCategory = tracer.getRegistry().get(TraceUtils.LATENCY);
}
// we have the permutation which has [pk locations, record location, optional:filter-location]
@@ -285,12 +295,12 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
}
}
- private void resetSearchPredicate(int tupleIndex) {
+ protected void resetSearchPredicate(int tupleIndex) {
key.reset(accessor, tupleIndex);
searchPred.reset(key, key, true, true, keySearchCmp, keySearchCmp);
}
- private void writeOutput(int tupleIndex, boolean recordWasInserted, boolean recordWasDeleted) throws IOException {
+ protected void writeOutput(int tupleIndex, boolean recordWasInserted, boolean recordWasDeleted) throws IOException {
if (recordWasInserted || recordWasDeleted) {
frameTuple.reset(accessor, tupleIndex);
for (int i = 0; i < frameTuple.getFieldCount(); i++) {
@@ -307,7 +317,7 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
}
}
- private static boolean isDeleteOperation(ITupleReference t1, int field) {
+ protected static boolean isDeleteOperation(ITupleReference t1, int field) {
return TypeTagUtil.isType(t1, field, ATypeTag.SERIALIZED_MISSING_TYPE_TAG);
}
@@ -326,7 +336,7 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
}
}
- private void appendFilterToOutput() throws IOException {
+ protected void appendFilterToOutput() throws IOException {
// if with filters, append the filter
if (isFiltered) {
dos.write(prevTuple.getFieldData(filterFieldIndex), prevTuple.getFieldStart(filterFieldIndex),
@@ -335,18 +345,18 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
}
}
- private void appendUpsertIndicator(boolean isUpsert) throws IOException {
+ protected void appendUpsertIndicator(boolean isUpsert) throws IOException {
recordDesc.getFields()[0].serialize(isUpsert ? ABoolean.TRUE : ABoolean.FALSE, dos);
tb.addFieldEndOffset();
}
- private void appendPrevRecord() throws IOException {
+ protected void appendPrevRecord() throws IOException {
dos.write(prevTuple.getFieldData(numOfPrimaryKeys), prevTuple.getFieldStart(numOfPrimaryKeys),
prevTuple.getFieldLength(numOfPrimaryKeys));
tb.addFieldEndOffset();
}
- private void appendPreviousMeta() throws IOException {
+ protected void appendPreviousMeta() throws IOException {
// if has meta, then append meta
if (hasMeta) {
dos.write(prevTuple.getFieldData(metaFieldIndex), prevTuple.getFieldStart(metaFieldIndex),
@@ -355,7 +365,7 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
}
}
- private void appendPreviousTupleAsMissing() throws IOException {
+ protected void appendPreviousTupleAsMissing() throws IOException {
prevTuple = null;
writeMissingField();
if (hasMeta) {
@@ -376,7 +386,7 @@ public class LSMPrimaryUpsertOperatorNodePushable extends LSMIndexInsertUpdateDe
appender.write(writer, true);
}
- private void appendFilterToPrevTuple() throws IOException {
+ protected void appendFilterToPrevTuple() throws IOException {
if (isFiltered) {
prevRecWithPKWithFilterValue.reset();
for (int i = 0; i < prevTuple.getFieldCount(); i++) {