Fix Row unpickling for long values that fit in int

src/csharp/Microsoft.Spark.UnitTest/Sql/RowTests.cs

@@ -1,8 +1,9 @@
 // Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
+// The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Collections;
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
@@ -62,6 +63,7 @@ public void RowTest()
             Assert.Equal(1, row.Get(0));
             Assert.Equal("abc", row.Get(1));
             Assert.Equal(1, row.GetAs<int>(0));
+            Assert.Equal("abc", row.GetAs<string>(1));
             Assert.ThrowsAny<Exception>(() => row.GetAs<string>(0));
             Assert.Equal("abc", row.GetAs<string>(1));
             Assert.ThrowsAny<Exception>(() => row.GetAs<int>(1));
@@ -70,6 +72,7 @@ public void RowTest()
             Assert.Equal(1, row.Get("col1"));
             Assert.Equal("abc", row.Get("col2"));
             Assert.Equal(1, row.GetAs<int>("col1"));
+            Assert.Equal("abc", row.GetAs<string>("col2"));
             Assert.ThrowsAny<Exception>(() => row.GetAs<string>("col1"));
             Assert.Equal("abc", row.GetAs<string>("col2"));
             Assert.ThrowsAny<Exception>(() => row.GetAs<int>("col2"));
@@ -81,11 +84,12 @@ public void RowConstructorTest()
             Pickler pickler = CreatePickler();
 
             var schema = (StructType)DataType.ParseDataType(_testJsonSchema);
+
             var row1 = new Row(new object[] { 10, "name1" }, schema);
             var row2 = new Row(new object[] { 15, "name2" }, schema);
             byte[] pickledBytes = pickler.dumps(new[] { row1, row2 });
 
-            // Note that the following will invoke RowConstructor.construct().
+            // Note that the following will invoke RowConstructor.ctor().
             object[] unpickledData = PythonSerDe.GetUnpickledObjects(
                 new MemoryStream(pickledBytes),
                 pickledBytes.Length);
@@ -116,7 +120,7 @@ public void RowCollectorTest()
             SerDe.Write(stream, batch2.Length);
             SerDe.Write(stream, batch2);
 
-            // Rewind the memory stream so that the row collect can read from beginning.
+            // Rewind the memory stream so that the row collector can read from beginning.
             stream.Seek(0, SeekOrigin.Begin);
 
             // Set up the mock to return memory stream to which pickled data is written.
@@ -156,9 +160,96 @@ public void GenericRowTest()
             Assert.Equal(1, row.Get(0));
             Assert.Equal("abc", row.Get(1));
             Assert.Equal(1, row.GetAs<int>(0));
+            Assert.Equal("abc", row.GetAs<string>(1));
             Assert.ThrowsAny<Exception>(() => row.GetAs<string>(0));
             Assert.Equal("abc", row.GetAs<string>(1));
             Assert.ThrowsAny<Exception>(() => row.GetAs<int>(1));
         }
+
+        /// <summary>
+        /// Verifies that Row correctly handles the case where Pickler serializes a long
+        /// value as int (because it fits in int). The schema says LongType, but the
+        /// unpickled value is a boxed int. Row.Convert() should coerce it to long.
+        /// </summary>
+        [Fact]
+        public void RowGetAsLongFromPickledIntTest()
+        {
+            var schema = new StructType(new List<StructField>()
+            {
+                new StructField("id", new LongType()),
+            });
+
+            // Simulate what Pickler does: serialize a long that fits in int as an int.
+            // This is the exact scenario described in issue #27.
+            int pickledAsInt = 42;
+            var row = new Row(new object[] { pickledAsInt }, schema);
+
+            // GetAs<long> should work — the schema says LongType, so Row.Convert()
+            // should have coerced the boxed int to long.
+            Assert.Equal(42L, row.GetAs<long>(0));
+            Assert.Equal(42L, row.GetAs<long>("id"));
+
+            // Direct unbox to long should also work now.
+            Assert.IsType<long>(row.Get(0));
+            Assert.Equal(42L, (long)row.Get(0));
+
+            // Verify that an already boxed long is returned as-is (no re-boxing).
+            long boxedLong = 100L;
+            var rowWithBoxedLong = new Row(new object[] { boxedLong }, schema);
+            Assert.Same(boxedLong, rowWithBoxedLong.Get(0));
+        }
+
+        /// <summary>
+        /// Verifies that Row correctly coerces int values to long inside ArrayType.
+        /// </summary>
+        [Fact]
+        public void RowLongTypeInArrayTest()
+        {
+            var schema = new StructType(new List<StructField>()
+            {
+                new StructField("ids", new ArrayType(new LongType())),
+            });
+
+            // Pickler serializes longs that fit in int as int values inside the ArrayList.
+            var pickledArray = new ArrayList { 1, 2, 1000000000 };
+            var row = new Row(new object[] { pickledArray }, schema);
+
+            var result = (ArrayList)row.Get(0);
+            Assert.Equal(3, result.Count);
+            Assert.IsType<long>(result[0]);
+            Assert.IsType<long>(result[1]);
+            Assert.IsType<long>(result[2]);
+            Assert.Equal(1L, result[0]);
+            Assert.Equal(2L, result[1]);
+            Assert.Equal(1000000000L, result[2]);
+        }
+
+        /// <summary>
+        /// Verifies that Row correctly coerces int values to long inside MapType.
+        /// </summary>
+        [Fact]
+        public void RowLongTypeInMapTest()
+        {
+            var schema = new StructType(new List<StructField>()
+            {
+                new StructField(
+                    "data",
+                    new MapType(new StringType(), new LongType())),
+            });
+
+            // Pickler serializes longs that fit in int as int values in the Hashtable.
+            var pickledMap = new Hashtable
+            {
+                { "a", 1 },
+                { "b", 2 },
+                { "c", 1000000000 },
+            };
+            var row = new Row(new object[] { pickledMap }, schema);
+
+            var result = (Hashtable)row.Get(0);
+            Assert.Equal(1L, result["a"]);
+            Assert.Equal(2L, result["b"]);
+            Assert.Equal(1000000000L, result["c"]);
+        }
     }
 }

src/csharp/Microsoft.Spark/Sql/Row.cs

@@ -102,9 +102,6 @@ public object Get(string columnName) =>
 
         /// <summary>
         /// Returns the column value at the given index, as a type T.
-        /// TODO: If the original type is "long" and its value can be
-        /// fit into the "int", Pickler will serialize the value as int.
-        /// Since the value is boxed, <see cref="GetAs{T}(int)"/> will throw an exception.
         /// </summary>
         /// <typeparam name="T">Type to convert to</typeparam>
         /// <param name="index">Index to look up</param>
@@ -113,9 +110,6 @@ public object Get(string columnName) =>
 
         /// <summary>
         /// Returns the column value whose column name is given, as a type T.
-        /// TODO: If the original type is "long" and its value can be
-        /// fit into the "int", Pickler will serialize the value as int.
-        /// Since the value is boxed, <see cref="GetAs{T}(string)"/> will throw an exception.
         /// </summary>
         /// <typeparam name="T">Type to convert to</typeparam>
         /// <param name="columnName">Column name to look up</param>

src/csharp/Microsoft.Spark/Sql/Types/SimpleTypes.cs

@@ -1,5 +1,5 @@
 // Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
+// The .NET Foundation licenses this file to you under the MIT License.
 // See the LICENSE file in the project root for more information.
 
 using System;
@@ -100,8 +100,8 @@ public sealed class TimestampType : AtomicType
         internal override bool NeedConversion() => true;
 
         /// <summary>
-        /// Internally, a timestamp is stored as the number of microseconds as long from the epoch
-        /// of 1970-01-01T00:00:00.000000Z(UTC+00:00). This will convert internal SQL TimestampType
+        /// Internally, a timestamp is stored as the number of microseconds as long from the
+        /// epoch of 1970-01-01T00:00:00.000000Z(UTC+00:00). This will convert internal SQL TimestampType
         /// objects from the number of microseconds into native C# Timestamp objects.
         /// </summary>
         internal override object FromInternal(object obj)
@@ -111,8 +111,8 @@ internal override object FromInternal(object obj)
                 return null;
             }
 
-            // Known issue that if the original type is "long" and its value can be fit into the
-            // "int", Pickler will serialize the value as int.
+            // Known issue that if the original type is "long" and its value can fit into the
+            // "int", the Pickler will serialize the value as int.
             long val = (obj is long v) ? v : (int)obj;
             return new Timestamp(
                 new DateTime(val * 10 + DateType.s_unixTimeEpoch.Ticks, DateTimeKind.Utc));
@@ -152,6 +152,28 @@ public sealed class IntegerType : IntegralType
     /// </summary>
     public sealed class LongType : IntegralType
     {
+        internal override bool NeedConversion() => true;
+
+        /// <summary>
+        /// Converts the internal object to a .NET long. If the original type is "long" and
+        /// its value can fit into "int", the Pickler will serialize the value as int. Since
+        /// the value is boxed, a direct unbox to long would fail. This method ensures the
+        /// value is always returned as a long regardless of how the Pickler serialized it.
+        /// </summary>
+        internal override object FromInternal(object obj)
+        {
+            if (obj == null)
+            {
+                return null;
+            }
+
+            if (obj is long l)
+            {
+                return obj;  // FIX: return obj (already boxed) instead of l (re-boxes)
+            }
+
+            return Convert.ToInt64(obj);
+        }
     }
 
     /// <summary>
@@ -166,8 +188,8 @@ public sealed class ShortType : IntegralType
     /// </summary>
     public sealed class DecimalType : FractionalType
     {
-        internal static Regex s_fixedDecimal =
-            new Regex(@"decimal\(\s*(\d+)\s*,\s*(\-?\d+)\s*\)", RegexOptions.Compiled);
+        internal static readonly Regex s_fixedDecimal =
+            new Regex(@"decimal\((\d+),\s*(\d+)\)", RegexOptions.Compiled);
 
         private readonly int _precision;
         private readonly int _scale;
@@ -176,13 +198,11 @@ public sealed class DecimalType : FractionalType
         /// Initializes the <see cref="DecimalType"/> instance.
         /// </summary>
         /// <remarks>
-        /// Default values of precision and scale are from Scala:
-        /// sql/catalyst/src/main/scala/org/apache/spark/sql/types/DecimalType.scala.
+        /// Default values of precision and scale are from Scale:
+        /// sql/catalog/src/main/scala/org/apache/spark/sql/types/DecimalType.scala.
         /// </remarks>
         /// <param name="precision">Number of digits in a number</param>
-        /// <param name="scale">
-        /// Number of digits to the right of the decimal point in a number
-        /// </param>
+        /// <param name="scale">Number of digits to the right of the decimal point in a number</param>
         public DecimalType(int precision = 10, int scale = 0)
         {
             _precision = precision;