You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ga...@apache.org on 2008/05/13 23:11:24 UTC
svn commit: r656011 [2/5] - in /incubator/pig/branches/types: ./ lib/
src/org/apache/pig/ src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/data/ src/org/apache/pig/impl/io/
src/org/apache/pig/impl/logicalLayer/ src/org/apache/pig/i...
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigInputFormat.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigInputFormat.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigInputFormat.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigInputFormat.java Tue May 13 14:11:21 2008
@@ -0,0 +1,358 @@
+/*
+ * 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.pig.impl.mapReduceLayer;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobConfigurable;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.data.TargetedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.BufferedPositionedInputStream;
+import org.apache.pig.impl.io.FileSpec;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+import org.apache.pig.impl.util.ObjectSerializer;
+import org.apache.tools.bzip2r.CBZip2InputStream;
+
+public class PigInputFormat implements InputFormat<Text, TargetedTuple>,
+ JobConfigurable {
+
+ public static final Log LOG = LogFactory
+ .getLog(PigInputFormat.class);
+
+ private static final PathFilter hiddenFileFilter = new PathFilter() {
+ public boolean accept(Path p) {
+ String name = p.getName();
+ return !name.startsWith("_") && !name.startsWith(".");
+ }
+ };
+
+ /**
+ * Is the given filename splitable? Usually, true, but if the file is stream
+ * compressed, it will not be.
+ *
+ * <code>FileInputFormat</code> implementations can override this and
+ * return <code>false</code> to ensure that individual input files are
+ * never split-up so that {@link Mapper}s process entire files.
+ *
+ * @param fs
+ * the file system that the file is on
+ * @param filename
+ * the file name to check
+ * @return is this file splitable?
+ */
+ protected boolean isSplitable(FileSystem fs, Path filename) {
+ return !filename.getName().endsWith(".gz");
+ }
+
+ /**
+ * List input directories. Subclasses may override to, e.g., select only
+ * files matching a regular expression.
+ *
+ * @param job
+ * the job to list input paths for
+ * @return array of Path objects
+ * @throws IOException
+ * if zero items.
+ */
+ protected Path[] listPaths(JobConf job) throws IOException {
+ Path[] dirs = job.getInputPaths();
+ if (dirs.length == 0) {
+ throw new IOException("No input paths specified in job");
+ }
+ List<Path> result = new ArrayList<Path>();
+ for (Path p : dirs) {
+ FileSystem fs = p.getFileSystem(job);
+ Path[] matches = fs.listPaths(fs.globPaths(p, hiddenFileFilter),
+ hiddenFileFilter);
+ for (Path match : matches) {
+ result.add(fs.makeQualified(match));
+ }
+ }
+
+ return result.toArray(new Path[result.size()]);
+ }
+
+ public void validateInput(JobConf job) throws IOException {
+ /*ArrayList<FileSpec> inputs = (ArrayList<FileSpec>) ObjectSerializer
+ .deserialize(job.get("pig.inputs"));
+ Path[] inputDirs = new Path[inputs.size()];
+ int i = 0;
+ for (FileSpec spec : inputs) {
+ inputDirs[i++] = new Path(spec.getFileName());
+ }
+
+ if (inputDirs.length == 0) {
+ throw new IOException("No input paths specified in input");
+ }
+
+ List<IOException> result = new ArrayList<IOException>();
+ int totalFiles = 0;
+ for (Path p : inputDirs) {
+ FileSystem fs = p.getFileSystem(job);
+ if (fs.exists(p)) {
+ // make sure all paths are files to avoid exception
+ // while generating splits
+ for (Path subPath : fs.listPaths(p, hiddenFileFilter)) {
+ FileSystem subFS = subPath.getFileSystem(job);
+ if (!subFS.exists(subPath)) {
+ result.add(new IOException(
+ "Input path does not exist: " + subPath));
+ } else {
+ totalFiles++;
+ }
+ }
+ } else {
+ Path[] paths = fs.globPaths(p, hiddenFileFilter);
+ if (paths.length == 0) {
+ result.add(new IOException("Input Pattern " + p
+ + " matches 0 files"));
+ } else {
+ // validate globbed paths
+ for (Path gPath : paths) {
+ FileSystem gPathFS = gPath.getFileSystem(job);
+ if (!gPathFS.exists(gPath)) {
+ result.add(new FileNotFoundException(
+ "Input path doesnt exist : " + gPath));
+ }
+ }
+ totalFiles += paths.length;
+ }
+ }
+ }
+ if (!result.isEmpty()) {
+ throw new InvalidInputException(result);
+ }
+ // send output to client.
+ LOG.info("Total input paths to process : " + totalFiles);*/
+ }
+
+ /**
+ * Creates input splits one per input and slices of it
+ * per DFS block of the input file. Configures the PigSplit
+ * and returns the list of PigSplits as an array
+ */
+ public InputSplit[] getSplits(JobConf job, int numSplits)
+ throws IOException {
+ ArrayList<FileSpec> inputs = (ArrayList<FileSpec>) ObjectSerializer
+ .deserialize(job.get("pig.inputs"));
+ ArrayList<List<OperatorKey>> inpTargets = (ArrayList<List<OperatorKey>>) ObjectSerializer
+ .deserialize(job.get("pig.inpTargets"));
+ PigContext pigContext = (PigContext) ObjectSerializer.deserialize(job
+ .get("pig.pigContext"));
+
+ ArrayList<InputSplit> splits = new ArrayList<InputSplit>();
+ for (int i = 0; i < inputs.size(); i++) {
+ Path path = new Path(inputs.get(i).getFileName());
+ String parser = inputs.get(i).getFuncSpec();
+ FileSystem fs = path.getFileSystem(job);
+
+ fs.setWorkingDirectory(new Path("/user", job.getUser()));
+ ArrayList<Path> paths = new ArrayList<Path>();
+ // If you give a non-glob name, globPaths returns a single
+ // element with just that name.
+ Path[] globPaths = fs.globPaths(path);
+ for (int m = 0; m < globPaths.length; m++)
+ paths.add(globPaths[m]);
+ // paths.add(path);
+ for (int j = 0; j < paths.size(); j++) {
+ Path fullPath = new Path(fs.getWorkingDirectory(), paths.get(j));
+ if (fs.getFileStatus(fullPath).isDir()) {
+ FileStatus children[] = fs.listStatus(fullPath);
+ for (int k = 0; k < children.length; k++) {
+ paths.add(children[k].getPath());
+ }
+ continue;
+ }
+ long bs = fs.getFileStatus(fullPath).getBlockSize();
+ long size = fs.getFileStatus(fullPath).getLen();
+ long pos = 0;
+ String name = paths.get(j).getName();
+ if (name.endsWith(".gz")) {
+ // Anything that ends with a ".gz" we must process as a
+ // complete file
+ splits.add(new PigSplit(pigContext, fs, fullPath, parser,
+ inpTargets.get(i), 0, size));
+ } else {
+ while (pos < size) {
+ if (pos + bs > size)
+ bs = size - pos;
+ splits.add(new PigSplit(pigContext, fs, fullPath,
+ parser, inpTargets.get(i), pos, bs));
+ pos += bs;
+ }
+ }
+ }
+ }
+ return splits.toArray(new PigSplit[splits.size()]);
+ }
+
+ public RecordReader<Text, TargetedTuple> getRecordReader(InputSplit split,
+ JobConf job, Reporter reporter) throws IOException {
+ PigRecordReader r = new PigRecordReader(job, (PigSplit) split,
+ compressionCodecs);
+ return r;
+ }
+
+ private CompressionCodecFactory compressionCodecs = null;
+
+ static public String codecList;
+
+ public void configure(JobConf conf) {
+ compressionCodecs = new CompressionCodecFactory(conf);
+ codecList = conf.get("io.compression.codecs", "none");
+ }
+
+ /**
+ * The class that actually reads the data off
+ * the filesystem configured in the split and
+ * parses tuples out of it
+ */
+ public static class PigRecordReader implements
+ RecordReader<Text, TargetedTuple> {
+
+ private InputStream is;
+
+ private FSDataInputStream fsis;
+
+ private BufferedPositionedInputStream bis;
+
+ private long end;
+
+ private PigSplit split;
+
+ LoadFunc loader;
+
+ CompressionCodecFactory compressionFactory;
+
+ JobConf job;
+
+ /**
+ * Uses the split to get information about the input
+ * and creates the inputstream to which the load func
+ * can bind to
+ * @param job
+ * @param split
+ * @param compressionFactory
+ * @throws IOException
+ */
+ PigRecordReader(JobConf job, PigSplit split,
+ CompressionCodecFactory compressionFactory) throws IOException {
+ this.split = split;
+ this.job = job;
+ this.compressionFactory = compressionFactory;
+ loader = split.getLoadFunction();
+ Path path = split.getPath();
+ FileSystem fs = path.getFileSystem(job);
+ fs.setWorkingDirectory(new Path("/user/" + job.getUser()));
+ CompressionCodec codec = compressionFactory.getCodec(split
+ .getPath());
+ long start = split.getStart();
+ fsis = fs.open(split.getPath());
+ fsis.seek(start);
+
+ if (codec != null) {
+ is = codec.createInputStream(fsis);
+ end = Long.MAX_VALUE;
+
+ } else {
+ end = start + split.getLength();
+
+ if (split.file.getName().endsWith(".bz")
+ || split.file.getName().endsWith(".bz2")) {
+ is = new CBZip2InputStream(fsis, 9);
+ } else {
+ is = fsis;
+ }
+ }
+ bis = new BufferedPositionedInputStream(is, start);
+ loader.bindTo(split.getPath().toString(), bis, start, end);
+
+ }
+
+ public JobConf getJobConf() {
+ return job;
+ }
+
+ /**
+ * Reads the next tuple using the load func and
+ * sets its target operators. The targeted tuple
+ * will then attached to those operators when the
+ * plan is executed
+ */
+ public boolean next(Text key, TargetedTuple value) throws IOException {
+ Tuple t = loader.getNext();
+ if (t == null) {
+ return false;
+ }
+
+ key.set(split.getPath().getName());
+
+ value.reference(t);
+ value.setTargetOps(split.getTargetOps());
+ return true;
+ }
+
+ public long getPos() throws IOException {
+ return fsis.getPos();
+ }
+
+ public void close() throws IOException {
+ is.close();
+ }
+
+ public PigSplit getPigFileSplit() {
+ return split;
+ }
+
+ public Text createKey() {
+ return new Text();
+ }
+
+ public TargetedTuple createValue() {
+ return new TargetedTuple();
+ }
+
+ public float getProgress() throws IOException {
+ float progress = getPos() - split.getStart();
+ float finish = split.getLength();
+ return progress / finish;
+ }
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapBase.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapBase.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapBase.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapBase.java Tue May 13 14:11:21 2008
@@ -0,0 +1,154 @@
+package org.apache.pig.impl.mapReduceLayer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.TargetedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+import org.apache.pig.impl.physicalLayer.POStatus;
+import org.apache.pig.impl.physicalLayer.Result;
+import org.apache.pig.impl.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.PhysicalOperator;
+import org.apache.pig.impl.util.ObjectSerializer;
+
+public abstract class PigMapBase extends MapReduceBase{
+ private final Log log = LogFactory.getLog(getClass());
+
+ protected String inputFile = null;
+
+ //Map Plan
+ protected PhysicalPlan<PhysicalOperator> mp;
+
+ //The reporter that handles communicating progress
+ protected RunnableReporter runnableReporter;
+
+ //The thread used to run the runnableReporter
+ protected Thread reporterThread;
+
+ /**
+ * Will be called when all the tuples in the input
+ * are done. So reporter thread should be closed.
+ */
+ @Override
+ public void close() throws IOException {
+ super.close();
+ if(runnableReporter!=null){
+ runnableReporter.setDone(true);
+ runnableReporter = null;
+ }
+ reporterThread = null;
+ mp = null;
+ inputFile = null;
+ }
+
+ /**
+ * Configures the mapper with the map plan and the
+ * reproter thread
+ */
+ @Override
+ public void configure(JobConf job) {
+ super.configure(job);
+ inputFile = job.get("map.input.file", "UNKNOWN");
+ try {
+ mp = (PhysicalPlan<PhysicalOperator>) ObjectSerializer.deserialize(job
+ .get("pig.mapPlan"));
+
+ // To be removed
+ if(mp.isEmpty())
+ log.debug("Map Plan empty!");
+ else{
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ mp.explain(baos);
+ log.debug(baos.toString());
+ }
+ // till here
+
+ long sleepTime = job.getLong("pig.reporter.sleep.time", 10000);
+ runnableReporter = new RunnableReporter(sleepTime);
+ reporterThread = new Thread(runnableReporter);
+ reporterThread.start();
+ } catch (IOException e) {
+ log.error(e.getMessage() + "was caused by:");
+ log.error(e.getCause().getMessage());
+ }
+ }
+
+ /**
+ * The map function that attaches the inpTuple appropriately
+ * and executes the map plan if its not empty. Collects the
+ * result of execution into oc or the input directly to oc
+ * if map plan empty. The collection is left abstract for the
+ * map-only or map-reduce job to implement. Map-only collects
+ * the tuple as-is whereas map-reduce collects it after extracting
+ * the key and indexed tuple.
+ */
+ public void map(Text key, TargetedTuple inpTuple,
+ OutputCollector<WritableComparable, Writable> oc,
+ Reporter reporter) throws IOException {
+
+ runnableReporter.setReporter(reporter);
+ runnableReporter.setInputFile(inputFile);
+
+ if(mp.isEmpty()){
+ try{
+ collect(oc,inpTuple.toTuple());
+ } catch (ExecException e) {
+ IOException ioe = new IOException(e.getMessage());
+ ioe.initCause(e.getCause());
+ throw ioe;
+ }
+ return;
+ }
+
+ for (OperatorKey targetKey : inpTuple.targetOps) {
+ PhysicalOperator<PhyPlanVisitor> target = mp.getOperator(targetKey);
+ Tuple t = inpTuple.toTuple();
+ target.attachInput(t);
+ }
+
+ List<PhysicalOperator> leaves = mp.getLeaves();
+
+ PhysicalOperator leaf = leaves.get(0);
+ try {
+ while(true){
+ Result res = leaf.getNext(inpTuple);
+ if(res.returnStatus==POStatus.STATUS_OK){
+ collect(oc,(Tuple)res.result);
+ continue;
+ }
+
+ if(res.returnStatus==POStatus.STATUS_EOP)
+ return;
+
+ if(res.returnStatus==POStatus.STATUS_NULL)
+ continue;
+
+ if(res.returnStatus==POStatus.STATUS_ERR){
+ IOException ioe = new IOException("Received Error while " +
+ "processing the map plan.");
+ throw ioe;
+ }
+ }
+ } catch (ExecException e) {
+ IOException ioe = new IOException(e.getMessage());
+ ioe.initCause(e.getCause());
+ throw ioe;
+ }
+ }
+
+ abstract public void collect(OutputCollector<WritableComparable, Writable> oc, Tuple tuple) throws ExecException, IOException;
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapOnly.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapOnly.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapOnly.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapOnly.java Tue May 13 14:11:21 2008
@@ -0,0 +1,72 @@
+/*
+ * 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.pig.impl.mapReduceLayer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.TargetedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+import org.apache.pig.impl.physicalLayer.POStatus;
+import org.apache.pig.impl.physicalLayer.Result;
+import org.apache.pig.impl.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.PhysicalOperator;
+import org.apache.pig.impl.util.ObjectSerializer;
+
+/**
+ * This class is the static Mapper class used by Pig
+ * to execute Pig map only jobs. It gets a TargetedTuple
+ * as input to the map function. Using the targets in it,
+ * attaches the tuple as input to the target operators.
+ *
+ * The map plan is then executed. The result is then collected
+ * into the output collector provide by Hadoop which is configured
+ * to write to the store file.
+ *
+ * There is a reporter running in a separate thread that keeps
+ * reporting progress to let the job tracker know that we are
+ * alive. The sleep time for the reporter thread can be configured
+ * per job via the "pig.reporter.sleep.time". By default it uses
+ * a 10 sec sleep time
+ */
+
+public class PigMapOnly {
+
+ public static class Map extends PigMapBase implements
+ Mapper<Text, TargetedTuple, WritableComparable, Writable> {
+
+ @Override
+ public void collect(OutputCollector<WritableComparable, Writable> oc, Tuple tuple) throws ExecException, IOException {
+ oc.collect(null, tuple);
+ }
+ }
+}
\ No newline at end of file
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapReduce.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapReduce.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapReduce.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigMapReduce.java Tue May 13 14:11:21 2008
@@ -0,0 +1,427 @@
+/*
+ * 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.pig.impl.mapReduceLayer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.IndexedTuple;
+import org.apache.pig.data.TargetedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+import org.apache.pig.impl.physicalLayer.POStatus;
+import org.apache.pig.impl.physicalLayer.Result;
+import org.apache.pig.impl.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POPackage;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.PhysicalOperator;
+import org.apache.pig.impl.util.ObjectSerializer;
+
+/**
+ * This class is the static Mapper & Reducer classes that
+ * are used by Pig to execute Pig Map Reduce jobs. Since
+ * there is a reduce phase, the leaf is bound to be a
+ * POLocalRearrange. So the map phase has to separate the
+ * key and indexed tuple and collect it into the output
+ * collector.
+ *
+ * The shuffle and sort phase sorts these key & indexed tuples
+ * and creates key, List<IndexedTuple> and passes the key and
+ * iterator to the list. The deserialized POPackage operator
+ * is used to package the key, List<IndexedTuple> into pigKey,
+ * Bag<Tuple> where pigKey is of the appropriate pig type and
+ * then the result of the package is attached to the reduce
+ * plan which is executed if its not empty. Either the result
+ * of the reduce plan or the package res is collected into
+ * the output collector.
+ *
+ */
+public class PigMapReduce {
+
+ public static class Map extends PigMapBase implements
+ Mapper<Text, TargetedTuple, WritableComparable, Writable> {
+
+ @Override
+ public void collect(OutputCollector<WritableComparable, Writable> oc, Tuple tuple) throws ExecException, IOException {
+ Object key = tuple.get(0);
+ IndexedTuple it = (IndexedTuple)tuple.get(1);
+ WritableComparable wcKey = DataType.getWritableComparableTypes(key);
+ oc.collect(wcKey, it);
+ }
+ /*private final Log log = LogFactory.getLog(getClass());
+
+ private String inputFile = null;
+
+ //Map Plan
+ private PhysicalPlan<PhysicalOperator> mp;
+
+ //The reporter that handles communicating progress
+ RunnableReporter runnableReporter;
+
+ //The thread used to run the runnableReporter
+ Thread reporterThread;
+
+ *//**
+ * Configures the mapper with the map plan and the
+ * reproter thread
+ *//*
+ @Override
+ public void configure(JobConf jConf) {
+ super.configure(jConf);
+ inputFile = jConf.get("map.input.file", "UNKNOWN");
+ try {
+ mp = (PhysicalPlan<PhysicalOperator>) ObjectSerializer.deserialize(jConf
+ .get("pig.mapPlan"));
+
+ // To be removed
+ if(mp.isEmpty())
+ log.debug("Map Plan empty!");
+ else{
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ mp.explain(baos);
+ log.debug(baos.toString());
+ }
+ // till here
+
+ long sleepTime = jConf.getLong("pig.reporter.sleep.time", 10000);
+ runnableReporter = new RunnableReporter(sleepTime);
+ reporterThread = new Thread(runnableReporter);
+ reporterThread.start();
+ } catch (IOException e) {
+ log.error(e.getMessage() + "was caused by:");
+ log.error(e.getCause().getMessage());
+ }
+ }
+
+ *//**
+ * The map function that attaches the inpTuple appropriately
+ * and executes the map plan if its not empty. Collects the
+ * result of execution into oc or the input directly to oc
+ * if map plan empty. The collection is done after extracting
+ * the key and indexed tuple.
+ *//*
+ public void map(Text key, TargetedTuple inpTuple,
+ OutputCollector<WritableComparable, Writable> oc,
+ Reporter reporter) throws IOException {
+
+ runnableReporter.setReporter(reporter);
+ runnableReporter.setInputFile(inputFile);
+
+ if(mp.isEmpty()){
+ try{
+ collectKeyAndTuple(oc,inpTuple.toTuple());
+ } catch (ExecException e) {
+ IOException ioe = new IOException(e.getMessage());
+ ioe.initCause(e.getCause());
+ throw ioe;
+ }
+ return;
+ }
+
+ for (OperatorKey targetKey : inpTuple.targetOps) {
+ PhysicalOperator<PhyPlanVisitor> target = mp.getOperator(targetKey);
+ Tuple t = inpTuple.toTuple();
+ target.attachInput(t);
+ }
+
+ List<PhysicalOperator> leaves = mp.getLeaves();
+
+ PhysicalOperator leaf = leaves.get(0);
+ try {
+ while(true){
+ Result res = leaf.getNext(inpTuple);
+ if(res.returnStatus==POStatus.STATUS_OK){
+ collectKeyAndTuple(oc,(Tuple)res.result);
+ continue;
+ }
+
+ if(res.returnStatus==POStatus.STATUS_EOP)
+ return;
+
+ if(res.returnStatus==POStatus.STATUS_NULL)
+ continue;
+
+ if(res.returnStatus==POStatus.STATUS_ERR){
+ IOException ioe = new IOException("Received Error while " +
+ "processing the map plan.");
+ throw ioe;
+ }
+ }
+ } catch (ExecException e) {
+ IOException ioe = new IOException(e.getMessage());
+ ioe.initCause(e.getCause());
+ throw ioe;
+ }
+ }
+
+ *//**
+ * Assumes that the tuple is of the form (key,indexedTuple) and
+ * extracts the key & indexed tuple. The key is then converted
+ * to the appropriate Hadoop type and the Hadoop type and IndexedTup
+ * are collected into the output collector
+ * @param oc - Output Collector
+ * @param tuple - The tuple which is the result of a LR either directly
+ * or by loading a file which has the output of a LR
+ * @throws ExecException
+ * @throws IOException
+ *//*
+ private void collectKeyAndTuple(OutputCollector<WritableComparable, Writable> oc, Tuple tuple) throws ExecException, IOException {
+ Object key = tuple.get(0);
+ IndexedTuple it = (IndexedTuple)tuple.get(1);
+ WritableComparable wcKey = DataType.getWritableComparableTypes(key);
+ oc.collect(wcKey, it);
+ }
+
+ *//**
+ * Will be called when all the tuples in the input
+ * are done. So reporter thread should be stopped.
+ *//*
+ @Override
+ public void close() throws IOException {
+ super.close();
+ if(runnableReporter!=null)
+ runnableReporter.setDone(true);
+ }*/
+
+ }
+
+ public static class Reduce extends MapReduceBase
+ implements
+ Reducer<WritableComparable, IndexedTuple, WritableComparable, Writable> {
+ private final Log log = LogFactory.getLog(getClass());
+
+ //The reduce plan
+ private PhysicalPlan<PhysicalOperator> rp;
+
+ //The POPackage operator which is the
+ //root of every Map Reduce plan is
+ //obtained through the job conf. The portion
+ //remaining after its removal is the reduce
+ //plan
+ private POPackage pack;
+
+ //The reporter that handles communicating progress
+ RunnableReporter runnableReporter;
+
+ //The thread used to run the runnableReporter
+ Thread reporterThread;
+
+ /**
+ * Configures the Reduce plan, the POPackage operator
+ * and the reporter thread
+ */
+ @Override
+ public void configure(JobConf jConf) {
+ super.configure(jConf);
+ try {
+ rp = (PhysicalPlan<PhysicalOperator>) ObjectSerializer.deserialize(jConf
+ .get("pig.reducePlan"));
+ pack = (POPackage)ObjectSerializer.deserialize(jConf.get("pig.reduce.package"));
+ // To be removed
+ if(rp.isEmpty())
+ log.debug("Reduce Plan empty!");
+ else{
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ rp.explain(baos);
+ log.debug(baos.toString());
+ }
+ // till here
+
+ long sleepTime = jConf.getLong("pig.reporter.sleep.time", 10000);
+ runnableReporter = new RunnableReporter(sleepTime);
+ reporterThread = new Thread(runnableReporter);
+ reporterThread.start();
+ } catch (IOException e) {
+ log.error(e.getMessage() + "was caused by:");
+ log.error(e.getCause().getMessage());
+ }
+ }
+
+ /**
+ * The reduce function which packages the key and List<IndexedTuple>
+ * into key, Bag<Tuple> after converting Hadoop type key into Pig type.
+ * The package result is either collected as is, if the reduce plan is
+ * empty or after passing through the reduce plan.
+ */
+ public void reduce(WritableComparable key,
+ Iterator<IndexedTuple> indInp,
+ OutputCollector<WritableComparable, Writable> oc,
+ Reporter reporter) throws IOException {
+
+ runnableReporter.setReporter(reporter);
+
+ Object k = DataType.convertToPigType(key);
+ pack.attachInput(k, indInp);
+
+ try {
+ Tuple t=null;
+ Result res = pack.getNext(t);
+ if(res.returnStatus==POStatus.STATUS_OK){
+ Tuple packRes = (Tuple)res.result;
+
+ if(rp.isEmpty()){
+ oc.collect(null, packRes);
+ return;
+ }
+
+ rp.attachInput(packRes);
+
+ List<PhysicalOperator> leaves = rp.getLeaves();
+
+ PhysicalOperator leaf = leaves.get(0);
+ while(true){
+ Result redRes = leaf.getNext(t);
+
+ if(redRes.returnStatus==POStatus.STATUS_OK){
+ oc.collect(null, (Tuple)redRes.result);
+ continue;
+ }
+
+ if(redRes.returnStatus==POStatus.STATUS_EOP)
+ return;
+
+ if(redRes.returnStatus==POStatus.STATUS_NULL)
+ continue;
+
+ if(redRes.returnStatus==POStatus.STATUS_ERR){
+ IOException ioe = new IOException("Received Error while " +
+ "processing the reduce plan.");
+ throw ioe;
+ }
+ }
+ }
+
+ if(res.returnStatus==POStatus.STATUS_NULL)
+ return;
+
+ if(res.returnStatus==POStatus.STATUS_ERR){
+ IOException ioe = new IOException("Packaging error while processing group");
+ throw ioe;
+ }
+
+
+ } catch (ExecException e) {
+ IOException ioe = new IOException(e.getMessage());
+ ioe.initCause(e.getCause());
+ throw ioe;
+ }
+ }
+
+
+ /**
+ * Will be called once all the intermediate keys and values are
+ * processed. So right place to stop the reporter thread.
+ */
+ @Override
+ public void close() throws IOException {
+ super.close();
+ if(runnableReporter!=null)
+ runnableReporter.setDone(true);
+ }
+ }
+
+ /*interface MapOutputCollector<K extends WritableComparable, V extends Writable>
+ extends OutputCollector<K, V> {
+
+ public void close() throws IOException;
+
+ public void flush() throws IOException;
+
+ }
+
+ static class DirectMapOutputCollector<K extends WritableComparable, V extends Writable>
+ implements MapOutputCollector<K, V> {
+
+ private RecordWriter<K, V> out = null;
+
+ private Reporter reporter = null;
+
+ @SuppressWarnings("unchecked")
+ public DirectMapOutputCollector(JobConf job, Reporter reporter)
+ throws IOException {
+ this.reporter = reporter;
+ String finalName = job.getOutputPath().toString();
+ FileSystem fs = FileSystem.get(job);
+
+ out = job.getOutputFormat().getRecordWriter(fs, job, finalName,
+ reporter);
+ }
+
+ public void close() throws IOException {
+ if (this.out != null) {
+ out.close(this.reporter);
+ }
+
+ }
+
+ public void flush() throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ public void collect(K key, V value) throws IOException {
+ System.out.println(value.toString());
+ }
+ }
+
+ public static void main(String[] args) throws IOException {
+ Random r = new Random();
+ PhysicalPlan<PhysicalOperator> rp = new PhysicalPlan<PhysicalOperator>();
+ POForEach fe = GenPhyOp.topForEachOPWithPlan(1);
+ rp.add(fe);
+ PigMapReduce.Reduce red = new PigMapReduce.Reduce();
+ POPackage pk = GenPhyOp.topPackageOp();
+ pk.setKeyType(DataType.INTEGER);
+ pk.setNumInps(1);
+ boolean[] inner = {false};
+ pk.setInner(inner);
+
+ JobConf jConf = new JobConf();
+ jConf.set("pig.reducePlan", ObjectSerializer.serialize(rp));
+ jConf.set("pig.reduce.package",ObjectSerializer.serialize(pk));
+ jConf.setOutputFormat(PigOutputFormat.class);
+ jConf.setOutputPath(new Path("pigmrtst1"));
+ red.configure(jConf);
+
+ WritableComparable key = new IntWritable(1);
+ List<IndexedTuple> itLst = new ArrayList<IndexedTuple>();
+ for(int i=0;i<2;i++){
+ Tuple t = TupleFactory.getInstance().newTuple();
+ t.append(GenRandomData.genRandString(r));
+ t.append(1);
+ IndexedTuple it = new IndexedTuple(t,0);
+ itLst.add(it);
+ }
+ red.reduce(key,itLst.iterator(),new DirectMapOutputCollector(jConf,reporter), reporter);
+ }*/
+}
\ No newline at end of file
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigOutputFormat.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigOutputFormat.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigOutputFormat.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigOutputFormat.java Tue May 13 14:11:21 2008
@@ -0,0 +1,123 @@
+/*
+ * 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.pig.impl.mapReduceLayer;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.OutputFormat;
+import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.util.Progressable;
+import org.apache.pig.StoreFunc;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.tools.bzip2r.CBZip2OutputStream;
+
+/**
+ * The better half of PigInputFormat which is responsible
+ * for the Store functionality. It is the exact mirror
+ * image of PigInputFormat having RecordWriter instead
+ * of a RecordReader.
+ */
+public class PigOutputFormat implements OutputFormat {
+ public static final String PIG_OUTPUT_FUNC = "pig.output.func";
+
+ public RecordWriter getRecordWriter(FileSystem fs, JobConf job,
+ String name, Progressable progress) throws IOException {
+ Path outputDir = job.getOutputPath();
+ return getRecordWriter(fs, job, outputDir, name, progress);
+ }
+
+ public PigRecordWriter getRecordWriter(FileSystem fs, JobConf job,
+ Path outputDir, String name, Progressable progress)
+ throws IOException {
+ StoreFunc store;
+ String storeFunc = job.get("pig.storeFunc", "");
+ if (storeFunc.length() == 0) {
+ store = new PigStorage();
+ } else {
+ try {
+ store = (StoreFunc) PigContext
+ .instantiateFuncFromSpec(storeFunc);
+ } catch (Exception e) {
+ RuntimeException re = new RuntimeException(e.getClass()
+ .getName()
+ + ": " + e.getMessage());
+ re.setStackTrace(e.getStackTrace());
+ throw re;
+ }
+ }
+ String parentName = outputDir.getParent().getName();
+ int suffixStart = parentName.lastIndexOf('.');
+ if (suffixStart != -1) {
+ String suffix = parentName.substring(suffixStart);
+ if (suffix.equals(".bz") || suffix.equals(".bz2")) {
+ name = name + suffix;
+ }
+ }
+ return new PigRecordWriter(fs, new Path(outputDir, name), store);
+ }
+
+ public void checkOutputSpecs(FileSystem fs, JobConf job) throws IOException {
+ // TODO We really should validate things here
+ return;
+ }
+
+ static public class PigRecordWriter implements RecordWriter {
+ private OutputStream os = null;
+
+ private StoreFunc sfunc = null;
+
+ public PigRecordWriter(FileSystem fs, Path file, StoreFunc sfunc)
+ throws IOException {
+ this.sfunc = sfunc;
+ fs.delete(file);
+ this.os = fs.create(file);
+ String name = file.getName();
+ if (name.endsWith(".bz") || name.endsWith(".bz2")) {
+ os = new CBZip2OutputStream(os);
+ }
+ this.sfunc.bindTo(os);
+ }
+
+ /**
+ * We only care about the values, so we are going to skip the keys when
+ * we write.
+ *
+ * @see org.apache.hadoop.mapred.RecordWriter#write(org.apache.hadoop.io.WritableComparable,
+ * org.apache.hadoop.io.Writable)
+ */
+ public void write(WritableComparable key, Writable value)
+ throws IOException {
+ this.sfunc.putNext((Tuple) value);
+ }
+
+ public void close(Reporter reporter) throws IOException {
+ sfunc.finish();
+ os.close();
+ }
+
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigSplit.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigSplit.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigSplit.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigSplit.java Tue May 13 14:11:21 2008
@@ -0,0 +1,178 @@
+/*
+ * 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.pig.impl.mapReduceLayer;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+
+/**
+ * The main split class that maintains important
+ * information about the input split.
+ *
+ */
+public class PigSplit implements InputSplit {
+ //The input file this split represents
+ Path file;
+
+ //The start pos from which this split starts
+ //in file
+ long start;
+
+ //The length of this split
+ long length;
+
+ //The operators to which the tuples from this
+ //input file are attached. These are the successors
+ //of the load operator representing this input
+ ArrayList<OperatorKey> targetOps;
+
+ //The load function that parses this split and
+ //produces tuples from it
+ String parser;
+
+ //The filesystem on which file resides
+ FileSystem fs;
+
+ //The context in which Pig was running when this
+ //job got submitted.
+ PigContext pigContext;
+
+ public PigSplit() {
+ }
+
+ public PigSplit(PigContext pigContext, FileSystem fs, Path path,
+ String parser, List<OperatorKey> targetOps, long start, long length) {
+ this.fs = fs;
+ this.file = path;
+ this.start = start;
+ this.length = length;
+ this.targetOps = (ArrayList) targetOps;
+ this.parser = parser;
+ this.pigContext = pigContext;
+ }
+
+ public String getParser() {
+ return parser;
+ }
+
+ public long getStart() {
+ return start;
+ }
+
+ public long getLength() {
+ return length;
+ }
+
+ public Path getPath() {
+ return file;
+ }
+
+ public List<OperatorKey> getTargetOps() {
+ return targetOps;
+ }
+
+ public LoadFunc getLoadFunction() {
+ LoadFunc loader = null;
+ if (this.parser == null) {
+ loader = new PigStorage();
+ } else {
+ try {
+ loader = (LoadFunc) PigContext
+ .instantiateFuncFromSpec(this.parser);
+ } catch (Exception exp) {
+ throw new RuntimeException("can't instantiate " + parser);
+ }
+ }
+ return loader;
+ }
+
+ public String[] getLocations() throws IOException {
+ String hints[][] = fs.getFileCacheHints(file, start, length);
+ int total = 0;
+ for (int i = 0; i < hints.length; i++) {
+ total += hints[i].length;
+ }
+ String locations[] = new String[total];
+ int count = 0;
+ for (int i = 0; i < hints.length; i++) {
+ for (int j = 0; j < hints[i].length; j++) {
+ locations[count++] = hints[i][j];
+ }
+ }
+ return locations;
+ }
+
+ public void readFields(DataInput is) throws IOException {
+ file = new Path(is.readUTF());
+ start = is.readLong();
+ length = is.readLong();
+ pigContext = (PigContext) readObject(is);
+ targetOps = (ArrayList<OperatorKey>) readObject(is);
+ parser = is.readUTF();
+ }
+
+ public void write(DataOutput os) throws IOException {
+ os.writeUTF(file.toString());
+ os.writeLong(start);
+ os.writeLong(length);
+ writeObject(pigContext, os);
+ writeObject(targetOps, os);
+ os.writeUTF(parser);
+ }
+
+ private void writeObject(Serializable obj, DataOutput os)
+ throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ oos.writeObject(obj);
+ byte[] bytes = baos.toByteArray();
+ os.writeInt(bytes.length);
+ os.write(bytes);
+ }
+
+ private Object readObject(DataInput is) throws IOException {
+ byte[] bytes = new byte[is.readInt()];
+ is.readFully(bytes);
+ ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(
+ bytes));
+ try {
+ return ois.readObject();
+ } catch (ClassNotFoundException cnfe) {
+ IOException newE = new IOException(cnfe.getMessage());
+ newE.initCause(cnfe);
+ throw newE;
+ }
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/RunnableReporter.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/RunnableReporter.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/RunnableReporter.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/RunnableReporter.java Tue May 13 14:11:21 2008
@@ -0,0 +1,64 @@
+/*
+ * 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.pig.impl.mapReduceLayer;
+
+import org.apache.hadoop.mapred.Reporter;
+
+public class RunnableReporter implements Runnable {
+ Reporter rep;
+ boolean done = false;
+ private long sleepTime;
+ String inputFile;
+
+ public RunnableReporter(long sleepTime) {
+ this.sleepTime = sleepTime;
+ }
+
+ /**
+ * While the task is not done, keep reporting
+ * status back every sleeptime millisecs
+ */
+ public void run() {
+ while(!done){
+ if(rep!=null)
+ rep.setStatus("Processing input file: " + inputFile);
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {}
+ }
+ }
+
+ public void setReporter(Reporter rep){
+ this.rep = rep;
+ }
+
+ public void setSleepTime(long sleepTime) {
+ this.sleepTime = sleepTime;
+ }
+
+ public void setInputFile(String inputFile) {
+ this.inputFile = inputFile;
+ }
+
+ public void setDone(boolean done) {
+ this.done = done;
+ }
+}
\ No newline at end of file
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROpPlanVisitor.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROpPlanVisitor.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROpPlanVisitor.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROpPlanVisitor.java Tue May 13 14:11:21 2008
@@ -0,0 +1,45 @@
+/*
+ * 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.pig.impl.mapReduceLayer.plans;
+
+import org.apache.pig.impl.mapReduceLayer.MapReduceOper;
+import org.apache.pig.impl.plan.PlanWalker;
+import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.impl.plan.PlanVisitor;
+
+/**
+ * A visitor for the MROperPlan class
+ *
+ */
+public class MROpPlanVisitor extends PlanVisitor<MapReduceOper, MROperPlan> {
+
+ public MROpPlanVisitor(MROperPlan plan, PlanWalker<MapReduceOper, MROperPlan> walker) {
+ super(plan, walker);
+ }
+
+ @Override
+ public void visit() throws VisitorException {
+ // TODO Auto-generated method stub
+
+ }
+
+ public void visitMROp(MapReduceOper mr) {
+ // do nothing
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROperPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROperPlan.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROperPlan.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/MROperPlan.java Tue May 13 14:11:21 2008
@@ -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.pig.impl.mapReduceLayer.plans;
+
+import org.apache.pig.impl.mapReduceLayer.MapReduceOper;
+import org.apache.pig.impl.plan.OperatorPlan;
+
+
+/**
+ * A Plan used to create the plan of
+ * Map Reduce Operators which can be
+ * converted into the Job Control
+ * object. This is necessary to capture
+ * the dependencies among jobs
+ */
+public class MROperPlan extends OperatorPlan<MapReduceOper> {
+
+ public MROperPlan() {
+ // TODO Auto-generated constructor stub
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/PigProgressable.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/PigProgressable.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/PigProgressable.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/PigProgressable.java Tue May 13 14:11:21 2008
@@ -0,0 +1,10 @@
+package org.apache.pig.impl.physicalLayer;
+
+public interface PigProgressable {
+ //Use to just inform that you are
+ //alive
+ public void progress();
+
+ //If you have a status to report
+ public void progress(String msg);
+}
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/ExprPlanVisitor.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/ExprPlanVisitor.java?rev=656011&r1=656010&r2=656011&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/ExprPlanVisitor.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/ExprPlanVisitor.java Tue May 13 14:11:21 2008
@@ -38,6 +38,7 @@
import org.apache.pig.impl.physicalLayer.topLevelOperators.expressionOperators.binaryExprOps.comparators.NotEqualToExpr;
import org.apache.pig.impl.physicalLayer.topLevelOperators.expressionOperators.unaryExprOps.PONegative;
import org.apache.pig.impl.plan.PlanWalker;
+import org.apache.pig.impl.plan.VisitorException;
/**
* The visitor to be used for visiting expression plans.
@@ -50,60 +51,60 @@
private final Log log = LogFactory.getLog(getClass());
public ExprPlanVisitor(ExprPlan plan,
- PlanWalker<ExpressionOperator, ExprPlan> walker) {
+ PlanWalker<ExpressionOperator, ExprPlan> walker) {
super(plan, walker);
// TODO Auto-generated constructor stub
}
- public void visitConstant(ConstantExpression cnst){
+ public void visitConstant(ConstantExpression cnst) throws VisitorException{
//do nothing
}
- public void visitProject(POProject proj){
+ public void visitProject(POProject proj) throws VisitorException{
//do nothing
}
- public void visitGreaterThan(GreaterThanExpr grt){
+ public void visitGreaterThan(GreaterThanExpr grt) throws VisitorException{
//do nothing
}
- public void visitLessThan(LessThanExpr lt){
+ public void visitLessThan(LessThanExpr lt) throws VisitorException{
//do nothing
}
- public void visitGTOrEqual(GTOrEqualToExpr gte){
+ public void visitGTOrEqual(GTOrEqualToExpr gte) throws VisitorException{
//do nothing
}
- public void visiLTOrEqual(LTOrEqualToExpr lte){
+ public void visiLTOrEqual(LTOrEqualToExpr lte) throws VisitorException{
//do nothing
}
- public void visitEqualTo(EqualToExpr eq){
+ public void visitEqualTo(EqualToExpr eq) throws VisitorException{
//do nothing
}
- public void visitNotEqualTo(NotEqualToExpr eq){
+ public void visitNotEqualTo(NotEqualToExpr eq) throws VisitorException{
//do nothing
}
- public void visitAdd(Add add) {
+ public void visitAdd(Add add) throws VisitorException{
//do nothing
}
- public void visitSubtract(Subtract sub) {
+ public void visitSubtract(Subtract sub) throws VisitorException {
//do nothing
}
- public void visitMultiply(Multiply mul) {
+ public void visitMultiply(Multiply mul) throws VisitorException {
//do nothing
}
- public void visitDivide(Divide dv) {
+ public void visitDivide(Divide dv) throws VisitorException {
//do nothing
}
- public void visitMod(Mod mod) {
+ public void visitMod(Mod mod) throws VisitorException {
//do nothing
}
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhyPlanVisitor.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhyPlanVisitor.java?rev=656011&r1=656010&r2=656011&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhyPlanVisitor.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhyPlanVisitor.java Tue May 13 14:11:21 2008
@@ -17,27 +17,30 @@
*/
package org.apache.pig.impl.physicalLayer.plans;
+import java.util.List;
+
import org.apache.pig.impl.physicalLayer.topLevelOperators.POFilter;
-import org.apache.pig.impl.physicalLayer.topLevelOperators.POGenerate;
-//import org.apache.pig.impl.physicalLayer.topLevelOperators.POGenerate;
-//import org.apache.pig.impl.physicalLayer.topLevelOperators.POGlobalRearrange;
-import org.apache.pig.impl.physicalLayer.topLevelOperators.PODistinct;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POForEach;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POGenerate;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POGlobalRearrange;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POLoad;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POLocalRearrange;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POPackage;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POSplit;
import org.apache.pig.impl.physicalLayer.topLevelOperators.PORead;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POSort;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.PODistinct;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POStore;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POUnion;
import org.apache.pig.impl.physicalLayer.topLevelOperators.POUserFunc;
import org.apache.pig.impl.physicalLayer.topLevelOperators.PhysicalOperator;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POUnion;
import org.apache.pig.impl.physicalLayer.topLevelOperators.expressionOperators.ExpressionOperator;
-//import org.apache.pig.impl.physicalLayer.topLevelOperators.StartMap;
+import org.apache.pig.impl.plan.DependencyOrderWalker;
import org.apache.pig.impl.plan.DepthFirstWalker;
-import org.apache.pig.impl.plan.Operator;
-import org.apache.pig.impl.plan.PlanVisitor;
import org.apache.pig.impl.plan.PlanWalker;
import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.impl.plan.PlanVisitor;
/**
* The visitor class for the Physical Plan. To use this,
@@ -59,11 +62,11 @@
super(plan, walker);
}
- public void visitLoad(POLoad ld){
+ public void visitLoad(POLoad ld) throws VisitorException{
//do nothing
}
- public void visitStore(POStore st){
+ public void visitStore(POStore st) throws VisitorException{
//do nothing
}
@@ -89,37 +92,48 @@
popWalker();
}
-// public void visitGlobalRearrange(POGlobalRearrange gr){
-// //do nothing
-// }
-//
-// public void visitStartMap(StartMap sm){
-// //do nothing
-// }
-//
- public void visitPackage(POPackage pkg){
+ public void visitGlobalRearrange(POGlobalRearrange gr) throws VisitorException{
+ //do nothing
+ }
+
+ public void visitPackage(POPackage pkg) throws VisitorException{
+ //do nothing
+ }
+
+ public void visitGenerate(POGenerate pogen) throws VisitorException {
+ List<ExprPlan> inpPlans = pogen.getInputPlans();
+ for (ExprPlan plan : inpPlans) {
+ ExprPlanVisitor epv = new ExprPlanVisitor(plan,new DependencyOrderWalker<ExpressionOperator, ExprPlan>(plan));
+ epv.visit();
+ }
+ }
+
+ public void visitUnion(POUnion un) throws VisitorException{
//do nothing
}
- public void visitGenerate(POGenerate pogen) {
+ public void visitSplit(POSplit spl) throws VisitorException{
//do nothing
}
- public void visitDistinct(PODistinct distinct) {
-
+ public void visitDistinct(PODistinct distinct) throws VisitorException {
+ //do nothing
}
- public void visitRead(PORead read) {
-
+ public void visitRead(PORead read) throws VisitorException {
+ //do nothing
}
- public void visitSort(POSort sort) {
-
+ public void visitSort(POSort sort) throws VisitorException {
+ List<ExprPlan> inpPlans = sort.getSortPlans();
+ for (ExprPlan plan : inpPlans) {
+ ExprPlanVisitor epv = new ExprPlanVisitor(plan,new DependencyOrderWalker<ExpressionOperator, ExprPlan>(plan));
+ epv.visit();
+ }
}
- public void visitUserFunc(POUserFunc userFunc) {
-
+ public void visitUserFunc(POUserFunc userFunc) throws VisitorException {
+ //do nothing
}
}
-
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhysicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhysicalPlan.java?rev=656011&r1=656010&r2=656011&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhysicalPlan.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PhysicalPlan.java Tue May 13 14:11:21 2008
@@ -17,6 +17,8 @@
*/
package org.apache.pig.impl.physicalLayer.plans;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
import java.io.OutputStream;
import java.util.List;
@@ -24,6 +26,7 @@
import org.apache.pig.impl.physicalLayer.topLevelOperators.PhysicalOperator;
import org.apache.pig.impl.plan.OperatorPlan;
import org.apache.pig.impl.plan.PlanException;
+import org.apache.pig.impl.plan.VisitorException;
/**
*
@@ -34,6 +37,11 @@
*/
public class PhysicalPlan<E extends PhysicalOperator> extends OperatorPlan<E> {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
public PhysicalPlan() {
super();
}
@@ -50,8 +58,18 @@
* @param out : OutputStream to which the visual representation is written
*/
public void explain(OutputStream out){
- //Use a plan visitor and output the current physical
- //plan into out
+ PlanPrinter<E, PhysicalPlan<E>> mpp = new PlanPrinter<E, PhysicalPlan<E>>(
+ this);
+
+ try {
+ mpp.print(out);
+ } catch (VisitorException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ } catch (IOException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
}
@Override
@@ -60,4 +78,47 @@
super.connect(from, to);
to.setInputs(getPredecessors(to));
}
+
+ /*public void connect(List<E> from, E to) throws IOException{
+ if(!to.supportsMultipleInputs()){
+ throw new IOException("Invalid Operation on " + to.name() + ". It doesn't support multiple inputs.");
+ }
+
+ }*/
+
+ @Override
+ public void remove(E op) {
+ op.setInputs(null);
+ List<E> sucs = getSuccessors(op);
+ if(sucs!=null && sucs.size()!=0){
+ for (E suc : sucs) {
+ suc.setInputs(null);
+ }
+ }
+ super.remove(op);
+ }
+
+ public boolean isEmpty() {
+ return (mOps.size() == 0);
+ }
+
+ @Override
+ public String toString() {
+ if(isEmpty())
+ return "Empty Plan!";
+ else{
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ explain(baos);
+ return baos.toString();
+ }
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ // TODO Auto-generated method stub
+ return super.equals(obj);
+ }
+
+
+
}
Added: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PlanPrinter.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PlanPrinter.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PlanPrinter.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/plans/PlanPrinter.java Tue May 13 14:11:21 2008
@@ -0,0 +1,250 @@
+/*
+ * 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.pig.impl.physicalLayer.plans;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POFilter;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POForEach;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POGenerate;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POGlobalRearrange;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POLoad;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POLocalRearrange;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POPackage;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POSort;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POStore;
+import org.apache.pig.impl.physicalLayer.topLevelOperators.POUnion;
+import org.apache.pig.impl.plan.DepthFirstWalker;
+import org.apache.pig.impl.plan.Operator;
+import org.apache.pig.impl.plan.OperatorPlan;
+import org.apache.pig.impl.plan.PlanVisitor;
+import org.apache.pig.impl.plan.VisitorException;
+
+public class PlanPrinter<O extends Operator, P extends OperatorPlan<O>> extends
+ PlanVisitor<O, P> {
+
+ String TAB1 = " ";
+
+ String TABMore = "| ";
+
+ String LSep = "|\n|---";
+
+ String USep = "| |\n| ";
+
+ int levelCntr = -1;
+
+ OutputStream printer;
+
+ public PlanPrinter(P plan) {
+ super(plan, new DepthFirstWalker<O, P>(plan));
+ }
+
+ @Override
+ public void visit() throws VisitorException {
+ try {
+ System.out.write(depthFirstPP().getBytes());
+ } catch (IOException e) {
+ throw new VisitorException(e.getMessage());
+ }
+ }
+
+ public void print(OutputStream printer) throws VisitorException, IOException {
+ this.printer = printer;
+ printer.write(depthFirstPP().getBytes());
+ }
+
+ protected void breadthFirst() throws VisitorException {
+ List<O> leaves = mPlan.getLeaves();
+ Set<O> seen = new HashSet<O>();
+ breadthFirst(leaves, seen);
+ }
+
+ private void breadthFirst(Collection<O> predecessors, Set<O> seen)
+ throws VisitorException {
+ ++levelCntr;
+ dispTabs();
+
+ List<O> newPredecessors = new ArrayList<O>();
+ for (O pred : predecessors) {
+ if (seen.add(pred)) {
+ List<O> predLst = mPlan.getPredecessors(pred);
+ if (predLst != null)
+ newPredecessors.addAll(predLst);
+
+ pred.visit(this);
+ }
+ }
+ if (newPredecessors.size() > 0) {
+ System.out.println();
+ breadthFirst(newPredecessors, seen);
+ }
+ }
+
+ protected String depthFirstPP() throws VisitorException {
+ StringBuilder sb = new StringBuilder();
+ List<O> leaves = mPlan.getLeaves();
+ Collections.sort(leaves);
+ for (O leaf : leaves) {
+ sb.append(depthFirst(leaf));
+ sb.append("\n");
+ }
+ sb.delete(sb.length() - "\n".length(), sb.length());
+ sb.delete(sb.length() - "\n".length(), sb.length());
+ return sb.toString();
+ }
+
+ private String planString(PhysicalPlan pp){
+ StringBuilder sb = new StringBuilder();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ if(pp!=null)
+ pp.explain(baos);
+ else
+ return "";
+ sb.append(USep);
+ sb.append(shiftStringByTabs(baos.toString(), 2));
+ return sb.toString();
+ }
+
+ private String planString(List<ExprPlan> lep){
+ StringBuilder sb = new StringBuilder();
+ if(lep!=null)
+ for (ExprPlan ep : lep) {
+ sb.append(planString(ep));
+ }
+ return sb.toString();
+ }
+
+ private String depthFirst(O node) throws VisitorException {
+ StringBuilder sb = new StringBuilder(node.name() + "\n");
+ if(node instanceof POFilter){
+ sb.append(planString(((POFilter)node).getPlan()));
+ }
+ else if(node instanceof POForEach){
+ sb.append(planString(((POForEach)node).getPlan()));
+ }
+ else if(node instanceof POGenerate){
+ sb.append(planString(((POGenerate)node).getInputPlans()));
+
+ }
+ else if(node instanceof POLocalRearrange){
+ sb.append(planString(((POLocalRearrange)node).getPlan()));
+ }
+ else if(node instanceof POSort){
+ sb.append(planString(((POSort)node).getSortPlans()));
+ }
+
+ List<O> predecessors = mPlan.getPredecessors(node);
+
+ if (predecessors == null)
+ return sb.toString();
+
+ Collections.sort(predecessors);
+ int i = 0;
+ for (O pred : predecessors) {
+ i++;
+ String DFStr = depthFirst(pred);
+ if (DFStr != null) {
+ sb.append(LSep);
+ if (i < predecessors.size())
+ sb.append(shiftStringByTabs(DFStr, 2));
+ else
+ sb.append(shiftStringByTabs(DFStr, 1));
+ }
+ }
+ return sb.toString();
+ }
+
+ private String shiftStringByTabs(String DFStr, int TabType) {
+ StringBuilder sb = new StringBuilder();
+ String[] spl = DFStr.split("\n");
+
+ String tab = (TabType == 1) ? TAB1 : TABMore;
+
+ sb.append(spl[0] + "\n");
+ for (int i = 1; i < spl.length; i++) {
+ sb.append(tab);
+ sb.append(spl[i]);
+ sb.append("\n");
+ }
+ return sb.toString();
+ }
+
+ private void dispTabs() {
+ for (int i = 0; i < levelCntr; i++)
+ System.out.print(TAB1);
+ }
+
+ public void visitLoad(POLoad op) {
+ System.out.print(op.name() + " ");
+ }
+
+ public void visitStore(POStore op) {
+ System.out.print(op.name() + " ");
+ }
+
+ public void visitFilter(POFilter op) {
+ System.out.print(op.name() + " ");
+ }
+
+ public void visitLocalRearrange(POLocalRearrange op) {
+ System.out.print(op.name() + " ");
+ }
+
+ public void visitGlobalRearrange(POGlobalRearrange op) {
+ System.out.print(op.name() + " ");
+ }
+
+ public void visitPackage(POPackage op) {
+ System.out.print(op.name() + " ");
+ }
+
+ public void visitStartMap(POUnion op) {
+ System.out.print(op.name() + " ");
+ }
+
+ /*public static void main(String[] args) throws PlanException, ExecException {
+ ExprPlan ep = GenPhyOp.arithPlan();
+ GreaterThanExpr gt = GenPhyOp.compGreaterThanExpr();
+ ConstantExpression ce = GenPhyOp.exprConst();
+ ce.setValue(50);
+ ep.add(ce);
+ ep.addAsLeaf(gt);
+
+ POFilter fil = GenPhyOp.topFilterOp();
+ fil.setPlan(ep);
+
+ PhysicalPlan php = new PhysicalPlan();
+ php.add(fil);
+ int fields[] = {0};
+ Tuple sample = TupleFactory.getInstance().newTuple();
+ sample.append("S");
+// POForEach fe = GenPhyOp.topForEachOPWithPlan(fields, sample);
+ POForEach fe = GenPhyOp.topForEachOp();
+ php.add(fe);
+ php.connect(fil, fe);
+ php.explain(System.out);
+ }*/
+}
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POFilter.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POFilter.java?rev=656011&r1=656010&r2=656011&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POFilter.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POFilter.java Tue May 13 14:11:21 2008
@@ -25,12 +25,12 @@
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.logicalLayer.OperatorKey;
-import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.physicalLayer.POStatus;
import org.apache.pig.impl.physicalLayer.Result;
import org.apache.pig.impl.physicalLayer.plans.ExprPlan;
import org.apache.pig.impl.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.impl.physicalLayer.topLevelOperators.expressionOperators.binaryExprOps.comparators.ComparisonOperator;
+import org.apache.pig.impl.plan.VisitorException;
/**
* This is an implementation of the Filter operator. It has an Expression Plan
@@ -45,6 +45,11 @@
*/
public class POFilter extends PhysicalOperator<PhyPlanVisitor> {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
private Log log = LogFactory.getLog(getClass());
// The expression plan
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POForEach.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POForEach.java?rev=656011&r1=656010&r2=656011&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POForEach.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POForEach.java Tue May 13 14:11:21 2008
@@ -41,6 +41,11 @@
*/
public class POForEach extends PhysicalOperator<PhyPlanVisitor> {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
private Log log = LogFactory.getLog(getClass());
PhysicalPlan<PhysicalOperator> plan;
@@ -116,14 +121,14 @@
if(res.returnStatus==POStatus.STATUS_OK){
return res;
}
- if(res.returnStatus==POStatus.STATUS_ERR)
- return res;
- if(res.returnStatus==POStatus.STATUS_NULL)
- continue;
if(res.returnStatus==POStatus.STATUS_EOP){
processingPlan = false;
break;
}
+ if(res.returnStatus==POStatus.STATUS_ERR)
+ return res;
+ if(res.returnStatus==POStatus.STATUS_NULL)
+ continue;
}
}
//The nested plan processing is done or is
@@ -133,15 +138,16 @@
while (true) {
inp = processInput();
if (inp.returnStatus == POStatus.STATUS_EOP || inp.returnStatus == POStatus.STATUS_ERR)
- break;
+ return inp;
if (inp.returnStatus == POStatus.STATUS_NULL)
continue;
plan.attachInput((Tuple) inp.result);
res = gen.getNext(t);
+
if (inp.returnStatus == POStatus.STATUS_EOP || inp.returnStatus == POStatus.STATUS_ERR)
- break;
+ return inp;
if(inp.returnStatus == POStatus.STATUS_NULL)
continue;
@@ -149,7 +155,6 @@
return res;
}
- return inp;
}
public PhysicalPlan<PhysicalOperator> getPlan() {
Modified: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGenerate.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGenerate.java?rev=656011&r1=656010&r2=656011&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGenerate.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGenerate.java Tue May 13 14:11:21 2008
@@ -38,11 +38,15 @@
import org.apache.pig.impl.physicalLayer.plans.ExprPlan;
import org.apache.pig.impl.physicalLayer.plans.PhyPlanVisitor;
import org.apache.pig.impl.physicalLayer.topLevelOperators.expressionOperators.ExpressionOperator;
-import org.apache.pig.impl.plan.PlanVisitor;
import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.impl.plan.PlanVisitor;
-public class POGenerate extends PhysicalOperator {
+public class POGenerate extends PhysicalOperator<PhyPlanVisitor> {
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
private List<Boolean> isToBeFlattened;
//private Boolean outputIsDataBag;
private List<Tuple> outputBuffer = new LinkedList<Tuple>();
@@ -280,11 +284,6 @@
}
@Override
- public void visit(PlanVisitor v) throws VisitorException {
- ((PhyPlanVisitor)v).visitGenerate(this);
- }
-
- @Override
public boolean supportsMultipleInputs() {
return false;
}
@@ -293,6 +292,10 @@
public boolean supportsMultipleOutputs() {
return false;
}
+
+ public List<ExprPlan> getInputPlans() {
+ return inputPlans;
+ }
Added: incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGlobalRearrange.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGlobalRearrange.java?rev=656011&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGlobalRearrange.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/topLevelOperators/POGlobalRearrange.java Tue May 13 14:11:21 2008
@@ -0,0 +1,88 @@
+/*
+ * 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.pig.impl.physicalLayer.topLevelOperators;
+
+import java.util.List;
+
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+import org.apache.pig.impl.physicalLayer.Result;
+import org.apache.pig.impl.physicalLayer.plans.PhyPlanVisitor;
+import org.apache.pig.impl.plan.VisitorException;
+
+/**
+ * Dummy operator to test MRCompiler.
+ * This will be a local operator and its
+ * getNext methods have to be implemented
+ *
+ */
+public class POGlobalRearrange extends PhysicalOperator<PhyPlanVisitor> {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ public POGlobalRearrange(OperatorKey k) {
+ this(k, -1, null);
+ }
+
+ public POGlobalRearrange(OperatorKey k, int rp) {
+ this(k, rp, null);
+ }
+
+ public POGlobalRearrange(OperatorKey k, List inp) {
+ this(k, -1, null);
+ }
+
+ public POGlobalRearrange(OperatorKey k, int rp, List inp) {
+ super(k, rp, inp);
+ }
+
+ @Override
+ public void visit(PhyPlanVisitor v) throws VisitorException {
+ v.visitGlobalRearrange(this);
+ }
+
+ @Override
+ public String name() {
+ return "Global Rearrange - " + mKey.toString();
+ }
+
+ @Override
+ public boolean supportsMultipleInputs() {
+ return true;
+ }
+
+ @Override
+ public boolean supportsMultipleOutputs() {
+ return false;
+ }
+
+ @Override
+ public boolean isBlocking() {
+ return true;
+ }
+
+ @Override
+ public Result getNext(Tuple t) throws ExecException {
+ // TODO Auto-generated method stub
+ return super.getNext(t);
+ }
+}