Avoid generating duplicate nan keys with MapGen(FloatGen)

[thirtiseven]

Fixes #9685
Fixes #9684

These two failed cases complain that "Duplicate map key NaN was found, please check the input data. "
This is because python dict and spark maptype handle NaN as key in different ways:

• In python dict, NaN as dict key are not equal to each other, so it is possible to have multiple NaN as keys in one dict.
• In Spark map, map treats NaN as a normal number when used as a key, so no duplicate NaN keys are allowed.

This PR avoids generating duplicate NaN keys with MapGen(FloatGen) in integration tests, should be able to fix these two test cases.

I'm surprised by this root cause because the probability of this case feels not very low:
When generating a dataframe of Map[Float, _], DataGen will generate 2048 maps, each of them containing 0~20 entries, NaN will appear in data with 1% chance. Will do the math later.

[thirtiseven]

build

[winningsix]

LGTM. Just nit left.

[winningsix]

Nit: could we mention more about the reason why at most one key is NaN?

e.g.,

In Spark, NaN = NaN returns true. At most one key is NaN to avoid duplicated key value.
This is different from python where NaN as dict key are not equal to each other, so it is possible to have multiple NaN as keys in one dict.

[thirtiseven]

Thanks for review, done.

[thirtiseven]

Seems it is more complicated for these two cases, the chance that datagen generated multiple NaNs is very high. However, it works fine and quiet in most cases when converting dict to dataframe.

>>> dict1 = {float('nan'): 1, float('nan'): 2, 0.3: 3}
>>> dict2 = {float('nan'): 3, float('nan'): 2, 0.3: 3}
>>> data = [dict1, dict2]
>>> df = spark.createDataFrame(SparkContext.getOrCreate().parallelize(data, numSlices=None), MapType(DoubleType(), IntegerType(), valueContainsNull=True))
>>> df.show()
+--------------------+
|               value|
+--------------------+
|{0.3 -> 3, NaN -> 1}|
|{0.3 -> 3, NaN -> 3}|
+--------------------+

And also, if we replace the command in test_case_when with a collect, the test will pass with the same seed.

Although the failure is from input data and the behavior is matched, I'd like to keep this PR on hold until I can make the root cause more clear.

The base branch was changed.

[thirtiseven]

The two failures are because there are multiple NaNs in the test Scalar that will be used directly as literal in query. It will be fine if we convert maps with multiple NaNs as key to dataframes.

Since we do not intend to create maps with multiple NaNs in DataGen, I think we can just keep the current solution to avoid generating such kind of data at all.

[revans2]

What happens on the GPU with this same issue? Do we do the same thing? I realize that python supports multiple NaN keys that will never compare as equal. But does the GPU do the same thing as the CPU does and throws an exception? If not we have to document this and at least file a follow on issue, even if it is a very low priority. If it does we probably want a test to verify that we continue to do this on all of the platforms we support.

To be clear I want to understand what happens when multiple NaN values are inserted as the keys, but also what happens when we try to look up the value stored under a NaN key.

[thirtiseven]

1. Create dataframe with dicts containing multiple NaN keys and look up value:

>>> from pyspark.sql.types import *
>>> dict1 = {float('nan'): 1, float('nan'): 2, 0.3: 3}
>>> dict2 = {float('nan'): 3, float('nan'): 2, 0.3: 3}
>>> data = [dict1, dict2]
>>> df = spark.createDataFrame(SparkContext.getOrCreate().parallelize(data, numSlices=None), MapType(DoubleType(), IntegerType(), valueContainsNull=True))
>>> df.show()
23/11/30 11:13:35 WARN GpuOverrides:
!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it
  @Partitioning <SinglePartition$> could run on GPU
  *Exec <ProjectExec> will run on GPU
    *Expression <Alias> cast(value#0 as string) AS value#3 will run on GPU
      *Expression <Cast> cast(value#0 as string) will run on GPU
    ! <RDDScanExec> cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.execution.RDDScanExec
      @Expression <AttributeReference> value#0 could run on GPU

+--------------------+
|               value|
+--------------------+
|{0.3 -> 3, NaN -> 1}|
|{0.3 -> 3, NaN -> 3}|
+--------------------+

>>> from pyspark.sql.functions import col
>>> df.select(col("value").getItem(float('nan'))).first()[0]
23/11/30 11:14:46 WARN GpuOverrides:
!Exec <CollectLimitExec> cannot run on GPU because the Exec CollectLimitExec has been disabled, and is disabled by default because Collect Limit replacement can be slower on the GPU, if huge number of rows in a batch it could help by limiting the number of rows transferred from GPU to CPU. Set spark.rapids.sql.exec.CollectLimitExec to true if you wish to enable it
  @Partitioning <SinglePartition$> could run on GPU
  *Exec <ProjectExec> will run on GPU
    *Expression <Alias> value#0[NaN] AS value[NaN]#7 will run on GPU
      *Expression <GetMapValue> value#0[NaN] will run on GPU
    ! <RDDScanExec> cannot run on GPU because GPU does not currently support the operator class org.apache.spark.sql.execution.RDDScanExec
      @Expression <AttributeReference> value#0 could run on GPU

1

The behavior is matched to pyspark. There is only one NaN key kept and it's value can be looked up normally.

pyspark will convert python object to java object first then create the dataframe, I think plugin is not touching related logic so we are good.

2. This issue's case, if we use a literal map containing multiple NaN keys in a query. I added a test and it throws same exception as CPU does.

[thirtiseven]

build