PARQUET-63: Enable dictionary encoding for FIXED.

parquet-column/src/main/java/parquet/column/Encoding.java

@@ -112,6 +112,7 @@ public ValuesReader getValuesReader(ColumnDescriptor descriptor, ValuesType valu
     public ValuesReader getDictionaryBasedValuesReader(ColumnDescriptor descriptor, ValuesType valuesType, Dictionary dictionary) {
       switch (descriptor.getType()) {
       case BINARY:
+      case FIXED_LEN_BYTE_ARRAY:
       case INT96:
       case INT64:
       case DOUBLE:
@@ -128,6 +129,8 @@ public Dictionary initDictionary(ColumnDescriptor descriptor, DictionaryPage dic
       switch (descriptor.getType()) {
       case BINARY:
         return new PlainBinaryDictionary(dictionaryPage);
+      case FIXED_LEN_BYTE_ARRAY:
+        return new PlainBinaryDictionary(dictionaryPage, descriptor.getTypeLength());
       case INT96:
         return new PlainBinaryDictionary(dictionaryPage, 12);
       case INT64:

parquet-column/src/main/java/parquet/column/ParquetProperties.java

@@ -119,7 +119,11 @@ public ValuesWriter getValuesWriter(ColumnDescriptor path, int initialSizePerCol
         return new PlainValuesWriter(initialSizePerCol);
       }
     case FIXED_LEN_BYTE_ARRAY:
-      return new FixedLenByteArrayPlainValuesWriter(path.getTypeLength(), initialSizePerCol);
+      if (enableDictionary && (writerVersion == WriterVersion.PARQUET_2_0)) {
+        return new PlainFixedLenArrayDictionaryValuesWriter(dictionaryPageSizeThreshold, initialSizePerCol, path.getTypeLength());
+      } else {
+        return new FixedLenByteArrayPlainValuesWriter(path.getTypeLength(), initialSizePerCol);
+      }
     default:
       return new PlainValuesWriter(initialSizePerCol);
     }

parquet-column/src/main/java/parquet/column/values/dictionary/DictionaryValuesWriter.java

@@ -364,7 +364,7 @@ public void writeBytes(Binary value) {
     public DictionaryPage createDictionaryPage() {
       if (lastUsedDictionarySize > 0) {
         // return a dictionary only if we actually used it
-        FixedLenByteArrayPlainValuesWriter dictionaryEncoder = new FixedLenByteArrayPlainValuesWriter(12, lastUsedDictionaryByteSize);
+        FixedLenByteArrayPlainValuesWriter dictionaryEncoder = new FixedLenByteArrayPlainValuesWriter(length, lastUsedDictionaryByteSize);
         Iterator<Binary> binaryIterator = binaryDictionaryContent.keySet().iterator();
         // write only the part of the dict that we used
         for (int i = 0; i < lastUsedDictionarySize; i++) {

parquet-column/src/main/java/parquet/example/data/simple/SimpleGroup.java

@@ -179,6 +179,7 @@ public void add(int fieldIndex, boolean value) {
   public void add(int fieldIndex, Binary value) {
     switch (getType().getType(fieldIndex).asPrimitiveType().getPrimitiveTypeName()) {
       case BINARY:
+      case FIXED_LEN_BYTE_ARRAY:
         add(fieldIndex, new BinaryValue(value));
         break;
       case INT96:

parquet-column/src/main/java/parquet/io/ValidatingRecordConsumer.java

@@ -27,6 +27,7 @@
 import parquet.schema.PrimitiveType.PrimitiveTypeName;
 import parquet.schema.Type.Repetition;
 
+import static parquet.schema.PrimitiveType.PrimitiveTypeName.*;
 
 /**
  * Wraps a record consumer
@@ -141,15 +142,15 @@ private void validate(PrimitiveTypeName p) {
         throw new InvalidRecordException("unknown repetition " + currentType.getRepetition() + " in " + currentType);
     }
     if (!currentType.isPrimitive() || currentType.asPrimitiveType().getPrimitiveTypeName() != p) {
-      throw new InvalidRecordException("expected type " + currentType + " but got "+ p);
+      throw new InvalidRecordException("expected type " + p + " but got "+ currentType);
     }
   }
 
-  private void validate(PrimitiveTypeName p1, PrimitiveTypeName p2) {
+  private void validate(PrimitiveTypeName... ptypes) {
     Type currentType = types.peek().asGroupType().getType(fields.peek());
     int c = fieldValueCount.pop() + 1;
     fieldValueCount.push(c);
-    if (DEBUG) LOG.debug("validate " + p1 + ", " + p2 + " for " + currentType.getName());
+    if (DEBUG) LOG.debug("validate " + Arrays.toString(ptypes) + " for " + currentType.getName());
     switch (currentType.getRepetition()) {
       case OPTIONAL:
       case REQUIRED:
@@ -162,60 +163,64 @@ private void validate(PrimitiveTypeName p1, PrimitiveTypeName p2) {
       default:
         throw new InvalidRecordException("unknown repetition " + currentType.getRepetition() + " in " + currentType);
     }
-    if (!currentType.isPrimitive() ||
-        (currentType.asPrimitiveType().getPrimitiveTypeName() != p1 &&
-         currentType.asPrimitiveType().getPrimitiveTypeName() != p2)) {
+    if (!currentType.isPrimitive()) {
       throw new InvalidRecordException(
-          "expected type " + currentType + " but got " + p1 + " or " + p2);
+          "expected type in " + Arrays.toString(ptypes) + " but got " + currentType);
     }
+    for (PrimitiveTypeName p : ptypes) {
+      if (currentType.asPrimitiveType().getPrimitiveTypeName() == p) {
+        return; // type is valid
+      }
+    }
+    throw new InvalidRecordException(
+        "expected type in " + Arrays.toString(ptypes) + " but got " + currentType);
   }
 
   /**
    * {@inheritDoc}
    */
   public void addInteger(int value) {
-    validate(PrimitiveTypeName.INT32);
+    validate(INT32);
     delegate.addInteger(value);
   }
 
   /**
    * {@inheritDoc}
    */
   public void addLong(long value) {
-    validate(PrimitiveTypeName.INT64);
+    validate(INT64);
     delegate.addLong(value);
   }
 
   /**
    * {@inheritDoc}
    */
   public void addBoolean(boolean value) {
-    validate(PrimitiveTypeName.BOOLEAN);
+    validate(BOOLEAN);
     delegate.addBoolean(value);
   }
 
   /**
    * {@inheritDoc}
    */
   public void addBinary(Binary value) {
-    // TODO: this is used for FIXED also
-    validate(PrimitiveTypeName.BINARY, PrimitiveTypeName.INT96);
+    validate(BINARY, INT96, FIXED_LEN_BYTE_ARRAY);
     delegate.addBinary(value);
   }
 
   /**
    * {@inheritDoc}
    */
   public void addFloat(float value) {
-    validate(PrimitiveTypeName.FLOAT);
+    validate(FLOAT);
     delegate.addFloat(value);
   }
 
   /**
    * {@inheritDoc}
    */
   public void addDouble(double value) {
-    validate(PrimitiveTypeName.DOUBLE);
+    validate(DOUBLE);
     delegate.addDouble(value);
   }
 

parquet-column/src/test/java/parquet/io/TestColumnIO.java

@@ -80,6 +80,7 @@ public class TestColumnIO {
   + "  required boolean e;\n"
   + "  required binary f;\n"
   + "  required int96 g;\n"
+  + "  required fixed_len_byte_array(3) h;\n"
   + "}\n";
 
   private static final String schemaString =
@@ -363,7 +364,8 @@ public void testOneOfEach() {
         .append("d", 4.0d)
         .append("e", true)
         .append("f", Binary.fromString("6"))
-        .append("g", new NanoTime(1234, System.currentTimeMillis() * 1000));
+        .append("g", new NanoTime(1234, System.currentTimeMillis() * 1000))
+        .append("h", Binary.fromString("abc"));
 
     testSchema(oneOfEachSchema, Arrays.asList(g1));
   }