package org.gbif.occurrence.hive.udf;

import org.gbif.common.parsers.core.OccurrenceParseResult;
import org.gbif.occurrence.processor.interpreting.TemporalInterpreter;
import org.gbif.occurrence.processor.interpreting.result.DateYearMonthDay;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.hive.ql.exec.Description;
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.ObjectInspectorConverters;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

/**
 * Parses year month and day only.
 */
@Description(
  name = "parseDate",
  value = "_FUNC_(year, month, day)")
public class DateParseUDF extends GenericUDF {

  private ObjectInspectorConverters.Converter[] converters;

  @Override
  public Object evaluate(GenericUDF.DeferredObject[] arguments) throws HiveException {
    assert arguments.length == 3;

    String year = getArgument(0, arguments);
    String month = getArgument(1, arguments);
    String day = getArgument(2, arguments);
    List<Object> result = new ArrayList<Object>(3);
    try {
      OccurrenceParseResult<DateYearMonthDay> parsed = TemporalInterpreter.interpretRecordedDate(year, month, day, null);
      if (parsed.isSuccessful() && parsed.getIssues().isEmpty()) {
        result.add(parsed.getPayload().getYear());
        result.add(parsed.getPayload().getMonth());
        result.add(parsed.getPayload().getDay());
      }
    } catch (Exception e) {
      // not much to do - indicates bad data
    }

    return result;
  }

  private String getArgument(int index, GenericUDF.DeferredObject[] arguments) throws HiveException {
    DeferredObject deferredObject = arguments[index];
    if (deferredObject == null) {
      return null;
    }

    Object convertedObject = converters[index].convert(deferredObject.get());
    return convertedObject == null ? null : convertedObject.toString();
  }

  @Override
  public String getDisplayString(String[] strings) {
    assert strings.length == 3;
    return "parseDate(" + strings[0] + ", " + strings[1] + ", " + strings[2] + ')';
  }

  @Override
  public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    if (arguments.length != 3) {
      throw new UDFArgumentException("parseDate takes three arguments");
    }

    converters = new ObjectInspectorConverters.Converter[arguments.length];
    for (int i = 0; i < arguments.length; i++) {
      converters[i] = ObjectInspectorConverters
        .getConverter(arguments[i], PrimitiveObjectInspectorFactory.writableStringObjectInspector);
    }

    return ObjectInspectorFactory
      .getStandardStructObjectInspector(Arrays.asList("year", "month", "day"),
        Arrays.<ObjectInspector>asList(PrimitiveObjectInspectorFactory.javaIntObjectInspector,
          PrimitiveObjectInspectorFactory.javaIntObjectInspector,
          PrimitiveObjectInspectorFactory.javaIntObjectInspector));
  }

}

@@ -977,16 +977,28 @@
 nth, day
+, event_date
 )%22)%0Apubl
@@ -1232,17 +1232,17 @@
 ngth == 
-3
+4
 ;%0A%0A    S
@@ -1367,24 +1367,75 @@
 arguments);%0A
+    String event_date = getArgument(3, arguments);%0A
     List%3CObj
@@ -1538,16 +1538,24 @@
 parsed =
+%0A       
  Tempora
@@ -1607,20 +1607,26 @@
 h, day, 
-null
+event_date
 );%0A     
@@ -2417,17 +2417,17 @@
 ngth == 
-3
+4
 ;%0A    re
@@ -2498,16 +2498,36 @@
 ngs%5B2%5D +
+ %22, %22 + strings%5B3%5D +
  ')';%0A  
@@ -2664,17 +2664,17 @@
 ngth != 
-3
+4
 ) %7B%0A    
@@ -2727,13 +2727,12 @@
 kes 
-three
+four
  arg
@@ -3061,23 +3061,16 @@
 rFactory
-%0A      
 .getStan
@@ -3135,10 +3135,31 @@
 day%22
-)
 ,
+ %22event_date%22), Arrays
 %0A   
@@ -3163,22 +3163,16 @@
         
-Arrays
 .%3CObject
@@ -3374,16 +3374,85 @@
 nspector
+,%0A          PrimitiveObjectInspectorFactory.javaStringObjectInspector
 ));%0A  %7D%0A

update dateparseUDF to accept a 4th arg - the eventDate as string