diff --git a/mllib/src/main/scala/org/apache/spark/ml/functions.scala b/mllib/src/main/scala/org/apache/spark/ml/functions.scala
index 87a8a7d98ea0c..12c073d8ae6ef 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/functions.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/functions.scala
@@ -18,10 +18,10 @@
 package org.apache.spark.ml
 
 import org.apache.spark.annotation.Since
-import org.apache.spark.ml.linalg.{DenseVector, SparseVector, Vector}
+import org.apache.spark.ml.linalg.{DenseVector, SparseVector, Vector, VectorUDT}
 import org.apache.spark.sql.{functions => sf}
 import org.apache.spark.sql.Column
-import org.apache.spark.sql.types.{ArrayType, IntegerType}
+import org.apache.spark.sql.types.{ArrayType, DoubleType, IntegerType}
 
 // scalastyle:off
 @Since("3.0.0")
@@ -65,24 +65,58 @@ object functions {
     Column.internalFn("ml_vector_dot_product", sf.unwrap_udt(left), sf.unwrap_udt(right))
 
   private[ml] def vector_dot_product(left: Column, right: Vector): Column = {
-    val rightStruct = right match {
-      case sparse: SparseVector =>
-        sf.struct(
-          sf.lit(0.toByte).alias("type"),
-          sf.lit(sparse.size).alias("size"),
-          sf.lit(sparse.indices).alias("indices"),
-          sf.lit(sparse.values).alias("values"))
-      case dense: DenseVector =>
-        sf.struct(
-          sf.lit(1.toByte).alias("type"),
-          sf.lit(null).cast(IntegerType).alias("size"),
-          sf.lit(null).cast(ArrayType(IntegerType)).alias("indices"),
-          sf.lit(dense.values).alias("values"))
-    }
     Column.internalFn(
       "ml_vector_dot_product",
       sf.unwrap_udt(left),
-      rightStruct)
+      vectorToStruct(right))
+  }
+
+  private[ml] def vector_scale_shift(
+      vector: Column,
+      scale: Column,
+      shift: Column): Column = {
+    val transformed = Column.internalFn(
+      "ml_vector_scale_shift",
+      sf.unwrap_udt(vector),
+      scale,
+      shift)
+    sf.wrap_udt(transformed, new VectorUDT)
+  }
+
+  private[ml] def vector_scale_shift(
+      vector: Column,
+      scale: Array[Double],
+      shift: Array[Double]): Column = {
+    val transformed = Column.internalFn(
+      "ml_vector_scale_shift",
+      sf.unwrap_udt(vector),
+      doubleArrayLiteral(scale),
+      doubleArrayLiteral(shift))
+    sf.wrap_udt(transformed, new VectorUDT)
+  }
+
+  private def doubleArrayLiteral(values: Array[Double]): Column = {
+    if (values == null) {
+      sf.lit(null).cast(ArrayType(DoubleType, containsNull = false))
+    } else {
+      sf.typedLit(values)
+    }
+  }
+
+  private def vectorToStruct(vector: Vector): Column = vector match {
+    case null => sf.lit(null).cast(new VectorUDT().sqlType)
+    case sparse: SparseVector =>
+      sf.struct(
+        sf.lit(0.toByte).alias("type"),
+        sf.lit(sparse.size).alias("size"),
+        sf.lit(sparse.indices).alias("indices"),
+        sf.lit(sparse.values).alias("values"))
+    case dense: DenseVector =>
+      sf.struct(
+        sf.lit(1.toByte).alias("type"),
+        sf.lit(null).cast(IntegerType).alias("size"),
+        sf.lit(null).cast(ArrayType(IntegerType)).alias("indices"),
+        sf.lit(dense.values).alias("values"))
   }
 
   private[ml] def array_binary_search(a: Column, v: Column): Column =
diff --git a/mllib/src/test/scala/org/apache/spark/ml/FunctionsSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/FunctionsSuite.scala
index ddc70e532b828..3ca352df81710 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/FunctionsSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/FunctionsSuite.scala
@@ -24,9 +24,9 @@ import org.apache.spark.ml.util.MLTest
 import org.apache.spark.mllib.linalg.{Matrices => OldMatrices, MatrixUDT => OldMatrixUDT,
   Vector => OldVector, Vectors => OldVectors, VectorUDT => OldVectorUDT}
 import org.apache.spark.sql.{AnalysisException, DataFrame, Row}
-import org.apache.spark.sql.catalyst.expressions.ml.VectorPosExplode
-import org.apache.spark.sql.functions.{col, unwrap_udt, wrap_udt}
-import org.apache.spark.sql.types.{StructField, StructType, UserDefinedType}
+import org.apache.spark.sql.catalyst.expressions.ml.{VectorPosExplode, VectorScaleShift}
+import org.apache.spark.sql.functions.{col, typedLit, unwrap_udt, wrap_udt}
+import org.apache.spark.sql.types.{ArrayType, DoubleType, StructField, StructType, UserDefinedType}
 
 class FunctionsSuite extends MLTest {
 
@@ -256,6 +256,131 @@ class FunctionsSuite extends MLTest {
     assert(error.getMessage.contains("vectors with non-matching sizes"))
   }
 
+  test("test vector_scale_shift") {
+    val df = Seq(
+      (Vectors.dense(1.0, 2.0), Array(2.0, 3.0), Array(4.0, 5.0)),
+      (Vectors.sparse(2, Seq((0, 1.0))), Array(2.0, 3.0), null),
+      (Vectors.dense(1.0, 2.0), null, Array(4.0, 5.0)),
+      (Vectors.sparse(2, Seq((0, 1.0))), Array(2.0, 3.0), Array(0.0, 0.0)),
+      (Vectors.sparse(2, Seq((0, 1.0))), null, null),
+      (null, Array(2.0, 3.0), Array(4.0, 5.0)))
+      .toDF("vector", "scale", "shift")
+
+    assert(df.schema("scale").dataType === ArrayType(DoubleType, containsNull = false))
+    assert(df.schema("shift").dataType === ArrayType(DoubleType, containsNull = false))
+    val transformed = df.select(vector_scale_shift($"vector", $"scale", $"shift"))
+    assert(transformed.schema.head.dataType === new VectorUDT)
+    assert(transformed.collect().map(_.get(0)).toSeq === Seq(
+      Vectors.dense(6.0, 11.0),
+      Vectors.sparse(2, Seq((0, 2.0))),
+      Vectors.dense(5.0, 7.0),
+      Vectors.dense(2.0, 0.0),
+      Vectors.sparse(2, Seq((0, 1.0))),
+      null))
+
+    val expressions = transformed.queryExecution.analyzed
+      .flatMap(_.expressions.flatMap(_.collect { case v: VectorScaleShift => v }))
+    assert(expressions.map(_.prettyName).distinct === Seq("ml_vector_scale_shift"))
+
+    val constantResult = df.limit(1)
+      .select(vector_scale_shift(
+        $"vector",
+        Array(2.0, 3.0),
+        Array(4.0, 5.0)))
+      .first()
+      .getAs[Vector](0)
+    assert(constantResult === Vectors.dense(6.0, 11.0))
+
+    val cachedScaleResult = df.limit(1)
+      .select(vector_scale_shift($"vector", typedLit(Array(2.0, 3.0)), $"shift"))
+      .first()
+      .getAs[Vector](0)
+    assert(cachedScaleResult === Vectors.dense(6.0, 11.0))
+
+    val cachedScaleWithNullShiftResult = df
+      .where($"scale".isNotNull && $"shift".isNull)
+      .select(vector_scale_shift($"vector", typedLit(Array(2.0, 3.0)), $"shift"))
+      .first()
+      .getAs[Vector](0)
+    assert(cachedScaleWithNullShiftResult === Vectors.sparse(2, Seq((0, 2.0))))
+
+    val cachedShiftResult = df.limit(1)
+      .select(vector_scale_shift($"vector", $"scale", typedLit(Array(4.0, 5.0))))
+      .first()
+      .getAs[Vector](0)
+    assert(cachedShiftResult === Vectors.dense(6.0, 11.0))
+
+    val nullScaleWithCachedShiftResult = df
+      .where($"scale".isNull && $"shift".isNotNull)
+      .select(vector_scale_shift($"vector", $"scale", typedLit(Array(4.0, 5.0))))
+      .first()
+      .getAs[Vector](0)
+    assert(nullScaleWithCachedShiftResult === Vectors.dense(5.0, 7.0))
+
+    val scaleOnlyConstantResult = df.limit(1)
+      .select(vector_scale_shift(
+        $"vector",
+        Array(2.0, 3.0),
+        null.asInstanceOf[Array[Double]]))
+      .first()
+      .getAs[Vector](0)
+    assert(scaleOnlyConstantResult === Vectors.dense(2.0, 6.0))
+
+    val shiftOnlyConstantResult = df.limit(1)
+      .select(vector_scale_shift(
+        $"vector",
+        null.asInstanceOf[Array[Double]],
+        Array(4.0, 5.0)))
+      .first()
+      .getAs[Vector](0)
+    assert(shiftOnlyConstantResult === Vectors.dense(5.0, 7.0))
+
+    val nullConstantsResult = df.limit(1)
+      .select(vector_scale_shift(
+        $"vector",
+        null.asInstanceOf[Array[Double]],
+        null.asInstanceOf[Array[Double]]))
+      .first()
+      .getAs[Vector](0)
+    assert(nullConstantsResult === Vectors.dense(1.0, 2.0))
+
+    val emptyConstantsResult = Seq(Tuple1(Vectors.dense(Array.emptyDoubleArray)))
+      .toDF("vector")
+      .select(vector_scale_shift($"vector", Array.emptyDoubleArray, Array.emptyDoubleArray))
+      .first()
+      .getAs[Vector](0)
+    assert(emptyConstantsResult === Vectors.dense(Array.emptyDoubleArray))
+
+    val specialValues = Seq(Double.NaN, Double.NegativeInfinity, Double.PositiveInfinity)
+    val specialValueRows = specialValues.flatMap { value =>
+      Seq(
+        (Vectors.dense(value), Array(1.0), Array(0.0)),
+        (Vectors.dense(1.0), Array(value), Array(0.0)),
+        (Vectors.dense(1.0), Array(1.0), Array(value)))
+    }
+    val specialValueResults = specialValueRows
+      .toDF("vector", "scale", "shift")
+      .select(vector_scale_shift($"vector", $"scale", $"shift"))
+      .collect()
+      .map(_.getAs[Vector](0)(0))
+    specialValueResults.zip(specialValues.flatMap(value => Seq.fill(3)(value)))
+      .foreach { case (actual, expected) =>
+        assert(java.lang.Double.compare(actual, expected) === 0)
+      }
+
+    Seq(
+      (Array(1.0, 2.0), Array(0.0)),
+      (Array(1.0), Array(0.0, 0.0))).foreach { case (scale, shift) =>
+      val error = intercept[IllegalArgumentException] {
+        Seq((Vectors.dense(1.0), scale, shift))
+          .toDF("vector", "scale", "shift")
+          .select(vector_scale_shift($"vector", $"scale", $"shift"))
+          .collect()
+      }
+      assert(error.getMessage.contains("inputs with non-matching sizes"))
+    }
+  }
+
   test("test get_vector") {
     val df = Seq(
       (Vectors.dense(1.0, 2.0, 3.0), 0),
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/ml/MLExpressionUtils.java b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/ml/MLExpressionUtils.java
new file mode 100644
index 0000000000000..1427775d28556
--- /dev/null
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/ml/MLExpressionUtils.java
@@ -0,0 +1,181 @@
+/*
+ * 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.catalyst.expressions.ml;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.catalyst.expressions.UnsafeArrayData;
+import org.apache.spark.sql.catalyst.util.ArrayData;
+
+public class MLExpressionUtils {
+  private static final byte SPARSE_VECTOR_TYPE = 0;
+  private static final byte DENSE_VECTOR_TYPE = 1;
+
+  private MLExpressionUtils() {}
+
+  public static InternalRow scaleShift(
+      InternalRow vector,
+      ArrayData scale,
+      ArrayData shift) {
+    return scaleShift(vector, scale, shift, null, null);
+  }
+
+  public static InternalRow scaleShift(
+      InternalRow vector,
+      ArrayData scale,
+      ArrayData shift,
+      double[] cachedScale,
+      double[] cachedShift) {
+    boolean hasScale = scale != null || cachedScale != null;
+    boolean hasShift = shift != null || cachedShift != null;
+    if (!hasScale && !hasShift) {
+      return vector;
+    }
+
+    byte vectorType = vector.getByte(0);
+    ArrayData vectorValues = vector.getArray(3);
+    int size;
+    if (vectorType == SPARSE_VECTOR_TYPE) {
+      size = vector.getInt(1);
+    } else if (vectorType == DENSE_VECTOR_TYPE) {
+      size = vectorValues.numElements();
+    } else {
+      throw new IllegalArgumentException("Unknown vector type " + vectorType + ".");
+    }
+
+    int scaleSize = !hasScale ? size :
+      (cachedScale == null ? scale.numElements() : cachedScale.length);
+    int shiftSize = !hasShift ? size :
+      (cachedShift == null ? shift.numElements() : cachedShift.length);
+    if (size != scaleSize || size != shiftSize) {
+      throw new IllegalArgumentException(
+        "requirement failed: VectorScaleShift was given inputs with non-matching sizes: " +
+          "vector.size = " + size + ", scale.size = " + scaleSize +
+          ", shift.size = " + shiftSize);
+    }
+
+    if (vectorType == SPARSE_VECTOR_TYPE && !hasShift) {
+      ArrayData vectorIndices = vector.getArray(2);
+      double[] resultValues = new double[vectorValues.numElements()];
+      if (cachedScale == null) {
+        for (int vectorIndex = 0; vectorIndex < resultValues.length; vectorIndex++) {
+          int featureIndex = vectorIndices.getInt(vectorIndex);
+          resultValues[vectorIndex] =
+            vectorValues.getDouble(vectorIndex) * scale.getDouble(featureIndex);
+        }
+      } else {
+        for (int vectorIndex = 0; vectorIndex < resultValues.length; vectorIndex++) {
+          int featureIndex = vectorIndices.getInt(vectorIndex);
+          resultValues[vectorIndex] =
+            vectorValues.getDouble(vectorIndex) * cachedScale[featureIndex];
+        }
+      }
+      return new GenericInternalRow(new Object[] {
+        SPARSE_VECTOR_TYPE,
+        size,
+        vectorIndices,
+        UnsafeArrayData.fromPrimitiveArray(resultValues)
+      });
+    }
+
+    double[] resultValues = new double[size];
+    if (vectorType == DENSE_VECTOR_TYPE) {
+      if (!hasScale) {
+        if (cachedShift == null) {
+          for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+            resultValues[featureIndex] =
+              vectorValues.getDouble(featureIndex) + shift.getDouble(featureIndex);
+          }
+        } else {
+          for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+            resultValues[featureIndex] =
+              vectorValues.getDouble(featureIndex) + cachedShift[featureIndex];
+          }
+        }
+      } else if (!hasShift) {
+        if (cachedScale == null) {
+          for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+            resultValues[featureIndex] =
+              vectorValues.getDouble(featureIndex) * scale.getDouble(featureIndex);
+          }
+        } else {
+          for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+            resultValues[featureIndex] =
+              vectorValues.getDouble(featureIndex) * cachedScale[featureIndex];
+          }
+        }
+      } else if (cachedScale != null && cachedShift != null) {
+        for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+          resultValues[featureIndex] = vectorValues.getDouble(featureIndex) *
+            cachedScale[featureIndex] + cachedShift[featureIndex];
+        }
+      } else if (cachedScale != null) {
+        for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+          resultValues[featureIndex] = vectorValues.getDouble(featureIndex) *
+            cachedScale[featureIndex] + shift.getDouble(featureIndex);
+        }
+      } else if (cachedShift != null) {
+        for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+          resultValues[featureIndex] = vectorValues.getDouble(featureIndex) *
+            scale.getDouble(featureIndex) + cachedShift[featureIndex];
+        }
+      } else {
+        for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+          resultValues[featureIndex] = vectorValues.getDouble(featureIndex) *
+            scale.getDouble(featureIndex) + shift.getDouble(featureIndex);
+        }
+      }
+    } else {
+      if (cachedShift == null) {
+        for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+          resultValues[featureIndex] = 0.0 + shift.getDouble(featureIndex);
+        }
+      } else {
+        for (int featureIndex = 0; featureIndex < size; featureIndex++) {
+          resultValues[featureIndex] = 0.0 + cachedShift[featureIndex];
+        }
+      }
+
+      ArrayData vectorIndices = vector.getArray(2);
+      if (!hasScale) {
+        for (int vectorIndex = 0; vectorIndex < vectorValues.numElements(); vectorIndex++) {
+          int featureIndex = vectorIndices.getInt(vectorIndex);
+          resultValues[featureIndex] += vectorValues.getDouble(vectorIndex);
+        }
+      } else if (cachedScale == null) {
+        for (int vectorIndex = 0; vectorIndex < vectorValues.numElements(); vectorIndex++) {
+          int featureIndex = vectorIndices.getInt(vectorIndex);
+          resultValues[featureIndex] +=
+            vectorValues.getDouble(vectorIndex) * scale.getDouble(featureIndex);
+        }
+      } else {
+        for (int vectorIndex = 0; vectorIndex < vectorValues.numElements(); vectorIndex++) {
+          int featureIndex = vectorIndices.getInt(vectorIndex);
+          resultValues[featureIndex] +=
+            vectorValues.getDouble(vectorIndex) * cachedScale[featureIndex];
+        }
+      }
+    }
+    return new GenericInternalRow(new Object[] {
+      DENSE_VECTOR_TYPE,
+      null,
+      null,
+      UnsafeArrayData.fromPrimitiveArray(resultValues)
+    });
+  }
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index a31e7ac4d74c7..26287782cb29a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -1222,8 +1222,9 @@ object FunctionRegistry {
     registerInternalExpression[NullIndex]("null_index")
     registerInternalExpression[CastTimestampNTZToLong]("timestamp_ntz_to_long")
     registerInternalExpression[ArrayBinarySearch]("array_binary_search")
-    registerInternalExpression[VectorPosExplode]("ml_vector_posexplode")
+    registerInternalExpression[VectorScaleShift]("ml_vector_scale_shift")
     registerInternalExpression[VectorDotProduct]("ml_vector_dot_product")
+    registerInternalExpression[VectorPosExplode]("ml_vector_posexplode")
   }
 
   registerInternalExpressions()
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ml/VectorScaleShift.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ml/VectorScaleShift.scala
new file mode 100644
index 0000000000000..8d9f758d898da
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ml/VectorScaleShift.scala
@@ -0,0 +1,134 @@
+/*
+ * 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.catalyst.expressions.ml
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes, Expression, Literal, TernaryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.catalyst.util.ArrayData
+import org.apache.spark.sql.types._
+
+/**
+ * Applies element-wise scaling and shifting to SQL struct representations of MLlib vectors:
+ * `vector(i) * scale(i) + shift(i)`. This expression is dedicated only for Spark ML and should be
+ * used together with `unwrap_udt` and `wrap_udt`. A null scale is treated as an identity scale, and
+ * a null shift is treated as a zero shift. If both are null, the input vector is returned
+ * unchanged.
+ */
+case class VectorScaleShift(
+    vector: Expression,
+    scale: Expression,
+    shift: Expression)
+  extends TernaryExpression with ExpectsInputTypes {
+
+  override def first: Expression = vector
+  override def second: Expression = scale
+  override def third: Expression = shift
+
+  override def prettyName: String = "ml_vector_scale_shift"
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(
+    VectorScaleShift.vectorSqlType,
+    VectorScaleShift.NonNullableDoubleArrayType,
+    VectorScaleShift.NonNullableDoubleArrayType)
+
+  override def dataType: DataType = VectorScaleShift.vectorSqlType
+
+  override def nullable: Boolean = vector.nullable
+
+  override def eval(input: InternalRow): Any = {
+    val vectorInput = vector.eval(input)
+    if (vectorInput == null) {
+      null
+    } else {
+      MLExpressionUtils.scaleShift(
+        vectorInput.asInstanceOf[InternalRow],
+        scale.eval(input).asInstanceOf[ArrayData],
+        shift.eval(input).asInstanceOf[ArrayData])
+    }
+  }
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val utils = classOf[MLExpressionUtils].getName
+    val vectorJavaType = CodeGenerator.javaType(dataType)
+    val arrayJavaType = CodeGenerator.javaType(VectorScaleShift.doubleArraySqlType)
+    val vectorGen = vector.genCode(ctx)
+    val scaleInput = ctx.freshName("scaleInput")
+    val shiftInput = ctx.freshName("shiftInput")
+    val (scaleCode, cachedScale) = scale match {
+      case Literal(value: ArrayData, _) =>
+        (code"$arrayJavaType $scaleInput = null;",
+          ctx.addReferenceObj("cachedScale", value.toDoubleArray(), "double[]"))
+      case _ =>
+        val scaleGen = scale.genCode(ctx)
+        (code"""
+          ${scaleGen.code}
+          $arrayJavaType $scaleInput = ${scaleGen.isNull} ? null : ${scaleGen.value};
+        """, "null")
+    }
+    val (shiftCode, cachedShift) = shift match {
+      case Literal(value: ArrayData, _) =>
+        (code"$arrayJavaType $shiftInput = null;",
+          ctx.addReferenceObj("cachedShift", value.toDoubleArray(), "double[]"))
+      case _ =>
+        val shiftGen = shift.genCode(ctx)
+        (code"""
+          ${shiftGen.code}
+          $arrayJavaType $shiftInput = ${shiftGen.isNull} ? null : ${shiftGen.value};
+        """, "null")
+    }
+
+    ev.copy(code = code"""
+      ${vectorGen.code}
+      boolean ${ev.isNull} = ${vectorGen.isNull};
+      $vectorJavaType ${ev.value} = null;
+      if (!${ev.isNull}) {
+        $scaleCode
+        $shiftCode
+        ${ev.value} = $utils.scaleShift(
+          ${vectorGen.value}, $scaleInput, $shiftInput, $cachedScale, $cachedShift);
+      }
+    """)
+  }
+
+  override protected def withNewChildrenInternal(
+      newVector: Expression,
+      newScale: Expression,
+      newShift: Expression): VectorScaleShift = {
+    copy(vector = newVector, scale = newScale, shift = newShift)
+  }
+}
+
+object VectorScaleShift {
+  private[ml] val vectorSqlType = StructType(Array(
+    StructField("type", ByteType, nullable = false),
+    StructField("size", IntegerType, nullable = true),
+    StructField("indices", ArrayType(IntegerType, containsNull = false), nullable = true),
+    StructField("values", ArrayType(DoubleType, containsNull = false), nullable = true)))
+
+  private[ml] val doubleArraySqlType = ArrayType(DoubleType, containsNull = false)
+
+  private object NonNullableDoubleArrayType extends AbstractDataType {
+    override private[sql] def defaultConcreteType: DataType = doubleArraySqlType
+
+    override private[sql] def acceptsType(other: DataType): Boolean = other == doubleArraySqlType
+
+    override private[spark] def simpleString: String = doubleArraySqlType.simpleString
+  }
+}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ml/VectorScaleShiftSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ml/VectorScaleShiftSuite.scala
new file mode 100644
index 0000000000000..3eedefa74e73c
--- /dev/null
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/ml/VectorScaleShiftSuite.scala
@@ -0,0 +1,179 @@
+/*
+ * 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.catalyst.expressions.ml
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{ExpressionEvalHelper, GenericInternalRow, Literal, UnsafeArrayData}
+import org.apache.spark.sql.types.{ArrayType, DoubleType}
+
+class VectorScaleShiftSuite extends SparkFunSuite with ExpressionEvalHelper {
+  private val vectorSqlType = VectorScaleShift.vectorSqlType
+  private val doubleArraySqlType = VectorScaleShift.doubleArraySqlType
+
+  private def denseRow(values: Double*): InternalRow = {
+    new GenericInternalRow(Array[Any](
+      1.toByte,
+      null,
+      null,
+      UnsafeArrayData.fromPrimitiveArray(values.toArray)))
+  }
+
+  private def dense(values: Double*): Literal = Literal(denseRow(values: _*), vectorSqlType)
+
+  private def sparseRow(size: Int, indices: Array[Int], values: Array[Double]): InternalRow = {
+    new GenericInternalRow(Array[Any](
+      0.toByte,
+      size,
+      UnsafeArrayData.fromPrimitiveArray(indices),
+      UnsafeArrayData.fromPrimitiveArray(values)))
+  }
+
+  private def sparse(size: Int, indices: Array[Int], values: Array[Double]): Literal = {
+    Literal(sparseRow(size, indices, values), vectorSqlType)
+  }
+
+  private def array(values: Double*): Literal = {
+    Literal(UnsafeArrayData.fromPrimitiveArray(values.toArray), doubleArraySqlType)
+  }
+
+  test("vector scale shift interpreted and code-generated evaluation") {
+    val expression = VectorScaleShift(
+      dense(1.0, 2.0, 3.0),
+      array(2.0, 3.0, 4.0),
+      array(5.0, 6.0, 7.0))
+    assert(expression.prettyName === "ml_vector_scale_shift")
+    checkEvaluation(expression, denseRow(7.0, 12.0, 19.0))
+
+    checkEvaluation(
+      VectorScaleShift(
+        dense(1.0, 2.0, 3.0),
+        array(2.0, 0.0, 4.0),
+        array(0.0, 1.0, 0.0)),
+      denseRow(2.0, 1.0, 12.0))
+  }
+
+  test("vector scale shift produces a dense vector for a non-null shift") {
+    val vector = sparse(3, Array(0, 2), Array(1.0, 3.0))
+
+    checkEvaluation(
+      VectorScaleShift(vector, array(2.0, 3.0, 4.0), array(0.0, 0.0, 0.0)),
+      denseRow(2.0, 0.0, 12.0))
+    checkEvaluation(
+      VectorScaleShift(
+        vector,
+        array(2.0, 3.0, 4.0),
+        array(0.0, 1.0, 0.0)),
+      denseRow(2.0, 1.0, 12.0))
+  }
+
+  test("vector scale shift with a null vector") {
+    val nullVector = Literal(null, vectorSqlType)
+    val values = array(1.0)
+
+    checkEvaluation(VectorScaleShift(nullVector, values, values), null)
+  }
+
+  test("vector scale shift with a null scale") {
+    val nullArray = Literal(null, doubleArraySqlType)
+
+    checkEvaluation(
+      VectorScaleShift(dense(1.0, 2.0), nullArray, array(3.0, 4.0)),
+      denseRow(4.0, 6.0))
+    checkEvaluation(
+      VectorScaleShift(
+        sparse(3, Array(0, 2), Array(1.0, 3.0)),
+        nullArray,
+        array(0.0, 2.0, 0.0)),
+      denseRow(1.0, 2.0, 3.0))
+  }
+
+  test("vector scale shift with a null shift") {
+    val nullArray = Literal(null, doubleArraySqlType)
+
+    checkEvaluation(
+      VectorScaleShift(dense(1.0, 2.0), array(3.0, 4.0), nullArray),
+      denseRow(3.0, 8.0))
+    checkEvaluation(
+      VectorScaleShift(
+        sparse(3, Array(0, 2), Array(1.0, 3.0)),
+        array(2.0, 3.0, 4.0),
+        nullArray),
+      sparseRow(3, Array(0, 2), Array(2.0, 12.0)))
+  }
+
+  test("vector scale shift with a null scale and shift") {
+    val nullArray = Literal(null, doubleArraySqlType)
+
+    checkEvaluation(
+      VectorScaleShift(dense(1.0, 2.0), nullArray, nullArray),
+      denseRow(1.0, 2.0))
+    checkEvaluation(
+      VectorScaleShift(
+        sparse(3, Array(0, 2), Array(1.0, 3.0)),
+        nullArray,
+        nullArray),
+      sparseRow(3, Array(0, 2), Array(1.0, 3.0)))
+  }
+
+  test("vector scale shift with empty vectors") {
+    val emptySparse = sparse(0, Array.emptyIntArray, Array.emptyDoubleArray)
+    val emptyArray = array()
+
+    checkEvaluation(
+      VectorScaleShift(dense(), emptyArray, emptyArray),
+      denseRow())
+    checkEvaluation(
+      VectorScaleShift(emptySparse, emptyArray, emptyArray),
+      denseRow())
+  }
+
+  test("vector scale shift with infinite and NaN values") {
+    Seq(Double.PositiveInfinity, Double.NegativeInfinity, Double.NaN).foreach { value =>
+      checkEvaluation(
+        VectorScaleShift(dense(value), array(1.0), array(0.0)),
+        denseRow(value))
+      checkEvaluation(
+        VectorScaleShift(dense(1.0), array(value), array(0.0)),
+        denseRow(value))
+      checkEvaluation(
+        VectorScaleShift(dense(1.0), array(1.0), array(value)),
+        denseRow(value))
+    }
+  }
+
+  test("vector scale shift rejects inputs with different sizes") {
+    checkExceptionInExpression[IllegalArgumentException](
+      VectorScaleShift(dense(1.0), array(1.0, 2.0), array(1.0)),
+      "inputs with non-matching sizes")
+    checkExceptionInExpression[IllegalArgumentException](
+      VectorScaleShift(dense(1.0), array(1.0), array(1.0, 2.0)),
+      "inputs with non-matching sizes")
+  }
+
+  test("vector scale shift requires arrays without null elements") {
+    val nullableArray = Literal(
+      UnsafeArrayData.fromPrimitiveArray(Array(1.0)),
+      ArrayType(DoubleType, containsNull = true))
+
+    assert(VectorScaleShift(dense(1.0), nullableArray, array(0.0))
+      .checkInputDataTypes().isFailure)
+    assert(VectorScaleShift(dense(1.0), array(1.0), nullableArray)
+      .checkInputDataTypes().isFailure)
+  }
+}