You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@vxquery.apache.org by pr...@apache.org on 2013/12/11 00:22:07 UTC
svn commit: r1550001 - in
/incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery:
compiler/rewriter/rules/ metadata/
Author: prestonc
Date: Tue Dec 10 23:22:07 2013
New Revision: 1550001
URL: http://svn.apache.org/r1550001
Log:
Merge unnest child statements into the data scan operator. Update the data source to include the path steps.
Added:
incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/compiler/rewriter/rules/ConsolidateDataScanUnnestRule.java (with props)
Modified:
incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionDataSource.java
incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java
incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryMetadataProvider.java
Added: incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/compiler/rewriter/rules/ConsolidateDataScanUnnestRule.java
URL: http://svn.apache.org/viewvc/incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/compiler/rewriter/rules/ConsolidateDataScanUnnestRule.java?rev=1550001&view=auto
==============================================================================
--- incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/compiler/rewriter/rules/ConsolidateDataScanUnnestRule.java (added)
+++ incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/compiler/rewriter/rules/ConsolidateDataScanUnnestRule.java Tue Dec 10 23:22:07 2013
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.vxquery.compiler.rewriter.rules;
+
+import org.apache.commons.lang3.mutable.Mutable;
+import org.apache.commons.lang3.mutable.MutableObject;
+import org.apache.vxquery.compiler.algebricks.VXQueryConstantValue;
+import org.apache.vxquery.compiler.rewriter.rules.util.ExpressionToolbox;
+import org.apache.vxquery.context.RootStaticContextImpl;
+import org.apache.vxquery.context.StaticContextImpl;
+import org.apache.vxquery.datamodel.accessors.TaggedValuePointable;
+import org.apache.vxquery.datamodel.values.XDMConstants;
+import org.apache.vxquery.functions.BuiltinOperators;
+import org.apache.vxquery.functions.Function;
+import org.apache.vxquery.metadata.VXQueryCollectionDataSource;
+import org.apache.vxquery.types.BuiltinTypeRegistry;
+import org.apache.vxquery.types.Quantifier;
+import org.apache.vxquery.types.SequenceType;
+
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalExpressionTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
+import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.AbstractFunctionCallExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ConstantExpression;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.DataSourceScanOperator;
+import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.UnnestOperator;
+import edu.uci.ics.hyracks.data.std.api.IPointable;
+import edu.uci.ics.hyracks.data.std.primitive.BooleanPointable;
+import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
+
+/**
+ * The rule searches for an unnest operator immediately following a data scan
+ * operator.
+ *
+ * <pre>
+ * Before
+ *
+ * plan__parent
+ * UNNEST( $v2 : child( $v1 ) )
+ * DATASCAN( $source : $v1 )
+ * plan__child
+ *
+ * Where $v1 is not used in plan__parent.
+ *
+ * After
+ *
+ * plan__parent
+ * DATASCAN( $source : $v1 )
+ * plan__child
+ *
+ * $source is encoded with the child parameters.
+ * </pre>
+ *
+ * @author prestonc
+ */
+public class ConsolidateDataScanUnnestRule extends AbstractUsedVariablesProcessingRule {
+ final StaticContextImpl dCtx = new StaticContextImpl(RootStaticContextImpl.INSTANCE);
+ final int ARG_DATA = 0;
+ final int ARG_TYPE = 1;
+
+ protected boolean processOperator(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
+ throws AlgebricksException {
+ AbstractLogicalOperator op0 = (AbstractLogicalOperator) opRef.getValue();
+ if (op0.getInputs().isEmpty()) {
+ return false;
+ }
+
+ AbstractLogicalOperator op1 = (AbstractLogicalOperator) op0.getInputs().get(0).getValue();
+ if (op1.getOperatorTag() != LogicalOperatorTag.UNNEST) {
+ return false;
+ }
+ UnnestOperator unnest = (UnnestOperator) op1;
+
+ AbstractLogicalOperator op2 = (AbstractLogicalOperator) unnest.getInputs().get(0).getValue();
+ if (op2.getOperatorTag() != LogicalOperatorTag.DATASOURCESCAN) {
+ return false;
+ }
+ DataSourceScanOperator datascan = (DataSourceScanOperator) op2;
+
+ if (!usedVariables.contains(datascan.getVariables())) {
+ // Check to see if the unnest expression is a child function.
+ ILogicalExpression logicalExpression = (ILogicalExpression) unnest.getExpressionRef().getValue();
+ if (logicalExpression.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
+ return false;
+ }
+ AbstractFunctionCallExpression functionCall = (AbstractFunctionCallExpression) logicalExpression;
+ Function functionInfo = (Function) functionCall.getFunctionInfo();
+ if (!functionInfo.getFunctionIdentifier().equals(BuiltinOperators.CHILD.getFunctionIdentifier())) {
+ return false;
+ }
+
+ // Find all child functions.
+ VXQueryCollectionDataSource ds = (VXQueryCollectionDataSource) datascan.getDataSource();
+ updateDataSource(ds, unnest.getExpressionRef());
+
+ // Replace unnest with noop assign. Keeps variable chain.
+ Mutable<ILogicalExpression> varExp = ExpressionToolbox.findVariableExpression(unnest.getExpressionRef(),
+ datascan.getVariables().get(0));
+ AssignOperator noOp = new AssignOperator(unnest.getVariable(), varExp);
+ noOp.getInputs().addAll(unnest.getInputs());
+ op0.getInputs().clear();
+ op0.getInputs().add(new MutableObject<ILogicalOperator>(noOp));
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * In reverse add them to the data source.
+ *
+ * @param ds
+ * @param expression
+ */
+ private void updateDataSource(VXQueryCollectionDataSource ds, Mutable<ILogicalExpression> expression) {
+ ILogicalExpression logicalExpression = (ILogicalExpression) expression.getValue();
+ if (logicalExpression.getExpressionTag() != LogicalExpressionTag.FUNCTION_CALL) {
+ return;
+ }
+ AbstractFunctionCallExpression functionCall = (AbstractFunctionCallExpression) logicalExpression;
+ Function functionInfo = (Function) functionCall.getFunctionInfo();
+ if (!functionInfo.getFunctionIdentifier().equals(BuiltinOperators.CHILD.getFunctionIdentifier())) {
+ return;
+ }
+ // Traverse down child function nesting.
+ updateDataSource(ds, functionCall.getArguments().get(ARG_DATA));
+
+ // Add the child type parameter to data source.
+ ILogicalExpression argType = functionCall.getArguments().get(ARG_TYPE).getValue();
+ if (argType.getExpressionTag() != LogicalExpressionTag.CONSTANT) {
+ return;
+ }
+ TaggedValuePointable tvp = new TaggedValuePointable();
+ ExpressionToolbox.getConstantAsPointable((ConstantExpression) argType, tvp);
+
+ IntegerPointable pTypeCode = (IntegerPointable) IntegerPointable.FACTORY.createPointable();
+ tvp.getValue(pTypeCode);
+ ds.addChildSeq(pTypeCode.getInteger());
+ }
+}
Propchange: incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/compiler/rewriter/rules/ConsolidateDataScanUnnestRule.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionDataSource.java
URL: http://svn.apache.org/viewvc/incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionDataSource.java?rev=1550001&r1=1550000&r2=1550001&view=diff
==============================================================================
--- incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionDataSource.java (original)
+++ incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionDataSource.java Tue Dec 10 23:22:07 2013
@@ -33,6 +33,7 @@ import edu.uci.ics.hyracks.algebricks.co
public class VXQueryCollectionDataSource implements IDataSource<String> {
private final int dataSourceId;
private final String collectionName;
+ private final List<Integer> childSeq;
private int totalDataSources;
private final Object[] types;
@@ -51,6 +52,7 @@ public class VXQueryCollectionDataSource
return vec;
}
};
+ this.childSeq = new ArrayList<Integer>();
}
public int getTotalDataSources() {
@@ -84,4 +86,12 @@ public class VXQueryCollectionDataSource
public void computeFDs(List<LogicalVariable> scanVariables, List<FunctionalDependency> fdList) {
}
+ public void addChildSeq(int integer) {
+ childSeq.add(integer);
+ }
+
+ public List<Integer> getChildSeq() {
+ return childSeq;
+ }
+
}
Modified: incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java
URL: http://svn.apache.org/viewvc/incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java?rev=1550001&r1=1550000&r2=1550001&view=diff
==============================================================================
--- incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java (original)
+++ incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java Tue Dec 10 23:22:07 2013
@@ -18,12 +18,17 @@ package org.apache.vxquery.metadata;
import java.io.File;
import java.nio.ByteBuffer;
+import java.util.List;
+import org.apache.vxquery.datamodel.accessors.TaggedValuePointable;
+import org.apache.vxquery.exceptions.SystemException;
+import org.apache.vxquery.runtime.functions.step.ChildPathStep;
import org.apache.vxquery.runtime.functions.util.FunctionHelper;
import org.apache.vxquery.xmlparser.ITreeNodeIdProvider;
import org.apache.vxquery.xmlparser.TreeNodeIdProvider;
import org.xml.sax.InputSource;
+import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
@@ -43,13 +48,15 @@ public class VXQueryCollectionOperatorDe
private short dataSourceId;
private short totalDataSources;
private String collectionName;
+ private List<Integer> childSeq;
- public VXQueryCollectionOperatorDescriptor(IOperatorDescriptorRegistry spec, String collectionName,
- int dataSourceId, int totalDataSources, RecordDescriptor rDesc) {
+ public VXQueryCollectionOperatorDescriptor(IOperatorDescriptorRegistry spec, VXQueryCollectionDataSource ds,
+ RecordDescriptor rDesc) {
super(spec, 1, 1);
- this.collectionName = collectionName;
- this.dataSourceId = (short) dataSourceId;
- this.totalDataSources = (short) totalDataSources;
+ collectionName = ds.getId();
+ dataSourceId = (short) ds.getDataSourceId();
+ totalDataSources = (short) ds.getTotalDataSources();
+ childSeq = ds.getChildSeq();
recordDescriptors[0] = rDesc;
}
@@ -68,6 +75,7 @@ public class VXQueryCollectionOperatorDe
final ITreeNodeIdProvider nodeIdProvider = new TreeNodeIdProvider(partitionId, dataSourceId, totalDataSources);
final String nodeId = ctx.getJobletContext().getApplicationContext().getNodeId();
final int frameSize = ctx.getFrameSize();
+ final IHyracksTaskContext ctx1 = ctx;
return new AbstractUnaryInputUnaryOutputOperatorNodePushable() {
@Override
@@ -98,31 +106,59 @@ public class VXQueryCollectionOperatorDe
for (File file : collectionDirectory.listFiles()) {
// Add the document node to the frame output.
if (FunctionHelper.readableXmlFile(file.getPath())) {
- // First copy all new fields over.
- tb.reset();
- if (fta.getFieldCount() > 0) {
- for (int f = 0; f < fta.getFieldCount(); ++f) {
- tb.addField(fta, t, f);
- }
- }
+ // TODO Make field addition based on output tuples instead of files.
// Now add new field.
abvsFileNode.reset();
try {
FunctionHelper.readInDocFromString(file.getPath(), in, abvsFileNode, nodeIdProvider);
+ // Can not fit XML file into frame.
+ if (frameSize <= (abvsFileNode.getLength() - abvsFileNode.getStartOffset())) {
+ throw new HyracksDataException(
+ "XML node is to large for the current frame size (VXQueryCollectionOperatorDescriptor.addXmlFile).");
+ }
} catch (Exception e) {
throw new HyracksDataException(e);
}
- tb.addField(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(),
- abvsFileNode.getLength());
-
- // Can not fit XML file into frame.
- if (frameSize <= (abvsFileNode.getLength() - abvsFileNode.getStartOffset())) {
- throw new HyracksDataException(
- "XML file is to large for the current frame size (VXQueryCollectionOperatorDescriptor.addXmlFile).");
+ if (childSeq.isEmpty()) {
+ // Return the whole XML file.
+ tb.addField(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(),
+ abvsFileNode.getLength());
+ } else {
+ // Process child nodes.
+ TaggedValuePointable tvp = (TaggedValuePointable) TaggedValuePointable.FACTORY
+ .createPointable();
+ tvp.set(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(),
+ abvsFileNode.getLength());
+ processChildStep(tb, tvp, t);
}
+ } else if (file.isDirectory()) {
+ // Consider all XML file in sub directories.
+ addXmlFile(file, t);
+ }
+ }
+ }
+ private void processChildStep(final ArrayTupleBuilder tb, TaggedValuePointable tvp, int t)
+ throws HyracksDataException {
+ TaggedValuePointable result = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
+ ChildPathStep childPathStep = new ChildPathStep(ctx1);
+ try {
+ childPathStep.init(tvp, childSeq);
+ } catch (SystemException e) {
+ throw new HyracksDataException("Child path step failed to load node tree.");
+ }
+ try {
+ while (childPathStep.step(result)) {
+ // First copy all new fields over.
+ tb.reset();
+ if (fta.getFieldCount() > 0) {
+ for (int f = 0; f < fta.getFieldCount(); ++f) {
+ tb.addField(fta, t, f);
+ }
+ }
+ tb.addField(result.getByteArray(), result.getStartOffset(), result.getLength());
// Send to the writer.
if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
FrameUtils.flushFrame(frame, writer);
@@ -132,10 +168,9 @@ public class VXQueryCollectionOperatorDe
"Could not write frame (VXQueryCollectionOperatorDescriptor.createPushRuntime).");
}
}
- } else if (file.isDirectory()) {
- // Consider all XML file in sub directories.
- addXmlFile(file, t);
}
+ } catch (AlgebricksException e) {
+ throw new HyracksDataException(e);
}
}
Modified: incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryMetadataProvider.java
URL: http://svn.apache.org/viewvc/incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryMetadataProvider.java?rev=1550001&r1=1550000&r2=1550001&view=diff
==============================================================================
--- incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryMetadataProvider.java (original)
+++ incubator/vxquery/trunk/vxquery/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryMetadataProvider.java Tue Dec 10 23:22:07 2013
@@ -67,10 +67,9 @@ public class VXQueryMetadataProvider imp
List<LogicalVariable> scanVariables, List<LogicalVariable> projectVariables, boolean projectPushed,
IOperatorSchema opSchema, IVariableTypeEnvironment typeEnv, JobGenContext context,
JobSpecification jobSpec, Object implConfig) throws AlgebricksException {
- VXQueryCollectionDataSource ds = (VXQueryCollectionDataSource) dataSource;
RecordDescriptor rDesc = new RecordDescriptor(new ISerializerDeserializer[opSchema.getSize()]);
- IOperatorDescriptor scanner = new VXQueryCollectionOperatorDescriptor(jobSpec, ds.getId(),
- ds.getDataSourceId(), ds.getTotalDataSources(), rDesc);
+ IOperatorDescriptor scanner = new VXQueryCollectionOperatorDescriptor(jobSpec,
+ (VXQueryCollectionDataSource) dataSource, rDesc);
AlgebricksAbsolutePartitionConstraint constraint = new AlgebricksAbsolutePartitionConstraint(nodeList);
return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(scanner, constraint);