apacheGH-38242: [Java] Fix incorrect internal struct accounting for D…

…UV getBufferSizeFor (apache#38242)

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

@@ -714,8 +714,22 @@ public int getBufferSizeFor(final int count) {
     if (count == 0) {
       return 0;
     }
+
+    int[] legCounts = new int[typeFields.length];
+    for (int i = 0; i < count; i++) {
+      byte tid = getTypeId(i);
+      legCounts[tid]++;
+    }
+
+    int legBytes = 0;
+    byte tid = 0;
+    for (ValueVector v : internalStruct) {
+      legBytes += v.getBufferSizeFor(legCounts[tid]);
+      tid++;
+    }
+
     return (int) (count * TYPE_WIDTH + (long) count * OFFSET_WIDTH
-        + DataSizeRoundingUtil.divideBy8Ceil(count) + internalStruct.getBufferSizeFor(count));
+        + DataSizeRoundingUtil.divideBy8Ceil(count) + legBytes);
   }
 
   @Override

java/vector/src/test/java/org/apache/arrow/vector/complex/TestDUVBufferSize.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.complex;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BaseValueVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.holders.NullableIntHolder;
+import org.apache.arrow.vector.holders.NullableVarBinaryHolder;
+import org.apache.arrow.vector.types.UnionMode;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.util.DataSizeRoundingUtil;
+import org.junit.jupiter.api.Test;
+
+public class TestDUVBufferSize {
+  @Test
+  public void testBufferSize() {
+    try (BufferAllocator allocator = new RootAllocator();
+         DenseUnionVector duv = new DenseUnionVector("duv", allocator,
+                 FieldType.nullable(new ArrowType.Union(UnionMode.Dense, null)), null)) {
+
+      List<Field> fields = Arrays.asList(
+              new Field("a", FieldType.notNullable(new ArrowType.Int(32, true)), null),
+              new Field("b", FieldType.notNullable(new ArrowType.Binary()), null)
+      );
+
+      duv.initializeChildrenFromFields(fields);
+
+      NullableIntHolder nih = new NullableIntHolder();
+      NullableVarBinaryHolder nvbh = new NullableVarBinaryHolder();
+
+      int ac = BaseValueVector.INITIAL_VALUE_ALLOCATION + 1;
+      for (int i = 0; i < ac; i++) {
+        duv.setTypeId(i, (byte) 0);
+        duv.setSafe(i, nih);
+      }
+
+      int bc = 1;
+      for (int i = 0; i < bc; i++) {
+        duv.setTypeId(i + ac, (byte) 1);
+        duv.setSafe(i + ac, nvbh);
+      }
+
+      int count = ac + bc;
+      duv.setValueCount(count);
+
+      // will not necessarily see an error unless bounds checking is on.
+      assertDoesNotThrow(duv::getBufferSize);
+
+      IntVector iv = duv.getIntVector((byte) 0);
+      VarBinaryVector vbv = duv.getVarBinaryVector((byte) 1);
+
+      long overhead = count * DenseUnionVector.TYPE_WIDTH +
+              (long) count * DenseUnionVector.OFFSET_WIDTH +
+              DataSizeRoundingUtil.divideBy8Ceil(count);
+
+      assertEquals(overhead + iv.getBufferSize() + vbv.getBufferSize(), duv.getBufferSize());
+    }
+  }
+}