content: /*
   * 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;
  
  import static java.util.Arrays.asList;
  import static org.junit.jupiter.api.Assertions.assertArrayEquals;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import java.nio.charset.StandardCharsets;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import org.apache.arrow.memory.BufferAllocator;
  import org.apache.arrow.memory.RootAllocator;
  import org.apache.arrow.vector.complex.DenseUnionVector;
  import org.apache.arrow.vector.complex.FixedSizeListVector;
  import org.apache.arrow.vector.complex.LargeListViewVector;
  import org.apache.arrow.vector.complex.ListVector;
  import org.apache.arrow.vector.complex.MapVector;
  import org.apache.arrow.vector.complex.StructVector;
  import org.apache.arrow.vector.complex.UnionVector;
  import org.apache.arrow.vector.types.pojo.ArrowType;
  import org.apache.arrow.vector.types.pojo.ArrowType.Struct;
  import org.apache.arrow.vector.types.pojo.Field;
  import org.apache.arrow.vector.types.pojo.FieldType;
  import org.apache.arrow.vector.util.TransferPair;
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  
  public class TestSplitAndTransfer {
    private BufferAllocator allocator;
  
    @BeforeEach
    public void init() {
      allocator = new RootAllocator(Long.MAX_VALUE);
    }
  
    @AfterEach
    public void terminate() throws Exception {
      allocator.close();
    }
  
    private void populateVarcharVector(
        final VarCharVector vector, int valueCount, String[] compareArray) {
      for (int i = 0; i < valueCount; i += 3) {
        final String s = String.format("%010d", i);
        vector.set(i, s.getBytes(StandardCharsets.UTF_8));
        if (compareArray != null) {
          compareArray[i] = s;
        }
      }
      vector.setValueCount(valueCount);
    }
  
    private void populateBaseVariableWidthViewVector(
        final BaseVariableWidthViewVector vector, int valueCount, String[] compareArray) {
      for (int i = 0; i < valueCount; i += 3) {
        final String s = String.format("%010d", i);
        vector.set(i, s.getBytes(StandardCharsets.UTF_8));
        if (compareArray != null) {
          compareArray[i] = s;
        }
      }
      vector.setValueCount(valueCount);
    }
  
    private void populateIntVector(final IntVector vector, int valueCount) {
      for (int i = 0; i < valueCount; i++) {
        vector.set(i, i);
      }
      vector.setValueCount(valueCount);
    }
  
    private void populateDenseUnionVector(final DenseUnionVector vector, int valueCount) {
      VarCharVector varCharVector =
          vector.addOrGet("varchar", FieldType.nullable(new ArrowType.Utf8()), VarCharVector.class);
      BigIntVector intVector =
          vector.addOrGet("int", FieldType.nullable(new ArrowType.Int(64, true)), BigIntVector.class);
  
      for (int i = 0; i < valueCount; i++) {
        vector.setTypeId(i, (byte) (i % 2));
        if (i % 2 == 0) {
          final String s = String.format("%010d", i);
          varCharVector.setSafe(i / 2, s.getBytes(StandardCharsets.UTF_8));
        } else {
          intVector.setSafe(i / 2, i);
        }
      }
      vector.setValueCount(valueCount);
    }
  
    @Test
    public void testWithEmptyVector() {
      // MapVector use TransferImpl from ListVector
      ListVector listVector = ListVector.empty("", allocator);
      TransferPair transferPair = listVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // BaseFixedWidthVector
      IntVector intVector = new IntVector("", allocator);
      transferPair = intVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // BaseVariableWidthVector
      VarCharVector varCharVector = new VarCharVector("", allocator);
      transferPair = varCharVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // BaseVariableWidthViewVector: ViewVarCharVector
      ViewVarCharVector viewVarCharVector = new ViewVarCharVector("", allocator);
      transferPair = viewVarCharVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // BaseVariableWidthVector: ViewVarBinaryVector
      ViewVarBinaryVector viewVarBinaryVector = new ViewVarBinaryVector("", allocator);
      transferPair = viewVarBinaryVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // BaseLargeVariableWidthVector
      LargeVarCharVector largeVarCharVector = new LargeVarCharVector("", allocator);
      transferPair = largeVarCharVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // StructVector
      StructVector structVector = StructVector.empty("", allocator);
      transferPair = structVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // FixedSizeListVector
      FixedSizeListVector fixedSizeListVector = FixedSizeListVector.empty("", 0, allocator);
      transferPair = fixedSizeListVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // FixedSizeBinaryVector
      FixedSizeBinaryVector fixedSizeBinaryVector = new FixedSizeBinaryVector("", allocator, 4);
      transferPair = fixedSizeBinaryVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // UnionVector
      UnionVector unionVector = UnionVector.empty("", allocator);
      transferPair = unionVector.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
      // DenseUnionVector
      DenseUnionVector duv = DenseUnionVector.empty("", allocator);
      transferPair = duv.getTransferPair(allocator);
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, transferPair.getTo().getValueCount());
  
      // non empty from vector
  
      // BaseFixedWidthVector
      IntVector fromIntVector = new IntVector("", allocator);
      fromIntVector.allocateNew(100);
      populateIntVector(fromIntVector, 100);
      transferPair = fromIntVector.getTransferPair(allocator);
      IntVector toIntVector = (IntVector) transferPair.getTo();
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, toIntVector.getValueCount());
  
      transferPair.splitAndTransfer(50, 0);
      assertEquals(0, toIntVector.getValueCount());
  
      transferPair.splitAndTransfer(100, 0);
      assertEquals(0, toIntVector.getValueCount());
      fromIntVector.clear();
      toIntVector.clear();
  
      // DenseUnionVector
      DenseUnionVector fromDuv = DenseUnionVector.empty("", allocator);
      populateDenseUnionVector(fromDuv, 100);
      transferPair = fromDuv.getTransferPair(allocator);
      DenseUnionVector toDUV = (DenseUnionVector) transferPair.getTo();
      transferPair.splitAndTransfer(0, 0);
      assertEquals(0, toDUV.getValueCount());
  
      transferPair.splitAndTransfer(50, 0);
      assertEquals(0, toDUV.getValueCount());
  
      transferPair.splitAndTransfer(100, 0);
      assertEquals(0, toDUV.getValueCount());
      fromDuv.clear();
      toDUV.clear();
    }
  
    @Test
    public void testWithNullVector() {
      int valueCount = 123;
      int startIndex = 10;
      NullVector fromNullVector = new NullVector("nullVector");
      fromNullVector.setValueCount(valueCount);
      TransferPair transferPair = fromNullVector.getTransferPair(fromNullVector.getAllocator());
      transferPair.splitAndTransfer(startIndex, valueCount - startIndex);
      NullVector toNullVector = (NullVector) transferPair.getTo();
  
      assertEquals(valueCount - startIndex, toNullVector.getValueCount());
      // no allocations to clear for NullVector
    }
  
    @Test
    public void testWithZeroVector() {
      ZeroVector fromZeroVector = new ZeroVector("zeroVector");
      TransferPair transferPair = fromZeroVector.getTransferPair(fromZeroVector.getAllocator());
      transferPair.splitAndTransfer(0, 0);
      ZeroVector toZeroVector = (ZeroVector) transferPair.getTo();
  
      assertEquals(0, toZeroVector.getValueCount());
      // no allocations to clear for ZeroVector
    }
  
    @Test
    public void testListVectorWithEmptyMapVector() {
      // List<element: Map(false)<entries: Struct<key: Utf8 not null, value: Utf8> not null>>
      int valueCount = 1;
      List<Field> children = new ArrayList<>();
      children.add(new Field("key", FieldType.notNullable(new ArrowType.Utf8()), null));
      children.add(new Field("value", FieldType.nullable(new ArrowType.Utf8()), null));
      Field structField =
          new Field("entries", FieldType.notNullable(ArrowType.Struct.INSTANCE), children);
  
      Field mapField =
          new Field("element", FieldType.notNullable(new ArrowType.Map(false)), asList(structField));
  
      Field listField = new Field("list", FieldType.nullable(new ArrowType.List()), asList(mapField));
  
      ListVector fromListVector = (ListVector) listField.createVector(allocator);
      fromListVector.allocateNew();
      fromListVector.setValueCount(valueCount);
  
      // child vector is empty
      MapVector dataVector = (MapVector) fromListVector.getDataVector();
      dataVector.allocateNew();
      // unset capacity to mimic observed failure mode
      dataVector.getOffsetBuffer().capacity(0);
  
      TransferPair transferPair = fromListVector.getTransferPair(fromListVector.getAllocator());
      transferPair.splitAndTransfer(0, valueCount);
      ListVector toListVector = (ListVector) transferPair.getTo();
  
      assertEquals(valueCount, toListVector.getValueCount());
      fromListVector.clear();
      toListVector.clear();
    }
  
    @Test /* VarCharVector */
    public void test() throws Exception {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator)) {
        varCharVector.allocateNew(10000, 1000);
  
        final int valueCount = 500;
        final String[] compareArray = new String[valueCount];
  
        populateVarcharVector(varCharVector, valueCount, compareArray);
  
        final TransferPair tp = varCharVector.getTransferPair(allocator);
        final VarCharVector newVarCharVector = (VarCharVector) tp.getTo();
        final int[][] startLengths = {{0, 201}, {201, 0}, {201, 200}, {401, 99}};
  
        for (final int[] startLength : startLengths) {
          final int start = startLength[0];
          final int length = startLength[1];
          tp.splitAndTransfer(start, length);
          for (int i = 0; i < length; i++) {
            final boolean expectedSet = ((start + i) % 3) == 0;
            if (expectedSet) {
              final byte[] expectedValue = compareArray[start + i].getBytes(StandardCharsets.UTF_8);
              assertFalse(newVarCharVector.isNull(i));
              assertArrayEquals(expectedValue, newVarCharVector.get(i));
            } else {
              assertTrue(newVarCharVector.isNull(i));
            }
          }
          newVarCharVector.clear();
        }
      }
    }
  
    private void testView(BaseVariableWidthViewVector vector) {
      vector.allocateNew(10000, 1000);
      final int valueCount = 500;
      final String[] compareArray = new String[valueCount];
  
      populateBaseVariableWidthViewVector(vector, valueCount, compareArray);
  
      final TransferPair tp = vector.getTransferPair(allocator);
      final BaseVariableWidthViewVector newVector = (BaseVariableWidthViewVector) tp.getTo();
      ;
      final int[][] startLengths = {{0, 201}, {201, 0}, {201, 200}, {401, 99}};
  
      for (final int[] startLength : startLengths) {
        final int start = startLength[0];
        final int length = startLength[1];
        tp.splitAndTransfer(start, length);
        for (int i = 0; i < length; i++) {
          final boolean expectedSet = ((start + i) % 3) == 0;
          if (expectedSet) {
            final byte[] expectedValue = compareArray[start + i].getBytes(StandardCharsets.UTF_8);
            assertFalse(newVector.isNull(i));
            assertArrayEquals(expectedValue, newVector.get(i));
          } else {
            assertTrue(newVector.isNull(i));
          }
        }
        newVector.clear();
      }
    }
  
    @Test
    public void testUtf8View() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        testView(viewVarCharVector);
      }
    }
  
    @Test
    public void testBinaryView() throws Exception {
      try (final ViewVarBinaryVector viewVarBinaryVector =
          new ViewVarBinaryVector("myvector", allocator)) {
        testView(viewVarBinaryVector);
      }
    }
  
    @Test
    public void testMemoryConstrainedTransfer() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator)) {
        allocator.setLimit(32768); /* set limit of 32KB */
  
        varCharVector.allocateNew(10000, 1000);
  
        final int valueCount = 1000;
  
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.getTransferPair(allocator);
        final VarCharVector newVarCharVector = (VarCharVector) tp.getTo();
        final int[][] startLengths = {{0, 700}, {700, 299}};
  
        for (final int[] startLength : startLengths) {
          final int start = startLength[0];
          final int length = startLength[1];
          tp.splitAndTransfer(start, length);
          newVarCharVector.clear();
        }
      }
    }
  
    private void testMemoryConstrainedTransferInViews(BaseVariableWidthViewVector vector) {
      // Here we have the target vector being transferred with a long string
      // hence, the data buffer will be allocated.
      // The default data buffer allocation takes
      // BaseVariableWidthViewVector.INITIAL_VIEW_VALUE_ALLOCATION *
      // BaseVariableWidthViewVector.ELEMENT_SIZE
      // set limit = BaseVariableWidthViewVector.INITIAL_VIEW_VALUE_ALLOCATION *
      // BaseVariableWidthViewVector.ELEMENT_SIZE
      final int setLimit =
          BaseVariableWidthViewVector.INITIAL_VIEW_VALUE_ALLOCATION
              * BaseVariableWidthViewVector.ELEMENT_SIZE;
      allocator.setLimit(setLimit);
  
      vector.allocateNew(16000, 1000);
  
      final int valueCount = 1000;
  
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.getTransferPair(allocator);
      final BaseVariableWidthViewVector newVector = (BaseVariableWidthViewVector) tp.getTo();
  
      final int[][] startLengths = {{0, 700}, {700, 299}};
  
      for (final int[] startLength : startLengths) {
        final int start = startLength[0];
        final int length = startLength[1];
        tp.splitAndTransfer(start, length);
        newVector.clear();
      }
    }
  
    @Test
    public void testMemoryConstrainedTransferInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        testMemoryConstrainedTransferInViews(viewVarCharVector);
      }
    }
  
    @Test
    public void testMemoryConstrainedTransferInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
          new ViewVarBinaryVector("myvector", allocator)) {
        testMemoryConstrainedTransferInViews(viewVarBinaryVector);
      }
    }
  
    @Test
    public void testTransfer() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator)) {
        varCharVector.allocateNew(10000, 1000);
  
        final int valueCount = 500;
        final String[] compareArray = new String[valueCount];
        populateVarcharVector(varCharVector, valueCount, compareArray);
  
        final TransferPair tp = varCharVector.getTransferPair(allocator);
        final VarCharVector newVarCharVector = (VarCharVector) tp.getTo();
        tp.transfer();
  
        assertEquals(0, varCharVector.valueCount);
        assertEquals(valueCount, newVarCharVector.valueCount);
  
        for (int i = 0; i < valueCount; i++) {
          final boolean expectedSet = (i % 3) == 0;
          if (expectedSet) {
            final byte[] expectedValue = compareArray[i].getBytes(StandardCharsets.UTF_8);
            assertFalse(newVarCharVector.isNull(i));
            assertArrayEquals(expectedValue, newVarCharVector.get(i));
          } else {
            assertTrue(newVarCharVector.isNull(i));
          }
        }
  
        newVarCharVector.clear();
      }
    }
  
    private void testTransferInViews(BaseVariableWidthViewVector vector) {
      vector.allocateNew(16000, 1000);
  
      final int valueCount = 500;
      final String[] compareArray = new String[valueCount];
      populateBaseVariableWidthViewVector(vector, valueCount, compareArray);
  
      final TransferPair tp = vector.getTransferPair(allocator);
      final BaseVariableWidthViewVector newVector = (BaseVariableWidthViewVector) tp.getTo();
      tp.transfer();
  
      assertEquals(0, vector.valueCount);
      assertEquals(valueCount, newVector.valueCount);
  
      for (int i = 0; i < valueCount; i++) {
        final boolean expectedSet = (i % 3) == 0;
        if (expectedSet) {
          final byte[] expectedValue = compareArray[i].getBytes(StandardCharsets.UTF_8);
          assertFalse(newVector.isNull(i));
          assertArrayEquals(expectedValue, newVector.get(i));
        } else {
          assertTrue(newVector.isNull(i));
        }
      }
  
      newVector.clear();
    }
  
    @Test
    public void testTransferInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        testTransferInViews(viewVarCharVector);
      }
    }
  
    @Test
    public void testTransferInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
          new ViewVarBinaryVector("myvector", allocator)) {
        testTransferInViews(viewVarBinaryVector);
      }
    }
  
    @Test
    public void testCopyValueSafe() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator);
          final VarCharVector newVarCharVector = new VarCharVector("newvector", allocator)) {
        varCharVector.allocateNew(10000, 1000);
  
        final int valueCount = 500;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.makeTransferPair(newVarCharVector);
  
        // new vector memory is not pre-allocated, we expect copyValueSafe work fine.
        for (int i = 0; i < valueCount; i++) {
          tp.copyValueSafe(i, i);
        }
        newVarCharVector.setValueCount(valueCount);
  
        for (int i = 0; i < valueCount; i++) {
          final boolean expectedSet = (i % 3) == 0;
          if (expectedSet) {
            assertFalse(varCharVector.isNull(i));
            assertFalse(newVarCharVector.isNull(i));
            assertArrayEquals(varCharVector.get(i), newVarCharVector.get(i));
          } else {
            assertTrue(newVarCharVector.isNull(i));
          }
        }
      }
    }
  
    @Test
    public void testSplitAndTransferNon() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator)) {
  
        varCharVector.allocateNew(10000, 1000);
        final int valueCount = 500;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.getTransferPair(allocator);
        VarCharVector newVarCharVector = (VarCharVector) tp.getTo();
  
        tp.splitAndTransfer(0, 0);
        assertEquals(0, newVarCharVector.getValueCount());
  
        newVarCharVector.clear();
      }
    }
  
    private void testSplitAndTransferNonInViews(BaseVariableWidthViewVector vector) {
      vector.allocateNew(16000, 1000);
      final int valueCount = 500;
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.getTransferPair(allocator);
      BaseVariableWidthViewVector newVector = (BaseVariableWidthViewVector) tp.getTo();
  
      tp.splitAndTransfer(0, 0);
      assertEquals(0, newVector.getValueCount());
  
      newVector.clear();
    }
  
    @Test
    public void testSplitAndTransferNonInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        testSplitAndTransferNonInViews(viewVarCharVector);
      }
    }
  
    @Test
    public void testSplitAndTransferNonInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
          new ViewVarBinaryVector("myvector", allocator)) {
        testSplitAndTransferNonInViews(viewVarBinaryVector);
      }
    }
  
    @Test
    public void testSplitAndTransferAll() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator)) {
  
        varCharVector.allocateNew(10000, 1000);
        final int valueCount = 500;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.getTransferPair(allocator);
        VarCharVector newVarCharVector = (VarCharVector) tp.getTo();
  
        tp.splitAndTransfer(0, valueCount);
        assertEquals(valueCount, newVarCharVector.getValueCount());
  
        newVarCharVector.clear();
      }
    }
  
    private void testSplitAndTransferAllInViews(BaseVariableWidthViewVector vector) {
      vector.allocateNew(16000, 1000);
      final int valueCount = 500;
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.getTransferPair(allocator);
      BaseVariableWidthViewVector newViewVarCharVector = (BaseVariableWidthViewVector) tp.getTo();
  
      tp.splitAndTransfer(0, valueCount);
      assertEquals(valueCount, newViewVarCharVector.getValueCount());
  
      newViewVarCharVector.clear();
    }
  
    @Test
    public void testSplitAndTransferAllInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        testSplitAndTransferAllInViews(viewVarCharVector);
      }
    }
  
    @Test
    public void testSplitAndTransferAllInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
          new ViewVarBinaryVector("myvector", allocator)) {
        testSplitAndTransferAllInViews(viewVarBinaryVector);
      }
    }
  
    @Test
    public void testInvalidStartIndex() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator);
          final VarCharVector newVarCharVector = new VarCharVector("newvector", allocator)) {
  
        varCharVector.allocateNew(10000, 1000);
        final int valueCount = 500;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.makeTransferPair(newVarCharVector);
  
        IllegalArgumentException e =
            assertThrows(IllegalArgumentException.class, () -> tp.splitAndTransfer(valueCount, 10));
  
        assertEquals(
            "Invalid parameters startIndex: 500, length: 10 for valueCount: 500", e.getMessage());
  
        newVarCharVector.clear();
      }
    }
  
    private void testInvalidStartIndexInViews(
        BaseVariableWidthViewVector vector, BaseVariableWidthViewVector newVector) {
      vector.allocateNew(16000, 1000);
      final int valueCount = 500;
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.makeTransferPair(newVector);
  
      IllegalArgumentException e =
          assertThrows(IllegalArgumentException.class, () -> tp.splitAndTransfer(valueCount, 10));
  
      assertEquals(
          "Invalid parameters startIndex: 500, length: 10 for valueCount: 500", e.getMessage());
  
      newVector.clear();
    }
  
    @Test
    public void testInvalidStartIndexInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator);
          final ViewVarCharVector newViewVarCharVector =
              new ViewVarCharVector("newvector", allocator)) {
        testInvalidStartIndexInViews(viewVarCharVector, newViewVarCharVector);
      }
    }
  
    @Test
    public void testInvalidStartIndexInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
              new ViewVarBinaryVector("myvector", allocator);
          final ViewVarBinaryVector newViewVarBinaryVector =
              new ViewVarBinaryVector("newvector", allocator)) {
        testInvalidStartIndexInViews(viewVarBinaryVector, newViewVarBinaryVector);
      }
    }
  
    @Test
    public void testInvalidLength() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator);
          final VarCharVector newVarCharVector = new VarCharVector("newvector", allocator)) {
  
        varCharVector.allocateNew(10000, 1000);
        final int valueCount = 500;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.makeTransferPair(newVarCharVector);
  
        IllegalArgumentException e =
            assertThrows(
                IllegalArgumentException.class, () -> tp.splitAndTransfer(0, valueCount * 2));
  
        assertEquals(
            "Invalid parameters startIndex: 0, length: 1000 for valueCount: 500", e.getMessage());
  
        newVarCharVector.clear();
      }
    }
  
    private void testInvalidLengthInViews(
        BaseVariableWidthViewVector vector, BaseVariableWidthViewVector newVector) {
      vector.allocateNew(16000, 1000);
      final int valueCount = 500;
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.makeTransferPair(newVector);
  
      IllegalArgumentException e =
          assertThrows(IllegalArgumentException.class, () -> tp.splitAndTransfer(0, valueCount * 2));
  
      assertEquals(
          "Invalid parameters startIndex: 0, length: 1000 for valueCount: 500", e.getMessage());
  
      newVector.clear();
    }
  
    @Test
    public void testInvalidLengthInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator);
          final ViewVarCharVector newViewVarCharVector =
              new ViewVarCharVector("newvector", allocator)) {
        testInvalidLengthInViews(viewVarCharVector, newViewVarCharVector);
      }
    }
  
    @Test
    public void testInvalidLengthInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
              new ViewVarBinaryVector("myvector", allocator);
          final ViewVarBinaryVector newViewVarBinaryVector =
              new ViewVarBinaryVector("newvector", allocator)) {
        testInvalidLengthInViews(viewVarBinaryVector, newViewVarBinaryVector);
      }
    }
  
    @Test
    public void testZeroStartIndexAndLength() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator);
          final VarCharVector newVarCharVector = new VarCharVector("newvector", allocator)) {
  
        varCharVector.allocateNew(0, 0);
        final int valueCount = 0;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.makeTransferPair(newVarCharVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newVarCharVector.getValueCount());
  
        newVarCharVector.clear();
      }
    }
  
    private void testZeroStartIndexAndLengthInViews(
        BaseVariableWidthViewVector vector, BaseVariableWidthViewVector newVector) {
      vector.allocateNew(0, 0);
      final int valueCount = 0;
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.makeTransferPair(newVector);
  
      tp.splitAndTransfer(0, 0);
      assertEquals(valueCount, newVector.getValueCount());
  
      newVector.clear();
    }
  
    @Test
    public void testZeroStartIndexAndLengthInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator);
          final ViewVarCharVector newViewVarCharVector =
              new ViewVarCharVector("newvector", allocator)) {
        testZeroStartIndexAndLengthInViews(viewVarCharVector, newViewVarCharVector);
      }
    }
  
    @Test
    public void testZeroStartIndexAndLengthInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
              new ViewVarBinaryVector("myvector", allocator);
          final ViewVarBinaryVector newViewVarBinaryVector =
              new ViewVarBinaryVector("newvector", allocator)) {
        testZeroStartIndexAndLengthInViews(viewVarBinaryVector, newViewVarBinaryVector);
      }
    }
  
    @Test
    public void testZeroLength() {
      try (final VarCharVector varCharVector = new VarCharVector("myvector", allocator);
          final VarCharVector newVarCharVector = new VarCharVector("newvector", allocator)) {
  
        varCharVector.allocateNew(10000, 1000);
        final int valueCount = 500;
        populateVarcharVector(varCharVector, valueCount, null);
  
        final TransferPair tp = varCharVector.makeTransferPair(newVarCharVector);
  
        tp.splitAndTransfer(500, 0);
        assertEquals(0, newVarCharVector.getValueCount());
  
        newVarCharVector.clear();
      }
    }
  
    private void testZeroLengthInViews(
        BaseVariableWidthViewVector vector, BaseVariableWidthViewVector newVector) {
      vector.allocateNew(16000, 1000);
      final int valueCount = 500;
      populateBaseVariableWidthViewVector(vector, valueCount, null);
  
      final TransferPair tp = vector.makeTransferPair(newVector);
  
      tp.splitAndTransfer(500, 0);
      assertEquals(0, newVector.getValueCount());
  
      newVector.clear();
    }
  
    @Test
    public void testZeroLengthInUtf8Views() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator);
          final ViewVarCharVector newViewVarCharVector =
              new ViewVarCharVector("newvector", allocator)) {
        testZeroLengthInViews(viewVarCharVector, newViewVarCharVector);
      }
    }
  
    @Test
    public void testZeroLengthInBinaryViews() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
              new ViewVarBinaryVector("myvector", allocator);
          final ViewVarBinaryVector newViewVarBinaryVector =
              new ViewVarBinaryVector("newvector", allocator)) {
        testZeroLengthInViews(viewVarBinaryVector, newViewVarBinaryVector);
      }
    }
  
    @Test
    public void testUnionVectorZeroStartIndexAndLength() {
      try (final UnionVector unionVector = UnionVector.empty("myvector", allocator);
          final UnionVector newUnionVector = UnionVector.empty("newvector", allocator)) {
  
        unionVector.allocateNew();
        final int valueCount = 0;
        unionVector.setValueCount(valueCount);
  
        final TransferPair tp = unionVector.makeTransferPair(newUnionVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newUnionVector.getValueCount());
  
        newUnionVector.clear();
      }
    }
  
    @Test
    public void testFixedWidthVectorZeroStartIndexAndLength() {
      try (final IntVector intVector = new IntVector("myvector", allocator);
          final IntVector newIntVector = new IntVector("newvector", allocator)) {
  
        intVector.allocateNew(0);
        final int valueCount = 0;
        intVector.setValueCount(valueCount);
  
        final TransferPair tp = intVector.makeTransferPair(newIntVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newIntVector.getValueCount());
  
        newIntVector.clear();
      }
    }
  
    @Test
    public void testBitVectorZeroStartIndexAndLength() {
      try (final BitVector bitVector = new BitVector("myvector", allocator);
          final BitVector newBitVector = new BitVector("newvector", allocator)) {
  
        bitVector.allocateNew(0);
        final int valueCount = 0;
        bitVector.setValueCount(valueCount);
  
        final TransferPair tp = bitVector.makeTransferPair(newBitVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newBitVector.getValueCount());
  
        newBitVector.clear();
      }
    }
  
    @Test
    public void testFixedSizeListVectorZeroStartIndexAndLength() {
      try (final FixedSizeListVector listVector = FixedSizeListVector.empty("list", 4, allocator);
          final FixedSizeListVector newListVector =
              FixedSizeListVector.empty("newList", 4, allocator)) {
  
        listVector.allocateNew();
        final int valueCount = 0;
        listVector.setValueCount(valueCount);
  
        final TransferPair tp = listVector.makeTransferPair(newListVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newListVector.getValueCount());
  
        newListVector.clear();
      }
    }
  
    @Test
    public void testListVectorZeroStartIndexAndLength() {
      try (final ListVector listVector = ListVector.empty("list", allocator);
          final ListVector newListVector = ListVector.empty("newList", allocator)) {
  
        listVector.allocateNew();
        final int valueCount = 0;
        listVector.setValueCount(valueCount);
  
        final TransferPair tp = listVector.makeTransferPair(newListVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newListVector.getValueCount());
  
        newListVector.clear();
      }
    }
  
    @Test
    public void testLargeListViewVectorZeroStartIndexAndLength() {
      try (final LargeListViewVector listVector =
              LargeListViewVector.empty("largelistview", allocator);
          final LargeListViewVector newListVector = LargeListViewVector.empty("newList", allocator)) {
  
        listVector.allocateNew();
        final int valueCount = 0;
        listVector.setValueCount(valueCount);
  
        final TransferPair tp = listVector.makeTransferPair(newListVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newListVector.getValueCount());
  
        newListVector.clear();
      }
    }
  
    @Test
    public void testStructVectorZeroStartIndexAndLength() {
      Map<String, String> metadata = new HashMap<>();
      metadata.put("k1", "v1");
      FieldType type = new FieldType(true, Struct.INSTANCE, null, metadata);
      try (final StructVector structVector = new StructVector("structvec", allocator, type, null);
          final StructVector newStructVector =
              new StructVector("newStructvec", allocator, type, null)) {
  
        structVector.allocateNew();
        final int valueCount = 0;
        structVector.setValueCount(valueCount);
  
        final TransferPair tp = structVector.makeTransferPair(newStructVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newStructVector.getValueCount());
  
        newStructVector.clear();
      }
    }
  
    @Test
    public void testMapVectorZeroStartIndexAndLength() {
      Map<String, String> metadata = new HashMap<>();
      metadata.put("k1", "v1");
      FieldType type = new FieldType(true, new ArrowType.Map(false), null, metadata);
      try (final MapVector mapVector = new MapVector("mapVec", allocator, type, null);
          final MapVector newMapVector = new MapVector("newMapVec", allocator, type, null)) {
  
        mapVector.allocateNew();
        final int valueCount = 0;
        mapVector.setValueCount(valueCount);
  
        final TransferPair tp = mapVector.makeTransferPair(newMapVector);
  
        tp.splitAndTransfer(0, 0);
        assertEquals(valueCount, newMapVector.getValueCount());
  
        newMapVector.clear();
      }
    }
  }
  