ARROW-7467: [Java] ComplexCopier does incorrect copy for Map nullable…

… info

Related to [ARROW-7467](https://issues.apache.org/jira/browse/ARROW-7467).

The MapVector and its 'value' vector are nullable, and its structVector and 'key' vector are non-nullable.
However, the MapVector generated by ComplexCopier has all nullable fields which is not correct.

Closes #6094 from tianchen92/ARROW-7467 and squashes the following commits:

7606a78 <tianchen> use UnionMapWriter API
2b68466 <tianchen> remove useless code in struct case
7cbd9bb <tianchen> resolve comments
e38e180 <tianchen> fix test
e058842 <tianchen> ARROW-7467:  ComplexCopier does incorrect copy for Map nullable info

Authored-by: tianchen <niki.lj@alibaba-inc.com>
Signed-off-by: Bryan Cutler <cutlerb@gmail.com>

java/vector/src/main/codegen/templates/ComplexCopier.java

@@ -15,6 +15,10 @@
  * limitations under the License.
  */
 
+import org.apache.arrow.vector.complex.MapVector;
+import org.apache.arrow.vector.complex.impl.UnionMapReader;
+import org.apache.arrow.vector.complex.impl.UnionMapWriter;
+import org.apache.arrow.vector.complex.reader.FieldReader;
 import org.apache.arrow.vector.complex.writer.FieldWriter;
 
 <@pp.dropOutputFile />
@@ -48,7 +52,6 @@ private static void writeValue(FieldReader reader, FieldWriter writer) {
       switch (mt) {
 
       case LIST:
-      case MAP:
       case FIXED_SIZE_LIST:
         if (reader.isSet()) {
           writer.startList();
@@ -58,6 +61,25 @@ private static void writeValue(FieldReader reader, FieldWriter writer) {
           writer.endList();
         }
         break;
+      case MAP:
+        if (reader.isSet()) {
+          UnionMapWriter mapWriter = (UnionMapWriter) writer;
+          UnionMapReader mapReader = (UnionMapReader) reader;
+
+          mapWriter.startMap();
+          while (mapReader.next()) {
+            FieldReader structReader = reader.reader();
+            UnionMapWriter structWriter = (UnionMapWriter) writer.struct();
+            if (structReader.isSet()) {
+              mapWriter.startEntry();
+              writeValue(mapReader.key(), getStructWriterForReader(mapReader.key(), structWriter.key(), MapVector.KEY_NAME));
+              writeValue(mapReader.value(), getStructWriterForReader(mapReader.value(), structWriter.value(), MapVector.VALUE_NAME));
+              mapWriter.endEntry();
+            }
+          }
+          mapWriter.endMap();
+        }
+        break;
       case STRUCT:
         if (reader.isSet()) {
           writer.start();

java/vector/src/main/java/org/apache/arrow/vector/types/pojo/FieldType.java

@@ -114,7 +114,7 @@ public boolean equals(Object obj) {
     if (!(obj instanceof FieldType)) {
       return false;
     }
-    Field that = (Field) obj;
+    FieldType that = (FieldType) obj;
     return Objects.equals(this.isNullable(), that.isNullable()) &&
         Objects.equals(this.getType(), that.getType()) &&
         Objects.equals(this.getDictionary(), that.getDictionary()) &&

java/vector/src/test/java/org/apache/arrow/vector/complex/impl/TestComplexCopier.java

@@ -20,11 +20,8 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
-import java.util.function.BiFunction;
-
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.RootAllocator;
-import org.apache.arrow.vector.ValueVector;
 import org.apache.arrow.vector.compare.VectorEqualsVisitor;
 import org.apache.arrow.vector.complex.FixedSizeListVector;
 import org.apache.arrow.vector.complex.ListVector;
@@ -43,9 +40,6 @@ public class TestComplexCopier {
 
   private static final int COUNT = 100;
 
-  private static final BiFunction<ValueVector, ValueVector, Boolean> TYPE_COMPARATOR =
-      (v1, v2) -> v1.getField().getType().equals(v2.getField().getType());
-
   @Before
   public void init() {
     allocator = new RootAllocator(Long.MAX_VALUE);
@@ -58,8 +52,8 @@ public void terminate() throws Exception {
 
   @Test
   public void testCopyFixedSizeListVector() {
-    try (FixedSizeListVector from = FixedSizeListVector.empty("from", 3, allocator);
-         FixedSizeListVector to = FixedSizeListVector.empty("to", 3, allocator)) {
+    try (FixedSizeListVector from = FixedSizeListVector.empty("v", 3, allocator);
+         FixedSizeListVector to = FixedSizeListVector.empty("v", 3, allocator)) {
 
       from.addOrGetVector(FieldType.nullable(Types.MinorType.INT.getType()));
       to.addOrGetVector(FieldType.nullable(Types.MinorType.INT.getType()));
@@ -86,15 +80,15 @@ public void testCopyFixedSizeListVector() {
       }
 
       // validate equals
-      assertTrue(VectorEqualsVisitor.vectorEquals(from, to, TYPE_COMPARATOR));
+      assertTrue(VectorEqualsVisitor.vectorEquals(from, to));
 
     }
   }
 
   @Test
   public void testInvalidCopyFixedSizeListVector() {
-    try (FixedSizeListVector from = FixedSizeListVector.empty("from", 3, allocator);
-         FixedSizeListVector to = FixedSizeListVector.empty("to", 2, allocator)) {
+    try (FixedSizeListVector from = FixedSizeListVector.empty("v", 3, allocator);
+         FixedSizeListVector to = FixedSizeListVector.empty("v", 2, allocator)) {
 
       from.addOrGetVector(FieldType.nullable(Types.MinorType.INT.getType()));
       to.addOrGetVector(FieldType.nullable(Types.MinorType.INT.getType()));
@@ -122,8 +116,8 @@ public void testInvalidCopyFixedSizeListVector() {
 
   @Test
   public void testCopyMapVector() {
-    try (final MapVector from = MapVector.empty("from", allocator, false);
-         final MapVector to = MapVector.empty("to", allocator, false)) {
+    try (final MapVector from = MapVector.empty("v", allocator, false);
+         final MapVector to = MapVector.empty("v", allocator, false)) {
 
       from.allocateNew();
 
@@ -155,15 +149,14 @@ public void testCopyMapVector() {
       to.setValueCount(COUNT);
 
       // validate equals
-      assertTrue(VectorEqualsVisitor.vectorEquals(from, to, TYPE_COMPARATOR));
-
+      assertTrue(VectorEqualsVisitor.vectorEquals(from, to));
     }
   }
 
   @Test
   public void testCopyListVector() {
-    try (ListVector from = ListVector.empty("from", allocator);
-         ListVector to = ListVector.empty("to", allocator)) {
+    try (ListVector from = ListVector.empty("v", allocator);
+         ListVector to = ListVector.empty("v", allocator)) {
 
       UnionListWriter listWriter = from.getWriter();
       listWriter.allocate();
@@ -202,7 +195,7 @@ public void testCopyListVector() {
       to.setValueCount(COUNT);
 
       // validate equals
-      assertTrue(VectorEqualsVisitor.vectorEquals(from, to, TYPE_COMPARATOR));
+      assertTrue(VectorEqualsVisitor.vectorEquals(from, to));
 
     }
   }