content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You 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.apache.carbondata.hive;
  
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import javax.annotation.Nullable;
  
  import org.apache.carbondata.common.logging.LogServiceFactory;
  import org.apache.carbondata.core.datastore.impl.FileFactory;
  import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.metadata.schema.SchemaReader;
  import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
  import org.apache.carbondata.core.metadata.schema.table.TableInfo;
  import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
  import org.apache.carbondata.core.util.CarbonUtil;
  import org.apache.carbondata.core.util.path.CarbonTablePath;
  
  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
  import org.apache.hadoop.hive.ql.io.parquet.serde.ArrayWritableObjectInspector;
  import org.apache.hadoop.hive.serde.serdeConstants;
  import org.apache.hadoop.hive.serde2.AbstractSerDe;
  import org.apache.hadoop.hive.serde2.SerDeException;
  import org.apache.hadoop.hive.serde2.SerDeSpec;
  import org.apache.hadoop.hive.serde2.SerDeStats;
  import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
  import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
  import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
  import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
  import org.apache.hadoop.hive.serde2.objectinspector.StructField;
  import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
  import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
  import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
  import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
  import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
  import org.apache.hadoop.io.ArrayWritable;
  import org.apache.hadoop.io.BytesWritable;
  import org.apache.hadoop.io.Writable;
  import org.apache.log4j.Logger;
  
  /**
   * A serde class for Carbondata.
   * It transparently passes the object to/from the Carbon file reader/writer.
   */
  @SerDeSpec(schemaProps = { serdeConstants.LIST_COLUMNS, serdeConstants.LIST_COLUMN_TYPES })
  public class CarbonHiveSerDe extends AbstractSerDe {
  
    private static final Logger LOGGER =
        LogServiceFactory.getLogService(CarbonHiveSerDe.class.getCanonicalName());
  
    private final SerDeStats stats;
    private ObjectInspector objInspector;
  
    private enum LAST_OPERATION {
      SERIALIZE, DESERIALIZE, UNKNOWN
    }
  
    private LAST_OPERATION status;
    private long serializedSize;
    private long deserializedSize;
  
    public CarbonHiveSerDe() {
      stats = new SerDeStats();
    }
  
    @Override
    public void initialize(@Nullable Configuration configuration, Properties tbl)
        throws SerDeException {
      final TypeInfo rowTypeInfo;
      final List<String> columnNames;
      final List<TypeInfo> columnTypes;
      // Get column names and sort order
      assert configuration != null;
      final String columnNameProperty = tbl.getProperty(serdeConstants.LIST_COLUMNS);
      final String columnTypeProperty = tbl.getProperty(serdeConstants.LIST_COLUMN_TYPES);
      if (columnNameProperty.length() == 0) {
        columnNames = new ArrayList<String>();
      } else {
        columnNames = Arrays.asList(columnNameProperty.split(","));
      }
      if (columnTypeProperty.length() == 0) {
        columnTypes = new ArrayList<TypeInfo>();
      } else {
        columnTypes = TypeInfoUtils.getTypeInfosFromTypeString(columnTypeProperty);
      }
      inferSchema(tbl, columnNames, columnTypes);
  
      // Create row related objects
      rowTypeInfo = TypeInfoFactory.getStructTypeInfo(columnNames, columnTypes);
      this.objInspector = new ArrayWritableObjectInspector((StructTypeInfo) rowTypeInfo);
      // Stats part
      serializedSize = 0;
      deserializedSize = 0;
      status = LAST_OPERATION.UNKNOWN;
    }
  
    private void inferSchema(Properties tbl, List<String> columnNames, List<TypeInfo> columnTypes) {
      if (columnNames.size() == 0 && columnTypes.size() == 0) {
        String external = tbl.getProperty("EXTERNAL");
        String location = CarbonUtil.checkAndAppendFileSystemURIScheme(
            tbl.getProperty(hive_metastoreConstants.META_TABLE_LOCATION));
        if (external != null && "TRUE".equals(external) && location != null) {
          String[] names =
              tbl.getProperty(hive_metastoreConstants.META_TABLE_NAME).split("\\.");
          if (names.length == 2) {
            AbsoluteTableIdentifier identifier =
                AbsoluteTableIdentifier.from(location, names[0], names[1]);
            String schemaPath = CarbonTablePath.getSchemaFilePath(identifier.getTablePath());
            try {
              TableInfo tableInfo = null;
              if (!FileFactory.isFileExist(schemaPath)) {
                tableInfo = SchemaReader.inferSchema(identifier, false);
              } else {
                tableInfo = SchemaReader.getTableInfo(identifier);
              }
              if (tableInfo != null) {
                CarbonTable carbonTable = CarbonTable.buildFromTableInfo(tableInfo);
                List<CarbonColumn> columns = carbonTable.getCreateOrderColumn();
                for (CarbonColumn column : columns) {
                  columnNames.add(column.getColName());
                  columnTypes.add(HiveDataTypeUtils.convertCarbonDataTypeToHive(column));
                }
              }
            } catch (Exception ex) {
              LOGGER.warn("Failed to infer schema: " + ex.getMessage());
            }
          }
        }
      }
    }
  
    @Override
    public Class<? extends Writable> getSerializedClass() {
      return ArrayWritable.class;
    }
  
    @Override
    public Writable serialize(Object obj, ObjectInspector objectInspector)
        throws SerDeException {
      if (!objInspector.getCategory().equals(ObjectInspector.Category.STRUCT)) {
        throw new SerDeException("Cannot serializeStartKey " + objInspector.getCategory()
            + ". Can only serializeStartKey a struct");
      }
      serializedSize += ((StructObjectInspector) objInspector).getAllStructFieldRefs().size();
      status = LAST_OPERATION.SERIALIZE;
      return createCarbonRow(obj, (StructObjectInspector) objectInspector);
    }
  
    private CarbonHiveRow createCarbonRow(Object obj, StructObjectInspector inspector)
        throws SerDeException {
      List fields = inspector.getAllStructFieldRefs();
      CarbonHiveRow carbonHiveRow = new CarbonHiveRow();
      for (int i = 0; i < fields.size(); i++) {
        StructField field = (StructField) fields.get(i);
        Object subObj = inspector.getStructFieldData(obj, field);
        ObjectInspector subInspector = field.getFieldObjectInspector();
        carbonHiveRow.addToRow(createObject(subObj, subInspector));
      }
      return carbonHiveRow;
    }
  
    private Object[] createStruct(Object obj, StructObjectInspector inspector)
        throws SerDeException {
      List fields = inspector.getAllStructFieldRefs();
      Object[] arr = new Object[fields.size()];
      for (int i = 0; i < fields.size(); i++) {
        StructField field = (StructField) fields.get(i);
        Object subObj = inspector.getStructFieldData(obj, field);
        ObjectInspector subInspector = field.getFieldObjectInspector();
        arr[i] = createObject(subObj, subInspector);
      }
      return arr;
    }
  
    private Object[] createArray(Object obj, ListObjectInspector inspector)
        throws SerDeException {
      List sourceArray = inspector.getList(obj);
      ObjectInspector subInspector = inspector.getListElementObjectInspector();
      List array = new ArrayList();
      Iterator iterator;
      if (sourceArray != null) {
        for (iterator = sourceArray.iterator(); iterator.hasNext(); ) {
          Object curObj = iterator.next();
          Object newObj = createObject(curObj, subInspector);
          if (newObj != null) {
            array.add(newObj);
          }
        }
      }
      return array.toArray();
    }
  
    private Object createPrimitive(Object obj, PrimitiveObjectInspector inspector)
        throws SerDeException {
      if (obj == null) {
        return null;
      }
      if (inspector.getTypeInfo().getTypeName().equalsIgnoreCase(DataTypes.BINARY.getName())) {
        BytesWritable primitiveWritableObject =
            (BytesWritable) inspector.getPrimitiveWritableObject(obj);
        byte[] bytes = primitiveWritableObject.getBytes();
        int length = primitiveWritableObject.getLength();
        return Arrays.copyOfRange(bytes, 0, length);
      }
      return inspector.getPrimitiveWritableObject(obj).toString();
    }
  
    private Object createObject(Object obj, ObjectInspector inspector) throws SerDeException {
      switch (inspector.getCategory()) {
        case STRUCT:
          return createStruct(obj, (StructObjectInspector) inspector);
        case LIST:
          return createArray(obj, (ListObjectInspector) inspector);
        case PRIMITIVE:
          return createPrimitive(obj, (PrimitiveObjectInspector) inspector);
        case MAP:
          return createMap(obj, (MapObjectInspector) inspector);
      }
      throw new SerDeException("Unknown data type" + inspector.getCategory());
    }
  
    private Object[] createMap(Object obj, MapObjectInspector inspector) throws SerDeException {
      Map<Writable, Writable> map = (Map<Writable, Writable>) inspector.getMap(obj);
      Object[] result = new Object[map.size()];
      int i = 0;
      for (Map.Entry<Writable, Writable> entry : map.entrySet()) {
        result[i++] =
            new Object[] { createObject(entry.getKey(), inspector.getMapKeyObjectInspector()),
                createObject(entry.getValue(), inspector.getMapValueObjectInspector()) };
      }
      return result;
    }
  
    @Override
    public SerDeStats getSerDeStats() {
      // must be different
      assert (status != LAST_OPERATION.UNKNOWN);
      if (status == LAST_OPERATION.SERIALIZE) {
        stats.setRawDataSize(serializedSize);
      } else {
        stats.setRawDataSize(deserializedSize);
      }
      return stats;
    }
  
    @Override
    public Object deserialize(Writable writable) {
      status = LAST_OPERATION.DESERIALIZE;
      if (writable instanceof ArrayWritable) {
        deserializedSize += ((StructObjectInspector) objInspector).getAllStructFieldRefs().size();
        return writable;
      } else {
        return null;
      }
    }
  
    @Override
    public ObjectInspector getObjectInspector() {
      return objInspector;
    }
  }