Skip to content

Add AvroConverter support for Decimal, Date, Time, and Timestamp logical types. #234

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

Merged
Ewen Cheslack-Postava (ewencp) merged 2 commits into from
Oct 9, 2015
Merged

Add AvroConverter support for Decimal, Date, Time, and Timestamp logical types. #234

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c34635d
Add AvroConverter support for Decimal, Date, Time, and Timestamp logi…
ewencp Oct 8, 2015
b88ec1e
Fix an incomplete comment and clarify an exception message.
ewencp Oct 9, 2015
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
204 changes: 180 additions & 24 deletions avro-converter/src/main/java/io/confluent/copycat/avro/AvroData.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,24 +18,28 @@

import io.confluent.kafka.serializers.NonRecordContainer;

import org.apache.avro.AvroRuntimeException;
import org.apache.avro.generic.GenericEnumSymbol;
import org.apache.avro.generic.GenericFixed;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
import org.apache.avro.generic.IndexedRecord;
import org.apache.kafka.copycat.data.Date;
import org.apache.kafka.copycat.data.Decimal;
import org.apache.kafka.copycat.data.Field;
import org.apache.kafka.copycat.data.Schema;
import org.apache.kafka.copycat.data.SchemaAndValue;
import org.apache.kafka.copycat.data.SchemaBuilder;
import org.apache.kafka.copycat.data.Struct;
import org.apache.kafka.copycat.data.Time;
import org.apache.kafka.copycat.data.Timestamp;
import org.apache.kafka.copycat.errors.DataException;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.node.ArrayNode;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;

import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
Expand Down Expand Up @@ -65,6 +69,7 @@ public class AvroData {
public static final String COPYCAT_DOC_PROP = "copycat.doc";
public static final String COPYCAT_VERSION_PROP = "copycat.version";
public static final String COPYCAT_DEFAULT_VALUE_PROP = "copycat.default";
public static final String COPYCAT_PARAMETERS_PROP = "copycat.parameters";

public static final String COPYCAT_TYPE_PROP = "copycat.type";

Expand Down Expand Up @@ -153,6 +158,94 @@ public class AvroData {
.getElementType();
}

// Convert values in Copycat form into their logical types. These logical converters are discovered by logical type
// names specified in the field
private static final HashMap<String, LogicalTypeConverter> TO_COPYCAT_LOGICAL_CONVERTERS = new HashMap<>();
static {
TO_COPYCAT_LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (value instanceof byte[]) {
return Decimal.toLogical(schema, (byte[]) value);
} else if (value instanceof ByteBuffer) {
return Decimal.toLogical(schema, ((ByteBuffer) value).array());
}
throw new DataException("Invalid type for Decimal, underlying representation should be bytes but was " + value.getClass());
}
});

TO_COPYCAT_LOGICAL_CONVERTERS.put(Date.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof Integer))
throw new DataException("Invalid type for Date, underlying representation should be int32 but was " + value.getClass());
return Date.toLogical(schema, (int) value);
}
});

TO_COPYCAT_LOGICAL_CONVERTERS.put(Time.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof Integer))
throw new DataException("Invalid type for Time, underlying representation should be int32 but was " + value.getClass());
return Time.toLogical(schema, (int) value);
}
});

TO_COPYCAT_LOGICAL_CONVERTERS.put(Timestamp.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof Long))
throw new DataException("Invalid type for Timestamp, underlying representation should be int64 but was " + value.getClass());
return Timestamp.toLogical(schema, (long) value);
}
});
}

private static final HashMap<String, LogicalTypeConverter> TO_AVRO_LOGICAL_CONVERTERS
= new HashMap<>();
static {
TO_AVRO_LOGICAL_CONVERTERS.put(Decimal.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof BigDecimal))
throw new DataException(
"Invalid type for Decimal, expected BigDecimal but was " + value.getClass());
return Decimal.fromLogical(schema, (BigDecimal) value);
}
});

TO_AVRO_LOGICAL_CONVERTERS.put(Date.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof java.util.Date))
throw new DataException(
"Invalid type for Date, expected Date but was " + value.getClass());
return Date.fromLogical(schema, (java.util.Date) value);
}
});

TO_AVRO_LOGICAL_CONVERTERS.put(Time.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof java.util.Date))
throw new DataException(
"Invalid type for Time, expected Date but was " + value.getClass());
return Time.fromLogical(schema, (java.util.Date) value);
}
});

TO_AVRO_LOGICAL_CONVERTERS.put(Timestamp.LOGICAL_NAME, new LogicalTypeConverter() {
@Override
public Object convert(Schema schema, Object value) {
if (!(value instanceof java.util.Date))
throw new DataException(
"Invalid type for Timestamp, expected Date but was " + value.getClass());
return Timestamp.fromLogical(schema, (java.util.Date) value);
}
});
}

/**
* Convert this object, in Copycat data format, into an Avro object.
*/
Expand All @@ -166,24 +259,32 @@ public static Object fromCopycatData(Schema schema, Object value) {
* been converted and makes the use of NonRecordContainer optional
* @param schema the Copycat schema
* @param avroSchema the corresponding
* @param value the Copycat data to convert
* @param logicalValue the Copycat data to convert, which may be a value for a logical type
* @param requireContainer if true, wrap primitives, maps, and arrays in a NonRecordContainer
* before returning them
* @return the converted data
*/
private static Object fromCopycatData(Schema schema, org.apache.avro.Schema avroSchema,
Object value, boolean requireContainer) {
Schema.Type schemaType = schema != null ? schema.type() : schemaTypeForSchemalessJavaType(value);
Object logicalValue, boolean requireContainer) {
Schema.Type schemaType = schema != null ? schema.type() : schemaTypeForSchemalessJavaType(logicalValue);
if (schemaType == null) {
// Schemaless null data
return maybeAddContainer(avroSchema, maybeWrapSchemaless(null, null, null), requireContainer);
}

boolean schemaOptional = schema != null ? schema.isOptional() : true;
if (!schemaOptional && value == null) {
if (!schemaOptional && logicalValue == null) {
throw new DataException("Found null value for non-optional schema");
}

// If this is a logical type, decode and

@Ishiihara Liquan Pei (Ishiihara) Oct 9, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

missing some comments here?

Object value = logicalValue;
if (schema != null && schema.name() != null) {
LogicalTypeConverter logicalConverter = TO_AVRO_LOGICAL_CONVERTERS.get(schema.name());
if (logicalConverter != null)
value = logicalConverter.convert(schema, logicalValue);
}

try {
switch (schemaType) {
case INT8: {
Expand Down Expand Up @@ -456,6 +557,9 @@ public static org.apache.avro.Schema fromCopycatSchema(Schema schema) {
baseSchema.addProp(COPYCAT_VERSION_PROP,
JsonNodeFactory.instance.numberNode(schema.version()));
}
if (schema.parameters() != null) {
baseSchema.addProp(COPYCAT_PARAMETERS_PROP, parametersFromCopycat(schema.parameters()));
}
if (schema.defaultValue() != null) {
baseSchema.addProp(COPYCAT_DEFAULT_VALUE_PROP,
defaultValueFromCopycat(schema, schema.defaultValue()));
Expand Down Expand Up @@ -582,6 +686,14 @@ private static JsonNode defaultValueFromCopycat(Schema schema, Object defaultVal
}
}

private static JsonNode parametersFromCopycat(Map<String, String> params) {
ObjectNode result = JsonNodeFactory.instance.objectNode();
for (Map.Entry<String, String> entry : params.entrySet()) {
result.put(entry.getKey(), entry.getValue());
}
return result;
}

/**
* Convert the given object, in Avro format, into an Copycat data object.
*/
Expand Down Expand Up @@ -657,57 +769,67 @@ private static Object toCopycatData(Schema schema, Object value) {
return null;
}

Object converted = null;
switch (schema.type()) {
// Pass through types
case INT32: {
Integer intValue = (Integer) value; // Validate type
return value;
converted = value;
break;
}
case INT64: {
Long longValue = (Long) value; // Validate type
return value;
converted = value;
break;
}
case FLOAT32: {
Float floatValue = (Float) value; // Validate type
return value;
converted = value;
break;
}
case FLOAT64: {
Double doubleValue = (Double) value; // Validate type
return value;
converted = value;
break;
}
case BOOLEAN: {
Boolean boolValue = (Boolean) value; // Validate type
return value;
converted = value;
break;
}

case INT8:
// Encoded as an Integer
return ((Integer) value).byteValue();
converted = ((Integer) value).byteValue();
break;
case INT16:
// Encoded as an Integer
return ((Integer) value).shortValue();
converted = ((Integer) value).shortValue();
break;

case STRING:
if (value instanceof String) {
return value;
converted = value;
} else if (value instanceof CharSequence ||
value instanceof GenericEnumSymbol ||
value instanceof Enum) {
return value.toString();
converted = value.toString();
} else {
throw new DataException("Invalid class for string type, expecting String or "
+ "CharSequence but found " + value.getClass());
}
break;

case BYTES:
if (value instanceof byte[]) {
return ByteBuffer.wrap((byte[]) value);
converted = ByteBuffer.wrap((byte[]) value);
} else if (value instanceof ByteBuffer) {
return value;
converted = value;
} else {
throw new DataException("Invalid class for bytes type, expecting byte[] or ByteBuffer "
+ "but found " + value.getClass());
}
break;

case ARRAY: {
Schema valueSchema = schema.valueSchema();
Expand All @@ -716,7 +838,8 @@ private static Object toCopycatData(Schema schema, Object value) {
for (Object elem : original) {
result.add(toCopycatData(valueSchema, elem));
}
return result;
converted = result;
break;
}

case MAP: {
Expand All @@ -731,7 +854,7 @@ private static Object toCopycatData(Schema schema, Object value) {
result.put(entry.getKey().toString(),
toCopycatData(valueSchema, entry.getValue()));
}
return result;
converted = result;
} else {
// Arbitrary keys
List<IndexedRecord> original = (List<IndexedRecord>) value;
Expand All @@ -743,8 +866,9 @@ private static Object toCopycatData(Schema schema, Object value) {
Object convertedValue = toCopycatData(valueSchema, entry.get(avroValueFieldIndex));
result.put(convertedKey, convertedValue);
}
return result;
converted = result;
}
break;
}

case STRUCT: {
Expand All @@ -760,12 +884,15 @@ private static Object toCopycatData(Schema schema, Object value) {

if (isInstanceOfAvroSchemaTypeForSimpleSchema(fieldSchema, value) ||
(valueRecordSchema != null && valueRecordSchema.equals(fieldSchema))) {
return new Struct(schema).put(unionMemberFieldName(fieldSchema),
toCopycatData(fieldSchema, value));
converted = new Struct(schema).put(unionMemberFieldName(fieldSchema),
toCopycatData(fieldSchema, value));
break;
}
}
throw new DataException(
"Did not find matching union field for data: " + value.toString());
if (converted == null) {
throw new DataException(
"Did not find matching union field for data: " + value.toString());
}
} else {
IndexedRecord original = (IndexedRecord) value;
Struct result = new Struct(schema);
Expand All @@ -775,13 +902,21 @@ private static Object toCopycatData(Schema schema, Object value) {
= toCopycatData(field.schema(), original.get(avroFieldIndex));
result.put(field, convertedFieldValue);
}
return result;
converted = result;
}
break;
}

default:
throw new DataException("Unknown Copycat schema type: " + schema.type());
}

if (schema != null && schema.name() != null) {
LogicalTypeConverter logicalConverter = TO_COPYCAT_LOGICAL_CONVERTERS.get(schema.name());
if (logicalConverter != null)
converted = logicalConverter.convert(schema, converted);
}
return converted;
} catch (ClassCastException e) {
throw new DataException("Invalid type for " + schema.type() + ": " + value.getClass());
}
Expand Down Expand Up @@ -950,6 +1085,23 @@ public static Schema toCopycatSchema(org.apache.avro.Schema schema, boolean forc
builder.version(version.asInt());
}

JsonNode parameters = schema.getJsonProp(COPYCAT_PARAMETERS_PROP);
if (parameters != null) {
if (!parameters.isObject()) {
throw new DataException("Expected map for schema parameters but found: " + parameters);

@Ishiihara Liquan Pei (Ishiihara) Oct 9, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Expected objectNode?

}
Iterator<Map.Entry<String, JsonNode>> paramIt = parameters.getFields();
while (paramIt.hasNext()) {
Map.Entry<String, JsonNode> field = paramIt.next();
JsonNode jsonValue = field.getValue();
if (!jsonValue.isTextual()) {
throw new DataException("Expected schema parameter values to be strings but found: " +
jsonValue);
}
builder.parameter(field.getKey(), jsonValue.getTextValue());
}
}

if (fieldDefaultVal == null) {
fieldDefaultVal = schema.getJsonProp(COPYCAT_DEFAULT_VALUE_PROP);
}
Expand Down Expand Up @@ -1130,4 +1282,8 @@ private static String[] splitName(String fullName) {
}
return result;
}

private interface LogicalTypeConverter {
Object convert(Schema schema, Object value);
}
}
Loading