[SPARK-59443][SQL] Reduce allocations in ArrowColumnVector decimal reads

sql/catalyst/src/main/java/org/apache/spark/sql/vectorized/ArrowColumnVector.java

@@ -17,6 +17,9 @@
 
 package org.apache.spark.sql.vectorized;
 
+import java.math.BigDecimal;
+import java.nio.ByteOrder;
+
 import org.apache.arrow.memory.ArrowBuf;
 import org.apache.arrow.vector.*;
 import org.apache.arrow.vector.complex.*;
@@ -202,7 +205,12 @@ void initAccessor(ValueVector vector) {
     } else if (vector instanceof Float8Vector float8Vector) {
       accessor = new DoubleAccessor(float8Vector);
     } else if (vector instanceof DecimalVector decimalVector) {
-      accessor = new DecimalAccessor(decimalVector);
+      int precision = decimalVector.getPrecision();
+      if (precision > 0 && precision <= Decimal.MAX_LONG_DIGITS()) {
+        accessor = new SmallDecimalAccessor(decimalVector, ByteOrder.nativeOrder());
+      } else {
+        accessor = new DecimalAccessor(decimalVector);
+      }
     } else if (vector instanceof VarCharVector varCharVector) {
       accessor = new StringAccessor(varCharVector);
     } else if (vector instanceof LargeVarCharVector largeVarCharVector) {
@@ -444,6 +452,37 @@ final double getDouble(int rowId) {
     }
   }
 
+  static class SmallDecimalAccessor extends ArrowVectorAccessor {
+
+    private final DecimalVector accessor;
+    private final int lowWordOffset;
+    private final int highWordOffset;
+
+    SmallDecimalAccessor(DecimalVector vector, ByteOrder byteOrder) {
+      super(vector);
+      this.accessor = vector;
+      this.lowWordOffset = byteOrder == ByteOrder.LITTLE_ENDIAN ? 0 : Long.BYTES;
+      this.highWordOffset = byteOrder == ByteOrder.LITTLE_ENDIAN ? Long.BYTES : 0;
+    }
+
+    @Override
+    final Decimal getDecimal(int rowId, int precision, int scale) {
+      if (isNullAt(rowId)) return null;
+      // Arrow stores Decimal128 values in native byte order.
+      long offset = (long) rowId * DecimalVector.TYPE_WIDTH;
+      ArrowBuf data = accessor.getDataBuffer();
+      long unscaled = data.getLong(offset + lowWordOffset);
+      long high = data.getLong(offset + highWordOffset);
+      // Preserve full-width values even when they exceed the declared source precision.
+      if (high == (unscaled >> 63)) {
+        // Keep BigDecimal-backed Decimal semantics, including checked integral conversions.
+        return Decimal.apply(
+          BigDecimal.valueOf(unscaled, accessor.getScale()), precision, scale);
+      }
+      return Decimal.apply(accessor.getObject(rowId), precision, scale);
+    }
+  }
+
   static class DecimalAccessor extends ArrowVectorAccessor {
 
     private final DecimalVector accessor;

sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/ArrowDecimalReadBenchmark.scala

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.benchmark
+
+import java.math.{BigDecimal => JavaBigDecimal}
+
+import scala.util.Using
+
+import org.apache.arrow.memory.RootAllocator
+import org.apache.arrow.vector.DecimalVector
+
+import org.apache.spark.benchmark.{Benchmark, BenchmarkBase}
+import org.apache.spark.sql.types.Decimal
+import org.apache.spark.sql.vectorized.ArrowColumnVector
+
+/**
+ * Measures Decimal128 reads through ArrowColumnVector, including scale conversion and wide values.
+ * To run this benchmark:
+ * {{{
+ *   build/sbt "sql/Test/runMain org.apache.spark.sql.execution.benchmark.ArrowDecimalReadBenchmark"
+ * }}}
+ * Set SPARK_GENERATE_BENCHMARK_FILES=1 to save results under benchmarks/.
+ */
+object ArrowDecimalReadBenchmark extends BenchmarkBase {
+  private val BatchSize = 4096
+  private val BatchesPerIteration = 256
+  private val results = new Array[Decimal](BatchSize)
+
+  private def readDecimals(precision: Int, sourceScale: Int, targetScale: Int): Unit = {
+    Using.resource(new RootAllocator()) { allocator =>
+      Using.resource(new DecimalVector("decimal", allocator, precision, sourceScale)) { vector =>
+        vector.allocateNew(BatchSize)
+        val base = BigInt(10).pow(precision - 1)
+        for (row <- 0 until BatchSize) {
+          val unscaled = (base + row) * (if (row % 2 == 0) 1 else -1)
+          vector.set(row, new JavaBigDecimal(unscaled.bigInteger, sourceScale))
+        }
+        vector.setValueCount(BatchSize)
+        val column = new ArrowColumnVector(vector)
+        val benchmark = new Benchmark(
+          s"decimal($precision,$sourceScale) to decimal($precision,$targetScale)",
+          BatchSize.toLong * BatchesPerIteration, output = output)
+        benchmark.addCase("Arrow getObject") { _ =>
+          var batch = 0
+          while (batch < BatchesPerIteration) {
+            var row = 0
+            while (row < BatchSize) {
+              results(row) = if (vector.isNull(row)) null else {
+                Decimal(vector.getObject(row), precision, targetScale)
+              }
+              row += 1
+            }
+            batch += 1
+          }
+        }
+        benchmark.addCase("ArrowColumnVector") { _ =>
+          var batch = 0
+          while (batch < BatchesPerIteration) {
+            var row = 0
+            while (row < BatchSize) {
+              results(row) = column.getDecimal(row, precision, targetScale)
+              row += 1
+            }
+            batch += 1
+          }
+        }
+        benchmark.run()
+      }
+    }
+  }
+
+  override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+    runBenchmark("Arrow decimal reads") {
+      readDecimals(18, 2, 2)
+      readDecimals(18, 2, 1)
+      readDecimals(38, 2, 2)
+    }
+  }
+}