[BEAM-7174] Add schema modification transforms

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/FieldAccessDescriptor.java

@@ -47,6 +47,7 @@
 import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ArrayListMultimap;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableList;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableMap;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Iterables;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Lists;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Maps;
 import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Multimap;
@@ -228,7 +229,7 @@ public static FieldAccessDescriptor withFields(Iterable<FieldDescriptor> fields)
  return builder().setFieldsAccessed(Sets.newLinkedHashSet(fields)).build();
  }
 
- // Union a set of FieldAccessDescriptors. This function currenty only supports descriptors with
+ // Union a set of FieldAccessDescriptors. This function currently only supports descriptors with
  // containing named fields, not those containing ids.
  private static FieldAccessDescriptor union(
  Iterable<FieldAccessDescriptor> fieldAccessDescriptors) {
@@ -323,6 +324,15 @@ public Set<Integer> fieldIdsAccessed() {
  .collect(Collectors.toSet());
  }
 
+ /**
+ * Return the field names accessed. Should not be called until after {@link #resolve} is called.
+ */
+ public Set<String> fieldNamesAccessed() {
+ return getFieldsAccessed().stream()
+ .map(FieldDescriptor::getFieldName)
+ .collect(Collectors.toSet());
+ }
+
  /**
  * Return the nested fields keyed by field ids. Should not be called until after {@link #resolve}
  * is called.
@@ -332,6 +342,32 @@ public Map<Integer, FieldAccessDescriptor> nestedFieldsById() {
  .collect(Collectors.toMap(f -> f.getKey().getFieldId(), f -> f.getValue()));
  }
 
+ /**
+ * Return the nested fields keyed by field name. Should not be called until after {@link #resolve}
+ * is called.
+ */
+ public Map<String, FieldAccessDescriptor> nestedFieldsByName() {
+ return getNestedFieldsAccessed().entrySet().stream()
+ .collect(Collectors.toMap(f -> f.getKey().getFieldName(), f -> f.getValue()));
+ }
+
+ /** Returns true if this descriptor references only a single, non-wildcard field. */
+ public boolean referencesSingleField() {
+ if (getAllFields()) {
+ return false;
+ }
+
+ if (getFieldsAccessed().size() == 1 && getNestedFieldsAccessed().isEmpty()) {
+ return true;
+ }
+
+ if (getFieldsAccessed().isEmpty() && getNestedFieldsAccessed().size() == 1) {
+ return getNestedFieldsAccessed().values().iterator().next().referencesSingleField();
+ }
+
+ return false;
+ }
+
  /**
  * Resolve the {@link FieldAccessDescriptor} against a schema.
  *
@@ -513,4 +549,22 @@ private static void validateFieldDescriptor(Schema schema, FieldDescriptor field
  }
  }
  }
+
+ @Override
+ public String toString() {
+ if (getAllFields()) {
+ return "*";
+ }
+
+ List<String> singleSelectors =
+ getFieldsAccessed().stream()
+ .map(FieldDescriptor::getFieldName)
+ .collect(Collectors.toList());
+ List<String> nestedSelectors =
+ getNestedFieldsAccessed().entrySet().stream()
+ .map(e -> e.getKey().getFieldName() + "." + e.getValue().toString())
+ .collect(Collectors.toList());
+ ;
+ return String.join(", ", Iterables.concat(singleSelectors, nestedSelectors));
+ }
 }

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java

@@ -196,6 +196,10 @@ public Builder addMapField(String name, FieldType keyType, FieldType valueType)
  return this;
  }
 
+ public int getLastFieldId() {
+ return fields.size() - 1;
+ }
+
  public Schema build() {
  return new Schema(fields);
  }