From bad99a190490319d86bd5c7f9f23ac2fc4e72598 Mon Sep 17 00:00:00 2001
From: Joshua Storck <joshuastorck@gmail.com>
Date: Thu, 20 Oct 2016 16:22:41 -0400
Subject: [PATCH] Adding operator+[=] and operator/[=] for FloatingArray and
 IntegerArray.

This includes a design change that obviates the need for an ArrayView.
Instead, every array has an internal offset. Shallow copy is achieved by copy
constructor, though the current set of copy constructors don't yet
support a slice. Deep copy is still achieved through the Copy virtual
function.

More detailed explanation of the changes:

* Adding copy/move constructors for {Floating,Integer,Numeric}Array
* Adding various method for marking/getting nulls (valid bits) in
  integer arrays
* Changing data() and mutable_data() in NumericArray so that they
  return a pointer that starts at the array's offset
* Addition of Addable/Divisable classes (similar to Boost operators)
  for easy support of operator[+/]
* Unit test scaffolding for testing permutations of left/right hand
  side types on arithmetic operators
* Implementing IntegerArray::operator/, IntegerArray::operator+=
  FloatingArray::operator/=, FloatingArray::operator+=
---
 CMakeLists.txt                 |   2 +-
 pandas/native.pxd              |   2 +-
 pandas/native.pyx              |   2 +-
 src/pandas/array-test.cc       | 171 +++++++++++++++++++
 src/pandas/array.cc            |   5 +-
 src/pandas/array.h             |  25 ++-
 src/pandas/type.h              |   4 +-
 src/pandas/types/category.cc   |   3 +
 src/pandas/types/category.h    |  11 +-
 src/pandas/types/numeric.cc    |  99 +++++++++--
 src/pandas/types/numeric.h     | 297 +++++++++++++++++++++++++++++++--
 src/pandas/types/numeric_fwd.h |  42 +++++
 12 files changed, 607 insertions(+), 56 deletions(-)
 create mode 100644 src/pandas/types/numeric_fwd.h

diff --git a/CMakeLists.txt b/CMakeLists.txt
index be51d7c96..4daeeab4f 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -74,7 +74,7 @@ endif()
 #     GCC cannot always verify whether strict aliasing rules are indeed followed due to
 #     fundamental limitations in escape analysis, which can result in subtle bad code generation.
 #     This has a small perf hit but worth it to avoid hard to debug crashes.
-set(CXX_COMMON_FLAGS "-std=c++11 -fno-strict-aliasing -msse4.2 -Wall -Wno-sign-compare -Wno-deprecated -pthread -D__STDC_FORMAT_MACROS")
+set(CXX_COMMON_FLAGS "-std=c++1y -fno-strict-aliasing -msse4.2 -Wall -Wno-sign-compare -Wno-deprecated -pthread -D__STDC_FORMAT_MACROS")
 
 # compiler flags for different build types (run 'cmake -DCMAKE_BUILD_TYPE=<type> .')
 # For all builds:
diff --git a/pandas/native.pxd b/pandas/native.pxd
index 429993804..8a624d04b 100644
--- a/pandas/native.pxd
+++ b/pandas/native.pxd
@@ -74,7 +74,7 @@ cdef extern from "pandas/api.h" namespace "pandas":
         c_bool Equals(const DataType& other)
         string ToString()
 
-    ctypedef shared_ptr[DataType] TypePtr
+    ctypedef shared_ptr[const DataType] TypePtr
 
     cdef cppclass Int8Type(DataType):
         pass
diff --git a/pandas/native.pyx b/pandas/native.pyx
index b700dedbe..9d7de4b8b 100644
--- a/pandas/native.pyx
+++ b/pandas/native.pyx
@@ -277,7 +277,7 @@ cdef Array wrap_array(const lp.ArrayPtr& arr):
 
 cdef PandasType wrap_type(const lp.TypePtr& sp_type):
     cdef:
-        lp.DataType* type = sp_type.get()
+        const lp.DataType* type = sp_type.get()
         PandasType result
 
     if type.type() == lp.TypeId_CATEGORY:
diff --git a/src/pandas/array-test.cc b/src/pandas/array-test.cc
index 04acf5ba9..fe050576a 100644
--- a/src/pandas/array-test.cc
+++ b/src/pandas/array-test.cc
@@ -45,6 +45,177 @@ TEST_F(TestArray, Attrs) {
   ASSERT_EQ(values_.size(), array_->length());
 }
 
+template <template <typename> class LEFT_ARRAY_TYPE, typename LEFT_DATA_TYPE,
+    template <typename> class RIGHT_ARRAY_TYPE, typename RIGHT_DATA_TYPE,
+    std::size_t LENGTH = 10>
+class OperatorTest {
+ public:
+  OperatorTest()
+      : left_buffer_(std::make_shared<Buffer>(
+            reinterpret_cast<const std::uint8_t*>(Initialize(left_data_)),
+            LENGTH * sizeof(LEFT_C_TYPE))),
+        right_buffer_(std::make_shared<Buffer>(
+            reinterpret_cast<const std::uint8_t*>(Initialize(right_data_)),
+            LENGTH * sizeof(RIGHT_C_TYPE))),
+        left_array_(LENGTH, left_buffer_),
+        right_array_(LENGTH, right_buffer_) {}
+
+  template <typename OPERATOR>
+  void TestOperator(OPERATOR& operation) {
+    auto result = operation(left_array_, right_array_);
+    for (auto ii = 0; ii < left_array_.length(); ++ii) {
+      ASSERT_EQ(result.data()[ii], operation(left_data_[ii], right_data_[ii]));
+    }
+  }
+
+  template <typename OPERATOR, typename INPLACE_OPERATOR>
+  void TestInplaceOperator(OPERATOR& operation, INPLACE_OPERATOR& inplace_operation) {
+    auto result = operation(left_array_, right_array_);
+    for (auto ii = 0; ii < left_array_.length(); ++ii) {
+      ASSERT_EQ(result.data()[ii], operation(left_data_[ii], right_data_[ii]));
+    }
+    inplace_operation(left_array_, right_array_);
+    for (auto ii = 0; ii < left_array_.length(); ++ii) {
+      ASSERT_EQ(left_array_.data()[ii], operation(left_data_[ii], right_data_[ii]));
+    }
+    for (auto ii = 0; ii < left_array_.length(); ++ii) {
+      ASSERT_EQ(left_array_.data()[ii], result.data()[ii]);
+    }
+  }
+
+ private:
+  template <typename C_TYPE>
+  static C_TYPE* Initialize(C_TYPE (&value)[LENGTH]) {
+    for (auto ii = 0; ii < LENGTH; ++ii) {
+      // Start at 1 so that we don't get FPE with operator/
+      value[ii] = static_cast<C_TYPE>(ii + 1);
+    }
+    return value;
+  }
+
+  using LEFT_C_TYPE = typename LEFT_DATA_TYPE::c_type;
+
+  using RIGHT_C_TYPE = typename RIGHT_DATA_TYPE::c_type;
+
+  LEFT_C_TYPE left_data_[LENGTH];
+
+  RIGHT_C_TYPE right_data_[LENGTH];
+
+  std::shared_ptr<Buffer> left_buffer_;
+
+  std::shared_ptr<Buffer> right_buffer_;
+
+  LEFT_ARRAY_TYPE<LEFT_DATA_TYPE> left_array_;
+
+  RIGHT_ARRAY_TYPE<RIGHT_DATA_TYPE> right_array_;
+};
+
+TEST(TestArrayOperators, Addition) {
+  auto plus = [](auto const& left, auto const& right) { return left + right; };
+  auto plus_inplace = [](auto& left, auto const& right) { left += right; };
+
+  OperatorTest<IntegerArray, UInt8Type, IntegerArray, UInt16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt16Type, IntegerArray, UInt8Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int8Type, IntegerArray, Int16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int16Type, IntegerArray, Int8Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt32Type, IntegerArray, UInt16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt64Type, IntegerArray, UInt32Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int32Type, IntegerArray, Int64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int64Type, IntegerArray, Int32Type>().TestInplaceOperator(
+      plus, plus_inplace);
+
+  OperatorTest<IntegerArray, Int8Type, IntegerArray, Int8Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int16Type, IntegerArray, Int16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int32Type, IntegerArray, Int32Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, Int64Type, IntegerArray, Int64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt8Type, IntegerArray, UInt8Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt16Type, IntegerArray, UInt16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt32Type, IntegerArray, UInt32Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<IntegerArray, UInt64Type, IntegerArray, UInt64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+
+  OperatorTest<FloatingArray, FloatType, IntegerArray, UInt8Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<FloatingArray, FloatType, IntegerArray, UInt64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<FloatingArray, FloatType, IntegerArray, Int8Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<FloatingArray, FloatType, IntegerArray, Int64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, UInt16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, UInt64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, Int16Type>().TestInplaceOperator(
+      plus, plus_inplace);
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, Int64Type>().TestInplaceOperator(
+      plus, plus_inplace);
+}
+
+TEST(TestArrayOperators, Division) {
+  auto divide = [](auto const& left, auto const& right) { return left / right; };
+  auto divide_inplace = [](auto& left, auto const& right) { left /= right; };
+
+  OperatorTest<IntegerArray, UInt8Type, IntegerArray, UInt16Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt16Type, IntegerArray, UInt8Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int8Type, IntegerArray, Int16Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int16Type, IntegerArray, Int8Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt32Type, IntegerArray, UInt16Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt64Type, IntegerArray, UInt32Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int32Type, IntegerArray, Int64Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int64Type, IntegerArray, Int32Type>().TestOperator(divide);
+
+  OperatorTest<IntegerArray, Int8Type, IntegerArray, Int8Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int16Type, IntegerArray, Int16Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int32Type, IntegerArray, Int32Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, Int64Type, IntegerArray, Int64Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt8Type, IntegerArray, UInt8Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt16Type, IntegerArray, UInt16Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt32Type, IntegerArray, UInt32Type>().TestOperator(divide);
+  OperatorTest<IntegerArray, UInt64Type, IntegerArray, UInt64Type>().TestOperator(divide);
+
+  OperatorTest<FloatingArray, FloatType, IntegerArray, UInt8Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, FloatType, IntegerArray, UInt64Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, FloatType, IntegerArray, Int8Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, FloatType, IntegerArray, Int64Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, FloatType, FloatingArray, FloatType>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, FloatType, FloatingArray, DoubleType>().TestInplaceOperator(
+      divide, divide_inplace);
+
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, UInt16Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, UInt64Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, Int16Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, DoubleType, IntegerArray, Int64Type>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, DoubleType, FloatingArray, FloatType>().TestInplaceOperator(
+      divide, divide_inplace);
+  OperatorTest<FloatingArray, DoubleType, FloatingArray, DoubleType>()
+      .TestInplaceOperator(divide, divide_inplace);
+}
+
 // ----------------------------------------------------------------------
 // Array view object
 
diff --git a/src/pandas/array.cc b/src/pandas/array.cc
index fbc499fa4..fa52e8e7e 100644
--- a/src/pandas/array.cc
+++ b/src/pandas/array.cc
@@ -11,11 +11,8 @@ namespace pandas {
 // ----------------------------------------------------------------------
 // Array
 
-Array::Array(const std::shared_ptr<DataType>& type, int64_t length)
-    : type_(type), length_(length) {}
-
 Status Array::Copy(std::shared_ptr<Array>* out) const {
-  return Copy(0, length_, out);
+  return Copy(0, length(), out);
 }
 
 // ----------------------------------------------------------------------
diff --git a/src/pandas/array.h b/src/pandas/array.h
index a22076f55..3bb56f127 100644
--- a/src/pandas/array.h
+++ b/src/pandas/array.h
@@ -20,9 +20,19 @@ class Array {
  public:
   virtual ~Array() {}
 
-  int64_t length() const { return length_; }
-  std::shared_ptr<DataType> type() const { return type_; }
-  DataType::TypeId type_id() const { return type_->type(); }
+  virtual int64_t length() const = 0;
+  // There are two methods to obtain the data type.
+  // The signature without a shared_ptr allows sub-classes
+  // to have a covariant return type, which eliminates the
+  // need/danger of doing a static_cast when dealing with
+  // a concrete sub-class. Ideally, the shared_ptr signature
+  // would suffice, but the compiler cannot treat a shared_ptr
+  // to a base class and a shared_ptr to a subclass as a
+  // covariant return type.
+  virtual TypePtr type() const = 0;
+  virtual const DataType& type_reference() const = 0;
+
+  DataType::TypeId type_id() const { return type()->type(); }
 
   // Copy a section of the array into a new output array
   virtual Status Copy(
@@ -42,13 +52,10 @@ class Array {
   virtual bool owns_data() const = 0;
 
  protected:
-  std::shared_ptr<DataType> type_;
-  int64_t length_;
+  Array() {}
 
-  Array(const std::shared_ptr<DataType>& type, int64_t length);
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(Array);
+  Array(const Array& other) = default;
+  Array(Array&& other) = default;
 };
 
 // An object that is a view on a section of another array (possibly the whole
diff --git a/src/pandas/type.h b/src/pandas/type.h
index a185f58f2..ed6bfcc60 100644
--- a/src/pandas/type.h
+++ b/src/pandas/type.h
@@ -57,7 +57,7 @@ class DataType {
 
   virtual std::string ToString() const = 0;
 
-  virtual bool Equals(const DataType& other) { return type_ == other.type_; }
+  virtual bool Equals(const DataType& other) const { return type_ == other.type_; }
 
   TypeId type() const { return type_; }
 
@@ -65,7 +65,7 @@ class DataType {
   TypeId type_;
 };
 
-typedef std::shared_ptr<DataType> TypePtr;
+using TypePtr = std::shared_ptr<const DataType>;
 
 class PANDAS_EXPORT TimestampType : public DataType {
  public:
diff --git a/src/pandas/types/category.cc b/src/pandas/types/category.cc
index e9089ad97..364389df8 100644
--- a/src/pandas/types/category.cc
+++ b/src/pandas/types/category.cc
@@ -5,6 +5,9 @@
 
 namespace pandas {
 
+CategoryArray::CategoryArray(ArrayView codes, const std::shared_ptr<CategoryType>& type)
+    : codes_(codes), type_(type) {}
+
 std::string CategoryType::ToString() const {
   std::stringstream s;
   s << "category<" << category_type()->ToString() << ">";
diff --git a/src/pandas/types/category.h b/src/pandas/types/category.h
index 5cfe4aaf9..ce780944f 100644
--- a/src/pandas/types/category.h
+++ b/src/pandas/types/category.h
@@ -20,7 +20,9 @@ struct CategoryType : public DataType {
 
   std::string ToString() const override;
 
-  std::shared_ptr<DataType> category_type() const { return categories_.data()->type(); }
+  std::shared_ptr<const DataType> category_type() const {
+    return categories_.data()->type();
+  }
 
   const ArrayView& categories() const { return categories_; }
 
@@ -30,14 +32,15 @@ struct CategoryType : public DataType {
 
 class CategoryArray : public Array {
  public:
+  CategoryArray(ArrayView codes, const std::shared_ptr<CategoryType>& type);
+
   const ArrayView& codes() const { return codes_; }
 
-  const ArrayView& categories() const {
-    return static_cast<CategoryType*>(type_.get())->categories();
-  }
+  const ArrayView& categories() const { return type_->categories(); }
 
  private:
   ArrayView codes_;
+  std::shared_ptr<CategoryType> type_;
 };
 
 }  // namespace pandas
diff --git a/src/pandas/types/numeric.cc b/src/pandas/types/numeric.cc
index 467a162de..fb375c62d 100644
--- a/src/pandas/types/numeric.cc
+++ b/src/pandas/types/numeric.cc
@@ -20,27 +20,60 @@ namespace pandas {
 // Generic numeric class
 
 template <typename TYPE>
-NumericArray<TYPE>::NumericArray(
-    const DataTypePtr& type, int64_t length, const std::shared_ptr<Buffer>& data)
-    : Array(type, length), data_(data) {}
+NumericArray<TYPE>::NumericArray(const DataTypePtr& type, int64_t length, int64_t offset,
+    const std::shared_ptr<Buffer>& data)
+    : type_(type), length_(length), offset_(offset), data_(data) {}
+
+template <typename TYPE>
+NumericArray<TYPE>::NumericArray(const NumericArray& other) = default;
+
+template <typename TYPE>
+NumericArray<TYPE>::NumericArray(NumericArray&& other) = default;
 
 template <typename TYPE>
 auto NumericArray<TYPE>::data() const -> const T* {
-  return reinterpret_cast<const T*>(data_->data());
+  return reinterpret_cast<const T*>(data_->data()) + offset_;
 }
 
 template <typename TYPE>
 auto NumericArray<TYPE>::mutable_data() const -> T* {
   auto mutable_buf = static_cast<MutableBuffer*>(data_.get());
-  return reinterpret_cast<T*>(mutable_buf->mutable_data());
+  return reinterpret_cast<T*>(mutable_buf->mutable_data()) + offset_;
+}
+
+template <typename TYPE>
+TypePtr NumericArray<TYPE>::type() const {
+  return type_;
+}
+
+template <typename TYPE>
+Status NumericArray<TYPE>::EnsureMutableAndCheckChange(bool& changed) {
+  if (this->data_.use_count() == 1) {
+    changed = false;
+    return Status::OK();
+  }
+  changed = true;
+  std::shared_ptr<Buffer> new_data;
+  RETURN_NOT_OK(this->data_->Copy(
+      this->offset_, this->length_ * sizeof(typename TYPE::c_type), &new_data));
+  this->data_ = new_data;
+  this->offset_ = 0;
+  return Status::OK();
 }
 
 // ----------------------------------------------------------------------
 // Floating point class
 
 template <typename TYPE>
-FloatingArray<TYPE>::FloatingArray(int64_t length, const std::shared_ptr<Buffer>& data)
-    : NumericArray<TYPE>(TYPE::SINGLETON, length, data) {}
+FloatingArray<TYPE>::FloatingArray(
+    int64_t length, const std::shared_ptr<Buffer>& data, int64_t offset)
+    : NumericArray<TYPE>(TYPE::SINGLETON, length, offset, data) {}
+
+template <typename TYPE>
+FloatingArray<TYPE>::FloatingArray(const FloatingArray& other) = default;
+
+template <typename TYPE>
+FloatingArray<TYPE>::FloatingArray(FloatingArray&& other) = default;
 
 template <typename TYPE>
 int64_t FloatingArray<TYPE>::GetNullCount() {
@@ -80,20 +113,28 @@ bool FloatingArray<TYPE>::owns_data() const {
 template class FloatingArray<FloatType>;
 template class FloatingArray<DoubleType>;
 
-// ----------------------------------------------------------------------
-// Types integers
+template class NumericArray<FloatType>;
+template class NumericArray<DoubleType>;
 
 // ----------------------------------------------------------------------
 // Typed integers
 
 template <typename TYPE>
-IntegerArray<TYPE>::IntegerArray(int64_t length, const std::shared_ptr<Buffer>& data)
-    : IntegerArray(length, data, nullptr) {}
+IntegerArray<TYPE>::IntegerArray(
+    int64_t length, const std::shared_ptr<Buffer>& data, int64_t offset)
+    : IntegerArray(length, data, nullptr, offset) {}
 
 template <typename TYPE>
 IntegerArray<TYPE>::IntegerArray(int64_t length, const std::shared_ptr<Buffer>& data,
-    const std::shared_ptr<Buffer>& valid_bits)
-    : NumericArray<TYPE>(TYPE::SINGLETON, length, data), valid_bits_(valid_bits) {}
+    const std::shared_ptr<Buffer>& valid_bits, int64_t offset)
+    : NumericArray<TYPE>(TYPE::SINGLETON, length, offset, data),
+      valid_bits_(valid_bits) {}
+
+template <typename TYPE>
+IntegerArray<TYPE>::IntegerArray(const IntegerArray& other) = default;
+
+template <typename TYPE>
+IntegerArray<TYPE>::IntegerArray(IntegerArray&& other) = default;
 
 template <typename TYPE>
 int64_t IntegerArray<TYPE>::GetNullCount() {
@@ -126,12 +167,14 @@ Status IntegerArray<TYPE>::Copy(
   std::shared_ptr<Buffer> copied_data;
   std::shared_ptr<Buffer> copied_valid_bits;
 
-  RETURN_NOT_OK(this->data_->Copy(offset * itemsize, length * itemsize, &copied_data));
+  RETURN_NOT_OK(this->data_->Copy(
+      (this->offset_ + offset) * itemsize, length * itemsize, &copied_data));
 
   if (valid_bits_) {
-    RETURN_NOT_OK(CopyBitmap(this->data_, offset, length, &copied_valid_bits));
+    RETURN_NOT_OK(CopyBitmap(
+        this->valid_bits_, this->offset_ + offset, length, &copied_valid_bits));
   }
-  *out = std::make_shared<FloatingArray<TYPE>>(length, copied_data);
+  *out = std::make_shared<IntegerArray<TYPE>>(length, copied_data);
   return Status::OK();
 }
 
@@ -176,6 +219,21 @@ Status IntegerArray<TYPE>::SetItem(int64_t i, PyObject* val) {
   return Status::OK();
 }
 
+template <typename TYPE>
+Status IntegerArray<TYPE>::EnsureMutable() {
+  bool changed;
+  RETURN_NOT_OK(NumericArray<TYPE>::EnsureMutableAndCheckChange(changed));
+  if (!changed) { return Status::OK(); }
+  // Copy the valid bits
+  if (valid_bits_) {
+    std::shared_ptr<Buffer> new_valid_bits;
+    RETURN_NOT_OK(
+        CopyBitmap(this->valid_bits_, this->offset_, this->length_, &new_valid_bits));
+    this->valid_bits_ = std::move(new_valid_bits);
+  }
+  return Status::OK();
+}
+
 // Instantiate templates
 template class IntegerArray<UInt8Type>;
 template class IntegerArray<Int8Type>;
@@ -186,4 +244,13 @@ template class IntegerArray<Int32Type>;
 template class IntegerArray<UInt64Type>;
 template class IntegerArray<Int64Type>;
 
+template class NumericArray<UInt8Type>;
+template class NumericArray<Int8Type>;
+template class NumericArray<UInt16Type>;
+template class NumericArray<Int16Type>;
+template class NumericArray<UInt32Type>;
+template class NumericArray<Int32Type>;
+template class NumericArray<UInt64Type>;
+template class NumericArray<Int64Type>;
+
 }  // namespace pandas
diff --git a/src/pandas/types/numeric.h b/src/pandas/types/numeric.h
index b5901102c..04e7f96ee 100644
--- a/src/pandas/types/numeric.h
+++ b/src/pandas/types/numeric.h
@@ -8,32 +8,94 @@
 #include "pandas/array.h"
 #include "pandas/common.h"
 #include "pandas/type.h"
+#include "pandas/types/numeric_fwd.h"
 
 namespace pandas {
 
+//
+// Helper class that will define operator+ if the derived
+// class defines operator+=
+//
+template <typename DERIVED>
+class Addable {
+ public:
+  template <typename OTHER_TYPE>
+  auto operator+(const OTHER_TYPE& other) const ->
+      typename std::remove_reference<decltype(
+          std::declval<DERIVED>() += std::declval<OTHER_TYPE>())>::type {
+    DERIVED copy(*static_cast<const DERIVED*>(this));
+    copy += other;
+    return std::move(copy);
+  }
+};
+
+//
+// Same as Addable above.
+//
+template <typename DERIVED>
+class Divisable {
+ public:
+  template <typename OTHER_TYPE>
+  auto operator/(const OTHER_TYPE& other) const ->
+      typename std::remove_reference<decltype(
+          std::declval<DERIVED>() /= std::declval<OTHER_TYPE>())>::type {
+    DERIVED copy(*static_cast<const DERIVED*>(this));
+    copy /= other;
+    return std::move(copy);
+  }
+};
+
 template <typename TYPE>
 class PANDAS_EXPORT NumericArray : public Array {
  public:
+  using expected_base = NumericType<TYPE, typename TYPE::c_type, TYPE::type_id>;
+  static_assert(std::is_assignable<expected_base, TYPE>::value,
+      "NumericArray's type must be a numeric type");
   using T = typename TYPE::c_type;
   using DataTypePtr = std::shared_ptr<TYPE>;
   using Array::Array;
 
-  NumericArray(
-      const DataTypePtr& type, int64_t length, const std::shared_ptr<Buffer>& data);
+  NumericArray(const DataTypePtr& type, int64_t length, int64_t offset,
+      const std::shared_ptr<Buffer>& data);
+
+  // By making this final, the compiler can inline it when
+  // we are invoking it through a reference to this class or
+  // a sub-class, which happens in a number of the operators
+  int64_t length() const final { return length_; }
 
   auto data() const -> const T*;
   auto mutable_data() const -> T*;
 
+  TypePtr type() const override;
+
+  // Performs copy-on-write semantics if needed, and sets
+  // changed to true if it occurred.
+  Status EnsureMutableAndCheckChange(bool& changed);
+
+  // Despite being virtual, compiler could inline this if
+  // the call is performed with a NumericArray reference
+  const TYPE& type_reference() const override { return *type_; }
+
  protected:
+  NumericArray(const NumericArray&);
+  NumericArray(NumericArray&&);
+
+  std::shared_ptr<TYPE> type_;
+  int64_t length_;
+  int64_t offset_;
   std::shared_ptr<Buffer> data_;
 };
 
 template <typename TYPE>
-class PANDAS_EXPORT IntegerArray : public NumericArray<TYPE> {
+class PANDAS_EXPORT IntegerArray : public NumericArray<TYPE>,
+                                   public Addable<IntegerArray<TYPE>> {
  public:
-  IntegerArray(int64_t length, const std::shared_ptr<Buffer>& data);
+  IntegerArray(int64_t length, const std::shared_ptr<Buffer>& data, int64_t offset = 0);
   IntegerArray(int64_t length, const std::shared_ptr<Buffer>& data,
-      const std::shared_ptr<Buffer>& valid_bits);
+      const std::shared_ptr<Buffer>& valid_bits, int64_t offset = 0);
+
+  IntegerArray(const IntegerArray& other);
+  IntegerArray(IntegerArray&& other);
 
   int64_t GetNullCount() override;
 
@@ -44,14 +106,153 @@ class PANDAS_EXPORT IntegerArray : public NumericArray<TYPE> {
 
   bool owns_data() const override;
 
+  Status EnsureMutable();
+
+  // In the case where there are any nulls in this array,
+  // this function will be faster than calling GetNullCount() > 0.
+  bool HasNulls() const {
+    for (auto ii = 0; ii < this->length_; ++ii) {
+      if (IsNull(ii)) { return true; }
+    }
+    return false;
+  }
+
+  // Marks the value at the specified index as null. A pre-condition
+  // is that the valid_bits for this object have already been
+  // initialized.
+  void SetNull(int64_t i) {
+    // Pre-condition is that valid_bits_ is non-null because
+    // operations that would set them (e.g. - operator+=)
+    // would first create the valid_bits_ if needed
+    BitUtil::SetBit(
+        static_cast<MutableBuffer&>(*valid_bits_).mutable_data(), this->offset_ + i);
+  }
+
+  // Marks the value at the specified index as null. A pre-condition
+  // is that the valid_bits for this object have already been
+  // initialized.
+  void SetValid(int64_t i) {
+    // Pre-condition is that valid_bits_ is non-null because
+    // operations that would set them (e.g. - operator+=)
+    // would first create the valid_bits_ if needed
+    BitUtil::ClearBit(
+        static_cast<MutableBuffer&>(*valid_bits_).mutable_data(), this->offset_ + i);
+  }
+
+  // Returns whether or not the value at the specified index has
+  // been marked as null.
+  bool IsNull(int64_t i) const {
+    if (valid_bits_) {
+      // TODO: range check?
+      return BitUtil::GetBit(valid_bits_->data(), this->offset_ + i);
+    } else {
+      return false;
+    }
+  }
+
+  // Copy the list of nulls (aka valid bits) to a buffer
+  Status CopyNulls(int64_t length, std::shared_ptr<Buffer>* out) const {
+    // Pre-condition is that we already have valid_bits_ set
+    return valid_bits_->Copy(this->offset_, BitUtil::BytesForBits(length), out);
+  }
+
+  template <typename OTHER_TYPE>
+  IntegerArray<TYPE>& operator+=(const IntegerArray<OTHER_TYPE>& other) {
+    // TODO: check Status and throw exception
+    EnsureMutable();
+    auto length = std::min(this->length_, other.length());
+    if (HasNulls()) {
+      // Perform a |= on the valid bits if the other array has them.
+      // TODO: do this more efficiently, perhaps by passing our
+      //       valid_bits_ like this: other.BitwiseOr(*valid_bits_).
+      //       It might also be useful to have a class that wraps
+      //       much of the valid bit logic.
+      if (other.HasNulls()) {
+        for (auto ii = 0; ii < length; ++ii) {
+          if (IsNull(ii) || other.IsNull(ii)) {
+            SetNull(ii);
+          } else {
+            SetValid(ii);
+          }
+        }
+      }
+    } else if (other.HasNulls()) {
+      std::shared_ptr<Buffer> new_valid_bits;
+      // TODO: check Status and throw an exception
+      other.CopyNulls(length, &new_valid_bits);
+    }
+    auto this_data = this->mutable_data();
+    auto other_data = other.data();
+    if (HasNulls()) {
+      for (auto ii = 0; ii < length; ++ii, ++this_data, ++other_data) {
+        if (!IsNull(ii)) { *this_data += *other_data; }
+      }
+    } else {
+      // Can just copy directly
+      // TODO: optimize with SIMD when TYPE::size == OTHER_TYPE::size
+      for (auto ii = 0; ii < length; ++ii, ++this_data, ++other_data) {
+        *this_data += *other_data;
+      }
+    }
+    return *this;
+  }
+
+  // Compile time logic for determining the result type of
+  // a division of two integers.
+  template <typename OTHER_TYPE>
+  class IntegerDivideInteger {
+   public:
+    static constexpr auto max_value =
+        (std::numeric_limits<typename TYPE::c_type>::max() >
+                    std::numeric_limits<typename OTHER_TYPE::c_type>::max()
+                ? std::numeric_limits<typename TYPE::c_type>::max()
+                : std::numeric_limits<typename OTHER_TYPE::c_type>::max());
+
+    static constexpr auto needs_double =
+        (max_value > std::numeric_limits<typename FloatType::c_type>::max());
+
+    using type = typename std::conditional<needs_double, DoubleType, FloatType>::type;
+  };
+
+  template <typename OTHER_TYPE>
+  FloatingArray<typename IntegerDivideInteger<OTHER_TYPE>::type> operator/(
+      const IntegerArray<OTHER_TYPE>& other) const {
+    using RESULT_TYPE = typename IntegerDivideInteger<OTHER_TYPE>::type;
+    using CAST_TYPE = typename RESULT_TYPE::c_type;
+    auto length = std::min(this->length_, other.length());
+    auto data = std::make_shared<PoolBuffer>();
+    // TODO: check Status and throw an exception
+    data->Resize(length);
+    FloatingArray<RESULT_TYPE> result(length, std::shared_ptr<Buffer>(std::move(data)));
+    // TODO: use SIMD (Intel's _mm*_div_p{s,d} intrinsics) depending on the size
+    //       of the resulting type
+    auto this_data = this->data();
+    auto other_data = other.data();
+    // The result was just created and has a zero offset, so no need to add it
+    auto result_data = result.mutable_data();
+    for (int64_t ii = 0; ii < length; ++ii, ++this_data, ++other_data, ++result_data) {
+      *result_data =
+          static_cast<CAST_TYPE>(*this_data) / static_cast<CAST_TYPE>(*other_data);
+    }
+    return result;
+  }
+
  private:
   std::shared_ptr<Buffer> valid_bits_;
 };
 
 template <typename TYPE>
-class PANDAS_EXPORT FloatingArray : public NumericArray<TYPE> {
+class PANDAS_EXPORT FloatingArray : public NumericArray<TYPE>,
+                                    public Addable<FloatingArray<TYPE>>,
+                                    public Divisable<FloatingArray<TYPE>> {
  public:
-  FloatingArray(int64_t length, const std::shared_ptr<Buffer>& data);
+  static_assert(std::is_floating_point<typename TYPE::c_type>::value,
+      "Only floating point types are allowed for FloatingArray's DataType");
+
+  FloatingArray(int64_t length, const std::shared_ptr<Buffer>& data, int64_t offset = 0);
+
+  FloatingArray(const FloatingArray& other);
+  FloatingArray(FloatingArray&& other);
 
   Status Copy(int64_t offset, int64_t length, std::shared_ptr<Array>* out) const override;
   PyObject* GetItem(int64_t i) override;
@@ -60,22 +261,71 @@ class PANDAS_EXPORT FloatingArray : public NumericArray<TYPE> {
   int64_t GetNullCount() override;
 
   bool owns_data() const override;
-};
 
-using FloatArray = FloatingArray<FloatType>;
-using DoubleArray = FloatingArray<DoubleType>;
+  //
+  // Arithmetic operators
+  //
 
-using Int8Array = IntegerArray<Int8Type>;
-using UInt8Array = IntegerArray<UInt8Type>;
+  template <template <typename> class ARRAY_TYPE, typename DATA_TYPE>
+  FloatingArray<TYPE>& operator/=(const ARRAY_TYPE<DATA_TYPE>& other) {
+    static auto divide_equals = [](
+        typename TYPE::c_type& left, auto const& right) { left /= right; };
+    return EvaluateBinaryOperator(divide_equals, other);
+  }
 
-using Int16Array = IntegerArray<Int16Type>;
-using UInt16Array = IntegerArray<UInt16Type>;
+  template <template <typename> class ARRAY_TYPE, typename DATA_TYPE>
+  FloatingArray<TYPE>& operator+=(const ARRAY_TYPE<DATA_TYPE>& other) {
+    static auto plus_equals = [](
+        typename TYPE::c_type& left, auto const& right) { left += right; };
+    return EvaluateBinaryOperator(plus_equals, other);
+  }
 
-using Int32Array = IntegerArray<Int32Type>;
-using UInt32Array = IntegerArray<UInt32Type>;
+ private:
+  // Evaluates an arithmetic operator. If the right hand side is an IntegerArray,
+  // this will evaluate whether or not the elements are null and then set
+  // the results to NaN.
+  template <typename OPERATOR, template <typename> class ARRAY_TYPE, typename DATA_TYPE>
+  FloatingArray<TYPE>& EvaluateBinaryOperator(
+      const OPERATOR& operation, const ARRAY_TYPE<DATA_TYPE>& other) {
+    bool changed;
+    this->EnsureMutableAndCheckChange(changed);
+    auto added = std::min(this->length_, other.length());
+    auto this_data = this->mutable_data();
+    auto other_data = other.data();
+    // TODO: use SIMD, special casing for the c_type of both
+    //       the left and right hand sides
+    for (auto ii = 0; ii < added; ++ii, ++this_data, ++other_data) {
+      // Handle integer NaN's by using SFINAE on the type. If the right
+      // hand side of this function is a FloatingArray, this will evaluate to
+      // false at compile time
+      if (IsNull(other, ii)) {
+        *this_data = std::numeric_limits<typename TYPE::c_type>::quiet_NaN();
+      } else {
+        operation(*this_data, *other_data);
+      }
+    }
+    return *this;
+  }
 
-using Int64Array = IntegerArray<Int64Type>;
-using UInt64Array = IntegerArray<UInt64Type>;
+  // SFINAE for IntegerArray
+  template <template <typename> class ARRAY_TYPE, typename DATA_TYPE>
+  typename std::enable_if<
+      std::is_assignable<IntegerArray<DATA_TYPE>, ARRAY_TYPE<DATA_TYPE>>::value,
+      bool>::type
+  IsNull(const ARRAY_TYPE<DATA_TYPE>& other, int64_t i) {
+    return other.IsNull(i);
+  }
+
+  // SFINAE for FloatingArray, compiler should remove any branches
+  // that depend on this function
+  template <template <typename> class ARRAY_TYPE, typename DATA_TYPE>
+  constexpr typename std::enable_if<
+      !std::is_assignable<IntegerArray<DATA_TYPE>, ARRAY_TYPE<DATA_TYPE>>::value,
+      bool>::type
+  IsNull(const ARRAY_TYPE<DATA_TYPE>& other, int64_t i) {
+    return false;
+  }
+};
 
 // Only instantiate these templates once
 extern template class PANDAS_EXPORT IntegerArray<Int8Type>;
@@ -89,4 +339,15 @@ extern template class PANDAS_EXPORT IntegerArray<UInt64Type>;
 extern template class PANDAS_EXPORT FloatingArray<FloatType>;
 extern template class PANDAS_EXPORT FloatingArray<DoubleType>;
 
+extern template class PANDAS_EXPORT NumericArray<Int8Type>;
+extern template class PANDAS_EXPORT NumericArray<UInt8Type>;
+extern template class PANDAS_EXPORT NumericArray<Int16Type>;
+extern template class PANDAS_EXPORT NumericArray<UInt16Type>;
+extern template class PANDAS_EXPORT NumericArray<Int32Type>;
+extern template class PANDAS_EXPORT NumericArray<UInt32Type>;
+extern template class PANDAS_EXPORT NumericArray<Int64Type>;
+extern template class PANDAS_EXPORT NumericArray<UInt64Type>;
+extern template class PANDAS_EXPORT NumericArray<FloatType>;
+extern template class PANDAS_EXPORT NumericArray<DoubleType>;
+
 }  // namespace pandas
diff --git a/src/pandas/types/numeric_fwd.h b/src/pandas/types/numeric_fwd.h
new file mode 100644
index 000000000..555379dd1
--- /dev/null
+++ b/src/pandas/types/numeric_fwd.h
@@ -0,0 +1,42 @@
+// This file is a part of pandas. See LICENSE for details about reuse and
+// copyright holders
+
+#pragma once
+
+namespace pandas {
+
+// Forward declarations for numeric types
+// and numeric arrays
+
+class FloatType;
+class DoubleType;
+class Int8Type;
+class UInt8Type;
+class Int16Type;
+class UInt16Type;
+class Int32Type;
+class UInt32Type;
+class Int64Type;
+class UInt64Type;
+
+template <typename TYPE>
+class IntegerArray;
+
+template <typename TYPE>
+class FloatingArray;
+
+using FloatArray = FloatingArray<FloatType>;
+using DoubleArray = FloatingArray<DoubleType>;
+
+using Int8Array = IntegerArray<Int8Type>;
+using UInt8Array = IntegerArray<UInt8Type>;
+
+using Int16Array = IntegerArray<Int16Type>;
+using UInt16Array = IntegerArray<UInt16Type>;
+
+using Int32Array = IntegerArray<Int32Type>;
+using UInt32Array = IntegerArray<UInt32Type>;
+
+using Int64Array = IntegerArray<Int64Type>;
+using UInt64Array = IntegerArray<UInt64Type>;
+}