Repository: jtablesaw/tablesaw
Pull request #562 (2019-08-03 00:52:32)

Speed up Joins
Thanks for contributing.

- [x] Tick to sign-off your agreement to the [Developer Certificate of Origin (DCO) 1.1](https://developercertificate.org)

## Description 

Fixes  #383

* Speeds up joins. Especially in the worst cases.
  * In one case a full outer join with a large number of columns went from over 5 mins to a few seconds.
* The performance is less sensitive to which table is on the left hand side of the join, but the join is still noticeably faster when the small table is on the left. This is a possible future improvement.
* Avoids using where and copying columns twice.
* Most of the join time is now spent calculating the cross product (as it should be).
* Makes no API changes.

I fixed a somewhat related issue. If the right hand side table has columns with duplicate names the duplicate columns will be renamed with a prefix. The old implementation mutated the columns on the original table. This PR changes that so the original columns are not changed. See the one added test.

There was good test coverage (Thanks Ben) and all tests are passing.

# Performance

Given a customers table and an orders table with a customerId uniformly distributed among the customerIds in the customers table

> Note: These performance tests do not use a mico benchmarking framework and have a ton of variance.

|num Order rows|num Customer rows|num Orders Cols|num Customers Columns|left hand table in join|new code (seconds)|old code (seconds)|
|---------------|-----------------|---------------|---------------------|-----------------------|------------------|------------------|
|1000000        |1000             |1              |1                    |customers              |0.143             |0.208             |
|1000000        |1000             |11             |1                    |customers              |0.901             |1.015             |
|1000000        |1000             |1              |11                   |customers              |0.364             |0.265             |
|1000000        |1000             |1              |1                    |customers              |0.083             |0.105             |
|1000000        |1000             |1              |6                    |customers              |0.244             |0.179             |
|100000         |1000             |1              |1                    |customers              |0.1               |0.009             |
|100000         |1000             |11             |1                    |customers              |0.405             |0.09              |
|100000         |1000             |1              |11                   |customers              |0.037             |0.039             |
|100000         |1000             |1              |1                    |customers              |0.019             |0.01              |
|100000         |1000             |1              |6                    |customers              |0.048             |0.022             |
|1000000        |1000             |1              |1                    |orders                 |0.884             |5.136             |
|1000000        |1000             |11             |1                    |orders                 |1.44              |18                |
|1000000        |1000             |1              |1                    |orders                 |0.21              |2.185             |
|1000000        |1000             |1              |6                    |orders                 |0.433             |10.182            |
|100000         |1000             |1              |1                    |orders                 |0.525             |0.227             |
|100000         |1000             |11             |1                    |orders                 |0.297             |1.709             |
|100000         |1000             |1              |1                    |orders                 |0.023             |0.211             |
|100000         |1000             |1              |6                    |orders                 |0.047             |0.882             |


## Testing

Yes. Added a unit test to show joining does not change the column names on the original table.

lwhite1: @ryancerf That was impressively fast :). The performance numbers look great.  I'm not sure the changes to the javadoc formatting improve readability, but since the core java library code uses both approaches (indenting/not-indenting on multiline comments) I think it's ok.  

@benmccann Would you like to take a look? 

Patch:
diff --git a/core/src/main/java/tech/tablesaw/api/DoubleColumn.java b/core/src/main/java/tech/tablesaw/api/DoubleColumn.java
index 0a9e91db..32d72580 100644
--- a/core/src/main/java/tech/tablesaw/api/DoubleColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/DoubleColumn.java
@@ -16,6 +16,7 @@ import tech.tablesaw.columns.Column;
 import tech.tablesaw.columns.AbstractColumnParser;
 import tech.tablesaw.columns.numbers.DoubleColumnType;
 import tech.tablesaw.columns.numbers.FloatColumnType;
+import tech.tablesaw.columns.numbers.NumberColumnFormatter;
 import tech.tablesaw.columns.numbers.NumberFillers;
 import tech.tablesaw.columns.numbers.fillers.DoubleRangeIterable;
 import tech.tablesaw.selection.Selection;
@@ -42,6 +43,7 @@ public class DoubleColumn extends NumberColumn<Double> implements NumberFillers<
 
     protected DoubleColumn(final String name, final DoubleArrayList data) {
         super(DoubleColumnType.instance(), name);
+        setPrintFormatter(NumberColumnFormatter.floatingPointDefault());
         this.data = data;
     }
 
@@ -51,7 +53,7 @@ public class DoubleColumn extends NumberColumn<Double> implements NumberFillers<
         if (DoubleColumnType.isMissingValue(value)) {
             return "";
         }
-        return String.valueOf(printFormatter.format(value));
+        return String.valueOf(getPrintFormatter().format(value));
     }
 
     @Override
@@ -71,6 +73,7 @@ public class DoubleColumn extends NumberColumn<Double> implements NumberFillers<
 
     protected DoubleColumn(final String name) {
         super(DoubleColumnType.instance(), name);
+        setPrintFormatter(NumberColumnFormatter.floatingPointDefault());
         this.data = new DoubleArrayList(DEFAULT_ARRAY_SIZE);
     }
 
diff --git a/core/src/main/java/tech/tablesaw/api/FloatColumn.java b/core/src/main/java/tech/tablesaw/api/FloatColumn.java
index 3f411555..750f2e42 100644
--- a/core/src/main/java/tech/tablesaw/api/FloatColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/FloatColumn.java
@@ -14,6 +14,7 @@ import it.unimi.dsi.fastutil.shorts.ShortArrayList;
 import tech.tablesaw.columns.Column;
 import tech.tablesaw.columns.AbstractColumnParser;
 import tech.tablesaw.columns.numbers.FloatColumnType;
+import tech.tablesaw.columns.numbers.NumberColumnFormatter;
 import tech.tablesaw.selection.Selection;
 
 import java.nio.ByteBuffer;
@@ -34,6 +35,7 @@ public class FloatColumn extends NumberColumn<Float> {
 
     private FloatColumn(final String name, FloatArrayList data) {
         super(FloatColumnType.instance(), name);
+        setPrintFormatter(NumberColumnFormatter.floatingPointDefault());
         this.data = data;
     }
 
@@ -43,7 +45,7 @@ public class FloatColumn extends NumberColumn<Float> {
         if (FloatColumnType.isMissingValue(value)) {
             return "";
         }
-        return String.valueOf(printFormatter.format(value));
+        return String.valueOf(getPrintFormatter().format(value));
     }
 
 
diff --git a/core/src/main/java/tech/tablesaw/api/IntColumn.java b/core/src/main/java/tech/tablesaw/api/IntColumn.java
index 02ad14eb..451fa68b 100644
--- a/core/src/main/java/tech/tablesaw/api/IntColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/IntColumn.java
@@ -36,7 +36,7 @@ public class IntColumn extends NumberColumn<Integer> implements CategoricalColum
 
     protected IntColumn(final String name, IntArrayList data) {
         super(IntColumnType.instance(), name);
-        this.printFormatter = NumberColumnFormatter.ints();
+        setPrintFormatter(NumberColumnFormatter.ints());
         this.data = data;
     }
 
@@ -272,7 +272,7 @@ public class IntColumn extends NumberColumn<Integer> implements CategoricalColum
         if (IntColumnType.isMissingValue(value)) {
             return "";
         }
-        return String.valueOf(printFormatter.format(value));
+        return String.valueOf(getPrintFormatter().format(value));
     }
 
     @Override
diff --git a/core/src/main/java/tech/tablesaw/api/LongColumn.java b/core/src/main/java/tech/tablesaw/api/LongColumn.java
index 8dfb6696..9a46ba07 100644
--- a/core/src/main/java/tech/tablesaw/api/LongColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/LongColumn.java
@@ -38,7 +38,7 @@ public class LongColumn extends NumberColumn<Long> implements CategoricalColumn<
 
     private LongColumn(final String name, LongArrayList data) {
         super(LongColumnType.instance(), name);
-        this.printFormatter = NumberColumnFormatter.ints();
+        setPrintFormatter(NumberColumnFormatter.ints());
         this.data = data;
     }
 
@@ -87,7 +87,7 @@ public class LongColumn extends NumberColumn<Long> implements CategoricalColumn<
         if (LongColumnType.isMissingValue(value)) {
             return "";
         }
-        return String.valueOf(printFormatter.format(value));
+        return String.valueOf(getPrintFormatter().format(value));
     }
 
     public static boolean valueIsMissing(long value) {
diff --git a/core/src/main/java/tech/tablesaw/api/NumberColumn.java b/core/src/main/java/tech/tablesaw/api/NumberColumn.java
index dea3550b..17e2ff61 100644
--- a/core/src/main/java/tech/tablesaw/api/NumberColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/NumberColumn.java
@@ -11,7 +11,7 @@ import java.util.function.DoublePredicate;
 
 public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> implements NumericColumn<T> {
 
-    protected NumberColumnFormatter printFormatter = new NumberColumnFormatter();
+    private NumberColumnFormatter printFormatter = new NumberColumnFormatter();
 
     protected Locale locale;
 
@@ -59,6 +59,10 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
         this.printFormatter = formatter;
     }
 
+    protected NumberColumnFormatter getPrintFormatter() {
+        return printFormatter;
+    }
+
     /**
      * Returns the largest ("top") n values in the column
      * TODO(lwhite): Consider whether this should exclude missing
diff --git a/core/src/main/java/tech/tablesaw/api/ShortColumn.java b/core/src/main/java/tech/tablesaw/api/ShortColumn.java
index 0516f6a3..49716b10 100644
--- a/core/src/main/java/tech/tablesaw/api/ShortColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/ShortColumn.java
@@ -35,7 +35,7 @@ public class ShortColumn extends NumberColumn<Short> implements CategoricalColum
 
     protected ShortColumn(final String name, ShortArrayList data) {
         super(ShortColumnType.instance(), name);
-        this.printFormatter = NumberColumnFormatter.ints();
+        setPrintFormatter(NumberColumnFormatter.ints());
         this.data = data;
     }
 
@@ -241,7 +241,7 @@ public class ShortColumn extends NumberColumn<Short> implements CategoricalColum
         if (ShortColumnType.isMissingValue(value)) {
             return "";
         }
-        return String.valueOf(printFormatter.format(value));
+        return String.valueOf(getPrintFormatter().format(value));
     }
 
     @Override
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/NumberColumnFormatter.java b/core/src/main/java/tech/tablesaw/columns/numbers/NumberColumnFormatter.java
index a914c830..9241863d 100644
--- a/core/src/main/java/tech/tablesaw/columns/numbers/NumberColumnFormatter.java
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/NumberColumnFormatter.java
@@ -1,6 +1,7 @@
 package tech.tablesaw.columns.numbers;
 
 import java.text.DecimalFormat;
+import java.text.DecimalFormatSymbols;
 import java.text.NumberFormat;
 import java.util.Locale;
 
@@ -17,6 +18,24 @@ public class NumberColumnFormatter {
         return new NumberColumnFormatter(format);
     }
 
+    /**
+     * Returns a formatter that prints floating point numbers with all precision
+     */
+    public static NumberColumnFormatter floatingPointDefault() {
+        NumberFormat format = new DecimalFormat("0", DecimalFormatSymbols.getInstance(Locale.getDefault()));
+        format.setMaximumFractionDigits(340);
+        format.setMaximumIntegerDigits(340);
+        format.setGroupingUsed(false);
+        return new NumberColumnFormatter(format);
+    }
+
+    /**
+     * Formats numbers using java default, so sometimes in scientific notation, sometimes not
+     */
+    public static NumberColumnFormatter standard() {
+        return new NumberColumnFormatter();
+    }
+
     public static NumberColumnFormatter ints() {
         NumberFormat format = new DecimalFormat();
         format.setGroupingUsed(false);
diff --git a/core/src/main/java/tech/tablesaw/io/jdbc/SqlResultSetReader.java b/core/src/main/java/tech/tablesaw/io/jdbc/SqlResultSetReader.java
index 2e984b79..a40e048f 100644
--- a/core/src/main/java/tech/tablesaw/io/jdbc/SqlResultSetReader.java
+++ b/core/src/main/java/tech/tablesaw/io/jdbc/SqlResultSetReader.java
@@ -16,6 +16,8 @@ package tech.tablesaw.io.jdbc;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
+
+import tech.tablesaw.api.BooleanColumn;
 import tech.tablesaw.api.ColumnType;
 import tech.tablesaw.api.DoubleColumn;
 import tech.tablesaw.api.FloatColumn;
@@ -108,15 +110,17 @@ public class SqlResultSetReader {
             for (int i = 1; i <= metaData.getColumnCount(); i++) {
                 Column<?> column = table.column(i - 1); // subtract 1 because results sets originate at 1 not 0
                 if (column instanceof ShortColumn) {
-                    column.appendObj(resultSet.getShort(i));
+                	appendToColumn(column, resultSet, resultSet.getShort(i));
                 } else if (column instanceof IntColumn) {
-                    column.appendObj(resultSet.getInt(i));
+                	appendToColumn(column, resultSet, resultSet.getInt(i));
                 } else if (column instanceof LongColumn) {
-                    column.appendObj(resultSet.getLong(i));
+                	appendToColumn(column, resultSet, resultSet.getLong(i));
                 } else if (column instanceof FloatColumn) {
-                    column.appendObj(resultSet.getFloat(i));
+                	appendToColumn(column, resultSet, resultSet.getFloat(i));
                 } else if (column instanceof DoubleColumn) {
-                    column.appendObj(resultSet.getDouble(i));
+                	appendToColumn(column, resultSet, resultSet.getDouble(i));
+                } else if (column instanceof BooleanColumn) {
+                	appendToColumn(column, resultSet, resultSet.getBoolean(i));                
                 } else {
                     column.appendObj(resultSet.getObject(i));
                 }
@@ -125,6 +129,14 @@ public class SqlResultSetReader {
         return table;
     }
     
+    protected static void appendToColumn(Column<?> column, ResultSet resultSet, Object value) throws SQLException {
+    	if (resultSet.wasNull()) {
+    		column.appendMissing();
+    	} else {
+    		column.appendObj(value);
+    	}
+    }
+    
     protected static ColumnType getColumnType(int columnType, int scale, int precision) {
         ColumnType type = SQL_TYPE_TO_TABLESAW_TYPE.get(columnType);
         // Try to improve on the initial type assigned to 'type' to minimize size/space of type needed.
diff --git a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
index bf059ef9..1b9329e8 100644
--- a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
+++ b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
@@ -1,6 +1,13 @@
 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.BooleanColumn;
 import tech.tablesaw.api.ColumnType;
 import tech.tablesaw.api.DateColumn;
@@ -36,37 +43,44 @@ import tech.tablesaw.index.IntIndex;
 import tech.tablesaw.index.LongIndex;
 import tech.tablesaw.index.ShortIndex;
 import tech.tablesaw.index.StringIndex;
-import tech.tablesaw.selection.BitmapBackedSelection;
 import tech.tablesaw.selection.Selection;
 
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
-
 public class DataFrameJoiner {
 
+    private enum JoinType {
+        INNER,
+        LEFT_OUTER,
+        RIGHT_OUTER,
+        FULL_OUTER
+    }
+
     private static final String TABLE_ALIAS = "T";
 
     private final Table table;
-    private Column<?>[] joinColumns;
-    private final String[] columnNames;
-    private AtomicInteger joinTableId = new AtomicInteger(2);
+    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 columnNames The column names to join on
+     * @param table The table to join on.
+     * @param joinColumnNames The join column names to join on.
      */
-    public DataFrameJoiner(Table table, String... columnNames) {
+    public DataFrameJoiner(Table table, String... joinColumnNames) {
         this.table = table;
-        joinColumns = new Column<?>[columnNames.length];
-        this.columnNames = columnNames;
-        for (int i = 0; i < this.columnNames.length; i++) {
-            String colName = this.columnNames[i];
-            this.joinColumns[i] = table.column(colName);
-        }
+        this.joinColumnNames = 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());
     }
 
     /**
@@ -81,15 +95,15 @@ public class DataFrameJoiner {
     /**
      * 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 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, false, allowDuplicateColumnNames, columnNames);
+            joined = joinInternal(joined, currT, JoinType.INNER, allowDuplicateColumnNames, joinColumnNames);
         }
         return joined;
     }
@@ -97,21 +111,21 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
-     public Table inner(Table table2, String col2Name) {
-         return inner(table2, false, col2Name);
-     }
+    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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table inner(Table table2, String[] col2Names) {
@@ -121,11 +135,11 @@ public class DataFrameJoiner {
     /**
      * 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 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*
+     * 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) {
@@ -135,205 +149,227 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table inner(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
-        return inner(table2, false, allowDuplicateColumnNames, col2Names);
+        Table joinedTable;
+        joinedTable = joinInternal(table, table2, JoinType.INNER, allowDuplicateColumnNames, col2Names);
+        return joinedTable;
     }
 
     /**
      * Joins the joiner to the table2, using the given columns for the second table and returns the resulting table
      *
-     * @param table2    The table to join with
-     * @param outer     True if this join is actually an outer join, left or right or full, otherwise false.
-     * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column have the same name
-     *                                  if {@code true} the join will succeed and duplicate columns are renamed*
+     * @param table2 The table to join with
+     * @param outer True if this join is actually an outer join, left or right or full, otherwise false.
+     * @param allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column
+     * have the same name if {@code true} the join will succeed and duplicate columns are renamed*
      * @param col2Names The columns to join on. If a name refers to a double column, the join is performed after
-     *                  rounding to integers.
+     * rounding to integers.
      * @return The resulting table
      */
+    @Deprecated
     public Table inner(Table table2, boolean outer, boolean allowDuplicateColumnNames, String... col2Names) {
+        JoinType joinType = JoinType.INNER;
+        if (outer) {
+            joinType = JoinType.LEFT_OUTER;
+        }
+
         Table joinedTable;
-        joinedTable = joinInternal(table, table2, outer, allowDuplicateColumnNames, col2Names);
+        joinedTable = joinInternal(table, table2, joinType, allowDuplicateColumnNames, col2Names);
         return joinedTable;
     }
 
     /**
-     * Joins the joiner to the {@code table2}, using the given columns for the second table and returns the resulting table
+     * Joins two tables.
      *
-     * @param table1    The base table to join on
-     * @param table2    The table to join with
-     * @param outer     True if this join is actually an outer join, left or right or full, otherwise false.
-     * @param 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 col2Names The columns to join on. If a col2Name refers to a double column, the join is performed after
-     *                  rounding to integers.
-     * @return The resulting table
+     * @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 table2JoinColumnNames The names of the columns in table2 to join on.
      */
-    private Table joinInternal(Table table1, Table table2, boolean outer, boolean allowDuplicates, String... col2Names) {
+    private Table joinInternal(Table table1, Table table2, JoinType joinType, boolean allowDuplicates, String... table2JoinColumnNames) {
+        List<Integer> table2JoinColumnIndexes = getJoinIndexes(table2, table2JoinColumnNames);
 
-        if (allowDuplicates) {
-            renameColumnsWithDuplicateNames(table1, table2, col2Names);
-        }
-        Table result = emptyTableFromColumns(table1, table2, col2Names);
-        Map<Column<?>, Index> columnIndexMap = new HashMap<>();
+        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);
 
-        for (int i = 0; i < joinColumns.length; i++) {
-            Column<?> col = joinColumns[i];
-            String col2Name = col2Names[i];
-            columnIndexMap.put(col, indexFor(table2, col2Name, col));
-        }
+        List<Index> table1Indexes = buildIndexesForJoinColumns(joinColumnIndexes, table1);
+        List<Index> table2Indexes = buildIndexesForJoinColumns(table2JoinColumnIndexes, table2);
 
+        Selection table1DoneRows = Selection.with();
+        Selection table2DoneRows = Selection.with();
         for (Row row : table1) {
             int ri = row.getRowNumber();
-            Table table1Rows = table1.where(Selection.with(ri));
-            Selection rowBitMapMultiCol = null;
-            for (int i = 0; i < joinColumns.length; i++) {
-
-                // Need to use the column from table1 that is the same column originally
-                // defined for this DataFrameJoiner. Column names must be unique within the
-                // same table, so use the original column's name to get the corresponding
-                // column out of the table1 input Table.
-                Column<?> column = joinColumns[i];
-                Column<?> table1Column = table1.column(column.name());
-
-                ColumnType type = table1Column.type();
-                // relies on both arrays, columns, and col2Names,
-                // having corresponding values at same index
-                Selection rowBitMapOneCol = null;
-                if (type instanceof DateColumnType) {
-                    IntIndex index = (IntIndex) columnIndexMap.get(column);
-                    DateColumn col1 = (DateColumn) table1Column;
-                    int value = col1.getIntInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof TimeColumnType) {
-                    IntIndex index = (IntIndex) columnIndexMap.get(column);
-                    TimeColumn col1 = (TimeColumn) table1Column;
-                    int value = col1.getIntInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof DateTimeColumnType) {
-                    LongIndex index = (LongIndex) columnIndexMap.get(column);
-                    DateTimeColumn col1 = (DateTimeColumn) table1Column;
-                    long value = col1.getLongInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof InstantColumnType) {
-                    LongIndex index = (LongIndex) columnIndexMap.get(column);
-                    InstantColumn col1 = (InstantColumn) table1Column;
-                    long value = col1.getLongInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof StringColumnType || type instanceof TextColumnType) {
-                    StringIndex index = (StringIndex) columnIndexMap.get(column);
-                    StringColumn col1 = (StringColumn) table1Column;
-                    String value = col1.get(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof IntColumnType) {
-                    IntIndex index = (IntIndex) columnIndexMap.get(column);
-                    IntColumn col1 = (IntColumn) table1Column;
-                    int value = col1.getInt(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof LongColumnType) {
-                    LongIndex index = (LongIndex) columnIndexMap.get(column);
-                    LongColumn col1 = (LongColumn) table1Column;
-                    long value = col1.getLong(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof ShortColumnType) {
-                    ShortIndex index = (ShortIndex) columnIndexMap.get(column);
-                    ShortColumn col1 = (ShortColumn) table1Column;
-                    short value = col1.getShort(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof BooleanColumnType) {
-                    ByteIndex index = (ByteIndex) columnIndexMap.get(column);
-                    BooleanColumn col1 = (BooleanColumn) table1Column;
-                    byte value = col1.getByte(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof DoubleColumnType) {
-                    DoubleIndex index = (DoubleIndex) columnIndexMap.get(column);
-                    DoubleColumn col1 = (DoubleColumn) table1Column;
-                    double value = col1.getDouble(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof FloatColumnType) {
-                    FloatIndex index = (FloatIndex) columnIndexMap.get(column);
-                    FloatColumn col1 = (FloatColumn) table1Column;
-                    float value = col1.getFloat(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else {
-                    throw new IllegalArgumentException(
-                            "Joining is supported on numeric, string, and date-like columns. Column "
-                                    + table1Column.name() + " is of type " + table1Column.type());
-                }
-                // combine Selection's into one big AND Selection
-                if (rowBitMapOneCol != null) {
-                    rowBitMapMultiCol =
-                            rowBitMapMultiCol != null
-                            ? rowBitMapMultiCol.and(rowBitMapOneCol)
-                            : rowBitMapOneCol;
-                }
+            if (table1DoneRows.contains(ri)) {
+                // Already processed a selection of table1 that contained this row.
+                continue;
             }
 
-            Table table2Rows = table2.where(rowBitMapMultiCol);
-            if (outer && table2Rows.isEmpty()) {
-                withMissingLeftJoin(result, table1Rows);
+            Selection table1Rows = createMultiColSelection(table1, ri, table1Indexes, table1.rowCount());
+            Selection table2Rows = createMultiColSelection(table1, ri, table2Indexes, table2.rowCount());
+
+            if ((joinType == JoinType.LEFT_OUTER || joinType == JoinType.FULL_OUTER) && table2Rows.isEmpty()) {
+                withMissingLeftJoin(result, table1, table1Rows, resultIgnoreColIndexes);
             } else {
-                crossProduct(result, table1Rows, table2Rows, col2Names);
+                crossProduct(result, table1, table2, table1Rows, table2Rows, resultIgnoreColIndexes);
+            }
+
+            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.
+                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);
+        result.removeColumns(Ints.toArray(resultIgnoreColIndexes));
         return result;
     }
 
-    private Index indexFor(Table table2, String col2Name, Column<?> col) {
-        ColumnType type = col.type();
+    /**
+     * 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(table2.dateColumn(col2Name));
-        }
-        if (type instanceof DateTimeColumnType) {
-            return new LongIndex(table2.dateTimeColumn(col2Name));
-        }
-        if (type instanceof InstantColumnType) {
-            return new LongIndex(table2.instantColumn(col2Name));
-        }
-        if (type instanceof TimeColumnType) {
-            return new IntIndex(table2.timeColumn(col2Name));
-        }
-        if (type instanceof StringColumnType || type instanceof TextColumnType) {
-            return new StringIndex(table2.stringColumn(col2Name));
-        }
-        if (type instanceof IntColumnType) {
-            return new IntIndex(table2.intColumn(col2Name));
-        }
-        if (type instanceof LongColumnType) {
-            return new LongIndex(table2.longColumn(col2Name));
-        }
-        if (type instanceof ShortColumnType) {
-            return new ShortIndex(table2.shortColumn(col2Name));
-        }
-        if (type instanceof BooleanColumnType) {
-            return new ByteIndex(table2.booleanColumn(col2Name));
-        }
-        if (type instanceof DoubleColumnType) {
-            return new DoubleIndex(table2.doubleColumn(col2Name));
-        }
-        if (type instanceof FloatColumnType) {
-            return new FloatIndex(table2.floatColumn(col2Name));
+            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 || type instanceof TextColumnType) {
+            return new StringIndex(table.stringColumn(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 " + col.type());
+            "Joining attempted on unsupported column type " + type);
     }
 
-    private void renameColumnsWithDuplicateNames(Table table1, Table table2, String... col2Names) {
-        String table2Alias = TABLE_ALIAS + joinTableId.getAndIncrement();
-        List<String> list = Arrays.asList(col2Names);
-        for (Column<?> table2Column : table2.columns()) {
-            String columnName = table2Column.name();
-            if (table1.columnNames().stream().anyMatch(columnName::equalsIgnoreCase)
-                    && !(list.stream().anyMatch(columnName::equalsIgnoreCase))) {
-                table2Column.setName(newName(table2Alias, columnName));
-            }
+    /**
+    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();
+        Selection selection = Selection.with();
+        if (type instanceof DateColumnType) {
+            IntIndex index = (IntIndex) rawIndex;
+            DateColumn typedValueColumn = (DateColumn) valueColumn;
+            int value = typedValueColumn.getIntInternal(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof TimeColumnType) {
+            IntIndex index = (IntIndex) rawIndex;
+            TimeColumn typedValueColumn = (TimeColumn) valueColumn;
+            int value = typedValueColumn.getIntInternal(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof DateTimeColumnType) {
+            LongIndex index = (LongIndex) rawIndex;
+            DateTimeColumn typedValueColumn = (DateTimeColumn) valueColumn;
+            long value = typedValueColumn.getLongInternal(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof InstantColumnType) {
+            LongIndex index = (LongIndex) rawIndex;
+            InstantColumn typedValueColumn = (InstantColumn) valueColumn;
+            long value = typedValueColumn.getLongInternal(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof StringColumnType || type instanceof TextColumnType) {
+            StringIndex index = (StringIndex) rawIndex;
+            StringColumn typedValueColumn = (StringColumn) valueColumn;
+            String value = typedValueColumn.get(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof IntColumnType) {
+            IntIndex index = (IntIndex) rawIndex;
+            IntColumn typedValueColumn = (IntColumn) valueColumn;
+            int value = typedValueColumn.getInt(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof LongColumnType) {
+            LongIndex index = (LongIndex) rawIndex;
+            LongColumn typedValueColumn = (LongColumn) valueColumn;
+            long value = typedValueColumn.getLong(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof ShortColumnType) {
+            ShortIndex index = (ShortIndex) rawIndex;
+            ShortColumn typedValueColumn = (ShortColumn) valueColumn;
+            short value = typedValueColumn.getShort(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof BooleanColumnType) {
+            ByteIndex index = (ByteIndex) rawIndex;
+            BooleanColumn typedValueColumn = (BooleanColumn) valueColumn;
+            byte value = typedValueColumn.getByte(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof DoubleColumnType) {
+            DoubleIndex index = (DoubleIndex) rawIndex;
+            DoubleColumn typedValueColumn = (DoubleColumn) valueColumn;
+            double value = typedValueColumn.getDouble(rowIndex);
+            selection = index.get(value);
+        } else if (type instanceof FloatColumnType) {
+            FloatIndex index = (FloatIndex) rawIndex;
+            FloatColumn typedValueColumn = (FloatColumn) valueColumn;
+            float value = typedValueColumn.getFloat(rowIndex);
+            selection = 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());
+        }
+        return selection;
+    }
+
+    /**
+    Create a big multicolumn selection for all join columns in the given table.
+     */
+    private Selection createMultiColSelection(Table table1, int ri, List<Index> indexes, int selectionSize) {
+        Selection multiColSelection = Selection.withRange(0, selectionSize);
+        int i = 0;
+        for (Integer joinColumnIndex : joinColumnIndexes) {
+            Column<?> col = table1.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) {
@@ -353,142 +389,31 @@ public class DataFrameJoiner {
     /**
      * 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 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;
+        Table joined = table;
 
         for (Table currT : tables) {
-            joined = fullOuter(joined, currT, allowDuplicateColumnNames, columnNames);
+            joined = joinInternal(joined, currT, JoinType.FULL_OUTER, allowDuplicateColumnNames, joinColumnNames);
         }
         return joined;
     }
 
     /**
-     * Full outer join the joiner to the table2, using the given column for the second table and returns the resulting table
+     * Full outer join the joiner to the table2, using the given column for the second table and returns the resulting
+     * table
      *
-     * @param table2   The table to join with
+     * @param 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table fullOuter(Table table2, String col2Name) {
-        return fullOuter(table, table2, false, col2Name);
-    }
-
-    /**
-     * Full outer join table1 to table2, using the given columns for the second table and returns the resulting table
-     *
-     * @param table1    The base table to join on
-     * @param table2    The table to join with
-     * @param 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 fullOuter(Table table1, Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
-        Table result = joinInternal(table1, table2, true, allowDuplicateColumnNames, col2Names);
-
-        Selection selection = new BitmapBackedSelection();
-        for (Row row : table2) {
-            int ri = row.getRowNumber();
-            Selection rowBitMapMultiCol = null;
-
-            for (int i = 0; i < joinColumns.length; i++) {
-
-                // Need to use the column from table1 that is the same column originally
-                // defined for this DataFrameJoiner. Column names must be unique within the
-                // same table, so use the original column's name to get the corresponding
-                // column out of the table1 input Table.
-                Column<?> column = joinColumns[i];
-                Column<?> table1Column = table1.column(column.name());
-
-                ColumnType type = table1Column.type();
-                // relies on both arrays, columns, and col2Names,
-                // having corresponding values at same index
-                String col2Name = col2Names[i];
-                Selection rowBitMapOneCol = null;
-
-                if (type instanceof DateColumnType) {
-                    IntIndex index = new IntIndex(result.dateColumn(col2Name));
-                    DateColumn col2 = (DateColumn) table2.column(col2Name);
-                    int value = col2.getIntInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof DateTimeColumnType) {
-                    LongIndex index = new LongIndex(result.dateTimeColumn(col2Name));
-                    DateTimeColumn col2 = (DateTimeColumn) table2.column(col2Name);
-                    long value = col2.getLongInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof InstantColumnType) {
-                    LongIndex index = new LongIndex(result.instantColumn(col2Name));
-                    InstantColumn col2 = (InstantColumn) table2.column(col2Name);
-                    long value = col2.getLongInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof TimeColumnType) {
-                    IntIndex index = new IntIndex(result.timeColumn(col2Name));
-                    TimeColumn col2 = (TimeColumn) table2.column(col2Name);
-                    int value = col2.getIntInternal(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof StringColumnType || type instanceof TextColumnType) {
-                    StringIndex index = new StringIndex(result.stringColumn(col2Name));
-                    StringColumn col2 = (StringColumn) table2.column(col2Name);
-                    String value = col2.get(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof IntColumnType) {
-                    IntIndex index = new IntIndex(result.intColumn(col2Name));
-                    IntColumn col2 = (IntColumn) table2.column(col2Name);
-                    int value = col2.getInt(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof LongColumnType) {
-                    LongIndex index = new LongIndex(result.longColumn(col2Name));
-                    LongColumn col2 = (LongColumn) table2.column(col2Name);
-                    long value = col2.getLong(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof ShortColumnType) {
-                    ShortIndex index = new ShortIndex(result.shortColumn(col2Name));
-                    ShortColumn col2 = (ShortColumn) table2.column(col2Name);
-                    short value = col2.getShort(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof BooleanColumnType) {
-                    ByteIndex index = new ByteIndex(result.booleanColumn(col2Name));
-                    BooleanColumn col2 = (BooleanColumn) table2.column(col2Name);
-                    byte value = col2.getByte(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof DoubleColumnType) {
-                    DoubleIndex index = new DoubleIndex(result.doubleColumn(col2Name));
-                    DoubleColumn col2 = (DoubleColumn) table2.column(col2Name);
-                    double value = col2.getDouble(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else if (type instanceof FloatColumnType) {
-                    FloatIndex index = new FloatIndex(result.floatColumn(col2Name));
-                    FloatColumn col2 = (FloatColumn) table2.column(col2Name);
-                    float value = col2.getFloat(ri);
-                    rowBitMapOneCol = index.get(value);
-                } else {
-                    throw new IllegalArgumentException(
-                            "Joining is supported on numeric, string, and date-like columns. Column "
-                                    + table1Column.name() + " is of type " + table1Column.type());
-                }
-                // combine Selections into one big AND Selection
-                if (rowBitMapOneCol != null) {
-                    rowBitMapMultiCol = rowBitMapMultiCol != null
-                            ? rowBitMapMultiCol.and(rowBitMapOneCol)
-                            : rowBitMapOneCol;
-                }
-            }
-            if (rowBitMapMultiCol.isEmpty()) {
-                selection.add(ri);
-            }
-        }
-        Table table2OnlyRows = table2.where(selection);
-        List<Column<?>> joinColumns = table2OnlyRows.columns(col2Names);
-        table2OnlyRows.removeColumns(col2Names);
-        withMissingRightJoin(result, joinColumns, table2OnlyRows);
-        return result;
+        return joinInternal(table, table2, JoinType.FULL_OUTER, false, col2Name);
     }
 
     /**
@@ -504,15 +429,15 @@ public class DataFrameJoiner {
     /**
      * 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 allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column
+     * have the same name if {@code true} the join will succeed and duplicate columns are renamed*
      * @param tables The tables to join with
      * @return The resulting table
      */
     public Table leftOuter(boolean allowDuplicateColumnNames, Table... tables) {
         Table joined = table;
         for (Table table2 : tables) {
-            joined = leftOuter(table2, allowDuplicateColumnNames, columnNames);
+            joined = leftOuter(table2, allowDuplicateColumnNames, joinColumnNames);
         }
         return joined;
     }
@@ -520,9 +445,9 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table leftOuter(Table table2, String[] col2Names) {
@@ -532,9 +457,9 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table leftOuter(Table table2, String col2Name) {
@@ -544,15 +469,15 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table leftOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
-        return joinInternal(table, table2, true, allowDuplicateColumnNames, col2Names);
+        return joinInternal(table, table2, JoinType.LEFT_OUTER, allowDuplicateColumnNames, col2Names);
     }
 
     /**
@@ -568,15 +493,15 @@ public class DataFrameJoiner {
     /**
      * 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 allowDuplicateColumnNames if {@code false} the join will fail if any columns other than the join column
+     * have the same name if {@code true} the join will succeed and duplicate columns are renamed
      * @param tables The tables to join with
      * @return The resulting table
      */
     public Table rightOuter(boolean allowDuplicateColumnNames, Table... tables) {
         Table joined = table;
         for (Table table2 : tables) {
-            joined = rightOuter(table2, allowDuplicateColumnNames, columnNames);
+            joined = rightOuter(table2, allowDuplicateColumnNames, joinColumnNames);
         }
         return joined;
     }
@@ -584,9 +509,9 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table rightOuter(Table table2, String col2Name) {
@@ -596,9 +521,9 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
     public Table rightOuter(Table table2, String[] col2Names) {
@@ -608,58 +533,100 @@ public class DataFrameJoiner {
     /**
      * 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 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.
+     * rounding to integers.
      * @return The resulting table
      */
-
     public Table rightOuter(Table table2, boolean allowDuplicateColumnNames, String... col2Names) {
-        Table leftOuter = table2.joinOn(col2Names).leftOuter(table, allowDuplicateColumnNames, columnNames);
-
-        // reverse the columns
-        Table result = Table.create(leftOuter.name());
-        // loop on table that was originally first (left) and add the left-joined matching columns by name
-        for (String name : table.columnNames()) {
-            try {
-                result.addColumns(leftOuter.column(name));
-            } catch (IllegalStateException e) {
-                // Can ignore this exception as it is anticipated.
-                // NOTE: DataFrameJoiner.rightOuter(): skipping left table's column,'"
-                //     +name+"', in favor of right table's matching column that was kept in join operation.");
+        return joinInternal(table, table2, JoinType.RIGHT_OUTER, allowDuplicateColumnNames, col2Names);
+    }
+
+    /**
+     * Adds empty columns to the destination table with the same type as columns in table1 and table2.
+     *
+     * 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) {
+
+        Column<?>[] cols = Streams.concat(table1.columns().stream(), table2.columns().stream())
+            .map(Column::emptyCopy).toArray(Column[]::new);
+
+        // 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)) {
+                    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);
+                }
             }
         }
-        for (String name : table2.columnNames()) {
-            if (!result.columnNames().stream().anyMatch(name::equalsIgnoreCase)) {
-                result.addColumns(leftOuter.column(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));
+                }
             }
         }
-        return result;
-    }
-
-    private Table emptyTableFromColumns(Table table1, Table table2, String... col2Names) {
-        Column<?>[] cols = Streams.concat(
-                table1.columns().stream(),
-                table2.columns().stream().filter(c -> !Arrays.asList(col2Names).stream().anyMatch(c.name()::equalsIgnoreCase))
-                ).map(Column::emptyCopy).toArray(Column[]::new);
-        return Table.create(table1.name(), cols);
+        destination.addColumns(cols);
+        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, String[] col2Names) {
-        int table2RowCount = table2.rowCount();
-        // Remove join columns
-        table2.removeColumns(col2Names);
+    private void crossProduct(Table destination, Table table1, Table table2,
+        Selection table1Rows, Selection table2Rows, Set<Integer> ignoreColumns) {
         for (int c = 0; c < table1.columnCount() + table2.columnCount(); c++) {
-            for (int r1 = 0; r1 < table1.rowCount(); r1++) {
-                for (int r2 = 0; r2 < table2RowCount; r2++) {
+            if (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(c - table1.columnCount());
+                        Column t2Col = table2.column(table2Index);
                         destination.column(c).append(t2Col, r2);
                     }
                 }
@@ -668,16 +635,22 @@ public class DataFrameJoiner {
     }
 
     /**
-     * Adds rows to destination for each row in table1, with the columns from table2 added as missing values in each
+     * 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) {
+    private void withMissingLeftJoin(Table destination, Table table1, Selection table1Rows,
+        Set<Integer> ignoreColumns) {
         for (int c = 0; c < destination.columnCount(); c++) {
+            if (ignoreColumns.contains(c)) {
+                continue;
+            }
             if (c < table1.columnCount()) {
                 Column t1Col = table1.column(c);
-                destination.column(c).append(t1Col);
+                for (int index : table1Rows) {
+                    destination.column(c).append(t1Col, index);
+                }
             } else {
-                for (int r1 = 0; r1 < table1.rowCount(); r1++) {
+                for (int r1 = 0; r1 < table1Rows.size(); r1++) {
                     destination.column(c).appendMissing();
                 }
             }
@@ -685,27 +658,34 @@ public class DataFrameJoiner {
     }
 
     /**
-     * Adds rows to destination for each row in the joinColumn and table2
+     * Adds rows to destination for each row in table2 with the columns from table1 added as missing values.
      */
     @SuppressWarnings({"rawtypes", "unchecked"})
-    private void withMissingRightJoin(Table destination, List<Column<?>> joinColumns, Table table2) {
-        int t2StartCol = destination.columnCount() - table2.columnCount();
-        for (int c = 0; c < destination.columnCount(); c++) {
-            boolean addedJoinColumns = false;
-            for (Column joinColumn : joinColumns) {
-                if (destination.column(c).name().equalsIgnoreCase(joinColumn.name())) {
-                    destination.column(c).append(joinColumn);
-                    addedJoinColumns = true;
+    private void withMissingRight(Table destination, int table1ColCount, Table table2, Selection table2Rows,
+        JoinType joinType, List<Integer> col2Indexes, Set<Integer> ignoreColumns) {
+
+        // 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);
                 }
             }
-            if (!addedJoinColumns) {
-                if (c < t2StartCol) {
-                    for (int r2 = 0; r2 < table2.rowCount(); r2++) {
-                        destination.column(c).appendMissing();
-                    }
-                } else {
-                    Column t2Col = table2.column(c - t2StartCol);
-                    destination.column(c).append(t2Col);
+        }
+
+        for (int c = 0; c < destination.columnCount(); c++) {
+            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);
                 }
             }
         }
diff --git a/core/src/main/java/tech/tablesaw/table/TableSliceGroup.java b/core/src/main/java/tech/tablesaw/table/TableSliceGroup.java
index 537e0e23..b6f45620 100644
--- a/core/src/main/java/tech/tablesaw/table/TableSliceGroup.java
+++ b/core/src/main/java/tech/tablesaw/table/TableSliceGroup.java
@@ -36,10 +36,12 @@ import java.util.Map;
  */
 public class TableSliceGroup implements Iterable<TableSlice> {
 
-    // A string that is used internally as a delimiter in creating a column name from all the grouping columns
+    // A string that is used internally as a delimiter in creating a column name from all the grouping
+    // columns
     protected static final String SPLIT_STRING = "~~~";
 
-    // A function that splits the group column name back into the original column names for the grouping columns
+    // A function that splits the group column name back into the original column names for the
+    // grouping columns
     private static final Splitter SPLITTER = Splitter.on(SPLIT_STRING);
 
     // The list of slices or views over the source table that I contain
@@ -54,16 +56,55 @@ public class TableSliceGroup implements Iterable<TableSlice> {
      * Returns an instance for calculating a single summary for the given table, with no sub-groupings
      */
     protected TableSliceGroup(Table original) {
-        sourceTable = original;
+        if (containsAnyTextColumns(original)) {
+            sourceTable = original.copy();
+            replaceTextColumnsWithStringColumns();
+        } else {
+            sourceTable = original;
+        }
         splitColumnNames = new String[0];
     }
 
+    private boolean containsAnyTextColumns(Table original) {
+        for (Column<?> column : original.columns()) {
+            if (column.type().equals(ColumnType.TEXT)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * Replace any textColumns in the table with stringColumns. We do this because TextColumns don't
+     * split correctly: The split algorithm uses a byte[] version of the elements to do it's magic,
+     * and text columns have variable sized strings, so variable sized byte arrays. Determining the
+     * correct array size (maybe the largest array size for the array?) would be somewhat fraught
+     * because the size depends on the encoding and the strings do not know they're own encoding. This
+     * would need to be detected using a 3rd party library.
+     *
+     * <p>So replace with the equivalent stringColumn instead.
+     */
+    private void replaceTextColumnsWithStringColumns() {
+        for (int i = 0; i < sourceTable.columnCount(); i++) {
+            if (sourceTable.column(i).type().equals(ColumnType.TEXT)) {
+                String originalName = sourceTable.column(i).name();
+                sourceTable.replaceColumn(i, sourceTable.textColumn(i).asStringColumn());
+                sourceTable.column(i).setName(originalName);
+            }
+        }
+    }
+
     /**
-     * Returns an instance for calculating subgroups,
-     * one for each combination of the given groupColumnNames that appear in the source table
+     * Returns an instance for calculating subgroups, one for each combination of the given
+     * groupColumnNames that appear in the source table
      */
     protected TableSliceGroup(Table sourceTable, String[] groupColumnNames) {
-        this.sourceTable = sourceTable;
+        if (containsAnyTextColumns(sourceTable)) {
+            this.sourceTable = sourceTable.copy();
+            replaceTextColumnsWithStringColumns();
+        } else {
+            this.sourceTable = sourceTable;
+        }
         this.splitColumnNames = groupColumnNames;
     }
 
@@ -103,8 +144,8 @@ public class TableSliceGroup implements Iterable<TableSlice> {
     }
 
     /**
-     * For a subtable that is grouped by the values in more than one column, split the grouping column into separate
-     * cols and return the revised view
+     * For a subtable that is grouped by the values in more than one column, split the grouping column
+     * into separate cols and return the revised view
      */
     private Table splitGroupingColumn(Table groupTable) {
 
@@ -115,7 +156,8 @@ public class TableSliceGroup implements Iterable<TableSlice> {
                 Column<?> newColumn = column.emptyCopy();
                 newColumns.add(newColumn);
             }
-            // iterate through the rows in the table and split each of the grouping columns into multiple columns
+            // iterate through the rows in the table and split each of the grouping columns into multiple
+            // columns
             for (int row = 0; row < groupTable.rowCount(); row++) {
                 List<String> strings = SPLITTER.splitToList(groupTable.stringColumn("Group").get(row));
                 for (int col = 0; col < newColumns.size(); col++) {
@@ -132,29 +174,31 @@ public class TableSliceGroup implements Iterable<TableSlice> {
     }
 
     /**
-     * Applies the given aggregation to the given column.
-     * The apply and combine steps of a split-apply-combine.
+     * Applies the given aggregation to the given column. The apply and combine steps of a
+     * split-apply-combine.
      */
-    public Table aggregate(String colName1, AggregateFunction<?,?>... functions) {
-        ArrayListMultimap<String, AggregateFunction<?,?>> columnFunctionMap = ArrayListMultimap.create();
+    public Table aggregate(String colName1, AggregateFunction<?, ?>... functions) {
+        ArrayListMultimap<String, AggregateFunction<?, ?>> columnFunctionMap =
+                ArrayListMultimap.create();
         columnFunctionMap.putAll(colName1, Lists.newArrayList(functions));
         return aggregate(columnFunctionMap);
     }
 
     /**
-     * Applies the given aggregations to the given columns.
-     * The apply and combine steps of a split-apply-combine.
+     * Applies the given aggregations to the given columns. The apply and combine steps of a
+     * split-apply-combine.
      *
      * @param functions map from column name to aggregation to apply on that function
      */
-    @SuppressWarnings({ "unchecked", "rawtypes" })
-    public Table aggregate(ListMultimap<String, AggregateFunction<?,?>> functions) {
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    public Table aggregate(ListMultimap<String, AggregateFunction<?, ?>> functions) {
         Preconditions.checkArgument(!getSlices().isEmpty());
         Table groupTable = summaryTableName(sourceTable);
         StringColumn groupColumn = StringColumn.create("Group");
         groupTable.addColumns(groupColumn);
         boolean firstFunction = true;
-        for (Map.Entry<String, Collection<AggregateFunction<?,?>>> entry : functions.asMap().entrySet()) {
+        for (Map.Entry<String, Collection<AggregateFunction<?, ?>>> entry :
+                functions.asMap().entrySet()) {
             String columnName = entry.getKey();
             for (AggregateFunction function : entry.getValue()) {
                 String colName = aggregateColumnName(columnName, function.functionName());


Test patch:
diff --git a/core/src/test/java/tech/tablesaw/api/TableTest.java b/core/src/test/java/tech/tablesaw/api/TableTest.java
index 7232dda7..0a6c70a2 100644
--- a/core/src/test/java/tech/tablesaw/api/TableTest.java
+++ b/core/src/test/java/tech/tablesaw/api/TableTest.java
@@ -66,7 +66,7 @@ public class TableTest {
         Table result = table.summarize("Injuries", mean, stdDev).by("State");
         assertEquals(49, result.rowCount());
         assertEquals(3, result.columnCount());
-        assertEquals("4.580805569368455", result.column(1).getString(0));
+        assertEquals("4.580805569368455", result.doubleColumn(1).getString(0));
     }
 
     @Test
diff --git a/core/src/test/java/tech/tablesaw/api/TextColumnTest.java b/core/src/test/java/tech/tablesaw/api/TextColumnTest.java
index 37745461..6f3dd495 100644
--- a/core/src/test/java/tech/tablesaw/api/TextColumnTest.java
+++ b/core/src/test/java/tech/tablesaw/api/TextColumnTest.java
@@ -381,6 +381,17 @@ public class TextColumnTest {
         assertEquals(result.get(1), "run--walk--swim");
     }
 
+    @Test
+    public void testAsStringColumn() throws Exception {
+        Table table = Table.read().csv("../data/first_names.csv");
+        StringColumn name = table.stringColumn("emma");
+        TextColumn name2 = name.asTextColumn();
+        StringColumn name3 = name2.asStringColumn();
+        for (int i = 0; i < table.rowCount(); i++) {
+            assertEquals(name.get(i), name3.get(i));
+        }
+    }
+
     @Test
     public void testTrim() {
         String[] words = {" running ", " run run run "};
@@ -427,6 +438,20 @@ public class TextColumnTest {
         assertEquals(result.get(1), "Backwards Writing");
     }
 
+    @Test
+    void testSort() throws Exception {
+        Table t = Table.read().csv("../data/bush.csv");
+        TextColumn whoText = t.stringColumn("who").asTextColumn();
+        whoText.setName("who text");
+        t.addColumns(whoText);
+        Table t2 = t.copy();
+        t.sortAscendingOn("who text");
+        t2.sortAscendingOn("who");
+        for (int i = 0; i < t.rowCount(); i++) {
+            assertEquals(t.row(i).getString("who text"), t2.row(i).getString("who"));
+        }
+    }
+
     @Test
     public void tokenizeAndSort1() {
         String[] words = {"Stop,Breaking,Down", "Writing Backwards"};
diff --git a/core/src/test/java/tech/tablesaw/io/jdbc/SqlResultSetReaderTest.java b/core/src/test/java/tech/tablesaw/io/jdbc/SqlResultSetReaderTest.java
index ff5f0b4d..f5f22d58 100644
--- a/core/src/test/java/tech/tablesaw/io/jdbc/SqlResultSetReaderTest.java
+++ b/core/src/test/java/tech/tablesaw/io/jdbc/SqlResultSetReaderTest.java
@@ -65,6 +65,9 @@ public class SqlResultSetReaderTest {
         // Build the OracleNumbers table.
         TestDb.buildNumbersTable(conn);
         
+        // Build the NullValues table.
+        TestDb.buildNullValuesTable(conn);
+        
         try (Statement stmt = conn.createStatement()) {
             String sql;
 
@@ -107,6 +110,20 @@ public class SqlResultSetReaderTest {
                 assertTrue(numbers.column("NumDouble7_8").type() instanceof DoubleColumnType);
                 assertTrue(numbers.column("NumDouble7_16").type() instanceof DoubleColumnType);
             }
+            
+            sql = "SELECT * FROM NullValues";
+            try (ResultSet rs = stmt.executeQuery(sql)) {
+                Table nullValues = SqlResultSetReader.read(rs);
+                assertEquals(8, nullValues.columnCount());
+                assertEquals(3, nullValues.rowCount());
+                assertEquals(2, nullValues.column("StringValue").removeMissing().size());
+                assertEquals(1, nullValues.column("DoubleValue").removeMissing().size());
+                assertEquals(2, nullValues.column("IntegerValue").removeMissing().size());
+                assertEquals(1, nullValues.column("ShortValue").removeMissing().size());
+                assertEquals(1, nullValues.column("LongValue").removeMissing().size());
+                assertEquals(1, nullValues.column("FloatValue").removeMissing().size());
+                assertEquals(1, nullValues.column("BooleanValue").removeMissing().size());
+            }
 
         }
     }
diff --git a/core/src/test/java/tech/tablesaw/io/string/DataFramePrinterTest.java b/core/src/test/java/tech/tablesaw/io/string/DataFramePrinterTest.java
index 2eeb9379..0084344d 100644
--- a/core/src/test/java/tech/tablesaw/io/string/DataFramePrinterTest.java
+++ b/core/src/test/java/tech/tablesaw/io/string/DataFramePrinterTest.java
@@ -2,7 +2,9 @@ package tech.tablesaw.io.string;
 
 import org.junit.jupiter.api.Test;
 import tech.tablesaw.api.DoubleColumn;
+import tech.tablesaw.api.FloatColumn;
 import tech.tablesaw.api.Table;
+import tech.tablesaw.columns.numbers.NumberColumnFormatter;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -30,5 +32,65 @@ public class DataFramePrinterTest {
         assertTrue(out.contains("fox"));
     }
 
-    
+    @Test
+    public void printWithSmallDoubleNumber() {
+        DoubleColumn col = DoubleColumn.create("testCol");
+        col.append(0.000003);
+        Table table = Table.create("small decimal table", col);
+        String out = table.print();
+        assertTrue(out.contains("0.000003"));
+    }
+
+    @Test
+    public void printWithSmallFloatNumber() {
+        FloatColumn col = FloatColumn.create("testCol");
+        col.append(0.000003f);
+        Table table = Table.create("small float table", col);
+        String out = table.print();
+        assertTrue(out.contains("0.000003"));
+    }
+
+    @Test
+    public void printWithExponent() {
+        DoubleColumn col = DoubleColumn.create("testCol");
+        col.append(0.000003);
+        col.append(123.000003);
+        col.setPrintFormatter(NumberColumnFormatter.standard());
+        Table table = Table.create("small float table", col);
+        String out = table.print();
+        assertTrue(out.contains("3.0E-6"));
+        assertTrue(out.contains("123.000003"));
+    }
+
+    @Test
+    public void printWithLargeDoubleNumber() {
+        DoubleColumn col = DoubleColumn.create("testCol");
+        col.append(33.333333333333328);
+        col.append(9007199254740992d);
+        col.append(900719925474099.1d);
+        col.append(90071992547409.11d);
+        Table table = Table.create("large float table", col);
+        String out = table.print();
+        assertTrue(out.contains("33.3333333333333"));
+        assertTrue(out.contains("9007199254740992"));
+        assertTrue(out.contains("900719925474099.1"));
+        assertTrue(out.contains("90071992547409.11"));
+    }
+
+    @Test
+    public void printWithLargeFloatNumber() {
+        FloatColumn col = FloatColumn.create("testCol");
+        col.append(33.333333333333328f);
+        col.append(900719925474f);
+        col.append(9007199254.1f);
+        col.append(90071992.11f);
+        col.append(90071.11f);
+        Table table = Table.create("large float table", col);
+        String out = table.print();
+        assertTrue(out.contains("33.33333206176758"));
+        assertTrue(out.contains("900719902720"));
+        assertTrue(out.contains("9007199232"));
+        assertTrue(out.contains("90071992"));
+        assertTrue(out.contains("90071.109375"));
+    }
 }
diff --git a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerPerformanceTest.java b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerPerformanceTest.java
new file mode 100644
index 00000000..2569d698
--- /dev/null
+++ b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerPerformanceTest.java
@@ -0,0 +1,84 @@
+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 50931899..db8b06df 100644
--- a/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
+++ b/core/src/test/java/tech/tablesaw/joining/DataFrameJoinerTest.java
@@ -1,9 +1,12 @@
 package tech.tablesaw.joining;
 
 import com.google.common.base.Joiner;
+import java.util.List;
+import java.util.stream.Collectors;
 import org.junit.jupiter.api.Test;
 import tech.tablesaw.api.Table;
 import tech.tablesaw.api.TextColumn;
+import tech.tablesaw.columns.Column;
 import tech.tablesaw.selection.Selection;
 
 import java.util.Arrays;
@@ -635,6 +638,18 @@ public class DataFrameJoinerTest {
         assertEquals(14, joined.rowCount());
     }
 
+    @Test
+    public void joiningShouldNotMutateColNamesOnOriginalTable() {
+        Table table1 = createSTUDENT();
+        Table table2 = createINSTRUCTOR();
+        List<String> originalColumns = table2.columns().stream().map(Column::name).collect(Collectors.toList());
+
+        table1.joinOn("Age").inner(true, table2);
+
+        List<String> newColumns = table2.columns().stream().map(Column::name).collect(Collectors.toList());
+        assertEquals(originalColumns, newColumns);
+    }
+
     @Test
     public void innerJoinInstructorStudentOnAge() {
         Table table1 = createINSTRUCTOR();
diff --git a/core/src/test/java/tech/tablesaw/table/TableSliceGroupTest.java b/core/src/test/java/tech/tablesaw/table/TableSliceGroupTest.java
index ac1adb0c..bf018196 100644
--- a/core/src/test/java/tech/tablesaw/table/TableSliceGroupTest.java
+++ b/core/src/test/java/tech/tablesaw/table/TableSliceGroupTest.java
@@ -28,6 +28,7 @@ import tech.tablesaw.aggregate.NumericAggregateFunction;
 import tech.tablesaw.api.NumericColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.Table;
+import tech.tablesaw.api.TextColumn;
 import tech.tablesaw.io.csv.CsvReadOptions;
 
 public class TableSliceGroupTest {
@@ -96,6 +97,18 @@ public class TableSliceGroupTest {
         assertEquals(aggregated.rowCount(), group.size());
     }
 
+    @Test
+    public void testCreateWithTextColumn() {
+        TextColumn whoText = table.stringColumn("who").asTextColumn();
+        whoText.setName("who text");
+        table.addColumns(whoText);
+        TableSliceGroup group1 = StandardTableSliceGroup.create(table, table.categoricalColumn("who text"));
+        TableSliceGroup group2 = StandardTableSliceGroup.create(table, table.categoricalColumn("who"));
+        Table aggregated1 = group1.aggregate("approval", exaggerate);
+        Table aggregated2 = group2.aggregate("approval", exaggerate);
+        assertEquals(aggregated1.rowCount(), aggregated2.rowCount());
+    }
+
     @Test
     public void aggregateWithMultipleColumns() {
         table.addColumns(table.categoricalColumn("approval").copy().setName("approval2"));
diff --git a/core/src/test/java/tech/tablesaw/util/TestDb.java b/core/src/test/java/tech/tablesaw/util/TestDb.java
index 23d443d9..79a2360b 100644
--- a/core/src/test/java/tech/tablesaw/util/TestDb.java
+++ b/core/src/test/java/tech/tablesaw/util/TestDb.java
@@ -101,6 +101,14 @@ public class TestDb {
                 // No need to report an error.
                 // The table simply did not exist.
             }
+            
+            try {
+                // Drop the NullValues table.
+                stmt.execute("DROP TABLE NullValues");
+            } catch (SQLException ex) {
+                // No need to report an error.
+                // The table simply did not exist.
+            }
         } catch (SQLException ex) {
             System.out.println("ERROR: " + ex.getMessage());
             ex.printStackTrace();
@@ -390,5 +398,50 @@ public class TestDb {
         }
     }
 
+    /**
+     * The buildNullValues method creates the
+     * NullValues table and adds some rows to it.
+     */
+    public static void buildNullValuesTable(Connection conn) {
+        try {
+            // Get a Statement object.
+            Statement stmt = conn.createStatement();
+
+            // Create the table.
+            stmt.execute("CREATE TABLE NullValues (" +
+                    "StringValue CHAR(25), " +
+                    "PrimaryValue CHAR(10) NOT NULL PRIMARY KEY, " +
+                    "DoubleValue DOUBLE, " +
+                    "IntegerValue INTEGER, " +
+                    "ShortValue SMALLINT, " +
+                    "LongValue BIGINT, " +
+                    "FloatValue FLOAT, " +
+                    "BooleanValue Boolean" +
+                    ")");
+
+            // Insert row #1.
+            stmt.execute("INSERT INTO NullValues VALUES ( " +
+                    "'Non Null Description', " +
+                    "'001', " +
+                    "8.95, " +
+                    "1, " +
+                    "1, " +
+                    "1, " +
+                    "3.14, " +
+                    "TRUE )");
+
+            // Insert row #2.
+            stmt.execute("INSERT INTO NullValues (PrimaryValue, IntegerValue) VALUES ( " +
+                    "'002', " +
+                    "2 )");
+
+            // Insert row #3.
+            stmt.execute("INSERT INTO NullValues (StringValue, PrimaryValue) VALUES ( " +
+                    "'Non Null Description', " +
+                    "'003')");
+        } catch (SQLException ex) {
+            System.out.println("ERROR: " + ex.getMessage());
+        }
+    }
 
 }
\ No newline at end of file
diff --git a/html/src/test/java/tech/tablesaw/io/html/HtmlWriterTest.java b/html/src/test/java/tech/tablesaw/io/html/HtmlWriterTest.java
index c8dc895f..05641c36 100644
--- a/html/src/test/java/tech/tablesaw/io/html/HtmlWriterTest.java
+++ b/html/src/test/java/tech/tablesaw/io/html/HtmlWriterTest.java
@@ -51,12 +51,12 @@ public class HtmlWriterTest {
                 " </thead>" + LINE_END +
                 " <tbody>" + LINE_END +
                 "  <tr>" + LINE_END +
-                "   <td>1.0</td>" + LINE_END +
-                "   <td>1.0</td>" + LINE_END +
+                "   <td>1</td>" + LINE_END +
+                "   <td>1</td>" + LINE_END +
                 "  </tr>" + LINE_END +
                 "  <tr>" + LINE_END +
-                "   <td>2.0</td>" + LINE_END +
-                "   <td>2.0</td>" + LINE_END +
+                "   <td>2</td>" + LINE_END +
+                "   <td>2</td>" + LINE_END +
                 "  </tr>" + LINE_END +
                 "  <tr>" + LINE_END +
                 "   <td></td>" + LINE_END +
@@ -87,12 +87,12 @@ public class HtmlWriterTest {
                 " </thead>" + LINE_END +
                 " <tbody>" + LINE_END +
                 "  <tr class=\"even\">" + LINE_END +
-                "   <td>1.0</td>" + LINE_END +
-                "   <td>1.0</td>" + LINE_END +
+                "   <td>1</td>" + LINE_END +
+                "   <td>1</td>" + LINE_END +
                 "  </tr>" + LINE_END +
                 "  <tr class=\"odd\">" + LINE_END +
-                "   <td>2.0</td>" + LINE_END +
-                "   <td>2.0</td>" + LINE_END +
+                "   <td>2</td>" + LINE_END +
+                "   <td>2</td>" + LINE_END +
                 "  </tr>" + LINE_END +
                 "  <tr class=\"even\">" + LINE_END +
                 "   <td></td>" + LINE_END +