This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/parquet-java.git
The following commit(s) were added to refs/heads/master by this push:
new b1e2d0358 GH-3816: Automatically detect list encodings in
AvroReadSupport (#3753)
b1e2d0358 is described below
commit b1e2d0358c4f8a652a44307a897944c4ca1e5e73
Author: Claire McGinty <[email protected]>
AuthorDate: Thu Sep 24 00:47:18 2026 -0400
GH-3816: Automatically detect list encodings in AvroReadSupport (#3753)
### Rationale for this change
parquet-avro supports writing both "old" and "new" list encodings via the
[parquet.avro.write-old-list-structure](https://github.com/apache/parquet-java/blob/apache-parquet-1.18.0/parquet-avro/src/main/java/org/apache/parquet/avro/AvroWriteSupport.java#L72-L73)
config. "old" encodings (aka "2-level"), which wrap the list in a `repeated
group array` schema, are the default; "new" encodings (aka "3-level") are
opt-in.
On the reader side, if you're using `ParquetAvroReader` to read data that
was written using `ParquetAvroWriter`, and don't specify a projection, both
type sof list encoding get parsed automatically from a combination of the file
schema + the `parquet.avro.schema` metadata key. There's no need to set
`parquet.avro.write-old-list-structure` key in your Configuration.
However, if you're either:
- specifying a projection (`AvroReadSupport.setRequestedProjection(...)`),
or
- reading data _not_ written using ParquetAvroWriter (and thus not
containing the `parquet.avro.schema` metadata key),
3-levle list encodings will not be parsed correctly - the reader will
inject an extra nested record, named `element`, into the list item type.
As a reader this introduces some pain, since you have to look up the
underlying file metadata of the upstream Parquet file, and risk reading
incorrect data. This PR attempts to automatically detect new list encodings
based on the writer file schema.
lmk what you think of this change. Automatic inference is always a bit
risky, but I tried to be conservative with the approach (only set the list
structure property if _all_ list fields in the schema use 3-level encoding;
don't override `parquet.avro.write-old-list-structure` if the user is already
setting it). any ideas for a better approach here are welcome - this is
becoming more of a pain point as 3-level lists become a more popular option
among other writer sdks.
### What changes are included in this PR?
A new read configuration property
`parquet.avro.read.autoDetectListStructure` (defaulting to true) that will
instruct AvroReadSupport to automatically set List configuration properties
based on parsing the writer file schema.
### Are these changes tested?
Yes, unit tests + locally on real data.
### Are there any user-facing changes?
Yes, since the new property defaults to `true` - it would impact anyone
who's reading 3-level list data without setting the
`parquet.avro.write-old-list-structure` key and who's relying on/working around
the incorrectly formatted data (e.g. `{"locations": [{"element": {"latitude":
0.0, "longitude": 180.0}}, ...]}` instead of `{"locations": [{"latitude": 0.0,
"longitude": 180.0}, ...]}` .
additionally, this change also modifies the underlying Configuration object
to add the properties.
Closes #3816
---
parquet-avro/README.md | 17 +-
.../org/apache/parquet/avro/AvroReadSupport.java | 85 ++++-
.../parquet/avro/TestArrayCompatibility.java | 379 +++++++++++++++++++++
3 files changed, 471 insertions(+), 10 deletions(-)
diff --git a/parquet-avro/README.md b/parquet-avro/README.md
index 644be9bfa..9bed4a6b3 100644
--- a/parquet-avro/README.md
+++ b/parquet-avro/README.md
@@ -26,14 +26,15 @@ Apache Avro integration
### Configuration for reading
-| Name | Type | Description
|
-|-----------------------------------------|-----------|----------------------------------------------------------------------|
-| `parquet.avro.data.supplier` | `Class` | The implementation of
the interface org.apache.parquet.avro.AvroDataSupplier. Available
implementations in the library: GenericDataSupplier, ReflectDataSupplier,
SpecificDataSupplier.<br/>The default value is
`org.apache.parquet.avro.SpecificDataSupplier` |
-| `parquet.avro.read.schema` | `String` | The Avro schema to be
used for reading. It shall be compatible with the file schema. The file schema
will be used directly if not set. |
-| `parquet.avro.projection` | `String` | The Avro schema to be
used for projection. |
-| `parquet.avro.compatible` | `boolean` | Flag for compatibility
mode. `true` for materializing Avro `IndexedRecord` objects, `false` for
materializing the related objects for either generic, specific, or reflect
records.<br/>The default value is `true`. |
-| `parquet.avro.readInt96AsFixed` | `boolean` | Flag for handling the
`INT96` Parquet types. `true` for converting it to the `fixed` Avro type,
`false` for not handling `INT96` types (throwing exception).<br/>The default
value is `false`.<br/>**NOTE: The `INT96` Parquet type is deprecated. This
option is only to support old data.** |
-| `parquet.avro.serializable.classes` | `String` | List of the fully
qualified class names separated by ',' that may be referenced from the Avro
schema by "java-class" or "java-key-class" and are allowed to be loaded. |
+| Name | Type | Description
|
+|---------------------------------------------|-----------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `parquet.avro.data.supplier` | `Class` | The implementation
of the interface org.apache.parquet.avro.AvroDataSupplier. Available
implementations in the library: GenericDataSupplier, ReflectDataSupplier,
SpecificDataSupplier.<br/>The default value is
`org.apache.parquet.avro.SpecificDataSupplier` |
+| `parquet.avro.read.schema` | `String` | The Avro schema to
be used for reading. It shall be compatible with the file schema. The file
schema will be used directly if not set.
|
+| `parquet.avro.projection` | `String` | The Avro schema to
be used for projection.
|
+| `parquet.avro.compatible` | `boolean` | Flag for
compatibility mode. `true` for materializing Avro `IndexedRecord` objects,
`false` for materializing the related objects for either generic, specific, or
reflect records.<br/>The default value is `true`.
|
+| `parquet.avro.readInt96AsFixed` | `boolean` | Flag for handling
the `INT96` Parquet types. `true` for converting it to the `fixed` Avro type,
`false` for not handling `INT96` types (throwing exception).<br/>The default
value is `false`.<br/>**NOTE: The `INT96` Parquet type is deprecated. This
option is only to support old data.** |
+| `parquet.avro.serializable.classes` | `String` | List of the fully
qualified class names separated by ',' that may be referenced from the Avro
schema by "java-class" or "java-key-class" and are allowed to be loaded.
|
+| `parquet.avro.read.autoDetectListStructure` | `boolean` | Automatically
detect whether the write schema uses 2- or 3-level list encoding and converts
the projection/read schema accordingly.<br/>The default value is `true`.
|
### Configuration for writing
diff --git
a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java
b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java
index 6d7ca398a..ee8f9ab27 100644
--- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java
+++ b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java
@@ -28,10 +28,14 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.parquet.conf.HadoopParquetConfiguration;
import org.apache.parquet.conf.ParquetConfiguration;
+import org.apache.parquet.conf.PlainParquetConfiguration;
import org.apache.parquet.hadoop.api.ReadSupport;
import org.apache.parquet.hadoop.util.ConfigurationUtil;
import org.apache.parquet.io.api.RecordMaterializer;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,6 +67,11 @@ public class AvroReadSupport<T> extends ReadSupport<T> {
public static final String READ_INT96_AS_FIXED =
"parquet.avro.readInt96AsFixed";
public static final boolean READ_INT96_AS_FIXED_DEFAULT = false;
+ // Automatically detect whether a Parquet file uses 2-level or 3-level
encoding;
+ // Ignored if AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE is also set
+ public static final String AUTO_DETECT_LIST_STRUCTURE =
"parquet.avro.read.autoDetectListStructure";
+ static final boolean AUTO_DETECT_LIST_STRUCTURE_DEFAULT = true;
+
/**
* List of the fully qualified class names separated by ',' that may be
referenced from the Avro schema by
* "java-class" or "java-key-class" and are allowed to be loaded.
@@ -131,7 +140,8 @@ public class AvroReadSupport<T> extends ReadSupport<T> {
String requestedProjectionString =
configuration.get(AVRO_REQUESTED_PROJECTION);
if (requestedProjectionString != null) {
Schema avroRequestedProjection = new
Schema.Parser().parse(requestedProjectionString);
- projection = new
AvroSchemaConverter(configuration).convert(avroRequestedProjection);
+ projection = new
AvroSchemaConverter(getDerivedListEncodingConf(configuration, fileSchema))
+ .convert(avroRequestedProjection);
}
String avroReadSchema = configuration.get(AVRO_READ_SCHEMA);
@@ -176,7 +186,8 @@ public class AvroReadSupport<T> extends ReadSupport<T> {
avroSchema = new
Schema.Parser().parse(keyValueMetaData.get(OLD_AVRO_SCHEMA_METADATA_KEY));
} else {
// default to converting the Parquet schema into an Avro schema
- avroSchema = new
AvroSchemaConverter(configuration).convert(parquetSchema);
+ avroSchema = new
AvroSchemaConverter(getDerivedListEncodingConf(configuration, fileSchema))
+ .convert(parquetSchema);
}
GenericData model = getDataModel(configuration, avroSchema);
@@ -230,4 +241,74 @@ public class AvroReadSupport<T> extends ReadSupport<T> {
return ReflectionUtils.newInstance(suppClass,
ConfigurationUtil.createHadoopConfiguration(conf))
.get();
}
+
+ // Returns a ParquetConfiguration with appropriate list-decoding properties
set, inferred from
+ // the file schema as well as user-supplied Configuration properties.
+ // If no configuration changes are required, the original
ParquetConfiguration object will be;
+ // returned; otherwise, a copy will be created with the correct properties.
+ private static ParquetConfiguration getDerivedListEncodingConf(
+ ParquetConfiguration configuration, MessageType fileSchema) {
+ final boolean autoDetectListStructure =
+ configuration.getBoolean(AUTO_DETECT_LIST_STRUCTURE,
AUTO_DETECT_LIST_STRUCTURE_DEFAULT);
+
+ if (!autoDetectListStructure
+ || configuration.get(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE) != null
+ || configuration.get(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS) !=
null
+ || !writesNewListStructure(fileSchema)) {
+ return configuration;
+ }
+
+ // Avoid mutating the original Configuration by creating a copy, with new
properties set
+ final ParquetConfiguration copiedConfiguration = new
PlainParquetConfiguration();
+ for (Map.Entry<String, String> property : configuration) {
+ copiedConfiguration.set(property.getKey(), property.getValue());
+ }
+ copiedConfiguration.setBoolean(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE,
false);
+
copiedConfiguration.setBoolean(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS,
false);
+
+ return copiedConfiguration;
+ }
+
+ private static boolean writesNewListStructure(MessageType schema) {
+ return Boolean.TRUE.equals(allListStructuresAreThreeLevel(schema));
+ }
+
+ // Given a Parquet schema, return true only if the schema:
+ // - contains one or more List fields
+ // - encodes every List field using 3-level list structure
+ private static Boolean allListStructuresAreThreeLevel(Type type) {
+ if (type.isPrimitive()) {
+ return null;
+ }
+ GroupType group = type.asGroupType();
+ if (group.getLogicalTypeAnnotation() instanceof
LogicalTypeAnnotation.ListLogicalTypeAnnotation) {
+ if (group.isRepetition(Type.Repetition.REPEATED) ||
group.getFieldCount() != 1) {
+ return false;
+ }
+ Type repeated = group.getType(0);
+ if (repeated.isPrimitive()
+ || !repeated.isRepetition(Type.Repetition.REPEATED)
+ || !repeated.getName().equals("list")
+ || repeated.asGroupType().getFieldCount() != 1) {
+ return false;
+ }
+ Type element = repeated.asGroupType().getType(0);
+ if (element.isRepetition(Type.Repetition.REPEATED)
+ || !element.getName().equals("element")) {
+ return false;
+ }
+ return !Boolean.FALSE.equals(allListStructuresAreThreeLevel(element));
+ }
+ Boolean result = null;
+ for (Type field : group.getFields()) {
+ Boolean fieldListStructuresAreThreeLevel =
allListStructuresAreThreeLevel(field);
+ if (Boolean.FALSE.equals(fieldListStructuresAreThreeLevel)) {
+ return false;
+ }
+ if (Boolean.TRUE.equals(fieldListStructuresAreThreeLevel)) {
+ result = true;
+ }
+ }
+ return result;
+ }
}
diff --git
a/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java
b/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java
index 3505b0265..242fd2fcf 100644
---
a/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java
+++
b/parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java
@@ -51,11 +51,14 @@ public class TestArrayCompatibility extends
DirectWriterTest {
public static final Configuration OLD_BEHAVIOR_CONF = new Configuration();
public static final Configuration NEW_BEHAVIOR_CONF = new Configuration();
+ public static final Configuration AUTO_DETECT_CONF = new Configuration();
@BeforeAll
public static void setupNewBehaviorConfiguration() {
OLD_BEHAVIOR_CONF.setBoolean(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS,
true);
+ OLD_BEHAVIOR_CONF.setBoolean(AvroReadSupport.AUTO_DETECT_LIST_STRUCTURE,
false);
NEW_BEHAVIOR_CONF.setBoolean(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS,
false);
+ AUTO_DETECT_CONF.setBoolean(AvroReadSupport.AUTO_DETECT_LIST_STRUCTURE,
true);
}
@Test
@@ -1256,6 +1259,382 @@ public class TestArrayCompatibility extends
DirectWriterTest {
.hasMessage("Parquet/Avro schema mismatch. Avro field 'element' not
found.");
}
+ @Test
+ public void testAutoDetectThreeLevelListEncoding() throws Exception {
+ Path test = writeDirect(
+ "message AutoDetectThreeLevel {"
+ + " optional group locations (LIST) {"
+ + " repeated group list {"
+ + " required group element {"
+ + " required double latitude;"
+ + " required double longitude;"
+ + " }"
+ + " }"
+ + " }"
+ + "}",
+ rc -> {
+ rc.startMessage();
+ rc.startField("locations", 0);
+
+ rc.startGroup();
+ rc.startField("list", 0);
+
+ rc.startGroup();
+ rc.startField("element", 0);
+ rc.startGroup();
+ rc.startField("latitude", 0);
+ rc.addDouble(0.0);
+ rc.endField("latitude", 0);
+ rc.startField("longitude", 1);
+ rc.addDouble(180.0);
+ rc.endField("longitude", 1);
+ rc.endGroup();
+ rc.endField("element", 0);
+ rc.endGroup();
+
+ rc.startGroup();
+ rc.startField("element", 0);
+ rc.startGroup();
+ rc.startField("latitude", 0);
+ rc.addDouble(0.0);
+ rc.endField("latitude", 0);
+ rc.startField("longitude", 1);
+ rc.addDouble(0.0);
+ rc.endField("longitude", 1);
+ rc.endGroup();
+ rc.endField("element", 0);
+ rc.endGroup();
+
+ rc.endField("list", 0);
+ rc.endGroup();
+
+ rc.endField("locations", 0);
+ rc.endMessage();
+ });
+
+ Schema location = record(
+ "element",
+ field("latitude", primitive(Schema.Type.DOUBLE)),
+ field("longitude", primitive(Schema.Type.DOUBLE)));
+
+ // without auto-detect, old behavior wraps repeated group in an extra
"element" record,
+ // e.g. {"locations": [{"element": {"latitude": 0.0, ... }}]}
+ Schema elementRecord = record("list", field("element", location));
+ Schema oldSchema = record("AutoDetectThreeLevel",
optionalField("locations", array(elementRecord)));
+ GenericRecord oldRecord = instance(
+ oldSchema,
+ "locations",
+ Arrays.asList(
+ instance(elementRecord, "element", instance(location, "latitude",
0.0, "longitude", 180.0)),
+ instance(elementRecord, "element", instance(location, "latitude",
0.0, "longitude", 0.0))));
+
+ Configuration nonAutoDetectConf = new Configuration(OLD_BEHAVIOR_CONF);
+ assertReaderContains(new AvroParquetReader<>(nonAutoDetectConf, test),
oldSchema, oldRecord);
+
+ // with auto-detect, repeated group is represented correctly,
+ // e.g. {"locations": [{"latitude": 0.0, ... }]}
+ Schema newSchema = record("AutoDetectThreeLevel",
optionalField("locations", array(location)));
+ GenericRecord newRecord = instance(
+ newSchema,
+ "locations",
+ Arrays.asList(
+ instance(location, "latitude", 0.0, "longitude", 180.0),
+ instance(location, "latitude", 0.0, "longitude", 0.0)));
+
+ assertReaderContains(autoDetectReader(test), newSchema, newRecord);
+ }
+
+ @Test
+ public void testAutoDetectThreeLevelListEncodingWithProjection() throws
Exception {
+ Path test = writeDirect(
+ "message AutoDetectThreeLevelProjection {"
+ + " required int64 year;"
+ + " optional group locations (LIST) {"
+ + " repeated group list {"
+ + " required group element {"
+ + " required double latitude;"
+ + " required double longitude;"
+ + " }"
+ + " }"
+ + " }"
+ + "}",
+ rc -> {
+ rc.startMessage();
+ rc.startField("year", 0);
+ rc.addLong(2010L);
+ rc.endField("year", 0);
+
+ rc.startField("locations", 1);
+
+ rc.startGroup();
+ rc.startField("list", 0);
+
+ rc.startGroup();
+ rc.startField("element", 0);
+ rc.startGroup();
+ rc.startField("latitude", 0);
+ rc.addDouble(0.0);
+ rc.endField("latitude", 0);
+ rc.startField("longitude", 1);
+ rc.addDouble(180.0);
+ rc.endField("longitude", 1);
+ rc.endGroup();
+ rc.endField("element", 0);
+ rc.endGroup();
+
+ rc.endField("list", 0);
+ rc.endGroup();
+
+ rc.endField("locations", 1);
+ rc.endMessage();
+ });
+
+ Schema location = record(
+ "element",
+ field("latitude", primitive(Schema.Type.DOUBLE)),
+ field("longitude", primitive(Schema.Type.DOUBLE)));
+
+ Schema projectionSchema = record("AutoDetectThreeLevelProjection",
optionalField("locations", array(location)));
+
+ Configuration autoDetectConf = new Configuration(AUTO_DETECT_CONF);
+ AvroReadSupport.setRequestedProjection(autoDetectConf, projectionSchema);
+
+ GenericRecord expectedRecord = instance(
+ projectionSchema, "locations", Arrays.asList(instance(location,
"latitude", 0.0, "longitude", 180.0)));
+
+ assertReaderContains(new AvroParquetReader<>(autoDetectConf, test),
projectionSchema, expectedRecord);
+ }
+
+ @Test
+ public void testAutoDetectMixedSiblingListStructuresUsesOldBehavior() {
+ MessageType fileSchema = MessageTypeParser.parseMessageType("message
MixedSiblingListStructures {"
+ + " optional group three_level (LIST) {"
+ + " repeated group list {"
+ + " required int32 element;"
+ + " }"
+ + " }"
+ + " optional group two_level (LIST) {"
+ + " repeated int32 element;"
+ + " }"
+ + "}");
+
+ assertListStructureIsNotAutoDetected(fileSchema);
+ }
+
+ @Test
+ public void testAutoDetectMixedNestedListStructuresUsesOldBehavior() {
+ MessageType fileSchema = MessageTypeParser.parseMessageType("message
MixedNestedListStructures {"
+ + " optional group outer_list (LIST) {"
+ + " repeated group list {"
+ + " required group element (LIST) {"
+ + " repeated int32 element;"
+ + " }"
+ + " }"
+ + " }"
+ + "}");
+
+ assertListStructureIsNotAutoDetected(fileSchema);
+ }
+
+ @Test
+ public void testAutoDetectMalformedThreeLevelListStructuresUsesOldBehavior()
{
+ for (String schema : Arrays.asList(
+ "message RepeatedOuterList {"
+ + " repeated group items (LIST) {"
+ + " repeated group list {"
+ + " required int32 element;"
+ + " }"
+ + " }"
+ + "}",
+ "message NonRepeatedMiddleList {"
+ + " optional group items (LIST) {"
+ + " optional group list {"
+ + " required int32 element;"
+ + " }"
+ + " }"
+ + "}",
+ "message RepeatedListElement {"
+ + " optional group items (LIST) {"
+ + " repeated group list {"
+ + " repeated int32 element;"
+ + " }"
+ + " }"
+ + "}")) {
+
assertListStructureIsNotAutoDetected(MessageTypeParser.parseMessageType(schema));
+ }
+ }
+
+ @Test
+ public void testAutoDetectTwoLevelListWithArrayGroupName() throws Exception {
+ // A 2-level list where the repeated group is named "array" (a standard
+ // backward-compat name). The "array" name causes isElementType to always
+ // treat the repeated group as the element, regardless of the list
structure flag.
+ Path test = writeDirect(
+ "message TwoLevelListWithArrayGroup {"
+ + " optional group list (LIST) {"
+ + " repeated group array {"
+ + " required int32 str;"
+ + " }"
+ + " }"
+ + "}",
+ rc -> {
+ rc.startMessage();
+ rc.startField("list", 0);
+
+ rc.startGroup();
+ rc.startField("array", 0);
+
+ rc.startGroup();
+ rc.startField("str", 0);
+ rc.addInteger(34);
+ rc.endField("str", 0);
+ rc.endGroup();
+
+ rc.startGroup();
+ rc.startField("str", 0);
+ rc.addInteger(35);
+ rc.endField("str", 0);
+ rc.endGroup();
+
+ rc.endField("array", 0);
+ rc.endGroup();
+
+ rc.endField("list", 0);
+ rc.endMessage();
+ });
+
+ // "array"-named group is always treated as the element type
+ Schema elementRecord = record("array", field("str",
primitive(Schema.Type.INT)));
+ Schema expectedSchema = record("TwoLevelListWithArrayGroup",
optionalField("list", array(elementRecord)));
+ GenericRecord expectedRecord = instance(
+ expectedSchema,
+ "list",
+ Arrays.asList(instance(elementRecord, "str", 34),
instance(elementRecord, "str", 35)));
+
+ // all three modes produce the same result
+ assertReaderContains(oldBehaviorReader(test), expectedSchema,
expectedRecord);
+ assertReaderContains(newBehaviorReader(test), expectedSchema,
expectedRecord);
+ assertReaderContains(new AvroParquetReader<>(newAutoDetectConf(), test),
expectedSchema, expectedRecord);
+ }
+
+ @Test
+ public void testAutoDetectTwoLevelListWithArrayGroupAndElementChild() throws
Exception {
+ // A 2-level list where the repeated group is named "array" and its single
+ // child is named "element". The "array" group name dominates:
isElementType
+ // treats the group as the element regardless of the child's name.
+ Path test = writeDirect(
+ "message TwoLevelListWithArrayGroupAndElementChild {"
+ + " optional group my_list (LIST) {"
+ + " repeated group array {"
+ + " required int32 element;"
+ + " }"
+ + " }"
+ + "}",
+ rc -> {
+ rc.startMessage();
+ rc.startField("my_list", 0);
+
+ rc.startGroup();
+ rc.startField("array", 0);
+
+ rc.startGroup();
+ rc.startField("element", 0);
+ rc.addInteger(34);
+ rc.endField("element", 0);
+ rc.endGroup();
+
+ rc.startGroup();
+ rc.startField("element", 0);
+ rc.addInteger(35);
+ rc.endField("element", 0);
+ rc.endGroup();
+
+ rc.endField("array", 0);
+ rc.endGroup();
+
+ rc.endField("my_list", 0);
+ rc.endMessage();
+ });
+
+ // "array"-named group is always the element, even with child named
"element"
+ Schema elementRecord = record("array", field("element",
primitive(Schema.Type.INT)));
+ Schema expectedSchema =
+ record("TwoLevelListWithArrayGroupAndElementChild",
optionalField("my_list", array(elementRecord)));
+ GenericRecord expectedRecord = instance(
+ expectedSchema,
+ "my_list",
+ Arrays.asList(instance(elementRecord, "element", 34),
instance(elementRecord, "element", 35)));
+
+ // all three modes produce the same result
+ assertReaderContains(oldBehaviorReader(test), expectedSchema,
expectedRecord);
+ assertReaderContains(newBehaviorReader(test), expectedSchema,
expectedRecord);
+ assertReaderContains(new AvroParquetReader<>(newAutoDetectConf(), test),
expectedSchema, expectedRecord);
+ }
+
+ @Test
+ public void testAutoDetectTwoLevelRepeatedPrimitive() throws Exception {
+ // The most basic backward-compat rule: a repeated primitive inside a LIST
+ // group is the element type directly, producing a required list of
+ // non-nullable elements.
+ Path test = writeDirect(
+ "message TwoLevelRepeatedPrimitive {"
+ + " required group my_list (LIST) {"
+ + " repeated int32 element;"
+ + " }"
+ + "}",
+ rc -> {
+ rc.startMessage();
+ rc.startField("my_list", 0);
+
+ rc.startGroup();
+ rc.startField("element", 0);
+
+ rc.addInteger(34);
+ rc.addInteger(35);
+ rc.addInteger(36);
+
+ rc.endField("element", 0);
+ rc.endGroup();
+
+ rc.endField("my_list", 0);
+ rc.endMessage();
+ });
+
+ Schema expectedSchema =
+ record("TwoLevelRepeatedPrimitive", field("my_list",
array(Schema.create(Schema.Type.INT))));
+ GenericRecord expectedRecord = instance(expectedSchema, "my_list",
Arrays.asList(34, 35, 36));
+
+ // all three modes produce the same result for repeated primitives
+ assertReaderContains(oldBehaviorReader(test), expectedSchema,
expectedRecord);
+ assertReaderContains(newBehaviorReader(test), expectedSchema,
expectedRecord);
+ assertReaderContains(new AvroParquetReader<>(newAutoDetectConf(), test),
expectedSchema, expectedRecord);
+ }
+
+ private static Configuration newAutoDetectConf() {
+ Configuration conf = new Configuration();
+ conf.setBoolean(AvroReadSupport.AUTO_DETECT_LIST_STRUCTURE, true);
+ return conf;
+ }
+
+ private static void assertListStructureIsNotAutoDetected(MessageType
fileSchema) {
+ Schema projection = record("Projection", field("items",
array(primitive(Schema.Type.INT))));
+ MessageType expectedProjection = new
AvroSchemaConverter().convert(projection);
+ Configuration conf = newAutoDetectConf();
+ AvroReadSupport.setRequestedProjection(conf, projection);
+
+ assertThat(new AvroReadSupport<>()
+ .init(conf, new HashMap<>(), fileSchema)
+ .getRequestedSchema())
+ .as("Should not infer the new list structure for %s",
fileSchema.getName())
+ .isEqualTo(expectedProjection);
+ assertThat(conf.get(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE)).isNull();
+
assertThat(conf.get(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS)).isNull();
+ }
+
+ public <T extends IndexedRecord> AvroParquetReader<T> autoDetectReader(Path
path) throws IOException {
+ return new AvroParquetReader<T>(AUTO_DETECT_CONF, path);
+ }
+
public <T extends IndexedRecord> AvroParquetReader<T> oldBehaviorReader(Path
path) throws IOException {
return new AvroParquetReader<T>(OLD_BEHAVIOR_CONF, path);
}
