Repository: jtablesaw/tablesaw
Pull request #377 (2018-09-05 07:11:29)

Add support to join tables on multiple columns.
Thanks for contributing.

- [x] Tick to sign-off your agreement to the [Developer Certificate of Origin (DCO) 1.1](https://developercertificate.org)

## Description

Enhanced API to accept multiple names of columns to join on.  Supported for all joins: inner, leftOuter, rightOuter, and fullOuter.  Added javadoc where missing for public methods.

## Testing

Added dozens of junit tests to cover all changed API.  Ran coverage tool in eclipse to ensure testing/exercising of all changes was being done.  Reached 100% coverage for all but one method.  Goal to reach 100% coverage exposed failure of LONG types to be parse from cvs input so added missing support for that.


benmccann: Wow, this is great!
gregorco: @benmccann  Thanks for the review.  I'll work on those changes. 
lwhite1: Joining on Boolean columns Isnt supported.  If it was done on the Boolean
col only you would get n squared rows frequently

On Wed, Sep 5, 2018 at 11:21 AM Gregor Okorn <notifications@github.com>
wrote:

> *@gregorco* commented on this pull request.
> ------------------------------
>
> In core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
> <https://github.com/jtablesaw/tablesaw/pull/377#discussion_r215315430>:
>
> >                  } else {
> -                    crossProduct(result, table1Rows, table2Rows);
> +                    throw new IllegalArgumentException(
> +                            "Joining is supported on numeric, string, and date-like columns. Column "
>
> If message was changed to say that only categorical columns are supported
> then BooleanColumn should already be accepted, but trying to join on a
> boolean column will currently filter to this else condition; I got this
> condition to be exercised with a junit test that joined on a boolean
> column. BooleanColumn implements CategoricalColumn though so should an if
> condition be added here for BooleanColumnType?
>
> —
> You are receiving this because you are subscribed to this thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/jtablesaw/tablesaw/pull/377#discussion_r215315430>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ADRXguDZRYYzS2w8T6DD1bVflEEoY05gks5uX-wWgaJpZM4WaTs4>
> .
>

gregorco: Thanks for the clarification.  Since BooleanColumn implements CategoricalColumn though, should the message in that IllegalArgumentException be changed - or left as is?
benmccann: Maybe change "numeric" to "integral" ?
gregorco: Sounds good.  Thanks. 

<sub>Sent with <a href="http://githawk.com">GitHawk</a></sub>
gregorco: @benmccann My app used to be able to join on a DoubleColumn, but now as I'm integrating recent changes into it, I'm seeing a failure that the DoubleColumn I'm trying to join on can't be cast to a CategoricalColumn.   I know I had the join working in the past.  At what point was that ability changed/restricted?  Is there any workaround to allow a join on a DoubleColumn?
lwhite1: it changed in master, i think.

(btw, I think this should be in a separate issue because it has nothing to
do with joining on multiple columns)

There was a concern that joining on double was a bit dangerous due to
rounding issues. We added joining on doubles when we didn't have any other
numeric column types. Usually people want to join on an int column.

Do you need to join on double or can your data be stored in an int column
before joining?

i'm not sure we should really prevent people from trying to join on double.
I'm also a bit uncertain that we should prevent people from joining on
booleans, because if they blow up their app it's on them. Perhaps we should
allow joining on any column type. Fewer special cases are better.


On Thu, Sep 6, 2018 at 4:07 AM Gregor Okorn <notifications@github.com>
wrote:

> @benmccann <https://github.com/benmccann> My app used to be able to join
> on a DoubleColumn, but now as I'm integrating recent changes into it, I'm
> seeing a failure that the DoubleColumn I'm trying to join on can't be cast
> to a CategoricalColumn. I know I had the join working in the past. At what
> point was that ability changed/restricted? Is there any workaround to allow
> a join on a DoubleColumn?
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/jtablesaw/tablesaw/pull/377#issuecomment-419004135>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ADRXggZ71sWzRh4oFoaGCn_Eos6YHYjhks5uYNfBgaJpZM4WaTs4>
> .
>

gregorco: @benmccann Don't know if you received notice of the reply comments I added yesterday regarding your comments for the slf4j and simplified for-loop suggestions.  I could add a logging mechanism, but am wondering if logback would be preferred over log4j.  The classic for-loop seemed to be the best solution for that situation.  Let me know what you think.  Thanks.
lwhite1: We don't use any library currently. Logging in java is a mess. In larger
projects you get multiple logging libraries all doing their own thing
because the libraries add them. I don't know that adding one is an
improvement or not.

I would prefer, I think, to write (relatively few) warnings/informational
output to standard out, and throw exceptions for real issues, but I'm open
to discussion. In the example code above, if the Illegal state causes real
issues we should let the client code deal with it.



On Thu, Sep 6, 2018 at 9:48 AM Gregor Okorn <notifications@github.com>
wrote:

> @benmccann <https://github.com/benmccann> Don't know if you received
> notice of the reply comments I added yesterday regarding your comments for
> the slf4j and simplified for-loop suggestions. I could add a logging
> mechanism, but am wondering if logback would be preferred over log4j. The
> classic for-loop seemed to be the best solution for that situation. Let me
> know what you think. Thanks.
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/jtablesaw/tablesaw/pull/377#issuecomment-419099400>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ADRXglgkN9r_l5aJPc6VWg736Y49J2FEks5uYSeegaJpZM4WaTs4>
> .
>

gregorco: In the above code the exception is anticipated and not a real concern.  The informational message isn't necessary, so I can remove it altogether. 
benmccann: 80%+ of Java libraries today use slf4j-api, which is a facade that delegates to other logging libraries. The end user can then pick any logging implementation they want to use and the log messages will be redirected to that underlying implementation. I definitely would not introduce any other library like log4j, logback, etc. because there's such widespread fragmentation between logging implementations that it will cause problems for some subset of users. But as long as you use slf4j-api it will work with any implementation that the user may choose.

The thing I don't like about writing to std out is that the end user has no control over configuring the output
lwhite1: yeah, in hindsight, i don't think we write to sout anywhere unless it's a
stray debugging statement that needs to be deleted. I'm trying to do to
many things at once to provide good feedback this morning.

Still not sure how i feel about adding slf4j as a dependency, or
philosophically how i feel about whether a library like this should be
doing logging.

On Thu, Sep 6, 2018 at 10:42 AM Ben McCann <notifications@github.com> wrote:

> 80%+ of Java libraries today use slf4j-api, which is a facade that
> delegates to other logging libraries. The end user can then pick any
> logging implementation they want to use and the log messages will be
> redirected to that underlying implementation. I definitely would not
> introduce any other library like log4j, logback, etc. because there's such
> widespread fragmentation between logging implementations that it will cause
> problems for some subset of users. But as long as you use slf4j-api it will
> work with any implementation that the user may choose.
>
> The thing I don't like about writing to std out is that the end user has
> no control over configuring the output
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/jtablesaw/tablesaw/pull/377#issuecomment-419119780>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ADRXgihAVaGj8Z2f_Noa08JWjqRsWAqEks5uYTRqgaJpZM4WaTs4>
> .
>

gregorco: The message that I output can simply be removed.  The exception was anticipated so not a real concern or need for action.  

<sub>Sent with <a href="http://githawk.com">GitHawk</a></sub>
gregorco: Thank you @benmccann .  I'll work on these changes.
gregorco: @benmccann , @lwhite1  I submitted the changes, but the travis-ci build failed with no clear reason.  I don't see a way for me to tell it to restart the build.  Are you able to restart the build?
lwhite1: I can do that    Travis is flakey

On Thu, Sep 6, 2018 at 10:39 PM Gregor Okorn <notifications@github.com>
wrote:

> @benmccann <https://github.com/benmccann> , @lwhite1
> <https://github.com/lwhite1> I submitted the changes, but the travis-ci
> build failed with no clear reason. I don't see a way for me to tell it to
> restart the build. Are you able to restart the build?
>
> —
> You are receiving this because you were mentioned.
>
>
> Reply to this email directly, view it on GitHub
> <https://github.com/jtablesaw/tablesaw/pull/377#issuecomment-419302690>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ADRXgjajBuQE2EAaiGyhMHPXv8dkZe5oks5uYdxzgaJpZM4WaTs4>
> .
>

gregorco: Looks like it tried again by itself already?  I saw this morning that the build had completed before your comment that you'd restart it.  Did you restart it earlier, or did it do so by itself?
lwhite1: It ran by itself.  All good. I will look at your pr as soon as time
allows.  Thanks very much

On Fri, Sep 7, 2018 at 9:44 AM Gregor Okorn <notifications@github.com>
wrote:

> Looks like it tried again by itself already? I saw this morning that the
> build had completed before your comment that you'd restart it. Did you
> restart it earlier, or did it do so by itself?
>
> —
> You are receiving this because you were mentioned.
>
>
> Reply to this email directly, view it on GitHub
> <https://github.com/jtablesaw/tablesaw/pull/377#issuecomment-419443868>,
> or mute the thread
> <https://github.com/notifications/unsubscribe-auth/ADRXgk3OeRuv9RFNzKgwGhGPI8q_IQZrks5uYnhOgaJpZM4WaTs4>
> .
>


Patch:
diff --git a/core/src/main/java/tech/tablesaw/api/DateTimeColumn.java b/core/src/main/java/tech/tablesaw/api/DateTimeColumn.java
index b4d12753..377299e8 100644
--- a/core/src/main/java/tech/tablesaw/api/DateTimeColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/DateTimeColumn.java
@@ -14,25 +14,7 @@
 
 package tech.tablesaw.api;
 
-import com.google.common.base.Preconditions;
-import it.unimi.dsi.fastutil.ints.IntComparator;
-import it.unimi.dsi.fastutil.longs.LongArrayList;
-import it.unimi.dsi.fastutil.longs.LongArrays;
-import it.unimi.dsi.fastutil.longs.LongComparator;
-import it.unimi.dsi.fastutil.longs.LongIterator;
-import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
-import it.unimi.dsi.fastutil.longs.LongSet;
-import tech.tablesaw.columns.AbstractColumn;
-import tech.tablesaw.columns.Column;
-import tech.tablesaw.columns.AbstractParser;
-import tech.tablesaw.columns.datetimes.DateTimeColumnFormatter;
-import tech.tablesaw.columns.datetimes.DateTimeColumnType;
-import tech.tablesaw.columns.datetimes.DateTimeFillers;
-import tech.tablesaw.columns.datetimes.DateTimeFilters;
-import tech.tablesaw.columns.datetimes.DateTimeMapFunctions;
-import tech.tablesaw.columns.datetimes.PackedLocalDateTime;
-import tech.tablesaw.selection.Selection;
-import tech.tablesaw.sorting.comparators.DescendingLongComparator;
+import static tech.tablesaw.api.ColumnType.LOCAL_DATE_TIME;
 
 import java.nio.ByteBuffer;
 import java.sql.Timestamp;
@@ -51,13 +33,33 @@ import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
-import static tech.tablesaw.api.ColumnType.LOCAL_DATE_TIME;
+import com.google.common.base.Preconditions;
+
+import it.unimi.dsi.fastutil.ints.IntComparator;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrays;
+import it.unimi.dsi.fastutil.longs.LongComparator;
+import it.unimi.dsi.fastutil.longs.LongIterator;
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongSet;
+import tech.tablesaw.columns.AbstractColumn;
+import tech.tablesaw.columns.AbstractParser;
+import tech.tablesaw.columns.Column;
+import tech.tablesaw.columns.datetimes.DateTimeColumnFormatter;
+import tech.tablesaw.columns.datetimes.DateTimeColumnType;
+import tech.tablesaw.columns.datetimes.DateTimeFillers;
+import tech.tablesaw.columns.datetimes.DateTimeFilters;
+import tech.tablesaw.columns.datetimes.DateTimeMapFunctions;
+import tech.tablesaw.columns.datetimes.PackedLocalDateTime;
+import tech.tablesaw.selection.Selection;
+import tech.tablesaw.sorting.comparators.DescendingLongComparator;
 
 /**
  * A column in a table that contains long-integer encoded (packed) local date-time values
  */
 public class DateTimeColumn extends AbstractColumn<LocalDateTime>
-    implements DateTimeMapFunctions, DateTimeFilters, DateTimeFillers<DateTimeColumn> {
+    implements DateTimeMapFunctions, DateTimeFilters, DateTimeFillers<DateTimeColumn>,
+        CategoricalColumn<LocalDateTime> {
 
     public static final long MISSING_VALUE = DateTimeColumnType.missingValueIndicator();
 
diff --git a/core/src/main/java/tech/tablesaw/api/LongColumn.java b/core/src/main/java/tech/tablesaw/api/LongColumn.java
index a8a48fc2..1452aaad 100644
--- a/core/src/main/java/tech/tablesaw/api/LongColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/LongColumn.java
@@ -348,7 +348,7 @@ public class LongColumn extends NumberColumn<Long> implements CategoricalColumn<
     @Override
     public LongColumn appendCell(final String value, AbstractParser<?> parser) {
         try {
-            return append(parser.parseInt(value));
+            return append(parser.parseLong(value));
         } catch (final NumberFormatException e) {
             throw new NumberFormatException("Error adding value to column " + name()  + ": " + e.getMessage());
         }
diff --git a/core/src/main/java/tech/tablesaw/api/Table.java b/core/src/main/java/tech/tablesaw/api/Table.java
index 09d53d9f..b510353a 100644
--- a/core/src/main/java/tech/tablesaw/api/Table.java
+++ b/core/src/main/java/tech/tablesaw/api/Table.java
@@ -865,12 +865,13 @@ public class Table extends Relation implements Iterable<Row> {
     }
 
     /**
-     * Join on the given {@code columnName}  
-     * @param columnName  The column name to join on
+     * Returns a new DataFrameJoiner initialized with multiple {@code columnNames}
+     * @param columnNames   Name of the columns to join on.
+     * @return              The new DataFrameJoiner
      */
-    public DataFrameJoiner join(String columnName) {
-        return new DataFrameJoiner(this, columnName);
-    }
+    public DataFrameJoiner join(String... columnNames) {
+        return new DataFrameJoiner(this, columnNames);
+	}
 
     public Table missingValueCounts() {
         return summarize(columnNames(), countMissing).apply();
diff --git a/core/src/main/java/tech/tablesaw/index/LongIndex.java b/core/src/main/java/tech/tablesaw/index/LongIndex.java
index 7f0ac956..759e1c2a 100644
--- a/core/src/main/java/tech/tablesaw/index/LongIndex.java
+++ b/core/src/main/java/tech/tablesaw/index/LongIndex.java
@@ -22,6 +22,7 @@ import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import it.unimi.dsi.fastutil.longs.Long2ObjectSortedMap;
 import tech.tablesaw.api.DateTimeColumn;
 import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.LongColumn;
 import tech.tablesaw.columns.datetimes.PackedLocalDateTime;
 import tech.tablesaw.selection.BitmapBackedSelection;
 import tech.tablesaw.selection.Selection;
@@ -68,6 +69,24 @@ public class LongIndex {
         }
         index = new Long2ObjectAVLTreeMap<>(tempMap);
     }
+    
+    public LongIndex(LongColumn column) {
+        int sizeEstimate = Integer.min(1_000_000, column.size() / 100);
+        Long2ObjectOpenHashMap<IntArrayList> tempMap = new Long2ObjectOpenHashMap<>(sizeEstimate);
+        for (int i = 0; i < column.size(); i++) {
+            long value = column.getLong(i);
+            IntArrayList recordIds = tempMap.get(value);
+            if (recordIds == null) {
+                recordIds = new IntArrayList();
+                recordIds.add(i);
+                tempMap.trim();
+                tempMap.put(value, recordIds);
+            } else {
+                recordIds.add(i);
+            }
+        }
+        index = new Long2ObjectAVLTreeMap<>(tempMap);
+    }
 
     private static void addAllToSelection(IntArrayList tableKeys, Selection selection) {
         for (int i : tableKeys) {
diff --git a/core/src/main/java/tech/tablesaw/io/csv/CsvReader.java b/core/src/main/java/tech/tablesaw/io/csv/CsvReader.java
index e522034f..f3631210 100644
--- a/core/src/main/java/tech/tablesaw/io/csv/CsvReader.java
+++ b/core/src/main/java/tech/tablesaw/io/csv/CsvReader.java
@@ -47,6 +47,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import static tech.tablesaw.api.ColumnType.BOOLEAN;
 import static tech.tablesaw.api.ColumnType.DOUBLE;
 import static tech.tablesaw.api.ColumnType.FLOAT;
+import static tech.tablesaw.api.ColumnType.LONG;
 import static tech.tablesaw.api.ColumnType.INTEGER;
 import static tech.tablesaw.api.ColumnType.LOCAL_DATE;
 import static tech.tablesaw.api.ColumnType.LOCAL_DATE_TIME;
@@ -92,7 +93,7 @@ public class CsvReader {
      * the integer test would never be evaluated and all the ints would be read as doubles.
      */
     private List<ColumnType> typeArray =
-            Lists.newArrayList(LOCAL_DATE_TIME, LOCAL_TIME, LOCAL_DATE, BOOLEAN, SHORT, INTEGER, FLOAT, DOUBLE, STRING, TEXT);
+            Lists.newArrayList(LOCAL_DATE_TIME, LOCAL_TIME, LOCAL_DATE, BOOLEAN, SHORT, INTEGER, LONG, FLOAT, DOUBLE, STRING, TEXT);
 
     /**
      * Constructs a CsvReader
diff --git a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
index 4e592e59..919484bc 100644
--- a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
+++ b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
@@ -1,12 +1,14 @@
 package tech.tablesaw.joining;
 
 import com.google.common.collect.Streams;
+
 import tech.tablesaw.api.CategoricalColumn;
 import tech.tablesaw.api.ColumnType;
 import tech.tablesaw.api.DateColumn;
 import tech.tablesaw.api.DateTimeColumn;
 import tech.tablesaw.api.IntColumn;
 import tech.tablesaw.api.LongColumn;
+import tech.tablesaw.api.Row;
 import tech.tablesaw.api.ShortColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.Table;
@@ -26,6 +28,8 @@ import tech.tablesaw.index.StringIndex;
 import tech.tablesaw.selection.BitmapBackedSelection;
 import tech.tablesaw.selection.Selection;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
 public class DataFrameJoiner {
@@ -33,12 +37,23 @@ public class DataFrameJoiner {
     private static final String TABLE_ALIAS = "T";
 
     private final Table table;
-    private final CategoricalColumn<?> column;
+    private final CategoricalColumn<?>[] columns;
+    private final String[] columnNames;
     private AtomicInteger joinTableId = new AtomicInteger(2);
 
-    public DataFrameJoiner(Table table, String column) {
+    /**
+     * Constructor.
+     * @param table       The table to join on
+     * @param columnNames The column names to join on
+     */
+    public DataFrameJoiner(Table table, String... columnNames) {
         this.table = table;
-        this.column = table.categoricalColumn(column);
+        columns = new CategoricalColumn<?>[columnNames.length];
+        this.columnNames = columnNames;
+        for (int i = 0; i < this.columnNames.length; i++) {
+            String colName = this.columnNames[i];
+            this.columns[i] = table.categoricalColumn(colName);
+        }
     }
 
     /**
@@ -61,7 +76,7 @@ public class DataFrameJoiner {
         Table joined = table;
 
         for (Table currT : tables) {
-            joined = joinInternal(joined, currT, column.name(), false, allowDuplicateColumnNames);
+            joined = joinInternal(joined, currT, false, allowDuplicateColumnNames, columnNames);
         }
         return joined;
     }
@@ -72,9 +87,22 @@ public class DataFrameJoiner {
      * @param table2   The table to join with
      * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
      *                 rounding to integers.
+     * @return The resulting table
      */
-    public Table inner(Table table2, String col2Name) {
-        return inner(table2, col2Name, false);
+     public Table inner(Table table2, String col2Name) {
+         return inner(table2, false, col2Name);
+     }
+
+    /**
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
+     *
+     * @param table2    The table to join with
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
+     */
+    public Table inner(Table table2, String[] col2Names) {
+        return inner(table2, false, col2Names);
     }
 
     /**
@@ -85,142 +113,145 @@ public class DataFrameJoiner {
      *                 rounding to integers.
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
      *                                  if {@code true} the join will succeed and duplicate columns are renamed*
+     * @return The resulting table
      */
     public Table inner(Table table2, String col2Name, boolean allowDuplicateColumnNames) {
-        return joinInternal(table2, col2Name, false, allowDuplicateColumnNames);
+        return inner(table2, allowDuplicateColumnNames, col2Name);
     }
 
-    private Table joinInternal(Table table2, String col2Name, boolean outer, boolean allowDuplicates) {
-        return joinInternal(table, table2, col2Name, outer, allowDuplicates);
+    /**
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
+     *
+     * @param table2    The table to join with
+     * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
+     *                                  if {@code true} the join will succeed and duplicate columns are renamed*
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
+     */
+    public Table inner(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+        return inner(table2, false, allowDuplicateColumnNames, col2Names);
     }
 
-    private Table joinInternal(Table table1, Table table2, String col2Name, boolean outer, boolean allowDuplicates) {
+    /**
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
+     *
+     * @param table2    The table to join with
+     * @param outer     True if this join is actually an outer join, left or right or full, otherwise false.
+     * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
+     *                                  if {@code true} the join will succeed and duplicate columns are renamed*
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
+     */
+    public Table inner(Table table2, boolean outer, boolean allowDuplicateColumnNames, String... col2Names) {
+        Table joinedTable;
+        joinedTable = joinInternal(table, table2, outer, allowDuplicateColumnNames, col2Names);
+        return joinedTable;
+    }
 
+    /**
+     * Joins the joiner to the {@code table2}, using the given columns for the second table and returns the resulting table
+     *
+     * @param table1    The base table to join on
+     * @param table2    The table to join with
+     * @param outer     True if this join is actually an outer join, left or right or full, otherwise false.
+     * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
+     *                                  if {@code true} the join will succeed and duplicate columns are renamed*
+     * @param col2Names The columns to join on. If a col2Name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
+     */
+    private Table joinInternal(Table table1, Table table2, boolean outer, boolean allowDuplicates, String... col2Names) {
         if (allowDuplicates) {
-            renameColumnsWithDuplicateNames(table1, table2, col2Name);
+            renameColumnsWithDuplicateNames(table1, table2, col2Names);
         }
-
-        Table result = emptyTableFromColumns(table1, table2, col2Name);
-
-        // Need to use the column from table1 that is the same column originally
-        // defined for this DataFrameJoiner. Column names must be unique within the
-        // same table, so use the original column's name to get the corresponding
-        // column out of the table1 input Table.
-        CategoricalColumn<?> table1Column = table1.categoricalColumn(column.name());
-        ColumnType type = table1Column.type();
-        if (type instanceof DateColumnType) {
-            IntIndex index = new IntIndex(table2.dateColumn(col2Name));
-            DateColumn col1 = (DateColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                int value = col1.getIntInternal(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
+        Table result = emptyTableFromColumns(table1, table2, col2Names);
+        for (Row row : table1) {
+            int ri = row.getRowNumber();
+            Table table1Rows = table1.where(Selection.with(ri));
+            Selection rowBitMapMultiCol = null;
+            for (int i = 0; i < columns.length; i++) {
+
+                // Need to use the column from table1 that is the same column originally
+                // defined for this DataFrameJoiner. Column names must be unique within the
+                // same table, so use the original column's name to get the corresponding
+                // column out of the table1 input Table.
+                CategoricalColumn<?> column = columns[i];
+                CategoricalColumn<?> table1Column = table1.categoricalColumn(column.name());
+
+                ColumnType type = table1Column.type();
+                // relies on both arrays, columns, and col2Names,
+                // having corresponding values at same index
+                String col2Name = col2Names[i];
+                Selection rowBitMapOneCol = null;
+                if (type instanceof DateColumnType) {
+                    IntIndex index = new IntIndex(table2.dateColumn(col2Name));
+                    DateColumn col1 = (DateColumn) table1Column;
+                    int value = col1.getIntInternal(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof DateTimeColumnType) {
+                    LongIndex index = new LongIndex(table2.dateTimeColumn(col2Name));
+                    DateTimeColumn col1 = (DateTimeColumn) table1Column;
+                    long value = col1.getLongInternal(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof TimeColumnType) {
+                    IntIndex index = new IntIndex(table2.timeColumn(col2Name));
+                    TimeColumn col1 = (TimeColumn) table1Column;
+                    int value = col1.getIntInternal(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof StringColumnType) {
+                    StringIndex index = new StringIndex(table2.stringColumn(col2Name));
+                    StringColumn col1 = (StringColumn) table1Column;
+                    String value = col1.get(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof IntColumnType) {
+                    IntIndex index = new IntIndex(table2.intColumn(col2Name));
+                    IntColumn col1 = (IntColumn) table1Column;
+                    int value = col1.getInt(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof LongColumnType) {
+                    LongIndex index = new LongIndex(table2.longColumn(col2Name));
+                    LongColumn col1 = (LongColumn) table1Column;
+                    long value = col1.getLong(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof ShortColumnType) {
+                    ShortIndex index = new ShortIndex(table2.shortColumn(col2Name));
+                    ShortColumn col1 = (ShortColumn) table1Column;
+                    short value = col1.getShort(ri);
+                    rowBitMapOneCol = index.get(value);
                 } else {
-                    crossProduct(result, table1Rows, table2Rows);
+                    throw new IllegalArgumentException(
+                            "Joining is supported on integral, string, and date-like columns. Column "
+                                    + table1Column.name() + " is of type " + table1Column.type());
                 }
-            }
-        } else if (type instanceof DateTimeColumnType) {
-            LongIndex index = new LongIndex(table2.dateTimeColumn(col2Name));
-            DateTimeColumn col1 = (DateTimeColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                long value = col1.getLongInternal(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
-                } else {
-                    crossProduct(result, table1Rows, table2Rows);
-                }
-            }
-        } else if (type instanceof TimeColumnType) {
-            IntIndex index = new IntIndex(table2.timeColumn(col2Name));
-            TimeColumn col1 = (TimeColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                int value = col1.getIntInternal(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
-                } else {
-                    crossProduct(result, table1Rows, table2Rows);
-                }
-            }
-        } else if (type instanceof StringColumnType) {
-            StringIndex index = new StringIndex(table2.stringColumn(col2Name));
-            StringColumn col1 = (StringColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                String value = col1.get(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
-                } else {
-                    crossProduct(result, table1Rows, table2Rows);
-                }
-            }
-        } else if (type instanceof IntColumnType) {
-            IntIndex index = new IntIndex(table2.intColumn(col2Name));
-            IntColumn col1 = (IntColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                int value = col1.getInt(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
-                } else {
-                    crossProduct(result, table1Rows, table2Rows);
-                }
-            }
-        } else if (type instanceof LongColumnType) {
-            LongIndex index = new LongIndex(table2.intColumn(col2Name));
-            LongColumn col1 = (LongColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                long value = col1.getLong(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
-                } else {
-                    crossProduct(result, table1Rows, table2Rows);
+                // combine Selection's into one big AND Selection
+                if (rowBitMapOneCol != null) {
+                    rowBitMapMultiCol = 
+                            rowBitMapMultiCol != null
+                            ? rowBitMapMultiCol.and(rowBitMapOneCol)
+                            : rowBitMapOneCol;
                 }
             }
-        } else if (type instanceof ShortColumnType) {
-            ShortIndex index = new ShortIndex(table2.shortColumn(col2Name));
-            ShortColumn col1 = (ShortColumn) table1Column;
-            for (int i = 0; i < col1.size(); i++) {
-                short value = col1.getShort(i);
-                Table table1Rows = table1.where(Selection.with(i));
-                Table table2Rows = table2.where(index.get(value));
-                table2Rows.removeColumns(col2Name);
-                if (outer && table2Rows.isEmpty()) {
-                    withMissingLeftJoin(result, table1Rows);
-                } else {
-                    crossProduct(result, table1Rows, table2Rows);
-                }
+
+            Table table2Rows = table2.where(rowBitMapMultiCol);
+            table2Rows.removeColumns(col2Names);
+            if (outer && table2Rows.isEmpty()) {
+                withMissingLeftJoin(result, table1Rows);
+            } else {
+                crossProduct(result, table1Rows, table2Rows);
             }
-        } else {
-            throw new IllegalArgumentException(
-                    "Joining is supported on integer, string, and date-like columns. Column "
-                            + column.name() + " is of type " + column.type());
         }
         return result;
     }
 
-    private void renameColumnsWithDuplicateNames(Table table1, Table table2, String col2Name) {
+    private void renameColumnsWithDuplicateNames(Table table1, Table table2, String... col2Names) {
         String table2Alias = TABLE_ALIAS + joinTableId.getAndIncrement();
-
+        List<String> list = Arrays.asList(col2Names);
         for (Column<?> table2Column : table2.columns()) {
             String columnName = table2Column.name();
-            if (table1.columnNames().contains(columnName)
-                    && !columnName.equalsIgnoreCase(col2Name)) {
+            if (table1.columnNames().stream().anyMatch(columnName::equalsIgnoreCase)
+                    && !(list.stream().anyMatch(columnName::equalsIgnoreCase))) {
                 table2Column.setName(newName(table2Alias, columnName));
             }
         }
@@ -231,128 +262,133 @@ public class DataFrameJoiner {
     }
 
     /**
-     * Joins to the given tables assuming that they have a column of the name we're joining on
+     * Full outer join to the given tables assuming that they have a column of the name we're joining on
      *
      * @param tables The tables to join with
+     * @return The resulting table
      */
     public Table fullOuter(Table... tables) {
         return fullOuter(false, tables);
     }
 
     /**
-     * Joins to the given tables assuming that they have a column of the name we're joining on
+     * Full outer join to the given tables assuming that they have a column of the name we're joining on
      *
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
      *                                  if {@code true} the join will succeed and duplicate columns are renamed*
      * @param tables The tables to join with
+     * @return The resulting table
      */
     public Table fullOuter(boolean allowDuplicateColumnNames, Table... tables) {
-        Table joined = table;
-        for (Table table2 : tables) {
-            joined = fullOuter(table2, column.name(), allowDuplicateColumnNames);
+       Table joined = table;
+
+        for (Table currT : tables) {
+            joined = fullOuter(joined, currT, allowDuplicateColumnNames, columnNames);
         }
         return joined;
     }
 
     /**
-     * Joins the joiner to the table2, using the given column for the second table and returns the resulting table
+     * Full outer join the joiner to the table2, using the given column for the second table and returns the resulting table
      *
      * @param table2   The table to join with
      * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
      *                 rounding to integers.
+     * @return The resulting table
      */
     public Table fullOuter(Table table2, String col2Name) {
-        return fullOuter(table2, col2Name, false);
+        return fullOuter(table, table2, false, col2Name);
     }
 
     /**
-     * Joins the joiner to the table2, using the given column for the second table and returns the resulting table
+     * Full outer join table1 to table2, using the given columns for the second table and returns the resulting table
      *
-     * @param table2   The table to join with
-     * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
-     *                 rounding to integers.
+     * @param table1    The base table to join on
+     * @param table2    The table to join with
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
-     *                                  if {@code true} the join will succeed and duplicate columns are renamed
+     *                                  if {@code true} the join will succeed and duplicate columns are renamed*
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
      */
-    public Table fullOuter(Table table2, String col2Name, boolean allowDuplicateColumnNames) {
-        Table result = joinInternal(table2, col2Name, true, allowDuplicateColumnNames);
+    public Table fullOuter(Table table1, Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+        Table result = joinInternal(table1, table2, true, allowDuplicateColumnNames, col2Names);
 
         Selection selection = new BitmapBackedSelection();
-        ColumnType type = column.type();
-
-        if (type instanceof DateColumnType) {
-            IntIndex index = new IntIndex(result.dateColumn(col2Name));
-            DateColumn col2 = (DateColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                int value = col2.getIntInternal(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
-                }
-            }
-        } else if (type instanceof DateTimeColumnType) {
-            LongIndex index = new LongIndex(result.dateTimeColumn(col2Name));
-            DateTimeColumn col2 = (DateTimeColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                long value = col2.getLongInternal(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
-                }
-            }
-        } else if (type instanceof TimeColumnType) {
-            IntIndex index = new IntIndex(result.timeColumn(col2Name));
-            TimeColumn col2 = (TimeColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                int value = col2.getIntInternal(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
-                }
-            }
-        } else if (type instanceof StringColumnType) {
-            StringIndex index = new StringIndex(result.stringColumn(col2Name));
-            StringColumn col2 = (StringColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                String value = col2.get(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
-                }
-            }
-        } else if (type instanceof IntColumnType) {
-            IntIndex index = new IntIndex(result.intColumn(col2Name));
-            IntColumn col2 = (IntColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                int value = col2.getInt(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
+        for (Row row : table2) {
+            int ri = row.getRowNumber();
+            Selection rowBitMapMultiCol = null;
+
+            for (int i = 0; i < columns.length; i++) {
+
+                // Need to use the column from table1 that is the same column originally
+                // defined for this DataFrameJoiner. Column names must be unique within the
+                // same table, so use the original column's name to get the corresponding
+                // column out of the table1 input Table.
+                CategoricalColumn<?> column = columns[i];
+                CategoricalColumn<?> table1Column = table1.categoricalColumn(column.name());
+
+                ColumnType type = table1Column.type();
+                // relies on both arrays, columns, and col2Names,
+                // having corresponding values at same index
+                String col2Name = col2Names[i];
+                Selection rowBitMapOneCol = null;
+
+                if (type instanceof DateColumnType) {
+                    IntIndex index = new IntIndex(result.dateColumn(col2Name));
+                    DateColumn col2 = (DateColumn) table2.column(col2Name);
+                    int value = col2.getIntInternal(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof DateTimeColumnType) {
+                    LongIndex index = new LongIndex(result.dateTimeColumn(col2Name));
+                    DateTimeColumn col2 = (DateTimeColumn) table2.column(col2Name);
+                    long value = col2.getLongInternal(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof TimeColumnType) {
+                    IntIndex index = new IntIndex(result.timeColumn(col2Name));
+                    TimeColumn col2 = (TimeColumn) table2.column(col2Name);
+                    int value = col2.getIntInternal(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof StringColumnType) {
+                    StringIndex index = new StringIndex(result.stringColumn(col2Name));
+                    StringColumn col2 = (StringColumn) table2.column(col2Name);
+                    String value = col2.get(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof IntColumnType) {
+                    IntIndex index = new IntIndex(result.intColumn(col2Name));
+                    IntColumn col2 = (IntColumn) table2.column(col2Name);
+                    int value = col2.getInt(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof LongColumnType) {
+                    LongIndex index = new LongIndex(result.longColumn(col2Name));
+                    LongColumn col2 = (LongColumn) table2.column(col2Name);
+                    long value = col2.getLong(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else if (type instanceof ShortColumnType) {
+                    ShortIndex index = new ShortIndex(result.shortColumn(col2Name));
+                    ShortColumn col2 = (ShortColumn) table2.column(col2Name);
+                    short value = col2.getShort(ri);
+                    rowBitMapOneCol = index.get(value);
+                } else {
+                    throw new IllegalArgumentException(
+                            "Joining is supported on integral, string, and date-like columns. Column "
+                                    + table1Column.name() + " is of type " + table1Column.type());
                 }
-            }
-        } else if (type instanceof LongColumnType) {
-            LongIndex index = new LongIndex(result.intColumn(col2Name));
-            LongColumn col2 = (LongColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                long value = col2.getLong(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
+                // combine Selections into one big AND Selection
+                if (rowBitMapOneCol != null) {
+                    rowBitMapMultiCol = rowBitMapMultiCol != null
+                            ? rowBitMapMultiCol.and(rowBitMapOneCol)
+                            : rowBitMapOneCol;
                 }
             }
-        } else if (type instanceof ShortColumnType) {
-            ShortIndex index = new ShortIndex(result.shortColumn(col2Name));
-            ShortColumn col2 = (ShortColumn) table2.column(col2Name);
-            for (int i = 0; i < col2.size(); i++) {
-                short value = col2.getShort(i);
-                if (index.get(value).isEmpty()) {
-                    selection.add(i);
-                }
+            if (rowBitMapMultiCol.isEmpty()) {
+                selection.add(ri);
             }
-        } else {
-            throw new IllegalArgumentException(
-                    "Joining is supported on numeric, string, and date-like columns. Column "
-                            + column.name() + " is of type " + column.type());
         }
-
         Table table2OnlyRows = table2.where(selection);
-        CategoricalColumn<?> joinColumn = table2OnlyRows.categoricalColumn(col2Name);
-        table2OnlyRows.removeColumns(joinColumn);
-        withMissingRightJoin(result, joinColumn, table2OnlyRows);        
+        List<CategoricalColumn<?>> joinColumns = table2OnlyRows.categoricalColumns(col2Names);
+        table2OnlyRows.removeColumns(col2Names);
+        withMissingRightJoin(result, joinColumns, table2OnlyRows);
         return result;
     }
 
@@ -360,6 +396,7 @@ public class DataFrameJoiner {
      * Joins to the given tables assuming that they have a column of the name we're joining on
      *
      * @param tables The tables to join with
+     * @return The resulting table
      */
     public Table leftOuter(Table... tables) {
         return leftOuter(false, tables);
@@ -371,43 +408,59 @@ public class DataFrameJoiner {
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
      *                                  if {@code true} the join will succeed and duplicate columns are renamed*
      * @param tables The tables to join with
+     * @return The resulting table
      */
     public Table leftOuter(boolean allowDuplicateColumnNames, Table... tables) {
         Table joined = table;
         for (Table table2 : tables) {
-          joined = leftOuter(table2, column.name(), allowDuplicateColumnNames);
+            joined = leftOuter(table2, allowDuplicateColumnNames, columnNames);
         }
         return joined;
     }
 
+    /**
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
+     *
+     * @param table2    The table to join with
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
+     */
+    public Table leftOuter(Table table2, String[] col2Names) {
+        return leftOuter(table2, false, col2Names);
+    }
+
     /**
      * Joins the joiner to the table2, using the given column for the second table and returns the resulting table
      *
      * @param table2   The table to join with
      * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
      *                 rounding to integers.
+     * @return The resulting table
      */
     public Table leftOuter(Table table2, String col2Name) {
-        return leftOuter(table2, col2Name, false);
+        return leftOuter(table2, false, col2Name);
     }
 
     /**
-     * Joins the joiner to the table2, using the given column for the second table and returns the resulting table
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
      *
-     * @param table2   The table to join with
-     * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
-     *                 rounding to integers.
+     * @param table2    The table to join with
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
      *                                  if {@code true} the join will succeed and duplicate columns are renamed
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
      */
-    public Table leftOuter(Table table2, String col2Name, boolean allowDuplicateColumnNames) {
-        return joinInternal(table2, col2Name, true, allowDuplicateColumnNames);
+    public Table leftOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+        return joinInternal(table, table2, true, allowDuplicateColumnNames, col2Names);
     }
 
     /**
      * Joins to the given tables assuming that they have a column of the name we're joining on
      *
      * @param tables The tables to join with
+     * @return The resulting table
      */
     public Table rightOuter(Table... tables) {
         return rightOuter(false, tables);
@@ -416,14 +469,15 @@ public class DataFrameJoiner {
     /**
      * Joins to the given tables assuming that they have a column of the name we're joining on
      *
-     * @param tables The tables to join with
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
      *                                  if {@code true} the join will succeed and duplicate columns are renamed
+     * @param tables The tables to join with
+     * @return The resulting table
      */
     public Table rightOuter(boolean allowDuplicateColumnNames, Table... tables) {
         Table joined = table;
         for (Table table2 : tables) {
-          joined = rightOuter(table2, column.name(), allowDuplicateColumnNames);
+            joined = rightOuter(table2, allowDuplicateColumnNames, columnNames);
         }
         return joined;
     }
@@ -434,42 +488,63 @@ public class DataFrameJoiner {
      * @param table2   The table to join with
      * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
      *                 rounding to integers.
+     * @return The resulting table
      */
     public Table rightOuter(Table table2, String col2Name) {
-        return rightOuter(table2, col2Name, false);
+        return rightOuter(table2, false, col2Name);
     }
 
     /**
-     * Joins the joiner to the table2, using the given column for the second table and returns the resulting table
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
+     *
+     * @param table2    The table to join with
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                  rounding to integers.
+     * @return The resulting table
+     */
+    public Table rightOuter(Table table2, String[] col2Names) {
+        return rightOuter(table2, false, col2Names);
+    }
+
+    /**
+     * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
      *
      * @param table2   The table to join with
-     * @param col2Name The column to join on. If col2Name refers to a double column, the join is performed after
-     *                 rounding to integers.
      * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
      *                                  if {@code true} the join will succeed and duplicate columns are renamed
+     * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
+     *                 rounding to integers.
+     * @return The resulting table
      */
 
-    public Table rightOuter(Table table2, String col2Name, boolean allowDuplicateColumnNames) {
-        Table leftOuter = table2.join(col2Name).leftOuter(table, column.name(), allowDuplicateColumnNames);
+    public Table rightOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+        Table leftOuter = table2.join(col2Names).leftOuter(table, allowDuplicateColumnNames, columnNames);
 
         // reverse the columns
         Table result = Table.create(leftOuter.name());
+        // loop on table that was originally first (left) and add the left-joined matching columns by name
         for (String name : table.columnNames()) {
-            result.addColumns(leftOuter.column(name));
+            try {
+                result.addColumns(leftOuter.column(name));
+            } catch (IllegalStateException e) {
+                // Can ignore this exception as it is anticipated.
+                // NOTE: DataFrameJoiner.rightOuter(): skipping left table's column,'"
+                //     +name+"', in favor of right table's matching column that was kept in join operation.");
+            }
         }
         for (String name : table2.columnNames()) {
-            if (!result.columnNames().contains(name)) {
+            if (!result.columnNames().stream().anyMatch(name::equalsIgnoreCase)) {
                 result.addColumns(leftOuter.column(name));
             }
         }
         return result;
     }
 
-    private Table emptyTableFromColumns(Table table1, Table table2, String col2Name) {
+    private Table emptyTableFromColumns(Table table1, Table table2, String... col2Names) {
         Column<?>[] cols = Streams.concat(
                 table1.columns().stream(),
-                table2.columns().stream().filter(c -> !c.name().equalsIgnoreCase(col2Name))
-        ).map(col -> col.emptyCopy(col.size())).toArray(Column[]::new);
+                table2.columns().stream().filter(c -> !Arrays.asList(col2Names).stream().anyMatch(c.name()::equalsIgnoreCase))
+                ).map(col -> col.emptyCopy(col.size())).toArray(Column[]::new);
         return Table.create(table1.name(), cols);
     }
 
@@ -511,21 +586,27 @@ public class DataFrameJoiner {
      * Adds rows to destination for each row in the joinColumn and table2
      */
     @SuppressWarnings({"rawtypes", "unchecked"})
-    private void withMissingRightJoin(Table destination, CategoricalColumn joinColumn, Table table2) {
+    private void withMissingRightJoin(Table destination, List<CategoricalColumn<?>> joinColumns, Table table2) {
         int t2StartCol = destination.columnCount() - table2.columnCount();
         for (int c = 0; c < destination.columnCount(); c++) {
-            if (destination.column(c).name().equalsIgnoreCase(joinColumn.name())) {
-                destination.column(c).append(joinColumn);
-                continue;
+            boolean addedJoinColumns = false;
+            for (CategoricalColumn joinColumn : joinColumns) {
+                if (destination.column(c).name().equalsIgnoreCase(joinColumn.name())) {
+                    destination.column(c).append(joinColumn);
+                    addedJoinColumns = true;
+                }
             }
-            if (c < t2StartCol) {
-                for (int r2 = 0; r2 < table2.rowCount(); r2++) {
-                    destination.column(c).appendMissing();
+            if (!addedJoinColumns) {
+                if (c < t2StartCol) {
+                    for (int r2 = 0; r2 < table2.rowCount(); r2++) {
+                        destination.column(c).appendMissing();
+                    }
+                } else {
+                    Column t2Col = table2.column(c - t2StartCol);
+                    destination.column(c).append(t2Col);
                 }
-            } else {
-                Column t2Col = table2.column(c - t2StartCol);
-                destination.column(c).append(t2Col);
             }
         }
     }
+
 }
diff --git a/core/src/main/java/tech/tablesaw/table/Relation.java b/core/src/main/java/tech/tablesaw/table/Relation.java
index c037ba1d..9f9cf030 100644
--- a/core/src/main/java/tech/tablesaw/table/Relation.java
+++ b/core/src/main/java/tech/tablesaw/table/Relation.java
@@ -383,6 +383,18 @@ public abstract class Relation {
         return (CategoricalColumn<?>) column(columnNumber);
     }
 
+    /**
+     * Returns the columns whose names are given in the input array
+     */
+    public List<CategoricalColumn<?>> categoricalColumns(String... columnName) {
+        List<CategoricalColumn<?>> cols = new ArrayList<>(columnName.length);
+        for (String aColumnName : columnName) {
+            cols.add(categoricalColumn(aColumnName));
+        }
+        return cols;
+    }
+
+    
     /**
      * Returns the column with the given name cast to a NumberColumn
      * <p>


Test patch:
diff --git a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
index 35d3ec6f..2f6b60b7 100644
--- a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
+++ b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
@@ -67,6 +67,30 @@ public class DataFrameJoinerTest {
                 "5.0,King"),
             "Dogs");
 
+    private static final Table DOUBLE_INDEXED_CATS = Table.read().csv(Joiner.on(System.lineSeparator()).join(
+            "ID,Cat Name",
+                    "1.0,Spot2",
+                    "2.0,Fido",
+                    "6.0,Sasha",
+                    "8.0,King2"),
+            "Cats");
+
+    private static final Table DOUBLE_INDEXED_FISH = Table.read().csv(Joiner.on(System.lineSeparator()).join(
+            "ID,Fish Name",
+                    "11.0,Spot3",
+                    "2.0,Fido",
+                    "4.0,Sasha",
+                    "6.0,King2"),
+            "Fish");
+
+    private static final Table DOUBLE_INDEXED_MICE = Table.read().csv(Joiner.on(System.lineSeparator()).join(
+            "ID,Mice_Name",
+                    "2.0,Jerry",
+                    "3.0,Fido",
+                    "6.0,Sasha",
+                    "9.0,Market"),
+            "Mice");
+
     private static final Table DUPLICATE_COL_NAME_DOGS = Table.read().csv(Joiner.on(System.lineSeparator()).join(
                 "ID,Dog Name, Good",
                 "1.0,Spot,true",
@@ -79,70 +103,275 @@ public class DataFrameJoinerTest {
                 "5.0,King,false"),
             "Dogs");
 
+    private static Table createHOUSE() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Style,Bedrooms,BuildDate,Owner",
+                        "Colonial,3,1976-06-02,Smith",
+                        "Gambrel,4,1982-11-18,Jones",
+                        "Contemporary,5,1980-03-24,White",
+                        "Split,2,1970-09-30,Brown"),
+                "House");
+    }
+
+    private static Table createBOAT() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Type,Bedrooms,SoldDate,Owner",
+                        "Yacht,2,1970-02-03,Jones",
+                        "Dinghy,0,1988-12-12,Trump",
+                        "HouseBoat,3,1981-04-21,Smith",
+                        "Contemporary,5,1980-05-17,White",
+                        "Cruise,200,1989-01-23,Brown"),
+                "Boat");
+    }
+
+    private static Table createHOUSE10() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Style,Bedrooms,BuildDate,Owner",
+                        "Colonial,3,1976-06-02,Smith",
+                        "Gambrel,4,1982-11-18,Jones",
+                        "Contemporary,5,1980-03-24,White",
+                        "Ranch,4,1982-11-18,Black",
+                        "Victorian,5,1980-03-24,Red",
+                        "Split,2,1970-09-30,Brown"),
+                "House10");
+    }
+
+    private static Table createBOAT10() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Type,Bedrooms,SoldDate,Owner",
+                        "Yacht,2,1970-02-03,Jones",
+                        "Dinghy,0,1988-12-12,White",
+                        "HouseBoat,3,1981-04-21,Smith",
+                        "Contemporary,5,1980-05-17,White",
+                        "Paddleboat,3,1981-04-21,Smith",
+                        "Rowboat,5,1980-05-17,White",
+                        "Sailboat,4,1980-05-17,Black",
+                        "Cruise,200,1989-01-23,Brown"),
+                "Boat10");
+    }
+
+    private static Table createBEDANDBREAKFAST() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Design,Bedrooms,SoldDate,Owner",
+                        "Colonial,5,1980-05-17,Smith",
+                        "Gambrel,4,1982-11-18,Jones",
+                        "Contemporary,5,1980-03-24,White",
+                        "Split,2,1970-09-30,Brown"),
+                "BedAndBreakfast");
+    }
+
     private static Table createSTUDENT() {
         return Table.read().csv(Joiner.on(System.lineSeparator()).join(
                 "ID,FirstName,LastName,City,State,Age,USID,GradYear",
-                "1,Bob,Barney,Burke,VA,17,11122,2019",
-                "2,Chris,Cabello,Canyonville,OR,17,22224,2019",
-                "3,Dan,Dirble,Denver,CO,17,33335,2020",
-                "4,Edward,Earhardt,Easterly,WA,18,44339,2021",
-                "5,Frank,Farnsworth,Fredriksburg,VA,18,55338,2019",
-                "6,George,Gabral,Garrisburg,MD,19,66337,2020",
-                "7,Michael,Marbury,Milton,NH,19,77330,2020",
-                "8,Robert,Riley,Roseburg,OR,19,88836,2020",
-                "9,Bob,Earhardt,Milton,NH,50,93333,2019",
-                "10,Dan,Gabral,Easterly,WA,20,13333,2020"),
-            "Student");
+                        "1,Bob,Barney,Burke,VA,20,11122,2019",
+                        "2,Chris,Cabello,Canyonville,OR,20,22224,2019",
+                        "3,Dan,Dirble,Denver,CO,21,33335,2020",
+                        "4,Edward,Earhardt,Easterly,WA,21,44339,2021",
+                        "5,Frank,Farnsworth,Fredriksburg,VA,22,55338,2019",
+                        "6,George,Gabral,Garrisburg,MD,22,66337,2020",
+                        "7,Michael,Marbury,Milton,NH,23,77330,2020",
+                        "8,Robert,Riley,Roseburg,OR,23,88836,2020",
+                        "9,Bob,Earhardt,Milton,NH,24,93333,2019",
+                        "10,Dan,Gabral,Easterly,WA,24,13333,2020"),
+                "Student");
     }
 
     private static Table createINSTRUCTOR() {
         return Table.read().csv(Joiner.on(System.lineSeparator()).join(
-                "ID,First,Last,Title,City,State,Age,USID,GradYear",
-                "1,Bob,Cabello,Prof,Burke,VA,20,11333,2019",
-                "2,Chris,Barney,TA,Canyonville,OR,20,22334,2019",
-                "3,Dan,Earhardt,Instructor,Denver,CO,22,33335,2020",
-                "4,Edward,Dirble,Prof,Easterly,WA,22,43339,2020",
-                "5,Farnsworth,Frank,Prof,Fredriksburg,VA,22,55338,2019",
-                "6,Gabral,George,TA,Garrisburg,MD,18,66337,2019",
-                "7,Robert,Marbury,TA,Msilton,NH,23,73330,2020",
-                "8,Michael,Riley,TA,Roseburg,OR,23,88336,2020",
-                "9,Bob,Riley,Prof,Milton,NH,50,99933,2020",
-                "10,Earhardt,Gabral,Prof,Easterly,WA,24,13333,2019"),
+            "ID,First,Last,Title,City,State,Age,USID,GradYear",
+                    "1,Bob,Cabello,Prof,Burke,VA,20,11333,2019",
+                    "2,Chris,Barney,TA,Canyonville,OR,20,22334,2019",
+                    "3,Dan,Earhardt,Instructor,Denver,CO,19,33335,2020",
+                    "4,Edward,Dirble,Prof,Easterly,WA,22,43339,2020",
+                    "5,Farnsworth,Frank,Prof,Fredriksburg,VA,18,55338,2019",
+                    "6,Gabral,George,TA,Garrisburg,MD,20,66337,2019",
+                    "7,Robert,Marbury,TA,Msilton,NH,22,73330,2020",
+                    "8,Michael,Riley,TA,Roseburg,OR,22,88336,2020",
+                    "9,Bob,Riley,Prof,Milton,NH,19,99933,2020",
+                    "10,Earhardt,Gabral,Prof,Easterly,WA,21,13333,2019"),
             "Instructor");
     }
 
     private static Table createDEPTHEAD() {
         return Table.read().csv(Joiner.on(System.lineSeparator()).join(
-                    "ID,First,Last,Dept,City,State,Age",
-                    "1,John,Cabello,ComputerScience,Burke,VA,20",
-                    "2,Samantha,Barney,Writing,Canyonville,OR,18",
-                    "3,Mark,Earhardt,Mathematics,Denver,CO,35",
-                    "4,Christie,Dirble,Architecture,Easterly,WA,36",
-                    "5,Bhawesh,Frank,Psychology,Fredriksburg,VA,37",
-                    "6,Robert,George,Sociology,Garrisburg,MD,38",
-                    "7,George,Marbury,Physics,Msilton,NH,39",
-                    "8,Zhongyu,Riley,Chemistry,Roseburg,OR,40",
-                    "9,Laura,Riley,Economics,Milton,NH,50",
-                    "10,Sally,Gabral,Marketing,Easterly,WA,42"),
+                "ID,First,Last,Dept,City,State,Age",
+                        "1,John,Cabello,ComputerScience,Burke,VA,20",
+                        "2,Samantha,Barney,Writing,Canyonville,OR,21",
+                        "3,Mark,Earhardt,Mathematics,Denver,CO,22",
+                        "4,Christie,Dirble,Architecture,Easterly,WA,23",
+                        "5,Bhawesh,Frank,Psychology,Fredriksburg,VA,24",
+                        "6,Robert,George,Sociology,Garrisburg,MD,25",
+                        "7,George,Marbury,Physics,Msilton,NH,26",
+                        "8,Zhongyu,Riley,Chemistry,Roseburg,OR,27",
+                        "9,Laura,Riley,Economics,Milton,NH,28",
+                        "10,Sally,Gabral,Marketing,Easterly,WA,29"),
                 "DepartmentHead");
     }
 
     private static Table createCLASS() {
         return Table.read().csv(Joiner.on(System.lineSeparator()).join(
                 "ID,ClassType,Name,Level,Description,StartDate,EndDate,Completed,Age",
-                "1001,Math,Calculus,101,Newton math,2018-09-20,2018-12-17,false,20",
-                "1002,Math,Calculus,102,Newton math,2019-01-06,2019-03-06,false,18",
-                "1003,Math,Calculus,103,Newton math,2019-03-10,2019-06-17,false,18",
-                "1004,Writing,Composition,101,Writing papers,2018-09-20,2018-12-17,false,23",
-                "1005,Writing,Composition,102,Writing papers,2019-01-06,2019-03-07,false,24",
-                "1006,Software,Programming,101,Programming basics,2018-09-22,2018-12-15,false,25",
-                "1007,Software,Programming,102,Programming basics,2019-01-05,2019-03-07,false,26",
-                "1008,Economics,Microeconomics,101,Basic micro economics,2018-09-20,2018-12-17,false,27",
-                "1009,Economics,Microeconomics,102,Basic micro economics,2018-01-05,2019-03-07,false,28",
-                "1010,Literature,Shakespeare,101,Understanding Shakespeare,2018-09-20,2018-12-17,false,50"),
+                        "1001,Math,Calculus,101,Newton math,2018-09-20,2018-12-17,false,16",
+                        "1002,Math,Calculus,102,Newton math,2019-01-06,2019-03-06,false,17",
+                        "1003,Math,Calculus,103,Newton math,2019-03-10,2019-06-17,false,18",
+                        "1004,Writing,Composition,101,Writing papers,2018-09-20,2018-12-17,false,19",
+                        "1005,Writing,Composition,102,Writing papers,2019-01-06,2019-03-07,false,20",
+                        "1006,Software,Programming,101,Programming basics,2018-09-22,2018-12-15,false,21",
+                        "1007,Software,Programming,102,Programming basics,2019-01-05,2019-03-07,false,22",
+                        "1008,Economics,Microeconomics,101,Basic micro economics,2018-09-20,2018-12-17,false,23",
+                        "1009,Economics,Microeconomics,102,Basic micro economics,2018-01-05,2019-03-07,false,24",
+                        "1010,Literature,Shakespeare,101,Understanding Shakespeare,2018-09-20,2018-12-17,false,25"),
             "Class");
     }
 
+    private static Table createANIMALHOMES() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+            "Animal,Name,Home,Age,MoveInDate",
+                    "Pig,Bob,Sty,5,5/5/2018",
+                    "Horse,James,Field,5,6/6/2018",
+                    "Goat,Samantha,Tree,10,7/7/2018",
+                    "Tigon,Rudhrani,Jungle,2,2/2/2018",
+                    "Chicken,Chuck,Pen,2,1/1/2018",
+                    "Squirrel,Sidney,Tree,10,3/3/2018",
+                    "Fox,Frankie,Forest,10,9/19/2018",
+                    "Rabbit,Taylor,Forest,10,4/4/2018"),
+            "Animal Homes");
+    }
+
+    private static Table createTREE() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Name,Home,Age",
+                        "Cherry,Frontyard,2",
+                        "Walnut,Field,3",
+                        "Birch,Forest,4",
+                        "Tallgreen,Jungle,5",
+                        "Apple,Orchard,6",
+                        "Orange,Orchard,9",
+                        "Hemlock,Forest,10"),
+                "Tree");
+    }
+
+    private static Table createFLOWER() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Name,Home,Age,Color",
+                        "Lily,Backyard,2,White",
+                        "VenusFlyTrap,Swamp,2,White",
+                        "Rose,Frontyard,4,Red",
+                        "Pansie,Meadow,5,Blue",
+                        "Daisy,Meadow,6,Yellow",
+                        "Dandelion,Field,7,Yellow",
+                        "Violet,Forest,10,Blue"),
+                "Flower");
+    }
+
+    private static Table createDOUBLEINDEXEDPEOPLENicknameDwellingYearsMoveInDate() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "ID,Nickname,Dwelling,Years,MoveInDate",
+                        "1.0,Bob,Jungle,5,2/2/2018",
+                        "2.0,James,Field,5,6/6/2018",
+                        "3.0,David,Jungle,5,6/6/2018",
+                        "4.0,Marty,Forest,10,2/2/2018",
+                        "5.0,Tarzan,Jungle,10,7/7/2018",
+                        "6.0,Samantha,Tree,10,5/5/2018"),
+                "People - Nickname Dwelling Years MoveInDate");
+    }
+
+    private static Table createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "ID,Name,Dwelling,Years,MoveInDate",
+                        "1.0,Bob,Jungle,5,2/2/2018",
+                        "2.0,James,Field,5,6/6/2018",
+                        "3.0,David,Jungle,5,6/6/2018",
+                        "4.0,Marty,Jungle,10,2/2/2018",
+                        "5.0,Tarzan,Jungle,10,7/7/2018",
+                        "6.0,Samantha,Tree,10,5/5/2018"),
+                "People - Name Dwelling Years MoveInDate");
+    }
+
+    private static Table createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+            "ID,Name,HOME,Age,MoveInDate",
+                    "1.0,Bob,Jungle,5,2/2/2018",
+                    "2.0,James,Field,5,6/6/2018",
+                    "3.0,David,Jungle,5,6/6/2018",
+                    "4.0,Marty,Jungle,10,2/2/2018",
+                    "5.0,Tarzan,Jungle,10,7/7/2018",
+                    "6.0,Samantha,Forest,10,5/5/2018"),
+            "People - Name Home Age MoveInDate");
+    }
+
+    private static Table createFOOTBALLSCHEDULE() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "TeamName,PlayDate,PlayTime,Location,HomeGame",
+                        "Lancers,2018-09-10,15:30,Springfield,true",
+                        "Tigers,2018-09-12,15:00,Detroit,false",
+                        "Patriots,2018-09-14,14:30,Boston,true",
+                        "Ravens,2018-09-10,12:30,Baltimore,true"),
+                "FootballSchedule");
+    }
+
+    private static Table createSOCCERSCHEDULE() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Mascot,PlayDate,PlayTime,Place",
+                        "Steelers,2018-09-10,15:30,Pittsburgh",
+                        "Dolphins,2018-09-12,15:00,Miami",
+                        "Patriots,2018-09-13,14:30,Boston",
+                        "Yankees,2018-09-10,12:00,NewYorkCity"),
+                "SoccerSchedule");
+    }
+
+    private static Table createFOOTBALLSCHEDULEDateTime() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "TeamName,PlayDateTime,Location,HomeGame,SeasonRevenue,AllTimeRevenue",
+                        "Lancers,2018-09-10T15:30,Springfield,true,2000000000,8500000000000",
+                        "Tigers,2018-09-12T15:00,Detroit,false,1500000000,9000000000000",
+                        "Patriots,2018-09-14T14:30,Boston,true,1400000000,8200000000000",
+                        "Ravens,2018-09-10T12:30,Baltimore,true,2000000000,7000000000000"),
+                "FootballSchedule2");
+    }
+
+    private static Table createBASEBALLSCHEDULEDateTime() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "TeamName,PlayDateTime,Location,HomeGame,SeasonRevenue,AllTimeRevenue",
+                        "RedSox,2018-09-10T15:30,Springfield,false,2000000000,7000000000000",
+                        "Marlins,2018-09-12T15:00,Detroit,true,1500000000,8500000000000",
+                        "Mariners,2018-09-14T14:30,Boston,true,1400000000,9000000000000",
+                        "Ravens,2018-09-10T12:30,Baltimore,false,2000000000,7000000000000"),
+                "FootballSchedule2");
+    }
+
+    private static Table createSOCCERSCHEDULEDateTime() {
+        return Table.read().csv(Joiner.on(System.lineSeparator()).join(
+                "Mascot,PlayDateTime,Place,SeasonRevenue,AllTimeRevenue",
+                        "Steelers,2018-09-10T15:30,Pittsburgh,2000000000,7500000000000",
+                        "Dolphins,2018-09-12T15:00,Miami,1500000000,8200000000000",
+                        "Patriots,2018-09-13T14:30,Boston,1500000000,9000000000000",
+                        "Yankees,2018-09-10T12:00,NewYorkCity,1300000000,7000000000000"),
+                "SoccerSchedule2");
+    }
+
+    @Test
+    public void innerJoinWithDoubleDogsCatsFish() {
+        Table joined = DOUBLE_INDEXED_MICE.join("ID").inner(new Table[] {DOUBLE_INDEXED_CATS,DOUBLE_INDEXED_FISH});
+        assertEquals(4, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinWithDoubleDogsCatsFishVarargs() {
+        Table joined = DOUBLE_INDEXED_MICE.join("ID").inner(DOUBLE_INDEXED_CATS,DOUBLE_INDEXED_FISH);
+        assertEquals(4, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinWithDoublesSimple() {
+        Table joined = DOUBLE_INDEXED_PEOPLE.join("ID").inner(DOUBLE_INDEXED_DOGS);
+        assertEquals(3, joined.columnCount());
+        assertEquals(3, joined.rowCount());
+        assertEquals(3, joined.column("ID").size());
+    }
+
     @Test
     public void innerJoinWithDoubles() {
         Table joined = DOUBLE_INDEXED_PEOPLE.join("ID").inner(DOUBLE_INDEXED_DOGS, "ID");
@@ -165,7 +394,7 @@ public class DataFrameJoinerTest {
         Table table1 = DUPLICATE_COL_NAME_DOGS.where(DUPLICATE_COL_NAME_DOGS.booleanColumn("Good").isTrue());
         Table table2 = DUPLICATE_COL_NAME_DOGS.where(DUPLICATE_COL_NAME_DOGS.booleanColumn("Good").isFalse());
 
-        Table joined = table1.join("ID").rightOuter(table2, "ID", true);
+        Table joined = table1.join("ID").rightOuter(table2, true, "ID");
         assertEquals(5, joined.columnCount());
         assertEquals(4, joined.rowCount());
     }
@@ -175,9 +404,11 @@ public class DataFrameJoinerTest {
         Table table1 = DUPLICATE_COL_NAME_DOGS.where(DUPLICATE_COL_NAME_DOGS.booleanColumn("Good").isTrue());
         Table table2 = DUPLICATE_COL_NAME_DOGS.where(DUPLICATE_COL_NAME_DOGS.booleanColumn("Good").isFalse());
 
-        Table joined = table1.join("ID").leftOuter(table2, "ID", true);
+        Table joined = table1.join("ID").leftOuter(table2, true, "ID");
         assertEquals(5, joined.columnCount());
         assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
+
     }
 
     @Test
@@ -185,6 +416,7 @@ public class DataFrameJoinerTest {
         Table joined = DOUBLE_INDEXED_PEOPLE.join("ID").leftOuter(DOUBLE_INDEXED_DOGS, "ID");
         assertEquals(3, joined.columnCount());
         assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
     }
 
     @Test
@@ -192,6 +424,22 @@ public class DataFrameJoinerTest {
         Table joined = DOUBLE_INDEXED_PEOPLE.join("ID").rightOuter(DOUBLE_INDEXED_DOGS, "ID");
         assertEquals(3, joined.columnCount());
         assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
+    }
+
+    @Test
+    public void rightOuterJoinWithDoubles2() {
+        Table joined = DOUBLE_INDEXED_PEOPLE.join("ID").rightOuter(DOUBLE_INDEXED_DOGS);
+        assertEquals(3, joined.columnCount());
+        assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
+    }
+
+    @Test
+    public void rightOuterJoinWithDoubles3() {
+        Table joined = DOUBLE_INDEXED_PEOPLE.join("ID").rightOuter(DOUBLE_INDEXED_DOGS, DOUBLE_INDEXED_CATS);
+        assertEquals(3, joined.columnCount());
+        assertEquals(4, joined.rowCount());
     }
 
     @Test
@@ -199,6 +447,23 @@ public class DataFrameJoinerTest {
         Table joined = DOUBLE_INDEXED_DOGS.join("ID").leftOuter(DOUBLE_INDEXED_PEOPLE, "ID");
         assertEquals(3, joined.columnCount());
         assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
+    }
+
+    @Test
+    public void leftOuterJoinWithDoubles3() {
+        Table joined = DOUBLE_INDEXED_DOGS.join("ID").leftOuter(DOUBLE_INDEXED_PEOPLE);
+        assertEquals(3, joined.columnCount());
+        assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
+    }
+
+    @Test
+    public void leftOuterJoinWithDoubles4() {
+        Table joined = DOUBLE_INDEXED_DOGS.join("ID").leftOuter(DOUBLE_INDEXED_PEOPLE, DOUBLE_INDEXED_CATS);
+        assertEquals(3, joined.columnCount());
+        assertEquals(4, joined.rowCount());
+        assertEquals(4, joined.column("ID").size());
     }
 
     @Test
@@ -213,6 +478,7 @@ public class DataFrameJoinerTest {
         Table joined = SP500.join("Date").leftOuter(ONE_YEAR, "Date");
         assertEquals(3, joined.columnCount());
         assertEquals(6, joined.rowCount());
+        assertEquals(6, joined.column("Date").size());
     }
 
     @Test
@@ -227,6 +493,7 @@ public class DataFrameJoinerTest {
         Table joined = ANIMAL_NAMES.join("Animal").leftOuter(ANIMAL_FEED, "Animal");
         assertEquals(3, joined.columnCount());
         assertEquals(6, joined.rowCount());
+        assertEquals(6, joined.column("Animal").size());
     }
 
     @Test
@@ -241,6 +508,20 @@ public class DataFrameJoinerTest {
         Table joined = ANIMAL_FEED.join("Animal").leftOuter(ANIMAL_NAMES, "Animal");
         assertEquals(3, joined.columnCount());
         assertEquals(6, joined.rowCount());
+        assertEquals(6, joined.column("Animal").size());
+    }
+
+    @Test
+    public void fullOuterJoinJustTable() {
+        Table joined = ANIMAL_FEED.join("Animal").fullOuter(ANIMAL_NAMES);
+        assertEquals(3, joined.columnCount());
+        assertEquals(8, joined.rowCount());
+        assertEquals(8, joined.column("Animal").size());
+        assertEquals(0, joined.column("Animal").countMissing());
+        assertEquals(8, joined.column("Name").size());
+        assertEquals(2, joined.column("Name").countMissing());
+        assertEquals(8, joined.column("Feed").size());
+        assertEquals(2, joined.column("Feed").countMissing());
     }
 
     @Test
@@ -256,53 +537,574 @@ public class DataFrameJoinerTest {
         assertEquals(2, joined.column("Feed").countMissing());
     }
 
+    @Test
+    public void fullOuterJoinNew() {
+        Table joined = ANIMAL_FEED.join("Animal").fullOuter(true, ANIMAL_NAMES);
+        assertEquals(3, joined.columnCount());
+        assertEquals(8, joined.rowCount());
+        assertEquals(8, joined.column("Animal").size());
+        assertEquals(0, joined.column("Animal").countMissing());
+        assertEquals(8, joined.column("Name").size());
+        assertEquals(2, joined.column("Name").countMissing());
+        assertEquals(8, joined.column("Feed").size());
+        assertEquals(2, joined.column("Feed").countMissing());
+    }
+
     @Test
     public void innerJoinStudentInstructorOnAge() {
-        Table STUDENT = createSTUDENT();
-        Table INSTRUCTOR = createINSTRUCTOR();
-        Table joined = STUDENT.join("Age").inner(true, INSTRUCTOR);
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join("Age").inner(true, table2);
         assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
                 "T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear"})));
         assertEquals(16, joined.columnCount());
-        assertEquals(5, joined.rowCount());
-    }    
+        assertEquals(14, joined.rowCount());
+    }
 
     @Test
     public void innerJoinInstructorStudentOnAge() {
-        Table STUDENT = createSTUDENT();
-        Table INSTRUCTOR = createINSTRUCTOR();
-        Table joined = INSTRUCTOR.join("Age").inner(true, STUDENT);
+        Table table1 = createINSTRUCTOR();
+        Table table2 = createSTUDENT();
+        Table joined = table1.join("Age").inner(true, table2);
         assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
                 "T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear"})));
         assertEquals(16, joined.columnCount());
-        assertEquals(5, joined.rowCount());
-    }    
+        assertEquals(14, joined.rowCount());
+    }
 
     @Test
     public void innerJoinStudentInstructorClassOnAge() {
-        Table STUDENT = createSTUDENT();
-        Table INSTRUCTOR = createINSTRUCTOR();
-        Table CLASS = createCLASS();
-        Table joined = STUDENT.join("Age").inner(true, INSTRUCTOR,CLASS);
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table table3 = createCLASS();
+        Table joined = table1.join("Age").inner(true, table2,table3);
         assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
                 "T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear",
         "T3.ID"})));
         assertEquals(24, joined.columnCount());
-        assertEquals(7, joined.rowCount());
+        assertEquals(14, joined.rowCount());
     }
 
     @Test
     public void innerJoinStudentInstructorClassDeptHeadOnAge() {
-        Table STUDENT = createSTUDENT();
-        Table INSTRUCTOR = createINSTRUCTOR();
-        Table CLASS = createCLASS();
-        Table DEPTHEAD = createDEPTHEAD();
-        Table joined = STUDENT.join("Age").inner(true, INSTRUCTOR,CLASS,DEPTHEAD);
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table table3 = createCLASS();
+        Table table4 = createDEPTHEAD();
+        Table joined = table1.join("Age").inner(true, table2,table3,table4);
         assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
                 "T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear",
                 "T3.ID", "T4.ID", "T4.First", "T4.Last", "T4.City", "T4.State"})));
         assertEquals(30, joined.columnCount());
+        assertEquals(14, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinStudentInstructorDeptHeadOnStateAge() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table table3 = createDEPTHEAD();
+        Table joined = table1.join(new String[] {"State","Age"})
+                .inner(true, table2,table3);
+        assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
+                "T2.ID", "T2.City", "T2.USID", "T2.GradYear",
+
+                "T3.ID", "T3.First","T3.Last","T3.City"})));
+        assertEquals(20, joined.columnCount());
+        assertEquals(1, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinStudentInstructorOnStateAge() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","Age"}).inner(true, table2);
+        assertEquals(15, joined.columnCount());
+        assertEquals(3, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinStudentInstructorOnStateAgeGradYear() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","Age","GradYear"})
+                .inner(true, table2);
+        assertEquals(14, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void leftJoinStudentInstructorOnStateAge() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","Age"})
+                .leftOuter(true, table2);
+        assertEquals(15, joined.columnCount());
+        assertEquals(10, joined.rowCount());
+        assertEquals(10, joined.column("State").size());
+        assertEquals(10, joined.column("Age").size());
+    }
+
+    @Test
+    public void innerJoinHouseBoatOnBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Bedrooms","Owner"})
+                .inner(table2,new String[] {"Bedrooms","Owner"});
+        assertEquals(6, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+        assertEquals(2, joined.column("Bedrooms").size());
+    }
+
+    @Test
+    public void innerJoinHouseBoatOnStyleTypeBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Style","Bedrooms","Owner"})
+                .inner(table2,new String[] {"Type","Bedrooms","Owner"});
+        assertEquals(5, joined.columnCount());
+        assertEquals(1, joined.rowCount());
+    }
+
+    @Test
+    public void fullJoinHouseBoatOnBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Bedrooms","Owner"})
+          .fullOuter(true,table2);
+        assertEquals(6, joined.columnCount());
         assertEquals(7, joined.rowCount());
+        assertEquals(7, joined.column("Style").size());
+        assertEquals(3, joined.column("Style").countMissing());
+        assertEquals(7, joined.column("Bedrooms").size());
+        assertEquals(0, joined.column("Bedrooms").countMissing());
+        assertEquals(7, joined.column("BuildDate").size());
+        assertEquals(3, joined.column("BuildDate").countMissing());
+        assertEquals(7, joined.column("Owner").size());
+        assertEquals(0, joined.column("Owner").countMissing());
+        assertEquals(7, joined.column("Type").size());
+        assertEquals(2, joined.column("Type").countMissing());
+        assertEquals(7, joined.column("SoldDate").size());
+        assertEquals(2, joined.column("SoldDate").countMissing());
+
     }
 
+    @Test
+    public void fullJoinHouse10Boat10OnBedroomsOwner() {
+        Table table1 = createHOUSE10();
+        Table table2 = createBOAT10();
+        Table joined = table1.join(new String[] {"Bedrooms","Owner"})
+                .fullOuter(true,table2);
+        assertEquals(6, joined.columnCount());
+        assertEquals(11, joined.rowCount());
+        assertEquals(11, joined.column("Bedrooms").size());
+        assertEquals(0, joined.column("Bedrooms").countMissing());
+        assertEquals(11, joined.column("Bedrooms").size());
+        assertEquals(0, joined.column("Bedrooms").countMissing());
+        assertEquals(11, joined.column("BuildDate").size());
+        assertEquals(3, joined.column("BuildDate").countMissing());
+        assertEquals(11, joined.column("Owner").size());
+        assertEquals(0, joined.column("Owner").countMissing());
+        assertEquals(11, joined.column("Type").size());
+        assertEquals(3, joined.column("Type").countMissing());
+        assertEquals(11, joined.column("SoldDate").size());
+        assertEquals(3, joined.column("SoldDate").countMissing());
+
+    }
+
+    @Test
+    public void fullJoinBnBBoat10OnBedroomsOwner() {
+        Table table1 = createBEDANDBREAKFAST();
+        Table table2 = createBOAT10();
+        Table joined = table1.join(new String[] {"Bedrooms","SoldDate"})
+                .fullOuter(true,table2);
+        assertEquals(6, joined.columnCount());
+        assertEquals(11, joined.rowCount());
+        assertEquals(11, joined.column("Design").size());
+        assertEquals(6, joined.column("Design").countMissing());
+        assertEquals(11, joined.column("Bedrooms").size());
+        assertEquals(0, joined.column("Bedrooms").countMissing());
+        assertEquals(11, joined.column("SoldDate").size());
+        assertEquals(0, joined.column("SoldDate").countMissing());
+        assertEquals(11, joined.column("Owner").size());
+        assertEquals(6, joined.column("Owner").countMissing());
+        assertEquals(11, joined.column("Type").size());
+        assertEquals(3, joined.column("Type").countMissing());
+        assertEquals(11, joined.column("T2.Owner").size());
+        assertEquals(3, joined.column("T2.Owner").countMissing());
+    }
+
+    @Test
+    public void leftJoinHouseBoatOnBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Bedrooms","Owner"})
+                .leftOuter(table2,new String[] {"Bedrooms","Owner"});
+        assertEquals(6, joined.columnCount());
+        assertEquals(4, joined.rowCount());
+    }
+
+    @Test
+    public void leftJoinHouseBoatOnStyleTypeBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Style","Bedrooms","Owner"})
+                .leftOuter(table2,new String[] {"Type","Bedrooms","Owner"});
+        assertEquals(5, joined.columnCount());
+        assertEquals(4, joined.rowCount());
+    }
+
+    @Test
+    public void leftJoinHouseBoatBnBOnStyleTypeBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Style","Bedrooms","Owner"})
+                .leftOuter(table2,new String[] {"Type","Bedrooms","Owner"});
+        assertEquals(5, joined.columnCount());
+        assertEquals(4, joined.rowCount());
+    }
+
+    @Test
+    public void rightJoinHouseBoatOnBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Bedrooms","Owner"})
+                .rightOuter(table2,new String[] {"Bedrooms","Owner"});
+        assertEquals(6, joined.columnCount());
+        assertEquals(5, joined.rowCount());
+    }
+
+    @Test
+    public void rightJoinHouseBoatOnStyleTypeBedroomsOwner() {
+        Table table1 = createHOUSE();
+        Table table2 = createBOAT();
+        Table joined = table1.join(new String[] {"Style","Bedrooms","Owner"})
+                .rightOuter(table2,new String[] {"Type","Bedrooms","Owner"});
+        assertEquals(5, joined.columnCount());
+        assertEquals(5, joined.rowCount());
+    }
+
+    @Test
+    public void rightJoinStudentInstructorOnStateAge() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","Age"})
+                .rightOuter(true, table2);
+        assertEquals(15, joined.columnCount());
+        assertEquals(10, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinStudentInstructorOnStateName() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","FirstName"})
+                .inner(table2, true, new String[] {"State","First"});
+        assertEquals(15, joined.columnCount());
+        assertEquals(5, joined.rowCount());
+    }
+
+    @Test
+    public void leftJoinStudentInstructorOnStateName() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","FirstName"})
+                .leftOuter(table2, true, new String[] {"State","First"});
+        assertEquals(15, joined.columnCount());
+        assertEquals(10, joined.rowCount());
+    }
+
+    @Test
+    public void rightJoinStudentInstructorOnStateName() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        Table joined = table1.join(new String[] {"State","FirstName"})
+                .rightOuter(table2, true, new String[] {"State","First"});
+        assertEquals(15, joined.columnCount());
+        assertEquals(10, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinOnAge() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table joined = table1.join("Age")
+                .inner(table2, "Age", true);
+        assertEquals(9, joined.columnCount());
+        assertEquals(18, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinAnimalPeopleOnAge() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table joined = table1.join("Age").inner(true, table2);
+        assertEquals(9, joined.columnCount());
+        assertEquals(18, joined.rowCount());
+    }
+    @Test
+    public void innerJoinAnimalTreeOnAge() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createTREE();
+        Table joined = table1.join("Age").inner(true, table2);
+        assertEquals(7, joined.columnCount());
+        assertEquals(8, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinAnimalPeopleTreeOnAge() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table table3 = createTREE();
+        Table table4 = createFLOWER();
+        Table joined = table1.join("Age")
+                .inner(true, table2, table3, table4);
+        assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
+                "T2.Name", "T2.HOME", "T2.MoveInDate",
+                "T3.Name", "T3.Home", "T4.Name","T4.Home","Color"})));
+        assertEquals(14, joined.columnCount());
+        assertEquals(18, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinAnimalPeopleTreeOnAgeHome() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table table3 = createTREE();
+        Table table4 = createFLOWER();
+        Table joined = table1.join(new String[] {"Age","Home"})
+                .inner(true, table2, table3, table4);
+        assert(joined.columnNames().containsAll(Arrays.asList(new String[] {
+                "Animal", "Name", "Home", "Age", "MoveInDate", "ID",
+                "T2.Name","T2.MoveInDate","T3.Name","T4.Name","Color"})));
+        assertEquals(11, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinOnNameHomeAge() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table joined = table1.join(new String[] {"Name", "Home", "Age"})
+                .inner(true, table2);
+        assertEquals(7, joined.columnCount());
+        assertEquals(1, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinOnAllMismatchedColNames() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENicknameDwellingYearsMoveInDate();
+        Table joined = table1.join(new String[] {"Name", "Home", "Age"})
+                .inner(table2,
+                true, new String[] {"Nickname", "Dwelling", "Years"});
+        assertEquals(7, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinOnPartiallyMismatchedColNames() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate();
+        Table joined = table1.join(new String[] {"Name", "Home", "Age"})
+                .inner(table2, true,
+                new String[] {"Name", "Dwelling", "Years"});
+        assert(joined.columnNames().containsAll(Arrays.asList(new String[] {"Name", "Home", "Age"})));
+        assertEquals(7, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void leftOuterJoinOnPartiallyMismatchedColNames() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate();
+        Table joined = table1.join(new String[] {"Name", "Home", "Age"})
+                .leftOuter(table2, true,
+                        new String[] {"Name", "Dwelling", "Years"});
+        assert(joined.columnNames().containsAll(Arrays.asList(new String[] {"Name", "Home", "Age"})));
+        assertEquals(7, joined.columnCount());
+        assertEquals(8, joined.rowCount());
+    }
+
+    @Test
+    public void rightOuterJoinOnPartiallyMismatchedColNames() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate();
+        Table joined = table1.join(new String[] {"Name", "Home", "Age"})
+                .rightOuter(table2, true,
+                        "Name", "Dwelling", "Years");
+        assert(joined.columnNames().containsAll(Arrays.asList(new String[] {"Name", "Dwelling", "Years"})));
+        assertEquals(7, joined.columnCount());
+        assertEquals(6, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinOnAgeMoveInDate() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table joined = table1.join(new String[] {"Age","MoveInDate"})
+                .inner(true, table2);
+        assertEquals(8, joined.columnCount());
+        assertEquals(3, joined.rowCount());
+    }
+
+    @Test
+    public void leftOuterJoinOnAgeMoveInDate() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table joined = table1.join(new String[] {"Age","MoveInDate"})
+                .leftOuter(true, table2);
+        assertEquals(8, joined.columnCount());
+        assertEquals(9, joined.rowCount());
+    }
+
+    @Test
+    public void rightOuterJoinOnAgeMoveInDate() {
+        Table table1 = createANIMALHOMES();
+        Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
+        Table joined = table1.join("Age","MoveInDate")
+                .rightOuter(true, table2);
+        assertEquals(8, joined.columnCount());
+        assertEquals(6, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinFootballSoccerOnPlayDate() {
+        Table table1 = createFOOTBALLSCHEDULE();
+        Table table2 = createSOCCERSCHEDULE();
+        Table joined = table1.join(new String[] {"PlayDate"})
+                .inner(true, table2);
+        assertEquals(8, joined.columnCount());
+        assertEquals(5, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinFootballSoccerOnPlayTime() {
+        Table table1 = createFOOTBALLSCHEDULE();
+        Table table2 = createSOCCERSCHEDULE();
+        Table joined = table1.join(new String[] {"PlayTime"})
+                .inner(true, table2);
+        assertEquals(8, joined.columnCount());
+        assertEquals(3, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinFootballSoccerOnPlayDatePlayTime() {
+        Table table1 = createFOOTBALLSCHEDULE();
+        Table table2 = createSOCCERSCHEDULE();
+        Table joined = table1.join(new String[] {"PlayDate","PlayTime"})
+                .inner(true, table2);
+        assertEquals(7, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void fullOuterJoinFootballSoccerOnPlayTime() {
+        Table table1 = createFOOTBALLSCHEDULE();
+        Table table2 = createSOCCERSCHEDULE();
+        Table joined = table1.join("PlayTime")
+                .fullOuter(true, table2);
+        assertEquals(8, joined.columnCount());
+        assertEquals(5, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinFootballSoccerOnPlayDatePlayDateTime() {
+        Table table1 = createFOOTBALLSCHEDULEDateTime();
+        Table table2 = createSOCCERSCHEDULEDateTime();
+        Table joined = table1.join("PlayDateTime")
+                .inner(true, table2);
+        assertEquals(10, joined.columnCount());
+        assertEquals(2, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinFootballSoccerOnSeasonRevenue() {
+        Table table1 = createFOOTBALLSCHEDULEDateTime();
+        Table table2 = createSOCCERSCHEDULEDateTime();
+        Table joined = table1.join(new String[] {"SeasonRevenue","AllTimeRevenue"})
+                .inner(true, table2);
+        assertEquals(9, joined.columnCount());
+        assertEquals(1, joined.rowCount());
+    }
+
+    @Test
+    public void innerJoinFootballSoccerOnHomeGame() {
+        Table table1 = createFOOTBALLSCHEDULEDateTime();
+        Table table2 = createSOCCERSCHEDULEDateTime();
+        try {
+            Table joined = table1.join(new String[] {"HomeGame"})
+                    .inner(true, table2);
+            assertEquals(9, joined.columnCount());
+            assertEquals(1, joined.rowCount());
+        } catch (IllegalArgumentException e) {
+            assertTrue(e.getMessage().contains("Joining is supported on integral, string, and date-like columns."));
+        }
+    }
+
+    @Test
+    public void fullOuterJoinFootballSoccerOnPlayDateTimeSeasonRevenue() {
+        Table table1 = createFOOTBALLSCHEDULEDateTime();
+        Table table2 = createSOCCERSCHEDULEDateTime();
+        Table joined = table1.join(new String[] {"PlayDateTime","SeasonRevenue"})
+                .fullOuter(true, table2);
+        assertEquals(9, joined.columnCount());
+        assertEquals(6, joined.rowCount());
+        assertEquals(6, joined.column("TeamName").size());
+        assertEquals(2, joined.column("TeamName").countMissing());
+        assertEquals(6, joined.column("PlayDateTime").size());
+        assertEquals(0, joined.column("PlayDateTime").countMissing());
+        assertEquals(6, joined.column("Location").size());
+        assertEquals(2, joined.column("Location").countMissing());
+        assertEquals(6, joined.column("HomeGame").size());
+        assertEquals(2, joined.column("HomeGame").countMissing());
+        assertEquals(6, joined.column("SeasonRevenue").size());
+        assertEquals(0, joined.column("SeasonRevenue").countMissing());
+        assertEquals(6, joined.column("Mascot").size());
+        assertEquals(2, joined.column("Mascot").countMissing());
+        assertEquals(6, joined.column("Place").size());
+        assertEquals(2, joined.column("Place").countMissing());
+    }
+
+    @Test
+    public void fullOuterJoinFootballSoccerOnPlayDateTimeAllTimeRevenue() {
+        Table table1 = createFOOTBALLSCHEDULEDateTime();
+        Table table2 = createSOCCERSCHEDULEDateTime();
+        Table joined = table1.join(new String[] {"AllTimeRevenue"})
+                .fullOuter(true, table2);
+        assertEquals(10, joined.columnCount());
+        assertEquals(5, joined.rowCount());
+        assertEquals(5, joined.column("TeamName").size());
+        assertEquals(1, joined.column("TeamName").countMissing());
+        assertEquals(5, joined.column("PlayDateTime").size());
+        assertEquals(1, joined.column("PlayDateTime").countMissing());
+        assertEquals(5, joined.column("Location").size());
+        assertEquals(1, joined.column("Location").countMissing());
+        assertEquals(5, joined.column("HomeGame").size());
+        assertEquals(1, joined.column("HomeGame").countMissing());
+        assertEquals(5, joined.column("SeasonRevenue").size());
+        assertEquals(1, joined.column("SeasonRevenue").countMissing());
+        assertEquals(5, joined.column("AllTimeRevenue").size());
+        assertEquals(0, joined.column("AllTimeRevenue").countMissing());
+        assertEquals(5, joined.column("Mascot").size());
+        assertEquals(1, joined.column("Mascot").countMissing());
+        assertEquals(5, joined.column("T2.PlayDateTime").size());
+        assertEquals(1, joined.column("T2.PlayDateTime").countMissing());
+        assertEquals(5, joined.column("Place").size());
+        assertEquals(1, joined.column("Place").countMissing());
+        assertEquals(5, joined.column("T2.SeasonRevenue").size());
+        assertEquals(1, joined.column("T2.SeasonRevenue").countMissing());
+    }
+
+    @Test
+    public void fullOuterJoinFootballBaseballException() {
+        Table table1 = createFOOTBALLSCHEDULE();
+        Table table2 = createBASEBALLSCHEDULEDateTime();
+        try {
+            table1.join(new String[] {"HomeGame"})
+                .fullOuter(true, table2);
+        } catch (IllegalArgumentException e) {
+            assertTrue(e.getMessage().contains("Joining is supported on integral, string, and date-like columns."));
+        }
+    }
 }