Skip to content

Commit

Permalink
added support for "avro.java.string" property
Browse files Browse the repository at this point in the history 
- string type support now adheres to the native avro implementation
- Utf8 is the default string type as in the native implementation (previously: java.lang.String)
- serializers use java.lang.CharSequence, thus support all types of provided string types
- minor fixes in test classes related to introduced string type support changes
  • Loading branch information
@flowenol
flowenol committed May 27, 2019
1 parent f7357ed commit 0221998
Show file tree
Hide file tree
Showing 11 changed files with 828 additions and 335 deletions.
2 changes: 1 addition & 1 deletion pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

<groupId>com.rtbhouse</groupId>
<artifactId>avro-fastserde</artifactId>
<version>1.0.4</version>
<version>1.0.5-SNAPSHOT</version>
<packaging>jar</packaging>

<name>avro-fastserde</name>
Expand Down
76 changes: 55 additions & 21 deletions src/main/java/com/rtbhouse/utils/avro/FastDeserializerGenerator.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,12 +53,15 @@ public class FastDeserializerGenerator<T> extends FastDeserializerGeneratorBase<
private Map<JMethod, Set<Class<? extends Exception>>> exceptionFromMethodMap = new HashMap<>();
private SchemaAssistant schemaAssistant;

private JClass string;

FastDeserializerGenerator(boolean useGenericTypes, Schema writer, Schema reader, File destination,
ClassLoader classLoader,
String compileClassPath) {
super(writer, reader, destination, classLoader, compileClassPath);
this.useGenericTypes = useGenericTypes;
this.schemaAssistant = new SchemaAssistant(codeModel, useGenericTypes);
this.schemaAssistant = new SchemaAssistant(codeModel, useGenericTypes, false);
this.string = codeModel.ref(String.class);
}

public FastDeserializer<T> generateDeserializer() {
Expand Down Expand Up @@ -88,9 +91,8 @@ public FastDeserializer<T> generateDeserializer() {
}

JClass readerSchemaClass = schemaAssistant.classFromSchema(reader);
JClass writerSchemaClass = schemaAssistant.classFromSchema(aliasedWriterSchema);

deserializerClass._implements(codeModel.ref(FastDeserializer.class).narrow(writerSchemaClass));
deserializerClass._implements(codeModel.ref(FastDeserializer.class).narrow(readerSchemaClass));
JMethod deserializeMethod = deserializerClass.method(JMod.PUBLIC, readerSchemaClass, "deserialize");

JBlock topLevelDeserializeBlock = new JBlock();
Expand Down Expand Up @@ -262,8 +264,12 @@ private void processRecord(JVar recordSchemaVar, String recordName, final Schema
processComplexType(fieldSchemaVar, field.name(), field.schema(), readerFieldSchema, methodBody, action,
putExpressionInRecord);
} else {
processSimpleType(field.schema(), methodBody, action, putExpressionInRecord);

// to preserve reader string specific options use reader field schema
if (action.getShouldRead() && Schema.Type.STRING.equals(field.schema().getType())) {
processSimpleType(readerFieldSchema, methodBody, action, putExpressionInRecord);
} else {
processSimpleType(field.schema(), methodBody, action, putExpressionInRecord);
}
}
}

Expand Down Expand Up @@ -489,7 +495,13 @@ private void processUnion(JVar unionSchemaVar, final String name, final Schema u
processComplexType(optionSchemaVar, optionName, optionSchema, readerOptionSchema, thenBlock,
unionAction, putValueIntoParent);
} else {
processSimpleType(optionSchema, thenBlock, unionAction, putValueIntoParent);
// to preserve reader string specific options use reader option schema
if (action.getShouldRead() && Schema.Type.STRING.equals(optionSchema.getType())) {
processSimpleType(readerOptionSchema, thenBlock, unionAction, putValueIntoParent);

} else {
processSimpleType(optionSchema, thenBlock, unionAction, putValueIntoParent);
}
}
}
}
Expand Down Expand Up @@ -518,14 +530,14 @@ private void processArray(JVar arraySchemaVar, final String name, final Schema a
action = FieldAction.fromValues(arraySchema.getElementType().getType(), false, EMPTY_SYMBOL);
}

final JVar arrayVar = action.getShouldRead() ? declareValueVar(name, arraySchema, parentBody) : null;
final JVar arrayVar = action.getShouldRead() ? declareValueVar(name, readerArraySchema, parentBody) : null;
JVar chunkLen = parentBody.decl(codeModel.LONG, getVariableName("chunkLen"),
JExpr.direct(DECODER + ".readArrayStart()"));

JConditional conditional = parentBody._if(chunkLen.gt(JExpr.lit(0)));
JBlock ifBlock = conditional._then();

JClass arrayClass = schemaAssistant.classFromSchema(arraySchema, false);
JClass arrayClass = schemaAssistant.classFromSchema(readerArraySchema, false);

if (action.getShouldRead()) {
JInvocation newArrayExp = JExpr._new(arrayClass);
Expand Down Expand Up @@ -560,14 +572,16 @@ private void processArray(JVar arraySchemaVar, final String name, final Schema a

if (SchemaAssistant.isComplexType(arraySchema.getElementType())) {
String elemName = name + "Elem";
Schema readerArrayElementSchema = null;
if (action.getShouldRead()) {
readerArrayElementSchema = readerArraySchema.getElementType();
}
Schema readerArrayElementSchema = action.getShouldRead() ? readerArraySchema.getElementType() : null;
processComplexType(elementSchemaVar, elemName, arraySchema.getElementType(), readerArrayElementSchema,
forBody, action, putValueInArray);
} else {
processSimpleType(arraySchema.getElementType(), forBody, action, putValueInArray);
// to preserve reader string specific options use reader array schema
if (action.getShouldRead() && Schema.Type.STRING.equals(arraySchema.getElementType().getType())) {
processSimpleType(readerArraySchema.getElementType(), forBody, action, putValueInArray);
} else {
processSimpleType(arraySchema.getElementType(), forBody, action, putValueInArray);
}
}
doLoop.body().assign(chunkLen, JExpr.direct(DECODER + ".arrayNext()"));

Expand Down Expand Up @@ -599,15 +613,15 @@ private void processMap(JVar mapSchemaVar, final String name, final Schema mapSc
action = FieldAction.fromValues(mapSchema.getValueType().getType(), false, EMPTY_SYMBOL);
}

final JVar mapVar = action.getShouldRead() ? declareValueVar(name, mapSchema, parentBody) : null;
final JVar mapVar = action.getShouldRead() ? declareValueVar(name, readerMapSchema, parentBody) : null;
JVar chunkLen = parentBody.decl(codeModel.LONG, getVariableName("chunkLen"),
JExpr.direct(DECODER + ".readMapStart()"));

JConditional conditional = parentBody._if(chunkLen.gt(JExpr.lit(0)));
JBlock ifBlock = conditional._then();

if (action.getShouldRead()) {
ifBlock.assign(mapVar, JExpr._new(schemaAssistant.classFromSchema(mapSchema, false)));
ifBlock.assign(mapVar, JExpr._new(schemaAssistant.classFromSchema(readerMapSchema, false)));
JBlock elseBlock = conditional._else();
elseBlock.assign(mapVar, codeModel.ref(Collections.class).staticInvoke("emptyMap"));
}
Expand All @@ -619,10 +633,13 @@ private void processMap(JVar mapSchemaVar, final String name, final Schema mapSc
forLoop.update(counter.incr());
JBlock forBody = forLoop.body();

JClass keyClass = schemaAssistant.keyClassFromMapSchema(mapSchema);
JExpression keyValueExpression = JExpr.direct(DECODER + ".readString()");
JClass keyClass = schemaAssistant.keyClassFromMapSchema(readerMapSchema);
JExpression keyValueExpression = (string.equals(keyClass)) ?
JExpr.direct(DECODER + ".readString()")
: JExpr.direct(DECODER + ".readString(null)");

if (SchemaAssistant.hasStringableKey(mapSchema)) {
keyValueExpression = JExpr._new(keyClass).arg(keyValueExpression);
keyValueExpression = JExpr._new(keyClass).arg(keyValueExpression.invoke("toString"));
}

JVar key = forBody.decl(keyClass, getVariableName("key"), keyValueExpression);
Expand All @@ -631,10 +648,12 @@ private void processMap(JVar mapSchemaVar, final String name, final Schema mapSc
mapValueSchemaVar = declareSchemaVar(mapSchema.getValueType(), name + "MapValueSchema",
mapSchemaVar.invoke("getValueType"));
}

BiConsumer<JBlock, JExpression> putValueInMap = null;
if (action.getShouldRead()) {
putValueInMap = (block, expression) -> block.invoke(mapVar, "put").arg(key).arg(expression);
}

if (SchemaAssistant.isComplexType(mapSchema.getValueType())) {
String valueName = name + "Value";
Schema readerMapValueSchema = null;
Expand All @@ -644,9 +663,15 @@ private void processMap(JVar mapSchemaVar, final String name, final Schema mapSc
processComplexType(mapValueSchemaVar, valueName, mapSchema.getValueType(), readerMapValueSchema, forBody,
action, putValueInMap);
} else {
processSimpleType(mapSchema.getValueType(), forBody, action, putValueInMap);
// to preserve reader string specific options use reader map schema
if (action.getShouldRead() && Schema.Type.STRING.equals(mapSchema.getValueType().getType())) {
processSimpleType(readerMapSchema.getValueType(), forBody, action, putValueInMap);
} else {
processSimpleType(mapSchema.getValueType(), forBody, action, putValueInMap);
}
}
doLoop.body().assign(chunkLen, JExpr.direct(DECODER + ".mapNext()"));

if (action.getShouldRead()) {
putMapIntoParent.accept(parentBody, mapVar);
}
Expand Down Expand Up @@ -727,7 +752,16 @@ private void processPrimitive(final Schema schema, JBlock body, FieldAction acti
String readFunction;
switch (schema.getType()) {
case STRING:
readFunction = action.getShouldRead() ? "readString()" : "skipString()";
if (action.getShouldRead()) {
if (string.equals(schemaAssistant.classFromSchema(schema))) {
readFunction = "readString()";
} else {
// reads as Utf8
readFunction = "readString(null)";
}
} else {
readFunction = "skipString()";
}
break;
case BYTES:
readFunction = "readBytes(null)";
Expand Down Expand Up @@ -756,7 +790,7 @@ private void processPrimitive(final Schema schema, JBlock body, FieldAction acti
if (action.getShouldRead()) {
if (schema.getType().equals(Schema.Type.STRING) && SchemaAssistant.isStringable(schema)) {
primitiveValueExpression = JExpr._new(schemaAssistant.classFromSchema(schema))
.arg(primitiveValueExpression);
.arg(primitiveValueExpression.invoke("toString"));
}
putValueIntoParent.accept(body, primitiveValueExpression);
} else {
Expand Down
Loading

0 comments on commit 0221998

Please sign in to comment.