/**
 * Copyright (C) 2014-2015 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 java.util.HashMap;
import java.util.Map;

import org.apache.avro.Schema.Type;
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.Schema;
import com.linkedin.pinot.core.data.GenericRow;
import com.linkedin.pinot.core.data.readers.AvroRecordReader;


public class AvroRecordToPinotRowGenerator {
  private final Schema indexingSchema;

  public AvroRecordToPinotRowGenerator(Schema indexingSchema) {
    this.indexingSchema = indexingSchema;
  }

  public GenericRow transform(GenericData.Record record, org.apache.avro.Schema schema) {
    Map<String, Object> rowEntries = new HashMap<String, Object>();
    for (String column : indexingSchema.getColumnNames()) {
      Object entry = record.get(column);
      if (entry instanceof Utf8) {
        entry = ((Utf8) entry).toString();
      }

      if (schema.getField(column).schema().getType() == Type.ENUM
          || schema.getField(column).schema().getType() == Type.BOOLEAN) {
        entry = entry.toString();
      }

      if (entry instanceof Array) {
        entry = AvroRecordReader.transformAvroArrayToObjectArray((Array) entry);
      }
      rowEntries.put(column, entry);
    }

    GenericRow row = new GenericRow();
    row.init(rowEntries);
    return row;
  }

  public GenericRow transform(GenericRecord avroRecord) {
    Map<String, Object> rowEntries = new HashMap<String, Object>();
    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);
      }
      rowEntries.put(column, entry);
    }

    GenericRow row = new GenericRow();
    row.init(rowEntries);
    return row;
  }
}

@@ -1577,36 +1577,37 @@
 ntry instanceof 
-Utf8
+Array
 ) %7B%0A        entr
@@ -1614,44 +1614,75 @@
 y = 
-((Utf8) entry).toString(
+AvroRecordReader.transformAvroArrayToObjectArray((Array) entry
 );%0A  
-    %7D%0A%0A
     
@@ -1743,16 +1743,18 @@
 pe.ENUM%0A
+  
         
@@ -1832,21 +1832,121 @@
     
+ 
-entry =
+ for (int i = 0; i %3C ((Object%5B%5D) entry).length; ++i) %7B%0A            ((Object%5B%5D) entry)%5Bi%5D = ((Object%5B%5D)
  entry
+)%5Bi%5D
 .toS
@@ -1956,27 +1956,57 @@
 ng();%0A      
+ 
-%7D%0A%0A
+   %7D%0A        %7D%0A      %7D else %7B%0A  
       if (en
@@ -2016,33 +2016,34 @@
  instanceof 
-Array
+Utf8
 ) %7B%0A
+  
         entr
@@ -2050,72 +2050,235 @@
 y = 
-AvroRecordReader.transformAvroArrayToObjectArray((Array) entry);
+((Utf8) entry).toString();%0A        %7D%0A        if (schema.getField(column).schema().getType() == Type.ENUM%0A            %7C%7C schema.getField(column).schema().getType() == Type.BOOLEAN) %7B%0A          entry = entry.toString();%0A        %7D
 %0A   

Adding enum to string converter for Object[].