diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/List3Levels.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/List3Levels.java index 297b063320..701b63e733 100644 --- a/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/List3Levels.java +++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/List3Levels.java @@ -43,6 +43,8 @@ class List3Levels { */ public List3Levels(GroupType list) { if (list.getLogicalTypeAnnotation() != LogicalTypeAnnotation.listType() + && !(list.getLogicalTypeAnnotation() + instanceof LogicalTypeAnnotation.VectorLogicalTypeAnnotation) || list.getFields().size() != 1) { throw new IllegalArgumentException("invalid list type: " + list); } diff --git a/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/SchemaConverter.java b/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/SchemaConverter.java index e779b6b8c2..d8985eeebd 100644 --- a/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/SchemaConverter.java +++ b/parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/SchemaConverter.java @@ -79,6 +79,7 @@ import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type; import org.apache.parquet.schema.Type.Repetition; +import org.apache.parquet.schema.TypeUtil; import org.apache.parquet.schema.Types; import org.apache.parquet.schema.Types.GroupBuilder; @@ -486,6 +487,13 @@ public Optional visit( return of(new ListTypeMapping(arrowField, list3Levels, child)); } + @Override + public Optional visit( + LogicalTypeAnnotation.VectorLogicalTypeAnnotation vectorLogicalType) { + TypeUtil.checkValidVectorSchema(type); + return visit(LogicalTypeAnnotation.listType()); + } + @Override public Optional visit( LogicalTypeAnnotation.MapLogicalTypeAnnotation mapLogicalType) { diff --git a/parquet-arrow/src/test/java/org/apache/parquet/arrow/schema/TestSchemaConverter.java b/parquet-arrow/src/test/java/org/apache/parquet/arrow/schema/TestSchemaConverter.java index 4d8a21a8d0..a8d517feaf 100644 --- a/parquet-arrow/src/test/java/org/apache/parquet/arrow/schema/TestSchemaConverter.java +++ b/parquet-arrow/src/test/java/org/apache/parquet/arrow/schema/TestSchemaConverter.java @@ -55,13 +55,33 @@ import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; import org.apache.parquet.schema.Types; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; /** * @see SchemaConverter */ public class TestSchemaConverter { + @ParameterizedTest + @ValueSource(strings = {"required", "optional"}) + public void testVectorAsList(String repetition) { + String children = "repeated group list { required float element; }"; + MessageType vector = MessageTypeParser.parseMessageType( + "message m { " + repetition + " group v (VECTOR(5)) { " + children + " } }"); + MessageType list = MessageTypeParser.parseMessageType( + "message m { " + repetition + " group v (LIST) { " + children + " } }"); + SchemaMapping mapping = converter.fromParquet(vector); + assertThat(mapping.getArrowSchema()) + .isEqualTo(converter.fromParquet(list).getArrowSchema()); + Field field = mapping.getArrowSchema().findField("v"); + assertThat(field.getType()).isEqualTo(new ArrowType.List()); + assertThat(field.getChildren()).hasSize(1); + assertThat(field.getChildren().get(0).getType()) + .isEqualTo(new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE)); + } private static Field field(String name, boolean nullable, ArrowType type, Field... children) { if (nullable) { 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 ee8f9ab272..62332b197e 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 @@ -281,7 +281,8 @@ private static Boolean allListStructuresAreThreeLevel(Type type) { return null; } GroupType group = type.asGroupType(); - if (group.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation) { + if (group.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation + || group.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.VectorLogicalTypeAnnotation) { if (group.isRepetition(Type.Repetition.REPEATED) || group.getFieldCount() != 1) { return false; } diff --git a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java index 782c009db4..8a395a2f2e 100644 --- a/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java +++ b/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java @@ -71,6 +71,7 @@ import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type; +import org.apache.parquet.schema.TypeUtil; import org.apache.parquet.schema.Types; /** @@ -470,7 +471,9 @@ public Optional visit( if (!repeatedType.isRepetition(REPEATED)) { throw new UnsupportedOperationException("Invalid list type " + parquetGroupType); } - if (isElementType(repeatedType, parquetGroupType.getName())) { + if (!(logicalTypeAnnotation + instanceof LogicalTypeAnnotation.VectorLogicalTypeAnnotation) + && isElementType(repeatedType, parquetGroupType.getName())) { // repeated element types are always required return of(Schema.createArray(convertField(repeatedType, names))); } else { @@ -484,6 +487,13 @@ public Optional visit( } } + @Override + public Optional visit( + LogicalTypeAnnotation.VectorLogicalTypeAnnotation vectorLogicalType) { + TypeUtil.checkValidVectorSchema(parquetGroupType); + return visit(LogicalTypeAnnotation.listType()); + } + @Override // for backward-compatibility public Optional visit( 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 242fd2fcf0..61c93362f2 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 @@ -39,13 +39,20 @@ import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.parquet.DirectWriterTest; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.io.InvalidRecordException; +import org.apache.parquet.io.LocalInputFile; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; public class TestArrayCompatibility extends DirectWriterTest { @@ -1610,6 +1617,49 @@ public void testAutoDetectTwoLevelRepeatedPrimitive() throws Exception { assertReaderContains(new AvroParquetReader<>(newAutoDetectConf(), test), expectedSchema, expectedRecord); } + @Test + public void testVectorWithoutAvroSchema(@TempDir java.nio.file.Path dir) throws IOException { + MessageType schema = MessageTypeParser.parseMessageType("message VectorWithoutAvroSchema {" + + " required group v (VECTOR(2)) { repeated group list { required float element; } }" + + " optional group o (VECTOR(1)) { repeated group list { required float element; } } }"); + Path path = new Path(dir.resolve("vector.parquet").toUri()); + SimpleGroupFactory factory = new SimpleGroupFactory(schema); + Group first = factory.newGroup(); + first.addGroup("v").addGroup("list").append("element", 1.0f); + first.getGroup("v", 0).addGroup("list").append("element", 2.0f); + first.addGroup("o").addGroup("list").append("element", 9.0f); + Group second = factory.newGroup(); + second.addGroup("v").addGroup("list").append("element", 3.0f); + second.getGroup("v", 0).addGroup("list").append("element", 4.0f); + try (ParquetWriter writer = ExampleParquetWriter.builder(path) + .withType(schema) + .withWriterVersion(org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_1_0) + .withDictionaryEncoding(false) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withValidation(false) + .build()) { + writer.write(first); + writer.write(second); + } + try (ParquetFileReader reader = ParquetFileReader.open(new LocalInputFile(dir.resolve("vector.parquet")))) { + assertThat(reader.getFileMetaData().getKeyValueMetaData()) + .doesNotContainKeys( + AvroReadSupport.AVRO_SCHEMA_METADATA_KEY, AvroReadSupport.OLD_AVRO_SCHEMA_METADATA_KEY); + } + Schema expectedSchema = record( + "VectorWithoutAvroSchema", + field("v", array(primitive(Schema.Type.FLOAT))), + optionalField("o", array(primitive(Schema.Type.FLOAT)))); + assertThat(new AvroSchemaConverter().convert(schema)).isEqualTo(expectedSchema); + try (AvroParquetReader reader = autoDetectReader(path)) { + assertReaderContains( + reader, + expectedSchema, + instance(expectedSchema, "v", Arrays.asList(1.0f, 2.0f), "o", Arrays.asList(9.0f)), + instance(expectedSchema, "v", Arrays.asList(3.0f, 4.0f), "o", null)); + } + } + private static Configuration newAutoDetectConf() { Configuration conf = new Configuration(); conf.setBoolean(AvroReadSupport.AUTO_DETECT_LIST_STRUCTURE, true); diff --git a/parquet-column/src/main/java/org/apache/parquet/io/ColumnIOFactory.java b/parquet-column/src/main/java/org/apache/parquet/io/ColumnIOFactory.java index 24868acaf1..a28d481cbf 100644 --- a/parquet-column/src/main/java/org/apache/parquet/io/ColumnIOFactory.java +++ b/parquet-column/src/main/java/org/apache/parquet/io/ColumnIOFactory.java @@ -24,6 +24,7 @@ import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.Type; +import org.apache.parquet.schema.TypeUtil; import org.apache.parquet.schema.TypeVisitor; /** @@ -53,6 +54,7 @@ private ColumnIOCreatorVisitor( @Override public void visit(MessageType messageType) { + TypeUtil.checkValidVectorSchema(messageType); columnIO = new MessageColumnIO(requestedSchema, validating, createdBy); visitChildren(columnIO, messageType, requestedSchema); columnIO.setLevels(); diff --git a/parquet-column/src/main/java/org/apache/parquet/io/FilteredRecordReader.java b/parquet-column/src/main/java/org/apache/parquet/io/FilteredRecordReader.java index f877842945..3c52368c9b 100644 --- a/parquet-column/src/main/java/org/apache/parquet/io/FilteredRecordReader.java +++ b/parquet-column/src/main/java/org/apache/parquet/io/FilteredRecordReader.java @@ -84,16 +84,19 @@ private void skipToMatch() { State currentState = getState(0); do { ColumnReader columnReader = currentState.column; + int d = columnReader.getCurrentDefinitionLevel(); + currentState.validateVector(d); // currentLevel = depth + 1 at this point // set the current value - if (columnReader.getCurrentDefinitionLevel() >= currentState.maxDefinitionLevel) { + if (d >= currentState.maxDefinitionLevel) { columnReader.skip(); } columnReader.consume(); // Based on repetition level work out next state to go to int nextR = currentState.maxRepetitionLevel == 0 ? 0 : columnReader.getCurrentRepetitionLevel(); + currentState.endVector(nextR); currentState = currentState.getNextState(nextR); } while (currentState != null); ++recordsRead; diff --git a/parquet-column/src/main/java/org/apache/parquet/io/MessageColumnIO.java b/parquet-column/src/main/java/org/apache/parquet/io/MessageColumnIO.java index 0ab2db39f6..7150182abb 100644 --- a/parquet-column/src/main/java/org/apache/parquet/io/MessageColumnIO.java +++ b/parquet-column/src/main/java/org/apache/parquet/io/MessageColumnIO.java @@ -47,6 +47,7 @@ import org.apache.parquet.io.api.RecordConsumer; import org.apache.parquet.io.api.RecordMaterializer; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.TypeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -508,7 +509,9 @@ public void flush() { public RecordConsumer getRecordWriter(ColumnWriteStore columns) { RecordConsumer recordWriter = new MessageColumnIORecordConsumer(columns); if (DEBUG) recordWriter = new RecordConsumerLoggingWrapper(recordWriter); - return validating ? new ValidatingRecordConsumer(recordWriter, getType()) : recordWriter; + return validating || TypeUtil.checkValidVectorSchema(getType()) + ? new ValidatingRecordConsumer(recordWriter, getType()) + : recordWriter; } void setLevels() { diff --git a/parquet-column/src/main/java/org/apache/parquet/io/RecordReaderImplementation.java b/parquet-column/src/main/java/org/apache/parquet/io/RecordReaderImplementation.java index 46c4b714a2..e05578c1fe 100644 --- a/parquet-column/src/main/java/org/apache/parquet/io/RecordReaderImplementation.java +++ b/parquet-column/src/main/java/org/apache/parquet/io/RecordReaderImplementation.java @@ -27,11 +27,15 @@ import java.util.Map; import org.apache.parquet.column.ColumnReader; import org.apache.parquet.column.impl.ColumnReadStoreImpl; +import org.apache.parquet.io.api.Binary; import org.apache.parquet.io.api.Converter; import org.apache.parquet.io.api.GroupConverter; import org.apache.parquet.io.api.PrimitiveConverter; import org.apache.parquet.io.api.RecordConsumer; import org.apache.parquet.io.api.RecordMaterializer; +import org.apache.parquet.schema.LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.Float16LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.VectorLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.slf4j.Logger; @@ -192,6 +196,9 @@ public static class State { private Case[][][] caseLookup; private List definedCases; private List undefinedCases; + private final int vectorLength; + private final boolean vectorFloat16; + private long vectorCount; private State( int id, @@ -213,6 +220,69 @@ private State( this.primitiveField = fieldPath[fieldPath.length - 1]; this.indexFieldPath = primitiveColumnIO.getIndexFieldPath(); this.primitiveFieldIndex = indexFieldPath[indexFieldPath.length - 1]; + VectorLogicalTypeAnnotation vector = null; + if (fieldPath.length >= 3) { + LogicalTypeAnnotation logical = + primitiveColumnIO.getParent().getParent().getType().getLogicalTypeAnnotation(); + if (logical instanceof VectorLogicalTypeAnnotation) { + vector = (VectorLogicalTypeAnnotation) logical; + } + } + this.vectorLength = vector == null ? 0 : vector.getNumElements(); + this.vectorFloat16 = vector != null + && primitiveColumnIO.getType().getLogicalTypeAnnotation() instanceof Float16LogicalTypeAnnotation; + } + + void validateVector(int d) { + if (vectorLength == 0) return; + int r = column.getCurrentRepetitionLevel(); + if (d < 0 || d > maxDefinitionLevel || r < 0 || r > maxRepetitionLevel) { + throw invalidVector("invalid levels"); + } + if (vectorCount == 0 && r == maxRepetitionLevel) { + throw invalidVector("invalid continuation"); + } + if (d < maxDefinitionLevel) { + if (vectorCount > 0) throw invalidVector("null within vector"); + if (d == maxDefinitionLevel - 1) throw invalidVector("empty vector"); + return; + } + vectorCount++; + if (vectorCount > vectorLength) { + throw invalidVector("expected " + vectorLength + " elements, got " + vectorCount); + } + if (primitive == PrimitiveTypeName.FLOAT && !Float.isFinite(column.getFloat()) + || primitive == PrimitiveTypeName.DOUBLE && !Double.isFinite(column.getDouble())) { + throw invalidVector("non-finite element"); + } + if (vectorFloat16) { + Binary value = column.getBinary(); + if (value.length() != 2) { + throw invalidVector("invalid FLOAT16 element length: " + value.length()); + } + if ((value.get2BytesLittleEndian() & 0x7c00) == 0x7c00) { + throw invalidVector("non-finite element"); + } + } + } + + void endVector(int nextR) { + if (vectorLength == 0) return; + if (nextR < 0 || nextR > maxRepetitionLevel) throw invalidVector("invalid levels"); + if (nextR == maxRepetitionLevel && vectorCount == 0) { + throw invalidVector("null within vector"); + } + if (nextR < maxRepetitionLevel && vectorCount > 0) { + if (vectorCount != vectorLength) { + throw invalidVector("expected " + vectorLength + " elements, got " + vectorCount); + } + vectorCount = 0; + } + } + + private ParquetDecodingException invalidVector(String reason) { + return new ParquetDecodingException( + "Invalid VECTOR data for " + Arrays.toString(fieldPath) + ": " + reason); } public int getDepth(int definitionLevel) { @@ -413,6 +483,7 @@ public T read() { do { ColumnReader columnReader = currentState.column; int d = columnReader.getCurrentDefinitionLevel(); + currentState.validateVector(d); // creating needed nested groups until the current field (opening tags) int depth = currentState.definitionLevelToDepth[d]; for (; currentLevel <= depth; ++currentLevel) { @@ -427,6 +498,7 @@ public T read() { columnReader.consume(); int nextR = currentState.maxRepetitionLevel == 0 ? 0 : columnReader.getCurrentRepetitionLevel(); + currentState.endVector(nextR); // level to go to close current groups int next = currentState.nextLevel[nextR]; for (; currentLevel > next; currentLevel--) { diff --git a/parquet-column/src/main/java/org/apache/parquet/io/ValidatingRecordConsumer.java b/parquet-column/src/main/java/org/apache/parquet/io/ValidatingRecordConsumer.java index 5a2aa20bf3..2e49f8cc25 100644 --- a/parquet-column/src/main/java/org/apache/parquet/io/ValidatingRecordConsumer.java +++ b/parquet-column/src/main/java/org/apache/parquet/io/ValidatingRecordConsumer.java @@ -32,10 +32,13 @@ import java.util.Deque; import org.apache.parquet.io.api.Binary; import org.apache.parquet.io.api.RecordConsumer; +import org.apache.parquet.schema.LogicalTypeAnnotation.Float16LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.VectorLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type; import org.apache.parquet.schema.Type.Repetition; +import org.apache.parquet.schema.TypeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,12 +55,17 @@ public class ValidatingRecordConsumer extends RecordConsumer { private Deque fields = new ArrayDeque<>(); private Deque previousField = new ArrayDeque<>(); private Deque fieldValueCount = new ArrayDeque<>(); + private VectorLogicalTypeAnnotation vectorType; + private String vectorName; + private long vectorCount; + private int vectorDepth; /** * @param delegate the consumer to pass down the event to * @param schema the schema to validate against */ public ValidatingRecordConsumer(RecordConsumer delegate, MessageType schema) { + TypeUtil.checkValidVectorSchema(schema); this.delegate = delegate; this.types.push(schema); } @@ -110,6 +118,13 @@ private void validateMissingFields(int index) { */ @Override public void endField(String field, int index) { + Type currentType = types.peek().asGroupType().getType(fields.peek()); + if (fieldValueCount.peek() == 0 + && currentType.isRepetition(Repetition.REQUIRED) + && (currentType.getLogicalTypeAnnotation() instanceof VectorLogicalTypeAnnotation + || vectorType != null && types.size() == vectorDepth + 1)) { + throw new InvalidRecordException("required field is missing " + currentType); + } delegate.endField(field, index); fieldValueCount.pop(); previousField.pop(); @@ -121,8 +136,26 @@ public void endField(String field, int index) { */ @Override public void startGroup() { + Type child = types.peek().asGroupType().getType(fields.peek()); + if (child.getLogicalTypeAnnotation() instanceof VectorLogicalTypeAnnotation) { + int count = fieldValueCount.pop() + 1; + fieldValueCount.push(count); + if (count > 1) { + throw new InvalidRecordException("repeated value when the type is not repeated in " + child); + } + vectorType = (VectorLogicalTypeAnnotation) child.getLogicalTypeAnnotation(); + vectorName = child.getName(); + vectorCount = 0; + vectorDepth = types.size() + 1; + } else if (vectorType != null && types.size() == vectorDepth) { + vectorCount++; + if (vectorCount > vectorType.getNumElements()) { + throw new InvalidRecordException("Invalid VECTOR length for " + vectorName + ": expected " + + vectorType.getNumElements() + ", got " + vectorCount); + } + } previousField.push(-1); - types.push(types.peek().asGroupType().getType(fields.peek())); + types.push(child); delegate.startGroup(); } @@ -131,10 +164,21 @@ public void startGroup() { */ @Override public void endGroup() { + boolean endingVector = vectorType != null && types.size() == vectorDepth; + if (endingVector && vectorCount != vectorType.getNumElements()) { + throw new InvalidRecordException("Invalid VECTOR length for " + vectorName + ": expected " + + vectorType.getNumElements() + ", got " + vectorCount); + } delegate.endGroup(); validateMissingFields(types.peek().asGroupType().getFieldCount()); types.pop(); previousField.pop(); + if (endingVector) { + vectorType = null; + vectorName = null; + vectorCount = 0; + vectorDepth = 0; + } } /** @@ -230,6 +274,16 @@ public void addBoolean(boolean value) { @Override public void addBinary(Binary value) { validate(BINARY, INT96, FIXED_LEN_BYTE_ARRAY); + if (vectorType != null + && types.peek().asGroupType().getType(fields.peek()).getLogicalTypeAnnotation() + instanceof Float16LogicalTypeAnnotation) { + if (value.length() != 2) { + throw new InvalidRecordException("Invalid FLOAT16 VECTOR element length: " + value.length()); + } + if ((value.get2BytesLittleEndian() & 0x7c00) == 0x7c00) { + throw new InvalidRecordException("Non-finite VECTOR element in " + vectorName); + } + } delegate.addBinary(value); } @@ -239,6 +293,9 @@ public void addBinary(Binary value) { @Override public void addFloat(float value) { validate(FLOAT); + if (vectorType != null && !Float.isFinite(value)) { + throw new InvalidRecordException("Non-finite VECTOR element in " + vectorName); + } delegate.addFloat(value); } @@ -248,6 +305,9 @@ public void addFloat(float value) { @Override public void addDouble(double value) { validate(DOUBLE); + if (vectorType != null && !Double.isFinite(value)) { + throw new InvalidRecordException("Non-finite VECTOR element in " + vectorName); + } delegate.addDouble(value); } } diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java b/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java index 625e9fd9d3..0fe4147610 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/LogicalTypeAnnotation.java @@ -59,6 +59,13 @@ protected LogicalTypeAnnotation fromString(List params) { return listType(); } }, + VECTOR { + @Override + protected LogicalTypeAnnotation fromString(List params) { + Preconditions.checkArgument(params.size() == 1, "Expecting 1 parameter for VECTOR: %s", params); + return vectorType(Integer.parseInt(params.get(0))); + } + }, VARIANT { @Override protected LogicalTypeAnnotation fromString(List params) { @@ -305,6 +312,10 @@ public static ListLogicalTypeAnnotation listType() { return ListLogicalTypeAnnotation.INSTANCE; } + public static VectorLogicalTypeAnnotation vectorType(int numElements) { + return new VectorLogicalTypeAnnotation(numElements); + } + public static VariantLogicalTypeAnnotation variantType(byte specVersion) { return new VariantLogicalTypeAnnotation(specVersion); } @@ -500,6 +511,50 @@ public int hashCode() { } } + public static class VectorLogicalTypeAnnotation extends LogicalTypeAnnotation { + private final int numElements; + + private VectorLogicalTypeAnnotation(int numElements) { + Preconditions.checkArgument(numElements > 0, "Invalid VECTOR length: %s", numElements); + this.numElements = numElements; + } + + public int getNumElements() { + return numElements; + } + + @Override + public OriginalType toOriginalType() { + return OriginalType.LIST; + } + + @Override + public Optional accept(LogicalTypeAnnotationVisitor visitor) { + return visitor.visit(this); + } + + @Override + LogicalTypeToken getType() { + return LogicalTypeToken.VECTOR; + } + + @Override + protected String typeParametersAsString() { + return "(" + numElements + ")"; + } + + @Override + public boolean equals(Object obj) { + return obj instanceof VectorLogicalTypeAnnotation + && numElements == ((VectorLogicalTypeAnnotation) obj).numElements; + } + + @Override + public int hashCode() { + return 31 * getClass().hashCode() + numElements; + } + } + public static class EnumLogicalTypeAnnotation extends LogicalTypeAnnotation { private static final EnumLogicalTypeAnnotation INSTANCE = new EnumLogicalTypeAnnotation(); @@ -1371,6 +1426,10 @@ default Optional visit(ListLogicalTypeAnnotation listLogicalType) { return empty(); } + default Optional visit(VectorLogicalTypeAnnotation vectorLogicalType) { + return empty(); + } + default Optional visit(VariantLogicalTypeAnnotation variantLogicalType) { return empty(); } diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/TypeUtil.java b/parquet-column/src/main/java/org/apache/parquet/schema/TypeUtil.java index bb1278fc57..b30e39cf8d 100644 --- a/parquet-column/src/main/java/org/apache/parquet/schema/TypeUtil.java +++ b/parquet-column/src/main/java/org/apache/parquet/schema/TypeUtil.java @@ -18,8 +18,104 @@ */ package org.apache.parquet.schema; +import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.Float16LogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.VectorLogicalTypeAnnotation; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.apache.parquet.schema.Type.Repetition; + public class TypeUtil { + public static boolean checkValidVectorSchema(Type type) { + boolean found = type.getLogicalTypeAnnotation() instanceof VectorLogicalTypeAnnotation; + if (found) { + if (type.isPrimitive()) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + type); + } + checkVector(type.asGroupType()); + } else if (!type.isPrimitive()) { + for (Type child : type.asGroupType().getFields()) { + found |= checkValidVectorSchema(child); + } + } + return found; + } + + private static void checkVector(GroupType vector) { + if (vector.getRepetition() == Repetition.REPEATED || vector.getFieldCount() != 1) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + vector); + } + Type listType = vector.getType(0); + if (listType.isPrimitive() + || !listType.getName().equals("list") + || listType.getRepetition() != Repetition.REPEATED + || listType.getLogicalTypeAnnotation() != null) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + vector); + } + GroupType list = listType.asGroupType(); + if (list.getFieldCount() != 1) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + vector); + } + Type elementType = list.getType(0); + if (!elementType.isPrimitive() + || !elementType.getName().equals("element") + || elementType.getRepetition() != Repetition.REQUIRED) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + vector); + } + PrimitiveType element = elementType.asPrimitiveType(); + PrimitiveTypeName physical = element.getPrimitiveTypeName(); + LogicalTypeAnnotation logical = element.getLogicalTypeAnnotation(); + boolean allowed; + switch (physical) { + case BOOLEAN: + case FLOAT: + case DOUBLE: + allowed = logical == null; + break; + case INT32: + case INT64: + allowed = logical == null + || logical instanceof IntLogicalTypeAnnotation + || logical instanceof DecimalLogicalTypeAnnotation; + break; + case FIXED_LEN_BYTE_ARRAY: + allowed = logical instanceof Float16LogicalTypeAnnotation + || logical instanceof DecimalLogicalTypeAnnotation; + break; + default: + allowed = false; + } + if (!allowed) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + vector); + } + try { + Types.PrimitiveBuilder builder = Types.primitive(physical, element.getRepetition()); + if (physical == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) { + builder.length(element.getTypeLength()); + } + if (logical != null) { + builder.as(logical); + } + builder.named(element.getName()); + } catch (IllegalArgumentException | IllegalStateException e) { + throw new InvalidSchemaException("Invalid VECTOR schema: " + vector, e); + } + } + + public static VectorLogicalTypeAnnotation getVectorType(GroupType schema, String[] path) { + if (path.length < 3) { + return null; + } + GroupType group = schema; + for (int i = 0; i < path.length - 2; i++) { + group = group.getType(path[i]).asGroupType(); + } + LogicalTypeAnnotation logical = group.getLogicalTypeAnnotation(); + return logical instanceof VectorLogicalTypeAnnotation ? (VectorLogicalTypeAnnotation) logical : null; + } + public static void checkValidWriteSchema(GroupType schema) { + checkValidVectorSchema(schema); schema.accept(new TypeVisitor() { @Override public void visit(GroupType groupType) { diff --git a/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java b/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java index 9a4c0eeecd..e34a373746 100644 --- a/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java +++ b/parquet-column/src/test/java/org/apache/parquet/io/TestColumnIO.java @@ -31,17 +31,25 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Iterator; import java.util.List; +import java.util.stream.Stream; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.HeapByteBufferAllocator; import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.column.ColumnWriteStore; import org.apache.parquet.column.ColumnWriter; +import org.apache.parquet.column.Encoding; import org.apache.parquet.column.ParquetProperties; import org.apache.parquet.column.impl.ColumnWriteStoreV1; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.column.page.mem.MemPageStore; +import org.apache.parquet.column.statistics.FloatStatistics; +import org.apache.parquet.column.values.plain.PlainValuesWriter; +import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesWriter; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.GroupFactory; import org.apache.parquet.example.data.GroupWriter; @@ -60,6 +68,8 @@ import org.apache.parquet.schema.Type.Repetition; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -67,6 +77,50 @@ public class TestColumnIO { private static final Logger LOG = LoggerFactory.getLogger(TestColumnIO.class); + private static Stream invalidVectorLevels() { + return Stream.of( + Arguments.of(new int[] {0}, new int[] {3}, "invalid levels"), + Arguments.of(new int[] {3}, new int[] {2}, "invalid levels"), + Arguments.of(new int[] {2}, new int[] {2}, "invalid continuation"), + Arguments.of(new int[] {0, 2}, new int[] {2, 1}, "null within vector"), + Arguments.of(new int[] {0, 2}, new int[] {0, 2}, "null within vector")); + } + + @ParameterizedTest + @MethodSource("invalidVectorLevels") + public void testInvalidVectorLevels(int[] repetition, int[] definition, String reason) throws IOException { + MessageType schema = MessageTypeParser.parseMessageType("message m { repeated group outer {" + + " required group v (VECTOR(2)) { repeated group list { required float element; } } } }"); + ColumnDescriptor descriptor = schema.getColumnDescription(new String[] {"outer", "v", "list", "element"}); + HeapByteBufferAllocator allocator = new HeapByteBufferAllocator(); + try (MemPageStore pages = new MemPageStore(1); + RunLengthBitPackingHybridValuesWriter rl = + new RunLengthBitPackingHybridValuesWriter(2, 64, 64, allocator); + RunLengthBitPackingHybridValuesWriter dl = + new RunLengthBitPackingHybridValuesWriter(2, 64, 64, allocator); + PlainValuesWriter values = new PlainValuesWriter(64, 64, allocator)) { + for (int i = 0; i < repetition.length; i++) { + rl.writeInteger(repetition[i]); + dl.writeInteger(definition[i]); + if (definition[i] >= 2) values.writeFloat(1.0f); + } + pages.getPageWriter(descriptor) + .writePage( + BytesInput.concat(rl.getBytes(), dl.getBytes(), values.getBytes()), + repetition.length, + new FloatStatistics(), + Encoding.RLE, + Encoding.RLE, + Encoding.PLAIN); + MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema); + RecordReaderImplementation reader = getRecordReader(columnIO, schema, pages); + assertThatThrownBy(reader::read) + .isInstanceOf(ParquetDecodingException.class) + .hasMessageContaining("[outer, v, list, element]") + .hasMessageContaining(reason); + } + } + private static final String oneOfEach = "message Document {\n" + " required int64 a;\n" + " required int32 b;\n" diff --git a/parquet-column/src/test/java/org/apache/parquet/io/ValidatingRecordConsumerTest.java b/parquet-column/src/test/java/org/apache/parquet/io/ValidatingRecordConsumerTest.java index 118fcbbcde..729e4104b8 100644 --- a/parquet-column/src/test/java/org/apache/parquet/io/ValidatingRecordConsumerTest.java +++ b/parquet-column/src/test/java/org/apache/parquet/io/ValidatingRecordConsumerTest.java @@ -21,16 +21,45 @@ import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; import static org.apache.parquet.schema.Type.Repetition.REQUIRED; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; import java.lang.reflect.Field; import java.util.Deque; import org.apache.parquet.io.api.Binary; import org.apache.parquet.io.api.RecordConsumer; import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; import org.apache.parquet.schema.PrimitiveType; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; public class ValidatingRecordConsumerTest { + @ParameterizedTest + @ValueSource(strings = {"v", "element"}) + public void testEmptyRequiredVectorField(String field) { + MessageType schema = MessageTypeParser.parseMessageType( + "message m { required group v (VECTOR(2)) {" + " repeated group list { required float element; } } }"); + RecordConsumer delegate = mock(RecordConsumer.class); + ValidatingRecordConsumer c = new ValidatingRecordConsumer(delegate, schema); + c.startMessage(); + c.startField("v", 0); + if (field.equals("element")) { + c.startGroup(); + c.startField("list", 0); + c.startGroup(); + c.startField("element", 0); + } + clearInvocations(delegate); + assertThatThrownBy(() -> c.endField(field, 0)) + .isInstanceOf(InvalidRecordException.class) + .hasMessage("required field is missing " + + (field.equals("v") ? schema.getType("v") : schema.getType("v", "list", "element"))); + verifyNoInteractions(delegate); + } private static final RecordConsumer NOOP = new RecordConsumer() { @Override diff --git a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java index fbc79cc81e..8fa7657448 100644 --- a/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java +++ b/parquet-column/src/test/java/org/apache/parquet/schema/TestTypeBuildersWithLogicalTypes.java @@ -31,6 +31,7 @@ import static org.apache.parquet.schema.LogicalTypeAnnotation.timeType; import static org.apache.parquet.schema.LogicalTypeAnnotation.timestampType; import static org.apache.parquet.schema.LogicalTypeAnnotation.uuidType; +import static org.apache.parquet.schema.LogicalTypeAnnotation.vectorType; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BOOLEAN; import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.DOUBLE; @@ -43,11 +44,91 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.stream.Stream; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type.Repetition; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; public class TestTypeBuildersWithLogicalTypes { + private static GroupType vector(Type element) { + return new GroupType(REQUIRED, "v", vectorType(5), new GroupType(Repetition.REPEATED, "list", element)); + } + + @ParameterizedTest + @ValueSource(ints = {1, Integer.MAX_VALUE}) + public void testVectorAnnotation(int length) { + GroupType vector = new GroupType( + REQUIRED, + "v", + vectorType(length), + new GroupType(Repetition.REPEATED, "list", Types.required(FLOAT).named("element"))); + MessageType schema = Types.buildMessage().addField(vector).named("m"); + TypeUtil.checkValidWriteSchema(schema); + assertThat(MessageTypeParser.parseMessageType(schema.toString())).isEqualTo(schema); + assertThat(vectorType(length).getNumElements()).isEqualTo(length); + assertThat(vectorType(length).toOriginalType()).isEqualTo(OriginalType.LIST); + assertThat(vectorType(length)).isEqualTo(vectorType(length)).hasSameHashCodeAs(vectorType(length)); + assertThat(vectorType(length)).isNotEqualTo(vectorType(length == 1 ? 5 : 1)); + assertThat(vectorType(length)).isNotEqualTo(LogicalTypeAnnotation.listType()); + } + + @ParameterizedTest + @ValueSource(ints = {0, -1}) + public void testInvalidVectorLength(int length) { + assertThatThrownBy(() -> vectorType(length)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid VECTOR length: " + length); + } + + private static Stream invalidVectors() { + PrimitiveType element = Types.required(FLOAT).named("element"); + GroupType list = new GroupType(Repetition.REPEATED, "list", element); + return Stream.of( + Arguments.of(new GroupType(Repetition.REPEATED, "v", vectorType(5), list)), + Arguments.of(new GroupType( + REQUIRED, "v", vectorType(5), new GroupType(Repetition.REPEATED, "items", element))), + Arguments.of(new GroupType(REQUIRED, "v", vectorType(5), list, list)), + Arguments.of(new GroupType(REQUIRED, "v", vectorType(5), new GroupType(REQUIRED, "list", element))), + Arguments.of(new GroupType( + REQUIRED, + "v", + vectorType(5), + new GroupType(Repetition.REPEATED, "list", LogicalTypeAnnotation.listType(), element))), + Arguments.of(new GroupType( + REQUIRED, "v", vectorType(5), new GroupType(Repetition.REPEATED, "list", element, element))), + Arguments.of(vector(Types.required(FLOAT).named("item"))), + Arguments.of(vector(Types.optional(FLOAT).named("element"))), + Arguments.of(vector(Types.repeated(FLOAT).named("element"))), + Arguments.of(vector(new GroupType(REQUIRED, "element", element))), + Arguments.of(vector(Types.required(BINARY).as(decimalType(2, 9)).named("element"))), + Arguments.of( + vector(Types.required(FIXED_LEN_BYTE_ARRAY).length(2).named("element"))), + Arguments.of(vector(Types.required(INT96).named("element"))), + Arguments.of(vector(Types.required(BINARY).as(stringType()).named("element"))), + Arguments.of(vector(Types.required(INT32).as(dateType()).named("element"))), + Arguments.of( + vector(Types.required(INT32).named("element").withLogicalTypeAnnotation(intType(64, true))))); + } + + @ParameterizedTest + @MethodSource("invalidVectors") + public void testInvalidVectorSchema(GroupType vector) { + assertThatThrownBy(() -> TypeUtil.checkValidVectorSchema(new MessageType("m", vector))) + .isInstanceOf(InvalidSchemaException.class) + .hasMessage("Invalid VECTOR schema: " + vector); + } + + @Test + public void testVectorOnPrimitive() { + assertThatThrownBy(() -> Types.required(FLOAT).as(vectorType(5)).named("v")) + .isInstanceOf(IllegalStateException.class) + .hasMessage("VECTOR(5) can not be applied to a primitive type"); + } + @Test public void testGroupTypeConstruction() { PrimitiveType f1 = Types.required(BINARY).as(stringType()).named("f1"); diff --git a/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java b/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java index 8956d3944e..4a534c8424 100644 --- a/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java +++ b/parquet-format-structures/src/main/java/org/apache/parquet/format/LogicalTypes.java @@ -38,6 +38,10 @@ public static LogicalType VARIANT(byte specificationVersion) { return LogicalType.VARIANT(type); } + public static LogicalType VECTOR(int numElements) { + return LogicalType.VECTOR(new VectorType(numElements)); + } + public static final LogicalType UTF8 = LogicalType.STRING(new StringType()); public static final LogicalType MAP = LogicalType.MAP(new MapType()); public static final LogicalType LIST = LogicalType.LIST(new ListType()); diff --git a/parquet-format-structures/src/main/thrift/parquet.thrift b/parquet-format-structures/src/main/thrift/parquet.thrift index bb582017bf..bcb32c4034 100644 --- a/parquet-format-structures/src/main/thrift/parquet.thrift +++ b/parquet-format-structures/src/main/thrift/parquet.thrift @@ -323,6 +323,10 @@ struct StringType {} // allowed for BYTE_ARRAY, must be encoded with UTF-8 struct UUIDType {} // allowed for FIXED[16], must be encoded as raw UUID bytes struct MapType {} // see LogicalTypes.md struct ListType {} // see LogicalTypes.md +// A fixed-length LIST with exactly num_elements required primitive elements. +struct VectorType { + 1: required i32 num_elements +} struct EnumType {} // allowed for BYTE_ARRAY, must be encoded with UTF-8 struct DateType {} // allowed for INT32 struct Float16Type {} // allowed for FIXED[2], must be encoded as raw FLOAT16 bytes (see LogicalTypes.md) @@ -513,6 +517,7 @@ union LogicalType { 17: GeometryType GEOMETRY // no compatible ConvertedType 18: GeographyType GEOGRAPHY // no compatible ConvertedType 19: FileType FILE // no compatible ConvertedType + 20: VectorType VECTOR // use ConvertedType LIST } /** diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java index bbbe1e78e4..2027e1eb2e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java @@ -131,13 +131,16 @@ import org.apache.parquet.io.api.Binary; import org.apache.parquet.schema.ColumnOrder.ColumnOrderName; import org.apache.parquet.schema.GroupType; +import org.apache.parquet.schema.InvalidSchemaException; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.LogicalTypeAnnotation.UUIDLogicalTypeAnnotation; +import org.apache.parquet.schema.LogicalTypeAnnotation.VectorLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.OriginalType; import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type.Repetition; +import org.apache.parquet.schema.TypeUtil; import org.apache.parquet.schema.TypeVisitor; import org.apache.parquet.schema.Types; import org.slf4j.Logger; @@ -419,6 +422,11 @@ public Optional visit(LogicalTypeAnnotation.ListLogicalTypeAnnota return of(ConvertedType.LIST); } + @Override + public Optional visit(VectorLogicalTypeAnnotation vectorLogicalType) { + return of(ConvertedType.LIST); + } + @Override public Optional visit(LogicalTypeAnnotation.EnumLogicalTypeAnnotation enumLogicalType) { return of(ConvertedType.ENUM); @@ -518,6 +526,11 @@ public Optional visit(LogicalTypeAnnotation.ListLogicalTypeAnnotati return of(LogicalTypes.LIST); } + @Override + public Optional visit(VectorLogicalTypeAnnotation vectorLogicalType) { + return of(LogicalTypes.VECTOR(vectorLogicalType.getNumElements())); + } + @Override public Optional visit(LogicalTypeAnnotation.EnumLogicalTypeAnnotation enumLogicalType) { return of(LogicalTypes.ENUM); @@ -1399,6 +1412,8 @@ LogicalTypeAnnotation getLogicalTypeAnnotation(LogicalType type) { return LogicalTypeAnnotation.jsonType(); case LIST: return LogicalTypeAnnotation.listType(); + case VECTOR: + return LogicalTypeAnnotation.vectorType(type.getVECTOR().getNum_elements()); case TIME: TimeType time = type.getTIME(); return LogicalTypeAnnotation.timeType(time.isAdjustedToUTC, convertTimeUnit(time.unit)); @@ -2086,13 +2101,22 @@ private static ColumnPath getPath(ColumnMetaData metaData) { MessageType fromParquetSchema(List schema, List columnOrders) { Iterator iterator = schema.iterator(); SchemaElement root = iterator.next(); + if (root.isSetLogicalType() && root.logicalType.isSetVECTOR()) { + throw new ParquetDecodingException("Invalid VECTOR metadata for " + root.name); + } Types.MessageTypeBuilder builder = Types.buildMessage(); if (root.isSetField_id()) { builder.id(root.field_id); } Iterator columnOrderIterator = columnOrders == null ? null : columnOrders.iterator(); buildChildren(builder, iterator, root.getNum_children(), columnOrderIterator); - return builder.named(root.name); + MessageType messageType = builder.named(root.name); + try { + TypeUtil.checkValidVectorSchema(messageType); + } catch (InvalidSchemaException e) { + throw new ParquetDecodingException("Invalid VECTOR schema in file metadata", e); + } + return messageType; } private void buildChildren( @@ -2102,6 +2126,14 @@ private void buildChildren( Iterator columnOrders) { for (int i = 0; i < childrenCount; i++) { SchemaElement schemaElement = schema.next(); + if (schemaElement.isSetLogicalType() + && schemaElement.logicalType.isSetVECTOR() + && (schemaElement.type != null + || schemaElement.converted_type != ConvertedType.LIST + || !schemaElement.logicalType.getVECTOR().isSetNum_elements() + || schemaElement.logicalType.getVECTOR().getNum_elements() <= 0)) { + throw new ParquetDecodingException("Invalid VECTOR metadata for " + schemaElement.name); + } // Create Parquet Type. Types.Builder childBuilder; diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index d29738a6d8..e7b651f28e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -118,8 +118,10 @@ import org.apache.parquet.io.ParquetDecodingException; import org.apache.parquet.io.ParquetFileRange; import org.apache.parquet.io.SeekableInputStream; +import org.apache.parquet.schema.LogicalTypeAnnotation.VectorLogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.TypeUtil; import org.apache.parquet.util.AutoCloseables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1159,6 +1161,45 @@ public PageReadStore readNextRowGroup() throws IOException { return currentRowGroup; } + private static void checkVectorCounts(ColumnDescriptor path, int numElements, long rows, long values, long nulls) { + long declaredNulls = nulls; + boolean invalid = rows < 0 || values < 0 || nulls < -1 || nulls > values; + if (!invalid && path.getMaxRepetitionLevel() == 1) { + if (path.getMaxDefinitionLevel() == 1) { + invalid = nulls > 0; + nulls = 0; + } + if (nulls >= 0) { + long present = values - nulls; + invalid |= nulls > rows || present % numElements != 0 || present / numElements != rows - nulls; + } else { + long extra = values - rows; + invalid |= values < rows + || (numElements == 1 + ? values != rows + : extra % (numElements - 1) != 0 || extra / (numElements - 1) > rows); + } + } else if (!invalid && nulls >= 0) { + invalid = (values - nulls) % numElements != 0; + } + if (invalid) { + throw new ParquetDecodingException("Invalid VECTOR counts for " + Arrays.toString(path.getPath()) + + ": rows=" + rows + ", values=" + values + ", nulls=" + declaredNulls + + ", length=" + numElements); + } + } + + private void checkVectorColumnCounts(ColumnChunkMetaData column, long rows) { + MessageType fileSchema = getFileMetaData().getSchema(); + String[] path = column.getPath().toArray(); + VectorLogicalTypeAnnotation vector = TypeUtil.getVectorType(fileSchema, path); + if (vector == null) return; + ColumnDescriptor descriptor = fileSchema.getColumnDescription(path); + org.apache.parquet.column.statistics.Statistics stats = column.getStatistics(); + long nulls = stats != null && stats.isNumNullsSet() ? stats.getNumNulls() : -1; + checkVectorCounts(descriptor, vector.getNumElements(), rows, column.getValueCount(), nulls); + } + private ColumnChunkPageReadStore internalReadRowGroup(int blockIndex) throws IOException { if (blockIndex < 0 || blockIndex >= blocks.size()) { return null; @@ -1176,6 +1217,7 @@ private ColumnChunkPageReadStore internalReadRowGroup(int blockIndex) throws IOE ColumnPath pathKey = mc.getPath(); ColumnDescriptor columnDescriptor = paths.get(pathKey); if (columnDescriptor != null) { + checkVectorColumnCounts(mc, block.getRowCount()); BenchmarkCounter.incrementTotalBytes(mc.getTotalSize()); long startingPos = mc.getStartingPos(); // first part or not consecutive => new list @@ -1574,6 +1616,7 @@ private ColumnChunkPageReadStore internalReadFilteredRowGroup( ColumnPath pathKey = mc.getPath(); ColumnDescriptor columnDescriptor = paths.get(pathKey); if (columnDescriptor != null) { + checkVectorColumnCounts(mc, block.getRowCount()); OffsetIndex offsetIndex = ciStore.getOffsetIndex(mc.getPath()); OffsetIndex filteredOffsetIndex = @@ -2136,6 +2179,10 @@ public ColumnChunkPageReader readAllPages( .getSchema() .getType(descriptor.col.getPath()) .asPrimitiveType(); + MessageType fileSchema = getFileMetaData().getSchema(); + String[] path = descriptor.col.getPath(); + VectorLogicalTypeAnnotation vector = TypeUtil.getVectorType(fileSchema, path); + ColumnDescriptor fileDescriptor = vector == null ? null : fileSchema.getColumnDescription(path); long valuesCountReadSoFar = 0L; int dataPageCountReadSoFar = 0; byte[] dataPageHeaderAAD = null; @@ -2210,12 +2257,35 @@ public ColumnChunkPageReader readAllPages( if (pageHeader.isSetCrc()) { dataPageV1.setCrc(pageHeader.getCrc()); } + if (vector != null && offsetIndex != null) { + long rows = offsetIndex.getLastRowIndex(dataPageCountReadSoFar, rowCount) + - offsetIndex.getFirstRowIndex(dataPageCountReadSoFar) + + 1; + org.apache.parquet.column.statistics.Statistics stats = dataPageV1.getStatistics(); + long nulls = stats != null && stats.isNumNullsSet() ? stats.getNumNulls() : -1; + checkVectorCounts( + fileDescriptor, vector.getNumElements(), rows, dataPageV1.getValueCount(), nulls); + } pagesInChunk.add(dataPageV1); valuesCountReadSoFar += dataHeaderV1.getNum_values(); ++dataPageCountReadSoFar; break; case DATA_PAGE_V2: DataPageHeaderV2 dataHeaderV2 = pageHeader.getData_page_header_v2(); + if (vector != null) { + if (dataHeaderV2.getNum_nulls() < 0) { + throw new ParquetDecodingException("Invalid VECTOR counts for " + Arrays.toString(path) + + ": rows=" + dataHeaderV2.getNum_rows() + ", values=" + + dataHeaderV2.getNum_values() + ", nulls=" + dataHeaderV2.getNum_nulls() + + ", length=" + vector.getNumElements()); + } + checkVectorCounts( + fileDescriptor, + vector.getNumElements(), + dataHeaderV2.getNum_rows(), + dataHeaderV2.getNum_values(), + dataHeaderV2.getNum_nulls()); + } int dataSize = compressedPageSize - dataHeaderV2.getRepetition_levels_byte_length() - dataHeaderV2.getDefinition_levels_byte_length(); diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java b/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java index a47166ae93..6977a89e59 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java @@ -68,6 +68,8 @@ import java.util.Random; import java.util.Set; import java.util.TreeSet; +import java.util.function.Consumer; +import java.util.stream.Stream; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.parquet.FixedBinaryTestUtils; @@ -133,6 +135,7 @@ import org.apache.parquet.internal.column.columnindex.ColumnIndexBuilder; import org.apache.parquet.internal.column.columnindex.OffsetIndex; import org.apache.parquet.internal.column.columnindex.OffsetIndexBuilder; +import org.apache.parquet.io.ParquetDecodingException; import org.apache.parquet.io.api.Binary; import org.apache.parquet.schema.ColumnOrder; import org.apache.parquet.schema.LogicalTypeAnnotation; @@ -144,8 +147,67 @@ import org.apache.parquet.schema.Types; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; public class TestParquetMetadataConverter { + private static MessageType vectorSchema() { + return parseMessageType("message m { required group v (VECTOR(5)) { repeated group list {" + + " required float element columnorder(TYPE_DEFINED_ORDER); } } }"); + } + + private static Stream invalidVectorMetadata() { + return Stream.of( + Arguments.of( + "missing length", + (Consumer>) + wire -> wire.get(1).getLogicalType().getVECTOR().unsetNum_elements(), + "Invalid VECTOR metadata for v"), + Arguments.of( + "zero length", + (Consumer>) + wire -> wire.get(1).getLogicalType().getVECTOR().setNum_elements(0), + "Invalid VECTOR metadata for v"), + Arguments.of( + "negative length", + (Consumer>) + wire -> wire.get(1).getLogicalType().getVECTOR().setNum_elements(-1), + "Invalid VECTOR metadata for v"), + Arguments.of( + "missing converted", + (Consumer>) wire -> wire.get(1).unsetConverted_type(), + "Invalid VECTOR metadata for v"), + Arguments.of( + "wrong converted", + (Consumer>) wire -> wire.get(1).setConverted_type(ConvertedType.MAP), + "Invalid VECTOR metadata for v"), + Arguments.of( + "primitive outer", + (Consumer>) wire -> wire.get(1).setType(Type.FLOAT), + "Invalid VECTOR metadata for v"), + Arguments.of( + "root", + (Consumer>) + wire -> wire.get(0).setLogicalType(org.apache.parquet.format.LogicalTypes.VECTOR(5)), + "Invalid VECTOR metadata for m"), + Arguments.of( + "element name", + (Consumer>) wire -> wire.get(3).setName("item"), + "Invalid VECTOR schema in file metadata")); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("invalidVectorMetadata") + public void testInvalidVectorMetadata(String name, Consumer> edit, String message) { + ParquetMetadataConverter converter = new ParquetMetadataConverter(); + List wire = converter.toParquetSchema(vectorSchema()); + edit.accept(wire); + assertThatThrownBy(() -> converter.fromParquetSchema(wire, null)) + .isInstanceOf(ParquetDecodingException.class) + .hasMessageStartingWith(message); + } + private static SecureRandom random = new SecureRandom(); private static final String CHAR_LOWER = "abcdefghijklmnopqrstuvwxyz"; private static final String CHAR_UPPER = CHAR_LOWER.toUpperCase(); diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestVectorReadWrite.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestVectorReadWrite.java new file mode 100644 index 0000000000..b18549f854 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestVectorReadWrite.java @@ -0,0 +1,591 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.hadoop; + +import static org.apache.parquet.column.Encoding.PLAIN; +import static org.apache.parquet.column.Encoding.RLE; +import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; +import static org.apache.parquet.filter.ColumnPredicates.equalTo; +import static org.apache.parquet.filter2.predicate.FilterApi.and; +import static org.apache.parquet.filter2.predicate.FilterApi.eq; +import static org.apache.parquet.filter2.predicate.FilterApi.gtEq; +import static org.apache.parquet.filter2.predicate.FilterApi.intColumn; +import static org.apache.parquet.filter2.predicate.FilterApi.ltEq; +import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.CREATE; +import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE; +import static org.apache.parquet.hadoop.ParquetWriter.MAX_PADDING_SIZE_DEFAULT; +import static org.apache.parquet.hadoop.VectorInteropFileGenerator.readFooter; +import static org.apache.parquet.hadoop.VectorInteropFileGenerator.rewriteFooter; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.stream.Stream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.column.page.DataPage; +import org.apache.parquet.column.page.DataPageV1; +import org.apache.parquet.column.page.DataPageV2; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.column.page.PageReader; +import org.apache.parquet.column.statistics.FloatStatistics; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroup; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.example.data.simple.convert.GroupRecordConverter; +import org.apache.parquet.filter2.columnindex.RowRanges; +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.format.ConvertedType; +import org.apache.parquet.format.LogicalTypes; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.io.ColumnIOFactory; +import org.apache.parquet.io.InvalidRecordException; +import org.apache.parquet.io.LocalInputFile; +import org.apache.parquet.io.MessageColumnIO; +import org.apache.parquet.io.ParquetDecodingException; +import org.apache.parquet.io.RecordReader; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.MethodSource; + +public class TestVectorReadWrite { + @TempDir + private java.nio.file.Path tempDir; + + private static MessageType schema(String outer, int length, boolean nested) { + String vector = outer + " group v (VECTOR(" + length + ")) { repeated group list { required float element; } }"; + return MessageTypeParser.parseMessageType( + nested ? "message m { repeated group outer { " + vector + " } }" : "message m { " + vector + " }"); + } + + private Path writeVectors(MessageType schema, WriterVersion version, boolean dictionary, List rows) + throws IOException { + Path path = new Path(tempDir.resolve(UUID.randomUUID() + ".parquet").toUri()); + VectorInteropFileGenerator.writeRows(path, schema, version, dictionary, rows); + return path; + } + + private static Path referenceFile(String name) throws java.net.URISyntaxException { + return new Path(java.nio.file.Path.of( + TestVectorReadWrite.class.getResource("/vector/" + name).toURI()) + .toUri()); + } + + private static List listRows(MessageType schema, int... lengths) { + List rows = new ArrayList<>(); + int value = 0; + for (int length : lengths) { + Group row = new SimpleGroupFactory(schema).newGroup(); + if (length >= 0) { + Group parent = schema.containsField("outer") ? row.addGroup("outer") : row; + Group vector = parent.addGroup("v"); + for (int i = 0; i < length; i++) { + vector.addGroup("list").append("element", (float) value++); + } + } + rows.add(row); + } + return rows; + } + + private static List readGroups(Path path, MessageType requested) throws IOException { + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri())))) { + reader.setRequestedSchema(requested); + try (PageReadStore pages = reader.readRowGroup(0)) { + MessageType fileSchema = reader.getFileMetaData().getSchema(); + MessageColumnIO columns = new ColumnIOFactory().getColumnIO(requested, fileSchema); + RecordReader records = columns.getRecordReader(pages, new GroupRecordConverter(requested)); + List result = new ArrayList<>(); + for (long i = 0; i < pages.getRowCount(); i++) result.add(records.read()); + return result; + } + } + } + + private static void assertVectorRows(List actual, int... lengths) { + assertThat(actual).hasSize(lengths.length); + int value = 0; + for (int row = 0; row < lengths.length; row++) { + Group vector = actual.get(row).getGroup("v", 0); + assertThat(vector.getFieldRepetitionCount("list")).isEqualTo(lengths[row]); + for (int i = 0; i < lengths[row]; i++) { + assertThat(vector.getGroup("list", i).getFloat("element", 0)).isEqualTo((float) value++); + } + } + } + + @ParameterizedTest + @CsvSource({"PARQUET_1_0,false", "PARQUET_1_0,true", "PARQUET_2_0,false", "PARQUET_2_0,true"}) + public void testVectorRoundTrip(WriterVersion version, boolean dictionary) throws IOException { + Path path = new Path(tempDir.resolve(UUID.randomUUID() + ".parquet").toUri()); + VectorInteropFileGenerator.writeMixed(path, version, dictionary); + verifyMixedFile(path, version, dictionary); + if (!dictionary) verifyFilteredRows(path); + if (version == WriterVersion.PARQUET_1_0 && !dictionary) { + Path legacy = new Path(tempDir.resolve("legacy-list.parquet").toUri()); + Files.copy(java.nio.file.Path.of(path.toUri()), java.nio.file.Path.of(legacy.toUri())); + rewriteFooter(legacy, footer -> footer.getSchema().stream() + .filter(element -> element.isSetLogicalType() + && element.getLogicalType().isSetVECTOR()) + .forEach(org.apache.parquet.format.SchemaElement::unsetLogicalType)); + List vectors = readFooter(legacy).getSchema().stream() + .filter(element -> element.getConverted_type() == org.apache.parquet.format.ConvertedType.LIST) + .toList(); + assertThat(vectors).hasSize(10); + try (ParquetReader reader = + ParquetReader.builder(new GroupReadSupport(), legacy).build()) { + for (Group expected : VectorInteropFileGenerator.mixedRows()) assertRow(reader.read(), expected); + assertThat(reader.read()).isNull(); + } + } + } + + private static void assertRow(Group actual, Group expected) { + assertThat(actual).isNotNull(); + assertThat(actual.getType().getFieldCount()) + .isEqualTo(expected.getType().getFieldCount()); + for (int field = 0; field < expected.getType().getFieldCount(); field++) { + assertThat(actual.getType().getFieldName(field)) + .isEqualTo(expected.getType().getFieldName(field)); + assertThat(actual.getFieldRepetitionCount(field)).isEqualTo(expected.getFieldRepetitionCount(field)); + for (int index = 0; index < expected.getFieldRepetitionCount(field); index++) { + if (!expected.getType().getType(field).isPrimitive()) { + assertRow(actual.getGroup(field, index), expected.getGroup(field, index)); + } else { + assertThat(((SimpleGroup) actual).getObject(field, index)) + .isEqualTo(((SimpleGroup) expected).getObject(field, index)); + } + } + } + } + + private static void verifyMixedFile(Path path, WriterVersion version, boolean dictionary) throws IOException { + MessageType schema = VectorInteropFileGenerator.mixedSchema(); + try (ParquetReader reader = + ParquetReader.builder(new GroupReadSupport(), path).build()) { + for (Group expected : VectorInteropFileGenerator.mixedRows()) assertRow(reader.read(), expected); + assertThat(reader.read()).isNull(); + } + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri())))) { + assertThat(reader.getFooter().getFileMetaData().getSchema()).isEqualTo(schema); + assertThat(reader.getFooter().getBlocks()).hasSize(1); + assertThat(reader.getFooter().getBlocks().get(0).getRowCount()).isEqualTo(6); + try (PageReadStore pages = reader.readRowGroup(0)) { + PageReader f = pages.getPageReader(schema.getColumns().get(1)); + if (dictionary) { + assertThat(f.readDictionaryPage()).isNotNull(); + } else { + assertThat(f.readDictionaryPage()).isNull(); + } + DataPage page = f.readPage(); + assertThat(page) + .isInstanceOf(version == WriterVersion.PARQUET_1_0 ? DataPageV1.class : DataPageV2.class); + } + } + List vectors = readFooter(path).getSchema().stream() + .filter(element -> + element.isSetLogicalType() && element.getLogicalType().isSetVECTOR()) + .toList(); + assertThat(vectors) + .hasSize(10) + .extracting(element -> element.getConverted_type()) + .containsOnly(ConvertedType.LIST); + assertThat((int) vectors.get(0).getLogicalType().getSetField().getThriftFieldId()) + .isEqualTo(20); + } + + private static void verifyFilteredRows(Path path) throws IOException { + MessageType schema = VectorInteropFileGenerator.mixedSchema(); + List expected = VectorInteropFileGenerator.mixedRows(); + var predicate = and(gtEq(intColumn("id"), 2), ltEq(intColumn("id"), 3)); + try (ParquetReader reader = ParquetReader.builder(new GroupReadSupport(), path) + .withFilter(FilterCompat.get(predicate)) + .useColumnIndexFilter(true) + .build()) { + assertRow(reader.read(), expected.get(2)); + assertRow(reader.read(), expected.get(3)); + assertThat(reader.read()).isNull(); + } + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri()))); + PageReadStore pages = reader.readFilteredRowGroup( + 0, RowRanges.builder().addSelectedRange(1, 4).build())) { + assertThat(pages.getRowIndexes()).isPresent(); + MessageColumnIO columns = new ColumnIOFactory().getColumnIO(schema, schema); + RecordReader records = columns.getRecordReader(pages, new GroupRecordConverter(schema)); + for (int i = 1; i <= 4; i++) assertRow(records.read(), expected.get(i)); + } + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri()))); + PageReadStore pages = reader.readRowGroup(0)) { + MessageColumnIO columns = new ColumnIOFactory().getColumnIO(schema, schema); + RecordReader records = columns.getRecordReader( + pages, + new GroupRecordConverter(schema), + FilterCompat.get(org.apache.parquet.filter.ColumnRecordFilter.column("id", equalTo(3)))); + assertRow(records.read(), expected.get(3)); + assertThat(records.read()).isNull(); + } + } + + private static Stream invalidVectorWrites() { + return Stream.of( + Arguments.of("float element", 2, List.of(), "Invalid VECTOR length for v: expected 2, got 0"), + Arguments.of("float element", 2, List.of(1.0f), "Invalid VECTOR length for v: expected 2, got 1"), + Arguments.of( + "float element", + 2, + List.of(1.0f, 2.0f, 3.0f), + "Invalid VECTOR length for v: expected 2, got 3"), + Arguments.of("float element", 1, List.of(Float.NaN), "Non-finite VECTOR element in v"), + Arguments.of("double element", 1, List.of(Double.POSITIVE_INFINITY), "Non-finite VECTOR element in v"), + Arguments.of( + "fixed_len_byte_array(2) element (FLOAT16)", + 1, + List.of(Binary.fromConstantByteArray(new byte[] {0, (byte) 0xfc})), + "Non-finite VECTOR element in v"), + Arguments.of( + "fixed_len_byte_array(2) element (FLOAT16)", + 1, + List.of(Binary.fromConstantByteArray(new byte[] {0})), + "Invalid FLOAT16 VECTOR element length: 1")); + } + + private void assertInvalidWrite(MessageType schema, Group row, String message) throws IOException { + Path path = new Path(tempDir.resolve(UUID.randomUUID() + ".parquet").toUri()); + try (ParquetWriter writer = ExampleParquetWriter.builder(path) + .withType(schema) + .withValidation(false) + .build()) { + assertThatThrownBy(() -> writer.write(row)) + .isInstanceOf(InvalidRecordException.class) + .hasMessage(message); + } + } + + @ParameterizedTest + @MethodSource("invalidVectorWrites") + public void testInvalidVectorWrite(String element, int length, List values, String message) throws IOException { + MessageType schema = MessageTypeParser.parseMessageType("message m { required group v (VECTOR(" + length + + ")) { repeated group list { required " + element + "; } } }"); + Group row = new SimpleGroupFactory(schema).newGroup(); + Group vector = row.addGroup("v"); + for (Object value : values) { + Group entry = vector.addGroup("list"); + if (value instanceof Float) { + entry.append("element", (Float) value); + } else if (value instanceof Double) { + entry.append("element", (Double) value); + } else { + entry.append("element", (Binary) value); + } + } + assertInvalidWrite(schema, row, message); + } + + @org.junit.jupiter.api.Test + public void testMandatoryVectorValidation() throws IOException { + MessageType missingId = MessageTypeParser.parseMessageType("message m { required int32 id;" + + " required group v (VECTOR(2)) { repeated group list { required float element; } } }"); + Group missingRow = new SimpleGroupFactory(missingId).newGroup(); + Group v = missingRow.addGroup("v"); + v.addGroup("list").append("element", 1.0f); + v.addGroup("list").append("element", 2.0f); + assertInvalidWrite(missingId, missingRow, "required field is missing required int32 id"); + + MessageType duplicate = schema("required", 2, false); + Group duplicateRow = new SimpleGroupFactory(duplicate).newGroup(); + for (int i = 0; i < 2; i++) { + Group entry = duplicateRow.addGroup("v"); + entry.addGroup("list").append("element", i == 0 ? 1.0f : 3.0f); + entry.addGroup("list").append("element", i == 0 ? 2.0f : 4.0f); + } + assertInvalidWrite( + duplicate, duplicateRow, "repeated value when the type is not repeated in " + duplicate.getType("v")); + } + + @org.junit.jupiter.api.Test + public void testVectorReadProjection() throws Exception { + MessageType list = MessageTypeParser.parseMessageType( + schema("required", 5, false).toString().replace("VECTOR(5)", "LIST")); + MessageType vector = schema("required", 5, false); + Path path = writeVectors(list, WriterVersion.PARQUET_1_0, false, listRows(list, 4, 6)); + assertVectorRows(readGroups(path, vector), 4, 6); + + rewriteFooter(path, footer -> footer.getSchema().get(1).setLogicalType(LogicalTypes.VECTOR(5))); + for (MessageType requested : List.of(vector, list)) { + assertThatThrownBy(() -> readGroups(path, requested)) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("Invalid VECTOR data for [v, list, element]: expected 5 elements, got 4"); + } + + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri()))); + PageReadStore pages = reader.readFilteredRowGroup(0, RowRanges.createSingle(1))) { + MessageColumnIO columns = new ColumnIOFactory() + .getColumnIO(vector, reader.getFileMetaData().getSchema()); + RecordReader records = columns.getRecordReader(pages, new GroupRecordConverter(vector)); + assertThatThrownBy(records::read) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("Invalid VECTOR data for [v, list, element]: expected 5 elements, got 4"); + } + } + + @ParameterizedTest + @CsvSource( + value = { + "short_v|all|expected 5 elements, got 4", + "long_v|all|expected 5 elements, got 6", + "empty_v|all|empty vector", + "nested|all|expected 5 elements, got 4", + "partial|all|expected 5 elements, got 3", + "nf_f|all|non-finite element", + "nf_d|all|non-finite element", + "nf_h|all|non-finite element", + "short_v|filter2|expected 5 elements, got 4", + "short_v|legacy|expected 5 elements, got 4" + }, + delimiter = '|') + public void testInvalidVectorReferenceFile(String field, String mode, String reason) throws Exception { + Path path = referenceFile("vector_invalid_v1_dictionary.parquet"); + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri())))) { + MessageType file = reader.getFileMetaData().getSchema(); + MessageType projected = new MessageType("m", file.getType("id"), file.getType(field)); + reader.setRequestedSchema(projected); + try (PageReadStore pages = reader.readRowGroup(0)) { + MessageColumnIO columns = new ColumnIOFactory().getColumnIO(projected, file); + FilterCompat.Filter filter = mode.equals("filter2") + ? FilterCompat.get(eq(intColumn("id"), 1)) + : mode.equals("legacy") + ? FilterCompat.get( + org.apache.parquet.filter.ColumnRecordFilter.column("id", equalTo(1))) + : FilterCompat.NOOP; + RecordReader records = + columns.getRecordReader(pages, new GroupRecordConverter(projected), filter); + String name = field.equals("nested") || field.equals("partial") ? field + ", v" : field; + assertThatThrownBy(() -> { + for (long i = 0; i < pages.getRowCount(); i++) records.read(); + }) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("Invalid VECTOR data for [" + name + ", list, element]: " + reason); + } + } + } + + private static org.apache.parquet.format.ColumnMetaData column(org.apache.parquet.format.FileMetaData footer) { + return footer.getRow_groups().get(0).getColumns().get(0).getMeta_data(); + } + + @ParameterizedTest + @CsvSource({"vector_mixed_v1_plain.parquet,PARQUET_1_0,false", "vector_mixed_v2_dictionary.parquet,PARQUET_2_0,true" + }) + public void testVectorReferenceFile(String name, WriterVersion version, boolean dictionary) throws Exception { + verifyMixedFile(referenceFile(name), version, dictionary); + } + + @org.junit.jupiter.api.Test + public void testSplitVectorFile() throws Exception { + Path path = referenceFile("vector_split_v1.parquet"); + MessageType schema = schema("required", 5, false); + assertVectorRows(readGroups(path, schema), 5, 5); + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri()))); + PageReadStore pages = reader.readRowGroup(0)) { + PageReader column = pages.getPageReader(schema.getColumns().get(0)); + for (int count : new int[] {3, 4, 3}) { + DataPageV1 page = (DataPageV1) column.readPage(); + assertThat(page.getValueCount()).isEqualTo(count); + } + assertThat(column.readPage()).isNull(); + } + } + + @ParameterizedTest + @CsvSource({ + "required,5,false,4,18,0,false,false", + "required,5,false,4,15,0,true,true", + "optional,5,false,3,11,0,false,true", + "optional,5,false,3,10,-1,false,false", + "required,1,false,3,2,0,false,false", + "required,5,false,9223372036854775807,15,0,false,false", + "required,2,true,3,5,0,false,false" + }) + public void testInvalidVectorFooter( + String outer, + int length, + boolean nested, + long rowCount, + long values, + long nulls, + boolean projectList, + boolean filtered) + throws IOException { + MessageType fileSchema = schema(outer, length, nested); + int count = rowCount == Long.MAX_VALUE ? 3 : (int) rowCount; + int[] lengths = new int[count]; + Arrays.fill(lengths, length); + if (outer.equals("optional")) lengths[0] = -1; + Path path = writeVectors(fileSchema, PARQUET_2_0, false, listRows(fileSchema, lengths)); + rewriteFooter(path, footer -> { + footer.setNum_rows(rowCount); + footer.getRow_groups().get(0).setNum_rows(rowCount); + column(footer).setNum_values(values); + if (nulls == -1) { + column(footer).getStatistics().unsetNull_count(); + } else { + column(footer).getStatistics().setNull_count(nulls); + } + }); + String expected = (nested ? "[outer, v, list, element]" : "[v, list, element]") + ": rows=" + rowCount + + ", values=" + values + ", nulls=" + nulls + ", length=" + length; + for (boolean select : filtered ? new boolean[] {false, true} : new boolean[] {false}) { + assertThatThrownBy(() -> { + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri())))) { + if (projectList) { + reader.setRequestedSchema(MessageTypeParser.parseMessageType( + "message m { repeated group v (LIST) { repeated group list {" + + " required float element; } } }")); + } + assertThat(reader.readOffsetIndex(reader.getFooter() + .getBlocks() + .get(0) + .getColumns() + .get(0))) + .isNotNull(); + if (select) { + reader.readFilteredRowGroup(0, RowRanges.createSingle(1)); + } else { + reader.readRowGroup(0); + } + } + }) + .isInstanceOf(ParquetDecodingException.class) + .hasMessageContaining("Invalid VECTOR counts for " + expected); + } + } + + @ParameterizedTest + @CsvSource({"false,3,10,0", "false,2,10,1", "false,-1,10,0", "false,2,10,-1", "true,2,9,0"}) + public void testInvalidVectorPageHeader(boolean nested, int headerRows, int headerValues, int headerNulls) + throws IOException { + MessageType fileSchema = schema("required", nested ? 2 : 5, nested); + Path path = new Path(tempDir.resolve(UUID.randomUUID() + ".parquet").toUri()); + try (ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, new Configuration()), + fileSchema, + CREATE, + DEFAULT_BLOCK_SIZE, + MAX_PADDING_SIZE_DEFAULT, + null, + ParquetProperties.builder().build())) { + FloatStatistics stats = new FloatStatistics(); + stats.updateStats(1.0f); + stats.setNumNulls(0); + writer.start(); + writer.startBlock(2); + writer.startColumn(fileSchema.getColumns().get(0), 10, CompressionCodecName.UNCOMPRESSED); + writer.writeDataPageV2( + headerRows, + headerNulls, + headerValues, + BytesInput.empty(), + BytesInput.empty(), + PLAIN, + BytesInput.empty(), + false, + 0, + stats); + writer.endColumn(); + writer.endBlock(); + writer.end(Collections.emptyMap()); + } + assertThatThrownBy(() -> { + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri())))) { + reader.readRowGroup(0); + } + }) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("Invalid VECTOR counts for " + + (nested ? "[outer, v, list, element]" : "[v, list, element]") + + ": rows=" + headerRows + ", values=" + headerValues + ", nulls=" + headerNulls + + ", length=" + (nested ? 2 : 5)); + } + + @org.junit.jupiter.api.Test + public void testInvalidIndexedVectorPage() throws IOException { + MessageType fileSchema = schema("required", 5, false); + Path path = new Path(tempDir.resolve("invalid-indexed.parquet").toUri()); + FloatStatistics stats = new FloatStatistics(); + stats.updateStats(1.0f); + stats.setNumNulls(0); + try (ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, new Configuration()), + fileSchema, + CREATE, + DEFAULT_BLOCK_SIZE, + MAX_PADDING_SIZE_DEFAULT, + null, + ParquetProperties.builder().build())) { + writer.start(); + writer.startBlock(4); + writer.startColumn(fileSchema.getColumns().get(0), 20, CompressionCodecName.UNCOMPRESSED); + writer.writeDataPage(9, 0, BytesInput.empty(), stats, 2, RLE, RLE, PLAIN); + writer.writeDataPage(11, 0, BytesInput.empty(), stats, 2, RLE, RLE, PLAIN); + writer.endColumn(); + writer.endBlock(); + writer.end(Collections.emptyMap()); + } + assertThatThrownBy(() -> { + try (ParquetFileReader reader = + ParquetFileReader.open(new LocalInputFile(java.nio.file.Path.of(path.toUri())))) { + assertThat(reader.readOffsetIndex(reader.getFooter() + .getBlocks() + .get(0) + .getColumns() + .get(0))) + .isNotNull(); + reader.readFilteredRowGroup(0, RowRanges.createSingle(1)); + } + }) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("Invalid VECTOR counts for [v, list, element]: rows=2, values=9, nulls=-1, length=5"); + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/VectorInteropFileGenerator.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/VectorInteropFileGenerator.java new file mode 100644 index 0000000000..f5217bb32e --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/VectorInteropFileGenerator.java @@ -0,0 +1,314 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.parquet.hadoop; + +import static org.apache.parquet.column.Encoding.BIT_PACKED; +import static org.apache.parquet.column.Encoding.PLAIN; +import static org.apache.parquet.column.Encoding.RLE; +import static org.apache.parquet.hadoop.ParquetFileWriter.Mode.CREATE; +import static org.apache.parquet.hadoop.ParquetWriter.DEFAULT_BLOCK_SIZE; +import static org.apache.parquet.hadoop.ParquetWriter.MAX_PADDING_SIZE_DEFAULT; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.function.Consumer; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.bytes.HeapByteBufferAllocator; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.column.values.ValuesWriter; +import org.apache.parquet.column.values.bitpacking.ByteBitPackingValuesWriter; +import org.apache.parquet.column.values.bitpacking.Packer; +import org.apache.parquet.column.values.plain.PlainValuesWriter; +import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesWriter; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.format.ConvertedType; +import org.apache.parquet.format.LogicalTypes; +import org.apache.parquet.format.Util; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.io.api.Binary; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; + +public final class VectorInteropFileGenerator { + private VectorInteropFileGenerator() {} + + static MessageType mixedSchema() { + return MessageTypeParser.parseMessageType("message m {" + + " required int32 id;" + + " required group f (VECTOR(5)) { repeated group list { required float element; } }" + + " optional group o (VECTOR(2)) { repeated group list { required int32 element; } }" + + " optional group a { required group h (VECTOR(2)) { repeated group list {" + + " required fixed_len_byte_array(2) element (FLOAT16); } } }" + + " repeated group outer { required group d (VECTOR(2)) { repeated group list {" + + " required double element; } } }" + + " required group b (VECTOR(2)) { repeated group list { required boolean element; } }" + + " required group l (VECTOR(1)) { repeated group list { required int64 element; } }" + + " required group s (VECTOR(1)) { repeated group list {" + + " required int32 element (INTEGER(16,true)); } }" + + " required group u (VECTOR(1)) { repeated group list {" + + " required int64 element (INTEGER(64,false)); } }" + + " required group dec (VECTOR(1)) { repeated group list {" + + " required int32 element (DECIMAL(9,2)); } }" + + " required group bytes (VECTOR(1)) { repeated group list {" + + " required fixed_len_byte_array(4) element (DECIMAL(9,2)); } } }"); + } + + static List mixedRows() { + List rows = new ArrayList<>(); + SimpleGroupFactory factory = new SimpleGroupFactory(mixedSchema()); + for (int i = 0; i < 6; i++) { + Group row = factory.newGroup().append("id", i); + Group f = row.addGroup("f"); + for (float value : new float[] {i + 0.5f, -0.0f, Float.MIN_VALUE, -0.25f, 100.0f}) { + f.addGroup("list").append("element", value); + } + if (i % 2 != 0) { + row.addGroup("o").addGroup("list").append("element", 7); + row.getGroup("o", 0).addGroup("list").append("element", 8); + } + if (i != 1 && i != 4) { + Group h = row.addGroup("a").addGroup("h"); + h.addGroup("list").append("element", Binary.fromConstantByteArray(new byte[] {0, (byte) 0x80})); + h.addGroup("list").append("element", Binary.fromConstantByteArray(new byte[] {1, 0})); + } + for (int j = 0; j < i % 3; j++) { + Group d = row.addGroup("outer").addGroup("d"); + d.addGroup("list").append("element", -0.0d); + d.addGroup("list").append("element", Double.MIN_VALUE); + } + Group b = row.addGroup("b"); + b.addGroup("list").append("element", true); + b.addGroup("list").append("element", false); + row.addGroup("l").addGroup("list").append("element", 9L); + row.addGroup("s").addGroup("list").append("element", -7); + row.addGroup("u").addGroup("list").append("element", 9L); + row.addGroup("dec").addGroup("list").append("element", 125); + row.addGroup("bytes") + .addGroup("list") + .append("element", Binary.fromConstantByteArray(new byte[] {0, 0, 0x7c, 0})); + rows.add(row); + } + return rows; + } + + static void writeMixed(Path path, WriterVersion version, boolean dictionary) throws IOException { + writeRows(path, mixedSchema(), version, dictionary, mixedRows()); + } + + static void writeRows(Path path, MessageType schema, WriterVersion version, boolean dictionary, List rows) + throws IOException { + try (ParquetWriter writer = ExampleParquetWriter.builder(path) + .withType(schema) + .withWriterVersion(version) + .withDictionaryEncoding(dictionary) + .withValidation(false) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withPageRowCountLimit(2) + .withMinRowCountForPageSizeCheck(1) + .withMaxRowCountForPageSizeCheck(1) + .build()) { + for (Group row : rows) writer.write(row); + } + } + + static MessageType invalidSchema() { + return MessageTypeParser.parseMessageType("message m { required int32 id;" + + " required group short_v (LIST) { repeated group list { required float element; } }" + + " required group long_v (LIST) { repeated group list { required float element; } }" + + " optional group empty_v (LIST) { repeated group list { required float element; } }" + + " repeated group nested { required group v (LIST) { repeated group list {" + + " required float element; } } }" + + " repeated group partial { required group v (LIST) { repeated group list {" + + " required float element; } } }" + + " required group nf_f (LIST) { repeated group list { required float element; } }" + + " required group nf_d (LIST) { repeated group list { required double element; } }" + + " required group nf_h (LIST) { repeated group list {" + + " required fixed_len_byte_array(2) element (FLOAT16); } } }"); + } + + static List invalidRows() { + MessageType schema = invalidSchema(); + SimpleGroupFactory factory = new SimpleGroupFactory(schema); + List rows = new ArrayList<>(); + int[] shortLengths = {4, 6, 5, 5}; + int[] longLengths = {6, 4, 5, 5}; + int[] emptyLengths = {0, 5, 5, 5}; + int[] partialLengths = {5, 5, 5, 3}; + byte[] halfNegativeInfinity = {0, (byte) 0xfc}; + byte[] halfZero = {0, 0}; + byte[] halfOne = {0, 0x3c}; + for (int i = 0; i < 4; i++) { + Group row = factory.newGroup().append("id", i); + for (String name : List.of("short_v", "long_v", "empty_v")) { + int length = name.equals("short_v") + ? shortLengths[i] + : name.equals("long_v") ? longLengths[i] : emptyLengths[i]; + Group vector = row.addGroup(name); + for (int k = 0; k < length; k++) { + vector.addGroup("list").append("element", (float) (100 * i + k)); + } + } + int[] nestedLengths = i == 0 ? new int[] {4, 6} : i == 1 ? new int[0] : new int[] {5}; + for (int j = 0; j < nestedLengths.length; j++) { + Group vector = row.addGroup("nested").addGroup("v"); + for (int k = 0; k < nestedLengths[j]; k++) { + vector.addGroup("list").append("element", (float) (100 * i + 10 * j + k)); + } + } + Group partial = row.addGroup("partial").addGroup("v"); + for (int k = 0; k < partialLengths[i]; k++) { + partial.addGroup("list").append("element", (float) (100 * i + k)); + } + Group nfFloat = row.addGroup("nf_f"); + Group nfDouble = row.addGroup("nf_d"); + Group nfHalf = row.addGroup("nf_h"); + for (int k = 0; k < 8; k++) { + nfFloat.addGroup("list").append("element", i == 0 && k == 0 ? Float.NaN : k == 0 ? 0.0f : 1.0f); + nfDouble.addGroup("list") + .append("element", i == 0 && k == 0 ? Double.POSITIVE_INFINITY : k == 0 ? 0.0d : 1.0d); + nfHalf.addGroup("list") + .append( + "element", + Binary.fromConstantByteArray( + i == 0 && k == 0 ? halfNegativeInfinity : k == 0 ? halfZero : halfOne)); + } + rows.add(row); + } + return rows; + } + + static void writeInvalid(Path path) throws IOException { + writeRows(path, invalidSchema(), WriterVersion.PARQUET_1_0, true, invalidRows()); + rewriteFooter(path, footer -> { + footer.getSchema().stream() + .filter(element -> element.getConverted_type() == ConvertedType.LIST) + .forEach(element -> element.setLogicalType( + LogicalTypes.VECTOR(element.getName().startsWith("nf_") ? 8 : 5))); + for (org.apache.parquet.format.ColumnChunk chunk : + footer.getRow_groups().get(0).getColumns()) { + chunk.getMeta_data().getStatistics().unsetNull_count(); + chunk.unsetColumn_index_offset(); + chunk.unsetColumn_index_length(); + chunk.unsetOffset_index_offset(); + chunk.unsetOffset_index_length(); + } + }); + } + + static org.apache.parquet.format.FileMetaData readFooter(Path path) throws IOException { + java.nio.file.Path local = java.nio.file.Path.of(path.toUri()); + byte[] original = Files.readAllBytes(local); + int footerLength = BytesUtils.readIntLittleEndian(original, original.length - 8); + int footerStart = original.length - 8 - footerLength; + return Util.readFileMetaData(new ByteArrayInputStream(original, footerStart, footerLength)); + } + + static void rewriteFooter(Path path, Consumer edit) throws IOException { + java.nio.file.Path local = java.nio.file.Path.of(path.toUri()); + byte[] original = Files.readAllBytes(local); + int footerLength = BytesUtils.readIntLittleEndian(original, original.length - 8); + int footerStart = original.length - 8 - footerLength; + org.apache.parquet.format.FileMetaData footer = readFooter(path); + edit.accept(footer); + ByteArrayOutputStream encodedFooter = new ByteArrayOutputStream(); + Util.writeFileMetaData(footer, encodedFooter); + ByteArrayOutputStream rewritten = new ByteArrayOutputStream(); + rewritten.write(original, 0, footerStart); + rewritten.write(encodedFooter.toByteArray()); + BytesUtils.writeIntLittleEndian(rewritten, encodedFooter.size()); + rewritten.write(original, original.length - 4, 4); + Files.write(local, rewritten.toByteArray()); + } + + static void writeSplit(Path path) throws IOException { + MessageType schema = MessageTypeParser.parseMessageType( + "message m { required group v (VECTOR(5)) {" + " repeated group list { required float element; } } }"); + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, new Configuration()), + schema, + CREATE, + DEFAULT_BLOCK_SIZE, + MAX_PADDING_SIZE_DEFAULT, + null, + ParquetProperties.builder().build()); + writer.start(); + writer.startBlock(2); + writer.startColumn(schema.getColumns().get(0), 10, CompressionCodecName.UNCOMPRESSED); + int start = 0; + for (int count : new int[] {3, 4, 3}) { + boolean packed = start == 3; + ValuesWriter rl = packed + ? new ByteBitPackingValuesWriter(1, Packer.BIG_ENDIAN) + : new RunLengthBitPackingHybridValuesWriter(1, 64, 64, new HeapByteBufferAllocator()); + ValuesWriter dl = packed + ? new ByteBitPackingValuesWriter(1, Packer.BIG_ENDIAN) + : new RunLengthBitPackingHybridValuesWriter(1, 64, 64, new HeapByteBufferAllocator()); + ValuesWriter values = new PlainValuesWriter(64, 64, new HeapByteBufferAllocator()); + try { + for (int i = start; i < start + count; i++) { + rl.writeInteger(i == 0 || i == 5 ? 0 : 1); + dl.writeInteger(1); + values.writeFloat(i); + } + BytesInput bytes = BytesInput.concat( + BytesInput.from(rl.getBytes().toByteArray()), + BytesInput.from(dl.getBytes().toByteArray()), + BytesInput.from(values.getBytes().toByteArray())); + Encoding levelEncoding = packed ? BIT_PACKED : RLE; + writer.writeDataPage(count, (int) bytes.size(), bytes, levelEncoding, levelEncoding, PLAIN); + } finally { + rl.close(); + dl.close(); + values.close(); + } + start += count; + } + writer.endColumn(); + writer.endBlock(); + writer.end(Collections.emptyMap()); + writer.close(); + } + + public static void main(String[] args) throws IOException { + if (args.length != 1) { + throw new IllegalArgumentException("Usage: VectorInteropFileGenerator "); + } + java.nio.file.Path dir = java.nio.file.Path.of(args[0]); + Files.createDirectories(dir); + writeMixed(new Path(dir.resolve("vector_mixed_v1_plain.parquet").toUri()), WriterVersion.PARQUET_1_0, false); + writeMixed( + new Path(dir.resolve("vector_mixed_v2_dictionary.parquet").toUri()), WriterVersion.PARQUET_2_0, true); + writeSplit(new Path(dir.resolve("vector_split_v1.parquet").toUri())); + writeInvalid( + new Path(dir.resolve("vector_invalid_v1_dictionary.parquet").toUri())); + } +} diff --git a/parquet-hadoop/src/test/resources/vector/README.md b/parquet-hadoop/src/test/resources/vector/README.md new file mode 100644 index 0000000000..6c06cce3f9 --- /dev/null +++ b/parquet-hadoop/src/test/resources/vector/README.md @@ -0,0 +1,91 @@ + + +# VECTOR reference files + +These files use the draft [VECTOR proposal](https://github.com/apache/parquet-format/pull/624) +at revision `3573b3e1357f6546f3710d8cb48103e61ad40350`. They are +parquet-java-generated examples, not independent interoperability evidence. Runtime +mixed-file tests exercise the writer; these fixed bytes exercise reader compatibility. +The valid files are intended for a later parquet-testing/data export. Malformed 4/6, +content, and count cases and Java exception policy are excluded from portable conformance. + +The current `VectorInteropFileGenerator` source blob is +`f5217bb32e86abfac965a3eebab9709adbaa549d`. The three valid files came from +the Round 2 generator source `2d496dad2213216ca6e75f1ab0b4249822829421`. +The malformed file came from the current source. Both used Java 17, Maven 3.9.16, +and Thrift 0.25.0. + +## Mixed files + +For each row `i=0..5`, `id=i`, `f=[i+0.5f,-0.0f,Float.MIN_VALUE,-0.25f,100.0f]`, +`o` is absent for even `i` and `[7,8]` for odd `i`, and `a` is absent for `i=1,4`. +When present, `a.h=[0080,0100]` in hex. There are `i%3` `outer` instances, +each with `d=[-0.0d,Double.MIN_VALUE]`. Every row has `b=[true,false]`, +`l=[9L]`, `s=[-7]`, `u=[9L]`, `dec=[125]` (logical 1.25), and +`bytes=[00007c00]` in hex (logical 317.44). + +## Split file + +The two rows are `[0f,1f,2f,3f,4f]` and `[5f,6f,7f,8f,9f]`. Their stored RLs +are `[0,1,1,1,1,0,1,1,1,1]`, with DL=1 throughout. Three unindexed V1 pages +contain 3/4/3 values. The first and last pages use RLE/RLE levels; the middle +uses BIT_PACKED/BIT_PACKED levels. Values are PLAIN, compression is UNCOMPRESSED, +and column and offset indexes are absent. + +## Malformed file + +`vector_invalid_v1_dictionary.parquet` is deliberately invalid VECTOR data. Its +ordinary V1 writer used a canonical LIST schema with `id=0..3`, then changed only +the footer to annotate eight LIST groups as VECTOR. `short_v` has lengths +`[4,6,5,5]`, `long_v` has `[6,4,5,5]`, `empty_v` has `[0,5,5,5]`, nested `v` +has per-row instance lengths `[[4,6],[],[5],[5]]`, and partial `v` has +`[[5],[5],[5],[3]]`. These groups declare VECTOR(5). Their finite element `k` +in instance `j` of row `i` is `100*i+10*j+k` as FLOAT. + +`nf_f`, `nf_d`, and `nf_h` declare VECTOR(8). Each row contains seven ones; +element zero is zero except in row zero, where the fields contain FLOAT NaN, +DOUBLE positive infinity, and FLOAT16 negative infinity respectively. The FLOAT16 +values use little-endian bytes `00fc`, `0000`, and `003c` for negative infinity, +zero, and one. The file is UNCOMPRESSED with a page row limit of two. `nf_f` and +`nf_d` each have a dictionary and two V1 PLAIN_DICTIONARY pages of 16 values; +`nf_h` has no dictionary and two V1 PLAIN pages of 16 values. All column-chunk +null counts are unset, and column and offset index references are absent. +Each projected malformed VECTOR field must be rejected. + +## SHA-256 + +```text +c0e182b7f019d394a26bf863626cd58fed3916d930195ab0741fe15b41afb6e8 vector_mixed_v1_plain.parquet +939ce11a1cbe84c68d867deeab85546162d04cab8fd191d5a8efe203bfbb0765 vector_mixed_v2_dictionary.parquet +a598252a6f5715742bdb0e52c86be992f609189096e1cdc518701f4520f4c657 vector_split_v1.parquet +a87042f46c3d56bb9b236795c9c44f13bd02eecb4107040d6bb4b245a73dd7fd vector_invalid_v1_dictionary.parquet +``` + +## Reproduce + +From the repository root, use a new empty `target/vector-reference` directory. +An occupied directory fails by design; the generator does not overwrite files. +The pinned source and toolchain reproduce the documented rows and layout; future +writer changes need not produce byte-identical files. + +```bash +JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 /home/div.arora/.m2/wrapper/dists/apache-maven-3.9.16/c51baac8/bin/mvn -Dthrift.executable=/home/div.arora/parquet-java-worktrees/vector/target/vector-tools/bin/thrift -pl parquet-hadoop -am -DskipTests package dependency:build-classpath -DincludeScope=test -Dmdep.outputFile=target/vector-test-classpath.txt -Dmdep.regenerateFile=true +/usr/lib/jvm/java-17-openjdk-amd64/bin/java -cp "parquet-hadoop/target/test-classes:parquet-hadoop/target/classes:$(cat parquet-hadoop/target/vector-test-classpath.txt)" org.apache.parquet.hadoop.VectorInteropFileGenerator target/vector-reference +``` diff --git a/parquet-hadoop/src/test/resources/vector/vector_invalid_v1_dictionary.parquet b/parquet-hadoop/src/test/resources/vector/vector_invalid_v1_dictionary.parquet new file mode 100644 index 0000000000..cbac842314 Binary files /dev/null and b/parquet-hadoop/src/test/resources/vector/vector_invalid_v1_dictionary.parquet differ diff --git a/parquet-hadoop/src/test/resources/vector/vector_mixed_v1_plain.parquet b/parquet-hadoop/src/test/resources/vector/vector_mixed_v1_plain.parquet new file mode 100644 index 0000000000..fc2e8aa647 Binary files /dev/null and b/parquet-hadoop/src/test/resources/vector/vector_mixed_v1_plain.parquet differ diff --git a/parquet-hadoop/src/test/resources/vector/vector_mixed_v2_dictionary.parquet b/parquet-hadoop/src/test/resources/vector/vector_mixed_v2_dictionary.parquet new file mode 100644 index 0000000000..ad809ff478 Binary files /dev/null and b/parquet-hadoop/src/test/resources/vector/vector_mixed_v2_dictionary.parquet differ diff --git a/parquet-hadoop/src/test/resources/vector/vector_split_v1.parquet b/parquet-hadoop/src/test/resources/vector/vector_split_v1.parquet new file mode 100644 index 0000000000..5b00013501 Binary files /dev/null and b/parquet-hadoop/src/test/resources/vector/vector_split_v1.parquet differ diff --git a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java index 0da0dfca96..c5a01352c2 100644 --- a/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java +++ b/parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoMessageConverter.java @@ -67,6 +67,7 @@ import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.Type; +import org.apache.parquet.schema.TypeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -268,6 +269,17 @@ public Optional visit(LogicalTypeAnnotation.ListLogicalTypeAnnotation return of(new ListConverter(parentBuilder, fieldDescriptor, parquetType)); } + @Override + public Optional visit( + LogicalTypeAnnotation.VectorLogicalTypeAnnotation vectorLogicalType) { + TypeUtil.checkValidVectorSchema(parquetType); + if (!fieldDescriptor.isRepeated()) { + throw new IncompatibleSchemaModificationException("VECTOR field " + + fieldDescriptor.getName() + " requires a repeated protobuf field"); + } + return visit(LogicalTypeAnnotation.listType()); + } + @Override public Optional visit(LogicalTypeAnnotation.MapLogicalTypeAnnotation mapLogicalType) { return of(new MapConverter(parentBuilder, fieldDescriptor, parquetType)); @@ -885,6 +897,7 @@ public ListConverter( Message.Builder parentBuilder, Descriptors.FieldDescriptor fieldDescriptor, Type parquetType) { LogicalTypeAnnotation logicalTypeAnnotation = parquetType.getLogicalTypeAnnotation(); if (!(logicalTypeAnnotation instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation) + && !(logicalTypeAnnotation instanceof LogicalTypeAnnotation.VectorLogicalTypeAnnotation) || parquetType.isPrimitive()) { throw new ParquetDecodingException( "Expected LIST wrapper. Found: " + logicalTypeAnnotation + " instead."); diff --git a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoRecordConverterTest.java b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoRecordConverterTest.java index 6a3560dc61..b7eb09dd16 100644 --- a/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoRecordConverterTest.java +++ b/parquet-protobuf/src/test/java/org/apache/parquet/proto/ProtoRecordConverterTest.java @@ -21,15 +21,65 @@ import static org.apache.parquet.proto.TestUtils.testData; import static org.apache.parquet.proto.test.TestProtobuf.SchemaConverterAllDatatypes; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.data.Offset.offset; import com.google.protobuf.ByteString; +import java.io.IOException; +import java.util.Collections; import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.proto.test.TestProto3; import org.apache.parquet.proto.test.TestProtobuf; +import org.apache.parquet.schema.IncompatibleSchemaModificationException; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; public class ProtoRecordConverterTest { + @Test + public void testVectorRepeatedPrimitive(@TempDir java.nio.file.Path dir) throws IOException { + MessageType schema = MessageTypeParser.parseMessageType("message m { required group repeatedInt" + + " (VECTOR(3)) { repeated group list { required int32 element; } } }"); + Path path = new Path(dir.resolve("vector.parquet").toUri()); + try (ParquetWriter writer = ExampleParquetWriter.builder(path) + .withType(schema) + .withWriterVersion(WriterVersion.PARQUET_1_0) + .withDictionaryEncoding(false) + .withCompressionCodec(CompressionCodecName.UNCOMPRESSED) + .withValidation(false) + .build()) { + for (int row = 0; row < 2; row++) { + Group group = new SimpleGroupFactory(schema).newGroup(); + Group vector = group.addGroup("repeatedInt"); + for (int i = 1; i <= 3; i++) vector.addGroup("list").append("element", row * 3 + i); + writer.write(group); + } + } + List messages = + TestUtils.readMessages(path, TestProtobuf.RepeatedIntMessage.class); + assertThat(messages).hasSize(2); + assertThat(messages.get(0).getRepeatedIntList()).containsExactly(1, 2, 3); + assertThat(messages.get(1).getRepeatedIntList()).containsExactly(4, 5, 6); + } + + @Test + public void testVectorRejectsScalarTarget() { + MessageType schema = MessageTypeParser.parseMessageType("message m { required group optionalInt32" + + " (VECTOR(3)) { repeated group list { required int32 element; } } }"); + assertThatThrownBy(() -> new ProtoRecordConverter<>( + new Configuration(), SchemaConverterAllDatatypes.class, schema, Collections.emptyMap())) + .isInstanceOf(IncompatibleSchemaModificationException.class) + .hasMessageContaining("VECTOR field optionalInt32 requires a repeated protobuf field"); + } @Test public void testAllTypes() throws Exception {