GH-3877: Add VECTOR logical type support

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);
     }

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<TypeMapping> visit(
               return of(new ListTypeMapping(arrowField, list3Levels, child));
             }
 
+            @Override
+            public Optional<TypeMapping> visit(
+                LogicalTypeAnnotation.VectorLogicalTypeAnnotation vectorLogicalType) {
+              TypeUtil.checkValidVectorSchema(type);
+              return visit(LogicalTypeAnnotation.listType());
+            }
+
             @Override
             public Optional<TypeMapping> visit(
                 LogicalTypeAnnotation.MapLogicalTypeAnnotation mapLogicalType) {

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) {

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;
       }

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<Schema> 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<Schema> visit(
                 }
               }
 
+              @Override
+              public Optional<Schema> visit(
+                  LogicalTypeAnnotation.VectorLogicalTypeAnnotation vectorLogicalType) {
+                TypeUtil.checkValidVectorSchema(parquetGroupType);
+                return visit(LogicalTypeAnnotation.listType());
+              }
+
               @Override
               // for backward-compatibility
               public Optional<Schema> visit(

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<Group> 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<GenericRecord> 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);

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();

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;

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() {

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<Case> definedCases;
     private List<Case> 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--) {