package org.apache.hadoop.hive.serde2.columnar;

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

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.serde2.lazy.ByteArrayRef;
import org.apache.hadoop.hive.serde2.lazy.LazyObjectBase;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;

import com.google.protobuf.Descriptors.FieldDescriptor;
import com.inmobi.dw.yoda.proto.NetworkObject.KeyLessNetworkObject;
import com.inmobi.yoda.cube.ddl.CubeDDL;

public class LazyNOBColumnarStruct extends ColumnarStructBase {
  private static final Log LOG = LogFactory.getLog(LazyNOBColumnarStruct.class);

  KeyLessNetworkObject.Builder nobBuilder = KeyLessNetworkObject.newBuilder();

  class NOBFieldInfo {
    Object value = null;
    boolean retrievedData = false;
    FieldDescriptor fd;
    KeyLessNetworkObject nob;

    protected Object getField() {
      if (!retrievedData) {
        if (fd != null && nob.hasField(fd)) {
          Object nobField = nob.getField(fd);
          value = nobField;
          if (nobField instanceof String) {
            if (CubeDDL.isMeasure(fd.getName())) {
              try {
                double val = Double.parseDouble((String) nobField);
                value = val;
              } catch (NumberFormatException e) {
                LOG.warn("Error parsing " + fd.getName() 
                    + ", value " + nob.getField(fd), e);
              }
            }
          } 
        } else {
          value = null;
        }
        retrievedData = true;
      }
      return value;
    }

    public void init(KeyLessNetworkObject nob,
        FieldDescriptor fieldDescriptor) {
      this.nob = nob;
      this.fd = fieldDescriptor;
      retrievedData  = false;
    }
  }

  private NOBFieldInfo[] fieldInfoList = null;
  private ArrayList<Object> cachedList;

  private int numCols;
  private int[] referedCols;
  public LazyNOBColumnarStruct(ObjectInspector oi, int numCols, int[] referedCols) {
    super(oi, new ArrayList<Integer>());
    this.numCols = numCols;
    fieldInfoList = new NOBFieldInfo[numCols];
    for (int i = 0; i < numCols; i++) {
      fieldInfoList[i] = new NOBFieldInfo();
    }
    this.referedCols = referedCols;
  }

  /**
   * Get one field out of the struct.
   *
   * @param fieldID
   *          The field ID
   * @return The field as a LazyObject
   */
  public Object getField(int fieldID) {
    LOG.info("Get field " + fieldID + " from lazy struct");
    Object o = fieldInfoList[fieldID].getField();
    return o;
  }


  /**
   * Get the values of the fields as an ArrayList.
   *
   * @return The values of the fields as an ArrayList.
   */
  public ArrayList<Object> getFieldsAsList() {
    if (cachedList == null) {
      cachedList = new ArrayList<Object>();
    } else {
      cachedList.clear();
    }
    for (int i = 0; i < fieldInfoList.length; i++) {
      cachedList.add(fieldInfoList[i].getField());
    }
    return cachedList;
  }

  public void init(BytesRefArrayWritable cols) {
    nobBuilder.clear();
    for (int i : referedCols) {
      BytesRefWritable bytesRef = cols.get(i);
      if (bytesRef.getLength() != 0) {
        try {
          nobBuilder.mergeFrom(bytesRef.getData(),
              bytesRef.getStart(), bytesRef.getLength());
        } catch (Exception e) {
          throw new RuntimeException("Could not initialize network object", e);
        }
      }
    }
    KeyLessNetworkObject nob = nobBuilder.build();

   // List<FieldDescriptor> fdList = new ArrayList<FieldDescriptor>();
   // fdList.addAll(nob.getDescriptorForType().getFields());
    for (int i = 0; i < numCols; i++) {
      fieldInfoList[i].init(nob, CubeDDL.getFdListMap().get(i));
    }
  }

  @Override
  protected LazyObjectBase createLazyObjectBase(ObjectInspector arg0) {
    return null;
  }

  @Override
  protected int getLength(ObjectInspector arg0, ByteArrayRef arg1, int arg2,
      int arg3) {
    return 0;
  }
}

@@ -73,33 +73,8 @@
 ist;
-%0Aimport java.util.Arrays;
 %0A%0Aim
@@ -2433,68 +2433,8 @@
 ) %7B%0A
-    LOG.info(%22Get field %22 + fieldID + %22 from lazy struct%22);%0A
     
@@ -3432,140 +3432,8 @@
 );%0A%0A
-   // List%3CFieldDescriptor%3E fdList = new ArrayList%3CFieldDescriptor%3E();%0A   // fdList.addAll(nob.getDescriptorForType().getFields());%0A
     

Remove debug logging