diff --git a/src/Microsoft.Data.Analysis/ArrowStringDataFrameColumn.cs b/src/Microsoft.Data.Analysis/ArrowStringDataFrameColumn.cs
index a80e85c173..d05912ca6f 100644
--- a/src/Microsoft.Data.Analysis/ArrowStringDataFrameColumn.cs
+++ b/src/Microsoft.Data.Analysis/ArrowStringDataFrameColumn.cs
@@ -460,34 +460,42 @@ private ArrowStringDataFrameColumn Clone(PrimitiveDataFrameColumn<int> mapIndice
         /// <inheritdoc/>
         public override DataFrame ValueCounts()
         {
-            Dictionary<string, ICollection<long>> groupedValues = GroupColumnValues<string>();
+            Dictionary<string, ICollection<long>> groupedValues = GroupColumnValues<string>(out HashSet<long> _);
             return StringDataFrameColumn.ValueCountsImplementation(groupedValues);
         }
 
         /// <inheritdoc/>
         public override GroupBy GroupBy(int columnIndex, DataFrame parent)
         {
-            Dictionary<string, ICollection<long>> dictionary = GroupColumnValues<string>();
+            Dictionary<string, ICollection<long>> dictionary = GroupColumnValues<string>(out HashSet<long> _);
             return new GroupBy<string>(parent, columnIndex, dictionary);
         }
 
         /// <inheritdoc/>
-        public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>()
+        public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>(out HashSet<long> nullIndices)
         {
             if (typeof(TKey) == typeof(string))
             {
+                nullIndices = new HashSet<long>();
                 Dictionary<string, ICollection<long>> multimap = new Dictionary<string, ICollection<long>>(EqualityComparer<string>.Default);
                 for (long i = 0; i < Length; i++)
                 {
-                    string str = this[i] ?? "__null__";
-                    bool containsKey = multimap.TryGetValue(str, out ICollection<long> values);
-                    if (containsKey)
+                    string str = this[i];
+                    if (str != null)
                     {
-                        values.Add(i);
+                        bool containsKey = multimap.TryGetValue(str, out ICollection<long> values);
+                        if (containsKey)
+                        {
+                            values.Add(i);
+                        }
+                        else
+                        {
+                            multimap.Add(str, new List<long>() { i });
+                        }
                     }
                     else
                     {
-                        multimap.Add(str, new List<long>() { i });
+                        nullIndices.Add(i);
                     }
                 }
                 return multimap as Dictionary<TKey, ICollection<long>>;
@@ -499,7 +507,7 @@ public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>()
         }
 
         /// <inheritdoc/>
-        public ArrowStringDataFrameColumn FillNulls(string value, bool inPlace = false) 
+        public ArrowStringDataFrameColumn FillNulls(string value, bool inPlace = false)
         {
             if (value == null)
             {
diff --git a/src/Microsoft.Data.Analysis/DataFrame.Join.cs b/src/Microsoft.Data.Analysis/DataFrame.Join.cs
index 381268dee2..da99e6254f 100644
--- a/src/Microsoft.Data.Analysis/DataFrame.Join.cs
+++ b/src/Microsoft.Data.Analysis/DataFrame.Join.cs
@@ -168,7 +168,7 @@ public DataFrame Merge<TKey>(DataFrame other, string leftJoinColumn, string righ
             {
                 // First hash other dataframe on the rightJoinColumn 
                 DataFrameColumn otherColumn = other.Columns[rightJoinColumn];
-                Dictionary<TKey, ICollection<long>> multimap = otherColumn.GroupColumnValues<TKey>();
+                Dictionary<TKey, ICollection<long>> multimap = otherColumn.GroupColumnValues<TKey>(out HashSet<long> otherColumnNullIndices);
 
                 // Go over the records in this dataframe and match with the dictionary 
                 DataFrameColumn thisColumn = Columns[leftJoinColumn];
@@ -176,74 +176,64 @@ public DataFrame Merge<TKey>(DataFrame other, string leftJoinColumn, string righ
                 for (long i = 0; i < thisColumn.Length; i++)
                 {
                     var thisColumnValue = thisColumn[i];
-                    TKey thisColumnValueOrDefault = (TKey)(thisColumnValue == null ? default(TKey) : thisColumnValue);
-                    if (multimap.TryGetValue(thisColumnValueOrDefault, out ICollection<long> rowNumbers))
+                    if (thisColumnValue != null)
                     {
-                        foreach (long row in rowNumbers)
+                        if (multimap.TryGetValue((TKey)thisColumnValue, out ICollection<long> rowNumbers))
                         {
-                            if (thisColumnValue == null)
+                            foreach (long row in rowNumbers)
                             {
-                                // Match only with nulls in otherColumn 
-                                if (otherColumn[row] == null)
-                                {
-                                    leftRowIndices.Append(i);
-                                    rightRowIndices.Append(row);
-                                }
-                            }
-                            else
-                            {
-                                // Cannot match nulls in otherColumn 
-                                if (otherColumn[row] != null)
-                                {
-                                    leftRowIndices.Append(i);
-                                    rightRowIndices.Append(row);
-                                }
+                                leftRowIndices.Append(i);
+                                rightRowIndices.Append(row);
                             }
                         }
+                        else
+                        {
+                            leftRowIndices.Append(i);
+                            rightRowIndices.Append(null);
+                        }
                     }
                     else
                     {
-                        leftRowIndices.Append(i);
-                        rightRowIndices.Append(null);
+                        foreach (long row in otherColumnNullIndices)
+                        {
+                            leftRowIndices.Append(i);
+                            rightRowIndices.Append(row);
+                        }
                     }
                 }
             }
             else if (joinAlgorithm == JoinAlgorithm.Right)
             {
                 DataFrameColumn thisColumn = Columns[leftJoinColumn];
-                Dictionary<TKey, ICollection<long>> multimap = thisColumn.GroupColumnValues<TKey>();
+                Dictionary<TKey, ICollection<long>> multimap = thisColumn.GroupColumnValues<TKey>(out HashSet<long> thisColumnNullIndices);
 
                 DataFrameColumn otherColumn = other.Columns[rightJoinColumn];
                 for (long i = 0; i < otherColumn.Length; i++)
                 {
                     var otherColumnValue = otherColumn[i];
-                    TKey otherColumnValueOrDefault = (TKey)(otherColumnValue == null ? default(TKey) : otherColumnValue);
-                    if (multimap.TryGetValue(otherColumnValueOrDefault, out ICollection<long> rowNumbers))
+                    if (otherColumnValue != null)
                     {
-                        foreach (long row in rowNumbers)
+                        if (multimap.TryGetValue((TKey)otherColumnValue, out ICollection<long> rowNumbers))
                         {
-                            if (otherColumnValue == null)
+                            foreach (long row in rowNumbers)
                             {
-                                if (thisColumn[row] == null)
-                                {
-                                    leftRowIndices.Append(row);
-                                    rightRowIndices.Append(i);
-                                }
-                            }
-                            else
-                            {
-                                if (thisColumn[row] != null)
-                                {
-                                    leftRowIndices.Append(row);
-                                    rightRowIndices.Append(i);
-                                }
+                                leftRowIndices.Append(row);
+                                rightRowIndices.Append(i);
                             }
                         }
+                        else
+                        {
+                            leftRowIndices.Append(null);
+                            rightRowIndices.Append(i);
+                        }
                     }
                     else
                     {
-                        leftRowIndices.Append(null);
-                        rightRowIndices.Append(i);
+                        foreach (long thisColumnNullIndex in thisColumnNullIndices)
+                        {
+                            leftRowIndices.Append(thisColumnNullIndex);
+                            rightRowIndices.Append(i);
+                        }
                     }
                 }
             }
@@ -253,97 +243,106 @@ public DataFrame Merge<TKey>(DataFrame other, string leftJoinColumn, string righ
                 long leftRowCount = Rows.Count;
                 long rightRowCount = other.Rows.Count;
 
-                var leftColumnIsSmaller = (leftRowCount <= rightRowCount);
+                bool leftColumnIsSmaller = leftRowCount <= rightRowCount;
                 DataFrameColumn hashColumn = leftColumnIsSmaller ? Columns[leftJoinColumn] : other.Columns[rightJoinColumn];
                 DataFrameColumn otherColumn = ReferenceEquals(hashColumn, Columns[leftJoinColumn]) ? other.Columns[rightJoinColumn] : Columns[leftJoinColumn];
-                Dictionary<TKey, ICollection<long>> multimap = hashColumn.GroupColumnValues<TKey>();
+                Dictionary<TKey, ICollection<long>> multimap = hashColumn.GroupColumnValues<TKey>(out HashSet<long> smallerDataFrameColumnNullIndices);
 
                 for (long i = 0; i < otherColumn.Length; i++)
                 {
                     var otherColumnValue = otherColumn[i];
-                    TKey otherColumnValueOrDefault = (TKey)(otherColumnValue == null ? default(TKey) : otherColumnValue);
-                    if (multimap.TryGetValue(otherColumnValueOrDefault, out ICollection<long> rowNumbers))
+                    if (otherColumnValue != null)
                     {
-                        foreach (long row in rowNumbers)
+                        if (multimap.TryGetValue((TKey)otherColumnValue, out ICollection<long> rowNumbers))
                         {
-                            if (otherColumnValue == null)
-                            {
-                                if (hashColumn[row] == null)
-                                {
-                                    leftRowIndices.Append(leftColumnIsSmaller ? row : i);
-                                    rightRowIndices.Append(leftColumnIsSmaller ? i : row);
-                                }
-                            }
-                            else
+                            foreach (long row in rowNumbers)
                             {
-                                if (hashColumn[row] != null)
-                                {
-                                    leftRowIndices.Append(leftColumnIsSmaller ? row : i);
-                                    rightRowIndices.Append(leftColumnIsSmaller ? i : row);
-                                }
+                                leftRowIndices.Append(leftColumnIsSmaller ? row : i);
+                                rightRowIndices.Append(leftColumnIsSmaller ? i : row);
                             }
                         }
                     }
+                    else
+                    {
+                        foreach (long nullIndex in smallerDataFrameColumnNullIndices)
+                        {
+                            leftRowIndices.Append(leftColumnIsSmaller ? nullIndex : i);
+                            rightRowIndices.Append(leftColumnIsSmaller ? i : nullIndex);
+                        }
+                    }
                 }
             }
             else if (joinAlgorithm == JoinAlgorithm.FullOuter)
             {
                 DataFrameColumn otherColumn = other.Columns[rightJoinColumn];
-                Dictionary<TKey, ICollection<long>> multimap = otherColumn.GroupColumnValues<TKey>();
+                Dictionary<TKey, ICollection<long>> multimap = otherColumn.GroupColumnValues<TKey>(out HashSet<long> otherColumnNullIndices);
                 Dictionary<TKey, long> intersection = new Dictionary<TKey, long>(EqualityComparer<TKey>.Default);
 
                 // Go over the records in this dataframe and match with the dictionary 
                 DataFrameColumn thisColumn = Columns[leftJoinColumn];
+                Int64DataFrameColumn thisColumnNullIndices = new Int64DataFrameColumn("ThisColumnNullIndices");
 
                 for (long i = 0; i < thisColumn.Length; i++)
                 {
                     var thisColumnValue = thisColumn[i];
-                    TKey thisColumnValueOrDefault = (TKey)(thisColumnValue == null ? default(TKey) : thisColumnValue);
-                    if (multimap.TryGetValue(thisColumnValueOrDefault, out ICollection<long> rowNumbers))
+                    if (thisColumnValue != null)
                     {
-                        foreach (long row in rowNumbers)
+                        if (multimap.TryGetValue((TKey)thisColumnValue, out ICollection<long> rowNumbers))
                         {
-                            if (thisColumnValue == null)
-                            {
-                                // Has to match only with nulls in otherColumn 
-                                if (otherColumn[row] == null)
-                                {
-                                    leftRowIndices.Append(i);
-                                    rightRowIndices.Append(row);
-                                    if (!intersection.ContainsKey(thisColumnValueOrDefault))
-                                    {
-                                        intersection.Add(thisColumnValueOrDefault, rowNumber);
-                                    }
-                                }
-                            }
-                            else
+                            foreach (long row in rowNumbers)
                             {
-                                // Cannot match to nulls in otherColumn 
-                                if (otherColumn[row] != null)
+                                leftRowIndices.Append(i);
+                                rightRowIndices.Append(row);
+                                if (!intersection.ContainsKey((TKey)thisColumnValue))
                                 {
-                                    leftRowIndices.Append(i);
-                                    rightRowIndices.Append(row);
-                                    if (!intersection.ContainsKey(thisColumnValueOrDefault))
-                                    {
-                                        intersection.Add(thisColumnValueOrDefault, rowNumber);
-                                    }
+                                    intersection.Add((TKey)thisColumnValue, rowNumber);
                                 }
                             }
                         }
+                        else
+                        {
+                            leftRowIndices.Append(i);
+                            rightRowIndices.Append(null);
+                        }
                     }
                     else
                     {
-                        leftRowIndices.Append(i);
-                        rightRowIndices.Append(null);
+                        thisColumnNullIndices.Append(i);
                     }
                 }
                 for (long i = 0; i < otherColumn.Length; i++)
                 {
-                    TKey value = (TKey)(otherColumn[i] ?? default(TKey));
-                    if (!intersection.ContainsKey(value))
+                    var value = otherColumn[i];
+                    if (value != null)
+                    {
+                        if (!intersection.ContainsKey((TKey)value))
+                        {
+                            leftRowIndices.Append(null);
+                            rightRowIndices.Append(i);
+                        }
+                    }
+                }
+
+                // Now handle the null rows
+                foreach (long? thisColumnNullIndex in thisColumnNullIndices)
+                {
+                    foreach (long otherColumnNullIndex in otherColumnNullIndices)
+                    {
+                        leftRowIndices.Append(thisColumnNullIndex.Value);
+                        rightRowIndices.Append(otherColumnNullIndex);
+                    }
+                    if (otherColumnNullIndices.Count == 0)
+                    {
+                        leftRowIndices.Append(thisColumnNullIndex.Value);
+                        rightRowIndices.Append(null);
+                    }
+                }
+                if (thisColumnNullIndices.Length == 0)
+                {
+                    foreach (long otherColumnNullIndex in otherColumnNullIndices)
                     {
                         leftRowIndices.Append(null);
-                        rightRowIndices.Append(i);
+                        rightRowIndices.Append(otherColumnNullIndex);
                     }
                 }
             }
diff --git a/src/Microsoft.Data.Analysis/DataFrameColumn.cs b/src/Microsoft.Data.Analysis/DataFrameColumn.cs
index bd21d6fe96..67e79cc300 100644
--- a/src/Microsoft.Data.Analysis/DataFrameColumn.cs
+++ b/src/Microsoft.Data.Analysis/DataFrameColumn.cs
@@ -203,7 +203,12 @@ public virtual DataFrameColumn Sort(bool ascending = true)
             return Clone(sortIndices, !ascending, NullCount);
         }
 
-        public virtual Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>() => throw new NotImplementedException();
+        /// <summary>
+        /// Groups the rows of this column by their value.
+        /// </summary>
+        /// <typeparam name="TKey">The type of data held by this column</typeparam>
+        /// <returns>A mapping of value(<typeparamref name="TKey"/>) to the indices containing this value</returns>
+        public virtual Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>(out HashSet<long> nullIndices) => throw new NotImplementedException();
 
         /// <summary>
         /// Returns a DataFrame containing counts of unique values
diff --git a/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs b/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs
index a7e7d20cb9..10f1627692 100644
--- a/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs
+++ b/src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs
@@ -313,7 +313,7 @@ protected override DataFrameColumn FillNullsImplementation(object value, bool in
 
         public override DataFrame ValueCounts()
         {
-            Dictionary<T, ICollection<long>> groupedValues = GroupColumnValues<T>();
+            Dictionary<T, ICollection<long>> groupedValues = GroupColumnValues<T>(out HashSet<long> _);
             PrimitiveDataFrameColumn<T> keys = new PrimitiveDataFrameColumn<T>("Values");
             PrimitiveDataFrameColumn<long> counts = new PrimitiveDataFrameColumn<long>("Counts");
             foreach (KeyValuePair<T, ICollection<long>> keyValuePair in groupedValues)
@@ -520,31 +520,40 @@ internal SingleDataFrameColumn CloneAsSingleColumn()
         /// <inheritdoc/>
         public override GroupBy GroupBy(int columnIndex, DataFrame parent)
         {
-            Dictionary<T, ICollection<long>> dictionary = GroupColumnValues<T>();
+            Dictionary<T, ICollection<long>> dictionary = GroupColumnValues<T>(out HashSet<long> _);
             return new GroupBy<T>(parent, columnIndex, dictionary);
         }
 
-        public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>()
+        public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>(out HashSet<long> nullIndices)
         {
             if (typeof(TKey) == typeof(T))
             {
                 Dictionary<T, ICollection<long>> multimap = new Dictionary<T, ICollection<long>>(EqualityComparer<T>.Default);
+                nullIndices = new HashSet<long>();
                 for (int b = 0; b < _columnContainer.Buffers.Count; b++)
                 {
                     ReadOnlyDataFrameBuffer<T> buffer = _columnContainer.Buffers[b];
                     ReadOnlySpan<T> readOnlySpan = buffer.ReadOnlySpan;
+                    ReadOnlySpan<byte> nullBitMapSpan = _columnContainer.NullBitMapBuffers[b].ReadOnlySpan;
                     long previousLength = b * ReadOnlyDataFrameBuffer<T>.MaxCapacity;
                     for (int i = 0; i < readOnlySpan.Length; i++)
                     {
                         long currentLength = i + previousLength;
-                        bool containsKey = multimap.TryGetValue(readOnlySpan[i], out ICollection<long> values);
-                        if (containsKey)
+                        if (_columnContainer.IsValid(nullBitMapSpan, i))
                         {
-                            values.Add(currentLength);
+                            bool containsKey = multimap.TryGetValue(readOnlySpan[i], out ICollection<long> values);
+                            if (containsKey)
+                            {
+                                values.Add(currentLength);
+                            }
+                            else
+                            {
+                                multimap.Add(readOnlySpan[i], new List<long>() { currentLength });
+                            }
                         }
                         else
                         {
-                            multimap.Add(readOnlySpan[i], new List<long>() { currentLength });
+                            nullIndices.Add(currentLength);
                         }
                     }
                 }
diff --git a/src/Microsoft.Data.Analysis/StringDataFrameColumn.cs b/src/Microsoft.Data.Analysis/StringDataFrameColumn.cs
index 7ada30e10c..761fbcda6b 100644
--- a/src/Microsoft.Data.Analysis/StringDataFrameColumn.cs
+++ b/src/Microsoft.Data.Analysis/StringDataFrameColumn.cs
@@ -398,31 +398,40 @@ internal static DataFrame ValueCountsImplementation(Dictionary<string, ICollecti
 
         public override DataFrame ValueCounts()
         {
-            Dictionary<string, ICollection<long>> groupedValues = GroupColumnValues<string>();
+            Dictionary<string, ICollection<long>> groupedValues = GroupColumnValues<string>(out HashSet<long> _);
             return ValueCountsImplementation(groupedValues);
         }
 
         public override GroupBy GroupBy(int columnIndex, DataFrame parent)
         {
-            Dictionary<string, ICollection<long>> dictionary = GroupColumnValues<string>();
+            Dictionary<string, ICollection<long>> dictionary = GroupColumnValues<string>(out HashSet<long> _);
             return new GroupBy<string>(parent, columnIndex, dictionary);
         }
 
-        public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>()
+        public override Dictionary<TKey, ICollection<long>> GroupColumnValues<TKey>(out HashSet<long> nullIndices)
         {
             if (typeof(TKey) == typeof(string))
             {
                 Dictionary<string, ICollection<long>> multimap = new Dictionary<string, ICollection<long>>(EqualityComparer<string>.Default);
+                nullIndices = new HashSet<long>();
                 for (long i = 0; i < Length; i++)
                 {
-                    bool containsKey = multimap.TryGetValue(this[i] ?? default, out ICollection<long> values);
-                    if (containsKey)
+                    string str = this[i];
+                    if (str != null)
                     {
-                        values.Add(i);
+                        bool containsKey = multimap.TryGetValue(str, out ICollection<long> values);
+                        if (containsKey)
+                        {
+                            values.Add(i);
+                        }
+                        else
+                        {
+                            multimap.Add(str, new List<long>() { i });
+                        }
                     }
                     else
                     {
-                        multimap.Add(this[i] ?? default, new List<long>() { i });
+                        nullIndices.Add(i);
                     }
                 }
                 return multimap as Dictionary<TKey, ICollection<long>>;
diff --git a/test/Microsoft.Data.Analysis.Tests/DataFrameTests.cs b/test/Microsoft.Data.Analysis.Tests/DataFrameTests.cs
index 72072fd533..348aeeea97 100644
--- a/test/Microsoft.Data.Analysis.Tests/DataFrameTests.cs
+++ b/test/Microsoft.Data.Analysis.Tests/DataFrameTests.cs
@@ -1152,7 +1152,7 @@ public void TestGroupBy()
                 if (originalColumn.Name == "Bool")
                     continue;
                 DataFrameColumn headColumn = head.Columns[originalColumn.Name];
-                Assert.Equal(originalColumn[5], headColumn[verify[5]]);
+                Assert.Equal(originalColumn[7], headColumn[verify[5]]);
             }
             Assert.Equal(6, head.Rows.Count);
 
@@ -1569,14 +1569,14 @@ public void TestSample()
 
             // all sampled rows should be unique.
             HashSet<int?> uniqueRowValues = new HashSet<int?>();
-            foreach(int? value in sampled.Columns["Int"])
+            foreach (int? value in sampled.Columns["Int"])
             {
                 uniqueRowValues.Add(value);
             }
             Assert.Equal(uniqueRowValues.Count, sampled.Rows.Count);
 
             // should throw exception as sample size is greater than dataframe rows
-            Assert.Throws<ArgumentException>(()=> df.Sample(13));
+            Assert.Throws<ArgumentException>(() => df.Sample(13));
         }
 
         [Theory]
@@ -1658,7 +1658,7 @@ public void TestMerge()
             Assert.Equal(16, merge.Rows.Count);
             Assert.Equal(merge.Columns.Count, left.Columns.Count + right.Columns.Count);
             Assert.Null(merge.Columns["Int_left"][12]);
-            Assert.Null(merge.Columns["Int_left"][5]);
+            Assert.Null(merge.Columns["Int_left"][15]);
             VerifyMerge(merge, left, right, JoinAlgorithm.FullOuter);
 
             // Inner merge 
@@ -1669,6 +1669,205 @@ public void TestMerge()
             VerifyMerge(merge, left, right, JoinAlgorithm.Inner);
         }
 
+        private void MatchRowsOnMergedDataFrame(DataFrame merge, DataFrame left, DataFrame right, long mergeRow, long? leftRow, long? rightRow)
+        {
+            Assert.Equal(merge.Columns.Count, left.Columns.Count + right.Columns.Count);
+            DataFrameRow dataFrameMergeRow = merge.Rows[mergeRow];
+            int columnIndex = 0;
+            foreach (object value in dataFrameMergeRow)
+            {
+                object compare = null;
+                if (columnIndex < left.Columns.Count)
+                {
+                    if (leftRow != null)
+                    {
+                        compare = left.Rows[leftRow.Value][columnIndex];
+                    }
+                }
+                else
+                {
+                    int rightColumnIndex = columnIndex - left.Columns.Count;
+                    if (rightRow != null)
+                    {
+                        compare = right.Rows[rightRow.Value][rightColumnIndex];
+                    }
+                }
+                Assert.Equal(value, compare);
+                columnIndex++;
+            }
+        }
+
+        [Theory]
+        [InlineData(10, 5, JoinAlgorithm.Left)]
+        [InlineData(5, 10, JoinAlgorithm.Right)]
+        public void TestMergeEdgeCases_LeftOrRight(int leftLength, int rightLength, JoinAlgorithm joinAlgorithm)
+        {
+            DataFrame left = MakeDataFrameWithAllMutableColumnTypes(leftLength);
+            if (leftLength > 5)
+            {
+                left["Int"][8] = null;
+            }
+            DataFrame right = MakeDataFrameWithAllMutableColumnTypes(rightLength);
+            if (rightLength > 5)
+            {
+                right["Int"][8] = null;
+            }
+
+            DataFrame merge = left.Merge<int>(right, "Int", "Int", joinAlgorithm: joinAlgorithm);
+            Assert.Equal(10, merge.Rows.Count);
+            Assert.Equal(merge.Columns.Count, left.Columns.Count + right.Columns.Count);
+            int[] matchedFullRows = new int[] { 0, 1, 3, 4 };
+            for (long i = 0; i < matchedFullRows.Length; i++)
+            {
+                int rowIndex = matchedFullRows[i];
+                MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, rowIndex, rowIndex);
+            }
+
+            int[] matchedLeftOrRightRowsNullOtherRows = new int[] { 2, 5, 6, 7, 8, 9 };
+            for (long i = 0; i < matchedLeftOrRightRowsNullOtherRows.Length; i++)
+            {
+                int rowIndex = matchedLeftOrRightRowsNullOtherRows[i];
+                if (leftLength > 5)
+                {
+                    MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, rowIndex, null);
+                }
+                else
+                {
+                    MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, null, rowIndex);
+                }
+            }
+        }
+
+        [Fact]
+        public void TestMergeEdgeCases_Inner()
+        {
+            DataFrame left = MakeDataFrameWithAllMutableColumnTypes(5);
+            DataFrame right = MakeDataFrameWithAllMutableColumnTypes(10);
+            left["Int"][3] = null;
+            right["Int"][6] = null;
+            // Creates this case:
+            /*
+             * Left:    Right:
+             * 0        0
+             * 1        1
+             * null(2)  2
+             * null(3)  3
+             * 4        4
+             *          null(5)
+             *          null(6)
+             *          7
+             *          8
+             *          9
+             */
+            /*
+             * Merge will result in a DataFrame like:
+             * Int_Left Int_Right
+             * 0        0
+             * 1        1
+             * 4        4
+             * null(2)  null(5)
+             * null(3)  null(5)
+             * null(2)  null(6)
+             * null(3)  null(6)
+             */
+
+            DataFrame merge = left.Merge<int>(right, "Int", "Int", joinAlgorithm: JoinAlgorithm.Inner);
+            Assert.Equal(7, merge.Rows.Count);
+            Assert.Equal(merge.Columns.Count, left.Columns.Count + right.Columns.Count);
+
+            int[] mergeRows = new int[] { 0, 1, 2, 3, 4, 5, 6 };
+            int[] leftRows = new int[] { 0, 1, 4, 2, 3, 2, 3 };
+            int[] rightRows = new int[] { 0, 1, 4, 5, 5, 6, 6 };
+            for (long i = 0; i < mergeRows.Length; i++)
+            {
+                int rowIndex = mergeRows[i];
+                int leftRowIndex = leftRows[i];
+                int rightRowIndex = rightRows[i];
+                MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, leftRowIndex, rightRowIndex);
+            }
+        }
+
+        [Fact]
+        public void TestMergeEdgeCases_Outer()
+        {
+            DataFrame left = MakeDataFrameWithAllMutableColumnTypes(5);
+            left["Int"][3] = null;
+            DataFrame right = MakeDataFrameWithAllMutableColumnTypes(5);
+            // Creates this case:
+            /*
+             * Left:    Right:
+             * 0        0
+             * 1        5
+             * null(2)  null(7)
+             * null(3)  null(8)
+             * 4        6
+             */
+            /*
+             * Merge will result in a DataFrame like:
+             * Int_Left     Int_Right
+             * 0            0
+             * 1            null
+             * 4            null 
+             * null         5
+             * null         6
+             * null(2)      null(7)
+             * null(2)      null(8)
+             * null(3)      null(7)
+             * null(3)      null(8)
+             */
+            right["Int"][1] = 5;
+            right["Int"][3] = null;
+            right["Int"][4] = 6;
+
+            DataFrame merge = left.Merge<int>(right, "Int", "Int", joinAlgorithm: JoinAlgorithm.FullOuter);
+            Assert.Equal(9, merge.Rows.Count);
+            Assert.Equal(merge.Columns.Count, left.Columns.Count + right.Columns.Count);
+
+            int[] mergeRows = new int[] { 0, 5, 6, 7, 8 };
+            int[] leftRows = new int[] { 0, 2, 2, 3, 3 };
+            int[] rightRows = new int[] { 0, 2, 3, 2, 3 };
+            for (long i = 0; i < mergeRows.Length; i++)
+            {
+                int rowIndex = mergeRows[i];
+                int leftRowIndex = leftRows[i];
+                int rightRowIndex = rightRows[i];
+                MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, leftRowIndex, rightRowIndex);
+            }
+
+            mergeRows = new int[] { 1, 2 };
+            leftRows = new int[] { 1, 4 };
+            for (long i = 0; i < mergeRows.Length; i++)
+            {
+                int rowIndex = mergeRows[i];
+                int leftRowIndex = leftRows[i];
+                MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, leftRowIndex, null);
+            }
+
+            mergeRows = new int[] { 3, 4 };
+            rightRows = new int[] { 1, 4 };
+            for (long i = 0; i < mergeRows.Length; i++)
+            {
+                int rowIndex = mergeRows[i];
+                int rightRowIndex = rightRows[i];
+                MatchRowsOnMergedDataFrame(merge, left, right, rowIndex, null, rightRowIndex);
+            }
+        }
+
+        [Fact]
+        public void TestMerge_Issue5778()
+        {
+            DataFrame left = MakeDataFrameWithAllMutableColumnTypes(2, false);
+            DataFrame right = MakeDataFrameWithAllMutableColumnTypes(1);
+
+            DataFrame merge = left.Merge<int>(right, "Int", "Int");
+
+            Assert.Equal(2, merge.Rows.Count);
+            Assert.Equal(0, (int)merge.Columns["Int_left"][0]);
+            Assert.Equal(1, (int)merge.Columns["Int_left"][1]);
+            MatchRowsOnMergedDataFrame(merge, left, right, 0, 0, 0);
+            MatchRowsOnMergedDataFrame(merge, left, right, 1, 1, 0);
+        }
+
         [Fact]
         public void TestDescription()
         {