Skip to content

Commit

Permalink
foo
Browse files Browse the repository at this point in the history
  • Loading branch information
@reuvenlax
reuvenlax committed Oct 12, 2024
1 parent 240d6a1 commit 29019bf
Show file tree
Hide file tree
Showing 18 changed files with 308 additions and 209 deletions.
5 changes: 3 additions & 2 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/AutoValueSchema.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ public List<FieldValueTypeInformation> get(TypeDescriptor<?> typeDescriptor) {
.filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
.collect(Collectors.toList());
List<FieldValueTypeInformation> types = Lists.newArrayListWithCapacity(methods.size());
Map<Type, Type> boundTypes = SchemaUtils.getAllBoundTypes(typeDescriptor);
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
for (int i = 0; i < methods.size(); ++i) {
types.add(FieldValueTypeInformation.forGetter(methods.get(i), i, boundTypes));
}
Expand Down Expand Up @@ -146,7 +146,8 @@ public SchemaUserTypeCreator schemaTypeCreator(

@Override
public <T> @Nullable Schema schemaFor(TypeDescriptor<T> typeDescriptor) {
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
return JavaBeanUtils.schemaFromJavaBeanClass(
typeDescriptor, AbstractGetterTypeSupplier.INSTANCE);
typeDescriptor, AbstractGetterTypeSupplier.INSTANCE, boundTypes);
}
}
75 changes: 40 additions & 35 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/FieldValueTypeInformation.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -127,19 +127,19 @@ public static FieldValueTypeInformation forOneOf(
}

public static FieldValueTypeInformation forField(
Field field, int index, Map<Type, Type> boundParameters) {
Field field, int index, Map<Type, Type> boundTypes) {
TypeDescriptor<?> type =
TypeDescriptor.of(SchemaUtils.resolveType(field.getGenericType(), boundParameters));
TypeDescriptor.of(ReflectUtils.resolveType(field.getGenericType(), boundTypes));
return new AutoValue_FieldValueTypeInformation.Builder()
.setName(getNameOverride(field.getName(), field))
.setNumber(getNumberOverride(index, field))
.setNullable(hasNullableAnnotation(field))
.setType(type)
.setRawType(type.getRawType())
.setField(field)
.setElementType(getIterableComponentType(field))
.setMapKeyType(getMapKeyType(field))
.setMapValueType(getMapValueType(field))
.setElementType(getIterableComponentType(field, boundTypes))
.setMapKeyType(getMapKeyType(field, boundTypes))
.setMapValueType(getMapValueType(field, boundTypes))
.setOneOfTypes(Collections.emptyMap())
.setDescription(getFieldDescription(field))
.build();
Expand Down Expand Up @@ -188,7 +188,7 @@ public static <T extends AnnotatedElement & Member> String getNameOverride(
}

public static FieldValueTypeInformation forGetter(
Method method, int index, Map<Type, Type> boundParameters) {
Method method, int index, Map<Type, Type> boundTypes) {
String name;
if (method.getName().startsWith("get")) {
name = ReflectUtils.stripPrefix(method.getName(), "get");
Expand All @@ -199,7 +199,7 @@ public static FieldValueTypeInformation forGetter(
}

TypeDescriptor<?> type =
TypeDescriptor.of(SchemaUtils.resolveType(method.getGenericReturnType(), boundParameters));
TypeDescriptor.of(ReflectUtils.resolveType(method.getGenericReturnType(), boundTypes));
boolean nullable = hasNullableReturnType(method);
return new AutoValue_FieldValueTypeInformation.Builder()
.setName(getNameOverride(name, method))
Expand All @@ -208,9 +208,9 @@ public static FieldValueTypeInformation forGetter(
.setType(type)
.setRawType(type.getRawType())
.setMethod(method)
.setElementType(getIterableComponentType(type))
.setMapKeyType(getMapKeyType(type))
.setMapValueType(getMapValueType(type))
.setElementType(getIterableComponentType(type, boundTypes))
.setMapKeyType(getMapKeyType(type, boundTypes))
.setMapValueType(getMapValueType(type, boundTypes))
.setOneOfTypes(Collections.emptyMap())
.setDescription(getFieldDescription(method))
.build();
Expand Down Expand Up @@ -263,7 +263,7 @@ public static FieldValueTypeInformation forSetter(
}

public static FieldValueTypeInformation forSetter(
Method method, String setterPrefix, Map<Type, Type> boundParameters) {
Method method, String setterPrefix, Map<Type, Type> boundTypes) {
String name;
if (method.getName().startsWith(setterPrefix)) {
name = ReflectUtils.stripPrefix(method.getName(), setterPrefix);
Expand All @@ -273,17 +273,17 @@ public static FieldValueTypeInformation forSetter(

TypeDescriptor<?> type =
TypeDescriptor.of(
SchemaUtils.resolveType(method.getGenericParameterTypes()[0], boundParameters));
ReflectUtils.resolveType(method.getGenericParameterTypes()[0], boundTypes));
boolean nullable = hasSingleNullableParameter(method);
return new AutoValue_FieldValueTypeInformation.Builder()
.setName(name)
.setNullable(nullable)
.setType(type)
.setRawType(type.getRawType())
.setMethod(method)
.setElementType(getIterableComponentType(type))
.setMapKeyType(getMapKeyType(type))
.setMapValueType(getMapValueType(type))
.setElementType(getIterableComponentType(type, boundTypes))
.setMapKeyType(getMapKeyType(type, boundTypes))
.setMapValueType(getMapValueType(type, boundTypes))
.setOneOfTypes(Collections.emptyMap())
.build();
}
Expand All @@ -292,13 +292,15 @@ public FieldValueTypeInformation withName(String name) {
return toBuilder().setName(name).build();
}

private static FieldValueTypeInformation getIterableComponentType(Field field) {
return getIterableComponentType(TypeDescriptor.of(field.getGenericType()));
private static FieldValueTypeInformation getIterableComponentType(
Field field, Map<Type, Type> boundTypes) {
return getIterableComponentType(TypeDescriptor.of(field.getGenericType()), boundTypes);
}

static @Nullable FieldValueTypeInformation getIterableComponentType(TypeDescriptor<?> valueType) {
static @Nullable FieldValueTypeInformation getIterableComponentType(
TypeDescriptor<?> valueType, Map<Type, Type> boundTypes) {
// TODO: Figure out nullable elements.
TypeDescriptor<?> componentType = ReflectUtils.getIterableComponentType(valueType);
TypeDescriptor<?> componentType = ReflectUtils.getIterableComponentType(valueType, boundTypes);
if (componentType == null) {
return null;
}
Expand All @@ -308,41 +310,44 @@ private static FieldValueTypeInformation getIterableComponentType(Field field) {
.setNullable(false)
.setType(componentType)
.setRawType(componentType.getRawType())
.setElementType(getIterableComponentType(componentType))
.setMapKeyType(getMapKeyType(componentType))
.setMapValueType(getMapValueType(componentType))
// ????? Is this a mistake?
.setElementType(getIterableComponentType(componentType, boundTypes))
.setMapKeyType(getMapKeyType(componentType, boundTypes))
.setMapValueType(getMapValueType(componentType, boundTypes))
.setOneOfTypes(Collections.emptyMap())
.build();
}

// If the Field is a map type, returns the key type, otherwise returns a null reference.

private static @Nullable FieldValueTypeInformation getMapKeyType(Field field) {
return getMapKeyType(TypeDescriptor.of(field.getGenericType()));
private static @Nullable FieldValueTypeInformation getMapKeyType(
Field field, Map<Type, Type> boundTypes) {
return getMapKeyType(TypeDescriptor.of(field.getGenericType()), boundTypes);
}

private static @Nullable FieldValueTypeInformation getMapKeyType(
TypeDescriptor<?> typeDescriptor) {
return getMapType(typeDescriptor, 0);
TypeDescriptor<?> typeDescriptor, Map<Type, Type> boundTypes) {
return getMapType(typeDescriptor, 0, boundTypes);
}

// If the Field is a map type, returns the value type, otherwise returns a null reference.

private static @Nullable FieldValueTypeInformation getMapValueType(Field field) {
return getMapType(TypeDescriptor.of(field.getGenericType()), 1);
private static @Nullable FieldValueTypeInformation getMapValueType(
Field field, Map<Type, Type> boundTypes) {
return getMapType(TypeDescriptor.of(field.getGenericType()), 1, boundTypes);
}

private static @Nullable FieldValueTypeInformation getMapValueType(
TypeDescriptor<?> typeDescriptor) {
return getMapType(typeDescriptor, 1);
TypeDescriptor<?> typeDescriptor, Map<Type, Type> boundTypes) {
return getMapType(typeDescriptor, 1, boundTypes);
}

// If the Field is a map type, returns the key or value type (0 is key type, 1 is value).
// Otherwise returns a null reference.
@SuppressWarnings("unchecked")
private static @Nullable FieldValueTypeInformation getMapType(
TypeDescriptor<?> valueType, int index) {
TypeDescriptor mapType = ReflectUtils.getMapType(valueType, index);
TypeDescriptor<?> valueType, int index, Map<Type, Type> boundTypes) {
TypeDescriptor mapType = ReflectUtils.getMapType(valueType, index, boundTypes);
if (mapType == null) {
return null;
}
Expand All @@ -351,9 +356,9 @@ private static FieldValueTypeInformation getIterableComponentType(Field field) {
.setNullable(false)
.setType(mapType)
.setRawType(mapType.getRawType())
.setElementType(getIterableComponentType(mapType))
.setMapKeyType(getMapKeyType(mapType))
.setMapValueType(getMapValueType(mapType))
.setElementType(getIterableComponentType(mapType, boundTypes))
.setMapKeyType(getMapKeyType(mapType, boundTypes))
.setMapValueType(getMapValueType(mapType, boundTypes))
.setOneOfTypes(Collections.emptyMap())
.build();
}
Expand Down
8 changes: 5 additions & 3 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/JavaBeanSchema.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ public List<FieldValueTypeInformation> get(TypeDescriptor<?> typeDescriptor) {
.filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
.collect(Collectors.toList());
List<FieldValueTypeInformation> types = Lists.newArrayListWithCapacity(methods.size());
Map<Type, Type> boundTypes = SchemaUtils.getAllBoundTypes(typeDescriptor);
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
for (int i = 0; i < methods.size(); ++i) {
types.add(FieldValueTypeInformation.forGetter(methods.get(i), i, boundTypes));
}
Expand Down Expand Up @@ -114,7 +114,7 @@ public static class SetterTypeSupplier implements FieldValueTypeSupplier {

@Override
public List<FieldValueTypeInformation> get(TypeDescriptor<?> typeDescriptor) {
Map<Type, Type> boundTypes = SchemaUtils.getAllBoundTypes(typeDescriptor);
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
return ReflectUtils.getMethods(typeDescriptor.getRawType()).stream()
.filter(ReflectUtils::isSetter)
.filter(m -> !m.isAnnotationPresent(SchemaIgnore.class))
Expand Down Expand Up @@ -160,8 +160,10 @@ public boolean equals(@Nullable Object obj) {

@Override
public <T> Schema schemaFor(TypeDescriptor<T> typeDescriptor) {
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
Schema schema =
JavaBeanUtils.schemaFromJavaBeanClass(typeDescriptor, GetterTypeSupplier.INSTANCE);
JavaBeanUtils.schemaFromJavaBeanClass(
typeDescriptor, GetterTypeSupplier.INSTANCE, boundTypes);

// If there are no creator methods, then validate that we have setters for every field.
// Otherwise, we will have no way of creating instances of the class.
Expand Down
6 changes: 4 additions & 2 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/JavaFieldSchema.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ public List<FieldValueTypeInformation> get(TypeDescriptor<?> typeDescriptor) {
.collect(Collectors.toList());

List<FieldValueTypeInformation> types = Lists.newArrayListWithCapacity(fields.size());
Map<Type, Type> boundTypes = SchemaUtils.getAllBoundTypes(typeDescriptor);
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
for (int i = 0; i < fields.size(); ++i) {
types.add(FieldValueTypeInformation.forField(fields.get(i), i, boundTypes));
}
Expand Down Expand Up @@ -115,7 +115,9 @@ private static void validateFieldNumbers(List<FieldValueTypeInformation> types)

@Override
public <T> Schema schemaFor(TypeDescriptor<T> typeDescriptor) {
return POJOUtils.schemaFromPojoClass(typeDescriptor, JavaFieldTypeSupplier.INSTANCE);
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(typeDescriptor);
return POJOUtils.schemaFromPojoClass(
typeDescriptor, JavaFieldTypeSupplier.INSTANCE, boundTypes);
}

@Override
Expand Down
50 changes: 0 additions & 50 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,8 @@
*/
package org.apache.beam.sdk.schemas;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.lang.reflect.TypeVariable;
import java.util.Map;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.Schema.LogicalType;
import org.apache.beam.sdk.util.Preconditions;
import org.apache.beam.sdk.values.TypeDescriptor;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Maps;

/** A set of utility functions for schemas. */
@SuppressWarnings({
Expand Down Expand Up @@ -129,47 +122,4 @@ public static <BaseT, InputT> InputT toLogicalInputType(
LogicalType<InputT, BaseT> logicalType, BaseT baseType) {
return logicalType.toInputType(baseType);
}

public static <T> Map<Type, Type> getAllBoundTypes(TypeDescriptor<T> typeDescriptor) {
Map<Type, Type> boundParameters = Maps.newHashMap();
TypeDescriptor<?> currentType = typeDescriptor;
do {
if (currentType.getType() instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) currentType.getType();
TypeVariable<?>[] typeVariables = currentType.getRawType().getTypeParameters();
Type[] typeArguments = parameterizedType.getActualTypeArguments();
;
if (typeArguments.length != typeVariables.length) {
throw new RuntimeException("Unmatching arguments lengths");
}
for (int i = 0; i < typeVariables.length; ++i) {
boundParameters.put(typeVariables[i], typeArguments[i]);
}
}
Type superClass = currentType.getRawType().getGenericSuperclass();
if (superClass == null || superClass.equals(Object.class)) {
break;
}
currentType = TypeDescriptor.of(superClass);
} while (true);
return boundParameters;
}

public static Type resolveType(Type type, Map<Type, Type> boundTypes) {
TypeDescriptor<?> typeDescriptor = TypeDescriptor.of(type);
if (typeDescriptor.isSubtypeOf(TypeDescriptor.of(Iterable.class))
|| typeDescriptor.isSubtypeOf(TypeDescriptor.of(Map.class))) {
// Don't resolve these as we special case map and interable.
return type;
}

if (type instanceof TypeVariable) {
TypeVariable<?> typeVariable = (TypeVariable<?>) type;
return Preconditions.checkArgumentNotNull(boundTypes.get(typeVariable));
} else if (type instanceof ParameterizedType) {
throw new RuntimeException("Parameterized types not supported yet.");
} else {
return type;
}
}
}
3 changes: 2 additions & 1 deletion .../java/core/src/main/java/org/apache/beam/sdk/schemas/transforms/providers/JavaRowUdf.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,8 @@ public FunctionAndType(Type outputType, Function<Row, Object> function) {

public FunctionAndType(TypeDescriptor<?> outputType, Function<Row, Object> function) {
this(
StaticSchemaInference.fieldFromType(outputType, new EmptyFieldValueTypeSupplier()),
StaticSchemaInference.fieldFromType(
outputType, new EmptyFieldValueTypeSupplier(), Collections.emptyMap()),
function);
}

Expand Down
3 changes: 1 addition & 2 deletions sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AutoValueUtils.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,6 @@
import org.apache.beam.sdk.schemas.FieldValueTypeInformation;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.SchemaUserTypeCreator;
import org.apache.beam.sdk.schemas.SchemaUtils;
import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.DefaultTypeConversionsFactory;
import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.InjectPackageStrategy;
import org.apache.beam.sdk.schemas.utils.ByteBuddyUtils.TypeConversion;
Expand Down Expand Up @@ -197,7 +196,7 @@ private static boolean matchConstructor(
return null;
}

Map<Type, Type> boundTypes = SchemaUtils.getAllBoundTypes(TypeDescriptor.of(builderClass));
Map<Type, Type> boundTypes = ReflectUtils.getAllBoundTypes(TypeDescriptor.of(builderClass));
Map<String, FieldValueTypeInformation> setterTypes =
ReflectUtils.getMethods(builderClass).stream()
.filter(ReflectUtils::isSetter)
Expand Down
Loading

0 comments on commit 29019bf

Please sign in to comment.