[SPARK-48665][PYTHON][CONNECT] Support providing a dict in pyspark lit to create a map.

[skanderboudawara]

What changes were proposed in this pull request?

Reopening the PR done by Ronserruya and addidng small changes.

Added the option to provide a dict to pyspark.sql.functions.lit in order to create a map

Why are the changes needed?

To make it easier to create a map in pyspark.
Currently, it is only possible via create_map which requires a sequence of key,value,key,value...
Scala already supports such functionality using typedLit

A similar PR was done in the past to add similar functionality for the creating of an array using a list, so I tried to follow all the changes done there as well.

Does this PR introduce any user-facing change?

Yes, docstring of lit was edited, and new functionality was added

Before:

from pyspark.sql import functions as F
F.lit({"a":1})
# pyspark.errors.exceptions.captured.SparkRuntimeException: [UNSUPPORTED_FEATURE.LITERAL_TYPE] The feature is not supported: Literal for '{asd=2}' of class java.util.HashMap.

After:

from pyspark.sql import functions as F
F.lit({"a":1, "b": 2})
# Column<'map(a, 1, b, 2)'>

How was this patch tested?

Manual tests + unittest in CI

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

No

[zhengruifeng]

how ansi affect this feature?

[skanderboudawara]

Without ansi the result is {“a”: 1, “b”: 2, “c”: None}
with the ansi {“a”:”1”, “b”:”2”, “c”: None} all the int are strings

[zhengruifeng]

do we require an active session here? @HyukjinKwon

[HyukjinKwon]

Let's use the default value of spark.sql.pyspark.inferNestedDictAsStruct.enabled when SparkSession.getActiveSession returns None. For Connect too.

[zhengruifeng]

this triggers a Config RPC, for nested cases, it will re-trigger multiple times.
We should cache the config, and make sure only at most one invocation.

[skanderboudawara]

shall I create a new cached function in utils

@lru_cache()
def __get_conf_nested():
        spark = SparkSession.getActiveSession()
        dict_as_struct = (
            spark.conf.get("spark.sql.pyspark.inferNestedDictAsStruct.enabled")
            if spark
            else “true"
        )
       return dict_as_struct

?