PARQUET-352: Add object model property to file footers.

WriteSupport now has a getName getter method that is added to the footer
if it returns a non-null string as writer.model.name. This is intended
to help identify files written by object models incorrectly.

Author: Ryan Blue <[email protected]>

Closes apache#289 from rdblue/PARQUET-352-add-object-model-property and squashes the following commits:

23f8f67 [Ryan Blue] PARQUET-352: Add object model property to file footers.

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

@@ -92,6 +92,11 @@ public AvroWriteSupport(MessageType schema, Schema avroSchema,
     this.model = model;
   }
 
+  @Override
+  public String getName() {
+    return "avro";
+  }
+
   /**
    * @see org.apache.parquet.avro.AvroParquetOutputFormat#setSchema(org.apache.hadoop.mapreduce.Job, org.apache.avro.Schema)
    */

parquet-cascading/src/main/java/org/apache/parquet/cascading/TupleWriteSupport.java

@@ -41,6 +41,11 @@ public class TupleWriteSupport extends WriteSupport<TupleEntry> {
   private MessageType rootSchema;
   public static final String PARQUET_CASCADING_SCHEMA = "parquet.cascading.schema";
 
+  @Override
+  public String getName() {
+    return "cascading";
+  }
+
   @Override
   public WriteContext init(Configuration configuration) {
     String schema = configuration.get(PARQUET_CASCADING_SCHEMA);

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java

@@ -119,6 +119,10 @@ public void close() throws IOException, InterruptedException {
     flushRowGroupToStore();
     FinalizedWriteContext finalWriteContext = writeSupport.finalizeWrite();
     Map<String, String> finalMetadata = new HashMap<String, String>(extraMetaData);
+    String modelName = writeSupport.getName();
+    if (modelName != null) {
+      finalMetadata.put(ParquetWriter.OBJECT_MODEL_NAME_PROP, modelName);
+    }
     finalMetadata.putAll(finalWriteContext.getExtraMetaData());
     parquetFileWriter.end(finalMetadata);
   }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java

@@ -45,6 +45,8 @@ public class ParquetWriter<T> implements Closeable {
   public static final WriterVersion DEFAULT_WRITER_VERSION =
       WriterVersion.PARQUET_1_0;
 
+  public static final String OBJECT_MODEL_NAME_PROP = "writer.model.name";
+
   // max size (bytes) to write as padding and the min size of a row group
   public static final int MAX_PADDING_SIZE_DEFAULT = 0;
 

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/api/DelegatingWriteSupport.java

@@ -54,6 +54,11 @@ public void write(T record) {
     delegate.write(record);
   }
 
+  @Override
+  public String getName() {
+    return delegate.getName();
+  }
+
   @Override
   public WriteSupport.FinalizedWriteContext finalizeWrite() {
     return delegate.finalizeWrite();

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/api/WriteSupport.java

@@ -120,6 +120,18 @@ public Map<String, String> getExtraMetaData() {
    */
   public abstract void write(T record);
 
+  /**
+   * Called to get a name to identify the WriteSupport object model.
+   * If not null, this is added to the file footer metadata.
+   * <p>
+   * Defining this method will be required in a future API version.
+   *
+   * @return a String name for file metadata.
+   */
+  public String getName() {
+    return null;
+  }
+
   /**
    * called once in the end after the last record was written
    * @return information to be added in the file

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/example/GroupWriteSupport.java

@@ -63,6 +63,11 @@ public GroupWriteSupport() {
     this.extraMetaData = extraMetaData;
   }
 
+  @Override
+  public String getName() {
+    return "example";
+  }
+
   @Override
   public org.apache.parquet.hadoop.api.WriteSupport.WriteContext init(Configuration configuration) {
     // if present, prefer the schema passed to the constructor

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriter.java

@@ -135,6 +135,9 @@ public void test() throws Exception {
             }
           }
         }
+        assertEquals("Object model property should be example",
+            "example", footer.getFileMetaData().getKeyValueMetaData()
+                .get(ParquetWriter.OBJECT_MODEL_NAME_PROP));
       }
     }
   }

parquet-pig/src/main/java/org/apache/parquet/pig/TupleWriteSupport.java

@@ -67,6 +67,11 @@ public TupleWriteSupport(Schema pigSchema) {
     this.rootPigSchema = pigSchema;
   }
 
+  @Override
+  public String getName() {
+    return "pig";
+  }
+
   public Schema getPigSchema() {
     return rootPigSchema;
   }

parquet-protobuf/src/main/java/org/apache/parquet/proto/ProtoWriteSupport.java

@@ -62,6 +62,11 @@ public ProtoWriteSupport(Class<? extends Message> protobufClass) {
     this.protoMessage = protobufClass;
   }
 
+  @Override
+  public String getName() {
+    return "protobuf";
+  }
+
   public static void setSchema(Configuration configuration, Class<? extends Message> protoClass) {
     configuration.setClass(PB_CLASS_WRITE, protoClass, Message.class);
   }

parquet-scrooge/src/main/java/org/apache/parquet/scrooge/ScroogeWriteSupport.java

@@ -48,6 +48,11 @@ public ScroogeWriteSupport(Class<T> thriftClass) {
     super(thriftClass);
   }
 
+  @Override
+  public String getName() {
+    return "scrooge";
+  }
+
   @Override
   protected StructType getThriftStruct() {
     ScroogeStructConverter schemaConverter = new ScroogeStructConverter();

parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/TBaseWriteSupport.java

@@ -45,6 +45,11 @@ public TBaseWriteSupport(Class<T> thriftClass) {
     super(thriftClass);
   }
 
+  @Override
+  public String getName() {
+    return "thrift";
+  }
+
   @Override
   protected StructType getThriftStruct() {
     return ThriftSchemaConverter.toStructType(thriftClass);

parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftBytesWriteSupport.java

@@ -92,6 +92,11 @@ public ThriftBytesWriteSupport(TProtocolFactory protocolFactory, Class<? extends
     }
   }
 
+  @Override
+  public String getName() {
+    return "thrift";
+  }
+
   @Override
   public WriteContext init(Configuration configuration) {
     if (this.protocolFactory == null) {

parquet-thrift/src/main/java/org/apache/parquet/hadoop/thrift/ThriftWriteSupport.java

@@ -58,6 +58,11 @@ public ThriftWriteSupport(Class<T> thriftClass) {
     this.writeSupport = new TBaseWriteSupport(thriftClass);
   }
 
+  @Override
+  public String getName() {
+    return writeSupport.getName();
+  }
+
   @Override
   public WriteContext init(Configuration configuration) {
     return this.writeSupport.init(configuration);

parquet-thrift/src/main/java/org/apache/parquet/thrift/pig/TupleToThriftWriteSupport.java

@@ -49,6 +49,11 @@ public TupleToThriftWriteSupport(String className) {
     this.className = className;
   }
 
+  @Override
+  public String getName() {
+    return "thrift";
+  }
+
   @SuppressWarnings({"rawtypes", "unchecked"})
   @Override
   public WriteContext init(Configuration configuration) {