ARROW-7284: [Java] ensure java implementation meets clarified dictionary spec

java/vector/src/main/java/org/apache/arrow/vector/ipc/ArrowReader.java

@@ -35,6 +35,7 @@
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.apache.arrow.vector.util.DictionaryUtility;
+import org.apache.arrow.vector.util.TransferPair;
 
 /**
  * Abstract class to read Schema and ArrowRecordBatches.
@@ -219,6 +220,18 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
       throw new IllegalArgumentException("Dictionary ID " + id + " not defined in schema");
     }
     FieldVector vector = dictionary.getVector();
+    // if is deltaVector, concat it with non-delta vector with the same ID.
+    if (dictionaryBatch.isDelta()) {
+      FieldVector deltaVector = vector.getField().createVector(allocator);
+      load(dictionaryBatch, deltaVector);
+      concatDeltaDictionary(vector, deltaVector);
+      return;
+    }
+
+    load(dictionaryBatch, vector);
+  }
+
+  private void load(ArrowDictionaryBatch dictionaryBatch, FieldVector vector) {
     VectorSchemaRoot root = new VectorSchemaRoot(
         Collections.singletonList(vector.getField()),
         Collections.singletonList(vector), 0);
@@ -229,4 +242,18 @@ protected void loadDictionary(ArrowDictionaryBatch dictionaryBatch) {
       dictionaryBatch.close();
     }
   }
+
+  /**
+   * Concat dictionary vector and delta dictionary vector.
+   */
+  private void concatDeltaDictionary(FieldVector vector, FieldVector deltaVector) {
+    final int valueCount = vector.getValueCount();
+    final int deltaValueCount = deltaVector.getValueCount();
+    final TransferPair transferPair = deltaVector.makeTransferPair(vector);
+    for (int i = 0; i < deltaValueCount; i++) {
+      transferPair.copyValueSafe(i, valueCount + i);
+    }
+    deltaVector.close();
+    vector.setValueCount(valueCount + deltaValueCount);
+  }
 }

java/vector/src/main/java/org/apache/arrow/vector/ipc/message/ArrowDictionaryBatch.java

@@ -30,10 +30,24 @@ public class ArrowDictionaryBatch implements ArrowMessage {
 
   private final long dictionaryId;
   private final ArrowRecordBatch dictionary;
+  private final boolean isDelta;
 
+  @Deprecated
   public ArrowDictionaryBatch(long dictionaryId, ArrowRecordBatch dictionary) {
+    this (dictionaryId, dictionary, false);
+  }
+
+  /**
+   * Constructs new instance.
+   */
+  public ArrowDictionaryBatch(long dictionaryId, ArrowRecordBatch dictionary, boolean isDelta) {
     this.dictionaryId = dictionaryId;
     this.dictionary = dictionary;
+    this.isDelta = isDelta;
+  }
+
+  public boolean isDelta() {
+    return isDelta;
   }
 
   public byte getMessageType() {
@@ -54,6 +68,7 @@ public int writeTo(FlatBufferBuilder builder) {
     DictionaryBatch.startDictionaryBatch(builder);
     DictionaryBatch.addId(builder, dictionaryId);
     DictionaryBatch.addData(builder, dataOffset);
+    DictionaryBatch.addIsDelta(builder, isDelta);
     return DictionaryBatch.endDictionaryBatch(builder);
   }
 

java/vector/src/main/java/org/apache/arrow/vector/ipc/message/MessageSerializer.java

@@ -491,7 +491,7 @@ public static ArrowDictionaryBatch deserializeDictionaryBatch(Message message, A
       throws IOException {
     DictionaryBatch dictionaryBatchFB = (DictionaryBatch) message.header(new DictionaryBatch());
     ArrowRecordBatch recordBatch = deserializeRecordBatch(dictionaryBatchFB.data(), bodyBuffer);
-    return new ArrowDictionaryBatch(dictionaryBatchFB.id(), recordBatch);
+    return new ArrowDictionaryBatch(dictionaryBatchFB.id(), recordBatch, dictionaryBatchFB.isDelta());
   }
 
   /**
@@ -570,7 +570,7 @@ public static ArrowDictionaryBatch deserializeDictionaryBatch(
     final ArrowBuf body = buffer.slice(block.getMetadataLength(),
         (int) totalLen - block.getMetadataLength());
     ArrowRecordBatch recordBatch = deserializeRecordBatch(dictionaryBatchFB.data(), body);
-    return new ArrowDictionaryBatch(dictionaryBatchFB.id(), recordBatch);
+    return new ArrowDictionaryBatch(dictionaryBatchFB.id(), recordBatch, dictionaryBatchFB.isDelta());
   }
 
   /**