@@ -329,7 +329,9 @@ object URLEncodingUtils {
 }
 
 private[ml] object ServiceAuthHeaders {
-  private[ml] case class Resolution(headers: Map[String, String], usesFabricFallback: Boolean)
+  private[ml] case class Resolution(headers: Map[String, String],
+                                    usesFabricFallback: Boolean,
+                                    usesSubscriptionKey: Boolean)
 
   private[ml] def nonBlank(value: String): Boolean = value != null && value.trim.nonEmpty
 
@@ -377,20 +379,15 @@ private[ml] object ServiceAuthHeaders {
     def isAuthHeaderName(name: String): Boolean =
       name.equalsIgnoreCase(subscriptionKeyHeaderName) || name.equalsIgnoreCase(aadHeaderName)
 
-    // Resolve exactly one auth header across every credential source in a single, case-insensitive
-    // precedence step. Blank values are skipped so they cannot suppress a valid lower-priority
-    // credential, and the automatic Fabric fallback ranks below a credential embedded in
-    // customHeaders (matching how management-index requests, which have no fallback, resolve auth).
-    // fabricFallbackAuthHeader is by-name and lowest priority, so it is evaluated only when every
-    // higher-priority source above is absent. A fallback that acquires a token (and may throw) is
-    // therefore never run while a subscription key, AAD token, explicit custom-auth header, or an
-    // embedded customHeaders credential is present.
-    val explicitAuthHeader: Option[(String, String)] = subscriptionKey.filter(nonBlank)
-      .map(value => subscriptionKeyHeaderName -> value)
-      .orElse(aadToken.filter(nonBlank).map(value => aadHeaderName -> ("Bearer " + value)))
-      .orElse(customAuthHeader.filter(nonBlank).map(value => aadHeaderName -> value))
-      .orElse(embeddedCredential(providedCustomHeaders, subscriptionKeyHeaderName))
-      .orElse(embeddedCredential(providedCustomHeaders, aadHeaderName))
+    // Select one non-blank credential and retain its source even when the auth header names coincide.
+    // The by-name Fabric fallback is evaluated only when no explicit or embedded credential exists.
+    val explicitAuth: Option[((String, String), Boolean)] = subscriptionKey.filter(nonBlank)
+      .map(value => (subscriptionKeyHeaderName -> value) -> true)
+      .orElse(aadToken.filter(nonBlank).map(value => (aadHeaderName -> ("Bearer " + value)) -> false))
+      .orElse(customAuthHeader.filter(nonBlank).map(value => (aadHeaderName -> value) -> false))
+      .orElse(embeddedCredential(providedCustomHeaders, subscriptionKeyHeaderName).map(_ -> true))
+      .orElse(embeddedCredential(providedCustomHeaders, aadHeaderName).map(_ -> false))
+    val explicitAuthHeader = explicitAuth.map(_._1)
     val fabricAuthHeader = if (explicitAuthHeader.isDefined) {
       None
     } else {
@@ -421,7 +418,8 @@ private[ml] object ServiceAuthHeaders {
 
     Resolution(
       new scala.collection.immutable.TreeMap[String, String]() ++ headers,
-      usesFabricFallback = fabricAuthHeader.isDefined)
+      usesFabricFallback = fabricAuthHeader.isDefined,
+      usesSubscriptionKey = explicitAuth.exists(_._2))
   }
 
   def build(subscriptionKey: Option[String],
 
@@ -0,0 +1,83 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.services.ai
+
+import org.apache.spark.sql.{Column, Row, functions => F}
+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, StructType}
+
+import java.util.{HashSet, Set => JavaSet, TreeSet}
+
+private[ai] final case class AIColumnPlan(input: StructType, outputs: Seq[StructField]) {
+  AIColumnPlan.distinct(input.fieldNames.toSeq, "input columns")
+  AIColumnPlan.distinct(outputs.map(_.name), "output roles")
+  require(outputs.forall(_.name.nonEmpty), "output roles must have non-empty names")
+  private val outputNames = outputs.map(_.name).toSet
+  private val replacements = input.fields.map(f =>
+    outputs.find(o => AIColumnPlan.matches(f.name, o.name)))
+  val schema: StructType = StructType(input.fields.zip(replacements).map {
+    case (original, replacement) => replacement.getOrElse(original)
+  } ++ outputs.filterNot(o => input.fields.exists(f => AIColumnPlan.matches(f.name, o.name))))
+
+  private val indices = schema.fields.map(f =>
+    if (outputNames(f.name)) -1 else input.fieldIndex(f.name))
+
+  def assemble(row: Row, values: Map[String, Any]): Row =
+    Row.fromSeq(schema.fields.zip(indices).map { case (field, index) =>
+      if (index < 0) values.getOrElse(field.name, None.orNull) else row.get(index)
+    })
+
+  def scratch(prefix: String, additional: String*): String =
+    AIColumnPlan.unused(prefix, schema.fieldNames.toSeq ++ input.fieldNames ++ additional)
+}
+
+private[ai] object AIColumnPlan {
+  def matches(left: String, right: String): Boolean = SQLConf.get.resolver(left, right)
+
+  def distinct(names: Seq[String], description: String): Unit = {
+    require(names.forall(_ != null), s"$description must not have null names")
+    // This comparator matches Spark's equalsIgnoreCase, including Unicode variants lost by lowercasing.
+    val seen: JavaSet[String] = if (SQLConf.get.caseSensitiveAnalysis) new HashSet[String]()
+    else new TreeSet[String](String.CASE_INSENSITIVE_ORDER)
+    require(!names.exists(name => !seen.add(name)),
+      s"Ambiguous $description under Spark's column resolver")
+  }
+
+  def resolve(schema: StructType, name: String): String =
+    schema.fieldNames.find(matches(_, name)).getOrElse(
+      throw new IllegalArgumentException(s"Input column '$name' does not exist"))
+
+  def unused(prefix: String, names: Seq[String]): String =
+    (Iterator(prefix) ++ Iterator.from(1).map(i => s"${prefix}_$i"))
+      .find(candidate => !names.exists(matches(_, candidate))).get
+
+  def column(name: String): Column = F.col("`" + name.replace("`", "``") + "`")
+
+  // Spark can eliminate serialization between Dataset maps and retain external JVM arrays.
+  def arrayValues(value: Any): scala.collection.Seq[Any] = value match {
+    case values: scala.collection.Seq[_] => values
+    case values: Array[_] => values.toSeq
+    case _ => throw new IllegalArgumentException("Expected a Spark array value")
+  }
+
+  // Fused Dataset maps can pass schema-less Rows directly to the protocol serializers.
+  def withSchema(row: Row, schema: StructType): Row =
+    new GenericRowWithSchema(schema.fields.zipWithIndex.map { case (field, index) =>
+      nestedSchema(row.get(index), field.dataType)
+    }, schema)
+
+  private def nestedSchema(value: Any, dataType: DataType): Any = Option(value).map { present =>
+    dataType match {
+      case struct: StructType => withSchema(present.asInstanceOf[Row], struct)
+      case ArrayType(element, _) =>
+        arrayValues(present).iterator.map(nestedSchema(_, element)).toVector
+      case MapType(key, element, _) =>
+        present.asInstanceOf[scala.collection.Map[Any, Any]].iterator.map { case (k, v) =>
+          nestedSchema(k, key) -> nestedSchema(v, element)
+        }.toMap
+      case _ => present
+    }
+  }.orNull
+}