diff --git a/parquet-column/src/main/java/parquet/io/ColumnIOFactory.java b/parquet-column/src/main/java/parquet/io/ColumnIOFactory.java index 7244b773b2..025273576d 100644 --- a/parquet-column/src/main/java/parquet/io/ColumnIOFactory.java +++ b/parquet-column/src/main/java/parquet/io/ColumnIOFactory.java @@ -43,10 +43,16 @@ public class ColumnIOCreatorVisitor implements TypeVisitor { private final MessageType requestedSchema; private int currentRequestedIndex; private Type currentRequestedType; + private boolean strictTypeChecking; public ColumnIOCreatorVisitor(boolean validating, MessageType requestedSchema) { + this(validating, requestedSchema, true); + } + + public ColumnIOCreatorVisitor(boolean validating, MessageType requestedSchema, boolean strictTypeChecking) { this.validating = validating; this.requestedSchema = requestedSchema; + this.strictTypeChecking = strictTypeChecking; } @Override @@ -86,7 +92,8 @@ private void visitChildren(GroupColumnIO newIO, GroupType groupType, GroupType r @Override public void visit(PrimitiveType primitiveType) { - if (!currentRequestedType.isPrimitive() || currentRequestedType.asPrimitiveType().getPrimitiveTypeName() != primitiveType.getPrimitiveTypeName()) { + if (!currentRequestedType.isPrimitive() || + (this.strictTypeChecking && currentRequestedType.asPrimitiveType().getPrimitiveTypeName() != primitiveType.getPrimitiveTypeName())) { incompatibleSchema(primitiveType, currentRequestedType); } PrimitiveColumnIO newIO = new PrimitiveColumnIO(primitiveType, current, currentRequestedIndex, leaves.size()); @@ -127,7 +134,17 @@ public ColumnIOFactory(boolean validating) { * @return the corresponding serializing/deserializing structure */ public MessageColumnIO getColumnIO(MessageType requestedSchema, MessageType fileSchema) { - ColumnIOCreatorVisitor visitor = new ColumnIOCreatorVisitor(validating, requestedSchema); + return getColumnIO(requestedSchema, fileSchema, true); + } + + /** + * @param schema the requestedSchema we want to read/write + * @param fileSchema the file schema (when reading it can be different from the requested schema) + * @param strict should file type and requested primitive types match + * @return the corresponding serializing/deserializing structure + */ + public MessageColumnIO getColumnIO(MessageType requestedSchema, MessageType fileSchema, boolean strict) { + ColumnIOCreatorVisitor visitor = new ColumnIOCreatorVisitor(validating, requestedSchema, strict); fileSchema.accept(visitor); return visitor.getColumnIO(); } diff --git a/parquet-column/src/main/java/parquet/schema/GroupType.java b/parquet-column/src/main/java/parquet/schema/GroupType.java index c4e5ff6f37..fcc801b70d 100644 --- a/parquet-column/src/main/java/parquet/schema/GroupType.java +++ b/parquet-column/src/main/java/parquet/schema/GroupType.java @@ -293,6 +293,11 @@ protected List convertChildren(List path, TypeConverter con @Override protected Type union(Type toMerge) { + return union(toMerge, true); + } + + @Override + protected Type union(Type toMerge, boolean strict) { if (toMerge.isPrimitive()) { throw new IncompatibleSchemaModificationException("can not merge primitive type " + toMerge + " into group type " + this); } @@ -305,6 +310,16 @@ protected Type union(Type toMerge) { * @return the merged list */ List mergeFields(GroupType toMerge) { + return mergeFields(toMerge, true); + } + + /** + * produces the list of fields resulting from merging toMerge into the fields of this + * @param toMerge the group containing the fields to merge + * @param strict should schema primitive types match + * @return the merged list + */ + List mergeFields(GroupType toMerge, boolean strict) { List newFields = new ArrayList(); // merge existing fields for (Type type : this.getFields()) { @@ -314,7 +329,7 @@ List mergeFields(GroupType toMerge) { if (fieldToMerge.getRepetition().isMoreRestrictiveThan(type.getRepetition())) { throw new IncompatibleSchemaModificationException("repetition constraint is more restrictive: can not merge type " + fieldToMerge + " into " + type); } - merged = type.union(fieldToMerge); + merged = type.union(fieldToMerge, strict); } else { merged = type; } diff --git a/parquet-column/src/main/java/parquet/schema/MessageType.java b/parquet-column/src/main/java/parquet/schema/MessageType.java index f71e3ce6a1..9c057475c3 100644 --- a/parquet-column/src/main/java/parquet/schema/MessageType.java +++ b/parquet-column/src/main/java/parquet/schema/MessageType.java @@ -137,7 +137,11 @@ public boolean containsPath(String[] path) { } public MessageType union(MessageType toMerge) { - return new MessageType(this.getName(), mergeFields(toMerge)); + return union(toMerge, true); + } + + public MessageType union(MessageType toMerge, boolean strict) { + return new MessageType(this.getName(), mergeFields(toMerge, strict)); } } diff --git a/parquet-column/src/main/java/parquet/schema/PrimitiveType.java b/parquet-column/src/main/java/parquet/schema/PrimitiveType.java index 5e1d9b3f58..3a734a9b63 100644 --- a/parquet-column/src/main/java/parquet/schema/PrimitiveType.java +++ b/parquet-column/src/main/java/parquet/schema/PrimitiveType.java @@ -276,7 +276,7 @@ abstract public void addValueToPrimitiveConverter( private final PrimitiveTypeName primitive; private final int length; private final DecimalMetadata decimalMeta; - + /** * @param repetition OPTIONAL, REPEATED, REQUIRED * @param primitive STRING, INT64, ... @@ -486,7 +486,12 @@ protected boolean containsPath(String[] path, int depth) { @Override protected Type union(Type toMerge) { - if (!toMerge.isPrimitive() || !primitive.equals(toMerge.asPrimitiveType().getPrimitiveTypeName())) { + return union(toMerge, true); + } + + @Override + protected Type union(Type toMerge, boolean strict) { + if (!toMerge.isPrimitive() || (strict && !primitive.equals(toMerge.asPrimitiveType().getPrimitiveTypeName()))) { throw new IncompatibleSchemaModificationException("can not merge type " + toMerge + " into " + this); } Types.PrimitiveBuilder builder = Types.primitive( diff --git a/parquet-column/src/main/java/parquet/schema/Type.java b/parquet-column/src/main/java/parquet/schema/Type.java index a3fe23caa6..a785394fa9 100644 --- a/parquet-column/src/main/java/parquet/schema/Type.java +++ b/parquet-column/src/main/java/parquet/schema/Type.java @@ -195,6 +195,13 @@ public boolean equals(Object other) { * @return the union result of merging toMerge into this */ protected abstract Type union(Type toMerge); + + /** + * @param toMerge the type to merge into this one + * @param strict should schema primitive types match + * @return the union result of merging toMerge into this + */ + protected abstract Type union(Type toMerge, boolean strict); /** * {@inheritDoc} diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java b/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java index f1c952cfd7..8d99a290f8 100644 --- a/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java +++ b/parquet-hadoop/src/main/java/parquet/hadoop/InternalParquetRecordReader.java @@ -37,6 +37,7 @@ import static java.lang.String.format; import static parquet.Log.DEBUG; +import static parquet.hadoop.ParquetInputFormat.STRICT_TYPE_CHECKING; class InternalParquetRecordReader { private static final Log LOG = Log.getLog(InternalParquetRecordReader.class); @@ -57,6 +58,7 @@ class InternalParquetRecordReader { private ParquetFileReader reader; private parquet.io.RecordReader recordReader; private UnboundRecordFilter recordFilter; + private boolean strictTypeChecking; private long totalTimeSpentReadingBytes; private long totalTimeSpentProcessingRecords; @@ -106,7 +108,7 @@ private void checkRead() throws IOException { BenchmarkCounter.incrementTime(timeSpentReading); LOG.info("block read in memory in " + timeSpentReading + " ms. row count = " + pages.getRowCount()); if (Log.DEBUG) LOG.debug("initializing Record assembly with requested schema " + requestedSchema); - MessageColumnIO columnIO = columnIOFactory.getColumnIO(requestedSchema, fileSchema); + MessageColumnIO columnIO = columnIOFactory.getColumnIO(requestedSchema, fileSchema, strictTypeChecking); recordReader = columnIO.getRecordReader(pages, recordConverter, recordFilter); startedAssemblingCurrentBlockAt = System.currentTimeMillis(); totalCountLoadedSoFar += pages.getRowCount(); @@ -142,7 +144,7 @@ public void initialize(MessageType requestedSchema, MessageType fileSchema, this.recordConverter = readSupport.prepareForRead( configuration, extraMetadata, fileSchema, new ReadSupport.ReadContext(requestedSchema, readSupportMetadata)); - + this.strictTypeChecking = configuration.getBoolean(STRICT_TYPE_CHECKING, true); List columns = requestedSchema.getColumns(); reader = new ParquetFileReader(configuration, file, blocks, columns); for (BlockMetaData block : blocks) { diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetFileWriter.java index f5fe167984..ff291797d1 100644 --- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetFileWriter.java @@ -83,7 +83,7 @@ public class ParquetFileWriter { private long currentChunkFirstDataPage; private long currentChunkDictionaryPageOffset; private long currentChunkValueCount; - + private Statistics currentStatistics; /** @@ -439,11 +439,16 @@ public long getPos() throws IOException { * @param footers the list files footers to merge * @return the global meta data for all the footers */ + static GlobalMetaData getGlobalMetaData(List