[BUG]: Error unpickling decimal array when array contains at least one '0' value

[akamor]

Describe the bug
Parquet file has a column of array<decimal<38,18>>. If row has an array such as array(0,1,2,3) we will get an exception under certain situations. The exception only happens when the array itself contains at least 1 element whose value is 0.

To Reproduce

Create a parquet file described as above then read it into a DF.

From the C# side do something like:

var df = spark.read.parquet("path to file")
var rows = df.toLocalIterator();

Once you try to enumerate rows you'll get the following exception:

Unhandled exception. System.FormatException: Input string was not in a correct format.
   at System.Number.ThrowOverflowOrFormatException(ParsingStatus status, TypeCode type)
   at System.Convert.ToDecimal(String value, IFormatProvider provider)
   at Razorvine.Pickle.Objects.DecimalConstructor.construct(Object[] args)
   at Razorvine.Pickle.UnpicklerImplementation`1.load_reduce()
   at Razorvine.Pickle.UnpicklerImplementation`1.Dispatch(Byte key)
   at Razorvine.Pickle.UnpicklerImplementation`1.Load()
   at Razorvine.Pickle.Unpickler.loads(Byte[] pickledata, Int32 stackCapacity)
   at Razorvine.Pickle.Unpickler.loads(ReadOnlyMemory`1 pickledata, Int32 stackCapacity)
   at Microsoft.Spark.Utils.PythonSerDe.GetUnpickledObjects(Stream stream, Int32 messageLength) in /mnt/c/Users/adam/Repos/spark/src/csharp/Microsoft.Spark/Utils/PythonSerDe.cs:line 65
   at Microsoft.Spark.Sql.RowCollector.Collect(ISocketWrapper socket)+MoveNext() in /mnt/c/Users/adam/Repos/spark/src/csharp/Microsoft.Spark/Sql/RowCollector.cs:line 33
   at Microsoft.Spark.Sql.RowCollector.LocalIteratorFromSocket.GetEnumerator()+MoveNext() in /mnt/c/Users/adam/Repos/spark/src/csharp/Microsoft.Spark/Sql/RowCollector.cs:line 117
   at Microsoft.Spark.Sql.DataFrame.ToLocalIterator(Boolean prefetchPartitions)+MoveNext() in /mnt/c/Users/adam/Repos/spark/src/csharp/Microsoft.Spark/Sql/DataFrame.cs:line 843

This currently is happening with v1.0.0.

[akamor]

To be clear, arrays that do not contain the value 0 seem to work just fine. Additionally, decimal columns (non-arrays) with 0 values also work just fine.

[akamor]

Issue also occurs using v2.1.0.

[imback82]

Could be a bug. Similar issue in #1006

[imback82]

@suhsteve Are you aware of this issue by any chance?

[akamor]

So we did more digging on our side. Decimal values defined as decimal(38,18) show values of 0 as 0E-18. More details here: apache/spark#22171 and this was also confirmed in our own testing.

The issue appears to be that System.Convert.ToDecimal("0E-18") throws the exception mentioned above (and in the linked issue).

[akamor]

Did more digging and I've filed a bug on Razorvine to consider using Decimal.Parse over Convert.ToDecimal as this would support numbers in scientific notation.

Is there any workaround that can be done in this repro in the meantime? If not, we'll likely need to fork Razorvine and use the forked version with the this spark driver.

irmen/pickle#8

[suhsteve]

@suhsteve Are you aware of this issue by any chance?

No sorry I'm unaware of this issue. But it seems like @akamor has a lead. If the fix makes it into the Razorvine package, although unsure, I suspect that it won't be a a simple direct replacement if we had to update to Razorvine 5.x.

[akamor]

At the moment I'm looking to use lib.Harmony to apply a Monkey patch in Microsoft.Spark.Worker. That seems to not be working as I'm now getting this very odd error when running a job.

Caused by: org.apache.spark.SparkException: 
Error from python worker:
  Unhandled exception. 
     Cannot print exception string because Exception.ToString() failed.
PYTHONPATH was:
  /mnt1/yarn/usercache/hadoop/filecache/1380/__spark_libs__5891156675855964406.zip/spark-core_2.12-3.0.0-amzn-0.jar
org.apache.spark.SparkException: EOFException occurred while reading the port number from pyspark.daemon's stdout
	at org.apache.spark.api.python.PythonWorkerFactory.startDaemon(PythonWorkerFactory.scala:228)
	at org.apache.spark.api.python.PythonWorkerFactory.createThroughDaemon(PythonWorkerFactory.scala:143)
	at org.apache.spark.api.python.PythonWorkerFactory.create(PythonWorkerFactory.scala:116)
	at org.apache.spark.SparkEnv.createPythonWorker(SparkEnv.scala:120)
	at org.apache.spark.api.python.BasePythonRunner.compute(PythonRunner.scala:131)
	at org.apache.spark.sql.execution.python.BatchEvalPythonExec.evaluate(BatchEvalPythonExec.scala:70)
	at org.apache.spark.sql.execution.python.EvalPythonExec.$anonfun$doExecute$2(EvalPythonExec.scala:128)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:837)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:837)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:349)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:313)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:127)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:444)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:447)
	... 3 more

Looks like my patch is maybe causing Microsoft.Spark.Worker to completely fail?

[suhsteve]

Looked into this a bit more. Haven't tested this but I think we can just create our own custom DecimalConstructor and register the constructor with the Unpickler.

[akamor]

@suhsteve What a great idea. Wish I had read the docs a bit more. :)

Registering a custom type constructor solved the problem and was much simpler than my other approach (forking and building a custom razorvine dll).

For those in the future that hit this issue here my custom constructor:

public class OverriddenDecimalConstructor : IObjectConstructor
{
    public object construct(object[] args)
    {
        if (args.Length == 1 && args[0] is string s)
        {
            if (s.ToLowerInvariant() == "nan") return double.NaN;
            //PATCH: //return Convert.ToDecimal((string)args[0], CultureInfo.InvariantCulture);
            return decimal.Parse(s, NumberStyles.Any);
        }

        throw new PickleException("invalid arguments for decimal constructor");
    }
}

You can register the constructor using

Unpickler.registerConstructor("decimal","Decimal",new OverriddenDecimalConstructor());

I put the registration in 2 places.

1\ Near the top of Program.cs in Microsoft.Spark.Worker
2\ At the top of the SparkSession constructor in Microsoft.Spark.Sql.

[suhsteve]

@akamor Glad it's working out for you. The steps you outlined is pretty much what I had in mind. Would you like to contribute the fix as a PR? I think the only missing piece are unit tests. They should be added to:

• Microsoft.Spark.UnitTest - to test the DecimalConstructor
• Microsoft.Spark.E2ETest - for end to end tests.

[akamor]

Sure, happy to write a PR. It might not happen for at least a few more days but folks in the meantime can definitely use the above.

Out of curiosity, why was the registration needed in both Program.cs and in the SparkSession constructor? I would've assumed it would only be needed in Program.cs

[suhsteve]

Out of curiosity, why was the registration needed in both Program.cs and in the SparkSession constructor? I would've assumed it would only be needed in Program.cs

I think it'll depend on where the unpickling is happening. From your example of

var df = spark.read.parquet("path to file")
var rows = df.toLocalIterator();

I think only the SparkSession part is necessary in your example, as there shouldn't be any C# Worker's spun up due to no UDF calls.

Registering the DecimalConstructor here in PythonSerDe.cs should be the only place where you'll need to.