fix out of bounds exception in array types

Author: binary-signal

fluss-common/src/main/java/org/apache/fluss/row/BinaryArray.java

@@ -474,7 +474,7 @@ public boolean[] toBooleanArray() {
         checkNoNull();
         boolean[] values = new boolean[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, BOOLEAN_ARRAY_OFFSET, size);
+                segments, elementOffset, values, 0, size);
         return values;
     }
 
@@ -483,7 +483,7 @@ public byte[] toByteArray() {
         checkNoNull();
         byte[] values = new byte[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, BYTE_ARRAY_BASE_OFFSET, size);
+                segments, elementOffset, values, 0, size);
         return values;
     }
 
@@ -492,7 +492,7 @@ public short[] toShortArray() {
         checkNoNull();
         short[] values = new short[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, SHORT_ARRAY_OFFSET, size * 2);
+                segments, elementOffset, values, 0, size * 2);
         return values;
     }
 
@@ -501,7 +501,7 @@ public int[] toIntArray() {
         checkNoNull();
         int[] values = new int[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, INT_ARRAY_OFFSET, size * 4);
+                segments, elementOffset, values, 0, size * 4);
         return values;
     }
 
@@ -510,7 +510,7 @@ public long[] toLongArray() {
         checkNoNull();
         long[] values = new long[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, LONG_ARRAY_OFFSET, size * 8);
+                segments, elementOffset, values, 0, size * 8);
         return values;
     }
 
@@ -519,7 +519,7 @@ public float[] toFloatArray() {
         checkNoNull();
         float[] values = new float[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, FLOAT_ARRAY_OFFSET, size * 4);
+                segments, elementOffset, values, 0, size * 4);
         return values;
     }
 
@@ -528,7 +528,7 @@ public double[] toDoubleArray() {
         checkNoNull();
         double[] values = new double[size];
         BinarySegmentUtils.copyToUnsafe(
-                segments, elementOffset, values, DOUBLE_ARRAY_OFFSET, size * 8);
+                segments, elementOffset, values, 0, size * 8);
         return values;
     }
 

fluss-common/src/test/java/org/apache/fluss/row/BinaryArrayTest.java

@@ -966,4 +966,117 @@ public void testCalculateFixLengthPartSizeForArray() {
         assertThat(BinaryArray.calculateFixLengthPartSize(DataTypes.CHAR(10))).isEqualTo(8);
         assertThat(BinaryArray.calculateFixLengthPartSize(DataTypes.BINARY(20))).isEqualTo(8);
     }
+
+    @Test
+    public void testToDoubleArrayWithLargeArray() {
+        // Test with 20 elements to verify fix for ArrayIndexOutOfBoundsException
+        double[] input = new double[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = i * 1.1;
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        double[] result = binaryArray.toDoubleArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo(i * 1.1);
+        }
+    }
+
+    @Test
+    public void testToFloatArrayWithLargeArray() {
+        float[] input = new float[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = i * 1.5f;
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        float[] result = binaryArray.toFloatArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo(i * 1.5f);
+        }
+    }
+
+    @Test
+    public void testToLongArrayWithLargeArray() {
+        long[] input = new long[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = i * 1000L;
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        long[] result = binaryArray.toLongArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo(i * 1000L);
+        }
+    }
+
+    @Test
+    public void testToIntArrayWithLargeArray() {
+        int[] input = new int[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = i * 100;
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        int[] result = binaryArray.toIntArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo(i * 100);
+        }
+    }
+
+    @Test
+    public void testToShortArrayWithLargeArray() {
+        short[] input = new short[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = (short) (i * 10);
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        short[] result = binaryArray.toShortArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo((short) (i * 10));
+        }
+    }
+
+    @Test
+    public void testToByteArrayWithLargeArray() {
+        byte[] input = new byte[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = (byte) i;
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        byte[] result = binaryArray.toByteArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo((byte) i);
+        }
+    }
+
+    @Test
+    public void testToBooleanArrayWithLargeArray() {
+        boolean[] input = new boolean[20];
+        for (int i = 0; i < 20; i++) {
+            input[i] = i % 2 == 0;
+        }
+        BinaryArray binaryArray = BinaryArray.fromPrimitiveArray(input);
+
+        boolean[] result = binaryArray.toBooleanArray();
+
+        assertThat(result).hasSize(20);
+        for (int i = 0; i < 20; i++) {
+            assertThat(result[i]).isEqualTo(i % 2 == 0);
+        }
+    }
 }