PARQUET-1502: Convert FIXED_LEN_BYTE_ARRAY to arrow type in logicalTypeAnnotation if it is not null (#593)

Author: yongyanw

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

@@ -498,7 +498,17 @@ public TypeMapping convertINT96(PrimitiveTypeName primitiveTypeName) throws Runt
 
       @Override
       public TypeMapping convertFIXED_LEN_BYTE_ARRAY(PrimitiveTypeName primitiveTypeName) throws RuntimeException {
-        return field(new ArrowType.Binary());
+        LogicalTypeAnnotation logicalTypeAnnotation = type.getLogicalTypeAnnotation();
+        if (logicalTypeAnnotation == null) {
+          return field(new ArrowType.Binary());
+        }
+
+        return logicalTypeAnnotation.accept(new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<TypeMapping>() {
+          @Override
+          public Optional<TypeMapping> visit(LogicalTypeAnnotation.DecimalLogicalTypeAnnotation decimalLogicalType) {
+            return of(decimal(decimalLogicalType.getPrecision(), decimalLogicalType.getScale()));
+          }
+        }).orElseThrow(() -> new IllegalArgumentException("illegal type " + type));
       }
 
       @Override

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

@@ -419,6 +419,26 @@ public void testParquetInt64TimeMicrosToArrow() {
     Assert.assertEquals(expected, converter.fromParquet(parquet).getArrowSchema());
   }
 
+  @Test
+  public void testParquetFixedBinaryToArrow() {
+    MessageType parquet = Types.buildMessage()
+      .addField(Types.optional(FIXED_LEN_BYTE_ARRAY).length(12).named("a")).named("root");
+    Schema expected = new Schema(asList(
+      field("a", new ArrowType.Binary())
+    ));
+    Assert.assertEquals(expected, converter.fromParquet(parquet).getArrowSchema());
+  }
+
+  @Test
+  public void testParquetFixedBinaryToArrowDecimal() {
+    MessageType parquet = Types.buildMessage()
+      .addField(Types.optional(FIXED_LEN_BYTE_ARRAY).length(5).as(DECIMAL).precision(8).scale(2).named("a")).named("root");
+    Schema expected = new Schema(asList(
+      field("a", new ArrowType.Decimal(8, 2))
+    ));
+    Assert.assertEquals(expected, converter.fromParquet(parquet).getArrowSchema());
+  }
+
   @Test(expected = IllegalStateException.class)
   public void testParquetInt64TimeMillisToArrow() {
     converter.fromParquet(Types.buildMessage()