/*
 * Licensed 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.gbif.occurrence.hive.udf;

import java.util.List;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
import org.apache.hadoop.io.BooleanWritable;

/**
 * A simple UDF that remove nulls from a list.
 */
public class StringArrayContainsGenericUDF extends GenericUDF {

  private StandardListObjectInspector retValInspector;
  private PrimitiveObjectInspector primitiveObjectInspector;
  private BooleanWritable result;

  @Override
  public Object evaluate(DeferredObject[] arguments) throws HiveException {
    List arrayValues = retValInspector.getList(arguments[0].get());
    String value = arguments[1].get().toString();
    boolean caseSensitive = Boolean.parseBoolean(arguments[2].get().toString());

    if (arrayValues == null || arrayValues.isEmpty()) {
      result.set(false);
      return result;
    }

    String lowerValue = value.toLowerCase();
    for (Object oElement : arrayValues) {
      Object stdObject =
          ObjectInspectorUtils.copyToStandardJavaObject(oElement, primitiveObjectInspector);
      if (stdObject != null && !((String) stdObject).trim().isEmpty()) {
        String arrayVal = (String) stdObject;

        boolean equal =
            caseSensitive ? arrayVal.equals(value) : arrayVal.toLowerCase().equals(lowerValue);

        if (equal) {
          result.set(true);
          return result;
        }
      }
    }

    result.set(false);
    return result;
  }

  @Override
  public String getDisplayString(String[] children) {
    return "stringArrayContains( " + children[0] + "," + children[1] + "," + children[2] + ")";
  }

  @Override
  public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    if (arguments.length != 3) {
      throw new UDFArgumentException("stringArrayContains takes an array, a value and a boolean as argument");
    }
    if (arguments[0].getCategory() != Category.LIST) {
      throw new UDFArgumentException("stringArrayContains takes an array as first argument");
    }
    if (!arguments[1].getTypeName().equals("string")) {
      throw new UDFArgumentTypeException(0, "stringArrayContains takes a string as second argument");
    }
    if (!arguments[2].getTypeName().equals("boolean")) {
      throw new UDFArgumentTypeException(0, "stringArrayContains takes a boolean as third argument");
    }

    retValInspector = (StandardListObjectInspector) ObjectInspectorUtils.getStandardObjectInspector(arguments[0]);
    if (retValInspector.getListElementObjectInspector().getCategory() != Category.PRIMITIVE) {
      primitiveObjectInspector = (PrimitiveObjectInspector) retValInspector.getListElementObjectInspector();
    }

    result = new BooleanWritable(false);

    return retValInspector;
  }

}

@@ -1233,32 +1233,205 @@
 bjectInspector;%0A
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.BooleanObjectInspector;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;%0A
 import org.apach
@@ -3300,32 +3300,33 @@
 %0A    %7D%0A    if (!
+(
 arguments%5B1%5D.get
@@ -3325,38 +3325,41 @@
 s%5B1%5D
-.getTypeName().equals(%22string%22
+ instanceof StringObjectInspector
 )) %7B
@@ -3378,36 +3378,32 @@
  new UDFArgument
-Type
 Exception(0, %22st
@@ -3388,35 +3388,32 @@
 gumentException(
-0, 
 %22stringArrayCont
@@ -3469,16 +3469,17 @@
    if (!
+(
 argument
@@ -3486,39 +3486,42 @@
 s%5B2%5D
-.getTypeName().equals(%22boolean%22
+ instanceof BooleanObjectInspector
 )) %7B
@@ -3540,36 +3540,32 @@
  new UDFArgument
-Type
 Exception(0, %22st
@@ -3562,11 +3562,8 @@
 ion(
-0, 
 %22str

fix StringArrayContainsGenericUDF