Skip to content

GH-3899: Remove Guava from parquet-protobuf main code #3905

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Fokko wants to merge 1 commit into from
+22 −24
Closed

GH-3899: Remove Guava from parquet-protobuf main code #3905

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 22 additions & 24 deletions parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoSchemaConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;

import com.google.common.collect.ImmutableSetMultimap;
import com.google.protobuf.BoolValue;
import com.google.protobuf.BytesValue;
import com.google.protobuf.Descriptors;
Expand All @@ -52,7 +51,11 @@
import com.google.type.Date;
import com.google.type.TimeOfDay;
import com.twitter.elephantbird.util.Protobufs;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.conf.HadoopParquetConfiguration;
Expand Down Expand Up @@ -194,7 +197,7 @@ public static void setMaxRecursion(Configuration config, int maxRecursion) {
public MessageType convert(Descriptors.Descriptor descriptor) {
// Remember classes seen with depths to avoid cycles.
int depth = 0;
ImmutableSetMultimap<String, Integer> seen = ImmutableSetMultimap.of(descriptor.getFullName(), depth);
Map<String, Set<Integer>> seen = Map.of(descriptor.getFullName(), Set.of(depth));
LOG.trace("convert:\n{}", descriptor.toProto());
MessageType messageType = convertFields(Types.buildMessage(), descriptor.getFields(), seen, depth)
.named(descriptor.getFullName());
Expand All @@ -218,7 +221,7 @@ public MessageType convert(Class<? extends Message> protobufClass) {
private <T> GroupBuilder<T> convertFields(
GroupBuilder<T> groupBuilder,
List<FieldDescriptor> fieldDescriptors,
ImmutableSetMultimap<String, Integer> seen,
Map<String, Set<Integer>> seen,
int depth) {
for (FieldDescriptor fieldDescriptor : fieldDescriptors) {
groupBuilder = addField(fieldDescriptor, groupBuilder, seen, depth)
Expand All @@ -239,10 +242,7 @@ private static Type.Repetition getRepetition(FieldDescriptor descriptor) {
}

private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addField(
FieldDescriptor descriptor,
final GroupBuilder<T> builder,
ImmutableSetMultimap<String, Integer> seen,
int depth) {
FieldDescriptor descriptor, final GroupBuilder<T> builder, Map<String, Set<Integer>> seen, int depth) {
if (descriptor.getJavaType() == JavaType.MESSAGE) {
if (unwrapProtoWrappers) {
Descriptor messageType = descriptor.getMessageType();
Expand Down Expand Up @@ -314,10 +314,7 @@ private static <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T
}

private <T> GroupBuilder<GroupBuilder<T>> addRepeatedMessage(
FieldDescriptor descriptor,
GroupBuilder<T> builder,
ImmutableSetMultimap<String, Integer> seen,
int depth) {
FieldDescriptor descriptor, GroupBuilder<T> builder, Map<String, Set<Integer>> seen, int depth) {
GroupBuilder<GroupBuilder<GroupBuilder<GroupBuilder<T>>>> result = builder.group(Type.Repetition.OPTIONAL)
.as(listType())
.group(Type.Repetition.REPEATED)
Expand All @@ -329,10 +326,7 @@ private <T> GroupBuilder<GroupBuilder<T>> addRepeatedMessage(
}

private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addMessageField(
FieldDescriptor descriptor,
final GroupBuilder<T> builder,
ImmutableSetMultimap<String, Integer> seen,
int depth) {
FieldDescriptor descriptor, final GroupBuilder<T> builder, Map<String, Set<Integer>> seen, int depth) {
// Terminate with proto bytes anything a static parquet schema cannot represent - recursion
// beyond maxRecursion and empty message types (parquet forbids empty groups) - preserving the
// field's repetition so the write path (Array/Repeated/MapWriter) still matches the schema.
Expand All @@ -348,7 +342,8 @@ private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addM
}

boolean emptyMessage = descriptor.getMessageType().getFields().isEmpty();
if (emptyMessage || (typeName != null && seen.get(typeName).size() > maxRecursion)) {
if (emptyMessage
|| (typeName != null && seen.getOrDefault(typeName, Set.of()).size() > maxRecursion)) {
if (descriptor.isRepeated() && parquetSpecsCompliant) {
// LIST-wrap the truncated bytes the same way any repeated primitive is wrapped
return addRepeatedPrimitive(BINARY, null, builder);
Expand All @@ -357,10 +352,7 @@ private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addM
return builder.primitive(BINARY, getRepetition(descriptor)).as((LogicalTypeAnnotation) null);
}

seen = ImmutableSetMultimap.<String, Integer>builder()
.putAll(seen)
.put(typeName, depth)
.build();
seen = withSeen(seen, typeName, depth);

if (descriptor.isRepeated() && parquetSpecsCompliant) {
// the old schema style did not include the LIST wrapper around repeated messages
Expand All @@ -373,6 +365,15 @@ private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addM
return group;
}

// Returns an immutable copy of seen with depth added for typeName, leaving seen untouched.
private static Map<String, Set<Integer>> withSeen(Map<String, Set<Integer>> seen, String typeName, int depth) {
Set<Integer> depths = new HashSet<>(seen.getOrDefault(typeName, Set.of()));
depths.add(depth);
Map<String, Set<Integer>> result = new HashMap<>(seen);
result.put(typeName, Set.copyOf(depths));
return Map.copyOf(result);
}

@Nullable private String getInnerTypeName(FieldDescriptor descriptor) {
if (descriptor.isMapField() && parquetSpecsCompliant) {
descriptor = descriptor.getMessageType().getFields().get(1);
Expand All @@ -387,10 +388,7 @@ private <T> Builder<? extends Builder<?, GroupBuilder<T>>, GroupBuilder<T>> addM
}

private <T> GroupBuilder<GroupBuilder<T>> addMapField(
FieldDescriptor descriptor,
final GroupBuilder<T> builder,
ImmutableSetMultimap<String, Integer> seen,
int depth) {
FieldDescriptor descriptor, final GroupBuilder<T> builder, Map<String, Set<Integer>> seen, int depth) {
List<FieldDescriptor> fields = descriptor.getMessageType().getFields();
if (fields.size() != 2) {
throw new UnsupportedOperationException("Expected two fields for the map (key/value), but got: " + fields);
Expand Down
Loading