[SPARK-49087][SQL][CONNECT] Distinguish UnresolvedFunction calling internal functions

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

This PR introduces a new Protobuf field `is_internal` to `UnresolvedFunction` message. This field is used to carry the info on whether an `UnresolvedFunction` is calling an internal function (using the `Column#internalFn` API), so that the Connect server could act appropriately:

- If `true`, then (inside Catalyst) mark the function as internal.
- If `false`, then mark the function as external (public or user-defined).
- (Current behaviour) If not set, then look it up in the internal registry: mark as "internal" if found, "external" otherwise.

TODO: [SPARK-50658](https://issues.apache.org/jira/browse/SPARK-50658): apply the same change to the Spark Connect Python client.

### Why are the changes needed?

Tidy up the code and make our first-party Connect client adhere to the best practices.

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

Nope.

### How was this patch tested?

New test is added.

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

Copilot helped generate protobuf comments and repeated test codes.

Closes apache#49274 from xupefei/internal-fn.

Lead-authored-by: Paddy Xu <xupaddy@gmail.com>
Co-authored-by: Hyukjin Kwon <gurwls223@gmail.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

Author: xupefei

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/internal/columnNodeSupport.scala

@@ -73,13 +73,19 @@ object ColumnNodeToProtoConverter extends (ColumnNode => proto.Expression) {
           .setColName(regex)
         planId.foreach(b.setPlanId)
 
-      case UnresolvedFunction(functionName, arguments, isDistinct, isUserDefinedFunction, _, _) =>
-        // TODO(SPARK-49087) use internal namespace.
+      case UnresolvedFunction(
+            functionName,
+            arguments,
+            isDistinct,
+            isUserDefinedFunction,
+            isInternal,
+            _) =>
         builder.getUnresolvedFunctionBuilder
           .setFunctionName(functionName)
           .setIsUserDefinedFunction(isUserDefinedFunction)
           .setIsDistinct(isDistinct)
           .addAllArguments(arguments.map(apply(_, e)).asJava)
+          .setIsInternal(isInternal)
 
       case Alias(child, name, metadata, _) =>
         val b = builder.getAliasBuilder.setExpr(apply(child, e))
@@ -156,6 +162,7 @@ object ColumnNodeToProtoConverter extends (ColumnNode => proto.Expression) {
       case CaseWhenOtherwise(branches, otherwise, _) =>
         val b = builder.getUnresolvedFunctionBuilder
           .setFunctionName("when")
+          .setIsInternal(false)
         branches.foreach { case (condition, value) =>
           b.addArguments(apply(condition, e))
           b.addArguments(apply(value, e))

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/internal/ColumnNodeToProtoConverterSuite.scala

@@ -128,19 +128,22 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
           .setFunctionName("+")
           .setIsDistinct(false)
           .addArguments(attribute("a"))
-          .addArguments(expr(_.getLiteralBuilder.setInteger(1)))))
+          .addArguments(expr(_.getLiteralBuilder.setInteger(1)))
+          .setIsInternal(false)))
     testConversion(
       UnresolvedFunction(
         "db1.myAgg",
         Seq(UnresolvedAttribute("a")),
         isDistinct = true,
-        isUserDefinedFunction = true),
+        isUserDefinedFunction = true,
+        isInternal = true),
       expr(
         _.getUnresolvedFunctionBuilder
           .setFunctionName("db1.myAgg")
           .setIsDistinct(true)
           .setIsUserDefinedFunction(true)
-          .addArguments(attribute("a"))))
+          .addArguments(attribute("a"))
+          .setIsInternal(true)))
   }
 
   test("alias") {
@@ -247,10 +250,12 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
       expr(
         _.getWindowBuilder
           .setWindowFunction(
-            expr(_.getUnresolvedFunctionBuilder
-              .setFunctionName("sum")
-              .setIsDistinct(false)
-              .addArguments(attribute("a"))))
+            expr(
+              _.getUnresolvedFunctionBuilder
+                .setFunctionName("sum")
+                .setIsDistinct(false)
+                .addArguments(attribute("a"))
+                .setIsInternal(false)))
           .addPartitionSpec(attribute("b"))
           .addPartitionSpec(attribute("c"))
           .addOrderSpec(proto.Expression.SortOrder
@@ -276,7 +281,8 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
               _.getUnresolvedFunctionBuilder
                 .setFunctionName("sum")
                 .setIsDistinct(false)
-                .addArguments(attribute("a"))))
+                .addArguments(attribute("a"))
+                .setIsInternal(false)))
           .addPartitionSpec(attribute("b"))
           .addPartitionSpec(attribute("c"))))
     testWindowFrame(
@@ -310,7 +316,8 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
               _.getUnresolvedFunctionBuilder
                 .setFunctionName("+")
                 .addArguments(expr(_.setUnresolvedNamedLambdaVariable(catX)))
-                .addArguments(attribute("y"))))
+                .addArguments(attribute("y"))
+                .setIsInternal(false)))
           .addArguments(catX)))
   }
 
@@ -330,7 +337,8 @@ class ColumnNodeToProtoConverterSuite extends ConnectFunSuite {
           .setFunctionName("when")
           .addArguments(attribute("c1"))
           .addArguments(expr(_.getLiteralBuilder.setString("r1")))
-          .addArguments(expr(_.getLiteralBuilder.setString("fallback")))))
+          .addArguments(expr(_.getLiteralBuilder.setString("fallback")))
+          .setIsInternal(false)))
   }
 
   test("extract field") {

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

@@ -847,6 +847,7 @@ class Expression(google.protobuf.message.Message):
         ARGUMENTS_FIELD_NUMBER: builtins.int
         IS_DISTINCT_FIELD_NUMBER: builtins.int
         IS_USER_DEFINED_FUNCTION_FIELD_NUMBER: builtins.int
+        IS_INTERNAL_FIELD_NUMBER: builtins.int
         function_name: builtins.str
         """(Required) name (or unparsed name for user defined function) for the unresolved function."""
         @property
@@ -864,27 +865,46 @@ class Expression(google.protobuf.message.Message):
         When it is not a user defined function, Connect will use the function name directly.
         When it is a user defined function, Connect will parse the function name first.
         """
+        is_internal: builtins.bool
+        """(Optional) Indicate if this function is defined in the internal function registry.
+        If not set, the server will try to look up the function in the internal function registry
+        and decide appropriately.
+        """
         def __init__(
             self,
             *,
             function_name: builtins.str = ...,
             arguments: collections.abc.Iterable[global___Expression] | None = ...,
             is_distinct: builtins.bool = ...,
             is_user_defined_function: builtins.bool = ...,
+            is_internal: builtins.bool | None = ...,
         ) -> None: ...
+        def HasField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_is_internal", b"_is_internal", "is_internal", b"is_internal"
+            ],
+        ) -> builtins.bool: ...
         def ClearField(
             self,
             field_name: typing_extensions.Literal[
+                "_is_internal",
+                b"_is_internal",
                 "arguments",
                 b"arguments",
                 "function_name",
                 b"function_name",
                 "is_distinct",
                 b"is_distinct",
+                "is_internal",
+                b"is_internal",
                 "is_user_defined_function",
                 b"is_user_defined_function",
             ],
         ) -> None: ...
+        def WhichOneof(
+            self, oneof_group: typing_extensions.Literal["_is_internal", b"_is_internal"]
+        ) -> typing_extensions.Literal["is_internal"] | None: ...
 
     class ExpressionString(google.protobuf.message.Message):
         """Expression as string."""

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

@@ -261,6 +261,11 @@ message Expression {
     // When it is not a user defined function, Connect will use the function name directly.
     // When it is a user defined function, Connect will parse the function name first.
     bool is_user_defined_function = 4;
+
+    // (Optional) Indicate if this function is defined in the internal function registry.
+    // If not set, the server will try to look up the function in the internal function registry
+    // and decide appropriately.
+    optional bool is_internal = 5;
   }
 
   // Expression as string.

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

@@ -22,7 +22,8 @@
           "unresolvedAttribute": {
             "unparsedIdentifier": "b"
           }
-        }]
+        }],
+        "isInternal": false
       }
     }]
   }

sql/connect/common/src/test/resources/query-tests/queries/column_add.json

@@ -25,7 +25,8 @@
               "literal": {
                 "integer": 10
               }
-            }]
+            }],
+            "isInternal": false
           }
         }, {
           "unresolvedFunction": {
@@ -38,9 +39,11 @@
               "literal": {
                 "double": 0.5
               }
-            }]
+            }],
+            "isInternal": false
           }
-        }]
+        }],
+        "isInternal": false
       }
     }]
   }

sql/connect/common/src/test/resources/query-tests/queries/column_add.proto.bin

@@ -25,7 +25,8 @@
               "literal": {
                 "integer": 10
               }
-            }]
+            }],
+            "isInternal": false
           }
         }, {
           "unresolvedFunction": {
@@ -38,9 +39,11 @@
               "literal": {
                 "integer": 20
               }
-            }]
+            }],
+            "isInternal": false
           }
-        }]
+        }],
+        "isInternal": false
       }
     }]
   }