From 0878f252ae6047ca3d6584317ae6ce7176e28dd7 Mon Sep 17 00:00:00 2001 From: Joseph Glanville Date: Sun, 11 Oct 2020 01:27:17 +0700 Subject: [PATCH 1/9] Avro union support --- .../data/input/AvroHadoopInputRowParser.java | 14 +++-- .../data/input/AvroStreamInputRowParser.java | 14 ++++- .../data/input/avro/AvroFlattenerMaker.java | 23 ++++---- .../data/input/avro/AvroOCFInputFormat.java | 11 +++- .../druid/data/input/avro/AvroOCFReader.java | 5 +- .../druid/data/input/avro/AvroParsers.java | 5 +- .../input/avro/GenericAvroJsonProvider.java | 56 ++++++++++++++++++- .../src/test/avro/some-datum.avsc | 6 ++ .../input/AvroHadoopInputRowParserTest.java | 2 +- .../input/AvroStreamInputRowParserTest.java | 14 ++++- .../input/avro/AvroFlattenerMakerTest.java | 21 ++++++- .../data/input/avro/AvroOCFReaderTest.java | 2 +- .../avro/ParquetAvroHadoopInputRowParser.java | 11 ++-- 13 files changed, 147 insertions(+), 37 deletions(-) diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java index 39ce48ec2b5e..183d7b5ee824 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java @@ -34,18 +34,24 @@ public class AvroHadoopInputRowParser implements InputRowParser { private final ParseSpec parseSpec; private final boolean fromPigAvroStorage; + private final boolean binaryAsString; + private final boolean explodeUnions; private final ObjectFlattener avroFlattener; private final MapInputRowParser mapParser; @JsonCreator public AvroHadoopInputRowParser( @JsonProperty("parseSpec") ParseSpec parseSpec, - @JsonProperty("fromPigAvroStorage") Boolean fromPigAvroStorage + @JsonProperty("fromPigAvroStorage") Boolean fromPigAvroStorage, + @JsonProperty("binaryAsString") Boolean binaryAsString, + @JsonProperty("explodeUnions") Boolean explodeUnions ) { this.parseSpec = parseSpec; - this.fromPigAvroStorage = fromPigAvroStorage == null ? false : fromPigAvroStorage; - this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, false); + this.fromPigAvroStorage = fromPigAvroStorage != null && fromPigAvroStorage; + this.binaryAsString = binaryAsString != null && binaryAsString; + this.explodeUnions = explodeUnions != null && explodeUnions; + this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, this.binaryAsString, this.explodeUnions); this.mapParser = new MapInputRowParser(parseSpec); } @@ -71,6 +77,6 @@ public boolean isFromPigAvroStorage() @Override public InputRowParser withParseSpec(ParseSpec parseSpec) { - return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage); + return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage, binaryAsString, explodeUnions); } } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java index f375c63b8b23..76cffc099352 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java @@ -36,6 +36,8 @@ public class AvroStreamInputRowParser implements ByteBufferInputRowParser { private final ParseSpec parseSpec; + private final Boolean binaryAsString; + private final Boolean explodeUnions; private final AvroBytesDecoder avroBytesDecoder; private final ObjectFlattener avroFlattener; private final MapInputRowParser mapParser; @@ -43,12 +45,16 @@ public class AvroStreamInputRowParser implements ByteBufferInputRowParser @JsonCreator public AvroStreamInputRowParser( @JsonProperty("parseSpec") ParseSpec parseSpec, - @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder + @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder, + @JsonProperty("binaryAsString") Boolean binaryAsString, + @JsonProperty("explodeUnions") Boolean explodeUnions ) { this.parseSpec = Preconditions.checkNotNull(parseSpec, "parseSpec"); this.avroBytesDecoder = Preconditions.checkNotNull(avroBytesDecoder, "avroBytesDecoder"); - this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, false); + this.binaryAsString = binaryAsString != null && binaryAsString; + this.explodeUnions = explodeUnions != null && explodeUnions; + this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, this.binaryAsString, this.explodeUnions); this.mapParser = new MapInputRowParser(parseSpec); } @@ -76,7 +82,9 @@ public ByteBufferInputRowParser withParseSpec(ParseSpec parseSpec) { return new AvroStreamInputRowParser( parseSpec, - avroBytesDecoder + avroBytesDecoder, + binaryAsString, + explodeUnions ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java index 8f6453200159..6b60f88144c4 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java @@ -44,13 +44,8 @@ public class AvroFlattenerMaker implements ObjectFlatteners.FlattenerMaker { - private static final JsonProvider AVRO_JSON_PROVIDER = new GenericAvroJsonProvider(); - private static final Configuration JSONPATH_CONFIGURATION = - Configuration.builder() - .jsonProvider(AVRO_JSON_PROVIDER) - .mappingProvider(new NotImplementedMappingProvider()) - .options(EnumSet.of(Option.SUPPRESS_EXCEPTIONS)) - .build(); + private final JsonProvider avroJsonProvider; + private final Configuration jsonPathConfiguration; private static final EnumSet ROOT_TYPES = EnumSet.of( Schema.Type.STRING, @@ -99,10 +94,18 @@ private static boolean isFieldPrimitive(Schema.Field field) * @param fromPigAvroStorage boolean to specify the data file is stored using AvroStorage * @param binaryAsString boolean to encode the byte[] as a string. */ - public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString) + public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString, final boolean explodeUnions) { this.fromPigAvroStorage = fromPigAvroStorage; this.binaryAsString = binaryAsString; + + this.avroJsonProvider= new GenericAvroJsonProvider(explodeUnions); + this.jsonPathConfiguration = + Configuration.builder() + .jsonProvider(avroJsonProvider) + .mappingProvider(new NotImplementedMappingProvider()) + .options(EnumSet.of(Option.SUPPRESS_EXCEPTIONS)) + .build(); } @Override @@ -126,7 +129,7 @@ public Object getRootField(final GenericRecord record, final String key) public Function makeJsonPathExtractor(final String expr) { final JsonPath jsonPath = JsonPath.compile(expr); - return record -> transformValue(jsonPath.read(record, JSONPATH_CONFIGURATION)); + return record -> transformValue(jsonPath.read(record, jsonPathConfiguration)); } @Override @@ -138,7 +141,7 @@ public Function makeJsonQueryExtractor(final String expr) @Override public JsonProvider getJsonProvider() { - return AVRO_JSON_PROVIDER; + return avroJsonProvider; } @Override diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java index 90c74f65e635..ee4c6589cc1d 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java @@ -33,6 +33,7 @@ import org.apache.druid.java.util.common.parsers.JSONPathSpec; import javax.annotation.Nullable; +import javax.validation.constraints.Null; import java.io.File; import java.util.Map; import java.util.Objects; @@ -42,6 +43,7 @@ public class AvroOCFInputFormat extends NestedInputFormat private static final Logger LOGGER = new Logger(AvroOCFInputFormat.class); private final boolean binaryAsString; + private final boolean explodeUnions; @Nullable private final Schema readerSchema; @@ -50,7 +52,8 @@ public AvroOCFInputFormat( @JacksonInject @Json ObjectMapper mapper, @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec, @JsonProperty("schema") @Nullable Map schema, - @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString + @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString, + @JsonProperty("explodeUnions") @Nullable Boolean explodeUnions ) throws Exception { super(flattenSpec); @@ -62,7 +65,8 @@ public AvroOCFInputFormat( } else { this.readerSchema = null; } - this.binaryAsString = binaryAsString == null ? false : binaryAsString; + this.binaryAsString = binaryAsString != null && binaryAsString; + this.explodeUnions = explodeUnions != null && explodeUnions; } @Override @@ -82,7 +86,8 @@ public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity temporaryDirectory, readerSchema, getFlattenSpec(), - binaryAsString + binaryAsString, + explodeUnions ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java index 97ab895225a8..2764a1f7cf89 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java @@ -56,14 +56,15 @@ public class AvroOCFReader extends IntermediateRowParsingReader File temporaryDirectory, @Nullable Schema readerSchema, JSONPathSpec flattenSpec, - boolean binaryAsString + boolean binaryAsString, + boolean explodeUnions ) { this.inputRowSchema = inputRowSchema; this.source = source; this.temporaryDirectory = temporaryDirectory; this.readerSchema = readerSchema; - this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString)); + this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, explodeUnions)); } @Override diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java index 9d5665e800b8..b11d65984465 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java @@ -39,7 +39,8 @@ private AvroParsers() public static ObjectFlattener makeFlattener( final ParseSpec parseSpec, final boolean fromPigAvroStorage, - final boolean binaryAsString + final boolean binaryAsString, + final boolean explodeUnions ) { final JSONPathSpec flattenSpec; @@ -49,7 +50,7 @@ public static ObjectFlattener makeFlattener( flattenSpec = JSONPathSpec.DEFAULT; } - return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString)); + return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString, explodeUnions)); } public static List parseGenericRecord( diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java index ab6a53e5c7a4..96f61447a7b2 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java @@ -19,15 +19,19 @@ package org.apache.druid.data.input.avro; +import com.google.common.collect.ImmutableMap; import com.jayway.jsonpath.InvalidJsonException; import com.jayway.jsonpath.spi.json.JsonProvider; import org.apache.avro.Schema; +import org.apache.avro.generic.GenericEnumSymbol; +import org.apache.avro.generic.GenericFixed; import org.apache.avro.generic.GenericRecord; import org.apache.avro.util.Utf8; import javax.annotation.Nullable; import java.io.InputStream; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -41,6 +45,11 @@ */ public class GenericAvroJsonProvider implements JsonProvider { + private final boolean explodeUnions; + GenericAvroJsonProvider(final boolean explodeUnions) { + this.explodeUnions = explodeUnions; + } + @Override public Object parse(final String s) throws InvalidJsonException { @@ -148,7 +157,11 @@ public Object getMapValue(final Object o, final String s) if (o == null) { return null; } else if (o instanceof GenericRecord) { - return ((GenericRecord) o).get(s); + final GenericRecord record = (GenericRecord) o; + if (explodeUnions && isExplodableUnion(record.getSchema().getField(s))) { + return explodeUnion(record.get(s)); + } + return record.get(s); } else if (o instanceof Map) { final Map theMap = (Map) o; if (theMap.containsKey(s)) { @@ -195,4 +208,45 @@ public Object unwrap(final Object o) { return o; } + + private boolean isExplodableUnion(final Schema.Field field) { + return field.schema().isUnion() && + field.schema().getTypes().stream().filter(type -> type.getType() != Schema.Type.NULL).count() > 1; + } + + private Map explodeUnion(final Object o) + { + // Primitive types and unnamped complex types are keyed their type name. + // Complex named types are keyed by their names. + // This is safe because an Avro union can only contain a single member of each unnamed type and duplicates + // of the same named type are not allowed. i.e only a single array is allowed, multiple records are allowed as + // long as each has a unique name. + // The Avro null type is elided as it's value can only ever be null + if (o instanceof Integer) { + return ImmutableMap.of("int", o); + } else if (o instanceof Long) { + return ImmutableMap.of("long", o); + } else if (o instanceof Float) { + return ImmutableMap.of("float", o); + } else if (o instanceof Double) { + return ImmutableMap.of("double", o); + } else if (o instanceof Boolean) { + return ImmutableMap.of("boolean", o); + } else if (o instanceof Utf8) { + return ImmutableMap.of("string", o); + } else if (o instanceof ByteBuffer) { + return ImmutableMap.of("bytes", o); + } else if (o instanceof Map) { + return ImmutableMap.of("map", o); + } else if (o instanceof List) { + return ImmutableMap.of("array", o); + } else if (o instanceof GenericRecord) { + return ImmutableMap.of(((GenericRecord) o).getSchema().getName(), o); + } else if (o instanceof GenericFixed) { + return ImmutableMap.of(((GenericFixed) o).getSchema().getName(), o); + } else if (o instanceof GenericEnumSymbol) { + return ImmutableMap.of(((GenericEnumSymbol) o).getSchema().getName(), o); + } + return ImmutableMap.of(); + } } diff --git a/extensions-core/avro-extensions/src/test/avro/some-datum.avsc b/extensions-core/avro-extensions/src/test/avro/some-datum.avsc index bf8ea6883ab1..72ffe7148a07 100644 --- a/extensions-core/avro-extensions/src/test/avro/some-datum.avsc +++ b/extensions-core/avro-extensions/src/test/avro/some-datum.avsc @@ -13,6 +13,12 @@ {"name":"someIntValueMap","type":{"type":"map","values":"int"}}, {"name":"someStringValueMap","type":{"type":"map","values":"string"}}, {"name":"someUnion","type":["null","string"]}, + {"name":"someMultiMemberUnion", "type": ["string", "int", { + "type": "record", "name": "UnionSubRecord", "fields": [ + {"name": "subString", "type": "string"}, + {"name": "subFloat", "type": "float"} + ] + }]}, {"name":"someNull","type":"null"}, {"name":"someFixed","type":{"type":"fixed","size":16,"name":"MyFixed"}}, {"name":"someBytes","type":"bytes"}, diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java index 6e3cdb011247..c6ade851b56b 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroHadoopInputRowParserTest.java @@ -62,7 +62,7 @@ public void testParseFromPigAvroStorage() throws IOException private void testParse(GenericRecord record, boolean fromPigAvroStorage) throws IOException { - AvroHadoopInputRowParser parser = new AvroHadoopInputRowParser(AvroStreamInputRowParserTest.PARSE_SPEC, fromPigAvroStorage); + AvroHadoopInputRowParser parser = new AvroHadoopInputRowParser(AvroStreamInputRowParserTest.PARSE_SPEC, fromPigAvroStorage, false, false); AvroHadoopInputRowParser parser2 = jsonMapper.readValue( jsonMapper.writeValueAsBytes(parser), AvroHadoopInputRowParser.class diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java index 0dbb6b95ff41..201e62334de6 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java @@ -151,6 +151,7 @@ public CharSequence apply(@Nullable CharSequence input) } ); private static final String SOME_UNION_VALUE = "string as union"; + private static final Integer SOME_UNION_MEMBER_VALUE = 1; private static final ByteBuffer SOME_BYTES_VALUE = ByteBuffer.allocate(8); private static final String SOME_RECORD_STRING_VALUE = "string in record"; private static final List SOME_RECORD_ARRAY_VALUE = Collections.singletonList(MyNestedRecord.newBuilder() @@ -176,7 +177,9 @@ public void testSerde() throws IOException Repository repository = new Avro1124RESTRepositoryClientWrapper("http://github.io"); AvroStreamInputRowParser parser = new AvroStreamInputRowParser( PARSE_SPEC, - new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository) + new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository), + false, + false ); ByteBufferInputRowParser parser2 = jsonMapper.readValue( jsonMapper.writeValueAsString(parser), @@ -193,7 +196,9 @@ public void testParse() throws SchemaValidationException, IOException Repository repository = new InMemoryRepository(null); AvroStreamInputRowParser parser = new AvroStreamInputRowParser( PARSE_SPEC, - new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository) + new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository), + false, + false ); ByteBufferInputRowParser parser2 = jsonMapper.readValue( jsonMapper.writeValueAsString(parser), @@ -234,7 +239,9 @@ public void testParseSchemaless() throws SchemaValidationException, IOException Repository repository = new InMemoryRepository(null); AvroStreamInputRowParser parser = new AvroStreamInputRowParser( PARSE_SPEC_SCHEMALESS, - new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository) + new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository), + false, + false ); ByteBufferInputRowParser parser2 = jsonMapper.readValue( jsonMapper.writeValueAsString(parser), @@ -370,6 +377,7 @@ public static SomeAvroDatum buildSomeAvroDatum() .setSomeIntValueMap(SOME_INT_VALUE_MAP_VALUE) .setSomeStringValueMap(SOME_STRING_VALUE_MAP_VALUE) .setSomeUnion(SOME_UNION_VALUE) + .setSomeMultiMemberUnion(SOME_UNION_MEMBER_VALUE) .setSomeFixed(SOME_FIXED_VALUE) .setSomeBytes(SOME_BYTES_VALUE) .setSomeNull(null) diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java index 076167269a74..634cdbbd3384 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java @@ -33,7 +33,7 @@ public class AvroFlattenerMakerTest public void getRootField() { final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum(); - final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false); + final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true); Assert.assertEquals( record.getTimestamp(), @@ -120,7 +120,7 @@ public void getRootField() public void makeJsonPathExtractor() { final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum(); - final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false); + final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true); Assert.assertEquals( record.getTimestamp(), @@ -162,6 +162,10 @@ public void makeJsonPathExtractor() record.getSomeUnion(), flattener.makeJsonPathExtractor("$.someUnion").apply(record) ); + Assert.assertEquals( + record.getSomeMultiMemberUnion(), + flattener.makeJsonPathExtractor("$.someMultiMemberUnion.int").apply(record) + ); Assert.assertEquals( record.getSomeNull(), flattener.makeJsonPathExtractor("$.someNull").apply(record) @@ -219,11 +223,22 @@ public void makeJsonPathExtractor() ); } + @Test + public void jsonPathExtractorExplodeUnions() { + final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum(); + final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true); + + Assert.assertEquals( + record.getSomeMultiMemberUnion(), + flattener.makeJsonPathExtractor("$.someMultiMemberUnion.int").apply(record) + ); + } + @Test(expected = UnsupportedOperationException.class) public void makeJsonQueryExtractor() { final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum(); - final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false); + final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, false); Assert.assertEquals( record.getTimestamp(), diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java index 4841483c4e1a..c28e0b0d55c9 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java @@ -201,7 +201,7 @@ private InputEntityReader createReader( final DimensionsSpec dimensionsSpec = new DimensionsSpec(DimensionsSpec.getDefaultSchemas(ImmutableList.of( "eventType"))); - final AvroOCFInputFormat inputFormat = new AvroOCFInputFormat(mapper, null, readerSchema, null); + final AvroOCFInputFormat inputFormat = new AvroOCFInputFormat(mapper, null, readerSchema, null, null); final InputRowSchema schema = new InputRowSchema(timestampSpec, dimensionsSpec, ColumnsFilter.all()); final FileEntity entity = new FileEntity(someAvroFile); return inputFormat.createReader(schema, entity, temporaryFolder.newFolder()); diff --git a/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java b/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java index c8e9a4510e11..a1c3060c7b3d 100755 --- a/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java +++ b/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java @@ -51,6 +51,7 @@ public class ParquetAvroHadoopInputRowParser implements InputRowParser recordFlattener; private final List dimensions; @@ -58,13 +59,15 @@ public class ParquetAvroHadoopInputRowParser implements InputRowParser Date: Mon, 22 Mar 2021 18:13:27 +0700 Subject: [PATCH 2/9] Document new union support --- docs/development/extensions-core/avro.md | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/docs/development/extensions-core/avro.md b/docs/development/extensions-core/avro.md index a200e8318724..a1bb0b694f68 100644 --- a/docs/development/extensions-core/avro.md +++ b/docs/development/extensions-core/avro.md @@ -39,13 +39,24 @@ Make sure to [include](../../development/extensions.md#loading-extensions) `drui Druid supports most Avro types natively, there are however some exceptions which are detailed here. -`union` types which aren't of the form `[null, otherType]` aren't supported at this time. +#### Unions +Druid has two modes for supporting `union` types. The original legacy mode can only support unions of the form `[null, otherType]`. +The newer mode can be enabled by setting `explodeUnions` on the Avro parser in which case unions will be expanded according to the following rules: +* Primitive types and unnamed complex types are keyed their type name. i.e `int`, `` +* Complex named types are keyed by their names. +* The Avro null type is elided as it's value can only ever be null +This is safe because an Avro union can only contain a single member of each unnamed type and duplicates of the same named type are not allowed. i.e only a single array is allowed, multiple records (or other named types) are allowed as long as each has a unique name. + +#### Binary types `bytes` and `fixed` Avro types will be returned by default as base64 encoded strings unless the `binaryAsString` option is enabled on the Avro parser. This setting will decode these types as UTF-8 strings. +#### Enums `enum` types will be returned as `string` of the enum symbol. +#### Complex types `record` and `map` types representing nested data can be ingested using [flattenSpec](../../ingestion/data-formats.md#flattenspec) on the parser. -Druid doesn't currently support Avro logical types, they will be ignored and fields will be handled according to the underlying primitive type. \ No newline at end of file +#### Logical types +Druid doesn't currently support Avro logical types, they will be ignored and fields will be handled according to the underlying primitive type. From fe3d3a1ac25fc7332bbb691e06f49bd7b8a4e394 Mon Sep 17 00:00:00 2001 From: Joseph Glanville Date: Wed, 14 Apr 2021 08:10:03 +0700 Subject: [PATCH 3/9] Add support for AvroStreamInputFormat and fix checkstyle --- .../druid/data/input/avro/AvroFlattenerMaker.java | 2 +- .../druid/data/input/avro/AvroOCFInputFormat.java | 1 - .../druid/data/input/avro/AvroStreamInputFormat.java | 10 +++++++--- .../apache/druid/data/input/avro/AvroStreamReader.java | 5 +++-- .../druid/data/input/avro/GenericAvroJsonProvider.java | 7 +++++-- .../druid/data/input/AvroStreamInputFormatTest.java | 4 ++++ .../druid/data/input/avro/AvroFlattenerMakerTest.java | 3 ++- 7 files changed, 22 insertions(+), 10 deletions(-) diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java index 6b60f88144c4..c68b38e39e1a 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java @@ -99,7 +99,7 @@ public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binary this.fromPigAvroStorage = fromPigAvroStorage; this.binaryAsString = binaryAsString; - this.avroJsonProvider= new GenericAvroJsonProvider(explodeUnions); + this.avroJsonProvider = new GenericAvroJsonProvider(explodeUnions); this.jsonPathConfiguration = Configuration.builder() .jsonProvider(avroJsonProvider) diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java index ee4c6589cc1d..ca2f7b84fd53 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java @@ -33,7 +33,6 @@ import org.apache.druid.java.util.common.parsers.JSONPathSpec; import javax.annotation.Nullable; -import javax.validation.constraints.Null; import java.io.File; import java.util.Map; import java.util.Objects; diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java index 3b59b368f0bf..2300d3ba1ae7 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java @@ -35,6 +35,7 @@ public class AvroStreamInputFormat extends NestedInputFormat { private final boolean binaryAsString; + private final boolean explodeUnions; private final AvroBytesDecoder avroBytesDecoder; @@ -42,12 +43,14 @@ public class AvroStreamInputFormat extends NestedInputFormat public AvroStreamInputFormat( @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec, @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder, - @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString + @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString, + @JsonProperty("explodeUnions") @Nullable Boolean explodeUnions ) { super(flattenSpec); this.avroBytesDecoder = avroBytesDecoder; - this.binaryAsString = binaryAsString == null ? false : binaryAsString; + this.binaryAsString = binaryAsString != null && binaryAsString; + this.explodeUnions = explodeUnions != null && explodeUnions; } @Override @@ -76,7 +79,8 @@ public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity source, avroBytesDecoder, getFlattenSpec(), - binaryAsString + binaryAsString, + explodeUnions ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java index d8d92eb5a044..ba94519d8fe8 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java @@ -52,13 +52,14 @@ public class AvroStreamReader extends IntermediateRowParsingReader type.getType() != Schema.Type.NULL).count() > 1; } diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java index 139616bfdbfa..918aa0abb513 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java @@ -118,6 +118,7 @@ public void testSerde() throws IOException AvroStreamInputFormat inputFormat = new AvroStreamInputFormat( flattenSpec, new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository), + false, false ); NestedInputFormat inputFormat2 = jsonMapper.readValue( @@ -134,6 +135,7 @@ public void testSerdeForSchemaRegistry() throws IOException AvroStreamInputFormat inputFormat = new AvroStreamInputFormat( flattenSpec, new SchemaRegistryBasedAvroBytesDecoder("http://test:8081", 100, null, null, null), + false, false ); NestedInputFormat inputFormat2 = jsonMapper.readValue( @@ -150,6 +152,7 @@ public void testParse() throws SchemaValidationException, IOException AvroStreamInputFormat inputFormat = new AvroStreamInputFormat( flattenSpec, new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository), + false, false ); NestedInputFormat inputFormat2 = jsonMapper.readValue( @@ -193,6 +196,7 @@ public void testParseSchemaless() throws SchemaValidationException, IOException AvroStreamInputFormat inputFormat = new AvroStreamInputFormat( flattenSpec, new SchemaRepoBasedAvroBytesDecoder<>(new Avro1124SubjectAndIdConverter(TOPIC), repository), + false, false ); NestedInputFormat inputFormat2 = jsonMapper.readValue( diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java index 634cdbbd3384..f3f5cce8740e 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java @@ -224,7 +224,8 @@ public void makeJsonPathExtractor() } @Test - public void jsonPathExtractorExplodeUnions() { + public void jsonPathExtractorExplodeUnions() + { final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum(); final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true); From 5335b8a9f84ee29b92655f81769de5a45966e5c5 Mon Sep 17 00:00:00 2001 From: Joseph Glanville Date: Wed, 14 Apr 2021 08:46:08 +0700 Subject: [PATCH 4/9] Extend multi-member union test schema and format --- .../src/test/avro/some-datum.avsc | 223 ++++++++++++++---- .../input/AvroStreamInputRowParserTest.java | 17 +- .../input/avro/AvroFlattenerMakerTest.java | 70 +++++- .../data/input/avro/AvroOCFReaderTest.java | 2 +- 4 files changed, 260 insertions(+), 52 deletions(-) diff --git a/extensions-core/avro-extensions/src/test/avro/some-datum.avsc b/extensions-core/avro-extensions/src/test/avro/some-datum.avsc index 72ffe7148a07..f03cb708df05 100644 --- a/extensions-core/avro-extensions/src/test/avro/some-datum.avsc +++ b/extensions-core/avro-extensions/src/test/avro/some-datum.avsc @@ -1,44 +1,181 @@ -[{ - "namespace": "org.apache.druid.data.input", - "name": "SomeAvroDatum", - "type": "record", - "fields" : [ - {"name":"timestamp","type":"long"}, - {"name":"eventType","type":"string"}, - {"name":"id","type":"long"}, - {"name":"someOtherId","type":"long"}, - {"name":"isValid","type":"boolean"}, - {"name":"someIntArray","type":{"type":"array","items":"int"}}, - {"name":"someStringArray","type":{"type":"array","items":"string"}}, - {"name":"someIntValueMap","type":{"type":"map","values":"int"}}, - {"name":"someStringValueMap","type":{"type":"map","values":"string"}}, - {"name":"someUnion","type":["null","string"]}, - {"name":"someMultiMemberUnion", "type": ["string", "int", { - "type": "record", "name": "UnionSubRecord", "fields": [ - {"name": "subString", "type": "string"}, - {"name": "subFloat", "type": "float"} - ] - }]}, - {"name":"someNull","type":"null"}, - {"name":"someFixed","type":{"type":"fixed","size":16,"name":"MyFixed"}}, - {"name":"someBytes","type":"bytes"}, - {"name":"someEnum","type":{"type":"enum","name":"MyEnum","symbols":["ENUM0","ENUM1","ENUM2"]}}, - {"name":"someRecord","type":{ - "type":"record","name":"MySubRecord","fields":[ - {"name":"subInt","type":"int"}, - {"name":"subLong","type":"long"} - ] - }}, - - {"name":"someLong","type":"long"}, - {"name":"someInt","type":"int"}, - {"name":"someFloat","type":"float"}, - {"name":"someRecordArray","type":{ - "type":"array","items":{ - "type":"record","name":"MyNestedRecord","fields":[ - {"name":"nestedString","type":"string"} +[ + { + "namespace": "org.apache.druid.data.input", + "name": "SomeAvroDatum", + "type": "record", + "fields": [ + { + "name": "timestamp", + "type": "long" + }, + { + "name": "eventType", + "type": "string" + }, + { + "name": "id", + "type": "long" + }, + { + "name": "someOtherId", + "type": "long" + }, + { + "name": "isValid", + "type": "boolean" + }, + { + "name": "someIntArray", + "type": { + "type": "array", + "items": "int" + } + }, + { + "name": "someStringArray", + "type": { + "type": "array", + "items": "string" + } + }, + { + "name": "someIntValueMap", + "type": { + "type": "map", + "values": "int" + } + }, + { + "name": "someStringValueMap", + "type": { + "type": "map", + "values": "string" + } + }, + { + "name": "someUnion", + "type": [ + "null", + "string" ] - } - }} - ] -}] + }, + { + "name": "someMultiMemberUnion", + "type": [ + "int", + "long", + "float", + "double", + "boolean", + "string", + "bytes", + { + "type": "map", + "values": "string" + }, + { + "type": "array", + "items": "int" + }, + { + "type": "record", + "name": "UnionSubRecord", + "fields": [ + { + "name": "subString", + "type": "string" + } + ] + }, + { + "name": "UnionSubFixed", + "type": "fixed", + "size": 10 + }, + { + "type": "enum", + "name": "UnionSubEnum", + "symbols": [ + "ENUM0", + "ENUM1", + "ENUM2" + ] + } + ] + }, + { + "name": "someNull", + "type": "null" + }, + { + "name": "someFixed", + "type": { + "type": "fixed", + "size": 16, + "name": "MyFixed" + } + }, + { + "name": "someBytes", + "type": "bytes" + }, + { + "name": "someEnum", + "type": { + "type": "enum", + "name": "MyEnum", + "symbols": [ + "ENUM0", + "ENUM1", + "ENUM2" + ] + } + }, + { + "name": "someRecord", + "type": { + "type": "record", + "name": "MySubRecord", + "fields": [ + { + "name": "subInt", + "type": "int" + }, + { + "name": "subLong", + "type": "long" + } + ] + } + }, + { + "name": "someLong", + "type": "long" + }, + { + "name": "someInt", + "type": "int" + }, + { + "name": "someFloat", + "type": "float" + }, + { + "name": "someRecordArray", + "type": { + "type": "array", + "items": { + "type": "record", + "name": "MyNestedRecord", + "fields": [ + { + "name": "nestedString", + "type": "string" + } + ] + } + } + } + ] + } +] diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java index 201e62334de6..8ed9f44fc062 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputRowParserTest.java @@ -361,7 +361,7 @@ public Integer apply(@Nullable String input) Assert.assertEquals(SOME_INT_VALUE, inputRow.getMetric("someInt")); } - public static SomeAvroDatum buildSomeAvroDatum() + private static SomeAvroDatum.Builder createSomeAvroDatumBuilderDefaults() { return SomeAvroDatum.newBuilder() .setTimestamp(DATE_TIME.toInstant().toEpochMilli()) @@ -383,7 +383,18 @@ public static SomeAvroDatum buildSomeAvroDatum() .setSomeNull(null) .setSomeEnum(MyEnum.ENUM1) .setSomeRecord(SOME_RECORD_VALUE) - .setSomeRecordArray(SOME_RECORD_ARRAY_VALUE) - .build(); + .setSomeRecordArray(SOME_RECORD_ARRAY_VALUE); + } + + public static SomeAvroDatum buildSomeAvroDatum() + { + return createSomeAvroDatumBuilderDefaults().build(); + } + + public static SomeAvroDatum buildSomeAvroDatumWithUnionValue(Object unionValue) + { + return createSomeAvroDatumBuilderDefaults() + .setSomeMultiMemberUnion(unionValue) + .build(); } } diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java index f3f5cce8740e..988c509b5f24 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java @@ -19,11 +19,19 @@ package org.apache.druid.data.input.avro; +import org.apache.avro.util.Utf8; import org.apache.druid.data.input.AvroStreamInputRowParserTest; import org.apache.druid.data.input.SomeAvroDatum; +import org.apache.druid.data.input.UnionSubEnum; +import org.apache.druid.data.input.UnionSubFixed; +import org.apache.druid.data.input.UnionSubRecord; import org.junit.Assert; import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; public class AvroFlattenerMakerTest @@ -226,13 +234,65 @@ public void makeJsonPathExtractor() @Test public void jsonPathExtractorExplodeUnions() { - final SomeAvroDatum record = AvroStreamInputRowParserTest.buildSomeAvroDatum(); final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true); - Assert.assertEquals( - record.getSomeMultiMemberUnion(), - flattener.makeJsonPathExtractor("$.someMultiMemberUnion.int").apply(record) - ); + // Unmamed types are accessed by type + + // int + Assert.assertEquals(1, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.int").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(1))); + + // long + Assert.assertEquals(1L, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.long").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(1L))); + + // float + Assert.assertEquals((float) 1.0, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.float").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue((float) 1.0))); + + // double + Assert.assertEquals(1.0, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.double").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(1.0))); + + // string + Assert.assertEquals("string", flattener.makeJsonPathExtractor("$.someMultiMemberUnion.string").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(new Utf8("string")))); + + // bytes + Assert.assertArrayEquals(new byte[] {1}, (byte[]) flattener.makeJsonPathExtractor("$.someMultiMemberUnion.bytes").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(ByteBuffer.wrap(new byte[] {1})))); + + // map + Assert.assertEquals(2, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.map.two").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(new HashMap() {{ + put("one", 1); + put("two", 2); + put("three", 3); + } + } + ))); + + // array + Assert.assertEquals(3, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.array[2]").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(Arrays.asList(1, 2, 3)))); + + // Named types are accessed by name + + // record + Assert.assertEquals("subRecordString", flattener.makeJsonPathExtractor("$.someMultiMemberUnion.UnionSubRecord.subString").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue( + UnionSubRecord.newBuilder() + .setSubString("subRecordString") + .build()))); + + // fixed + final byte[] fixedBytes = new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; + Assert.assertEquals(fixedBytes, flattener.makeJsonPathExtractor("$.someMultiMemberUnion.UnionSubFixed").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(new UnionSubFixed(fixedBytes)))); + + // enum + Assert.assertEquals(String.valueOf(UnionSubEnum.ENUM1), flattener.makeJsonPathExtractor("$.someMultiMemberUnion.UnionSubEnum").apply( + AvroStreamInputRowParserTest.buildSomeAvroDatumWithUnionValue(UnionSubEnum.ENUM1))); } @Test(expected = UnsupportedOperationException.class) diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java index c28e0b0d55c9..b00c1e4ad7a1 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroOCFReaderTest.java @@ -144,7 +144,7 @@ public void testSample() throws Exception Assert.assertFalse(iterator.hasNext()); final Map rawColumns = row.getRawValues(); Assert.assertNotNull(rawColumns); - Assert.assertEquals(19, rawColumns.size()); + Assert.assertEquals(20, rawColumns.size()); final List inputRows = row.getInputRows(); Assert.assertNotNull(inputRows); final InputRow inputRow = Iterables.getOnlyElement(inputRows); From cde6041505b7701aa2ca7d7a3253f82e46edba9f Mon Sep 17 00:00:00 2001 From: Joseph Glanville Date: Thu, 15 Apr 2021 06:27:53 +0700 Subject: [PATCH 5/9] Some additional docs and add Enums to spelling --- docs/development/extensions-core/avro.md | 4 ++-- website/.spelling | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/development/extensions-core/avro.md b/docs/development/extensions-core/avro.md index a1bb0b694f68..f460c3b095f7 100644 --- a/docs/development/extensions-core/avro.md +++ b/docs/development/extensions-core/avro.md @@ -42,8 +42,8 @@ Druid supports most Avro types natively, there are however some exceptions which #### Unions Druid has two modes for supporting `union` types. The original legacy mode can only support unions of the form `[null, otherType]`. The newer mode can be enabled by setting `explodeUnions` on the Avro parser in which case unions will be expanded according to the following rules: -* Primitive types and unnamed complex types are keyed their type name. i.e `int`, `` -* Complex named types are keyed by their names. +* Primitive types and unnamed complex types are keyed their type name. i.e `int`, `string` +* Complex named types are keyed by their names, this includes `record`, `fixed` and `enum`. * The Avro null type is elided as it's value can only ever be null This is safe because an Avro union can only contain a single member of each unnamed type and duplicates of the same named type are not allowed. i.e only a single array is allowed, multiple records (or other named types) are allowed as long as each has a unique name. diff --git a/website/.spelling b/website/.spelling index 4434cd0b1cc5..fde9407a63eb 100644 --- a/website/.spelling +++ b/website/.spelling @@ -74,6 +74,7 @@ EMR EMRFS ETL Elasticsearch +Enums FirehoseFactory FlattenSpec Float.NEGATIVE_INFINITY From 79c1fbfe0f0882d7e33f680e3d03f96b62a26b6f Mon Sep 17 00:00:00 2001 From: Joseph Glanville Date: Thu, 3 Jun 2021 19:25:55 +0700 Subject: [PATCH 6/9] Rename explodeUnions -> extractUnions --- docs/development/extensions-core/avro.md | 2 +- .../druid/data/input/AvroHadoopInputRowParser.java | 10 +++++----- .../druid/data/input/AvroStreamInputRowParser.java | 10 +++++----- .../druid/data/input/avro/AvroFlattenerMaker.java | 4 ++-- .../druid/data/input/avro/AvroOCFInputFormat.java | 8 ++++---- .../apache/druid/data/input/avro/AvroOCFReader.java | 4 ++-- .../org/apache/druid/data/input/avro/AvroParsers.java | 4 ++-- .../druid/data/input/avro/AvroStreamInputFormat.java | 8 ++++---- .../apache/druid/data/input/avro/AvroStreamReader.java | 4 ++-- .../druid/data/input/avro/GenericAvroJsonProvider.java | 8 ++++---- .../druid/data/input/avro/AvroFlattenerMakerTest.java | 2 +- .../parquet/avro/ParquetAvroHadoopInputRowParser.java | 10 +++++----- 12 files changed, 37 insertions(+), 37 deletions(-) diff --git a/docs/development/extensions-core/avro.md b/docs/development/extensions-core/avro.md index f460c3b095f7..61980919815a 100644 --- a/docs/development/extensions-core/avro.md +++ b/docs/development/extensions-core/avro.md @@ -41,7 +41,7 @@ Druid supports most Avro types natively, there are however some exceptions which #### Unions Druid has two modes for supporting `union` types. The original legacy mode can only support unions of the form `[null, otherType]`. -The newer mode can be enabled by setting `explodeUnions` on the Avro parser in which case unions will be expanded according to the following rules: +The newer mode can be enabled by setting `extractUnions` on the Avro parser in which case unions will be expanded according to the following rules: * Primitive types and unnamed complex types are keyed their type name. i.e `int`, `string` * Complex named types are keyed by their names, this includes `record`, `fixed` and `enum`. * The Avro null type is elided as it's value can only ever be null diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java index 183d7b5ee824..b665912992e8 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java @@ -35,7 +35,7 @@ public class AvroHadoopInputRowParser implements InputRowParser private final ParseSpec parseSpec; private final boolean fromPigAvroStorage; private final boolean binaryAsString; - private final boolean explodeUnions; + private final boolean extractUnions; private final ObjectFlattener avroFlattener; private final MapInputRowParser mapParser; @@ -44,14 +44,14 @@ public AvroHadoopInputRowParser( @JsonProperty("parseSpec") ParseSpec parseSpec, @JsonProperty("fromPigAvroStorage") Boolean fromPigAvroStorage, @JsonProperty("binaryAsString") Boolean binaryAsString, - @JsonProperty("explodeUnions") Boolean explodeUnions + @JsonProperty("extractUnions") Boolean extractUnions ) { this.parseSpec = parseSpec; this.fromPigAvroStorage = fromPigAvroStorage != null && fromPigAvroStorage; this.binaryAsString = binaryAsString != null && binaryAsString; - this.explodeUnions = explodeUnions != null && explodeUnions; - this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, this.binaryAsString, this.explodeUnions); + this.extractUnions = extractUnions != null && extractUnions; + this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, this.binaryAsString, this.extractUnions); this.mapParser = new MapInputRowParser(parseSpec); } @@ -77,6 +77,6 @@ public boolean isFromPigAvroStorage() @Override public InputRowParser withParseSpec(ParseSpec parseSpec) { - return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage, binaryAsString, explodeUnions); + return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage, binaryAsString, extractUnions); } } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java index 76cffc099352..ec0acebb3920 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java @@ -37,7 +37,7 @@ public class AvroStreamInputRowParser implements ByteBufferInputRowParser { private final ParseSpec parseSpec; private final Boolean binaryAsString; - private final Boolean explodeUnions; + private final Boolean extractUnions; private final AvroBytesDecoder avroBytesDecoder; private final ObjectFlattener avroFlattener; private final MapInputRowParser mapParser; @@ -47,14 +47,14 @@ public AvroStreamInputRowParser( @JsonProperty("parseSpec") ParseSpec parseSpec, @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder, @JsonProperty("binaryAsString") Boolean binaryAsString, - @JsonProperty("explodeUnions") Boolean explodeUnions + @JsonProperty("extractUnions") Boolean extractUnions ) { this.parseSpec = Preconditions.checkNotNull(parseSpec, "parseSpec"); this.avroBytesDecoder = Preconditions.checkNotNull(avroBytesDecoder, "avroBytesDecoder"); this.binaryAsString = binaryAsString != null && binaryAsString; - this.explodeUnions = explodeUnions != null && explodeUnions; - this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, this.binaryAsString, this.explodeUnions); + this.extractUnions = extractUnions != null && extractUnions; + this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, this.binaryAsString, this.extractUnions); this.mapParser = new MapInputRowParser(parseSpec); } @@ -84,7 +84,7 @@ public ByteBufferInputRowParser withParseSpec(ParseSpec parseSpec) parseSpec, avroBytesDecoder, binaryAsString, - explodeUnions + extractUnions ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java index c68b38e39e1a..f8c1b8616ef8 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java @@ -94,12 +94,12 @@ private static boolean isFieldPrimitive(Schema.Field field) * @param fromPigAvroStorage boolean to specify the data file is stored using AvroStorage * @param binaryAsString boolean to encode the byte[] as a string. */ - public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString, final boolean explodeUnions) + public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString, final boolean extractUnions) { this.fromPigAvroStorage = fromPigAvroStorage; this.binaryAsString = binaryAsString; - this.avroJsonProvider = new GenericAvroJsonProvider(explodeUnions); + this.avroJsonProvider = new GenericAvroJsonProvider(extractUnions); this.jsonPathConfiguration = Configuration.builder() .jsonProvider(avroJsonProvider) diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java index ca2f7b84fd53..19517fdda481 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java @@ -42,7 +42,7 @@ public class AvroOCFInputFormat extends NestedInputFormat private static final Logger LOGGER = new Logger(AvroOCFInputFormat.class); private final boolean binaryAsString; - private final boolean explodeUnions; + private final boolean extractUnions; @Nullable private final Schema readerSchema; @@ -52,7 +52,7 @@ public AvroOCFInputFormat( @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec, @JsonProperty("schema") @Nullable Map schema, @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString, - @JsonProperty("explodeUnions") @Nullable Boolean explodeUnions + @JsonProperty("extractUnions") @Nullable Boolean extractUnions ) throws Exception { super(flattenSpec); @@ -65,7 +65,7 @@ public AvroOCFInputFormat( this.readerSchema = null; } this.binaryAsString = binaryAsString != null && binaryAsString; - this.explodeUnions = explodeUnions != null && explodeUnions; + this.extractUnions = extractUnions != null && extractUnions; } @Override @@ -86,7 +86,7 @@ public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity readerSchema, getFlattenSpec(), binaryAsString, - explodeUnions + extractUnions ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java index 2764a1f7cf89..f47a77acfec9 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java @@ -57,14 +57,14 @@ public class AvroOCFReader extends IntermediateRowParsingReader @Nullable Schema readerSchema, JSONPathSpec flattenSpec, boolean binaryAsString, - boolean explodeUnions + boolean extractUnions ) { this.inputRowSchema = inputRowSchema; this.source = source; this.temporaryDirectory = temporaryDirectory; this.readerSchema = readerSchema; - this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, explodeUnions)); + this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, extractUnions)); } @Override diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java index b11d65984465..b8bc7f0dcb54 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java @@ -40,7 +40,7 @@ public static ObjectFlattener makeFlattener( final ParseSpec parseSpec, final boolean fromPigAvroStorage, final boolean binaryAsString, - final boolean explodeUnions + final boolean extractUnions ) { final JSONPathSpec flattenSpec; @@ -50,7 +50,7 @@ public static ObjectFlattener makeFlattener( flattenSpec = JSONPathSpec.DEFAULT; } - return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString, explodeUnions)); + return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString, extractUnions)); } public static List parseGenericRecord( diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java index 2300d3ba1ae7..afb820060764 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java @@ -35,7 +35,7 @@ public class AvroStreamInputFormat extends NestedInputFormat { private final boolean binaryAsString; - private final boolean explodeUnions; + private final boolean extractUnions; private final AvroBytesDecoder avroBytesDecoder; @@ -44,13 +44,13 @@ public AvroStreamInputFormat( @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec, @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder, @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString, - @JsonProperty("explodeUnions") @Nullable Boolean explodeUnions + @JsonProperty("extractUnions") @Nullable Boolean extractUnions ) { super(flattenSpec); this.avroBytesDecoder = avroBytesDecoder; this.binaryAsString = binaryAsString != null && binaryAsString; - this.explodeUnions = explodeUnions != null && explodeUnions; + this.extractUnions = extractUnions != null && extractUnions; } @Override @@ -80,7 +80,7 @@ public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity avroBytesDecoder, getFlattenSpec(), binaryAsString, - explodeUnions + extractUnions ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java index ba94519d8fe8..a8f10098b770 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java @@ -53,13 +53,13 @@ public class AvroStreamReader extends IntermediateRowParsingReader recordFlattener; private final List dimensions; @@ -60,14 +60,14 @@ public class ParquetAvroHadoopInputRowParser implements InputRowParser Date: Fri, 11 Jun 2021 06:11:16 +0700 Subject: [PATCH 7/9] explode -> extract --- .../druid/data/input/avro/GenericAvroJsonProvider.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java index 191a52c44eb9..11a02ed9ad37 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/GenericAvroJsonProvider.java @@ -160,8 +160,8 @@ public Object getMapValue(final Object o, final String s) return null; } else if (o instanceof GenericRecord) { final GenericRecord record = (GenericRecord) o; - if (extractUnions && isExplodableUnion(record.getSchema().getField(s))) { - return explodeUnion(record.get(s)); + if (extractUnions && isExtractableUnion(record.getSchema().getField(s))) { + return extractUnion(record.get(s)); } return record.get(s); } else if (o instanceof Map) { @@ -211,13 +211,13 @@ public Object unwrap(final Object o) return o; } - private boolean isExplodableUnion(final Schema.Field field) + private boolean isExtractableUnion(final Schema.Field field) { return field.schema().isUnion() && field.schema().getTypes().stream().filter(type -> type.getType() != Schema.Type.NULL).count() > 1; } - private Map explodeUnion(final Object o) + private Map extractUnion(final Object o) { // Primitive types and unnamped complex types are keyed their type name. // Complex named types are keyed by their names. From 3ebd657d19191c8f55542a797fb5f230b026a22d Mon Sep 17 00:00:00 2001 From: Joseph Glanville Date: Fri, 11 Jun 2021 07:41:45 +0700 Subject: [PATCH 8/9] ByType --- docs/development/extensions-core/avro.md | 14 ++++++++++---- .../druid/data/input/AvroHadoopInputRowParser.java | 10 +++++----- .../druid/data/input/AvroStreamInputRowParser.java | 10 +++++----- .../druid/data/input/avro/AvroFlattenerMaker.java | 4 ++-- .../druid/data/input/avro/AvroOCFInputFormat.java | 8 ++++---- .../druid/data/input/avro/AvroOCFReader.java | 4 ++-- .../apache/druid/data/input/avro/AvroParsers.java | 4 ++-- .../data/input/avro/AvroStreamInputFormat.java | 8 ++++---- .../druid/data/input/avro/AvroStreamReader.java | 4 ++-- .../data/input/avro/GenericAvroJsonProvider.java | 12 ++++++------ .../data/input/avro/AvroFlattenerMakerTest.java | 2 +- .../avro/ParquetAvroHadoopInputRowParser.java | 10 +++++----- 12 files changed, 48 insertions(+), 42 deletions(-) diff --git a/docs/development/extensions-core/avro.md b/docs/development/extensions-core/avro.md index 61980919815a..333f0dfb560c 100644 --- a/docs/development/extensions-core/avro.md +++ b/docs/development/extensions-core/avro.md @@ -40,13 +40,19 @@ Make sure to [include](../../development/extensions.md#loading-extensions) `drui Druid supports most Avro types natively, there are however some exceptions which are detailed here. #### Unions -Druid has two modes for supporting `union` types. The original legacy mode can only support unions of the form `[null, otherType]`. -The newer mode can be enabled by setting `extractUnions` on the Avro parser in which case unions will be expanded according to the following rules: +Druid has two modes for supporting `union` types. + +The default mode will treat unions as a single value regardless of the type it is poulated with. + +If you wish to operate on each different member of a union however you can set `extractUnionsByType` on the Avro parser in which case unions will be expanded into nested objects according to the following rules: * Primitive types and unnamed complex types are keyed their type name. i.e `int`, `string` * Complex named types are keyed by their names, this includes `record`, `fixed` and `enum`. -* The Avro null type is elided as it's value can only ever be null +* The Avro null type is elided as its value can only ever be null + +This is safe because an Avro union can only contain a single member of each unnamed type and duplicates of the same named type are not allowed. +i.e only a single array is allowed, multiple records (or other named types) are allowed as long as each has a unique name. -This is safe because an Avro union can only contain a single member of each unnamed type and duplicates of the same named type are not allowed. i.e only a single array is allowed, multiple records (or other named types) are allowed as long as each has a unique name. +The members can then be accessed using a [flattenSpec](../../ingestion/data-formats.md#flattenspec) similar other nested types. #### Binary types `bytes` and `fixed` Avro types will be returned by default as base64 encoded strings unless the `binaryAsString` option is enabled on the Avro parser. diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java index b665912992e8..165f6a40a6e4 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroHadoopInputRowParser.java @@ -35,7 +35,7 @@ public class AvroHadoopInputRowParser implements InputRowParser private final ParseSpec parseSpec; private final boolean fromPigAvroStorage; private final boolean binaryAsString; - private final boolean extractUnions; + private final boolean extractUnionsByType; private final ObjectFlattener avroFlattener; private final MapInputRowParser mapParser; @@ -44,14 +44,14 @@ public AvroHadoopInputRowParser( @JsonProperty("parseSpec") ParseSpec parseSpec, @JsonProperty("fromPigAvroStorage") Boolean fromPigAvroStorage, @JsonProperty("binaryAsString") Boolean binaryAsString, - @JsonProperty("extractUnions") Boolean extractUnions + @JsonProperty("extractUnionsByType") Boolean extractUnionsByType ) { this.parseSpec = parseSpec; this.fromPigAvroStorage = fromPigAvroStorage != null && fromPigAvroStorage; this.binaryAsString = binaryAsString != null && binaryAsString; - this.extractUnions = extractUnions != null && extractUnions; - this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, this.binaryAsString, this.extractUnions); + this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType; + this.avroFlattener = AvroParsers.makeFlattener(parseSpec, this.fromPigAvroStorage, this.binaryAsString, this.extractUnionsByType); this.mapParser = new MapInputRowParser(parseSpec); } @@ -77,6 +77,6 @@ public boolean isFromPigAvroStorage() @Override public InputRowParser withParseSpec(ParseSpec parseSpec) { - return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage, binaryAsString, extractUnions); + return new AvroHadoopInputRowParser(parseSpec, fromPigAvroStorage, binaryAsString, extractUnionsByType); } } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java index ec0acebb3920..fcb8bd833222 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/AvroStreamInputRowParser.java @@ -37,7 +37,7 @@ public class AvroStreamInputRowParser implements ByteBufferInputRowParser { private final ParseSpec parseSpec; private final Boolean binaryAsString; - private final Boolean extractUnions; + private final Boolean extractUnionsByType; private final AvroBytesDecoder avroBytesDecoder; private final ObjectFlattener avroFlattener; private final MapInputRowParser mapParser; @@ -47,14 +47,14 @@ public AvroStreamInputRowParser( @JsonProperty("parseSpec") ParseSpec parseSpec, @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder, @JsonProperty("binaryAsString") Boolean binaryAsString, - @JsonProperty("extractUnions") Boolean extractUnions + @JsonProperty("extractUnionsByType") Boolean extractUnionsByType ) { this.parseSpec = Preconditions.checkNotNull(parseSpec, "parseSpec"); this.avroBytesDecoder = Preconditions.checkNotNull(avroBytesDecoder, "avroBytesDecoder"); this.binaryAsString = binaryAsString != null && binaryAsString; - this.extractUnions = extractUnions != null && extractUnions; - this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, this.binaryAsString, this.extractUnions); + this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType; + this.avroFlattener = AvroParsers.makeFlattener(parseSpec, false, this.binaryAsString, this.extractUnionsByType); this.mapParser = new MapInputRowParser(parseSpec); } @@ -84,7 +84,7 @@ public ByteBufferInputRowParser withParseSpec(ParseSpec parseSpec) parseSpec, avroBytesDecoder, binaryAsString, - extractUnions + extractUnionsByType ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java index f8c1b8616ef8..8526a04c9125 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroFlattenerMaker.java @@ -94,12 +94,12 @@ private static boolean isFieldPrimitive(Schema.Field field) * @param fromPigAvroStorage boolean to specify the data file is stored using AvroStorage * @param binaryAsString boolean to encode the byte[] as a string. */ - public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString, final boolean extractUnions) + public AvroFlattenerMaker(final boolean fromPigAvroStorage, final boolean binaryAsString, final boolean extractUnionsByType) { this.fromPigAvroStorage = fromPigAvroStorage; this.binaryAsString = binaryAsString; - this.avroJsonProvider = new GenericAvroJsonProvider(extractUnions); + this.avroJsonProvider = new GenericAvroJsonProvider(extractUnionsByType); this.jsonPathConfiguration = Configuration.builder() .jsonProvider(avroJsonProvider) diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java index 19517fdda481..7417477ea505 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFInputFormat.java @@ -42,7 +42,7 @@ public class AvroOCFInputFormat extends NestedInputFormat private static final Logger LOGGER = new Logger(AvroOCFInputFormat.class); private final boolean binaryAsString; - private final boolean extractUnions; + private final boolean extractUnionsByType; @Nullable private final Schema readerSchema; @@ -52,7 +52,7 @@ public AvroOCFInputFormat( @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec, @JsonProperty("schema") @Nullable Map schema, @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString, - @JsonProperty("extractUnions") @Nullable Boolean extractUnions + @JsonProperty("extractUnionsByType") @Nullable Boolean extractUnionsByType ) throws Exception { super(flattenSpec); @@ -65,7 +65,7 @@ public AvroOCFInputFormat( this.readerSchema = null; } this.binaryAsString = binaryAsString != null && binaryAsString; - this.extractUnions = extractUnions != null && extractUnions; + this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType; } @Override @@ -86,7 +86,7 @@ public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity readerSchema, getFlattenSpec(), binaryAsString, - extractUnions + extractUnionsByType ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java index f47a77acfec9..dc2a6d7eaad5 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroOCFReader.java @@ -57,14 +57,14 @@ public class AvroOCFReader extends IntermediateRowParsingReader @Nullable Schema readerSchema, JSONPathSpec flattenSpec, boolean binaryAsString, - boolean extractUnions + boolean extractUnionsByType ) { this.inputRowSchema = inputRowSchema; this.source = source; this.temporaryDirectory = temporaryDirectory; this.readerSchema = readerSchema; - this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, extractUnions)); + this.recordFlattener = ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(false, binaryAsString, extractUnionsByType)); } @Override diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java index b8bc7f0dcb54..6d399a8dc679 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroParsers.java @@ -40,7 +40,7 @@ public static ObjectFlattener makeFlattener( final ParseSpec parseSpec, final boolean fromPigAvroStorage, final boolean binaryAsString, - final boolean extractUnions + final boolean extractUnionsByType ) { final JSONPathSpec flattenSpec; @@ -50,7 +50,7 @@ public static ObjectFlattener makeFlattener( flattenSpec = JSONPathSpec.DEFAULT; } - return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString, extractUnions)); + return ObjectFlatteners.create(flattenSpec, new AvroFlattenerMaker(fromPigAvroStorage, binaryAsString, extractUnionsByType)); } public static List parseGenericRecord( diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java index afb820060764..d42f3593da6b 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamInputFormat.java @@ -35,7 +35,7 @@ public class AvroStreamInputFormat extends NestedInputFormat { private final boolean binaryAsString; - private final boolean extractUnions; + private final boolean extractUnionsByType; private final AvroBytesDecoder avroBytesDecoder; @@ -44,13 +44,13 @@ public AvroStreamInputFormat( @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec, @JsonProperty("avroBytesDecoder") AvroBytesDecoder avroBytesDecoder, @JsonProperty("binaryAsString") @Nullable Boolean binaryAsString, - @JsonProperty("extractUnions") @Nullable Boolean extractUnions + @JsonProperty("extractUnionsByType") @Nullable Boolean extractUnionsByType ) { super(flattenSpec); this.avroBytesDecoder = avroBytesDecoder; this.binaryAsString = binaryAsString != null && binaryAsString; - this.extractUnions = extractUnions != null && extractUnions; + this.extractUnionsByType = extractUnionsByType != null && extractUnionsByType; } @Override @@ -80,7 +80,7 @@ public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity avroBytesDecoder, getFlattenSpec(), binaryAsString, - extractUnions + extractUnionsByType ); } diff --git a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java index a8f10098b770..c04ec14c5f9e 100644 --- a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java +++ b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/AvroStreamReader.java @@ -53,13 +53,13 @@ public class AvroStreamReader extends IntermediateRowParsingReader type.getType() != Schema.Type.NULL).count() > 1; } - private Map extractUnion(final Object o) + private Map extractUnionTypes(final Object o) { // Primitive types and unnamped complex types are keyed their type name. // Complex named types are keyed by their names. diff --git a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java index db1387f235d9..e064e7af080b 100644 --- a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java +++ b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/AvroFlattenerMakerTest.java @@ -232,7 +232,7 @@ public void makeJsonPathExtractor() } @Test - public void jsonPathExtractorextractUnions() + public void jsonPathExtractorExtractUnionsByType() { final AvroFlattenerMaker flattener = new AvroFlattenerMaker(false, false, true); diff --git a/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java b/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java index f4cb5d6f6b5b..796891ea8ef8 100755 --- a/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java +++ b/extensions-core/parquet-extensions/src/main/java/org/apache/druid/data/input/parquet/avro/ParquetAvroHadoopInputRowParser.java @@ -51,7 +51,7 @@ public class ParquetAvroHadoopInputRowParser implements InputRowParser recordFlattener; private final List dimensions; @@ -60,14 +60,14 @@ public class ParquetAvroHadoopInputRowParser implements InputRowParser Date: Sun, 27 Jun 2021 14:14:05 +0700 Subject: [PATCH 9/9] Correct spelling error --- docs/development/extensions-core/avro.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/development/extensions-core/avro.md b/docs/development/extensions-core/avro.md index 333f0dfb560c..75704cfcc0c9 100644 --- a/docs/development/extensions-core/avro.md +++ b/docs/development/extensions-core/avro.md @@ -42,7 +42,7 @@ Druid supports most Avro types natively, there are however some exceptions which #### Unions Druid has two modes for supporting `union` types. -The default mode will treat unions as a single value regardless of the type it is poulated with. +The default mode will treat unions as a single value regardless of the type it is populated with. If you wish to operate on each different member of a union however you can set `extractUnionsByType` on the Avro parser in which case unions will be expanded into nested objects according to the following rules: * Primitive types and unnamed complex types are keyed their type name. i.e `int`, `string`