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 2014/01/31 08:21:27 UTC
[11/13] git commit: Removed recursion and set up the file list to use
an iterator.
Removed recursion and set up the file list to use an iterator.
Project: http://git-wip-us.apache.org/repos/asf/incubator-vxquery/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-vxquery/commit/74542f99
Tree: http://git-wip-us.apache.org/repos/asf/incubator-vxquery/tree/74542f99
Diff: http://git-wip-us.apache.org/repos/asf/incubator-vxquery/diff/74542f99
Branch: refs/heads/prestonc/benchmarks
Commit: 74542f992eff8fc6a4f3f671bb568ada61577373
Parents: e19b609
Author: Preston Carman <pr...@apache.org>
Authored: Thu Jan 30 17:55:16 2014 -0800
Committer: Preston Carman <pr...@apache.org>
Committed: Thu Jan 30 17:55:16 2014 -0800
----------------------------------------------------------------------
.../VXQueryCollectionOperatorDescriptor.java | 112 ++++++++++---------
.../vxquery/metadata/VXQueryIOFileFilter.java | 37 ++++++
.../FnCollectionScalarEvaluatorFactory.java | 31 ++---
.../runtime/functions/util/FunctionHelper.java | 27 ++---
4 files changed, 121 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-vxquery/blob/74542f99/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java
----------------------------------------------------------------------
diff --git a/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java b/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java
index 976b2a7..33d6498 100644
--- a/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java
+++ b/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryCollectionOperatorDescriptor.java
@@ -18,8 +18,11 @@ package org.apache.vxquery.metadata;
import java.io.File;
import java.nio.ByteBuffer;
+import java.util.Iterator;
import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.vxquery.datamodel.accessors.PointablePool;
import org.apache.vxquery.datamodel.accessors.PointablePoolFactory;
import org.apache.vxquery.datamodel.accessors.TaggedValuePointable;
@@ -47,6 +50,7 @@ import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperat
public class VXQueryCollectionOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
private static final long serialVersionUID = 1L;
+ private static final String[] EXTENTIONS = { "xml" };
private short dataSourceId;
private short totalDataSources;
private String[] collectionPartitions;
@@ -79,7 +83,7 @@ public class VXQueryCollectionOperatorDescriptor extends AbstractSingleActivityO
final int frameSize = ctx.getFrameSize();
final PointablePool ppool = PointablePoolFactory.INSTANCE.createPointablePool();
final ChildPathStep childPathStep = new ChildPathStep(ctx, ppool);
-
+
final String collectionName = collectionPartitions[partition % collectionPartitions.length];
return new AbstractUnaryInputUnaryOutputOperatorNodePushable() {
@@ -98,7 +102,12 @@ public class VXQueryCollectionOperatorDescriptor extends AbstractSingleActivityO
// Go through each tuple.
if (collectionDirectory.isDirectory()) {
for (int t = 0; t < fta.getTupleCount(); ++t) {
- addNextXmlNode(collectionDirectory, t);
+ @SuppressWarnings("unchecked")
+ Iterator<File> it = FileUtils.iterateFiles(collectionDirectory, new VXQueryIOFileFilter(),
+ TrueFileFilter.INSTANCE);
+ while (it.hasNext()) {
+ addNextXmlNode(it.next(), t);
+ }
}
} else {
throw new HyracksDataException(
@@ -106,42 +115,34 @@ public class VXQueryCollectionOperatorDescriptor extends AbstractSingleActivityO
}
}
- private void addNextXmlNode(File collectionDirectory, int t) throws HyracksDataException {
- // Add a field for the document node to each tuple.
- for (File file : collectionDirectory.listFiles()) {
- // Add the document node to the frame output.
- if (FunctionHelper.readableXmlFile(file.getPath())) {
- // 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);
- } catch (Exception e) {
- throw new HyracksDataException(e);
- }
+ /**
+ * Add the document node to the frame output.
+ */
+ private void addNextXmlNode(File file, int t) throws HyracksDataException {
+ // TODO Make field addition based on output tuples instead of files.
- if (childSeq.isEmpty()) {
- // 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).");
- }
- // 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.
- addNextXmlNode(file, t);
+ // Now add new field.
+ abvsFileNode.reset();
+ try {
+ FunctionHelper.readInDocFromString(file, in, abvsFileNode, nodeIdProvider);
+ } catch (Exception e) {
+ throw new HyracksDataException(e);
+ }
+
+ if (childSeq.isEmpty()) {
+ // 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).");
}
+ TaggedValuePointable result = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
+ result.set(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(), abvsFileNode.getLength());
+ addNodeToTuple(tb, result, t);
+ } else {
+ // Process child nodes.
+ TaggedValuePointable tvp = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
+ tvp.set(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(), abvsFileNode.getLength());
+ processChildStep(tb, tvp, t);
}
}
@@ -155,29 +156,34 @@ public class VXQueryCollectionOperatorDescriptor extends AbstractSingleActivityO
}
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);
- appender.reset(frame, true);
- if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
- throw new HyracksDataException(
- "Could not write frame (VXQueryCollectionOperatorDescriptor.createPushRuntime).");
- }
- }
+ addNodeToTuple(tb, result, t);
}
} catch (AlgebricksException e) {
throw new HyracksDataException(e);
}
}
+ private void addNodeToTuple(final ArrayTupleBuilder tb, TaggedValuePointable result, int t)
+ throws HyracksDataException {
+ // 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);
+ appender.reset(frame, true);
+ if (!appender.append(tb.getFieldEndOffsets(), tb.getByteArray(), 0, tb.getSize())) {
+ throw new HyracksDataException(
+ "Could not write frame (VXQueryCollectionOperatorDescriptor.createPushRuntime).");
+ }
+ }
+ }
+
@Override
public void fail() throws HyracksDataException {
writer.fail();
http://git-wip-us.apache.org/repos/asf/incubator-vxquery/blob/74542f99/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryIOFileFilter.java
----------------------------------------------------------------------
diff --git a/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryIOFileFilter.java b/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryIOFileFilter.java
new file mode 100644
index 0000000..104f254
--- /dev/null
+++ b/vxquery-core/src/main/java/org/apache/vxquery/metadata/VXQueryIOFileFilter.java
@@ -0,0 +1,37 @@
+/*
+ * 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.metadata;
+
+import java.io.File;
+
+import org.apache.commons.io.filefilter.IOFileFilter;
+
+public class VXQueryIOFileFilter implements IOFileFilter {
+
+ @Override
+ public boolean accept(final File file) {
+ return accept(file, file.getName());
+ }
+
+ @Override
+ public boolean accept(final File file, final String name) {
+ if (name.toLowerCase().endsWith(".xml") || name.toLowerCase().endsWith(".xml.gz")) {
+ return true;
+ }
+ return false;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-vxquery/blob/74542f99/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/node/FnCollectionScalarEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/node/FnCollectionScalarEvaluatorFactory.java b/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/node/FnCollectionScalarEvaluatorFactory.java
index d843d78..bde1874 100644
--- a/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/node/FnCollectionScalarEvaluatorFactory.java
+++ b/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/node/FnCollectionScalarEvaluatorFactory.java
@@ -21,12 +21,16 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
+import java.util.Iterator;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.filefilter.TrueFileFilter;
import org.apache.vxquery.datamodel.accessors.TaggedValuePointable;
import org.apache.vxquery.datamodel.builders.sequence.SequenceBuilder;
import org.apache.vxquery.datamodel.values.ValueTag;
import org.apache.vxquery.exceptions.ErrorCode;
import org.apache.vxquery.exceptions.SystemException;
+import org.apache.vxquery.metadata.VXQueryIOFileFilter;
import org.apache.vxquery.runtime.functions.base.AbstractTaggedValueArgumentScalarEvaluator;
import org.apache.vxquery.runtime.functions.base.AbstractTaggedValueArgumentScalarEvaluatorFactory;
import org.apache.vxquery.runtime.functions.util.FunctionHelper;
@@ -79,6 +83,8 @@ public class FnCollectionScalarEvaluatorFactory extends AbstractTaggedValueArgum
bbis.setByteBuffer(ByteBuffer.wrap(Arrays.copyOfRange(stringp.getByteArray(),
stringp.getStartOffset(), stringp.getLength() + stringp.getStartOffset())), 0);
collectionName = di.readUTF();
+ di.close();
+ bbis.close();
} catch (IOException e) {
throw new SystemException(ErrorCode.SYSE0001, e);
}
@@ -90,7 +96,12 @@ public class FnCollectionScalarEvaluatorFactory extends AbstractTaggedValueArgum
try {
abvs.reset();
sb.reset(abvs);
- addXmlFiles(collectionDirectory);
+ @SuppressWarnings("unchecked")
+ Iterator<File> it = FileUtils.iterateFiles(collectionDirectory, new VXQueryIOFileFilter(),
+ TrueFileFilter.INSTANCE);
+ while (it.hasNext()) {
+ addNextXmlNode(it.next());
+ }
sb.finish();
result.set(abvs);
} catch (IOException e) {
@@ -98,19 +109,11 @@ public class FnCollectionScalarEvaluatorFactory extends AbstractTaggedValueArgum
}
}
- private void addXmlFiles(File collectionDirectory) throws SystemException, IOException {
- for (File file : collectionDirectory.listFiles()) {
- // Add the document node to the sequence.
- if (FunctionHelper.readableXmlFile(file.getPath())) {
- abvsFileNode.reset();
- FunctionHelper.readInDocFromString(file.getPath(), in, abvsFileNode, nodeIdProvider);
- nodep.set(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(), abvsFileNode.getLength());
- sb.addItem(nodep);
- } else if (file.isDirectory()) {
- // Consider all XML file in sub directories.
- addXmlFiles(file);
- }
- }
+ private void addNextXmlNode(File file) throws SystemException, IOException {
+ abvsFileNode.reset();
+ FunctionHelper.readInDocFromString(file, in, abvsFileNode, nodeIdProvider);
+ nodep.set(abvsFileNode.getByteArray(), abvsFileNode.getStartOffset(), abvsFileNode.getLength());
+ sb.addItem(nodep);
}
};
}
http://git-wip-us.apache.org/repos/asf/incubator-vxquery/blob/74542f99/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/util/FunctionHelper.java
----------------------------------------------------------------------
diff --git a/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/util/FunctionHelper.java b/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/util/FunctionHelper.java
index 035a6ad..c827e60 100644
--- a/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/util/FunctionHelper.java
+++ b/vxquery-core/src/main/java/org/apache/vxquery/runtime/functions/util/FunctionHelper.java
@@ -18,6 +18,7 @@ package org.apache.vxquery.runtime.functions.util;
import java.io.DataInputStream;
import java.io.DataOutput;
+import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
@@ -1664,37 +1665,25 @@ public class FunctionHelper {
System.err.println(" printUTF8String END");
}
- /**
- * Checks the path has an acceptable file name extension to read in.
- *
- * @param path
- * @return
- */
- public static boolean readableXmlFile(String path) {
- if (path.toLowerCase().endsWith(".xml") || path.toLowerCase().endsWith(".xml.gz")) {
- return true;
- }
- return false;
- }
-
public static void readInDocFromPointable(UTF8StringPointable stringp, InputSource in, ByteBufferInputStream bbis,
DataInputStream di, ArrayBackedValueStorage abvs, ITreeNodeIdProvider treeNodeIdProvider)
throws SystemException {
String fName = getStringFromPointable(stringp, bbis, di);
- readInDocFromString(fName, in, abvs, treeNodeIdProvider);
+ File file = new File(fName);
+ readInDocFromString(file, in, abvs, treeNodeIdProvider);
}
- public static void readInDocFromString(String fName, InputSource in, ArrayBackedValueStorage abvs,
+ public static void readInDocFromString(File file, InputSource in, ArrayBackedValueStorage abvs,
ITreeNodeIdProvider treeNodeIdProvider) throws SystemException {
try {
- if (fName.toLowerCase().endsWith(".xml.gz")) {
- in.setCharacterStream(new InputStreamReader(new GZIPInputStream(new FileInputStream(fName))));
+ if (file.getName().toLowerCase().endsWith(".xml.gz")) {
+ in.setCharacterStream(new InputStreamReader(new GZIPInputStream(new FileInputStream(file))));
} else {
- in.setCharacterStream(new InputStreamReader(new FileInputStream(fName)));
+ in.setCharacterStream(new InputStreamReader(new FileInputStream(file)));
}
XMLParser.parseInputSource(in, abvs, false, treeNodeIdProvider);
} catch (FileNotFoundException e) {
- throw new RuntimeException("The file (" + fName + ") does not exist.");
+ throw new RuntimeException("The file (" + file.getName() + ") does not exist.");
} catch (IOException e) {
throw new SystemException(ErrorCode.SYSE0001, e);
}