[FEA] Support spark.sql.mapKeyDedupPolicy=LAST_WIN for TransformKeys

[viadea]

I wish we can support spark.sql.mapKeyDedupPolicy=LAST_WIN to do deduplicate of the map keys.

For example:

from pyspark.sql.types import *
from pyspark.sql.functions import *
jsonString="""{"Zipcode":704,"ZipCodeType":"STANDARD","City":"PARC PARQUE","State":"PR","Zipcode":999}"""
df=spark.createDataFrame([(1, jsonString)],["id","value"])
df.show(truncate=False)
df2=df.withColumn("value",from_json(df.value,MapType(StringType(),StringType())))
df2.collect()

spark.conf.set("spark.sql.mapKeyDedupPolicy","LAST_WIN")
df2.withColumn('newcol', expr("""transform_keys(value, (k, v) -> split(k, "code")[0])""")).collect()

Not-supported-messages:

!Expression <TransformKeys> transform_keys(value#160, lambdafunction(split(lambda k#254, code, 2)[0], lambda k#254, lambda v#255, false)) cannot run on GPU because LAST_WIN is not supported for config setting spark.sql.mapKeyDedupPolicy

[andygrove]

I updated the title to make it clear that this is asking for support for LAST_WIN for transform_keys. We actually already support LAST_WIN for create_map so we should have everything we need from cuDF.

For create_map we call dropListDuplicatesWithKeysValues to remove duplicates and this already supports the same semantics as LAST_WIN:

    // Apache Spark desires to keep the last duplicate element.
    auto [out_keys, out_vals] =
        cudf::lists::drop_list_duplicates(keys, vals, cudf::duplicate_keep_option::KEEP_LAST);