(parquet-avro) Automatically detect list encodings in AvroReadSupport #3753
+370
−8
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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. | ||
|
|
@@ -128,6 +136,18 @@ public ReadContext init( | |
| MessageType projection = fileSchema; | ||
| Map<String, String> metadata = new LinkedHashMap<String, String>(); | ||
|
|
||
| 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); | ||
|
Member
There was a problem hiding this comment. This writes the inferred mode back into the shared Configuration. ParquetReader reuses that configuration across files, so after a 3-level file, a later 2-level file (especially with a projection) is still converted as 3-level. Could this stay per-file/read-context instead of mutating the caller's configuration?
Contributor
Author
There was a problem hiding this comment. Yeah - I was a bit worried about that. Mutating the caller conf is simplest because we want to have these properties set in both public ReadContext init(
ParquetConfiguration configuration, Map<String, String> keyValueMetaData, MessageType fileSchema) {
MessageType projection = fileSchema;
Map<String, String> metadata = new LinkedHashMap<String, String>();
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);
+ metadata.put("inferred." + AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false");
+ metadata.put("inferred." + AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS, "false");
}
}
+ // ...read those properties in prepareForRead() and apply to copied ConfigurationOr - we could just re-compute |
||
| 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() | ||
|
Member
There was a problem hiding this comment. I may be misremembering the LIST compatibility rules, so I wanted to check this edge case: could a legacy 2-level list also use the
Contributor
Author
There was a problem hiding this comment. added a few more test cases here and the auto-detect correctly returns false for those cases! lmk if you had any other cases in mind... |
||
| && 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; | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is it a good time to make a shift on the default value of
AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE? It has caused a lot of troubles.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
it would be nice to have the more modern encoding as default! though I'm a little nervous about the implications of changing the schema for existing datasets - I know we have some downstream use cases where a single Parquet reader is reading a globbed filepattern matching multiple partitions of the same dataset - not sure what would happen if some partitions used 2-level encoding and some used 3-level encoding. Same concern about Parquet datasets that are the source of truth for external tables in Snowflake/BigQuery/etc.
maybe I can create an issue for this and request a bit more investigation on possible implications of this change?