[SPARK-57036][SQL] Use intrinsic bulk-fill APIs for constant-value WritableColumnVector methods

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, byte-fill methods (putBytes / putBooleans / putNulls) write bytes in
+  // an inline loop. At or above this count, they call 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 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;
   }
@@ -151,9 +163,13 @@ public void putBoolean(int rowId, boolean value) {
 
   @Override
   public void putBooleans(int rowId, int count, boolean value) {
-    byte v = (byte)((value) ? 1 : 0);
-    for (int i = 0; i < count; ++i) {
-      Platform.putByte(null, data + rowId + i, v);
+    byte v = (byte) (value ? 1 : 0);
+    if (count < SET_MEMORY_THRESHOLD) {
+      for (int i = 0; i < count; ++i) {
+        Platform.putByte(null, data + rowId + i, v);
+      }
+    } else {
+      Platform.setMemory(data + rowId, v, count);
     }
   }
 
@@ -193,8 +209,12 @@ public void putByte(int rowId, byte value) {
 
   @Override
   public void putBytes(int rowId, int count, byte value) {
-    for (int i = 0; i < count; ++i) {
-      Platform.putByte(null, data + rowId + i, value);
+    if (count < SET_MEMORY_THRESHOLD) {
+      for (int i = 0; i < count; ++i) {
+        Platform.putByte(null, data + rowId + i, value);
+      }
+    } else {
+      Platform.setMemory(data + rowId, value, 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;
   }
 
@@ -146,10 +144,8 @@ public void putBoolean(int rowId, boolean value) {
 
   @Override
   public void putBooleans(int rowId, int count, boolean value) {
-    byte v = (byte)((value) ? 1 : 0);
-    for (int i = 0; i < count; ++i) {
-      byteData[i + rowId] = v;
-    }
+    byte v = (byte) (value ? 1 : 0);
+    java.util.Arrays.fill(byteData, rowId, rowId + count, v);
   }
 
   @Override
@@ -191,9 +187,7 @@ public void putByte(int rowId, byte value) {
 
   @Override
   public void putBytes(int rowId, int count, byte value) {
-    for (int i = 0; i < count; ++i) {
-      byteData[i + rowId] = value;
-    }
+    java.util.Arrays.fill(byteData, rowId, rowId + count, value);
   }
 
   @Override
@@ -253,9 +247,7 @@ public void putShort(int rowId, short value) {
 
   @Override
   public void putShorts(int rowId, int count, short value) {
-    for (int i = 0; i < count; ++i) {
-      shortData[i + rowId] = value;
-    }
+    java.util.Arrays.fill(shortData, rowId, rowId + count, value);
   }
 
   @Override
@@ -319,9 +311,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
@@ -395,9 +385,7 @@ public void putLong(int rowId, long value) {
 
   @Override
   public void putLongs(int rowId, int count, long value) {
-    for (int i = 0; i < count; ++i) {
-      longData[i + rowId] = value;
-    }
+    java.util.Arrays.fill(longData, rowId, rowId + count, value);
   }
 
   @Override