/*
 * 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.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.hive.serde2.objectinspector.primitive.BooleanObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
import org.apache.hadoop.io.BooleanWritable;

/** A UDF that customizes the array contains to take into account case sensitivity. */
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;
    }

    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.equalsIgnoreCase(value);

        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] instanceof StringObjectInspector)) {
      throw new UDFArgumentException("stringArrayContains takes a string as second argument");
    }
    if (!(arguments[2] instanceof BooleanObjectInspector)) {
      throw new UDFArgumentException("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 PrimitiveObjectInspectorFactory.writableBooleanObjectInspector;
  }
}

@@ -598,32 +598,8 @@
 f;%0A%0A
-import java.util.List;%0A%0A
 impo
@@ -768,16 +768,90 @@
 ricUDF;%0A
+import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;%0A
 import o
@@ -995,244 +995,8 @@
 ry;%0A
-import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;%0Aimport org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;%0A
 impo
@@ -1472,72 +1472,22 @@
 ate 
-StandardListObjectInspector retValInspector;%0A  private Primitive
+transient List
 Obje
@@ -1498,25 +1498,20 @@
 spector 
-primitive
+list
 ObjectIn
@@ -1650,51 +1650,23 @@
     
-Lis
+Objec
 t array
-Values = retValInspector.getList(
+ = 
 argu
@@ -1679,17 +1679,16 @@
 0%5D.get()
-)
 ;%0A    St
@@ -1830,22 +1830,16 @@
 rray
-Values
  == null
  %7C%7C 
@@ -1838,250 +1838,278 @@
 null
- %7C%7C arrayValues.isEmpty()) %7B%0A      result.set(false);%0A      return result;%0A    %7D%0A%0A    for (Object oElement : arrayValues) %7B%0A      Object stdObject =%0A          ObjectInspectorUtils.copyToStandardJavaObject(oElement, primitiveObjectInspector
+) %7B%0A      return result;%0A    %7D%0A%0A    int arrayLength = listObjectInspector.getListLength(array);%0A    if (arrayLength %3C= 0) %7B%0A      return result;%0A    %7D%0A%0A    for (int i = 0; i %3C arrayLength; ++i) %7B%0A      Object listElement = listObjectInspector.getListElement(array, i
 );%0A
+%0A
     
@@ -2114,24 +2114,26 @@
     if (
-stdObjec
+listElemen
 t != nul
@@ -2142,27 +2142,29 @@
 && !
-((String) stdObject
+listElement.toString(
 ).tr
@@ -2200,39 +2200,45 @@
 ing 
-array
+string
 Val
+ue
  = 
-(String) stdObject;%0A
+listElement.toString();
 %0A   
@@ -2257,16 +2257,28 @@
  equal =
+%0A           
  caseSen
@@ -2286,24 +2286,27 @@
 itive ? 
-array
+string
 Val
+ue
 .equals(
@@ -2318,16 +2318,19 @@
 ) : 
-array
+string
 Val
+ue
 .equ
@@ -3420,346 +3420,65 @@
     
-retValInspector =%0A        (StandardListObjectInspector) ObjectInspectorUtils.getStandardObjectInspector(arguments%5B0%5D);%0A    if (retValInspector.getListElementObjectInspector().getCategory() != Category.PRIMITIVE) %7B%0A      primitiveObjectInspector =%0A          (PrimitiveObjectInspector) retValInspector.getListElementObjectInspector();%0A    %7D
+listObjectInspector = (ListObjectInspector) arguments%5B0%5D;
 %0A%0A  

fix StringArrayContainsGenericUDF