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 org.apache.arrow.vector.BitVectorHelper.getValidityBufferSizeFromCount;
  import static org.apache.arrow.vector.TestUtils.newVector;
  import static org.apache.arrow.vector.TestUtils.newViewVarBinaryVector;
  import static org.apache.arrow.vector.TestUtils.newViewVarCharVector;
  import static org.apache.arrow.vector.testing.ValueVectorDataPopulator.setVector;
  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.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertNull;
  import static org.junit.jupiter.api.Assertions.assertSame;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Random;
  import java.util.function.BiConsumer;
  import java.util.function.Function;
  import java.util.stream.IntStream;
  import java.util.stream.Stream;
  import org.apache.arrow.memory.ArrowBuf;
  import org.apache.arrow.memory.BufferAllocator;
  import org.apache.arrow.memory.RootAllocator;
  import org.apache.arrow.memory.rounding.DefaultRoundingPolicy;
  import org.apache.arrow.memory.util.ArrowBufPointer;
  import org.apache.arrow.memory.util.CommonUtil;
  import org.apache.arrow.util.AutoCloseables;
  import org.apache.arrow.vector.holders.NullableViewVarBinaryHolder;
  import org.apache.arrow.vector.holders.NullableViewVarCharHolder;
  import org.apache.arrow.vector.holders.ValueHolder;
  import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
  import org.apache.arrow.vector.testing.ValueVectorDataPopulator;
  import org.apache.arrow.vector.types.Types;
  import org.apache.arrow.vector.types.pojo.Field;
  import org.apache.arrow.vector.types.pojo.Schema;
  import org.apache.arrow.vector.util.ReusableByteArray;
  import org.apache.arrow.vector.util.Text;
  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;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.Arguments;
  import org.junit.jupiter.params.provider.MethodSource;
  
  public class TestVariableWidthViewVector {
  
    // short string (length <= 12)
    private static final byte[] STR0 = "0123456".getBytes(StandardCharsets.UTF_8);
    // short string (length <= 12)
    private static final byte[] STR1 = "012345678912".getBytes(StandardCharsets.UTF_8);
    // long string (length > 12)
    private static final byte[] STR2 = "0123456789123".getBytes(StandardCharsets.UTF_8);
    // long string (length > 12)
    private static final byte[] STR3 = "01234567891234567".getBytes(StandardCharsets.UTF_8);
    // short string (length <= 12)
    private static final byte[] STR4 = "01234567".getBytes(StandardCharsets.UTF_8);
    // short string (length <= 12)
    private static final byte[] STR5 = "A1234A".getBytes(StandardCharsets.UTF_8);
    // short string (length <= 12)
    private static final byte[] STR6 = "B1234567B".getBytes(StandardCharsets.UTF_8);
    // long string (length > 12)
    private static final byte[] STR7 = "K01234567891234567K".getBytes(StandardCharsets.UTF_8);
    // long string (length > 12)
    private static final byte[] STR8 = "M012345678912345678M".getBytes(StandardCharsets.UTF_8);
    private static final String EMPTY_SCHEMA_PATH = "";
  
    private BufferAllocator allocator;
  
    private Random random;
  
    @BeforeEach
    public void prepare() {
      allocator = new RootAllocator(Integer.MAX_VALUE);
      random = new Random();
    }
  
    @AfterEach
    public void shutdown() {
      allocator.close();
    }
  
    public static void setBytes(int index, byte[] bytes, ViewVarCharVector vector) {
      BitVectorHelper.setBit(vector.validityBuffer, index);
      vector.setBytes(index, bytes, 0, bytes.length);
    }
  
    @Test
    public void testInlineAllocation() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        viewVarCharVector.allocateNew(48, 3);
        final int valueCount = 3;
        viewVarCharVector.set(0, STR0);
        viewVarCharVector.set(1, STR1);
        viewVarCharVector.set(2, STR4);
        viewVarCharVector.setValueCount(valueCount);
  
        byte[] view1 = viewVarCharVector.get(0);
        byte[] view2 = viewVarCharVector.get(1);
        byte[] view3 = viewVarCharVector.get(2);
  
        assertNotNull(view1);
        assertNotNull(view2);
        assertNotNull(view3);
  
        String str1 = new String(STR0, StandardCharsets.UTF_8);
        String str2 = new String(STR1, StandardCharsets.UTF_8);
        String str3 = new String(STR4, StandardCharsets.UTF_8);
  
        assertEquals(new String(view1, StandardCharsets.UTF_8), str1);
        assertEquals(new String(view2, StandardCharsets.UTF_8), str2);
        assertEquals(new String(view3, StandardCharsets.UTF_8), str3);
  
        assertTrue(viewVarCharVector.dataBuffers.isEmpty());
  
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(0)).getBuffer(),
                StandardCharsets.UTF_8),
            str1);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(1)).getBuffer(),
                StandardCharsets.UTF_8),
            str2);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(2)).getBuffer(),
                StandardCharsets.UTF_8),
            str3);
      }
    }
  
    @Test
    public void testDataBufferBasedAllocationInSameBuffer() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        viewVarCharVector.allocateNew(48, 4);
        final int valueCount = 4;
        String str4 = TestUtils.generateRandomString(34);
        viewVarCharVector.set(0, STR1);
        viewVarCharVector.set(1, STR2);
        viewVarCharVector.set(2, STR3);
        viewVarCharVector.set(3, str4.getBytes(StandardCharsets.UTF_8));
        viewVarCharVector.setValueCount(valueCount);
  
        byte[] view1 = viewVarCharVector.get(0);
        byte[] view2 = viewVarCharVector.get(1);
        byte[] view3 = viewVarCharVector.get(2);
        byte[] view4 = viewVarCharVector.get(3);
  
        assertNotNull(view1);
        assertNotNull(view2);
        assertNotNull(view3);
        assertNotNull(view4);
  
        String str1 = new String(STR1, StandardCharsets.UTF_8);
        String str2 = new String(STR2, StandardCharsets.UTF_8);
        String str3 = new String(STR3, StandardCharsets.UTF_8);
  
        assertEquals(new String(view1, StandardCharsets.UTF_8), str1);
        assertEquals(new String(view2, StandardCharsets.UTF_8), str2);
        assertEquals(new String(view3, StandardCharsets.UTF_8), str3);
        assertEquals(new String(view4, StandardCharsets.UTF_8), str4);
  
        assertEquals(1, viewVarCharVector.dataBuffers.size());
  
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(0)).getBuffer(),
                StandardCharsets.UTF_8),
            str1);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(1)).getBuffer(),
                StandardCharsets.UTF_8),
            str2);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(2)).getBuffer(),
                StandardCharsets.UTF_8),
            str3);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(3)).getBuffer(),
                StandardCharsets.UTF_8),
            str4);
      }
    }
  
    @Test
    public void testDataBufferBasedAllocationInOtherBuffer() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        viewVarCharVector.allocateNew(48, 4);
        final int valueCount = 4;
        String str4 = TestUtils.generateRandomString(35);
        viewVarCharVector.set(0, STR1);
        viewVarCharVector.set(1, STR2);
        viewVarCharVector.set(2, STR3);
        viewVarCharVector.set(3, str4.getBytes(StandardCharsets.UTF_8));
        viewVarCharVector.setValueCount(valueCount);
  
        byte[] view1 = viewVarCharVector.get(0);
        byte[] view2 = viewVarCharVector.get(1);
        byte[] view3 = viewVarCharVector.get(2);
        byte[] view4 = viewVarCharVector.get(3);
  
        assertNotNull(view1);
        assertNotNull(view2);
        assertNotNull(view3);
        assertNotNull(view4);
  
        String str1 = new String(STR1, StandardCharsets.UTF_8);
        String str2 = new String(STR2, StandardCharsets.UTF_8);
        String str3 = new String(STR3, StandardCharsets.UTF_8);
  
        assertEquals(new String(view1, StandardCharsets.UTF_8), str1);
        assertEquals(new String(view2, StandardCharsets.UTF_8), str2);
        assertEquals(new String(view3, StandardCharsets.UTF_8), str3);
        assertEquals(new String(view4, StandardCharsets.UTF_8), str4);
  
        assertEquals(2, viewVarCharVector.dataBuffers.size());
  
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(0)).getBuffer(),
                StandardCharsets.UTF_8),
            str1);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(1)).getBuffer(),
                StandardCharsets.UTF_8),
            str2);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(2)).getBuffer(),
                StandardCharsets.UTF_8),
            str3);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(3)).getBuffer(),
                StandardCharsets.UTF_8),
            str4);
      }
    }
  
    @Test
    public void testSetSafe() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        viewVarCharVector.allocateNew(1, 1);
        byte[] str6 = TestUtils.generateRandomString(40).getBytes();
        final List<byte[]> strings = List.of(STR0, STR1, STR2, STR3, STR4, STR5, str6);
  
        // set data to a position out of capacity index
        Map<Integer, byte[]> expected = new HashMap<>();
        for (byte[] string : strings) {
          int cap = viewVarCharVector.getValueCapacity();
          expected.put(cap, string);
          viewVarCharVector.setSafe(cap, string);
        }
        int nullIndex = viewVarCharVector.getValueCapacity();
        viewVarCharVector.setNull(nullIndex);
        int valueCount = nullIndex + 1;
        viewVarCharVector.setValueCount(valueCount);
        assertEquals(viewVarCharVector.getNullCount(), valueCount - strings.size());
  
        assertEquals(128, viewVarCharVector.getValueCapacity());
        assertEquals(2, viewVarCharVector.dataBuffers.size());
  
        for (int i = 0; i < viewVarCharVector.getValueCapacity(); i++) {
          if (expected.containsKey(i)) {
            assertArrayEquals(expected.get(i), viewVarCharVector.get(i));
          } else {
            assertNull(viewVarCharVector.get(i));
          }
        }
      }
    }
  
    @Test
    public void testMixedAllocation() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        viewVarCharVector.allocateNew(128, 6);
        final int valueCount = 6;
        String str4 = TestUtils.generateRandomString(35);
        String str6 = TestUtils.generateRandomString(40);
        viewVarCharVector.set(0, STR1);
        viewVarCharVector.set(1, STR2);
        viewVarCharVector.set(2, STR3);
        viewVarCharVector.set(3, str4.getBytes(StandardCharsets.UTF_8));
        viewVarCharVector.set(4, STR1);
        viewVarCharVector.set(5, str6.getBytes(StandardCharsets.UTF_8));
        viewVarCharVector.setValueCount(valueCount);
  
        byte[] view1 = viewVarCharVector.get(0);
        byte[] view2 = viewVarCharVector.get(1);
        byte[] view3 = viewVarCharVector.get(2);
        byte[] view4 = viewVarCharVector.get(3);
        byte[] view5 = viewVarCharVector.get(4);
        byte[] view6 = viewVarCharVector.get(5);
  
        assertNotNull(view1);
        assertNotNull(view2);
        assertNotNull(view3);
        assertNotNull(view4);
        assertNotNull(view5);
        assertNotNull(view6);
  
        String str1 = new String(STR1, StandardCharsets.UTF_8);
        String str2 = new String(STR2, StandardCharsets.UTF_8);
        String str3 = new String(STR3, StandardCharsets.UTF_8);
  
        assertEquals(new String(view1, StandardCharsets.UTF_8), str1);
        assertEquals(new String(view2, StandardCharsets.UTF_8), str2);
        assertEquals(new String(view3, StandardCharsets.UTF_8), str3);
        assertEquals(new String(view4, StandardCharsets.UTF_8), str4);
        assertEquals(new String(view5, StandardCharsets.UTF_8), str1);
        assertEquals(new String(view6, StandardCharsets.UTF_8), str6);
  
        assertEquals(1, viewVarCharVector.dataBuffers.size());
  
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(0)).getBuffer(),
                StandardCharsets.UTF_8),
            str1);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(1)).getBuffer(),
                StandardCharsets.UTF_8),
            str2);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(2)).getBuffer(),
                StandardCharsets.UTF_8),
            str3);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(3)).getBuffer(),
                StandardCharsets.UTF_8),
            str4);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(4)).getBuffer(),
                StandardCharsets.UTF_8),
            str1);
        assertEquals(
            new String(
                Objects.requireNonNull(viewVarCharVector.getObject(5)).getBuffer(),
                StandardCharsets.UTF_8),
            str6);
      }
    }
  
    @Test
    public void testSetNullableViewVarCharHolder() {
      try (final ViewVarCharVector viewVarCharVector = new ViewVarCharVector("myvector", allocator)) {
        viewVarCharVector.allocateNew(0, 0);
        final List<byte[]> strings = List.of(STR0, STR1, STR2, STR3, STR4, STR5);
  
        NullableViewVarCharHolder stringHolder = new NullableViewVarCharHolder();
  
        // set not null
        int size = strings.size();
        for (int i = 0; i < size; i++) {
          setAndCheck(viewVarCharVector, i, strings.get(i), stringHolder);
        }
  
        // set null
        setAndCheck(viewVarCharVector, 6, null, stringHolder);
  
        // copy by holder
        // len < 12
        copyAndCheck(viewVarCharVector, stringHolder, 0, 7);
        // len > 12
        copyAndCheck(viewVarCharVector, stringHolder, 2, 8);
        // null
        copyAndCheck(viewVarCharVector, stringHolder, 6, 9);
  
        // test overwrite
        for (int i = 0; i < size; i++) {
          setAndCheck(viewVarCharVector, i, strings.get(size - i - 1), stringHolder);
        }
  
        String longString = TestUtils.generateRandomString(128);
        setAndCheck(viewVarCharVector, 6, longString.getBytes(), stringHolder);
      }
    }
  
    @Test
    public void testSetNullableViewVarBinaryHolder() {
      try (final ViewVarBinaryVector viewVarBinaryVector =
          new ViewVarBinaryVector("myvector", allocator)) {
        viewVarBinaryVector.allocateNew(0, 0);
        final List<byte[]> strings = List.of(STR0, STR1, STR2, STR3, STR4, STR5);
  
        NullableViewVarBinaryHolder holder = new NullableViewVarBinaryHolder();
  
        // set not null
        int size = strings.size();
        for (int i = 0; i < size; i++) {
          setAndCheck(viewVarBinaryVector, i, strings.get(i), holder);
        }
  
        // set null
        setAndCheck(viewVarBinaryVector, 6, null, holder);
  
        // copy by holder
        // len < 12
        copyAndCheck(viewVarBinaryVector, holder, 0, 7);
        // len > 12
        copyAndCheck(viewVarBinaryVector, holder, 2, 8);
        // null
        copyAndCheck(viewVarBinaryVector, holder, 6, 9);
  
        // test overwrite
        for (int i = 0; i < size; i++) {
          setAndCheck(viewVarBinaryVector, i, strings.get(size - i - 1), holder);
        }
  
        String longString = TestUtils.generateRandomString(128);
        setAndCheck(viewVarBinaryVector, 6, longString.getBytes(), holder);
      }
    }
  
    private static void copyAndCheck(
        BaseVariableWidthViewVector vector, ValueHolder holder, int fromIndex, int toIndex) {
      if (vector instanceof ViewVarCharVector) {
        ViewVarCharVector viewVarCharVector = (ViewVarCharVector) vector;
        NullableViewVarCharHolder stringHolder = (NullableViewVarCharHolder) holder;
        viewVarCharVector.get(fromIndex, stringHolder);
        viewVarCharVector.setSafe(toIndex, stringHolder);
      }
  
      if (vector instanceof ViewVarBinaryVector) {
        ViewVarBinaryVector viewVarBinaryVector = (ViewVarBinaryVector) vector;
        NullableViewVarBinaryHolder binaryHolder = (NullableViewVarBinaryHolder) holder;
        viewVarBinaryVector.get(fromIndex, binaryHolder);
        viewVarBinaryVector.setSafe(toIndex, binaryHolder);
      }
  
      assertArrayEquals(vector.get(fromIndex), vector.get(toIndex));
    }
  
    private void setAndCheck(
        ViewVarCharVector vector, int index, byte[] str, NullableViewVarCharHolder stringHolder) {
      ArrowBuf buf = null;
      if (null == str) {
        stringHolder.isSet = 0;
      } else {
        buf = allocator.buffer(str.length);
        buf.setBytes(0, str);
        stringHolder.isSet = 1;
        stringHolder.start = 0;
        stringHolder.end = str.length;
        stringHolder.buffer = buf;
      }
      vector.setSafe(index, stringHolder);
  
      // verify results
      assertArrayEquals(str, vector.get(index));
      AutoCloseables.closeNoChecked(buf);
    }
  
    private void setAndCheck(
        ViewVarBinaryVector vector, int index, byte[] str, NullableViewVarBinaryHolder binaryHolder) {
      ArrowBuf buf = null;
      if (null == str) {
        binaryHolder.isSet = 0;
      } else {
        buf = allocator.buffer(str.length);
        buf.setBytes(0, str);
        binaryHolder.isSet = 1;
        binaryHolder.start = 0;
        binaryHolder.end = str.length;
        binaryHolder.buffer = buf;
      }
      vector.setSafe(index, binaryHolder);
  
      // verify results
      assertArrayEquals(str, vector.get(index));
      AutoCloseables.closeNoChecked(buf);
    }
  
    @Test
    public void testAllocationIndexOutOfBounds() {
      assertThrows(
          IndexOutOfBoundsException.class,
          () -> {
            try (final ViewVarCharVector viewVarCharVector =
                new ViewVarCharVector("myvector", allocator)) {
              viewVarCharVector.allocateNew(32, 3);
              final int valueCount = 3;
              viewVarCharVector.set(0, STR1);
              viewVarCharVector.set(1, STR2);
              viewVarCharVector.set(2, STR2);
              viewVarCharVector.setValueCount(valueCount);
            }
          });
    }
  
    @Test
    public void testSizeOfViewBufferElements() {
      try (final ViewVarCharVector vector = new ViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        int valueCount = 100;
        int currentSize = 0;
        vector.setInitialCapacity(valueCount);
        vector.allocateNew();
        vector.setValueCount(valueCount);
        for (int i = 0; i < valueCount; i++) {
          currentSize += i;
          vector.setSafe(i, new byte[i]);
        }
        assertEquals(currentSize, vector.sizeOfViewBufferElements());
      }
    }
  
    @Test
    public void testNullableVarType1() {
  
      // Create a new value vector for 1024 integers.
      try (final ViewVarCharVector vector = newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        vector.allocateNew(1024 * 10, 1024);
  
        vector.set(0, STR1);
        vector.set(1, STR2);
        vector.set(2, STR3);
        vector.setSafe(3, STR3, 1, STR3.length - 1);
        vector.setSafe(4, STR3, 2, STR3.length - 2);
        ByteBuffer str3ByteBuffer = ByteBuffer.wrap(STR3);
        vector.setSafe(5, str3ByteBuffer, 1, STR3.length - 1);
        vector.setSafe(6, str3ByteBuffer, 2, STR3.length - 2);
  
        // Set with convenience function
        Text txt = new Text("foo");
        vector.setSafe(7, txt.getBytes(), 0, (int) txt.getLength());
  
        // Check the sample strings.
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(Arrays.copyOfRange(STR3, 1, STR3.length), vector.get(3));
        assertArrayEquals(Arrays.copyOfRange(STR3, 2, STR3.length), vector.get(4));
        assertArrayEquals(Arrays.copyOfRange(STR3, 1, STR3.length), vector.get(5));
        assertArrayEquals(Arrays.copyOfRange(STR3, 2, STR3.length), vector.get(6));
  
        // Check returning a Text object
        assertEquals(txt, vector.getObject(7));
  
        // Ensure null value throws.
        assertNull(vector.get(8));
      }
    }
  
    @Test
    public void testGetTextRepeatedly() {
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        ValueVectorDataPopulator.setVector(vector, STR1, STR2);
        vector.setValueCount(2);
  
        /* check the vector output */
        Text text = new Text();
        vector.read(0, text);
        assertArrayEquals(STR1, text.getBytes());
        vector.read(1, text);
        assertArrayEquals(STR2, text.getBytes());
      }
    }
  
    @Test
    public void testNullableVarType2() {
      try (final ViewVarBinaryVector vector = newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
        vector.allocateNew(1024 * 10, 1024);
        vector.set(0, STR1);
        vector.set(1, STR2);
        vector.set(2, STR3);
        vector.setSafe(3, STR3, 1, STR3.length - 1);
        vector.setSafe(4, STR3, 2, STR3.length - 2);
        ByteBuffer str3ByteBuffer = ByteBuffer.wrap(STR3);
        vector.setSafe(5, str3ByteBuffer, 1, STR3.length - 1);
        vector.setSafe(6, str3ByteBuffer, 2, STR3.length - 2);
  
        // Check the sample strings.
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(Arrays.copyOfRange(STR3, 1, STR3.length), vector.get(3));
        assertArrayEquals(Arrays.copyOfRange(STR3, 2, STR3.length), vector.get(4));
        assertArrayEquals(Arrays.copyOfRange(STR3, 1, STR3.length), vector.get(5));
        assertArrayEquals(Arrays.copyOfRange(STR3, 2, STR3.length), vector.get(6));
  
        // Ensure null value throws.
        assertNull(vector.get(7));
      }
    }
  
    @Test
    public void testGetBytesRepeatedly() {
      try (ViewVarBinaryVector vector = new ViewVarBinaryVector("", allocator)) {
        vector.allocateNew(5, 1);
  
        final String str = "hello world!!!";
        final String str2 = "foo";
        vector.setSafe(0, str.getBytes(StandardCharsets.UTF_8));
        vector.setSafe(1, str2.getBytes(StandardCharsets.UTF_8));
  
        // verify results
        ReusableByteArray reusableByteArray = new ReusableByteArray();
        vector.read(0, reusableByteArray);
        assertArrayEquals(
            str.getBytes(StandardCharsets.UTF_8),
            Arrays.copyOfRange(
                reusableByteArray.getBuffer(), 0, (int) reusableByteArray.getLength()));
        byte[] oldBuffer = reusableByteArray.getBuffer();
  
        vector.read(1, reusableByteArray);
        assertArrayEquals(
            str2.getBytes(StandardCharsets.UTF_8),
            Arrays.copyOfRange(
                reusableByteArray.getBuffer(), 0, (int) reusableByteArray.getLength()));
  
        // There should not have been any reallocation since the newer value is smaller in length.
        assertSame(oldBuffer, reusableByteArray.getBuffer());
      }
    }
  
    @Test
    public void testReAllocVariableWidthViewVector() {
      try (final ViewVarCharVector vector =
          newVector(
              ViewVarCharVector.class, EMPTY_SCHEMA_PATH, Types.MinorType.VIEWVARCHAR, allocator)) {
        final int capacityLimit = 4095;
        final int overLimitIndex = 200;
        vector.setInitialCapacity(capacityLimit);
        vector.allocateNew();
  
        int initialCapacity = vector.getValueCapacity();
        assertTrue(initialCapacity >= capacityLimit);
  
        /* Put values in indexes that fall within the initial allocation */
        vector.setSafe(0, STR1, 0, STR1.length);
        vector.setSafe(initialCapacity - 1, STR2, 0, STR2.length);
  
        /* the set calls above should NOT have triggered a realloc */
        assertEquals(initialCapacity, vector.getValueCapacity());
  
        /* Now try to put values in space that falls beyond the initial allocation */
        vector.setSafe(initialCapacity + overLimitIndex, STR3, 0, STR3.length);
  
        /* Check valueCapacity is more than initial allocation */
        assertTrue(initialCapacity * 2 <= vector.getValueCapacity());
  
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(initialCapacity - 1));
        assertArrayEquals(STR3, vector.get(initialCapacity + overLimitIndex));
  
        // Set the valueCount to be more than valueCapacity of current allocation. This is possible
        // for ValueVectors
        // as we don't call setSafe for null values, but we do call setValueCount when the current
        // batch is processed.
        vector.setValueCount(vector.getValueCapacity() + overLimitIndex);
      }
    }
  
    @Test
    public void testSetSafeWithArrowBufNoExcessAllocs() {
      final int numValues = BaseVariableWidthViewVector.INITIAL_VALUE_ALLOCATION * 2;
      final byte[] valueBytes = "hello world!!!".getBytes(StandardCharsets.UTF_8);
      final int valueBytesLength = valueBytes.length;
      final int isSet = 1;
      try (final ViewVarCharVector fromVector =
              newVector(
                  ViewVarCharVector.class,
                  EMPTY_SCHEMA_PATH,
                  Types.MinorType.VIEWVARCHAR,
                  allocator);
          final ViewVarCharVector toVector =
              newVector(
                  ViewVarCharVector.class,
                  EMPTY_SCHEMA_PATH,
                  Types.MinorType.VIEWVARCHAR,
                  allocator)) {
        /*
         * Populate the `fromVector` with `numValues` with byte-arrays, each of size `valueBytesLength`.
         */
        fromVector.setInitialCapacity(numValues);
        fromVector.allocateNew();
        for (int i = 0; i < numValues; ++i) {
          fromVector.setSafe(i, valueBytes, 0 /*start*/, valueBytesLength);
        }
        fromVector.setValueCount(numValues);
        ArrowBuf fromDataBuffer = fromVector.getDataBuffer();
        assertTrue(numValues * valueBytesLength <= fromDataBuffer.capacity());
  
        /*
         * Copy the entries one-by-one from 'fromVector' to 'toVector', but use the setSafe with
         * ArrowBuf API (instead of setSafe with byte-array).
         */
        toVector.setInitialCapacity(numValues);
        toVector.allocateNew();
        for (int i = 0; i < numValues; i++) {
          int start = fromVector.getTotalValueLengthUpToIndex(i);
          // across variable
          // width implementations
          int end = fromVector.getTotalValueLengthUpToIndex(i + 1);
          toVector.setSafe(i, isSet, start, end, fromDataBuffer);
        }
  
        /*
         * Since the 'fromVector' and 'toVector' have the same initial capacity, and were populated
         * with the same varchar elements, the allocations and hence, the final capacity should be
         * the same.
         */
        assertEquals(fromDataBuffer.capacity(), toVector.getDataBuffer().capacity());
      }
    }
  
    @Test
    public void testSetLastSetUsage() {
      try (final ViewVarCharVector vector = new ViewVarCharVector("myvector", allocator)) {
        vector.allocateNew(1024 * 10, 1024);
  
        setBytes(0, STR1, vector);
        setBytes(1, STR2, vector);
        setBytes(2, STR3, vector);
        setBytes(3, STR4, vector);
  
        /* Check current lastSet */
        assertEquals(-1, vector.getLastSet());
  
        /* Check the vector output */
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(STR4, vector.get(3));
  
        /*
         * If we don't do setLastSe(3) before setValueCount(), then the latter will corrupt
         * the value vector by filling in all positions [0,valuecount-1] will empty byte arrays.
         * Run the test by commenting on the next line, and we should see incorrect vector output.
         */
        vector.setLastSet(3);
        vector.setValueCount(20);
  
        /* Check current lastSet */
        assertEquals(19, vector.getLastSet());
  
        /* Check the vector output again */
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(STR4, vector.get(3));
  
        assertEquals(0, vector.getValueLength(4));
        assertEquals(0, vector.getValueLength(5));
        assertEquals(0, vector.getValueLength(6));
        assertEquals(0, vector.getValueLength(7));
        assertEquals(0, vector.getValueLength(8));
        assertEquals(0, vector.getValueLength(9));
        assertEquals(0, vector.getValueLength(10));
        assertEquals(0, vector.getValueLength(11));
        assertEquals(0, vector.getValueLength(12));
        assertEquals(0, vector.getValueLength(13));
        assertEquals(0, vector.getValueLength(14));
        assertEquals(0, vector.getValueLength(15));
        assertEquals(0, vector.getValueLength(16));
        assertEquals(0, vector.getValueLength(17));
        assertEquals(0, vector.getValueLength(18));
        assertEquals(0, vector.getValueLength(19));
      }
    }
  
    @Test
    public void testFillEmptiesUsage() {
      try (final ViewVarCharVector vector = new ViewVarCharVector("myvector", allocator)) {
        vector.allocateNew(1024 * 10, 1024);
  
        setBytes(0, STR1, vector);
        setBytes(1, STR2, vector);
        setBytes(2, STR3, vector);
        setBytes(3, STR4, vector);
  
        /* Check current lastSet */
        assertEquals(-1, vector.getLastSet());
  
        /* Check the vector output */
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(STR4, vector.get(3));
  
        vector.setLastSet(3);
        /* fill empty byte arrays from index [4, 9] */
        vector.fillEmpties(10);
  
        /* Check current lastSet */
        assertEquals(9, vector.getLastSet());
  
        /* Check the vector output */
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(STR4, vector.get(3));
        assertEquals(0, vector.getValueLength(4));
        assertEquals(0, vector.getValueLength(5));
        assertEquals(0, vector.getValueLength(6));
        assertEquals(0, vector.getValueLength(7));
        assertEquals(0, vector.getValueLength(8));
        assertEquals(0, vector.getValueLength(9));
  
        setBytes(10, STR1, vector);
        setBytes(11, STR2, vector);
  
        vector.setLastSet(11);
        /* fill empty byte arrays from index [12, 14] */
        vector.setValueCount(15);
  
        /* Check current lastSet */
        assertEquals(14, vector.getLastSet());
  
        /* Check the vector output */
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(STR4, vector.get(3));
        assertEquals(0, vector.getValueLength(4));
        assertEquals(0, vector.getValueLength(5));
        assertEquals(0, vector.getValueLength(6));
        assertEquals(0, vector.getValueLength(7));
        assertEquals(0, vector.getValueLength(8));
        assertEquals(0, vector.getValueLength(9));
        assertArrayEquals(STR1, vector.get(10));
        assertArrayEquals(STR2, vector.get(11));
        assertEquals(0, vector.getValueLength(12));
        assertEquals(0, vector.getValueLength(13));
        assertEquals(0, vector.getValueLength(14));
      }
    }
  
    @Test
    public void testGetBufferAddress1() {
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
  
        setVector(vector, STR1, STR2, STR3, STR4);
        vector.setValueCount(15);
  
        /* check the vector output */
        assertArrayEquals(STR1, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR3, vector.get(2));
        assertArrayEquals(STR4, vector.get(3));
  
        List<ArrowBuf> buffers = vector.getFieldBuffers();
        long bitAddress = vector.getValidityBufferAddress();
        long dataAddress = vector.getDataBufferAddress();
  
        assertEquals(3, buffers.size());
        assertEquals(bitAddress, buffers.get(0).memoryAddress());
        assertEquals(dataAddress, buffers.get(1).memoryAddress());
      }
    }
  
    @Test
    public void testSetInitialCapacityInViews() {
      try (final ViewVarCharVector vector = new ViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
  
        /* use the default 16 data bytes on average per element */
        final int viewSize = BaseVariableWidthViewVector.ELEMENT_SIZE;
        int defaultCapacity = BaseVariableWidthViewVector.INITIAL_VIEW_VALUE_ALLOCATION / viewSize;
        vector.setInitialCapacity(defaultCapacity);
        vector.allocateNew();
        assertEquals(defaultCapacity, vector.getValueCapacity());
        assertEquals(
            CommonUtil.nextPowerOfTwo(defaultCapacity * viewSize), vector.getDataBuffer().capacity());
  
        double density = 4.0;
        final int valueCount = 5;
        vector.setInitialCapacity(valueCount, density);
        vector.allocateNew();
        assertEquals(8, vector.getValueCapacity());
        assertEquals(128, vector.getDataBuffer().capacity());
        int initialDataBufferSize = (int) (valueCount * density);
        // making sure a databuffer is allocated
        vector.set(4, "01234567890123456".getBytes(StandardCharsets.UTF_8));
        assertEquals(vector.dataBuffers.size(), 1);
        ArrowBuf dataBuf = vector.dataBuffers.get(0);
        try (ArrowBuf tempBuf = vector.allocator.buffer(initialDataBufferSize)) {
          // replicating a new buffer allocation process when a new buffer is added to the
          // data buffer when inserting an element with length > 12
          assertEquals(tempBuf.capacity(), dataBuf.capacity());
        }
      }
    }
  
    @Test
    public void testGetPointerVariableWidthViews() {
      final String[] sampleData =
          new String[] {
            "abc",
            "1234567890123",
            "def",
            null,
            "hello world java",
            "aaaaa",
            "world",
            "2019",
            null,
            "0717"
          };
  
      try (ViewVarCharVector vec1 = new ViewVarCharVector("vec1", allocator);
          ViewVarCharVector vec2 = new ViewVarCharVector("vec2", allocator)) {
  
        vec1.allocateNew((long) sampleData.length * 16, sampleData.length);
        vec2.allocateNew((long) sampleData.length * 16, sampleData.length);
  
        for (int i = 0; i < sampleData.length; i++) {
          String str = sampleData[i];
          if (str != null) {
            vec1.set(i, sampleData[i].getBytes(StandardCharsets.UTF_8));
            vec2.set(i, sampleData[i].getBytes(StandardCharsets.UTF_8));
          } else {
            vec1.setNull(i);
  
            vec2.setNull(i);
          }
        }
  
        ArrowBufPointer ptr1 = new ArrowBufPointer();
        ArrowBufPointer ptr2 = new ArrowBufPointer();
  
        for (int i = 0; i < sampleData.length; i++) {
          vec1.getDataPointer(i, ptr1);
          vec2.getDataPointer(i, ptr2);
  
          assertTrue(ptr1.equals(ptr2));
          assertTrue(ptr2.equals(ptr2));
        }
      }
    }
  
    @Test
    public void testGetNullFromVariableWidthViewVector() {
      try (final ViewVarCharVector varCharViewVector =
              new ViewVarCharVector("viewvarcharvec", allocator);
          final ViewVarBinaryVector varBinaryViewVector =
              new ViewVarBinaryVector("viewvarbinary", allocator)) {
        varCharViewVector.allocateNew(16, 1);
        varBinaryViewVector.allocateNew(16, 1);
  
        varCharViewVector.setNull(0);
        varBinaryViewVector.setNull(0);
  
        assertNull(varCharViewVector.get(0));
        assertNull(varBinaryViewVector.get(0));
      }
    }
  
    @Test
    public void testVariableWidthViewVectorNullHashCode() {
      try (ViewVarCharVector viewVarChar = new ViewVarCharVector("view var char vector", allocator)) {
        viewVarChar.allocateNew(100, 1);
        viewVarChar.setValueCount(1);
  
        viewVarChar.set(0, "abc".getBytes(StandardCharsets.UTF_8));
        viewVarChar.setNull(0);
  
        assertEquals(0, viewVarChar.hashCode(0));
      }
    }
  
    @Test
    public void testUnloadVariableWidthViewVector() {
      try (final ViewVarCharVector viewVarCharVector =
          new ViewVarCharVector("view var char", allocator)) {
        viewVarCharVector.allocateNew(16, 2);
        viewVarCharVector.setValueCount(2);
        viewVarCharVector.set(0, "abcd".getBytes(StandardCharsets.UTF_8));
  
        List<ArrowBuf> bufs = viewVarCharVector.getFieldBuffers();
        assertEquals(2, bufs.size());
  
        ArrowBuf viewBuf = bufs.get(1);
  
        assertEquals(32, viewBuf.writerIndex());
        final String longString = "012345678901234";
        viewVarCharVector.set(1, longString.getBytes(StandardCharsets.UTF_8));
  
        bufs = viewVarCharVector.getFieldBuffers();
        assertEquals(3, bufs.size());
  
        ArrowBuf referenceBuf = bufs.get(2);
        assertEquals(longString.length(), referenceBuf.writerIndex());
      }
    }
  
    @Test
    public void testUnSupportedOffSet() {
      // offset is not a feature required in ViewVarCharVector
      assertThrows(
          UnsupportedOperationException.class,
          () -> {
            try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
  
              setVector(vector, STR1, STR2);
              vector.setValueCount(2);
  
              /* check the vector output */
              assertArrayEquals(STR1, vector.get(0));
              assertArrayEquals(STR2, vector.get(1));
  
              vector.getOffsetBuffer();
            }
          });
    }
  
    private void validateViewBuffer(
        int index,
        ViewVarCharVector vector,
        byte[] expectedData,
        int expectedBufId,
        int expectedOffSet) {
      final ArrowBuf viewBuffer = vector.viewBuffer;
      int writePosition = index * BaseVariableWidthViewVector.ELEMENT_SIZE;
      final int prefixBufWidth = BaseVariableWidthViewVector.PREFIX_WIDTH;
      final int lengthBufWidth = BaseVariableWidthViewVector.LENGTH_WIDTH;
      int length = viewBuffer.getInt(writePosition);
  
      // validate length of the view
      assertEquals(expectedData.length, length);
  
      byte[] prefixBytes = new byte[prefixBufWidth];
      viewBuffer.getBytes(writePosition + lengthBufWidth, prefixBytes);
  
      // validate the prefix
      byte[] expectedPrefixBytes = new byte[prefixBufWidth];
      System.arraycopy(expectedData, 0, expectedPrefixBytes, 0, prefixBufWidth);
      assertArrayEquals(expectedPrefixBytes, prefixBytes);
  
      if (length > 12) {
        /// validate bufId
        int bufId = viewBuffer.getInt(writePosition + lengthBufWidth + prefixBufWidth);
        assertEquals(expectedBufId, bufId);
        // validate offset
        int offset =
            viewBuffer.getInt(
                writePosition
                    + lengthBufWidth
                    + prefixBufWidth
                    + BaseVariableWidthViewVector.BUF_INDEX_WIDTH);
        assertEquals(expectedOffSet, offset);
      }
      // validate retrieved data
      assertArrayEquals(expectedData, vector.get(index));
    }
  
    @Test
    public void testOverwriteShortFromLongString() {
      /*NA: not applicable */
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set short string
        vector.set(0, STR0);
        vector.setValueCount(1);
        assertEquals(0, vector.dataBuffers.size());
        assertArrayEquals(STR0, vector.get(0));
  
        validateViewBuffer(0, vector, STR0, /*NA*/ -1, /*NA*/ -1);
  
        // set long string
        vector.set(0, STR3);
        vector.setValueCount(1);
        assertEquals(1, vector.dataBuffers.size());
        assertArrayEquals(STR3, vector.get(0));
  
        validateViewBuffer(0, vector, STR3, 0, 0);
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all shorts.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(48, 3);
        // set short string 1
        vector.set(0, STR0);
        // set short string 2
        vector.set(1, STR5);
        // set short string 3
        vector.set(2, STR6);
        vector.setValueCount(3);
  
        // overwrite index 1 with a long string
        vector.set(1, STR7);
        vector.setValueCount(3);
  
        validateViewBuffer(0, vector, STR0, /*NA*/ -1, /*NA*/ -1);
        validateViewBuffer(1, vector, STR7, 0, 0);
        validateViewBuffer(2, vector, STR6, /*NA*/ -1, /*NA*/ -1);
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(80, 5);
        // set short string 1
        vector.set(0, STR0);
        // set long string 1
        vector.set(1, STR3);
        // set short string 2
        vector.set(2, STR5);
        // set short string 3
        vector.set(3, STR6);
        // set long string 2
        vector.set(4, STR7);
        vector.setValueCount(5);
  
        // overwrite index 2 with a long string
        vector.set(2, STR8);
        vector.setValueCount(5);
  
        validateViewBuffer(0, vector, STR0, /*NA*/ -1, /*NA*/ -1);
        validateViewBuffer(1, vector, STR3, 0, 0);
        // Since we did overwrite index 2 with STR8, and as we are using append-only approach,
        // it will be appended to the data buffer.
        // Thus, it will be stored in the dataBuffer in order i.e. [STR3, STR7, STR8].
        validateViewBuffer(2, vector, STR8, 0, STR3.length + STR7.length);
        validateViewBuffer(3, vector, STR6, /*NA*/ -1, /*NA*/ -1);
        validateViewBuffer(4, vector, STR7, 0, STR3.length);
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      // Here the short string is overwritten with a long string, and its length is larger than
      // the remaining capacity of the existing data buffer.
      // This would allocate a new buffer in the data buffers.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(80, 5);
        // set short string 1
        vector.set(0, STR0);
        // set long string 1
        vector.set(1, STR3);
        // set short string 2
        vector.set(2, STR5);
        // set short string 3
        vector.set(3, STR6);
        // set long string 2
        vector.set(4, STR7);
  
        vector.setValueCount(5);
  
        // overwrite index 2 with a long string
        String longString = TestUtils.generateRandomString(128);
        byte[] longStringBytes = longString.getBytes(StandardCharsets.UTF_8);
        // since the append-only approach is used and the remaining capacity
        // is not enough to store the new string; a new buffer will be allocated.
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        assertTrue(remainingCapacity < longStringBytes.length);
        vector.set(2, longStringBytes);
        vector.setValueCount(5);
  
        validateViewBuffer(0, vector, STR0, /*NA*/ -1, /*NA*/ -1);
        validateViewBuffer(1, vector, STR3, 0, 0);
        // overwritten long string will be stored in the new data buffer.
        validateViewBuffer(2, vector, longStringBytes, 1, 0);
        validateViewBuffer(3, vector, STR6, /*NA*/ -1, /*NA*/ -1);
        validateViewBuffer(4, vector, STR7, 0, STR3.length);
      }
    }
  
    @Test
    public void testOverwriteLongFromShortString() {
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set short string
        vector.set(0, STR3);
        vector.setValueCount(1);
        // set long string
        vector.set(0, STR0);
        vector.setValueCount(1);
  
        validateViewBuffer(0, vector, STR0, /*NA*/ -1, /*NA*/ -1);
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all longs.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(48, 3);
        // set long string 1
        vector.set(0, STR3);
        // set long string 2
        vector.set(1, STR8);
        // set long string 3
        vector.set(2, STR7);
        vector.setValueCount(3);
  
        // overwrite index 1 with a short string
        vector.set(1, STR6);
        vector.setValueCount(3);
  
        validateViewBuffer(0, vector, STR3, 0, 0);
        validateViewBuffer(1, vector, STR6, /*NA*/ -1, /*NA*/ -1);
        // since the append-only approach is used,
        // STR8 will still be in the first data buffer in dataBuffers.
        validateViewBuffer(2, vector, STR7, 0, STR3.length + STR8.length);
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(80, 5);
        // set long string 1
        vector.set(0, STR3);
        // set short string 1
        vector.set(1, STR5);
        // set long string 2
        vector.set(2, STR7);
        // set long string 3
        vector.set(3, STR8);
        // set short string 2
        vector.set(4, STR6);
        vector.setValueCount(5);
  
        // overwrite index 2 with a short string
        vector.set(2, STR0);
        vector.setValueCount(5);
  
        validateViewBuffer(0, vector, STR3, 0, 0);
        validateViewBuffer(1, vector, STR5, /*NA*/ -1, /*NA*/ -1);
        validateViewBuffer(2, vector, STR0, /*NA*/ -1, /*NA*/ -1);
        // since the append-only approach is used,
        // STR7 will still be in the first data buffer in dataBuffers.
        validateViewBuffer(3, vector, STR8, 0, STR3.length + STR7.length);
        validateViewBuffer(4, vector, STR6, /*NA*/ -1, /*NA*/ -1);
      }
    }
  
    @Test
    public void testOverwriteLongFromAShorterLongString() {
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set long string
        vector.set(0, STR7);
        vector.setValueCount(1);
        // set shorter long string, since append-only approach is used and the remaining capacity
        // is not enough to store the new string; a new buffer will be allocated.
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        assertTrue(remainingCapacity < STR3.length);
        // set shorter long string
        vector.set(0, STR3);
        vector.setValueCount(1);
  
        validateViewBuffer(0, vector, STR3, 1, 0);
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all longs.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        // extra memory is allocated
        vector.allocateNew(128, 3);
        // set long string 1
        vector.set(0, STR3);
        // set long string 2
        vector.set(1, STR8);
        // set long string 3
        vector.set(2, STR7);
        vector.setValueCount(3);
  
        // overwrite index 1 with a shorter long string
        // Since append-only approach is used
        // and the remaining capacity is enough to store in the same data buffer.;
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        assertTrue(remainingCapacity > STR2.length);
        vector.set(1, STR2);
        vector.setValueCount(3);
  
        validateViewBuffer(0, vector, STR3, 0, 0);
        // since the append-only approach is used,
        // STR8 will still be in the first data buffer in dataBuffers.
        validateViewBuffer(1, vector, STR2, 0, STR3.length + STR8.length + STR7.length);
        validateViewBuffer(2, vector, STR7, 0, STR3.length + STR8.length);
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(128, 5);
        // set long string 1
        vector.set(0, STR3);
        // set short string 1
        vector.set(1, STR5);
        // set long string 2
        vector.set(2, STR7);
        // set long string 3
        vector.set(3, STR8);
        // set short string 2
        vector.set(4, STR6);
        vector.setValueCount(5);
  
        // overwrite index 2 with a shorter long string
        // Since append-only approach is used
        // and the remaining capacity is enough to store in the same data buffer.;
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        assertTrue(remainingCapacity > STR2.length);
        vector.set(2, STR2);
        vector.setValueCount(5);
  
        validateViewBuffer(0, vector, STR3, 0, 0);
        validateViewBuffer(1, vector, STR5, /*NA*/ -1, /*NA*/ -1);
        // since the append-only approach is used,
        // STR7 will still be in the first data buffer in dataBuffers.
        validateViewBuffer(2, vector, STR2, 0, STR3.length + STR7.length + STR8.length);
        validateViewBuffer(3, vector, STR8, 0, STR3.length + STR7.length);
        validateViewBuffer(4, vector, STR6, /*NA*/ -1, /*NA*/ -1);
      }
    }
  
    @Test
    public void testOverwriteLongFromALongerLongString() {
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set long string
        vector.set(0, STR3);
        vector.setValueCount(1);
        // set longer long string, since append-only approach is used and the remaining capacity
        // is not enough to store the new string; a new buffer will be allocated.
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        assertTrue(remainingCapacity < STR7.length);
        // set longer long string
        vector.set(0, STR7);
        vector.setValueCount(1);
  
        validateViewBuffer(0, vector, STR7, 1, 0);
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all longs.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        // extra memory is allocated
        vector.allocateNew(48, 3);
        // set long string 1
        vector.set(0, STR3);
        // set long string 2
        vector.set(1, STR8);
        // set long string 3
        vector.set(2, STR7);
        vector.setValueCount(3);
  
        // overwrite index 1 with a longer long string
        // the remaining capacity is not enough to store in the same data buffer
        // since a new buffer is added to the dataBuffers
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        String longerString = TestUtils.generateRandomString(35);
        byte[] longerStringBytes = longerString.getBytes(StandardCharsets.UTF_8);
        assertTrue(remainingCapacity < longerStringBytes.length);
  
        vector.set(1, longerStringBytes);
        vector.setValueCount(3);
  
        validateViewBuffer(0, vector, STR3, 0, 0);
        validateViewBuffer(1, vector, longerStringBytes, 1, 0);
        // since the append-only approach is used,
        // STR8 will still be in the first data buffer in dataBuffers.
        validateViewBuffer(2, vector, STR7, 0, STR3.length + STR8.length);
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(128, 5);
        // set long string 1
        vector.set(0, STR3);
        // set short string 1
        vector.set(1, STR5);
        // set long string 2
        vector.set(2, STR7);
        // set long string 3
        vector.set(3, STR2);
        // set short string 2
        vector.set(4, STR6);
        vector.setValueCount(5);
  
        // overwrite index 2 with a longer long string
        // the remaining capacity is enough to store in the same data buffer
        final ArrowBuf currentDataBuf = vector.dataBuffers.get(0);
        final long remainingCapacity = currentDataBuf.capacity() - currentDataBuf.writerIndex();
        String longerString = TestUtils.generateRandomString(24);
        byte[] longerStringBytes = longerString.getBytes(StandardCharsets.UTF_8);
        assertTrue(remainingCapacity > longerStringBytes.length);
  
        vector.set(2, longerStringBytes);
        vector.setValueCount(5);
  
        validateViewBuffer(0, vector, STR3, 0, 0);
        validateViewBuffer(1, vector, STR5, /*NA*/ -1, /*NA*/ -1);
        // since the append-only approach is used,
        // STR7 will still be in the first data buffer in dataBuffers.
        validateViewBuffer(2, vector, longerStringBytes, 0, STR3.length + STR7.length + STR2.length);
        validateViewBuffer(3, vector, STR2, 0, STR3.length + STR7.length);
        validateViewBuffer(4, vector, STR6, /*NA*/ -1, /*NA*/ -1);
      }
    }
  
    @Test
    public void testSafeOverwriteShortFromLongString() {
      /*NA: not applicable */
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set short string
        vector.setSafe(0, STR0);
        vector.setValueCount(1);
        assertEquals(0, vector.dataBuffers.size());
        assertArrayEquals(STR0, vector.get(0));
  
        // set long string
        vector.setSafe(0, STR3);
        vector.setValueCount(1);
        assertEquals(1, vector.dataBuffers.size());
        assertArrayEquals(STR3, vector.get(0));
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all shorts.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 3);
        // set short string 1
        vector.setSafe(0, STR0);
        // set short string 2
        vector.setSafe(1, STR5);
        // set short string 3
        vector.setSafe(2, STR6);
        vector.setValueCount(3);
  
        // overwrite index 1 with a long string
        vector.setSafe(1, STR7);
        vector.setValueCount(3);
  
        assertArrayEquals(STR0, vector.get(0));
        assertArrayEquals(STR7, vector.get(1));
        assertArrayEquals(STR6, vector.get(2));
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 5);
        // set short string 1
        vector.setSafe(0, STR0);
        // set long string 1
        vector.setSafe(1, STR3);
        // set short string 2
        vector.setSafe(2, STR5);
        // set short string 3
        vector.setSafe(3, STR6);
        // set long string 2
        vector.setSafe(4, STR7);
        vector.setValueCount(5);
  
        // overwrite index 2 with a long string
        vector.setSafe(2, STR8);
        vector.setValueCount(5);
  
        assertArrayEquals(STR0, vector.get(0));
        assertArrayEquals(STR3, vector.get(1));
        assertArrayEquals(STR8, vector.get(2));
        assertArrayEquals(STR6, vector.get(3));
        assertArrayEquals(STR7, vector.get(4));
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 5);
        // set short string 1
        vector.setSafe(0, STR0);
        // set long string 1
        vector.setSafe(1, STR3);
        // set short string 2
        vector.setSafe(2, STR5);
        // set short string 3
        vector.setSafe(3, STR6);
        // set long string 2
        vector.setSafe(4, STR7);
  
        vector.setValueCount(5);
  
        // overwrite index 2 with a long string
        String longString = TestUtils.generateRandomString(128);
        byte[] longStringBytes = longString.getBytes(StandardCharsets.UTF_8);
  
        vector.setSafe(2, longStringBytes);
        vector.setValueCount(5);
  
        assertArrayEquals(STR0, vector.get(0));
        assertArrayEquals(STR3, vector.get(1));
        assertArrayEquals(longStringBytes, vector.get(2));
        assertArrayEquals(STR6, vector.get(3));
        assertArrayEquals(STR7, vector.get(4));
      }
    }
  
    @Test
    public void testSafeOverwriteLongFromShortString() {
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set short string
        vector.setSafe(0, STR3);
        vector.setValueCount(1);
        // set long string
        vector.setSafe(0, STR0);
        vector.setValueCount(1);
  
        assertArrayEquals(STR0, vector.get(0));
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all longs.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 3);
        // set long string 1
        vector.setSafe(0, STR3);
        // set long string 2
        vector.setSafe(1, STR8);
        // set long string 3
        vector.setSafe(2, STR7);
        vector.setValueCount(3);
  
        // overwrite index 1 with a short string
        vector.setSafe(1, STR6);
        vector.setValueCount(3);
  
        assertArrayEquals(STR3, vector.get(0));
        assertArrayEquals(STR6, vector.get(1));
        assertArrayEquals(STR7, vector.get(2));
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 5);
        // set long string 1
        vector.setSafe(0, STR3);
        // set short string 1
        vector.setSafe(1, STR5);
        // set long string 2
        vector.setSafe(2, STR7);
        // set long string 3
        vector.setSafe(3, STR8);
        // set short string 2
        vector.setSafe(4, STR6);
        vector.setValueCount(5);
  
        // overwrite index 2 with a short string
        vector.setSafe(2, STR0);
        vector.setValueCount(5);
  
        assertArrayEquals(STR3, vector.get(0));
        assertArrayEquals(STR5, vector.get(1));
        assertArrayEquals(STR0, vector.get(2));
        assertArrayEquals(STR8, vector.get(3));
        assertArrayEquals(STR6, vector.get(4));
      }
    }
  
    @Test
    public void testSafeOverwriteLongFromAShorterLongString() {
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set long string
        vector.setSafe(0, STR7);
        vector.setValueCount(1);
        // set shorter long string
        vector.setSafe(0, STR3);
        vector.setValueCount(1);
  
        assertArrayEquals(STR3, vector.get(0));
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all longs.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        // extra memory is allocated
        vector.allocateNew(16, 3);
        // set long string 1
        vector.setSafe(0, STR3);
        // set long string 2
        vector.setSafe(1, STR8);
        // set long string 3
        vector.setSafe(2, STR7);
        vector.setValueCount(3);
  
        // overwrite index 1 with a shorter long string
        vector.setSafe(1, STR2);
        vector.setValueCount(3);
  
        assertArrayEquals(STR3, vector.get(0));
        assertArrayEquals(STR2, vector.get(1));
        assertArrayEquals(STR7, vector.get(2));
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 5);
        // set long string 1
        vector.setSafe(0, STR3);
        // set short string 1
        vector.setSafe(1, STR5);
        // set long string 2
        vector.setSafe(2, STR7);
        // set long string 3
        vector.setSafe(3, STR8);
        // set short string 2
        vector.setSafe(4, STR6);
        vector.setValueCount(5);
  
        // overwrite index 2 with a shorter long string
        vector.setSafe(2, STR2);
        vector.setValueCount(5);
  
        assertArrayEquals(STR3, vector.get(0));
        assertArrayEquals(STR5, vector.get(1));
        assertArrayEquals(STR2, vector.get(2));
        assertArrayEquals(STR8, vector.get(3));
        assertArrayEquals(STR6, vector.get(4));
      }
    }
  
    @Test
    public void testSafeOverwriteLongFromALongerLongString() {
      // Overwriting at the beginning of the buffer.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 1);
        // set long string
        vector.setSafe(0, STR3);
        vector.setValueCount(1);
        // set longer long string
        vector.setSafe(0, STR7);
        vector.setValueCount(1);
  
        assertArrayEquals(STR7, vector.get(0));
      }
  
      // Overwriting in the middle of the buffer when existing buffers are all longs.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        // extra memory is allocated
        vector.allocateNew(16, 3);
        // set long string 1
        vector.setSafe(0, STR3);
        // set long string 2
        vector.setSafe(1, STR8);
        // set long string 3
        vector.setSafe(2, STR7);
        vector.setValueCount(3);
  
        String longerString = TestUtils.generateRandomString(35);
        byte[] longerStringBytes = longerString.getBytes(StandardCharsets.UTF_8);
  
        vector.setSafe(1, longerStringBytes);
        vector.setValueCount(3);
  
        assertArrayEquals(STR3, vector.get(0));
        assertArrayEquals(longerStringBytes, vector.get(1));
        assertArrayEquals(STR7, vector.get(2));
      }
  
      // Overwriting in the middle of the buffer with a mix of short and long strings.
      try (final ViewVarCharVector vector = new ViewVarCharVector("myviewvector", allocator)) {
        vector.allocateNew(16, 5);
        // set long string 1
        vector.setSafe(0, STR3);
        // set short string 1
        vector.setSafe(1, STR5);
        // set long string 2
        vector.setSafe(2, STR7);
        // set long string 3
        vector.setSafe(3, STR2);
        // set short string 2
        vector.setSafe(4, STR6);
        vector.setValueCount(5);
  
        String longerString = TestUtils.generateRandomString(24);
        byte[] longerStringBytes = longerString.getBytes(StandardCharsets.UTF_8);
  
        vector.setSafe(2, longerStringBytes);
        vector.setValueCount(5);
  
        assertArrayEquals(STR3, vector.get(0));
        assertArrayEquals(STR5, vector.get(1));
        assertArrayEquals(longerStringBytes, vector.get(2));
        assertArrayEquals(STR2, vector.get(3));
        assertArrayEquals(STR6, vector.get(4));
      }
    }
  
    @Test
    public void testVectorLoadUnloadInLine() {
  
      try (final ViewVarCharVector vector1 = new ViewVarCharVector("myvector", allocator)) {
  
        setVector(vector1, STR0, STR1, STR4, STR5, STR6);
  
        assertEquals(4, vector1.getLastSet());
        vector1.setValueCount(15);
        assertEquals(14, vector1.getLastSet());
  
        /* Check the vector output */
        assertArrayEquals(STR0, vector1.get(0));
        assertArrayEquals(STR1, vector1.get(1));
        assertArrayEquals(STR4, vector1.get(2));
        assertArrayEquals(STR5, vector1.get(3));
        assertArrayEquals(STR6, vector1.get(4));
  
        Field field = vector1.getField();
        String fieldName = field.getName();
  
        List<Field> fields = new ArrayList<>();
        List<FieldVector> fieldVectors = new ArrayList<>();
  
        fields.add(field);
        fieldVectors.add(vector1);
  
        Schema schema = new Schema(fields);
  
        VectorSchemaRoot schemaRoot1 =
            new VectorSchemaRoot(schema, fieldVectors, vector1.getValueCount());
        VectorUnloader vectorUnloader = new VectorUnloader(schemaRoot1);
  
        try (ArrowRecordBatch recordBatch = vectorUnloader.getRecordBatch();
            BufferAllocator finalVectorsAllocator =
                allocator.newChildAllocator("new vector", 0, Long.MAX_VALUE);
            VectorSchemaRoot schemaRoot2 = VectorSchemaRoot.create(schema, finalVectorsAllocator); ) {
  
          VectorLoader vectorLoader = new VectorLoader(schemaRoot2);
          vectorLoader.load(recordBatch);
  
          ViewVarCharVector vector2 = (ViewVarCharVector) schemaRoot2.getVector(fieldName);
          /*
           * lastSet would have internally been set by VectorLoader.load() when it invokes
           * loadFieldBuffers.
           */
          assertEquals(14, vector2.getLastSet());
          vector2.setValueCount(25);
          assertEquals(24, vector2.getLastSet());
  
          /* Check the vector output */
          assertArrayEquals(STR0, vector2.get(0));
          assertArrayEquals(STR1, vector2.get(1));
          assertArrayEquals(STR4, vector2.get(2));
          assertArrayEquals(STR5, vector2.get(3));
          assertArrayEquals(STR6, vector2.get(4));
        }
      }
    }
  
    @Test
    public void testVectorLoadUnload() {
  
      try (final ViewVarCharVector vector1 = new ViewVarCharVector("myvector", allocator)) {
  
        setVector(vector1, STR1, STR2, STR3, STR4, STR5, STR6);
  
        assertEquals(5, vector1.getLastSet());
        vector1.setValueCount(15);
        assertEquals(14, vector1.getLastSet());
  
        /* Check the vector output */
        assertArrayEquals(STR1, vector1.get(0));
        assertArrayEquals(STR2, vector1.get(1));
        assertArrayEquals(STR3, vector1.get(2));
        assertArrayEquals(STR4, vector1.get(3));
        assertArrayEquals(STR5, vector1.get(4));
        assertArrayEquals(STR6, vector1.get(5));
  
        Field field = vector1.getField();
        String fieldName = field.getName();
  
        List<Field> fields = new ArrayList<>();
        List<FieldVector> fieldVectors = new ArrayList<>();
  
        fields.add(field);
        fieldVectors.add(vector1);
  
        Schema schema = new Schema(fields);
  
        VectorSchemaRoot schemaRoot1 =
            new VectorSchemaRoot(schema, fieldVectors, vector1.getValueCount());
        VectorUnloader vectorUnloader = new VectorUnloader(schemaRoot1);
  
        try (ArrowRecordBatch recordBatch = vectorUnloader.getRecordBatch();
            BufferAllocator finalVectorsAllocator =
                allocator.newChildAllocator("new vector", 0, Long.MAX_VALUE);
            VectorSchemaRoot schemaRoot2 = VectorSchemaRoot.create(schema, finalVectorsAllocator); ) {
  
          VectorLoader vectorLoader = new VectorLoader(schemaRoot2);
          vectorLoader.load(recordBatch);
  
          ViewVarCharVector vector2 = (ViewVarCharVector) schemaRoot2.getVector(fieldName);
          /*
           * lastSet would have internally been set by VectorLoader.load() when it invokes
           * loadFieldBuffers.
           */
          assertEquals(14, vector2.getLastSet());
          vector2.setValueCount(25);
          assertEquals(24, vector2.getLastSet());
  
          /* Check the vector output */
          assertArrayEquals(STR1, vector2.get(0));
          assertArrayEquals(STR2, vector2.get(1));
          assertArrayEquals(STR3, vector2.get(2));
          assertArrayEquals(STR4, vector2.get(3));
          assertArrayEquals(STR5, vector2.get(4));
          assertArrayEquals(STR6, vector2.get(5));
        }
      }
    }
  
    static Stream<Arguments> vectorCreatorProvider() {
      return Stream.of(
          Arguments.of(
              (Function<BufferAllocator, BaseVariableWidthViewVector>)
                  (allocator ->
                      newVector(
                          ViewVarBinaryVector.class,
                          EMPTY_SCHEMA_PATH,
                          Types.MinorType.VIEWVARBINARY,
                          allocator))),
          Arguments.of(
              (Function<BufferAllocator, BaseVariableWidthViewVector>)
                  (allocator ->
                      newVector(
                          ViewVarCharVector.class,
                          EMPTY_SCHEMA_PATH,
                          Types.MinorType.VIEWVARCHAR,
                          allocator))));
    }
  
    @ParameterizedTest
    @MethodSource({"vectorCreatorProvider"})
    public void testCopyFromWithNulls(
        Function<BufferAllocator, BaseVariableWidthViewVector> vectorCreator) {
      try (final BaseVariableWidthViewVector vector = vectorCreator.apply(allocator);
          final BaseVariableWidthViewVector vector2 = vectorCreator.apply(allocator)) {
        final int initialCapacity = 1024;
        vector.setInitialCapacity(initialCapacity);
        vector.allocateNew();
        int capacity = vector.getValueCapacity();
        assertTrue(capacity >= initialCapacity);
  
        // setting number of values such that we have enough space in the initial allocation
        // to avoid re-allocation. This is to test copyFrom() without re-allocation.
        final int numberOfValues = initialCapacity / 2 / ViewVarCharVector.ELEMENT_SIZE;
  
        final String prefixString = TestUtils.generateRandomString(12);
  
        for (int i = 0; i < numberOfValues; i++) {
          if (i % 3 == 0) {
            // null values
            vector.setNull(i);
          } else if (i % 3 == 1) {
            // short strings
            byte[] b = Integer.toString(i).getBytes(StandardCharsets.UTF_8);
            vector.set(i, b, 0, b.length);
          } else {
            // long strings
            byte[] b = (i + prefixString).getBytes(StandardCharsets.UTF_8);
            vector.set(i, b, 0, b.length);
          }
        }
  
        assertEquals(capacity, vector.getValueCapacity());
  
        vector.setValueCount(numberOfValues);
  
        for (int i = 0; i < numberOfValues; i++) {
          if (i % 3 == 0) {
            assertNull(vector.getObject(i));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector.get(i),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector.get(i),
                "unexpected value at index: " + i);
          }
        }
  
        vector2.setInitialCapacity(initialCapacity);
        vector2.allocateNew();
        int capacity2 = vector2.getValueCapacity();
        assertEquals(capacity2, capacity);
  
        for (int i = 0; i < numberOfValues; i++) {
          vector2.copyFrom(i, i, vector);
          if (i % 3 == 0) {
            assertNull(vector2.getObject(i));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          }
        }
  
        assertEquals(capacity, vector2.getValueCapacity());
  
        vector2.setValueCount(numberOfValues);
  
        for (int i = 0; i < numberOfValues; i++) {
          if (i % 3 == 0) {
            assertNull(vector2.getObject(i));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          }
        }
      }
    }
  
    @ParameterizedTest
    @MethodSource("vectorCreatorProvider")
    public void testCopyFromSafeWithNulls(
        Function<BufferAllocator, BaseVariableWidthViewVector> vectorCreator) {
      try (final BaseVariableWidthViewVector vector = vectorCreator.apply(allocator);
          final BaseVariableWidthViewVector vector2 = vectorCreator.apply(allocator)) {
  
        final int initialCapacity = 4096;
        vector.setInitialCapacity(initialCapacity);
        vector.allocateNew();
        int capacity = vector.getValueCapacity();
        assertTrue(capacity >= initialCapacity);
  
        final int numberOfValues = initialCapacity / ViewVarCharVector.ELEMENT_SIZE;
  
        final String prefixString = TestUtils.generateRandomString(12);
  
        for (int i = 0; i < numberOfValues; i++) {
          if (i % 3 == 0) {
            // null values
            vector.setNull(i);
          } else if (i % 3 == 1) {
            // short strings
            byte[] b = Integer.toString(i).getBytes(StandardCharsets.UTF_8);
            vector.setSafe(i, b, 0, b.length);
          } else {
            // long strings
            byte[] b = (i + prefixString).getBytes(StandardCharsets.UTF_8);
            vector.setSafe(i, b, 0, b.length);
          }
        }
  
        /* NO reAlloc() should have happened in setSafe() */
        assertEquals(capacity, vector.getValueCapacity());
  
        vector.setValueCount(numberOfValues);
  
        for (int i = 0; i < numberOfValues; i++) {
          if (i % 3 == 0) {
            assertNull(vector.getObject(i));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector.get(i),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector.get(i),
                "unexpected value at index: " + i);
          }
        }
  
        vector2.setInitialCapacity(initialCapacity);
        vector2.allocateNew();
        int capacity2 = vector2.getValueCapacity();
        assertEquals(capacity2, capacity);
  
        for (int i = 0; i < numberOfValues; i++) {
          vector2.copyFromSafe(i, i, vector);
          if (i % 3 == 0) {
            assertNull(vector2.getObject(i));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          }
        }
  
        /* NO reAlloc() should have happened in setSafe() */
        assertEquals(capacity, vector2.getValueCapacity());
  
        vector2.setValueCount(numberOfValues);
  
        for (int i = 0; i < numberOfValues; i++) {
          if (i % 3 == 0) {
            assertNull(vector2.getObject(i));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector2.get(i),
                "unexpected value at index: " + i);
          }
        }
  
        // make it reallocate
        int valueCapacity = vector2.getValueCapacity();
        for (int i = 0; i < numberOfValues; i++) {
          int thisIndex = i + valueCapacity;
          vector2.copyFromSafe(i, thisIndex, vector);
          if (i % 3 == 0) {
            assertNull(vector2.getObject(thisIndex));
          } else if (i % 3 == 1) {
            assertArrayEquals(
                Integer.toString(i).getBytes(StandardCharsets.UTF_8),
                vector2.get(thisIndex),
                "unexpected value at index: " + i);
          } else {
            assertArrayEquals(
                (i + prefixString).getBytes(StandardCharsets.UTF_8),
                vector2.get(thisIndex),
                "unexpected value at index: " + i);
          }
        }
  
        // test target vector with different initialCapacity
        try (final BaseVariableWidthViewVector vector3 = vectorCreator.apply(allocator)) {
          vector3.setInitialCapacity(16);
          vector3.allocateNew();
          for (int i = 0; i < numberOfValues; i++) {
            vector3.copyFromSafe(i, i, vector);
            if (i % 3 == 0) {
              assertNull(vector3.getObject(i));
            } else {
              assertArrayEquals(vector.get(i), vector3.get(i));
            }
          }
        }
  
        // test overwrite a used vector by copy
        try (final BaseVariableWidthViewVector targetVector = vectorCreator.apply(allocator)) {
  
          targetVector.setInitialCapacity(initialCapacity);
          targetVector.allocateNew();
  
          // source vector: null, short, long...
          // target vector: long, null, short...
          for (int i = 0; i < numberOfValues; i++) {
            if (i % 3 == 0) {
              // long strings
              byte[] b = (i + prefixString).getBytes(StandardCharsets.UTF_8);
              targetVector.set(i, b, 0, b.length);
            } else if (i % 3 == 1) {
              // null values
              targetVector.setNull(i);
            } else {
              // short strings
              byte[] b = Integer.toString(i).getBytes(StandardCharsets.UTF_8);
              targetVector.set(i, b, 0, b.length);
            }
          }
          targetVector.setValueCount(numberOfValues);
  
          for (int i = 0; i < numberOfValues; i++) {
            targetVector.copyFromSafe(i, i, vector);
            if (i % 3 == 0) {
              assertNull(targetVector.getObject(i));
            } else {
              assertArrayEquals(vector.get(i), targetVector.get(i));
            }
          }
        }
      }
    }
  
    public byte[] generateRandomBinaryData(int size) {
      byte[] binaryData = new byte[size];
      random.nextBytes(binaryData);
      return binaryData;
    }
  
    private byte[][] generateBinaryDataArray(int size, int length) {
      byte[][] binaryDataArray = new byte[size][];
      for (int i = 0; i < size; i++) {
        binaryDataArray[i] = generateRandomBinaryData(length);
      }
      return binaryDataArray;
    }
  
    private void testSplitAndTransferOnSlicedBufferHelper(
        BaseVariableWidthViewVector targetVector,
        BaseVariableWidthViewVector sourceVector,
        int startIndex,
        int length,
        byte[][] data) {
      sourceVector.allocateNew(1024 * 10, 1024);
  
      for (int i = 0; i < data.length; i++) {
        sourceVector.set(i, data[i]);
      }
      sourceVector.setValueCount(data.length);
  
      final long allocatedMem = allocator.getAllocatedMemory();
      final int validityRefCnt = sourceVector.getValidityBuffer().refCnt();
      final int dataRefCnt = sourceVector.getDataBuffer().refCnt();
  
      sourceVector.splitAndTransferTo(startIndex, length, targetVector);
      // we allocate view and data buffers for the target vector
      assertTrue(allocatedMem < allocator.getAllocatedMemory());
  
      // The validity buffer is sliced from the same buffer.See
      // BaseFixedWidthViewVector#allocateBytes.
      // Therefore, the refcnt of the validity buffer is increased once since the startIndex is 0.
      assertEquals(validityRefCnt + 1, sourceVector.getValidityBuffer().refCnt());
      // since the new view buffer is allocated, the refcnt is the same as the source vector.
      assertEquals(dataRefCnt, sourceVector.getDataBuffer().refCnt());
    }
  
    /**
     * ARROW-7831: this checks a slice taken off a buffer is still readable after that buffer's
     * allocator is closed. With short strings.
     */
    @Test
    public void testSplitAndTransferWithShortStringOnSlicedBuffer() {
      final byte[][] data = new byte[][] {STR4, STR5, STR6};
      final int startIndex = 0;
      final int length = 2;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (targetVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], targetVector.get(i - startIndex)));
          };
  
      try (final ViewVarCharVector targetVector = newViewVarCharVector("split-target", allocator)) {
        try (final ViewVarCharVector sourceVector =
            newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
          testSplitAndTransferOnSlicedBufferHelper(
              targetVector, sourceVector, startIndex, length, data);
        }
        validateVector.accept(targetVector, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(3, 10);
  
      try (final ViewVarBinaryVector targetVector =
          newViewVarBinaryVector("split-target", allocator)) {
        try (final ViewVarBinaryVector sourceVector =
            newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
          testSplitAndTransferOnSlicedBufferHelper(
              targetVector, sourceVector, startIndex, length, binaryData);
        }
        validateVector.accept(targetVector, binaryData);
      }
    }
  
    /**
     * ARROW-7831: this checks a slice taken off a buffer is still readable after that buffer's
     * allocator is closed. With a long string included.
     */
    @Test
    public void testSplitAndTransferWithLongStringsOnSlicedBuffer() {
      final byte[][] data = new byte[][] {STR2, STR5, STR6};
      final int startIndex = 0;
      final int length = 2;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (targetVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], targetVector.get(i - startIndex)));
          };
  
      try (final ViewVarCharVector targetVector = newViewVarCharVector("split-target", allocator)) {
        try (final ViewVarCharVector sourceVector =
            newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
          testSplitAndTransferOnSlicedBufferHelper(
              targetVector, sourceVector, startIndex, length, data);
        }
        validateVector.accept(targetVector, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(3, 18);
      try (final ViewVarBinaryVector targetVector =
          newViewVarBinaryVector("split-target", allocator)) {
        try (final ViewVarBinaryVector sourceVector =
            newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
          testSplitAndTransferOnSlicedBufferHelper(
              targetVector, sourceVector, startIndex, length, binaryData);
        }
        validateVector.accept(targetVector, binaryData);
      }
    }
  
    private void testSplitAndTransferOnSlicedVectorHelper(
        BaseVariableWidthViewVector sourceVector,
        BaseVariableWidthViewVector targetVector,
        int startIndex,
        int length,
        byte[][] data) {
      sourceVector.allocateNew(1024 * 10, 1024);
  
      for (int i = 0; i < data.length; i++) {
        sourceVector.set(i, data[i]);
      }
      sourceVector.setValueCount(data.length);
  
      final long allocatedMem = allocator.getAllocatedMemory();
      final int validityRefCnt = sourceVector.getValidityBuffer().refCnt();
      final int dataRefCnt = sourceVector.getDataBuffer().refCnt();
  
      sourceVector.splitAndTransferTo(startIndex, length, targetVector);
      // we allocate view and data buffers for the target vector
      assertTrue(allocatedMem < allocator.getAllocatedMemory());
      // The validity buffer is sliced from the same buffer.See
      // BaseFixedWidthViewVector#allocateBytes.
      // Therefore, the refcnt of the validity buffer is increased once since the startIndex is 0.
      assertEquals(validityRefCnt + 1, sourceVector.getValidityBuffer().refCnt());
      // since the new view buffer is allocated, the refcnt is the same as the source vector.
      assertEquals(dataRefCnt, sourceVector.getDataBuffer().refCnt());
  
      for (int i = startIndex; i < length; i++) {
        assertArrayEquals(data[i], targetVector.get(i - startIndex));
      }
    }
  
    /**
     * ARROW-7831: this checks a vector that got sliced is still readable after the slice's allocator
     * got closed. With short strings.
     */
    @Test
    public void testSplitAndTransferWithShortStringsOnSlicedVector() {
      byte[][] data = new byte[][] {STR4, STR5, STR6};
      final int startIndex = 0;
      final int length = 2;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (sourceVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], sourceVector.get(i)));
          };
  
      try (final ViewVarCharVector sourceVector =
          newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        try (final ViewVarCharVector targetVector = newViewVarCharVector("split-target", allocator)) {
          testSplitAndTransferOnSlicedVectorHelper(
              sourceVector, targetVector, startIndex, length, data);
        }
        validateVector.accept(sourceVector, data);
      }
  
      byte[][] binaryData = generateBinaryDataArray(3, 10);
      try (final ViewVarBinaryVector sourceVector =
          newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
        try (final ViewVarBinaryVector targetVector =
            newViewVarBinaryVector("split-target", allocator)) {
          testSplitAndTransferOnSlicedVectorHelper(
              sourceVector, targetVector, startIndex, length, binaryData);
        }
        validateVector.accept(sourceVector, binaryData);
      }
    }
  
    /**
     * ARROW-7831: this checks a vector that got sliced is still readable after the slice's allocator
     * got closed. With a long string included.
     */
    @Test
    public void testSplitAndTransferWithLongStringsOnSlicedVector() {
      final byte[][] data = new byte[][] {STR2, STR5, STR6};
      final int startIndex = 0;
      final int length = 2;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (sourceVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], sourceVector.get(i)));
          };
  
      try (final ViewVarCharVector sourceVector =
          newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        try (final ViewVarCharVector targetVector = newViewVarCharVector("split-target", allocator)) {
          testSplitAndTransferOnSlicedVectorHelper(
              sourceVector, targetVector, startIndex, length, data);
        }
        validateVector.accept(sourceVector, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(3, 20);
      try (final ViewVarBinaryVector sourceVector =
          newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
        try (final ViewVarBinaryVector targetVector =
            newViewVarBinaryVector("split-target", allocator)) {
          testSplitAndTransferOnSlicedVectorHelper(
              sourceVector, targetVector, startIndex, length, binaryData);
        }
        validateVector.accept(sourceVector, binaryData);
      }
    }
  
    private void testSplitAndTransferOnValiditySplitHelper(
        BaseVariableWidthViewVector targetVector,
        BaseVariableWidthViewVector sourceVector,
        int startIndex,
        int length,
        byte[][] data) {
      sourceVector.allocateNew(1024 * 10, 1024);
  
      sourceVector.set(0, new byte[0]);
      sourceVector.setNull(1);
      for (int i = 0; i < data.length; i++) {
        if (data[i] == null) {
          sourceVector.setNull(i);
        } else {
          sourceVector.set(i, data[i]);
        }
      }
      sourceVector.setValueCount(data.length);
  
      final long allocatedMem = allocator.getAllocatedMemory();
      final int validityRefCnt = sourceVector.getValidityBuffer().refCnt();
      final int dataRefCnt = sourceVector.getDataBuffer().refCnt();
  
      sourceVector.splitAndTransferTo(startIndex, length, targetVector);
      // the allocation only consists in the size needed for the validity buffer
      final long validitySize =
          DefaultRoundingPolicy.DEFAULT_ROUNDING_POLICY.getRoundedSize(
              getValidityBufferSizeFromCount(2));
      // we allocate view and data buffers for the target vector
      assertTrue(allocatedMem + validitySize < allocator.getAllocatedMemory());
      // The validity is sliced from the same buffer.See BaseFixedWidthViewVector#allocateBytes.
      // Since values up to the startIndex are empty/null validity refcnt should not change.
      assertEquals(validityRefCnt, sourceVector.getValidityBuffer().refCnt());
      // since the new view buffer is allocated, the refcnt is the same as the source vector.
      assertEquals(dataRefCnt, sourceVector.getDataBuffer().refCnt());
  
      for (int i = startIndex; i < startIndex + length; i++) {
        assertArrayEquals(data[i], targetVector.get(i - startIndex));
      }
  
      for (int i = 0; i < data.length; i++) {
        if (data[i] == null) {
          assertTrue(sourceVector.isNull(i));
        } else {
          assertArrayEquals(data[i], sourceVector.get(i));
        }
      }
    }
  
    /**
     * ARROW-7831: this checks a validity splitting where the validity buffer is sliced from the same
     * buffer. In the case where all the values up to the start of the slice are null/empty. With
     * short strings.
     */
    @Test
    public void testSplitAndTransferWithShortStringsOnValiditySplit() {
      final byte[][] data = new byte[][] {new byte[0], null, STR4, STR5, STR6};
      final int startIndex = 2;
      final int length = 2;
  
      try (final ViewVarCharVector targetVector = newViewVarCharVector("split-target", allocator);
          final ViewVarCharVector sourceVector = newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        testSplitAndTransferOnValiditySplitHelper(
            targetVector, sourceVector, startIndex, length, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(5, 10);
      binaryData[0] = new byte[0];
      binaryData[1] = null;
      try (final ViewVarBinaryVector targetVector =
              newViewVarBinaryVector("split-target", allocator);
          final ViewVarBinaryVector sourceVector =
              newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
        testSplitAndTransferOnValiditySplitHelper(
            targetVector, sourceVector, startIndex, length, binaryData);
      }
    }
  
    /**
     * ARROW-7831: this checks a validity splitting where the validity buffer is sliced from the same
     * buffer. In the case where all the values up to the start of the slice are null/empty. With long
     * strings.
     */
    @Test
    public void testSplitAndTransferWithLongStringsOnValiditySplit() {
      final byte[][] data = new byte[][] {new byte[0], null, STR1, STR2, STR3};
      final int startIndex = 2;
      final int length = 2;
  
      try (final ViewVarCharVector targetVector = newViewVarCharVector("split-target", allocator);
          final ViewVarCharVector sourceVector = newViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        testSplitAndTransferOnValiditySplitHelper(
            targetVector, sourceVector, startIndex, length, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(5, 18);
      binaryData[0] = new byte[0];
      binaryData[1] = null;
  
      try (final ViewVarBinaryVector targetVector =
              newViewVarBinaryVector("split-target", allocator);
          final ViewVarBinaryVector sourceVector =
              newViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
        testSplitAndTransferOnValiditySplitHelper(
            targetVector, sourceVector, startIndex, length, data);
      }
    }
  
    private void testSplitAndTransferOnAllocatorToAllocator(
        BaseVariableWidthViewVector targetVector,
        BaseVariableWidthViewVector sourceVector,
        int startIndex,
        int length,
        byte[][] data) {
      sourceVector.allocateNew(50, data.length);
  
      for (int i = 0; i < data.length; i++) {
        sourceVector.set(i, data[i]);
      }
      sourceVector.setValueCount(data.length);
  
      final long allocatedMem = allocator.getAllocatedMemory();
      final int validityRefCnt = sourceVector.getValidityBuffer().refCnt();
      final int dataRefCnt = sourceVector.getDataBuffer().refCnt();
  
      sourceVector.splitAndTransferTo(startIndex, length, targetVector);
  
      if (sourceVector.getDataBuffers().isEmpty()) {
        // no extra allocation as strings are all inline
        assertEquals(allocatedMem, allocator.getAllocatedMemory());
      } else {
        // extra allocation as some strings are not inline
        assertTrue(allocatedMem < allocator.getAllocatedMemory());
      }
  
      // the refcnts of each buffer for this test should be the same as what
      // the source allocator ended up with.
      assertEquals(validityRefCnt, sourceVector.getValidityBuffer().refCnt());
      // since the new view buffer is allocated, the refcnt is the same as the source vector.
      assertEquals(dataRefCnt, sourceVector.getDataBuffer().refCnt());
  
      for (int i = 0; i < data.length; i++) {
        assertArrayEquals(data[i], sourceVector.get(i));
      }
    }
  
    /**
     * ARROW-7831: ensures that data is transferred from one allocator to another in case of 0-index
     * start special cases. With short strings.
     */
    @Test
    public void testSplitAndTransferWithShortStringsOnAllocatorToAllocator() {
      final int maxAllocation = 512;
      final byte[][] data = new byte[][] {STR4, STR5, STR6};
      final int startIndex = 0;
      final int length = 2;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (targetVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], targetVector.get(i - startIndex)));
          };
  
      try (final BufferAllocator targetAllocator =
              allocator.newChildAllocator("target-alloc", 256, maxAllocation);
          final ViewVarCharVector targetVector =
              newViewVarCharVector("split-target", targetAllocator)) {
        try (final BufferAllocator sourceAllocator =
                allocator.newChildAllocator("source-alloc", 256, maxAllocation);
            final ViewVarCharVector sourceVector =
                newViewVarCharVector(EMPTY_SCHEMA_PATH, sourceAllocator)) {
          testSplitAndTransferOnAllocatorToAllocator(
              targetVector, sourceVector, startIndex, length, data);
        }
        validateVector.accept(targetVector, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(3, 10);
      try (final BufferAllocator targetAllocator =
              allocator.newChildAllocator("target-alloc", 256, maxAllocation);
          final ViewVarBinaryVector targetVector =
              newViewVarBinaryVector("split-target", targetAllocator)) {
        try (final BufferAllocator sourceAllocator =
                allocator.newChildAllocator("source-alloc", 256, maxAllocation);
            final ViewVarBinaryVector sourceVector =
                newViewVarBinaryVector(EMPTY_SCHEMA_PATH, sourceAllocator)) {
          testSplitAndTransferOnAllocatorToAllocator(
              targetVector, sourceVector, startIndex, length, binaryData);
        }
        validateVector.accept(targetVector, binaryData);
      }
    }
  
    /**
     * ARROW-7831: ensures that data is transferred from one allocator to another in case of 0-index
     * start special cases. With long strings.
     */
    @Test
    public void testSplitAndTransferWithLongStringsOnAllocatorToAllocator() {
      final int initialReservation = 1024;
      // 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
      final byte[][] data = new byte[][] {STR1, STR2, STR3};
      final int startIndex = 0;
      final int length = 2;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (targetVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], targetVector.get(i - startIndex)));
          };
  
      final int maxAllocation =
          initialReservation
              + BaseVariableWidthViewVector.INITIAL_VIEW_VALUE_ALLOCATION
                  * BaseVariableWidthViewVector.ELEMENT_SIZE;
      try (final BufferAllocator targetAllocator =
              allocator.newChildAllocator("target-alloc", initialReservation, maxAllocation);
          final ViewVarCharVector targetVector =
              newViewVarCharVector("split-target", targetAllocator)) {
        try (final BufferAllocator sourceAllocator =
                allocator.newChildAllocator("source-alloc", initialReservation, maxAllocation);
            final ViewVarCharVector sourceVector =
                newViewVarCharVector(EMPTY_SCHEMA_PATH, sourceAllocator)) {
          testSplitAndTransferOnAllocatorToAllocator(
              targetVector, sourceVector, startIndex, length, data);
        }
        validateVector.accept(targetVector, data);
      }
  
      final byte[][] binaryData = generateBinaryDataArray(3, 18);
  
      try (final BufferAllocator targetAllocator =
              allocator.newChildAllocator("target-alloc", initialReservation, maxAllocation);
          final ViewVarBinaryVector targetVector =
              newViewVarBinaryVector("split-target", targetAllocator)) {
        try (final BufferAllocator sourceAllocator =
                allocator.newChildAllocator("source-alloc", initialReservation, maxAllocation);
            final ViewVarBinaryVector sourceVector =
                newViewVarBinaryVector(EMPTY_SCHEMA_PATH, sourceAllocator)) {
          testSplitAndTransferOnAllocatorToAllocator(
              targetVector, sourceVector, startIndex, length, binaryData);
        }
        validateVector.accept(targetVector, binaryData);
      }
    }
  
    private void testReallocAfterVectorTransferHelper(
        BaseVariableWidthViewVector vector, byte[] str1, byte[] str2) {
      /* 4096 values with 16 bytes per record */
      final int bytesPerRecord = 32;
      vector.allocateNew(4096 * bytesPerRecord, 4096);
      int valueCapacity = vector.getValueCapacity();
      assertTrue(valueCapacity >= 4096);
  
      /* populate the vector */
      for (int i = 0; i < valueCapacity; i++) {
        if ((i & 1) == 1) {
          vector.set(i, str1);
        } else {
          vector.set(i, str2);
        }
      }
  
      /* Check the vector output */
      for (int i = 0; i < valueCapacity; i++) {
        if ((i & 1) == 1) {
          assertArrayEquals(str1, vector.get(i));
        } else {
          assertArrayEquals(str2, vector.get(i));
        }
      }
  
      /* trigger first realloc */
      vector.setSafe(valueCapacity, str2, 0, str2.length);
      assertTrue(vector.getValueCapacity() >= 2 * valueCapacity);
      while (vector.getByteCapacity() < bytesPerRecord * vector.getValueCapacity()) {
        vector.reallocViewBuffer();
        vector.reallocViewDataBuffer();
      }
  
      /* populate the remaining vector */
      for (int i = valueCapacity; i < vector.getValueCapacity(); i++) {
        if ((i & 1) == 1) {
          vector.set(i, str1);
        } else {
          vector.set(i, str2);
        }
      }
  
      /* Check the vector output */
      valueCapacity = vector.getValueCapacity();
      for (int i = 0; i < valueCapacity; i++) {
        if ((i & 1) == 1) {
          assertArrayEquals(str1, vector.get(i));
        } else {
          assertArrayEquals(str2, vector.get(i));
        }
      }
  
      /* trigger second realloc */
      vector.setSafe(valueCapacity + bytesPerRecord, str2, 0, str2.length);
      assertTrue(vector.getValueCapacity() >= 2 * valueCapacity);
      while (vector.getByteCapacity() < bytesPerRecord * vector.getValueCapacity()) {
        vector.reallocViewBuffer();
        vector.reallocViewDataBuffer();
      }
  
      /* populate the remaining vector */
      for (int i = valueCapacity; i < vector.getValueCapacity(); i++) {
        if ((i & 1) == 1) {
          vector.set(i, str1);
        } else {
          vector.set(i, str2);
        }
      }
  
      /* Check the vector output */
      valueCapacity = vector.getValueCapacity();
      for (int i = 0; i < valueCapacity; i++) {
        if ((i & 1) == 1) {
          assertArrayEquals(str1, vector.get(i));
        } else {
          assertArrayEquals(str2, vector.get(i));
        }
      }
  
      /* We are potentially working with 4x the size of vector buffer
       * that we initially started with.
       * Now let's transfer the vector.
       */
  
      TransferPair transferPair = vector.getTransferPair(allocator);
      transferPair.transfer();
      BaseVariableWidthViewVector toVector = (BaseVariableWidthViewVector) transferPair.getTo();
      valueCapacity = toVector.getValueCapacity();
  
      for (int i = 0; i < valueCapacity; i++) {
        if ((i & 1) == 1) {
          assertArrayEquals(str1, toVector.get(i));
        } else {
          assertArrayEquals(str2, toVector.get(i));
        }
      }
      toVector.close();
    }
  
    @Test
    public void testReallocAfterVectorTransfer() {
      try (final ViewVarCharVector vector = new ViewVarCharVector(EMPTY_SCHEMA_PATH, allocator)) {
        testReallocAfterVectorTransferHelper(vector, STR1, STR2);
      }
  
      try (final ViewVarBinaryVector vector = new ViewVarBinaryVector(EMPTY_SCHEMA_PATH, allocator)) {
        testReallocAfterVectorTransferHelper(
            vector, generateRandomBinaryData(12), generateRandomBinaryData(13));
      }
    }
  
    private void testSplitAndTransferWithMultipleDataBuffersHelper(
        BaseVariableWidthViewVector sourceVector,
        BaseVariableWidthViewVector targetVector,
        int startIndex,
        int length,
        byte[][] data) {
      sourceVector.allocateNew(48, 4);
  
      for (int i = 0; i < data.length; i++) {
        sourceVector.set(i, data[i]);
      }
      sourceVector.setValueCount(data.length);
  
      // we should have multiple data buffers
      assertTrue(sourceVector.getDataBuffers().size() > 1);
  
      final long allocatedMem = allocator.getAllocatedMemory();
      final int validityRefCnt = sourceVector.getValidityBuffer().refCnt();
      final int dataRefCnt = sourceVector.getDataBuffer().refCnt();
  
      // split and transfer with slice starting at the beginning:
      // this should not allocate anything new
      sourceVector.splitAndTransferTo(startIndex, length, targetVector);
      // we allocate view and data buffers for the target vector
      assertTrue(allocatedMem < allocator.getAllocatedMemory());
  
      // the refcnts of each buffer for this test should be the same as what
      // the source allocator ended up with.
      assertEquals(validityRefCnt, sourceVector.getValidityBuffer().refCnt());
      // since the new view buffer is allocated, the refcnt is the same as the source vector.
      assertEquals(dataRefCnt, sourceVector.getDataBuffer().refCnt());
  
      for (int i = 0; i < data.length; i++) {
        assertArrayEquals(data[i], sourceVector.get(i));
      }
    }
  
    /**
     * ARROW-7831: ensures that data is transferred from one allocator to another in case of 0-index
     * start special cases. With long strings and multiple data buffers. Check multi-data buffer
     * source copying
     */
    @Test
    public void testSplitAndTransferWithMultipleDataBuffers() {
      final String str4 = TestUtils.generateRandomString(35);
      final byte[][] data = new byte[][] {STR1, STR2, STR3, str4.getBytes(StandardCharsets.UTF_8)};
      final int startIndex = 1;
      final int length = 3;
  
      BiConsumer<BaseVariableWidthViewVector, byte[][]> validateVector =
          (targetVector, expectedData) -> {
            IntStream.range(startIndex, length)
                .forEach(i -> assertArrayEquals(expectedData[i], targetVector.get(i - startIndex)));
          };
  
      try (final ViewVarCharVector targetVector = new ViewVarCharVector("target", allocator)) {
        try (final ViewVarCharVector sourceVector = new ViewVarCharVector("source", allocator)) {
          testSplitAndTransferWithMultipleDataBuffersHelper(
              sourceVector, targetVector, startIndex, length, data);
        }
        validateVector.accept(targetVector, data);
      }
  
      try (final ViewVarBinaryVector targetVector = new ViewVarBinaryVector("target", allocator)) {
        try (final ViewVarBinaryVector sourceVector = new ViewVarBinaryVector("source", allocator)) {
          testSplitAndTransferWithMultipleDataBuffersHelper(
              sourceVector, targetVector, startIndex, length, data);
        }
        validateVector.accept(targetVector, data);
      }
    }
  
    @Test
    public void testVectorLoadUnloadOnMixedTypes() {
  
      try (final IntVector vector1 = new IntVector("myvector", allocator);
          final ViewVarCharVector vector2 = new ViewVarCharVector("myviewvector", allocator)) {
  
        final int valueCount = 15;
  
        setVector(vector1, 1, 2, 3, 4, 5, 6);
        vector1.setValueCount(valueCount);
  
        setVector(vector2, STR1, STR2, STR3, STR4, STR5, STR6);
        vector1.setValueCount(valueCount);
  
        /* Check the vector output */
        assertEquals(1, vector1.get(0));
        assertEquals(2, vector1.get(1));
        assertEquals(3, vector1.get(2));
        assertEquals(4, vector1.get(3));
        assertEquals(5, vector1.get(4));
        assertEquals(6, vector1.get(5));
  
        Field field1 = vector1.getField();
        String fieldName1 = field1.getName();
  
        Field field2 = vector2.getField();
        String fieldName2 = field2.getName();
  
        List<Field> fields = new ArrayList<>(2);
        List<FieldVector> fieldVectors = new ArrayList<>(2);
  
        fields.add(field1);
        fields.add(field2);
        fieldVectors.add(vector1);
        fieldVectors.add(vector2);
  
        Schema schema = new Schema(fields);
  
        VectorSchemaRoot schemaRoot1 = new VectorSchemaRoot(schema, fieldVectors, valueCount);
        VectorUnloader vectorUnloader = new VectorUnloader(schemaRoot1);
  
        try (ArrowRecordBatch recordBatch = vectorUnloader.getRecordBatch();
            BufferAllocator finalVectorsAllocator =
                allocator.newChildAllocator("new vector", 0, Long.MAX_VALUE);
            VectorSchemaRoot schemaRoot2 = VectorSchemaRoot.create(schema, finalVectorsAllocator); ) {
  
          // validating recordBatch contains an output for variadicBufferCounts
          assertFalse(recordBatch.getVariadicBufferCounts().isEmpty());
          assertEquals(1, recordBatch.getVariadicBufferCounts().size());
  
          VectorLoader vectorLoader = new VectorLoader(schemaRoot2);
          vectorLoader.load(recordBatch);
  
          IntVector vector3 = (IntVector) schemaRoot2.getVector(fieldName1);
          vector3.setValueCount(25);
  
          /* Check the vector output */
          assertEquals(1, vector3.get(0));
          assertEquals(2, vector3.get(1));
          assertEquals(3, vector3.get(2));
          assertEquals(4, vector3.get(3));
          assertEquals(5, vector3.get(4));
          assertEquals(6, vector3.get(5));
  
          ViewVarCharVector vector4 = (ViewVarCharVector) schemaRoot2.getVector(fieldName2);
          vector4.setValueCount(25);
  
          /* Check the vector output */
          assertArrayEquals(STR1, vector4.get(0));
          assertArrayEquals(STR2, vector4.get(1));
          assertArrayEquals(STR3, vector4.get(2));
          assertArrayEquals(STR4, vector4.get(3));
          assertArrayEquals(STR5, vector4.get(4));
          assertArrayEquals(STR6, vector4.get(5));
        }
      }
    }
  }
  