(parquet-avro) Automatically detect list encodings in AvroReadSupport

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

@@ -31,7 +31,10 @@
 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 +66,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 +139,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 +185,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 +240,67 @@ private GenericData getDataModel(ParquetConfiguration conf, Schema schema) {
     return ReflectionUtils.newInstance(suppClass, ConfigurationUtil.createHadoopConfiguration(conf))
         .get();
   }
+
+  // Creates a copy of the user's Configuration with derived list encoding properties
+  private static ParquetConfiguration getDerivedListEncodingConf(
+      ParquetConfiguration configuration, MessageType fileSchema) {
+    Configuration copiedConfiguration =
+        new Configuration(ConfigurationUtil.createHadoopConfiguration(configuration));
+
+    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)) {
+      copiedConfiguration.setBoolean(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, false);
+      copiedConfiguration.setBoolean(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS, false);
+    }
+
+    return new HadoopParquetConfiguration(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;
+  }
 }