Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions
  /*
   * 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;
 
 
Core logic for applying an accumulative/algebraic math function to a bag of doubles.
 
 public abstract class AlgebraicIntMathBase extends AlgebraicMathBase<Integerimplements Accumulator<Integer> {
 
     protected static Integer getSeed(KNOWN_OP op) {
         switch (op) {
         case return 0;
         case return .;
         case return .;
         defaultreturn null;
         }
     }
 
     private static Integer doWork(Integer arg1Integer arg2KNOWN_OP op) {
         if (arg1 == null) {
             return arg2;
         } else if (arg2 == null) {
             return arg1;
         } else {
             switch (op) {
             case return Math.max(arg1arg2);
             case return Math.min(arg1arg2);
             case return arg1 + arg2;
             defaultreturn null;
             }
         }
     }
 
     protected static Integer doTupleWork(Tuple inputKnownOpProvider opProviderthrows 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;
         }
         int sofar = AlgebraicIntMathBase.getSeed(opProvider.getOp());
         boolean sawNonNull = false;
         for (Iterator<Tupleit = values.iterator(); it.hasNext();) {
             Tuple t = it.next();
             try {
                 Integer d = (Integer)(t.get(0));
                 if (d == nullcontinue;
                 sawNonNull = true;
                 sofar = doWork(sofardopProvider.getOp());
             }catch(RuntimeException exp) {
                 int errCode = 2103;
                 throw new ExecException("Problem doing work on Doubles"errCode.exp);
             }
         }
         return sawNonNull ? sofar : null;
     }
 
     @Override
     public Integer exec(Tuple inputthrows IOException {
         try {
             return doTupleWork(input);
         } catch (ExecException ee) {
             throw ee;
         } catch (Exception e) {
             int errCode = 2106;
             throw new ExecException("Error executing function on Doubles"errCode.e);
         }
     }
 
 
    static public abstract class Intermediate extends AlgebraicMathBase.Intermediate {
        private static TupleFactory tfact = TupleFactory.getInstance();
        @Override
        public Tuple exec(Tuple inputthrows IOException {
            try {
                return .newTuple(doTupleWork(inputthis));
            } catch (ExecException ee) {
                throw ee;
            } catch (Exception e) {
                int errCode = 2106;
                throw new ExecException("Error executing function "errCode.e);
            }
        }
    }
    static public abstract class Final extends AlgebraicMathBase.Final<Integer> {
        @Override
        public Integer exec(Tuple inputthrows IOException {
            try {
                return doTupleWork(inputthis);
            } catch (ExecException ee) {
                throw ee;
            } catch (Exception e) {
                int errCode = 2106;
                throw new ExecException("Error executing function "errCode.e);
            }
        }
    }
    @Override
    public Schema outputSchema(Schema input) {
        return new Schema(new Schema.FieldSchema(null.));
    }
    /* Accumulator interface implementation*/
    private Integer intermediateVal = null;
    @Override
    public void accumulate(Tuple bthrows IOException {
        try {
            Integer curVal = doTupleWork(b);
            if (curVal == null) {
                return;
            }
            if ( == null) {
                 = getSeed(.getOp());
            }
             = doWork(curVal.getOp());
        } catch (ExecException ee) {
            throw ee;
        } catch (Exception e) {
            int errCode = 2106;
            throw new ExecException("Error executing function on Doubles"errCode.e);
        }
    }
    @Override
    public void cleanup() {
         = null;
    }
    @Override
    public Integer getValue() {
        return ;
    }
New to GrepCode? Check out our FAQ X