Repository: jtablesaw/tablesaw
Pull request #350 (2018-08-19 05:56:43)

Add LongColumn and remove the need for data wrappers
Thanks for contributing.

- [X] Tick to sign-off your agreement to the [Developer Certificate of Origin (DCO) 1.1](https://developercertificate.org)

## Description

Combined columns and data wrappers. Change saves abut 700 lines of code and also identified and fixed a few bugs

Added new `LongColumn` type fixing https://github.com/jtablesaw/tablesaw/issues/349

benmccann: @lwhite1 there are two big changes here, but I made them in separate commits in case you want to review separately

Patch:
diff --git a/core/src/main/java/tech/tablesaw/api/BooleanColumn.java b/core/src/main/java/tech/tablesaw/api/BooleanColumn.java
index c7d54e61..36f21010 100644
--- a/core/src/main/java/tech/tablesaw/api/BooleanColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/BooleanColumn.java
@@ -627,15 +627,6 @@ public class BooleanColumn extends AbstractColumn<Boolean> implements BooleanMap
         return data.iterator();
     }
 
-    @Override
-    public int[] asIntArray() {
-        int[] output = new int[data.size()];
-        for (int i = 0; i < data.size(); i++) {
-            output[i] = data.getByte(i);
-        }
-        return output;
-    }
-
     public DoubleColumn asNumberColumn() {
         DoubleColumn numberColumn = DoubleColumn.create(this.name() + ": ints", size());
         ByteArrayList data = data();
diff --git a/core/src/main/java/tech/tablesaw/api/CategoricalColumn.java b/core/src/main/java/tech/tablesaw/api/CategoricalColumn.java
index 3a4f2c20..0325f0d7 100644
--- a/core/src/main/java/tech/tablesaw/api/CategoricalColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/CategoricalColumn.java
@@ -14,14 +14,12 @@
 
 package tech.tablesaw.api;
 
-import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import java.util.Map;
+
 import it.unimi.dsi.fastutil.objects.Object2IntMap;
 import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
 import tech.tablesaw.columns.Column;
 
-import java.util.Map;
-import java.util.Set;
-
 /**
  * A column type that can be summarized, or serve as a grouping variable in cross tabs or other aggregation operations.
  * <p>
@@ -69,9 +67,4 @@ public interface CategoricalColumn<T> extends Column<T> {
         return t;
     }
 
-    int[] asIntArray();
-
-    default Set<Integer> asIntegerSet() {
-        return new IntOpenHashSet(asIntArray());
-    }
 }
diff --git a/core/src/main/java/tech/tablesaw/api/ColumnType.java b/core/src/main/java/tech/tablesaw/api/ColumnType.java
index 204906cd..ba05468e 100644
--- a/core/src/main/java/tech/tablesaw/api/ColumnType.java
+++ b/core/src/main/java/tech/tablesaw/api/ColumnType.java
@@ -1,6 +1,10 @@
 package tech.tablesaw.api;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import com.google.common.base.Preconditions;
+
 import tech.tablesaw.columns.Column;
 import tech.tablesaw.columns.SkipColumnType;
 import tech.tablesaw.columns.StringParser;
@@ -10,19 +14,18 @@ import tech.tablesaw.columns.datetimes.DateTimeColumnType;
 import tech.tablesaw.columns.numbers.DoubleColumnType;
 import tech.tablesaw.columns.numbers.FloatColumnType;
 import tech.tablesaw.columns.numbers.IntColumnType;
+import tech.tablesaw.columns.numbers.LongColumnType;
 import tech.tablesaw.columns.strings.StringColumnType;
 import tech.tablesaw.columns.times.TimeColumnType;
 import tech.tablesaw.io.csv.CsvReadOptions;
 
-import java.util.HashMap;
-import java.util.Map;
-
 public interface ColumnType {
 
     Map<String, ColumnType> values = new HashMap<>();
 
     // standard column types
     IntColumnType INTEGER = IntColumnType.INSTANCE;
+    LongColumnType LONG = LongColumnType.INSTANCE;
     FloatColumnType FLOAT = FloatColumnType.INSTANCE;
     BooleanColumnType BOOLEAN = BooleanColumnType.INSTANCE;
     StringColumnType STRING = StringColumnType.INSTANCE;
diff --git a/core/src/main/java/tech/tablesaw/api/DateColumn.java b/core/src/main/java/tech/tablesaw/api/DateColumn.java
index 9409e013..cecfae5e 100644
--- a/core/src/main/java/tech/tablesaw/api/DateColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/DateColumn.java
@@ -596,11 +596,6 @@ public class DateColumn extends AbstractColumn<LocalDate> implements DateFilters
         return this;
     }
 
-    @Override
-    public int[] asIntArray() {
-        return data.toArray(new int[size()]);
-    }
-
     @Override
     public Object[] asObjectArray() {
         final LocalDate[] output = new LocalDate[data.size()];
diff --git a/core/src/main/java/tech/tablesaw/api/DoubleColumn.java b/core/src/main/java/tech/tablesaw/api/DoubleColumn.java
index 267964ae..cb44a13a 100644
--- a/core/src/main/java/tech/tablesaw/api/DoubleColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/DoubleColumn.java
@@ -1,10 +1,6 @@
 package tech.tablesaw.api;
 
-import static tech.tablesaw.api.ColumnType.DOUBLE;
-
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
+import java.nio.ByteBuffer;
 import java.util.Iterator;
 import java.util.List;
 import java.util.function.DoubleConsumer;
@@ -15,35 +11,39 @@ import java.util.function.ToDoubleFunction;
 import com.google.common.base.Preconditions;
 
 import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.doubles.DoubleArrays;
+import it.unimi.dsi.fastutil.doubles.DoubleComparator;
+import it.unimi.dsi.fastutil.doubles.DoubleListIterator;
 import it.unimi.dsi.fastutil.doubles.DoubleOpenHashSet;
 import it.unimi.dsi.fastutil.doubles.DoubleSet;
 import tech.tablesaw.columns.Column;
-import tech.tablesaw.columns.numbers.DoubleDataWrapper;
+import tech.tablesaw.columns.StringParser;
+import tech.tablesaw.columns.numbers.DoubleColumnType;
+import tech.tablesaw.columns.numbers.FloatColumnType;
 import tech.tablesaw.columns.numbers.NumberFillers;
 import tech.tablesaw.columns.numbers.NumberIterable;
 import tech.tablesaw.columns.numbers.NumberIterator;
-import tech.tablesaw.columns.numbers.NumericDataWrapper;
 
 public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<Double>, NumberFillers<DoubleColumn> {
 
-    protected DoubleColumn(final String name, final DoubleArrayList data) {
-        super(DOUBLE, name);
-        setDataWrapper(new DoubleDataWrapper(data));
-    }
+    private static final ColumnType COLUMN_TYPE = ColumnType.DOUBLE;
+
+    /**
+     * Compares two doubles, such that a sort based on this comparator would sort in descending order
+     */
+    private final DoubleComparator descendingComparator = (o2, o1) -> (Double.compare(o1, o2));
 
-    protected DoubleColumn(final String name, final NumericDataWrapper data) {
-        super(data.type(), name);
-        setDataWrapper(data);
+    private final DoubleArrayList data;
+
+    protected DoubleColumn(final String name, final DoubleArrayList data) {
+        super(COLUMN_TYPE, name, data);
+        this.data = data;
     }
 
     public static DoubleColumn create(final String name, final double[] arr) {
         return new DoubleColumn(name, new DoubleArrayList(arr));
     }
 
-    public static DoubleColumn create(final String name, final NumericDataWrapper data) {
-        return new DoubleColumn(name, data);
-    }
-
     public static DoubleColumn create(final String name) {
         return new DoubleColumn(name, new DoubleArrayList());
     }
@@ -94,13 +94,13 @@ public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<
     }
 
     @Override
-    protected NumberColumn<Double> createCol(String name, NumericDataWrapper data) {
-	return DoubleColumn.create(name, data);
+    public DoubleColumn createCol(final String name, final int initialSize) {
+        return create(name, initialSize);
     }
 
     @Override
     public Double get(int index) {
-	return getDouble(index);
+        return getDouble(index);
     }
 
     @Override
@@ -116,18 +116,71 @@ public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<
         return column;
     }
 
+    @Override
+    public DoubleColumn top(int n) {
+        DoubleArrayList top = new DoubleArrayList();
+        double[] values = data.toDoubleArray();
+        DoubleArrays.parallelQuickSort(values, descendingComparator);
+        for (int i = 0; i < n && i < values.length; i++) {
+            top.add(values[i]);
+        }
+        return new DoubleColumn(name() + "[Top " + n  + "]", top);
+    }
+
+    @Override
+    public DoubleColumn bottom(final int n) {
+        DoubleArrayList bottom = new DoubleArrayList();
+        double[] values = data.toDoubleArray();
+        DoubleArrays.parallelQuickSort(values);
+        for (int i = 0; i < n && i < values.length; i++) {
+            bottom.add(values[i]);
+        }
+        return new DoubleColumn(name() + "[Bottoms " + n  + "]", bottom);
+    }    
+
+    @Override
+    public DoubleColumn lag(int n) {
+        final int srcPos = n >= 0 ? 0 : 0 - n;
+        final double[] dest = new double[size()];
+        final int destPos = n <= 0 ? 0 : n;
+        final int length = n >= 0 ? size() - n : size() + n;
+
+        for (int i = 0; i < size(); i++) {
+            dest[i] = FloatColumnType.missingValueIndicator();
+        }
+
+        double[] array = data.toDoubleArray();
+
+        System.arraycopy(array, srcPos, dest, destPos, length);
+        return new DoubleColumn(name() + " lag(" + n + ")", new DoubleArrayList(dest));
+    }
+
+    @Override
+    public DoubleColumn removeMissing() {
+        DoubleColumn result = copy();
+        result.clear();
+        DoubleListIterator iterator = data.iterator();
+        while (iterator.hasNext()) {
+            double v = iterator.nextDouble();
+            if (!isMissingValue(v)) {
+                result.append(v);
+            }
+        }
+        return result;
+    }
+
     public double firstElement() {
         if (size() > 0) {
             return getDouble(0);
         }
-        return data.missingValueIndicator();
+        return (Double) COLUMN_TYPE.getMissingValueIndicator();
     }
 
     /**
      * Adds the given float to this column
      */
     public DoubleColumn append(final float f) {
-        data.append(f);
+        data.add(f);
         return this;
     }
 
@@ -135,12 +188,12 @@ public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<
      * Adds the given double to this column
      */
     public DoubleColumn append(double d) {
-        data.append(d);
+        data.add(d);
         return this;
     }
 
     public DoubleColumn append(int i) {
-        data.append(i);
+        data.add(i);
         return this;
     }
 
@@ -162,37 +215,15 @@ public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<
 
     @Override
     public DoubleColumn emptyCopy(final int rowSize) {
-	return (DoubleColumn) super.emptyCopy(rowSize);
+        return (DoubleColumn) super.emptyCopy(rowSize);
     }
 
     @Override
     public DoubleColumn copy() {
-        return (DoubleColumn) super.copy();
-    }
-
-    /**
-     * Returns a DateTimeColumn where each value is the LocalDateTime represented by the values in this column
-     * <p>
-     * The values in this column must be longs that represent the time in milliseconds from the epoch as in standard
-     * Java date/time calculations
-     *
-     * @param offset The ZoneOffset to use in the calculation
-     * @return A column of LocalDateTime values
-     */
-    public DateTimeColumn asDateTimes(ZoneOffset offset) {
-        DateTimeColumn column = DateTimeColumn.create(name() + ": date time");
-        NumberIterator it = numberIterator();
-        while (it.hasNext()) {
-            double d = it.next();
-            LocalDateTime dateTime =
-                    Instant.ofEpochMilli((long) d).atZone(offset).toLocalDateTime();
-            column.append(dateTime);
-        }
-        return column;
+        return new DoubleColumn(name(), data.clone());
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public Iterator<Double> iterator() {
         return (Iterator<Double>) data.iterator();
     }
@@ -213,7 +244,12 @@ public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<
 
     @Override
     public DoubleColumn set(int i, Double val) {
-        return (DoubleColumn) set(i, (double) val);
+        return set(i, (double) val);
+    }
+
+    public DoubleColumn set(int i, double val) {
+        data.set(i, val);
+        return this;
     }
 
     @Override
@@ -301,4 +337,83 @@ public class DoubleColumn extends NumberColumn<Double> implements NumericColumn<
         return result;
     }
 
+    @Override
+    public byte[] asBytes(int rowNumber) {
+        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putDouble(getDouble(rowNumber)).array();
+    }
+
+    @Override
+    public int countUnique() {
+        DoubleSet uniqueElements = new DoubleOpenHashSet();
+        for (int i = 0; i < size(); i++) {
+            if (!isMissing(i)) {
+                uniqueElements.add(getDouble(i));
+            }
+        }
+        return uniqueElements.size();
+    }
+
+    @Override
+    public double getDouble(int row) {
+        return data.getDouble(row);
+    }
+
+    public boolean isMissingValue(double value) {
+        return DoubleColumnType.isMissingValue(value);
+    }
+
+    @Override
+    public boolean isMissing(int rowNumber) {
+        return isMissingValue(getDouble(rowNumber));
+    }
+
+    @Override
+    public NumberIterator numberIterator() {
+        return new NumberIterator(data);
+    }
+
+    @Override
+    public void sortAscending() {
+        DoubleArrays.parallelQuickSort(data.elements());
+    }
+
+    @Override
+    public void sortDescending() {
+        DoubleArrays.parallelQuickSort(data.elements(), descendingComparator);
+    }
+
+    @Override
+    public DoubleColumn appendMissing() {
+        return append(DoubleColumnType.missingValueIndicator());
+    }
+
+    @Override
+    public DoubleColumn appendObj(Object obj) {
+        if (obj == null) {
+            return appendMissing();
+        }
+        if (obj instanceof Double) {
+            return append((double) obj);
+        }
+        throw new IllegalArgumentException("Could not append " + obj.getClass());
+    }
+
+    @Override
+    public DoubleColumn appendCell(final String value) {
+        try {
+            return append(DoubleColumnType.DEFAULT_PARSER.parseDouble(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name() + ": " + e.getMessage());
+        }
+    }
+
+    @Override
+    public DoubleColumn appendCell(final String value, StringParser<?> parser) {
+        try {
+            return append(parser.parseDouble(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name()  + ": " + e.getMessage());
+        }
+    }
+
 }
diff --git a/core/src/main/java/tech/tablesaw/api/FloatColumn.java b/core/src/main/java/tech/tablesaw/api/FloatColumn.java
index 88ec0766..f83593e6 100644
--- a/core/src/main/java/tech/tablesaw/api/FloatColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/FloatColumn.java
@@ -1,31 +1,36 @@
 package tech.tablesaw.api;
 
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+
 import com.google.common.base.Preconditions;
+
 import it.unimi.dsi.fastutil.floats.FloatArrayList;
-import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
-import it.unimi.dsi.fastutil.ints.IntSet;
+import it.unimi.dsi.fastutil.floats.FloatArrays;
+import it.unimi.dsi.fastutil.floats.FloatComparator;
+import it.unimi.dsi.fastutil.floats.FloatListIterator;
+import it.unimi.dsi.fastutil.floats.FloatOpenHashSet;
+import it.unimi.dsi.fastutil.floats.FloatSet;
 import tech.tablesaw.columns.Column;
-import tech.tablesaw.columns.numbers.FloatDataWrapper;
-import tech.tablesaw.columns.numbers.NumericDataWrapper;
-
-import java.util.Iterator;
-
-import static tech.tablesaw.api.ColumnType.FLOAT;
+import tech.tablesaw.columns.StringParser;
+import tech.tablesaw.columns.numbers.DoubleColumnType;
+import tech.tablesaw.columns.numbers.FloatColumnType;
+import tech.tablesaw.columns.numbers.NumberIterator;
 
 public class FloatColumn extends NumberColumn<Float> implements NumericColumn<Float> {
 
-    protected FloatColumn(final String name, FloatArrayList data) {
-        super(FLOAT, name);
-        setDataWrapper(new FloatDataWrapper(data));
-    }
+    private static final FloatColumnType COLUMN_TYPE = ColumnType.FLOAT;
 
-    protected FloatColumn(final String name, final NumericDataWrapper data) {
-        super(data.type(), name);
-        setDataWrapper(data);
-    }
+    /**
+     * Compares two doubles, such that a sort based on this comparator would sort in descending order
+     */
+    private final FloatComparator descendingComparator = (o2, o1) -> (Float.compare(o1, o2));
+
+    private final FloatArrayList data;    
 
-    public static FloatColumn create(final String name, final NumericDataWrapper data) {
-        return new FloatColumn(name, data);
+    protected FloatColumn(final String name, FloatArrayList data) {
+        super(COLUMN_TYPE, name, data);
+        this.data = data;
     }
 
     public static FloatColumn create(final String name) {
@@ -41,32 +46,85 @@ public class FloatColumn extends NumberColumn<Float> implements NumericColumn<Fl
     }
 
     @Override
-    protected NumberColumn<Float> createCol(String name, NumericDataWrapper data) {
-	return FloatColumn.create(name, data);
+    public FloatColumn createCol(final String name, final int initialSize) {
+        return create(name, initialSize);
     }
 
     @Override
     public Float get(int index) {
-	return data.getFloat(index);
+        return data.getFloat(index);
     }
 
     @Override
-    public FloatColumn unique() {  // todo, this is broken. these should be floats
-        final IntSet values = new IntOpenHashSet();
+    public FloatColumn unique() {
+        final FloatSet values = new FloatOpenHashSet();
         for (int i = 0; i < size(); i++) {
             if (!isMissing(i)) {
-                values.add(getInt(i));
+                values.add(getFloat(i));
             }
         }
         final FloatColumn column = FloatColumn.create(name() + " Unique values", values.size());
-        for (int value : values) {
+        for (float value : values) {
             column.append(value);
         }
         return column;
     }
 
+    @Override
+    public FloatColumn top(int n) {
+        FloatArrayList top = new FloatArrayList();
+        float[] values = data.toFloatArray();
+        FloatArrays.parallelQuickSort(values, descendingComparator);
+        for (int i = 0; i < n && i < values.length; i++) {
+            top.add(values[i]);
+        }
+        return new FloatColumn(name() + "[Top " + n  + "]", top);
+    }
+
+    @Override
+    public FloatColumn bottom(final int n) {
+        FloatArrayList bottom = new FloatArrayList();
+        float[] values = data.toFloatArray();
+        FloatArrays.parallelQuickSort(values);
+        for (int i = 0; i < n && i < values.length; i++) {
+            bottom.add(values[i]);
+        }
+        return new FloatColumn(name() + "[Bottoms " + n  + "]", bottom);
+    }
+
+    @Override
+    public FloatColumn lag(int n) {
+        final int srcPos = n >= 0 ? 0 : 0 - n;
+        final float[] dest = new float[size()];
+        final int destPos = n <= 0 ? 0 : n;
+        final int length = n >= 0 ? size() - n : size() + n;
+
+        for (int i = 0; i < size(); i++) {
+            dest[i] = FloatColumnType.missingValueIndicator();
+        }
+
+        float[] array = data.toFloatArray();
+
+        System.arraycopy(array, srcPos, dest, destPos, length);
+        return new FloatColumn(name() + " lag(" + n + ")", new FloatArrayList(dest));
+    }
+
+    @Override
+    public FloatColumn removeMissing() {
+        FloatColumn result = copy();
+        result.clear();
+        FloatListIterator iterator = data.iterator();
+        while (iterator.hasNext()) {
+            final float v = iterator.nextFloat();
+            if (!isMissingValue(v)) {
+                result.append(v);
+            }
+        }
+        return result;
+    }
+
     public FloatColumn append(float i) {
-        data.append(i);
+        data.add(i);
         return this;
     }
 
@@ -82,25 +140,24 @@ public class FloatColumn extends NumberColumn<Float> implements NumericColumn<Fl
 
     @Override
     public FloatColumn emptyCopy(final int rowSize) {
-	return (FloatColumn) super.emptyCopy(rowSize);
+        return (FloatColumn) super.emptyCopy(rowSize);
     }
 
     @Override
     public FloatColumn copy() {
-        return (FloatColumn) super.copy();
+        return new FloatColumn(name(), data.clone());
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public Iterator<Float> iterator() {
-        return (Iterator<Float>) data.iterator();
+        return data.iterator();
     }
 
     @Override
     public Object[] asObjectArray() {
-        final Integer[] output = new Integer[size()];
+        final Float[] output = new Float[size()];
         for (int i = 0; i < size(); i++) {
-            output[i] = getInt(i);
+            output[i] = getFloat(i);
         }
         return output;
     }
@@ -112,7 +169,12 @@ public class FloatColumn extends NumberColumn<Float> implements NumericColumn<Fl
 
     @Override
     public FloatColumn set(int i, Float val) {
-        return (FloatColumn) set(i, (float) val);
+        return set(i, (float) val);
+    }
+
+    public FloatColumn set(int i, float val) {
+        data.set(i, val);
+        return this;
     }
 
     @Override
@@ -125,4 +187,97 @@ public class FloatColumn extends NumberColumn<Float> implements NumericColumn<Fl
         return this;
     }
 
+    @Override
+    public byte[] asBytes(int rowNumber) {
+        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putFloat(getFloat(rowNumber)).array();
+    }
+
+    @Override
+    public int countUnique() {
+        FloatSet uniqueElements = new FloatOpenHashSet();
+        for (int i = 0; i < size(); i++) {
+            if (!isMissing(i)) {
+                uniqueElements.add(getFloat(i));
+            }
+        }
+        return uniqueElements.size();
+    }
+
+    @Override
+    public double getDouble(int row) {
+        float value = data.getFloat(row);
+        if (isMissingValue(value)) {
+            return DoubleColumnType.missingValueIndicator();
+        }
+        return value;
+    }
+
+    /**
+     * Returns a float representation of the data at the given index. Some precision may be lost, and if the value is
+     * to large to be cast to a float, an exception is thrown.
+     *
+     * @throws  ClassCastException if the value can't be cast to ta float
+     */
+    public float getFloat(int row) {
+        return data.getFloat(row);
+    }
+
+    public boolean isMissingValue(float value) {
+        return FloatColumnType.isMissingValue(value);
+    }
+
+    @Override
+    public boolean isMissing(int rowNumber) {
+        return isMissingValue(getFloat(rowNumber));
+    }
+
+    @Override
+    public NumberIterator numberIterator() {
+        return new NumberIterator(data);
+    }
+
+    @Override
+    public void sortAscending() {
+        FloatArrays.parallelQuickSort(data.elements());
+    }
+
+    @Override
+    public void sortDescending() {
+        FloatArrays.parallelQuickSort(data.elements(), descendingComparator);
+    }
+
+    @Override
+    public FloatColumn appendMissing() {
+        return append(FloatColumnType.missingValueIndicator());
+    }
+
+    @Override
+    public FloatColumn appendObj(Object obj) {
+        if (obj == null) {
+            return appendMissing();
+        }
+        if (obj instanceof Float) {
+            return append((float) obj);
+        }
+        throw new IllegalArgumentException("Could not append " + obj.getClass());
+    }
+
+    @Override
+    public FloatColumn appendCell(final String value) {
+        try {
+            return append(FloatColumnType.DEFAULT_PARSER.parseFloat(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name() + ": " + e.getMessage());
+        }
+    }
+
+    @Override
+    public FloatColumn appendCell(final String value, StringParser<?> parser) {
+        try {
+            return append(parser.parseFloat(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name()  + ": " + e.getMessage());
+        }
+    }    
+
 }
diff --git a/core/src/main/java/tech/tablesaw/api/IntColumn.java b/core/src/main/java/tech/tablesaw/api/IntColumn.java
index 43ba7b29..1b03c9ec 100644
--- a/core/src/main/java/tech/tablesaw/api/IntColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/IntColumn.java
@@ -1,34 +1,38 @@
 package tech.tablesaw.api;
 
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+
 import com.google.common.base.Preconditions;
+
 import it.unimi.dsi.fastutil.ints.IntArrayList;
+import it.unimi.dsi.fastutil.ints.IntArrays;
+import it.unimi.dsi.fastutil.ints.IntComparator;
+import it.unimi.dsi.fastutil.ints.IntListIterator;
 import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
 import it.unimi.dsi.fastutil.ints.IntSet;
 import tech.tablesaw.columns.Column;
+import tech.tablesaw.columns.StringParser;
+import tech.tablesaw.columns.numbers.DoubleColumnType;
 import tech.tablesaw.columns.numbers.IntColumnType;
-import tech.tablesaw.columns.numbers.IntDataWrapper;
 import tech.tablesaw.columns.numbers.NumberColumnFormatter;
-import tech.tablesaw.columns.numbers.NumericDataWrapper;
+import tech.tablesaw.columns.numbers.NumberIterator;
 
-import java.util.Iterator;
+public class IntColumn extends NumberColumn<Integer> implements NumericColumn<Integer>, CategoricalColumn<Integer> {
 
-import static tech.tablesaw.api.ColumnType.INTEGER;
+    private static final IntColumnType COLUMN_TYPE = ColumnType.INTEGER;
 
-public class IntColumn extends NumberColumn<Integer> implements NumericColumn<Integer>, CategoricalColumn<Integer> {
+    /**
+     * Compares two ints, such that a sort based on this comparator would sort in descending order
+     */
+    private final IntComparator descendingComparator = (o2, o1) -> (Integer.compare(o1, o2));
+
+    private final IntArrayList data;
 
     protected IntColumn(final String name, IntArrayList data) {
-        super(INTEGER, name);
+        super(COLUMN_TYPE, name, data);
         this.printFormatter = NumberColumnFormatter.ints();
-        setDataWrapper(new IntDataWrapper(data));
-    }
-
-    protected IntColumn(final String name, final NumericDataWrapper data) {
-        super(data.type(), name);
-        setDataWrapper(data);
-    }
-
-    public static IntColumn create(final String name, final NumericDataWrapper data) {
-        return new IntColumn(name, data);
+        this.data = data;
     }
 
     public static IntColumn create(final String name) {
@@ -43,6 +47,11 @@ public class IntColumn extends NumberColumn<Integer> implements NumericColumn<In
         return new IntColumn(name, new IntArrayList(initialSize));
     }
 
+    @Override
+    public IntColumn createCol(final String name, final int initialSize) {
+        return create(name, initialSize);
+    }
+
     /**
      * Returns a new numeric column initialized with the given name and size. The values in the column are
      * integers beginning at startsWith and continuing through size (exclusive), monotonically increasing by 1
@@ -60,14 +69,9 @@ public class IntColumn extends NumberColumn<Integer> implements NumericColumn<In
         return value == IntColumnType.missingValueIndicator();
     }
 
-    @Override
-    protected NumberColumn<Integer> createCol(String name, NumericDataWrapper data) {
-	return IntColumn.create(name, data);
-    }
-
     @Override
     public Integer get(int index) {
-	return data.getInt(index);
+        return getInt(index);
     }
 
     @Override
@@ -85,8 +89,61 @@ public class IntColumn extends NumberColumn<Integer> implements NumericColumn<In
         return column;
     }
 
+    @Override
+    public IntColumn top(int n) {
+        final IntArrayList top = new IntArrayList();
+        final int[] values = data.toIntArray();
+        IntArrays.parallelQuickSort(values, descendingComparator);
+        for (int i = 0; i < n && i < values.length; i++) {
+            top.add(values[i]);
+        }
+        return new IntColumn(name() + "[Top " + n  + "]", top);
+    }
+
+    @Override
+    public IntColumn bottom(final int n) {
+        final IntArrayList bottom = new IntArrayList();
+        final int[] values = data.toIntArray();
+        IntArrays.parallelQuickSort(values);
+        for (int i = 0; i < n && i < values.length; i++) {
+            bottom.add(values[i]);
+        }
+        return new IntColumn(name() + "[Bottoms " + n  + "]", bottom);
+    }
+
+    @Override
+    public IntColumn lag(int n) {
+        final int srcPos = n >= 0 ? 0 : 0 - n;
+        final int[] dest = new int[size()];
+        final int destPos = n <= 0 ? 0 : n;
+        final int length = n >= 0 ? size() - n : size() + n;
+
+        for (int i = 0; i < size(); i++) {
+            dest[i] = IntColumnType.missingValueIndicator();
+        }
+
+        int[] array = data.toIntArray();
+
+        System.arraycopy(array, srcPos, dest, destPos, length);
+        return new IntColumn(name() + " lag(" + n + ")", new IntArrayList(dest));
+    }
+
+    @Override
+    public IntColumn removeMissing() {
+        IntColumn result = copy();
+        result.clear();
+        IntListIterator iterator = data.iterator();
+        while (iterator.hasNext()) {
+            final int v = iterator.nextInt();
+            if (!isMissingValue(v)) {
+                result.append(v);
+            }
+        }
+        return result;
+    }
+
     public IntColumn append(int i) {
-        data.append(i);
+        data.add(i);
         return this;
     }
 
@@ -102,27 +159,17 @@ public class IntColumn extends NumberColumn<Integer> implements NumericColumn<In
 
     @Override
     public IntColumn emptyCopy(final int rowSize) {
-	return (IntColumn) super.emptyCopy(rowSize);
+        return (IntColumn) super.emptyCopy(rowSize);
     }
 
     @Override
     public IntColumn copy() {
-        return (IntColumn) super.copy();
-    }
-
-    @Override
-    public int[] asIntArray() {  // TODO: Need to figure out how to handle NaN -> Maybe just use a list with nulls?
-        final int[] result = new int[size()];
-        for (int i = 0; i < size(); i++) {
-            result[i] = roundInt(i);
-        }
-        return result;
+        return new IntColumn(name(), data.clone());
     }
 
     @Override
-    @SuppressWarnings("unchecked")
     public Iterator<Integer> iterator() {
-        return (Iterator<Integer>) data.iterator();
+        return data.iterator();
     }
 
     @Override
@@ -141,7 +188,12 @@ public class IntColumn extends NumberColumn<Integer> implements NumericColumn<In
 
     @Override
     public IntColumn set(int i, Integer val) {
-        return (IntColumn) set(i, (int) val);
+        return set(i, (int) val);
+    }
+
+    public IntColumn set(int i, int val) {
+        data.set(i, val);
+        return this;
     }
 
     @Override
@@ -154,4 +206,116 @@ public class IntColumn extends NumberColumn<Integer> implements NumericColumn<In
         return this;
     }
 
+    @Override
+    public IntColumn appendMissing() {
+        return append(IntColumnType.missingValueIndicator());
+    }
+
+    @Override
+    public byte[] asBytes(int rowNumber) {
+        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putInt(getInt(rowNumber)).array();
+    }
+
+    @Override
+    public int countUnique() {
+        IntSet uniqueElements = new IntOpenHashSet();
+        for (int i = 0; i < size(); i++) {
+            if (!isMissingValue(getInt(i))) {
+                uniqueElements.add(getInt(i));
+            }
+        }
+        return uniqueElements.size();
+    }
+
+    /**
+     * Returns the value at the given index. The actual value is returned if the ColumnType is INTEGER. Otherwise the
+     * value is rounded as described below.
+     *
+     * Returns the closest {@code int} to the argument, with ties
+     * rounding to positive infinity.
+     *
+     * <p>
+     * Special cases:
+     * <ul><li>If the argument is NaN, the result is 0.
+     * <li>If the argument is positive infinity or any value greater than or
+     * equal to the value of {@code Integer.MAX_VALUE}, an error will be thrown
+     *
+     * @param   row the index of the value to be rounded to an integer.
+     * @return  the value of the argument rounded to the nearest
+     *          {@code int} value.
+     * @throws  ClassCastException if the absolute value of the value to be rounded is too large to be cast to an int
+     */
+    public int getInt(int row) {
+        return data.getInt(row);
+    }
+
+    @Override
+    public double getDouble(int row) {
+        int value = data.getInt(row);
+        if (isMissingValue(value)) {
+            return DoubleColumnType.missingValueIndicator();
+        }
+        return value;
+    }
+
+    public boolean isMissingValue(int value) {
+        return IntColumnType.isMissingValue(value);
+    }
+
+    @Override
+    public boolean isMissing(int rowNumber) {
+        return isMissingValue(getInt(rowNumber));
+    }
+
+    @Override
+    public NumberIterator numberIterator() {
+        return new NumberIterator(data);
+    }
+
+    @Override
+    public void sortAscending() {
+        IntArrays.parallelQuickSort(data.elements());
+    }
+
+    @Override
+    public void sortDescending() {
+        IntArrays.parallelQuickSort(data.elements(), descendingComparator);
+    }
+
+    @Override
+    public IntColumn appendObj(Object obj) {
+        if (obj == null) {
+            return appendMissing();
+        }
+        if (obj instanceof Integer) {
+            return append((int) obj);
+        }
+        throw new IllegalArgumentException("Could not append " + obj.getClass());
+    }
+
+    @Override
+    public IntColumn appendCell(final String value) {
+        try {
+            return append(IntColumnType.DEFAULT_PARSER.parseInt(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name() + ": " + e.getMessage());
+        }
+    }
+
+    @Override
+    public IntColumn appendCell(final String value, StringParser<?> parser) {
+        try {
+            return append(parser.parseInt(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name()  + ": " + e.getMessage());
+        }
+    }
+
+    public int firstElement() {
+        if (size() > 0) {
+            return getInt(0);
+        }
+        return (Integer) COLUMN_TYPE.getMissingValueIndicator();
+    }
+
 }
diff --git a/core/src/main/java/tech/tablesaw/api/LongColumn.java b/core/src/main/java/tech/tablesaw/api/LongColumn.java
new file mode 100644
index 00000000..98738374
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/api/LongColumn.java
@@ -0,0 +1,348 @@
+package tech.tablesaw.api;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.util.Iterator;
+
+import com.google.common.base.Preconditions;
+
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrays;
+import it.unimi.dsi.fastutil.longs.LongComparator;
+import it.unimi.dsi.fastutil.longs.LongListIterator;
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongSet;
+import tech.tablesaw.columns.Column;
+import tech.tablesaw.columns.StringParser;
+import tech.tablesaw.columns.numbers.DoubleColumnType;
+import tech.tablesaw.columns.numbers.LongColumnType;
+import tech.tablesaw.columns.numbers.NumberColumnFormatter;
+import tech.tablesaw.columns.numbers.NumberIterator;
+
+public class LongColumn extends NumberColumn<Long> implements NumericColumn<Long>, CategoricalColumn<Long> {
+
+    private static final LongColumnType COLUMN_TYPE = ColumnType.LONG;
+
+    /**
+     * Compares two ints, such that a sort based on this comparator would sort in descending order
+     */
+    private final LongComparator descendingComparator = (o2, o1) -> (Long.compare(o1, o2));
+
+    private final LongArrayList data;
+
+    protected LongColumn(final String name, LongArrayList data) {
+        super(COLUMN_TYPE, name, data);
+        this.printFormatter = NumberColumnFormatter.ints();
+        this.data = data;
+    }
+
+    public static LongColumn create(final String name) {
+        return new LongColumn(name, new LongArrayList());
+    }
+
+    public static LongColumn create(final String name, final long[] arr) {
+        return new LongColumn(name, new LongArrayList(arr));
+    }
+
+    public static LongColumn create(final String name, final int initialSize) {
+        return new LongColumn(name, new LongArrayList(initialSize));
+    }
+
+    @Override
+    public LongColumn createCol(final String name, final int initialSize) {
+        return create(name, initialSize);
+    }
+
+    /**
+     * Returns a new numeric column initialized with the given name and size. The values in the column are
+     * integers beginning at startsWith and continuing through size (exclusive), monotonically increasing by 1
+     * TODO consider a generic fill function including steps or random samples from various distributions
+     */
+    public static LongColumn indexColumn(final String columnName, final int size, final int startsWith) {
+        final LongColumn indexColumn = LongColumn.create(columnName, size);
+        for (int i = 0; i < size; i++) {
+            indexColumn.append(i + startsWith);
+        }
+        return indexColumn;
+    }
+
+    public static boolean valueIsMissing(long value) {
+        return value == LongColumnType.missingValueIndicator();
+    }
+
+    @Override
+    public Long get(int index) {
+        return getLong(index);
+    }
+
+    @Override
+    public LongColumn unique() {
+        final LongSet values = new LongOpenHashSet();
+        for (int i = 0; i < size(); i++) {
+            if (!isMissing(i)) {
+                values.add(getLong(i));
+            }
+        }
+        final LongColumn column = LongColumn.create(name() + " Unique values", values.size());
+        for (long value : values) {
+            column.append(value);
+        }
+        return column;
+    }
+
+    @Override
+    public LongColumn top(int n) {
+        final LongArrayList top = new LongArrayList();
+        final long[] values = data.toLongArray();
+        LongArrays.parallelQuickSort(values, descendingComparator);
+        for (int i = 0; i < n && i < values.length; i++) {
+            top.add(values[i]);
+        }
+        return new LongColumn(name() + "[Top " + n  + "]", top);
+    }
+
+    @Override
+    public LongColumn bottom(final int n) {
+        final LongArrayList bottom = new LongArrayList();
+        final long[] values = data.toLongArray();
+        LongArrays.parallelQuickSort(values);
+        for (int i = 0; i < n && i < values.length; i++) {
+            bottom.add(values[i]);
+        }
+        return new LongColumn(name() + "[Bottoms " + n  + "]", bottom);
+    }
+
+    @Override
+    public LongColumn lag(int n) {
+        final int srcPos = n >= 0 ? 0 : 0 - n;
+        final long[] dest = new long[size()];
+        final int destPos = n <= 0 ? 0 : n;
+        final int length = n >= 0 ? size() - n : size() + n;
+
+        for (int i = 0; i < size(); i++) {
+            dest[i] = LongColumnType.missingValueIndicator();
+        }
+
+        long[] array = data.toLongArray();
+
+        System.arraycopy(array, srcPos, dest, destPos, length);
+        return new LongColumn(name() + " lag(" + n + ")", new LongArrayList(dest));
+    }
+
+    @Override
+    public LongColumn removeMissing() {
+        LongColumn result = copy();
+        result.clear();
+        LongListIterator iterator = data.iterator();
+        while (iterator.hasNext()) {
+            final long v = iterator.nextLong();
+            if (!isMissingValue(v)) {
+                result.append(v);
+            }
+        }
+        return result;
+    }
+
+    public LongColumn append(long i) {
+        data.add(i);
+        return this;
+    }
+
+    public LongColumn append(Long val) {
+        this.append(val.longValue());
+        return this;
+    }
+
+    @Override
+    public LongColumn emptyCopy() {
+        return (LongColumn) super.emptyCopy();
+    }
+
+    @Override
+    public LongColumn emptyCopy(final int rowSize) {
+        return (LongColumn) super.emptyCopy(rowSize);
+    }
+
+    @Override
+    public LongColumn copy() {
+        return new LongColumn(name(), data.clone());
+    }
+
+    public long[] asLongArray() {  // TODO: Need to figure out how to handle NaN -> Maybe just use a list with nulls?
+        final long[] result = new long[size()];
+        for (int i = 0; i < size(); i++) {
+            result[i] = getLong(i);
+        }
+        return result;
+    }
+
+    /**
+     * Returns a DateTimeColumn where each value is the LocalDateTime represented by the values in this column
+     * <p>
+     * The values in this column must be longs that represent the time in milliseconds from the epoch as in standard
+     * Java date/time calculations
+     *
+     * @param offset The ZoneOffset to use in the calculation
+     * @return A column of LocalDateTime values
+     */
+    public DateTimeColumn asDateTimes(ZoneOffset offset) {
+        DateTimeColumn column = DateTimeColumn.create(name() + ": date time");
+        for (int i = 0; i < size(); i++) {
+            column.append(Instant.ofEpochMilli(getLong(i)).atZone(offset).toLocalDateTime());
+        }
+        return column;
+    }
+
+    @Override
+    public Iterator<Long> iterator() {
+        return data.iterator();
+    }
+
+    @Override
+    public Object[] asObjectArray() {
+        final Long[] output = new Long[size()];
+        for (int i = 0; i < size(); i++) {
+            output[i] = getLong(i);
+        }
+        return output;
+    }
+
+    @Override
+    public int compare(Long o1, Long o2) {
+        return Long.compare(o1, o2);
+    }
+
+    @Override
+    public LongColumn set(int i, Long val) {
+        return set(i, (long) val);
+    }
+
+    public LongColumn set(int i, long val) {
+        data.set(i, val);
+        return this;
+    }
+
+    @Override
+    public LongColumn append(final Column<Long> column) {
+        Preconditions.checkArgument(column.type() == this.type());
+        final LongColumn numberColumn = (LongColumn) column;
+        for (int i = 0; i < numberColumn.size(); i++) {
+            append(numberColumn.getLong(i));
+        }
+        return this;
+    }
+
+    @Override
+    public LongColumn appendMissing() {
+        return append(LongColumnType.missingValueIndicator());
+    }
+
+    @Override
+    public byte[] asBytes(int rowNumber) {
+        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putLong(getLong(rowNumber)).array();
+    }
+
+    @Override
+    public int countUnique() {
+        LongSet uniqueElements = new LongOpenHashSet();
+        for (int i = 0; i < size(); i++) {
+            if (!isMissingValue(getLong(i))) {
+                uniqueElements.add(getLong(i));
+            }
+        }
+        return uniqueElements.size();
+    }
+
+    /**
+     * Returns the value at the given index. The actual value is returned if the ColumnType is INTEGER. Otherwise the
+     * value is rounded as described below.
+     *
+     * Returns the closest {@code int} to the argument, with ties
+     * rounding to positive infinity.
+     *
+     * <p>
+     * Special cases:
+     * <ul><li>If the argument is NaN, the result is 0.
+     * <li>If the argument is positive infinity or any value greater than or
+     * equal to the value of {@code Integer.MAX_VALUE}, an error will be thrown
+     *
+     * @param   row the index of the value to be rounded to an integer.
+     * @return  the value of the argument rounded to the nearest
+     *          {@code int} value.
+     * @throws  ClassCastException if the absolute value of the value to be rounded is too large to be cast to an int
+     */
+    public long getLong(int row) {
+        return data.getLong(row);
+    }
+
+    @Override
+    public double getDouble(int row) {
+        long value = data.getLong(row);
+        if (isMissingValue(value)) {
+            return DoubleColumnType.missingValueIndicator();
+        }
+        return value;
+    }
+
+    public boolean isMissingValue(long value) {
+        return LongColumnType.isMissingValue(value);
+    }
+
+    @Override
+    public boolean isMissing(int rowNumber) {
+        return isMissingValue(getLong(rowNumber));
+    }
+
+    @Override
+    public NumberIterator numberIterator() {
+        return new NumberIterator(data);
+    }
+
+    @Override
+    public void sortAscending() {
+        LongArrays.parallelQuickSort(data.elements());
+    }
+
+    @Override
+    public void sortDescending() {
+        LongArrays.parallelQuickSort(data.elements(), descendingComparator);
+    }
+
+    @Override
+    public LongColumn appendObj(Object obj) {
+        if (obj == null) {
+            return appendMissing();
+        }
+        if (obj instanceof Long) {
+            return append((long) obj);
+        }
+        throw new IllegalArgumentException("Could not append " + obj.getClass());
+    }
+
+    @Override
+    public LongColumn appendCell(final String value) {
+        try {
+            return append(LongColumnType.DEFAULT_PARSER.parseLong(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name() + ": " + e.getMessage());
+        }
+    }
+
+    @Override
+    public LongColumn appendCell(final String value, StringParser<?> parser) {
+        try {
+            return append(parser.parseInt(value));
+        } catch (final NumberFormatException e) {
+            throw new NumberFormatException("Error adding value to column " + name()  + ": " + e.getMessage());
+        }
+    }
+
+    public long firstElement() {
+        if (size() > 0) {
+            return getLong(0);
+        }
+        return (Long) COLUMN_TYPE.getMissingValueIndicator();
+    }
+
+}
diff --git a/core/src/main/java/tech/tablesaw/api/NumberColumn.java b/core/src/main/java/tech/tablesaw/api/NumberColumn.java
index 06d0bfb6..10d48734 100644
--- a/core/src/main/java/tech/tablesaw/api/NumberColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/NumberColumn.java
@@ -1,40 +1,23 @@
 package tech.tablesaw.api;
 
-import static tech.tablesaw.api.ColumnType.DOUBLE;
-import static tech.tablesaw.api.ColumnType.FLOAT;
-import static tech.tablesaw.api.ColumnType.INTEGER;
-
 import java.text.NumberFormat;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
+import java.util.List;
 import java.util.Locale;
 import java.util.function.DoublePredicate;
 
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
-import it.unimi.dsi.fastutil.floats.FloatArrayList;
-import it.unimi.dsi.fastutil.ints.IntArrayList;
 import it.unimi.dsi.fastutil.ints.IntComparator;
-import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
-import it.unimi.dsi.fastutil.ints.IntSet;
 import tech.tablesaw.columns.AbstractColumn;
-import tech.tablesaw.columns.StringParser;
-import tech.tablesaw.columns.numbers.DoubleDataWrapper;
-import tech.tablesaw.columns.numbers.FloatDataWrapper;
-import tech.tablesaw.columns.numbers.IntDataWrapper;
+import tech.tablesaw.columns.numbers.DoubleColumnType;
 import tech.tablesaw.columns.numbers.NumberColumnFormatter;
-import tech.tablesaw.columns.numbers.NumberIterator;
-import tech.tablesaw.columns.numbers.NumericDataWrapper;
-import tech.tablesaw.selection.Selection;
 
 public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> implements NumericColumn<T> {
 
-    protected NumericDataWrapper data;
-
     protected NumberColumnFormatter printFormatter = new NumberColumnFormatter();
 
     protected Locale locale;
 
+    private List<T> data;
+
     protected final IntComparator comparator = new IntComparator() {
 
         @Override
@@ -45,44 +28,12 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
         }
     };
 
-    protected NumberColumn(final ColumnType type, final String name) {
+    protected NumberColumn(final ColumnType type, final String name, List<T> data) {
         super(type, name);
+        this.data = data;
     }
 
-    protected NumberColumn(final String name, final DoubleArrayList data) {
-        super(DOUBLE, name);
-        setDataWrapper(new DoubleDataWrapper(data));
-    }
-
-    protected NumberColumn(final String name, final FloatArrayList data) {
-        super(FLOAT, name);
-        setDataWrapper(new FloatDataWrapper(data));
-    }
-
-    protected NumberColumn(final String name, IntArrayList data) {
-        super(INTEGER, name);
-        this.printFormatter = NumberColumnFormatter.ints();
-        setDataWrapper(new IntDataWrapper(data));
-    }
-
-    protected NumberColumn(final String name, final NumericDataWrapper data) {
-        super(data.type(), name);
-        setDataWrapper(data);
-    }
-
-    protected abstract NumberColumn<T> createCol(final String name, final NumericDataWrapper data);
-
-    protected void setDataWrapper(NumericDataWrapper wrapper) {
-        if (wrapper instanceof IntDataWrapper) {
-            printFormatter = NumberColumnFormatter.ints();
-        }
-        this.data = wrapper;
-    }
-
-    @Override
-    public boolean isMissing(final int rowNumber) {
-        return data.isMissingValue(getDouble(rowNumber));
-    }
+    protected abstract NumberColumn<T> createCol(final String name, int size);
 
     public void setPrintFormatter(final NumberFormat format, final String missingValueString) {
         this.printFormatter = new NumberColumnFormatter(format, missingValueString);
@@ -92,11 +43,6 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
         this.printFormatter = formatter;
     }
 
-    @Override
-    public int size() {
-        return data.size();
-    }
-
     /**
      * Returns the largest ("top") n values in the column
      * TODO(lwhite): Consider whether this should exclude missing
@@ -105,9 +51,7 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
      *          number of observations in the column
      * @return A list, possibly empty, of the largest observations
      */
-    public NumericColumn<T> top(final int n) {
-        return createCol(name() + "[Top " + n  + "]", data.top(n));
-    }
+    public abstract NumericColumn<T> top(final int n);
 
     /**
      * Returns the smallest ("bottom") n values in the column
@@ -117,59 +61,25 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
      *          number of observations in the column
      * @return A list, possibly empty, of the smallest n observations
      */
-    public NumericColumn<T> bottom(final int n) {
-        return createCol(name() + "[Bottoms " + n  + "]", data.bottom(n));
-    }
+    public abstract NumericColumn<T> bottom(final int n);
 
     @Override
     public String getString(final int row) {
         final double value = getDouble(row);
-        if (data.isMissingValue(value)) {
+        if (DoubleColumnType.isMissingValue(value)) {
             return "";
         }
         return String.valueOf(printFormatter.format(value));
     }
 
     @Override
-    public double getDouble(final int row) {
-        return data.getDouble(row);
-    }
-
-    /**
-     * Returns the value at the given index. The actual value is returned if the ColumnType is INTEGER. Otherwise the
-     * value is rounded as described below.
-     *
-     * Returns the closest {@code int} to the argument, with ties
-     * rounding to positive infinity.
-     *
-     * <p>
-     * Special cases:
-     * <ul><li>If the argument is NaN, the result is 0.
-     * <li>If the argument is positive infinity or any value greater than or
-     * equal to the value of {@code Integer.MAX_VALUE}, an error will be thrown
-     *
-     * @param   row the index of the value to be rounded to an integer.
-     * @return  the value of the argument rounded to the nearest
-     *          {@code int} value.
-     * @throws  ClassCastException if the absolute value of the value to be rounded is too large to be cast to an int
-     */
-    public int getInt(final int row) {
-        return data.getInt(row);
-    }
-
-    /**
-     * Returns a float representation of the data at the given index. Some precision may be lost, and if the value is
-     * to large to be cast to a float, an exception is thrown.
-     *
-     * @throws  ClassCastException if the value can't be cast to ta float
-     */
-    public float getFloat(final int index) {
-        return data.getFloat(index);
+    public int size() {
+        return data.size();
     }
 
     @Override
     public NumberColumn<T> emptyCopy() {
-        final NumberColumn<T> column = createCol(name(), data.emptyCopy(DEFAULT_ARRAY_SIZE));
+        final NumberColumn<T> column = createCol(name(), DEFAULT_ARRAY_SIZE);
         column.setPrintFormatter(printFormatter);
         column.locale = locale;
         return column;
@@ -177,71 +87,13 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
 
     @Override
     public NumberColumn<T> emptyCopy(final int rowSize) {
-        final NumberColumn<T> column = createCol(name(), data.emptyCopy(rowSize));
+        final NumberColumn<T> column = createCol(name(), rowSize);
         column.setPrintFormatter(printFormatter);
         column.locale = locale;
         return column;
     }
 
-    @Override
-    public NumberColumn<T> copy() {
-        return createCol(name(), data.copy());
-    }
-
-    @Override
-    public void clear() {
-        data.clear();
-    }
-
-    @Override
-    public void sortAscending() {
-        data.sortAscending();
-    }
-
-    @Override
-    public void sortDescending() {
-        data.sortDescending();
-    }
-
-    @Override
-    public NumberColumn<T> appendCell(final String object) {
-        try {
-            data.appendCell(object);
-        } catch (final NumberFormatException e) {
-            throw new NumberFormatException("Error adding value to column " + name() + ": " + e.getMessage());
-        }
-        return this;
-    }
-
-    @Override
-    public NumberColumn<T> appendCell(final String object, StringParser<?> parser) {
-        try {
-            data.appendCell(object, parser);
-        } catch (final NumberFormatException e) {
-            throw new NumberFormatException("Error adding value to column " + name()  + ": " + e.getMessage());
-        }
-        return this;
-    }
-
-    /**
-     * Returns the rounded value as an int
-     *
-     * @throws ClassCastException if the returned value will not fit in an int
-     */
-    public Integer roundInt(final int i) {
-        return getInt(i);
-    }
-
-    /**
-     * Returns the value of the ith element rounded to the nearest long
-     *
-     * @param i the index in the column
-     * @return the value at i, rounded to the nearest integer
-     */
-    public long getLong(final int i) {
-        final double value = getDouble(i);
-        return data.isMissingValue(value) ? DateTimeColumn.MISSING_VALUE : Math.round(value);
-    }
+    public abstract NumberColumn<T> copy();
 
     /**
      * Compares the given ints, which refer to the indexes of the doubles in this column, according to the values of the
@@ -252,40 +104,12 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
         return comparator;
     }
 
-    public NumberColumn<T> set(final int r, final double value) {
-        data.set(r, value);
-        return this;
-    }
-
-    public NumberColumn<T> set(final int r, final int value) {
-        data.set(r, value);
-        return this;
-    }
-
-    public NumberColumn<T> set(final int r, final float value) {
-        data.set(r, value);
-        return this;
-    }
-
-    /**
-     * Conditionally update this column, replacing current values with newValue for all rows where the current value
-     * matches the selection criteria
-     * <p>
-     * Example:
-     * myColumn.set(4.0, myColumn.valueIsMissing()); // no more missing values
-     */
-    public NumberColumn<T> set(final Selection rowSelection, final double newValue) {
-        for (final int row : rowSelection) {
-            set(row, newValue);
-        }
-        return this;
-    }
-
-    public boolean contains(final double value) {
-        return data.contains(value);
+    @Override
+    public void clear() {
+        data.clear();
     }
-
-    public boolean contains(final int value) {
+    
+    public boolean contains(T value) {
         return data.contains(value);
     }
 
@@ -298,84 +122,23 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
      * Returns the contents of the cell at rowNumber as a byte[]
      */
     @Override
-    public byte[] asBytes(final int rowNumber) {
-        return data.asBytes(rowNumber);
-    }
+    public abstract byte[] asBytes(final int rowNumber);
 
     @Override
-    public NumberColumn<T> appendMissing() {
-        data.appendMissing();
-        return this;
-    }
+    public abstract NumberColumn<T> appendMissing();
 
     /**
      * Returns the count of missing values in this column
      */
     @Override
     public int countMissing() {
-        return data.countMissing();
-    }
-
-    /**
-     * Returns a DateTimeColumn where each value is the LocalDateTime represented by the values in this column
-     * <p>
-     * The values in this column must be longs that represent the time in milliseconds from the epoch as in standard
-     * Java date/time calculations
-     *
-     * @param offset The ZoneOffset to use in the calculation
-     * @return A column of LocalDateTime values
-     */
-    public DateTimeColumn asDateTimes(ZoneOffset offset) {
-        DateTimeColumn column = DateTimeColumn.create(name() + ": date time");
-        NumberIterator it = numberIterator();
-        while (it.hasNext()) {
-            double d = it.next();
-            LocalDateTime dateTime =
-                    Instant.ofEpochMilli((long) d).atZone(offset).toLocalDateTime();
-            column.append(dateTime);
-        }
-        return column;
-    }
-
-    /**
-     * Returns the number of unique values in this column, excluding missing values
-     */
-    @Override
-    public int countUnique() {
-        return data.countUnique();
-    }
-
-    @Override
-    public NumberColumn<T> appendObj(Object obj) {
-        data.appendObj(obj);
-        return this;
-    }
-
-    @Override
-    public NumberColumn<T> lag(final int n) {
-        return createCol(name() + " lag(" + n + ")", data.lag(n));
-    }
-
-    @Override
-    public NumberColumn<T> removeMissing() {
-        return createCol(name(), data.removeMissing());
-    }
-
-    @Override
-    public NumberIterator numberIterator() {
-        return data.numberIterator();
-    }
-
-    public IntSet asIntegerSet() {
-        final IntSet ints = new IntOpenHashSet();
-        NumberIterator it = numberIterator();
-        while (it.hasNext()) {
-            double d = it.next();
-            if (!data.isMissingValue(d)) {
-                ints.add((int) Math.round(d));
+        int count = 0;
+        for (int i = 0; i < size(); i++) {
+            if (isMissing(i)) {
+                count++;
             }
         }
-        return ints;
+        return count;
     }
 
     @Override
@@ -416,5 +179,4 @@ public abstract class NumberColumn<T extends Number> extends AbstractColumn<T> i
     public boolean noneMatch(DoublePredicate test) {
         return count(test, 1) == 0;
     }
-
 }
diff --git a/core/src/main/java/tech/tablesaw/api/NumericColumn.java b/core/src/main/java/tech/tablesaw/api/NumericColumn.java
index 5fa43711..491de676 100644
--- a/core/src/main/java/tech/tablesaw/api/NumericColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/NumericColumn.java
@@ -49,9 +49,6 @@ import tech.tablesaw.selection.Selection;
 
 public interface NumericColumn<T> extends Column<T>, NumberMapFunctions, NumberFilters {
 
-    @Override
-    T get(final int index);
-
     @Override
     default boolean isEmpty() {
         return size() == 0;
diff --git a/core/src/main/java/tech/tablesaw/api/Row.java b/core/src/main/java/tech/tablesaw/api/Row.java
index c45ccee7..375fb4fc 100644
--- a/core/src/main/java/tech/tablesaw/api/Row.java
+++ b/core/src/main/java/tech/tablesaw/api/Row.java
@@ -18,6 +18,7 @@ public class Row implements Iterator<Row> {
     private final Map<String, DateColumn> dateColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
     private final Map<String, DoubleColumn> doubleColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
     private final Map<String, IntColumn> intColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+    private final Map<String, LongColumn> longColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
     private final Map<String, FloatColumn> floatColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
     private final Map<String, StringColumn> stringColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
     private final Map<String, BooleanColumn> booleanColumnMap = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
@@ -37,6 +38,9 @@ public class Row implements Iterator<Row> {
             if (column instanceof IntColumn) {
                 intColumnMap.put(column.name(), (IntColumn) column);
             }
+            if (column instanceof LongColumn) {
+                longColumnMap.put(column.name(), (LongColumn) column);
+            }
             if (column instanceof FloatColumn) {
                 floatColumnMap.put(column.name(), (FloatColumn) column);
             }
@@ -97,11 +101,19 @@ public class Row implements Iterator<Row> {
         return getInt(columnNames[columnIndex]);
     }
 
-    public int getFloat(String columnName) {
-	return floatColumnMap.get(columnName).getInt(rowNumber);
+    public long getLong(String columnName) {
+	return longColumnMap.get(columnName).getLong(rowNumber);
+    }
+
+    public long getLong(int columnIndex) {
+        return getLong(columnNames[columnIndex]);
+    }
+
+    public float getFloat(String columnName) {
+	return floatColumnMap.get(columnName).getFloat(rowNumber);
     }
 
-    public int getFloat(int columnIndex) {
+    public float getFloat(int columnIndex) {
         return getFloat(columnNames[columnIndex]);
     }
     
diff --git a/core/src/main/java/tech/tablesaw/api/StringColumn.java b/core/src/main/java/tech/tablesaw/api/StringColumn.java
index f4e6eb90..079629c8 100644
--- a/core/src/main/java/tech/tablesaw/api/StringColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/StringColumn.java
@@ -248,11 +248,6 @@ public class StringColumn extends AbstractColumn<String>
         return t;
     }
 
-    @Override
-    public int[] asIntArray() {
-        return data().toArray(new int[size()]);
-    }
-
     @Override
     public void clear() {
         values.clear();
diff --git a/core/src/main/java/tech/tablesaw/api/TimeColumn.java b/core/src/main/java/tech/tablesaw/api/TimeColumn.java
index e5239325..ec8edd13 100644
--- a/core/src/main/java/tech/tablesaw/api/TimeColumn.java
+++ b/core/src/main/java/tech/tablesaw/api/TimeColumn.java
@@ -601,11 +601,6 @@ public class TimeColumn extends AbstractColumn<LocalTime>
         return this;
     }
 
-    @Override
-    public int[] asIntArray() {
-        return data.toArray(new int[size()]);
-    }
-
     @Override
     public int compare(LocalTime o1, LocalTime o2) {
         return o1.compareTo(o2);
diff --git a/core/src/main/java/tech/tablesaw/columns/Column.java b/core/src/main/java/tech/tablesaw/columns/Column.java
index 93cf31fd..9c18af04 100644
--- a/core/src/main/java/tech/tablesaw/columns/Column.java
+++ b/core/src/main/java/tech/tablesaw/columns/Column.java
@@ -298,9 +298,9 @@ public interface Column<T> extends Iterable<T>, Comparator<T> {
         Column<T> newCol = emptyCopy();
         for (int i = 0; i < this.size(); i++) {
             if (isMissing(i)) {
-                newCol.appendCell(other.getUnformattedString(i));
+                newCol.appendObj(other.get(i));
             } else {
-                newCol.appendCell(getUnformattedString(i));
+                newCol.appendObj(get(i));
             }
         }
         return newCol;
@@ -313,9 +313,9 @@ public interface Column<T> extends Iterable<T>, Comparator<T> {
         Column<T> newCol = emptyCopy();
         for (int i = 0; i < this.size(); i++) {
             if (isMissing(i)) {
-                newCol.append(defaultVal);
+                newCol.appendObj(defaultVal);
             } else {
-                newCol.appendCell(getUnformattedString(i));
+                newCol.appendObj(get(i));
             }
         }
         return newCol;
diff --git a/core/src/main/java/tech/tablesaw/columns/StringParser.java b/core/src/main/java/tech/tablesaw/columns/StringParser.java
index ece46c51..312c3ef3 100644
--- a/core/src/main/java/tech/tablesaw/columns/StringParser.java
+++ b/core/src/main/java/tech/tablesaw/columns/StringParser.java
@@ -48,6 +48,10 @@ public abstract class StringParser<T> {
         throw new UnsupportedOperationException(this.getClass().getSimpleName() + " doesn't support parsing to ints");
     }
 
+    public long parseLong(String s) {
+        throw new UnsupportedOperationException(this.getClass().getSimpleName() + " doesn't support parsing to longs");
+    }
+
     public double parseDouble(String s) {
         throw new UnsupportedOperationException(this.getClass().getSimpleName() + " doesn't support parsing to doubles");
     }
diff --git a/core/src/main/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctions.java b/core/src/main/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctions.java
index cb51ddaf..139af5e4 100644
--- a/core/src/main/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctions.java
+++ b/core/src/main/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctions.java
@@ -14,60 +14,78 @@
 
 package tech.tablesaw.columns.datetimes;
 
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.asLocalDateTime;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.daysUntil;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getDayOfMonth;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getDayOfWeek;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getDayOfYear;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getHour;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getMinute;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getMinuteOfDay;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getMonthValue;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getQuarter;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getSecondOfDay;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getWeekOfYear;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.getYear;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.hoursUntil;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.minutesUntil;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.monthsUntil;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.pack;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.weeksUntil;
+import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.yearsUntil;
+
+import java.time.LocalDateTime;
+import java.time.Month;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.UnsupportedTemporalTypeException;
+
 import com.google.common.base.Strings;
+
 import tech.tablesaw.api.DateColumn;
 import tech.tablesaw.api.DateTimeColumn;
-import tech.tablesaw.api.DoubleColumn;
+import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.LongColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.TimeColumn;
 import tech.tablesaw.columns.Column;
 import tech.tablesaw.columns.dates.PackedLocalDate;
-import tech.tablesaw.columns.numbers.DoubleColumnType;
 import tech.tablesaw.columns.numbers.NumberColumnFormatter;
 
-import java.time.LocalDateTime;
-import java.time.Month;
-import java.time.temporal.ChronoUnit;
-import java.time.temporal.UnsupportedTemporalTypeException;
-
-import static tech.tablesaw.api.DateTimeColumn.MISSING_VALUE;
-import static tech.tablesaw.columns.datetimes.PackedLocalDateTime.*;
-
 public interface DateTimeMapFunctions extends Column<LocalDateTime> {
 
-    default DoubleColumn differenceInMilliseconds(DateTimeColumn column2) {
+    default LongColumn differenceInMilliseconds(DateTimeColumn column2) {
         return difference(column2, ChronoUnit.MILLIS);
     }
 
-    default DoubleColumn differenceInSeconds(DateTimeColumn column2) {
+    default LongColumn differenceInSeconds(DateTimeColumn column2) {
         return difference(column2, ChronoUnit.SECONDS);
     }
 
-    default DoubleColumn differenceInMinutes(DateTimeColumn column2) {
+    default LongColumn differenceInMinutes(DateTimeColumn column2) {
         return difference(column2, ChronoUnit.MINUTES);
     }
 
-    default DoubleColumn differenceInHours(DateTimeColumn column2) {
+    default LongColumn differenceInHours(DateTimeColumn column2) {
         return difference(column2, ChronoUnit.HOURS);
     }
 
-    default DoubleColumn differenceInDays(DateTimeColumn column2) {
+    default LongColumn differenceInDays(DateTimeColumn column2) {
         return difference(column2, ChronoUnit.DAYS);
     }
 
-    default DoubleColumn differenceInYears(DateTimeColumn column2) {
+    default LongColumn differenceInYears(DateTimeColumn column2) {
         return difference(column2, ChronoUnit.YEARS);
     }
 
-    default DoubleColumn difference(DateTimeColumn column2, ChronoUnit unit) {
-        DoubleColumn newColumn = DoubleColumn.create(name() + " - " + column2.name());
+    default LongColumn difference(DateTimeColumn column2, ChronoUnit unit) {
+	LongColumn newColumn = LongColumn.create(name() + " - " + column2.name());
 
         for (int r = 0; r < size(); r++) {
-            long c1 = this.getLongInternal(r);
-            long c2 = column2.getLongInternal(r);
-            if (c1 == MISSING_VALUE || c2 == MISSING_VALUE) {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+            if (this.isMissing(r) || column2.isMissing(r)) {
+                newColumn.appendMissing();
             } else {
+                long c1 = this.getLongInternal(r);
+                long c2 = column2.getLongInternal(r);
                 LocalDateTime value1 = asLocalDateTime(c1);
                 LocalDateTime value2 = asLocalDateTime(c2);
                 if (value1 != null && value2 != null) {
@@ -80,40 +98,40 @@ public interface DateTimeMapFunctions extends Column<LocalDateTime> {
         return newColumn;
     }
 
-    default DoubleColumn hour() {
-        DoubleColumn newColumn = DoubleColumn.create(name() + "[" + "hour" + "]");
+    default IntColumn hour() {
+	IntColumn newColumn = IntColumn.create(name() + "[" + "hour" + "]");
         for (int r = 0; r < size(); r++) {
-            long c1 = getLongInternal(r);
-            if (c1 != MISSING_VALUE) {
+            if (!isMissing(r)) {
+                long c1 = getLongInternal(r);
                 newColumn.append(getHour(c1));
             } else {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+                newColumn.appendMissing();
             }
         }
         return newColumn;
     }
 
-    default DoubleColumn minuteOfDay() {
-        DoubleColumn newColumn = DoubleColumn.create(name() + "[" + "minute-of-day" + "]");
+    default IntColumn minuteOfDay() {
+	IntColumn newColumn = IntColumn.create(name() + "[" + "minute-of-day" + "]");
         for (int r = 0; r < size(); r++) {
-            long c1 = getLongInternal(r);
-            if (c1 != MISSING_VALUE) {
+            if (!isMissing(r)) {
+                long c1 = getLongInternal(r);
                 newColumn.append((short) getMinuteOfDay(c1));
             } else {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+                newColumn.appendMissing();
             }
         }
         return newColumn;
     }
 
-    default DoubleColumn secondOfDay() {
-        DoubleColumn newColumn = DoubleColumn.create(name() + "[" + "second-of-day" + "]");
+    default IntColumn secondOfDay() {
+	IntColumn newColumn = IntColumn.create(name() + "[" + "second-of-day" + "]");
         for (int r = 0; r < size(); r++) {
-            long c1 = getLongInternal(r);
-            if (c1 != MISSING_VALUE) {
+            if (!isMissing(r)) {
+                long c1 = getLongInternal(r);
                 newColumn.append(getSecondOfDay(c1));
             } else {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+                newColumn.appendMissing();
             }
         }
         return newColumn;
@@ -145,13 +163,13 @@ public interface DateTimeMapFunctions extends Column<LocalDateTime> {
         return newColumn;
     }
 
-    default DoubleColumn monthValue() {
-        DoubleColumn newColumn = DoubleColumn.create(this.name() + " month");
+    default IntColumn monthValue() {
+	IntColumn newColumn = IntColumn.create(this.name() + " month");
         for (int r = 0; r < this.size(); r++) {
-            long c1 = getLongInternal(r);
-            if (DateTimeColumn.valueIsMissing(c1)) {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+            if (isMissing(r)) {
+                newColumn.appendMissing();
             } else {
+                long c1 = getLongInternal(r);
                 newColumn.append((short) getMonthValue(c1));
             }
         }
@@ -299,23 +317,23 @@ public interface DateTimeMapFunctions extends Column<LocalDateTime> {
     default DateColumn date() {
         DateColumn newColumn = DateColumn.create(this.name() + " date");
         for (int r = 0; r < this.size(); r++) {
-            long c1 = getLongInternal(r);
-            if (DateTimeColumn.valueIsMissing(c1)) {
-                newColumn.appendInternal(DateColumn.MISSING_VALUE);
+            if (isMissing(r)) {
+                newColumn.appendMissing();
             } else {
+                long c1 = getLongInternal(r);
                 newColumn.appendInternal(PackedLocalDateTime.date(c1));
             }
         }
         return newColumn;
     }
 
-    default DoubleColumn year() {
-        DoubleColumn newColumn = DoubleColumn.create(this.name() + " year");
+    default IntColumn year() {
+	IntColumn newColumn = IntColumn.create(this.name() + " year");
         for (int r = 0; r < this.size(); r++) {
-            long c1 = getLongInternal(r);
-            if (DateTimeColumn.valueIsMissing(c1)) {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+            if (isMissing(r)) {
+                newColumn.appendMissing();
             } else {
+                long c1 = getLongInternal(r);
                 newColumn.append(PackedLocalDate.getYear(PackedLocalDateTime.date(c1)));
             }
         }
@@ -335,39 +353,39 @@ public interface DateTimeMapFunctions extends Column<LocalDateTime> {
         return newColumn;
     }
 
-    default DoubleColumn dayOfWeekValue() {
-        DoubleColumn newColumn = DoubleColumn.create(this.name() + " day of week", this.size());
+    default IntColumn dayOfWeekValue() {
+	IntColumn newColumn = IntColumn.create(this.name() + " day of week", this.size());
         for (int r = 0; r < this.size(); r++) {
-            long c1 = this.getLongInternal(r);
-            if (DateTimeColumn.valueIsMissing(c1)) {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+            if (isMissing(r)) {
+                newColumn.appendMissing();
             } else {
+                long c1 = this.getLongInternal(r);
                 newColumn.append((short) getDayOfWeek(c1).getValue());
             }
         }
         return newColumn;
     }
 
-    default DoubleColumn dayOfYear() {
-        DoubleColumn newColumn = DoubleColumn.create(this.name() + " day of year", this.size());
+    default IntColumn dayOfYear() {
+	IntColumn newColumn = IntColumn.create(this.name() + " day of year", this.size());
         for (int r = 0; r < this.size(); r++) {
-            long c1 = this.getLongInternal(r);
-            if (DateTimeColumn.valueIsMissing(c1)) {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+            if (isMissing(r)) {
+                newColumn.appendMissing();
             } else {
+                long c1 = this.getLongInternal(r);
                 newColumn.append((short) getDayOfYear(c1));
             }
         }
         return newColumn;
     }
 
-    default DoubleColumn dayOfMonth() {
-        DoubleColumn newColumn = DoubleColumn.create(this.name() + " day of month");
+    default IntColumn dayOfMonth() {
+	IntColumn newColumn = IntColumn.create(this.name() + " day of month");
         for (int r = 0; r < this.size(); r++) {
-            long c1 = this.getLongInternal(r);
-            if (DateTimeColumn.valueIsMissing(c1)) {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+            if (isMissing(r)) {
+                newColumn.isMissing();
             } else {
+                long c1 = this.getLongInternal(r);
                 newColumn.append(getDayOfMonth(c1));
             }
         }
@@ -384,10 +402,10 @@ public interface DateTimeMapFunctions extends Column<LocalDateTime> {
      * @param n     The number of units in each group.
      * @param start The starting point of the first group; group boundaries are offsets from this point
      */
-    default DoubleColumn timeWindow(ChronoUnit unit, int n, LocalDateTime start) {
+    default LongColumn timeWindow(ChronoUnit unit, int n, LocalDateTime start) {
         String newColumnName = "" +  n + " " + unit.toString() + " window [" + name() + "]";
         long packedStartDate = pack(start);
-        DoubleColumn numberColumn = DoubleColumn.create(newColumnName, size());
+        LongColumn numberColumn = LongColumn.create(newColumnName, size());
         for (int i = 0; i < size(); i++) {
             long packedDate = getLongInternal(i);
             long result;
@@ -419,20 +437,20 @@ public interface DateTimeMapFunctions extends Column<LocalDateTime> {
         return numberColumn;
     }
 
-    default DoubleColumn minute() {
-        DoubleColumn newColumn = DoubleColumn.create(name() + "[" + "minute" + "]");
+    default IntColumn minute() {
+	IntColumn newColumn = IntColumn.create(name() + "[" + "minute" + "]");
         for (int r = 0; r < size(); r++) {
-            long c1 = getLongInternal(r);
-            if (!DateTimeColumn.valueIsMissing(c1)) {
+            if (!isMissing(r)) {
+                long c1 = getLongInternal(r);
                 newColumn.append(getMinute(c1));
             } else {
-                newColumn.append(DoubleColumnType.missingValueIndicator());
+                newColumn.appendMissing();
             }
         }
         return newColumn;
     }
 
-    default DoubleColumn timeWindow(ChronoUnit unit, int n) {
+    default LongColumn timeWindow(ChronoUnit unit, int n) {
         return timeWindow(unit, n, min());
     }
 
diff --git a/core/src/main/java/tech/tablesaw/columns/datetimes/PackedLocalDateTime.java b/core/src/main/java/tech/tablesaw/columns/datetimes/PackedLocalDateTime.java
index bde6da61..a4d7c6b2 100644
--- a/core/src/main/java/tech/tablesaw/columns/datetimes/PackedLocalDateTime.java
+++ b/core/src/main/java/tech/tablesaw/columns/datetimes/PackedLocalDateTime.java
@@ -392,7 +392,7 @@ public class PackedLocalDateTime {
         return PackedLocalTime.getSecond(time(packedLocalDateTime));
     }
 
-    public static double getSecondOfDay(long packedLocalDateTime) {
+    public static int getSecondOfDay(long packedLocalDateTime) {
         return PackedLocalTime.getSecondOfDay(time(packedLocalDateTime));
     }
 
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/DoubleDataWrapper.java b/core/src/main/java/tech/tablesaw/columns/numbers/DoubleDataWrapper.java
deleted file mode 100644
index e62a7878..00000000
--- a/core/src/main/java/tech/tablesaw/columns/numbers/DoubleDataWrapper.java
+++ /dev/null
@@ -1,245 +0,0 @@
-package tech.tablesaw.columns.numbers;
-
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
-import it.unimi.dsi.fastutil.doubles.DoubleArrays;
-import it.unimi.dsi.fastutil.doubles.DoubleComparator;
-import it.unimi.dsi.fastutil.doubles.DoubleOpenHashSet;
-import it.unimi.dsi.fastutil.doubles.DoubleSet;
-import tech.tablesaw.api.ColumnType;
-import tech.tablesaw.columns.StringParser;
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Iterator;
-
-public class DoubleDataWrapper implements NumericDataWrapper {
-
-    private static final ColumnType COLUMN_TYPE = ColumnType.DOUBLE;
-
-    /**
-     * Compares two doubles, such that a sort based on this comparator would sort in descending order
-     */
-    private final DoubleComparator descendingComparator = (o2, o1) -> (Double.compare(o1, o2));
-
-    private final DoubleArrayList data;
-
-    public DoubleDataWrapper(DoubleArrayList data) {
-        this.data = data;
-    }
-
-    @Override
-    public NumberIterator numberIterator() {
-        return new NumberIterator(data);
-    }
-
-    @Override
-    public Iterator<Double> iterator() {
-        return data.iterator();
-    }
-
-    @Override
-    public NumericDataWrapper top(int n) {
-        final DoubleArrayList top = new DoubleArrayList();
-        final double[] values = data.toDoubleArray();
-        DoubleArrays.parallelQuickSort(values, descendingComparator);
-        for (int i = 0; i < n && i < values.length; i++) {
-            top.add(values[i]);
-        }
-        return new DoubleDataWrapper(top);
-    }
-
-    @Override
-    public NumericDataWrapper bottom(int n) {
-
-        final DoubleArrayList bottom = new DoubleArrayList();
-        final double[] values = data.toDoubleArray();
-        DoubleArrays.parallelQuickSort(values);
-        for (int i = 0; i < n && i < values.length; i++) {
-            bottom.add(values[i]);
-        }
-        return new DoubleDataWrapper(bottom);
-    }
-
-    @Override
-    public int size() {
-        return data.size();
-    }
-
-    public void append(final float f) {
-        data.add(f);
-    }
-
-    @Override
-    public void append(double d) {
-        data.add(d);
-    }
-
-    @Override
-    public void append(int i) {
-        data.add(i);
-    }
-
-    @Override
-    public void appendCell(String value) {
-        append(DoubleColumnType.DEFAULT_PARSER.parseDouble(value));
-    }
-
-    @Override
-    public void appendCell(String value, StringParser<?> parser) {
-        append(parser.parseDouble(value));
-    }
-
-    @Override
-    public int countMissing() {
-        int count = 0;
-        for (int i = 0; i < size(); i++) {
-            if (isMissingValue(getDouble(i))) {
-                count++;
-            }
-        }
-        return count;
-    }
-
-    @Override
-    public double getDouble(int row) {
-        return data.getDouble(row);
-    }
-
-    @Override
-    public float getFloat(int row) {
-        return (float) data.getDouble(row);
-    }
-
-    @Override
-    public DoubleDataWrapper copy() {
-        return new DoubleDataWrapper(data.clone());
-    }
-
-    @Override
-    public DoubleDataWrapper emptyCopy() {
-        return new DoubleDataWrapper(new DoubleArrayList());
-    }
-
-    @Override
-    public DoubleDataWrapper emptyCopy(int size) {
-        return new DoubleDataWrapper(new DoubleArrayList(size));
-    }
-
-    @Override
-    public void clear() {
-        this.data.clear();
-    }
-
-    @Override
-    public void sortAscending() {
-        Arrays.parallelSort(data.elements());
-    }
-
-    @Override
-    public void sortDescending() {
-        DoubleArrays.parallelQuickSort(data.elements(), descendingComparator);
-    }
-
-    @Override
-    public void set(int r, double value) {
-        this.data.set(r, value);
-    }
-
-    @Override
-    public boolean contains(double value) {
-        return data.contains(value);
-    }
-
-    @Override
-    public boolean contains(int value) {
-        return data.contains(value);
-    }
-
-    @Override
-    public NumericDataWrapper lag(int n) {
-        final int srcPos = n >= 0 ? 0 : 0 - n;
-        final double[] dest = new double[size()];
-        final int destPos = n <= 0 ? 0 : n;
-        final int length = n >= 0 ? size() - n : size() + n;
-
-        for (int i = 0; i < size(); i++) {
-            dest[i] = DoubleColumnType.missingValueIndicator();
-        }
-
-        double[] array = data.toDoubleArray();
-
-        System.arraycopy(array, srcPos, dest, destPos, length);
-        return new DoubleDataWrapper(new DoubleArrayList(dest));
-    }
-
-    @Override
-    public NumericDataWrapper lead(int n) {
-        return lag(-n);
-    }
-
-    @Override
-    public NumericDataWrapper removeMissing() {
-        NumericDataWrapper wrapper = copy();
-        wrapper.clear();
-        ;
-        final NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            final double v = iterator.next();
-            if (!isMissingValue(v)) {
-                wrapper.append(v);
-            }
-        }
-        return wrapper;
-    }
-
-    @Override
-    public int countUnique() {
-        DoubleSet doubles = new DoubleOpenHashSet();
-        for (int i = 0; i < size(); i++) {
-            if (!isMissingValue(getDouble(i))) {
-                doubles.add(getDouble(i));
-            }
-        }
-        return doubles.size();
-    }
-
-    /**
-     * Returns the closest {@code int} to the argument, with ties
-     * rounding to positive infinity.
-     *
-     * <p>
-     * Special cases:
-     * <ul><li>If the argument is NaN, the result is 0.
-     * <li>If the argument is positive infinity or any value greater than or
-     * equal to the value of {@code Integer.MAX_VALUE}, an error will be thrown
-     *
-     * @param   index the index of the value to be rounded to an integer.
-     * @return  the value of the argument rounded to the nearest
-     *          {@code int} value.
-     * @throws  ClassCastException if the absolute value of the value to be rounded is too large to be cast to an int
-     */
-    @Override
-    public int getInt(int index) {
-        return (int) Math.round(data.getDouble(index));
-    }
-
-    @Override
-    public void appendMissing() {
-        data.add(DoubleColumnType.missingValueIndicator());
-    }
-
-    @Override
-    public byte[] asBytes(int rowNumber) {
-        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putDouble(getDouble(rowNumber)).array();
-    }
-
-    @Override
-    public double missingValueIndicator() {
-        return DoubleColumnType.missingValueIndicator();
-    }
-
-    @Override
-    public DoubleColumnType type() {
-        return ColumnType.DOUBLE;
-    }
-}
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/FloatDataWrapper.java b/core/src/main/java/tech/tablesaw/columns/numbers/FloatDataWrapper.java
deleted file mode 100644
index 442ae413..00000000
--- a/core/src/main/java/tech/tablesaw/columns/numbers/FloatDataWrapper.java
+++ /dev/null
@@ -1,233 +0,0 @@
-package tech.tablesaw.columns.numbers;
-
-import it.unimi.dsi.fastutil.floats.FloatArrayList;
-import it.unimi.dsi.fastutil.floats.FloatArrays;
-import it.unimi.dsi.fastutil.floats.FloatComparator;
-import it.unimi.dsi.fastutil.floats.FloatOpenHashSet;
-import it.unimi.dsi.fastutil.floats.FloatSet;
-import tech.tablesaw.api.ColumnType;
-import tech.tablesaw.columns.StringParser;
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Iterator;
-
-public class FloatDataWrapper implements NumericDataWrapper {
-
-    private FloatColumnType COLUMN_TYPE = ColumnType.FLOAT;
-    /**
-     * Compares two doubles, such that a sort based on this comparator would sort in descending order
-     */
-    private final FloatComparator descendingComparator = (o2, o1) -> (Float.compare(o1, o2));
-
-    private final FloatArrayList data;
-
-    public FloatDataWrapper(FloatArrayList data) {
-        this.data = data;
-    }
-
-    @Override
-    public NumberIterator numberIterator() {
-        return new NumberIterator(data);
-    }
-
-    @Override
-    public Iterator<Float> iterator() {
-        return data.iterator();
-    }
-
-    @Override
-    public NumericDataWrapper top(int n) {
-        final FloatArrayList top = new FloatArrayList();
-        final float[] values = data.toFloatArray();
-        FloatArrays.parallelQuickSort(values, descendingComparator);
-        for (int i = 0; i < n && i < values.length; i++) {
-            top.add(values[i]);
-        }
-        return new FloatDataWrapper(top);
-    }
-
-    @Override
-    public NumericDataWrapper bottom(int n) {
-
-        final FloatArrayList bottom = new FloatArrayList();
-        final float[] values = data.toFloatArray();
-        FloatArrays.parallelQuickSort(values);
-        for (int i = 0; i < n && i < values.length; i++) {
-            bottom.add(values[i]);
-        }
-        return new FloatDataWrapper(bottom);
-    }
-
-    @Override
-    public int size() {
-        return data.size();
-    }
-
-    public void append(final float f) {
-        data.add(f);
-    }
-
-    @Override
-    public void append(double d) {
-        data.add((float) d);
-    }
-
-    @Override
-    public void append(int i) {
-        data.add(i);
-    }
-
-    @Override
-    public void appendCell(String value) {
-        append(FloatColumnType.DEFAULT_PARSER.parseFloat(value));
-    }
-
-    @Override
-    public void appendCell(String value, StringParser<?> parser) {
-        append(parser.parseFloat(value));
-    }
-
-    @Override
-    public int countMissing() {
-        int count = 0;
-        for (int i = 0; i < size(); i++) {
-            if (isMissingValue(getFloat(i))) {
-                count++;
-            }
-        }
-        return count;
-    }
-
-    @Override
-    public double getDouble(int row) {
-        return data.getFloat(row);
-    }
-
-    @Override
-    public FloatDataWrapper copy() {
-        return new FloatDataWrapper(data.clone());
-    }
-
-    @Override
-    public FloatDataWrapper emptyCopy() {
-        return new FloatDataWrapper(new FloatArrayList());
-    }
-
-    @Override
-    public FloatDataWrapper emptyCopy(int size) {
-        return new FloatDataWrapper(new FloatArrayList(size));
-    }
-
-    @Override
-    public void clear() {
-        this.data.clear();
-    }
-
-    @Override
-    public void sortAscending() {
-        Arrays.parallelSort(data.elements());
-    }
-
-    @Override
-    public void sortDescending() {
-        FloatArrays.parallelQuickSort(data.elements(), descendingComparator);
-    }
-
-    @Override
-    public void set(int r, double value) {
-        this.data.set(r, (float) value);
-    }
-
-    @Override
-    public boolean contains(double value) {
-        return data.contains((float) value);
-    }
-
-    @Override
-    public boolean contains(int value) {
-        return data.contains(value);
-    }
-
-    @Override
-    public NumericDataWrapper lag(int n) {
-        final int srcPos = n >= 0 ? 0 : 0 - n;
-        final float[] dest = new float[size()];
-        final int destPos = n <= 0 ? 0 : n;
-        final int length = n >= 0 ? size() - n : size() + n;
-
-        for (int i = 0; i < size(); i++) {
-            dest[i] = FloatColumnType.missingValueIndicator();
-        }
-
-        float[] array = data.toFloatArray();
-
-        System.arraycopy(array, srcPos, dest, destPos, length);
-        return new FloatDataWrapper(new FloatArrayList(dest));
-    }
-
-    @Override
-    public NumericDataWrapper lead(int n) {
-        return lag(-n);
-    }
-
-    @Override
-    public int countUnique() {
-        FloatSet floats = new FloatOpenHashSet();
-        for (int i = 0; i < size(); i++) {
-            if (!isMissingValue(getFloat(i))) {
-                floats.add(getFloat(i));
-            }
-        }
-        return floats.size();
-    }
-
-    @Override
-    public void appendMissing() {
-        data.add(FloatColumnType.missingValueIndicator());
-    }
-
-    public float getFloat(int index) {
-        return data.getFloat(index);
-    }
-
-    @Override
-    public byte[] asBytes(int rowNumber) {
-        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putFloat(getFloat(rowNumber)).array();
-    }
-
-    @Override
-    public ColumnType type() {
-        return ColumnType.FLOAT;
-    }
-
-    /**
-     * Returns the closest {@code int} to the argument, with ties
-     * rounding to positive infinity.
-     *
-     * <p>
-     * Special cases:
-     * <ul><li>If the argument is NaN, the result is 0.
-     * <li>If the argument is negative infinity or any value less than or
-     * equal to the value of {@code Integer.MIN_VALUE}, the result is
-     * equal to the value of {@code Integer.MIN_VALUE}.
-     * <li>If the argument is positive infinity or any value greater than or
-     * equal to the value of {@code Integer.MAX_VALUE}, the result is
-     * equal to the value of {@code Integer.MAX_VALUE}.</ul>
-     *
-     * @param   index the index of the value to be rounded to an integer.
-     * @return  the value of the argument rounded to the nearest
-     *          {@code int} value.
-     * @see     java.lang.Integer#MAX_VALUE
-     * @see     java.lang.Integer#MIN_VALUE
-     */
-    @Override
-    public int getInt(int index) {
-        return Math.round(data.getFloat(index));
-    }
-
-    @Override
-    public double missingValueIndicator() {
-        return FloatColumnType.missingValueIndicator();
-    }
-}
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/IntDataWrapper.java b/core/src/main/java/tech/tablesaw/columns/numbers/IntDataWrapper.java
deleted file mode 100644
index 3033a1ea..00000000
--- a/core/src/main/java/tech/tablesaw/columns/numbers/IntDataWrapper.java
+++ /dev/null
@@ -1,246 +0,0 @@
-package tech.tablesaw.columns.numbers;
-
-import it.unimi.dsi.fastutil.ints.IntArrayList;
-import it.unimi.dsi.fastutil.ints.IntArrays;
-import it.unimi.dsi.fastutil.ints.IntComparator;
-import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
-import it.unimi.dsi.fastutil.ints.IntSet;
-import tech.tablesaw.api.ColumnType;
-import tech.tablesaw.columns.StringParser;
-
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.Iterator;
-
-public class IntDataWrapper implements NumericDataWrapper {
-
-    private static final IntColumnType COLUMN_TYPE = ColumnType.INTEGER;
-
-    /**
-     * Compares two ints, such that a sort based on this comparator would sort in descending order
-     */
-    private final IntComparator descendingComparator = (o2, o1) -> (Integer.compare(o1, o2));
-
-    private final IntArrayList data;
-
-    public IntDataWrapper(IntArrayList data) {
-        this.data = data;
-    }
-
-    @Override
-    public NumberIterator numberIterator() {
-        return new NumberIterator(data);
-    }
-
-    @Override
-    public Iterator<Integer> iterator() {
-        return data.iterator();
-    }
-
-    @Override
-    public NumericDataWrapper top(int n) {
-        final IntArrayList top = new IntArrayList();
-        final int[] values = data.toIntArray();
-        IntArrays.parallelQuickSort(values, descendingComparator);
-        for (int i = 0; i < n && i < values.length; i++) {
-            top.add(values[i]);
-        }
-        return new IntDataWrapper(top);
-    }
-
-    @Override
-    public NumericDataWrapper bottom(int n) {
-
-        final IntArrayList bottom = new IntArrayList();
-        final int[] values = data.toIntArray();
-        IntArrays.parallelQuickSort(values);
-        for (int i = 0; i < n && i < values.length; i++) {
-            bottom.add(values[i]);
-        }
-        return new IntDataWrapper(bottom);
-    }
-
-    @Override
-    public NumericDataWrapper removeMissing() {
-        NumericDataWrapper wrapper = copy();
-        wrapper.clear();
-        final NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            final int v = iterator.nextInt();
-            if (!isMissingValue(v)) {
-                wrapper.append(v);
-            }
-        }
-        return wrapper;
-    }
-
-    @Override
-    public int size() {
-        return data.size();
-    }
-
-    public void append(final float f) {
-        append((double) f);
-    }
-
-    @Override
-    public void append(double d) {
-        if (isMissingValue(d)) {
-            appendMissing();
-        } else if (d == (int) d) {
-            data.add((int) d);
-        } else {
-            throw new RuntimeException("Incompatible numeric type. Attempting to add a float to a column of integers.");
-        }
-    }
-
-    @Override
-    public void append(int i) {
-        data.add(i);
-    }
-
-    @Override
-    public void appendCell(String value) {
-        append(IntColumnType.DEFAULT_PARSER.parseInt(value));
-    }
-
-    @Override
-    public void appendCell(String value, StringParser<?> parser) {
-        append(parser.parseInt(value));
-    }
-
-    @Override
-    public int countMissing() {
-        int count = 0;
-        for (int i = 0; i < size(); i++) {
-            if (isMissingValue(getInt(i))) {
-                count++;
-            }
-        }
-        return count;
-    }
-
-    @Override
-    public double getDouble(int row) {
-        int value = data.getInt(row);
-        if (isMissingValue(value)) {
-            return DoubleColumnType.missingValueIndicator();
-        }
-        return value;
-    }
-
-    @Override
-    public float getFloat(int row) {
-        int value = data.getInt(row);
-        if (isMissingValue(value)) {
-            return FloatColumnType.missingValueIndicator();
-        }
-        return value;
-    }
-
-    @Override
-    public IntDataWrapper copy() {
-        return new IntDataWrapper(data.clone());
-    }
-
-    @Override
-    public IntDataWrapper emptyCopy() {
-        return new IntDataWrapper(new IntArrayList());
-    }
-
-    @Override
-    public IntDataWrapper emptyCopy(int size) {
-        return new IntDataWrapper(new IntArrayList(size));
-    }
-
-    @Override
-    public void clear() {
-        this.data.clear();
-    }
-
-    @Override
-    public void sortAscending() {
-        Arrays.parallelSort(data.elements());
-    }
-
-    @Override
-    public void sortDescending() {
-        IntArrays.parallelQuickSort(data.elements(), descendingComparator);
-    }
-
-    @Override
-    public void set(int r, double value) {
-        if (value == (int) value) {
-            data.set(r, (int) value);
-        } else {
-            throw new RuntimeException("Incompatible numeric type. Attempting to add a float to a column of integers.");
-        }
-    }
-
-    @Override
-    public boolean contains(double value) {
-        return (value == (int) value) && data.contains((int) value);
-    }
-
-    @Override
-    public boolean contains(int value) {
-        return data.contains(value);
-    }
-
-    @Override
-    public IntDataWrapper lag(int n) {
-        final int srcPos = n >= 0 ? 0 : 0 - n;
-        final int[] dest = new int[size()];
-        final int destPos = n <= 0 ? 0 : n;
-        final int length = n >= 0 ? size() - n : size() + n;
-
-        for (int i = 0; i < size(); i++) {
-            dest[i] = IntColumnType.missingValueIndicator();
-        }
-
-        int[] array = data.toIntArray();
-
-        System.arraycopy(array, srcPos, dest, destPos, length);
-        return new IntDataWrapper(new IntArrayList(dest));
-    }
-
-    @Override
-    public IntDataWrapper lead(int n) {
-        return lag(-n);
-    }
-
-    @Override
-    public int countUnique() {
-        IntSet uniqueElements = new IntOpenHashSet();
-        for (int i = 0; i < size(); i++) {
-            if (!isMissingValue(getInt(i))) {
-                uniqueElements.add(getInt(i));
-            }
-        }
-        return uniqueElements.size();
-    }
-
-    public int getInt(int index) {
-        return data.getInt(index);
-    }
-
-    @Override
-    public void appendMissing() {
-        append(IntColumnType.missingValueIndicator());
-    }
-
-    @Override
-    public byte[] asBytes(int rowNumber) {
-        return ByteBuffer.allocate(COLUMN_TYPE.byteSize()).putInt(getInt(rowNumber)).array();
-    }
-
-    @Override
-    public IntColumnType type() {
-        return ColumnType.INTEGER;
-    }
-
-    @Override
-    public double missingValueIndicator() {
-        return IntColumnType.missingValueIndicator();
-    }
-}
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/LongColumnType.java b/core/src/main/java/tech/tablesaw/columns/numbers/LongColumnType.java
new file mode 100644
index 00000000..ef562853
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/LongColumnType.java
@@ -0,0 +1,47 @@
+package tech.tablesaw.columns.numbers;
+
+import tech.tablesaw.api.ColumnType;
+import tech.tablesaw.api.LongColumn;
+import tech.tablesaw.columns.AbstractColumnType;
+import tech.tablesaw.io.csv.CsvReadOptions;
+
+public class LongColumnType extends AbstractColumnType {
+
+    public static final LongStringParser DEFAULT_PARSER = new LongStringParser(ColumnType.LONG);
+
+    private static final int BYTE_SIZE = 8;
+
+    public static final LongColumnType INSTANCE =
+            new LongColumnType(BYTE_SIZE, "LONG", "Long");
+
+    private LongColumnType(int byteSize, String name, String printerFriendlyName) {
+        super(byteSize, name, printerFriendlyName);
+    }
+
+    @Override
+    public LongColumn create(String name) {
+        return LongColumn.create(name);
+    }
+
+    public LongStringParser defaultParser() {
+        return DEFAULT_PARSER;
+    }
+
+    @Override
+    public LongStringParser customParser(CsvReadOptions options) {
+        return new LongStringParser(this, options);
+    }
+
+    public static boolean isMissingValue(long value) {
+	return value == missingValueIndicator();
+    }
+
+    public static long missingValueIndicator() {
+        return Long.MIN_VALUE;
+    }
+
+    @Override
+    public Comparable<?> getMissingValueIndicator() {
+        return missingValueIndicator();
+    }
+}
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/LongStringParser.java b/core/src/main/java/tech/tablesaw/columns/numbers/LongStringParser.java
new file mode 100644
index 00000000..afb3b6d2
--- /dev/null
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/LongStringParser.java
@@ -0,0 +1,79 @@
+package tech.tablesaw.columns.numbers;
+
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import com.google.common.collect.Lists;
+
+import tech.tablesaw.api.ColumnType;
+import tech.tablesaw.columns.StringParser;
+import tech.tablesaw.io.csv.CsvReadOptions;
+
+public class LongStringParser extends StringParser<Long> {
+
+    private static final Pattern COMMA_PATTERN = Pattern.compile(",");
+
+    public LongStringParser(ColumnType columnType) {
+        super(columnType);
+    }
+
+    public LongStringParser(LongColumnType columnType, CsvReadOptions readOptions) {
+        super(columnType);
+        if (readOptions.missingValueIndicator() != null) {
+            missingValueStrings = Lists.newArrayList(readOptions.missingValueIndicator());
+        }
+    }
+
+    @Override
+    public boolean canParse(String str) {
+        if (isMissing(str)) {
+            return true;
+        }
+        String s = str;
+        try {
+            if (s.endsWith(".0")) {
+                s = s.replaceFirst(".0$", "");
+            }
+            Long.parseLong(s);
+            return true;
+        } catch (NumberFormatException e) {
+            // it's all part of the plan
+            return false;
+        }
+    }
+
+    @Override
+    public Long parse(String s) {
+        if (isMissing(s)) {
+            return LongColumnType.missingValueIndicator();
+        }
+        final Matcher matcher = COMMA_PATTERN.matcher(s);
+        return Long.parseLong(matcher.replaceAll(""));
+    }
+
+    @Override
+    public double parseDouble(String str) {
+        if (isMissing(str)) {
+            return IntColumnType.missingValueIndicator();
+        }
+        String s = str;
+        if (s.endsWith(".0")) {
+            s = s.replaceFirst(".0$", "");
+        }
+        final Matcher matcher = COMMA_PATTERN.matcher(s);
+        return Integer.parseInt(matcher.replaceAll(""));
+    }
+
+    @Override
+    public long parseLong(String str) {
+        if (isMissing(str)) {
+            return LongColumnType.missingValueIndicator();
+        }
+        String s = str;
+        if (s.endsWith(".0")) {
+            s = s.replaceFirst(".0$", "");
+        }
+        final Matcher matcher = COMMA_PATTERN.matcher(s);
+        return Long.parseLong(matcher.replaceAll(""));
+    }
+}
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/NumberIterator.java b/core/src/main/java/tech/tablesaw/columns/numbers/NumberIterator.java
index 98f70861..f3bb0c4c 100644
--- a/core/src/main/java/tech/tablesaw/columns/numbers/NumberIterator.java
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/NumberIterator.java
@@ -9,11 +9,14 @@ import it.unimi.dsi.fastutil.floats.FloatArrayList;
 import it.unimi.dsi.fastutil.floats.FloatIterator;
 import it.unimi.dsi.fastutil.ints.IntArrayList;
 import it.unimi.dsi.fastutil.ints.IntIterator;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.longs.LongIterator;
 
 public class NumberIterator {
 
-    private DoubleArrayList dList;
     private IntArrayList iList;
+    private LongArrayList lList;
+    private DoubleArrayList dList;
     private FloatArrayList fList;
 
     private final ListIterator<?> iterator;
@@ -33,6 +36,11 @@ public class NumberIterator {
         this.iterator = list.iterator();
     }
 
+    public NumberIterator(LongArrayList list) {
+        this.lList = list;
+        this.iterator = list.iterator();
+    }
+
     public int nextIndex() {
         return iterator.nextIndex();
     }
@@ -47,9 +55,12 @@ public class NumberIterator {
         } else if (iList != null){
             int nextInt = ((IntIterator) iterator).nextInt();
             return IntColumnType.isMissingValue(nextInt) ? DoubleColumnType.missingValueIndicator() : nextInt;
-        } else {
+        } else if (fList != null) {
             float nextFloat = ((FloatIterator) iterator).nextFloat();
             return FloatColumnType.isMissingValue(nextFloat) ? DoubleColumnType.missingValueIndicator() : nextFloat;
+        } else {
+            long nextLong = ((LongIterator) iterator).nextLong();
+            return LongColumnType.isMissingValue(nextLong) ? DoubleColumnType.missingValueIndicator() : nextLong;
         }
     }
 
@@ -58,8 +69,10 @@ public class NumberIterator {
             return (int) ((DoubleIterator) iterator).nextDouble();
         } else if (iList != null){
             return ((IntIterator) iterator).nextInt();
-        } else {
+        } else if (fList != null) {
             return (int) ((FloatIterator) iterator).nextFloat();
+        } else {
+            return (int) ((LongIterator) iterator).nextLong();
         }
     }
 
@@ -68,8 +81,22 @@ public class NumberIterator {
             return (float) ((DoubleIterator) iterator).nextDouble();
         } else if (iList != null){
             return ((IntIterator) iterator).nextInt();
-        } else {
+        } else if (fList != null) {
             return ((FloatIterator) iterator).nextFloat();
+        } else {
+            return ((LongIterator) iterator).nextLong();
+        }
+    }
+
+    public long nextLong() {
+        if (dList != null) {
+            return (long) ((DoubleIterator) iterator).nextDouble();
+        } else if (iList != null){
+            return ((IntIterator) iterator).nextInt();
+        } else if (fList != null) {
+            return (long) ((FloatIterator) iterator).nextFloat();
+        } else {
+            return ((LongIterator) iterator).nextLong();
         }
     }
 
@@ -93,6 +120,19 @@ public class NumberIterator {
                     return (double) intIterator.nextInt();
                 }
             };
+        } else if (lList != null) {
+            return new Iterator<Double>() {
+                LongIterator longIterator = lList.iterator();
+                @Override
+                public boolean hasNext() {
+                    return longIterator.hasNext();
+                }
+
+                @Override
+                public Double next() {
+                    return (double) longIterator.nextLong();
+                }
+            };
         } else {
             return new Iterator<Double>() {
                 FloatIterator floatIterator = fList.iterator();
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java b/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java
index fb753040..fc060bb2 100644
--- a/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/NumberMapFunctions.java
@@ -46,11 +46,9 @@ public interface NumberMapFunctions extends NumberIterable {
     default DoubleColumn asRatio() {
       DoubleColumn pctColumn = DoubleColumn.create(name() + " percents");
         double total = sum();
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
+        for (int i = 0; i < size(); i++) {
             if (total != 0) {
-                pctColumn.append((float) value / total);
+                pctColumn.append(getDouble(i) / total);
             } else {
                 pctColumn.append(DoubleColumnType.missingValueIndicator());
             }
@@ -67,11 +65,9 @@ public interface NumberMapFunctions extends NumberIterable {
     default DoubleColumn asPercent() {
         DoubleColumn pctColumn = DoubleColumn.create(name() + " percents");
         double total = sum();
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
+        for (int i = 0; i < size(); i++) {
             if (total != 0) {
-                pctColumn.append(((float) value / total) * 100);
+                pctColumn.append((getDouble(i) / total) * 100);
             } else {
                 pctColumn.append(DoubleColumnType.missingValueIndicator());
             }
@@ -204,10 +200,8 @@ public interface NumberMapFunctions extends NumberIterable {
      */
     default DoubleColumn power(double power) {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[pow]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.pow(value, power));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.pow(getDouble(i), power));
         }
         return newColumn;
     }    
@@ -223,20 +217,16 @@ public interface NumberMapFunctions extends NumberIterable {
 
     default DoubleColumn sqrt() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[sqrt]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.sqrt(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.sqrt(getDouble(i)));
         }
         return newColumn;
     }
 
     default DoubleColumn cubeRoot() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[cbrt]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.cbrt(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.cbrt(getDouble(i)));
         }
         return newColumn;
     }
@@ -268,10 +258,8 @@ public interface NumberMapFunctions extends NumberIterable {
     default DoubleColumn logN() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[logN]", size());
 
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.log(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.log(getDouble(i)));
         }
         return newColumn;
     }
@@ -281,11 +269,8 @@ public interface NumberMapFunctions extends NumberIterable {
      */
     default DoubleColumn log10() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[log10]", size());
-
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.log10(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.log10(getDouble(i)));
         }
         return newColumn;
     }
@@ -296,20 +281,16 @@ public interface NumberMapFunctions extends NumberIterable {
      */
     default DoubleColumn log1p() {
       DoubleColumn newColumn = DoubleColumn.create(name() + "[1og1p]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.log1p(value));
+      for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.log1p(getDouble(i)));
         }
         return newColumn;
     }
 
     default DoubleColumn round() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[rounded]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.round(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.round(getDouble(i)));
         }
         return newColumn;
     }
@@ -321,10 +302,8 @@ public interface NumberMapFunctions extends NumberIterable {
      */
     default DoubleColumn roundInt() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[rounded]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append((int) Math.round(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append((int) Math.round(getDouble(i)));
         }
         return newColumn;
     }
@@ -335,10 +314,8 @@ public interface NumberMapFunctions extends NumberIterable {
      */
     default DoubleColumn abs() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[abs]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(Math.abs(value));
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(Math.abs(getDouble(i)));
         }
         return newColumn;
     }
@@ -352,10 +329,8 @@ public interface NumberMapFunctions extends NumberIterable {
      */
     default DoubleColumn neg() {
         DoubleColumn newColumn = DoubleColumn.create(name() + "[neg]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
-            newColumn.append(value * -1);
+        for (int i = 0; i < size(); i++) {
+            newColumn.append(getDouble(i) * -1);
         }
         return newColumn;
     }
@@ -378,9 +353,8 @@ public interface NumberMapFunctions extends NumberIterable {
     default DoubleColumn cumSum() {
         double total = 0.0;
         DoubleColumn newColumn = DoubleColumn.create(name() + "[cumSum]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
+        for (int i = 0; i < size(); i++) {
+            double value = getDouble(i);
             if (DoubleColumnType.isMissingValue(value)) {
                 newColumn.append(DoubleColumnType.missingValueIndicator());
             } else {
@@ -397,9 +371,8 @@ public interface NumberMapFunctions extends NumberIterable {
     default DoubleColumn cumProd() {
         double total = 1.0;
         DoubleColumn newColumn = DoubleColumn.create(name() + "[cumProd]", size());
-        NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            double value = iterator.next();
+        for (int i = 0; i < size(); i++) {
+            double value = getDouble(i);
             if (DoubleColumnType.isMissingValue(value)) {
                 newColumn.append(DoubleColumnType.missingValueIndicator());
             } else {
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/NumericDataWrapper.java b/core/src/main/java/tech/tablesaw/columns/numbers/NumericDataWrapper.java
deleted file mode 100644
index 25ac90dd..00000000
--- a/core/src/main/java/tech/tablesaw/columns/numbers/NumericDataWrapper.java
+++ /dev/null
@@ -1,120 +0,0 @@
-package tech.tablesaw.columns.numbers;
-
-import java.util.Iterator;
-
-import tech.tablesaw.api.ColumnType;
-import tech.tablesaw.columns.StringParser;
-
-public interface NumericDataWrapper extends NumberIterable {
-
-    int size();
-
-    void append(final float f);
-
-    void append(double d);
-
-    void append(int i);
-
-    void appendCell(String value);
-
-    double getDouble(final int row);
-
-    float getFloat(final int row);
-
-    /**
-     * Returns an int from the data at the given index.
-     *
-     * NOTE: If the data is floating point, it is cast to int and precision may be lost
-     */
-    int getInt(int index);
-
-    NumericDataWrapper copy();
-
-    NumericDataWrapper emptyCopy();
-
-    NumericDataWrapper emptyCopy(int rowSize);
-
-    void clear();
-
-    void sortAscending();
-
-    void sortDescending();
-
-    void set(final int r, final double value);
-
-    boolean contains(final double value);
-
-    boolean contains(final int value);
-
-    NumericDataWrapper lag(final int n);
-
-    NumericDataWrapper lead(final int n);
-
-    NumberIterator numberIterator();
-
-    Iterator<?> iterator();
-
-    NumericDataWrapper top(final int n);
-
-    NumericDataWrapper bottom(final int n);
-
-    default NumericDataWrapper removeMissing() {
-        NumericDataWrapper wrapper = copy();
-        wrapper.clear();
-        final NumberIterator iterator = numberIterator();
-        while (iterator.hasNext()) {
-            final double v = iterator.next();
-            if (!isMissingValue(v)) {
-                wrapper.append(v);
-            }
-        }
-        return wrapper;
-    }
-
-    int countMissing();
-
-    default boolean isMissingValue(double value) {
-        return value != value;
-    }
-
-    default boolean isMissingValue(float value) {
-        return value != value;
-    }
-
-    default boolean isMissingValue(int value) {
-        return value == IntColumnType.missingValueIndicator();
-    }
-
-    /**
-     * Returns the count of unique, non-missing values in the dataset.
-     */
-    int countUnique();
-
-    void appendMissing();
-
-    ColumnType type();
-
-    /**
-     * Returns the contents of the cell at rowNumber as a byte[]
-     */
-    byte[] asBytes(final int rowNumber);
-
-    double missingValueIndicator();
-
-    void appendCell(final String object, StringParser<?> parser);
-
-    default void appendObj(Object obj) {
-        if (obj instanceof Double) {
-            append((double) obj);
-        }
-        else if (obj instanceof Float) {
-            append((float) obj);
-        }
-        else if (obj instanceof Integer) {
-            append((int) obj);
-        }
-        else {
-            throw new IllegalArgumentException();
-        }
-    }
-}
diff --git a/core/src/main/java/tech/tablesaw/columns/numbers/Stats.java b/core/src/main/java/tech/tablesaw/columns/numbers/Stats.java
index 2148c8da..fb08b710 100644
--- a/core/src/main/java/tech/tablesaw/columns/numbers/Stats.java
+++ b/core/src/main/java/tech/tablesaw/columns/numbers/Stats.java
@@ -21,9 +21,6 @@ import tech.tablesaw.api.NumericColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.Table;
 
-/**
- *
- */
 public class Stats {
 
     private long n;
@@ -47,9 +44,8 @@ public class Stats {
 
     public static Stats create(final NumericColumn<?> values) {
         SummaryStatistics summaryStatistics = new SummaryStatistics();
-        NumberIterator itr = values.numberIterator();
-        while (itr.hasNext()) {
-            summaryStatistics.addValue(itr.next());
+        for (int i = 0; i < values.size(); i++) {
+            summaryStatistics.addValue(values.getDouble(i));
         }
         return getStats(values, summaryStatistics);
     }
diff --git a/core/src/main/java/tech/tablesaw/columns/times/TimeMapFunctions.java b/core/src/main/java/tech/tablesaw/columns/times/TimeMapFunctions.java
index 6aeb8d30..ce587de6 100644
--- a/core/src/main/java/tech/tablesaw/columns/times/TimeMapFunctions.java
+++ b/core/src/main/java/tech/tablesaw/columns/times/TimeMapFunctions.java
@@ -20,7 +20,6 @@ import tech.tablesaw.api.IntColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.TimeColumn;
 import tech.tablesaw.columns.Column;
-import tech.tablesaw.columns.numbers.DoubleColumnType;
 import tech.tablesaw.columns.numbers.IntColumnType;
 import tech.tablesaw.columns.numbers.NumberColumnFormatter;
 
diff --git a/core/src/main/java/tech/tablesaw/index/FloatIndex.java b/core/src/main/java/tech/tablesaw/index/FloatIndex.java
index e177a6b4..04ebd5b2 100644
--- a/core/src/main/java/tech/tablesaw/index/FloatIndex.java
+++ b/core/src/main/java/tech/tablesaw/index/FloatIndex.java
@@ -18,7 +18,7 @@ import it.unimi.dsi.fastutil.floats.Float2ObjectAVLTreeMap;
 import it.unimi.dsi.fastutil.floats.Float2ObjectOpenHashMap;
 import it.unimi.dsi.fastutil.floats.Float2ObjectSortedMap;
 import it.unimi.dsi.fastutil.ints.IntArrayList;
-import tech.tablesaw.api.DoubleColumn;
+import tech.tablesaw.api.FloatColumn;
 import tech.tablesaw.selection.BitmapBackedSelection;
 import tech.tablesaw.selection.Selection;
 
@@ -29,7 +29,7 @@ public class FloatIndex {
 
     private final Float2ObjectAVLTreeMap<IntArrayList> index;
 
-    public FloatIndex(DoubleColumn column) {
+    public FloatIndex(FloatColumn column) {
         int sizeEstimate = Integer.min(1_000_000, column.size() / 100);
         Float2ObjectOpenHashMap<IntArrayList> tempMap = new Float2ObjectOpenHashMap<>(sizeEstimate);
         for (int i = 0; i < column.size(); i++) {
diff --git a/core/src/main/java/tech/tablesaw/index/LongIndex.java b/core/src/main/java/tech/tablesaw/index/LongIndex.java
index d99745c8..7f0ac956 100644
--- a/core/src/main/java/tech/tablesaw/index/LongIndex.java
+++ b/core/src/main/java/tech/tablesaw/index/LongIndex.java
@@ -14,18 +14,18 @@
 
 package tech.tablesaw.index;
 
+import java.time.LocalDateTime;
+
 import it.unimi.dsi.fastutil.ints.IntArrayList;
 import it.unimi.dsi.fastutil.longs.Long2ObjectAVLTreeMap;
 import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
 import it.unimi.dsi.fastutil.longs.Long2ObjectSortedMap;
 import tech.tablesaw.api.DateTimeColumn;
-import tech.tablesaw.api.DoubleColumn;
+import tech.tablesaw.api.IntColumn;
 import tech.tablesaw.columns.datetimes.PackedLocalDateTime;
 import tech.tablesaw.selection.BitmapBackedSelection;
 import tech.tablesaw.selection.Selection;
 
-import java.time.LocalDateTime;
-
 /**
  * An index for eight-byte long and long backed columns (datetime)
  */
@@ -51,11 +51,11 @@ public class LongIndex {
         index = new Long2ObjectAVLTreeMap<>(tempMap);
     }
 
-    public LongIndex(DoubleColumn column) {
+    public LongIndex(IntColumn column) {
         int sizeEstimate = Integer.min(1_000_000, column.size() / 100);
         Long2ObjectOpenHashMap<IntArrayList> tempMap = new Long2ObjectOpenHashMap<>(sizeEstimate);
         for (int i = 0; i < column.size(); i++) {
-            long value = column.getLong(i);
+            long value = column.getInt(i);
             IntArrayList recordIds = tempMap.get(value);
             if (recordIds == null) {
                 recordIds = new IntArrayList();
diff --git a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
index 24228bc4..896ffadc 100644
--- a/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
+++ b/core/src/main/java/tech/tablesaw/joining/DataFrameJoiner.java
@@ -8,7 +8,6 @@ import tech.tablesaw.api.CategoricalColumn;
 import tech.tablesaw.api.ColumnType;
 import tech.tablesaw.api.DateColumn;
 import tech.tablesaw.api.DateTimeColumn;
-import tech.tablesaw.api.DoubleColumn;
 import tech.tablesaw.api.IntColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.Table;
@@ -276,7 +275,7 @@ public class DataFrameJoiner {
             }
         } else if (type instanceof IntColumnType) {
             IntIndex index = new IntIndex(result.intColumn(col2Name));
-            DoubleColumn col2 = (DoubleColumn) table2.column(col2Name);
+            IntColumn col2 = (IntColumn) table2.column(col2Name);
             for (int i = 0; i < col2.size(); i++) {
                 int value = col2.getInt(i);
                 if (index.get(value).isEmpty()) {
diff --git a/core/src/main/java/tech/tablesaw/table/Relation.java b/core/src/main/java/tech/tablesaw/table/Relation.java
index c1893289..69a2b874 100644
--- a/core/src/main/java/tech/tablesaw/table/Relation.java
+++ b/core/src/main/java/tech/tablesaw/table/Relation.java
@@ -28,6 +28,7 @@ import tech.tablesaw.api.DateTimeColumn;
 import tech.tablesaw.api.DoubleColumn;
 import tech.tablesaw.api.FloatColumn;
 import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.LongColumn;
 import tech.tablesaw.api.NumberColumn;
 import tech.tablesaw.api.StringColumn;
 import tech.tablesaw.api.Table;
@@ -363,6 +364,14 @@ public abstract class Relation {
         return (IntColumn) column(columnIndex);
     }
     
+    public LongColumn longColumn(String columnName) {
+        return longColumn(columnIndex(columnName));
+    }
+
+    public LongColumn longColumn(int columnIndex) {
+        return (LongColumn) column(columnIndex);
+    }
+    
     public FloatColumn floatColumn(String columnName) {
         return floatColumn(columnIndex(columnName));
     }


Test patch:
diff --git a/core/src/test/java/tech/tablesaw/aggregate/CrossTabTest.java b/core/src/test/java/tech/tablesaw/aggregate/CrossTabTest.java
index a7df3b31..b8e9961a 100644
--- a/core/src/test/java/tech/tablesaw/aggregate/CrossTabTest.java
+++ b/core/src/test/java/tech/tablesaw/aggregate/CrossTabTest.java
@@ -65,7 +65,7 @@ public class CrossTabTest {
         Table bush = Table.read().csv("../data/bush.csv");
         bush.addColumns(bush.dateColumn("date").year());
         Table xtab = CrossTab.columnPercents(bush, "who", "date year");
-        assertEquals(1, xtab.doubleColumn(1).getInt(xtab.rowCount() - 1));
+        assertEquals(1.0, xtab.doubleColumn(1).getDouble(xtab.rowCount() - 1), 0.00001);
     }
 
     @Test
@@ -73,7 +73,7 @@ public class CrossTabTest {
         Table bush = Table.read().csv("../data/bush.csv");
         bush.addColumns(bush.dateColumn("date").year());
         Table xtab = CrossTab.rowPercents(bush, "who", "date year");
-        assertEquals(1, xtab.doubleColumn(xtab.columnCount() - 1).getInt(0));
+        assertEquals(1.0, xtab.doubleColumn(xtab.columnCount() - 1).getDouble(0), 0.00001);
     }
 
     @Test
@@ -81,6 +81,6 @@ public class CrossTabTest {
         Table bush = Table.read().csv("../data/bush.csv");
         bush.addColumns(bush.dateColumn("date").year());
         Table xtab = CrossTab.tablePercents(bush, "who", "date year");
-        assertEquals(1, xtab.doubleColumn(xtab.columnCount() - 1).getInt(xtab.rowCount()-1));
+        assertEquals(1.0, xtab.doubleColumn(xtab.columnCount() - 1).getDouble(xtab.rowCount()-1), 0.00001);
     }
 }
\ No newline at end of file
diff --git a/core/src/test/java/tech/tablesaw/api/BooleanColumnTest.java b/core/src/test/java/tech/tablesaw/api/BooleanColumnTest.java
index af4b3469..9be9e2f4 100644
--- a/core/src/test/java/tech/tablesaw/api/BooleanColumnTest.java
+++ b/core/src/test/java/tech/tablesaw/api/BooleanColumnTest.java
@@ -191,18 +191,6 @@ public class BooleanColumnTest {
         assertEquals("5.0", summary.get(1, 1));
     }
 
-    @Test
-    public void testConversionToInt() {
-        int[] array = column.asIntArray();
-        assertTrue(array[0] == 0);
-        assertTrue(array[1] == 0);
-        assertTrue(array[2] == 0);
-        assertTrue(array[3] == 0);
-        assertTrue(array[4] == 1);
-        assertTrue(array[5] == 1);
-        assertTrue(array[6] == 0);
-    }
-
     @Test
     public void testSelectionMethods() {
         assertEquals(5, column.isFalse().size());
diff --git a/core/src/test/java/tech/tablesaw/api/DateTimeColumnTest.java b/core/src/test/java/tech/tablesaw/api/DateTimeColumnTest.java
index 46d72060..e18b90c0 100644
--- a/core/src/test/java/tech/tablesaw/api/DateTimeColumnTest.java
+++ b/core/src/test/java/tech/tablesaw/api/DateTimeColumnTest.java
@@ -57,8 +57,8 @@ public class DateTimeColumnTest {
     @Test
     public void testConvertMillisSinceEpoch() {
         long millis = 1503952123189L;
-        DoubleColumn dc = DoubleColumn.create("test");
-        dc.appendCell(Long.toString(millis));
+        LongColumn dc = LongColumn.create("test");
+        dc.append(millis);
         DateTimeColumn column2 = dc.asDateTimes(ZoneOffset.UTC);
 
         assertEquals(1, column2.size());
diff --git a/core/src/test/java/tech/tablesaw/api/TimeColumnTest.java b/core/src/test/java/tech/tablesaw/api/TimeColumnTest.java
index 74b5fd69..fbed4699 100644
--- a/core/src/test/java/tech/tablesaw/api/TimeColumnTest.java
+++ b/core/src/test/java/tech/tablesaw/api/TimeColumnTest.java
@@ -410,11 +410,6 @@ public class TimeColumnTest {
         assertNull(col.get(0));
     }
 
-    private void check(DoubleColumn numberColumn, double expected) {
-        assertEquals(expected, numberColumn.min(), .0001);
-        assertEquals(expected, numberColumn.max(), .0001);
-    }
-
     private void check(IntColumn numberColumn, int expected) {
         assertEquals(expected, (int) numberColumn.min());
         assertEquals(expected, (int) numberColumn.max());
diff --git a/core/src/test/java/tech/tablesaw/columns/ColumnTest.java b/core/src/test/java/tech/tablesaw/columns/ColumnTest.java
index d1c2357f..cb1141a6 100644
--- a/core/src/test/java/tech/tablesaw/columns/ColumnTest.java
+++ b/core/src/test/java/tech/tablesaw/columns/ColumnTest.java
@@ -126,9 +126,9 @@ public class ColumnTest {
         DoubleColumn dc1 = DoubleColumn.create("t1", d1);
         DoubleColumn dc2 = DoubleColumn.create("t2", d2);
         DoubleColumn dc3 = (DoubleColumn) dc1.min(dc2);
-        assertTrue(dc3.contains(1));
-        assertTrue(dc3.contains(-4));
-        assertTrue(dc3.contains(-1));
+        assertTrue(dc3.contains(1.0));
+        assertTrue(dc3.contains(-4.0));
+        assertTrue(dc3.contains(-1.0));
     }
 
     @Test
@@ -139,9 +139,9 @@ public class ColumnTest {
         DoubleColumn dc1 = DoubleColumn.create("t1", d1);
         DoubleColumn dc2 = DoubleColumn.create("t2", d2);
         DoubleColumn dc3 = (DoubleColumn) dc1.max(dc2);
-        assertTrue(dc3.contains(2));
-        assertTrue(dc3.contains(0));
-        assertTrue(dc3.contains(3));
+        assertTrue(dc3.contains(2.0));
+        assertTrue(dc3.contains(0.0));
+        assertTrue(dc3.contains(3.0));
     }
     
     // Functional methods
diff --git a/core/src/test/java/tech/tablesaw/columns/dates/DateFiltersTest.java b/core/src/test/java/tech/tablesaw/columns/dates/DateFiltersTest.java
index f81a45ba..609c0f89 100644
--- a/core/src/test/java/tech/tablesaw/columns/dates/DateFiltersTest.java
+++ b/core/src/test/java/tech/tablesaw/columns/dates/DateFiltersTest.java
@@ -139,23 +139,23 @@ public class DateFiltersTest {
         IntColumn index = IntColumn.indexColumn("index", t.rowCount(), 0);
         t.addColumns(index);
 
-        assertTrue(t.where(t.dateColumn("test").isInJanuary()).numberColumn("index").contains(0.0));
-        assertTrue(t.where(t.dateColumn("test").isInFebruary()).numberColumn("index").contains(1.0));
-        assertTrue(t.where(t.dateColumn("test").isInMarch()).numberColumn("index").contains(2.0));
-        assertTrue(t.where(t.dateColumn("test").isInApril()).numberColumn("index").contains(3.0));
-        assertTrue(t.where(t.dateColumn("test").isInMay()).numberColumn("index").contains(4.0));
-        assertTrue(t.where(t.dateColumn("test").isInJune()).numberColumn("index").contains(5.0));
-        assertTrue(t.where(t.dateColumn("test").isInJuly()).numberColumn("index").contains(6.0));
-        assertTrue(t.where(t.dateColumn("test").isInAugust()).numberColumn("index").contains(7.0));
-        assertTrue(t.where(t.dateColumn("test").isInSeptember()).numberColumn("index").contains(8.0));
-        assertTrue(t.where(t.dateColumn("test").isInOctober()).numberColumn("index").contains(9.0));
-        assertTrue(t.where(t.dateColumn("test").isInNovember()).numberColumn("index").contains(10.0));
-        assertTrue(t.where(t.dateColumn("test").isInDecember()).numberColumn("index").contains(11.0));
-
-        assertTrue(t.where(t.dateColumn("test").isInQ1()).nCol("index").contains(2));
-        assertTrue(t.where(t.dateColumn("test").isInQ2()).nCol("index").contains(4));
-        assertTrue(t.where(t.dateColumn("test").isInQ3()).nCol("index").contains(8));
-        assertTrue(t.where(t.dateColumn("test").isInQ4()).nCol("index").contains(11));
+        assertTrue(t.where(t.dateColumn("test").isInJanuary()).intColumn("index").contains(0));
+        assertTrue(t.where(t.dateColumn("test").isInFebruary()).intColumn("index").contains(1));
+        assertTrue(t.where(t.dateColumn("test").isInMarch()).intColumn("index").contains(2));
+        assertTrue(t.where(t.dateColumn("test").isInApril()).intColumn("index").contains(3));
+        assertTrue(t.where(t.dateColumn("test").isInMay()).intColumn("index").contains(4));
+        assertTrue(t.where(t.dateColumn("test").isInJune()).intColumn("index").contains(5));
+        assertTrue(t.where(t.dateColumn("test").isInJuly()).intColumn("index").contains(6));
+        assertTrue(t.where(t.dateColumn("test").isInAugust()).intColumn("index").contains(7));
+        assertTrue(t.where(t.dateColumn("test").isInSeptember()).intColumn("index").contains(8));
+        assertTrue(t.where(t.dateColumn("test").isInOctober()).intColumn("index").contains(9));
+        assertTrue(t.where(t.dateColumn("test").isInNovember()).intColumn("index").contains(10));
+        assertTrue(t.where(t.dateColumn("test").isInDecember()).intColumn("index").contains(11));
+
+        assertTrue(t.where(t.dateColumn("test").isInQ1()).intColumn("index").contains(2));
+        assertTrue(t.where(t.dateColumn("test").isInQ2()).intColumn("index").contains(4));
+        assertTrue(t.where(t.dateColumn("test").isInQ3()).intColumn("index").contains(8));
+        assertTrue(t.where(t.dateColumn("test").isInQ4()).intColumn("index").contains(11));
     }
 
     @Test
@@ -198,19 +198,19 @@ public class DateFiltersTest {
         IntColumn index = IntColumn.indexColumn("index", dateColumn.size(), 0);
         Table t = Table.create("test", dateColumn, index);
 
-        assertTrue(t.where(dateColumn.isBefore(packed)).nCol("index").contains(0));
-        assertTrue(t.where(dateColumn.isEqualTo(packed)).nCol("index").contains(1));
-        assertTrue(t.where(dateColumn.isAfter(packed)).nCol("index").contains(2));
+        assertTrue(t.where(dateColumn.isBefore(packed)).intColumn("index").contains(0));
+        assertTrue(t.where(dateColumn.isEqualTo(packed)).intColumn("index").contains(1));
+        assertTrue(t.where(dateColumn.isAfter(packed)).intColumn("index").contains(2));
         assertTrue(t.where(t.dateColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(1));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(1));
         assertTrue(t.where(t.dateColumn("test")
-                .isBetweenIncluding(beforeDate, afterDate)).nCol("index").contains(2));
+                .isBetweenIncluding(beforeDate, afterDate)).intColumn("index").contains(2));
         assertTrue(t.where(t.dateColumn("test")
-                .isBetweenIncluding(beforeDate, afterDate)).nCol("index").contains(0));
+                .isBetweenIncluding(beforeDate, afterDate)).intColumn("index").contains(0));
         assertFalse(t.where(t.dateColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(2));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(2));
         assertFalse(t.where(t.dateColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(0));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(0));
     }
 
     @Test
diff --git a/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeFiltersTest.java b/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeFiltersTest.java
index 619c28b4..05fd9f46 100644
--- a/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeFiltersTest.java
+++ b/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeFiltersTest.java
@@ -182,47 +182,47 @@ public class DateTimeFiltersTest {
         assertTrue(dateTimeColumn.isBefore(dateTime).contains(0));
         assertTrue(dateTimeColumn.isBefore(date).contains(0));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isBefore(dateTime)).nCol("index").contains(0));
+                .isBefore(dateTime)).intColumn("index").contains(0));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isBefore(date)).nCol("index").contains(0));
+                .isBefore(date)).intColumn("index").contains(0));
 
         assertTrue(dateTimeColumn.isEqualTo(dateTime).contains(1));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isEqualTo(dateTime)).nCol("index").contains(1));
+                .isEqualTo(dateTime)).intColumn("index").contains(1));
 
         assertTrue(dateTimeColumn.isAfter(dateTime).contains(2));
         assertTrue(dateTimeColumn.isAfter(date).contains(2));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isAfter(dateTime)).nCol("index").contains(2));
+                .isAfter(dateTime)).intColumn("index").contains(2));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isAfter(date)).nCol("index").contains(2));
+                .isAfter(date)).intColumn("index").contains(2));
 
         assertTrue(dateTimeColumn.isNotEqualTo(dateTime).contains(2));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isNotEqualTo(dateTime)).nCol("index").contains(2));
+                .isNotEqualTo(dateTime)).intColumn("index").contains(2));
 
         assertTrue(dateTimeColumn.isBetweenExcluding(beforeDate, afterDate).contains(1));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(1));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(1));
         assertTrue(dateTimeColumn.isBetweenExcluding(beforeDate, afterDate).contains(1));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(1));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(1));
 
         assertTrue(dateTimeColumn.isBetweenIncluding(beforeDate, afterDate).contains(2));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isBetweenIncluding(beforeDate, afterDate)).nCol("index").contains(2));
+                .isBetweenIncluding(beforeDate, afterDate)).intColumn("index").contains(2));
 
         assertTrue(dateTimeColumn.isBetweenIncluding(beforeDate, afterDate).contains(0));
         assertTrue(t.where(t.dateTimeColumn("test")
-                .isBetweenIncluding(beforeDate, afterDate)).nCol("index").contains(0));
+                .isBetweenIncluding(beforeDate, afterDate)).intColumn("index").contains(0));
 
         assertFalse(dateTimeColumn.isBetweenExcluding(beforeDate, afterDate).contains(2));
         assertFalse(t.where(t.dateTimeColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(2));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(2));
 
         assertFalse(dateTimeColumn.isBetweenExcluding(beforeDate, afterDate).contains(0));
         assertFalse(t.where(t.dateTimeColumn("test")
-                .isBetweenExcluding(beforeDate, afterDate)).nCol("index").contains(0));
+                .isBetweenExcluding(beforeDate, afterDate)).intColumn("index").contains(0));
     }
 
 
diff --git a/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctionsTest.java b/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctionsTest.java
index 38190eaf..efb961e9 100644
--- a/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctionsTest.java
+++ b/core/src/test/java/tech/tablesaw/columns/datetimes/DateTimeMapFunctionsTest.java
@@ -14,13 +14,7 @@
 
 package tech.tablesaw.columns.datetimes;
 
-import com.google.common.base.Strings;
-import org.junit.Test;
-import tech.tablesaw.api.DateColumn;
-import tech.tablesaw.api.DateTimeColumn;
-import tech.tablesaw.api.DoubleColumn;
-import tech.tablesaw.api.StringColumn;
-import tech.tablesaw.api.TimeColumn;
+import static org.junit.Assert.assertEquals;
 
 import java.time.LocalDateTime;
 import java.time.temporal.ChronoField;
@@ -29,7 +23,16 @@ import java.time.temporal.TemporalField;
 import java.time.temporal.WeekFields;
 import java.util.Locale;
 
-import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+
+import com.google.common.base.Strings;
+
+import tech.tablesaw.api.DateColumn;
+import tech.tablesaw.api.DateTimeColumn;
+import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.LongColumn;
+import tech.tablesaw.api.StringColumn;
+import tech.tablesaw.api.TimeColumn;
 
 /**
  * Tests for DateTimeMapFunctions
@@ -45,7 +48,7 @@ public class DateTimeMapFunctionsTest {
         LocalDateTime stop = start.plus(100_000L, ChronoUnit.MILLIS);
         startCol.append(start);
         stopCol.append(stop);
-        DoubleColumn result = startCol.differenceInMilliseconds(stopCol);
+        LongColumn result = startCol.differenceInMilliseconds(stopCol);
         assertEquals(100_000L, result.firstElement(),0.01);
     }
 
@@ -56,7 +59,7 @@ public class DateTimeMapFunctionsTest {
         startCol.append(start);
         stopCol.append(stop);
 
-        DoubleColumn result = startCol.differenceInSeconds(stopCol);
+        LongColumn result = startCol.differenceInSeconds(stopCol);
         assertEquals(100_000L, result.firstElement(), 0.01);
     }
 
@@ -67,7 +70,7 @@ public class DateTimeMapFunctionsTest {
         startCol.append(start);
         stopCol.append(stop);
 
-        DoubleColumn result = startCol.differenceInMinutes(stopCol);
+        LongColumn result = startCol.differenceInMinutes(stopCol);
         assertEquals(100_000L, result.firstElement(), 0.01);
     }
 
@@ -78,7 +81,7 @@ public class DateTimeMapFunctionsTest {
         startCol.append(start);
         stopCol.append(stop);
 
-        DoubleColumn result = startCol.differenceInHours(stopCol);
+        LongColumn result = startCol.differenceInHours(stopCol);
         assertEquals(100_000L, result.firstElement(), 0.01);
 
     }
@@ -90,7 +93,7 @@ public class DateTimeMapFunctionsTest {
         startCol.append(start);
         stopCol.append(stop);
 
-        DoubleColumn result = startCol.differenceInDays(stopCol);
+        LongColumn result = startCol.differenceInDays(stopCol);
         assertEquals(100_000L, result.firstElement(), 0.01);
     }
 
@@ -101,49 +104,49 @@ public class DateTimeMapFunctionsTest {
         startCol.append(start);
         stopCol.append(stop);
 
-        DoubleColumn result = startCol.differenceInYears(stopCol);
+        LongColumn result = startCol.differenceInYears(stopCol);
         assertEquals(10_000L, result.firstElement(), 0.01);
     }
 
     @Test
     public void testHour() {
         startCol.append(LocalDateTime.of(1984, 12, 12, 7, 30));
-        DoubleColumn hour = startCol.hour();
+        IntColumn hour = startCol.hour();
         assertEquals(7, hour.firstElement(), 0.0001);
     }
 
     @Test
     public void testYear() {
         startCol.append(LocalDateTime.of(1984, 12, 12, 7, 30));
-        DoubleColumn year = startCol.year();
+        IntColumn year = startCol.year();
         assertEquals(1984, year.firstElement(), 0.0001);
     }
 
     @Test
     public void testDayOfYear() {
         startCol.append(LocalDateTime.of(1984, 1, 5, 7, 30));
-        DoubleColumn dayOfYear = startCol.dayOfYear();
+        IntColumn dayOfYear = startCol.dayOfYear();
         assertEquals(5, dayOfYear.firstElement(), 0.0001);
     }
 
     @Test
     public void testDayOfMonth() {
         startCol.append(LocalDateTime.of(1984, 1, 22, 7, 30));
-        DoubleColumn dayOfMonth = startCol.dayOfMonth();
+        IntColumn dayOfMonth = startCol.dayOfMonth();
         assertEquals(22, dayOfMonth.firstElement(), 0.0001);
     }
 
     @Test
     public void testMinute() {
         startCol.append(LocalDateTime.of(1984, 1, 22, 7, 30));
-        DoubleColumn minute = startCol.minute();
+        IntColumn minute = startCol.minute();
         assertEquals(30, minute.firstElement(), 0.0001);
     }
 
     @Test
     public void testDayOfWeekValue() {
         startCol.append(LocalDateTime.of(2018, 4, 10, 7, 30));
-        DoubleColumn dayOfWeekValue = startCol.dayOfWeekValue();
+        IntColumn dayOfWeekValue = startCol.dayOfWeekValue();
         assertEquals(2, dayOfWeekValue.firstElement(), 0.0001);
     }
 
@@ -211,7 +214,7 @@ public class DateTimeMapFunctionsTest {
     public void testMonthValue() {
         LocalDateTime dateTime = LocalDateTime.of(2018, 4, 10, 7, 30);
         startCol.append(dateTime);
-        DoubleColumn month = startCol.monthValue();
+        IntColumn month = startCol.monthValue();
         assertEquals(4, month.get(0), 0.0001);
     }
 
@@ -219,7 +222,7 @@ public class DateTimeMapFunctionsTest {
     public void testMinuteOfDay() {
         LocalDateTime dateTime = LocalDateTime.of(2018, 4, 10, 7, 30);
         startCol.append(dateTime);
-        DoubleColumn minuteOfDay = startCol.minuteOfDay();
+        IntColumn minuteOfDay = startCol.minuteOfDay();
         assertEquals((7 * 60) + 30, minuteOfDay.get(0), 0.0001);
     }
 
@@ -227,7 +230,7 @@ public class DateTimeMapFunctionsTest {
     public void testSecondOfDay() {
         LocalDateTime dateTime = LocalDateTime.of(2018, 4, 10, 7, 30);
         startCol.append(dateTime);
-        DoubleColumn secondOfDay = startCol.secondOfDay();
+        IntColumn secondOfDay = startCol.secondOfDay();
         assertEquals(dateTime.get(ChronoField.SECOND_OF_DAY), secondOfDay.get(0), 0.0001);
     }
 
@@ -255,7 +258,7 @@ public class DateTimeMapFunctionsTest {
             dateTime = dateTime.plusDays(1);
             startCol.append(dateTime);
         }
-        DoubleColumn timeWindows = startCol.timeWindow(ChronoUnit.DAYS, 5);
+        LongColumn timeWindows = startCol.timeWindow(ChronoUnit.DAYS, 5);
         assertEquals(0, timeWindows.get(0), 0.0001);
         assertEquals(9, timeWindows.max(), 0.0001);
 
diff --git a/core/src/test/java/tech/tablesaw/columns/numbers/FloatDataWrapperTest.java b/core/src/test/java/tech/tablesaw/columns/numbers/FloatDataWrapperTest.java
deleted file mode 100644
index 4e1edc6c..00000000
--- a/core/src/test/java/tech/tablesaw/columns/numbers/FloatDataWrapperTest.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package tech.tablesaw.columns.numbers;
-
-import static org.junit.Assert.assertEquals;
-
-import org.junit.Test;
-
-import tech.tablesaw.api.FloatColumn;
-
-public class FloatDataWrapperTest {
-
-    @Test
-    public void getInt() {
-        float[] data = {1.2f, 1.8f};
-        FloatColumn column = FloatColumn.create("test", data);
-        assertEquals(1.0, column.getInt(0), 0.00001);
-        assertEquals(2.0, column.getInt(1), 0.00001);
-    }
-}
\ No newline at end of file
diff --git a/core/src/test/java/tech/tablesaw/io/csv/CsvReaderTest.java b/core/src/test/java/tech/tablesaw/io/csv/CsvReaderTest.java
index d98f9e36..cb5c0631 100644
--- a/core/src/test/java/tech/tablesaw/io/csv/CsvReaderTest.java
+++ b/core/src/test/java/tech/tablesaw/io/csv/CsvReaderTest.java
@@ -48,8 +48,8 @@ import org.junit.rules.ExpectedException;
 import tech.tablesaw.api.ColumnType;
 import tech.tablesaw.api.DateColumn;
 import tech.tablesaw.api.DateTimeColumn;
-import tech.tablesaw.api.DoubleColumn;
 import tech.tablesaw.api.IntColumn;
+import tech.tablesaw.api.LongColumn;
 import tech.tablesaw.api.Table;
 
 /**
@@ -143,9 +143,8 @@ public class CsvReaderTest {
 
     @Test
     public void testMillis() {
-
         long[] times = {1530486314124L, 1530488214124L};
-        DoubleColumn d = DoubleColumn.create("times", times);
+        LongColumn d = LongColumn.create("times", times);
         DateTimeColumn column = d.asDateTimes(ZoneOffset.UTC);
         assertEquals(1530486314124L, column.get(0).toInstant(ZoneOffset.UTC).toEpochMilli());
     }