GH-3877: Add FINITE_VECTOR logical type support

parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/List3Levels.java

@@ -42,7 +42,8 @@ class List3Levels {
    * @param list the Parquet List
    */
   public List3Levels(GroupType list) {
-    if (list.getLogicalTypeAnnotation() != LogicalTypeAnnotation.listType()
+    LogicalTypeAnnotation logicalType = list.getLogicalTypeAnnotation();
+    if (!(logicalType instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation)
         || list.getFields().size() != 1) {
       throw new IllegalArgumentException("invalid list type: " + list);
     }

parquet-arrow/src/main/java/org/apache/parquet/arrow/schema/SchemaConverter.java

@@ -486,6 +486,21 @@ public Optional<TypeMapping> visit(
               return of(new ListTypeMapping(arrowField, list3Levels, child));
             }
 
+            @Override
+            public Optional<TypeMapping> visit(
+                LogicalTypeAnnotation.FiniteVectorLogicalTypeAnnotation vectorLogicalType) {
+              List3Levels list3Levels = new List3Levels(type);
+              Type element = list3Levels.getElement();
+              TypeMapping child = fromParquet(element, null, element.getRepetition());
+              ArrowType.FixedSizeList fixedSizeList =
+                  new ArrowType.FixedSizeList(vectorLogicalType.getNumElements());
+              Field arrowField = new Field(
+                  name,
+                  new FieldType(type.isRepetition(OPTIONAL), fixedSizeList, null),
+                  Collections.singletonList(child.getArrowField()));
+              return of(new ListTypeMapping(arrowField, list3Levels, child));
+            }
+
             @Override
             public Optional<TypeMapping> visit(
                 LogicalTypeAnnotation.MapLogicalTypeAnnotation mapLogicalType) {

parquet-arrow/src/test/java/org/apache/parquet/arrow/schema/TestSchemaConverter.java

@@ -55,13 +55,34 @@
 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 testVectorAsFixedSizeList(String repetition) {
+    String children = "repeated group list { required float element; }";
+    MessageType vector = MessageTypeParser.parseMessageType(
+        "message m { " + repetition + " group v (FINITE_VECTOR(5)) { " + children + " } }");
+    SchemaMapping mapping = converter.fromParquet(vector);
+    Field field = mapping.getArrowSchema().findField("v");
+    assertThat(field.getType()).isEqualTo(new ArrowType.FixedSizeList(5));
+    assertThat(field.isNullable()).isEqualTo(repetition.equals("optional"));
+    assertThat(field.getChildren()).hasSize(1);
+    assertThat(field.getChildren().get(0).getType())
+        .isEqualTo(new ArrowType.FloatingPoint(FloatingPointPrecision.SINGLE));
+    assertThat(field.getChildren().get(0).isNullable()).isFalse();
+    assertThat(mapping.getParquetSchema()).isEqualTo(vector);
+    assertThat(converter.map(mapping.getArrowSchema(), vector).getParquetSchema())
+        .isEqualTo(vector);
+  }
 
   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) {
+    LogicalTypeAnnotation logicalType = group.getLogicalTypeAnnotation();
+    if (logicalType instanceof LogicalTypeAnnotation.ListLogicalTypeAnnotation) {
       if (group.isRepetition(Type.Repetition.REPEATED) || group.getFieldCount() != 1) {
         return false;
       }

parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java

@@ -470,7 +470,7 @@ public Optional<Schema> visit(
                 if (!repeatedType.isRepetition(REPEATED)) {
                   throw new UnsupportedOperationException("Invalid list type " + parquetGroupType);
                 }
-                if (isElementType(repeatedType, parquetGroupType.getName())) {
+                if (isElementType(repeatedType, parquetGroupType)) {
                   // repeated element types are always required
                   return of(Schema.createArray(convertField(repeatedType, names)));
                 } else {
@@ -654,7 +654,10 @@ public Optional<LogicalType> visit(UUIDLogicalTypeAnnotation uuidLogicalType) {
    * the same. This ensures that the LIST rules are followed when there is no
    * schema and that a schema can be provided to override the default behavior.
    */
-  private boolean isElementType(Type repeatedType, String parentName) {
+  private boolean isElementType(Type repeatedType, GroupType parent) {
+    if (parent.getLogicalTypeAnnotation() instanceof LogicalTypeAnnotation.FiniteVectorLogicalTypeAnnotation) {
+      return false;
+    }
     return (
     // can't be a synthetic layer because it would be invalid
     repeatedType.isPrimitive()
@@ -663,7 +666,7 @@ private boolean isElementType(Type repeatedType, String parentName) {
         ||
         // known patterns without the synthetic layer
         repeatedType.getName().equals("array")
-        || repeatedType.getName().equals(parentName + "_tuple")
+        || repeatedType.getName().equals(parent.getName() + "_tuple")
         ||
         // default assumption
         assumeRepeatedIsListElement);

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,53 @@ 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 (FINITE_VECTOR(2)) { repeated group list { required float element; } }"
+        + " optional group o (FINITE_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);
+    for (boolean assumeRepeatedIsListElement : new boolean[] {true, false}) {
+      assertThat(new AvroSchemaConverter(assumeRepeatedIsListElement, false).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/schema/LogicalTypeAnnotation.java

@@ -62,6 +62,13 @@ protected LogicalTypeAnnotation fromString(List<String> params) {
         return listType();
       }
     },
+    FINITE_VECTOR {
+      @Override
+      protected LogicalTypeAnnotation fromString(List<String> params) {
+        Preconditions.checkArgument(params.size() == 1, "Expecting 1 parameter for FINITE_VECTOR: %s", params);
+        return finiteVectorType(Integer.parseInt(params.get(0)));
+      }
+    },
     VARIANT {
       @Override
       protected LogicalTypeAnnotation fromString(List<String> params) {
@@ -314,6 +321,10 @@ public static ListLogicalTypeAnnotation listType() {
     return ListLogicalTypeAnnotation.INSTANCE;
   }
 
+  public static FiniteVectorLogicalTypeAnnotation finiteVectorType(int numElements) {
+    return new FiniteVectorLogicalTypeAnnotation(numElements);
+  }
+
   public static VariantLogicalTypeAnnotation variantType(byte specVersion) {
     return new VariantLogicalTypeAnnotation(specVersion);
   }
@@ -503,7 +514,7 @@ LogicalTypeToken getType() {
 
     @Override
     public boolean equals(Object obj) {
-      return obj instanceof ListLogicalTypeAnnotation;
+      return obj instanceof ListLogicalTypeAnnotation && !(obj instanceof FiniteVectorLogicalTypeAnnotation);
     }
 
     @Override
@@ -513,6 +524,45 @@ public int hashCode() {
     }
   }
 
+  public static class FiniteVectorLogicalTypeAnnotation extends ListLogicalTypeAnnotation {
+    private final int numElements;
+
+    private FiniteVectorLogicalTypeAnnotation(int numElements) {
+      Preconditions.checkArgument(numElements > 0, "Invalid FINITE_VECTOR length: %s", numElements);
+      this.numElements = numElements;
+    }
+
+    public int getNumElements() {
+      return numElements;
+    }
+
+    @Override
+    public <T> Optional<T> accept(LogicalTypeAnnotationVisitor<T> visitor) {
+      return visitor.visit(this);
+    }
+
+    @Override
+    LogicalTypeToken getType() {
+      return LogicalTypeToken.FINITE_VECTOR;
+    }
+
+    @Override
+    protected String typeParametersAsString() {
+      return "(" + numElements + ")";
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return obj instanceof FiniteVectorLogicalTypeAnnotation
+          && numElements == ((FiniteVectorLogicalTypeAnnotation) obj).numElements;
+    }
+
+    @Override
+    public int hashCode() {
+      return 31 * getClass().hashCode() + numElements;
+    }
+  }
+
   public static class EnumLogicalTypeAnnotation extends LogicalTypeAnnotation {
     private static final EnumLogicalTypeAnnotation INSTANCE = new EnumLogicalTypeAnnotation();
 
@@ -1536,6 +1586,10 @@ default Optional<T> visit(ListLogicalTypeAnnotation listLogicalType) {
       return empty();
     }
 
+    default Optional<T> visit(FiniteVectorLogicalTypeAnnotation vectorLogicalType) {
+      return visit((ListLogicalTypeAnnotation) vectorLogicalType);
+    }
+
     default Optional<T> visit(VariantLogicalTypeAnnotation variantLogicalType) {
       return empty();
     }

parquet-column/src/main/java/org/apache/parquet/schema/TypeUtil.java

@@ -18,11 +18,77 @@
  */
 package org.apache.parquet.schema;
 
+import org.apache.parquet.schema.LogicalTypeAnnotation.DecimalLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.FiniteVectorLogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.Float16LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.IntLogicalTypeAnnotation;
+
 public class TypeUtil {
+  public static boolean hasFiniteVector(Type type) {
+    if (type.getLogicalTypeAnnotation() instanceof FiniteVectorLogicalTypeAnnotation) return true;
+    if (!type.isPrimitive()) {
+      for (Type child : type.asGroupType().getFields()) {
+        if (hasFiniteVector(child)) return true;
+      }
+    }
+    return false;
+  }
+
+  public static FiniteVectorLogicalTypeAnnotation getFiniteVectorType(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 FiniteVectorLogicalTypeAnnotation
+        ? (FiniteVectorLogicalTypeAnnotation) logical
+        : null;
+  }
+
+  static void checkFiniteVector(GroupType vector) {
+    if (vector.isRepetition(Type.Repetition.REPEATED) || vector.getFieldCount() != 1) {
+      throw new InvalidSchemaException("Invalid FINITE_VECTOR schema: " + vector);
+    }
+    Type list = vector.getType(0);
+    if (list.isPrimitive()
+        || !list.getName().equals("list")
+        || !list.isRepetition(Type.Repetition.REPEATED)
+        || list.getLogicalTypeAnnotation() != null
+        || list.asGroupType().getFieldCount() != 1) {
+      throw new InvalidSchemaException("Invalid FINITE_VECTOR schema: " + vector);
+    }
+    Type element = list.asGroupType().getType(0);
+    if (!element.isPrimitive()
+        || !element.getName().equals("element")
+        || !element.isRepetition(Type.Repetition.REQUIRED)) {
+      throw new InvalidSchemaException("Invalid FINITE_VECTOR schema: " + vector);
+    }
+    LogicalTypeAnnotation logical = element.getLogicalTypeAnnotation();
+    boolean allowed = switch (element.asPrimitiveType().getPrimitiveTypeName()) {
+      case BOOLEAN, FLOAT, DOUBLE -> logical == null;
+      case INT32, INT64 ->
+        logical == null
+            || logical instanceof IntLogicalTypeAnnotation
+            || logical instanceof DecimalLogicalTypeAnnotation;
+      case FIXED_LEN_BYTE_ARRAY ->
+        logical instanceof Float16LogicalTypeAnnotation || logical instanceof DecimalLogicalTypeAnnotation;
+      default -> false;
+    };
+    if (!allowed) {
+      throw new InvalidSchemaException("Invalid FINITE_VECTOR schema: " + vector);
+    }
+  }
+
   public static void checkValidWriteSchema(GroupType schema) {
     schema.accept(new TypeVisitor() {
       @Override
       public void visit(GroupType groupType) {
+        if (groupType.getLogicalTypeAnnotation() instanceof FiniteVectorLogicalTypeAnnotation) {
+          checkFiniteVector(groupType);
+        }
         if (groupType.getFieldCount() <= 0) {
           throw new InvalidSchemaException("Cannot write a schema with an empty group: " + groupType);
         }

parquet-column/src/main/java/org/apache/parquet/schema/Types.java

@@ -29,6 +29,7 @@
 import java.util.stream.Collectors;
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.schema.ColumnOrder.ColumnOrderName;
+import org.apache.parquet.schema.LogicalTypeAnnotation.FiniteVectorLogicalTypeAnnotation;
 import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
 import org.apache.parquet.schema.Type.ID;
 import org.slf4j.Logger;
@@ -846,11 +847,13 @@ public THIS addFields(Type... types) {
 
     @Override
     protected GroupType build(String name) {
-      if (newLogicalTypeSet) {
-        return new GroupType(repetition, name, logicalTypeAnnotation, fields, id);
-      } else {
-        return new GroupType(repetition, name, getOriginalType(), fields, id);
+      GroupType group = newLogicalTypeSet
+          ? new GroupType(repetition, name, logicalTypeAnnotation, fields, id)
+          : new GroupType(repetition, name, getOriginalType(), fields, id);
+      if (logicalTypeAnnotation instanceof FiniteVectorLogicalTypeAnnotation) {
+        TypeUtil.checkFiniteVector(group);
       }
+      return group;
     }
 
     public MapBuilder<THIS> map(Type.Repetition repetition) {