diff --git a/src/Microsoft.Data.Analysis/BooleanDataFrameColumn.cs b/src/Microsoft.Data.Analysis/BooleanDataFrameColumn.cs
new file mode 100644
index 0000000000..982463d182
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/BooleanDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class BooleanDataFrameColumn : PrimitiveDataFrameColumn<bool>
+    {
+        public BooleanDataFrameColumn(string name, IEnumerable<bool?> values) : base(name, values) { }
+
+        public BooleanDataFrameColumn(string name, IEnumerable<bool> values) : base(name, values) { }
+
+        public BooleanDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public BooleanDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal BooleanDataFrameColumn(string name, PrimitiveColumnContainer<bool> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/ByteDataFrameColumn.cs b/src/Microsoft.Data.Analysis/ByteDataFrameColumn.cs
new file mode 100644
index 0000000000..369e6d33ee
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/ByteDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class ByteDataFrameColumn : PrimitiveDataFrameColumn<byte>
+    {
+        public ByteDataFrameColumn(string name, IEnumerable<byte?> values) : base(name, values) { }
+
+        public ByteDataFrameColumn(string name, IEnumerable<byte> values) : base(name, values) { }
+
+        public ByteDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public ByteDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal ByteDataFrameColumn(string name, PrimitiveColumnContainer<byte> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/CharDataFrameColumn.cs b/src/Microsoft.Data.Analysis/CharDataFrameColumn.cs
new file mode 100644
index 0000000000..1dcedc4fc9
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/CharDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class CharDataFrameColumn : PrimitiveDataFrameColumn<char>
+    {
+        public CharDataFrameColumn(string name, IEnumerable<char?> values) : base(name, values) { }
+
+        public CharDataFrameColumn(string name, IEnumerable<char> values) : base(name, values) { }
+
+        public CharDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public CharDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal CharDataFrameColumn(string name, PrimitiveColumnContainer<char> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/ColumnArithmeticTemplate.ttinclude b/src/Microsoft.Data.Analysis/ColumnArithmeticTemplate.ttinclude
index 55960ee376..f97eea5a1e 100644
--- a/src/Microsoft.Data.Analysis/ColumnArithmeticTemplate.ttinclude
+++ b/src/Microsoft.Data.Analysis/ColumnArithmeticTemplate.ttinclude
@@ -38,6 +38,56 @@
         return $"{keyword} (typeof(T) == typeof({type.TypeName}))";
     }
 
+        // A way to discern implicit conversions. For ex: short has primitivitty 2. int has primitivitty 3. primitivitty(short) + primitivitty(int) > 2 * primitivitty(short) implying that a conversion has to take place 
+    public Dictionary<string, int> primitiveTypeToPrimitivityLevelMap = new Dictionary<string, int> {
+        {"byte", 1},
+        {"sbyte", 1},
+        {"short", 2},
+        {"ushort", 2},
+        {"int", 3},
+        {"uint", 3},
+        {"long", 4},
+        {"ulong", 4},
+        {"float", 5},
+        {"double", 6},
+        {"decimal", 7}
+    };
+
+    public string GetCapitalizedPrimitiveTypes(string type)
+    {
+        string typeFirstCharUpper;
+        if (type.First() == 'u' || type == "sbyte")
+        {
+            typeFirstCharUpper = type[0].ToString().ToUpper() + type[1].ToString().ToUpper() + type.Substring(2);
+        }
+        else
+        {
+            typeFirstCharUpper = type[0].ToString().ToUpper() + type.Substring(1);
+        }
+        return typeFirstCharUpper;
+    }
+
+    public bool IsMixedSignedAndUnsignedTypePair(string t1, string t2)
+    {
+        if (t1 == "byte" && t2 == "sbyte")
+        {
+            return true;
+        }
+        if (t2 == "byte" && t1 == "sbyte")
+        {
+            return true;
+        }
+        if (("u" + t1) == t2)
+        {
+            return true;
+        }
+        if (("u" + t2) == t1)
+        {
+            return true;
+        }
+        return false;
+    }
+
     public TypeConfiguration[] typeConfiguration = new []
     {
         new TypeConfiguration("bool", oneLiteral:"true", zeroLiteral:"false", supportsNumeric: false, unsupportedMethods: new[] {"LeftShift", "RightShift"}),
@@ -55,6 +105,30 @@
         new TypeConfiguration("ushort", classPrefix:"UShort", unsupportedMethods: new[] {"UnaryMinus", "All", "Any"})
     };
 
+    public string GetBinaryOperationReturnType(TypeConfiguration t1, TypeConfiguration t2)
+    {
+        int t1Level;
+        if (!primitiveTypeToPrimitivityLevelMap.TryGetValue(t1.TypeName, out t1Level))
+        {
+            throw new Exception("Unknown type");
+        }
+        int t2Level;
+        if (!primitiveTypeToPrimitivityLevelMap.TryGetValue(t2.TypeName, out t2Level))
+        {
+            throw new Exception("Unknown type");
+        }
+        if (t1Level + t2Level <= 2 * t1Level)
+        {
+            return t1.TypeName;
+        }
+        if (t1Level + t2Level <= 2 * t2Level)
+        {
+            return t2.TypeName;
+        }
+        throw new Exception("Bug in GetBinaryOperationReturnType");
+        return "";
+    }
+
     public enum MethodType
     {
        Unary,
diff --git a/src/Microsoft.Data.Analysis/Converters.cs b/src/Microsoft.Data.Analysis/Converters.cs
new file mode 100644
index 0000000000..a3d2abc67c
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/Converters.cs
@@ -0,0 +1,948 @@
+
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+// Generated from Converters.tt. Do not modify directly
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.Data.Analysis
+{
+    internal interface IByteConverter<T>
+    {
+        byte GetByte(T value);
+    }
+    internal static class ByteConverter<T>
+    {
+        public static IByteConverter<T> Instance { get; } = ByteConverter.GetByteConverter<T>();
+    }
+    internal static class ByteConverter
+    {
+        public static IByteConverter<T> GetByteConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IByteConverter<T>)new ByteByteConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IByteConverter<T>)new SByteByteConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteByteConverter : IByteConverter<byte>
+    {
+        public byte GetByte(byte value)
+        {
+            return (byte)value;
+        }
+    }
+    internal class SByteByteConverter : IByteConverter<sbyte>
+    {
+        public byte GetByte(sbyte value)
+        {
+            return (byte)value;
+        }
+    }
+    internal interface ISByteConverter<T>
+    {
+        sbyte GetSByte(T value);
+    }
+    internal static class SByteConverter<T>
+    {
+        public static ISByteConverter<T> Instance { get; } = SByteConverter.GetSByteConverter<T>();
+    }
+    internal static class SByteConverter
+    {
+        public static ISByteConverter<T> GetSByteConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (ISByteConverter<T>)new ByteSByteConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (ISByteConverter<T>)new SByteSByteConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteSByteConverter : ISByteConverter<byte>
+    {
+        public sbyte GetSByte(byte value)
+        {
+            return (sbyte)value;
+        }
+    }
+    internal class SByteSByteConverter : ISByteConverter<sbyte>
+    {
+        public sbyte GetSByte(sbyte value)
+        {
+            return (sbyte)value;
+        }
+    }
+    internal interface IShortConverter<T>
+    {
+        short GetShort(T value);
+    }
+    internal static class ShortConverter<T>
+    {
+        public static IShortConverter<T> Instance { get; } = ShortConverter.GetShortConverter<T>();
+    }
+    internal static class ShortConverter
+    {
+        public static IShortConverter<T> GetShortConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IShortConverter<T>)new ByteShortConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IShortConverter<T>)new SByteShortConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IShortConverter<T>)new ShortShortConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IShortConverter<T>)new UShortShortConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteShortConverter : IShortConverter<byte>
+    {
+        public short GetShort(byte value)
+        {
+            return (short)value;
+        }
+    }
+    internal class SByteShortConverter : IShortConverter<sbyte>
+    {
+        public short GetShort(sbyte value)
+        {
+            return (short)value;
+        }
+    }
+    internal class ShortShortConverter : IShortConverter<short>
+    {
+        public short GetShort(short value)
+        {
+            return (short)value;
+        }
+    }
+    internal class UShortShortConverter : IShortConverter<ushort>
+    {
+        public short GetShort(ushort value)
+        {
+            return (short)value;
+        }
+    }
+    internal interface IUShortConverter<T>
+    {
+        ushort GetUShort(T value);
+    }
+    internal static class UShortConverter<T>
+    {
+        public static IUShortConverter<T> Instance { get; } = UShortConverter.GetUShortConverter<T>();
+    }
+    internal static class UShortConverter
+    {
+        public static IUShortConverter<T> GetUShortConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IUShortConverter<T>)new ByteUShortConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IUShortConverter<T>)new SByteUShortConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IUShortConverter<T>)new ShortUShortConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IUShortConverter<T>)new UShortUShortConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteUShortConverter : IUShortConverter<byte>
+    {
+        public ushort GetUShort(byte value)
+        {
+            return (ushort)value;
+        }
+    }
+    internal class SByteUShortConverter : IUShortConverter<sbyte>
+    {
+        public ushort GetUShort(sbyte value)
+        {
+            return (ushort)value;
+        }
+    }
+    internal class ShortUShortConverter : IUShortConverter<short>
+    {
+        public ushort GetUShort(short value)
+        {
+            return (ushort)value;
+        }
+    }
+    internal class UShortUShortConverter : IUShortConverter<ushort>
+    {
+        public ushort GetUShort(ushort value)
+        {
+            return (ushort)value;
+        }
+    }
+    internal interface IIntConverter<T>
+    {
+        int GetInt(T value);
+    }
+    internal static class IntConverter<T>
+    {
+        public static IIntConverter<T> Instance { get; } = IntConverter.GetIntConverter<T>();
+    }
+    internal static class IntConverter
+    {
+        public static IIntConverter<T> GetIntConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IIntConverter<T>)new ByteIntConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IIntConverter<T>)new SByteIntConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IIntConverter<T>)new ShortIntConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IIntConverter<T>)new UShortIntConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (IIntConverter<T>)new IntIntConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (IIntConverter<T>)new UIntIntConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteIntConverter : IIntConverter<byte>
+    {
+        public int GetInt(byte value)
+        {
+            return (int)value;
+        }
+    }
+    internal class SByteIntConverter : IIntConverter<sbyte>
+    {
+        public int GetInt(sbyte value)
+        {
+            return (int)value;
+        }
+    }
+    internal class ShortIntConverter : IIntConverter<short>
+    {
+        public int GetInt(short value)
+        {
+            return (int)value;
+        }
+    }
+    internal class UShortIntConverter : IIntConverter<ushort>
+    {
+        public int GetInt(ushort value)
+        {
+            return (int)value;
+        }
+    }
+    internal class IntIntConverter : IIntConverter<int>
+    {
+        public int GetInt(int value)
+        {
+            return (int)value;
+        }
+    }
+    internal class UIntIntConverter : IIntConverter<uint>
+    {
+        public int GetInt(uint value)
+        {
+            return (int)value;
+        }
+    }
+    internal interface IUIntConverter<T>
+    {
+        uint GetUInt(T value);
+    }
+    internal static class UIntConverter<T>
+    {
+        public static IUIntConverter<T> Instance { get; } = UIntConverter.GetUIntConverter<T>();
+    }
+    internal static class UIntConverter
+    {
+        public static IUIntConverter<T> GetUIntConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IUIntConverter<T>)new ByteUIntConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IUIntConverter<T>)new SByteUIntConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IUIntConverter<T>)new ShortUIntConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IUIntConverter<T>)new UShortUIntConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (IUIntConverter<T>)new IntUIntConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (IUIntConverter<T>)new UIntUIntConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteUIntConverter : IUIntConverter<byte>
+    {
+        public uint GetUInt(byte value)
+        {
+            return (uint)value;
+        }
+    }
+    internal class SByteUIntConverter : IUIntConverter<sbyte>
+    {
+        public uint GetUInt(sbyte value)
+        {
+            return (uint)value;
+        }
+    }
+    internal class ShortUIntConverter : IUIntConverter<short>
+    {
+        public uint GetUInt(short value)
+        {
+            return (uint)value;
+        }
+    }
+    internal class UShortUIntConverter : IUIntConverter<ushort>
+    {
+        public uint GetUInt(ushort value)
+        {
+            return (uint)value;
+        }
+    }
+    internal class IntUIntConverter : IUIntConverter<int>
+    {
+        public uint GetUInt(int value)
+        {
+            return (uint)value;
+        }
+    }
+    internal class UIntUIntConverter : IUIntConverter<uint>
+    {
+        public uint GetUInt(uint value)
+        {
+            return (uint)value;
+        }
+    }
+    internal interface ILongConverter<T>
+    {
+        long GetLong(T value);
+    }
+    internal static class LongConverter<T>
+    {
+        public static ILongConverter<T> Instance { get; } = LongConverter.GetLongConverter<T>();
+    }
+    internal static class LongConverter
+    {
+        public static ILongConverter<T> GetLongConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (ILongConverter<T>)new ByteLongConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (ILongConverter<T>)new SByteLongConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (ILongConverter<T>)new ShortLongConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (ILongConverter<T>)new UShortLongConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (ILongConverter<T>)new IntLongConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (ILongConverter<T>)new UIntLongConverter();
+            }
+            if (typeof(T) == typeof(long))
+            {
+                return (ILongConverter<T>)new LongLongConverter();
+            }
+            if (typeof(T) == typeof(ulong))
+            {
+                return (ILongConverter<T>)new ULongLongConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteLongConverter : ILongConverter<byte>
+    {
+        public long GetLong(byte value)
+        {
+            return (long)value;
+        }
+    }
+    internal class SByteLongConverter : ILongConverter<sbyte>
+    {
+        public long GetLong(sbyte value)
+        {
+            return (long)value;
+        }
+    }
+    internal class ShortLongConverter : ILongConverter<short>
+    {
+        public long GetLong(short value)
+        {
+            return (long)value;
+        }
+    }
+    internal class UShortLongConverter : ILongConverter<ushort>
+    {
+        public long GetLong(ushort value)
+        {
+            return (long)value;
+        }
+    }
+    internal class IntLongConverter : ILongConverter<int>
+    {
+        public long GetLong(int value)
+        {
+            return (long)value;
+        }
+    }
+    internal class UIntLongConverter : ILongConverter<uint>
+    {
+        public long GetLong(uint value)
+        {
+            return (long)value;
+        }
+    }
+    internal class LongLongConverter : ILongConverter<long>
+    {
+        public long GetLong(long value)
+        {
+            return (long)value;
+        }
+    }
+    internal class ULongLongConverter : ILongConverter<ulong>
+    {
+        public long GetLong(ulong value)
+        {
+            return (long)value;
+        }
+    }
+    internal interface IULongConverter<T>
+    {
+        ulong GetULong(T value);
+    }
+    internal static class ULongConverter<T>
+    {
+        public static IULongConverter<T> Instance { get; } = ULongConverter.GetULongConverter<T>();
+    }
+    internal static class ULongConverter
+    {
+        public static IULongConverter<T> GetULongConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IULongConverter<T>)new ByteULongConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IULongConverter<T>)new SByteULongConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IULongConverter<T>)new ShortULongConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IULongConverter<T>)new UShortULongConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (IULongConverter<T>)new IntULongConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (IULongConverter<T>)new UIntULongConverter();
+            }
+            if (typeof(T) == typeof(long))
+            {
+                return (IULongConverter<T>)new LongULongConverter();
+            }
+            if (typeof(T) == typeof(ulong))
+            {
+                return (IULongConverter<T>)new ULongULongConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteULongConverter : IULongConverter<byte>
+    {
+        public ulong GetULong(byte value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class SByteULongConverter : IULongConverter<sbyte>
+    {
+        public ulong GetULong(sbyte value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class ShortULongConverter : IULongConverter<short>
+    {
+        public ulong GetULong(short value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class UShortULongConverter : IULongConverter<ushort>
+    {
+        public ulong GetULong(ushort value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class IntULongConverter : IULongConverter<int>
+    {
+        public ulong GetULong(int value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class UIntULongConverter : IULongConverter<uint>
+    {
+        public ulong GetULong(uint value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class LongULongConverter : IULongConverter<long>
+    {
+        public ulong GetULong(long value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal class ULongULongConverter : IULongConverter<ulong>
+    {
+        public ulong GetULong(ulong value)
+        {
+            return (ulong)value;
+        }
+    }
+    internal interface IFloatConverter<T>
+    {
+        float GetFloat(T value);
+    }
+    internal static class FloatConverter<T>
+    {
+        public static IFloatConverter<T> Instance { get; } = FloatConverter.GetFloatConverter<T>();
+    }
+    internal static class FloatConverter
+    {
+        public static IFloatConverter<T> GetFloatConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IFloatConverter<T>)new ByteFloatConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IFloatConverter<T>)new SByteFloatConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IFloatConverter<T>)new ShortFloatConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IFloatConverter<T>)new UShortFloatConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (IFloatConverter<T>)new IntFloatConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (IFloatConverter<T>)new UIntFloatConverter();
+            }
+            if (typeof(T) == typeof(long))
+            {
+                return (IFloatConverter<T>)new LongFloatConverter();
+            }
+            if (typeof(T) == typeof(ulong))
+            {
+                return (IFloatConverter<T>)new ULongFloatConverter();
+            }
+            if (typeof(T) == typeof(float))
+            {
+                return (IFloatConverter<T>)new FloatFloatConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteFloatConverter : IFloatConverter<byte>
+    {
+        public float GetFloat(byte value)
+        {
+            return (float)value;
+        }
+    }
+    internal class SByteFloatConverter : IFloatConverter<sbyte>
+    {
+        public float GetFloat(sbyte value)
+        {
+            return (float)value;
+        }
+    }
+    internal class ShortFloatConverter : IFloatConverter<short>
+    {
+        public float GetFloat(short value)
+        {
+            return (float)value;
+        }
+    }
+    internal class UShortFloatConverter : IFloatConverter<ushort>
+    {
+        public float GetFloat(ushort value)
+        {
+            return (float)value;
+        }
+    }
+    internal class IntFloatConverter : IFloatConverter<int>
+    {
+        public float GetFloat(int value)
+        {
+            return (float)value;
+        }
+    }
+    internal class UIntFloatConverter : IFloatConverter<uint>
+    {
+        public float GetFloat(uint value)
+        {
+            return (float)value;
+        }
+    }
+    internal class LongFloatConverter : IFloatConverter<long>
+    {
+        public float GetFloat(long value)
+        {
+            return (float)value;
+        }
+    }
+    internal class ULongFloatConverter : IFloatConverter<ulong>
+    {
+        public float GetFloat(ulong value)
+        {
+            return (float)value;
+        }
+    }
+    internal class FloatFloatConverter : IFloatConverter<float>
+    {
+        public float GetFloat(float value)
+        {
+            return (float)value;
+        }
+    }
+    internal interface IDoubleConverter<T>
+    {
+        double GetDouble(T value);
+    }
+    internal static class DoubleConverter<T>
+    {
+        public static IDoubleConverter<T> Instance { get; } = DoubleConverter.GetDoubleConverter<T>();
+    }
+    internal static class DoubleConverter
+    {
+        public static IDoubleConverter<T> GetDoubleConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IDoubleConverter<T>)new ByteDoubleConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IDoubleConverter<T>)new SByteDoubleConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IDoubleConverter<T>)new ShortDoubleConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IDoubleConverter<T>)new UShortDoubleConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (IDoubleConverter<T>)new IntDoubleConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (IDoubleConverter<T>)new UIntDoubleConverter();
+            }
+            if (typeof(T) == typeof(long))
+            {
+                return (IDoubleConverter<T>)new LongDoubleConverter();
+            }
+            if (typeof(T) == typeof(ulong))
+            {
+                return (IDoubleConverter<T>)new ULongDoubleConverter();
+            }
+            if (typeof(T) == typeof(float))
+            {
+                return (IDoubleConverter<T>)new FloatDoubleConverter();
+            }
+            if (typeof(T) == typeof(double))
+            {
+                return (IDoubleConverter<T>)new DoubleDoubleConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteDoubleConverter : IDoubleConverter<byte>
+    {
+        public double GetDouble(byte value)
+        {
+            return (double)value;
+        }
+    }
+    internal class SByteDoubleConverter : IDoubleConverter<sbyte>
+    {
+        public double GetDouble(sbyte value)
+        {
+            return (double)value;
+        }
+    }
+    internal class ShortDoubleConverter : IDoubleConverter<short>
+    {
+        public double GetDouble(short value)
+        {
+            return (double)value;
+        }
+    }
+    internal class UShortDoubleConverter : IDoubleConverter<ushort>
+    {
+        public double GetDouble(ushort value)
+        {
+            return (double)value;
+        }
+    }
+    internal class IntDoubleConverter : IDoubleConverter<int>
+    {
+        public double GetDouble(int value)
+        {
+            return (double)value;
+        }
+    }
+    internal class UIntDoubleConverter : IDoubleConverter<uint>
+    {
+        public double GetDouble(uint value)
+        {
+            return (double)value;
+        }
+    }
+    internal class LongDoubleConverter : IDoubleConverter<long>
+    {
+        public double GetDouble(long value)
+        {
+            return (double)value;
+        }
+    }
+    internal class ULongDoubleConverter : IDoubleConverter<ulong>
+    {
+        public double GetDouble(ulong value)
+        {
+            return (double)value;
+        }
+    }
+    internal class FloatDoubleConverter : IDoubleConverter<float>
+    {
+        public double GetDouble(float value)
+        {
+            return (double)value;
+        }
+    }
+    internal class DoubleDoubleConverter : IDoubleConverter<double>
+    {
+        public double GetDouble(double value)
+        {
+            return (double)value;
+        }
+    }
+    internal interface IDecimalConverter<T>
+    {
+        decimal GetDecimal(T value);
+    }
+    internal static class DecimalConverter<T>
+    {
+        public static IDecimalConverter<T> Instance { get; } = DecimalConverter.GetDecimalConverter<T>();
+    }
+    internal static class DecimalConverter
+    {
+        public static IDecimalConverter<T> GetDecimalConverter<T>()
+        {
+            if (typeof(T) == typeof(byte))
+            {
+                return (IDecimalConverter<T>)new ByteDecimalConverter();
+            }
+            if (typeof(T) == typeof(sbyte))
+            {
+                return (IDecimalConverter<T>)new SByteDecimalConverter();
+            }
+            if (typeof(T) == typeof(short))
+            {
+                return (IDecimalConverter<T>)new ShortDecimalConverter();
+            }
+            if (typeof(T) == typeof(ushort))
+            {
+                return (IDecimalConverter<T>)new UShortDecimalConverter();
+            }
+            if (typeof(T) == typeof(int))
+            {
+                return (IDecimalConverter<T>)new IntDecimalConverter();
+            }
+            if (typeof(T) == typeof(uint))
+            {
+                return (IDecimalConverter<T>)new UIntDecimalConverter();
+            }
+            if (typeof(T) == typeof(long))
+            {
+                return (IDecimalConverter<T>)new LongDecimalConverter();
+            }
+            if (typeof(T) == typeof(ulong))
+            {
+                return (IDecimalConverter<T>)new ULongDecimalConverter();
+            }
+            if (typeof(T) == typeof(float))
+            {
+                return (IDecimalConverter<T>)new FloatDecimalConverter();
+            }
+            if (typeof(T) == typeof(double))
+            {
+                return (IDecimalConverter<T>)new DoubleDecimalConverter();
+            }
+            if (typeof(T) == typeof(decimal))
+            {
+                return (IDecimalConverter<T>)new DecimalDecimalConverter();
+            }
+            throw new NotSupportedException();
+        }
+    }
+    internal class ByteDecimalConverter : IDecimalConverter<byte>
+    {
+        public decimal GetDecimal(byte value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class SByteDecimalConverter : IDecimalConverter<sbyte>
+    {
+        public decimal GetDecimal(sbyte value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class ShortDecimalConverter : IDecimalConverter<short>
+    {
+        public decimal GetDecimal(short value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class UShortDecimalConverter : IDecimalConverter<ushort>
+    {
+        public decimal GetDecimal(ushort value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class IntDecimalConverter : IDecimalConverter<int>
+    {
+        public decimal GetDecimal(int value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class UIntDecimalConverter : IDecimalConverter<uint>
+    {
+        public decimal GetDecimal(uint value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class LongDecimalConverter : IDecimalConverter<long>
+    {
+        public decimal GetDecimal(long value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class ULongDecimalConverter : IDecimalConverter<ulong>
+    {
+        public decimal GetDecimal(ulong value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class FloatDecimalConverter : IDecimalConverter<float>
+    {
+        public decimal GetDecimal(float value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class DoubleDecimalConverter : IDecimalConverter<double>
+    {
+        public decimal GetDecimal(double value)
+        {
+            return (decimal)value;
+        }
+    }
+    internal class DecimalDecimalConverter : IDecimalConverter<decimal>
+    {
+        public decimal GetDecimal(decimal value)
+        {
+            return (decimal)value;
+        }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/Converters.tt b/src/Microsoft.Data.Analysis/Converters.tt
new file mode 100644
index 0000000000..c21661f75e
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/Converters.tt
@@ -0,0 +1,84 @@
+<#@ template debug="false" hostspecific="false" language="C#" #>
+<#@ assembly name="System.Core" #>
+<#@ import namespace="System.Linq" #>
+<#@ import namespace="System.Text" #>
+<#@ import namespace="System.Collections.Generic" #>
+<#@ output extension=".cs" #>
+<#@ include file="ColumnArithmeticTemplate.ttinclude" #>
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+// Generated from Converters.tt. Do not modify directly
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.Data.Analysis
+{
+<#
+    foreach (var typeToLevel in primitiveTypeToPrimitivityLevelMap)
+    {
+        string type = typeToLevel.Key;
+        int level = typeToLevel.Value;
+        string typeFirstCharUpper = GetCapitalizedPrimitiveTypes(type);
+#>
+    internal interface I<#=typeFirstCharUpper#>Converter<T>
+    {
+        <#=type#> Get<#=typeFirstCharUpper#>(T value);
+    }
+    internal static class <#=typeFirstCharUpper#>Converter<T>
+    {
+        public static I<#=typeFirstCharUpper#>Converter<T> Instance { get; } = <#=typeFirstCharUpper#>Converter.Get<#=typeFirstCharUpper#>Converter<T>();
+    }
+    internal static class <#=typeFirstCharUpper#>Converter
+    {
+        public static I<#=typeFirstCharUpper#>Converter<T> Get<#=typeFirstCharUpper#>Converter<T>()
+        {
+<#
+    foreach (var typeLoop in primitiveTypeToPrimitivityLevelMap)
+    {
+        int loopLevel = typeLoop.Value;
+        if (loopLevel > level)
+        {
+            continue;
+        }
+#>
+            if (typeof(T) == typeof(<#=typeLoop.Key#>))
+            {
+<#
+                string capitalizedLoopPrimitive = GetCapitalizedPrimitiveTypes(typeLoop.Key);
+#>
+                return (I<#=typeFirstCharUpper#>Converter<T>)new <#=capitalizedLoopPrimitive#><#=typeFirstCharUpper#>Converter();
+            }
+<#
+    }
+#>
+            throw new NotSupportedException();
+        }
+    }
+<#
+    // Generate the individual type converter classes
+    foreach (var typeLoop in primitiveTypeToPrimitivityLevelMap)
+    {
+        int loopLevel = typeLoop.Value;
+        if (loopLevel > level)
+        {
+            continue;
+        }
+        string capitalizedLoopPrimitive = GetCapitalizedPrimitiveTypes(typeLoop.Key);
+#>
+    internal class <#=capitalizedLoopPrimitive#><#=typeFirstCharUpper#>Converter : I<#=typeFirstCharUpper#>Converter<<#=typeLoop.Key#>>
+    {
+        public <#=type#> Get<#=typeFirstCharUpper#>(<#=typeLoop.Key#> value)
+        {
+            return (<#=type#>)value;
+        }
+    }
+<#
+    }
+#>
+<#
+    }
+#>
+}
diff --git a/src/Microsoft.Data.Analysis/DecimalConverter.cs b/src/Microsoft.Data.Analysis/DecimalConverter.cs
deleted file mode 100644
index 97240969b0..0000000000
--- a/src/Microsoft.Data.Analysis/DecimalConverter.cs
+++ /dev/null
@@ -1,170 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace Microsoft.Data.Analysis
-{
-    internal interface IDecimalConverter<T>
-    {
-        decimal GetDecimal(T value);
-    }
-    internal static class DecimalConverter<T>
-    {
-        public static IDecimalConverter<T> Instance { get; } = DecimalConverter.GetDecimalConverter<T>();
-    }
-    internal static class DecimalConverter
-    {
-        public static IDecimalConverter<T> GetDecimalConverter<T>()
-        {
-            if (typeof(T) == typeof(bool))
-            {
-                throw new NotImplementedException();
-            }
-            else if (typeof(T) == typeof(byte))
-            {
-                return (IDecimalConverter<T>)new ByteDecimalConverter();
-            }
-            else if (typeof(T) == typeof(char))
-            {
-                return (IDecimalConverter<T>)new CharDecimalConverter();
-            }
-            else if (typeof(T) == typeof(decimal))
-            {
-                return (IDecimalConverter<T>)new DecimalDecimalConverter();
-            }
-            else if (typeof(T) == typeof(double))
-            {
-                return (IDecimalConverter<T>)new DoubleDecimalConverter();
-            }
-            else if (typeof(T) == typeof(float))
-            {
-                return (IDecimalConverter<T>)new FloatDecimalConverter();
-            }
-            else if (typeof(T) == typeof(int))
-            {
-                return (IDecimalConverter<T>)new IntDecimalConverter();
-            }
-            else if (typeof(T) == typeof(long))
-            {
-                return (IDecimalConverter<T>)new LongDecimalConverter();
-            }
-            else if (typeof(T) == typeof(sbyte))
-            {
-                return (IDecimalConverter<T>)new SByteDecimalConverter();
-            }
-            else if (typeof(T) == typeof(short))
-            {
-                return (IDecimalConverter<T>)new ShortDecimalConverter();
-            }
-            else if (typeof(T) == typeof(uint))
-            {
-                return (IDecimalConverter<T>)new UIntDecimalConverter();
-            }
-            else if (typeof(T) == typeof(ulong))
-            {
-                return (IDecimalConverter<T>)new ULongDecimalConverter();
-            }
-            else if (typeof(T) == typeof(ushort))
-            {
-                return (IDecimalConverter<T>)new UShortDecimalConverter();
-            }
-            throw new NotSupportedException();
-        }
-    }
-
-    internal class ByteDecimalConverter : IDecimalConverter<byte>
-    {
-        public decimal GetDecimal(byte value)
-        {
-            return value;
-        }
-    }
-
-    internal class CharDecimalConverter : IDecimalConverter<char>
-    {
-        public decimal GetDecimal(char value)
-        {
-            return value;
-        }
-    }
-
-    internal class DecimalDecimalConverter : IDecimalConverter<decimal>
-    {
-        public decimal GetDecimal(decimal value)
-        {
-            return value;
-        }
-    }
-
-    internal class DoubleDecimalConverter : IDecimalConverter<double>
-    {
-        public decimal GetDecimal(double value)
-        {
-            return (decimal)value;
-        }
-    }
-
-    internal class FloatDecimalConverter : IDecimalConverter<float>
-    {
-        public decimal GetDecimal(float value)
-        {
-            return (decimal)value;
-        }
-    }
-
-    internal class IntDecimalConverter : IDecimalConverter<int>
-    {
-        public decimal GetDecimal(int value)
-        {
-            return value;
-        }
-    }
-
-    internal class LongDecimalConverter : IDecimalConverter<long>
-    {
-        public decimal GetDecimal(long value)
-        {
-            return value;
-        }
-    }
-
-    internal class SByteDecimalConverter : IDecimalConverter<sbyte>
-    {
-        public decimal GetDecimal(sbyte value)
-        {
-            return value;
-        }
-    }
-
-    internal class ShortDecimalConverter : IDecimalConverter<short>
-    {
-        public decimal GetDecimal(short value)
-        {
-            return value;
-        }
-    }
-
-    internal class UIntDecimalConverter : IDecimalConverter<uint>
-    {
-        public decimal GetDecimal(uint value)
-        {
-            return value;
-        }
-    }
-
-    internal class ULongDecimalConverter : IDecimalConverter<ulong>
-    {
-        public decimal GetDecimal(ulong value)
-        {
-            return value;
-        }
-    }
-
-    internal class UShortDecimalConverter : IDecimalConverter<ushort>
-    {
-        public decimal GetDecimal(ushort value)
-        {
-            return value;
-        }
-    }
-}
diff --git a/src/Microsoft.Data.Analysis/DecimalDataFrameColumn.cs b/src/Microsoft.Data.Analysis/DecimalDataFrameColumn.cs
new file mode 100644
index 0000000000..b0b4fafd04
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/DecimalDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class DecimalDataFrameColumn : PrimitiveDataFrameColumn<decimal>
+    {
+        public DecimalDataFrameColumn(string name, IEnumerable<decimal?> values) : base(name, values) { }
+
+        public DecimalDataFrameColumn(string name, IEnumerable<decimal> values) : base(name, values) { }
+
+        public DecimalDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public DecimalDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal DecimalDataFrameColumn(string name, PrimitiveColumnContainer<decimal> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/DoubleConverter.cs b/src/Microsoft.Data.Analysis/DoubleConverter.cs
deleted file mode 100644
index 399c3ffb25..0000000000
--- a/src/Microsoft.Data.Analysis/DoubleConverter.cs
+++ /dev/null
@@ -1,162 +0,0 @@
-using System;
-using System.Collections.Generic;
-using System.Text;
-
-namespace Microsoft.Data.Analysis
-{
-    internal interface IDoubleConverter<T>
-    {
-        double GetDouble(T value);
-    }
-    internal static class DoubleConverter<T>
-    {
-        public static IDoubleConverter<T> Instance { get; } = DoubleConverter.GetDoubleConverter<T>();
-    }
-    internal static class DoubleConverter
-    {
-        public static IDoubleConverter<T> GetDoubleConverter<T>()
-        {
-            if (typeof(T) == typeof(bool))
-            {
-                throw new NotImplementedException();
-            }
-            else if (typeof(T) == typeof(byte))
-            {
-                return (IDoubleConverter<T>)new ByteDoubleConverter();
-            }
-            else if (typeof(T) == typeof(char))
-            {
-                return (IDoubleConverter<T>)new CharDoubleConverter();
-            }
-            else if (typeof(T) == typeof(decimal))
-            {
-                throw new NotImplementedException();
-            }
-            else if (typeof(T) == typeof(double))
-            {
-                return (IDoubleConverter<T>)new DoubleDoubleConverter();
-            }
-            else if (typeof(T) == typeof(float))
-            {
-                return (IDoubleConverter<T>)new FloatDoubleConverter();
-            }
-            else if (typeof(T) == typeof(int))
-            {
-                return (IDoubleConverter<T>)new IntDoubleConverter();
-            }
-            else if (typeof(T) == typeof(long))
-            {
-                return (IDoubleConverter<T>)new LongDoubleConverter();
-            }
-            else if (typeof(T) == typeof(sbyte))
-            {
-                return (IDoubleConverter<T>)new SByteDoubleConverter();
-            }
-            else if (typeof(T) == typeof(short))
-            {
-                return (IDoubleConverter<T>)new ShortDoubleConverter();
-            }
-            else if (typeof(T) == typeof(uint))
-            {
-                return (IDoubleConverter<T>)new UIntDoubleConverter();
-            }
-            else if (typeof(T) == typeof(ulong))
-            {
-                return (IDoubleConverter<T>)new ULongDoubleConverter();
-            }
-            else if (typeof(T) == typeof(ushort))
-            {
-                return (IDoubleConverter<T>)new UShortDoubleConverter();
-            }
-            throw new NotSupportedException();
-        }
-    }
-
-    internal class ByteDoubleConverter : IDoubleConverter<byte>
-    {
-        public double GetDouble(byte value)
-        {
-            return value;
-        }
-    }
-
-    internal class CharDoubleConverter : IDoubleConverter<char>
-    {
-        public double GetDouble(char value)
-        {
-            return value;
-        }
-    }
-
-    internal class DoubleDoubleConverter : IDoubleConverter<double>
-    {
-        public double GetDouble(double value)
-        {
-            return value;
-        }
-    }
-
-    internal class FloatDoubleConverter : IDoubleConverter<float>
-    {
-        public double GetDouble(float value)
-        {
-            return value;
-        }
-    }
-
-    internal class IntDoubleConverter : IDoubleConverter<int>
-    {
-        public double GetDouble(int value)
-        {
-            return value;
-        }
-    }
-
-    internal class LongDoubleConverter : IDoubleConverter<long>
-    {
-        public double GetDouble(long value)
-        {
-            return value;
-        }
-    }
-
-    internal class SByteDoubleConverter : IDoubleConverter<sbyte>
-    {
-        public double GetDouble(sbyte value)
-        {
-            return value;
-        }
-    }
-
-    internal class ShortDoubleConverter : IDoubleConverter<short>
-    {
-        public double GetDouble(short value)
-        {
-            return value;
-        }
-    }
-
-    internal class UIntDoubleConverter : IDoubleConverter<uint>
-    {
-        public double GetDouble(uint value)
-        {
-            return value;
-        }
-    }
-
-    internal class ULongDoubleConverter : IDoubleConverter<ulong>
-    {
-        public double GetDouble(ulong value)
-        {
-            return value;
-        }
-    }
-
-    internal class UShortDoubleConverter : IDoubleConverter<ushort>
-    {
-        public double GetDouble(ushort value)
-        {
-            return value;
-        }
-    }
-}
diff --git a/src/Microsoft.Data.Analysis/DoubleDataFrameColumn.cs b/src/Microsoft.Data.Analysis/DoubleDataFrameColumn.cs
new file mode 100644
index 0000000000..6a97300376
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/DoubleDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class DoubleDataFrameColumn : PrimitiveDataFrameColumn<double>
+    {
+        public DoubleDataFrameColumn(string name, IEnumerable<double?> values) : base(name, values) { }
+
+        public DoubleDataFrameColumn(string name, IEnumerable<double> values) : base(name, values) { }
+
+        public DoubleDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public DoubleDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal DoubleDataFrameColumn(string name, PrimitiveColumnContainer<double> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/Int16DataFrameColumn.cs b/src/Microsoft.Data.Analysis/Int16DataFrameColumn.cs
new file mode 100644
index 0000000000..79893b1899
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/Int16DataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class Int16DataFrameColumn : PrimitiveDataFrameColumn<short>
+    {
+        public Int16DataFrameColumn(string name, IEnumerable<short?> values) : base(name, values) { }
+
+        public Int16DataFrameColumn(string name, IEnumerable<short> values) : base(name, values) { }
+
+        public Int16DataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public Int16DataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal Int16DataFrameColumn(string name, PrimitiveColumnContainer<short> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/Int32DataFrameColumn.cs b/src/Microsoft.Data.Analysis/Int32DataFrameColumn.cs
new file mode 100644
index 0000000000..91b15764cd
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/Int32DataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class Int32DataFrameColumn : PrimitiveDataFrameColumn<int>
+    {
+        public Int32DataFrameColumn(string name, IEnumerable<int?> values) : base(name, values) { }
+
+        public Int32DataFrameColumn(string name, IEnumerable<int> values) : base(name, values) { }
+
+        public Int32DataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public Int32DataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal Int32DataFrameColumn(string name, PrimitiveColumnContainer<int> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/Int64DataFrameColumn.cs b/src/Microsoft.Data.Analysis/Int64DataFrameColumn.cs
new file mode 100644
index 0000000000..f955404152
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/Int64DataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class Int64DataFrameColumn : PrimitiveDataFrameColumn<long>
+    {
+        public Int64DataFrameColumn(string name, IEnumerable<long?> values) : base(name, values) { }
+
+        public Int64DataFrameColumn(string name, IEnumerable<long> values) : base(name, values) { }
+
+        public Int64DataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public Int64DataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal Int64DataFrameColumn(string name, PrimitiveColumnContainer<long> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/Microsoft.Data.Analysis.csproj b/src/Microsoft.Data.Analysis/Microsoft.Data.Analysis.csproj
index 71490c0883..7071a50700 100644
--- a/src/Microsoft.Data.Analysis/Microsoft.Data.Analysis.csproj
+++ b/src/Microsoft.Data.Analysis/Microsoft.Data.Analysis.csproj
@@ -11,6 +11,14 @@
     <PackageReleaseNotes>Initial preview of robust and extensible types and algorithms for manipulating structured data that supports aggregations, statistical funtions, sorting, grouping, joins, merges, handling missing values and more. </PackageReleaseNotes>
     <PackageTags>ML.NET ML Machine Learning Data Science DataFrame Preparation DataView Analytics Exploration</PackageTags>
   </PropertyGroup>
+  
+  <ItemGroup>
+    <None Include="Converters.cs">
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+      <DependentUpon>Converters.tt</DependentUpon>
+    </None>
+  </ItemGroup>
 
   <ItemGroup>
     <PackageReference Include="Apache.Arrow" Version="0.14.1" />
@@ -22,6 +30,14 @@
   </ItemGroup>
   
   <ItemGroup>
+    <None Update="Converters.tt">
+      <Generator>TextTemplatingFileGenerator</Generator>
+      <LastGenOutput>Converters.cs</LastGenOutput>
+    </None>
+    <None Update="DataFrameColumn.BinaryOperations.ExplodedColumns.tt">
+      <Generator>TextTemplatingFileGenerator</Generator>
+      <LastGenOutput>DataFrameColumn.BinaryOperations.ExplodedColumns.cs</LastGenOutput>
+    </None>
     <None Update="DataFrameColumn.BinaryOperations.tt">
       <Generator>TextTemplatingFileGenerator</Generator>
       <LastGenOutput>DataFrameColumn.BinaryOperations.cs</LastGenOutput>
@@ -50,10 +66,23 @@
       <Generator>TextTemplatingFileGenerator</Generator>
       <LastGenOutput>ColumnArithmeticTemplate.cs</LastGenOutput>
     </None>
+    <None Update="PrimitiveDataFrameColumn.BinaryOperations.Combinations.tt">
+      <Generator>TextTemplatingFileGenerator</Generator>
+      <LastGenOutput>PrimitiveDataFrameColumn.BinaryOperations.Combinations.ttinclude</LastGenOutput>
+    </None>
+    <None Update="PrimitiveDataFrameColumn.BinaryOperations.Combinations.ttinclude">
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+      <DependentUpon>PrimitiveDataFrameColumn.BinaryOperations.Combinations.tt</DependentUpon>
+    </None>
     <None Update="PrimitiveDataFrameColumn.BinaryOperations.tt">
       <Generator>TextTemplatingFileGenerator</Generator>
       <LastGenOutput>PrimitiveDataFrameColumn.BinaryOperations.cs</LastGenOutput>
     </None>
+    <None Update="PrimitiveDataFrameColumn.BinaryOperators.tt">
+      <Generator>TextTemplatingFileGenerator</Generator>
+      <LastGenOutput>PrimitiveDataFrameColumn.BinaryOperators.cs</LastGenOutput>
+    </None>
     <None Update="PrimitiveDataFrameColumn.Computations.tt">
       <Generator>TextTemplatingFileGenerator</Generator>
       <LastGenOutput>PrimitiveDataFrameColumn.Computations.cs</LastGenOutput>
@@ -89,6 +118,16 @@
   </ItemGroup>
 
   <ItemGroup>
+    <Compile Update="ColumnArithmeticTemplate.cs">
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+      <DependentUpon>ColumnArithmeticTemplate.ttinclude</DependentUpon>
+    </Compile>
+    <Compile Update="Converters.cs">
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+      <DependentUpon>Converters.tt</DependentUpon>
+    </Compile>
     <Compile Update="DataFrameColumn.BinaryOperations.cs">
       <DesignTime>True</DesignTime>
       <AutoGen>True</AutoGen>
@@ -124,6 +163,11 @@
       <AutoGen>True</AutoGen>
       <DependentUpon>PrimitiveDataFrameColumn.BinaryOperations.tt</DependentUpon>
     </Compile>
+    <Compile Update="PrimitiveDataFrameColumn.BinaryOperators.cs">
+      <DesignTime>True</DesignTime>
+      <AutoGen>True</AutoGen>
+      <DependentUpon>PrimitiveDataFrameColumn.BinaryOperators.tt</DependentUpon>
+    </Compile>
     <Compile Update="PrimitiveDataFrameColumn.Computations.cs">
       <DesignTime>True</DesignTime>
       <AutoGen>True</AutoGen>
diff --git a/src/Microsoft.Data.Analysis/PrimitiveColumnContainer.cs b/src/Microsoft.Data.Analysis/PrimitiveColumnContainer.cs
index 767200a818..45521c137e 100644
--- a/src/Microsoft.Data.Analysis/PrimitiveColumnContainer.cs
+++ b/src/Microsoft.Data.Analysis/PrimitiveColumnContainer.cs
@@ -617,6 +617,46 @@ internal PrimitiveColumnContainer<bool> CloneAsBoolContainer()
             return ret;
         }
 
+        internal PrimitiveColumnContainer<byte> CloneAsByteContainer()
+        {
+            var ret = new PrimitiveColumnContainer<byte>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<byte> newBuffer = new DataFrameBuffer<byte>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(ByteConverter<T>.Instance.GetByte(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<sbyte> CloneAsSByteContainer()
+        {
+            var ret = new PrimitiveColumnContainer<sbyte>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<sbyte> newBuffer = new DataFrameBuffer<sbyte>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(SByteConverter<T>.Instance.GetSByte(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
         internal PrimitiveColumnContainer<double> CloneAsDoubleContainer()
         {
             var ret = new PrimitiveColumnContainer<double>();
@@ -656,5 +696,145 @@ internal PrimitiveColumnContainer<decimal> CloneAsDecimalContainer()
             ret.NullCount = NullCount;
             return ret;
         }
+
+        internal PrimitiveColumnContainer<short> CloneAsShortContainer()
+        {
+            var ret = new PrimitiveColumnContainer<short>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<short> newBuffer = new DataFrameBuffer<short>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(ShortConverter<T>.Instance.GetShort(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<ushort> CloneAsUShortContainer()
+        {
+            var ret = new PrimitiveColumnContainer<ushort>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<ushort> newBuffer = new DataFrameBuffer<ushort>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(UShortConverter<T>.Instance.GetUShort(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<int> CloneAsIntContainer()
+        {
+            var ret = new PrimitiveColumnContainer<int>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<int> newBuffer = new DataFrameBuffer<int>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(IntConverter<T>.Instance.GetInt(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<uint> CloneAsUIntContainer()
+        {
+            var ret = new PrimitiveColumnContainer<uint>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<uint> newBuffer = new DataFrameBuffer<uint>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(UIntConverter<T>.Instance.GetUInt(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<long> CloneAsLongContainer()
+        {
+            var ret = new PrimitiveColumnContainer<long>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<long> newBuffer = new DataFrameBuffer<long>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(LongConverter<T>.Instance.GetLong(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<ulong> CloneAsULongContainer()
+        {
+            var ret = new PrimitiveColumnContainer<ulong>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<ulong> newBuffer = new DataFrameBuffer<ulong>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(ULongConverter<T>.Instance.GetULong(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
+
+        internal PrimitiveColumnContainer<float> CloneAsFloatContainer()
+        {
+            var ret = new PrimitiveColumnContainer<float>();
+            foreach (ReadOnlyDataFrameBuffer<T> buffer in Buffers)
+            {
+                ret.Length += buffer.Length;
+                DataFrameBuffer<float> newBuffer = new DataFrameBuffer<float>();
+                ret.Buffers.Add(newBuffer);
+                newBuffer.EnsureCapacity(buffer.Length);
+                ReadOnlySpan<T> span = buffer.ReadOnlySpan;
+                for (int i = 0; i < span.Length; i++)
+                {
+                    newBuffer.Append(FloatConverter<T>.Instance.GetFloat(span[i]));
+                }
+            }
+            ret.NullBitMapBuffers = CloneNullBitMapBuffers();
+            ret.NullCount = NullCount;
+            return ret;
+        }
     }
 }
diff --git a/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs b/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs
index cf6e6d2b8f..18a9903342 100644
--- a/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs
+++ b/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs
@@ -429,22 +429,77 @@ public PrimitiveDataFrameColumn<T> Clone(IEnumerable<long> mapIndices)
             return ret;
         }
 
-        internal PrimitiveDataFrameColumn<bool> CloneAsBoolColumn()
+        internal BooleanDataFrameColumn CloneAsBoolColumn()
         {
             PrimitiveColumnContainer<bool> newColumnContainer = _columnContainer.CloneAsBoolContainer();
-            return new PrimitiveDataFrameColumn<bool>(Name, newColumnContainer);
+            return new BooleanDataFrameColumn(Name, newColumnContainer);
         }
 
-        internal PrimitiveDataFrameColumn<double> CloneAsDoubleColumn()
+        internal ByteDataFrameColumn CloneAsByteColumn()
+        {
+            PrimitiveColumnContainer<byte> newColumnContainer = _columnContainer.CloneAsByteContainer();
+            return new ByteDataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal SByteDataFrameColumn CloneAsSByteColumn()
+        {
+            PrimitiveColumnContainer<sbyte> newColumnContainer = _columnContainer.CloneAsSByteContainer();
+            return new SByteDataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal DoubleDataFrameColumn CloneAsDoubleColumn()
         {
             PrimitiveColumnContainer<double> newColumnContainer = _columnContainer.CloneAsDoubleContainer();
-            return new PrimitiveDataFrameColumn<double>(Name, newColumnContainer);
+            return new DoubleDataFrameColumn(Name, newColumnContainer);
         }
 
-        internal PrimitiveDataFrameColumn<decimal> CloneAsDecimalColumn()
+        internal DecimalDataFrameColumn CloneAsDecimalColumn()
         {
             PrimitiveColumnContainer<decimal> newColumnContainer = _columnContainer.CloneAsDecimalContainer();
-            return new PrimitiveDataFrameColumn<decimal>(Name, newColumnContainer);
+            return new DecimalDataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal Int16DataFrameColumn CloneAsShortColumn()
+        {
+            PrimitiveColumnContainer<short> newColumnContainer = _columnContainer.CloneAsShortContainer();
+            return new Int16DataFrameColumn(Name, newColumnContainer);
+        }
+
+
+        internal UInt16DataFrameColumn CloneAsUShortColumn()
+        {
+            PrimitiveColumnContainer<ushort> newColumnContainer = _columnContainer.CloneAsUShortContainer();
+            return new UInt16DataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal Int32DataFrameColumn CloneAsIntColumn()
+        {
+            PrimitiveColumnContainer<int> newColumnContainer = _columnContainer.CloneAsIntContainer();
+            return new Int32DataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal UInt32DataFrameColumn CloneAsUIntColumn()
+        {
+            PrimitiveColumnContainer<uint> newColumnContainer = _columnContainer.CloneAsUIntContainer();
+            return new UInt32DataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal Int64DataFrameColumn CloneAsLongColumn()
+        {
+            PrimitiveColumnContainer<long> newColumnContainer = _columnContainer.CloneAsLongContainer();
+            return new Int64DataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal UInt64DataFrameColumn CloneAsULongColumn()
+        {
+            PrimitiveColumnContainer<ulong> newColumnContainer = _columnContainer.CloneAsULongContainer();
+            return new UInt64DataFrameColumn(Name, newColumnContainer);
+        }
+
+        internal SingleDataFrameColumn CloneAsFloatColumn()
+        {
+            PrimitiveColumnContainer<float> newColumnContainer = _columnContainer.CloneAsFloatContainer();
+            return new SingleDataFrameColumn(Name, newColumnContainer);
         }
 
         public override GroupBy GroupBy(int columnIndex, DataFrame parent)
diff --git a/src/Microsoft.Data.Analysis/SByteDataFrameColumn.cs b/src/Microsoft.Data.Analysis/SByteDataFrameColumn.cs
new file mode 100644
index 0000000000..731614da72
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/SByteDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class SByteDataFrameColumn : PrimitiveDataFrameColumn<sbyte>
+    {
+        public SByteDataFrameColumn(string name, IEnumerable<sbyte?> values) : base(name, values) { }
+
+        public SByteDataFrameColumn(string name, IEnumerable<sbyte> values) : base(name, values) { }
+
+        public SByteDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public SByteDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal SByteDataFrameColumn(string name, PrimitiveColumnContainer<sbyte> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/SingleDataFrameColumn.cs b/src/Microsoft.Data.Analysis/SingleDataFrameColumn.cs
new file mode 100644
index 0000000000..e6be6255d0
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/SingleDataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class SingleDataFrameColumn : PrimitiveDataFrameColumn<float>
+    {
+        public SingleDataFrameColumn(string name, IEnumerable<float?> values) : base(name, values) { }
+
+        public SingleDataFrameColumn(string name, IEnumerable<float> values) : base(name, values) { }
+
+        public SingleDataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public SingleDataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal SingleDataFrameColumn(string name, PrimitiveColumnContainer<float> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/UInt16DataFrameColumn.cs b/src/Microsoft.Data.Analysis/UInt16DataFrameColumn.cs
new file mode 100644
index 0000000000..e56cdca7f3
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/UInt16DataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class UInt16DataFrameColumn : PrimitiveDataFrameColumn<ushort>
+    {
+        public UInt16DataFrameColumn(string name, IEnumerable<ushort?> values) : base(name, values) { }
+
+        public UInt16DataFrameColumn(string name, IEnumerable<ushort> values) : base(name, values) { }
+
+        public UInt16DataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public UInt16DataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal UInt16DataFrameColumn(string name, PrimitiveColumnContainer<ushort> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/UInt32DataFrameColumn.cs b/src/Microsoft.Data.Analysis/UInt32DataFrameColumn.cs
new file mode 100644
index 0000000000..af114ff41e
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/UInt32DataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class UInt32DataFrameColumn : PrimitiveDataFrameColumn<uint>
+    {
+        public UInt32DataFrameColumn(string name, IEnumerable<uint?> values) : base(name, values) { }
+
+        public UInt32DataFrameColumn(string name, IEnumerable<uint> values) : base(name, values) { }
+
+        public UInt32DataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public UInt32DataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal UInt32DataFrameColumn(string name, PrimitiveColumnContainer<uint> values) : base(name, values) { }
+    }
+}
diff --git a/src/Microsoft.Data.Analysis/UInt64DataFrameColumn.cs b/src/Microsoft.Data.Analysis/UInt64DataFrameColumn.cs
new file mode 100644
index 0000000000..cb74704319
--- /dev/null
+++ b/src/Microsoft.Data.Analysis/UInt64DataFrameColumn.cs
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// 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.Generic;
+using System.Text;
+
+namespace Microsoft.Data.Analysis
+{
+    public partial class UInt64DataFrameColumn : PrimitiveDataFrameColumn<ulong>
+    {
+        public UInt64DataFrameColumn(string name, IEnumerable<ulong?> values) : base(name, values) { }
+
+        public UInt64DataFrameColumn(string name, IEnumerable<ulong> values) : base(name, values) { }
+
+        public UInt64DataFrameColumn(string name, long length = 0) : base(name, length) { }
+
+        public UInt64DataFrameColumn(string name, ReadOnlyMemory<byte> buffer, ReadOnlyMemory<byte> nullBitMap, int length = 0, int nullCount = 0) : base(name, buffer, nullBitMap, length, nullCount) { }
+
+        internal UInt64DataFrameColumn(string name, PrimitiveColumnContainer<ulong> values) : base(name, values) { }
+    }
+}
diff --git a/tests/Microsoft.Data.Analysis.Tests/DataFrameTests.cs b/tests/Microsoft.Data.Analysis.Tests/DataFrameTests.cs
index bdf383425e..f715dfa17d 100644
--- a/tests/Microsoft.Data.Analysis.Tests/DataFrameTests.cs
+++ b/tests/Microsoft.Data.Analysis.Tests/DataFrameTests.cs
@@ -16,8 +16,8 @@ public partial class DataFrameTests
     {
         public static DataFrame MakeDataFrameWithTwoColumns(int length, bool withNulls = true)
         {
-            DataFrameColumn dataFrameColumn1 = new PrimitiveDataFrameColumn<int>("Int1", Enumerable.Range(0, length).Select(x => x));
-            DataFrameColumn dataFrameColumn2 = new PrimitiveDataFrameColumn<int>("Int2", Enumerable.Range(10, length).Select(x => x));
+            DataFrameColumn dataFrameColumn1 = new Int32DataFrameColumn("Int1", Enumerable.Range(0, length).Select(x => x));
+            DataFrameColumn dataFrameColumn2 = new Int32DataFrameColumn("Int2", Enumerable.Range(10, length).Select(x => x));
             if (withNulls)
             {
                 dataFrameColumn1[length / 2] = null;
@@ -83,7 +83,7 @@ public static DataFrame MakeDataFrameWithAllColumnTypes(int length, bool withNul
         public static DataFrame MakeDataFrameWithAllMutableColumnTypes(int length, bool withNulls = true)
         {
             DataFrame df = MakeDataFrameWithNumericAndStringColumns(length, withNulls);
-            DataFrameColumn boolColumn = new PrimitiveDataFrameColumn<bool>("Bool", Enumerable.Range(0, length).Select(x => x % 2 == 0));
+            DataFrameColumn boolColumn = new BooleanDataFrameColumn("Bool", Enumerable.Range(0, length).Select(x => x % 2 == 0));
             df.Columns.Insert(df.Columns.Count, boolColumn);
             if (withNulls)
             {
@@ -95,7 +95,7 @@ public static DataFrame MakeDataFrameWithAllMutableColumnTypes(int length, bool
         public static DataFrame MakeDataFrameWithNumericAndBoolColumns(int length)
         {
             DataFrame df = MakeDataFrameWithNumericColumns(length);
-            DataFrameColumn boolColumn = new PrimitiveDataFrameColumn<bool>("Bool", Enumerable.Range(0, length).Select(x => x % 2 == 0));
+            DataFrameColumn boolColumn = new BooleanDataFrameColumn("Bool", Enumerable.Range(0, length).Select(x => x % 2 == 0));
             df.Columns.Insert(df.Columns.Count, boolColumn);
             boolColumn[length / 2] = null;
             return df;
@@ -104,14 +104,14 @@ public static DataFrame MakeDataFrameWithNumericAndBoolColumns(int length)
         public static DataFrame MakeDataFrameWithNumericAndStringColumns(int length, bool withNulls = true)
         {
             DataFrame df = MakeDataFrameWithNumericColumns(length, withNulls);
-            DataFrameColumn stringColumn = DataFrameColumn.Create("String", Enumerable.Range(0, length).Select(x => x.ToString()));
+            DataFrameColumn stringColumn = new StringDataFrameColumn("String", Enumerable.Range(0, length).Select(x => x.ToString()));
             df.Columns.Insert(df.Columns.Count, stringColumn);
             if (withNulls)
             {
                 stringColumn[length / 2] = null;
             }
 
-            DataFrameColumn charColumn = DataFrameColumn.Create("Char", Enumerable.Range(0, length).Select(x => (char)(x + 65)));
+            DataFrameColumn charColumn = new CharDataFrameColumn("Char", Enumerable.Range(0, length).Select(x => (char)(x + 65)));
             df.Columns.Insert(df.Columns.Count, charColumn);
             if (withNulls)
             {
@@ -122,17 +122,17 @@ public static DataFrame MakeDataFrameWithNumericAndStringColumns(int length, boo
 
         public static DataFrame MakeDataFrameWithNumericColumns(int length, bool withNulls = true)
         {
-            DataFrameColumn byteColumn = DataFrameColumn.Create("Byte", Enumerable.Range(0, length).Select(x => (byte)x));
-            DataFrameColumn decimalColumn = DataFrameColumn.Create("Decimal", Enumerable.Range(0, length).Select(x => (decimal)x));
-            DataFrameColumn doubleColumn = DataFrameColumn.Create("Double", Enumerable.Range(0, length).Select(x => (double)x));
-            DataFrameColumn floatColumn = DataFrameColumn.Create("Float", Enumerable.Range(0, length).Select(x => (float)x));
-            DataFrameColumn intColumn = DataFrameColumn.Create("Int", Enumerable.Range(0, length).Select(x => x));
-            DataFrameColumn longColumn = DataFrameColumn.Create("Long", Enumerable.Range(0, length).Select(x => (long)x));
-            DataFrameColumn sbyteColumn = DataFrameColumn.Create("Sbyte", Enumerable.Range(0, length).Select(x => (sbyte)x));
-            DataFrameColumn shortColumn = DataFrameColumn.Create("Short", Enumerable.Range(0, length).Select(x => (short)x));
-            DataFrameColumn uintColumn = DataFrameColumn.Create("Uint", Enumerable.Range(0, length).Select(x => (uint)x));
-            DataFrameColumn ulongColumn = DataFrameColumn.Create("Ulong", Enumerable.Range(0, length).Select(x => (ulong)x));
-            DataFrameColumn ushortColumn = DataFrameColumn.Create("Ushort", Enumerable.Range(0, length).Select(x => (ushort)x));
+            DataFrameColumn byteColumn = new ByteDataFrameColumn("Byte", Enumerable.Range(0, length).Select(x => (byte)x));
+            DataFrameColumn decimalColumn = new DecimalDataFrameColumn("Decimal", Enumerable.Range(0, length).Select(x => (decimal)x));
+            DataFrameColumn doubleColumn = new DoubleDataFrameColumn("Double", Enumerable.Range(0, length).Select(x => (double)x));
+            DataFrameColumn floatColumn = new SingleDataFrameColumn("Float", Enumerable.Range(0, length).Select(x => (float)x));
+            DataFrameColumn intColumn = new Int32DataFrameColumn("Int", Enumerable.Range(0, length).Select(x => x));
+            DataFrameColumn longColumn = new Int64DataFrameColumn("Long", Enumerable.Range(0, length).Select(x => (long)x));
+            DataFrameColumn sbyteColumn = new SByteDataFrameColumn("Sbyte", Enumerable.Range(0, length).Select(x => (sbyte)x));
+            DataFrameColumn shortColumn = new Int16DataFrameColumn("Short", Enumerable.Range(0, length).Select(x => (short)x));
+            DataFrameColumn uintColumn = new UInt32DataFrameColumn("Uint", Enumerable.Range(0, length).Select(x => (uint)x));
+            DataFrameColumn ulongColumn = new UInt64DataFrameColumn("Ulong", Enumerable.Range(0, length).Select(x => (ulong)x));
+            DataFrameColumn ushortColumn = new UInt16DataFrameColumn("Ushort", Enumerable.Range(0, length).Select(x => (ushort)x));
 
             DataFrame dataFrame = new DataFrame(new List<DataFrameColumn> { byteColumn, decimalColumn, doubleColumn, floatColumn, intColumn, longColumn, sbyteColumn, shortColumn, uintColumn, ulongColumn, ushortColumn });
 
@@ -197,7 +197,7 @@ public void TestIndexer()
             var row = dataFrame.Rows[4];
             Assert.Equal(14, (int)row[1]);
 
-            var column = dataFrame["Int2"] as PrimitiveDataFrameColumn<int>;
+            var column = dataFrame["Int2"] as Int32DataFrameColumn;
             Assert.Equal(1000, (int)column[2]);
 
             Assert.Throws<ArgumentException>(() => dataFrame["Int5"]);
@@ -206,8 +206,8 @@ public void TestIndexer()
         [Fact]
         public void ColumnAndTableCreationTest()
         {
-            DataFrameColumn intColumn = new PrimitiveDataFrameColumn<int>("IntColumn", Enumerable.Range(0, 10).Select(x => x));
-            DataFrameColumn floatColumn = new PrimitiveDataFrameColumn<float>("FloatColumn", Enumerable.Range(0, 10).Select(x => (float)x));
+            DataFrameColumn intColumn = new Int32DataFrameColumn("IntColumn", Enumerable.Range(0, 10).Select(x => x));
+            DataFrameColumn floatColumn = new SingleDataFrameColumn("FloatColumn", Enumerable.Range(0, 10).Select(x => (float)x));
             DataFrame dataFrame = new DataFrame();
             dataFrame.Columns.Insert(0, intColumn);
             dataFrame.Columns.Insert(1, floatColumn);
@@ -218,17 +218,17 @@ public void ColumnAndTableCreationTest()
             Assert.Equal(10, dataFrame.Columns[1].Length);
             Assert.Equal("FloatColumn", dataFrame.Columns[1].Name);
 
-            DataFrameColumn bigColumn = new PrimitiveDataFrameColumn<float>("BigColumn", Enumerable.Range(0, 11).Select(x => (float)x));
-            DataFrameColumn repeatedName = new PrimitiveDataFrameColumn<float>("FloatColumn", Enumerable.Range(0, 10).Select(x => (float)x));
+            DataFrameColumn bigColumn = new SingleDataFrameColumn("BigColumn", Enumerable.Range(0, 11).Select(x => (float)x));
+            DataFrameColumn repeatedName = new SingleDataFrameColumn("FloatColumn", Enumerable.Range(0, 10).Select(x => (float)x));
             Assert.Throws<ArgumentException>(() => dataFrame.Columns.Insert(2, bigColumn));
             Assert.Throws<ArgumentException>(() => dataFrame.Columns.Insert(2, repeatedName));
             Assert.Throws<ArgumentOutOfRangeException>(() => dataFrame.Columns.Insert(10, repeatedName));
 
             Assert.Equal(2, dataFrame.Columns.Count);
-            DataFrameColumn intColumnCopy = new PrimitiveDataFrameColumn<int>("IntColumn", Enumerable.Range(0, 10).Select(x => x));
+            DataFrameColumn intColumnCopy = new Int32DataFrameColumn("IntColumn", Enumerable.Range(0, 10).Select(x => x));
             Assert.Throws<ArgumentException>(() => dataFrame.Columns[1] = intColumnCopy);
 
-            DataFrameColumn differentIntColumn = new PrimitiveDataFrameColumn<int>("IntColumn1", Enumerable.Range(0, 10).Select(x => x));
+            DataFrameColumn differentIntColumn = new Int32DataFrameColumn("IntColumn1", Enumerable.Range(0, 10).Select(x => x));
             dataFrame.Columns[1] = differentIntColumn;
             Assert.True(object.ReferenceEquals(differentIntColumn, dataFrame.Columns[1]));
 
@@ -247,7 +247,7 @@ public void ColumnAndTableCreationTest()
         public void InsertAndRemoveColumnTests()
         {
             DataFrame dataFrame = MakeDataFrameWithAllMutableColumnTypes(10);
-            DataFrameColumn intColumn = new PrimitiveDataFrameColumn<int>("IntColumn", Enumerable.Range(0, 10).Select(x => x));
+            DataFrameColumn intColumn = new Int32DataFrameColumn("IntColumn", Enumerable.Range(0, 10).Select(x => x));
             DataFrameColumn charColumn = dataFrame["Char"];
             int insertedIndex = dataFrame.Columns.Count;
             dataFrame.Columns.Insert(dataFrame.Columns.Count, intColumn);
@@ -390,7 +390,7 @@ public void TestBinaryOperationsWithConversions()
             // int + bool should throw
             Assert.Throws<NotSupportedException>(() => df.Add(true));
 
-            var dataFrameColumn1 = new PrimitiveDataFrameColumn<double>("Double1", Enumerable.Range(0, 10).Select(x => (double)x));
+            var dataFrameColumn1 = new DoubleDataFrameColumn("Double1", Enumerable.Range(0, 10).Select(x => (double)x));
             df.Columns[0] = dataFrameColumn1;
             // Double + comparison ops should throw
             Assert.Throws<NotSupportedException>(() => df.And(true));
@@ -400,8 +400,8 @@ public void TestBinaryOperationsWithConversions()
         public void TestBinaryOperationsOnBoolColumn()
         {
             var df = new DataFrame();
-            var dataFrameColumn1 = new PrimitiveDataFrameColumn<bool>("Bool1", Enumerable.Range(0, 10).Select(x => true));
-            var dataFrameColumn2 = new PrimitiveDataFrameColumn<bool>("Bool2", Enumerable.Range(0, 10).Select(x => true));
+            var dataFrameColumn1 = new BooleanDataFrameColumn("Bool1", Enumerable.Range(0, 10).Select(x => true));
+            var dataFrameColumn2 = new BooleanDataFrameColumn("Bool2", Enumerable.Range(0, 10).Select(x => true));
             df.Columns.Insert(0, dataFrameColumn1);
             df.Columns.Insert(1, dataFrameColumn2);
 
@@ -623,8 +623,8 @@ public void TestBinaryOperatorsWithConversions()
         public void TestBinaryOperatorsOnBoolColumns()
         {
             var df = new DataFrame();
-            var dataFrameColumn1 = new PrimitiveDataFrameColumn<bool>("Bool1", Enumerable.Range(0, 10).Select(x => x % 2 == 0 ? true : false));
-            var dataFrameColumn2 = new PrimitiveDataFrameColumn<bool>("Bool2", Enumerable.Range(0, 10).Select(x => x % 2 == 0 ? true : false));
+            var dataFrameColumn1 = new BooleanDataFrameColumn("Bool1", Enumerable.Range(0, 10).Select(x => x % 2 == 0 ? true : false));
+            var dataFrameColumn2 = new BooleanDataFrameColumn("Bool2", Enumerable.Range(0, 10).Select(x => x % 2 == 0 ? true : false));
             df.Columns.Insert(0, dataFrameColumn1);
             df.Columns.Insert(1, dataFrameColumn2);
 
@@ -668,7 +668,7 @@ public void TestBinaryOperatorsOnBoolColumns()
         [Fact]
         public void TestBinaryOperationsOnColumns()
         {
-            PrimitiveDataFrameColumn<int> column = new PrimitiveDataFrameColumn<int>("Int", Enumerable.Range(0, 10));
+            Int32DataFrameColumn column = new Int32DataFrameColumn("Int", Enumerable.Range(0, 10));
             Assert.ThrowsAny<ArgumentException>(() => column.Add(5.5, inPlace: true));
             Assert.ThrowsAny<ArgumentException>(() => column.ReverseAdd(5.5, inPlace: true));
             string str = "A String";
@@ -679,7 +679,7 @@ public void TestBinaryOperationsOnColumns()
         [Fact]
         public void TestColumnReverseOrderState()
         {
-            var column = new PrimitiveDataFrameColumn<int>("Int", Enumerable.Range(0, 10));
+            var column = new Int32DataFrameColumn("Int", Enumerable.Range(0, 10));
             var newColumn = 1 - column;
             var checkOrderColumn = 1 - newColumn;
             Assert.True(checkOrderColumn.ElementwiseEquals(column).All());
@@ -935,13 +935,13 @@ public void TestStringColumnSort()
         public void TestPrimitiveColumnSort(int numberOfNulls)
         {
             // Primitive Column Sort
-            PrimitiveDataFrameColumn<int> intColumn = new PrimitiveDataFrameColumn<int>("Int", 0);
+            Int32DataFrameColumn intColumn = new Int32DataFrameColumn("Int", 0);
             Assert.Equal(0, intColumn.NullCount);
             intColumn.AppendMany(null, numberOfNulls);
             Assert.Equal(numberOfNulls, intColumn.NullCount);
 
             // Should handle all nulls
-            PrimitiveDataFrameColumn<int> sortedIntColumn = intColumn.Sort() as PrimitiveDataFrameColumn<int>;
+            PrimitiveDataFrameColumn<int> sortedIntColumn = intColumn.Sort();
             Assert.Equal(numberOfNulls, sortedIntColumn.NullCount);
             Assert.Null(sortedIntColumn[0]);
 
@@ -952,19 +952,19 @@ public void TestPrimitiveColumnSort(int numberOfNulls)
             Assert.Equal(numberOfNulls, intColumn.NullCount);
 
             // Ascending sort
-            sortedIntColumn = intColumn.Sort() as PrimitiveDataFrameColumn<int>;
+            sortedIntColumn = intColumn.Sort();
             Assert.Equal(0, sortedIntColumn[0]);
             Assert.Null(sortedIntColumn[9]);
 
             // Descending sort
-            sortedIntColumn = intColumn.Sort(false) as PrimitiveDataFrameColumn<int>;
+            sortedIntColumn = intColumn.Sort(ascending: false);
             Assert.Equal(4, sortedIntColumn[0]);
             Assert.Null(sortedIntColumn[9]);
         }
 
         private void VerifyJoin(DataFrame join, DataFrame left, DataFrame right, JoinAlgorithm joinAlgorithm)
         {
-            PrimitiveDataFrameColumn<long> mapIndices = new PrimitiveDataFrameColumn<long>("map", join.Rows.Count);
+            Int64DataFrameColumn mapIndices = new Int64DataFrameColumn("map", join.Rows.Count);
             for (long i = 0; i < join.Rows.Count; i++)
             {
                 mapIndices[i] = i;
@@ -1379,7 +1379,7 @@ public void TestIEnumerable()
             }
             Assert.Equal("foofoofoofoofoo<null>foofoofoofoo", actualStrings.ToString());
 
-            PrimitiveDataFrameColumn<float> floatColumn = (PrimitiveDataFrameColumn<float>)df["Float"];
+            SingleDataFrameColumn floatColumn = (SingleDataFrameColumn)df["Float"];
             actualStrings.Clear();
             foreach (float? value in floatColumn)
             {
@@ -1394,7 +1394,7 @@ public void TestIEnumerable()
             }
             Assert.Equal("01234<null>6789", actualStrings.ToString());
 
-            PrimitiveDataFrameColumn<int> intColumn = (PrimitiveDataFrameColumn<int>)df["Int"];
+            Int32DataFrameColumn intColumn = (Int32DataFrameColumn)df["Int"];
             actualStrings.Clear();
             foreach (int? value in intColumn)
             {
@@ -1808,7 +1808,7 @@ public void TestValueCounts()
         public void TestApplyElementwiseNullCount()
         {
             DataFrame df = MakeDataFrameWithTwoColumns(10);
-            PrimitiveDataFrameColumn<int> column = df["Int1"] as PrimitiveDataFrameColumn<int>;
+            Int32DataFrameColumn column = df["Int1"] as Int32DataFrameColumn;
             Assert.Equal(1, column.NullCount);
 
             // Change all existing values to null
@@ -1850,7 +1850,7 @@ public void TestClone(int dfLength, int intDfLength)
         {
             DataFrame df = MakeDataFrameWithAllColumnTypes(dfLength, withNulls: true);
             DataFrame intDf = MakeDataFrameWithTwoColumns(intDfLength, false);
-            PrimitiveDataFrameColumn<int> intColumn = intDf["Int1"] as PrimitiveDataFrameColumn<int>;
+            Int32DataFrameColumn intColumn = intDf["Int1"] as Int32DataFrameColumn;
             DataFrame clone = df[intColumn];
             Assert.Equal(intDfLength, clone.Rows.Count);
             Assert.Equal(df.Columns.Count, clone.Columns.Count);
@@ -1869,7 +1869,7 @@ public void TestClone(int dfLength, int intDfLength)
         public void TestColumnCreationFromExisitingColumn()
         {
             DataFrame df = MakeDataFrameWithAllColumnTypes(10);
-            PrimitiveDataFrameColumn<bool> bigInts = new PrimitiveDataFrameColumn<bool>("BigInts", df["Int"].ElementwiseGreaterThan(5));
+            BooleanDataFrameColumn bigInts = new BooleanDataFrameColumn("BigInts", df["Int"].ElementwiseGreaterThan(5));
             for (int i = 0; i < 10; i++)
             {
                 if (i <= 5)
@@ -2126,7 +2126,7 @@ public void TestAppendEmptyValue()
         public void TestApply()
         {
             int[] values = { 1, 2, 3, 4, 5 };
-            var col = new PrimitiveDataFrameColumn<int>("Ints", values);
+            var col = new Int32DataFrameColumn("Ints", values);
             PrimitiveDataFrameColumn<double> newCol = col.Apply(i => i + 0.5d);
 
             Assert.Equal(values.Length, newCol.Length);
@@ -2137,5 +2137,44 @@ public void TestApply()
                 Assert.Equal(newCol[i], values[i] + 0.5d);
             }
         }
+
+        [Fact]
+        public void TestDataFrameCreate()
+        {
+            int length = 10;
+            void AssertLengthTypeAndValues(DataFrameColumn column, Type type)
+            {
+                Assert.Equal(column.DataType, type);
+                Assert.Equal(length, column.Length);
+                for (long i = 0; i < column.Length; i++)
+                {
+                    Assert.Equal(i.ToString(), column[i].ToString());
+                }
+            }
+            DataFrameColumn stringColumn = DataFrameColumn.Create("String", Enumerable.Range(0, length).Select(x => x.ToString()));
+            AssertLengthTypeAndValues(stringColumn, typeof(string));
+            DataFrameColumn byteColumn = DataFrameColumn.Create("Byte", Enumerable.Range(0, length).Select(x => (byte)x));
+            AssertLengthTypeAndValues(byteColumn, typeof(byte));
+            DataFrameColumn decimalColumn = DataFrameColumn.Create("Decimal", Enumerable.Range(0, length).Select(x => (decimal)x));
+            AssertLengthTypeAndValues(decimalColumn, typeof(decimal));
+            DataFrameColumn doubleColumn = DataFrameColumn.Create("Double", Enumerable.Range(0, length).Select(x => (double)x));
+            AssertLengthTypeAndValues(doubleColumn, typeof(double));
+            DataFrameColumn floatColumn = DataFrameColumn.Create("Float", Enumerable.Range(0, length).Select(x => (float)x));
+            AssertLengthTypeAndValues(floatColumn, typeof(float));
+            DataFrameColumn intColumn = DataFrameColumn.Create("Int", Enumerable.Range(0, length).Select(x => x));
+            AssertLengthTypeAndValues(intColumn, typeof(int));
+            DataFrameColumn longColumn = DataFrameColumn.Create("Long", Enumerable.Range(0, length).Select(x => (long)x));
+            AssertLengthTypeAndValues(longColumn, typeof(long));
+            DataFrameColumn sbyteColumn = DataFrameColumn.Create("Sbyte", Enumerable.Range(0, length).Select(x => (sbyte)x));
+            AssertLengthTypeAndValues(sbyteColumn, typeof(sbyte));
+            DataFrameColumn shortColumn = DataFrameColumn.Create("Short", Enumerable.Range(0, length).Select(x => (short)x));
+            AssertLengthTypeAndValues(shortColumn, typeof(short));
+            DataFrameColumn uintColumn = DataFrameColumn.Create("Uint", Enumerable.Range(0, length).Select(x => (uint)x));
+            AssertLengthTypeAndValues(uintColumn, typeof(uint));
+            DataFrameColumn ulongColumn = DataFrameColumn.Create("Ulong", Enumerable.Range(0, length).Select(x => (ulong)x));
+            AssertLengthTypeAndValues(ulongColumn, typeof(ulong));
+            DataFrameColumn ushortColumn = DataFrameColumn.Create("Ushort", Enumerable.Range(0, length).Select(x => (ushort)x));
+            AssertLengthTypeAndValues(ushortColumn, typeof(ushort));
+        }
     }
 }