[FLINK-38138] Fix OOB error in ColumnarArrayData

[voonhous]

What is the purpose of the change

We have noticed that an OOB error is being thrown when trying to invoke getBinary when the following conditions are met:

1. There are multiple elements in a HeapArrayVector
2. When trying to get an element from a non-zero offset of the array
3. When trying to convert this array that we have obtained to a binary

Caused by: java.lang.IllegalArgumentException: 72 > 36
	at java.util.Arrays.copyOfRange(Arrays.java:3519)
	at org.apache.flink.table.data.columnar.ColumnarArrayData.getBinary(ColumnarArrayData.java:138)
	at org.apache.hudi.table.format.cow.vector.ColumnarGroupRowData.getBinary(ColumnarGroupRowData.java:121)
	at org.apache.flink.table.data.RowData.lambda$createFieldGetter$245ca7d1$3(RowData.java:228)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.toBinaryRow(RowDataSerializer.java:207)
	at org.apache.flink.table.data.writer.AbstractBinaryWriter.writeRow(AbstractBinaryWriter.java:147)
	at org.apache.flink.table.data.writer.BinaryArrayWriter.writeRow(BinaryArrayWriter.java:30)
	at org.apache.flink.table.data.writer.BinaryWriter.write(BinaryWriter.java:155)
	at org.apache.flink.table.runtime.typeutils.MapDataSerializer.toBinaryMap(MapDataSerializer.java:179)
	at org.apache.flink.table.runtime.typeutils.MapDataSerializer.copy(MapDataSerializer.java:113)
	at org.apache.flink.table.runtime.typeutils.MapDataSerializer.copy(MapDataSerializer.java:44)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copyRowData(RowDataSerializer.java:170)
	at org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:131)

The error below is caused by the code below:

@Override
public byte[] getBinary(int pos) {
    BytesColumnVector.Bytes byteArray = getByteArray(pos);
    if (byteArray.len == byteArray.data.length) {
        return byteArray.data;
    } else {
        return Arrays.copyOfRange(byteArray.data, byteArray.offset, byteArray.len);
    }
} 

The function signature of Arrays.copyOfRange is as such:

public static byte[] copyOfRange(byte[] original, int from, int to) 

The java.util.Arrays.copyOfRange(original, from, to) method copies the specified range of the original array into a new array.

1. from: The initial index of the range to be copied, inclusive.
2. to: The final index of the range to be copied, exclusive.

Hence, the correct way of invoking `Arrays.copyOfRange`` should be:

return Arrays.copyOfRange(
        byteArray.data, byteArray.offset, byteArray.offset + byteArray.len);  

Brief change log

• Fix the OOB error by ensuring that getBinary is calling Arrays.copyOfRange with the correct arguments.

Verifying this change

• Added unit tests for ColumnarArrayData to verify fix. Test fails before fix, but passes after

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? (no)

[flinkbot]

CI report:

• 3920c8d Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[snuyanzin]

we are trying to go away from junit assertions to assertj
can you please replace it here as well?

[voonhous]

Sure!