Export constant columns as 1-element scalar arrays for native expansion

Instead of materializing full N-element Arrow arrays for partition and
missing columns, export 1-element arrays from JVM and expand them on
the native side using ScalarValue. This avoids O(N) memory allocation
and copying for constant columns.

- Add CometConstantVector: stores a single value, lazily creates a
  1-element Arrow vector for FFI export, returns constant for all rowIds
- Modify ImmutableConstantColumnReader to produce CometConstantVector
- Add CometConstantVector case in NativeUtil.exportBatch() to skip
  row count validation for 1-element vectors
- In scan.rs, detect 1-element arrays and expand via ScalarValue when
  actual_num_rows > 1; skip take() for scalar columns with selection
  vectors since constants are unaffected by row deletion

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: andygrove

common/src/main/java/org/apache/comet/parquet/ImmutableConstantColumnReader.java

@@ -19,9 +19,6 @@
 
 package org.apache.comet.parquet;
 
-import org.apache.arrow.memory.BufferAllocator;
-import org.apache.arrow.memory.RootAllocator;
-import org.apache.arrow.vector.*;
 import org.apache.arrow.vector.types.DateUnit;
 import org.apache.arrow.vector.types.FloatingPointPrecision;
 import org.apache.arrow.vector.types.TimeUnit;
@@ -31,9 +28,8 @@
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns;
 import org.apache.spark.sql.types.*;
-import org.apache.spark.unsafe.types.UTF8String;
 
-import org.apache.comet.vector.CometPlainVector;
+import org.apache.comet.vector.CometConstantVector;
 import org.apache.comet.vector.CometVector;
 
 /**
@@ -74,8 +70,6 @@ public static boolean isTypeSupported(DataType type) {
     return false;
   }
 
-  private final BufferAllocator allocator = new RootAllocator();
-
   /** Whether all the values in this constant column are nulls */
   private boolean isNull;
 
@@ -143,174 +137,9 @@ protected void initNative() {
     nativeHandle = 0;
   }
 
-  /** Creates a constant Arrow vector with the specified number of rows. */
+  /** Creates a constant vector with the specified logical row count. */
   private CometVector createConstantVector(int numRows) {
-    ValueVector arrowVector = createArrowVector(numRows);
-    return new CometPlainVector(arrowVector, useDecimal128);
-  }
-
-  /** Creates an Arrow vector filled with constant values. */
-  private ValueVector createArrowVector(int numRows) {
-    if (isNull) {
-      return createNullVector(numRows);
-    }
-
-    if (type == DataTypes.BooleanType) {
-      return createBooleanVector(numRows, (Boolean) value);
-    } else if (type == DataTypes.ByteType) {
-      return createByteVector(numRows, (Byte) value);
-    } else if (type == DataTypes.ShortType) {
-      return createShortVector(numRows, (Short) value);
-    } else if (type == DataTypes.IntegerType) {
-      return createIntVector(numRows, (Integer) value);
-    } else if (type == DataTypes.LongType) {
-      return createLongVector(numRows, (Long) value);
-    } else if (type == DataTypes.FloatType) {
-      return createFloatVector(numRows, (Float) value);
-    } else if (type == DataTypes.DoubleType) {
-      return createDoubleVector(numRows, (Double) value);
-    } else if (type == DataTypes.StringType) {
-      return createStringVector(numRows, (UTF8String) value);
-    } else if (type == DataTypes.BinaryType) {
-      return createBinaryVector(numRows, (byte[]) value);
-    } else if (type == DataTypes.DateType) {
-      return createDateVector(numRows, (Integer) value);
-    } else if (type == DataTypes.TimestampType || type == TimestampNTZType$.MODULE$) {
-      return createTimestampVector(numRows, (Long) value);
-    } else if (type instanceof DecimalType) {
-      return createDecimalVector(numRows, (Decimal) value, (DecimalType) type);
-    } else {
-      throw new UnsupportedOperationException("Unsupported Spark type: " + type);
-    }
-  }
-
-  private ValueVector createNullVector(int numRows) {
-    NullVector vector = new NullVector(arrowField.getName(), numRows);
-    return vector;
-  }
-
-  private ValueVector createBooleanVector(int numRows, boolean value) {
-    BitVector vector = new BitVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value ? 1 : 0);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createByteVector(int numRows, byte value) {
-    TinyIntVector vector = new TinyIntVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createShortVector(int numRows, short value) {
-    SmallIntVector vector = new SmallIntVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createIntVector(int numRows, int value) {
-    IntVector vector = new IntVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createLongVector(int numRows, long value) {
-    BigIntVector vector = new BigIntVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createFloatVector(int numRows, float value) {
-    Float4Vector vector = new Float4Vector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createDoubleVector(int numRows, double value) {
-    Float8Vector vector = new Float8Vector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createStringVector(int numRows, UTF8String value) {
-    VarCharVector vector = new VarCharVector(arrowField, allocator);
-    byte[] bytes = value.getBytes();
-    vector.allocateNew((long) bytes.length * numRows, numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, bytes);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createBinaryVector(int numRows, byte[] value) {
-    VarBinaryVector vector = new VarBinaryVector(arrowField, allocator);
-    vector.allocateNew((long) value.length * numRows, numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createDateVector(int numRows, int value) {
-    DateDayVector vector = new DateDayVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createTimestampVector(int numRows, long value) {
-    TimeStampMicroTZVector vector = new TimeStampMicroTZVector(arrowField, allocator);
-    vector.allocateNew(numRows);
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, value);
-    }
-    vector.setValueCount(numRows);
-    return vector;
-  }
-
-  private ValueVector createDecimalVector(int numRows, Decimal value, DecimalType dt) {
-    DecimalVector vector =
-        new DecimalVector(arrowField.getName(), allocator, dt.precision(), dt.scale());
-    vector.allocateNew(numRows);
-
-    java.math.BigDecimal bigDecimal = value.toJavaBigDecimal();
-    for (int i = 0; i < numRows; i++) {
-      vector.set(i, bigDecimal);
-    }
-    vector.setValueCount(numRows);
-    return vector;
+    return new CometConstantVector(type, arrowField, useDecimal128, value, isNull, numRows);
   }
 
   /** Converts a Spark StructField to an Arrow Field. */