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/07 00:09:55 UTC
svn commit: r619213 [4/5] - in /incubator/pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/executionengine/
src/org/apache/pig/backend/hadoop/datastorage/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/...
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/GenerateSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/GenerateSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/GenerateSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/GenerateSpec.java Wed Feb 6 15:09:43 2008
@@ -34,123 +34,123 @@
public class GenerateSpec extends EvalSpec {
- private static final long serialVersionUID = 1L;
-
- protected List<EvalSpec> specs = new ArrayList<EvalSpec>();
+ private static final long serialVersionUID = 1L;
+
+ protected List<EvalSpec> specs = new ArrayList<EvalSpec>();
- protected int driver;
+ protected int driver;
public GenerateSpec(List<EvalSpec> specs){
- this.specs = specs;
- selectDriver();
+ this.specs = specs;
+ selectDriver();
}
public GenerateSpec(EvalSpec col){
- specs.add(col);
+ specs.add(col);
}
public GenerateSpec getGroupBySpec(){
-
- //Adding a new start spec to get the group by spec. The new star spec that
- //we are adding should not contain a schema since that can cause conflicts.
-
- StarSpec ss = new StarSpec();
- ss.setSchema(new TupleSchema());
- specs.add(ss);
- return new GenerateSpec(specs);
+
+ //Adding a new start spec to get the group by spec. The new star spec that
+ //we are adding should not contain a schema since that can cause conflicts.
+
+ StarSpec ss = new StarSpec();
+ ss.setSchema(new TupleSchema());
+ specs.add(ss);
+ return new GenerateSpec(specs);
}
@Override
protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
- return new DataCollector(endOfPipe){
- LinkedList<CrossProductItem> pendingCrossProducts = new LinkedList<CrossProductItem>();
-
- @Override
- public void add(Datum d) {
-
- if (checkDelimiter(d))
- throw new RuntimeException("Internal error: not expecting a delimiter tuple");
-
- // general case (use driver method):
- CrossProductItem cpi = new CrossProductItem(d, successor);
-
- pendingCrossProducts.addLast(cpi);
-
-
- //Since potentially can return without filling output, mark output as stale
- //the exec method of CrossProductItem will mark output as not stale
- successor.markStale(true);
- while (!pendingCrossProducts.isEmpty() && pendingCrossProducts.peek().isReady()){
- pendingCrossProducts.remove().exec();
- }
- }
-
- @Override
- protected void finish() {
- for (EvalSpec spec: specs){
- if (specs.get(driver)!=spec)
- spec.finish();
- }
-
- while (!pendingCrossProducts.isEmpty()){
- CrossProductItem cpi = pendingCrossProducts.remove();
- cpi.waitToBeReady();
- cpi.exec();
- }
-
- specs.get(driver).finish();
- }
-
- };
+ return new DataCollector(endOfPipe){
+ LinkedList<CrossProductItem> pendingCrossProducts = new LinkedList<CrossProductItem>();
+
+ @Override
+ public void add(Datum d) {
+
+ if (checkDelimiter(d))
+ throw new RuntimeException("Internal error: not expecting a delimiter tuple");
+
+ // general case (use driver method):
+ CrossProductItem cpi = new CrossProductItem(d, successor);
+
+ pendingCrossProducts.addLast(cpi);
+
+
+ //Since potentially can return without filling output, mark output as stale
+ //the exec method of CrossProductItem will mark output as not stale
+ successor.markStale(true);
+ while (!pendingCrossProducts.isEmpty() && pendingCrossProducts.peek().isReady()){
+ pendingCrossProducts.remove().exec();
+ }
+ }
+
+ @Override
+ protected void finish() {
+ for (EvalSpec spec: specs){
+ if (specs.get(driver)!=spec)
+ spec.finish();
+ }
+
+ while (!pendingCrossProducts.isEmpty()){
+ CrossProductItem cpi = pendingCrossProducts.remove();
+ cpi.waitToBeReady();
+ cpi.exec();
+ }
+
+ specs.get(driver).finish();
+ }
+
+ };
}
private class DatumBag extends DataCollector{
- DataBag bag;
- public DatumBag(){
- super(null);
- bag = BagFactory.getInstance().newDefaultBag();
- }
-
- @Override
- public void add(Datum d){
- bag.add(new Tuple(d));
- }
-
- public Iterator<Datum> content(){
- return new Iterator<Datum>(){
- Iterator<Tuple> iter;
- {
- iter = bag.iterator();
- }
- public boolean hasNext() {
- return iter.hasNext();
- }
- public Datum next() {
- try{
- return iter.next().getField(0);
- }catch(IOException e){
- throw new RuntimeException(e);
- }
- }
- public void remove() {
- throw new RuntimeException("Can't remove from read-only iterator");
- }
- };
- }
-
+ DataBag bag;
+ public DatumBag(){
+ super(null);
+ bag = BagFactory.getInstance().newDefaultBag();
+ }
+
+ @Override
+ public void add(Datum d){
+ bag.add(new Tuple(d));
+ }
+
+ public Iterator<Datum> content(){
+ return new Iterator<Datum>(){
+ Iterator<Tuple> iter;
+ {
+ iter = bag.iterator();
+ }
+ public boolean hasNext() {
+ return iter.hasNext();
+ }
+ public Datum next() {
+ try{
+ return iter.next().getField(0);
+ }catch(IOException e){
+ throw new RuntimeException(e);
+ }
+ }
+ public void remove() {
+ throw new RuntimeException("Can't remove from read-only iterator");
+ }
+ };
+ }
+
}
private class CrossProductItem extends DataCollector{
- DatumBag[] toBeCrossed;
- Datum cpiInput;
-
- public CrossProductItem(Datum driverInput, DataCollector successor){
- super(successor);
- this.cpiInput = driverInput;
-
- // materialize data for all to-be-crossed items
+ DatumBag[] toBeCrossed;
+ Datum cpiInput;
+
+ public CrossProductItem(Datum driverInput, DataCollector successor){
+ super(successor);
+ this.cpiInput = driverInput;
+
+ // materialize data for all to-be-crossed items
// (except driver, which is done in streaming fashion)
toBeCrossed = new DatumBag[specs.size()];
for (int i = 0; i < specs.size(); i++) {
@@ -160,22 +160,22 @@
specs.get(i).setupPipe(toBeCrossed[i]).add(cpiInput);
}
- }
-
- @Override
- public void add(Datum d){
- if (checkDelimiter(d))
- throw new RuntimeException("Internal error: not expecting a delimiter tuple");
- int numItems = specs.size();
-
+ }
+
+ @Override
+ public void add(Datum d){
+ if (checkDelimiter(d))
+ throw new RuntimeException("Internal error: not expecting a delimiter tuple");
+ int numItems = specs.size();
+
// create one iterator per to-be-crossed bag
Iterator<Datum>[] its = new Iterator[numItems];
for (int i = 0; i < numItems; i++) {
- if (i != driver){
- its[i] = toBeCrossed[i].content();
- if (!its[i].hasNext())
- return; // one of inputs is empty, so cross-prod yields empty result
- }
+ if (i != driver){
+ its[i] = toBeCrossed[i].content();
+ if (!its[i].hasNext())
+ return; // one of inputs is empty, so cross-prod yields empty result
+ }
}
Datum[] lastOutput = null;
@@ -188,7 +188,7 @@
if (i == driver)
outData[i] = d;
else
- outData[i] = its[i].next();
+ outData[i] = its[i].next();
}
} else {
boolean needToAdvance = true;
@@ -222,55 +222,55 @@
Tuple outTuple = new Tuple();
for (int i=0; i< numItems; i++){
- if (specs.get(i).isFlattened() && outData[i] instanceof Tuple){
- Tuple t = (Tuple)outData[i];
- try{
- for (int j=0; j < t.arity(); j++){
- outTuple.appendField(t.getField(j));
- }
- }catch (IOException e){
- throw new RuntimeException(e);
- }
- }else{
- outTuple.appendField(outData[i]);
- }
+ if (specs.get(i).isFlattened() && outData[i] instanceof Tuple){
+ Tuple t = (Tuple)outData[i];
+ try{
+ for (int j=0; j < t.arity(); j++){
+ outTuple.appendField(t.getField(j));
+ }
+ }catch (IOException e){
+ throw new RuntimeException(e);
+ }
+ }else{
+ outTuple.appendField(outData[i]);
+ }
}
successor.add(outTuple);
lastOutput = outData;
}
- }
+ }
- public boolean isReady(){
- for (int i=0; i<toBeCrossed.length; i++){
- if (i!=driver && toBeCrossed[i].isStale())
- return false;
- }
- return true;
- }
-
- public void waitToBeReady(){
- for (int i=0; i<toBeCrossed.length; i++){
- if (i!=driver){
- synchronized(toBeCrossed[i]){
- while (toBeCrossed[i].isStale()){
- try{
- toBeCrossed[i].wait();
- }catch (InterruptedException e){}
- }
- }
- }
- }
-
- }
-
- public void exec(){
+ public boolean isReady(){
+ for (int i=0; i<toBeCrossed.length; i++){
+ if (i!=driver && toBeCrossed[i].isStale())
+ return false;
+ }
+ return true;
+ }
+
+ public void waitToBeReady(){
+ for (int i=0; i<toBeCrossed.length; i++){
+ if (i!=driver){
+ synchronized(toBeCrossed[i]){
+ while (toBeCrossed[i].isStale()){
+ try{
+ toBeCrossed[i].wait();
+ }catch (InterruptedException e){}
+ }
+ }
+ }
+ }
+
+ }
+
+ public void exec(){
specs.get(driver).setupPipe(this).add(cpiInput);
//System.err.println(Thread.currentThread().getName() + ": Executing driver on " + cpiInput);
successor.markStale(false);
- }
-
+ }
+
}
/**
@@ -285,38 +285,38 @@
for (int i = 0; i < specs.size(); i++) {
EvalSpec spec = specs.get(i);
if (spec.isFlattened() || spec.isAsynchronous()){
- //This is just a heuristic that if its a flattened bag eval function, it should
- //be chosen as the driver
- if (spec instanceof CompositeEvalSpec)
- spec = ((CompositeEvalSpec)spec).getSpecs().get(0);
- if (spec instanceof FuncEvalSpec && ((FuncEvalSpec)spec).getFunc().getReturnType() == DataBag.class) { // trumps 'em all
+ //This is just a heuristic that if its a flattened bag eval function, it should
+ //be chosen as the driver
+ if (spec instanceof CompositeEvalSpec)
+ spec = ((CompositeEvalSpec)spec).getSpecs().get(0);
+ if (spec instanceof FuncEvalSpec && ((FuncEvalSpec)spec).getFunc().getReturnType() == DataBag.class) { // trumps 'em all
driver = i;
return;
}
- driver = i; // we'll use this as the driver, unless something better comes along
+ driver = i; // we'll use this as the driver, unless something better comes along
}
}
}
@Override
- public String toString() {
+ public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("GENERATE ");
sb.append("{");
boolean first = true;
for (EvalSpec spec: specs){
- if (!first)
- sb.append(",");
- else
- first = false;
- sb.append(spec);
+ if (!first)
+ sb.append(",");
+ else
+ first = false;
+ sb.append(spec);
}
sb.append("}");
return sb.toString();
}
@Override
- public List<String> getFuncs() {
+ public List<String> getFuncs() {
List<String> funcs = new ArrayList<String>();
for (EvalSpec spec: specs)
funcs.addAll(spec.getFuncs());
@@ -324,50 +324,50 @@
}
@Override
- protected TupleSchema mapInputSchema(Schema input) {
+ protected TupleSchema mapInputSchema(Schema input) {
TupleSchema output = new TupleSchema();
for (EvalSpec spec: specs) {
- Schema schema = spec.getOutputSchemaForPipe(input).copy();
-
- if (spec.isFlattened()){
- List<Schema> flattenedSchema = schema.flatten();
- if (flattenedSchema.size() == 0){
- output.add(new TupleSchema(),true);
- continue;
- }
- for (Schema flattenedItem: flattenedSchema){
- output.add(flattenedItem,true);
- }
- }else{
- output.add(schema,false);
- }
+ Schema schema = spec.getOutputSchemaForPipe(input).copy();
+
+ if (spec.isFlattened()){
+ List<Schema> flattenedSchema = schema.flatten();
+ if (flattenedSchema.size() == 0){
+ output.add(new TupleSchema(),true);
+ continue;
+ }
+ for (Schema flattenedItem: flattenedSchema){
+ output.add(flattenedItem,true);
+ }
+ }else{
+ output.add(schema,false);
+ }
}
return output;
}
@Override
public boolean isAsynchronous() {
- for (EvalSpec es: specs)
- if (es.isAsynchronous())
- return true;
- return false;
- }
-
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- for (EvalSpec es: specs)
- es.instantiateFunc(instantiaor);
- }
-
- public List<EvalSpec> getSpecs() {
- return specs;
- }
-
- @Override
- public void visit(EvalSpecVisitor v) {
- v.visitGenerate(this);
- }
+ for (EvalSpec es: specs)
+ if (es.isAsynchronous())
+ return true;
+ return false;
+ }
+
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ for (EvalSpec es: specs)
+ es.instantiateFunc(instantiaor);
+ }
+
+ public List<EvalSpec> getSpecs() {
+ return specs;
+ }
+
+ @Override
+ public void visit(EvalSpecVisitor v) {
+ v.visitGenerate(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/MapLookupSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/MapLookupSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/MapLookupSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/MapLookupSpec.java Wed Feb 6 15:09:43 2008
@@ -27,52 +27,52 @@
public class MapLookupSpec extends SimpleEvalSpec {
-
- /**
- *
- */
- private static final long serialVersionUID = 1L;
- protected String keyToLookup;
-
- public MapLookupSpec(String keyToLookup){
- this.keyToLookup = keyToLookup;
- }
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+ protected String keyToLookup;
+
+ public MapLookupSpec(String keyToLookup){
+ this.keyToLookup = keyToLookup;
+ }
- @Override
- protected Datum eval(Datum d) {
- if (!(d instanceof DataMap))
- throw new RuntimeException("Attempt to lookup on data of type " + d.getClass().getName());
- return ((DataMap)d).get(keyToLookup);
- }
-
- @Override
- public List<String> getFuncs() {
- return new ArrayList<String>();
- }
-
- @Override
- protected Schema mapInputSchema(Schema schema) {
- //TODO: until we have map schemas
- return new TupleSchema();
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("[");
- sb.append("#'");
- sb.append(keyToLookup);
- sb.append("'");
- sb.append("]");
- return sb.toString();
- }
+ @Override
+ protected Datum eval(Datum d) {
+ if (!(d instanceof DataMap))
+ throw new RuntimeException("Attempt to lookup on data of type " + d.getClass().getName());
+ return ((DataMap)d).get(keyToLookup);
+ }
+
+ @Override
+ public List<String> getFuncs() {
+ return new ArrayList<String>();
+ }
+
+ @Override
+ protected Schema mapInputSchema(Schema schema) {
+ //TODO: until we have map schemas
+ return new TupleSchema();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("[");
+ sb.append("#'");
+ sb.append(keyToLookup);
+ sb.append("'");
+ sb.append("]");
+ return sb.toString();
+ }
- @Override
- public void visit(EvalSpecVisitor v) {
- v.visitMapLookup(this);
- }
+ @Override
+ public void visit(EvalSpecVisitor v) {
+ v.visitMapLookup(this);
+ }
- public String key() { return keyToLookup; }
+ public String key() { return keyToLookup; }
-
+
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/ProjectSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/ProjectSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/ProjectSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/ProjectSpec.java Wed Feb 6 15:09:43 2008
@@ -28,93 +28,93 @@
public class ProjectSpec extends SimpleEvalSpec {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 1L;
- protected List<Integer> cols;
- protected boolean wrapInTuple;
-
-
- public List<Integer> getCols() {
- return cols;
- }
-
- public ProjectSpec(List<Integer> cols){
- this.cols = cols;
- }
-
- public ProjectSpec(int col){
- cols = new ArrayList<Integer>();
- cols.add(col);
- }
-
- @Override
- public List<String> getFuncs() {
- return new ArrayList<String>();
- }
-
- @Override
- protected Schema mapInputSchema(Schema schema) {
- if (!wrapInTuple && cols.size()==1){
- return maskNullSchema(schema.schemaFor(cols.get(0)));
- }else{
- TupleSchema output = new TupleSchema();
- for (int i: cols){
- output.add(maskNullSchema(schema.schemaFor(i)));
- }
- return output;
- }
- }
-
- private Schema maskNullSchema(Schema s){
- if (s == null)
- return new TupleSchema();
- else
- return s;
-
- }
-
- @Override
- protected Datum eval(Datum d){
- if (!(d instanceof Tuple)){
- throw new RuntimeException("Project operator expected a Tuple, found a " + d.getClass().getSimpleName());
- }
- Tuple t = (Tuple)d;
-
- try{
- if (!wrapInTuple && cols.size() == 1){
- return t.getField(cols.get(0));
- }else{
- Tuple out = new Tuple();
- for (int i: cols){
- out.appendField(t.getField(i));
- }
- return out;
- }
- }catch (IOException e){
- //TODO: Based on a strictness level, insert null values here
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("[");
- if (isFlattened)
- sb.append("FLATTEN ");
- sb.append("PROJECT ");
- boolean first = true;
- for (int i: cols){
- if (!first)
- sb.append(",");
- else
- first = false;
- sb.append("$");
- sb.append(i);
- }
- sb.append("]");
- return sb.toString();
- }
+ protected List<Integer> cols;
+ protected boolean wrapInTuple;
+
+
+ public List<Integer> getCols() {
+ return cols;
+ }
+
+ public ProjectSpec(List<Integer> cols){
+ this.cols = cols;
+ }
+
+ public ProjectSpec(int col){
+ cols = new ArrayList<Integer>();
+ cols.add(col);
+ }
+
+ @Override
+ public List<String> getFuncs() {
+ return new ArrayList<String>();
+ }
+
+ @Override
+ protected Schema mapInputSchema(Schema schema) {
+ if (!wrapInTuple && cols.size()==1){
+ return maskNullSchema(schema.schemaFor(cols.get(0)));
+ }else{
+ TupleSchema output = new TupleSchema();
+ for (int i: cols){
+ output.add(maskNullSchema(schema.schemaFor(i)));
+ }
+ return output;
+ }
+ }
+
+ private Schema maskNullSchema(Schema s){
+ if (s == null)
+ return new TupleSchema();
+ else
+ return s;
+
+ }
+
+ @Override
+ protected Datum eval(Datum d){
+ if (!(d instanceof Tuple)){
+ throw new RuntimeException("Project operator expected a Tuple, found a " + d.getClass().getSimpleName());
+ }
+ Tuple t = (Tuple)d;
+
+ try{
+ if (!wrapInTuple && cols.size() == 1){
+ return t.getField(cols.get(0));
+ }else{
+ Tuple out = new Tuple();
+ for (int i: cols){
+ out.appendField(t.getField(i));
+ }
+ return out;
+ }
+ }catch (IOException e){
+ //TODO: Based on a strictness level, insert null values here
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("[");
+ if (isFlattened)
+ sb.append("FLATTEN ");
+ sb.append("PROJECT ");
+ boolean first = true;
+ for (int i: cols){
+ if (!first)
+ sb.append(",");
+ else
+ first = false;
+ sb.append("$");
+ sb.append(i);
+ }
+ sb.append("]");
+ return sb.toString();
+ }
public int numCols() {
return cols.size();
@@ -125,21 +125,21 @@
throw new RuntimeException("Internal error: improper use of getColumn in " + ProjectSpec.class.getName());
else return cols.get(i);
}
-
- public int getCol(){
- if (cols.size()!=1)
- throw new RuntimeException("Internal error: improper use of getColumn in " + ProjectSpec.class.getName());
- return cols.get(0);
- }
-
- public void setWrapInTuple(boolean wrapInTuple) {
- this.wrapInTuple = wrapInTuple;
- }
+
+ public int getCol(){
+ if (cols.size()!=1)
+ throw new RuntimeException("Internal error: improper use of getColumn in " + ProjectSpec.class.getName());
+ return cols.get(0);
+ }
+
+ public void setWrapInTuple(boolean wrapInTuple) {
+ this.wrapInTuple = wrapInTuple;
+ }
- @Override
- public void visit(EvalSpecVisitor v) {
- v.visitProject(this);
- }
+ @Override
+ public void visit(EvalSpecVisitor v) {
+ v.visitProject(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/SimpleEvalSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/SimpleEvalSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/SimpleEvalSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/SimpleEvalSpec.java Wed Feb 6 15:09:43 2008
@@ -24,25 +24,25 @@
public abstract class SimpleEvalSpec extends EvalSpec {
- @Override
- protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
- return new DataCollector(endOfPipe){
- @Override
- public void add(Datum d) {
- if (checkDelimiter(d))
- addToSuccessor(d);
- else
- addToSuccessor(eval(d));
- }
-
- @Override
- protected boolean needFlatteningLocally() {
- return true;
- }
- };
- }
-
- protected abstract Datum eval(Datum d);
+ @Override
+ protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
+ return new DataCollector(endOfPipe){
+ @Override
+ public void add(Datum d) {
+ if (checkDelimiter(d))
+ addToSuccessor(d);
+ else
+ addToSuccessor(eval(d));
+ }
+
+ @Override
+ protected boolean needFlatteningLocally() {
+ return true;
+ }
+ };
+ }
+
+ protected abstract Datum eval(Datum d);
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/SortDistinctSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/SortDistinctSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/SortDistinctSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/SortDistinctSpec.java Wed Feb 6 15:09:43 2008
@@ -31,105 +31,105 @@
public class SortDistinctSpec extends EvalSpec {
- private static final long serialVersionUID = 1L;
- transient DataBag bag;
- protected EvalSpec sortSpec;
- protected boolean eliminateDuplicates;
-
-
- public SortDistinctSpec(boolean eliminateDuplicates, EvalSpec sortSpec){
- this.eliminateDuplicates = eliminateDuplicates;
- this.sortSpec = sortSpec;
- }
-
- @Override
- public List<String> getFuncs() {
- if (sortSpec!=null)
- return sortSpec.getFuncs();
- else
- return new ArrayList<String>();
- }
+ private static final long serialVersionUID = 1L;
+ transient DataBag bag;
+ protected EvalSpec sortSpec;
+ protected boolean eliminateDuplicates;
+
+
+ public SortDistinctSpec(boolean eliminateDuplicates, EvalSpec sortSpec){
+ this.eliminateDuplicates = eliminateDuplicates;
+ this.sortSpec = sortSpec;
+ }
+
+ @Override
+ public List<String> getFuncs() {
+ if (sortSpec!=null)
+ return sortSpec.getFuncs();
+ else
+ return new ArrayList<String>();
+ }
- @Override
- protected Schema mapInputSchema(Schema schema) {
- return schema;
- }
+ @Override
+ protected Schema mapInputSchema(Schema schema) {
+ return schema;
+ }
- @Override
- protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
- return new DataCollector(endOfPipe){
-
- @Override
- public void add(Datum d) {
- if (inTheMiddleOfBag){
- if (checkDelimiter(d)){
- addToSuccessor(bag);
- }else{
- if (d instanceof Tuple){
- bag.add((Tuple)d);
- }else{
- bag.add(new Tuple(d));
- }
- }
- }else{
- if (checkDelimiter(d)){
- //Bag must have started now
- if (eliminateDuplicates) {
- bag = BagFactory.getInstance().newDistinctBag();
- } else {
- bag = BagFactory.getInstance().newSortedBag(sortSpec);
- }
- }else{
- addToSuccessor(d);
- }
- }
- }
-
- @Override
- protected boolean needFlatteningLocally() {
- return true;
- }
-
-
- @Override
- protected void finish() {
-
- /*
- * To clear the temporary files if it was a big bag
- */
- if (bag!=null)
- bag.clear();
-
- }
- };
- }
+ @Override
+ protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
+ return new DataCollector(endOfPipe){
+
+ @Override
+ public void add(Datum d) {
+ if (inTheMiddleOfBag){
+ if (checkDelimiter(d)){
+ addToSuccessor(bag);
+ }else{
+ if (d instanceof Tuple){
+ bag.add((Tuple)d);
+ }else{
+ bag.add(new Tuple(d));
+ }
+ }
+ }else{
+ if (checkDelimiter(d)){
+ //Bag must have started now
+ if (eliminateDuplicates) {
+ bag = BagFactory.getInstance().newDistinctBag();
+ } else {
+ bag = BagFactory.getInstance().newSortedBag(sortSpec);
+ }
+ }else{
+ addToSuccessor(d);
+ }
+ }
+ }
+
+ @Override
+ protected boolean needFlatteningLocally() {
+ return true;
+ }
+
+
+ @Override
+ protected void finish() {
+
+ /*
+ * To clear the temporary files if it was a big bag
+ */
+ if (bag!=null)
+ bag.clear();
+
+ }
+ };
+ }
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("[");
- sb.append(eliminateDuplicates?"DISTINCT ":"SORT ");
- if (sortSpec!=null)
- sb.append(sortSpec.toString());
- sb.append("]");
- return sb.toString();
- }
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("[");
+ sb.append(eliminateDuplicates?"DISTINCT ":"SORT ");
+ if (sortSpec!=null)
+ sb.append(sortSpec.toString());
+ sb.append("]");
+ return sb.toString();
+ }
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- if (sortSpec!=null)
- sortSpec.instantiateFunc(instantiaor);
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ if (sortSpec!=null)
+ sortSpec.instantiateFunc(instantiaor);
+ }
- @Override
- public void visit(EvalSpecVisitor v) {
- v.visitSortDistinct(this);
- }
+ @Override
+ public void visit(EvalSpecVisitor v) {
+ v.visitSortDistinct(this);
+ }
- public EvalSpec getSortSpec() { return sortSpec; }
+ public EvalSpec getSortSpec() { return sortSpec; }
- public boolean distinct() { return eliminateDuplicates; }
+ public boolean distinct() { return eliminateDuplicates; }
-
+
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/StarSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/StarSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/StarSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/StarSpec.java Wed Feb 6 15:09:43 2008
@@ -25,32 +25,32 @@
final public class StarSpec extends EvalSpec {
- private static final long serialVersionUID = 1L;
-
- @Override
- public String toString() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public String toString() {
return "[*]";
}
@Override
- public List<String> getFuncs() {
+ public List<String> getFuncs() {
return new ArrayList<String>();
}
@Override
- protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
- return endOfPipe;
+ protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
+ return endOfPipe;
}
@Override
- protected Schema mapInputSchema(Schema input){
- return input;
+ protected Schema mapInputSchema(Schema input){
+ return input;
}
- @Override
- public void visit(EvalSpecVisitor v) {
- v.visitStar(this);
- }
+ @Override
+ public void visit(EvalSpecVisitor v) {
+ v.visitStar(this);
+ }
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/DataCollector.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/DataCollector.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/DataCollector.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/DataCollector.java Wed Feb 6 15:09:43 2008
@@ -31,103 +31,103 @@
* or a file.
*/
public abstract class DataCollector {
- Integer staleCount = 0;
- protected boolean inTheMiddleOfBag = false;
-
- protected DataCollector successor = null;
-
- public DataCollector(DataCollector successor){
- this.successor = successor;
- }
-
- /**
+ Integer staleCount = 0;
+ protected boolean inTheMiddleOfBag = false;
+
+ protected DataCollector successor = null;
+
+ public DataCollector(DataCollector successor){
+ this.successor = successor;
+ }
+
+ /**
* Add a tuple to the collector.
- */
- public abstract void add(Datum d);
-
- private boolean needsFlattening(){
- if (needFlatteningLocally() || (successor!=null && successor.needsFlattening()))
- return true;
- else
- return false;
- }
-
- /*
- * Whether this collector needs flattened bags to operate
- */
- protected boolean needFlatteningLocally(){
- return false;
- }
-
-
- protected boolean checkDelimiter(Datum d){
- if (d instanceof DataBag.BagDelimiterTuple){
- if (d instanceof DataBag.StartBag){
- if (inTheMiddleOfBag)
- throw new RuntimeException("Internal error: Found a flattened bag inside another");
- else
- inTheMiddleOfBag = true;
- }else{
- if (!(d instanceof DataBag.EndBag))
- throw new RuntimeException("Internal error: Unknown bag delimiter type");
- if (!inTheMiddleOfBag)
- throw new RuntimeException("Internal error: Improper nesting of bag delimiter tuples");
- inTheMiddleOfBag = false;
- }
- return true;
- }
- return false;
- }
+ */
+ public abstract void add(Datum d);
+
+ private boolean needsFlattening(){
+ if (needFlatteningLocally() || (successor!=null && successor.needsFlattening()))
+ return true;
+ else
+ return false;
+ }
+
+ /*
+ * Whether this collector needs flattened bags to operate
+ */
+ protected boolean needFlatteningLocally(){
+ return false;
+ }
+
+
+ protected boolean checkDelimiter(Datum d){
+ if (d instanceof DataBag.BagDelimiterTuple){
+ if (d instanceof DataBag.StartBag){
+ if (inTheMiddleOfBag)
+ throw new RuntimeException("Internal error: Found a flattened bag inside another");
+ else
+ inTheMiddleOfBag = true;
+ }else{
+ if (!(d instanceof DataBag.EndBag))
+ throw new RuntimeException("Internal error: Unknown bag delimiter type");
+ if (!inTheMiddleOfBag)
+ throw new RuntimeException("Internal error: Improper nesting of bag delimiter tuples");
+ inTheMiddleOfBag = false;
+ }
+ return true;
+ }
+ return false;
+ }
- protected void addToSuccessor(Datum d){
- if (d instanceof DataBag && !inTheMiddleOfBag && successor!=null && successor.needsFlattening()){
- DataBag bag = (DataBag)d;
- //flatten the bag and send it through the pipeline
- successor.add(DataBag.startBag);
- Iterator<Tuple> iter = bag.iterator();
- while(iter.hasNext())
- successor.add(iter.next());
- successor.add(DataBag.endBag);
- }else{
- //simply add the datum
- successor.add(d);
- }
- }
-
- public void markStale(boolean stale){
- synchronized(staleCount){
- if (stale){
- staleCount++;
- }else{
- if (staleCount > 0){
- synchronized(this){
- staleCount--;
- notifyAll();
- }
- }
- }
- }
- if (successor!=null)
- successor.markStale(stale);
- }
-
- public void setSuccessor(DataCollector output){
- this.successor = output;
- }
-
- public boolean isStale(){
- synchronized(staleCount){
- return staleCount>0;
- }
- }
-
- public void finishPipe(){
- finish();
- if (successor!=null)
- successor.finishPipe();
- }
-
- protected void finish(){}
-
+ protected void addToSuccessor(Datum d){
+ if (d instanceof DataBag && !inTheMiddleOfBag && successor!=null && successor.needsFlattening()){
+ DataBag bag = (DataBag)d;
+ //flatten the bag and send it through the pipeline
+ successor.add(DataBag.startBag);
+ Iterator<Tuple> iter = bag.iterator();
+ while(iter.hasNext())
+ successor.add(iter.next());
+ successor.add(DataBag.endBag);
+ }else{
+ //simply add the datum
+ successor.add(d);
+ }
+ }
+
+ public void markStale(boolean stale){
+ synchronized(staleCount){
+ if (stale){
+ staleCount++;
+ }else{
+ if (staleCount > 0){
+ synchronized(this){
+ staleCount--;
+ notifyAll();
+ }
+ }
+ }
+ }
+ if (successor!=null)
+ successor.markStale(stale);
+ }
+
+ public void setSuccessor(DataCollector output){
+ this.successor = output;
+ }
+
+ public boolean isStale(){
+ synchronized(staleCount){
+ return staleCount>0;
+ }
+ }
+
+ public void finishPipe(){
+ finish();
+ if (successor!=null)
+ successor.finishPipe();
+ }
+
+ protected void finish(){}
+
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/FlattenCollector.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/FlattenCollector.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/FlattenCollector.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/FlattenCollector.java Wed Feb 6 15:09:43 2008
@@ -21,22 +21,22 @@
public class FlattenCollector extends DataCollector {
- public FlattenCollector(DataCollector successor){
- super(successor);
- }
-
- @Override
- public void add(Datum d) {
- if (checkDelimiter(d))
- return;
- else
- successor.add(d);
- }
+ public FlattenCollector(DataCollector successor){
+ super(successor);
+ }
+
+ @Override
+ public void add(Datum d) {
+ if (checkDelimiter(d))
+ return;
+ else
+ successor.add(d);
+ }
- @Override
- protected boolean needFlatteningLocally() {
- return true;
- }
-
+ @Override
+ protected boolean needFlatteningLocally() {
+ return true;
+ }
+
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/UnflattenCollector.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/UnflattenCollector.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/UnflattenCollector.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/collector/UnflattenCollector.java Wed Feb 6 15:09:43 2008
@@ -26,34 +26,34 @@
public class UnflattenCollector extends DataCollector {
- DataBag bag;
-
- public UnflattenCollector(DataCollector successor){
- super(successor);
- }
-
- @Override
- public void add(Datum d) {
- if (inTheMiddleOfBag){
- if (checkDelimiter(d)){
- successor.add(bag);
- }else{
- if (d instanceof Tuple){
- bag.add((Tuple)d);
- }else{
- bag.add(new Tuple(d));
- }
- }
- }else{
- if (checkDelimiter(d)){
- //Bag must have started now
- bag = BagFactory.getInstance().newDefaultBag();
- }else{
- successor.add(d);
- }
- }
- }
-
-
+ DataBag bag;
+
+ public UnflattenCollector(DataCollector successor){
+ super(successor);
+ }
+
+ @Override
+ public void add(Datum d) {
+ if (inTheMiddleOfBag){
+ if (checkDelimiter(d)){
+ successor.add(bag);
+ }else{
+ if (d instanceof Tuple){
+ bag.add((Tuple)d);
+ }else{
+ bag.add(new Tuple(d));
+ }
+ }
+ }else{
+ if (checkDelimiter(d)){
+ //Bag must have started now
+ bag = BagFactory.getInstance().newDefaultBag();
+ }else{
+ successor.add(d);
+ }
+ }
+ }
+
+
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/AndCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/AndCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/AndCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/AndCond.java Wed Feb 6 15:09:43 2008
@@ -28,15 +28,15 @@
public class AndCond extends Cond {
- private static final long serialVersionUID = 1L;
- public List<Cond> cList;
-
- public AndCond(List<Cond> cList) {
- this.cList = cList;
- }
-
+ private static final long serialVersionUID = 1L;
+ public List<Cond> cList;
+
+ public AndCond(List<Cond> cList) {
+ this.cList = cList;
+ }
+
@Override
- public List<String> getFuncs() {
+ public List<String> getFuncs() {
List<String> funcs = new ArrayList<String>();
for (Iterator<Cond> it = cList.iterator(); it.hasNext(); ) {
funcs.addAll(it.next().getFuncs());
@@ -45,7 +45,7 @@
}
@Override
- public boolean eval(Datum input){
+ public boolean eval(Datum input){
for (Iterator<Cond> it = cList.iterator(); it.hasNext(); ) {
if (it.next().eval(input) == false) return false;
}
@@ -53,7 +53,7 @@
}
@Override
- public String toString() {
+ public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("(");
for (Iterator<Cond> it = cList.iterator(); it.hasNext(); ) {
@@ -66,15 +66,15 @@
@Override
public void finish() {
- for (Cond c: cList)
- c.finish();
+ for (Cond c: cList)
+ c.finish();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- for (Cond c: cList)
- c.instantiateFunc(instantiaor);
-
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ for (Cond c: cList)
+ c.instantiateFunc(instantiaor);
+
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/CompCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/CompCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/CompCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/CompCond.java Wed Feb 6 15:09:43 2008
@@ -29,21 +29,21 @@
public class CompCond extends Cond {
- private static final long serialVersionUID = 1L;
-
+ private static final long serialVersionUID = 1L;
+
public String op; // one of "<", ">", "==", etc.
public EvalSpec left, right;
- public CompCond(EvalSpec left, String op, EvalSpec right) throws IOException{
+ public CompCond(EvalSpec left, String op, EvalSpec right) throws IOException{
this.op = op.toLowerCase();
this.left = left;
this.right = right;
if (left.isAsynchronous() || right.isAsynchronous()){
- throw new IOException("Can't compare the output of an asynchronous function");
+ throw new IOException("Can't compare the output of an asynchronous function");
}
}
@@ -54,18 +54,18 @@
@Override
public boolean eval(Datum input) {
-
- Datum d1 = left.simpleEval(input);
- Datum d2 = right.simpleEval(input);
-
- if (!(d1 instanceof DataAtom) || !(d2 instanceof DataAtom)){
- throw new RuntimeException("Builtin functions cannot be used to compare non-atomic values. Use a filter function instead.");
- }
-
- DataAtom da1 = (DataAtom)d1;
- DataAtom da2 = (DataAtom)d2;
-
-
+
+ Datum d1 = left.simpleEval(input);
+ Datum d2 = right.simpleEval(input);
+
+ if (!(d1 instanceof DataAtom) || !(d2 instanceof DataAtom)){
+ throw new RuntimeException("Builtin functions cannot be used to compare non-atomic values. Use a filter function instead.");
+ }
+
+ DataAtom da1 = (DataAtom)d1;
+ DataAtom da2 = (DataAtom)d2;
+
+
char op1 = op.charAt(0);
char op2 = op.length() >= 2 ? op.charAt(1) : '0';
char op3 = op.length() == 3 ? op.charAt(2) : '0';
@@ -127,21 +127,21 @@
}
@Override
- public String toString() {
+ public String toString() {
return "(" + left + " " + op + " " + right + ")";
}
@Override
public void finish() {
- left.finish();
- right.finish();
+ left.finish();
+ right.finish();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- left.instantiateFunc(instantiaor);
- right.instantiateFunc(instantiaor);
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ left.instantiateFunc(instantiaor);
+ right.instantiateFunc(instantiaor);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/Cond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/Cond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/Cond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/Cond.java Wed Feb 6 15:09:43 2008
@@ -28,12 +28,12 @@
public abstract class Cond implements Serializable{
- public abstract List<String> getFuncs();
+ public abstract List<String> getFuncs();
public abstract boolean eval(Datum input);
public abstract void finish();
- abstract public void instantiateFunc(FunctionInstantiator instantiaor) throws IOException;
+ abstract public void instantiateFunc(FunctionInstantiator instantiaor) throws IOException;
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FalseCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FalseCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FalseCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FalseCond.java Wed Feb 6 15:09:43 2008
@@ -27,7 +27,7 @@
public class FalseCond extends Cond {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 1L;
@Override
public boolean eval(Datum input) {
@@ -43,11 +43,11 @@
return new ArrayList<String>();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- // TODO Auto-generated method stub
-
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FuncCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FuncCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FuncCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/FuncCond.java Wed Feb 6 15:09:43 2008
@@ -31,25 +31,25 @@
public class FuncCond extends Cond {
- private static final long serialVersionUID = 1L;
-
- public String funcName;
+ private static final long serialVersionUID = 1L;
+
+ public String funcName;
transient public FilterFunc func;
public EvalSpec args;
- public FuncCond(FunctionInstantiator fInstantiaor, String funcName, EvalSpec args) throws IOException{
- this.funcName = funcName;
- this.args = args;
- if (args!=null && args.isAsynchronous())
- throw new IOException("Can't use the output of an asynchronous function as an argument");
- instantiateFunc(fInstantiaor);
-
+ public FuncCond(FunctionInstantiator fInstantiaor, String funcName, EvalSpec args) throws IOException{
+ this.funcName = funcName;
+ this.args = args;
+ if (args!=null && args.isAsynchronous())
+ throw new IOException("Can't use the output of an asynchronous function as an argument");
+ instantiateFunc(fInstantiaor);
+
}
@Override
- public void instantiateFunc(FunctionInstantiator instantiaor) throws IOException{
- if(instantiaor != null)
- func = (FilterFunc)instantiaor.instantiateFuncFromAlias(funcName);
+ public void instantiateFunc(FunctionInstantiator instantiaor) throws IOException{
+ if(instantiaor != null)
+ func = (FilterFunc)instantiaor.instantiateFuncFromAlias(funcName);
}
@Override
@@ -61,16 +61,16 @@
@Override
public boolean eval(Datum input){
- try {
-
- Datum d = null;
- if (args!=null)
- d = args.simpleEval(input);
-
- if (d!=null && !(d instanceof Tuple))
- throw new RuntimeException("Internal error: Non-tuple returned on evaluation of arguments.");
+ try {
+
+ Datum d = null;
+ if (args!=null)
+ d = args.simpleEval(input);
+
+ if (d!=null && !(d instanceof Tuple))
+ throw new RuntimeException("Internal error: Non-tuple returned on evaluation of arguments.");
- return func.exec((Tuple)d);
+ return func.exec((Tuple)d);
} catch (IOException e) {
System.out.println("Warning: filter function " + funcName + " failed. Substituting default value \'false\'.");
return false;
@@ -78,14 +78,14 @@
}
@Override
- public String toString() {
+ public String toString() {
return funcName + "(" + args + ")";
}
@Override
public void finish() {
- if (args!=null)
- args.finish();
- func.finish();
+ if (args!=null)
+ args.finish();
+ func.finish();
}
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/NotCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/NotCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/NotCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/NotCond.java Wed Feb 6 15:09:43 2008
@@ -26,38 +26,38 @@
public class NotCond extends Cond {
- private static final long serialVersionUID = 1L;
-
- public Cond cond;
+ private static final long serialVersionUID = 1L;
+
+ public Cond cond;
- public NotCond(Cond cond) {
+ public NotCond(Cond cond) {
this.cond = cond;
}
@Override
- public List<String> getFuncs() {
+ public List<String> getFuncs() {
return cond.getFuncs();
}
@Override
- public boolean eval(Datum input){
+ public boolean eval(Datum input){
return !cond.eval(input);
}
@Override
- public String toString() {
+ public String toString() {
return "(NOT " + cond + ")";
}
@Override
public void finish() {
- cond.finish();
+ cond.finish();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- cond.instantiateFunc(instantiaor);
-
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ cond.instantiateFunc(instantiaor);
+
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/OrCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/OrCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/OrCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/OrCond.java Wed Feb 6 15:09:43 2008
@@ -28,17 +28,17 @@
public class OrCond extends Cond {
- private static final long serialVersionUID = 1L;
-
- public List<Cond> cList;
+ private static final long serialVersionUID = 1L;
+
+ public List<Cond> cList;
- public OrCond(List<Cond> cList) {
+ public OrCond(List<Cond> cList) {
this.cList = cList;
}
@Override
- public List<String> getFuncs() {
+ public List<String> getFuncs() {
List<String> funcs = new ArrayList<String>();
for (Iterator<Cond> it = cList.iterator(); it.hasNext(); ) {
funcs.addAll(it.next().getFuncs());
@@ -47,7 +47,7 @@
}
@Override
- public boolean eval(Datum input){
+ public boolean eval(Datum input){
for (Iterator<Cond> it = cList.iterator(); it.hasNext(); ) {
if (it.next().eval(input) == true) return true;
}
@@ -55,7 +55,7 @@
}
@Override
- public String toString() {
+ public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("(");
for (Iterator<Cond> it = cList.iterator(); it.hasNext(); ) {
@@ -68,16 +68,16 @@
@Override
public void finish() {
- for (Cond c: cList)
- c.finish();
+ for (Cond c: cList)
+ c.finish();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- for (Cond c: cList)
- c.instantiateFunc(instantiaor);
-
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ for (Cond c: cList)
+ c.instantiateFunc(instantiaor);
+
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/RegexpCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/RegexpCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/RegexpCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/RegexpCond.java Wed Feb 6 15:09:43 2008
@@ -28,18 +28,18 @@
public class RegexpCond extends Cond {
- private static final long serialVersionUID = 1L;
-
- public EvalSpec left;
- public String re;
+ private static final long serialVersionUID = 1L;
+
+ public EvalSpec left;
+ public String re;
- public RegexpCond(EvalSpec left, String re) throws IOException{
+ public RegexpCond(EvalSpec left, String re) throws IOException{
this.left = left;
this.re = re;
if (left.isAsynchronous())
- throw new IOException("Can't compare the output of an asynchronous function.");
+ throw new IOException("Can't compare the output of an asynchronous function.");
}
@Override
@@ -49,30 +49,30 @@
@Override
public boolean eval(Datum input){
-
- Datum d = left.simpleEval(input);
-
- if (!(d instanceof DataAtom))
- throw new RuntimeException("Cannot match non-atomic value against a regular expression. Use a filter function instead.");
+
+ Datum d = left.simpleEval(input);
+
+ if (!(d instanceof DataAtom))
+ throw new RuntimeException("Cannot match non-atomic value against a regular expression. Use a filter function instead.");
- return ((DataAtom)d).strval().matches(re);
+ return ((DataAtom)d).strval().matches(re);
}
@Override
- public String toString() {
+ public String toString() {
return "(" + left + " MATCHES " + "'" + re + "')";
}
@Override
public void finish() {
- left.finish();
+ left.finish();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- left.instantiateFunc(instantiaor);
-
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ left.instantiateFunc(instantiaor);
+
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/TrueCond.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/TrueCond.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/TrueCond.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/cond/TrueCond.java Wed Feb 6 15:09:43 2008
@@ -26,7 +26,7 @@
public class TrueCond extends Cond {
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 1L;
@Override
public boolean eval(Datum input) {
@@ -42,11 +42,11 @@
return new ArrayList<String>();
}
- @Override
- public void instantiateFunc(FunctionInstantiator instantiaor)
- throws IOException {
- // TODO Auto-generated method stub
-
- }
+ @Override
+ public void instantiateFunc(FunctionInstantiator instantiaor)
+ throws IOException {
+ // TODO Auto-generated method stub
+
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TimeWindowSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TimeWindowSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TimeWindowSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TimeWindowSpec.java Wed Feb 6 15:09:43 2008
@@ -29,22 +29,22 @@
public class TimeWindowSpec extends WindowSpec {
- private static final long serialVersionUID = 1L;
- double duration; // duration in seconds
- transient List<TimestampedTuple> window;
+ private static final long serialVersionUID = 1L;
+ double duration; // duration in seconds
+ transient List<TimestampedTuple> window;
- public TimeWindowSpec(windowType type, double duration){
- super(type);
- this.duration = duration;
+ public TimeWindowSpec(windowType type, double duration){
+ super(type);
+ this.duration = duration;
window = new LinkedList<TimestampedTuple>();
}
-
+
@Override
- protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
+ protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
return new DataCollector(endOfPipe) {
@Override
- public void add(Datum d){
+ public void add(Datum d){
boolean changed = false;
@@ -79,13 +79,13 @@
@Override
public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("[WINDOW ");
- sb.append(type);
- sb.append(" TIME ");
- sb.append(duration);
- sb.append("]");
- return sb.toString();
+ StringBuilder sb = new StringBuilder();
+ sb.append("[WINDOW ");
+ sb.append(type);
+ sb.append(" TIME ");
+ sb.append(duration);
+ sb.append("]");
+ return sb.toString();
}
public void visit(EvalSpecVisitor v) {
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TupleWindowSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TupleWindowSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TupleWindowSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/window/TupleWindowSpec.java Wed Feb 6 15:09:43 2008
@@ -25,54 +25,54 @@
import org.apache.pig.impl.eval.EvalSpecVisitor;
public class TupleWindowSpec extends WindowSpec {
- private static final long serialVersionUID = 1L;
-
- int numTuples;
+ private static final long serialVersionUID = 1L;
+
+ int numTuples;
transient List<Datum> window;
-
- public TupleWindowSpec(windowType type, int numTuples){
- super(type);
- this.numTuples = numTuples;
+
+ public TupleWindowSpec(windowType type, int numTuples){
+ super(type);
+ this.numTuples = numTuples;
window = new LinkedList<Datum>();
- }
-
- @Override
- protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
- return new DataCollector(endOfPipe) {
+ }
+
+ @Override
+ protected DataCollector setupDefaultPipe(DataCollector endOfPipe) {
+ return new DataCollector(endOfPipe) {
- @Override
- public void add(Datum d) {
- if (d!=null){
-
- if (d instanceof TimestampedTuple){
- if (((TimestampedTuple) d).isHeartBeat()) return;
- }
-
- window.add(d);
- while (window.size() > numTuples) {
- window.remove(0);
- }
- }
-
- // emit entire window content to output collector
+ @Override
+ public void add(Datum d) {
+ if (d!=null){
+
+ if (d instanceof TimestampedTuple){
+ if (((TimestampedTuple) d).isHeartBeat()) return;
+ }
+
+ window.add(d);
+ while (window.size() > numTuples) {
+ window.remove(0);
+ }
+ }
+
+ // emit entire window content to output collector
for (Iterator<Datum> it = window.iterator(); it.hasNext(); ) {
addToSuccessor(it.next());
}
- }
-
-
- };
+ }
+
+
+ };
}
-
+
@Override
public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("[WINDOW ");
- sb.append(type);
- sb.append(" TUPLES ");
- sb.append(numTuples);
- sb.append("]");
- return sb.toString();
+ StringBuilder sb = new StringBuilder();
+ sb.append("[WINDOW ");
+ sb.append(type);
+ sb.append(" TUPLES ");
+ sb.append(numTuples);
+ sb.append("]");
+ return sb.toString();
}
public void visit(EvalSpecVisitor v) {
Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/window/WindowSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/window/WindowSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/eval/window/WindowSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/eval/window/WindowSpec.java Wed Feb 6 15:09:43 2008
@@ -25,22 +25,22 @@
public abstract class WindowSpec extends EvalSpec {
- public enum windowType {SLIDING, HOPPING};
- protected windowType type;
-
- public WindowSpec(windowType type){
- this.type = type;
- }
+ public enum windowType {SLIDING, HOPPING};
+ protected windowType type;
+
+ public WindowSpec(windowType type){
+ this.type = type;
+ }
- @Override
- public List<String> getFuncs() {
- return new ArrayList<String>();
- }
+ @Override
+ public List<String> getFuncs() {
+ return new ArrayList<String>();
+ }
- @Override
- protected Schema mapInputSchema(Schema input){
- return input;
- }
-
-
+ @Override
+ protected Schema mapInputSchema(Schema input){
+ return input;
+ }
+
+
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/io/BufferedPositionedInputStream.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/io/BufferedPositionedInputStream.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/io/BufferedPositionedInputStream.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/io/BufferedPositionedInputStream.java Wed Feb 6 15:09:43 2008
@@ -30,24 +30,24 @@
import org.apache.tools.bzip2r.CBZip2InputStream;
public class BufferedPositionedInputStream extends InputStream {
- long pos;
+ long pos;
InputStream in;
- public BufferedPositionedInputStream(InputStream in, long pos) {
- this.in = in;
- this.pos = pos;
- }
-
- public BufferedPositionedInputStream(InputStream in){
- this(in,0);
- }
-
- @Override
- public int read() throws IOException {
+ public BufferedPositionedInputStream(InputStream in, long pos) {
+ this.in = in;
+ this.pos = pos;
+ }
+
+ public BufferedPositionedInputStream(InputStream in){
+ this(in,0);
+ }
+
+ @Override
+ public int read() throws IOException {
int c = in.read();
pos++;
return c;
- }
+ }
@Override
public int read(byte b[], int off, int len) throws IOException {
@@ -55,9 +55,9 @@
pos += read;
return read;
}
-
+
@Override
- public long skip(long n) throws IOException {
+ public long skip(long n) throws IOException {
long rc = in.skip(n);
pos += rc;
return rc;
@@ -68,8 +68,8 @@
*/
public long getPosition() throws IOException{
if (in instanceof CBZip2InputStream)
- return ((CBZip2InputStream)in).getPos();
- return pos;
+ return ((CBZip2InputStream)in).getPos();
+ return pos;
}
/*
Modified: incubator/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java Wed Feb 6 15:09:43 2008
@@ -72,7 +72,7 @@
}
@Override
- public int read() throws IOException {
+ public int read() throws IOException {
while (!isEOF()) {
int rc = current.read();
if (rc != -1)
@@ -83,14 +83,14 @@
}
@Override
- public int available() throws IOException {
+ public int available() throws IOException {
if (isEOF())
return 0;
return current.available();
}
@Override
- public void close() throws IOException {
+ public void close() throws IOException {
if (current != null) {
current.close();
current = null;
@@ -99,7 +99,7 @@
}
@Override
- public int read(byte[] b, int off, int len) throws IOException {
+ public int read(byte[] b, int off, int len) throws IOException {
int count = 0;
while (!isEOF() && len > 0) {
int rc = current.read(b, off, len);
@@ -115,12 +115,12 @@
}
@Override
- public int read(byte[] b) throws IOException {
+ public int read(byte[] b) throws IOException {
return read(b, 0, b.length);
}
@Override
- public long skip(long n) throws IOException {
+ public long skip(long n) throws IOException {
while (!isEOF() && n > 0) {
n -= current.skip(n);
}
@@ -143,13 +143,13 @@
*/
public static InputStream openDFSFile(String fileName) throws IOException{
- PigRecordReader prr = PigInputFormat.PigRecordReader.getPigRecordReader();
-
- if (prr == null)
- throw new RuntimeException("can't open DFS file while executing locally");
-
- return openDFSFile(fileName, prr.getJobConf());
-
+ PigRecordReader prr = PigInputFormat.PigRecordReader.getPigRecordReader();
+
+ if (prr == null)
+ throw new RuntimeException("can't open DFS file while executing locally");
+
+ return openDFSFile(fileName, prr.getJobConf());
+
}
public static InputStream openDFSFile(String fileName, JobConf conf) throws IOException{
@@ -172,13 +172,13 @@
if (elem.exists()) {
try {
if(! elem.getDataStorage().isContainer(elem.toString())) {
- return elem.open();
- }
+ return elem.open();
+ }
}
catch (DataStorageException e) {
- IOException ioe = new IOException("Failed to determine if elem=" + elem + " is container");
- ioe.initCause(e);
- throw ioe;
+ IOException ioe = new IOException("Failed to determine if elem=" + elem + " is container");
+ ioe.initCause(e);
+ throw ioe;
}
ArrayList<ElementDescriptor> arrayList =
@@ -194,11 +194,11 @@
arrayList.toArray(elements);
} else {
- // It might be a glob
- if (!globMatchesFiles(elem, elem.getDataStorage())) {
- throw new IOException(elem.toString() + " does not exist");
- }
- }
+ // It might be a glob
+ if (!globMatchesFiles(elem, elem.getDataStorage())) {
+ throw new IOException(elem.toString() + " does not exist");
+ }
+ }
return new DataStorageInputStreamIterator(elements);
}
@@ -226,7 +226,7 @@
}
static public OutputStream create(String fileSpec, PigContext pigContext) throws IOException{
- return create(fileSpec,false,pigContext);
+ return create(fileSpec,false,pigContext);
}
static public OutputStream create(String fileSpec, boolean append, PigContext pigContext) throws IOException {
@@ -246,12 +246,12 @@
else {
fileSpec = fileSpec.substring(LOCAL_PREFIX.length());
- // TODO probably this should be replaced with the local file system
- File f = (new File(fileSpec)).getParentFile();
- if (f!=null){
- f.mkdirs();
- }
-
+ // TODO probably this should be replaced with the local file system
+ File f = (new File(fileSpec)).getParentFile();
+ if (f!=null){
+ f.mkdirs();
+ }
+
return new FileOutputStream(fileSpec,append);
}
}
@@ -377,30 +377,30 @@
}
}
- private static boolean globMatchesFiles(ElementDescriptor elem,
- DataStorage fs)
- throws IOException
- {
- try {
- // Currently, if you give a glob with non-special glob characters, hadoop
- // returns an array with your file name in it. So check for that.
- ElementDescriptor[] elems = fs.asCollection(elem.toString());
- switch (elems.length) {
- case 0:
- return false;
+ private static boolean globMatchesFiles(ElementDescriptor elem,
+ DataStorage fs)
+ throws IOException
+ {
+ try {
+ // Currently, if you give a glob with non-special glob characters, hadoop
+ // returns an array with your file name in it. So check for that.
+ ElementDescriptor[] elems = fs.asCollection(elem.toString());
+ switch (elems.length) {
+ case 0:
+ return false;
- case 1:
- return !elems[0].equals(elem);
+ case 1:
+ return !elems[0].equals(elem);
- default:
- return true;
- }
- }
- catch (DataStorageException e) {
- IOException ioe = new IOException("Unable to get collect for pattern " + elem.toString());
- ioe.initCause(e);
- throw ioe;
- }
- }
+ default:
+ return true;
+ }
+ }
+ catch (DataStorageException e) {
+ IOException ioe = new IOException("Unable to get collect for pattern " + elem.toString());
+ ioe.initCause(e);
+ throw ioe;
+ }
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/io/FileSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/io/FileSpec.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/io/FileSpec.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/io/FileSpec.java Wed Feb 6 15:09:43 2008
@@ -28,34 +28,34 @@
*
*/
public class FileSpec implements Serializable {
-
- private static final long serialVersionUID = 1L;
- String fileName;
+
+ private static final long serialVersionUID = 1L;
+ String fileName;
- String funcSpec;
-
- public FileSpec(String fileName, String funcSpec){
- this.fileName = fileName;
- this.funcSpec = funcSpec;
- }
-
- public String getFileName(){
- return fileName;
- }
-
- public String getFuncSpec(){
- return funcSpec;
- }
-
- @Override
- public String toString(){
- return fileName + ":" + funcSpec;
- }
-
- public String getFuncName(){
- return PigContext.getClassNameFromSpec(funcSpec);
- }
- public int getSize() {
- throw new UnsupportedOperationException("File Size not implemented yet");
- }
+ String funcSpec;
+
+ public FileSpec(String fileName, String funcSpec){
+ this.fileName = fileName;
+ this.funcSpec = funcSpec;
+ }
+
+ public String getFileName(){
+ return fileName;
+ }
+
+ public String getFuncSpec(){
+ return funcSpec;
+ }
+
+ @Override
+ public String toString(){
+ return fileName + ":" + funcSpec;
+ }
+
+ public String getFuncName(){
+ return PigContext.getClassNameFromSpec(funcSpec);
+ }
+ public int getSize() {
+ throw new UnsupportedOperationException("File Size not implemented yet");
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/io/PigFile.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/io/PigFile.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/io/PigFile.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/io/PigFile.java Wed Feb 6 15:09:43 2008
@@ -41,7 +41,7 @@
}
public PigFile(String filename){
- file = filename;
+ file = filename;
}
public DataBag load(LoadFunc lfunc, PigContext pigContext) throws IOException {
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCogroup.java Wed Feb 6 15:09:43 2008
@@ -48,8 +48,8 @@
@Override
public String name() {
- if (inputs.size() == 1) return "Group " + scope + "-" + id;
- else return "CoGroup " + scope + "-" + id;
+ if (inputs.size() == 1) return "Group " + scope + "-" + id;
+ else return "CoGroup " + scope + "-" + id;
}
@Override
public String arguments() {
@@ -180,8 +180,8 @@
return specs;
}
- public void visit(LOVisitor v) {
- v.visitCogroup(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitCogroup(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOEval.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOEval.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOEval.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOEval.java Wed Feb 6 15:09:43 2008
@@ -89,7 +89,7 @@
return spec;
}
- public void visit(LOVisitor v) {
- v.visitEval(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitEval(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOLoad.java Wed Feb 6 15:09:43 2008
@@ -120,7 +120,7 @@
return funcs;
}
- public void visit(LOVisitor v) {
- v.visitLoad(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitLoad(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSort.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSort.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSort.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSort.java Wed Feb 6 15:09:43 2008
@@ -79,8 +79,8 @@
return sortSpec;
}
- public void visit(LOVisitor v) {
- v.visitSort(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitSort(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSplit.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSplit.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSplit.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOSplit.java Wed Feb 6 15:09:43 2008
@@ -59,8 +59,8 @@
return "Split " + scope + "-" + id;
}
- public void visit(LOVisitor v) {
- v.visitSplit(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitSplit(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOStore.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOStore.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOStore.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOStore.java Wed Feb 6 15:09:43 2008
@@ -109,7 +109,7 @@
return append;
}
- public void visit(LOVisitor v) {
- v.visitStore(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitStore(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOUnion.java Wed Feb 6 15:09:43 2008
@@ -82,7 +82,7 @@
return outputType;
}
- public void visit(LOVisitor v) {
- v.visitUnion(this);
- }
+ public void visit(LOVisitor v) {
+ v.visitUnion(this);
+ }
}
Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java
URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java?rev=619213&r1=619212&r2=619213&view=diff
==============================================================================
--- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java (original)
+++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java Wed Feb 6 15:09:43 2008
@@ -29,82 +29,82 @@
import org.apache.pig.backend.executionengine.ExecLogicalPlan;
public class LogicalPlan implements Serializable, ExecLogicalPlan{
- private static final long serialVersionUID = 1L;
+ private static final long serialVersionUID = 1L;
- protected OperatorKey root;
- protected Map<OperatorKey, LogicalOperator> opTable;
- protected PigContext pigContext = null;
-
- protected String alias = null;
-
- public OperatorKey getRoot() {
- return root;
- }
-
- public LogicalPlan(OperatorKey rootIn,
- Map<OperatorKey, LogicalOperator> opTable ,
- PigContext pigContext) {
- this.pigContext = pigContext;
- this.root = rootIn;
- this.opTable = opTable;
+ protected OperatorKey root;
+ protected Map<OperatorKey, LogicalOperator> opTable;
+ protected PigContext pigContext = null;
+
+ protected String alias = null;
+
+ public OperatorKey getRoot() {
+ return root;
+ }
+
+ public LogicalPlan(OperatorKey rootIn,
+ Map<OperatorKey, LogicalOperator> opTable ,
+ PigContext pigContext) {
+ this.pigContext = pigContext;
+ this.root = rootIn;
+ this.opTable = opTable;
alias = opTable.get(root).alias;
- }
-
- public Map<OperatorKey, LogicalOperator> getOpTable() {
- return this.opTable;
- }
-
+ }
+
+ public Map<OperatorKey, LogicalOperator> getOpTable() {
+ return this.opTable;
+ }
+
public void setRoot(OperatorKey newRoot) {
root = newRoot;
}
- public PigContext getPigContext() {
- return pigContext;
- }
-
- public String getAlias() {
- return alias;
- }
-
- public void setAlias(String newAlias) {
- alias = newAlias;
- }
+ public PigContext getPigContext() {
+ return pigContext;
+ }
+
+ public String getAlias() {
+ return alias;
+ }
- public List<String> getFuncs() {
+ public void setAlias(String newAlias) {
+ alias = newAlias;
+ }
+
+ public List<String> getFuncs() {
if (root == null) return new LinkedList<String>();
else return opTable.get(root).getFuncs();
}
-
- // indentation for root is 0
- @Override
- public String toString() {
- StringBuffer sb = new StringBuffer();
- sb.append(opTable.get(root).name() +"(" + opTable.get(root).arguments() +")\n");
- sb.append(appendChildren(root, 1));
- return sb.toString();
- }
- public String appendChildren(OperatorKey parent, int indentation) {
- StringBuffer sb = new StringBuffer();
- List<OperatorKey> children = opTable.get(parent).getInputs();
- for(int i=0; i != children.size(); i++) {
- for(int j=0; j != indentation; j++) {
- sb.append("\t");
- }
-
- sb.append(opTable.get(children.get(i)).name() +
- "(" + opTable.get(children.get(i)).arguments()+ ")\n");
- sb.append(appendChildren(children.get(i), indentation+1));
- }
- return sb.toString();
- }
-
- public int getOutputType(){
- return opTable.get(root).getOutputType();
- }
-
- public void explain(OutputStream out) {
- LOVisitor lprinter = new LOPrinter(new PrintStream(out));
-
- opTable.get(root).visit(lprinter);
- }
+
+ // indentation for root is 0
+ @Override
+ public String toString() {
+ StringBuffer sb = new StringBuffer();
+ sb.append(opTable.get(root).name() +"(" + opTable.get(root).arguments() +")\n");
+ sb.append(appendChildren(root, 1));
+ return sb.toString();
+ }
+ public String appendChildren(OperatorKey parent, int indentation) {
+ StringBuffer sb = new StringBuffer();
+ List<OperatorKey> children = opTable.get(parent).getInputs();
+ for(int i=0; i != children.size(); i++) {
+ for(int j=0; j != indentation; j++) {
+ sb.append("\t");
+ }
+
+ sb.append(opTable.get(children.get(i)).name() +
+ "(" + opTable.get(children.get(i)).arguments()+ ")\n");
+ sb.append(appendChildren(children.get(i), indentation+1));
+ }
+ return sb.toString();
+ }
+
+ public int getOutputType(){
+ return opTable.get(root).getOutputType();
+ }
+
+ public void explain(OutputStream out) {
+ LOVisitor lprinter = new LOPrinter(new PrintStream(out));
+
+ opTable.get(root).visit(lprinter);
+ }
}