You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ol...@apache.org on 2008/02/01 04:11:39 UTC
svn commit: r617338 [3/5] - in /incubator/pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/ src/org/apache/pig/backend/datastorage/
src/org/apache/pig/backend/executionengine/
src/org/apache/pig/backend/hadoop/ src/org/apache/pig/backend/ha...
Added: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigCombine.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,141 @@
+/*
+ * 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.backend.hadoop.executionengine.mapreduceExec;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Datum;
+import org.apache.pig.data.IndexedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.eval.EvalSpec;
+import org.apache.pig.impl.eval.collector.DataCollector;
+import org.apache.pig.impl.io.FileSpec;
+import org.apache.pig.impl.util.ObjectSerializer;
+
+
+public class PigCombine implements Reducer {
+
+ private JobConf job;
+ private CombineDataOutputCollector finalout;
+ private DataCollector evalPipe;
+ private OutputCollector oc;
+ private int index;
+ private int inputCount;
+ private DataBag bags[];
+
+ public void reduce(WritableComparable key, Iterator values, OutputCollector output, Reporter reporter)
+ throws IOException {
+
+ try {
+ PigContext pigContext = (PigContext) ObjectSerializer.deserialize(job.get("pig.pigContext"));
+ if (evalPipe == null) {
+ inputCount = ((ArrayList<FileSpec>)ObjectSerializer.deserialize(job.get("pig.inputs"))).size();
+ oc = output;
+ finalout = new CombineDataOutputCollector(oc);
+ String evalSpec = job.get("pig.combineFunc", "");
+ EvalSpec esp = (EvalSpec)ObjectSerializer.deserialize(evalSpec);
+ if(esp != null) esp.instantiateFunc(pigContext);
+ evalPipe = esp.setupPipe(finalout);
+ //throw new RuntimeException("combine spec: " + evalSpec + " combine pipe: " + esp.toString());
+
+ bags = new DataBag[inputCount];
+ for (int i = 0; i < inputCount; i++) {
+ bags[i] = BagFactory.getInstance().newDefaultBag();
+ }
+ }
+
+ PigSplit split = PigInputFormat.PigRecordReader.getPigRecordReader().getPigFileSplit();
+ index = split.getIndex();
+
+ Datum groupName = ((Tuple) key).getField(0);
+ finalout.group = ((Tuple) key);
+ finalout.index = index;
+
+ Tuple t = new Tuple(1 + inputCount);
+ t.setField(0, groupName);
+ for (int i = 1; i < 1 + inputCount; i++) {
+ bags[i - 1].clear();
+ t.setField(i, bags[i - 1]);
+ }
+
+ while (values.hasNext()) {
+ IndexedTuple it = (IndexedTuple) values.next();
+ t.getBagField(it.index + 1).add(it.toTuple());
+ }
+ for (int i = 0; i < inputCount; i++) { // XXX: shouldn't we only do this if INNER flag is set?
+ if (t.getBagField(1 + i).size() == 0)
+ return;
+ }
+// throw new RuntimeException("combine input: " + t.toString());
+ evalPipe.add(t);
+ // evalPipe.add(null); // EOF marker
+ } catch (Throwable tr) {
+ tr.printStackTrace();
+ RuntimeException exp = new RuntimeException(tr.getMessage());
+ exp.setStackTrace(tr.getStackTrace());
+ throw exp;
+ }
+ }
+
+ /**
+ * Just save off the PigJobConf for later use.
+ */
+ public void configure(JobConf job) {
+ this.job = job;
+ }
+
+ /**
+ * Nothing happens here.
+ */
+ public void close() throws IOException {
+ }
+
+ private static class CombineDataOutputCollector extends DataCollector {
+ OutputCollector oc = null;
+ Tuple group = null;
+ int index = -1;
+
+ public CombineDataOutputCollector(OutputCollector oc) {
+ super(null);
+ this.oc = oc;
+ }
+
+ @Override
+ public void add(Datum d){
+ if (d == null) return; // EOF marker from eval pipeline; ignore
+ try{
+ // oc.collect(group, new IndexedTuple(((Tuple)d).getTupleField(0),index));
+ oc.collect(group, new IndexedTuple(((Tuple)d),index));
+ }catch (IOException e){
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigInputFormat.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,228 @@
+/*
+ * 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.backend.hadoop.executionengine.mapreduceExec;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+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.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.eval.EvalSpec;
+import org.apache.pig.impl.io.BufferedPositionedInputStream;
+import org.apache.pig.impl.io.FileSpec;
+import org.apache.pig.impl.util.ObjectSerializer;
+import org.apache.tools.bzip2r.CBZip2InputStream;
+
+
+public class PigInputFormat implements InputFormat<Text, Tuple>, JobConfigurable {
+
+ public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
+
+ ArrayList<FileSpec> inputs = (ArrayList<FileSpec>) ObjectSerializer.deserialize(job.get("pig.inputs"));
+ ArrayList<EvalSpec> mapFuncs = (ArrayList<EvalSpec>) ObjectSerializer.deserialize(job.get("pig.mapFuncs",""));
+ ArrayList<EvalSpec> groupFuncs = (ArrayList<EvalSpec>) ObjectSerializer.deserialize(job.get("pig.groupFuncs", ""));
+ PigContext pigContext = (PigContext)ObjectSerializer.deserialize(job.get("pig.pigContext"));
+ //TODO: don't understand this code
+ // added for UNION: set group func arity to match arity of inputs
+ if (groupFuncs!=null && groupFuncs.size() != inputs.size()) {
+ groupFuncs = new ArrayList<EvalSpec>();
+ for (int i = 0; i < groupFuncs.size(); i++) {
+ groupFuncs.set(i,null);
+ }
+ }
+
+ if (inputs.size() != mapFuncs.size()) {
+ throw new IOException("number of inputs != number of map functions: " + inputs.size() + " != "
+ + mapFuncs.size() + ": " + job.get("pig.mapFuncs", "missing"));
+ }
+
+ if (groupFuncs!= null && inputs.size() != groupFuncs.size()) {
+ throw new IOException("number of inputs != number of group functions: " + inputs.size() + " != "
+ + groupFuncs.size());
+ }
+
+ 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, groupFuncs==null ? null : groupFuncs.get(i), mapFuncs.get(i), i, 0, size));
+ } else {
+ while (pos < size) {
+ if (pos + bs > size)
+ bs = size - pos;
+ splits.add(new PigSplit(pigContext, fs, fullPath, parser, groupFuncs==null ? null : groupFuncs.get(i), mapFuncs.get(i), i, pos, bs));
+ pos += bs;
+ }
+ }
+ }
+ }
+ return splits.toArray(new PigSplit[splits.size()]);
+ }
+
+ public RecordReader<Text, Tuple> 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");
+ }
+
+ public static class PigRecordReader implements RecordReader<Text, Tuple> {
+ /**
+ * This is a tremendously ugly hack to get around the fact that mappers do not have access
+ * to their readers. We take advantage of the fact that RecordReader.next and Mapper.map is
+ * run on same the thread to share information through a thread local variable.
+ */
+ static ThreadLocal<PigRecordReader> myReader = new ThreadLocal<PigRecordReader>();
+
+ public static PigRecordReader getPigRecordReader() {
+ return myReader.get();
+ }
+
+ private InputStream is;
+ private FSDataInputStream fsis;
+ private long end;
+ private PigSplit split;
+ LoadFunc loader;
+ CompressionCodecFactory compressionFactory;
+ JobConf job;
+
+ 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;
+ }
+ }
+ myReader.set(this);
+ loader.bindTo(split.getPath().toString(), new BufferedPositionedInputStream(is, start), start, end);
+
+ }
+
+ public JobConf getJobConf(){
+ return job;
+ }
+
+ public boolean next(Text key, Tuple value) throws IOException {
+ Tuple t = loader.getNext();
+ if (t == null) {
+ return false;
+ }
+
+ key.set(split.getPath().getName());
+ value.copyFrom(t);
+ 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 Tuple createValue() {
+ return new Tuple();
+ }
+
+ public float getProgress() throws IOException {
+ float progress = getPos() - split.getStart();
+ float finish = split.getLength();
+ return progress/finish;
+ }
+ }
+
+ public void validateInput(JobConf arg0) throws IOException {
+ }
+
+ }
Added: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigMapReduce.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,337 @@
+/*
+ * 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.backend.hadoop.executionengine.mapreduceExec;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.util.ArrayList;
+import java.util.Iterator;
+
+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.MapRunnable;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.pig.backend.hadoop.executionengine.SplitSpec;
+import org.apache.pig.backend.hadoop.executionengine.mapreduceExec.PigOutputFormat.PigRecordWriter;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Datum;
+import org.apache.pig.data.IndexedTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.eval.EvalSpec;
+import org.apache.pig.impl.eval.StarSpec;
+import org.apache.pig.impl.eval.collector.DataCollector;
+import org.apache.pig.impl.eval.cond.Cond;
+import org.apache.pig.impl.io.FileSpec;
+import org.apache.pig.impl.logicalLayer.LOCogroup;
+import org.apache.pig.impl.util.ObjectSerializer;
+import org.apache.pig.tools.timer.PerformanceTimerFactory;
+
+
+/**
+ * This class is a wrapper of sorts for Pig Map/Reduce jobs. Both the Mapper and the Reducer are
+ * implemented by this class. The methods of this class are driven by job configuration variables:
+ * <dl>
+ * <dt>pig.inputs</dt>
+ * <dd>A semi-colon separated list of inputs. If an input uses a special parser, it will be
+ * specified by adding a colon and the name of the parser to the input. For example:
+ * /tmp/names.txt;/tmp/logs.dat:com.yahoo.research.pig.parser.LogParser will parse /tmp/names.txt
+ * using the default parser and /tmp/logs.dat using com.yahoo.research.pig.parser.LogParser.</dd>
+ * <dt>pig.mapFuncs</dt>
+ * <dd>A semi-colon separated list of functions-specification to be applied to the inputs in the
+ * Map phase. This list must have the same number of items as pig.inputs because the each
+ * functions-spectification will be matched to the corresponding input.</dd>
+ * <dt>pig.groupFuncs</dt>
+ * <dd>A semi-colon separated list of group functions. As with pig.mapFuncs, this list must have
+ * the same number of items as pig.inputs because the each group function will be matched to the
+ * corresponding input.</dd>
+ * <dt>pig.reduceFuncs</dt>
+ * <dd>functions-specification to be applied to the tuples passed into the Reduce phase.</dd>
+ * </dl>
+ *
+ * @author breed
+ */
+public class PigMapReduce implements MapRunnable, Reducer {
+
+ public static Reporter reporter = null;
+
+ JobConf job;
+ private DataCollector evalPipe;
+ private OutputCollector oc;
+ private EvalSpec group;
+ private PigRecordWriter pigWriter;
+ private int index;
+ private int inputCount;
+ private boolean isInner[];
+ private static PigContext pigContext = null;
+ ArrayList<PigRecordWriter> sideFileWriters = new ArrayList<PigRecordWriter>();
+
+ /**
+ * This function is called in MapTask by Hadoop as the Mapper.run() method. We basically pull
+ * the tuples from our PigRecordReader (see ugly ThreadLocal hack), pipe the tuples through the
+ * function pipeline and then close the writer.
+ */
+ public void run(RecordReader input, OutputCollector output, Reporter reporter) throws IOException {
+ PigMapReduce.reporter = reporter;
+
+ oc = output;
+
+ setupMapPipe(reporter);
+
+ // allocate key & value instances that are re-used for all entries
+ WritableComparable key = input.createKey();
+ Writable value = input.createValue();
+ while (input.next(key, value)) {
+ evalPipe.add((Tuple) value);
+ }
+ evalPipe.finishPipe(); // EOF marker
+ evalPipe= null;
+ if (pigWriter != null) {
+ pigWriter.close(reporter);
+ }
+ }
+
+ public void reduce(WritableComparable key, Iterator values, OutputCollector output, Reporter reporter)
+ throws IOException {
+
+ PigMapReduce.reporter = reporter;
+
+ try {
+ oc = output;
+ if (evalPipe == null) {
+ setupReducePipe();
+ }
+
+ DataBag[] bags = new DataBag[inputCount];
+ Datum groupName = ((Tuple) key).getField(0);
+ Tuple t = new Tuple(1 + inputCount);
+ t.setField(0, groupName);
+ for (int i = 1; i < 1 + inputCount; i++) {
+ bags[i - 1] = BagFactory.getInstance().newDefaultBag();
+ t.setField(i, bags[i - 1]);
+ }
+
+ while (values.hasNext()) {
+ IndexedTuple it = (IndexedTuple) values.next();
+ t.getBagField(it.index + 1).add(it.toTuple());
+ }
+
+ for (int i = 0; i < inputCount; i++) {
+ if (isInner[i] && t.getBagField(1 + i).size() == 0)
+ return;
+ }
+
+ evalPipe.add(t);
+ } catch (Throwable tr) {
+ tr.printStackTrace();
+ RuntimeException exp = new RuntimeException(tr.getMessage());
+ exp.setStackTrace(tr.getStackTrace());
+ throw exp;
+ }
+ }
+
+ /**
+ * Just save off the PigJobConf for later use.
+ */
+ public void configure(JobConf job) {
+ this.job = job;
+ }
+
+ /**
+ * Nothing happens here.
+ */
+ public void close() throws IOException {
+ if (evalPipe!=null)
+ evalPipe.finishPipe();
+ }
+
+ public static PigContext getPigContext() {
+ if (pigContext!=null)
+ return pigContext;
+ try {
+ InputStream is = PigMapReduce.class.getResourceAsStream("/pigContext");
+ if (is == null) throw new RuntimeException("/pigContext not found!");
+ return (PigContext)new ObjectInputStream(is).readObject();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private String getTaskId(){
+ String taskidParts[] = job.get("mapred.task.id").split("_"); // The format is
+ // tip_job_m_partition_try
+ return taskidParts[taskidParts.length - 2];
+ }
+
+ private void setupMapPipe(Reporter reporter) throws IOException {
+ PigSplit split = PigInputFormat.PigRecordReader.getPigRecordReader().getPigFileSplit();
+ pigContext = (PigContext)ObjectSerializer.deserialize(job.get("pig.pigContext"));
+ index = split.getIndex();
+ EvalSpec evalSpec = split.getEvalSpec();
+
+ EvalSpec groupSpec = split.getGroupbySpec();
+ if (groupSpec == null) {
+
+ String fileName = "map-" + getTaskId();
+
+ SplitSpec splitSpec = (SplitSpec) ObjectSerializer.deserialize(job.get("pig.splitSpec", ""));
+ if(splitSpec != null) splitSpec.instantiateFunc(pigContext);
+
+ if (splitSpec == null){
+ pigWriter = (PigRecordWriter) job.getOutputFormat().getRecordWriter(FileSystem.get(job), job, fileName,
+ reporter);
+ oc = new OutputCollector() {
+ public void collect(WritableComparable key, Writable value) throws IOException {
+ pigWriter.write(key, value);
+ }
+ };
+ }else{
+ oc = getSplitCollector(splitSpec);
+ }
+ evalPipe = evalSpec.setupPipe(new MapDataOutputCollector());
+ } else {
+ group = groupSpec;
+ DataCollector groupInput = group.setupPipe(new MapDataOutputCollector());
+ evalPipe = evalSpec.setupPipe(groupInput);
+ }
+
+ }
+
+ private OutputCollector getSplitCollector(final SplitSpec splitSpec) throws IOException{
+
+ PigOutputFormat outputFormat = (PigOutputFormat)job.getOutputFormat();
+
+ for (String name: splitSpec.tempFiles){
+ sideFileWriters.add( outputFormat.getRecordWriter(FileSystem.get(job), job, new Path(name), "split-" + getTaskId(), reporter));
+ }
+ return new OutputCollector(){
+ public void collect(WritableComparable key, Writable value) throws IOException {
+ Tuple t = (Tuple) value;
+ ArrayList<Cond> conditions = splitSpec.conditions;
+ for (int i=0; i< conditions.size(); i++){
+ Cond cond = conditions.get(i);
+ if (cond.eval(t)){
+ //System.out.println("Writing " + t + " to condition " + cond);
+ sideFileWriters.get(i).write(null, t);
+ }
+ }
+
+ }
+ };
+ }
+
+ private void setupReducePipe() throws IOException {
+ pigContext = (PigContext)ObjectSerializer.deserialize(job.get("pig.pigContext"));
+ EvalSpec evalSpec = (EvalSpec)ObjectSerializer.deserialize(job.get("pig.reduceFunc", ""));
+
+ if (evalSpec == null)
+ evalSpec = new StarSpec();
+ else
+ evalSpec.instantiateFunc(pigContext);
+
+ ArrayList<EvalSpec> groupSpecs = (ArrayList<EvalSpec>) ObjectSerializer.deserialize(job.get("pig.groupFuncs", ""));
+
+ isInner = new boolean[groupSpecs.size()];
+ for (int i = 0; i < groupSpecs.size(); i++) {
+ isInner[i] = groupSpecs.get(i).isInner();
+
+ }
+
+ SplitSpec splitSpec = (SplitSpec) ObjectSerializer.deserialize(job.get("pig.splitSpec", ""));
+ if (splitSpec != null){
+ splitSpec.instantiateFunc(pigContext);
+ oc = getSplitCollector(splitSpec);
+ }
+
+ evalPipe = evalSpec.setupPipe(new ReduceDataOutputCollector());
+
+ inputCount = ((ArrayList<FileSpec>)ObjectSerializer.deserialize(job.get("pig.inputs"))).size();
+
+ }
+
+ public void closeSideFiles(){
+ for (PigRecordWriter writer: sideFileWriters){
+ try{
+ writer.close(reporter);
+ }catch(IOException e){
+ e.printStackTrace();
+ }
+ }
+ }
+
+ class MapDataOutputCollector extends DataCollector {
+
+ public MapDataOutputCollector(){
+ super(null);
+ }
+
+ @Override
+ public void add(Datum d){
+ try{
+ if (group == null) {
+ oc.collect(null, (Tuple)d);
+ } else {
+ Datum[] groupAndTuple = LOCogroup.getGroupAndTuple(d);
+ // wrap group label in a tuple, so it becomes writable.
+ oc.collect(new Tuple(groupAndTuple[0]), new IndexedTuple((Tuple)groupAndTuple[1], index));
+ }
+ }catch(IOException e){
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void finish(){
+ closeSideFiles();
+ if (group != null)
+ group.finish();
+ }
+ }
+
+ class ReduceDataOutputCollector extends DataCollector {
+
+ public ReduceDataOutputCollector(){
+ super(null);
+ }
+
+ @Override
+ public void add(Datum d){
+ try{
+ //System.out.println("Adding " + d + " to reduce output");
+ oc.collect(null, (Tuple)d);
+ }catch(IOException e){
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ protected void finish(){
+ closeSideFiles();
+ }
+ }
+
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigOutputFormat.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,112 @@
+/*
+ * 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.backend.hadoop.executionengine.mapreduceExec;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.hadoop.fs.FSDataOutputStream;
+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.BZip2Constants;
+import org.apache.tools.bzip2r.CBZip2OutputStream;
+
+
+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/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/PigSplit.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,170 @@
+/*
+ * 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.backend.hadoop.executionengine.mapreduceExec;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
+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.eval.EvalSpec;
+import org.apache.pig.impl.util.ObjectSerializer;
+
+
+public class PigSplit implements InputSplit {
+ Path file;
+ long start;
+ long length;
+ EvalSpec groupbySpec;
+ EvalSpec evalSpec;
+ int index;
+ String parser;
+ FileSystem fs;
+ PigContext pigContext;
+
+ public PigSplit() {}
+
+ public PigSplit(PigContext pigContext, FileSystem fs, Path path, String parser, EvalSpec groupbySpec, EvalSpec evalSpec, int index, long start, long length) {
+ this.fs = fs;
+ this.file = path;
+ this.start = start;
+ this.length = length;
+ this.groupbySpec = groupbySpec;
+ this.evalSpec = evalSpec;
+ this.index = index;
+ this.parser = parser;
+ this.pigContext = pigContext;
+ }
+
+ public String getParser() {
+ return parser;
+ }
+
+ public long getStart() {
+ return start;
+ }
+
+ public long getLength() {
+ return length;
+ }
+ public EvalSpec getGroupbySpec() {
+ return groupbySpec;
+ }
+
+ public EvalSpec getEvalSpec() {
+ return evalSpec;
+ }
+
+ public Path getPath() {
+ return file;
+ }
+
+ public int getIndex() {
+ return index;
+ }
+
+ 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);
+ groupbySpec = (EvalSpec)readObject(is);
+ if(groupbySpec != null ) groupbySpec.instantiateFunc(pigContext);
+ evalSpec = (EvalSpec) readObject(is);
+ if(evalSpec != null) evalSpec.instantiateFunc(pigContext);
+ index = is.readInt();
+ parser = is.readUTF();
+ }
+
+ public void write(DataOutput os) throws IOException {
+ os.writeUTF(file.toString());
+ os.writeLong(start);
+ os.writeLong(length);
+ writeObject(pigContext, os);
+ writeObject(groupbySpec, os);
+ writeObject(evalSpec, os);
+ os.writeInt(index);
+ 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/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapreduceExec/SortPartitioner.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,80 @@
+/*
+ * 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.backend.hadoop.executionengine.mapreduceExec;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Partitioner;
+import org.apache.pig.builtin.BinStorage;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.io.BufferedPositionedInputStream;
+import org.apache.pig.impl.io.FileLocalizer;
+
+
+public class SortPartitioner implements Partitioner {
+ Tuple[] quantiles;
+ WritableComparator comparator;
+
+ public int getPartition(WritableComparable key, Writable value,
+ int numPartitions) {
+ try{
+ Tuple keyTuple = (Tuple)key;
+ int index = Arrays.binarySearch(quantiles, keyTuple.getTupleField(0), comparator);
+ if (index < 0)
+ index = -index-1;
+ return Math.min(index, numPartitions - 1);
+ }catch(IOException e){
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void configure(JobConf job) {
+ String quantilesFile = job.get("pig.quantilesFile", "");
+ if (quantilesFile.length() == 0)
+ throw new RuntimeException("Sort paritioner used but no quantiles found");
+
+ try{
+ InputStream is = FileLocalizer.openDFSFile(quantilesFile,job);
+ BinStorage loader = new BinStorage();
+ loader.bindTo(quantilesFile, new BufferedPositionedInputStream(is), 0, Long.MAX_VALUE);
+
+ Tuple t;
+ ArrayList<Tuple> quantiles = new ArrayList<Tuple>();
+
+ while(true){
+ t = loader.getNext();
+ if (t==null)
+ break;
+ quantiles.add(t);
+ }
+ this.quantiles = quantiles.toArray(new Tuple[0]);
+ }catch (IOException e){
+ throw new RuntimeException(e);
+ }
+
+ comparator = job.getOutputKeyComparator();
+ }
+
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDataStorage.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,149 @@
+package org.apache.pig.backend.local.datastorage;
+
+import java.io.IOException;
+import java.io.File;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.pig.backend.datastorage.*;
+
+public class LocalDataStorage implements DataStorage {
+
+ protected File workingDir;
+
+ public LocalDataStorage() {
+ workingDir = new File(System.getProperty("user.dir"));
+ }
+
+ @Override
+ public void init() {
+ ;
+ }
+
+ @Override
+ public void close() throws IOException {
+ ;
+ }
+
+ @Override
+ public Properties getConfiguration() {
+ Properties config = new Properties();
+
+ config.put(DEFAULT_REPLICATION_FACTOR_KEY, (new Integer(1)).toString());
+
+ return config;
+ }
+
+ @Override
+ public void updateConfiguration(Properties newConfiguration)
+ throws DataStorageException {
+ ;
+ }
+
+ @Override
+ public Map<String, Object> getStatistics() throws IOException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public LocalPath asElement(String name)
+ throws DataStorageException {
+ if (this.isContainer(name)) {
+ return new LocalDir(this, name);
+ }
+ else {
+ return new LocalFile(this, name);
+ }
+ }
+
+ @Override
+ public LocalPath asElement(ElementDescriptor element)
+ throws DataStorageException {
+ return asElement(element.toString());
+ }
+
+ @Override
+ public LocalPath asElement(String parent,
+ String child)
+ throws DataStorageException {
+ return asElement((new File(parent, child).toString()));
+ }
+
+ @Override
+ public LocalPath asElement(ContainerDescriptor parent,
+ String child)
+ throws DataStorageException {
+ return asElement(parent.toString(), child);
+ }
+
+ @Override
+ public LocalPath asElement(ContainerDescriptor parent,
+ ElementDescriptor child)
+ throws DataStorageException {
+ return asElement(parent.toString(), child.toString());
+ }
+
+ @Override
+ public LocalDir asContainer(String name)
+ throws DataStorageException {
+ return new LocalDir(this, name);
+ }
+
+ @Override
+ public LocalDir asContainer(ContainerDescriptor container)
+ throws DataStorageException {
+ return new LocalDir(this, container.toString());
+ }
+
+ @Override
+ public LocalDir asContainer(String parent,
+ String child)
+ throws DataStorageException {
+ return new LocalDir(this, parent, child);
+ }
+
+ @Override
+ public LocalDir asContainer(ContainerDescriptor parent,
+ String child)
+ throws DataStorageException {
+ return new LocalDir(this, parent.toString(), child);
+ }
+
+ @Override
+ public LocalDir asContainer(ContainerDescriptor parent,
+ ContainerDescriptor child)
+ throws DataStorageException {
+ return new LocalDir(this, parent.toString(), child.toString());
+ }
+
+ @Override
+ public boolean isContainer(String name) throws DataStorageException {
+ boolean isContainer = false;
+ File file = new File(name);
+
+ if (file.exists() && file.isDirectory()) {
+ isContainer = true;
+ }
+
+ return isContainer;
+ }
+
+ @Override
+ public void setActiveContainer(ContainerDescriptor container) {
+ this.workingDir = new File(container.toString());
+ }
+
+ @Override
+ public ContainerDescriptor getActiveContainer() {
+ return new LocalDir(this, this.workingDir.getPath());
+ }
+
+ @Override
+ public LocalPath[] asCollection(String pattern) throws DataStorageException {
+ throw new UnsupportedOperationException();
+ }
+
+ public File getWorkingDir() {
+ return this.workingDir;
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalDir.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,173 @@
+package org.apache.pig.backend.local.datastorage;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.LinkedList;
+import java.util.Properties;
+import java.util.Iterator;
+
+import org.apache.pig.backend.datastorage.ElementDescriptor;
+import org.apache.pig.backend.datastorage.ContainerDescriptor;
+import org.apache.pig.backend.datastorage.DataStorageException;
+import org.apache.pig.backend.datastorage.ImmutableOutputStream;
+import org.apache.pig.backend.datastorage.SeekableInputStream;
+
+public class LocalDir extends LocalPath
+ implements ContainerDescriptor {
+
+ public LocalDir(LocalDataStorage fs, String path) {
+ super(fs, path);
+ }
+
+ public LocalDir(LocalDataStorage fs, File path) {
+ super(fs, path);
+ }
+
+ public LocalDir(LocalDataStorage fs, String parent, String child) {
+ super(fs, parent, child);
+ }
+
+ public LocalDir(LocalDataStorage fs, File parent, File child) {
+ super(fs,
+ parent.getPath(),
+ child.getPath());
+ }
+
+ public LocalDir(LocalDataStorage fs, File parent, String child) {
+ this(fs, parent.getPath(), child);
+ }
+
+ public LocalDir(LocalDataStorage fs, String parent, File child) {
+ this(fs, parent, child.getPath());
+ }
+
+ @Override
+ public OutputStream create(Properties configuration)
+ throws IOException {
+ if (! getCurPath().mkdirs()) {
+ throw new IOException("Unable to create dirs for: " + this.path);
+ }
+
+ return new ImmutableOutputStream(path.toString());
+ }
+
+ @Override
+ public void copy(ElementDescriptor dstName,
+ Properties dstConfiguration,
+ boolean removeSrc)
+ throws IOException {
+ try {
+ if (! dstName.getDataStorage().isContainer(dstName.toString())) {
+ dstName = dstName.getDataStorage().asContainer(dstName.toString());
+ }
+ }
+ catch (DataStorageException e) {
+ throw new IOException("Failed to get container for " + dstName.toString(), e);
+ }
+
+ copy((ContainerDescriptor) dstName,
+ dstConfiguration,
+ removeSrc);
+ }
+
+
+ public void copy(ContainerDescriptor dstName,
+ Properties dstConfiguration,
+ boolean removeSrc)
+ throws IOException {
+ if (dstName == null) {
+ return;
+ }
+
+ if (!exists()) {
+ throw new IOException("Source does not exist " +
+ this);
+ }
+
+ if (dstName.exists()) {
+ throw new IOException("Destination already exists " +
+ dstName);
+ }
+
+ dstName.create();
+
+ Iterator<ElementDescriptor> elems = iterator();
+
+ try {
+ while (elems.hasNext()) {
+ ElementDescriptor curElem = elems.next();
+
+ if (curElem instanceof ContainerDescriptor) {
+ ContainerDescriptor dst =
+ dstName.getDataStorage().asContainer(dstName,
+ ((LocalPath)curElem).getPath().getName());
+
+ curElem.copy(dst, dstConfiguration, removeSrc);
+
+ if (removeSrc) {
+ curElem.delete();
+ }
+ }
+ else {
+ ElementDescriptor dst =
+ dstName.getDataStorage().asElement(dstName,
+ ((LocalPath)curElem).getPath().getName());
+
+ curElem.copy(dst, dstConfiguration, removeSrc);
+ }
+ }
+ }
+ catch (DataStorageException e) {
+ throw new IOException("Failed to copy " + this + " to " + dstName, e);
+ }
+
+ if (removeSrc) {
+ delete();
+ }
+ }
+
+ @Override
+ public InputStream open(Properties configuration) throws IOException {
+ return open();
+ }
+
+ @Override
+ public InputStream open() throws IOException {
+ throw new IOException("Cannot open dir " + path);
+ }
+
+ @Override
+ public SeekableInputStream sopen(Properties configuration) throws IOException {
+ return sopen();
+ }
+
+ @Override
+ public SeekableInputStream sopen() throws IOException {
+ throw new IOException("Cannot sopen dir " + path);
+ }
+
+ @Override
+ public Iterator<ElementDescriptor> iterator() {
+ LinkedList<ElementDescriptor> elements =
+ new LinkedList<ElementDescriptor>();
+
+ try {
+ File[] files = getCurPath().listFiles();
+
+ for (File f : files) {
+ if (f.isFile()) {
+ elements.add(fs.asElement(f.getPath()));
+ }
+ else {
+ elements.add(fs.asContainer(f.getPath()));
+ }
+ }
+ }
+ catch (DataStorageException e) {
+ }
+
+ return elements.iterator();
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalFile.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,125 @@
+
+package org.apache.pig.backend.local.datastorage;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.FileNotFoundException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.FileOutputStream;
+import java.io.FileInputStream;
+import java.util.Properties;
+
+import org.apache.pig.backend.datastorage.*;
+
+public class LocalFile extends LocalPath {
+
+ public LocalFile(LocalDataStorage fs, String path) {
+ super(fs, path);
+ }
+
+ public LocalFile(LocalDataStorage fs, File path) {
+ super(fs, path);
+ }
+
+ public LocalFile(LocalDataStorage fs, String parent, String child) {
+ super(fs, parent, child);
+ }
+
+ public LocalFile(LocalDataStorage fs, File parent, File child) {
+ super(fs,
+ parent.getPath(),
+ child.getPath());
+ }
+
+ public LocalFile(LocalDataStorage fs, File parent, String child) {
+ this(fs, parent.getPath(), child);
+ }
+
+ public LocalFile(LocalDataStorage fs, String parent, File child) {
+ this(fs, parent, child.getPath());
+ }
+
+ @Override
+ public OutputStream create(Properties configuration)
+ throws IOException {
+ if (! getCurPath().createNewFile()) {
+ throw new IOException("Failed to create file " + this.path);
+ }
+
+ return new FileOutputStream(getCurPath());
+ }
+
+ @Override
+ public void copy(ElementDescriptor dstName,
+ Properties dstConfiguration,
+ boolean removeSrc)
+ throws IOException {
+ if (dstName == null) {
+ return;
+ }
+
+ if (!exists()) {
+ throw new IOException("Source does not exist " +
+ this);
+ }
+
+ if (dstName.exists()) {
+ if (dstName instanceof ContainerDescriptor) {
+ try {
+ dstName = dstName.getDataStorage().
+ asElement((ContainerDescriptor) dstName,
+ path.getName());
+ }
+ catch (DataStorageException e) {
+ throw new IOException("Unable to generate element name (src: " +
+ this + ", dst: " + dstName + ")",
+ e);
+ }
+ }
+ }
+
+ InputStream in = null;
+ OutputStream out = null;
+
+ in = this.open();
+ out = dstName.create(dstConfiguration);
+
+ byte[] data = new byte[4 * 1024];
+ int bc;
+ while((bc = in.read(data)) != -1) {
+ out.write(data, 0, bc);
+ }
+
+ out.close();
+
+ if (removeSrc) {
+ delete();
+ }
+ }
+
+ @Override
+ public InputStream open (Properties configuration) throws IOException {
+ return open();
+ }
+
+ @Override
+ public InputStream open () throws IOException {
+ return new FileInputStream(this.path);
+ }
+
+ @Override
+ public SeekableInputStream sopen(Properties configuration) throws IOException {
+ return sopen();
+ }
+
+ @Override
+ public SeekableInputStream sopen() throws IOException {
+ try {
+ return new LocalSeekableInputStream(this.path);
+ }
+ catch (FileNotFoundException e) {
+ throw new IOException("Unable to find " + this.path, e);
+ }
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalPath.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,155 @@
+package org.apache.pig.backend.local.datastorage;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.File;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.Properties;
+
+import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.datastorage.ElementDescriptor;
+import org.apache.pig.backend.datastorage.SeekableInputStream;
+
+public abstract class LocalPath implements ElementDescriptor {
+
+ protected DataStorage fs;
+ protected File path;
+
+ protected File getCurPath() {
+ File path;
+
+ if (this.path.isAbsolute()) {
+ path = this.path;
+ }
+ else {
+ path = new File(fs.getActiveContainer().toString(),
+ this.path.getPath());
+ }
+
+ return path;
+ }
+
+ public LocalPath(LocalDataStorage fs, String path) {
+ this.fs = fs;
+ this.path = new File(path);
+ }
+
+ public LocalPath(LocalDataStorage fs, File path) {
+ this.fs = fs;
+ this.path = new File(path.getPath());
+ }
+
+ public LocalPath(LocalDataStorage fs, String parent, String child) {
+ this.fs = fs;
+ this.path = new File(parent, child);
+ }
+
+ public LocalPath(LocalDataStorage fs, File parent, File child) {
+ this.fs = fs;
+ this.path = new File(parent.getPath(),
+ child.getPath());
+ }
+
+ public LocalPath(LocalDataStorage fs, File parent, String child) {
+ this(fs, parent.getPath(), child);
+ }
+
+ public LocalPath(LocalDataStorage fs, String parent, File child) {
+ this(fs, parent, child.getPath());
+ }
+
+ @Override
+ public DataStorage getDataStorage() {
+ return fs;
+ }
+
+ public File getPath() {
+ return this.path;
+ }
+
+ @Override
+ public abstract OutputStream create(Properties configuration)
+ throws IOException;
+
+ @Override
+ public OutputStream create()
+ throws IOException {
+ return create(null);
+ }
+
+ @Override
+ public abstract void copy(ElementDescriptor dstName,
+ Properties dstConfiguration,
+ boolean removeSrc)
+ throws IOException;
+
+ @Override
+ public void copy(ElementDescriptor dstName,
+ boolean removeSrc) throws IOException {
+ copy(dstName, null, removeSrc);
+ }
+
+ @Override
+ public abstract InputStream open() throws IOException;
+
+ @Override
+ public abstract SeekableInputStream sopen() throws IOException;
+
+ @Override
+ public boolean exists() throws IOException {
+ return getCurPath().exists();
+ }
+
+ @Override
+ public void rename(ElementDescriptor newName)
+ throws IOException {
+ if (! this.path.renameTo(((LocalPath)newName).path)) {
+ throw new IOException("Unalbe to rename " + this.path +
+ "to " + ((LocalPath)newName).path);
+ }
+ }
+
+ @Override
+ public void delete() throws IOException {
+ getCurPath().delete();
+ }
+
+ @Override
+ public Properties getConfiguration() throws IOException {
+ Properties props = new Properties();
+
+ props.put(BLOCK_REPLICATION_KEY, "1");
+
+ return props;
+ }
+
+ @Override
+ public void updateConfiguration(Properties newConfig)
+ throws IOException {
+ ;
+ }
+
+ @Override
+ public Map<String, Object> getStatistics() throws IOException {
+ Map<String, Object> stats = new HashMap<String, Object>();
+
+ long size = this.path.length();
+ stats.put(LENGTH_KEY , (new Long(size)).toString());
+
+ long lastModified = this.path.lastModified();
+ stats.put(MODIFICATION_TIME_KEY, (new Long(lastModified)).toString());
+
+ return stats;
+ }
+
+ @Override
+ public int compareTo(ElementDescriptor other) {
+ return this.path.compareTo(((LocalPath)other).path);
+ }
+
+ public String toString() {
+ return this.path.toString();
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalSeekableInputStream.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalSeekableInputStream.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalSeekableInputStream.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/datastorage/LocalSeekableInputStream.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,107 @@
+package org.apache.pig.backend.local.datastorage;
+
+import java.io.RandomAccessFile;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import org.apache.pig.backend.datastorage.*;
+
+public class LocalSeekableInputStream extends SeekableInputStream {
+
+ protected RandomAccessFile file;
+ protected long curMark;
+
+ public LocalSeekableInputStream(File file) throws FileNotFoundException {
+ this.file = new RandomAccessFile(file, "r");
+ this.curMark = 0;
+ }
+
+ @Override
+ public void seek(long offset, FLAGS whence) throws IOException {
+ long targetPos;
+
+ switch (whence) {
+ case SEEK_SET: {
+ targetPos = offset;
+ break;
+ }
+ case SEEK_CUR: {
+ targetPos = this.file.getFilePointer() + offset;
+ break;
+ }
+ case SEEK_END: {
+ targetPos = this.file.length() + offset;
+ break;
+ }
+ default: {
+ throw new IOException("Invalid seek option: " + whence);
+ }
+ }
+
+ this.file.seek(targetPos);
+ }
+
+ @Override
+ public long tell() throws IOException {
+ return this.file.getFilePointer();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return this.file.read();
+ }
+
+ @Override
+ public int read(byte[] b) throws IOException {
+ return this.file.read(b);
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len ) throws IOException {
+ return this.file.read(b, off, len);
+ }
+
+ @Override
+ public int available() throws IOException {
+ throw new IOException("No information on available bytes");
+ }
+
+ @Override
+ public long skip(long n) throws IOException {
+ long skipped = 0;
+
+ if (n > 0) {
+ skipped = this.file.length() - tell();
+
+ seek(n, FLAGS.SEEK_CUR);
+ }
+
+ return skipped;
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.file.close();
+ }
+
+ @Override
+ public void mark(int readlimit) {
+ try {
+ this.curMark = tell();
+ }
+ catch (IOException e) {
+ ;
+ }
+ }
+
+ @Override
+ public void reset() throws IOException {
+ seek(this.curMark, FLAGS.SEEK_SET);
+ }
+
+ @Override
+ public boolean markSupported() {
+ return true;
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,290 @@
+package org.apache.pig.backend.local.executionengine;
+
+import java.util.Collection;
+import java.util.Properties;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.HashSet;
+
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.backend.datastorage.DataStorage;
+import org.apache.pig.backend.executionengine.ExecutionEngine;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.executionengine.ExecJob;
+import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
+import org.apache.pig.backend.executionengine.ExecLogicalPlan;
+import org.apache.pig.backend.executionengine.ExecPhysicalOperator;
+import org.apache.pig.backend.executionengine.ExecScopedLogicalOperator;
+import org.apache.pig.backend.executionengine.ExecPhysicalPlan;
+import org.apache.pig.impl.logicalLayer.*;
+import org.apache.pig.impl.physicalLayer.PhysicalOperator;
+import org.apache.pig.impl.logicalLayer.parser.NodeIdGenerator;
+import org.apache.pig.impl.eval.cond.Cond;
+import org.apache.pig.impl.io.FileSpec;
+import java.util.Iterator;
+
+
+public class LocalExecutionEngine implements ExecutionEngine {
+
+ protected PigContext pigContext;
+ protected DataStorage ds;
+ protected NodeIdGenerator nodeIdGenerator;
+
+ // key: the operator key from the logical plan that originated the physical plan
+ // val: the operator key for the root of the phyisical plan
+ protected Map<OperatorKey, OperatorKey> logicalToPhysicalKeys;
+
+ protected Map<OperatorKey, ExecPhysicalOperator> physicalOpTable;
+
+ // map from LOGICAL key to into about the execution
+ protected Map<OperatorKey, LocalResult> materializedResults;
+
+ public LocalExecutionEngine(PigContext pigContext) {
+ this.pigContext = pigContext;
+ this.ds = pigContext.getLfs();
+ this.nodeIdGenerator = NodeIdGenerator.getGenerator();
+ this.logicalToPhysicalKeys = new HashMap<OperatorKey, OperatorKey>();
+ this.physicalOpTable = new HashMap<OperatorKey, ExecPhysicalOperator>();
+ this.materializedResults = new HashMap<OperatorKey, LocalResult>();
+ }
+
+ @Override
+ public DataStorage getDataStorage() {
+ return this.ds;
+ }
+
+ @Override
+ public void init() throws ExecException {
+ ;
+ }
+
+ @Override
+ public void close() throws ExecException {
+ ;
+ }
+
+ @Override
+ public Properties getConfiguration() throws ExecException {
+ Properties conf = new Properties();
+ return conf;
+ }
+
+ @Override
+ public void updateConfiguration(Properties newConfiguration)
+ throws ExecException {
+ ;
+ }
+
+ @Override
+ public Map<String, Object> getStatistics() throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+
+ @Override
+ public LocalPhysicalPlan compile(ExecLogicalPlan plan,
+ Properties properties)
+ throws ExecException {
+ if (plan == null) {
+ throw new ExecException("No Plan to compile");
+ }
+
+ return compile(new ExecLogicalPlan[]{ plan } , properties);
+ }
+
+ @Override
+ public LocalPhysicalPlan compile(ExecLogicalPlan[] plans,
+ Properties properties)
+ throws ExecException {
+ if (plans == null) {
+ throw new ExecException("No Plans to compile");
+ }
+
+ OperatorKey physicalKey = null;
+ for (int i = 0; i < plans.length; ++i) {
+ ExecLogicalPlan curPlan = null;
+
+ curPlan = plans[ i ];
+
+ OperatorKey logicalKey = curPlan.getRoot();
+
+ physicalKey = logicalToPhysicalKeys.get(logicalKey);
+
+ if (physicalKey == null) {
+ physicalKey = doCompile(curPlan.getRoot(),
+ curPlan.getOpTable(),
+ properties);
+
+ logicalToPhysicalKeys.put(logicalKey, physicalKey);
+ }
+ }
+
+ return new LocalPhysicalPlan(physicalKey, physicalOpTable);
+ }
+
+ @Override
+ public LocalJob execute(ExecPhysicalPlan plan) throws ExecException {
+ DataBag results = BagFactory.getInstance().newDefaultBag();
+ try {
+ PhysicalOperator pp = (PhysicalOperator)physicalOpTable.get(plan.getRoot());
+
+ pp.open();
+
+ Tuple t;
+ while ((t = (Tuple) pp.getNext()) != null) {
+ results.add(t);
+ }
+
+ pp.close();
+ }
+ catch (IOException e) {
+ throw new ExecException(e);
+ }
+
+ return new LocalJob(results, JOB_STATUS.COMPLETED);
+ }
+
+ @Override
+ public LocalJob submit(ExecPhysicalPlan plan) throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Collection<ExecJob> runningJobs(Properties properties) throws ExecException {
+ return new HashSet<ExecJob>();
+ }
+
+ @Override
+ public Collection<String> activeScopes() throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void reclaimScope(String scope) throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ private OperatorKey doCompile(OperatorKey logicalKey,
+ Map<OperatorKey, LogicalOperator> logicalOpTable,
+ Properties properties)
+ throws ExecException {
+
+ LocalResult materializedResult = materializedResults.get(logicalKey);
+
+ if (materializedResult != null) {
+ ExecPhysicalOperator pp = new POLoad(logicalKey.getScope(),
+ nodeIdGenerator.getNextNodeId(logicalKey.getScope()),
+ physicalOpTable,
+ pigContext,
+ materializedResult.outFileSpec,
+ LogicalOperator.FIXED);
+
+ OperatorKey ppKey = new OperatorKey(pp.getScope(), pp.getId());
+
+ return ppKey;
+ }
+
+ OperatorKey physicalKey = new OperatorKey();
+
+ if (compileOperator(logicalKey, logicalOpTable, properties, physicalKey)) {
+ for (int i = 0; i < logicalOpTable.get(logicalKey).getInputs().size(); ++i) {
+ ((PhysicalOperator)physicalOpTable.get(physicalKey)).inputs[i] =
+ doCompile(logicalOpTable.get(logicalKey).getInputs().get(i), logicalOpTable, properties);
+ }
+ }
+
+ return physicalKey;
+ }
+
+ private boolean compileOperator(OperatorKey logicalKey,
+ Map<OperatorKey, LogicalOperator> logicalOpTable,
+ Properties properties,
+ OperatorKey physicalKey)
+ throws ExecException {
+ ExecPhysicalOperator pp;
+ LogicalOperator lo = logicalOpTable.get(logicalKey);
+ String scope = lo.getScope();
+ boolean compileInputs = true;
+
+ if (lo instanceof LOEval) {
+
+ pp = new POEval(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ ((LOEval) lo).getSpec(),
+ lo.getOutputType());
+ }
+ else if (lo instanceof LOCogroup) {
+ pp = new POCogroup(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ ((LOCogroup) lo).getSpecs(),
+ lo.getOutputType());
+ }
+ else if (lo instanceof LOLoad) {
+ pp = new POLoad(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ pigContext,
+ ((LOLoad)lo).getInputFileSpec(),
+ lo.getOutputType());
+ }
+ else if (lo instanceof LOSplitOutput) {
+ LOSplitOutput loso = (LOSplitOutput)lo;
+ LOSplit los = ((LOSplit)(logicalOpTable.get(loso.getInputs().get(0))));
+
+ pp = new POSplit(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ doCompile(los.getInputs().get(0),
+ logicalOpTable,
+ properties),
+ los.getConditions(),
+ loso.getReadFrom(),
+ lo.getOutputType());
+
+ compileInputs = false;
+ }
+ else if (lo instanceof LOStore) {
+ pp = new POStore(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ lo.getInputs().get(0),
+ materializedResults,
+ ((LOStore)lo).getOutputFileSpec(),
+ ((LOStore)lo).isAppend(),
+ pigContext);
+ }
+ else if (lo instanceof LOUnion) {
+ pp = new POUnion(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ ((LOUnion)lo).getInputs().size(),
+ lo.getOutputType());
+ }
+ else if (lo instanceof LOSort) {
+ pp = new POSort(scope,
+ nodeIdGenerator.getNextNodeId(scope),
+ physicalOpTable,
+ ((LOSort)lo).getSortSpec(),
+ lo.getOutputType());
+ }
+ else {
+ throw new ExecException("Internal error: Unknown logical operator.");
+ }
+
+ physicalKey.scope = pp.getScope();
+ physicalKey.id = pp.getId();
+
+ logicalToPhysicalKeys.put(logicalKey, physicalKey);
+
+ return compileInputs;
+ }
+}
+
+
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalJob.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,73 @@
+package org.apache.pig.backend.local.executionengine;
+
+import java.io.OutputStream;
+import java.util.Iterator;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.executionengine.ExecJob;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.DataBag;
+
+public class LocalJob implements ExecJob {
+
+ protected DataBag results;
+ protected JOB_STATUS status;
+
+ public LocalJob(DataBag results, JOB_STATUS status) {
+ this.results = results;
+ this.status = status;
+ }
+
+ @Override
+ public JOB_STATUS getStatus() {
+ return status;
+ }
+
+ @Override
+ public boolean hasCompleted() throws ExecException {
+ return true;
+ }
+
+ @Override
+ public Iterator<Tuple> getResults() throws ExecException {
+ return this.results.content();
+ }
+
+ @Override
+ public Properties getContiguration() {
+ Properties props = new Properties();
+ return props;
+ }
+
+ @Override
+ public Map<String, Object> getStatistics() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void completionNotification(Object cookie) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void kill() throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void getLogs(OutputStream log) throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void getSTDOut(OutputStream out) throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void getSTDError(OutputStream error) throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,53 @@
+package org.apache.pig.backend.local.executionengine;
+
+import java.io.OutputStream;
+import java.io.PrintStream;
+import java.util.Properties;
+import java.util.Map;
+
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.backend.executionengine.ExecPhysicalOperator;
+import org.apache.pig.backend.executionengine.ExecPhysicalPlan;
+import org.apache.pig.impl.physicalLayer.PhysicalOperator;
+import org.apache.pig.impl.logicalLayer.OperatorKey;
+
+public class LocalPhysicalPlan implements ExecPhysicalPlan {
+ private static final long serialVersionUID = 1;
+
+ protected OperatorKey root;
+ protected Map<OperatorKey, ExecPhysicalOperator> opTable;
+
+ LocalPhysicalPlan(OperatorKey root,
+ Map<OperatorKey, ExecPhysicalOperator> opTable) {
+ this.root = root;
+ this.opTable = opTable;
+ }
+
+ @Override
+ public Properties getConfiguration() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void updateConfiguration(Properties configuration)
+ throws ExecException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void explain(OutputStream out) {
+ POVisitor lprinter = new POVisitor(new PrintStream(out));
+
+ ((PhysicalOperator)opTable.get(root)).visit(lprinter, "");
+ }
+
+ @Override
+ public Map<OperatorKey, ExecPhysicalOperator> getOpTable() {
+ return opTable;
+ }
+
+ @Override
+ public OperatorKey getRoot() {
+ return root;
+ }
+}
Added: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalResult.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalResult.java?rev=617338&view=auto
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalResult.java (added)
+++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalResult.java Thu Jan 31 19:11:28 2008
@@ -0,0 +1,21 @@
+package org.apache.pig.backend.local.executionengine;
+
+import java.util.Iterator;
+
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.io.FileSpec;
+
+public class LocalResult {
+ // could record other information like when file was first
+ // created, how many times it's been re-used (in this context we
+ // are inside the same scope)
+ // ...
+ //
+ public FileSpec outFileSpec;
+
+ public LocalResult(FileSpec outFileSpec) {
+ this.outFileSpec = outFileSpec;
+ }
+}
+