Fix Row unpickling for long values that fit in int

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

@@ -160,5 +160,33 @@ public void GenericRowTest()
             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));
+        }
     }
 }

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

@@ -145,20 +145,82 @@ public sealed class ByteType : IntegralType
     /// </summary>
     public sealed class IntegerType : IntegralType
     {
+        internal override bool NeedConversion() => true;
+
+        /// <summary>
+        /// Converts the internal object to a .NET int. The Pickler may serialize a short
+        /// or byte value as its native type, so we need to ensure the value is an int
+        /// to match the schema.
+        /// </summary>
+        internal override object FromInternal(object obj)
+        {
+            if (obj == null)
+            {
+                return null;
+            }
+
+            if (obj is int i)
+            {
+                return i;
+            }
+
+            return Convert.ToInt32(obj);
+        }
     }
 
     /// <summary>
     /// Represents a long type.
     /// </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 l;
+            }
+
+            return Convert.ToInt64(obj);
+        }
     }
 
     /// <summary>
     /// Represents a short type.
     /// </summary>
     public sealed class ShortType : IntegralType
     {
+        internal override bool NeedConversion() => true;
+
+        /// <summary>
+        /// Converts the internal object to a .NET short.
+        /// </summary>
+        internal override object FromInternal(object obj)
+        {
+            if (obj == null)
+            {
+                return null;
+            }
+
+            if (obj is short s)
+            {
+                return s;
+            }
+
+            return Convert.ToInt16(obj);
+        }
     }
 
     /// <summary>