PARQUET-22: Backport of HIVE-6938 adding rename support for parquet

This patch was included in hive after the moving the Serde to hive (included in hive 0.14+).  Backport is required for use with previous versions.

Author: Daniel Weeks <[email protected]>

Closes apache#13 from dcw-netflix/backport-hive-6938-rename and squashes the following commits:

453367b [Daniel Weeks] Backport of HIVE-6938 adding rename support for parquet

Author: Daniel Weeks

parquet-hive/parquet-hive-storage-handler/src/main/java/org/apache/hadoop/hive/ql/io/parquet/read/DataWritableReadSupport.java

@@ -45,6 +45,7 @@ public class DataWritableReadSupport extends ReadSupport<ArrayWritable> {
 
   private static final String TABLE_SCHEMA = "table_schema";
   public static final String HIVE_SCHEMA_KEY = "HIVE_TABLE_SCHEMA";
+  public static final String PARQUET_COLUMN_INDEX_ACCESS = "parquet.column.index.access";
 
   /**
    * From a string which columns names (including hive column), return a list
@@ -93,7 +94,8 @@ public parquet.hadoop.api.ReadSupport.ReadContext init(final Configuration confi
       for (final Integer idx : indexColumnsWanted) {
         typeListWanted.add(tableSchema.getType(listColumns.get(idx)));
       }
-      requestedSchemaByUser = new MessageType(fileSchema.getName(), typeListWanted);
+      requestedSchemaByUser = resolveSchemaAccess(new MessageType(fileSchema.getName(),
+              typeListWanted), fileSchema, configuration);
 
       return new ReadContext(requestedSchemaByUser, contextMetadata);
     } else {
@@ -121,8 +123,31 @@ public RecordMaterializer<ArrayWritable> prepareForRead(final Configuration conf
       throw new IllegalStateException("ReadContext not initialized properly. " +
         "Don't know the Hive Schema.");
     }
-    final MessageType tableSchema = MessageTypeParser.
-        parseMessageType(metadata.get(HIVE_SCHEMA_KEY));
+    final MessageType tableSchema = resolveSchemaAccess(MessageTypeParser.
+        parseMessageType(metadata.get(HIVE_SCHEMA_KEY)), fileSchema, configuration);
+
     return new DataWritableRecordConverter(readContext.getRequestedSchema(), tableSchema);
   }
+
+  /**
+  * Determine the file column names based on the position within the requested columns and
+  * use that as the requested schema.
+  */
+  private MessageType resolveSchemaAccess(MessageType requestedSchema, MessageType fileSchema,
+          Configuration configuration) {
+    if(configuration.getBoolean(PARQUET_COLUMN_INDEX_ACCESS, false)) {
+      final List<String> listColumns = getColumns(configuration.get(IOConstants.COLUMNS));
+
+      List<Type> requestedTypes = new ArrayList<Type>();
+
+      for(Type t : requestedSchema.getFields()) {
+        int index = listColumns.indexOf(t.getName());
+        requestedTypes.add(fileSchema.getType(index));
+      }
+
+      requestedSchema = new MessageType(requestedSchema.getName(), requestedTypes);
+    }
+
+    return requestedSchema;
+  }
 }