Support adding fields to nested Parquet structs

presto-hive/src/main/java/com/facebook/presto/hive/HiveSplitManager.java

@@ -35,9 +35,13 @@
 import com.google.common.collect.Ordering;
 import io.airlift.concurrent.BoundedExecutor;
 import org.apache.hadoop.hive.metastore.ProtectMode;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
 
 import javax.inject.Inject;
 
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -248,7 +252,24 @@ private Iterable<HivePartitionMetadata> getPartitionMetadata(SemiTransactionalHi
                 for (int i = 0; i < min(partitionColumns.size(), tableColumns.size()); i++) {
                     HiveType tableType = tableColumns.get(i).getType();
                     HiveType partitionType = partitionColumns.get(i).getType();
-                    if (!tableType.equals(partitionType)) {
+                    if (isStruct(tableType) && isStruct(partitionType)) {
+                        ArrayList<TypeInfo> fromFieldTypes = getStructFields(partitionType);
+                        ArrayList<TypeInfo> toFieldTypes = getStructFields(tableType);
+                        if (!toFieldTypes.subList(0, fromFieldTypes.size()).equals(fromFieldTypes)) {
+                            throw new PrestoException(HIVE_PARTITION_SCHEMA_MISMATCH, format("" +
+                                            "There is a mismatch between the table and partition schemas. " +
+                                            "The structs are incompatible and cannot be coerced. " +
+                                            "The column '%s' in table '%s' is declared as type '%s', " +
+                                            "but partition '%s' declared column '%s' as type '%s'.",
+                                    tableColumns.get(i).getName(),
+                                    tableName,
+                                    tableType,
+                                    partName,
+                                    partitionColumns.get(i).getName(),
+                                    partitionType));
+                        }
+                    }
+                    else if (!tableType.equals(partitionType)) {
                         if (!coercionPolicy.canCoerce(partitionType, tableType)) {
                             throw new PrestoException(HIVE_PARTITION_SCHEMA_MISMATCH, format("" +
                                             "There is a mismatch between the table and partition schemas. " +
@@ -335,4 +356,14 @@ public void execute(Runnable command)
             }
         }
     }
+
+    private static boolean isStruct(HiveType type)
+    {
+        return type.getCategory() == ObjectInspector.Category.STRUCT;
+    }
+
+    private static ArrayList<TypeInfo> getStructFields(HiveType structHiveType)
+    {
+        return ((StructTypeInfo) structHiveType.getTypeInfo()).getAllStructFieldTypeInfos();
+    }
 }

presto-hive/src/main/java/com/facebook/presto/hive/parquet/ParquetHiveRecordCursor.java

@@ -741,7 +741,7 @@ public ParquetStructConverter(Type prestoType, String columnName, GroupType entr
             List<Type> prestoTypeParameters = prestoType.getTypeParameters();
             List<parquet.schema.Type> fieldTypes = entryType.getFields();
             checkArgument(
-                    prestoTypeParameters.size() == fieldTypes.size(),
+                    prestoTypeParameters.size() >= fieldTypes.size(),
                     "Schema mismatch, metastore schema for row column %s has %s fields but parquet schema has %s fields",
                     columnName,
                     prestoTypeParameters.size(),
@@ -751,7 +751,7 @@ public ParquetStructConverter(Type prestoType, String columnName, GroupType entr
             this.fieldIndex = fieldIndex;
 
             ImmutableList.Builder<BlockConverter> converters = ImmutableList.builder();
-            for (int i = 0; i < prestoTypeParameters.size(); i++) {
+            for (int i = 0; i < fieldTypes.size(); i++) {
                 parquet.schema.Type fieldType = fieldTypes.get(i);
                 converters.add(createConverter(prestoTypeParameters.get(i), columnName + "." + fieldType.getName(), fieldType, i));
             }
@@ -796,7 +796,7 @@ public void end()
             for (BlockConverter converter : converters) {
                 converter.afterValue();
             }
-            while (currentEntryBuilder.getPositionCount() < converters.size()) {
+            while (currentEntryBuilder.getPositionCount() < rowType.getTypeParameters().size()) {
                 currentEntryBuilder.appendNull();
             }
 

presto-hive/src/test/java/com/facebook/presto/hive/AbstractTestHiveClient.java

@@ -79,6 +79,7 @@
 import com.facebook.presto.testing.TestingConnectorSession;
 import com.facebook.presto.type.ArrayType;
 import com.facebook.presto.type.MapType;
+import com.facebook.presto.type.RowType;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMultimap;
@@ -2742,6 +2743,22 @@ else if (rowNumber % 39 == 1) {
                     }
                 }
 
+                // STRUCT<s_string: STRING, s_double:DOUBLE>
+                index = columnIndex.get("t_struct");
+                if (index != null) {
+                    if ((rowNumber % 31) == 0) {
+                        assertNull(row.getField(index));
+                    }
+                    else {
+                        assertTrue(row.getField(index) instanceof List);
+                        List values = (List) row.getField(index);
+                        assertEquals(values.size(), 3);
+                        assertEquals(values.get(0), "test abc");
+                        assertEquals(values.get(1), 0.1);
+                        assertNull(values.get(2));
+                    }
+                }
+
                 // MAP<INT, ARRAY<STRUCT<s_string: STRING, s_double:DOUBLE>>>
                 index = columnIndex.get("t_complex");
                 if (index != null) {
@@ -2980,7 +2997,7 @@ else if (TIMESTAMP.equals(column.getType())) {
                 else if (DATE.equals(column.getType())) {
                     assertInstanceOf(value, SqlDate.class);
                 }
-                else if (column.getType() instanceof ArrayType) {
+                else if (column.getType() instanceof ArrayType || column.getType() instanceof RowType) {
                     assertInstanceOf(value, List.class);
                 }
                 else if (column.getType() instanceof MapType) {

presto-hive/src/test/sql/create-test-hive13.sql

@@ -98,12 +98,15 @@ CREATE TABLE presto_test_types_parquet (
 , t_binary BINARY
 , t_map MAP<STRING, STRING>
 , t_array_string ARRAY<STRING>
-, t_array_struct ARRAY<STRUCT<s_string: STRING, s_double:DOUBLE>>
+, t_array_struct ARRAY<STRUCT<s_string:STRING, s_double:DOUBLE>>
+, t_struct STRUCT<s_string:STRING, s_double:DOUBLE>
 )
+PARTITIONED BY (dummy INT)
 STORED AS PARQUET
 ;
 
 INSERT INTO TABLE presto_test_types_parquet
+PARTITION (dummy=0)
 SELECT
   t_string
 , t_varchar
@@ -119,9 +122,12 @@ SELECT
 , t_map
 , t_array_string
 , t_array_struct
+, t_array_struct[0]
 FROM presto_test_types_textfile
 ;
 
+ALTER TABLE presto_test_types_parquet
+CHANGE COLUMN t_struct t_struct STRUCT<s_string:STRING, s_double:DOUBLE, s_boolean:BOOLEAN>;
 
 ALTER TABLE presto_test_types_textfile ADD COLUMNS (new_column INT);
 ALTER TABLE presto_test_types_sequencefile ADD COLUMNS (new_column INT);