[SPARK-51326][CONNECT] Remove LazyExpression proto message

### What changes were proposed in this pull request?

Removes `LazyExpression` proto message.

As any feature using this proto message is not released yet, it's safe to remove it from master and branch-4.0.

### Why are the changes needed?

Discussed offline and found out that Spark Connect should work without it.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

The existing tests should pass.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#50093 from ueshin/issues/SPARK-51326/lazy_expression.

Authored-by: Takuya Ueshin <[email protected]>
Signed-off-by: Takuya Ueshin <[email protected]>

Author: ueshin

python/pyspark/sql/connect/column.py

@@ -44,7 +44,6 @@
     Expression,
     UnresolvedFunction,
     UnresolvedExtractValue,
-    LazyExpression,
     LiteralExpression,
     CaseWhen,
     SortOrder,
@@ -459,7 +458,7 @@ def over(self, window: "WindowSpec") -> ParentColumn:  # type: ignore[override]
         return Column(WindowExpression(windowFunction=self._expr, windowSpec=window))
 
     def outer(self) -> ParentColumn:
-        return Column(LazyExpression(self._expr))
+        return Column(self._expr)
 
     def isin(self, *cols: Any) -> ParentColumn:
         if len(cols) == 1 and isinstance(cols[0], (list, set)):

python/pyspark/sql/connect/expressions.py

@@ -1233,25 +1233,6 @@ def __repr__(self) -> str:
         return f"{self._key} => {self._value}"
 
 
-class LazyExpression(Expression):
-    def __init__(self, expr: Expression):
-        assert isinstance(expr, Expression)
-        super().__init__()
-        self._expr = expr
-
-    def to_plan(self, session: "SparkConnectClient") -> proto.Expression:
-        expr = self._create_proto_expression()
-        expr.lazy_expression.child.CopyFrom(self._expr.to_plan(session))
-        return expr
-
-    @property
-    def children(self) -> Sequence["Expression"]:
-        return [self._expr]
-
-    def __repr__(self) -> str:
-        return f"lazy({self._expr})"
-
-
 class SubqueryExpression(Expression):
     def __init__(
         self,

python/pyspark/sql/connect/proto/expressions_pb2.py

@@ -1289,7 +1289,6 @@ class Expression(google.protobuf.message.Message):
     NAMED_ARGUMENT_EXPRESSION_FIELD_NUMBER: builtins.int
     MERGE_ACTION_FIELD_NUMBER: builtins.int
     TYPED_AGGREGATE_EXPRESSION_FIELD_NUMBER: builtins.int
-    LAZY_EXPRESSION_FIELD_NUMBER: builtins.int
     SUBQUERY_EXPRESSION_FIELD_NUMBER: builtins.int
     EXTENSION_FIELD_NUMBER: builtins.int
     @property
@@ -1335,8 +1334,6 @@ class Expression(google.protobuf.message.Message):
     @property
     def typed_aggregate_expression(self) -> global___TypedAggregateExpression: ...
     @property
-    def lazy_expression(self) -> global___LazyExpression: ...
-    @property
     def subquery_expression(self) -> global___SubqueryExpression: ...
     @property
     def extension(self) -> google.protobuf.any_pb2.Any:
@@ -1367,7 +1364,6 @@ class Expression(google.protobuf.message.Message):
         named_argument_expression: global___NamedArgumentExpression | None = ...,
         merge_action: global___MergeAction | None = ...,
         typed_aggregate_expression: global___TypedAggregateExpression | None = ...,
-        lazy_expression: global___LazyExpression | None = ...,
         subquery_expression: global___SubqueryExpression | None = ...,
         extension: google.protobuf.any_pb2.Any | None = ...,
     ) -> None: ...
@@ -1392,8 +1388,6 @@ class Expression(google.protobuf.message.Message):
             b"extension",
             "lambda_function",
             b"lambda_function",
-            "lazy_expression",
-            b"lazy_expression",
             "literal",
             b"literal",
             "merge_action",
@@ -1445,8 +1439,6 @@ class Expression(google.protobuf.message.Message):
             b"extension",
             "lambda_function",
             b"lambda_function",
-            "lazy_expression",
-            b"lazy_expression",
             "literal",
             b"literal",
             "merge_action",
@@ -1500,7 +1492,6 @@ class Expression(google.protobuf.message.Message):
             "named_argument_expression",
             "merge_action",
             "typed_aggregate_expression",
-            "lazy_expression",
             "subquery_expression",
             "extension",
         ]
@@ -1931,25 +1922,6 @@ class MergeAction(google.protobuf.message.Message):
 
 global___MergeAction = MergeAction
 
-class LazyExpression(google.protobuf.message.Message):
-    DESCRIPTOR: google.protobuf.descriptor.Descriptor
-
-    CHILD_FIELD_NUMBER: builtins.int
-    @property
-    def child(self) -> global___Expression:
-        """(Required) The expression to be marked as lazy."""
-    def __init__(
-        self,
-        *,
-        child: global___Expression | None = ...,
-    ) -> None: ...
-    def HasField(
-        self, field_name: typing_extensions.Literal["child", b"child"]
-    ) -> builtins.bool: ...
-    def ClearField(self, field_name: typing_extensions.Literal["child", b"child"]) -> None: ...
-
-global___LazyExpression = LazyExpression
-
 class SubqueryExpression(google.protobuf.message.Message):
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
 

python/pyspark/sql/connect/proto/expressions_pb2.pyi

@@ -52,8 +52,7 @@ message Expression {
     NamedArgumentExpression named_argument_expression = 17;
     MergeAction merge_action = 19;
     TypedAggregateExpression typed_aggregate_expression = 20;
-    LazyExpression lazy_expression = 21;
-    SubqueryExpression subquery_expression = 22;
+    SubqueryExpression subquery_expression = 21;
 
     // This field is used to mark extensions to the protocol. When plugins generate arbitrary
     // relations they can add them here. During the planning the correct resolution is done.
@@ -474,11 +473,6 @@ message MergeAction {
   }
 }
 
-message LazyExpression {
-  // (Required) The expression to be marked as lazy.
-  Expression child = 1;
-}
-
 message SubqueryExpression {
   // (Required) The ID of the corresponding connect plan.
   int64 plan_id = 1;

sql/connect/common/src/main/protobuf/spark/connect/expressions.proto

@@ -216,7 +216,7 @@ object ColumnNodeToProtoConverter extends (ColumnNode => proto.Expression) {
         }
 
       case LazyExpression(child, _) =>
-        builder.getLazyExpressionBuilder.setChild(apply(child, e))
+        return apply(child, e)
 
       case SubqueryExpressionNode(relation, subqueryType, _) =>
         val b = builder.getSubqueryExpressionBuilder

sql/connect/common/src/main/scala/org/apache/spark/sql/connect/columnNodeSupport.scala

@@ -46,7 +46,7 @@ import org.apache.spark.internal.LogKeys.{DATAFRAME_ID, SESSION_ID}
 import org.apache.spark.resource.{ExecutorResourceRequest, ResourceProfile, TaskResourceProfile, TaskResourceRequest}
 import org.apache.spark.sql.{Column, Encoders, ForeachWriter, Observation, Row}
 import org.apache.spark.sql.catalyst.{expressions, AliasIdentifier, FunctionIdentifier, QueryPlanningTracker}
-import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, GlobalTempView, LazyExpression, LocalTempView, MultiAlias, NameParameterizedQuery, PosParameterizedQuery, UnresolvedAlias, UnresolvedAttribute, UnresolvedDataFrameStar, UnresolvedDeserializer, UnresolvedExtractValue, UnresolvedFunction, UnresolvedPlanId, UnresolvedRegex, UnresolvedRelation, UnresolvedStar, UnresolvedStarWithColumns, UnresolvedStarWithColumnsRenames, UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction, UnresolvedTranspose}
+import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, GlobalTempView, LocalTempView, MultiAlias, NameParameterizedQuery, PosParameterizedQuery, UnresolvedAlias, UnresolvedAttribute, UnresolvedDataFrameStar, UnresolvedDeserializer, UnresolvedExtractValue, UnresolvedFunction, UnresolvedPlanId, UnresolvedRegex, UnresolvedRelation, UnresolvedStar, UnresolvedStarWithColumns, UnresolvedStarWithColumnsRenames, UnresolvedSubqueryColumnAliases, UnresolvedTableValuedFunction, UnresolvedTranspose}
 import org.apache.spark.sql.catalyst.encoders.{encoderFor, AgnosticEncoder, ExpressionEncoder, RowEncoder}
 import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.{ProductEncoder, RowEncoder => AgnosticRowEncoder, StringEncoder, UnboundRowEncoder}
 import org.apache.spark.sql.catalyst.expressions._
@@ -1694,8 +1694,6 @@ class SparkConnectPlanner(
         transformMergeAction(exp.getMergeAction)
       case proto.Expression.ExprTypeCase.TYPED_AGGREGATE_EXPRESSION =>
         transformTypedAggregateExpression(exp.getTypedAggregateExpression, baseRelationOpt)
-      case proto.Expression.ExprTypeCase.LAZY_EXPRESSION =>
-        transformLazyExpression(exp.getLazyExpression)
       case proto.Expression.ExprTypeCase.SUBQUERY_EXPRESSION =>
         transformSubqueryExpression(exp.getSubqueryExpression)
       case _ =>
@@ -3939,10 +3937,6 @@ class SparkConnectPlanner(
     }
   }
 
-  private def transformLazyExpression(getLazyExpression: proto.LazyExpression): Expression = {
-    LazyExpression(transformExpression(getLazyExpression.getChild))
-  }
-
   private def transformSubqueryExpression(
       getSubqueryExpression: proto.SubqueryExpression): Expression = {
     val planId = getSubqueryExpression.getPlanId