[BEAM-13697] Add SchemaFieldNumber annotation

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

@@ -19,11 +19,11 @@
 
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.util.Comparator;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
-import org.apache.beam.sdk.schemas.annotations.SchemaFieldName;
 import org.apache.beam.sdk.schemas.annotations.SchemaIgnore;
 import org.apache.beam.sdk.schemas.utils.AutoValueUtils;
 import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.DefaultTypeConversionsFactory;
@@ -32,6 +32,8 @@
 import org.apache.beam.sdk.schemas.utils.ReflectUtils;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /** A {@link SchemaProvider} for AutoValue classes. */
@@ -48,22 +50,44 @@ public static class AbstractGetterTypeSupplier implements FieldValueTypeSupplier
 
     @Override
     public List<FieldValueTypeInformation> get(Class<?> clazz) {
+
       // If the generated class is passed in, we want to look at the base class to find the getters.
       Class<?> targetClass = AutoValueUtils.getBaseAutoValueClass(clazz);
-      return ReflectUtils.getMethods(targetClass).stream()
-          .filter(ReflectUtils::isGetter)
-          // All AutoValue getters are marked abstract.
-          .filter(m -> Modifier.isAbstract(m.getModifiers()))
-          .filter(m -> !Modifier.isPrivate(m.getModifiers()))
-          .filter(m -> !Modifier.isProtected(m.getModifiers()))
-          .filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
-          .map(FieldValueTypeInformation::forGetter)
-          .map(
-              t -> {
-                SchemaFieldName fieldName = t.getMethod().getAnnotation(SchemaFieldName.class);
-                return (fieldName == null) ? t : t.withName(fieldName.value());
-              })
-          .collect(Collectors.toList());
+
+      List<Method> methods =
+          ReflectUtils.getMethods(targetClass).stream()
+              .filter(ReflectUtils::isGetter)
+              // All AutoValue getters are marked abstract.
+              .filter(m -> Modifier.isAbstract(m.getModifiers()))
+              .filter(m -> !Modifier.isPrivate(m.getModifiers()))
+              .filter(m -> !Modifier.isProtected(m.getModifiers()))
+              .filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
+              .collect(Collectors.toList());
+      List<FieldValueTypeInformation> types = Lists.newArrayListWithCapacity(methods.size());
+      for (int i = 0; i < methods.size(); ++i) {
+        types.add(FieldValueTypeInformation.forGetter(methods.get(i), i));
+      }
+      types.sort(Comparator.comparing(FieldValueTypeInformation::getNumber));
+      validateFieldNumbers(types);
+      return types;
+    }
+  }
+
+  private static void validateFieldNumbers(List<FieldValueTypeInformation> types) {
+    for (int i = 0; i < types.size(); ++i) {
+      FieldValueTypeInformation type = types.get(i);
+      @javax.annotation.Nullable Integer number = type.getNumber();
+      if (number == null) {
+        throw new RuntimeException("Unexpected null number for " + type.getName());
+      }
+      Preconditions.checkState(
+          number == i,
+          "Expected field number "
+              + i
+              + " for field + "
+              + type.getName()
+              + " instead got "
+              + number);
     }
   }
 

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

@@ -30,6 +30,7 @@
 import java.util.stream.Stream;
 import org.apache.beam.sdk.schemas.annotations.SchemaCaseFormat;
 import org.apache.beam.sdk.schemas.annotations.SchemaFieldName;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
 import org.apache.beam.sdk.schemas.logicaltypes.OneOfType;
 import org.apache.beam.sdk.schemas.utils.ReflectUtils;
 import org.apache.beam.sdk.values.TypeDescriptor;
@@ -43,6 +44,9 @@
   "rawtypes"
 })
 public abstract class FieldValueTypeInformation implements Serializable {
+  /** Optionally returns the field index. */
+  public abstract @Nullable Integer getNumber();
+
   /** Returns the field name. */
   public abstract String getName();
 
@@ -74,6 +78,8 @@ public abstract class FieldValueTypeInformation implements Serializable {
 
   @AutoValue.Builder
   public abstract static class Builder {
+    public abstract Builder setNumber(@Nullable Integer number);
+
     public abstract Builder setName(String name);
 
     public abstract Builder setNullable(boolean nullable);
@@ -113,10 +119,11 @@ public static FieldValueTypeInformation forOneOf(
         .build();
   }
 
-  public static FieldValueTypeInformation forField(Field field) {
+  public static FieldValueTypeInformation forField(Field field, int index) {
     TypeDescriptor type = TypeDescriptor.of(field.getGenericType());
     return new AutoValue_FieldValueTypeInformation.Builder()
         .setName(getNameOverride(field.getName(), field))
+        .setNumber(getNumberOverride(index, field))
         .setNullable(hasNullableAnnotation(field))
         .setType(type)
         .setRawType(type.getRawType())
@@ -128,10 +135,19 @@ public static FieldValueTypeInformation forField(Field field) {
         .build();
   }
 
+  public static <T extends AnnotatedElement & Member> int getNumberOverride(int index, T member) {
+    @Nullable SchemaFieldNumber fieldNumber = member.getAnnotation(SchemaFieldNumber.class);
+    if (fieldNumber == null) {
+      return index;
+    }
+    return Integer.parseInt(fieldNumber.value());
+  }
+
   public static <T extends AnnotatedElement & Member> String getNameOverride(
       String original, T member) {
-    SchemaFieldName fieldName = member.getAnnotation(SchemaFieldName.class);
-    SchemaCaseFormat caseFormatAnnotation = member.getAnnotation(SchemaCaseFormat.class);
+    @Nullable SchemaFieldName fieldName = member.getAnnotation(SchemaFieldName.class);
+    @Nullable SchemaCaseFormat caseFormatAnnotation = member.getAnnotation(SchemaCaseFormat.class);
+    @Nullable
     SchemaCaseFormat classCaseFormatAnnotation =
         member.getDeclaringClass().getAnnotation(SchemaCaseFormat.class);
     if (fieldName != null) {
@@ -151,7 +167,7 @@ public static <T extends AnnotatedElement & Member> String getNameOverride(
     }
   }
 
-  public static FieldValueTypeInformation forGetter(Method method) {
+  public static FieldValueTypeInformation forGetter(Method method, int index) {
     String name;
     if (method.getName().startsWith("get")) {
       name = ReflectUtils.stripPrefix(method.getName(), "get");
@@ -165,6 +181,7 @@ public static FieldValueTypeInformation forGetter(Method method) {
     boolean nullable = hasNullableReturnType(method);
     return new AutoValue_FieldValueTypeInformation.Builder()
         .setName(getNameOverride(name, method))
+        .setNumber(getNumberOverride(index, method))
         .setNullable(nullable)
         .setType(type)
         .setRawType(type.getRawType())

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

@@ -19,19 +19,23 @@
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Method;
+import java.util.Comparator;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
 import org.apache.beam.sdk.schemas.annotations.SchemaCaseFormat;
 import org.apache.beam.sdk.schemas.annotations.SchemaFieldName;
+import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber;
 import org.apache.beam.sdk.schemas.annotations.SchemaIgnore;
 import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.DefaultTypeConversionsFactory;
 import org.apache.beam.sdk.schemas.utils.FieldValueTypeSupplier;
 import org.apache.beam.sdk.schemas.utils.JavaBeanUtils;
 import org.apache.beam.sdk.schemas.utils.ReflectUtils;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
@@ -60,11 +64,36 @@ public static class GetterTypeSupplier implements FieldValueTypeSupplier {
 
     @Override
     public List<FieldValueTypeInformation> get(Class<?> clazz) {
-      return ReflectUtils.getMethods(clazz).stream()
-          .filter(ReflectUtils::isGetter)
-          .filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
-          .map(FieldValueTypeInformation::forGetter)
-          .collect(Collectors.toList());
+      List<Method> methods =
+          ReflectUtils.getMethods(clazz).stream()
+              .filter(ReflectUtils::isGetter)
+              .filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
+              .collect(Collectors.toList());
+      List<FieldValueTypeInformation> types = Lists.newArrayListWithCapacity(methods.size());
+      for (int i = 0; i < methods.size(); ++i) {
+        types.add(FieldValueTypeInformation.forGetter(methods.get(i), i));
+      }
+      types.sort(Comparator.comparing(FieldValueTypeInformation::getNumber));
+      validateFieldNumbers(types);
+      return types;
+    }
+
+    private static void validateFieldNumbers(List<FieldValueTypeInformation> types) {
+      for (int i = 0; i < types.size(); ++i) {
+        FieldValueTypeInformation type = types.get(i);
+        @javax.annotation.Nullable Integer number = type.getNumber();
+        if (number == null) {
+          throw new RuntimeException("Unexpected null number for " + type.getName());
+        }
+        Preconditions.checkState(
+            number == i,
+            "Expected field number "
+                + i
+                + " for field: "
+                + type.getName()
+                + " instead got "
+                + number);
+      }
     }
 
     @Override
@@ -91,6 +120,12 @@ public List<FieldValueTypeInformation> get(Class<?> clazz) {
           .map(FieldValueTypeInformation::forSetter)
           .map(
               t -> {
+                if (t.getMethod().getAnnotation(SchemaFieldNumber.class) != null) {
+                  throw new RuntimeException(
+                      String.format(
+                          "@SchemaFieldNumber can only be used on getters in Java Beans. Found on setter '%s'",
+                          t.getMethod().getName()));
+                }
                 if (t.getMethod().getAnnotation(SchemaFieldName.class) != null) {
                   throw new RuntimeException(
                       String.format(

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

@@ -21,19 +21,22 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import org.apache.beam.sdk.annotations.Experimental;
 import org.apache.beam.sdk.annotations.Experimental.Kind;
-import org.apache.beam.sdk.schemas.annotations.SchemaFieldName;
 import org.apache.beam.sdk.schemas.annotations.SchemaIgnore;
 import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.DefaultTypeConversionsFactory;
 import org.apache.beam.sdk.schemas.utils.FieldValueTypeSupplier;
 import org.apache.beam.sdk.schemas.utils.POJOUtils;
 import org.apache.beam.sdk.schemas.utils.ReflectUtils;
 import org.apache.beam.sdk.values.TypeDescriptor;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
 
 /**
  * A {@link SchemaProvider} for Java POJO objects.
@@ -58,16 +61,16 @@ public static class JavaFieldTypeSupplier implements FieldValueTypeSupplier {
 
     @Override
     public List<FieldValueTypeInformation> get(Class<?> clazz) {
-      List<FieldValueTypeInformation> types =
+      List<Field> fields =
           ReflectUtils.getFields(clazz).stream()
-              .filter(f -> !f.isAnnotationPresent(SchemaIgnore.class))
-              .map(FieldValueTypeInformation::forField)
-              .map(
-                  t -> {
-                    SchemaFieldName fieldName = t.getField().getAnnotation(SchemaFieldName.class);
-                    return (fieldName != null) ? t.withName(fieldName.value()) : t;
-                  })
+              .filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
               .collect(Collectors.toList());
+      List<FieldValueTypeInformation> types = Lists.newArrayListWithCapacity(fields.size());
+      for (int i = 0; i < fields.size(); ++i) {
+        types.add(FieldValueTypeInformation.forField(fields.get(i), i));
+      }
+      types.sort(Comparator.comparing(FieldValueTypeInformation::getNumber));
+      validateFieldNumbers(types);
 
       // If there are no creators registered, then make sure none of the schema fields are final,
       // as we (currently) have no way of creating classes in this case.
@@ -91,6 +94,24 @@ public List<FieldValueTypeInformation> get(Class<?> clazz) {
     }
   }
 
+  private static void validateFieldNumbers(List<FieldValueTypeInformation> types) {
+    for (int i = 0; i < types.size(); ++i) {
+      FieldValueTypeInformation type = types.get(i);
+      @Nullable Integer number = type.getNumber();
+      if (number == null) {
+        throw new RuntimeException("Unexpected null number for " + type.getName());
+      }
+      Preconditions.checkState(
+          number == i,
+          "Expected field number "
+              + i
+              + " for field + "
+              + type.getName()
+              + " instead got "
+              + number);
+    }
+  }
+
   @Override
   public <T> Schema schemaFor(TypeDescriptor<T> typeDescriptor) {
     return POJOUtils.schemaFromPojoClass(

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

@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.schemas.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import javax.annotation.Nonnull;
+import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.sdk.annotations.Experimental.Kind;
+
+/**
+ * When used on a {@link org.apache.beam.sdk.schemas.JavaFieldSchema POJO} field, a {@link
+ * org.apache.beam.sdk.schemas.JavaBeanSchema Java Bean} getter, or an {@link
+ * org.apache.beam.sdk.schemas.AutoValueSchema AutoValue} getter, the generated field will have the
+ * specified index. There cannot be "gaps" in field numbers, or schema inference will fail. If used,
+ * all fields (or getters in the case of a bean) must be annotated.
+ *
+ * <p>For example, say we have a Java POJO with a field that we want in our schema but under a
+ * different name:
+ *
+ * <pre><code>
+ *  {@literal @}DefaultSchema(JavaFieldSchema.class)
+ *   class MyClass {
+ *     {@literal @}SchemaFieldNumber(1)
+ *     public String user;
+ *
+ *    {@literal @}SchemaFieldNumber(0)
+ *     public int ageInYears;
+ *   }
+ * </code></pre>
+ *
+ * <p>The resulting schema will have ageInYears first followed by user.
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.FIELD, ElementType.METHOD})
+@SuppressWarnings({
+  "rawtypes" // TODO(https://issues.apache.org/jira/browse/BEAM-10556)
+})
+@Experimental(Kind.SCHEMAS)
+public @interface SchemaFieldNumber {
+
+  /** The name to use for the generated schema field. */
+  @Nonnull
+  String value();
+}