/**
 * Copyright (C) 2014-2016 LinkedIn Corp. (pinot-core@linkedin.com)
 *
 * 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 com.linkedin.pinot.core.realtime.impl.kafka;

import com.linkedin.pinot.common.data.TimeFieldSpec;
import java.util.HashMap;
import java.util.Map;

import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericData.Array;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.util.Utf8;

import com.linkedin.pinot.common.data.FieldSpec;
import com.linkedin.pinot.common.data.FieldSpec.DataType;
import com.linkedin.pinot.common.data.Schema;
import com.linkedin.pinot.core.data.GenericRow;
import com.linkedin.pinot.core.data.readers.AvroRecordReader;


public class AvroRecordToPinotRowGenerator {
  private final Schema indexingSchema;
  private final String timeColumnName;
  private final FieldSpec timeFieldSpec;

  public AvroRecordToPinotRowGenerator(Schema indexingSchema) {
    this.indexingSchema = indexingSchema;
    timeColumnName = indexingSchema.getTimeFieldSpec().getIncomingTimeColumnName();
    if (indexingSchema.getTimeColumnName().equals(timeColumnName)) {
      timeFieldSpec = new TimeFieldSpec(indexingSchema.getTimeFieldSpec().getIncomingGranularitySpec());
    } else {
      timeFieldSpec = indexingSchema.getTimeFieldSpec();
    }
  }

  public GenericRow transform(GenericData.Record record, org.apache.avro.Schema schema, GenericRow destination) {
    FieldSpec fieldSpec;
    for (String column : indexingSchema.getColumnNames()) {
      if (column.equals(indexingSchema.getTimeColumnName())
          && (!column.equals(indexingSchema.getTimeFieldSpec().getIncomingTimeColumnName()))){
        column = timeColumnName;
        fieldSpec = timeFieldSpec;
      } else {
        fieldSpec = indexingSchema.getFieldSpecFor(column);
      }
      Object entry = record.get(column);

      if (entry != null) {
        if (entry instanceof Array) {
          entry = AvroRecordReader.transformAvroArrayToObjectArray((Array) entry, fieldSpec);
          if (fieldSpec.getDataType() == DataType.STRING || fieldSpec.getDataType() == DataType.STRING_ARRAY) {
            for (int i = 0; i < ((Object[]) entry).length; ++i) {
              if (((Object[]) entry)[i] != null) {
                ((Object[]) entry)[i] = ((Object[]) entry)[i].toString();
              }
            }
          }
        } else {
          if (entry instanceof Utf8) {
            entry = ((Utf8) entry).toString();
          }
          if (fieldSpec.getDataType() == DataType.STRING) {
            entry = entry.toString();
          }
        }
      } else {
        // entry was null.
        if (fieldSpec.isSingleValueField()) {
          entry = AvroRecordReader.getDefaultNullValue(fieldSpec);
        } else {
          // A multi-value field, and null. Any of the instanceof checks above will not match, so we need to repeat some
          // of the logic above here.
          entry = AvroRecordReader.transformAvroArrayToObjectArray((Array) entry, fieldSpec);
          if (fieldSpec.getDataType() == DataType.STRING || fieldSpec.getDataType() == DataType.STRING_ARRAY) {
            for (int i = 0; i < ((Object[]) entry).length; ++i) {
              if (((Object[]) entry)[i] != null) {
                ((Object[]) entry)[i] = ((Object[]) entry)[i].toString();
              }
            }
          }
        }
      }
      destination.putField(column, entry);
    }

    return destination;
  }

  public GenericRow transform(GenericRecord avroRecord, GenericRow destination) {
    for (String column : indexingSchema.getColumnNames()) {
      Object entry = avroRecord.get(column);
      if (entry instanceof Utf8) {
        entry = ((Utf8) entry).toString();
      }
      if (entry instanceof Array) {
        entry = AvroRecordReader.transformAvroArrayToObjectArray((Array) entry, indexingSchema.getFieldSpecFor(column));
      }
      if (entry == null && indexingSchema.getFieldSpecFor(column).isSingleValueField()) {
        entry = AvroRecordReader.getDefaultNullValue(indexingSchema.getFieldSpecFor(column));
      }
      destination.putField(column, entry);
    }

    return destination;
  }
}

@@ -741,56 +741,8 @@
 pec;
-%0Aimport java.util.HashMap;%0Aimport java.util.Map;
 %0A%0Aim
@@ -1286,17 +1286,25 @@
  String 
-t
+incomingT
 imeColum
@@ -1336,17 +1336,25 @@
 eldSpec 
-t
+incomingT
 imeField
@@ -1470,17 +1470,25 @@
 ma;%0A    
-t
+incomingT
 imeColum
@@ -1608,17 +1608,25 @@
 .equals(
-t
+incomingT
 imeColum
@@ -1633,33 +1633,41 @@
 nName)) %7B%0A      
-t
+incomingT
 imeFieldSpec = n
@@ -1767,17 +1767,25 @@
 %7B%0A      
-t
+incomingT
 imeField
@@ -2079,73 +2079,8 @@
 ema.
-getTimeColumnName())%0A          && (!column.equals(indexingSchema.
 getT
@@ -2089,37 +2089,37 @@
 eFieldSpec().get
-Incom
+Outgo
 ingTimeColumnNam
@@ -2123,17 +2123,17 @@
 Name()))
-)
+ 
 %7B%0A      
@@ -2143,17 +2143,25 @@
 olumn = 
-t
+incomingT
 imeColum
@@ -2187,17 +2187,25 @@
 dSpec = 
-t
+incomingT
 imeField

Address comments for outgoing time column (#1389)