Untitled

package com.adition.pig.filtering.string;

import java.io.IOException;
import java.util.List;
import java.util.ArrayList;

import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.data.DataType;
import org.apache.pig.FuncSpec;
import org.apache.pig.PigWarning;
import org.apache.pig.FilterFunc;
import org.apache.pig.data.Tuple;

public class CONTAINS extends FilterFunc {
        public Boolean exec(Tuple input) throws IOException {
                if (input == null || input.size() == 0  || input.get(0) == null
                                || input.get(1) == null) {
                        warn("Ignoring input as it is 'null' or of size 0.", PigWarning.UDF_WARNING_1);
                        return false;
                }
                String str = null;
                try {
                        try {
                                str = (String)input.get(0);
                        } catch (ClassCastException e) {
                                warn("Unable to cast input " + input.get(0) + " of class " +
                                                input.get(0).getClass() + " to String.", PigWarning.UDF_WARNING_2);
                                return false;
                        }
                        try {
                                return str.contains((String)input.get(1));
                        } catch (ClassCastException e) {
                                warn("Unable to cast input " + input.get(1) + " of class " +
                                                        input.get(1).getClass() + " to String.", PigWarning.UDF_WARNING_2);
                                return false;
                        }
                } catch(Exception e) {
                        throw new IOException("Could not process input ", e);
                }
        }

        @Override
        public Schema outputSchema(Schema input) {
                return new Schema(
                                new Schema.FieldSchema(
                                        getSchemaName(
                                                this.getClass().getName().toLowerCase(),
                                                input
                                        ),
                                        DataType.BOOLEAN
                                )
                );
        }

}