diff --git a/sql/core/src/main/scala/org/apache/spark/sql/KeyValueGroupedDataset.scala b/sql/core/src/main/scala/org/apache/spark/sql/KeyValueGroupedDataset.scala
index 5cce4a09c705c..4094b06c4c9e4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/KeyValueGroupedDataset.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/KeyValueGroupedDataset.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.catalyst.expressions.{Ascending, Attribute, Expressi
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.execution.QueryExecution
 import org.apache.spark.sql.expressions.ReduceAggregator
+import org.apache.spark.sql.internal.ExpressionColumnNode
 import org.apache.spark.sql.internal.TypedAggUtils.{aggKeyColumn, withInputType}
 import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout, OutputMode, StatefulProcessor, StatefulProcessorWithInitialState, TimeMode}
 
@@ -1213,7 +1214,7 @@ class KeyValueGroupedDataset[K, V] private[sql](
   private def scopeTypedColumn(
       scope: Seq[Attribute])(
       typedCol: TypedColumn[_, _]): TypedColumn[_, _] =
-    new TypedColumn(ScopedExpression(typedCol.expr, scope), typedCol.encoder)
+    new TypedColumn(ExpressionColumnNode(ScopedExpression(typedCol.expr, scope)), typedCol.encoder)
 
 
   override def toString: String = {