diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializer.scala
index 9529e85698537..9de3130310535 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializer.scala
@@ -31,7 +31,7 @@ import org.apache.arrow.vector.ipc.message.{ArrowRecordBatch, MessageSerializer}
 import org.apache.spark.{SparkException, TaskContext}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSeq}
 import org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
 import org.apache.spark.sql.columnar.{CachedBatch, SimpleMetricsCachedBatchSerializer}
@@ -144,8 +144,12 @@ class ArrowCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
       conf: SQLConf): RDD[ColumnarBatch] = {
     val cacheSchema = DataTypeUtils.fromAttributes(cacheAttributes)
     val selectedSchema = DataTypeUtils.fromAttributes(selectedAttributes)
+    // Use AttributeSeq's cached exprId -> ordinal map so each selected attribute is a
+    // constant-time lookup, instead of rebuilding the exprId list and linear-scanning it
+    // per selected attribute.
+    val cacheAttributeSeq = AttributeSeq(cacheAttributes)
     val columnIndices =
-      selectedAttributes.map(a => cacheAttributes.map(o => o.exprId).indexOf(a.exprId)).toArray
+      selectedAttributes.map(a => cacheAttributeSeq.indexOf(a.exprId)).toArray
     // Capture config on driver
     val timeZoneId = conf.sessionLocalTimeZone
     val prefetchEnabled = conf.arrowCachePrefetchEnabled
@@ -170,9 +174,12 @@ class ArrowCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
     val selectedSchema = DataTypeUtils.fromAttributes(selectedAttributes)
     val timeZoneId = conf.sessionLocalTimeZone
 
-    // Calculate column indices for projection
+    // Calculate column indices for projection. Use AttributeSeq's cached exprId -> ordinal
+    // map so each selected attribute is a constant-time lookup, instead of linear-scanning
+    // the cache attributes per selected attribute.
+    val cacheAttributeSeq = AttributeSeq(cacheAttributes)
     val selectedIndices = selectedAttributes.map { attr =>
-      cacheAttributes.indexWhere(_.exprId == attr.exprId)
+      cacheAttributeSeq.indexOf(attr.exprId)
     }.toArray
 
     // Check if all selected types can use the fast path.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
index e79fad6c249d9..e99b6d9b06ef8 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala
@@ -197,8 +197,12 @@ class DefaultCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
       conf: SQLConf): RDD[ColumnarBatch] = {
     val offHeapColumnVectorEnabled = conf.offHeapColumnVectorEnabled
     val outputSchema = DataTypeUtils.fromAttributes(selectedAttributes)
+    // Use AttributeSeq's cached exprId -> ordinal map so each selected attribute is a
+    // constant-time lookup, instead of rebuilding the exprId list and linear-scanning it
+    // per selected attribute.
+    val cacheAttributeSeq = AttributeSeq(cacheAttributes)
     val columnIndices =
-      selectedAttributes.map(a => cacheAttributes.map(o => o.exprId).indexOf(a.exprId)).toArray
+      selectedAttributes.map(a => cacheAttributeSeq.indexOf(a.exprId)).toArray
 
     def createAndDecompressColumn(cb: CachedBatch): ColumnarBatch = {
       val cachedColumnarBatch = cb.asInstanceOf[DefaultCachedBatch]
@@ -230,10 +234,13 @@ class DefaultCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
       cacheAttributes: Seq[Attribute],
       selectedAttributes: Seq[Attribute],
       conf: SQLConf): RDD[InternalRow] = {
-    // Find the ordinals and data types of the requested columns.
+    // Find the ordinals and data types of the requested columns. Use AttributeSeq's cached
+    // exprId -> ordinal map so each requested column is a constant-time lookup, instead of
+    // rebuilding the exprId list and linear-scanning it per requested column.
+    val cacheAttributeSeq = AttributeSeq(cacheAttributes)
     val (requestedColumnIndices, requestedColumnDataTypes) =
       selectedAttributes.map { a =>
-        cacheAttributes.map(_.exprId).indexOf(a.exprId) -> a.dataType
+        cacheAttributeSeq.indexOf(a.exprId) -> a.dataType
       }.unzip
 
     val columnTypes = requestedColumnDataTypes.map {