@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;
}
 

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, ",");
}
 

private static void addField(RequiredField field, List<String> fields, FieldAlias fa, String currentNode) {
    if (field.getSubFields() != null && !field.getSubFields().isEmpty()) {
        for (RequiredField subField : field.getSubFields()) {
            addField(subField, fields, fa, currentNode + "." + fa.toES(subField.getAlias()));
        }
    }

    else {
        fields.add(fa.toES(field.getAlias()));
    }
}