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

Author: clairemcginty

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
 

parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java

@@ -28,10 +28,14 @@
 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 ReadContext init(
     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 RecordMaterializer<T> prepareForRead(
       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 @@ private GenericData getDataModel(ParquetConfiguration conf, Schema schema) {
     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;
+  }
 }