[SPARK-57024][SQL] Use bulk fill APIs to materialize RLE runs in Parquet vectorized reader

sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java

@@ -298,9 +298,7 @@ private void readPackedBatch(
         } while (currentBufferIdx < bufEnd
             && currentBuffer[currentBufferIdx] != maxDefLevel);
         int runLen = currentBufferIdx - runStart;
-        for (int k = 0; k < runLen; k++) {
-          nulls.putNull(valueOff + k);
-        }
+        nulls.putNulls(valueOff, runLen);
         valueOff += runLen;
       }
     }
@@ -714,14 +712,10 @@ private void readPackedBatchWithDefLevels(
           updater.readValues(runLen, valueOff, values, valueReader);
         }
       } else {
-        for (int k = 0; k < runLen; k++) {
-          nulls.putNull(valueOff + k);
-        }
+        nulls.putNulls(valueOff, runLen);
       }
       valueOff += runLen;
-      for (int k = 0; k < runLen; k++) {
-        defLevels.putInt(levelIdx + k, runValue);
-      }
+      defLevels.putInts(levelIdx, runLen, runValue);
       levelIdx += runLen;
     }
     state.valueOffset = valueOff;

sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java

@@ -33,6 +33,14 @@ public final class OffHeapColumnVector extends WritableColumnVector {
   private static final boolean bigEndianPlatform =
     ByteOrder.nativeOrder().equals(ByteOrder.BIG_ENDIAN);
 
+  // Below this count, putNulls writes bytes in an inline loop. At or above this count, it
+  // calls Platform.setMemory which lowers to a native memset. The JNI fixed cost of
+  // setMemory dominates for very short fills; on the benchmarked hardware (Apple M4 Max +
+  // OpenJDK 21) the crossover sits between 64 and 512, so 128 is a conservative choice
+  // that avoids regression at small counts (common for random null patterns where RLE
+  // runs are short) while retaining the bulk of the asymptotic gain.
+  private static final int SET_MEMORY_THRESHOLD = 128;
+
   /**
    * Allocates columns to store elements of each field of the schema off heap.
    * Capacity is the initial capacity of the vector and it will grow as necessary. Capacity is
@@ -119,9 +127,13 @@ public void putNull(int rowId) {
   @Override
   public void putNulls(int rowId, int count) {
     if (isAllNull()) return; // Skip writing nulls to all-null vector.
-    long offset = nulls + rowId;
-    for (int i = 0; i < count; ++i, ++offset) {
-      Platform.putByte(null, offset, (byte) 1);
+    if (count < SET_MEMORY_THRESHOLD) {
+      long offset = nulls + rowId;
+      for (int i = 0; i < count; ++i, ++offset) {
+        Platform.putByte(null, offset, (byte) 1);
+      }
+    } else {
+      Platform.setMemory(nulls + rowId, (byte) 1, count);
     }
     numNulls += count;
   }

sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java

@@ -116,9 +116,7 @@ public void putNull(int rowId) {
   @Override
   public void putNulls(int rowId, int count) {
     if (isAllNull()) return; // Skip writing nulls to all-null vector.
-    for (int i = 0; i < count; ++i) {
-      nulls[rowId + i] = (byte)1;
-    }
+    java.util.Arrays.fill(nulls, rowId, rowId + count, (byte) 1);
     numNulls += count;
   }
 
@@ -319,9 +317,7 @@ public void putInt(int rowId, int value) {
 
   @Override
   public void putInts(int rowId, int count, int value) {
-    for (int i = 0; i < count; ++i) {
-      intData[i + rowId] = value;
-    }
+    java.util.Arrays.fill(intData, rowId, rowId + count, value);
   }
 
   @Override