Fix DataFrame Merge issue

# Conflicts:
#	src/Microsoft.Data.Analysis/BooleanDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/ByteDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/CharDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/DateTimeDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/DecimalDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/DoubleDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/Int16DataFrameColumn.cs
#	src/Microsoft.Data.Analysis/Int32DataFrameColumn.cs
#	src/Microsoft.Data.Analysis/Int64DataFrameColumn.cs
#	src/Microsoft.Data.Analysis/SByteDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/SingleDataFrameColumn.cs
#	src/Microsoft.Data.Analysis/UInt16DataFrameColumn.cs
#	src/Microsoft.Data.Analysis/UInt32DataFrameColumn.cs
#	src/Microsoft.Data.Analysis/UInt64DataFrameColumn.cs
#	test/Microsoft.Data.Analysis.Tests/DataFrame.IOTests.cs

Author: asmirnov82

src/Microsoft.Data.Analysis/DataFrameColumnCollection.cs

@@ -220,6 +220,23 @@ public PrimitiveDataFrameColumn<T> GetPrimitiveColumn<T>(string name)
             throw new ArgumentException(string.Format(Strings.BadColumnCast, column.DataType, typeof(T)), nameof(T));
         }
 
+        /// <summary>
+        /// Gets the <see cref="PrimitiveDataFrameColumn{T}"/> with the specified <paramref name="name"/>.
+        /// </summary>
+        /// <param name="name">The name of the column</param>
+        /// <returns><see cref="PrimitiveDataFrameColumn{T}"/>.</returns>
+        /// <exception cref="ArgumentException">A column named <paramref name="name"/> cannot be found, or if the column's type doesn't match.</exception>
+        public PrimitiveDataFrameColumn<DateTime> GetDateTimeColumn(string name)
+        {
+            DataFrameColumn column = this[name];
+            if (column is PrimitiveDataFrameColumn<DateTime> ret)
+            {
+                return ret;
+            }
+
+            throw new ArgumentException(string.Format(Strings.BadColumnCast, column.DataType, typeof(DateTime)));
+        }
+
         /// <summary>
         /// Gets the <see cref="ArrowStringDataFrameColumn"/> with the specified <paramref name="name"/>.
         /// </summary>

src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs

@@ -220,6 +220,16 @@ protected override IReadOnlyList<object> GetValues(long startIndex, int length)
             return ret;
         }
 
+        internal virtual PrimitiveDataFrameColumn<T> CreateNewColumn(string name, PrimitiveColumnContainer<T> container)
+        {
+            return new PrimitiveDataFrameColumn<T>(name, container);
+        }
+
+        protected virtual PrimitiveDataFrameColumn<T> CreateNewColumn(string name, long length = 0)
+        {
+            return new PrimitiveDataFrameColumn<T>(name, length);
+        }
+
         internal T? GetTypedValue(long rowIndex) => _columnContainer[rowIndex];
 
         protected override object GetValue(long rowIndex) => GetTypedValue(rowIndex);
@@ -411,7 +421,7 @@ private PrimitiveDataFrameColumn<T> Clone(PrimitiveDataFrameColumn<bool> boolCol
         {
             if (boolColumn.Length > Length)
                 throw new ArgumentException(Strings.MapIndicesExceedsColumnLenth, nameof(boolColumn));
-            PrimitiveDataFrameColumn<T> ret = new PrimitiveDataFrameColumn<T>(Name);
+            PrimitiveDataFrameColumn<T> ret = CreateNewColumn(Name);
             for (long i = 0; i < boolColumn.Length; i++)
             {
                 bool? value = boolColumn[i];
@@ -438,7 +448,8 @@ private PrimitiveDataFrameColumn<T> CloneImplementation<U>(PrimitiveDataFrameCol
             }
             else
                 throw new NotImplementedException();
-            PrimitiveDataFrameColumn<T> ret = new PrimitiveDataFrameColumn<T>(Name, retContainer);
+
+            PrimitiveDataFrameColumn<T> ret = CreateNewColumn(Name, retContainer);
             return ret;
         }
 
@@ -447,7 +458,7 @@ public PrimitiveDataFrameColumn<T> Clone(PrimitiveDataFrameColumn<long> mapIndic
             if (mapIndices is null)
             {
                 PrimitiveColumnContainer<T> newColumnContainer = _columnContainer.Clone();
-                return new PrimitiveDataFrameColumn<T>(Name, newColumnContainer);
+                return CreateNewColumn(Name, newColumnContainer);
             }
             else
             {

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

@@ -2766,6 +2766,56 @@ public void TestMerge_Issue5778()
             MatchRowsOnMergedDataFrame(merge, left, right, 1, 1, 0);
         }
 
+        [Fact]
+        //Issue 6127
+        public void TestMerge_CorrectColumnTypes()
+        {
+            DataFrame left = MakeDataFrameWithAllMutableColumnTypes(2, false);
+            DataFrame right = MakeDataFrameWithAllMutableColumnTypes(1);
+
+            DataFrame merge = left.Merge<int>(right, "Int", "Int");
+
+            Assert.NotNull(merge.Columns.GetBooleanColumn("Bool_left"));
+            Assert.NotNull(merge.Columns.GetBooleanColumn("Bool_right"));
+
+            Assert.NotNull(merge.Columns.GetDecimalColumn("Decimal_left"));
+            Assert.NotNull(merge.Columns.GetDecimalColumn("Decimal_right"));
+
+            Assert.NotNull(merge.Columns.GetSingleColumn("Float_left"));
+            Assert.NotNull(merge.Columns.GetSingleColumn("Float_right"));
+
+            Assert.NotNull(merge.Columns.GetDoubleColumn("Double_left"));
+            Assert.NotNull(merge.Columns.GetDoubleColumn("Double_right"));
+
+            Assert.NotNull(merge.Columns.GetByteColumn("Byte_left"));
+            Assert.NotNull(merge.Columns.GetByteColumn("Byte_right"));
+
+            Assert.NotNull(merge.Columns.GetCharColumn("Char_left"));
+            Assert.NotNull(merge.Columns.GetCharColumn("Char_right"));
+
+            Assert.NotNull(merge.Columns.GetInt16Column("Short_left"));
+            Assert.NotNull(merge.Columns.GetInt16Column("Short_right"));
+
+            Assert.NotNull(merge.Columns.GetUInt16Column("Ushort_left"));
+            Assert.NotNull(merge.Columns.GetUInt16Column("Ushort_right"));
+
+            Assert.NotNull(merge.Columns.GetInt32Column("Int_left"));
+            Assert.NotNull(merge.Columns.GetInt32Column("Int_right"));
+
+            Assert.NotNull(merge.Columns.GetUInt32Column("Uint_left"));
+            Assert.NotNull(merge.Columns.GetUInt32Column("Uint_right"));
+
+            Assert.NotNull(merge.Columns.GetInt64Column("Long_left"));
+            Assert.NotNull(merge.Columns.GetInt64Column("Long_right"));
+
+            Assert.NotNull(merge.Columns.GetUInt64Column("Ulong_left"));
+            Assert.NotNull(merge.Columns.GetUInt64Column("Ulong_right"));
+
+            Assert.NotNull(merge.Columns.GetDateTimeColumn("DateTime_left"));
+            Assert.NotNull(merge.Columns.GetDateTimeColumn("DateTime_right"));
+
+        }
+
         [Fact]
         public void TestDescription()
         {