Merge branch 'master' into bloom_filter_length

Author: zhangjiashen

parquet-arrow/pom.xml

@@ -33,7 +33,7 @@
   <url>https://parquet.apache.org</url>
 
   <properties>
-    <arrow.version>0.10.0</arrow.version>
+    <arrow.version>14.0.1</arrow.version>
   </properties>
 
   <dependencies>

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

@@ -21,7 +21,7 @@
 import static org.apache.parquet.schema.Type.Repetition.REPEATED;
 
 import org.apache.parquet.schema.GroupType;
-import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
 import org.apache.parquet.schema.Type;
 
 /**
@@ -41,7 +41,7 @@ class List3Levels {
    * @param list the Parquet List
    */
   public List3Levels(GroupType list) {
-    if (list.getOriginalType() != OriginalType.LIST || list.getFields().size() != 1) {
+    if (list.getLogicalTypeAnnotation() != LogicalTypeAnnotation.listType() || list.getFields().size() != 1) {
       throw new IllegalArgumentException("invalid list type: " + list);
     }
     this.list = list;

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

@@ -0,0 +1,86 @@
+/*
+ * 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.arrow.schema;
+
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.Type;
+
+import static org.apache.parquet.schema.Type.Repetition.REPEATED;
+
+/**
+ * Represents a standard 3 levels Parquet map
+ * - optional map
+ * - repeated key_value
+ * - required key, optional value
+ */
+class Map3Levels {
+  private final GroupType map;
+  private final GroupType repeated;
+  private final Type key;
+  private final Type value;
+
+  /**
+   * Will validate the structure of the map
+   * @param map the Parquet map
+   */
+  public Map3Levels(GroupType map) {
+    if (map.getLogicalTypeAnnotation() != LogicalTypeAnnotation.mapType() || map.getFields().size() != 1) {
+      throw new IllegalArgumentException("invalid map type: " + map);
+    }
+    this.map = map;
+    Type repeatedField = map.getFields().get(0);
+    if (repeatedField.isPrimitive() || !repeatedField.isRepetition(REPEATED) || repeatedField.asGroupType().getFields().size() != 2) {
+      throw new IllegalArgumentException("invalid map key: " + map);
+    }
+    this.repeated = repeatedField.asGroupType();
+    this.key = repeated.getFields().get(0);
+    this.value = repeated.getFields().get(1);
+  }
+
+  /**
+   * @return the root map element (an optional group with two children)
+   */
+  public GroupType getMap() {
+    return map;
+  }
+
+  /**
+   * @return repeated level, single child of map
+   */
+  public GroupType getRepeated() {
+    return repeated;
+  }
+
+  /**
+   * @return the key level
+   */
+  public Type getKey() {
+    return key;
+  }
+
+  /**
+   * @return the element level, single child of repeated.
+   */
+  public Type getValue() {
+    return value;
+  }
+
+}

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

@@ -42,6 +42,7 @@
 import static org.apache.parquet.schema.Type.Repetition.REQUIRED;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 
@@ -64,6 +65,7 @@
 import org.apache.arrow.vector.types.pojo.ArrowType.Union;
 import org.apache.arrow.vector.types.pojo.ArrowType.Utf8;
 import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.apache.parquet.arrow.schema.SchemaMapping.ListTypeMapping;
 import org.apache.parquet.arrow.schema.SchemaMapping.PrimitiveTypeMapping;
@@ -158,6 +160,11 @@ public TypeMapping visit(org.apache.arrow.vector.types.pojo.ArrowType.List type)
         return createListTypeMapping();
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.LargeList largeList) {
+        return createListTypeMapping();
+      }
+
       @Override
       public TypeMapping visit(org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeList type) {
         return createListTypeMapping();
@@ -179,6 +186,17 @@ public TypeMapping visit(Union type) {
         return new UnionTypeMapping(field, addToBuilder(parquetTypes, Types.buildGroup(OPTIONAL)).named(fieldName), parquetTypes);
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.Map map) {
+        if (children.size() != 2) {
+          throw new IllegalArgumentException("Map fields must have exactly two children: " + field);
+        }
+        TypeMapping keyChild = fromArrow(children.get(0), "key");
+        TypeMapping valueChild = fromArrow(children.get(1), "value");
+        GroupType groupType = Types.optionalMap().key(keyChild.getParquetType()).value(valueChild.getParquetType()).named(fieldName);
+        return new SchemaMapping.MapTypeMapping(field, new Map3Levels(groupType), keyChild, valueChild);
+      }
+
       @Override
       public TypeMapping visit(Int type) {
         boolean signed = type.getIsSigned();
@@ -214,11 +232,21 @@ public TypeMapping visit(Utf8 type) {
         return primitive(BINARY, stringType());
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.LargeUtf8 largeUtf8) {
+        return primitive(BINARY, stringType());
+      }
+
       @Override
       public TypeMapping visit(Binary type) {
         return primitive(BINARY);
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.LargeBinary largeBinary) {
+        return primitive(BINARY);
+      }
+
       @Override
       public TypeMapping visit(Bool type) {
         return primitive(BOOLEAN);
@@ -289,6 +317,16 @@ public TypeMapping visit(Interval type) {
         return primitiveFLBA(12, LogicalTypeAnnotation.IntervalLogicalTypeAnnotation.getInstance());
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.Duration duration) {
+        return primitiveFLBA(12, LogicalTypeAnnotation.IntervalLogicalTypeAnnotation.getInstance());
+      }
+
+      @Override
+      public TypeMapping visit(ArrowType.ExtensionType type) {
+        return ArrowTypeVisitor.super.visit(type);
+      }
+
       @Override
       public TypeMapping visit(ArrowType.FixedSizeBinary fixedSizeBinary) {
         return primitive(BINARY);
@@ -358,7 +396,7 @@ private TypeMapping fromParquet(Type type, String name, Repetition repetition) {
     if (repetition == REPEATED) {
       // case where we have a repeated field that is not in a List/Map
       TypeMapping child = fromParquet(type, null, REQUIRED);
-      Field arrowField = new Field(name, false, new ArrowType.List(), asList(child.getArrowField()));
+      Field arrowField = new Field(name, FieldType.notNullable(new ArrowType.List()), Collections.singletonList(child.getArrowField()));
       return new RepeatedTypeMapping(arrowField, type, child);
     }
     if (type.isPrimitive()) {
@@ -376,18 +414,32 @@ private TypeMapping fromParquet(Type type, String name, Repetition repetition) {
   private TypeMapping fromParquetGroup(GroupType type, String name) {
     LogicalTypeAnnotation logicalType = type.getLogicalTypeAnnotation();
     if (logicalType == null) {
+      final FieldType field;
+      if (type.isRepetition(OPTIONAL)) {
+        field = FieldType.nullable(new Struct());
+      } else {
+        field = FieldType.notNullable(new Struct());
+      }
       List<TypeMapping> typeMappings = fromParquet(type.getFields());
-      Field arrowField = new Field(name, type.isRepetition(OPTIONAL), new Struct(), fields(typeMappings));
+      Field arrowField = new Field(name, field, fields(typeMappings));
       return new StructTypeMapping(arrowField, type, typeMappings);
     } else {
       return logicalType.accept(new LogicalTypeAnnotation.LogicalTypeAnnotationVisitor<TypeMapping>() {
         @Override
         public Optional<TypeMapping> visit(LogicalTypeAnnotation.ListLogicalTypeAnnotation listLogicalType) {
           List3Levels list3Levels = new List3Levels(type);
           TypeMapping child = fromParquet(list3Levels.getElement(), null, list3Levels.getElement().getRepetition());
-          Field arrowField = new Field(name, type.isRepetition(OPTIONAL), new ArrowType.List(), asList(child.getArrowField()));
+          Field arrowField = new Field(name, FieldType.nullable(new ArrowType.List()), Collections.singletonList(child.getArrowField()));
           return of(new ListTypeMapping(arrowField, list3Levels, child));
         }
+        @Override
+        public Optional<TypeMapping> visit(LogicalTypeAnnotation.MapLogicalTypeAnnotation mapLogicalType) {
+          Map3Levels map3levels = new Map3Levels(type);
+          TypeMapping keyType = fromParquet(map3levels.getKey(), null, map3levels.getKey().getRepetition());
+          TypeMapping valueType = fromParquet(map3levels.getValue(), null, map3levels.getValue().getRepetition());
+          Field arrowField = new Field(name, FieldType.nullable(new ArrowType.Map(false)), asList(keyType.getArrowField(), valueType.getArrowField()));
+          return of(new SchemaMapping.MapTypeMapping(arrowField, map3levels, keyType, valueType));
+        }
       }).orElseThrow(() -> new UnsupportedOperationException("Unsupported type " + type));
     }
   }
@@ -401,7 +453,12 @@ private TypeMapping fromParquetPrimitive(final PrimitiveType type, final String
     return type.getPrimitiveTypeName().convert(new PrimitiveType.PrimitiveTypeNameConverter<TypeMapping, RuntimeException>() {
 
       private TypeMapping field(ArrowType arrowType) {
-        Field field = new Field(name, type.isRepetition(OPTIONAL), arrowType, null);
+        final Field field;
+        if (type.isRepetition(OPTIONAL)) {
+          field = Field.nullable(name, arrowType);
+        } else {
+          field = Field.notNullable(name, arrowType);
+        }
         return new PrimitiveTypeMapping(field, type);
       }
 
@@ -607,6 +664,11 @@ public TypeMapping visit(org.apache.arrow.vector.types.pojo.ArrowType.List type)
         return createListTypeMapping(type);
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.LargeList largeList) {
+        return createListTypeMapping(largeList);
+      }
+
       @Override
       public TypeMapping visit(org.apache.arrow.vector.types.pojo.ArrowType.FixedSizeList type) {
         return createListTypeMapping(type);
@@ -639,6 +701,26 @@ public TypeMapping visit(Union type) {
         return new UnionTypeMapping(arrowField, groupType, map(arrowField.getChildren(), groupType.getFields()));
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.Map map) {
+        if (arrowField.getChildren().size() != 2) {
+          throw new IllegalArgumentException("Invalid map type: " + map);
+        }
+        if (parquetField.isPrimitive()) {
+          throw new IllegalArgumentException("Parquet type not a group: " + parquetField);
+        }
+        Map3Levels map3levels = new Map3Levels(parquetField.asGroupType());
+        if (arrowField.getChildren().size() != 2) {
+          throw new IllegalArgumentException("invalid arrow map: " + arrowField);
+        }
+        Field keyChild = arrowField.getChildren().get(0);
+        Field valueChild = arrowField.getChildren().get(1);
+        return new SchemaMapping.MapTypeMapping(arrowField, map3levels,
+          map(keyChild, map3levels.getKey()),
+          map(valueChild, map3levels.getValue())
+        );
+      }
+
       @Override
       public TypeMapping visit(Int type) {
         return primitive();
@@ -654,11 +736,21 @@ public TypeMapping visit(Utf8 type) {
         return primitive();
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.LargeUtf8 largeUtf8) {
+        return primitive();
+      }
+
       @Override
       public TypeMapping visit(Binary type) {
         return primitive();
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.LargeBinary largeBinary) {
+        return primitive();
+      }
+
       @Override
       public TypeMapping visit(Bool type) {
         return primitive();
@@ -689,6 +781,11 @@ public TypeMapping visit(Interval type) {
         return primitive();
       }
 
+      @Override
+      public TypeMapping visit(ArrowType.Duration duration) {
+        return primitive();
+      }
+
       @Override
       public TypeMapping visit(ArrowType.FixedSizeBinary fixedSizeBinary) {
         return primitive();