Add ability to merge using multiple columns in JOIN condition

src/Microsoft.Data.Analysis/ArrowStringDataFrameColumn.cs

@@ -660,5 +660,10 @@ public ArrowStringDataFrameColumn Apply(Func<string, string> func)
             }
             return ret;
         }
+
+        public override Dictionary<long, ICollection<long>> GetGroupedOccurrences(DataFrameColumn other, out HashSet<long> otherColumnNullIndices)
+        {
+            return GetGroupedOccurrences<string>(other, out otherColumnNullIndices);
+        }
     }
 }

src/Microsoft.Data.Analysis/DataFrameColumn.cs

@@ -210,6 +210,44 @@ public virtual DataFrameColumn Sort(bool ascending = true)
         /// <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>
+        /// Get occurences of each value from this column in other column, grouped by this value
+        /// </summary>
+        /// <param name="other"></param>
+        /// <param name="otherColumnNullIndices"></param>
+        /// <returns>A mapping of index from this column to the indices of same value in other column</returns>
+        public abstract Dictionary<long, ICollection<long>> GetGroupedOccurrences(DataFrameColumn other, out HashSet<long> otherColumnNullIndices);
+
+        /// <summary>
+        /// Get occurences of each value from this column in other column, grouped by this value
+        /// </summary>
+        /// <typeparam name="TKey"></typeparam>
+        /// <param name="other"></param>
+        /// <param name="otherColumnNullIndices"></param>
+        /// <returns>A mapping of index from this column to the indices of same value in other column</returns>
+        protected Dictionary<long, ICollection<long>> GetGroupedOccurrences<TKey>(DataFrameColumn other, out HashSet<long> otherColumnNullIndices)
+        {
+            if (this.DataType != other.DataType)
+                throw new ArgumentException(String.Format(Strings.MismatchedColumnValueType, this.DataType), nameof(other));
+
+            // First hash other column   
+            Dictionary<TKey, ICollection<long>> multimap = other.GroupColumnValues<TKey>(out otherColumnNullIndices);
+
+            var ret = new Dictionary<long, ICollection<long>>();
+
+            //For each value in this column find rows from other column with equal value
+            for (int i = 0; i < this.Length; i++)
+            {
+                var value = this[i];
+                if (value != null && multimap.TryGetValue((TKey)value, out ICollection<long> otherRowIndices))
+                {
+                    ret.Add(i, otherRowIndices);
+                }
+            }
+
+            return ret;
+        }
+
         /// <summary>
         /// Returns a DataFrame containing counts of unique values
         /// </summary>

src/Microsoft.Data.Analysis/PrimitiveDataFrameColumn.cs

@@ -810,5 +810,10 @@ protected internal override Delegate GetValueGetterUsingCursor(DataViewRowCursor
         {
             return cursor.GetGetter<T>(schemaColumn);
         }
+
+        public override Dictionary<long, ICollection<long>> GetGroupedOccurrences(DataFrameColumn other, out HashSet<long> otherColumnNullIndices)
+        {
+            return GetGroupedOccurrences<T>(other, out otherColumnNullIndices);
+        }
     }
 }

src/Microsoft.Data.Analysis/StringDataFrameColumn.cs

@@ -511,5 +511,10 @@ protected internal override Delegate GetValueGetterUsingCursor(DataViewRowCursor
         {
             return cursor.GetGetter<ReadOnlyMemory<char>>(schemaColumn);
         }
+
+        public override Dictionary<long, ICollection<long>> GetGroupedOccurrences(DataFrameColumn other, out HashSet<long> otherColumnNullIndices)
+        {
+            return GetGroupedOccurrences<string>(other, out otherColumnNullIndices);
+        }
     }
 }

src/Microsoft.Data.Analysis/Strings.resx

@@ -201,6 +201,9 @@
   <data name="MapIndicesExceedsColumnLenth" xml:space="preserve">
     <value>MapIndices exceeds column length</value>
   </data>
+  <data name="MismatchedArrayLengths" xml:space="preserve">
+    <value>Array lengths are mistmached</value>
+  </data>
   <data name="MismatchedColumnLengths" xml:space="preserve">
     <value>Column lengths are mismatched</value>
   </data>

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

@@ -91,7 +91,6 @@ public void TestGroupingWithTKeyPrimitiveType()
         [Fact]
         public void TestGroupingWithTKeyOfWrongType()
         {           
-
             var message = string.Empty;
 
             //Create test dataframe (numbers starting from 0 up to lenght)