PySpark invokes UnresolvedNamedLambdaVariable.freshVarName in JVM to get a unique variable name

object UnresolvedNamedLambdaVariable {

  // Counter to ensure lambda variable names are unique
  private val nextVarNameId = new AtomicInteger(0)

  def freshVarName(name: String): String = {
    s"${name}_${nextVarNameId.getAndIncrement()}"
  }
}

I think from looking at this the last time, the reason for the variables is mostly to create unique aliases but not to reference them actually in the plan. Because the expression itself must be a boolean expression. And the lambda function we pass in is really just a expression transformation.

Can we see if we can do without this one? This is very much an implementation detail.

will remove it.

Can we just make this an unresolved attribute? And do the heavy lifting in the connect planner?

Unless there is special resolution in Catalyst on this type otherwise +1 to re-use unresolved attribute

Can someone give me a good example what the variable is supposed to do?

it are literally the argument names in a lambda, for example x => x + 1, the UnresolvedNamedLambdaVariable is x.

I am on the fence about adding this message. We could go the CaseWhen route here as well. The only argument against that are the arguments them self :).

How would the lambda function look like for using the CaseWhen route?

ok, that would be simpler

How would the lambda function look like for using the CaseWhen route?

this one #38956, we build the expressions in Connect Planner other than let FunctionRegistry do this

actually, we can remove ExpressionString, UnresolvedStar, UnresolvedRegex, Cast in the same way.

But now I'm not sure whether it is the correct way.

I think this is tricky because it's hard to provide clear guidance on what the right approach is. Generally, the downside of the unresolved function approach is that you're using a magic value that someone has to understand. This knowledge is now embedded in the client and cannot be inferred when looking at the protos.

will add LambdaFunction back in the protos

I think this is tricky because it's hard to provide clear guidance on what the right approach is. Generally, the downside of the unresolved function approach is that you're using a magic value that someone has to understand. This knowledge is now embedded in the client and cannot be inferred when looking at the protos.

Agreed, that is what I feel. We should avoid abusing unresolved function

If all arguments must be UnresolvedNamedLambdaVariables. why not define this as
repeated UnresolvedNamedLambdaVariables arguments = 2?

I was hitting something weird if using UnresolvedNamedLambdaVariables here

In the scala side, a LambdaFunction actually accepts NamedExpressions as argument;
In PySpark, an argument is always a UnresolvedNamedLambdaVariable;

Here we use Expression instead of UnresolvedAttributes, a benefit I can image is, if we need to support more types of Expression, we do not need to change the proto

Shall we simply use repeated String ...? If we think about the SQL API exists(array_col, c -> ...), what we need to provide here is the argument name list, which is really just a list of string.

In general I'm in favor of repeated string since it's clearer about the intention. I think this function is a good example where we need to put guidance on where the correct resolution of function arguments happens.

yes, that's simpler. At least repeated String is enough for existing pyspark's implementations. will update

How would the lambda function look like for using the CaseWhen route?

Can someone give me a good example what the variable is supposed to do?

I think from looking at this the last time, the reason for the variables is mostly to create unique aliases but not to reference them actually in the plan. Because the expression itself must be a boolean expression. And the lambda function we pass in is really just a expression transformation.

This branch deserves it's own function please.

+1 will add it back

There are two interesting issues here:

1. When someone submits to the API an expression that does not transform into UnresolvedExpression this would throw a weird error message about the name parts but actually the type does not match.
2. Why the restriction to single part names? Is this a Spark limitation?

1, the error message also mentioned UnresolvedAttribute

2, existing implementation in PySpark only use single part name

Adding these assertions here is helpful in the Python client but the server side does not do the same assertion. What happens if we drop the assertion on ColumnReference what would happen on the server?

Is the analysis exception not better than the Python assertion>

we also check UnresolvedAttribute in the server side

one line?

done

existing pyspark only uses single name part, so check it in the server side

since we support up to 3 parameters, shall we add an assert here to avoid future mistakes?

@cloud-fan Here is the validation on number of paramters

In general I'm in favor of repeated string since it's clearer about the intention. I think this function is a good example where we need to put guidance on where the correct resolution of function arguments happens.

Do we get an analysis exception if more than three arguments are submitted?

There is a validation in the client side already but I guess we need that in the server side as well?

ok, let me add it in server side

IIUC, we use Column with UnresolvedAttribute to invoke the function, then at the server side, we replace UnresolvedAttribute with LambdaVariable.

Shall we use Column with LambdaVariable to invoke the function in client side directly?

the first commit used UnresolvedNamedLambdaVariable

    arg_cols: List[Column] = []
    for arg in arg_names[: len(parameters)]:
        # TODO: How to make sure lambda variable names are unique? RPC for increasing ID?
        _uuid = str(uuid.uuid4()).replace("-", "_")
        arg_cols.append(Column(UnresolvedNamedLambdaVariable([f"{arg}_{_uuid}"])))

    result = f(*arg_cols)

    if not isinstance(result, Column):
        raise ValueError(f"Callable {f} should return Column, got {type(result)}")

    return LambdaFunction(result._expr, [arg._expr for arg in arg_cols])

then switched to UnresolvedAttribute according to the comments #39068 (comment)

I don't think this can simplify the client-side implementation, but it does save one proto message. At least we should document this in the proto message LambdaFunction: the function body should use UnresolvedAttribute as arguments to build the query plan.

done