You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by pi...@apache.org on 2008/07/12 17:26:37 UTC
svn commit: r676185 [1/4] - in /incubator/pig/branches/types:
src/org/apache/pig/ src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/local/executionengine/
src/org/apache/pig/builtin/ src/org/apache/pig/data/ src/org/apache/p...
Author: pisong
Date: Sat Jul 12 08:26:35 2008
New Revision: 676185
URL: http://svn.apache.org/viewvc?rev=676185&view=rev
Log:
PIG-276 UDF overloading
Added:
incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java
incubator/pig/branches/types/src/org/apache/pig/builtin/BagSize.java
incubator/pig/branches/types/src/org/apache/pig/builtin/CONCAT.java
incubator/pig/branches/types/src/org/apache/pig/builtin/ConstantSize.java
incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleAvg.java
incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMax.java
incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMin.java
incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleSum.java
incubator/pig/branches/types/src/org/apache/pig/builtin/FloatAvg.java
incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMax.java
incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMin.java
incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java
incubator/pig/branches/types/src/org/apache/pig/builtin/IntAvg.java
incubator/pig/branches/types/src/org/apache/pig/builtin/IntMax.java
incubator/pig/branches/types/src/org/apache/pig/builtin/IntMin.java
incubator/pig/branches/types/src/org/apache/pig/builtin/IntSum.java
incubator/pig/branches/types/src/org/apache/pig/builtin/LongAvg.java
incubator/pig/branches/types/src/org/apache/pig/builtin/LongMax.java
incubator/pig/branches/types/src/org/apache/pig/builtin/LongMin.java
incubator/pig/branches/types/src/org/apache/pig/builtin/LongSum.java
incubator/pig/branches/types/src/org/apache/pig/builtin/MapSize.java
incubator/pig/branches/types/src/org/apache/pig/builtin/SIZE.java
incubator/pig/branches/types/src/org/apache/pig/builtin/StringConcat.java
incubator/pig/branches/types/src/org/apache/pig/builtin/StringMax.java
incubator/pig/branches/types/src/org/apache/pig/builtin/StringMin.java
incubator/pig/branches/types/src/org/apache/pig/builtin/StringSize.java
incubator/pig/branches/types/src/org/apache/pig/builtin/TupleSize.java
incubator/pig/branches/types/test/org/apache/pig/test/TestFuncSpec.java
Modified:
incubator/pig/branches/types/src/org/apache/pig/EvalFunc.java
incubator/pig/branches/types/src/org/apache/pig/PigServer.java
incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java
incubator/pig/branches/types/src/org/apache/pig/builtin/AVG.java
incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java
incubator/pig/branches/types/src/org/apache/pig/builtin/MAX.java
incubator/pig/branches/types/src/org/apache/pig/builtin/MIN.java
incubator/pig/branches/types/src/org/apache/pig/builtin/SUM.java
incubator/pig/branches/types/src/org/apache/pig/data/DataByteArray.java
incubator/pig/branches/types/src/org/apache/pig/data/DataType.java
incubator/pig/branches/types/src/org/apache/pig/impl/PigContext.java
incubator/pig/branches/types/src/org/apache/pig/impl/io/FileSpec.java
incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/LOSort.java
incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/LOUserFunc.java
incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt
incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/schema/Schema.java
incubator/pig/branches/types/src/org/apache/pig/impl/logicalLayer/validators/TypeCheckingVisitor.java
incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/JobControlCompiler.java
incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/MRCompiler.java
incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigInputFormat.java
incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/PigSplit.java
incubator/pig/branches/types/src/org/apache/pig/impl/mapReduceLayer/plans/UDFFinder.java
incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/expressionOperators/POUserComparisonFunc.java
incubator/pig/branches/types/src/org/apache/pig/impl/physicalLayer/expressionOperators/POUserFunc.java
incubator/pig/branches/types/test/org/apache/pig/test/TestBuiltin.java
incubator/pig/branches/types/test/org/apache/pig/test/TestInputOutputFileValidator.java
incubator/pig/branches/types/test/org/apache/pig/test/TestJobSubmission.java
incubator/pig/branches/types/test/org/apache/pig/test/TestLoad.java
incubator/pig/branches/types/test/org/apache/pig/test/TestLocalJobSubmission.java
incubator/pig/branches/types/test/org/apache/pig/test/TestLogToPhyCompiler.java
incubator/pig/branches/types/test/org/apache/pig/test/TestMRCompiler.java
incubator/pig/branches/types/test/org/apache/pig/test/TestPOSort.java
incubator/pig/branches/types/test/org/apache/pig/test/TestPOUserFunc.java
incubator/pig/branches/types/test/org/apache/pig/test/TestPigSplit.java
incubator/pig/branches/types/test/org/apache/pig/test/TestStore.java
incubator/pig/branches/types/test/org/apache/pig/test/TestTypeCheckingValidator.java
incubator/pig/branches/types/test/org/apache/pig/test/TestTypeCheckingValidatorNoSchema.java
incubator/pig/branches/types/test/org/apache/pig/test/TestUnion.java
incubator/pig/branches/types/test/org/apache/pig/test/Util.java
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Arithmetic.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/BinCond.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Cogroup.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/ComplexForeach.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Distinct.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Generate.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/IsNull1.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/IsNull2.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Sort.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Split1.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Split2.gld
incubator/pig/branches/types/test/org/apache/pig/test/data/GoldenFiles/Union.gld
incubator/pig/branches/types/test/org/apache/pig/test/utils/GenPhyOp.java
incubator/pig/branches/types/test/org/apache/pig/test/utils/TypeCheckingTestUtil.java
incubator/pig/branches/types/test/org/apache/pig/test/utils/dotGraph/LogicalPlanLoader.java
Modified: incubator/pig/branches/types/src/org/apache/pig/EvalFunc.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/EvalFunc.java?rev=676185&r1=676184&r2=676185&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/EvalFunc.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/EvalFunc.java Sat Jul 12 08:26:35 2008
@@ -21,9 +21,11 @@
import java.io.IOException;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
+import java.util.List;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.physicalLayer.PigProgressable;
@@ -80,20 +82,20 @@
Algebraic a = (Algebraic)this;
errMsg = "function of " + getClass().getName() + " is not of the expected type.";
- if (getReturnTypeFromSpec(a.getInitial()) != Tuple.class)
+ if (getReturnTypeFromSpec(new FuncSpec(a.getInitial())) != Tuple.class)
throw new RuntimeException("Initial " + errMsg);
- if (getReturnTypeFromSpec(a.getIntermed()) != Tuple.class)
+ if (getReturnTypeFromSpec(new FuncSpec(a.getIntermed())) != Tuple.class)
throw new RuntimeException("Intermediate " + errMsg);
- if (getReturnTypeFromSpec(a.getFinal()) != returnType)
+ if (getReturnTypeFromSpec(new FuncSpec(a.getFinal())) != returnType)
throw new RuntimeException("Final " + errMsg);
}
}
- private Type getReturnTypeFromSpec(String funcSpec){
+ private Type getReturnTypeFromSpec(FuncSpec funcSpec){
try{
- return ((EvalFunc)PigContext.instantiateFuncFromSpec(funcSpec)).getReturnType();
+ return ((EvalFunc<?>)PigContext.instantiateFuncFromSpec(funcSpec)).getReturnType();
}catch (ClassCastException e){
throw new RuntimeException(funcSpec + " does not specify an eval func", e);
}
@@ -156,4 +158,12 @@
public void setReporter(PigProgressable reporter) {
this.reporter = reporter;
}
+
+ /**
+ * @return A List containing FuncSpec objects representing the Function class
+ * which can handle the inputs corresponding to the schema in the objects
+ */
+ public List<FuncSpec> getArgToFuncMapping() throws FrontendException{
+ return null;
+ }
}
Added: incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/FuncSpec.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,199 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+/**
+ * Class to represent a UDF specification - essentially
+ * encapsulates the class name and the arguments to the constructor
+ */
+/**
+ *
+ */
+public class FuncSpec implements Serializable{
+
+ private static final long serialVersionUID = 2L;
+ String className = null;
+ String[] ctorArgs = null;
+ Schema inputArgsSchema = null;
+
+ /**
+ * @param className the name of the class for the udf
+ * @param ctorArgs the arguments for the constructor for the above class
+ */
+ public FuncSpec(String className, String[] ctorArgs) {
+ this.className = className;
+ this.ctorArgs = ctorArgs;
+ }
+
+ /**
+ * @param className the name of the class for the udf
+ * @param ctorArgs the arguments for the constructor for the above class
+ * @param inputArgsSchema schema for input args taken by this Function
+ */
+ public FuncSpec(String className, String[] ctorArgs, Schema inputArgsSchema) {
+ this(className, ctorArgs);
+ this.inputArgsSchema = inputArgsSchema;
+ }
+
+ /**
+ * @param funcSpec the name of the function and any arguments.
+ * It should have the form: classname('arg1', 'arg2', ...)
+ */
+ public FuncSpec(String funcSpec) {
+ this.className = getClassNameFromSpec(funcSpec);
+ List<String> args = parseArguments(getArgStringFromSpec(funcSpec));
+ if(args.size() > 0) {
+ this.ctorArgs = new String[args.size()];
+ int i = 0;
+ for (Iterator<String> iterator = args.iterator(); iterator.hasNext();) {
+ this.ctorArgs[i++] = iterator.next();
+
+ }
+ }
+
+ }
+
+ /**
+ *
+ * @param funcSpec funcSpec the name of the function and any arguments.
+ * It should have the form: classname('arg1', 'arg2', ...)
+ * @param inputArgsSchema schema for input args taken by this Function
+ */
+ public FuncSpec(String funcSpec, Schema inputArgsSchema) {
+ this(funcSpec);
+ this.inputArgsSchema = inputArgsSchema;
+ }
+
+ public static String getClassNameFromSpec(String funcSpec){
+ int paren = funcSpec.indexOf('(');
+ if (paren!=-1)
+ return funcSpec.substring(0, paren);
+ else
+ return funcSpec;
+ }
+
+ public static String getArgStringFromSpec(String funcSpec){
+ int paren = funcSpec.indexOf('(');
+ if (paren!=-1)
+ return funcSpec.substring(paren+1);
+ else
+ return "";
+ }
+
+ /**
+ * Function to parse the arguments from a function specification argument list
+ * @param argString should be of the form "'arg1', 'arg2', ..."
+ * @return List of the different argument strings
+ */
+ public static List<String> parseArguments(String argString){
+ List<String> args = new ArrayList<String>();
+
+ int startIndex = 0;
+ int endIndex;
+ while (startIndex < argString.length()) {
+ while (startIndex < argString.length() && argString.charAt(startIndex++) != '\'')
+ ;
+ endIndex = startIndex;
+ while (endIndex < argString.length() && argString.charAt(endIndex) != '\'') {
+ if (argString.charAt(endIndex) == '\\')
+ endIndex++;
+ endIndex++;
+ }
+ if (endIndex < argString.length()) {
+ args.add(argString.substring(startIndex, endIndex));
+ }
+ startIndex = endIndex + 1;
+ }
+ return args;
+ }
+
+ /**
+ * @return the className
+ */
+ public String getClassName() {
+ return className;
+ }
+ /**
+ * @param className the className to set
+ */
+ public void setClassName(String className) {
+ this.className = className;
+ }
+ /**
+ * @return the ctorArgs
+ */
+ public String[] getCtorArgs() {
+ return ctorArgs;
+ }
+ /**
+ * @param ctorArgs the ctorArgs to set
+ */
+ public void setCtorArgs(String[] ctorArgs) {
+ this.ctorArgs = ctorArgs;
+ }
+
+ /* (non-Javadoc)
+ * @see java.lang.Object#toString()
+ */
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append(className);
+
+
+ if(ctorArgs != null) {
+ sb.append("(");
+ for (int i = 0; i < ctorArgs.length; i++) {
+ sb.append("'");
+ sb.append(ctorArgs[i]);
+ sb.append("'");
+ if(i != ctorArgs.length - 1) {
+ sb.append(",");
+ }
+ }
+ sb.append(")");
+
+ }
+
+
+ return sb.toString();
+ }
+
+ /**
+ * @return the inputArgsSchema
+ */
+ public Schema getInputArgsSchema() {
+ return inputArgsSchema;
+ }
+
+ /**
+ * @param inputArgsSchema the inputArgsSchema to set
+ */
+ public void setInputArgsSchema(Schema inputArgsSchema) {
+ this.inputArgsSchema = inputArgsSchema;
+ }
+
+}
Modified: incubator/pig/branches/types/src/org/apache/pig/PigServer.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/PigServer.java?rev=676185&r1=676184&r2=676185&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/PigServer.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/PigServer.java Sat Jul 12 08:26:35 2008
@@ -27,6 +27,7 @@
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
@@ -144,8 +145,22 @@
* @param functionSpec - the name of the function and any arguments.
* It should have the form: classname('arg1', 'arg2', ...)
*/
+ @Deprecated
public void registerFunction(String function, String functionSpec) {
- pigContext.registerFunction(function, functionSpec);
+ registerFunction(function, new FuncSpec(functionSpec));
+ }
+
+ /**
+ * Defines an alias for the given function spec. This
+ * is useful for functions that require arguments to the
+ * constructor.
+ *
+ * @param function - the new function alias to define.
+ * @param funcSpec - the FuncSpec object representing the name of
+ * the function class and any arguments to constructor.
+ */
+ public void registerFunction(String function, FuncSpec funcSpec) {
+ pigContext.registerFunction(function, funcSpec);
}
private URL locateJarFromResources(String jarName) throws IOException {
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java?rev=676185&r1=676184&r2=676185&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Sat Jul 12 08:26:35 2008
@@ -39,6 +39,7 @@
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobSubmissionProtocol;
import org.apache.hadoop.mapred.JobTracker;
+import org.apache.pig.FuncSpec;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.executionengine.ExecJob;
@@ -246,7 +247,7 @@
str.setPc(pigContext);
spec = new FileSpec(FileLocalizer.getTemporaryPath(null,
pigContext).toString(),
- BinStorage.class.getName());
+ new FuncSpec(BinStorage.class.getName()));
str.setSFile(spec);
plan.addAsLeaf(str);
}
@@ -479,7 +480,7 @@
str.setPc(pigContext);
spec = new FileSpec(FileLocalizer.getTemporaryPath(null,
pigContext).toString(),
- BinStorage.class.getName());
+ new FuncSpec(BinStorage.class.getName()));
str.setSFile(spec);
plan.addAsLeaf(str);
} else{
Modified: incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java?rev=676185&r1=676184&r2=676185&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/backend/local/executionengine/LocalExecutionEngine.java Sat Jul 12 08:26:35 2008
@@ -27,6 +27,7 @@
import java.util.HashMap;
import java.util.HashSet;
+import org.apache.pig.FuncSpec;
import org.apache.pig.impl.PigContext;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.BagFactory;
@@ -136,7 +137,7 @@
str.setPc(pigContext);
spec = new FileSpec(FileLocalizer.getTemporaryPath(null,
pigContext).toString(),
- BinStorage.class.getName());
+ new FuncSpec(BinStorage.class.getName()));
str.setSFile(spec);
plan.addAsLeaf(str);
}
Modified: incubator/pig/branches/types/src/org/apache/pig/builtin/AVG.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/AVG.java?rev=676185&r1=676184&r2=676185&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/AVG.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/AVG.java Sat Jul 12 08:26:35 2008
@@ -18,14 +18,18 @@
package org.apache.pig.builtin;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Iterator;
+import java.util.List;
import org.apache.pig.Algebraic;
import org.apache.pig.EvalFunc;
+import org.apache.pig.FuncSpec;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.backend.executionengine.ExecException;
@@ -41,14 +45,19 @@
@Override
public Double exec(Tuple input) throws IOException {
try {
- double sum = sum(input);
+ Double sum = sum(input);
+ if(sum == null) {
+ // either we were handed an empty bag or a bag
+ // filled with nulls - return null in this case
+ return null;
+ }
double count = count(input);
- double avg = 0;
+ Double avg = null;
if (count > 0)
- avg = sum / count;
+ avg = new Double(sum / count);
- return new Double(avg);
+ return avg;
} catch (ExecException ee) {
IOException oughtToBeEE = new IOException();
oughtToBeEE.initCause(ee);
@@ -73,7 +82,7 @@
public Tuple exec(Tuple input) throws IOException {
try {
Tuple t = mTupleFactory.newTuple(2);
- t.set(0, new Double(sum(input)));
+ t.set(0, sum(input));
t.set(1, new Long(count(input)));
return t;
} catch(RuntimeException t) {
@@ -108,14 +117,17 @@
DataBag b = (DataBag)input.get(0);
Tuple combined = combine(b);
- double sum = (Double)combined.get(0);
+ Double sum = (Double)combined.get(0);
+ if(sum == null) {
+ return null;
+ }
double count = (Long)combined.get(1);
- double avg = 0;
+ Double avg = null;
if (count > 0) {
- avg = sum / count;
+ avg = new Double(sum / count);
}
- return new Double(avg);
+ return avg;
} catch (ExecException ee) {
IOException oughtToBeEE = new IOException();
oughtToBeEE.initCause(ee);
@@ -128,15 +140,27 @@
double sum = 0;
long count = 0;
- Tuple output = mTupleFactory.newTuple(2);
+ // combine is called from Intermediate and Final
+ // In either case, Initial would have been called
+ // before and would have sent in valid tuples
+ // Hence we don't need to check if incoming bag
+ // is empty
+ Tuple output = mTupleFactory.newTuple(2);
+ boolean sawNonNull = false;
for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
Tuple t = it.next();
- sum += (Double)t.get(0);
+ Double d = (Double)t.get(0);
+ if(d == null) continue;
+ sawNonNull = true;
+ sum += d;
count += (Long)t.get(1);
}
-
- output.set(0, new Double(sum));
+ if(sawNonNull) {
+ output.set(0, new Double(sum));
+ } else {
+ output.set(0, null);
+ }
output.set(1, new Long(count));
return output;
}
@@ -146,18 +170,38 @@
return values.size();
}
- static protected double sum(Tuple input) throws ExecException, IOException {
+ static protected Double sum(Tuple input) throws ExecException, IOException {
DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ if(values.size() == 0) {
+ return null;
+ }
double sum = 0;
+ boolean sawNonNull = false;
for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
Tuple t = it.next();
- Double d = DataType.toDouble(t.get(0));
- if (d == null) continue;
- sum += d;
+ try{
+ Double d = DataType.toDouble(t.get(0));
+ if (d == null) continue;
+ sawNonNull = true;
+ sum += d;
+ }catch(NumberFormatException nfe){
+ // do nothing - essentially treat this
+ // particular input as null
+ }catch(RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
}
- return sum;
+ if(sawNonNull) {
+ return new Double(sum);
+ } else {
+ return null;
+ }
}
@Override
@@ -165,4 +209,17 @@
return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
}
+ /* (non-Javadoc)
+ * @see org.apache.pig.EvalFunc#getArgToFuncMapping()
+ */
+ @Override
+ public List<FuncSpec> getArgToFuncMapping() throws FrontendException {
+ List<FuncSpec> funcList = new ArrayList<FuncSpec>();
+ funcList.add(new FuncSpec(this.getClass().getName(), Schema.generateNestedSchema(DataType.BAG, DataType.BYTEARRAY)));
+ funcList.add(new FuncSpec(DoubleAvg.class.getName(), Schema.generateNestedSchema(DataType.BAG, DataType.DOUBLE)));
+ funcList.add(new FuncSpec(FloatAvg.class.getName(), Schema.generateNestedSchema(DataType.BAG, DataType.FLOAT)));
+ funcList.add(new FuncSpec(IntAvg.class.getName(), Schema.generateNestedSchema(DataType.BAG, DataType.INTEGER)));
+ funcList.add(new FuncSpec(LongAvg.class.getName(), Schema.generateNestedSchema(DataType.BAG, DataType.LONG)));
+ return funcList;
+ }
}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/BagSize.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/BagSize.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/BagSize.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/BagSize.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,53 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+
+/**
+ * Generates the size of the first field of a tuple.
+ */
+public class BagSize extends EvalFunc<Long> {
+
+ @Override
+ public Long exec(Tuple input) throws IOException {
+ try {
+ DataBag bag = (DataBag)(input.get(0));
+ return bag == null ? null : new Long(bag.size());
+ } catch (ExecException exp) {
+ IOException oughtToBeEE =
+ new IOException("Error processing: " +
+ input.toString() + exp.getMessage());
+ oughtToBeEE.initCause(exp);
+ throw oughtToBeEE;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.LONG));
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/CONCAT.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/CONCAT.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/CONCAT.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/CONCAT.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,78 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.pig.EvalFunc;
+import org.apache.pig.FuncSpec;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+
+/**
+ * Generates the concatenation of the first two fields of a tuple.
+ */
+public class CONCAT extends EvalFunc<DataByteArray> {
+
+ @Override
+ public DataByteArray exec(Tuple input) throws IOException {
+ try {
+ DataByteArray dba1 = (DataByteArray)input.get(0);
+ DataByteArray dba2 = (DataByteArray)input.get(1);
+ if(dba1 != null && dba2 != null) {
+ return new DataByteArray(dba1, dba2);
+ } else {
+ return null;
+ }
+ } catch (ExecException exp) {
+ IOException oughtToBeEE = new IOException("Error processing: " +
+ input.toString() + exp.getMessage());
+ oughtToBeEE.initCause(exp);
+ throw oughtToBeEE;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.BYTEARRAY));
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.EvalFunc#getArgToFuncMapping()
+ */
+ @Override
+ public List<FuncSpec> getArgToFuncMapping() throws FrontendException {
+ List<FuncSpec> funcList = new ArrayList<FuncSpec>();
+ Schema s = new Schema();
+ s.add(new Schema.FieldSchema(null, DataType.BYTEARRAY));
+ s.add(new Schema.FieldSchema(null, DataType.BYTEARRAY));
+ funcList.add(new FuncSpec(this.getClass().getName(), s));
+ s = new Schema();
+ s.add(new Schema.FieldSchema(null, DataType.CHARARRAY));
+ s.add(new Schema.FieldSchema(null, DataType.CHARARRAY));
+ funcList.add(new FuncSpec(StringConcat.class.getName(), s));
+ return funcList;
+ }
+}
Modified: incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java?rev=676185&r1=676184&r2=676185&view=diff
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java (original)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/COUNT.java Sat Jul 12 08:26:35 2008
@@ -35,6 +35,7 @@
* implemenation, so if possible the execution will be split into a local and global functions
*/
public class COUNT extends EvalFunc<Long> implements Algebraic{
+ private static TupleFactory mTupleFactory = TupleFactory.getInstance();
@Override
public Long exec(Tuple input) throws IOException {
@@ -60,12 +61,11 @@
}
static public class Initial extends EvalFunc<Tuple> {
- TupleFactory tfact = TupleFactory.getInstance();
@Override
public Tuple exec(Tuple input) throws IOException {
try {
- return tfact.newTuple(count(input));
+ return mTupleFactory.newTuple(count(input));
} catch (ExecException ee) {
IOException oughtToBeEE = new IOException();
oughtToBeEE.initCause(ee);
@@ -75,12 +75,11 @@
}
static public class Intermed extends EvalFunc<Tuple> {
- TupleFactory tfact = TupleFactory.getInstance();
@Override
public Tuple exec(Tuple input) throws IOException {
try {
- return tfact.newTuple(count(input));
+ return mTupleFactory.newTuple(count(input));
} catch (ExecException ee) {
IOException oughtToBeEE = new IOException();
oughtToBeEE.initCause(ee);
@@ -130,6 +129,4 @@
return new Schema(new Schema.FieldSchema(null, DataType.LONG));
}
- private static int count = 1;
-
}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/ConstantSize.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/ConstantSize.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/ConstantSize.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/ConstantSize.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,51 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+
+/**
+ * Generates the size of the first field of a tuple.
+ */
+public class ConstantSize extends EvalFunc<Long> {
+
+ @Override
+ public Long exec(Tuple input) throws IOException {
+ try {
+ return input.get(0) == null ? null : new Long(1);
+ } catch (ExecException exp) {
+ IOException oughtToBeEE = new IOException("Error processing: " +
+ input.toString() + exp.getMessage());
+ oughtToBeEE.initCause(exp);
+ throw oughtToBeEE;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.LONG));
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleAvg.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleAvg.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleAvg.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleAvg.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,207 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.FuncSpec;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.backend.executionengine.ExecException;
+
+
+/**
+ * Generates the average of the values of the first field of a tuple. This class is Algebraic in
+ * implemenation, so if possible the execution will be split into a local and global application
+ */
+public class DoubleAvg extends EvalFunc<Double> implements Algebraic {
+
+ private static TupleFactory mTupleFactory = TupleFactory.getInstance();
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ Double sum = sum(input);
+ if(sum == null) {
+ // either we were handed an empty bag or a bag
+ // filled with nulls - return null in this case
+ return null;
+ }
+ double count = count(input);
+
+ Double avg = null;
+ if (count > 0)
+ avg = new Double(sum / count);
+
+ return avg;
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Intermed.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ Tuple t = mTupleFactory.newTuple(2);
+ t.set(0, sum(input));
+ t.set(1, new Long(count(input)));
+ return t;
+ } catch(RuntimeException t) {
+ throw new RuntimeException(t.getMessage() + ": " + input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+
+ }
+ }
+
+ static public class Intermed extends EvalFunc<Tuple> {
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ DataBag b = (DataBag)input.get(0);
+ return combine(b);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ DataBag b = (DataBag)input.get(0);
+ Tuple combined = combine(b);
+
+ Double sum = (Double)combined.get(0);
+ if(sum == null) {
+ return null;
+ }
+ double count = (Long)combined.get(1);
+
+ Double avg = null;
+ if (count > 0) {
+ avg = new Double(sum / count);
+ }
+ return avg;
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Tuple combine(DataBag values) throws ExecException {
+ double sum = 0;
+ long count = 0;
+
+ // combine is called from Intermediate and Final
+ // In either case, Initial would have been called
+ // before and would have sent in valid tuples
+ // Hence we don't need to check if incoming bag
+ // is empty
+
+ Tuple output = mTupleFactory.newTuple(2);
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ Double d = (Double)t.get(0);
+ if(d == null) continue;
+ sawNonNull = true;
+ sum += d;
+ count += (Long)t.get(1);
+ }
+ if(sawNonNull) {
+ output.set(0, new Double(sum));
+ } else {
+ output.set(0, null);
+ }
+ output.set(1, new Long(count));
+ return output;
+ }
+
+ static protected long count(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+ return values.size();
+ }
+
+ static protected Double sum(Tuple input) throws ExecException, IOException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ if(values.size() == 0) {
+ return null;
+ }
+
+ double sum = 0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try{
+ Double d = (Double)(t.get(0));
+ if (d == null) continue;
+ sawNonNull = true;
+ sum += d;
+ }catch(RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Double(sum);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMax.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMax.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMax.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMax.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,126 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+/**
+ * Generates the max of the values of the first field of a tuple.
+ */
+public class DoubleMax extends EvalFunc<Double> implements Algebraic {
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return max(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ private static TupleFactory tfact = TupleFactory.getInstance();
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ return tfact.newTuple(max(input));
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return max(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Double max(Tuple input) throws ExecException {
+
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ double curMax = Double.NEGATIVE_INFINITY;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try {
+ Double d = (Double)(t.get(0));
+ if (d == null) continue;
+ sawNonNull = true;
+ curMax = java.lang.Math.max(curMax, d);
+ } catch (RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Double(curMax);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMin.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMin.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMin.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleMin.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+/**
+ * Generates the min of the Double values in the first field of a tuple.
+ */
+public class DoubleMin extends EvalFunc<Double> implements Algebraic {
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return min(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ private static TupleFactory tfact = TupleFactory.getInstance();
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ return tfact.newTuple(min(input));
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return min(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Double min(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ double curMin = Double.POSITIVE_INFINITY;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try {
+ Double d = (Double)(t.get(0));
+ if (d == null) continue;
+ sawNonNull = true;
+ curMin = java.lang.Math.min(curMin, d);
+ } catch (RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Double(curMin);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleSum.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleSum.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleSum.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/DoubleSum.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,128 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.FuncSpec;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+
+/**
+ * Generates the sum of the values of the first field of a tuple.
+ */
+public class DoubleSum extends EvalFunc<Double> implements Algebraic {
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return sum(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ private static TupleFactory tfact = TupleFactory.getInstance();
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ return tfact.newTuple(sum(input));
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return sum(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Double sum(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ double sum = 0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try {
+ Double d = (Double)(t.get(0));
+ if (d == null) continue;
+ sawNonNull = true;
+ sum += d;
+ }catch(RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Double(sum);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/FloatAvg.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/FloatAvg.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/FloatAvg.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/FloatAvg.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,205 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.backend.executionengine.ExecException;
+
+
+/**
+ * Generates the average of the values of the first field of a tuple. This class is Algebraic in
+ * implementation, so if possible the execution will be split into a local and global application
+ */
+public class FloatAvg extends EvalFunc<Double> implements Algebraic {
+
+ private static TupleFactory mTupleFactory = TupleFactory.getInstance();
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ Double sum = sum(input);
+ if(sum == null) {
+ // either we were handed an empty bag or a bag
+ // filled with nulls - return null in this case
+ return null;
+ }
+ double count = count(input);
+
+ Double avg = null;
+ if (count > 0)
+ avg = new Double(sum / count);
+
+ return avg;
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Intermed.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ Tuple t = mTupleFactory.newTuple(2);
+ t.set(0, sum(input));
+ t.set(1, new Long(count(input)));
+ return t;
+ } catch(RuntimeException t) {
+ throw new RuntimeException(t.getMessage() + ": " + input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+
+ }
+ }
+
+ static public class Intermed extends EvalFunc<Tuple> {
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ DataBag b = (DataBag)input.get(0);
+ return combine(b);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ DataBag b = (DataBag)input.get(0);
+ Tuple combined = combine(b);
+
+ Double sum = (Double)combined.get(0);
+ if(sum == null) {
+ return null;
+ }
+ double count = (Long)combined.get(1);
+
+ Double avg = null;
+ if (count > 0) {
+ avg = new Double(sum / count);
+ }
+ return avg;
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Tuple combine(DataBag values) throws ExecException {
+ double sum = 0;
+ long count = 0;
+
+ // combine is called from Intermediate and Final
+ // In either case, Initial would have been called
+ // before and would have sent in valid tuples
+ // Hence we don't need to check if incoming bag
+ // is empty
+
+ Tuple output = mTupleFactory.newTuple(2);
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ Double d = (Double)t.get(0);
+ if(d == null) continue;
+ sawNonNull = true;
+ sum += d;
+ count += (Long)t.get(1);
+ }
+ if(sawNonNull) {
+ output.set(0, new Double(sum));
+ } else {
+ output.set(0, null);
+ }
+ output.set(1, new Long(count));
+ return output;
+ }
+
+ static protected long count(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+ return values.size();
+ }
+
+ static protected Double sum(Tuple input) throws ExecException, IOException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ if(values.size() == 0) {
+ return null;
+ }
+
+ double sum = 0.0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try {
+ Float f = (Float)(t.get(0));
+ if (f == null) continue;
+ sawNonNull = true;
+ sum += f;
+ }catch(RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Double(sum);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMax.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMax.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMax.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMax.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+/**
+ * Generates the max of the values of the first field of a tuple.
+ */
+public class FloatMax extends EvalFunc<Float> implements Algebraic {
+
+ @Override
+ public Float exec(Tuple input) throws IOException {
+ try {
+ return max(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ private static TupleFactory tfact = TupleFactory.getInstance();
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ return tfact.newTuple(max(input));
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+ static public class Final extends EvalFunc<Float> {
+ @Override
+ public Float exec(Tuple input) throws IOException {
+ try {
+ return max(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Float max(Tuple input) throws ExecException {
+ DataBag values = (DataBag)(input.get(0));
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ float curMax = Float.NEGATIVE_INFINITY;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = (Tuple)it.next();
+ try {
+ Float f = (Float)(t.get(0));
+ if (f == null) continue;
+ sawNonNull = true;
+ curMax = java.lang.Math.max(curMax, f);
+ } catch (RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Float(curMax);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.FLOAT));
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMin.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMin.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMin.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/FloatMin.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+/**
+ * Generates the min of the Float values in the first field of a tuple.
+ */
+public class FloatMin extends EvalFunc<Float> implements Algebraic {
+
+ @Override
+ public Float exec(Tuple input) throws IOException {
+ try {
+ return min(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ private static TupleFactory tfact = TupleFactory.getInstance();
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ return tfact.newTuple(min(input));
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+ static public class Final extends EvalFunc<Float> {
+ @Override
+ public Float exec(Tuple input) throws IOException {
+ try {
+ return min(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Float min(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ float curMin = Float.POSITIVE_INFINITY;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try {
+ Float f = (Float)(t.get(0));
+ if (f == null) continue;
+ sawNonNull = true;
+ curMin = java.lang.Math.min(curMin, f);
+ } catch (RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Float(curMin);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.FLOAT));
+ }
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/FloatSum.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,127 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+
+
+/**
+ * Generates the sum of the Float values in the first field of a tuple.
+ */
+public class FloatSum extends EvalFunc<Double> implements Algebraic {
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return sum(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Initial.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ private static TupleFactory tfact = TupleFactory.getInstance();
+
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ return tfact.newTuple(sum(input));
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ return sum(input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Double sum(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ // this is in compliance with SQL standard
+ if(values.size() == 0) {
+ return null;
+ }
+
+ double sum = 0.0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = (Tuple) it.next();
+ try {
+ Float f = (Float)(t.get(0));
+ if (f == null) continue;
+ sawNonNull = true;
+ sum += f;
+ }catch(RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+
+ if(sawNonNull) {
+ return new Double(sum);
+ } else {
+ return null;
+ }
+
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+
+}
Added: incubator/pig/branches/types/src/org/apache/pig/builtin/IntAvg.java
URL: http://svn.apache.org/viewvc/incubator/pig/branches/types/src/org/apache/pig/builtin/IntAvg.java?rev=676185&view=auto
==============================================================================
--- incubator/pig/branches/types/src/org/apache/pig/builtin/IntAvg.java (added)
+++ incubator/pig/branches/types/src/org/apache/pig/builtin/IntAvg.java Sat Jul 12 08:26:35 2008
@@ -0,0 +1,205 @@
+/*
+ * 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.builtin;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.pig.Algebraic;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.backend.executionengine.ExecException;
+
+
+/**
+ * Generates the average of the values of the first field of a tuple. This class is Algebraic in
+ * implementation, so if possible the execution will be split into a local and global application
+ */
+public class IntAvg extends EvalFunc<Double> implements Algebraic {
+
+ private static TupleFactory mTupleFactory = TupleFactory.getInstance();
+
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ Long sum = sum(input);
+ if(sum == null) {
+ // either we were handed an empty bag or a bag
+ // filled with nulls - return null in this case
+ return null;
+ }
+ double count = count(input);
+
+ Double avg = null;
+ if (count > 0)
+ avg = new Double(sum / count);
+
+ return avg;
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+
+ public String getInitial() {
+ return Initial.class.getName();
+ }
+
+ public String getIntermed() {
+ return Intermed.class.getName();
+ }
+
+ public String getFinal() {
+ return Final.class.getName();
+ }
+
+ static public class Initial extends EvalFunc<Tuple> {
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ Tuple t = mTupleFactory.newTuple(2);
+ t.set(0, sum(input));
+ t.set(1, new Long(count(input)));
+ return t;
+ } catch(RuntimeException t) {
+ throw new RuntimeException(t.getMessage() + ": " + input);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+
+ }
+ }
+
+ static public class Intermed extends EvalFunc<Tuple> {
+ @Override
+ public Tuple exec(Tuple input) throws IOException {
+ try {
+ DataBag b = (DataBag)input.get(0);
+ return combine(b);
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static public class Final extends EvalFunc<Double> {
+ @Override
+ public Double exec(Tuple input) throws IOException {
+ try {
+ DataBag b = (DataBag)input.get(0);
+ Tuple combined = combine(b);
+
+ Long sum = (Long)combined.get(0);
+ if(sum == null) {
+ return null;
+ }
+ double count = (Long)combined.get(1);
+
+ Double avg = null;
+ if (count > 0) {
+ avg = new Double(sum / count);
+ }
+ return avg;
+ } catch (ExecException ee) {
+ IOException oughtToBeEE = new IOException();
+ oughtToBeEE.initCause(ee);
+ throw oughtToBeEE;
+ }
+ }
+ }
+
+ static protected Tuple combine(DataBag values) throws ExecException {
+ long sum = 0;
+ long count = 0;
+
+ // combine is called from Intermediate and Final
+ // In either case, Initial would have been called
+ // before and would have sent in valid tuples
+ // Hence we don't need to check if incoming bag
+ // is empty
+
+ Tuple output = mTupleFactory.newTuple(2);
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ Long l = (Long)t.get(0);
+ if(l == null) continue;
+ sawNonNull = true;
+ sum += l;
+ count += (Long)t.get(1);
+ }
+ if(sawNonNull) {
+ output.set(0, new Long(sum));
+ } else {
+ output.set(0, null);
+ }
+ output.set(1, new Long(count));
+ return output;
+ }
+
+ static protected long count(Tuple input) throws ExecException {
+ DataBag values = (DataBag)input.get(0);
+ return values.size();
+ }
+
+ static protected Long sum(Tuple input) throws ExecException, IOException {
+ DataBag values = (DataBag)input.get(0);
+
+ // if we were handed an empty bag, return NULL
+ if(values.size() == 0) {
+ return null;
+ }
+
+ long sum = 0;
+ boolean sawNonNull = false;
+ for (Iterator<Tuple> it = values.iterator(); it.hasNext();) {
+ Tuple t = it.next();
+ try {
+ Integer i = (Integer)(t.get(0));
+ if (i == null) continue;
+ sawNonNull = true;
+ sum += i;
+ }catch(RuntimeException exp) {
+ ExecException newE = new ExecException("Error processing: " +
+ t.toString() + exp.getMessage(), exp);
+ throw newE;
+ }
+ }
+
+ if(sawNonNull) {
+ return new Long(sum);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public Schema outputSchema(Schema input) {
+ return new Schema(new Schema.FieldSchema(null, DataType.DOUBLE));
+ }
+
+}