import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
static String asProjection(RequiredFieldList list, Properties props) {
    List<String> fields = new ArrayList<String>();
    FieldAlias alias = alias(new PropertiesSettings(props));
    for (RequiredField field : list.getFields()) {
        addField(field, fields, alias, "");
    }

    return StringUtils.concatenateAndUriEncode(fields, ",");
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
/**
 * @param configuration configuration for the current job
 * @return List of required fields from pushProjection
 */
static RequiredFieldList getRequiredFields(Configuration configuration) {
  String requiredFieldString = configuration.get(PARQUET_PIG_REQUIRED_FIELDS);

  if(requiredFieldString == null) {
    return null;
  }

  try {
    return (RequiredFieldList) ObjectSerializer.deserialize(requiredFieldString);
  } catch (IOException iOException) {
    throw new RuntimeException("Failed to deserialize pushProjection");
  }
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
@Override
public ReadContext init(InitContext initContext) {
  Schema pigSchema = getPigSchema(initContext.getConfiguration());
  RequiredFieldList requiredFields = getRequiredFields(initContext.getConfiguration());
  boolean columnIndexAccess = initContext.getConfiguration().getBoolean(PARQUET_COLUMN_INDEX_ACCESS, false);

  if (pigSchema == null) {
    return new ReadContext(initContext.getFileSchema());
  } else {

    // project the file schema according to the requested Pig schema
    MessageType parquetRequestedSchema = new PigSchemaConverter(columnIndexAccess).filter(initContext.getFileSchema(), pigSchema, requiredFields);
    return new ReadContext(parquetRequestedSchema);
  }
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
@Override
public List<Type> filterTupleSchema(GroupType schemaToFilter, Schema pigSchema, RequiredFieldList requiredFieldsList) {
  List<Type> newFields = new ArrayList<Type>();
  List<Pair<FieldSchema,Integer>> indexedFields = new ArrayList<Pair<FieldSchema,Integer>>();

  try {
    if(requiredFieldsList == null) {
      int index = 0;
      for(FieldSchema fs : pigSchema.getFields()) {
        indexedFields.add(new Pair<FieldSchema, Integer>(fs, index++));
      }
    } else {
      for(RequiredField rf : requiredFieldsList.getFields()) {
        indexedFields.add(new Pair<FieldSchema, Integer>(pigSchema.getField(rf.getAlias()), rf.getIndex()));
      }
    }

    for (Pair<FieldSchema, Integer> p : indexedFields) {
      FieldSchema fieldSchema = pigSchema.getField(p.first.alias);
      if (p.second < schemaToFilter.getFieldCount()) {
        Type type = schemaToFilter.getFields().get(p.second);
        newFields.add(filter(type, fieldSchema));
      }
    }
  } catch (FrontendException e) {
      throw new RuntimeException("Failed to filter requested fields", e);
  }
  return newFields;
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
public static RequiredFieldList deserializeRequiredFieldList(String requiredFieldString) {
  if(requiredFieldString == null) {
      return null;
  }

  try {
    return (RequiredFieldList) ObjectSerializer.deserialize(requiredFieldString);
  } catch (IOException e) {
    throw new RuntimeException("Failed to deserialize pushProjection", e);
  }
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
static String serializeRequiredFieldList(RequiredFieldList requiredFieldList) {
  try {
    return ObjectSerializer.serialize(requiredFieldList);
  } catch (IOException e) {
    throw new RuntimeException("Failed to searlize required fields.", e);
  }
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
/**
 * filters a Parquet schema based on a pig schema for projection
 * @param schemaToFilter the schema to be filter
 * @param requestedPigSchema the pig schema to filter it with
 * @param requiredFieldList projected required fields
 * @return the resulting filtered schema
 */
public MessageType filter(MessageType schemaToFilter, Schema requestedPigSchema, RequiredFieldList requiredFieldList) {
  try {
    if (LOG.isDebugEnabled()) LOG.debug("filtering schema:\n" + schemaToFilter + "\nwith requested pig schema:\n " + requestedPigSchema);
    List<Type> result = columnAccess.filterTupleSchema(schemaToFilter, requestedPigSchema, requiredFieldList);
    if (LOG.isDebugEnabled()) LOG.debug("schema:\n" + schemaToFilter + "\nfiltered to:\n" + result);
    return new MessageType(schemaToFilter.getName(), result);
  } catch (RuntimeException e) {
    throw new RuntimeException("can't filter " + schemaToFilter + " with " + requestedPigSchema, e);
  }
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
static String asProjection(RequiredFieldList list, Properties props) {
    List<String> fields = new ArrayList<String>();
    FieldAlias alias = alias(new PropertiesSettings(props));
    for (RequiredField field : list.getFields()) {
        addField(field, fields, alias, "");
    }

    return StringUtils.concatenate(fields, ",");
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
@Override
public RequiredFieldResponse pushProjection(RequiredFieldList requiredFieldList) throws FrontendException {
    aliases = new String[requiredFieldList.getFields().size()];
    for (int i=0; i<requiredFieldList.getFields().size(); i++) {
        RequiredField fs = requiredFieldList.getFields().get(i);
        aliases[i] = fs.getAlias();
    }
    try {
        UDFContext.getUDFContext().getUDFProperties(this.getClass()).setProperty(signature, ObjectSerializer.serialize(aliases));
    } catch (IOException e) {
        throw new FrontendException(e);
    }
    return new RequiredFieldResponse(true);
}
 

import org.apache.pig.LoadPushDown.RequiredFieldList; //导入依赖的package包/类
/**
 * Takes an Avro Schema and a Pig RequiredFieldList and returns a new schema
 * with only the requried fields, or no if the function can't extract only
 * those fields. Useful for push down projections.
 * @param oldSchema The avro schema from which to extract the schema
 * @param rfl the Pig required field list
 * @return the new schema, or null
 */
public static Schema newSchemaFromRequiredFieldList(
    final Schema oldSchema, final RequiredFieldList rfl) {
  return newSchemaFromRequiredFieldList(oldSchema, rfl.getFields());
}