import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
ParquetGroupConverter(
    OutputMutator mutator,
    GroupType schema,
    Collection<SchemaPath> columns,
    OptionManager options,
    List<Field> arrowSchema,
    Function<String, String> childNameResolver,
    ParquetReaderUtility.DateCorruptionStatus containsCorruptedDates,
    boolean readInt96AsTimeStamp) {
  this.converters = Lists.newArrayList();
  this.mutator = mutator;
  this.schema = schema;
  this.columns = columns;
  this.options = options;
  this.containsCorruptedDates = containsCorruptedDates;
  this.arrowSchema = arrowSchema;
  this.childNameResolver = childNameResolver;
  this.readInt96AsTimeStamp = readInt96AsTimeStamp;
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private Converter groupConverter(OutputMutator mutator,
    List<Field> arrowSchema, GroupType groupType, PathSegment colNextChild, final String nameForChild) {
  Collection<SchemaPath> c = new ArrayList<>();

  while (colNextChild != null) {
    if (colNextChild.isNamed()) {
      break;
    }
    colNextChild = colNextChild.getChild();
  }

  if (colNextChild != null) {
    SchemaPath s = new SchemaPath(colNextChild.getNameSegment());
    c.add(s);
  }

  if (arrowSchema != null) {
    return groupConverterFromArrowSchema(nameForChild, groupType.getName(), groupType, c);
  }

  return defaultGroupConverter(mutator, groupType, nameForChild, c, null);
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
/**
 * Changes the list inner '$data$' vector name to 'element' in the schema
 */
private Type renameChildTypeToElement(Type childType) {
  if (childType.isPrimitive()) {
    PrimitiveType childPrimitiveType = childType.asPrimitiveType();
    return new PrimitiveType(childType.getRepetition(),
      childPrimitiveType.getPrimitiveTypeName(),
      childPrimitiveType.getTypeLength(),
      "element",
      childPrimitiveType.getOriginalType(),
      childPrimitiveType.getDecimalMetadata(),
      null);
  } else {
    GroupType childGroupType = childType.asGroupType();
    return new GroupType(childType.getRepetition(),
      "element",
      childType.getOriginalType(),
      childGroupType.getFields());
  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
public ParquetValueConverter(GroupType schema, ParentContainerUpdater updater)
{
    super(updater);

    ArrayList<String> fieldNames = new ArrayList<>();
    for (Type type : schema.getFields()) {
        fieldNames.add(type.getName());
    }
    this.currentMap = new InternalMap(fieldNames);

    this.fieldConverters = new Converter[schema.getFieldCount()];
    int i = 0;
    for (Type field : schema.getFields()) {
        InternalMapUpdater update = new InternalMapUpdater(currentMap, i);
        fieldConverters[i++] = newFieldConverter(field, update);
    }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private ParquetArrayConverter(GroupType schema, ParentContainerUpdater updater)
{
    super(updater);

    Type repeatedType = schema.getType(0);
    if (isElementType(repeatedType, schema.getName())) {
        // If the repeated field corresponds to the element type, creates a new converter using the
        // type of the repeated field.
        elementConverter = newConverter(repeatedType, new ParentContainerUpdater.Noop() {
            @Override
            public void set(Value value)
            {
                ParquetArrayConverter.this.currentArray.add(value);
            }
        });
    }
    else {
        // If the repeated field corresponds to the syntactic group in the standard 3-level Parquet
        // LIST layout, creates a new converter using the only child field of the repeated field.
        elementConverter = new ElementConverter(repeatedType.asGroupType().getType(0));
    }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private ColTypeInfo getColTypeInfo(MessageType schema, Type type, String[] path, int depth) {
  if (type.isPrimitive()) {
    PrimitiveType primitiveType = (PrimitiveType) type;
    int precision = 0;
    int scale = 0;
    if (primitiveType.getDecimalMetadata() != null) {
      precision = primitiveType.getDecimalMetadata().getPrecision();
      scale = primitiveType.getDecimalMetadata().getScale();
    }

    int repetitionLevel = schema.getMaxRepetitionLevel(path);
    int definitionLevel = schema.getMaxDefinitionLevel(path);

    return new ColTypeInfo(type.getOriginalType(), precision, scale, repetitionLevel, definitionLevel);
  }
  Type t = ((GroupType) type).getType(path[depth]);
  return getColTypeInfo(schema, t, path, depth + 1);
}
 

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

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private Map<Integer, Converter> buildFieldToConverter(final GroupType schema) {
    final Map<Integer, Converter> fieldToConverter = new HashMap<>(fieldCount);
    int i = 0;
    for (final Type field : schema.getFields()) {
        final String[] newColumnPath = new String[columnPath.length + 1];
        int j = 0;
        for (final String part : columnPath) {
            newColumnPath[j] = part;
            j++;
        }
        newColumnPath[j] = field.getName();
        if (field.isPrimitive()) {
            fieldToConverter.put(i, new PrimitiveConverter(parquetColumnToObject, field.asPrimitiveType().getPrimitiveTypeName().javaType.getSimpleName(), newColumnPath, field.getOriginalType()));
        } else {
            fieldToConverter.put(i, new BypassGroupConverter(parquetColumnToObject, field.asGroupType(), newColumnPath));
        }
        i++;
    }
    return fieldToConverter;
}
 

import org.apache.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;
    }
    Type fieldType = fields.get(index);
    if (!tuple.isBlankOrNull(tajoIndex)) {
      recordConsumer.startField(fieldType.getName(), index);
      writeValue(column, tuple, tajoIndex);
      recordConsumer.endField(fieldType.getName(), index);
    } else if (fieldType.isRepetition(Type.Repetition.REQUIRED)) {
      throw new RuntimeException("Null-value for required field: " +
          column.getSimpleName());
    }
    ++index;
  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private void writeGroup(Group group, GroupType type) {
  int fieldCount = type.getFieldCount();
  for (int field = 0; field < fieldCount; ++field) {
    int valueCount = group.getFieldRepetitionCount(field);
    if (valueCount > 0) {
      Type fieldType = type.getType(field);
      String fieldName = fieldType.getName();
      recordConsumer.startField(fieldName, field);
      for (int index = 0; index < valueCount; ++index) {
        if (fieldType.isPrimitive()) {
          group.writeValue(field, index, recordConsumer);
        } else {
          recordConsumer.startGroup();
          writeGroup(group.getGroup(field, index), fieldType.asGroupType());
          recordConsumer.endGroup();
        }
      }
      recordConsumer.endField(fieldName, field);
    }
  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
SimpleGroupConverter(SimpleGroupConverter parent, int index, GroupType schema) {
  this.parent = parent;
  this.index = index;

  converters = new Converter[schema.getFieldCount()];

  for (int i = 0; i < converters.length; i++) {
    final Type type = schema.getType(i);
    if (type.isPrimitive()) {
      converters[i] = new SimplePrimitiveConverter(this, i);
    } else {
      converters[i] = new SimpleGroupConverter(this, i, type.asGroupType());
    }

  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private void visitChildren(GroupColumnIO newIO, GroupType groupType, GroupType requestedGroupType) {
  GroupColumnIO oldIO = current;
  current = newIO;
  for (Type type : groupType.getFields()) {
    // if the file schema does not contain the field it will just stay null
    if (requestedGroupType.containsField(type.getName())) {
      currentRequestedIndex = requestedGroupType.getFieldIndex(type.getName());
      currentRequestedType = requestedGroupType.getType(currentRequestedIndex);
      if (currentRequestedType.getRepetition().isMoreRestrictiveThan(type.getRepetition())) {
        incompatibleSchema(type, currentRequestedType);
      }
      type.accept(this);
    }
  }
  current = oldIO;
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
@Test
public void testBadWriteSchema() throws IOException {
  final File file = temp.newFile("test.parquet");
  file.delete();

  TestUtils.assertThrows("Should reject a schema with an empty group",
      InvalidSchemaException.class, new Callable<Void>() {
        @Override
        public Void call() throws IOException {
          ExampleParquetWriter.builder(new Path(file.toString()))
              .withType(Types.buildMessage()
                  .addField(new GroupType(REQUIRED, "invalid_group"))
                  .named("invalid_message"))
              .build();
          return null;
        }
      });

  Assert.assertFalse("Should not create a file when schema is rejected",
      file.exists());
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
public ElementConverter(String listName, List<TProtocol> listEvents,
                        GroupType repeatedType, ThriftField thriftElement) {
  this.listEvents = listEvents;
  this.elementEvents = new ArrayList<TProtocol>();
  Type elementType = repeatedType.getType(0);
  if (elementType.isRepetition(Type.Repetition.OPTIONAL)) {
    if (ignoreNullElements) {
      LOG.warn("List " + listName +
          " has optional elements: null elements are ignored.");
    } else {
      throw new ParquetDecodingException("Cannot read list " + listName +
          " with optional elements: set " + IGNORE_NULL_LIST_ELEMENTS +
          " to ignore nulls.");
    }
  }
  elementConverter = newConverter(elementEvents, elementType, thriftElement);
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private boolean hasMissingRequiredFieldInGroupType(GroupType requested, GroupType fullSchema) {
  for (Type field : fullSchema.getFields()) {

    if (requested.containsField(field.getName())) {
      Type requestedType = requested.getType(field.getName());
      // if a field is in requested schema and the type of it is a group type, then do recursive check
      if (!field.isPrimitive()) {
        if (hasMissingRequiredFieldInGroupType(requestedType.asGroupType(), field.asGroupType())) {
          return true;
        } else {
          continue;// check next field
        }
      }
    } else {
      if (field.getRepetition() == Type.Repetition.REQUIRED) {
        return true; // if a field is missing in requested schema and it's required
      } else {
        continue; // the missing field is not required, then continue checking next field
      }
    }
  }

  return false;
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
public DataWritableGroupConverter(final GroupType selectedGroupType,
    final HiveGroupConverter parent, final int index, final GroupType containingGroupType) {
  this.parent = parent;
  this.index = index;
  final int totalFieldCount = containingGroupType.getFieldCount();
  final int selectedFieldCount = selectedGroupType.getFieldCount();

  currentArr = new Object[totalFieldCount];
  converters = new Converter[selectedFieldCount];

  List<Type> selectedFields = selectedGroupType.getFields();
  for (int i = 0; i < selectedFieldCount; i++) {
    Type subtype = selectedFields.get(i);
    if (containingGroupType.getFields().contains(subtype)) {
      converters[i] = getConverterFromDescription(subtype,
          containingGroupType.getFieldIndex(subtype.getName()), this);
    } else {
      throw new IllegalStateException("Group type [" + containingGroupType +
          "] does not contain requested field: " + subtype);
    }
  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
@SuppressWarnings("unchecked")
MessageWriter(Descriptors.Descriptor descriptor, GroupType schema) {
  List<Descriptors.FieldDescriptor> fields = descriptor.getFields();
  fieldWriters = (FieldWriter[]) Array.newInstance(FieldWriter.class, fields.size());

  for (Descriptors.FieldDescriptor fieldDescriptor: fields) {
    String name = fieldDescriptor.getName();
    Type type = schema.getType(name);
    FieldWriter writer = createWriter(fieldDescriptor, type);

    if(fieldDescriptor.isRepeated()) {
     writer = new ArrayWriter(writer);
    }

    writer.setFieldName(name);
    writer.setIndex(schema.getFieldIndex(name));

    fieldWriters[fieldDescriptor.getIndex()] = writer;
  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
/**
 *
 * @param alias
 * @param fieldSchema
 * @return an optional group containing one repeated group field (key, value)
 * @throws FrontendException
 */
private GroupType convertMap(String alias, FieldSchema fieldSchema) {
  Schema innerSchema = fieldSchema.schema;
  if (innerSchema == null || innerSchema.size() != 1) {
    throw new SchemaConversionException("Invalid map Schema, schema should contain exactly one field: " + fieldSchema);
  }
  FieldSchema innerField = null;
  try {
    innerField = innerSchema.getField(0);
  } catch (FrontendException fe) {
    throw new SchemaConversionException("Invalid map schema, cannot infer innerschema: ", fe);
  }
  Type convertedValue = convertWithName(innerField, "value");
  return ConversionPatterns.stringKeyMapType(Repetition.OPTIONAL, alias, name(innerField.alias, "map"),
      convertedValue);
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
@Test
public void testListsOfPrimitive() throws Exception {
  for (Type.Repetition repetition : Type.Repetition.values()) {
    for (Type.Repetition valueRepetition : Type.Repetition.values()) {
      for (PrimitiveType.PrimitiveTypeName primitiveTypeName : PrimitiveType.PrimitiveTypeName.values()) {
        if (primitiveTypeName != PrimitiveType.PrimitiveTypeName.INT96) { // INT96 is NYI
          Types.PrimitiveBuilder<PrimitiveType> value = Types.primitive(primitiveTypeName, valueRepetition);
          if (primitiveTypeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
            value.length(1);
          GroupType type = Types.buildGroup(repetition).addField(value.named("b")).as(OriginalType.LIST).named("a");
          pigSchemaConverter.convertField(type); // no exceptions, please
        }
      }
    }
  }
}
 

import org.apache.parquet.schema.GroupType; //导入依赖的package包/类
private void writeRecordFields(GroupType schema, Schema avroSchema,
                               Object 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 = model.getField(record, avroField.name(), avroIndex);
    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++;
  }
}