AvroUtils { public static DataType dataTypeFor(Schema schemaType) { LOG.trace("Converting Schema[{}] to DataType", schemaType); boolean isOptional = isNullable(schemaType); if (isOptional) { if (schemaType.getTypes().size() == 2) { LOG.trace("Unwrapping simple 'optional' union for {}", schemaType); for (Schema s : schemaType.getTypes()) { if (s.getType().equals(NULL)) { continue; } schemaType = s; break; } } } if (null != schemaType.getLogicalType()) { LogicalType logicalType = schemaType.getLogicalType(); switch (logicalType.getName()) { case "date" : return DataTypes.DateType; case "timestamp-millis" : return DataTypes.TimestampType; case "decimal" : LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) logicalType; return DataTypes.createDecimalType(decimal.getPrecision(), decimal.getScale()); default: LOG.warn("Unsupported LogicalType[{}], continuing with underlying base type", logicalType.getName()); } } switch (schemaType.getType()) { case RECORD: List<StructField> structFieldList = Lists.newArrayListWithCapacity(schemaType.getFields().size()); for (Field f : schemaType.getFields()) { structFieldList.add(DataTypes.createStructField(f.name(), dataTypeFor(f.schema()), isNullable(f.schema()))); } return DataTypes.createStructType(structFieldList); case ARRAY: Schema elementType = schemaType.getElementType(); return DataTypes.createArrayType(dataTypeFor(elementType), isNullable(elementType)); case MAP: Schema valueType = schemaType.getValueType(); return DataTypes.createMapType(DataTypes.StringType, dataTypeFor(valueType), isNullable(valueType)); case UNION: List<StructField> unionFieldList = Lists.newArrayListWithCapacity(schemaType.getTypes().size()); int m = 0; for (Schema u : schemaType.getTypes()) { unionFieldList.add(DataTypes.createStructField("member" + m++, dataTypeFor(u), isNullable(u))); } return DataTypes.createStructType(unionFieldList); case FIXED: case BYTES: return DataTypes.BinaryType; case ENUM: case STRING: return DataTypes.StringType; case INT: return DataTypes.IntegerType; case LONG: return DataTypes.LongType; case FLOAT: return DataTypes.FloatType; case DOUBLE: return DataTypes.DoubleType; case BOOLEAN: return DataTypes.BooleanType; case NULL: return DataTypes.NullType; default: throw new RuntimeException(String.format("Unrecognized or unsupported Avro Type conversion: %s", schemaType)); } }  static Schema schemaFor(StructType structType); static Schema schemaFor(StructType structType, String record, String namespace); static Schema typeFor(DataType dataType); static Schema typeFor(DataType dataType, boolean isOptional); static StructType structTypeFor(Schema schema); static DataType dataTypeFor(Schema schemaType);  }
@Test public void toDataTypeNull() { assertEquals("Invalid DataType", DataTypes.NullType, AvroUtils.dataTypeFor(SchemaBuilder.builder().nullType())); }
@Test public void toDataTypeInt() { assertEquals("Invalid DataType", DataTypes.IntegerType, AvroUtils.dataTypeFor(SchemaBuilder.builder().intType())); }
@Test public void toDataTypeFloat() { assertEquals("Invalid DataType", DataTypes.FloatType, AvroUtils.dataTypeFor(SchemaBuilder.builder().floatType())); }
@Test public void toDataTypeEnum() { assertEquals("Invalid DataType", DataTypes.StringType, AvroUtils.dataTypeFor(SchemaBuilder.enumeration("Enum").symbols("FOO"))); }
@Test public void toDataTypeFixed() { assertEquals("Invalid DataType", DataTypes.BinaryType, AvroUtils.dataTypeFor(SchemaBuilder.fixed("Fixed").size(16))); }
@Test public void toDataTypeString() { assertEquals("Invalid DataType", DataTypes.StringType, AvroUtils.dataTypeFor(SchemaBuilder.builder().stringType())); }
@Test public void toDataTypeBytes() { assertEquals("Invalid DataType", DataTypes.BinaryType, AvroUtils.dataTypeFor(SchemaBuilder.builder().bytesType())); }
@Test public void toDataTypeLong() { assertEquals("Invalid DataType", DataTypes.LongType, AvroUtils.dataTypeFor(SchemaBuilder.builder().longType())); }
@Test public void toDataTypeDouble() { assertEquals("Invalid DataType", DataTypes.DoubleType, AvroUtils.dataTypeFor(SchemaBuilder.builder().doubleType())); }
@Test public void toDataTypeBoolean() { assertEquals("Invalid DataType", DataTypes.BooleanType, AvroUtils.dataTypeFor(SchemaBuilder.builder().booleanType())); }
@Test public void toDataTypeArray() { assertEquals("Invalid DataType", DataTypes.createArrayType(DataTypes.IntegerType, false), AvroUtils.dataTypeFor(SchemaBuilder.array().items().intType())); }
@Test public void toDataTypeArrayContainsNull() { assertEquals("Invalid DataType", DataTypes.createArrayType(DataTypes.IntegerType, true), AvroUtils.dataTypeFor(SchemaBuilder.array().items().nullable().intType())); }
@Test public void toDataTypeArrayNested() { Schema inner = SchemaBuilder.array().items().intType(); Schema outer = SchemaBuilder.array().items(inner); assertEquals("Invalid DataType", DataTypes.createArrayType( DataTypes.createArrayType(DataTypes.IntegerType, false), false ), AvroUtils.dataTypeFor(outer)); }
@Test public void toDataTypeMap() { assertEquals("Invalid DataType", DataTypes.createMapType(DataTypes.StringType, DataTypes.IntegerType, false), AvroUtils.dataTypeFor(SchemaBuilder.map().values().intType())); }
@Test public void toDataTypeMapContainsNull() { assertEquals("Invalid DataType", DataTypes.createMapType(DataTypes.StringType, DataTypes.IntegerType, true), AvroUtils.dataTypeFor(SchemaBuilder.map().values().nullable().intType())); }
@Test public void toDataTypeMapNested() { Schema inner = SchemaBuilder.map().values().intType(); Schema outer = SchemaBuilder.map().values(inner); assertEquals("Invalid DataType", DataTypes.createMapType(DataTypes.StringType, DataTypes.createMapType(DataTypes.StringType, DataTypes.IntegerType, false), false ), AvroUtils.dataTypeFor(outer)); }
@Test public void toDataTypeRecord() { Schema record = SchemaBuilder.record("test").fields() .name("field1").type().intType().noDefault() .endRecord(); assertEquals("Invalid DataType", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.IntegerType, false) )), AvroUtils.dataTypeFor(record)); }
@Test public void toDataTypeRecordNullable() { Schema record = SchemaBuilder.record("test").fields() .name("field1").type().nullable().intType().noDefault() .endRecord(); assertEquals("Invalid DataType", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.IntegerType, true) )), AvroUtils.dataTypeFor(record)); }
@Test public void toDataTypeRecordOptional() { Schema record = SchemaBuilder.record("test").fields() .name("field1").type().optional().intType() .endRecord(); assertEquals("Invalid DataType", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.IntegerType, true) )), AvroUtils.dataTypeFor(record)); }
@Test public void toDataTypeRecordNested() { Schema inner = SchemaBuilder.record("inner").fields() .name("field1").type().intType().noDefault() .endRecord(); Schema outer = SchemaBuilder.record("outer").fields() .name("inner").type(inner).noDefault() .endRecord(); assertEquals("Invalid DataType", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("inner", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.IntegerType, false) )), false)) ), AvroUtils.dataTypeFor(outer)); }
@Test public void toDataTypeUnion() { Schema union = SchemaBuilder.unionOf() .intType().and() .floatType() .endUnion(); assertEquals("Invalid DataType", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("member0", DataTypes.IntegerType, false), DataTypes.createStructField("member1", DataTypes.FloatType, false) )), AvroUtils.dataTypeFor(union)); }
@Test public void toDataTypeUnionAsSimpleOptional() { Schema union = SchemaBuilder.unionOf() .intType().and() .nullType() .endUnion(); assertEquals("Invalid DataType", DataTypes.IntegerType, AvroUtils.dataTypeFor(union)); }
@Test public void toDataTypeUnionIncludingNull() { Schema union = SchemaBuilder.unionOf() .intType().and() .floatType().and() .nullType() .endUnion(); assertEquals("Invalid DataType", DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("member0", DataTypes.IntegerType, false), DataTypes.createStructField("member1", DataTypes.FloatType, false), DataTypes.createStructField("member2", DataTypes.NullType, false) )), AvroUtils.dataTypeFor(union)); }
@Test public void toDataTypeDate() { Schema logicalType = LogicalTypes.date().addToSchema(SchemaBuilder.builder().intType()); assertEquals("Invalid DataType", DataTypes.DateType, AvroUtils.dataTypeFor(logicalType)); }
@Test public void toDataTypeTimestamp() { Schema logicalType = LogicalTypes.timestampMillis().addToSchema(SchemaBuilder.builder().longType()); assertEquals("Invalid DataType", DataTypes.TimestampType, AvroUtils.dataTypeFor(logicalType)); }
@Test public void toDataTypeDecimal() { Schema logicalType = LogicalTypes.decimal(4, 2).addToSchema(SchemaBuilder.builder().bytesType()); assertEquals("Invalid DataType", DataTypes.createDecimalType(4, 2), AvroUtils.dataTypeFor(logicalType)); }
@Test public void toDataTypeIgnoredLogicalType() { Schema logicalType = LogicalTypes.timeMicros().addToSchema(SchemaBuilder.builder().longType()); assertEquals("Invalid DataType", DataTypes.LongType, AvroUtils.dataTypeFor(logicalType)); }