Use dataTypes if it passed in to LoadCsv (dotnet#2791)

* Fix LoadCsv to use dataType if it passed in

* sq

* Don't read the full file after guessRows lines have been read

* Address feedback

* Last round of feedback

Author: Prashanth Govindarajan

src/Microsoft.Data.Analysis/DataFrame.IO.cs

@@ -5,7 +5,6 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
-using System.Text;
 
 namespace Microsoft.Data.Analysis
 {
@@ -104,6 +103,77 @@ public static DataFrame LoadCsv(string filename,
             }
         }
 
+        private static DataFrameColumn CreateColumn(Type kind, string[] columnNames, int columnIndex)
+        {
+            PrimitiveDataFrameColumn<T> CreatePrimitiveDataFrameColumn<T>()
+                where T : unmanaged
+            {
+                return new PrimitiveDataFrameColumn<T>(columnNames == null ? "Column" + columnIndex.ToString() : columnNames[columnIndex]);
+            }
+            DataFrameColumn ret;
+            if (kind == typeof(bool))
+            {
+                ret = CreatePrimitiveDataFrameColumn<bool>();
+            }
+            else if (kind == typeof(int))
+            {
+                ret = CreatePrimitiveDataFrameColumn<int>();
+            }
+            else if (kind == typeof(float))
+            {
+                ret = CreatePrimitiveDataFrameColumn<float>();
+            }
+            else if (kind == typeof(string))
+            {
+                ret = new StringDataFrameColumn(columnNames == null ? "Column" + columnIndex.ToString() : columnNames[columnIndex], 0);
+            }
+            else if (kind == typeof(long))
+            {
+                ret = CreatePrimitiveDataFrameColumn<long>();
+            }
+            else if (kind == typeof(decimal))
+            {
+                ret = CreatePrimitiveDataFrameColumn<decimal>();
+            }
+            else if (kind == typeof(byte))
+            {
+                ret = CreatePrimitiveDataFrameColumn<byte>();
+            }
+            else if (kind == typeof(char))
+            {
+                ret = CreatePrimitiveDataFrameColumn<char>();
+            }
+            else if (kind == typeof(double))
+            {
+                ret = CreatePrimitiveDataFrameColumn<double>();
+            }
+            else if (kind == typeof(sbyte))
+            {
+                ret = CreatePrimitiveDataFrameColumn<sbyte>();
+            }
+            else if (kind == typeof(short))
+            {
+                ret = CreatePrimitiveDataFrameColumn<short>();
+            }
+            else if (kind == typeof(uint))
+            {
+                ret = CreatePrimitiveDataFrameColumn<uint>();
+            }
+            else if (kind == typeof(ulong))
+            {
+                ret = CreatePrimitiveDataFrameColumn<ulong>();
+            }
+            else if (kind == typeof(ushort))
+            {
+                ret = CreatePrimitiveDataFrameColumn<ushort>();
+            }
+            else
+            {
+                throw new NotSupportedException(nameof(kind));
+            }
+            return ret;
+        }
+
         /// <summary>
         /// Reads a seekable stream of CSV data into a DataFrame.
         /// Follows pandas API.
@@ -116,7 +186,7 @@ public static DataFrame LoadCsv(string filename,
         /// <param name="numberOfRowsToRead">number of rows to read not including the header(if present)</param>
         /// <param name="guessRows">number of rows used to guess types</param>
         /// <param name="addIndexColumn">add one column with the row index</param>
-        /// <returns>DataFrame</returns>
+        /// <returns><see cref="DataFrame"/></returns>
         public static DataFrame LoadCsv(Stream csvStream,
                                 char separator = ',', bool header = true,
                                 string[] columnNames = null, Type[] dataTypes = null,
@@ -127,7 +197,7 @@ public static DataFrame LoadCsv(Stream csvStream,
 
             var linesForGuessType = new List<string[]>();
             long rowline = 0;
-            int numberOfColumns = 0;
+            int numberOfColumns = dataTypes?.Length ?? 0;
 
             if (header == true && numberOfRowsToRead != -1)
                 numberOfRowsToRead++;
@@ -137,60 +207,52 @@ public static DataFrame LoadCsv(Stream csvStream,
             // First pass: schema and number of rows.
             using (var streamReader = new StreamReader(csvStream, encoding: null, detectEncodingFromByteOrderMarks: true, bufferSize: -1, leaveOpen: true))
             {
-                string line = streamReader.ReadLine();
-                while (line != null)
+                string line = null;
+                if (dataTypes == null)
                 {
-                    if ((numberOfRowsToRead == -1) || rowline < numberOfRowsToRead)
+                    line = streamReader.ReadLine();
+                    while (line != null)
                     {
-                        if (linesForGuessType.Count < guessRows)
+                        if ((numberOfRowsToRead == -1) || rowline < numberOfRowsToRead)
                         {
-                            var spl = line.Split(separator);
-                            if (header && rowline == 0)
+                            if (linesForGuessType.Count < guessRows)
                             {
-                                if (columnNames == null)
-                                    columnNames = spl;
-                            }
-                            else
-                            {
-                                linesForGuessType.Add(spl);
-                                numberOfColumns = Math.Max(numberOfColumns, spl.Length);
+                                var spl = line.Split(separator);
+                                if (header && rowline == 0)
+                                {
+                                    if (columnNames == null)
+                                        columnNames = spl;
+                                }
+                                else
+                                {
+                                    linesForGuessType.Add(spl);
+                                    numberOfColumns = Math.Max(numberOfColumns, spl.Length);
+                                }
                             }
                         }
+                        ++rowline;
+                        if (rowline == guessRows)
+                        {
+                            break;
+                        }
+                        line = streamReader.ReadLine();
                     }
-                    ++rowline;
-                    if (rowline == numberOfRowsToRead)
-                        break;
-                    line = streamReader.ReadLine();
-                }
 
-                if (linesForGuessType.Count == 0)
-                    throw new FormatException(Strings.EmptyFile);
+                    if (linesForGuessType.Count == 0)
+                    {
+                        throw new FormatException(Strings.EmptyFile);
+                    }
+                }
 
                 columns = new List<DataFrameColumn>(numberOfColumns);
-
-                // Guesses types and adds columns.
+                // Guesses types or looks up dataTypes and adds columns.
                 for (int i = 0; i < numberOfColumns; ++i)
                 {
-                    Type kind = GuessKind(i, linesForGuessType);
-                    if (kind == typeof(bool))
-                    {
-                        DataFrameColumn boolColumn = new PrimitiveDataFrameColumn<bool>(columnNames == null ? "Column" + i.ToString() : columnNames[i], header == true ? rowline - 1 : rowline);
-                        columns.Add(boolColumn);
-                    }
-                    else if (kind == typeof(float))
-                    {
-                        DataFrameColumn floatColumn = new PrimitiveDataFrameColumn<float>(columnNames == null ? "Column" + i.ToString() : columnNames[i], header == true ? rowline - 1 : rowline);
-                        columns.Add(floatColumn);
-                    }
-                    else if (kind == typeof(string))
-                    {
-                        DataFrameColumn stringColumn = new StringDataFrameColumn(columnNames == null ? "Column" + i.ToString() : columnNames[i], header == true ? rowline - 1 : rowline);
-                        columns.Add(stringColumn);
-                    }
-                    else
-                        throw new NotSupportedException(nameof(kind));
+                    Type kind = dataTypes == null ? GuessKind(i, linesForGuessType) : dataTypes[i];
+                    columns.Add(CreateColumn(kind, columnNames, i));
                 }
 
+                DataFrame ret = new DataFrame(columns);
                 line = null;
                 streamReader.DiscardBufferedData();
                 streamReader.BaseStream.Seek(streamStart, SeekOrigin.Begin);
@@ -207,7 +269,7 @@ public static DataFrame LoadCsv(Stream csvStream,
                     }
                     else
                     {
-                        AppendRow(columns, header == true ? rowline - 1 : rowline, spl);
+                        ret.Append(spl);
                     }
                     ++rowline;
                     line = streamReader.ReadLine();
@@ -222,61 +284,7 @@ public static DataFrame LoadCsv(Stream csvStream,
                     }
                     columns.Insert(0, indexColumn);
                 }
-            }
-            return new DataFrame(columns);
-        }
-
-        private static void AppendRow(List<DataFrameColumn> columns, long rowIndex, string[] values)
-        {
-            for (int i = 0; i < columns.Count; i++)
-            {
-                DataFrameColumn column = columns[i];
-                string val = values[i];
-                Type dType = column.DataType;
-                if (dType == typeof(bool))
-                {
-                    bool boolParse = bool.TryParse(val, out bool boolResult);
-                    if (boolParse)
-                    {
-                        column[rowIndex] = boolResult;
-                        continue;
-                    }
-                    else
-                    {
-                        if (string.IsNullOrEmpty(val))
-                        {
-                            column[rowIndex] = null;
-                            continue;
-                        }
-                        throw new ArgumentException(string.Format(Strings.MismatchedValueType, typeof(bool)), nameof(val));
-                    }
-                }
-                else if (dType == typeof(float))
-                {
-                    bool floatParse = float.TryParse(val, out float floatResult);
-                    if (floatParse)
-                    {
-                        column[rowIndex] = floatResult;
-                        continue;
-                    }
-                    else
-                    {
-                        if (string.IsNullOrEmpty(val))
-                        {
-                            column[rowIndex] = null;
-                            continue;
-                        }
-                        throw new ArgumentException(string.Format(Strings.MismatchedValueType, typeof(float)), nameof(val));
-                    }
-                }
-                else if (dType == typeof(string))
-                {
-                    column[rowIndex] = values[i];
-                }
-                else
-                {
-                    throw new NotImplementedException();
-                }
+                return ret;
             }
         }
     }

src/Microsoft.Data.Analysis/DataFrame.cs

@@ -445,6 +445,11 @@ public void Append(IEnumerable<object> row = null)
                 {
                     DataFrameColumn column = columnEnumerator.Current;
                     object value = rowEnumerator.Current;
+                    // StringDataFrameColumn can accept empty strings. The other columns interpret empty values as nulls
+                    if (value is string stringValue && string.IsNullOrEmpty(stringValue) && column.DataType != typeof(string))
+                    {
+                        value = null;
+                    }
                     if (value != null)
                     {
                         value = Convert.ChangeType(value, column.DataType);

tests/Microsoft.Data.Analysis.Tests/DataFrame.IOTests.cs

@@ -4,7 +4,6 @@
 
 using System;
 using System.IO;
-using System.Runtime.CompilerServices;
 using System.Text;
 using Xunit;
 
@@ -58,5 +57,52 @@ Stream GetStream(string streamData)
             Assert.Equal(7, reducedRows.Columns.Count);
             Assert.Equal("CMT", reducedRows["Column0"][2]);
         }
+
+        [Fact]
+        public void TestReadCsvWithTypes()
+        {
+            string data = @"vendor_id,rate_code,passenger_count,trip_time_in_secs,trip_distance,payment_type,fare_amount
+CMT,1,1,1271,3.8,CRD,17.5
+CMT,1,1,474,1.5,CRD,8
+CMT,1,1,637,1.4,CRD,8.5
+,,,,,,
+CMT,1,1,181,0.6,CSH,4.5";
+
+            Stream GetStream(string streamData)
+            {
+                return new MemoryStream(Encoding.Default.GetBytes(streamData));
+            }
+            DataFrame df = DataFrame.LoadCsv(GetStream(data), dataTypes: new Type[] { typeof(string), typeof(short), typeof(int), typeof(long), typeof(float), typeof(string), typeof(double) });
+            Assert.Equal(5, df.RowCount);
+            Assert.Equal(7, df.Columns.Count);
+
+            Assert.True(typeof(string) == df.Columns[0].DataType);
+            Assert.True(typeof(short) == df.Columns[1].DataType);
+            Assert.True(typeof(int) == df.Columns[2].DataType);
+            Assert.True(typeof(long) == df.Columns[3].DataType);
+            Assert.True(typeof(float) == df.Columns[4].DataType);
+            Assert.True(typeof(string) == df.Columns[5].DataType);
+            Assert.True(typeof(double) == df.Columns[6].DataType);
+
+            foreach (var column in df.Columns)
+            {
+                if (column.DataType != typeof(string))
+                {
+                    Assert.Equal(1, column.NullCount);
+                }
+                else
+                {
+                    Assert.Equal(0, column.NullCount);
+                }
+            }
+            var nullRow = df[3];
+            Assert.Equal("", nullRow[0]);
+            Assert.Null(nullRow[1]);
+            Assert.Null(nullRow[2]);
+            Assert.Null(nullRow[3]);
+            Assert.Null(nullRow[4]);
+            Assert.Equal("", nullRow[5]);
+            Assert.Null(nullRow[6]);
+        }
     }
 }

tests/Microsoft.Data.Analysis.Tests/DataFrameTests.cs

@@ -1926,5 +1926,28 @@ public void TestAppendRow()
             Assert.Equal(5, df.Columns[0].NullCount);
             Assert.Equal(6, df.Columns[1].NullCount);
         }
+
+        [Fact]
+        public void TestAppendEmptyValue()
+        {
+            DataFrame df = MakeDataFrame<int, bool>(10);
+            df.Append(new List<object> { "", true });
+            Assert.Equal(11, df.RowCount);
+            Assert.Equal(2, df.Columns[0].NullCount);
+            Assert.Equal(1, df.Columns[1].NullCount);
+
+            StringDataFrameColumn column = new StringDataFrameColumn("Strings", Enumerable.Range(0, 11).Select(x => x.ToString()));
+            df.Columns.Add(column);
+
+            df.Append(new List<object> { 1, true, "" });
+            Assert.Equal(12, df.RowCount);
+            Assert.Equal(2, df.Columns[0].NullCount);
+            Assert.Equal(1, df.Columns[1].NullCount);
+            Assert.Equal(0, df.Columns[2].NullCount);
+
+            df.Append(new List<object> { 1, true, null });
+            Assert.Equal(13, df.RowCount);
+            Assert.Equal(1, df.Columns[2].NullCount);
+        }
     }
 }