Repository: jtablesaw/tablesaw
Pull request #1107 (2022-04-11 23:24:23)

Added support for joins using Sort-Merge-Join Algorithm
Thanks for contributing.

- [x] Tick to sign-off your agreement to the [Developer Certificate of Origin (DCO) 1.1](https://developercertificate.org)

## Description

Added an implementation of Sort-Merge-Join to address the On^2 behavior of the existing Nested-Loop-Join (aka CrossProductJoin) under certain circumstances. 

Separate the DataFramJoiner class from the join algorithm implementation, making it possible to support multiple algorithms

Revised Join API to eliminate the 25 methods covering the four join types (INNER, etc.) and their optional params. 

Deprecated the old API methods.

Wrote a first pass method for selecting one of the two join algorithms based on the data being joined. More work required.

## Testing

Did you add a unit test? Yes. 



Patch:
diff --git a/core/src/main/java/tech/tablesaw/api/BooleanColumn.java b/core/src/main/java/tech/tablesaw/api/BooleanColumn.java
index 3135a35f..6775b313 100644
--- a/core/src/main/java/tech/tablesaw/api/BooleanColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/BooleanColumn.java
@@ -492,7 +492,7 @@ public class BooleanColumn extends AbstractColumn<BooleanColumn, Boolean>
   }
 
   /** Sets the value at i to b, and returns this column */
-  private BooleanColumn set(int i, byte b) {
+  public BooleanColumn set(int i, byte b) {
     data.set(i, b);
     return this;
   }
diff --git a/core/src/main/java/tech/tablesaw/api/Row.java b/core/src/main/java/tech/tablesaw/api/Row.java
index bf8d1eba..44a659be 100644
--- a/core/src/main/java/tech/tablesaw/api/Row.java
+++ b/core/src/main/java/tech/tablesaw/api/Row.java
@@ -168,6 +168,10 @@ public class Row implements Iterator<Row> {
     }
   }
 
+  public ColumnType type(int columnIndex) {
+    return tableSlice.column(columnIndex).type();
+  }
+
   /** Moves this Row to the given 0-based row index */
   public void at(int rowNumber) {
     this.rowNumber = rowNumber;
@@ -567,6 +571,10 @@ public class Row implements Iterator<Row> {
     setDate(columnNames[columnIndex], value);
   }
 
+  public void setPackedDate(int columnIndex, int value) {
+    setPackedDate(columnNames[columnIndex], value);
+  }
+
   /**
    * Sets the value of the column with the given name at this Row to the given value. An
    * IllegalStateException is * thrown if the column is not present in the Row and an
@@ -577,6 +585,42 @@ public class Row implements Iterator<Row> {
     dateColumnMap.get(columnName).set(getIndex(rowNumber), value);
   }
 
+  public void setPackedDate(String columnName, int value) {
+    dateColumnMap.get(columnName).set(getIndex(rowNumber), value);
+  }
+
+  public void setPackedTime(int columnIndex, int value) {
+    setPackedTime(columnNames[columnIndex], value);
+  }
+
+  public void setPackedDateTime(int columnIndex, long value) {
+    setPackedDateTime(columnNames[columnIndex], value);
+  }
+
+  public void setPackedInstant(int columnIndex, long value) {
+    setPackedInstant(columnNames[columnIndex], value);
+  }
+
+  public void setPackedTime(String columnName, int value) {
+    timeColumnMap.get(columnName).set(getIndex(rowNumber), value);
+  }
+
+  public void setPackedDateTime(String columnName, long value) {
+    dateTimeColumnMap.get(columnName).set(getIndex(rowNumber), value);
+  }
+
+  public void setPackedInstant(String columnName, long value) {
+    instantColumnMap.get(columnName).set(getIndex(rowNumber), value);
+  }
+
+  public void setBooleanAsByte(String columnName, byte value) {
+    booleanColumnMap.get(columnName).set(getIndex(rowNumber), value);
+  }
+
+  public void setBooleanAsByte(int columnIndex, byte value) {
+    setBooleanAsByte(columnNames[columnIndex], value);
+  }
+
   /**
    * Sets the value of the column at the given index and this Row to the given value. An
    * IllegalStateException is * thrown if the column is not present in the Row and an
@@ -777,6 +821,14 @@ public class Row implements Iterator<Row> {
     return tableSlice.mappedRowNumber(rowNumber);
   }
 
+  /**
+   * Returns the row number in the table backing the slice behind this row. This value may differ
+   * from the rowNumber() if the slice covers less than the entire table
+   */
+  public int getBackingRowNumber() {
+    return getIndex(getRowNumber());
+  }
+
   /**
    * Returns a double representing the value held in the column with the given name at this row, for
    * any numeric column type
@@ -794,7 +846,7 @@ public class Row implements Iterator<Row> {
     return tableSlice.column(columnIndex).type();
   }
 
-  Column<?> column(int columnIndex) {
+  public Column<?> column(int columnIndex) {
     return tableSlice.column(columnIndex);
   }
 
diff --git a/core/src/main/java/tech/tablesaw/api/Table.java b/core/src/main/java/tech/tablesaw/api/Table.java
index eb372678..3e1b90b7 100644
--- a/core/src/main/java/tech/tablesaw/api/Table.java
+++ b/core/src/main/java/tech/tablesaw/api/Table.java
@@ -219,7 +219,7 @@ public class Table extends Relation implements Iterable<Row> {
     return key;
   }
 
-  /** Returns an object that an be used to read data from a file into a new Table */
+  /** Returns an object that can be used to read data from a file into a new Table */
   public static DataFrameReader read() {
     return new DataFrameReader(defaultReaderRegistry);
   }
diff --git a/core/src/main/java/tech/tablesaw/joining/AbstractJoiner.java b/core/src/main/java/tech/tablesaw/joining/AbstractJoiner.java
new file mode 100644
index 00000000..73ca96f8
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/AbstractJoiner.java
@@ -0,0 +1,439 @@
+package tech.tablesaw.joining;
+
+import tech.tablesaw.api.Table;
+
+public abstract class AbstractJoiner {
+
+  public abstract DataFrameJoiner type(JoinType joinType);
+
+  public abstract DataFrameJoiner keepAllJoinKeyColumns(boolean keep);
+
+  public abstract DataFrameJoiner allowDuplicateColumnNames(boolean allow);
+
+  public abstract DataFrameJoiner rightJoinColumns(String... rightJoinColumnNames);
+
+  public abstract DataFrameJoiner with(Table... tables);
+
+  public abstract Table join();
+
+  public AbstractJoiner() {}
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table inner(Table... tables) {
+    type(JoinType.INNER);
+    with(tables);
+    return join();
+  }
+
+  /**
+   * Joins 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
+   */
+  @Deprecated
+  public Table inner(boolean allowDuplicateColumnNames, Table... tables) {
+    type(JoinType.INNER);
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    with(tables);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  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
+   */
+  @Deprecated
+  public Table inner(Table table2, String[] col2Names) {
+    return inner(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.
+   * @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
+   */
+  @Deprecated
+  public Table inner(Table table2, String col2Name, boolean allowDuplicateColumnNames) {
+    return inner(table2, allowDuplicateColumnNames, 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 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
+   */
+  @Deprecated
+  public Table inner(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.INNER);
+    rightJoinColumns(col2Names);
+    with(table2);
+    return join();
+  }
+
+  /**
+   * 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
+   *     table1 if {@code true} the join will return all join key columns in both table, which may
+   *     have difference when there are null values
+   * @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
+   */
+  @Deprecated
+  public Table inner(
+      Table table2,
+      boolean allowDuplicateColumnNames,
+      boolean keepAllJoinKeyColumns,
+      String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    keepAllJoinKeyColumns(keepAllJoinKeyColumns);
+    type(JoinType.INNER);
+    rightJoinColumns(col2Names);
+    with(table2);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table fullOuter(Table... tables) {
+    return fullOuter(false, tables);
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table fullOuter(boolean allowDuplicateColumnNames, Table... tables) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.FULL_OUTER);
+    with(tables);
+    return join();
+  }
+
+  /**
+   * 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
+   *     table1 if {@code true} the join will return all join key columns in both table, which may
+   *     have difference when there are null values
+   * @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
+   */
+  @Deprecated
+  public Table fullOuter(
+      Table table2,
+      boolean allowDuplicateColumnNames,
+      boolean keepAllJoinKeyColumns,
+      String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.FULL_OUTER);
+    rightJoinColumns(col2Names);
+    keepAllJoinKeyColumns(keepAllJoinKeyColumns);
+    with(table2);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table fullOuter(Table table2, String col2Name) {
+    type(JoinType.FULL_OUTER);
+    with(table2);
+    rightJoinColumns(col2Name);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table leftOuter(Table... tables) {
+    return leftOuter(false, tables);
+  }
+
+  /**
+   * Joins 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
+   */
+  @Deprecated
+  public Table leftOuter(boolean allowDuplicateColumnNames, Table... tables) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.LEFT_OUTER);
+    with(tables);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  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
+   */
+  @Deprecated
+  public Table leftOuter(Table table2, String col2Name) {
+    return leftOuter(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 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
+   */
+  @Deprecated
+  public Table leftOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.LEFT_OUTER);
+    with(table2);
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    rightJoinColumns(col2Names);
+    return join();
+  }
+
+  /**
+   * 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
+   *     table1 if {@code true} the join will return all join key columns in both table, which may
+   *     have difference when there are null values
+   * @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
+   */
+  @Deprecated
+  public Table leftOuter(
+      Table table2,
+      boolean allowDuplicateColumnNames,
+      boolean keepAllJoinKeyColumns,
+      String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.LEFT_OUTER);
+    with(table2);
+    keepAllJoinKeyColumns(keepAllJoinKeyColumns);
+    rightJoinColumns(col2Names);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table rightOuter(Table... tables) {
+    return rightOuter(false, tables);
+  }
+
+  /**
+   * Joins 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
+   */
+  @Deprecated
+  public Table rightOuter(boolean allowDuplicateColumnNames, Table... tables) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.RIGHT_OUTER);
+    with(tables);
+    return join();
+  }
+
+  /**
+   * 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
+   */
+  @Deprecated
+  public Table rightOuter(Table table2, String col2Name) {
+    return rightOuter(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
+   */
+  @Deprecated
+  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 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.
+   */
+  @Deprecated
+  public Table rightOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    type(JoinType.RIGHT_OUTER);
+    with(table2);
+    rightJoinColumns(col2Names);
+    return join();
+  }
+
+  /**
+   * 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
+   *     table1 if {@code true} the join will return all join key columns in both table, which may
+   *     have difference when there are null values
+   * @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
+   */
+  @Deprecated
+  public Table rightOuter(
+      Table table2,
+      boolean allowDuplicateColumnNames,
+      boolean keepAllJoinKeyColumns,
+      String... col2Names) {
+    allowDuplicateColumnNames(allowDuplicateColumnNames);
+    keepAllJoinKeyColumns(keepAllJoinKeyColumns);
+    type(JoinType.RIGHT_OUTER);
+    with(table2);
+    rightJoinColumns(col2Names);
+    return join();
+  }
+
+  abstract Table joinInternal(
+      Table table,
+      Table table2,
+      JoinType rightOuter,
+      boolean allowDuplicateColumnNames,
+      boolean keepAllJoinKeyColumns,
+      String[] col2Names);
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/ColumnIndexPair.java b/core/src/main/java/tech/tablesaw/joining/ColumnIndexPair.java
new file mode 100644
index 00000000..61df50ca
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/ColumnIndexPair.java
@@ -0,0 +1,30 @@
+package tech.tablesaw.joining;
+
+import tech.tablesaw.api.ColumnType;
+
+/**
+ * Describes two columns that are to be compared in a sort The columns are expected to be referenced
+ * in two separate rows. The values of left and right provide the column index (position) in each of
+ * the two rows.
+ */
+public class ColumnIndexPair {
+  final ColumnType type;
+  final int left;
+  final int right;
+
+  public ColumnIndexPair(ColumnType type, int left, int right) {
+    this.type = type;
+    this.left = left;
+    this.right = right;
+  }
+
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder("ColumnIndexPair{");
+    sb.append("type=").append(type);
+    sb.append(", left=").append(left);
+    sb.append(", right=").append(right);
+    sb.append('}');
+    return sb.toString();
+  }
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/CrossProductJoin.java b/core/src/main/java/tech/tablesaw/joining/CrossProductJoin.java
new file mode 100644
index 00000000..a7280078
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/CrossProductJoin.java
@@ -0,0 +1,528 @@
+package tech.tablesaw.joining;
+
+import com.google.common.collect.Streams;
+import com.google.common.primitives.Ints;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import tech.tablesaw.api.*;
+import tech.tablesaw.columns.Column;
+import tech.tablesaw.columns.booleans.BooleanColumnType;
+import tech.tablesaw.columns.dates.DateColumnType;
+import tech.tablesaw.columns.datetimes.DateTimeColumnType;
+import tech.tablesaw.columns.instant.InstantColumnType;
+import tech.tablesaw.columns.numbers.*;
+import tech.tablesaw.columns.strings.StringColumnType;
+import tech.tablesaw.columns.strings.TextColumnType;
+import tech.tablesaw.columns.times.TimeColumnType;
+import tech.tablesaw.index.*;
+import tech.tablesaw.selection.Selection;
+
+/** Implements joins between two or more Tables */
+public class CrossProductJoin implements JoinStrategy {
+
+  private static final String TABLE_ALIAS = "T";
+
+  private List<Integer> joinColumnIndexes;
+  private final AtomicInteger joinTableId = new AtomicInteger(2);
+
+  /**
+   * Constructor.
+   *
+   * @param table The table to join on.
+   * @param joinColumnNames The join column names to join on.
+   */
+  public CrossProductJoin(Table table, String... joinColumnNames) {
+    this.joinColumnIndexes = getJoinIndexes(table, joinColumnNames);
+  }
+
+  /**
+   * Finds the index of the columns corresponding to the columnNames. E.G. The column named "ID" is
+   * located at index 5 in table.
+   *
+   * @param table the table that contains the columns.
+   * @param columnNames the column names to find indexes of.
+   * @return a list of column indexes within the table.
+   */
+  private List<Integer> getJoinIndexes(Table table, String[] columnNames) {
+    return Arrays.stream(columnNames).map(table::columnIndex).collect(Collectors.toList());
+  }
+
+  /**
+   * Joins two tables.
+   *
+   * @param table1 the table on the left side of the join.
+   * @param table2 the table on the right side of the join.
+   * @param joinType the type of join.
+   * @param allowDuplicates 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
+   *     table1 if {@code true} the join will return all join key columns in both table, which may
+   *     have difference when there are null values
+   * @param table2JoinColumnNames The names of the columns in table2 to join on.
+   * @return the joined table
+   */
+  public Table performJoin(
+      Table table1,
+      Table table2,
+      JoinType joinType,
+      boolean allowDuplicates,
+      boolean keepAllJoinKeyColumns,
+      int[] leftJoinColumnIndexes,
+      String... table2JoinColumnNames) {
+
+    this.joinColumnIndexes =
+        Arrays.stream(leftJoinColumnIndexes).boxed().collect(Collectors.toList());
+    List<Integer> table2JoinColumnIndexes = getJoinIndexes(table2, table2JoinColumnNames);
+    List<Index> table1Indexes = buildIndexesForJoinColumns(joinColumnIndexes, table1);
+    List<Index> table2Indexes = buildIndexesForJoinColumns(table2JoinColumnIndexes, table2);
+
+    // A set of column indexes in the result table that can be ignored. They are duplicate join
+    // keys.
+
+    // collect all the column name in both tables
+    Column<?>[] cols =
+        Streams.concat(table1.columns().stream(), table2.columns().stream())
+            .map(Column::emptyCopy)
+            .toArray(Column[]::new);
+
+    Set<Integer> resultIgnoreColIndexes = new HashSet<>();
+    if (!keepAllJoinKeyColumns) {
+      resultIgnoreColIndexes = getIgnoreColumns(table1, joinType, table2JoinColumnIndexes, cols);
+    }
+
+    Table result = emptyTableFromColumns(table1, allowDuplicates, cols);
+
+    validateIndexes(table1Indexes, table2Indexes);
+    if (table1.rowCount() == 0 && (joinType == JoinType.LEFT_OUTER || joinType == JoinType.INNER)) {
+      // Handle special case of empty table here so it doesn't fall through to the behavior
+      // that adds rows for full outer and right outer joins
+      if (!keepAllJoinKeyColumns) {
+        result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
+      }
+      return result;
+    }
+
+    Selection table1DoneRows = Selection.with();
+    Selection table2DoneRows = Selection.with();
+    // use table 2 for row iteration, which can significantly increase performance
+    if (table1.rowCount() > table2.rowCount() && joinType == JoinType.INNER) {
+      for (Row row : table2) {
+        int ri = row.getRowNumber();
+        if (table2DoneRows.contains(ri)) {
+          // Already processed a selection of table1 that contained this row.
+          continue;
+        }
+        Selection table1Rows =
+            createMultiColSelection(
+                table2, ri, table1Indexes, table1.rowCount(), table2JoinColumnIndexes);
+        Selection table2Rows =
+            createMultiColSelection(
+                table2, ri, table2Indexes, table2.rowCount(), table2JoinColumnIndexes);
+        crossProduct(
+            result,
+            table1,
+            table2,
+            table1Rows,
+            table2Rows,
+            resultIgnoreColIndexes,
+            keepAllJoinKeyColumns);
+
+        table2DoneRows = table2DoneRows.or(table2Rows);
+        if (table2DoneRows.size() == table2.rowCount()) {
+          // Processed all the rows in table1 exit early.
+          if (!keepAllJoinKeyColumns) {
+            result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
+          }
+          return result;
+        }
+      }
+    } else {
+      for (Row row : table1) {
+        int ri = row.getRowNumber();
+        if (table1DoneRows.contains(ri)) {
+          // Already processed a selection of table1 that contained this row.
+          continue;
+        }
+        Selection table1Rows =
+            createMultiColSelection(
+                table1, ri, table1Indexes, table1.rowCount(), joinColumnIndexes);
+        Selection table2Rows =
+            createMultiColSelection(
+                table1, ri, table2Indexes, table2.rowCount(), joinColumnIndexes);
+        if ((joinType == JoinType.LEFT_OUTER || joinType == JoinType.FULL_OUTER)
+            && table2Rows.isEmpty()) {
+          withMissingLeftJoin(
+              result, table1, table1Rows, resultIgnoreColIndexes, keepAllJoinKeyColumns);
+        } else {
+          crossProduct(
+              result,
+              table1,
+              table2,
+              table1Rows,
+              table2Rows,
+              resultIgnoreColIndexes,
+              keepAllJoinKeyColumns);
+        }
+        table1DoneRows = table1DoneRows.or(table1Rows);
+        if (joinType == JoinType.FULL_OUTER || joinType == JoinType.RIGHT_OUTER) {
+          // Update done rows in table2 for full Outer.
+          table2DoneRows = table2DoneRows.or(table2Rows);
+        } else if (table1DoneRows.size() == table1.rowCount()) {
+          // Processed all the rows in table1 exit early.
+          if (!keepAllJoinKeyColumns) {
+            result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
+          }
+          return result;
+        }
+      }
+    }
+
+    // Add all rows from table2 that were not handled already.
+    Selection table2Rows = table2DoneRows.flip(0, table2.rowCount());
+    withMissingRight(
+        result,
+        table1.columnCount(),
+        table2,
+        table2Rows,
+        joinType,
+        table2JoinColumnIndexes,
+        resultIgnoreColIndexes,
+        keepAllJoinKeyColumns);
+    if (!keepAllJoinKeyColumns) {
+      result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
+    }
+    return result;
+  }
+
+  private void validateIndexes(List<Index> table1Indexes, List<Index> table2Indexes) {
+    if (table1Indexes.size() != table2Indexes.size()) {
+      throw new IllegalArgumentException(
+          "Cannot join using a different number of indices on each table: "
+              + table1Indexes
+              + " and "
+              + table2Indexes);
+    }
+    for (int i = 0; i < table1Indexes.size(); i++) {
+      if (!table1Indexes.get(i).getClass().equals(table2Indexes.get(i).getClass())) {
+        throw new IllegalArgumentException(
+            "Cannot join using different index types: " + table1Indexes + " and " + table2Indexes);
+      }
+    }
+  }
+
+  /** Build a reverse index for every join column in the table. */
+  private List<Index> buildIndexesForJoinColumns(List<Integer> joinColumnIndexes, Table table) {
+    return joinColumnIndexes.stream().map(c -> indexFor(table, c)).collect(Collectors.toList());
+  }
+
+  /** Create a reverse index for a given column. */
+  private Index indexFor(Table table, int colIndex) {
+    ColumnType type = table.column(colIndex).type();
+    if (type instanceof DateColumnType) {
+      return new IntIndex(table.dateColumn(colIndex));
+    } else if (type instanceof DateTimeColumnType) {
+      return new LongIndex(table.dateTimeColumn(colIndex));
+    } else if (type instanceof InstantColumnType) {
+      return new LongIndex(table.instantColumn(colIndex));
+    } else if (type instanceof TimeColumnType) {
+      return new IntIndex(table.timeColumn(colIndex));
+    } else if (type instanceof StringColumnType) {
+      return new StringIndex(table.stringColumn(colIndex));
+    } else if (type instanceof TextColumnType) {
+      return new StringIndex(table.textColumn(colIndex));
+    } else if (type instanceof IntColumnType) {
+      return new IntIndex(table.intColumn(colIndex));
+    } else if (type instanceof LongColumnType) {
+      return new LongIndex(table.longColumn(colIndex));
+    } else if (type instanceof ShortColumnType) {
+      return new ShortIndex(table.shortColumn(colIndex));
+    } else if (type instanceof BooleanColumnType) {
+      return new ByteIndex(table.booleanColumn(colIndex));
+    } else if (type instanceof DoubleColumnType) {
+      return new DoubleIndex(table.doubleColumn(colIndex));
+    } else if (type instanceof FloatColumnType) {
+      return new FloatIndex(table.floatColumn(colIndex));
+    }
+    throw new IllegalArgumentException("Joining attempted on unsupported column type " + type);
+  }
+
+  /**
+   * Given a reverse index find a selection of rows that have the same value as the supplied column
+   * does in the given row index.
+   */
+  private Selection selectionForColumn(Column<?> valueColumn, int rowIndex, Index rawIndex) {
+
+    ColumnType type = valueColumn.type();
+    if (type instanceof DateColumnType) {
+      IntIndex index = (IntIndex) rawIndex;
+      int value = ((DateColumn) valueColumn).getIntInternal(rowIndex);
+      return index.get(value);
+    } else if (type instanceof TimeColumnType) {
+      IntIndex index = (IntIndex) rawIndex;
+      int value = ((TimeColumn) valueColumn).getIntInternal(rowIndex);
+      return index.get(value);
+    } else if (type instanceof DateTimeColumnType) {
+      LongIndex index = (LongIndex) rawIndex;
+      long value = ((DateTimeColumn) valueColumn).getLongInternal(rowIndex);
+      return index.get(value);
+    } else if (type instanceof InstantColumnType) {
+      LongIndex index = (LongIndex) rawIndex;
+      long value = ((InstantColumn) valueColumn).getLongInternal(rowIndex);
+      return index.get(value);
+    } else if (type instanceof StringColumnType) {
+      StringIndex index = (StringIndex) rawIndex;
+      String value = ((StringColumn) valueColumn).get(rowIndex);
+      return index.get(value);
+    } else if (type instanceof TextColumnType) {
+      StringIndex index = (StringIndex) rawIndex;
+      String value = ((TextColumn) valueColumn).get(rowIndex);
+      return index.get(value);
+    } else if (type instanceof IntColumnType) {
+      IntIndex index = (IntIndex) rawIndex;
+      int value = ((IntColumn) valueColumn).getInt(rowIndex);
+      return index.get(value);
+    } else if (type instanceof LongColumnType) {
+      LongIndex index = (LongIndex) rawIndex;
+      long value = ((LongColumn) valueColumn).getLong(rowIndex);
+      return index.get(value);
+    } else if (type instanceof ShortColumnType) {
+      ShortIndex index = (ShortIndex) rawIndex;
+      short value = ((ShortColumn) valueColumn).getShort(rowIndex);
+      return index.get(value);
+    } else if (type instanceof BooleanColumnType) {
+      ByteIndex index = (ByteIndex) rawIndex;
+      byte value = ((BooleanColumn) valueColumn).getByte(rowIndex);
+      return index.get(value);
+    } else if (type instanceof DoubleColumnType) {
+      DoubleIndex index = (DoubleIndex) rawIndex;
+      double value = ((DoubleColumn) valueColumn).getDouble(rowIndex);
+      return index.get(value);
+    } else if (type instanceof FloatColumnType) {
+      FloatIndex index = (FloatIndex) rawIndex;
+      float value = ((FloatColumn) valueColumn).getFloat(rowIndex);
+      return index.get(value);
+    } else {
+      throw new IllegalArgumentException(
+          "Joining is supported on numeric, string, and date-like columns. Column "
+              + valueColumn.name()
+              + " is of type "
+              + valueColumn.type());
+    }
+  }
+  /**
+   * Create a big multicolumn selection for all join columns in the given table. Joins two tables.
+   *
+   * @param table the table that used to generate Selection.
+   * @param ri row number of row in table.
+   * @param indexes a reverse index for every join column in the table.
+   * @param selectionSize max size in table .
+   * @param joinColumnIndexes the column index of join key in tables
+   * @return selection created
+   */
+  private Selection createMultiColSelection(
+      Table table,
+      int ri,
+      List<Index> indexes,
+      int selectionSize,
+      List<Integer> joinColumnIndexes) {
+    Selection multiColSelection = Selection.withRange(0, selectionSize);
+    int i = 0;
+    for (Integer joinColumnIndex : joinColumnIndexes) {
+      Column<?> col = table.column(joinColumnIndex);
+      Selection oneColSelection = selectionForColumn(col, ri, indexes.get(i));
+      // and the selections.
+      multiColSelection = multiColSelection.and(oneColSelection);
+      i++;
+    }
+    return multiColSelection;
+  }
+
+  private String newName(String table2Alias, String columnName) {
+    return table2Alias + "." + columnName;
+  }
+
+  /**
+   * Adds empty columns to the destination table with the same type as columns in table1 and table2.
+   *
+   * <p>For inner, left and full outer join types the join columns in table2 are not needed and will
+   * be marked as placeholders. The indexes of those columns will be returned. The downstream logic
+   * is easier if we wait to remove the redundant columns until the last step.
+   *
+   * @param table1 the table on left side of the join.
+   * @param allowDuplicates whether to allow duplicates. If yes rename columns in table2 that have
+   *     the same name as columns in table1 with the exception of join columns in table2 when
+   *     performing a right join.
+   * @return A
+   */
+  private Table emptyTableFromColumns(Table table1, boolean allowDuplicates, Column<?>[] cols) {
+
+    Table destination = Table.create(table1.name());
+
+    // Rename duplicate columns in second table
+    if (allowDuplicates) {
+      Set<String> table1ColNames =
+          Arrays.stream(cols)
+              .map(Column::name)
+              .map(String::toLowerCase)
+              .limit(table1.columnCount())
+              .collect(Collectors.toSet());
+
+      String table2Alias = TABLE_ALIAS + joinTableId.getAndIncrement();
+      for (int c = table1.columnCount(); c < cols.length; c++) {
+        String columnName = cols[c].name();
+        if (table1ColNames.contains(columnName.toLowerCase())) {
+          cols[c].setName(newName(table2Alias, columnName));
+        }
+      }
+    }
+    destination.addColumns(cols);
+    return destination;
+  }
+
+  /**
+   * For inner join, left join and full outer join mark the join columns in table2 as placeholders.
+   *
+   * <p>For right join mark the join columns in table1 as placeholders. Keep track of which join
+   * columns are placeholders so they can be ignored.
+   */
+  private Set<Integer> getIgnoreColumns(
+      Table table1, JoinType joinType, List<Integer> table2JoinColumnIndexes, Column<?>[] cols) {
+    Set<Integer> ignoreColumns = new HashSet<>();
+    for (int c = 0; c < cols.length; c++) {
+      if (joinType == JoinType.RIGHT_OUTER) {
+        if (c < table1.columnCount() && joinColumnIndexes.contains(c)) {
+          cols[c].setName("Placeholder_" + ignoreColumns.size());
+          ignoreColumns.add(c);
+        }
+      } else {
+        int table2Index = c - table1.columnCount();
+        if (c >= table1.columnCount() && table2JoinColumnIndexes.contains(table2Index)) {
+          cols[c].setName("Placeholder_" + ignoreColumns.size());
+          ignoreColumns.add(c);
+        }
+      }
+    }
+    return ignoreColumns;
+  }
+
+  /**
+   * Creates cross product for the selection of two tables.
+   *
+   * @param destination the destination table.
+   * @param table1 the table on left of join.
+   * @param table2 the table on right of join.
+   * @param table1Rows the selection of rows in table1.
+   * @param table2Rows the selection of rows in table2.
+   * @param ignoreColumns a set of column indexes in the result to ignore. They are redundant join
+   *     columns.
+   */
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private void crossProduct(
+      Table destination,
+      Table table1,
+      Table table2,
+      Selection table1Rows,
+      Selection table2Rows,
+      Set<Integer> ignoreColumns,
+      boolean keepTable2JoinKeyColumns) {
+    for (int c = 0; c < table1.columnCount() + table2.columnCount(); c++) {
+      if (!keepTable2JoinKeyColumns && ignoreColumns.contains(c)) {
+        continue;
+      }
+      int table2Index = c - table1.columnCount();
+      for (int r1 : table1Rows) {
+        for (int r2 : table2Rows) {
+          if (c < table1.columnCount()) {
+            Column t1Col = table1.column(c);
+            destination.column(c).append(t1Col, r1);
+          } else {
+            Column t2Col = table2.column(table2Index);
+            destination.column(c).append(t2Col, r2);
+          }
+        }
+      }
+    }
+  }
+
+  /**
+   * Adds rows to destination for each row in table1 with the columns from table2 added as missing
+   * values.
+   */
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private void withMissingLeftJoin(
+      Table destination,
+      Table table1,
+      Selection table1Rows,
+      Set<Integer> ignoreColumns,
+      boolean keepTable2JoinKeyColumns) {
+    for (int c = 0; c < destination.columnCount(); c++) {
+      if (!keepTable2JoinKeyColumns && ignoreColumns.contains(c)) {
+        continue;
+      }
+      if (c < table1.columnCount()) {
+        Column t1Col = table1.column(c);
+        for (int index : table1Rows) {
+          destination.column(c).append(t1Col, index);
+        }
+      } else {
+        for (int r1 = 0; r1 < table1Rows.size(); r1++) {
+          destination.column(c).appendMissing();
+        }
+      }
+    }
+  }
+
+  /**
+   * Adds rows to destination for each row in table2 with the columns from table1 added as missing
+   * values.
+   */
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  private void withMissingRight(
+      Table destination,
+      int table1ColCount,
+      Table table2,
+      Selection table2Rows,
+      JoinType joinType,
+      List<Integer> col2Indexes,
+      Set<Integer> ignoreColumns,
+      boolean keepTable2JoinKeyColumns) {
+
+    // Add index data from table2 into join column positions in table one.
+    if (joinType == JoinType.FULL_OUTER) {
+      for (int i = 0; i < col2Indexes.size(); i++) {
+        Column t2Col = table2.column(col2Indexes.get(i));
+        for (int index : table2Rows) {
+          destination.column(joinColumnIndexes.get(i)).append(t2Col, index);
+        }
+      }
+    }
+
+    for (int c = 0; c < destination.columnCount(); c++) {
+      if (!keepTable2JoinKeyColumns) {
+        if (ignoreColumns.contains(c) || joinColumnIndexes.contains(c)) {
+          continue;
+        }
+      }
+      if (c < table1ColCount) {
+        for (int r1 = 0; r1 < table2Rows.size(); r1++) {
+          destination.column(c).appendMissing();
+        }
+      } else {
+        Column t2Col = table2.column(c - table1ColCount);
+        for (int index : table2Rows) {
+          destination.column(c).append(t2Col, index);
+        }
+      }
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "CrossProductJoin";
+  }
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
index 4495302f..5c711c07 100644
--- a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
+++ b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
@@ -1,946 +1,285 @@
 package tech.tablesaw.joining;
 
-import com.google.common.collect.Streams;
-import com.google.common.primitives.Ints;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
+import static tech.tablesaw.joining.JoinType.INNER;
+
+import com.google.common.base.Preconditions;
+import java.util.*;
 import java.util.stream.Collectors;
 import tech.tablesaw.api.*;
-import tech.tablesaw.columns.Column;
-import tech.tablesaw.columns.booleans.BooleanColumnType;
-import tech.tablesaw.columns.dates.DateColumnType;
-import tech.tablesaw.columns.datetimes.DateTimeColumnType;
-import tech.tablesaw.columns.instant.InstantColumnType;
-import tech.tablesaw.columns.numbers.*;
-import tech.tablesaw.columns.strings.StringColumnType;
-import tech.tablesaw.columns.strings.TextColumnType;
-import tech.tablesaw.columns.times.TimeColumnType;
-import tech.tablesaw.index.*;
-import tech.tablesaw.selection.Selection;
 
 /** Implements joins between two or more Tables */
-public class DataFrameJoiner {
-
-  /** The types of joins that are supported */
-  private enum JoinType {
-    INNER,
-    LEFT_OUTER,
-    RIGHT_OUTER,
-    FULL_OUTER
-  }
+public class DataFrameJoiner extends AbstractJoiner {
 
-  private static final String TABLE_ALIAS = "T";
+  /** The join algorithm to be used */
+  private JoinStrategy strategy;
 
+  /** The first (left) table named in the join statement */
   private final Table table;
-  private final String[] joinColumnNames;
-  private final List<Integer> joinColumnIndexes;
-  private final AtomicInteger joinTableId = new AtomicInteger(2);
 
-  /**
-   * Constructor.
-   *
-   * @param table The table to join on.
-   * @param joinColumnNames The join column names to join on.
-   */
-  public DataFrameJoiner(Table table, String... joinColumnNames) {
-    this.table = table;
-    this.joinColumnNames = joinColumnNames;
-    this.joinColumnIndexes = getJoinIndexes(table, joinColumnNames);
-  }
+  /** The names of the columns to be used for the first (left) table */
+  private final String[] leftJoinColumnNames;
 
   /**
-   * Finds the index of the columns corresponding to the columnNames. E.G. The column named "ID" is
-   * located at index 5 in table.
-   *
-   * @param table the table that contains the columns.
-   * @param columnNames the column names to find indexes of.
-   * @return a list of column indexes within the table.
+   * The names of the columns to be joined on in the second (right) table. If these are not
+   * explicitly provided, they default to the names used for the left table.
    */
-  private List<Integer> getJoinIndexes(Table table, String[] columnNames) {
-    return Arrays.stream(columnNames).map(table::columnIndex).collect(Collectors.toList());
-  }
+  private String[] rightJoinColumnNames;
 
-  /**
-   * 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
-   */
-  public Table inner(Table... tables) {
-    return inner(false, tables);
-  }
-
-  /**
-   * Joins 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
-   */
-  public Table inner(boolean allowDuplicateColumnNames, Table... tables) {
-    Table joined = table;
-    for (Table currT : tables) {
-      joined =
-          joinInternal(
-              joined, currT, JoinType.INNER, allowDuplicateColumnNames, false, joinColumnNames);
-    }
-    return joined;
-  }
-
-  /**
-   * 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 inner(Table table2, String col2Name) {
-    return inner(table2, false, col2Name);
-  }
+  /** The positions (indexes) in the table of the columns used in the first table */
+  private int[] leftJoinColumnPositions;
 
   /**
-   * 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
+   * The table(s) to be used on the right side. If more than one table is provided, the join is
+   * executed repeatedly, merging the next right table with the prior results
    */
-  public Table inner(Table table2, String[] col2Names) {
-    return inner(table2, false, col2Names);
-  }
+  private List<Table> rightTables = new ArrayList<>();
 
-  /**
-   * 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.
-   * @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 inner(table2, allowDuplicateColumnNames, col2Name);
-  }
+  /** The type of join to be performed (INNER, LEFT_OUTER, RIGHT_OUTER, or FULL_OUTER */
+  private JoinType joinType = INNER;
 
   /**
-   * 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
+   * When this is false, columns in the second (and subsequent) join tables are excluded from the
+   * results if they have the same name as a column in the any prior table. When it is true, they
+   * are give a prefix and included. The prefix used is "Tn." where n is the number of the table in
+   * the join. The second table is (T2.column_name), for example.
    */
-  public Table inner(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
-    Table joinedTable;
-    joinedTable =
-        joinInternal(table, table2, JoinType.INNER, allowDuplicateColumnNames, false, col2Names);
-    return joinedTable;
-  }
+  private boolean allowDuplicateColumnNames = false;
 
   /**
-   * Joins the joiner to the table2, using the given columns for the second table and returns the
-   * resulting table
+   * When this is true, the columns of the second (and subsequent) join tables are included in the
+   * results, even when they're identical in name and data with the first join table. When false,
+   * only the first join columns are retained.
    *
-   * @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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
-   *     table1 if {@code true} the join will return all join key columns in both table, which may
-   *     have difference when there are null values
-   * @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
+   * <p>If the second (or any subsequent) table has the same join column names as the first (or any
+   * prior) table, the same scheme used for non-join columns is used, and each column with a
+   * duplicate name gets a prefix of "Tn." where n is the number of the table in the join.
    */
-  public Table inner(
-      Table table2,
-      boolean allowDuplicateColumnNames,
-      boolean keepAllJoinKeyColumns,
-      String... col2Names) {
-    return joinInternal(
-        table, table2, JoinType.INNER, allowDuplicateColumnNames, keepAllJoinKeyColumns, col2Names);
-  }
+  private boolean keepAllJoinKeyColumns = false;
 
   /**
-   * Joins two tables.
+   * Constructor.
    *
-   * @param table1 the table on the left side of the join.
-   * @param table2 the table on the right side of the join.
-   * @param joinType the type of join.
-   * @param allowDuplicates 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
-   *     table1 if {@code true} the join will return all join key columns in both table, which may
-   *     have difference when there are null values
-   * @param table2JoinColumnNames The names of the columns in table2 to join on.
-   * @return the joined table
+   * @param table The table to join on.
+   * @param leftJoinColumnNames The join column names in that table to be used. These names also
+   *     serve as the default for the second table, unless other names are explicitly provided.
    */
-  private Table joinInternal(
-      Table table1,
-      Table table2,
-      JoinType joinType,
-      boolean allowDuplicates,
-      boolean keepAllJoinKeyColumns,
-      String... table2JoinColumnNames) {
-
-    List<Integer> table2JoinColumnIndexes = getJoinIndexes(table2, table2JoinColumnNames);
-    List<Index> table1Indexes = buildIndexesForJoinColumns(joinColumnIndexes, table1);
-    List<Index> table2Indexes = buildIndexesForJoinColumns(table2JoinColumnIndexes, table2);
-
-    Table result = Table.create(table1.name());
-    // A set of column indexes in the result table that can be ignored. They are duplicate join
-    // keys.
-    Set<Integer> resultIgnoreColIndexes =
-        emptyTableFromColumns(
-            result,
-            table1,
-            table2,
-            joinType,
-            allowDuplicates,
-            table2JoinColumnIndexes,
-            keepAllJoinKeyColumns);
-
-    validateIndexes(table1Indexes, table2Indexes);
-    if (table1.rowCount() == 0 && (joinType == JoinType.LEFT_OUTER || joinType == JoinType.INNER)) {
-      // Handle special case of empty table here so it doesn't fall through to the behavior
-      // that adds rows for full outer and right outer joins
-      if (!keepAllJoinKeyColumns) {
-        result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
-      }
-      return result;
-    }
-
-    Selection table1DoneRows = Selection.with();
-    Selection table2DoneRows = Selection.with();
-    // use table 2 for row iteration, which can significantly increase performance
-    if (table1.rowCount() > table2.rowCount() && joinType == JoinType.INNER) {
-      for (Row row : table2) {
-        int ri = row.getRowNumber();
-        if (table2DoneRows.contains(ri)) {
-          // Already processed a selection of table1 that contained this row.
-          continue;
-        }
-        Selection table1Rows =
-            createMultiColSelection(
-                table2, ri, table1Indexes, table1.rowCount(), table2JoinColumnIndexes);
-        Selection table2Rows =
-            createMultiColSelection(
-                table2, ri, table2Indexes, table2.rowCount(), table2JoinColumnIndexes);
-        crossProduct(
-            result,
-            table1,
-            table2,
-            table1Rows,
-            table2Rows,
-            resultIgnoreColIndexes,
-            keepAllJoinKeyColumns);
-
-        table2DoneRows = table2DoneRows.or(table2Rows);
-        if (table2DoneRows.size() == table2.rowCount()) {
-          // Processed all the rows in table1 exit early.
-          if (!keepAllJoinKeyColumns) {
-            result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
-          }
-          return result;
-        }
-      }
-    } else {
-      for (Row row : table1) {
-        int ri = row.getRowNumber();
-        if (table1DoneRows.contains(ri)) {
-          // Already processed a selection of table1 that contained this row.
-          continue;
-        }
-        Selection table1Rows =
-            createMultiColSelection(
-                table1, ri, table1Indexes, table1.rowCount(), joinColumnIndexes);
-        Selection table2Rows =
-            createMultiColSelection(
-                table1, ri, table2Indexes, table2.rowCount(), joinColumnIndexes);
-        if ((joinType == JoinType.LEFT_OUTER || joinType == JoinType.FULL_OUTER)
-            && table2Rows.isEmpty()) {
-          withMissingLeftJoin(
-              result, table1, table1Rows, resultIgnoreColIndexes, keepAllJoinKeyColumns);
-        } else {
-          crossProduct(
-              result,
-              table1,
-              table2,
-              table1Rows,
-              table2Rows,
-              resultIgnoreColIndexes,
-              keepAllJoinKeyColumns);
-        }
-        table1DoneRows = table1DoneRows.or(table1Rows);
-        if (joinType == JoinType.FULL_OUTER || joinType == JoinType.RIGHT_OUTER) {
-          // Update done rows in table2 for full Outer.
-          table2DoneRows = table2DoneRows.or(table2Rows);
-        } else if (table1DoneRows.size() == table1.rowCount()) {
-          // Processed all the rows in table1 exit early.
-          if (!keepAllJoinKeyColumns) {
-            result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
-          }
-          return result;
-        }
-      }
-    }
-
-    // Add all rows from table2 that were not handled already.
-    Selection table2Rows = table2DoneRows.flip(0, table2.rowCount());
-    withMissingRight(
-        result,
-        table1.columnCount(),
-        table2,
-        table2Rows,
-        joinType,
-        table2JoinColumnIndexes,
-        resultIgnoreColIndexes,
-        keepAllJoinKeyColumns);
-    if (!keepAllJoinKeyColumns) {
-      result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
-    }
-    return result;
-  }
-
-  private void validateIndexes(List<Index> table1Indexes, List<Index> table2Indexes) {
-    if (table1Indexes.size() != table2Indexes.size()) {
-      throw new IllegalArgumentException(
-          "Cannot join using a different number of indices on each table: "
-              + table1Indexes
-              + " and "
-              + table2Indexes);
-    }
-    for (int i = 0; i < table1Indexes.size(); i++) {
-      if (!table1Indexes.get(i).getClass().equals(table2Indexes.get(i).getClass())) {
-        throw new IllegalArgumentException(
-            "Cannot join using different index types: " + table1Indexes + " and " + table2Indexes);
-      }
-    }
-  }
+  public DataFrameJoiner(Table table, String... leftJoinColumnNames) {
+    this.table = table;
+    this.leftJoinColumnNames = leftJoinColumnNames;
 
-  /** Build a reverse index for every join column in the table. */
-  private List<Index> buildIndexesForJoinColumns(List<Integer> joinColumnIndexes, Table table) {
-    return joinColumnIndexes.stream().map(c -> indexFor(table, c)).collect(Collectors.toList());
-  }
+    // we assume the join columns for both tables have the same names,
+    // unless names for the right table are explicitly set
+    this.rightJoinColumnNames = leftJoinColumnNames;
 
-  /** Create a reverse index for a given column. */
-  private Index indexFor(Table table, int colIndex) {
-    ColumnType type = table.column(colIndex).type();
-    if (type instanceof DateColumnType) {
-      return new IntIndex(table.dateColumn(colIndex));
-    } else if (type instanceof DateTimeColumnType) {
-      return new LongIndex(table.dateTimeColumn(colIndex));
-    } else if (type instanceof InstantColumnType) {
-      return new LongIndex(table.instantColumn(colIndex));
-    } else if (type instanceof TimeColumnType) {
-      return new IntIndex(table.timeColumn(colIndex));
-    } else if (type instanceof StringColumnType) {
-      return new StringIndex(table.stringColumn(colIndex));
-    } else if (type instanceof TextColumnType) {
-      return new StringIndex(table.textColumn(colIndex));
-    } else if (type instanceof IntColumnType) {
-      return new IntIndex(table.intColumn(colIndex));
-    } else if (type instanceof LongColumnType) {
-      return new LongIndex(table.longColumn(colIndex));
-    } else if (type instanceof ShortColumnType) {
-      return new ShortIndex(table.shortColumn(colIndex));
-    } else if (type instanceof BooleanColumnType) {
-      return new ByteIndex(table.booleanColumn(colIndex));
-    } else if (type instanceof DoubleColumnType) {
-      return new DoubleIndex(table.doubleColumn(colIndex));
-    } else if (type instanceof FloatColumnType) {
-      return new FloatIndex(table.floatColumn(colIndex));
-    }
-    throw new IllegalArgumentException("Joining attempted on unsupported column type " + type);
+    this.leftJoinColumnPositions = getJoinIndexes(table, leftJoinColumnNames);
   }
 
   /**
-   * Given a reverse index find a selection of rows that have the same value as the supplied column
-   * does in the given row index.
-   */
-  private Selection selectionForColumn(Column<?> valueColumn, int rowIndex, Index rawIndex) {
-
-    ColumnType type = valueColumn.type();
-    if (type instanceof DateColumnType) {
-      IntIndex index = (IntIndex) rawIndex;
-      int value = ((DateColumn) valueColumn).getIntInternal(rowIndex);
-      return index.get(value);
-    } else if (type instanceof TimeColumnType) {
-      IntIndex index = (IntIndex) rawIndex;
-      int value = ((TimeColumn) valueColumn).getIntInternal(rowIndex);
-      return index.get(value);
-    } else if (type instanceof DateTimeColumnType) {
-      LongIndex index = (LongIndex) rawIndex;
-      long value = ((DateTimeColumn) valueColumn).getLongInternal(rowIndex);
-      return index.get(value);
-    } else if (type instanceof InstantColumnType) {
-      LongIndex index = (LongIndex) rawIndex;
-      long value = ((InstantColumn) valueColumn).getLongInternal(rowIndex);
-      return index.get(value);
-    } else if (type instanceof StringColumnType) {
-      StringIndex index = (StringIndex) rawIndex;
-      String value = ((StringColumn) valueColumn).get(rowIndex);
-      return index.get(value);
-    } else if (type instanceof TextColumnType) {
-      StringIndex index = (StringIndex) rawIndex;
-      String value = ((TextColumn) valueColumn).get(rowIndex);
-      return index.get(value);
-    } else if (type instanceof IntColumnType) {
-      IntIndex index = (IntIndex) rawIndex;
-      int value = ((IntColumn) valueColumn).getInt(rowIndex);
-      return index.get(value);
-    } else if (type instanceof LongColumnType) {
-      LongIndex index = (LongIndex) rawIndex;
-      long value = ((LongColumn) valueColumn).getLong(rowIndex);
-      return index.get(value);
-    } else if (type instanceof ShortColumnType) {
-      ShortIndex index = (ShortIndex) rawIndex;
-      short value = ((ShortColumn) valueColumn).getShort(rowIndex);
-      return index.get(value);
-    } else if (type instanceof BooleanColumnType) {
-      ByteIndex index = (ByteIndex) rawIndex;
-      byte value = ((BooleanColumn) valueColumn).getByte(rowIndex);
-      return index.get(value);
-    } else if (type instanceof DoubleColumnType) {
-      DoubleIndex index = (DoubleIndex) rawIndex;
-      double value = ((DoubleColumn) valueColumn).getDouble(rowIndex);
-      return index.get(value);
-    } else if (type instanceof FloatColumnType) {
-      FloatIndex index = (FloatIndex) rawIndex;
-      float value = ((FloatColumn) valueColumn).getFloat(rowIndex);
-      return index.get(value);
-    } else {
-      throw new IllegalArgumentException(
-          "Joining is supported on numeric, string, and date-like columns. Column "
-              + valueColumn.name()
-              + " is of type "
-              + valueColumn.type());
-    }
-  }
-  /**
-   * Create a big multicolumn selection for all join columns in the given table. Joins two tables.
+   * Sets the type of join, which defaults to INNER if not provided.
    *
-   * @param table the table that used to generate Selection.
-   * @param ri row number of row in table.
-   * @param indexes a reverse index for every join column in the table.
-   * @param selectionSize max size in table .
-   * @param joinColumnIndexes the column index of join key in tables
-   * @return selection created
+   * @param joinType The type of join to perform (INNER, LEFT_OUTER, RIGHT_OUTER, FULL_OUTER)
+   * @return This joiner object.
    */
-  private Selection createMultiColSelection(
-      Table table,
-      int ri,
-      List<Index> indexes,
-      int selectionSize,
-      List<Integer> joinColumnIndexes) {
-    Selection multiColSelection = Selection.withRange(0, selectionSize);
-    int i = 0;
-    for (Integer joinColumnIndex : joinColumnIndexes) {
-      Column<?> col = table.column(joinColumnIndex);
-      Selection oneColSelection = selectionForColumn(col, ri, indexes.get(i));
-      // and the selections.
-      multiColSelection = multiColSelection.and(oneColSelection);
-      i++;
-    }
-    return multiColSelection;
-  }
-
-  private String newName(String table2Alias, String columnName) {
-    return table2Alias + "." + columnName;
+  public DataFrameJoiner type(JoinType joinType) {
+    this.joinType = joinType;
+    return this;
   }
 
   /**
-   * Full outer join to the given tables assuming that they have a column of the name we're joining
-   * on
+   * When the argument is true, the join columns of the second (and subsequent) tables are included
+   * in the results, even when they're identical in name and data with the first join table. When
+   * false, only one set of join columns is retained in the result.
    *
-   * @param tables The tables to join with
-   * @return The resulting table
-   */
-  public Table fullOuter(Table... tables) {
-    return fullOuter(false, tables);
-  }
-
-  /**
-   * Full outer join to the given tables assuming that they have a column of the name we're joining
-   * on
+   * <p>Note that if the second (or any subsequent) table has the same join column names as the
+   * first (or any prior) table, the same scheme used for non-join columns is used, and each column
+   * with a duplicate name gets a prefix of "Tn." where n is the number of the table in the join.
    *
-   * @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 currT : tables) {
-      joined =
-          joinInternal(
-              joined,
-              currT,
-              JoinType.FULL_OUTER,
-              allowDuplicateColumnNames,
-              false,
-              joinColumnNames);
-    }
-    return joined;
-  }
-
-  /**
-   * Joins the joiner to the table2, using the given columns for the second table and returns the
-   * resulting table
+   * <p>If this method is not called, the default is false
    *
-   * @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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
-   *     table1 if {@code true} the join will return all join key columns in both table, which may
-   *     have difference when there are null values
-   * @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
+   * @param keep true or false
+   * @return this DataFrameJoiner instance
    */
-  public Table fullOuter(
-      Table table2,
-      boolean allowDuplicateColumnNames,
-      boolean keepAllJoinKeyColumns,
-      String... col2Names) {
-    return joinInternal(
-        table,
-        table2,
-        JoinType.FULL_OUTER,
-        allowDuplicateColumnNames,
-        keepAllJoinKeyColumns,
-        col2Names);
+  public DataFrameJoiner keepAllJoinKeyColumns(boolean keep) {
+    this.keepAllJoinKeyColumns = keep;
+    return this;
   }
 
   /**
-   * Full outer join the joiner to the table2, using the given column for the second table and
-   * returns the resulting table
+   * 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 and included in
+   * the results. Specifically, the renamed columns are given a are give a prefix and the prefix
+   * used is "Tn." where n is the number of the table in the join. The second table is
+   * (T2.column_name), for example.
    *
-   * @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 joinInternal(table, table2, JoinType.FULL_OUTER, false, false, col2Name);
-  }
-
-  /**
-   * Joins to the given tables assuming that they have a column of the name we're joining on
+   * <p>See also {@link DataFrameJoiner#keepAllJoinKeyColumns(boolean)} to determine whether to
+   * retain the join columns from the second table
    *
-   * @param tables The tables to join with
-   * @return The resulting table
+   * @param allow true, if columns with duplicate names are to be retained; false otherwise. Default
+   *     is false
+   * @return this DataFrameJoiner instance
    */
-  public Table leftOuter(Table... tables) {
-    return leftOuter(false, tables);
+  public DataFrameJoiner allowDuplicateColumnNames(boolean allow) {
+    this.allowDuplicateColumnNames = allow;
+    return this;
   }
 
   /**
-   * Joins to the given tables assuming that they have a column of the name we're joining on
+   * The names of the columns to be joined on in the second (right) table. If this method is not
+   * called, they default to the names used for the left table.
    *
-   * @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
+   * @param rightJoinColumnNames The names to be used
+   * @return This DataFrameJoiner instance
    */
-  public Table leftOuter(boolean allowDuplicateColumnNames, Table... tables) {
-    Table joined = table;
-    for (Table table2 : tables) {
-      joined =
-          joinInternal(
-              joined,
-              table2,
-              JoinType.LEFT_OUTER,
-              allowDuplicateColumnNames,
-              false,
-              joinColumnNames);
-    }
-    return joined;
+  public DataFrameJoiner rightJoinColumns(String... rightJoinColumnNames) {
+    Preconditions.checkNotNull(rightJoinColumnNames);
+    this.rightJoinColumnNames = rightJoinColumnNames;
+    return this;
   }
 
   /**
-   * Joins the joiner to the table2, using the given columns for the second table and returns the
-   * resulting table
+   * The table or tables to be used on the right side of the join. If more than one table is
+   * provided, the join is executed repeatedly, merging the next right table with the prior results
    *
-   * @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
+   * @param tables The table or tables to be used on the right side
+   * @return This DataFrameJoiner instance
    */
-  public Table leftOuter(Table table2, String[] col2Names) {
-    return leftOuter(table2, false, col2Names);
+  public DataFrameJoiner with(Table... tables) {
+    Preconditions.checkNotNull(tables);
+    this.rightTables = Arrays.stream(tables).collect(Collectors.toList());
+    return this;
   }
 
   /**
-   * Joins the joiner to the table2, using the given column for the second table and returns the
-   * resulting table
+   * Performs the actual join and returns the results
    *
-   * @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
+   * @return The combined table
    */
-  public Table leftOuter(Table table2, String col2Name) {
-    return leftOuter(table2, false, col2Name);
-  }
+  public Table join() {
 
-  /**
-   * 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 leftOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
-    return joinInternal(
-        table, table2, JoinType.LEFT_OUTER, allowDuplicateColumnNames, false, col2Names);
-  }
+    selectJoinStrategy();
 
-  /**
-   * 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
-   *     table1 if {@code true} the join will return all join key columns in both table, which may
-   *     have difference when there are null values
-   * @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,
-      boolean allowDuplicateColumnNames,
-      boolean keepAllJoinKeyColumns,
-      String... col2Names) {
-    return joinInternal(
-        table,
-        table2,
-        JoinType.LEFT_OUTER,
-        allowDuplicateColumnNames,
-        keepAllJoinKeyColumns,
-        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);
-  }
+    if (!allowDuplicateColumnNames) {
+      Set<String> rightJoinColumns = Set.of(rightJoinColumnNames);
+      Set<String> leftJoinColumns = Set.of(leftJoinColumnNames);
+      Set<String> nonJoinColumns =
+          table.columnNames().stream()
+              .filter(e -> !leftJoinColumns.contains(e))
+              .collect(Collectors.toSet());
 
-  /**
-   * Joins 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 rightOuter(boolean allowDuplicateColumnNames, Table... tables) {
-    Table joined = table;
-    for (Table table2 : tables) {
-      joined =
-          joinInternal(
-              joined,
-              table2,
-              JoinType.RIGHT_OUTER,
-              allowDuplicateColumnNames,
-              false,
-              joinColumnNames);
-      joinColumnIndexes.clear();
-      joinColumnIndexes.addAll(getJoinIndexes(joined, joinColumnNames));
+      for (Table t : rightTables) {
+        List<String> names =
+            t.columnNames().stream()
+                .filter(e -> !rightJoinColumns.contains(e))
+                .collect(Collectors.toList());
+        for (String nm : names) {
+          if (!nonJoinColumns.contains(nm)) {
+            nonJoinColumns.add(nm);
+          } else {
+            throw new IllegalArgumentException(
+                "Attempting to join tables containing non-join columns with at least one name: "
+                    + nm
+                    + " appears in more than one table. "
+                    + "If you would like to join tables containing columns with duplicate names, "
+                    + " the value of 'allowDuplicateColumnNames' must be true");
+          }
+        }
+      }
     }
-    return joined;
+    return performJoin(table, rightTables);
   }
 
-  /**
-   * 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 rightOuter(Table table2, String col2Name) {
-    return rightOuter(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 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 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, boolean allowDuplicateColumnNames, String... col2Names) {
-    return joinInternal(
-        table, table2, JoinType.RIGHT_OUTER, allowDuplicateColumnNames, false, col2Names);
-  }
+  private void selectJoinStrategy() {
+    // System.out.println(table);
+    // System.out.println(rightTables.get(0));
 
-  /**
-   * 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
-   *     table1 if {@code true} the join will return all join key columns in both table, which may
-   *     have difference when there are null values
-   * @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,
-      boolean allowDuplicateColumnNames,
-      boolean keepAllJoinKeyColumns,
-      String... col2Names) {
-    return joinInternal(
-        table,
-        table2,
-        JoinType.RIGHT_OUTER,
-        allowDuplicateColumnNames,
-        keepAllJoinKeyColumns,
-        col2Names);
-  }
-
-  /**
-   * Adds empty columns to the destination table with the same type as columns in table1 and table2.
-   *
-   * <p>For inner, left and full outer join types the join columns in table2 are not needed and will
-   * be marked as placeholders. The indexes of those columns will be returned. The downstream logic
-   * is easier if we wait to remove the redundant columns until the last step.
-   *
-   * @param destination the table to fill up with columns. Will be mutated in place.
-   * @param table1 the table on left side of the join.
-   * @param table2 the table on the right side of the join.
-   * @param joinType the type of join.
-   * @param allowDuplicates whether to allow duplicates. If yes rename columns in table2 that have
-   *     the same name as columns in table1 with the exception of join columns in table2 when
-   *     performing a right join.
-   * @param table2JoinColumnIndexes the index locations of the table2 join columns.
-   * @return A
-   */
-  private Set<Integer> emptyTableFromColumns(
-      Table destination,
-      Table table1,
-      Table table2,
-      JoinType joinType,
-      boolean allowDuplicates,
-      List<Integer> table2JoinColumnIndexes,
-      boolean keepTable2JoinKeyColumns) {
+    int leftRowCount = table.rowCount();
+    int rightRowCount = rightTables.get(0).rowCount();
 
-    Column<?>[] cols =
-        Streams.concat(table1.columns().stream(), table2.columns().stream())
-            .map(Column::emptyCopy)
-            .toArray(Column[]::new);
+    int minCardinalityLeft = Integer.MAX_VALUE;
+    int minCardinalityRight = Integer.MAX_VALUE;
 
-    // For inner join, left join and full outer join mark the join columns in table2 as
-    // placeholders.
-    // For right join mark the join columns in table1 as placeholders.
-    // Keep track of which join columns are placeholders so they can be ignored.
-    Set<Integer> ignoreColumns = new HashSet<>();
-    for (int c = 0; c < cols.length; c++) {
-      if (joinType == JoinType.RIGHT_OUTER) {
-        if (c < table1.columnCount() && joinColumnIndexes.contains(c)) {
-          if (!keepTable2JoinKeyColumns) {
-            cols[c].setName("Placeholder_" + ignoreColumns.size());
-          }
-          ignoreColumns.add(c);
-        }
-      } else {
-        int table2Index = c - table1.columnCount();
-        if (c >= table1.columnCount() && table2JoinColumnIndexes.contains(table2Index)) {
-          if (!keepTable2JoinKeyColumns) {
-            cols[c].setName("Placeholder_" + ignoreColumns.size());
-          }
-          ignoreColumns.add(c);
-        }
+    for (int i = 0; i < rightJoinColumnNames.length; i++) {
+      int cardinality = table.column(leftJoinColumnNames[i]).countUnique();
+      if (cardinality < minCardinalityLeft) {
+        minCardinalityLeft = cardinality;
       }
     }
-
-    // Rename duplicate columns in second table
-    if (allowDuplicates) {
-      Set<String> table1ColNames =
-          Arrays.stream(cols)
-              .map(Column::name)
-              .map(String::toLowerCase)
-              .limit(table1.columnCount())
-              .collect(Collectors.toSet());
-
-      String table2Alias = TABLE_ALIAS + joinTableId.getAndIncrement();
-      for (int c = table1.columnCount(); c < cols.length; c++) {
-        String columnName = cols[c].name();
-        if (table1ColNames.contains(columnName.toLowerCase())) {
-          cols[c].setName(newName(table2Alias, columnName));
-        }
+    for (String rightJoinColumnName : rightJoinColumnNames) {
+      int cardinality = rightTables.get(0).column(rightJoinColumnName).countUnique();
+      if (cardinality < minCardinalityRight) {
+        minCardinalityRight = cardinality;
       }
     }
-    destination.addColumns(cols);
-    return ignoreColumns;
+    //    System.out.println("min cardinality left " + minCardinalityLeft);
+    //    System.out.println("min cardinality right " + minCardinalityRight);
+    //    System.out.println("Avg values left " + (leftRowCount/(minCardinalityLeft * 1.0)));
+    //    System.out.println("Avg values right " + rightRowCount / minCardinalityRight);
+    if ((leftRowCount / (minCardinalityLeft * 1.0)) > 1000
+        || (rightRowCount / (minCardinalityRight * 1.0) > 1000)) {
+      this.strategy = new SortMergeJoin(table, leftJoinColumnNames);
+    } else {
+      this.strategy = new CrossProductJoin(table, leftJoinColumnNames);
+    }
+    //    System.out.println(strategy + " selected.");
   }
 
   /**
-   * Creates cross product for the selection of two tables.
+   * Finds the position of the columns corresponding to the columnNames. E.G. The column named "ID"
+   * is located at position 5 (0-based) in the table.
    *
-   * @param destination the destination table.
-   * @param table1 the table on left of join.
-   * @param table2 the table on right of join.
-   * @param table1Rows the selection of rows in table1.
-   * @param table2Rows the selection of rows in table2.
-   * @param ignoreColumns a set of column indexes in the result to ignore. They are redundant join
-   *     columns.
+   * @param table the table that contains the columns.
+   * @param columnNames the column names to find position of.
+   * @return a list of column indexes within the table.
    */
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private void crossProduct(
-      Table destination,
-      Table table1,
-      Table table2,
-      Selection table1Rows,
-      Selection table2Rows,
-      Set<Integer> ignoreColumns,
-      boolean keepTable2JoinKeyColumns) {
-    for (int c = 0; c < table1.columnCount() + table2.columnCount(); c++) {
-      if (!keepTable2JoinKeyColumns && ignoreColumns.contains(c)) {
-        continue;
-      }
-      int table2Index = c - table1.columnCount();
-      for (int r1 : table1Rows) {
-        for (int r2 : table2Rows) {
-          if (c < table1.columnCount()) {
-            Column t1Col = table1.column(c);
-            destination.column(c).append(t1Col, r1);
-          } else {
-            Column t2Col = table2.column(table2Index);
-            destination.column(c).append(t2Col, r2);
-          }
-        }
-      }
+  private int[] getJoinIndexes(Table table, String[] columnNames) {
+    int[] results = new int[columnNames.length];
+    for (int i = 0; i < columnNames.length; i++) {
+      String nm = columnNames[i];
+      results[i] = table.columnIndex(nm);
     }
+    return results;
   }
 
   /**
-   * Adds rows to destination for each row in table1 with the columns from table2 added as missing
-   * values.
-   */
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private void withMissingLeftJoin(
-      Table destination,
-      Table table1,
-      Selection table1Rows,
-      Set<Integer> ignoreColumns,
-      boolean keepTable2JoinKeyColumns) {
-    for (int c = 0; c < destination.columnCount(); c++) {
-      if (!keepTable2JoinKeyColumns && ignoreColumns.contains(c)) {
-        continue;
-      }
-      if (c < table1.columnCount()) {
-        Column t1Col = table1.column(c);
-        for (int index : table1Rows) {
-          destination.column(c).append(t1Col, index);
-        }
-      } else {
-        for (int r1 = 0; r1 < table1Rows.size(); r1++) {
-          destination.column(c).appendMissing();
-        }
-      }
+   * Recursively joins the table on the left with each of the tables on the right, substituting the
+   * result of the nth join as the left table for the nth + 1 join
+   *
+   * @param left The first (left) table participating in the join
+   * @param rightTables One or more tables to be joined with the first (left) table
+   * @return this DataFrameJoiner instance
+   */
+  private Table performJoin(Table left, List<Table> rightTables) {
+    Table result =
+        joinInternal(
+            left,
+            rightTables.remove(0),
+            joinType,
+            allowDuplicateColumnNames,
+            keepAllJoinKeyColumns,
+            rightJoinColumnNames);
+    if (rightTables.isEmpty()) {
+      return result;
+    } else {
+      // on subsequent calls, the left column may have a new structure
+      this.leftJoinColumnPositions = getJoinIndexes(result, leftJoinColumnNames);
+      return performJoin(result, rightTables);
     }
   }
 
-  /**
-   * Adds rows to destination for each row in table2 with the columns from table1 added as missing
-   * values.
-   */
-  @SuppressWarnings({"rawtypes", "unchecked"})
-  private void withMissingRight(
-      Table destination,
-      int table1ColCount,
+  Table joinInternal(
+      Table table1,
       Table table2,
-      Selection table2Rows,
       JoinType joinType,
-      List<Integer> col2Indexes,
-      Set<Integer> ignoreColumns,
-      boolean keepTable2JoinKeyColumns) {
-
-    // Add index data from table2 into join column positions in table one.
-    if (joinType == JoinType.FULL_OUTER) {
-      for (int i = 0; i < col2Indexes.size(); i++) {
-        Column t2Col = table2.column(col2Indexes.get(i));
-        for (int index : table2Rows) {
-          destination.column(joinColumnIndexes.get(i)).append(t2Col, index);
-        }
-      }
-    }
-
-    for (int c = 0; c < destination.columnCount(); c++) {
-      if (!keepTable2JoinKeyColumns) {
-        if (ignoreColumns.contains(c) || joinColumnIndexes.contains(c)) {
-          continue;
-        }
-      }
-      if (c < table1ColCount) {
-        for (int r1 = 0; r1 < table2Rows.size(); r1++) {
-          destination.column(c).appendMissing();
-        }
-      } else {
-        Column t2Col = table2.column(c - table1ColCount);
-        for (int index : table2Rows) {
-          destination.column(c).append(t2Col, index);
-        }
-      }
-    }
+      boolean allowDuplicates,
+      boolean keepAllJoinKeyColumns,
+      String... rightJoinColumnPositions) {
+    return strategy.performJoin(
+        table1,
+        table2,
+        joinType,
+        allowDuplicates,
+        keepAllJoinKeyColumns,
+        leftJoinColumnPositions,
+        rightJoinColumnPositions);
   }
 }
diff --git a/core/src/main/java/tech/tablesaw/joining/JoinStrategy.java b/core/src/main/java/tech/tablesaw/joining/JoinStrategy.java
new file mode 100644
index 00000000..ecdb92b1
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/JoinStrategy.java
@@ -0,0 +1,15 @@
+package tech.tablesaw.joining;
+
+import tech.tablesaw.api.Table;
+
+interface JoinStrategy {
+
+  Table performJoin(
+      Table table1,
+      Table table2,
+      JoinType joinType,
+      boolean allowDuplicates,
+      boolean keepAllJoinKeyColumns,
+      int[] leftJoinColumnIndexes,
+      String... table2JoinColumnNames);
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/JoinType.java b/core/src/main/java/tech/tablesaw/joining/JoinType.java
new file mode 100644
index 00000000..c0cd3788
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/JoinType.java
@@ -0,0 +1,9 @@
+package tech.tablesaw.joining;
+
+/** The types of joins that are supported */
+public enum JoinType {
+  INNER,
+  LEFT_OUTER,
+  RIGHT_OUTER,
+  FULL_OUTER
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/RowComparatorChain.java b/core/src/main/java/tech/tablesaw/joining/RowComparatorChain.java
new file mode 100644
index 00000000..808a5c21
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/RowComparatorChain.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package tech.tablesaw.joining;
+
+import java.util.*;
+import tech.tablesaw.api.Row;
+
+/**
+ * A comparator for comparing two Row objects on one or more Columns. The two rows are expected to
+ * be from different tables, but they should have the same number and type of Columns being used for
+ * their sorts The implementation is based on Apache Commons Collections
+ */
+class RowComparatorChain implements Comparator<Row> {
+
+  private final List<Comparator<Row>> comparatorChain;
+  private BitSet orderingBits;
+  private boolean isLocked;
+
+  /** Constructs a comparator chain with the argument as the first node in the chain */
+  public RowComparatorChain(Comparator<Row> comparator) {
+    this(comparator, false);
+  }
+
+  private RowComparatorChain(Comparator<Row> comparator, boolean reverse) {
+    this.orderingBits = null;
+    this.isLocked = false;
+    this.comparatorChain = new ArrayList<>(1);
+    this.comparatorChain.add(comparator);
+    this.orderingBits = new BitSet(1);
+    if (reverse) {
+      this.orderingBits.set(0);
+    }
+  }
+
+  /** Appends the comparator to the end of the chain */
+  public void addComparator(Comparator<Row> comparator) {
+    this.comparatorChain.add(comparator);
+  }
+
+  /** Returns the number of comparators in the chain */
+  public int size() {
+    return this.comparatorChain.size();
+  }
+
+  private void checkChainIntegrity() {
+    if (this.comparatorChain.isEmpty()) {
+      throw new UnsupportedOperationException(
+          "ComparatorChains must contain at least one Comparator");
+    }
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * @throws UnsupportedOperationException
+   */
+  @Override
+  public int compare(Row o1, Row o2) throws UnsupportedOperationException {
+    if (!this.isLocked) {
+      this.checkChainIntegrity();
+      this.isLocked = true;
+    }
+
+    Iterator<Comparator<Row>> comparators = this.comparatorChain.iterator();
+
+    for (int comparatorIndex = 0; comparators.hasNext(); ++comparatorIndex) {
+      Comparator<Row> comparator = comparators.next();
+      int retval = comparator.compare(o1, o2);
+      if (retval != 0) {
+        if (this.orderingBits.get(comparatorIndex)) {
+          if (retval > 0) {
+            retval = -1;
+          } else {
+            retval = 1;
+          }
+        }
+        return retval;
+      }
+    }
+    return 0;
+  }
+
+  @Override
+  public int hashCode() {
+    int hash = 0;
+    if (null != this.comparatorChain) {
+      hash ^= this.comparatorChain.hashCode();
+    }
+
+    if (null != this.orderingBits) {
+      hash ^= this.orderingBits.hashCode();
+    }
+    return hash;
+  }
+
+  @Override
+  public boolean equals(Object object) {
+    if (this == object) {
+      return true;
+    } else if (null == object) {
+      return false;
+    } else if (!object.getClass().equals(this.getClass())) {
+      return false;
+    } else {
+      label48:
+      {
+        label32:
+        {
+          RowComparatorChain chain = (RowComparatorChain) object;
+          if (null == this.orderingBits) {
+            if (null != chain.orderingBits) {
+              break label32;
+            }
+          } else if (!this.orderingBits.equals(chain.orderingBits)) {
+            break label32;
+          }
+
+          if (null == this.comparatorChain) {
+            if (null == chain.comparatorChain) {
+              break label48;
+            }
+          } else if (this.comparatorChain.equals(chain.comparatorChain)) {
+            break label48;
+          }
+        }
+        return false;
+      }
+      return true;
+    }
+  }
+
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder("RowComparatorChain{");
+    sb.append("comparatorChain=").append(comparatorChain);
+    sb.append('}');
+    return sb.toString();
+  }
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/SortKey.java b/core/src/main/java/tech/tablesaw/joining/SortKey.java
new file mode 100644
index 00000000..5c89f64f
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/SortKey.java
@@ -0,0 +1,183 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package tech.tablesaw.joining;
+
+import com.google.common.base.MoreObjects;
+import java.util.*;
+import tech.tablesaw.api.ColumnType;
+import tech.tablesaw.api.Row;
+
+/**
+ * SortKey is basically a specification for a sort. It defines the sort required for a
+ * merge-sort-join. The sort order is defined such that the tables being joined are both sorted
+ * independently on the join columns. All columns being sorted are sorted in ascending order
+ */
+class SortKey implements Iterable<ColumnIndexPair> {
+
+  /** Describes how the tables are to be sorted */
+  private final ArrayList<ColumnIndexPair> sortOrder = new ArrayList<>();
+
+  private SortKey(ColumnIndexPair pair) {
+    next(pair);
+  }
+
+  /**
+   * Returns a new SortKey defining the first sort (for the first join column)
+   *
+   * @param pair The details of the sort, i.e. what type of column and the index of the columns in
+   *     the respective tables.
+   */
+  public static SortKey on(ColumnIndexPair pair) {
+    return new SortKey(pair);
+  }
+
+  /**
+   * Returns a new SortKey defining an additional sort clause
+   *
+   * @param pair The details of the sort, i.e. what type of column and the index of the columns in
+   *     the respective tables.
+   */
+  public SortKey next(ColumnIndexPair pair) {
+    sortOrder.add(pair);
+    return this;
+  }
+
+  /** Returns true if no order has been set */
+  public boolean isEmpty() {
+    return sortOrder.isEmpty();
+  }
+
+  /** Returns the number of columns used in this sort */
+  public int size() {
+    return sortOrder.size();
+  }
+
+  /**
+   * Returns a new SortKey for the given ColumnIndexPairs. A table being sorted on three columns,
+   * will have three pairs in the SortKey
+   */
+  public static SortKey create(List<ColumnIndexPair> pairs) {
+    SortKey key = null;
+
+    for (ColumnIndexPair pair : pairs) {
+      if (key == null) { // key will be null the first time through
+        key = new SortKey(pair);
+      } else {
+        key.next(pair);
+      }
+    }
+    return key;
+  }
+
+  /**
+   * Returns a ComparatorChain consisting of one or more comparators as specified in the given
+   * SortKey
+   */
+  static RowComparatorChain getChain(SortKey key) {
+    Iterator<ColumnIndexPair> entries = key.iterator();
+    ColumnIndexPair sort = entries.next();
+    Comparator<Row> comparator = comparator(sort);
+
+    RowComparatorChain chain = new RowComparatorChain(comparator);
+    while (entries.hasNext()) {
+      sort = entries.next();
+      chain.addComparator(comparator(sort));
+    }
+    return chain;
+  }
+
+  /** Returns a comparator for a given ColumnIndexPair */
+  private static Comparator<Row> comparator(ColumnIndexPair pair) {
+    if (pair.type.equals(ColumnType.INTEGER)) {
+      return (r11, r21) -> {
+        int b1 = r11.getInt(pair.left);
+        int b2 = r21.getInt(pair.right);
+        return Integer.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.LOCAL_DATE)) {
+      return (r11, r21) -> {
+        int b1 = r11.getPackedDate(pair.left);
+        int b2 = r21.getPackedDate(pair.right);
+        return Integer.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.LOCAL_TIME)) {
+      return (r11, r21) -> {
+        int b1 = r11.getPackedTime(pair.left);
+        int b2 = r21.getPackedTime(pair.right);
+        return Integer.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.LONG)) {
+      return (r11, r21) -> {
+        long b1 = r11.getLong(pair.left);
+        long b2 = r21.getLong(pair.right);
+        return Long.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.LOCAL_DATE_TIME)) {
+      return (r11, r21) -> {
+        long b1 = r11.getPackedDateTime(pair.left);
+        long b2 = r21.getPackedDateTime(pair.right);
+        return Long.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.INSTANT)) {
+      return (r11, r21) -> {
+        long b1 = r11.getPackedInstant(pair.left);
+        long b2 = r21.getPackedInstant(pair.right);
+        return Long.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.DOUBLE)) {
+      return (r11, r21) -> {
+        double b1 = r11.getDouble(pair.left);
+        double b2 = r21.getDouble(pair.right);
+        return Double.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.FLOAT)) {
+      return (r11, r21) -> {
+        float b1 = r11.getFloat(pair.left);
+        float b2 = r21.getFloat(pair.right);
+        return Float.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.BOOLEAN)) {
+      return (r11, r21) -> {
+        byte b1 = r11.getBooleanAsByte(pair.left);
+        byte b2 = r21.getBooleanAsByte(pair.right);
+        return Byte.compare(b1, b2);
+      };
+    } else if (pair.type.equals(ColumnType.STRING)) {
+      return (r11, r21) -> {
+        String b1 = r11.getString(pair.left);
+        String b2 = r21.getString(pair.right);
+        return b1.compareTo(b2);
+      };
+    } else if (pair.type.equals(ColumnType.TEXT)) {
+      return (r11, r21) -> {
+        String b1 = r11.getText(pair.left);
+        String b2 = r21.getText(pair.right);
+        return b1.compareTo(b2);
+      };
+    }
+    throw new RuntimeException("Unhandled ColumnType in SortKey.");
+  }
+
+  /** Returns the iterator for the SortKey */
+  @Override
+  public Iterator<ColumnIndexPair> iterator() {
+    return sortOrder.iterator();
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this).add("order", sortOrder).toString();
+  }
+}
diff --git a/core/src/main/java/tech/tablesaw/joining/SortMergeJoin.java b/core/src/main/java/tech/tablesaw/joining/SortMergeJoin.java
new file mode 100644
index 00000000..c06ab822
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/joining/SortMergeJoin.java
@@ -0,0 +1,460 @@
+package tech.tablesaw.joining;
+
+import com.google.common.collect.Streams;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import tech.tablesaw.api.ColumnType;
+import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.Row;
+import tech.tablesaw.api.Table;
+import tech.tablesaw.columns.Column;
+import tech.tablesaw.selection.Selection;
+
+/** Implements joins between two or more Tables */
+class SortMergeJoin implements JoinStrategy {
+
+  private static final String LEFT_RECORD_ID_NAME = "_left_record_id_";
+  private static final String RIGHT_RECORD_ID_NAME = "_right_record_id_";
+
+  private static final String TABLE_ALIAS = "T";
+  public static final String PLACEHOLDER_COL_PREFIX = "Placeholder_";
+
+  private final String[] leftjoinColumnNames;
+  private int[] leftJoinColumnPositions;
+  private int[] rightJoinColumnPositions;
+
+  private final AtomicInteger joinTableId = new AtomicInteger(1);
+
+  /**
+   * Constructor.
+   *
+   * @param table The table to join on.
+   * @param joinColumnNames The join column names to join on.
+   */
+  public SortMergeJoin(Table table, String... joinColumnNames) {
+    this.leftJoinColumnPositions = getJoinIndexes(table, joinColumnNames);
+    this.leftjoinColumnNames = joinColumnNames;
+  }
+
+  /**
+   * Finds the index of the columns corresponding to the columnNames. E.G. The column named "ID" is
+   * located at index 5 in table.
+   *
+   * @param table the table that contains the columns.
+   * @param columnNames the column names to find indexes of.
+   * @return a list of column indexes within the table.
+   */
+  private int[] getJoinIndexes(Table table, String[] columnNames) {
+    int[] results = new int[columnNames.length];
+    for (int i = 0; i < columnNames.length; i++) {
+      String nm = columnNames[i];
+      results[i] = table.columnIndex(nm);
+    }
+    return results;
+  }
+
+  /**
+   * Joins two tables.
+   *
+   * @param t1 the table on the left side of the join.
+   * @param t2 the table on the right side of the join.
+   * @param joinType the type of join.
+   * @param allowDuplicates 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 keepAllJoinKeyColumns if {@code false} the join will only keep join key columns in
+   *     table1 if {@code true} the join will return all join key columns in both table, which may
+   *     have difference when there are null values
+   * @param table2JoinColumnNames The names of the columns in table2 to join on.
+   * @return the joined table
+   */
+  public Table performJoin(
+      Table t1,
+      Table t2,
+      JoinType joinType,
+      boolean allowDuplicates,
+      boolean keepAllJoinKeyColumns,
+      int[] leftJoinColumnIndexes,
+      String... table2JoinColumnNames) {
+
+    this.leftJoinColumnPositions = leftJoinColumnIndexes;
+    rightJoinColumnPositions = getJoinIndexes(t2, table2JoinColumnNames);
+
+    Table table1 = t1.sortAscendingOn(leftjoinColumnNames);
+    Table table2 = t2.sortAscendingOn(table2JoinColumnNames);
+
+    Column<?>[] cols =
+        Streams.concat(table1.columns().stream(), table2.columns().stream())
+            .map(Column::emptyCopy)
+            .toArray(Column[]::new);
+
+    // A set of column indexes in the result table that can be ignored. They are duplicate join
+    // keys.
+    int[] resultIgnoreColIndexes =
+        keepAllJoinKeyColumns ? new int[0] : getIgnoredColumns(table1, joinType, cols);
+
+    Table result = emptyTableFromColumns(table1, allowDuplicates, cols);
+
+    // add indexes for outer join processing
+    IntColumn indexLeft = IntColumn.indexColumn(LEFT_RECORD_ID_NAME, table1.rowCount(), 0);
+    table1.addColumns(indexLeft);
+    result.addColumns(IntColumn.create(LEFT_RECORD_ID_NAME));
+
+    IntColumn indexRight = IntColumn.indexColumn(RIGHT_RECORD_ID_NAME, table2.rowCount(), 0);
+    table2.addColumns(indexRight);
+    result.addColumns(IntColumn.create(RIGHT_RECORD_ID_NAME));
+
+    validateJoinColumns(table1, table2);
+
+    if (table1.rowCount() == 0 && (joinType == JoinType.LEFT_OUTER || joinType == JoinType.INNER)) {
+      // Handle special case of empty table here so it doesn't fall through to the behavior
+      // that adds rows for full outer and right outer joins
+      if (!keepAllJoinKeyColumns) {
+        result.removeColumns(resultIgnoreColIndexes);
+      }
+      return result;
+    }
+    if (joinType == JoinType.INNER) {
+      joinInner(result, table1, table2, resultIgnoreColIndexes);
+    } else if (joinType == JoinType.LEFT_OUTER) {
+      joinLeft(result, table1, table2, resultIgnoreColIndexes);
+    } else if (joinType == JoinType.RIGHT_OUTER) {
+      joinRight(result, table1, table2, resultIgnoreColIndexes);
+    } else if (joinType == JoinType.FULL_OUTER) {
+      joinFull(result, table1, table2, resultIgnoreColIndexes);
+    }
+    result.removeColumns(LEFT_RECORD_ID_NAME, RIGHT_RECORD_ID_NAME);
+
+    if (!keepAllJoinKeyColumns) {
+      result = result.removeColumns(resultIgnoreColIndexes);
+    } else {
+      renameJoinColumns(result, table1, resultIgnoreColIndexes);
+    }
+    return result;
+  }
+
+  /**
+   * Renames the column indexes for the second table from Placeholder_X to their original names
+   *
+   * @param resultIgnoreColIndexes The positions of the secondary join columns
+   */
+  private void renameJoinColumns(Table result, Table left, int[] resultIgnoreColIndexes) {
+
+    String table2Alias = TABLE_ALIAS + joinTableId.get();
+
+    for (int position : resultIgnoreColIndexes) {
+      String realName = result.column(position).name().replace(PLACEHOLDER_COL_PREFIX, "");
+      if (position >= left.columnCount()) {
+        if (result.containsColumn(realName.toLowerCase())) {
+          result.column(position).setName(newName(table2Alias, realName));
+        } else {
+          result.column(position).setName(realName);
+        }
+      } else {
+        result.column(position).setName(realName);
+      }
+    }
+  }
+
+  private String newName(String table2Alias, String columnName) {
+    return table2Alias + "." + columnName;
+  }
+
+  /**
+   * Adds empty columns to the destination table with the same type as columns in table1 and table2.
+   *
+   * <p>For inner, left and full outer join types the join columns in table2 are not needed and will
+   * be marked as placeholders. The indexes of those columns will be returned. The downstream logic
+   * is easier if we wait to remove the redundant columns until the last step.
+   *
+   * @param table1 the table on left side of the join.
+   * @param allowDuplicates whether to allow duplicates. If yes rename columns in table2 that have
+   *     the same name as columns in table1, with the exception of join columns in table2 when
+   *     performing a right join.
+   * @param cols An array of columns from both join tables
+   * @return the table to use for the join results
+   */
+  Table emptyTableFromColumns(Table table1, boolean allowDuplicates, Column<?>[] cols) {
+
+    Table destination = Table.create(table1.name());
+
+    // Rename duplicate columns in second table
+    if (allowDuplicates) {
+      Set<String> table1ColNames =
+          Arrays.stream(cols)
+              .map(Column::name)
+              .map(String::toLowerCase)
+              .limit(table1.columnCount())
+              .collect(Collectors.toSet());
+
+      String table2Alias = TABLE_ALIAS + joinTableId.incrementAndGet();
+      for (int c = table1.columnCount(); c < cols.length; c++) {
+        String columnName = cols[c].name();
+        if (table1ColNames.contains(columnName.toLowerCase())) {
+          cols[c].setName(newName(table2Alias, columnName));
+        }
+      }
+    }
+    destination.addColumns(cols);
+    return destination;
+  }
+
+  /**
+   * Returns the positions of columns that can be ignored in the result table
+   *
+   * <p>For inner join, left join and full outer join mark the join columns in table2 as
+   * placeholders.
+   *
+   * <p>For right join, mark the join columns in table1 as placeholders. Keep track of which join
+   * columns are placeholders so they can be ignored.
+   */
+  private int[] getIgnoredColumns(Table table1, JoinType joinType, Column<?>[] cols) {
+    int[] ignoreColumns = new int[leftJoinColumnPositions.length];
+    int ignoreIndex = 0;
+    for (int c = 0; c < cols.length; c++) {
+      if (joinType == JoinType.RIGHT_OUTER) {
+        if (c < table1.columnCount() && indexesContainsValue(leftJoinColumnPositions, c)) {
+          ignoreColumns[ignoreIndex] = c;
+          cols[c].setName(PLACEHOLDER_COL_PREFIX + cols[c].name());
+          ignoreIndex++;
+        }
+      } else { // JoinType is LEFT, INNER, or FULL
+        int table2Index = c + table1.columnCount();
+        if (indexesContainsValue(rightJoinColumnPositions, c)) {
+          ignoreColumns[ignoreIndex] = table2Index;
+          cols[table2Index].setName(PLACEHOLDER_COL_PREFIX + cols[table2Index].name());
+          ignoreIndex++;
+        }
+      }
+    }
+    return ignoreColumns;
+  }
+
+  private void joinInner(Table destination, Table left, Table right, int[] ignoreColumns) {
+
+    Comparator<Row> comparator = getRowComparator(left, rightJoinColumnPositions);
+
+    Row leftRow = left.row(0);
+    Row rightRow = right.row(0);
+
+    // Marks the position of the first record in right table that matches a specific join value
+    int mark = -1;
+
+    while (leftRow.hasNext() || rightRow.hasNext()) {
+      if (mark == -1) {
+        while (comparator.compare(leftRow, rightRow) < 0 && leftRow.hasNext()) leftRow.next();
+        while (comparator.compare(leftRow, rightRow) > 0 && rightRow.hasNext()) rightRow.next();
+        // set the position of the first matching record on the right side
+        mark = rightRow.getRowNumber();
+      }
+      if (comparator.compare(leftRow, rightRow) == 0 && (leftRow.hasNext() || rightRow.hasNext())) {
+        addValues(destination, leftRow, rightRow);
+        if (rightRow.hasNext()) {
+          rightRow.next();
+        } else {
+          rightRow.at(mark);
+          if (leftRow.hasNext()) {
+            leftRow.next();
+          }
+          mark = -1;
+        }
+      } else {
+        if (rightRow.hasNext() && leftRow.hasNext()) {
+          rightRow.at(mark);
+          leftRow.next();
+          mark = -1;
+        } else {
+          if (leftRow.hasNext()) leftRow.next();
+          if (!leftRow.hasNext()) {
+            break;
+          }
+        }
+      }
+    }
+    // add the last value if you end on a match
+    if (comparator.compare(leftRow, rightRow) == 0) {
+      addValues(destination, leftRow, rightRow);
+    }
+  }
+
+  private void joinLeft(Table destination, Table left, Table right, int[] ignoreColumns) {
+
+    joinInner(destination, left, right, ignoreColumns);
+    Selection unmatched =
+        left.intColumn(LEFT_RECORD_ID_NAME)
+            .isNotIn(destination.intColumn(LEFT_RECORD_ID_NAME).unique());
+    addLeftOnlyValues(destination, left, unmatched);
+  }
+
+  private void joinRight(Table destination, Table left, Table right, int[] ignoreColumns) {
+    joinInner(destination, left, right, ignoreColumns);
+    Selection unmatched =
+        right
+            .intColumn(RIGHT_RECORD_ID_NAME)
+            .isNotIn(destination.intColumn(RIGHT_RECORD_ID_NAME).unique());
+    addRightOnlyValues(destination, left, right, unmatched);
+  }
+
+  private void joinFull(Table destination, Table left, Table right, int[] ignoreColumns) {
+
+    Table tempDestination = destination.emptyCopy();
+
+    joinInner(destination, left, right, ignoreColumns);
+
+    Selection unmatchedLeft =
+        left.intColumn(LEFT_RECORD_ID_NAME)
+            .isNotIn(destination.intColumn(LEFT_RECORD_ID_NAME).unique());
+    addLeftOnlyValues(destination, left, unmatchedLeft);
+
+    Selection unmatchedRight =
+        right
+            .intColumn(RIGHT_RECORD_ID_NAME)
+            .isNotIn(destination.intColumn(RIGHT_RECORD_ID_NAME).unique());
+    addRightOnlyValues(tempDestination, left, right, unmatchedRight);
+    for (int i = 0; i < ignoreColumns.length; i++) {
+      String name = tempDestination.columnNames().get(leftJoinColumnPositions[i]);
+      tempDestination.replaceColumn(
+          leftJoinColumnPositions[i],
+          tempDestination.column(ignoreColumns[i]).copy().setName(name));
+    }
+    destination.append(tempDestination);
+  }
+
+  private void addLeftOnlyValues(Table destination, Table left, Selection unmatched) {
+    for (Row leftRow : left.where(unmatched)) {
+      Row destRow = destination.appendRow();
+      for (int c = 0; c < leftRow.columnCount() - 1; c++) {
+        updateDestinationRow(destRow, leftRow, c, c);
+      }
+      // update the index column putting it at the end of the destination table
+      updateDestinationRow(destRow, leftRow, destRow.columnCount() - 2, leftRow.columnCount() - 1);
+    }
+  }
+
+  private void addRightOnlyValues(Table destination, Table left, Table right, Selection unmatched) {
+    int leftColumnCount = left.columnCount();
+    for (Row rightRow : right.where(unmatched)) {
+      Row destRow = destination.appendRow();
+      for (int c = 0; c < rightRow.columnCount() - 1; c++) {
+        updateDestinationRow(destRow, rightRow, c + leftColumnCount - 1, c);
+      }
+      // update the index column putting it at the end of the destination table
+      updateDestinationRow(
+          destRow, rightRow, destRow.columnCount() - 1, rightRow.columnCount() - 1);
+    }
+  }
+
+  private Comparator<Row> getRowComparator(Table left, int[] rightJoinColumnIndexes) {
+    List<ColumnIndexPair> pairs = createJoinColumnPairs(left, rightJoinColumnIndexes);
+    return SortKey.getChain(SortKey.create(pairs));
+  }
+
+  private List<ColumnIndexPair> createJoinColumnPairs(Table left, int[] rightJoinColumnIndexes) {
+    List<ColumnIndexPair> pairs = new ArrayList<>();
+    for (int i = 0; i < leftJoinColumnPositions.length; i++) {
+      ColumnIndexPair columnIndexPair =
+          new ColumnIndexPair(
+              left.column(leftJoinColumnPositions[i]).type(),
+              leftJoinColumnPositions[i],
+              rightJoinColumnIndexes[i]);
+      pairs.add(columnIndexPair);
+    }
+    return pairs;
+  }
+
+  private void updateDestinationRow(
+      Row destRow, Row sourceRow, int destColumnPosition, int sourceColumnPosition) {
+    ColumnType type = destRow.getColumnType(destColumnPosition);
+    if (type.equals(ColumnType.INTEGER)) {
+      destRow.setInt(destColumnPosition, sourceRow.getInt(sourceColumnPosition));
+    } else if (type.equals(ColumnType.LONG)) {
+      destRow.setLong(destColumnPosition, sourceRow.getLong(sourceColumnPosition));
+    } else if (type.equals(ColumnType.SHORT)) {
+      destRow.setShort(destColumnPosition, sourceRow.getShort(sourceColumnPosition));
+    } else if (type.equals(ColumnType.STRING)) {
+      destRow.setString(destColumnPosition, sourceRow.getString(sourceColumnPosition));
+    } else if (type.equals(ColumnType.LOCAL_DATE)) {
+      destRow.setPackedDate(destColumnPosition, sourceRow.getPackedDate(sourceColumnPosition));
+    } else if (type.equals(ColumnType.LOCAL_TIME)) {
+      destRow.setPackedTime(destColumnPosition, sourceRow.getPackedTime(sourceColumnPosition));
+    } else if (type.equals(ColumnType.LOCAL_DATE_TIME)) {
+      destRow.setPackedDateTime(
+          destColumnPosition, sourceRow.getPackedDateTime(sourceColumnPosition));
+    } else if (type.equals(ColumnType.INSTANT)) {
+      destRow.setPackedInstant(
+          destColumnPosition, sourceRow.getPackedInstant(sourceColumnPosition));
+    } else if (type.equals(ColumnType.DOUBLE)) {
+      destRow.setDouble(destColumnPosition, sourceRow.getDouble(sourceColumnPosition));
+    } else if (type.equals(ColumnType.FLOAT)) {
+      destRow.setFloat(destColumnPosition, sourceRow.getFloat(sourceColumnPosition));
+    } else if (type.equals(ColumnType.BOOLEAN)) {
+      destRow.setBooleanAsByte(
+          destColumnPosition, sourceRow.getBooleanAsByte(sourceColumnPosition));
+    } else if (type.equals(ColumnType.TEXT)) {
+      destRow.setText(destColumnPosition, sourceRow.getText(sourceColumnPosition));
+    }
+  }
+
+  private void addValues(Table destination, Row leftRow, Row rightRow) {
+
+    Row destRow = destination.appendRow();
+
+    // update positionally, but take into account the RECORD_ID COLUMNS at the end of the dest table
+    int leftColumnCount = leftRow.columnCount();
+    int rightColumnCount = rightRow.columnCount();
+
+    // update from the left table first (everythint but the RECORD_ID column)
+    for (int destIdx1 = 0; destIdx1 < leftColumnCount - 1; destIdx1++) {
+      updateDestinationRow(destRow, leftRow, destIdx1, destIdx1);
+    }
+
+    // update from the right table (everythint but the RECORD_ID column)
+    for (int destIdx2 = (leftColumnCount - 1);
+        destIdx2 < (leftColumnCount + rightColumnCount) - 2;
+        destIdx2++) {
+      int rightIndex = destIdx2 - (leftColumnCount - 1);
+      updateDestinationRow(destRow, rightRow, destIdx2, rightIndex);
+    }
+
+    // update the RECORD_ID columns
+    updateDestinationRow(destRow, leftRow, destRow.columnCount() - 2, leftColumnCount - 1);
+    updateDestinationRow(destRow, rightRow, destRow.columnCount() - 1, rightColumnCount - 1);
+  }
+
+  private boolean indexesContainsValue(int[] joinColumnIndexes, int columnIndex) {
+    for (int i : joinColumnIndexes) {
+      if (columnIndex == i) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private void validateJoinColumns(Table table1, Table table2) {
+    if (leftJoinColumnPositions.length != rightJoinColumnPositions.length) {
+      throw new IllegalArgumentException(
+          "Cannot join using a different number of indices on each table: "
+              + Arrays.toString(leftJoinColumnPositions)
+              + " and "
+              + Arrays.toString(rightJoinColumnPositions));
+    }
+    for (int i = 0; i < leftJoinColumnPositions.length; i++) {
+      if (!table1
+          .column(leftJoinColumnPositions[i])
+          .getClass()
+          .equals(table2.column(rightJoinColumnPositions[i]).getClass())) {
+        throw new IllegalArgumentException(
+            "Cannot join using different index types: "
+                + Arrays.toString(leftJoinColumnPositions)
+                + " and "
+                + Arrays.toString(rightJoinColumnPositions));
+      }
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "SortMergeJoin";
+  }
+}
diff --git a/core/src/main/java/tech/tablesaw/table/Relation.java b/core/src/main/java/tech/tablesaw/table/Relation.java
index f7d81f13..f7f0820a 100644
--- a/core/src/main/java/tech/tablesaw/table/Relation.java
+++ b/core/src/main/java/tech/tablesaw/table/Relation.java
@@ -14,6 +14,8 @@
 
 package tech.tablesaw.table;
 
+import static tech.tablesaw.joining.JoinType.FULL_OUTER;
+
 import it.unimi.dsi.fastutil.ints.IntArrays;
 import it.unimi.dsi.fastutil.ints.IntComparators;
 import java.io.ByteArrayOutputStream;
@@ -299,7 +301,12 @@ public abstract class Relation implements Iterable<Row> {
       Table columnSummary = this.column(i).summary();
       columnSummary.column(1).setName(this.column(i).name());
       summaryTable =
-          summaryTable.joinOn("Measure").fullOuter(columnSummary, columnSummary.column(0).name());
+          summaryTable
+              .joinOn("Measure")
+              .with(columnSummary)
+              .rightJoinColumns(columnSummary.column(0).name())
+              .type(FULL_OUTER)
+              .join();
     }
     summaryTable.column(0).setName("Summary");
     return summaryTable;


Test patch:
diff --git a/core/src/test/java/tech/tablesaw/api/TextColumnTest.java b/core/src/test/java/tech/tablesaw/api/TextColumnTest.java
index e49ffb3b..a5a8f2e3 100644
--- a/core/src/test/java/tech/tablesaw/api/TextColumnTest.java
+++ b/core/src/test/java/tech/tablesaw/api/TextColumnTest.java
@@ -31,6 +31,7 @@ import tech.tablesaw.TestDataUtil;
 import tech.tablesaw.columns.strings.StringColumnFormatter;
 import tech.tablesaw.columns.strings.StringParser;
 import tech.tablesaw.io.csv.CsvReadOptions;
+import tech.tablesaw.joining.JoinType;
 import tech.tablesaw.selection.Selection;
 
 public class TextColumnTest {
@@ -107,7 +108,7 @@ public class TextColumnTest {
                                 "13:21:52.451,26.1"))
                     .columnTypesToDetect(Arrays.asList(ColumnType.DOUBLE, ColumnType.TEXT)));
 
-    Table joined = t1.joinOn("TIME").fullOuter(t2);
+    Table joined = t1.joinOn("TIME").with(t2).type(JoinType.FULL_OUTER).join();
     assertEquals(3, joined.columnCount());
   }
 
diff --git a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerPerformanceTest.java b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerPerformanceTest.java
deleted file mode 100644
index 5510818a..00000000
--- a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerPerformanceTest.java
+++ /dev/null
@@ -1,89 +0,0 @@
-package tech.tablesaw.joining;
-
-import static org.junit.jupiter.api.Assertions.assertTimeout;
-
-import java.time.Duration;
-import java.util.Arrays;
-import java.util.Random;
-import java.util.stream.IntStream;
-import org.junit.jupiter.api.Tag;
-import org.junit.jupiter.api.Test;
-import tech.tablesaw.api.IntColumn;
-import tech.tablesaw.api.Table;
-
-@Tag("Slow")
-@Tag("Flaky")
-public class DataFrameJoinerPerformanceTest {
-
-  private static final long SEED = 200L;
-
-  private static Table addFillerColumn(Table table, int numberColumnsToAdd, String prefix) {
-    int[] filler = new int[table.rowCount()];
-    Arrays.fill(filler, 1);
-    IntColumn col = IntColumn.create("temp", filler);
-    for (int i = 0; i < numberColumnsToAdd; i++) {
-      table.addColumns(col.copy().setName(prefix + "_appendColumn" + i));
-    }
-    return table;
-  }
-
-  private static Table createCustomersTable(int numberCustomers) {
-    Table customersTable = Table.create("customers");
-    IntColumn customerIds =
-        IntColumn.create("customerId", IntStream.range(0, numberCustomers).toArray());
-    customersTable.addColumns(customerIds);
-    return customersTable;
-  }
-
-  private static Table createOrdersTable(int numberOrders, int numberCustomers) {
-    Table ordersTable = Table.create("orders");
-    Random random = new Random(SEED);
-    IntColumn orderCustomerIds =
-        IntColumn.create(
-            "customerId",
-            random
-                .doubles()
-                .limit(numberOrders)
-                .mapToInt(randomDouble -> (int) Math.floor(randomDouble * numberCustomers))
-                .toArray());
-    ordersTable.addColumns(orderCustomerIds);
-    return ordersTable;
-  }
-
-  @Test
-  public void innerJoinCustomersFirst() {
-    int numberOrders = 10_000;
-    int numberCustomers = 1_000;
-    Table customers = createCustomersTable(numberCustomers);
-    addFillerColumn(customers, 5, "customer");
-    Table orders = createOrdersTable(numberOrders, numberCustomers);
-    addFillerColumn(orders, 5, "order");
-
-    assertTimeout(Duration.ofSeconds(1), () -> customers.joinOn("customerId").inner(orders));
-  }
-
-  @Test
-  public void leftOuterOrdersFirst() {
-    int numberOrders = 10_000;
-    Table customers = createCustomersTable(1_000);
-    addFillerColumn(customers, 5, "customer");
-    // Number customers here is larger. Will create rows orders without matching customers.
-    Table orders = createOrdersTable(numberOrders, 2_000);
-    addFillerColumn(orders, 5, "order");
-
-    assertTimeout(Duration.ofSeconds(1), () -> orders.joinOn("customerId").leftOuter(customers));
-  }
-
-  @Test
-  public void fullOuterJoin() {
-    int numberOrders = 10_000;
-    int numberCustomers = 1_000;
-    Table customers = createCustomersTable(numberCustomers);
-    addFillerColumn(customers, 5, "customer");
-    // Number customers here is larger. Will create orders without matching customers.
-    Table orders = createOrdersTable(numberOrders, 2_000);
-    addFillerColumn(orders, 5, "order");
-
-    assertTimeout(Duration.ofSeconds(1), () -> customers.joinOn("customerId").fullOuter(orders));
-  }
-}
diff --git a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
index 6e10312d..91890429 100644
--- a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
+++ b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
@@ -2,6 +2,7 @@ package tech.tablesaw.joining;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static tech.tablesaw.joining.JoinType.*;
 
 import com.google.common.base.Joiner;
 import java.util.Arrays;
@@ -514,7 +515,11 @@ public class DataFrameJoinerTest {
   @Test
   public void innerJoinWithDoubleBirdsCatsFishDouble() {
     Table joined =
-        DOUBLE_INDEXED_BIRDS.joinOn("ID").inner(DOUBLE_INDEXED_CATS, DOUBLE_INDEXED_FISH);
+        DOUBLE_INDEXED_BIRDS
+            .joinOn("ID")
+            .type(INNER)
+            .with(DOUBLE_INDEXED_CATS, DOUBLE_INDEXED_FISH)
+            .join();
     assertEquals(4, joined.columnCount());
     assertEquals(1, joined.rowCount());
   }
@@ -522,21 +527,26 @@ public class DataFrameJoinerTest {
   @Test
   public void innerJoinWithDoubleDogsCatsBirdsDouble() {
     Table joined =
-        DOUBLE_INDEXED_FISH.joinOn("ID").inner(DOUBLE_INDEXED_CATS, DOUBLE_INDEXED_BIRDS);
+        DOUBLE_INDEXED_FISH
+            .joinOn("ID")
+            .type(INNER)
+            .with(DOUBLE_INDEXED_CATS, DOUBLE_INDEXED_BIRDS)
+            .join();
     assertEquals(4, joined.columnCount());
     assertEquals(1, joined.rowCount());
   }
 
   @Test
   public void innerJoinWithDoubleDogsCatsFishVarargs() {
-    Table joined = DOUBLE_INDEXED_MICE.joinOn("ID").inner(DOUBLE_INDEXED_CATS, DOUBLE_INDEXED_FISH);
+    Table joined =
+        DOUBLE_INDEXED_MICE.joinOn("ID").with(DOUBLE_INDEXED_CATS, DOUBLE_INDEXED_FISH).join();
     assertEquals(4, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
 
   @Test
   public void innerJoinWithDoublesSimple() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").inner(DOUBLE_INDEXED_DOGS);
+    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").with(DOUBLE_INDEXED_DOGS).join();
     assertEquals(3, joined.columnCount());
     assertEquals(3, joined.rowCount());
     assertEquals(3, joined.column("ID").size());
@@ -544,7 +554,8 @@ public class DataFrameJoinerTest {
 
   @Test
   public void innerJoinWithDoubles() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").inner(DOUBLE_INDEXED_DOGS, "ID");
+    Table joined =
+        DOUBLE_INDEXED_PEOPLE.joinOn("ID").with(DOUBLE_INDEXED_DOGS).rightJoinColumns("ID").join();
     assertEquals(3, joined.columnCount());
     assertEquals(3, joined.rowCount());
   }
@@ -556,7 +567,13 @@ public class DataFrameJoinerTest {
     Table table2 =
         DUPLICATE_COL_NAME_DOGS.where(DUPLICATE_COL_NAME_DOGS.booleanColumn("Good").isFalse());
 
-    Table joined = table1.joinOn("ID").inner(table2, "ID", true);
+    Table joined =
+        table1
+            .joinOn("ID")
+            .with(table2)
+            .rightJoinColumns("ID")
+            .allowDuplicateColumnNames(true)
+            .join();
     assertEquals(5, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
@@ -568,27 +585,47 @@ public class DataFrameJoinerTest {
     Table table2 =
         DUPLICATE_COL_NAME_DOGS.where(DUPLICATE_COL_NAME_DOGS.booleanColumn("Good").isFalse());
 
-    Table joined = table1.joinOn("ID").rightOuter(table2, true, "ID");
+    Table joined =
+        table1
+            .joinOn("ID")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .rightJoinColumns("ID")
+            .type(RIGHT_OUTER)
+            .join();
     assertEquals(5, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoinWithDuplicateColumnNames() {
+  public void leftOuterWithDuplicateColumnNames() {
     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.joinOn("ID").leftOuter(table2, true, "ID");
+    Table joined =
+        table1
+            .joinOn("ID")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .type(LEFT_OUTER)
+            .rightJoinColumns("ID")
+            .join();
     assertEquals(5, joined.columnCount());
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
   }
 
   @Test
-  public void leftOuterJoinWithDoubles() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").leftOuter(DOUBLE_INDEXED_DOGS, "ID");
+  public void leftOuterWithDoubles() {
+    Table joined =
+        DOUBLE_INDEXED_PEOPLE
+            .joinOn("ID")
+            .with(DOUBLE_INDEXED_DOGS)
+            .rightJoinColumns("ID")
+            .type(LEFT_OUTER)
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
@@ -596,15 +633,23 @@ public class DataFrameJoinerTest {
 
   @Test
   public void rightOuterJoinWithDoubles() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").rightOuter(DOUBLE_INDEXED_DOGS, "ID");
+    Table joined =
+        DOUBLE_INDEXED_PEOPLE
+            .joinOn("ID")
+            .with(DOUBLE_INDEXED_DOGS)
+            .type(RIGHT_OUTER)
+            .rightJoinColumns("ID")
+            .join();
     assertEquals(3, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
   }
 
   @Test
   public void rightOuterJoinWithDoubles2() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").rightOuter(DOUBLE_INDEXED_DOGS);
+    Table joined =
+        DOUBLE_INDEXED_PEOPLE.joinOn("ID").with(DOUBLE_INDEXED_DOGS).type(RIGHT_OUTER).join();
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
@@ -612,7 +657,12 @@ public class DataFrameJoinerTest {
 
   @Test
   public void rightOuterJoinWithDoubles2Reverse() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").rightOuter(DOUBLE_INDEXED_DOGS_REVERSE);
+    Table joined =
+        DOUBLE_INDEXED_PEOPLE
+            .joinOn("ID")
+            .with(DOUBLE_INDEXED_DOGS_REVERSE)
+            .type(RIGHT_OUTER)
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
@@ -621,7 +671,11 @@ public class DataFrameJoinerTest {
   @Test
   public void rightOuterJoinWithDoubles3() {
     Table joined =
-        DOUBLE_INDEXED_PEOPLE.joinOn("ID").rightOuter(DOUBLE_INDEXED_DOGS, DOUBLE_INDEXED_CATS);
+        DOUBLE_INDEXED_PEOPLE
+            .joinOn("ID")
+            .with(DOUBLE_INDEXED_DOGS, DOUBLE_INDEXED_CATS)
+            .type(RIGHT_OUTER)
+            .join();
     assertTrue(
         joined.columnNames().containsAll(Arrays.asList("ID", "Name", "Dog Name", "Cat Name")));
     assertEquals(4, joined.rowCount());
@@ -637,7 +691,8 @@ public class DataFrameJoinerTest {
 
   @Test
   public void rightOuterJoinWithDoubles4() {
-    Table joined = DOUBLE_INDEXED_PEOPLE.joinOn("ID").rightOuter(DOUBLE_INDEXED_DOGS);
+    Table joined =
+        DOUBLE_INDEXED_PEOPLE.joinOn("ID").with(DOUBLE_INDEXED_DOGS).type(RIGHT_OUTER).join();
     assertTrue(joined.columnNames().containsAll(Arrays.asList("ID", "Name", "Dog Name")));
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
@@ -649,25 +704,36 @@ public class DataFrameJoinerTest {
   }
 
   @Test
-  public void leftOuterJoinWithDoubles2() {
-    Table joined = DOUBLE_INDEXED_DOGS.joinOn("ID").leftOuter(DOUBLE_INDEXED_PEOPLE, "ID");
+  public void leftOuterWithDoubles2() {
+    Table joined =
+        DOUBLE_INDEXED_DOGS
+            .joinOn("ID")
+            .with(DOUBLE_INDEXED_PEOPLE)
+            .rightJoinColumns("ID")
+            .type(LEFT_OUTER)
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
   }
 
   @Test
-  public void leftOuterJoinWithDoubles3() {
-    Table joined = DOUBLE_INDEXED_DOGS.joinOn("ID").leftOuter(DOUBLE_INDEXED_PEOPLE);
+  public void leftOuterWithDoubles3() {
+    Table joined =
+        DOUBLE_INDEXED_DOGS.joinOn("ID").with(DOUBLE_INDEXED_PEOPLE).type(LEFT_OUTER).join();
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
     assertEquals(4, joined.column("ID").size());
   }
 
   @Test
-  public void leftOuterJoinWithDoubles4() {
+  public void leftOuterWithDoubles4() {
     Table joined =
-        DOUBLE_INDEXED_DOGS.joinOn("ID").leftOuter(DOUBLE_INDEXED_PEOPLE, DOUBLE_INDEXED_CATS);
+        DOUBLE_INDEXED_DOGS
+            .joinOn("ID")
+            .with(DOUBLE_INDEXED_PEOPLE, DOUBLE_INDEXED_CATS)
+            .type(LEFT_OUTER)
+            .join();
     assertTrue(
         joined.columnNames().containsAll(Arrays.asList("ID", "Dog Name", "Name", "Cat Name")));
     assertEquals(4, joined.column("ID").size());
@@ -682,22 +748,26 @@ public class DataFrameJoinerTest {
 
   @Test
   public void innerJoin() {
-    Table joined = SP500.joinOn("Date").inner(ONE_YEAR, "Date");
+    Table joined = SP500.joinOn("Date").with(ONE_YEAR).rightJoinColumns("Date").join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(3, joined.columnCount());
     assertEquals(5, joined.rowCount());
   }
 
   @Test
   public void innerJoinSingleColumn() {
-    Table joined = SP500.selectColumns("Date").joinOn("Date").inner(ONE_YEAR.selectColumns("Date"));
+    Table joined =
+        SP500.selectColumns("Date").joinOn("Date").with(ONE_YEAR.selectColumns("Date")).join();
     assertEquals(5, joined.rowCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(1, joined.columnCount());
   }
 
   @Test
   public void innerJoinSingleColumnOnRight() {
-    Table joined = SP500.joinOn("Date").inner(ONE_YEAR.selectColumns("Date"));
+    Table joined = SP500.joinOn("Date").with(ONE_YEAR.selectColumns("Date")).join();
     assertEquals(5, joined.rowCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(2, joined.columnCount());
   }
 
@@ -709,7 +779,8 @@ public class DataFrameJoinerTest {
         Table.create(
             ONE_YEAR.dateColumn("Date"), ONE_YEAR.dateColumn("Date").copy().setName("Date2"));
 
-    Table joined = table1.joinOn("Date", "Date2").inner(table2);
+    Table joined = table1.joinOn("Date", "Date2").with(table2).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(5, joined.rowCount());
     assertEquals(2, joined.columnCount());
   }
@@ -717,29 +788,43 @@ public class DataFrameJoinerTest {
   @Test
   public void innerJoinWithBoolean() {
     Table joined =
-        DUPLICATE_COL_NAME_DOGS.joinOn("Good").inner(true, DUPLICATE_COL_NAME_DOGS.copy());
+        DUPLICATE_COL_NAME_DOGS
+            .joinOn("Good")
+            .allowDuplicateColumnNames(true)
+            .with(DUPLICATE_COL_NAME_DOGS)
+            .join();
     assertEquals(5, joined.columnCount());
     assertEquals(32, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoin() {
-    Table joined = SP500.joinOn("Date").leftOuter(ONE_YEAR, "Date");
+  public void leftOuter() {
+    Table joined =
+        SP500.joinOn("Date").with(ONE_YEAR).rightJoinColumns("Date").type(LEFT_OUTER).join();
     assertEquals(3, joined.columnCount());
     assertEquals(6, joined.rowCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(6, joined.column("Date").size());
   }
 
   @Test
   public void innerJoinDuplicateKeysFirstTable() {
-    Table joined = ANIMAL_NAMES.joinOn("Animal").inner(ANIMAL_FEED, "Animal");
+    Table joined =
+        ANIMAL_NAMES.joinOn("Animal").with(ANIMAL_FEED).rightJoinColumns("Animal").join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoinDuplicateKeysFirstTable() {
-    Table joined = ANIMAL_NAMES.joinOn("Animal").leftOuter(ANIMAL_FEED, "Animal");
+  public void leftOuterDuplicateKeysFirstTable() {
+    Table joined =
+        ANIMAL_NAMES
+            .joinOn("Animal")
+            .with(ANIMAL_FEED)
+            .type(LEFT_OUTER)
+            .rightJoinColumns("Animal")
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(6, joined.rowCount());
     assertEquals(6, joined.column("Animal").size());
@@ -747,7 +832,9 @@ public class DataFrameJoinerTest {
 
   @Test
   public void innerJoinDuplicateKeysSecondTable() {
-    Table joined = ANIMAL_FEED.joinOn("Animal").inner(ANIMAL_NAMES, "Animal");
+    Table joined =
+        ANIMAL_FEED.joinOn("Animal").with(ANIMAL_NAMES).rightJoinColumns("Animal").join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
@@ -760,14 +847,22 @@ public class DataFrameJoinerTest {
     TextColumn nameCol = names.stringColumn("Animal").asTextColumn();
     nameCol = nameCol.where(Selection.withRange(0, feed.rowCount()));
     feed.replaceColumn("Animal", nameCol);
-    Table joined = ANIMAL_FEED.joinOn("Animal").inner(ANIMAL_NAMES, "Animal");
+    Table joined =
+        ANIMAL_FEED.joinOn("Animal").with(ANIMAL_NAMES).rightJoinColumns("Animal").join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(3, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoinDuplicateKeysSecondTable() {
-    Table joined = ANIMAL_FEED.joinOn("Animal").leftOuter(ANIMAL_NAMES, "Animal");
+  public void leftOuterDuplicateKeysSecondTable() {
+    Table joined =
+        ANIMAL_FEED
+            .joinOn("Animal")
+            .with(ANIMAL_NAMES)
+            .rightJoinColumns("Animal")
+            .type(LEFT_OUTER)
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(6, joined.rowCount());
     assertEquals(6, joined.column("Animal").size());
@@ -775,7 +870,7 @@ public class DataFrameJoinerTest {
 
   @Test
   public void fullOuterJoinJustTable() {
-    Table joined = ANIMAL_FEED.joinOn("Animal").fullOuter(ANIMAL_NAMES);
+    Table joined = ANIMAL_FEED.joinOn("Animal").with(ANIMAL_NAMES).type(FULL_OUTER).join();
     assertEquals(3, joined.columnCount());
     assertEquals(8, joined.rowCount());
     assertEquals(8, joined.column("Animal").size());
@@ -788,7 +883,13 @@ public class DataFrameJoinerTest {
 
   @Test
   public void fullOuterJoin() {
-    Table joined = ANIMAL_FEED.joinOn("Animal").fullOuter(ANIMAL_NAMES, "Animal");
+    Table joined =
+        ANIMAL_FEED
+            .joinOn("Animal")
+            .with(ANIMAL_NAMES)
+            .type(FULL_OUTER)
+            .rightJoinColumns("Animal")
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(8, joined.rowCount());
     assertEquals(8, joined.column("Animal").size());
@@ -801,14 +902,25 @@ public class DataFrameJoinerTest {
 
   @Test
   public void fullOuterJoinColTwoOnlyJoinKeys() {
-    Table joined = ANIMAL_FEED.joinOn("Animal").fullOuter(ANIMAL_NAMES.selectColumns("Animal"));
+    Table joined =
+        ANIMAL_FEED
+            .joinOn("Animal")
+            .with(ANIMAL_NAMES.selectColumns("Animal"))
+            .type(FULL_OUTER)
+            .join();
     assertEquals(2, joined.columnCount());
     assertEquals(8, joined.rowCount());
   }
 
   @Test
   public void fullOuterJoinNew() {
-    Table joined = ANIMAL_FEED.joinOn("Animal").fullOuter(true, ANIMAL_NAMES);
+    Table joined =
+        ANIMAL_FEED
+            .joinOn("Animal")
+            .allowDuplicateColumnNames(true)
+            .type(FULL_OUTER)
+            .with(ANIMAL_NAMES)
+            .join();
     assertEquals(3, joined.columnCount());
     assertEquals(8, joined.rowCount());
     assertEquals(8, joined.column("Animal").size());
@@ -824,7 +936,13 @@ public class DataFrameJoinerTest {
     Table table1 = createGOODS1();
     Table table2 = createGOODS2();
     Table table3 = createGOODS3();
-    Table joined = table1.joinOn("Name").fullOuter(true, table2, table3);
+    Table joined =
+        table1
+            .joinOn("Name")
+            .allowDuplicateColumnNames(true)
+            .with(table2, table3)
+            .type(FULL_OUTER)
+            .join();
     assertEquals(7, joined.columnCount());
     assertTrue(
         joined
@@ -850,7 +968,13 @@ public class DataFrameJoinerTest {
     Table table3 = createGOODS3();
     Table table2 = createGOODS2();
     Table table1 = createGOODS1();
-    Table joined = table3.joinOn("Name").fullOuter(true, table2, table1);
+    Table joined =
+        table3
+            .joinOn("Name")
+            .allowDuplicateColumnNames(true)
+            .type(FULL_OUTER)
+            .with(table2, table1)
+            .join();
     assertEquals(7, joined.columnCount());
     assertTrue(
         joined
@@ -876,7 +1000,13 @@ public class DataFrameJoinerTest {
     Table table2 = createGOODS2();
     Table table3 = createGOODS3();
     Table table1 = createGOODS1();
-    Table joined = table2.joinOn("Name").fullOuter(true, table3, table1);
+    Table joined =
+        table2
+            .joinOn("Name")
+            .allowDuplicateColumnNames(true)
+            .type(FULL_OUTER)
+            .with(table3, table1)
+            .join();
     assertEquals(7, joined.columnCount());
     assertTrue(
         joined
@@ -901,7 +1031,7 @@ public class DataFrameJoinerTest {
   public void innerJoinStudentInstructorOnAge() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("Age").inner(true, table2);
+    Table joined = table1.joinOn("Age").allowDuplicateColumnNames(true).with(table2).join();
     assert (joined
         .columnNames()
         .containsAll(Arrays.asList("T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear")));
@@ -916,7 +1046,7 @@ public class DataFrameJoinerTest {
     List<String> originalColumns =
         table2.columns().stream().map(Column::name).collect(Collectors.toList());
 
-    table1.joinOn("Age").inner(true, table2);
+    table1.joinOn("Age").with(table2).allowDuplicateColumnNames(true).join();
 
     List<String> newColumns =
         table2.columns().stream().map(Column::name).collect(Collectors.toList());
@@ -927,10 +1057,11 @@ public class DataFrameJoinerTest {
   public void innerJoinInstructorStudentOnAge() {
     Table table1 = createINSTRUCTOR();
     Table table2 = createSTUDENT();
-    Table joined = table1.joinOn("Age").inner(true, table2);
+    Table joined = table1.joinOn("Age").with(table2).allowDuplicateColumnNames(true).join();
     assert (joined
         .columnNames()
         .containsAll(Arrays.asList("T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(16, joined.columnCount());
     assertEquals(14, joined.rowCount());
   }
@@ -940,12 +1071,13 @@ public class DataFrameJoinerTest {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
     Table table3 = createCLASS();
-    Table joined = table1.joinOn("Age").inner(true, table2, table3);
+    Table joined = table1.joinOn("Age").with(table2, table3).allowDuplicateColumnNames(true).join();
     assert (joined
         .columnNames()
         .containsAll(
             Arrays.asList("T2.ID", "T2.City", "T2.State", "T2.USID", "T2.GradYear", "T3.ID")));
     assertEquals(24, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(14, joined.rowCount());
   }
 
@@ -955,7 +1087,8 @@ public class DataFrameJoinerTest {
     Table table2 = createINSTRUCTOR();
     Table table3 = createCLASS();
     Table table4 = createDEPTHEAD();
-    Table joined = table1.joinOn("Age").inner(true, table2, table3, table4);
+    Table joined =
+        table1.joinOn("Age").with(table2, table3, table4).allowDuplicateColumnNames(true).join();
     assert (joined
         .columnNames()
         .containsAll(
@@ -971,6 +1104,7 @@ public class DataFrameJoinerTest {
                 "T4.Last",
                 "T4.City",
                 "T4.State")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(30, joined.columnCount());
     assertEquals(14, joined.rowCount());
   }
@@ -981,7 +1115,8 @@ public class DataFrameJoinerTest {
     Table table2 = createINSTRUCTOR();
     Table table3 = createCLASS();
     Table table4 = createDEPTHEAD();
-    Table joined = table1.joinOn("Age").inner(true, table2, table3, table4);
+    Table joined =
+        table1.joinOn("Age").with(table2, table3, table4).allowDuplicateColumnNames(true).join();
     List<String> expectedCol =
         Arrays.asList(
             "ID",
@@ -1016,6 +1151,7 @@ public class DataFrameJoinerTest {
             "T4.State");
     assert (String.join("", joined.columnNames()).equals(String.join("", expectedCol)));
     assertEquals(30, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(28, joined.rowCount());
   }
 
@@ -1024,7 +1160,8 @@ public class DataFrameJoinerTest {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
     Table table3 = createDEPTHEAD();
-    Table joined = table1.joinOn("State", "Age").inner(true, table2, table3);
+    Table joined =
+        table1.joinOn("State", "Age").with(table2, table3).allowDuplicateColumnNames(true).join();
     assert (joined
         .columnNames()
         .containsAll(
@@ -1038,6 +1175,7 @@ public class DataFrameJoinerTest {
                 "T3.Last",
                 "T3.City")));
     assertEquals(20, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(1, joined.rowCount());
   }
   // Tests for left table is larger than right table, multiple cols
@@ -1048,7 +1186,8 @@ public class DataFrameJoinerTest {
     Table table3 = createDEPTHEAD();
     // table1 join table 2 will have 6 rows, so drop last 5 rows of table3 for test
     table3 = table3.dropRows(5, 6, 7, 8, 9);
-    Table joined = table1.joinOn("State", "Age").inner(true, table2, table3);
+    Table joined =
+        table1.joinOn("State", "Age").with(table2, table3).allowDuplicateColumnNames(true).join();
     List<String> expectedCol =
         Arrays.asList(
             "ID",
@@ -1072,6 +1211,7 @@ public class DataFrameJoinerTest {
             "Dept",
             "T3.City");
     // test for col order
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assert (String.join("", joined.columnNames()).equals(String.join("", expectedCol)));
     assertEquals(20, joined.columnCount());
     assertEquals(2, joined.rowCount());
@@ -1081,7 +1221,9 @@ public class DataFrameJoinerTest {
   public void innerJoinStudentInstructorOnStateAge() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "Age").inner(true, table2);
+    Table joined =
+        table1.joinOn("State", "Age").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(15, joined.columnCount());
     assertEquals(3, joined.rowCount());
   }
@@ -1090,7 +1232,12 @@ public class DataFrameJoinerTest {
   public void innerJoinStudentInstructorOnStateAgeGradYear() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "Age", "GradYear").inner(true, table2);
+    Table joined =
+        table1
+            .joinOn("State", "Age", "GradYear")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .join();
     assertEquals(14, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
@@ -1099,7 +1246,13 @@ public class DataFrameJoinerTest {
   public void leftJoinStudentInstructorOnStateAge() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "Age").leftOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("State", "Age")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(LEFT_OUTER)
+            .join();
     assertEquals(15, joined.columnCount());
     assertEquals(10, joined.rowCount());
     assertEquals(10, joined.column("State").size());
@@ -1111,7 +1264,13 @@ public class DataFrameJoinerTest {
     Table table1 = createGOODS1();
     Table table2 = createGOODS2();
     Table table3 = createGOODS3();
-    Table joined = table1.joinOn("Name").leftOuter(true, table2, table3);
+    Table joined =
+        table1
+            .joinOn("Name")
+            .allowDuplicateColumnNames(true)
+            .with(table2, table3)
+            .type(LEFT_OUTER)
+            .join();
     assertEquals(7, joined.columnCount());
     assertTrue(
         joined
@@ -1119,6 +1278,7 @@ public class DataFrameJoinerTest {
             .containsAll(
                 Arrays.asList("ID", "Name", "Price", "T2.ID", "T2.Price", "T3.ID", "T3.Price")));
     assertEquals(6, joined.rowCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(6, joined.column("Price").size());
     assertEquals(6, joined.column("T2.ID").size());
     assertEquals(2, joined.column("T2.ID").countMissing());
@@ -1131,8 +1291,14 @@ public class DataFrameJoinerTest {
     Table table1 = createHOUSE();
     Table table2 = createBOAT();
     Table joined =
-        table1.joinOn("Bedrooms", "Owner").inner(table2, new String[] {"Bedrooms", "Owner"});
+        table1
+            .joinOn("Bedrooms", "Owner")
+            .with(table2)
+            .rightJoinColumns("Bedrooms", "Owner")
+            .join();
     assertEquals(6, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(2, joined.rowCount());
     assertEquals(2, joined.column("Bedrooms").size());
   }
@@ -1144,7 +1310,11 @@ public class DataFrameJoinerTest {
     Table joined =
         table1
             .joinOn("Style", "Bedrooms", "Owner")
-            .inner(table2, new String[] {"Type", "Bedrooms", "Owner"});
+            .with(table2)
+            .rightJoinColumns("Type", "Bedrooms", "Owner")
+            .join();
+
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(5, joined.columnCount());
     assertEquals(1, joined.rowCount());
   }
@@ -1153,7 +1323,14 @@ public class DataFrameJoinerTest {
   public void fullJoinHouseBoatOnBedroomsOwner() {
     Table table1 = createHOUSE();
     Table table2 = createBOAT();
-    Table joined = table1.joinOn("Bedrooms", "Owner").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("Bedrooms", "Owner")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(FULL_OUTER)
+            .join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(6, joined.columnCount());
     assertEquals(7, joined.rowCount());
     assertEquals(7, joined.column("Style").size());
@@ -1174,7 +1351,13 @@ public class DataFrameJoinerTest {
   public void fullJoinHouse10Boat10OnBedroomsOwner() {
     Table table1 = createHOUSE10();
     Table table2 = createBOAT10();
-    Table joined = table1.joinOn("Bedrooms", "Owner").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("Bedrooms", "Owner")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(FULL_OUTER)
+            .join();
     assertEquals(6, joined.columnCount());
     assertEquals(11, joined.rowCount());
     assertEquals(11, joined.column("Bedrooms").size());
@@ -1195,7 +1378,13 @@ public class DataFrameJoinerTest {
   public void fullJoinBnBBoat10OnBedroomsOwner() {
     Table table1 = createBEDANDBREAKFAST();
     Table table2 = createBOAT10();
-    Table joined = table1.joinOn("Bedrooms", "SoldDate").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("Bedrooms", "SoldDate")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(FULL_OUTER)
+            .join();
     assertEquals(6, joined.columnCount());
     assertEquals(11, joined.rowCount());
     assertEquals(11, joined.column("Design").size());
@@ -1217,7 +1406,12 @@ public class DataFrameJoinerTest {
     Table table1 = createHOUSE();
     Table table2 = createBOAT();
     Table joined =
-        table1.joinOn("Bedrooms", "Owner").leftOuter(table2, new String[] {"Bedrooms", "Owner"});
+        table1
+            .joinOn("Bedrooms", "Owner")
+            .with(table2)
+            .type(LEFT_OUTER)
+            .rightJoinColumns("Bedrooms", "Owner")
+            .join();
     assertEquals(6, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
@@ -1229,7 +1423,10 @@ public class DataFrameJoinerTest {
     Table joined =
         table1
             .joinOn("Style", "Bedrooms", "Owner")
-            .leftOuter(table2, new String[] {"Type", "Bedrooms", "Owner"});
+            .with(table2)
+            .type(LEFT_OUTER)
+            .rightJoinColumns("Type", "Bedrooms", "Owner")
+            .join();
     assertEquals(5, joined.columnCount());
     assertEquals(4, joined.rowCount());
   }
@@ -1239,7 +1436,12 @@ public class DataFrameJoinerTest {
     Table table1 = createHOUSE();
     Table table2 = createBOAT();
     Table joined =
-        table1.joinOn("Bedrooms", "Owner").rightOuter(table2, new String[] {"Bedrooms", "Owner"});
+        table1
+            .joinOn("Bedrooms", "Owner")
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .rightJoinColumns("Bedrooms", "Owner")
+            .join();
     assertEquals(6, joined.columnCount());
     assertEquals(5, joined.rowCount());
   }
@@ -1251,7 +1453,10 @@ public class DataFrameJoinerTest {
     Table joined =
         table1
             .joinOn("Style", "Bedrooms", "Owner")
-            .rightOuter(table2, new String[] {"Type", "Bedrooms", "Owner"});
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .rightJoinColumns("Type", "Bedrooms", "Owner")
+            .join();
     assertEquals(5, joined.columnCount());
     assertEquals(5, joined.rowCount());
   }
@@ -1260,7 +1465,13 @@ public class DataFrameJoinerTest {
   public void rightJoinStudentInstructorOnStateAge() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "Age").rightOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("State", "Age")
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .join();
     assertEquals(15, joined.columnCount());
     assertEquals(10, joined.rowCount());
   }
@@ -1269,7 +1480,13 @@ public class DataFrameJoinerTest {
   public void innerJoinStudentInstructorOnStateName() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "FirstName").inner(table2, true, "State", "First");
+    Table joined =
+        table1
+            .joinOn("State", "FirstName")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .rightJoinColumns("State", "First")
+            .join();
     assertEquals(15, joined.columnCount());
     assertEquals(5, joined.rowCount());
   }
@@ -1278,7 +1495,14 @@ public class DataFrameJoinerTest {
   public void leftJoinStudentInstructorOnStateName() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "FirstName").leftOuter(table2, true, "State", "First");
+    Table joined =
+        table1
+            .joinOn("State", "FirstName")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .type(LEFT_OUTER)
+            .rightJoinColumns("State", "First")
+            .join();
     assertEquals(15, joined.columnCount());
     assertEquals(10, joined.rowCount());
   }
@@ -1287,7 +1511,14 @@ public class DataFrameJoinerTest {
   public void rightJoinStudentInstructorOnStateName() {
     Table table1 = createSTUDENT();
     Table table2 = createINSTRUCTOR();
-    Table joined = table1.joinOn("State", "FirstName").rightOuter(table2, true, "State", "First");
+    Table joined =
+        table1
+            .joinOn("State", "FirstName")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .rightJoinColumns("State", "First")
+            .type(RIGHT_OUTER)
+            .join();
     assertEquals(15, joined.columnCount());
     assertEquals(10, joined.rowCount());
   }
@@ -1296,7 +1527,14 @@ public class DataFrameJoinerTest {
   public void innerJoinOnAge() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
-    Table joined = table1.joinOn("Age").inner(table2, "Age", true);
+    Table joined =
+        table1
+            .joinOn("Age")
+            .with(table2)
+            .rightJoinColumns("Age")
+            .allowDuplicateColumnNames(true)
+            .join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(9, joined.columnCount());
     assertEquals(18, joined.rowCount());
   }
@@ -1305,7 +1543,8 @@ public class DataFrameJoinerTest {
   public void innerJoinAnimalPeopleOnAge() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
-    Table joined = table1.joinOn("Age").inner(true, table2);
+    Table joined = table1.joinOn("Age").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(9, joined.columnCount());
     assertEquals(18, joined.rowCount());
   }
@@ -1314,7 +1553,8 @@ public class DataFrameJoinerTest {
   public void innerJoinAnimalTreeOnAge() {
     Table table1 = createANIMALHOMES();
     Table table2 = createTREE();
-    Table joined = table1.joinOn("Age").inner(true, table2);
+    Table joined = table1.joinOn("Age").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(7, joined.columnCount());
     assertEquals(8, joined.rowCount());
   }
@@ -1325,7 +1565,8 @@ public class DataFrameJoinerTest {
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
     Table table3 = createTREE();
     Table table4 = createFLOWER();
-    Table joined = table1.joinOn("Age").inner(true, table2, table3, table4);
+    Table joined =
+        table1.joinOn("Age").with(table2, table3, table4).allowDuplicateColumnNames(true).join();
     assert (joined
         .columnNames()
         .containsAll(
@@ -1338,6 +1579,8 @@ public class DataFrameJoinerTest {
                 "T4.Name",
                 "T4.Home",
                 "Color")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(14, joined.columnCount());
     assertEquals(18, joined.rowCount());
   }
@@ -1348,7 +1591,12 @@ public class DataFrameJoinerTest {
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
     Table table3 = createTREE();
     Table table4 = createFLOWER();
-    Table joined = table1.joinOn("Age", "Home").inner(true, table2, table3, table4);
+    Table joined =
+        table1
+            .joinOn("Age", "Home")
+            .with(table2, table3, table4)
+            .allowDuplicateColumnNames(true)
+            .join();
     assert (joined
         .columnNames()
         .containsAll(
@@ -1364,6 +1612,7 @@ public class DataFrameJoinerTest {
                 "T3.Name",
                 "T4.Name",
                 "Color")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(11, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
@@ -1372,7 +1621,9 @@ public class DataFrameJoinerTest {
   public void innerJoinOnNameHomeAge() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
-    Table joined = table1.joinOn("Name", "Home", "Age").inner(true, table2);
+    Table joined =
+        table1.joinOn("Name", "Home", "Age").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(7, joined.columnCount());
     assertEquals(1, joined.rowCount());
   }
@@ -1382,7 +1633,13 @@ public class DataFrameJoinerTest {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENicknameDwellingYearsMoveInDate();
     Table joined =
-        table1.joinOn("Name", "Home", "Age").inner(table2, true, "Nickname", "Dwelling", "Years");
+        table1
+            .joinOn("Name", "Home", "Age")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .rightJoinColumns("Nickname", "Dwelling", "Years")
+            .join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(7, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
@@ -1392,18 +1649,30 @@ public class DataFrameJoinerTest {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate();
     Table joined =
-        table1.joinOn("Name", "Home", "Age").inner(table2, true, "Name", "Dwelling", "Years");
+        table1
+            .joinOn("Name", "Home", "Age")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .rightJoinColumns("Name", "Dwelling", "Years")
+            .join();
     assert (joined.columnNames().containsAll(Arrays.asList("Name", "Home", "Age")));
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(7, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoinOnPartiallyMismatchedColNames() {
+  public void leftOuterOnPartiallyMismatchedColNames() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate();
     Table joined =
-        table1.joinOn("Name", "Home", "Age").leftOuter(table2, true, "Name", "Dwelling", "Years");
+        table1
+            .joinOn("Name", "Home", "Age")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .type(LEFT_OUTER)
+            .rightJoinColumns("Name", "Dwelling", "Years")
+            .join();
     assert (joined.columnNames().containsAll(Arrays.asList("Name", "Home", "Age")));
     assertEquals(7, joined.columnCount());
     assertEquals(8, joined.rowCount());
@@ -1414,7 +1683,13 @@ public class DataFrameJoinerTest {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameDwellingYearsMoveInDate();
     Table joined =
-        table1.joinOn("Name", "Home", "Age").rightOuter(table2, true, "Name", "Dwelling", "Years");
+        table1
+            .joinOn("Name", "Home", "Age")
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .rightJoinColumns("Name", "Dwelling", "Years")
+            .join();
     assert (joined.columnNames().containsAll(Arrays.asList("Name", "Dwelling", "Years")));
     assertEquals(7, joined.columnCount());
     assertEquals(6, joined.rowCount());
@@ -1424,26 +1699,41 @@ public class DataFrameJoinerTest {
   public void innerJoinOnAgeMoveInDate() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
-    Table joined = table1.joinOn("Age", "MoveInDate").inner(true, table2);
+    Table joined =
+        table1.joinOn("Age", "MoveInDate").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(8, joined.columnCount());
     assertEquals(3, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoinOnAgeMoveInDate() {
+  public void leftOuterOnAgeMoveInDate() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
-    Table joined = table1.joinOn("Age", "MoveInDate").leftOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("Age", "MoveInDate")
+            .type(LEFT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .join();
     assertEquals(8, joined.columnCount());
     assertEquals(9, joined.rowCount());
   }
 
   @Test
-  public void leftOuterJoin_keepAllJoinKeyColumns() {
+  public void leftOuterKeepAllJoinKeyColumns() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
     Table joined =
-        table1.joinOn("Age", "MoveInDate").leftOuter(table2, true, true, "Age", "MoveInDate");
+        table1
+            .joinOn("Age", "MoveInDate")
+            .with(table2)
+            .type(LEFT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .keepAllJoinKeyColumns(true)
+            .rightJoinColumns("Age", "MoveInDate")
+            .join();
     assertEquals(10, joined.columnCount());
     assertEquals(9, joined.rowCount());
   }
@@ -1453,7 +1743,14 @@ public class DataFrameJoinerTest {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
     Table joined =
-        table1.joinOn("Age", "MoveInDate").rightOuter(table2, true, true, "Age", "MoveInDate");
+        table1
+            .joinOn("Age", "MoveInDate")
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .keepAllJoinKeyColumns(true)
+            .rightJoinColumns("Age", "MoveInDate")
+            .join();
     assertEquals(10, joined.columnCount());
     assertEquals(6, joined.rowCount());
   }
@@ -1463,7 +1760,14 @@ public class DataFrameJoinerTest {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
     Table joined =
-        table1.joinOn("Age", "MoveInDate").fullOuter(table2, true, true, "Age", "MoveInDate");
+        table1
+            .joinOn("Age", "MoveInDate")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .keepAllJoinKeyColumns(true)
+            .type(FULL_OUTER)
+            .rightJoinColumns("Age", "MoveInDate")
+            .join();
     assertEquals(10, joined.columnCount());
     assertEquals(12, joined.rowCount());
   }
@@ -1473,8 +1777,15 @@ public class DataFrameJoinerTest {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
     Table joined =
-        table1.joinOn("Name", "Home", "Age").inner(table2, true, true, "Name", "Home", "Age");
+        table1
+            .joinOn("Name", "Home", "Age")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .keepAllJoinKeyColumns(true)
+            .rightJoinColumns("Name", "Home", "Age")
+            .join();
     assertEquals(10, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(1, joined.rowCount());
   }
 
@@ -1482,7 +1793,13 @@ public class DataFrameJoinerTest {
   public void rightOuterJoinOnAgeMoveInDate() {
     Table table1 = createANIMALHOMES();
     Table table2 = createDOUBLEINDEXEDPEOPLENameHomeAgeMoveInDate();
-    Table joined = table1.joinOn("Age", "MoveInDate").rightOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("Age", "MoveInDate")
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .join();
     assertEquals(8, joined.columnCount());
     assertEquals(6, joined.rowCount());
   }
@@ -1491,7 +1808,8 @@ public class DataFrameJoinerTest {
   public void innerJoinFootballSoccerOnPlayDate() {
     Table table1 = createFOOTBALLSCHEDULE();
     Table table2 = createSOCCERSCHEDULE();
-    Table joined = table1.joinOn("PlayDate").inner(true, table2);
+    Table joined = table1.joinOn("PlayDate").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(8, joined.columnCount());
     assertEquals(5, joined.rowCount());
   }
@@ -1500,7 +1818,8 @@ public class DataFrameJoinerTest {
   public void innerJoinFootballSoccerOnPlayTime() {
     Table table1 = createFOOTBALLSCHEDULE();
     Table table2 = createSOCCERSCHEDULE();
-    Table joined = table1.joinOn("PlayTime").inner(true, table2);
+    Table joined = table1.joinOn("PlayTime").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(8, joined.columnCount());
     assertEquals(3, joined.rowCount());
   }
@@ -1509,7 +1828,8 @@ public class DataFrameJoinerTest {
   public void innerJoinFootballSoccerOnPlayDatePlayTime() {
     Table table1 = createFOOTBALLSCHEDULE();
     Table table2 = createSOCCERSCHEDULE();
-    Table joined = table1.joinOn("PlayDate", "PlayTime").inner(true, table2);
+    Table joined =
+        table1.joinOn("PlayDate", "PlayTime").with(table2).allowDuplicateColumnNames(true).join();
     assertEquals(7, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
@@ -1518,7 +1838,13 @@ public class DataFrameJoinerTest {
   public void fullOuterJoinFootballSoccerOnPlayTime() {
     Table table1 = createFOOTBALLSCHEDULE();
     Table table2 = createSOCCERSCHEDULE();
-    Table joined = table1.joinOn("PlayTime").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("PlayTime")
+            .allowDuplicateColumnNames(true)
+            .type(FULL_OUTER)
+            .with(table2)
+            .join();
     assertEquals(8, joined.columnCount());
     assertEquals(5, joined.rowCount());
   }
@@ -1527,7 +1853,9 @@ public class DataFrameJoinerTest {
   public void innerJoinFootballSoccerOnPlayDatePlayDateTime() {
     Table table1 = createFOOTBALLSCHEDULEDateTime();
     Table table2 = createSOCCERSCHEDULEDateTime();
-    Table joined = table1.joinOn("PlayDateTime").inner(true, table2);
+    Table joined =
+        table1.joinOn("PlayDateTime").with(table2).allowDuplicateColumnNames(true).join();
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(10, joined.columnCount());
     assertEquals(2, joined.rowCount());
   }
@@ -1536,8 +1864,14 @@ public class DataFrameJoinerTest {
   public void innerJoinFootballSoccerOnSeasonRevenue() {
     Table table1 = createFOOTBALLSCHEDULEDateTime();
     Table table2 = createSOCCERSCHEDULEDateTime();
-    Table joined = table1.joinOn("SeasonRevenue", "AllTimeRevenue").inner(true, table2);
+    Table joined =
+        table1
+            .joinOn("SeasonRevenue", "AllTimeRevenue")
+            .with(table2)
+            .allowDuplicateColumnNames(true)
+            .join();
     assertEquals(9, joined.columnCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     assertEquals(1, joined.rowCount());
   }
 
@@ -1545,7 +1879,13 @@ public class DataFrameJoinerTest {
   public void fullOuterJoinFootballSoccerOnPlayDateTimeSeasonRevenue() {
     Table table1 = createFOOTBALLSCHEDULEDateTime();
     Table table2 = createSOCCERSCHEDULEDateTime();
-    Table joined = table1.joinOn("PlayDateTime", "SeasonRevenue").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("PlayDateTime", "SeasonRevenue")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(FULL_OUTER)
+            .join();
     assertEquals(9, joined.columnCount());
     assertEquals(6, joined.rowCount());
     assertEquals(6, joined.column("TeamName").size());
@@ -1565,10 +1905,16 @@ public class DataFrameJoinerTest {
   }
 
   @Test
-  public void fullOuterJoinFootballSoccerOnPlayDateTimeAllTimeRevenue() {
+  public void fullOuterJoinFootballSoccerOnAllTimeRevenue() {
     Table table1 = createFOOTBALLSCHEDULEDateTime();
     Table table2 = createSOCCERSCHEDULEDateTime();
-    Table joined = table1.joinOn("AllTimeRevenue").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("AllTimeRevenue")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(FULL_OUTER)
+            .join();
     assertEquals(10, joined.columnCount());
     assertEquals(5, joined.rowCount());
     assertEquals(5, joined.column("TeamName").size());
@@ -1593,40 +1939,73 @@ public class DataFrameJoinerTest {
     assertEquals(1, joined.column("T2.SeasonRevenue").countMissing());
   }
 
+  @Test
+  public void rightOuterJoinFootballSoccerOnAllTimeRevenue() {
+    Table table1 = createFOOTBALLSCHEDULEDateTime();
+    Table table2 = createSOCCERSCHEDULEDateTime();
+    Table joined =
+        table1
+            .joinOn("AllTimeRevenue")
+            .with(table2)
+            .type(RIGHT_OUTER)
+            .allowDuplicateColumnNames(true)
+            .join();
+    assertEquals(10, joined.columnCount());
+    assertEquals(4, joined.rowCount());
+    assertEquals(4, joined.column("SeasonRevenue").size());
+    assertEquals(1, joined.column("SeasonRevenue").countMissing());
+  }
+
+  @Test
+  public void innerJoinFootballSoccerOnAllTimeRevenue() {
+    Table table1 = createFOOTBALLSCHEDULEDateTime();
+    Table table2 = createSOCCERSCHEDULEDateTime();
+    Table joined =
+        table1.joinOn("AllTimeRevenue").with(table2).allowDuplicateColumnNames(true).join();
+    assertEquals(10, joined.columnCount());
+    assertEquals(3, joined.rowCount());
+    assertEquals(3, joined.column("SeasonRevenue").size());
+  }
+
   @Test
   public void fullOuterJoinFootballBaseballBoolean() {
     Table table1 = createFOOTBALLSCHEDULE();
     Table table2 = createBASEBALLSCHEDULEDateTime();
-    Table joined = table1.joinOn("HomeGame").fullOuter(true, table2);
+    Table joined =
+        table1
+            .joinOn("HomeGame")
+            .allowDuplicateColumnNames(true)
+            .with(table2)
+            .type(FULL_OUTER)
+            .join();
     assertEquals(10, joined.columnCount());
     assertEquals(8, joined.rowCount());
   }
 
   @Test
   public void differentColumnTypes() {
-    Table table1 = STRING_INDEXED_PEOPLE;
-    Table table2 = DOUBLE_INDEXED_PEOPLE;
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () -> {
-          table1.joinOn("ID").inner(table2);
+          STRING_INDEXED_PEOPLE.joinOn("ID").with(DOUBLE_INDEXED_PEOPLE).join();
         });
   }
 
   @Test
   public void innerJoinEmptyLeftTable() {
     Table leftTable = Table.create(StringColumn.create("Animal"));
-    Table joined = leftTable.joinOn("Animal").inner(ANIMAL_NAMES);
+    Table joined = leftTable.joinOn("Animal").with(ANIMAL_NAMES).join();
     assertEquals(0, joined.rowCount());
+    assertTrue(joined.columnNames().stream().noneMatch(e -> e.startsWith("Placeholder_")));
     for (Column<?> column : joined.columnArray()) {
       assertEquals(0, column.size());
     }
   }
 
   @Test
-  public void leftOuterJoinEmptyLeftTable() {
+  public void leftOuterEmptyLeftTable() {
     Table leftTable = Table.create(StringColumn.create("Animal"));
-    Table joined = leftTable.joinOn("Animal").inner(ANIMAL_NAMES);
+    Table joined = leftTable.joinOn("Animal").with(ANIMAL_NAMES).join();
     assertEquals(0, joined.rowCount());
     for (Column<?> column : joined.columnArray()) {
       assertEquals(0, column.size());
diff --git a/core/src/test/java/tech/tablesaw/perf/DataFrameJoinerPerformanceTest.java b/core/src/test/java/tech/tablesaw/perf/DataFrameJoinerPerformanceTest.java
new file mode 100644
index 00000000..1044db02
--- /dev/null
+++ b/core/src/test/java/tech/tablesaw/perf/DataFrameJoinerPerformanceTest.java
@@ -0,0 +1,159 @@
+package tech.tablesaw.perf;
+
+import static org.junit.jupiter.api.Assertions.assertTimeout;
+import static tech.tablesaw.joining.JoinType.FULL_OUTER;
+import static tech.tablesaw.joining.JoinType.LEFT_OUTER;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.Table;
+
+@Tag("Slow")
+@Tag("Flaky")
+public class DataFrameJoinerPerformanceTest {
+
+  private static final long SEED = 200L;
+
+  private static final int CUSTOMER_COUNT = 1_000;
+  private static final int ORDER_COUNT = 10_000;
+
+  private static final int CUSTOMER_FILL_COL_COUNT = 5;
+  private static final int ORDER_FILL_COL_COUNT = 5;
+
+  private static final int TIME_OUT_MILLIES = 1000;
+
+  private static Table customers;
+  private static Table orders;
+
+  private static final Map<Integer, Integer> REGION_MAP = new HashMap<>();
+
+  @BeforeAll
+  static void setup() {
+    customers = createCustomersTable(CUSTOMER_COUNT);
+    orders = createOrdersTable(ORDER_COUNT, CUSTOMER_COUNT);
+    addFillerColumns(customers, orders, CUSTOMER_FILL_COL_COUNT, ORDER_FILL_COL_COUNT);
+  }
+
+  private static void addFillerColumn(Table table, int numberColumnsToAdd, String prefix) {
+    int[] filler = new int[table.rowCount()];
+    Arrays.fill(filler, 1);
+    IntColumn col = IntColumn.create("temp", filler);
+    for (int i = 0; i < numberColumnsToAdd; i++) {
+      table.addColumns(col.copy().setName(prefix + "_appendColumn" + i));
+    }
+  }
+
+  private static Table createCustomersTable(int numberCustomers) {
+    Random random = new Random(SEED);
+
+    Table customersTable = Table.create("customers");
+    IntColumn customerIds =
+        IntColumn.create("customerId", IntStream.range(0, numberCustomers).toArray());
+    IntColumn regions = IntColumn.create("region", numberCustomers);
+    for (int i = 0; i < numberCustomers; i++) {
+      int val = random.nextInt(49);
+      REGION_MAP.put(customerIds.get(i), val);
+      regions.set(i, val);
+    }
+    customersTable.addColumns(customerIds, regions);
+    return customersTable;
+  }
+
+  private static Table createOrdersTable(int numberOrders, int numberCustomers) {
+    Table ordersTable = Table.create("orders");
+    Random random = new Random(SEED);
+    IntColumn orderCustomerIds =
+        IntColumn.create(
+            "customerId",
+            random
+                .doubles()
+                .limit(numberOrders)
+                .mapToInt(randomDouble -> (int) Math.floor(randomDouble * numberCustomers))
+                .toArray());
+
+    IntColumn regions = IntColumn.create("region", numberOrders);
+    for (int i = 0; i < numberOrders; i++) {
+      int customer = orderCustomerIds.getInt(i);
+      int region = REGION_MAP.get(customer);
+      regions.set(i, region);
+    }
+    ordersTable.addColumns(orderCustomerIds, regions);
+    return ordersTable;
+  }
+
+  @Test
+  public void innerJoinCustomersFirst() {
+    assertTimeout(
+        Duration.ofMillis(TIME_OUT_MILLIES),
+        () -> customers.joinOn("customerId").with(orders).allowDuplicateColumnNames(true).join());
+  }
+
+  @Test
+  public void innerJoinCustomersFirst2() {
+    assertTimeout(
+        Duration.ofMillis(TIME_OUT_MILLIES),
+        () ->
+            customers
+                .joinOn("customerId", "region")
+                .with(orders)
+                .allowDuplicateColumnNames(true)
+                .join());
+  }
+
+  @Test
+  public void innerJoinOrdersFirst() {
+    assertTimeout(
+        Duration.ofMillis(TIME_OUT_MILLIES),
+        () -> orders.joinOn("customerId").with(customers).allowDuplicateColumnNames(true).join());
+  }
+
+  @Test
+  public void leftOuterOrdersFirst() {
+    assertTimeout(
+        Duration.ofMillis(TIME_OUT_MILLIES),
+        () ->
+            orders
+                .joinOn("customerId")
+                .with(customers)
+                .type(LEFT_OUTER)
+                .allowDuplicateColumnNames(true)
+                .join());
+  }
+
+  @Test
+  public void leftOuterCustomersFirst() {
+    assertTimeout(
+        Duration.ofMillis(TIME_OUT_MILLIES),
+        () ->
+            customers
+                .joinOn("customerId")
+                .with(orders)
+                .type(LEFT_OUTER)
+                .allowDuplicateColumnNames(true)
+                .join());
+  }
+
+  @Test
+  public void fullOuterJoin() {
+    assertTimeout(
+        Duration.ofMillis(TIME_OUT_MILLIES),
+        () ->
+            customers
+                .joinOn("customerId")
+                .with(orders)
+                .type(FULL_OUTER)
+                .allowDuplicateColumnNames(true)
+                .join());
+  }
+
+  private static void addFillerColumns(
+      Table customers, Table orders, int customerFillCols, int orderFillCols) {
+    addFillerColumn(customers, customerFillCols, "customer");
+    addFillerColumn(orders, orderFillCols, "order");
+  }
+}
diff --git a/core/src/test/java/tech/tablesaw/perf/JoinPerfTest.java b/core/src/test/java/tech/tablesaw/perf/JoinPerfTest.java
new file mode 100644
index 00000000..ced88621
--- /dev/null
+++ b/core/src/test/java/tech/tablesaw/perf/JoinPerfTest.java
@@ -0,0 +1,60 @@
+package tech.tablesaw.perf;
+
+import static org.junit.jupiter.api.Assertions.assertTimeout;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.Test;
+import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.Table;
+import tech.tablesaw.testutil.NanoBench;
+
+public class JoinPerfTest {
+
+  private static Table createXYZTable(String name, int cardinality) {
+    Table table = Table.create(name);
+    int[] filler = new int[cardinality];
+    Arrays.fill(filler, 1);
+    IntColumn x = IntColumn.create("x", IntStream.range(0, cardinality).toArray());
+    IntColumn y = IntColumn.create("y", IntStream.range(0, cardinality).toArray());
+    IntColumn z = IntColumn.create("z", filler);
+    // IntColumn z = IntColumn.create("z", IntStream.range(0, cardinality).toArray());
+    table.addColumns(x, y, z);
+    return table;
+  }
+
+  @Test
+  void testTimeout() {
+    Table a = createXYZTable("a", 500);
+    Table b = createXYZTable("b", 500);
+    assertTimeout(Duration.ofSeconds(1), () -> a.joinOn("x", "y", "z").with(b).join());
+  }
+
+  @Test
+  void testPerf() {
+    NanoBench bench = NanoBench.create();
+    bench.warmUps(5);
+    bench.measurements(50);
+    Runnable r = new Runner();
+    bench.measure("test", r);
+  }
+
+  static class Runner extends Thread {
+    Table a = createXYZTable("a", 50000);
+    Table b = createXYZTable("b", 50000);
+
+    {
+      for (int i = 2; i <= 14; i++) {
+        a.addColumns(IntColumn.indexColumn(String.valueOf(i), a.rowCount(), 1));
+      }
+      b.addColumns(IntColumn.indexColumn("1", b.rowCount(), 0));
+    }
+
+    public void run() {
+      a.sortDescendingOn("x", "y", "z");
+      b.sortDescendingOn("x", "y", "z");
+      a.joinOn("x", "y", "z").with(b).join();
+    }
+  }
+}