From fdbb36e867ffff8646bac3d6c046dbb82a737af8 Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Tue, 25 Aug 2026 12:08:52 -0400 Subject: [PATCH 1/2] (parquet-avro) Automatically detect list encodings in AvroReadSupport --- parquet-avro/README.md | 17 +- .../apache/parquet/avro/AvroReadSupport.java | 55 +++++++ .../parquet/avro/TestArrayCompatibility.java | 153 ++++++++++++++++++ 3 files changed, 217 insertions(+), 8 deletions(-) diff --git a/parquet-avro/README.md b/parquet-avro/README.md index 644be9bfa3..9bed4a6b39 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.
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.
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).
The default value is `false`.
**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.
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.
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).
The default value is `false`.
**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.
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 6d7ca398a4..da7755d9bb 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 @@ -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 extends ReadSupport { 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. @@ -128,6 +136,18 @@ public ReadContext init( MessageType projection = fileSchema; Map metadata = new LinkedHashMap(); + 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) { + if (writesNewListStructure(fileSchema)) { + configuration.setBoolean(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, false); + configuration.setBoolean(AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS, false); + } + } + String requestedProjectionString = configuration.get(AVRO_REQUESTED_PROJECTION); if (requestedProjectionString != null) { Schema avroRequestedProjection = new Schema.Parser().parse(requestedProjectionString); @@ -230,4 +250,39 @@ private GenericData getDataModel(ParquetConfiguration conf, Schema schema) { return ReflectionUtils.newInstance(suppClass, ConfigurationUtil.createHadoopConfiguration(conf)) .get(); } + + 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.getFieldCount() != 1) { + return false; + } + Type repeated = group.getType(0); + return !repeated.isPrimitive() + && repeated.getName().equals("list") + && repeated.asGroupType().getFieldCount() == 1 + && repeated.asGroupType().getType(0).getName().equals("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 3505b0265f..68eee63aff 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,156 @@ public void testIsElementTypeFailsInvalidSchema() throws Exception { .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); + } + + public AvroParquetReader autoDetectReader(Path path) throws IOException { + return new AvroParquetReader(AUTO_DETECT_CONF, path); + } + public AvroParquetReader oldBehaviorReader(Path path) throws IOException { return new AvroParquetReader(OLD_BEHAVIOR_CONF, path); } From bf9d6b4a8322d4d4805623f62afb1680876dc30a Mon Sep 17 00:00:00 2001 From: Claire McGinty Date: Thu, 3 Sep 2026 12:48:42 -0400 Subject: [PATCH 2/2] Test more backwards-compat use cases --- .../parquet/avro/TestArrayCompatibility.java | 153 ++++++++++++++++++ 1 file changed, 153 insertions(+) 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 68eee63aff..b2bed01c1a 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 @@ -1405,6 +1405,159 @@ public void testAutoDetectThreeLevelListEncodingWithProjection() throws Exceptio assertReaderContains(new AvroParquetReader<>(autoDetectConf, test), projectionSchema, expectedRecord); } + @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; + } + public AvroParquetReader autoDetectReader(Path path) throws IOException { return new AvroParquetReader(AUTO_DETECT_CONF, path); }