FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

(parquet-avro) Automatically detect list encodings in AvroReadSupport by clairemcginty · Pull Request #3753 · apache/parquet-java · GitHub

/ parquet-java Public

(parquet-avro) Automatically detect list encodings in AvroReadSupport #3753

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
clairemcginty wants to merge 1 commit into
base: master
Choose a base branch
Could not load branches
Branch not found: {{ refName }}

Are you sure you want to change the base?

Some commits from the old base branch may be removed from the timeline, and old review comments may become outdated.
from
+217 −8
Conversation 0 Commits 1 Checks 0 Files changed 3
Draft

(parquet-avro) Automatically detect list encodings in AvroReadSupport #3753

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension .java  (2) .md  (1) All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
17 changes: 9 additions & 8 deletions parquet-avro/README.md
Show comments
View file
Open in desktop
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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
55 changes: 55 additions & 0 deletions parquet-avro/src/main/java/org/apache/parquet/avro/AvroReadSupport.java
Show comments
View file
Open in desktop
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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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);
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);
Expand Down Expand Up @@ -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;
}
}
153 changes: 153 additions & 0 deletions parquet-avro/src/test/java/org/apache/parquet/avro/TestArrayCompatibility.java
Show comments
View file
Open in desktop
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
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 <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);
}
Expand Down
Back | FazBrowse Home | New Git URL