import parquet.schema.GroupType; //导入依赖的package包/类
private parquet.schema.Type getType(MaterializedField field) {
  MinorType minorType = field.getType().getMinorType();
  DataMode dataMode = field.getType().getMode();
  switch(minorType) {
    case MAP:
      List<parquet.schema.Type> types = Lists.newArrayList();
      for (MaterializedField childField : field.getChildren()) {
        types.add(getType(childField));
      }
      return new GroupType(dataMode == DataMode.REPEATED ? Repetition.REPEATED : Repetition.OPTIONAL, field.getLastName(), types);
    case LIST:
      throw new UnsupportedOperationException("Unsupported type " + minorType);
    default:
      return getPrimitiveType(field);
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private void writeTuple(Tuple tuple, GroupType type) {
    for (int index = 0; index < type.getFieldCount(); index++) {
        Type fieldType = type.getType(index);
        String fieldName = fieldType.getName();
        // empty fields have to be omitted
        if (tuple.isNull(index))
            continue;
        recordConsumer.startField(fieldName, index);
        if (fieldType.isPrimitive()) {
            tuple.writePrimitiveValue(recordConsumer, index, (PrimitiveType)fieldType);
        }
        else {
            recordConsumer.startGroup();
            writeTuple(tuple.getTuple(index), fieldType.asGroupType());
            recordConsumer.endGroup();
        }
        recordConsumer.endField(fieldName, index);
    }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
/**
 * transform data in group into cells(List<cell> - > {@link org.apache.hadoop.hbase.client.Result}</>)
 * @param group
 * @return
 */
public static List<Cell> groupToCells(Group group){

    List<Cell> cells = new LinkedList<>();
    if(group != null){
        cells = new LinkedList<>();
        GroupType groupType = group.getType();
        List<Type> types = groupType.getFields();
        byte [] rowKey = group.getBinary(HConstants.ROW_KEY, 0).getBytes();

        long timestamp = group.getLong(HConstants.TIME_STAMP, 0);

        for(Type t : types){
            if(! t.getName().equals(HConstants.ROW_KEY) && ! t.getName().equals(HConstants.TIME_STAMP)){
                String name = t.getName();
                String [] names = name.split(":");
                if(names.length == 2) {
                    byte[] value = group.getBinary(name, 0).getBytes();
                    Cell cell = new KeyValue(rowKey, names[0].getBytes(), names[1].getBytes(), timestamp, value);
                    cells.add(cell);
                }
            }
        }
    }
    return cells;
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public ParquetStructConverter(Type prestoType, String columnName, GroupType entryType, int fieldIndex)
{
    checkArgument(ROW.equals(prestoType.getTypeSignature().getBase()));
    List<Type> prestoTypeParameters = prestoType.getTypeParameters();
    List<parquet.schema.Type> fieldTypes = entryType.getFields();
    checkArgument(prestoTypeParameters.size() == fieldTypes.size());

    this.rowType = prestoType;
    this.fieldIndex = fieldIndex;

    ImmutableList.Builder<BlockConverter> converters = ImmutableList.builder();
    for (int i = 0; i < prestoTypeParameters.size(); i++) {
        parquet.schema.Type fieldType = fieldTypes.get(i);
        converters.add(createConverter(prestoTypeParameters.get(i), columnName + "." + fieldType.getName(), fieldType, i));
    }
    this.converters = converters.build();
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private void writeRecordFields(GroupType schema, Schema tajoSchema,
                               Tuple tuple) {
  List<Type> fields = schema.getFields();
  // Parquet ignores Tajo NULL_TYPE columns, so the index may differ.
  int index = 0;
  for (int tajoIndex = 0; tajoIndex < tajoSchema.size(); ++tajoIndex) {
    Column column = tajoSchema.getColumn(tajoIndex);
    if (column.getDataType().getType() == TajoDataTypes.Type.NULL_TYPE) {
      continue;
    }
    Datum datum = tuple.get(tajoIndex);
    Type fieldType = fields.get(index);
    if (!tuple.isNull(tajoIndex)) {
      recordConsumer.startField(fieldType.getName(), index);
      writeValue(fieldType, column, datum);
      recordConsumer.endField(fieldType.getName(), index);
    } else if (fieldType.isRepetition(Type.Repetition.REQUIRED)) {
      throw new RuntimeException("Null-value for required field: " +
          column.getSimpleName());
    }
    ++index;
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
@Override
Object convertField(JsonElement value) {
  ParquetGroup r1 = new ParquetGroup((GroupType) schema());
  JsonObject inputRecord = value.getAsJsonObject();
  for (Map.Entry<String, JsonElement> entry : inputRecord.entrySet()) {
    String key = entry.getKey();
    JsonElementConverter converter = this.converters.get(key);
    Object convertedValue = converter.convert(entry.getValue());
    boolean valueIsNull = convertedValue == null;
    Type.Repetition repetition = converter.jsonSchema.optionalOrRequired();
    if (valueIsNull && repetition.equals(OPTIONAL)) {
      continue;
    }
    r1.add(key, convertedValue);
  }
  return r1;
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private Type buildSchema() {
  JsonArray inputSchema = this.jsonSchema.getDataTypeValues();
  List<Type> parquetTypes = new ArrayList<>();
  for (JsonElement element : inputSchema) {
    JsonObject map = (JsonObject) element;
    JsonSchema elementSchema = new JsonSchema(map);
    String columnName = elementSchema.getColumnName();
    JsonElementConverter converter = JsonElementConversionFactory.getConverter(elementSchema, false);
    Type schemaType = converter.schema();
    this.converters.put(columnName, converter);
    parquetTypes.add(schemaType);
  }
  String docName = this.jsonSchema.getColumnName();
  switch (recordType) {
    case ROOT:
      return new MessageType(docName, parquetTypes);
    case CHILD:
      return new GroupType(this.jsonSchema.optionalOrRequired(), docName, parquetTypes);
    default:
      throw new RuntimeException("Unsupported Record type");
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public AvroUnionConverter(ParentValueContainer parent, Type parquetSchema,
                          Schema avroSchema) {
  this.parent = parent;
  GroupType parquetGroup = parquetSchema.asGroupType();
  this.memberConverters = new Converter[ parquetGroup.getFieldCount()];

  int parquetIndex = 0;
  for (int index = 0; index < avroSchema.getTypes().size(); index++) {
    Schema memberSchema = avroSchema.getTypes().get(index);
    if (!memberSchema.getType().equals(Schema.Type.NULL)) {
      Type memberType = parquetGroup.getType(parquetIndex);
      memberConverters[parquetIndex] = newConverter(memberSchema, memberType, new ParentValueContainer() {
        @Override
        void add(Object value) {
          Preconditions.checkArgument(memberValue==null, "Union is resolving to more than one type");
          memberValue = value;
        }
      });
      parquetIndex++; // Note for nulls the parquetIndex id not increased
    }
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private void writeRecordFields(GroupType schema, Schema avroSchema,
                               Map record) {
  List<Type> fields = schema.getFields();
  List<Schema.Field> avroFields = avroSchema.getFields();
  int index = 0; // parquet ignores Avro nulls, so index may differ
  for (int avroIndex = 0; avroIndex < avroFields.size(); avroIndex++) {
    Schema.Field avroField = avroFields.get(avroIndex);
    if (avroField.schema().getType().equals(Schema.Type.NULL)) {
      continue;
    }
    Type fieldType = fields.get(index);
    Object value = record.get(fieldKeyword.invoke(avroField));
    if (value != null) {
      recordConsumer.startField(fieldType.getName(), index);
      writeValue(fieldType, avroField.schema(), value);
      recordConsumer.endField(fieldType.getName(), index);
    } else if (fieldType.isRepetition(Type.Repetition.REQUIRED)) {
      throw new RuntimeException("Null-value for required field: " + avroField.name());
    }
    index++;
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private static Type getType(String[] pathSegments, int depth, MessageType schema) {
  Type type = schema.getType(Arrays.copyOfRange(pathSegments, 0, depth + 1));
  if (depth + 1 == pathSegments.length) {
    return type;
  } else {
    Preconditions.checkState(!type.isPrimitive());
    return new GroupType(type.getRepetition(), type.getName(), getType(pathSegments, depth + 1, schema));
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private OriginalType getOriginalType(Type type, String[] path, int depth) {
  if (type.isPrimitive()) {
    return type.getOriginalType();
  }
  Type t = ((GroupType) type).getType(path[depth]);
  return getOriginalType(t, path, depth + 1);
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private boolean contains(GroupType group, String[] path, int index) {
    if (index == path.length) {
        return false;
    }
    if (group.containsField(path[index])) {
        Type type = group.getType(path[index]);
        if (type.isPrimitive()) {
            return index + 1 == path.length;
        } else {
            return contains(type.asGroupType(), path, index + 1);
        }
    }
    return false;
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private static GroupedConverter createGroupConverter(Type prestoType, String columnName, parquet.schema.Type parquetType, int fieldIndex)
{
    GroupType groupType = parquetType.asGroupType();
    switch (prestoType.getTypeSignature().getBase()) {
        case ARRAY:
            return new ParquetListConverter(prestoType, columnName, groupType, fieldIndex);
        case MAP:
            return new ParquetMapConverter(prestoType, columnName, groupType, fieldIndex);
        case ROW:
            return new ParquetStructConverter(prestoType, columnName, groupType, fieldIndex);
        default:
            throw new IllegalArgumentException("Column " + columnName + " type " + parquetType.getOriginalType() + " not supported");
    }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public ParquetListConverter(Type prestoType, String columnName, GroupType listType, int fieldIndex)
{
    checkArgument(listType.getFieldCount() == 1,
            "Expected LIST column '%s' to only have one field, but has %s fields",
            columnName,
            listType.getFieldCount());
    checkArgument(ARRAY.equals(prestoType.getTypeSignature().getBase()));

    this.arrayType = prestoType;
    this.fieldIndex = fieldIndex;

    // The Parquet specification requires that the element value of a
    // LIST type be wrapped in an inner repeated group, like so:
    //
    // optional group listField (LIST) {
    //   repeated group list {
    //     optional int element
    //   }
    // }
    //
    // However, some parquet libraries don't follow this spec. The
    // compatibility rules used here are specified in the Parquet
    // documentation at http://git.io/vOpNz.
    parquet.schema.Type elementType = listType.getType(0);
    if (isElementType(elementType, listType.getName())) {
        elementConverter = createConverter(prestoType.getTypeParameters().get(0), columnName + ".element", elementType, 0);
    }
    else {
        elementConverter = new ParquetListEntryConverter(prestoType.getTypeParameters().get(0), columnName, elementType.asGroupType());
    }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public ParquetListEntryConverter(Type prestoType, String columnName, GroupType elementType)
{
    checkArgument(elementType.getOriginalType() == null,
            "Expected LIST column '%s' field to be type STRUCT, but is %s",
            columnName,
            elementType);

    checkArgument(elementType.getFieldCount() == 1,
            "Expected LIST column '%s' element to have one field, but has %s fields",
            columnName,
            elementType.getFieldCount());

    elementConverter = createConverter(prestoType, columnName + ".element", elementType.getType(0), 0);
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public ParquetMapConverter(Type type, String columnName, GroupType mapType, int fieldIndex)
{
    checkArgument(mapType.getFieldCount() == 1,
            "Expected MAP column '%s' to only have one field, but has %s fields",
            mapType.getName(),
            mapType.getFieldCount());

    this.mapType = type;
    this.fieldIndex = fieldIndex;

    parquet.schema.Type entryType = mapType.getFields().get(0);

    entryConverter = new ParquetMapEntryConverter(type, columnName + ".entry", entryType.asGroupType());
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public ParquetMapEntryConverter(Type prestoType, String columnName, GroupType entryType)
{
    checkArgument(MAP.equals(prestoType.getTypeSignature().getBase()));
    // original version of parquet used null for entry due to a bug
    if (entryType.getOriginalType() != null) {
        checkArgument(entryType.getOriginalType() == MAP_KEY_VALUE,
                "Expected MAP column '%s' field to be type %s, but is %s",
                columnName,
                MAP_KEY_VALUE,
                entryType);
    }

    GroupType entryGroupType = entryType.asGroupType();
    checkArgument(entryGroupType.getFieldCount() == 2,
            "Expected MAP column '%s' entry to have two fields, but has %s fields",
            columnName,
            entryGroupType.getFieldCount());
    checkArgument(entryGroupType.getFieldName(0).equals("key"),
            "Expected MAP column '%s' entry field 0 to be named 'key', but is named %s",
            columnName,
            entryGroupType.getFieldName(0));
    checkArgument(entryGroupType.getFieldName(1).equals("value"),
            "Expected MAP column '%s' entry field 1 to be named 'value', but is named %s",
            columnName,
            entryGroupType.getFieldName(1));
    checkArgument(entryGroupType.getType(0).isPrimitive(),
            "Expected MAP column '%s' entry field 0 to be primitive, but is %s",
            columnName,
            entryGroupType.getType(0));

    keyConverter = createConverter(prestoType.getTypeParameters().get(0), columnName + ".key", entryGroupType.getFields().get(0), 0);
    valueConverter = createConverter(prestoType.getTypeParameters().get(1), columnName + ".value", entryGroupType.getFields().get(1), 1);
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
/**
 * Creates a new TajoRecordConverter.
 *
 * @param parquetSchema The Parquet schema of the projection.
 * @param tajoReadSchema The Tajo schema of the table.
 * @param projectionMap An array mapping the projection column to the column
 *                      index in the table.
 */
public TajoRecordConverter(GroupType parquetSchema, Schema tajoReadSchema,
                           int[] projectionMap) {
  this.parquetSchema = parquetSchema;
  this.tajoReadSchema = tajoReadSchema;
  this.projectionMap = projectionMap;
  this.tupleSize = tajoReadSchema.size();

  // The projectionMap.length does not match parquetSchema.getFieldCount()
  // when the projection contains NULL_TYPE columns. We will skip over the
  // NULL_TYPE columns when we construct the converters and populate the
  // NULL_TYPE columns with NullDatums in start().
  int index = 0;
  this.converters = new Converter[parquetSchema.getFieldCount()];
  for (int i = 0; i < projectionMap.length; ++i) {
    final int projectionIndex = projectionMap[i];
    Column column = tajoReadSchema.getColumn(projectionIndex);
    if (column.getDataType().getType() == TajoDataTypes.Type.NULL_TYPE) {
      continue;
    }
    Type type = parquetSchema.getType(index);
    converters[index] = newConverter(column, type, new ParentValueContainer() {
      @Override
      void add(Object value) {
        TajoRecordConverter.this.set(projectionIndex, value);
      }
    });
    ++index;
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
public ParquetGroup(GroupType schema) {
  this.schema = schema;
  this.data = new List[schema.getFields().size()];

  for (int i = 0; i < schema.getFieldCount(); ++i) {
    this.data[i] = new ArrayList();
  }
}
 

import parquet.schema.GroupType; //导入依赖的package包/类
private void writeArray(GroupType schema, Schema avroSchema,
                            List array) {
  recordConsumer.startGroup(); // group wrapper (original type LIST)
  if (array.iterator().hasNext()) {
    recordConsumer.startField("array", 0);
    for (Object elt : array) {
      writeValue(schema.getType(0), avroSchema.getElementType(), elt);
    }
    recordConsumer.endField("array", 0);
  }
  recordConsumer.endGroup();
}