You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by dl...@apache.org on 2018/05/30 17:44:30 UTC
[03/24] asterixdb git commit: [ASTERIXDB-2393][COMP][RT] Add source
location to error messages
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ConstantExpression.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ConstantExpression.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ConstantExpression.java
index 689d51c..42ff3c0 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ConstantExpression.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ConstantExpression.java
@@ -174,6 +174,7 @@ public final class ConstantExpression extends AbstractLogicalExpression {
annotationMap.forEach((key, value1) -> m.put(key, value1.copy()));
ConstantExpression c = new ConstantExpression(value);
c.annotationMap = m;
+ c.setSourceLocation(sourceLoc);
return c;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ScalarFunctionCallExpression.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ScalarFunctionCallExpression.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ScalarFunctionCallExpression.java
index 6308636..f8b25e2 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ScalarFunctionCallExpression.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/ScalarFunctionCallExpression.java
@@ -47,6 +47,7 @@ public class ScalarFunctionCallExpression extends AbstractFunctionCallExpression
ScalarFunctionCallExpression funcExpr = new ScalarFunctionCallExpression(finfo, clonedArgs);
funcExpr.getAnnotations().putAll(cloneAnnotations());
funcExpr.setOpaqueParameters(this.getOpaqueParameters());
+ funcExpr.setSourceLocation(sourceLoc);
return funcExpr;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/StatefulFunctionCallExpression.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/StatefulFunctionCallExpression.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/StatefulFunctionCallExpression.java
index 661c389..0d310f7 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/StatefulFunctionCallExpression.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/StatefulFunctionCallExpression.java
@@ -52,7 +52,10 @@ public class StatefulFunctionCallExpression extends AbstractFunctionCallExpressi
public StatefulFunctionCallExpression cloneExpression() {
cloneAnnotations();
List<Mutable<ILogicalExpression>> clonedArgs = cloneArguments();
- return new StatefulFunctionCallExpression(finfo, propertiesComputer, clonedArgs);
+ StatefulFunctionCallExpression clonedExpr =
+ new StatefulFunctionCallExpression(finfo, propertiesComputer, clonedArgs);
+ clonedExpr.setSourceLocation(sourceLoc);
+ return clonedExpr;
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/UnnestingFunctionCallExpression.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/UnnestingFunctionCallExpression.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/UnnestingFunctionCallExpression.java
index ac539ec..3dad669 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/UnnestingFunctionCallExpression.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/UnnestingFunctionCallExpression.java
@@ -50,6 +50,7 @@ public class UnnestingFunctionCallExpression extends AbstractFunctionCallExpress
UnnestingFunctionCallExpression ufce = new UnnestingFunctionCallExpression(finfo, clonedArgs);
ufce.setReturnsUniqueValues(returnsUniqueValues);
ufce.setOpaqueParameters(this.getOpaqueParameters());
+ ufce.setSourceLocation(sourceLoc);
return ufce;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/VariableReferenceExpression.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/VariableReferenceExpression.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/VariableReferenceExpression.java
index c1342cc..1c7da34 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/VariableReferenceExpression.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/expressions/VariableReferenceExpression.java
@@ -95,7 +95,9 @@ public class VariableReferenceExpression extends AbstractLogicalExpression {
@Override
public AbstractLogicalExpression cloneExpression() {
- return new VariableReferenceExpression(variable);
+ VariableReferenceExpression varRef = new VariableReferenceExpression(variable);
+ varRef.setSourceLocation(sourceLoc);
+ return varRef;
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/AbstractLogicalOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/AbstractLogicalOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/AbstractLogicalOperator.java
index 4686f32..1dbf15e 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/AbstractLogicalOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/AbstractLogicalOperator.java
@@ -40,6 +40,7 @@ import org.apache.hyracks.algebricks.core.algebra.typing.ITypingContext;
import org.apache.hyracks.algebricks.core.algebra.typing.OpRefTypeEnvPointer;
import org.apache.hyracks.algebricks.core.algebra.typing.PropagatingTypeEnvironment;
import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
+import org.apache.hyracks.api.exceptions.SourceLocation;
public abstract class AbstractLogicalOperator implements ILogicalOperator {
@@ -69,6 +70,8 @@ public abstract class AbstractLogicalOperator implements ILogicalOperator {
protected final List<Mutable<ILogicalOperator>> inputs;
protected List<LogicalVariable> schema;
+ private SourceLocation sourceLoc;
+
public AbstractLogicalOperator() {
inputs = new ArrayList<>();
}
@@ -195,4 +198,13 @@ public abstract class AbstractLogicalOperator implements ILogicalOperator {
public boolean requiresVariableReferenceExpressions() {
return true;
}
+
+ @Override
+ public SourceLocation getSourceLocation() {
+ return sourceLoc;
+ }
+
+ public void setSourceLocation(SourceLocation sourceLoc) {
+ this.sourceLoc = sourceLoc;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalExpressionDeepCopyWithNewVariablesVisitor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalExpressionDeepCopyWithNewVariablesVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalExpressionDeepCopyWithNewVariablesVisitor.java
index 24c5162..6be8ee0 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalExpressionDeepCopyWithNewVariablesVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalExpressionDeepCopyWithNewVariablesVisitor.java
@@ -30,6 +30,7 @@ import org.apache.hyracks.algebricks.core.algebra.base.ILogicalExpression;
import org.apache.hyracks.algebricks.core.algebra.base.IVariableContext;
import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
+import org.apache.hyracks.algebricks.core.algebra.expressions.AbstractLogicalExpression;
import org.apache.hyracks.algebricks.core.algebra.expressions.AggregateFunctionCallExpression;
import org.apache.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
import org.apache.hyracks.algebricks.core.algebra.expressions.IExpressionAnnotation;
@@ -80,6 +81,10 @@ public class LogicalExpressionDeepCopyWithNewVariablesVisitor
dest.setOpaqueParameters(newOpaqueParameters);
}
+ private void copySourceLocation(ILogicalExpression src, AbstractLogicalExpression dest) {
+ dest.setSourceLocation(src.getSourceLocation());
+ }
+
public MutableObject<ILogicalExpression> deepCopyExpressionReference(Mutable<ILogicalExpression> exprRef)
throws AlgebricksException {
return new MutableObject<>(deepCopy(exprRef.getValue()));
@@ -104,12 +109,15 @@ public class LogicalExpressionDeepCopyWithNewVariablesVisitor
exprCopy.setStepTwoAggregate(expr.getStepTwoAggregate());
deepCopyAnnotations(expr, exprCopy);
deepCopyOpaqueParameters(expr, exprCopy);
+ copySourceLocation(expr, exprCopy);
return exprCopy;
}
@Override
public ILogicalExpression visitConstantExpression(ConstantExpression expr, Void arg) throws AlgebricksException {
- return new ConstantExpression(expr.getValue());
+ ConstantExpression exprCopy = new ConstantExpression(expr.getValue());
+ copySourceLocation(expr, exprCopy);
+ return exprCopy;
}
@Override
@@ -119,6 +127,7 @@ public class LogicalExpressionDeepCopyWithNewVariablesVisitor
deepCopyExpressionReferenceList(expr.getArguments()));
deepCopyAnnotations(expr, exprCopy);
deepCopyOpaqueParameters(expr, exprCopy);
+ copySourceLocation(expr, exprCopy);
return exprCopy;
}
@@ -130,6 +139,7 @@ public class LogicalExpressionDeepCopyWithNewVariablesVisitor
expr.getPropertiesComputer(), deepCopyExpressionReferenceList(expr.getArguments()));
deepCopyAnnotations(expr, exprCopy);
deepCopyOpaqueParameters(expr, exprCopy);
+ copySourceLocation(expr, exprCopy);
return exprCopy;
}
@@ -140,6 +150,7 @@ public class LogicalExpressionDeepCopyWithNewVariablesVisitor
deepCopyExpressionReferenceList(expr.getArguments()));
deepCopyAnnotations(expr, exprCopy);
deepCopyOpaqueParameters(expr, exprCopy);
+ copySourceLocation(expr, exprCopy);
return exprCopy;
}
@@ -153,13 +164,17 @@ public class LogicalExpressionDeepCopyWithNewVariablesVisitor
LogicalVariable givenVarReplacement = inVarMapping.get(var);
if (givenVarReplacement != null) {
outVarMapping.put(var, givenVarReplacement);
- return new VariableReferenceExpression(givenVarReplacement);
+ VariableReferenceExpression varRef = new VariableReferenceExpression(givenVarReplacement);
+ copySourceLocation(expr, varRef);
+ return varRef;
}
LogicalVariable varCopy = outVarMapping.get(var);
if (varCopy == null) {
varCopy = varContext.newVar();
outVarMapping.put(var, varCopy);
}
- return new VariableReferenceExpression(varCopy);
+ VariableReferenceExpression varRef = new VariableReferenceExpression(varCopy);
+ copySourceLocation(expr, varRef);
+ return varRef;
}
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
index d6062ee..e0210cc 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/logical/visitors/LogicalOperatorDeepCopyWithNewVariablesVisitor.java
@@ -146,6 +146,10 @@ public class LogicalOperatorDeepCopyWithNewVariablesVisitor
dest.getAnnotations().putAll(src.getAnnotations());
}
+ private void copySourceLocation(ILogicalOperator src, AbstractLogicalOperator dest) {
+ dest.setSourceLocation(src.getSourceLocation());
+ }
+
public ILogicalOperator deepCopy(ILogicalOperator op) throws AlgebricksException {
// The deep copy call outside this visitor always has a null argument.
return deepCopy(op, null);
@@ -269,6 +273,7 @@ public class LogicalOperatorDeepCopyWithNewVariablesVisitor
AbstractLogicalOperator opCopy) throws AlgebricksException {
deepCopyInputs(op, opCopy, arg);
copyAnnotations(op, opCopy);
+ copySourceLocation(op, opCopy);
opCopy.setExecutionMode(op.getExecutionMode());
}
@@ -337,6 +342,7 @@ public class LogicalOperatorDeepCopyWithNewVariablesVisitor
@Override
public ILogicalOperator visitEmptyTupleSourceOperator(EmptyTupleSourceOperator op, ILogicalOperator arg) {
EmptyTupleSourceOperator opCopy = new EmptyTupleSourceOperator();
+ copySourceLocation(op, opCopy);
opCopy.setExecutionMode(op.getExecutionMode());
return opCopy;
}
@@ -371,6 +377,7 @@ public class LogicalOperatorDeepCopyWithNewVariablesVisitor
deepCopyOperatorReference(op.getInputs().get(0), arg),
deepCopyOperatorReference(op.getInputs().get(1), arg));
copyAnnotations(op, opCopy);
+ copySourceLocation(op, opCopy);
opCopy.setExecutionMode(op.getExecutionMode());
return opCopy;
}
@@ -383,6 +390,7 @@ public class LogicalOperatorDeepCopyWithNewVariablesVisitor
deepCopyOperatorReference(op.getInputs().get(0), arg),
deepCopyOperatorReference(op.getInputs().get(1), arg));
copyAnnotations(op, opCopy);
+ copySourceLocation(op, opCopy);
opCopy.setExecutionMode(op.getExecutionMode());
return opCopy;
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractUnnestPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractUnnestPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractUnnestPOperator.java
index 4a177f7..0602258 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractUnnestPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AbstractUnnestPOperator.java
@@ -81,6 +81,7 @@ public abstract class AbstractUnnestPOperator extends AbstractScanPOperator {
int[] projectionList = JobGenHelper.projectAllVariables(opSchema);
UnnestRuntimeFactory unnestRuntime = new UnnestRuntimeFactory(outCol, unnestingFactory, projectionList,
unnest.getPositionWriter(), leftOuter, context.getMissingWriterFactory());
+ unnestRuntime.setSourceLocation(unnest.getSourceLocation());
RecordDescriptor recDesc = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
builder.contributeMicroOperator(unnest, unnestRuntime, recDesc);
ILogicalOperator src = unnest.getInputs().get(0).getValue();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AggregatePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AggregatePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AggregatePOperator.java
index 147d5cc..2736b6f 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AggregatePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AggregatePOperator.java
@@ -102,6 +102,7 @@ public class AggregatePOperator extends AbstractPhysicalOperator {
}
AggregateRuntimeFactory runtime = new AggregateRuntimeFactory(aggFactories);
+ runtime.setSourceLocation(aggOp.getSourceLocation());
// contribute one Asterix framewriter
RecordDescriptor recDesc = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AssignPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AssignPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AssignPOperator.java
index ccd27f4..2df5a4e 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AssignPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/AssignPOperator.java
@@ -90,6 +90,7 @@ public class AssignPOperator extends AbstractPhysicalOperator {
AssignRuntimeFactory runtime =
new AssignRuntimeFactory(outColumns, evalFactories, projectionList, flushFramesRapidly);
+ runtime.setSourceLocation(assign.getSourceLocation());
// contribute one Asterix framewriter
RecordDescriptor recDesc = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
@@ -103,7 +104,6 @@ public class AssignPOperator extends AbstractPhysicalOperator {
// and contribute one edge from its child
ILogicalOperator src = assign.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, assign, 0);
-
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/BulkloadPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/BulkloadPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/BulkloadPOperator.java
index 2204637..bb0f08b 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/BulkloadPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/BulkloadPOperator.java
@@ -107,8 +107,10 @@ public class BulkloadPOperator extends AbstractPhysicalOperator {
Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> runtimeAndConstraints =
mp.getInsertRuntime(dataSource, propagatedSchema, typeEnv, primaryKeys, payload,
additionalFilteringKeys, additionalNonFilterVars, inputDesc, context, spec, true);
- builder.contributeHyracksOperator(insertDeleteOp, runtimeAndConstraints.first);
- builder.contributeAlgebricksPartitionConstraint(runtimeAndConstraints.first, runtimeAndConstraints.second);
+ IOperatorDescriptor opDesc = runtimeAndConstraints.first;
+ opDesc.setSourceLocation(insertDeleteOp.getSourceLocation());
+ builder.contributeHyracksOperator(insertDeleteOp, opDesc);
+ builder.contributeAlgebricksPartitionConstraint(opDesc, runtimeAndConstraints.second);
ILogicalOperator src = insertDeleteOp.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, insertDeleteOp, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
index e8c5c64..3ddb233 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DataSourceScanPOperator.java
@@ -119,9 +119,11 @@ public class DataSourceScanPOperator extends AbstractScanPOperator {
Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> p = mp.getScannerRuntime(dataSource, vars, projectVars,
scan.isProjectPushed(), scan.getMinFilterVars(), scan.getMaxFilterVars(), tupleFilterFactory,
scan.getOutputLimit(), opSchema, typeEnv, context, builder.getJobSpec(), implConfig);
- builder.contributeHyracksOperator(scan, p.first);
+ IOperatorDescriptor opDesc = p.first;
+ opDesc.setSourceLocation(scan.getSourceLocation());
+ builder.contributeHyracksOperator(scan, opDesc);
if (p.second != null) {
- builder.contributeAlgebricksPartitionConstraint(p.first, p.second);
+ builder.contributeAlgebricksPartitionConstraint(opDesc, p.second);
}
ILogicalOperator srcExchange = scan.getInputs().get(0).getValue();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DistributeResultPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DistributeResultPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DistributeResultPOperator.java
index 178f2a1..34dc0d4 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DistributeResultPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/DistributeResultPOperator.java
@@ -105,8 +105,10 @@ public class DistributeResultPOperator extends AbstractPhysicalOperator {
Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> runtimeAndConstraints =
mp.getResultHandleRuntime(resultOp.getDataSink(), columns, pf, inputDesc, true, spec);
+ IOperatorDescriptor opDesc = runtimeAndConstraints.first;
+ opDesc.setSourceLocation(resultOp.getSourceLocation());
+ builder.contributeHyracksOperator(resultOp, opDesc);
- builder.contributeHyracksOperator(resultOp, runtimeAndConstraints.first);
ILogicalOperator src = resultOp.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, resultOp, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/EmptyTupleSourcePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/EmptyTupleSourcePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/EmptyTupleSourcePOperator.java
index 0718d13..502d022 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/EmptyTupleSourcePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/EmptyTupleSourcePOperator.java
@@ -63,6 +63,7 @@ public class EmptyTupleSourcePOperator extends AbstractPhysicalOperator {
IOperatorSchema propagatedSchema, IOperatorSchema[] inputSchemas, IOperatorSchema outerPlanSchema)
throws AlgebricksException {
EmptyTupleSourceRuntimeFactory runtime = new EmptyTupleSourceRuntimeFactory();
+ runtime.setSourceLocation(op.getSourceLocation());
RecordDescriptor recDesc = new RecordDescriptor(new ISerializerDeserializer[] {});
builder.contributeMicroOperator(op, runtime, recDesc);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ExternalGroupByPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ExternalGroupByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ExternalGroupByPOperator.java
index 5ee967d..edc9d08 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ExternalGroupByPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ExternalGroupByPOperator.java
@@ -62,6 +62,7 @@ import org.apache.hyracks.api.dataflow.value.IBinaryHashFunctionFamily;
import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
+import org.apache.hyracks.dataflow.std.group.AbstractAggregatorDescriptorFactory;
import org.apache.hyracks.dataflow.std.group.HashSpillableTableFactory;
import org.apache.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
import org.apache.hyracks.dataflow.std.group.external.ExternalGroupOperatorDescriptor;
@@ -253,8 +254,10 @@ public class ExternalGroupByPOperator extends AbstractPhysicalOperator {
merges[i] = expressionRuntimeProvider.createSerializableAggregateFunctionFactory(mergeFun, aggOpInputEnv,
localInputSchemas, context);
}
- IAggregatorDescriptorFactory aggregatorFactory = new SerializableAggregatorDescriptorFactory(aff);
- IAggregatorDescriptorFactory mergeFactory = new SerializableAggregatorDescriptorFactory(merges);
+ AbstractAggregatorDescriptorFactory aggregatorFactory = new SerializableAggregatorDescriptorFactory(aff);
+ aggregatorFactory.setSourceLocation(gby.getSourceLocation());
+ AbstractAggregatorDescriptorFactory mergeFactory = new SerializableAggregatorDescriptorFactory(merges);
+ mergeFactory.setSourceLocation(gby.getSourceLocation());
INormalizedKeyComputerFactory normalizedKeyFactory =
JobGenHelper.variablesToAscNormalizedKeyComputerFactory(gbyCols, aggOpInputEnv, context);
@@ -268,6 +271,7 @@ public class ExternalGroupByPOperator extends AbstractPhysicalOperator {
ExternalGroupOperatorDescriptor gbyOpDesc = new ExternalGroupOperatorDescriptor(spec, hashTableSize, inputSize,
keyAndDecFields, frameLimit, comparatorFactories, normalizedKeyFactory, aggregatorFactory, mergeFactory,
recordDescriptor, recordDescriptor, new HashSpillableTableFactory(hashFunctionFactories));
+ gbyOpDesc.setSourceLocation(gby.getSourceLocation());
contributeOpDesc(builder, gby, gbyOpDesc);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/HybridHashJoinPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/HybridHashJoinPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/HybridHashJoinPOperator.java
index 301b8f1..d06dde5 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/HybridHashJoinPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/HybridHashJoinPOperator.java
@@ -152,6 +152,7 @@ public class HybridHashJoinPOperator extends AbstractHashJoinPOperator {
opDesc = generateHashJoinRuntime(context, inputSchemas, keysLeft, keysRight, hashFunFactories,
comparatorFactories, predEvaluatorFactory, recDescriptor, spec);
}
+ opDesc.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);
ILogicalOperator src1 = op.getInputs().get(0).getValue();
@@ -164,68 +165,49 @@ public class HybridHashJoinPOperator extends AbstractHashJoinPOperator {
int[] keysLeft, int[] keysRight, IBinaryHashFunctionFactory[] hashFunFactories,
IBinaryComparatorFactory[] comparatorFactories, IPredicateEvaluatorFactory predEvaluatorFactory,
RecordDescriptor recDescriptor, IOperatorDescriptorRegistry spec) throws AlgebricksException {
- IOperatorDescriptor opDesc;
- try {
- switch (kind) {
- case INNER:
- opDesc = new HybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(), maxInputBuildSizeInFrames,
- aveRecordsPerFrame, getFudgeFactor(), keysLeft, keysRight, hashFunFactories,
- comparatorFactories, recDescriptor, predEvaluatorFactory, false, null);
- break;
- case LEFT_OUTER:
- IMissingWriterFactory[] nonMatchWriterFactories =
- new IMissingWriterFactory[inputSchemas[1].getSize()];
- for (int j = 0; j < nonMatchWriterFactories.length; j++) {
- nonMatchWriterFactories[j] = context.getMissingWriterFactory();
- }
- opDesc = new HybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(), maxInputBuildSizeInFrames,
- aveRecordsPerFrame, getFudgeFactor(), keysLeft, keysRight, hashFunFactories,
- comparatorFactories, recDescriptor, predEvaluatorFactory, true, nonMatchWriterFactories);
- break;
- default:
- throw new NotImplementedException();
- }
- } catch (HyracksDataException e) {
- throw new AlgebricksException(e);
+ switch (kind) {
+ case INNER:
+ return new HybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(), maxInputBuildSizeInFrames,
+ aveRecordsPerFrame, getFudgeFactor(), keysLeft, keysRight, hashFunFactories,
+ comparatorFactories, recDescriptor, predEvaluatorFactory, false, null);
+ case LEFT_OUTER:
+ IMissingWriterFactory[] nonMatchWriterFactories = new IMissingWriterFactory[inputSchemas[1].getSize()];
+ for (int j = 0; j < nonMatchWriterFactories.length; j++) {
+ nonMatchWriterFactories[j] = context.getMissingWriterFactory();
+ }
+ return new HybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(), maxInputBuildSizeInFrames,
+ aveRecordsPerFrame, getFudgeFactor(), keysLeft, keysRight, hashFunFactories,
+ comparatorFactories, recDescriptor, predEvaluatorFactory, true, nonMatchWriterFactories);
+ default:
+ throw new NotImplementedException();
}
- return opDesc;
}
private IOperatorDescriptor generateOptimizedHashJoinRuntime(JobGenContext context, IOperatorSchema[] inputSchemas,
int[] keysLeft, int[] keysRight, IBinaryHashFunctionFamily[] hashFunFamilies,
IBinaryComparatorFactory[] comparatorFactories, IPredicateEvaluatorFactory predEvaluatorFactory,
RecordDescriptor recDescriptor, IOperatorDescriptorRegistry spec) throws AlgebricksException {
- IOperatorDescriptor opDesc;
- try {
- switch (kind) {
- case INNER:
- opDesc = new OptimizedHybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
- maxInputBuildSizeInFrames, getFudgeFactor(), keysLeft, keysRight, hashFunFamilies,
- comparatorFactories, recDescriptor,
- new JoinMultiComparatorFactory(comparatorFactories, keysLeft, keysRight),
- new JoinMultiComparatorFactory(comparatorFactories, keysRight, keysLeft),
- predEvaluatorFactory);
- break;
- case LEFT_OUTER:
- IMissingWriterFactory[] nonMatchWriterFactories =
- new IMissingWriterFactory[inputSchemas[1].getSize()];
- for (int j = 0; j < nonMatchWriterFactories.length; j++) {
- nonMatchWriterFactories[j] = context.getMissingWriterFactory();
- }
- opDesc = new OptimizedHybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
- maxInputBuildSizeInFrames, getFudgeFactor(), keysLeft, keysRight, hashFunFamilies,
- comparatorFactories, recDescriptor,
- new JoinMultiComparatorFactory(comparatorFactories, keysLeft, keysRight),
- new JoinMultiComparatorFactory(comparatorFactories, keysRight, keysLeft),
- predEvaluatorFactory, true, nonMatchWriterFactories);
- break;
- default:
- throw new NotImplementedException();
- }
- } catch (HyracksDataException e) {
- throw new AlgebricksException(e);
+ switch (kind) {
+ case INNER:
+ return new OptimizedHybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
+ maxInputBuildSizeInFrames, getFudgeFactor(), keysLeft, keysRight, hashFunFamilies,
+ comparatorFactories, recDescriptor,
+ new JoinMultiComparatorFactory(comparatorFactories, keysLeft, keysRight),
+ new JoinMultiComparatorFactory(comparatorFactories, keysRight, keysLeft), predEvaluatorFactory);
+ case LEFT_OUTER:
+ IMissingWriterFactory[] nonMatchWriterFactories = new IMissingWriterFactory[inputSchemas[1].getSize()];
+ for (int j = 0; j < nonMatchWriterFactories.length; j++) {
+ nonMatchWriterFactories[j] = context.getMissingWriterFactory();
+ }
+ return new OptimizedHybridHashJoinOperatorDescriptor(spec, getMemSizeInFrames(),
+ maxInputBuildSizeInFrames, getFudgeFactor(), keysLeft, keysRight, hashFunFamilies,
+ comparatorFactories, recDescriptor,
+ new JoinMultiComparatorFactory(comparatorFactories, keysLeft, keysRight),
+ new JoinMultiComparatorFactory(comparatorFactories, keysRight, keysLeft), predEvaluatorFactory,
+ true, nonMatchWriterFactories);
+ default:
+ throw new NotImplementedException();
}
- return opDesc;
}
@Override
@@ -264,7 +246,7 @@ public class HybridHashJoinPOperator extends AbstractHashJoinPOperator {
}
/**
- * {@ ITuplePairComparatorFactory} implementation for optimized hybrid hash join.
+ * {@code ITuplePairComparatorFactory} implementation for optimized hybrid hash join.
*/
class JoinMultiComparatorFactory implements ITuplePairComparatorFactory {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InMemoryHashJoinPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InMemoryHashJoinPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InMemoryHashJoinPOperator.java
index 9c29c53..580d8e1 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InMemoryHashJoinPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InMemoryHashJoinPOperator.java
@@ -105,15 +105,14 @@ public class InMemoryHashJoinPOperator extends AbstractHashJoinPOperator {
RecordDescriptor recDescriptor =
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), propagatedSchema, context);
IOperatorDescriptorRegistry spec = builder.getJobSpec();
- IOperatorDescriptor opDesc = null;
+ IOperatorDescriptor opDesc;
switch (kind) {
- case INNER: {
+ case INNER:
opDesc = new InMemoryHashJoinOperatorDescriptor(spec, keysLeft, keysRight, hashFunFactories,
comparatorFactories, recDescriptor, tableSize, predEvaluatorFactory, memSizeInFrames);
break;
- }
- case LEFT_OUTER: {
+ case LEFT_OUTER:
IMissingWriterFactory[] nonMatchWriterFactories = new IMissingWriterFactory[inputSchemas[1].getSize()];
for (int j = 0; j < nonMatchWriterFactories.length; j++) {
nonMatchWriterFactories[j] = context.getMissingWriterFactory();
@@ -122,11 +121,11 @@ public class InMemoryHashJoinPOperator extends AbstractHashJoinPOperator {
comparatorFactories, predEvaluatorFactory, recDescriptor, true, nonMatchWriterFactories,
tableSize, memSizeInFrames);
break;
- }
- default: {
+ default:
throw new NotImplementedException();
- }
}
+
+ opDesc.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);
ILogicalOperator src1 = op.getInputs().get(0).getValue();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
index fa0fb1a..6512700 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexBulkloadPOperator.java
@@ -135,8 +135,10 @@ public class IndexBulkloadPOperator extends AbstractPhysicalOperator {
Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> runtimeAndConstraints =
mp.getIndexInsertRuntime(dataSourceIndex, propagatedSchema, inputSchemas, typeEnv, primaryKeys,
secondaryKeys, additionalFilteringKeys, filterExpr, inputDesc, context, spec, true);
- builder.contributeHyracksOperator(indexInsertDeleteOp, runtimeAndConstraints.first);
- builder.contributeAlgebricksPartitionConstraint(runtimeAndConstraints.first, runtimeAndConstraints.second);
+ IOperatorDescriptor opDesc = runtimeAndConstraints.first;
+ opDesc.setSourceLocation(indexInsertDeleteOp.getSourceLocation());
+ builder.contributeHyracksOperator(indexInsertDeleteOp, opDesc);
+ builder.contributeAlgebricksPartitionConstraint(opDesc, runtimeAndConstraints.second);
ILogicalOperator src = indexInsertDeleteOp.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, indexInsertDeleteOp, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexInsertDeleteUpsertPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexInsertDeleteUpsertPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexInsertDeleteUpsertPOperator.java
index a66db35..92fa86d 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexInsertDeleteUpsertPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IndexInsertDeleteUpsertPOperator.java
@@ -118,19 +118,30 @@ public class IndexInsertDeleteUpsertPOperator extends AbstractPhysicalOperator {
Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> runtimeAndConstraints = null;
IVariableTypeEnvironment typeEnv = context.getTypeEnvironment(insertDeleteUpsertOp);
- if (insertDeleteUpsertOp.getOperation() == Kind.INSERT) {
- runtimeAndConstraints = mp.getIndexInsertRuntime(dataSourceIndex, propagatedSchema, inputSchemas, typeEnv,
- primaryKeys, secondaryKeys, additionalFilteringKeys, filterExpr, inputDesc, context, spec, false);
- } else if (insertDeleteUpsertOp.getOperation() == Kind.DELETE) {
- runtimeAndConstraints = mp.getIndexDeleteRuntime(dataSourceIndex, propagatedSchema, inputSchemas, typeEnv,
- primaryKeys, secondaryKeys, additionalFilteringKeys, filterExpr, inputDesc, context, spec);
- } else if (insertDeleteUpsertOp.getOperation() == Kind.UPSERT) {
- runtimeAndConstraints = mp.getIndexUpsertRuntime(dataSourceIndex, propagatedSchema, inputSchemas, typeEnv,
- primaryKeys, secondaryKeys, additionalFilteringKeys, filterExpr, prevSecondaryKeys,
- prevAdditionalFilteringKey, inputDesc, context, spec);
+ Kind operation = insertDeleteUpsertOp.getOperation();
+ switch (operation) {
+ case INSERT:
+ runtimeAndConstraints =
+ mp.getIndexInsertRuntime(dataSourceIndex, propagatedSchema, inputSchemas, typeEnv, primaryKeys,
+ secondaryKeys, additionalFilteringKeys, filterExpr, inputDesc, context, spec, false);
+ break;
+ case DELETE:
+ runtimeAndConstraints =
+ mp.getIndexDeleteRuntime(dataSourceIndex, propagatedSchema, inputSchemas, typeEnv, primaryKeys,
+ secondaryKeys, additionalFilteringKeys, filterExpr, inputDesc, context, spec);
+ break;
+ case UPSERT:
+ runtimeAndConstraints = mp.getIndexUpsertRuntime(dataSourceIndex, propagatedSchema, inputSchemas,
+ typeEnv, primaryKeys, secondaryKeys, additionalFilteringKeys, filterExpr, prevSecondaryKeys,
+ prevAdditionalFilteringKey, inputDesc, context, spec);
+ break;
+ default:
+ throw new AlgebricksException("Unsupported Operation " + operation);
}
- builder.contributeHyracksOperator(insertDeleteUpsertOp, runtimeAndConstraints.first);
- builder.contributeAlgebricksPartitionConstraint(runtimeAndConstraints.first, runtimeAndConstraints.second);
+ IOperatorDescriptor opDesc = runtimeAndConstraints.first;
+ opDesc.setSourceLocation(insertDeleteUpsertOp.getSourceLocation());
+ builder.contributeHyracksOperator(insertDeleteUpsertOp, opDesc);
+ builder.contributeAlgebricksPartitionConstraint(opDesc, runtimeAndConstraints.second);
ILogicalOperator src = insertDeleteUpsertOp.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, insertDeleteUpsertOp, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
index 6ded4a3..927fb66 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/InsertDeleteUpsertPOperator.java
@@ -106,22 +106,28 @@ public class InsertDeleteUpsertPOperator extends AbstractPhysicalOperator {
RecordDescriptor inputDesc = JobGenHelper.mkRecordDescriptor(
context.getTypeEnvironment(op.getInputs().get(0).getValue()), inputSchemas[0], context);
- Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> runtimeAndConstraints = null;
- if (operation == Kind.INSERT) {
- runtimeAndConstraints = mp.getInsertRuntime(dataSource, propagatedSchema, typeEnv, keys, payload,
- additionalFilteringKeys, additionalNonFilteringFields, inputDesc, context, spec, false);
- } else if (operation == Kind.DELETE) {
- runtimeAndConstraints = mp.getDeleteRuntime(dataSource, propagatedSchema, typeEnv, keys, payload,
- additionalFilteringKeys, inputDesc, context, spec);
- } else if (operation == Kind.UPSERT) {
- runtimeAndConstraints = mp.getUpsertRuntime(dataSource, inputSchemas[0], typeEnv, keys, payload,
- additionalFilteringKeys, additionalNonFilteringFields, inputDesc, context, spec);
- } else {
- throw new AlgebricksException("Unsupported Operation " + operation);
+ Pair<IOperatorDescriptor, AlgebricksPartitionConstraint> runtimeAndConstraints;
+ switch (operation) {
+ case INSERT:
+ runtimeAndConstraints = mp.getInsertRuntime(dataSource, propagatedSchema, typeEnv, keys, payload,
+ additionalFilteringKeys, additionalNonFilteringFields, inputDesc, context, spec, false);
+ break;
+ case DELETE:
+ runtimeAndConstraints = mp.getDeleteRuntime(dataSource, propagatedSchema, typeEnv, keys, payload,
+ additionalFilteringKeys, inputDesc, context, spec);
+ break;
+ case UPSERT:
+ runtimeAndConstraints = mp.getUpsertRuntime(dataSource, inputSchemas[0], typeEnv, keys, payload,
+ additionalFilteringKeys, additionalNonFilteringFields, inputDesc, context, spec);
+ break;
+ default:
+ throw new AlgebricksException("Unsupported Operation " + operation);
}
- builder.contributeHyracksOperator(insertDeleteOp, runtimeAndConstraints.first);
- builder.contributeAlgebricksPartitionConstraint(runtimeAndConstraints.first, runtimeAndConstraints.second);
+ IOperatorDescriptor opDesc = runtimeAndConstraints.first;
+ opDesc.setSourceLocation(insertDeleteOp.getSourceLocation());
+ builder.contributeHyracksOperator(insertDeleteOp, opDesc);
+ builder.contributeAlgebricksPartitionConstraint(opDesc, runtimeAndConstraints.second);
ILogicalOperator src = insertDeleteOp.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, insertDeleteOp, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IntersectPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IntersectPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IntersectPOperator.java
index 1d36cc0..544c546 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IntersectPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/IntersectPOperator.java
@@ -151,7 +151,7 @@ public class IntersectPOperator extends AbstractPhysicalOperator {
} catch (HyracksException e) {
throw new AlgebricksException(e);
}
-
+ opDescriptor.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, opDescriptor);
for (int i = 0; i < op.getInputs().size(); i++) {
builder.contributeGraphEdge(op.getInputs().get(i).getValue(), 0, op, i);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MaterializePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MaterializePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MaterializePOperator.java
index a48e3c2..fe8985a 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MaterializePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MaterializePOperator.java
@@ -74,6 +74,7 @@ public class MaterializePOperator extends AbstractPhysicalOperator {
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), propagatedSchema, context);
MaterializingOperatorDescriptor materializationOpDesc =
new MaterializingOperatorDescriptor(builder.getJobSpec(), recDescriptor, isSingleActivity);
+ materializationOpDesc.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, materializationOpDesc);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreSortedDistinctByPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreSortedDistinctByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreSortedDistinctByPOperator.java
index 94d5fd4..fda879c 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreSortedDistinctByPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreSortedDistinctByPOperator.java
@@ -34,7 +34,7 @@ import org.apache.hyracks.algebricks.runtime.operators.aggreg.SimpleAlgebricksAc
import org.apache.hyracks.algebricks.runtime.operators.group.MicroPreClusteredGroupRuntimeFactory;
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
-import org.apache.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
+import org.apache.hyracks.dataflow.std.group.AbstractAggregatorDescriptorFactory;
public class MicroPreSortedDistinctByPOperator extends AbstractPreSortedDistinctByPOperator {
@@ -62,8 +62,9 @@ public class MicroPreSortedDistinctByPOperator extends AbstractPreSortedDistinct
IBinaryComparatorFactory[] comparatorFactories = JobGenHelper
.variablesToAscBinaryComparatorFactories(columnList, context.getTypeEnvironment(op), context);
IAggregateEvaluatorFactory[] aggFactories = new IAggregateEvaluatorFactory[] {};
- IAggregatorDescriptorFactory aggregatorFactory =
+ AbstractAggregatorDescriptorFactory aggregatorFactory =
new SimpleAlgebricksAccumulatingAggregatorFactory(aggFactories, keysAndDecs);
+ aggregatorFactory.setSourceLocation(op.getSourceLocation());
RecordDescriptor recordDescriptor =
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
@@ -73,6 +74,7 @@ public class MicroPreSortedDistinctByPOperator extends AbstractPreSortedDistinct
/* make fd columns part of the key but the comparator only compares the distinct key columns */
MicroPreClusteredGroupRuntimeFactory runtime = new MicroPreClusteredGroupRuntimeFactory(keysAndDecs,
comparatorFactories, aggregatorFactory, inputRecordDesc, recordDescriptor, null);
+ runtime.setSourceLocation(op.getSourceLocation());
builder.contributeMicroOperator(op, runtime, recordDescriptor);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreclusteredGroupByPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreclusteredGroupByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreclusteredGroupByPOperator.java
index 629afa3..13308a1 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreclusteredGroupByPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroPreclusteredGroupByPOperator.java
@@ -74,6 +74,7 @@ public class MicroPreclusteredGroupByPOperator extends AbstractPreclusteredGroup
context.getTypeEnvironment(op.getInputs().get(0).getValue()), inputSchemas[0], context);
MicroPreClusteredGroupRuntimeFactory runtime = new MicroPreClusteredGroupRuntimeFactory(keys,
comparatorFactories, aggregatorFactory, inputRecordDesc, recordDescriptor, null);
+ runtime.setSourceLocation(gby.getSourceLocation());
builder.contributeMicroOperator(gby, runtime, recordDescriptor);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroUnionAllPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroUnionAllPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroUnionAllPOperator.java
index f5e992e..da48049 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroUnionAllPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroUnionAllPOperator.java
@@ -49,6 +49,7 @@ public class MicroUnionAllPOperator extends AbstractUnionAllPOperator {
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
MicroUnionAllRuntimeFactory runtime = new MicroUnionAllRuntimeFactory(op.getInputs().size());
+ runtime.setSourceLocation(op.getSourceLocation());
builder.contributeMicroOperator(op, runtime, recordDescriptor);
super.contributeRuntimeOperator(builder, context, op, opSchema, inputSchemas, outerPlanSchema);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedLoopJoinPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedLoopJoinPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedLoopJoinPOperator.java
index 4d7bd7e..83cea4a 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedLoopJoinPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedLoopJoinPOperator.java
@@ -139,15 +139,14 @@ public class NestedLoopJoinPOperator extends AbstractJoinPOperator {
ITuplePairComparatorFactory comparatorFactory =
new TuplePairEvaluatorFactory(cond, context.getBinaryBooleanInspectorFactory());
IOperatorDescriptorRegistry spec = builder.getJobSpec();
- IOperatorDescriptor opDesc = null;
+ IOperatorDescriptor opDesc;
switch (kind) {
- case INNER: {
+ case INNER:
opDesc = new NestedLoopJoinOperatorDescriptor(spec, comparatorFactory, recDescriptor, memSize, false,
null);
break;
- }
- case LEFT_OUTER: {
+ case LEFT_OUTER:
IMissingWriterFactory[] nonMatchWriterFactories = new IMissingWriterFactory[inputSchemas[1].getSize()];
for (int j = 0; j < nonMatchWriterFactories.length; j++) {
nonMatchWriterFactories[j] = context.getMissingWriterFactory();
@@ -155,12 +154,12 @@ public class NestedLoopJoinPOperator extends AbstractJoinPOperator {
opDesc = new NestedLoopJoinOperatorDescriptor(spec, comparatorFactory, recDescriptor, memSize, true,
nonMatchWriterFactories);
break;
- }
- default: {
+ default:
throw new NotImplementedException();
- }
}
- contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);
+
+ opDesc.setSourceLocation(join.getSourceLocation());
+ contributeOpDesc(builder, join, opDesc);
ILogicalOperator src1 = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src1, 0, op, 0);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
index 179bb73..7971e78 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/NestedTupleSourcePOperator.java
@@ -98,6 +98,7 @@ public class NestedTupleSourcePOperator extends AbstractPhysicalOperator {
throws AlgebricksException {
propagatedSchema.addAllVariables(outerPlanSchema);
NestedTupleSourceRuntimeFactory runtime = new NestedTupleSourceRuntimeFactory();
+ runtime.setSourceLocation(op.getSourceLocation());
RecordDescriptor recDesc =
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), propagatedSchema, context);
builder.contributeMicroOperator(op, runtime, recDesc);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
index 54e577f..dd4c65f 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreSortedDistinctByPOperator.java
@@ -34,7 +34,7 @@ import org.apache.hyracks.algebricks.runtime.operators.aggreg.SimpleAlgebricksAc
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
+import org.apache.hyracks.dataflow.std.group.AbstractAggregatorDescriptorFactory;
import org.apache.hyracks.dataflow.std.group.preclustered.PreclusteredGroupOperatorDescriptor;
public class PreSortedDistinctByPOperator extends AbstractPreSortedDistinctByPOperator {
@@ -64,14 +64,16 @@ public class PreSortedDistinctByPOperator extends AbstractPreSortedDistinctByPOp
IBinaryComparatorFactory[] comparatorFactories = JobGenHelper
.variablesToAscBinaryComparatorFactories(columnList, context.getTypeEnvironment(op), context);
IAggregateEvaluatorFactory[] aggFactories = new IAggregateEvaluatorFactory[] {};
- IAggregatorDescriptorFactory aggregatorFactory =
+ AbstractAggregatorDescriptorFactory aggregatorFactory =
new SimpleAlgebricksAccumulatingAggregatorFactory(aggFactories, keysAndDecs);
+ aggregatorFactory.setSourceLocation(op.getSourceLocation());
RecordDescriptor recordDescriptor =
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
/* make fd columns part of the key but the comparator only compares the distinct key columns */
PreclusteredGroupOperatorDescriptor opDesc = new PreclusteredGroupOperatorDescriptor(spec, keysAndDecs,
comparatorFactories, aggregatorFactory, recordDescriptor);
+ opDesc.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreclusteredGroupByPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreclusteredGroupByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreclusteredGroupByPOperator.java
index 0e0953c..e5076ce 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreclusteredGroupByPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/PreclusteredGroupByPOperator.java
@@ -27,7 +27,6 @@ import org.apache.hyracks.algebricks.core.algebra.base.ILogicalPlan;
import org.apache.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.GroupByOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
@@ -38,7 +37,7 @@ import org.apache.hyracks.algebricks.runtime.operators.aggreg.NestedPlansRunning
import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
+import org.apache.hyracks.dataflow.std.group.AbstractAggregatorDescriptorFactory;
import org.apache.hyracks.dataflow.std.group.preclustered.PreclusteredGroupOperatorDescriptor;
public class PreclusteredGroupByPOperator extends AbstractPreclusteredGroupByPOperator {
@@ -71,7 +70,7 @@ public class PreclusteredGroupByPOperator extends AbstractPreclusteredGroupByPOp
int fdColumns[] = getFdColumns(gby, inputSchemas[0]);
// compile subplans and set the gby op. schema accordingly
AlgebricksPipeline[] subplans = compileSubplans(inputSchemas[0], gby, opSchema, context);
- IAggregatorDescriptorFactory aggregatorFactory;
+ AbstractAggregatorDescriptorFactory aggregatorFactory;
List<ILogicalPlan> nestedPlans = gby.getNestedPlans();
if (!nestedPlans.isEmpty() && nestedPlans.get(0).getRoots().get(0).getValue()
@@ -80,6 +79,7 @@ public class PreclusteredGroupByPOperator extends AbstractPreclusteredGroupByPOp
} else {
aggregatorFactory = new NestedPlansAccumulatingAggregatorFactory(subplans, keys, fdColumns);
}
+ aggregatorFactory.setSourceLocation(gby.getSourceLocation());
IOperatorDescriptorRegistry spec = builder.getJobSpec();
IBinaryComparatorFactory[] comparatorFactories = JobGenHelper
@@ -89,8 +89,9 @@ public class PreclusteredGroupByPOperator extends AbstractPreclusteredGroupByPOp
PreclusteredGroupOperatorDescriptor opDesc = new PreclusteredGroupOperatorDescriptor(spec, keys,
comparatorFactories, aggregatorFactory, recordDescriptor, groupAll, framesLimit);
+ opDesc.setSourceLocation(gby.getSourceLocation());
- contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);
+ contributeOpDesc(builder, gby, opDesc);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ReplicatePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ReplicatePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ReplicatePOperator.java
index 25d31d2..f34cecc 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ReplicatePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/ReplicatePOperator.java
@@ -22,7 +22,6 @@ import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.core.algebra.base.IHyracksJobBuilder;
import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.ReplicateOperator;
import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
@@ -50,9 +49,10 @@ public class ReplicatePOperator extends AbstractReplicatePOperator {
int outputArity = rop.getOutputArity();
boolean[] outputMaterializationFlags = rop.getOutputMaterializationFlags();
- ReplicateOperatorDescriptor splitOpDesc =
+ ReplicateOperatorDescriptor ropDesc =
new ReplicateOperatorDescriptor(spec, recDescriptor, outputArity, outputMaterializationFlags);
- contributeOpDesc(builder, (AbstractLogicalOperator) op, splitOpDesc);
+ ropDesc.setSourceLocation(rop.getSourceLocation());
+ contributeOpDesc(builder, rop, ropDesc);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/RunningAggregatePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/RunningAggregatePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/RunningAggregatePOperator.java
index 3a6ba74..d68be20 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/RunningAggregatePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/RunningAggregatePOperator.java
@@ -91,6 +91,7 @@ public class RunningAggregatePOperator extends AbstractPhysicalOperator {
RunningAggregateRuntimeFactory runtime =
new RunningAggregateRuntimeFactory(outColumns, runningAggFuns, projectionList);
+ runtime.setSourceLocation(ragg.getSourceLocation());
// contribute one Asterix framewriter
RecordDescriptor recDesc = JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkPOperator.java
index 5084c18..6c9f6e2 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkPOperator.java
@@ -79,6 +79,7 @@ public class SinkPOperator extends AbstractPhysicalOperator {
IOperatorDescriptorRegistry spec = builder.getJobSpec();
SinkOperatorDescriptor opDesc = new SinkOperatorDescriptor(spec, op.getInputs().size());
+ opDesc.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, opDesc);
for (int i = 0; i < op.getInputs().size(); i++) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkWritePOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkWritePOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkWritePOperator.java
index f76b69b..6e37b3a 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkWritePOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SinkWritePOperator.java
@@ -101,10 +101,12 @@ public class SinkWritePOperator extends AbstractPhysicalOperator {
IMetadataProvider<?, ?> mp = context.getMetadataProvider();
- Pair<IPushRuntimeFactory, AlgebricksPartitionConstraint> runtime =
+ Pair<IPushRuntimeFactory, AlgebricksPartitionConstraint> runtimeAndConstraints =
mp.getWriteFileRuntime(write.getDataSink(), columns, pf, inputDesc);
+ IPushRuntimeFactory runtime = runtimeAndConstraints.first;
+ runtime.setSourceLocation(write.getSourceLocation());
- builder.contributeMicroOperator(write, runtime.first, recDesc, runtime.second);
+ builder.contributeMicroOperator(write, runtime, recDesc, runtimeAndConstraints.second);
ILogicalOperator src = write.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, write, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
index 1aeeca9..967e7e6 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortGroupByPOperator.java
@@ -63,7 +63,7 @@ import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
-import org.apache.hyracks.dataflow.std.group.IAggregatorDescriptorFactory;
+import org.apache.hyracks.dataflow.std.group.AbstractAggregatorDescriptorFactory;
import org.apache.hyracks.dataflow.std.group.sort.SortGroupByOperatorDescriptor;
public class SortGroupByPOperator extends AbstractPhysicalOperator {
@@ -258,10 +258,12 @@ public class SortGroupByPOperator extends AbstractPhysicalOperator {
RecordDescriptor partialAggRecordDescriptor =
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), localInputSchemas[0], context);
- IAggregatorDescriptorFactory aggregatorFactory =
+ AbstractAggregatorDescriptorFactory aggregatorFactory =
new SimpleAlgebricksAccumulatingAggregatorFactory(aff, keyAndDecFields);
- IAggregatorDescriptorFactory mergeFactory =
+ aggregatorFactory.setSourceLocation(gby.getSourceLocation());
+ AbstractAggregatorDescriptorFactory mergeFactory =
new SimpleAlgebricksAccumulatingAggregatorFactory(merges, keyAndDecFields);
+ mergeFactory.setSourceLocation(gby.getSourceLocation());
INormalizedKeyComputerFactory normalizedKeyFactory = null;
INormalizedKeyComputerFactoryProvider nkcfProvider = context.getNormalizedKeyComputerFactoryProvider();
@@ -275,6 +277,7 @@ public class SortGroupByPOperator extends AbstractPhysicalOperator {
SortGroupByOperatorDescriptor gbyOpDesc = new SortGroupByOperatorDescriptor(spec, frameLimit, keys,
keyAndDecFields, normalizedKeyFactory, compFactories, aggregatorFactory, mergeFactory,
partialAggRecordDescriptor, recordDescriptor, false);
+ gbyOpDesc.setSourceLocation(gby.getSourceLocation());
contributeOpDesc(builder, gby, gbyOpDesc);
ILogicalOperator src = op.getInputs().get(0).getValue();
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SplitPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SplitPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SplitPOperator.java
index c9fde4b..2f02eba 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SplitPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SplitPOperator.java
@@ -60,10 +60,11 @@ public class SplitPOperator extends AbstractReplicatePOperator {
IBinaryIntegerInspectorFactory intInsepctorFactory = context.getBinaryIntegerInspectorFactory();
- SplitOperatorDescriptor splitOpDesc = new SplitOperatorDescriptor(spec, recDescriptor, outputArity,
+ SplitOperatorDescriptor sopDesc = new SplitOperatorDescriptor(spec, recDescriptor, outputArity,
brachingExprEvalFactory, intInsepctorFactory, defaultBranch, propageToAllBranchAsDefault);
+ sopDesc.setSourceLocation(sop.getSourceLocation());
- contributeOpDesc(builder, (AbstractLogicalOperator) op, splitOpDesc);
+ contributeOpDesc(builder, sop, sopDesc);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
index 3a4249b..269a809 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
@@ -37,6 +37,7 @@ import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
+import org.apache.hyracks.dataflow.std.sort.AbstractSorterOperatorDescriptor;
import org.apache.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
import org.apache.hyracks.dataflow.std.sort.TopKSorterOperatorDescriptor;
@@ -98,22 +99,21 @@ public class StableSortPOperator extends AbstractStableSortPOperator {
i++;
}
+ AbstractSorterOperatorDescriptor sortOpDesc;
// topK == -1 means that a topK value is not provided.
if (topK == -1) {
- ExternalSortOperatorDescriptor sortOpDesc =
+ sortOpDesc =
new ExternalSortOperatorDescriptor(spec, maxNumberOfFrames, sortFields, nkcf, comps, recDescriptor);
- contributeOpDesc(builder, (AbstractLogicalOperator) op, sortOpDesc);
- ILogicalOperator src = op.getInputs().get(0).getValue();
- builder.contributeGraphEdge(src, 0, op, 0);
} else {
// Since topK value is provided, topK optimization is possible.
// We call topKSorter instead of calling ExternalSortOperator.
- TopKSorterOperatorDescriptor sortOpDesc = new TopKSorterOperatorDescriptor(spec, maxNumberOfFrames, topK,
- sortFields, nkcf, comps, recDescriptor);
- contributeOpDesc(builder, (AbstractLogicalOperator) op, sortOpDesc);
- ILogicalOperator src = op.getInputs().get(0).getValue();
- builder.contributeGraphEdge(src, 0, op, 0);
+ sortOpDesc = new TopKSorterOperatorDescriptor(spec, maxNumberOfFrames, topK, sortFields, nkcf, comps,
+ recDescriptor);
}
+ sortOpDesc.setSourceLocation(op.getSourceLocation());
+ contributeOpDesc(builder, (AbstractLogicalOperator) op, sortOpDesc);
+ ILogicalOperator src = op.getInputs().get(0).getValue();
+ builder.contributeGraphEdge(src, 0, op, 0);
}
@Override
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/ee54cc02/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StreamLimitPOperator.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StreamLimitPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StreamLimitPOperator.java
index da75da8..90732ce 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StreamLimitPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StreamLimitPOperator.java
@@ -98,6 +98,7 @@ public class StreamLimitPOperator extends AbstractPhysicalOperator {
JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), propagatedSchema, context);
StreamLimitRuntimeFactory runtime = new StreamLimitRuntimeFactory(maxObjectsFact, offsetFact, null,
context.getBinaryIntegerInspectorFactory());
+ runtime.setSourceLocation(limit.getSourceLocation());
builder.contributeMicroOperator(limit, runtime, recDesc);
// and contribute one edge from its child
ILogicalOperator src = limit.getInputs().get(0).getValue();