ARROW-5855: [Python] Support for Duration (timedelta) type

cpp/src/arrow/python/arrow_to_pandas.cc

@@ -131,6 +131,7 @@ static inline bool ListTypeSupported(const DataType& type) {
     case Type::TIME32:
     case Type::TIME64:
     case Type::TIMESTAMP:
+    case Type::DURATION:
     case Type::NA:  // empty list
       // The above types are all supported.
       return true;
@@ -230,6 +231,29 @@ inline void set_numpy_metadata(int type, const DataType* datatype, PyArray_Descr
       // datatype->type == Type::DATE64
       date_dtype->meta.base = NPY_FR_D;
     }
+  } else if (type == NPY_TIMEDELTA) {
+    auto date_dtype = reinterpret_cast<PyArray_DatetimeDTypeMetaData*>(out->c_metadata);
+    if (datatype->id() == Type::DURATION) {
+      const auto& duration_type = checked_cast<const DurationType&>(*datatype);
+
+      switch (duration_type.unit()) {
+        case DurationType::Unit::SECOND:
+          date_dtype->meta.base = NPY_FR_s;
+          break;
+        case DurationType::Unit::MILLI:
+          date_dtype->meta.base = NPY_FR_ms;
+          break;
+        case DurationType::Unit::MICRO:
+          date_dtype->meta.base = NPY_FR_us;
+          break;
+        case DurationType::Unit::NANO:
+          date_dtype->meta.base = NPY_FR_ns;
+          break;
+      }
+    } else {
+      // datatype->type == Type::DATE64
+      date_dtype->meta.base = NPY_FR_D;
+    }
   }
 }
 
@@ -285,6 +309,7 @@ class PandasBlock {
     BOOL,
     DATETIME,
     DATETIME_WITH_TZ,
+    TIMEDELTA,
     CATEGORICAL
   };
 
@@ -721,6 +746,19 @@ inline void ConvertDatetimeNanos(const ChunkedArray& data, int64_t* out_values)
   }
 }
 
+template <typename T, int64_t SHIFT>
+inline void ConvertTimedeltaNanos(const ChunkedArray& data, int64_t* out_values) {
+  for (int c = 0; c < data.num_chunks(); c++) {
+    const auto& arr = *data.chunk(c);
+    const T* in_values = GetPrimitiveValues<T>(arr);
+
+    for (int64_t i = 0; i < arr.length(); ++i) {
+      *out_values++ = arr.IsNull(i) ? kPandasTimestampNull
+                                    : (static_cast<int64_t>(in_values[i]) * SHIFT);
+    }
+  }
+}
+
 template <typename Type>
 static Status ConvertDates(const PandasOptions& options, const ChunkedArray& data,
                            PyObject** out_values) {
@@ -846,6 +884,7 @@ class ObjectBlock : public PandasBlock {
         CONVERTLISTSLIKE_CASE(Time32Type, TIME32)
         CONVERTLISTSLIKE_CASE(Time64Type, TIME64)
         CONVERTLISTSLIKE_CASE(TimestampType, TIMESTAMP)
+        CONVERTLISTSLIKE_CASE(DurationType, DURATION)
         CONVERTLISTSLIKE_CASE(FloatType, FLOAT)
         CONVERTLISTSLIKE_CASE(DoubleType, DOUBLE)
         CONVERTLISTSLIKE_CASE(DecimalType, DECIMAL)
@@ -1082,6 +1121,53 @@ class DatetimeBlock : public PandasBlock {
   }
 };
 
+class TimedeltaBlock : public PandasBlock {
+ public:
+  using PandasBlock::PandasBlock;
+  Status AllocateDatetime(int ndim) {
+    RETURN_NOT_OK(AllocateNDArray(NPY_TIMEDELTA, ndim));
+
+    PyAcquireGIL lock;
+    auto date_dtype = reinterpret_cast<PyArray_DatetimeDTypeMetaData*>(
+        PyArray_DESCR(reinterpret_cast<PyArrayObject*>(block_arr_.obj()))->c_metadata);
+    date_dtype->meta.base = NPY_FR_ns;
+    return Status::OK();
+  }
+
+  Status Allocate() override { return AllocateDatetime(2); }
+
+  Status Write(std::shared_ptr<ChunkedArray> data, int64_t abs_placement,
+               int64_t rel_placement) override {
+    Type::type type = data->type()->id();
+
+    int64_t* out_buffer =
+        reinterpret_cast<int64_t*>(block_data_) + rel_placement * num_rows_;
+
+    if (type == Type::DURATION) {
+      const auto& ts_type = checked_cast<const DurationType&>(*data->type());
+
+      if (ts_type.unit() == TimeUnit::NANO) {
+        ConvertNumericNullable<int64_t>(*data, kPandasTimestampNull, out_buffer);
+      } else if (ts_type.unit() == TimeUnit::MICRO) {
+        ConvertTimedeltaNanos<int64_t, 1000L>(*data, out_buffer);
+      } else if (ts_type.unit() == TimeUnit::MILLI) {
+        ConvertTimedeltaNanos<int64_t, 1000000L>(*data, out_buffer);
+      } else if (ts_type.unit() == TimeUnit::SECOND) {
+        ConvertTimedeltaNanos<int64_t, 1000000000L>(*data, out_buffer);
+      } else {
+        return Status::NotImplemented("Unsupported time unit");
+      }
+    } else {
+      return Status::NotImplemented("Cannot write Arrow data of type ",
+                                    data->type()->ToString(),
+                                    " to a Pandas timedelta block.");
+    }
+
+    placement_data_[rel_placement] = abs_placement;
+    return Status::OK();
+  }
+};
+
 class DatetimeTZBlock : public DatetimeBlock {
  public:
   DatetimeTZBlock(const PandasOptions& options, const std::string& timezone,
@@ -1411,6 +1497,7 @@ Status MakeBlock(const PandasOptions& options, PandasBlock::type type, int64_t n
     BLOCK_CASE(DOUBLE, Float64Block);
     BLOCK_CASE(BOOL, BoolBlock);
     BLOCK_CASE(DATETIME, DatetimeBlock);
+    BLOCK_CASE(TIMEDELTA, TimedeltaBlock);
     default:
       return Status::NotImplemented("Unsupported block type");
   }
@@ -1488,6 +1575,9 @@ static Status GetPandasBlockType(const ChunkedArray& data, const PandasOptions&
         *output_type = PandasBlock::DATETIME;
       }
     } break;
+    case Type::DURATION:
+      *output_type = PandasBlock::TIMEDELTA;
+      break;
     case Type::LIST: {
       auto list_type = std::static_pointer_cast<ListType>(data.type());
       if (!ListTypeSupported(*list_type->value_type())) {
@@ -1795,7 +1885,9 @@ class ArrowDeserializer {
   }
 
   template <typename Type>
-  typename std::enable_if<std::is_base_of<TimestampType, Type>::value, Status>::type
+  typename std::enable_if<std::is_base_of<TimestampType, Type>::value ||
+                              std::is_base_of<DurationType, Type>::value,
+                          Status>::type
   Visit(const Type& type) {
     if (options_.zero_copy_only) {
       return Status::Invalid("Copy Needed, but zero_copy_only was True");
@@ -1968,6 +2060,7 @@ class ArrowDeserializer {
       CONVERTVALUES_LISTSLIKE_CASE(Time32Type, TIME32)
       CONVERTVALUES_LISTSLIKE_CASE(Time64Type, TIME64)
       CONVERTVALUES_LISTSLIKE_CASE(TimestampType, TIMESTAMP)
+      CONVERTVALUES_LISTSLIKE_CASE(DurationType, DURATION)
       CONVERTVALUES_LISTSLIKE_CASE(FloatType, FLOAT)
       CONVERTVALUES_LISTSLIKE_CASE(DoubleType, DOUBLE)
       CONVERTVALUES_LISTSLIKE_CASE(BinaryType, BINARY)

cpp/src/arrow/python/numpy_convert.cc

@@ -173,6 +173,28 @@ Status NumPyDtypeToArrow(PyArray_Descr* descr, std::shared_ptr<DataType>* out) {
           return Status::NotImplemented("Unsupported datetime64 time unit");
       }
     } break;
+    case NPY_TIMEDELTA: {
+      auto timedelta_dtype =
+          reinterpret_cast<PyArray_DatetimeDTypeMetaData*>(descr->c_metadata);
+      switch (timedelta_dtype->meta.base) {
+        case NPY_FR_s:
+          *out = duration(TimeUnit::SECOND);
+          break;
+        case NPY_FR_ms:
+          *out = duration(TimeUnit::MILLI);
+          break;
+        case NPY_FR_us:
+          *out = duration(TimeUnit::MICRO);
+          break;
+        case NPY_FR_ns:
+          *out = duration(TimeUnit::NANO);
+          break;
+        case NPY_FR_GENERIC:
+          return Status::NotImplemented("Unbound or generic timedelta64 time unit");
+        default:
+          return Status::NotImplemented("Unsupported timedelta64 time unit");
+      }
+    } break;
     default: {
       return Status::NotImplemented("Unsupported numpy type ", descr->type_num);
     }

cpp/src/arrow/python/numpy_internal.h

@@ -104,6 +104,7 @@ static inline std::string GetNumPyTypeName(int npy_type) {
     TYPE_CASE(FLOAT32, "float32")
     TYPE_CASE(FLOAT64, "float64")
     TYPE_CASE(DATETIME, "datetime64")
+    TYPE_CASE(TIMEDELTA, "timedelta64")
     TYPE_CASE(OBJECT, "object")
     TYPE_CASE(VOID, "void")
     default:
@@ -144,6 +145,7 @@ inline Status VisitNumpyArrayInline(PyArrayObject* arr, VISITOR* visitor) {
     TYPE_VISIT_INLINE(FLOAT32);
     TYPE_VISIT_INLINE(FLOAT64);
     TYPE_VISIT_INLINE(DATETIME);
+    TYPE_VISIT_INLINE(TIMEDELTA);
     TYPE_VISIT_INLINE(OBJECT);
   }
   return Status::NotImplemented("NumPy type not implemented: ",

cpp/src/arrow/python/numpy_to_arrow.cc

@@ -125,7 +125,7 @@ class NumPyNullsConverter {
 
     const bool null_sentinels_possible =
         // Always treat Numpy's NaT as null
-        TYPE == NPY_DATETIME ||
+        TYPE == NPY_DATETIME || TYPE == NPY_TIMEDELTA ||
         // Observing pandas's null sentinels
         (from_pandas_ && traits::supports_nulls);
 
@@ -218,6 +218,7 @@ class NumPyConverter {
   Status Visit(const TimestampType& type) { return VisitNative<TimestampType>(); }
   Status Visit(const Time32Type& type) { return VisitNative<Int32Type>(); }
   Status Visit(const Time64Type& type) { return VisitNative<Int64Type>(); }
+  Status Visit(const DurationType& type) { return VisitNative<DurationType>(); }
 
   Status Visit(const NullType& type) { return TypeNotImplemented(type.ToString()); }
 

cpp/src/arrow/python/type_traits.h

@@ -130,6 +130,23 @@ struct npy_traits<NPY_DATETIME> {
   }
 };
 
+template <>
+struct npy_traits<NPY_TIMEDELTA> {
+  typedef int64_t value_type;
+  using TypeClass = DurationType;
+  using BuilderClass = DurationBuilder;
+
+  static constexpr bool supports_nulls = true;
+
+  static inline bool isnull(int64_t v) {
+    // NaT = -2**63
+    // = -0x8000000000000000
+    // = -9223372036854775808;
+    // = std::numeric_limits<int64_t>::min()
+    return v == std::numeric_limits<int64_t>::min();
+  }
+};
+
 template <>
 struct npy_traits<NPY_OBJECT> {
   typedef PyObject* value_type;
@@ -208,6 +225,16 @@ struct arrow_traits<Type::TIMESTAMP> {
   typedef typename npy_traits<NPY_DATETIME>::value_type T;
 };
 
+template <>
+struct arrow_traits<Type::DURATION> {
+  static constexpr int npy_type = NPY_TIMEDELTA;
+  static constexpr int64_t npy_shift = 1;
+
+  static constexpr bool supports_nulls = true;
+  static constexpr int64_t na_value = kPandasTimestampNull;
+  typedef typename npy_traits<NPY_TIMEDELTA>::value_type T;
+};
+
 template <>
 struct arrow_traits<Type::DATE32> {
   // Data stores as FR_D day unit

python/pyarrow/__init__.py

@@ -50,7 +50,7 @@ from pyarrow.lib import cpu_count, set_cpu_count
 from pyarrow.lib import (null, bool_,
                          int8, int16, int32, int64,
                          uint8, uint16, uint32, uint64,
-                         time32, time64, timestamp, date32, date64,
+                         time32, time64, timestamp, date32, date64, duration,
                          float16, float32, float64,
                          binary, string, utf8,
                          large_binary, large_string, large_utf8,
@@ -58,8 +58,8 @@ from pyarrow.lib import (null, bool_,
                          list_, large_list, struct, union, dictionary, field,
                          type_for_alias,
                          DataType, DictionaryType, StructType,
-                         ListType, LargeListType,
-                         UnionType, TimestampType, Time32Type, Time64Type,
+                         ListType, LargeListType, UnionType,
+                         TimestampType, Time32Type, Time64Type, DurationType,
                          FixedSizeBinaryType, Decimal128Type,
                          BaseExtensionType, ExtensionType,
                          PyExtensionType, UnknownExtensionType,
@@ -84,8 +84,8 @@ from pyarrow.lib import (null, bool_,
                          LargeBinaryArray, LargeStringArray,
                          FixedSizeBinaryArray,
                          DictionaryArray,
-                         Date32Array, Date64Array,
-                         TimestampArray, Time32Array, Time64Array,
+                         Date32Array, Date64Array, TimestampArray,
+                         Time32Array, Time64Array, DurationArray,
                          Decimal128Array, StructArray, ExtensionArray,
                          ArrayValue, Scalar, NA, _NULL as NULL,
                          BooleanValue,
@@ -99,7 +99,7 @@ from pyarrow.lib import (null, bool_,
                          DecimalValue, UnionValue, StructValue, DictionaryValue,
                          Date32Value, Date64Value,
                          Time32Value, Time64Value,
-                         TimestampValue)
+                         TimestampValue, DurationValue)
 
 # Buffers, allocation
 from pyarrow.lib import (Buffer, ResizableBuffer, foreign_buffer, py_buffer,

python/pyarrow/array.pxi

@@ -1117,6 +1117,11 @@ cdef class Time64Array(NumericArray):
     """
 
 
+cdef class DurationArray(NumericArray):
+    """
+    Concrete class for Arrow arrays of duration data type.
+    """
+
 cdef class HalfFloatArray(FloatingPointArray):
     """
     Concrete class for Arrow arrays of float16 data type.
@@ -1694,6 +1699,7 @@ cdef dict _array_classes = {
     _Type_TIMESTAMP: TimestampArray,
     _Type_TIME32: Time32Array,
     _Type_TIME64: Time64Array,
+    _Type_DURATION: DurationArray,
     _Type_HALF_FLOAT: HalfFloatArray,
     _Type_FLOAT: FloatArray,
     _Type_DOUBLE: DoubleArray,

python/pyarrow/includes/libarrow.pxd

@@ -63,6 +63,8 @@ cdef extern from "arrow/api.h" namespace "arrow" nogil:
         _Type_TIMESTAMP" arrow::Type::TIMESTAMP"
         _Type_TIME32" arrow::Type::TIME32"
         _Type_TIME64" arrow::Type::TIME64"
+        _Type_DURATION" arrow::Type::DURATION"
+
         _Type_BINARY" arrow::Type::BINARY"
         _Type_STRING" arrow::Type::STRING"
         _Type_LARGE_BINARY" arrow::Type::LARGE_BINARY"
@@ -199,6 +201,11 @@ cdef extern from "arrow/api.h" namespace "arrow" nogil:
     shared_ptr[CDataType] ctime32" arrow::time32"(TimeUnit unit)
     shared_ptr[CDataType] ctime64" arrow::time64"(TimeUnit unit)
 
+    cdef cppclass CDurationType" arrow::DurationType"(CFixedWidthType):
+        TimeUnit unit()
+
+    shared_ptr[CDataType] cduration" arrow::duration"(TimeUnit unit)
+
     cdef cppclass CDictionaryType" arrow::DictionaryType"(CFixedWidthType):
         CDictionaryType(const shared_ptr[CDataType]& index_type,
                         const shared_ptr[CDataType]& value_type,
@@ -407,6 +414,9 @@ cdef extern from "arrow/api.h" namespace "arrow" nogil:
     cdef cppclass CTimestampArray" arrow::TimestampArray"(CArray):
         int64_t Value(int i)
 
+    cdef cppclass CDurationArray" arrow::DurationArray"(CArray):
+        int64_t Value(int i)
+
     cdef cppclass CHalfFloatArray" arrow::HalfFloatArray"(CArray):
         uint16_t Value(int i)
 

python/pyarrow/lib.pxd

@@ -105,6 +105,11 @@ cdef class Time64Type(DataType):
         const CTime64Type* time_type
 
 
+cdef class DurationType(DataType):
+    cdef:
+        const CDurationType* duration_type
+
+
 cdef class FixedSizeBinaryType(DataType):
     cdef:
         const CFixedSizeBinaryType* fixed_size_binary_type

python/pyarrow/public-api.pxi

@@ -92,6 +92,8 @@ cdef api object pyarrow_wrap_data_type(
         out = UnionType.__new__(UnionType)
     elif type.get().id() == _Type_TIMESTAMP:
         out = TimestampType.__new__(TimestampType)
+    elif type.get().id() == _Type_DURATION:
+        out = DurationType.__new__(DurationType)
     elif type.get().id() == _Type_FIXED_SIZE_BINARY:
         out = FixedSizeBinaryType.__new__(FixedSizeBinaryType)
     elif type.get().id() == _Type_DECIMAL:

python/pyarrow/scalar.pxi

@@ -405,6 +405,79 @@ cdef class TimestampValue(ArrayValue):
         return converter(value, tzinfo=tzinfo)
 
 
+cdef dict _TIMEDELTA_CONVERSION_FUNCTIONS = {}
+cdef c_bool _timedelta_conversion_initialized = False
+
+
+def _nanoseconds_to_timedelta_safe(v):
+    if v % 1000 != 0:
+        raise ValueError(
+            "Nanosecond duration {} is not safely convertible to microseconds "
+            "to convert to datetime.timedelta. Install pandas to return as "
+            "Timedelta with nanosecond support or access the .value "
+            "attribute.".format(v))
+    micros = v // 1000
+
+    return datetime.timedelta(microseconds=micros)
+
+
+def _timedelta_conversion_functions():
+    global _timedelta_conversion_initialized
+    if _timedelta_conversion_initialized:
+        return _TIMEDELTA_CONVERSION_FUNCTIONS
+
+    _TIMEDELTA_CONVERSION_FUNCTIONS.update({
+        TimeUnit_SECOND: lambda v: datetime.timedelta(seconds=v),
+        TimeUnit_MILLI: lambda v: datetime.timedelta(milliseconds=v),
+        TimeUnit_MICRO: lambda v: datetime.timedelta(microseconds=v),
+        TimeUnit_NANO: _nanoseconds_to_timedelta_safe
+    })
+
+    try:
+        import pandas as pd
+        _TIMEDELTA_CONVERSION_FUNCTIONS[TimeUnit_NANO] = (
+            lambda v: pd.Timedelta(v, unit='ns')
+        )
+    except ImportError:
+        pass
+
+    _timedelta_conversion_initialized = True
+    return _TIMEDELTA_CONVERSION_FUNCTIONS
+
+
+cdef class DurationValue(ArrayValue):
+    """
+    Concrete class for duration array elements.
+    """
+
+    @property
+    def value(self):
+        cdef CDurationArray* ap = <CDurationArray*> self.sp_array.get()
+        return ap.Value(self.index)
+
+    def as_py(self):
+        """
+        Return this value as a Pandas Timestamp instance (if available),
+        otherwise as a Python datetime.timedelta instance.
+        """
+        cdef CDurationArray* ap = <CDurationArray*> self.sp_array.get()
+        cdef CDurationType* dtype = <CDurationType*> ap.type().get()
+
+        cdef int64_t value = ap.Value(self.index)
+
+        # if dtype.unit() == TimeUnit_SECOND:
+        #     return datetime.timedelta(seconds=val)
+        # elif dtype.unit() == TimeUnit_MILLI:
+        #     return datetime.timedelta(milliseconds=val)
+        # elif dtype.unit() == TimeUnit_MICRO:
+        #     return datetime.timedelta(microseconds=val)
+        # elif dtype.unit() == TimeUnit_NANO:
+        #     raise NotImplementedError
+
+        converter = _timedelta_conversion_functions()[dtype.unit()]
+        return converter(value)
+
+
 cdef class HalfFloatValue(ArrayValue):
     """
     Concrete class for float16 array elements.
@@ -810,6 +883,7 @@ cdef dict _array_value_classes = {
     _Type_TIME32: Time32Value,
     _Type_TIME64: Time64Value,
     _Type_TIMESTAMP: TimestampValue,
+    _Type_DURATION: DurationValue,
     _Type_HALF_FLOAT: HalfFloatValue,
     _Type_FLOAT: FloatValue,
     _Type_DOUBLE: DoubleValue,

python/pyarrow/tests/test_array.py

@@ -823,6 +823,7 @@ def test_cast_from_null():
         pa.timestamp('us'),
         pa.timestamp('us', tz='UTC'),
         pa.timestamp('us', tz='Europe/Paris'),
+        # pa.duration('us'),
         pa.struct([pa.field('a', pa.int32()),
                    pa.field('b', pa.list_(pa.int8())),
                    pa.field('c', pa.string())]),
@@ -926,6 +927,16 @@ def test_cast_date32_to_int():
     assert result2.equals(arr)
 
 
+@pytest.mark.xfail(strict=True)  # TODO implement duration cast
+def test_cast_duration_to_int():
+    arr = pa.array(np.array([0, 1, 2], dtype='int64'),
+                   type=pa.duration('us'))
+    expected = pa.array([0, 1, 2], type='i8')
+
+    result = arr.cast('i8')
+    assert result.equals(expected)
+
+
 def test_cast_binary_to_utf8():
     binary_arr = pa.array([b'foo', b'bar', b'baz'], type=pa.binary())
     utf8_arr = binary_arr.cast(pa.utf8())
@@ -1149,6 +1160,7 @@ def test_pandas_null_sentinels_raise_error():
         ([0, np.nan], pa.time32('s')),
         ([0, np.nan], pa.time64('us')),
         ([0, np.nan], pa.timestamp('us')),
+        # ([0, np.nan], pa.duration('us')),
     ]
     for case, ty in cases:
         # Both types of exceptions are raised. May want to clean that up
@@ -1229,6 +1241,34 @@ def test_array_from_timestamp_with_generic_unit():
         pa.array([n, x, y])
 
 
+@pytest.mark.parametrize(('dtype', 'type'), [
+    ('timedelta64[s]', pa.duration('s')),
+    ('timedelta64[ms]', pa.duration('ms')),
+    ('timedelta64[us]', pa.duration('us')),
+    ('timedelta64[ns]', pa.duration('ns'))
+])
+def test_array_from_numpy_timedelta(dtype, type):
+    data = [
+        None,
+        datetime.timedelta(1),
+        datetime.timedelta(0, 1)
+    ]
+
+    # from numpy array
+    np_arr = np.array(data, dtype=dtype)
+    arr = pa.array(np_arr)
+    assert isinstance(arr, pa.DurationArray)
+    assert arr.type == type
+    # TODO implement python conversion
+    # expected = pa.array(data, type=type)
+    # assert arr.equals(expected)
+    np.testing.assert_array_equal(np.array(arr), np_arr)
+
+    # # from list of numpy scalars
+    # arr = pa.array(list(np.array(data, dtype=dtype)))
+    # assert arr.equals(expected)
+
+
 def test_array_from_numpy_ascii():
     arr = np.array(['abcde', 'abc', ''], dtype='|S5')
 

python/pyarrow/tests/test_misc.py

@@ -85,6 +85,7 @@ def test_cpu_count():
     pa.TimestampArray,
     pa.Time32Array,
     pa.Time64Array,
+    pa.DurationArray,
     pa.Decimal128Array,
     pa.StructArray,
     pa.ArrayValue,
@@ -106,6 +107,7 @@ def test_cpu_count():
     pa.Time32Value,
     pa.Time64Value,
     pa.TimestampValue,
+    pa.DurationValue,
     pa.StringValue,
     pa.BinaryValue,
     pa.FixedSizeBinaryValue,

python/pyarrow/tests/test_pandas.py

@@ -1383,6 +1383,31 @@ class TestConvertDateTimeLikeTypes(object):
         _check_pandas_roundtrip(df)
         _check_serialize_components_roundtrip(df)
 
+    def test_timedeltas_no_nulls(self):
+        df = pd.DataFrame({
+            'timedelta64': np.array([0, 3600000000000, 7200000000000],
+                                    dtype='timedelta64[ns]')
+        })
+        field = pa.field('timedelta64', pa.duration('ns'))
+        schema = pa.schema([field])
+        _check_pandas_roundtrip(
+            df,
+            expected_schema=schema,
+        )
+
+    def test_timedeltas_nulls(self):
+        df = pd.DataFrame({
+            'timedelta64': np.array([0, None, 7200000000000],
+                                    dtype='timedelta64[ns]')
+        })
+        field = pa.field('timedelta64', pa.duration('ns'))
+        schema = pa.schema([field])
+        _check_pandas_roundtrip(
+            df,
+            expected_schema=schema,
+        )
+
+
 # ----------------------------------------------------------------------
 # Conversion tests for string and binary types.
 
@@ -2184,6 +2209,10 @@ class TestZeroCopyConversion(object):
         arr = np.array(['2007-07-13'], dtype='datetime64[ns]')
         self.check_zero_copy_failure(pa.array(arr))
 
+    def test_zero_copy_failure_on_duration_types(self):
+        arr = np.array([1], dtype='timedelta64[ns]')
+        self.check_zero_copy_failure(pa.array(arr))
+
 
 # This function must be at the top-level for Python 2.7's multiprocessing
 def _non_threaded_conversion():
@@ -2542,16 +2571,13 @@ def _pytime_to_micros(pytime):
 def test_convert_unsupported_type_error_message():
     # ARROW-1454
 
+    # period as yet unsupported
     df = pd.DataFrame({
-        't1': pd.date_range('2000-01-01', periods=20),
-        't2': pd.date_range('2000-05-01', periods=20)
+        'a': pd.period_range('2000-01-01', periods=20),
     })
 
-    # timedelta64 as yet unsupported
-    df['diff'] = df.t2 - df.t1
-
-    expected_msg = 'Conversion failed for column diff with type timedelta64'
-    with pytest.raises(pa.ArrowNotImplementedError, match=expected_msg):
+    expected_msg = 'Conversion failed for column a with type period'
+    with pytest.raises(pa.ArrowInvalid, match=expected_msg):
         pa.Table.from_pandas(df)
 
 

python/pyarrow/tests/test_scalars.py

@@ -299,6 +299,48 @@ class TestScalars(unittest.TestCase):
         result = arr.to_pylist()
         assert result == timestamp_rows
 
+    def test_duration(self):
+        arr = np.array([0, 3600000000000], dtype='timedelta64[ns]')
+
+        units = ['us', 'ms', 's']
+
+        for i, unit in enumerate(units):
+            dtype = 'timedelta64[{0}]'.format(unit)
+            arrow_arr = pa.array(arr.astype(dtype))
+            expected = datetime.timedelta(seconds=60*60)
+            assert isinstance(arrow_arr[1].as_py(), datetime.timedelta)
+            assert arrow_arr[1].as_py() == expected
+            assert (arrow_arr[1].value * 1000**(i+1) ==
+                    expected.total_seconds() * 1e9)
+
+    @pytest.mark.pandas
+    def test_duration_nanos_pandas(self):
+        import pandas as pd
+        arr = np.array([0, 3600000000000], dtype='timedelta64[ns]')
+        arrow_arr = pa.array(arr)
+        expected = pd.Timedelta('1 hour')
+        assert isinstance(arrow_arr[1].as_py(), pd.Timedelta)
+        assert arrow_arr[1].as_py() == expected
+        assert arrow_arr[1].value == expected.value
+
+        # Non-zero nanos work fine
+        arr = pa.array(np.array([946684800000000001], dtype='timedelta64[ns]'))
+        arr[0].as_py() == pd.Timedelta(946684800000000001, unit='ns')
+
+    @pytest.mark.nopandas
+    def test_duration_nanos_nopandas(self):
+        arr = np.array([0, 3600000000000], dtype='timedelta64[ns]')
+        arrow_arr = pa.array(arr)
+        expected = datetime.timedelta(seconds=60*60)
+        assert isinstance(arrow_arr[1].as_py(), datetime.timedelta)
+        assert arrow_arr[1].as_py() == expected
+        assert arrow_arr[1].value == expected.total_seconds() * 1e9
+
+        # Non-zero nanos yields ValueError
+        arr = pa.array(np.array([946684800000000001], dtype='timedelta64[ns]'))
+        with pytest.raises(ValueError):
+            arr[0].as_py()
+
     @pytest.mark.pandas
     def test_dictionary(self):
         import pandas as pd

python/pyarrow/tests/test_types.py

@@ -42,6 +42,7 @@ def get_many_types():
         pa.timestamp('us'),
         pa.timestamp('us', tz='UTC'),
         pa.timestamp('us', tz='Europe/Paris'),
+        pa.duration('s'),
         pa.float16(),
         pa.float32(),
         pa.float64(),
@@ -244,6 +245,16 @@ def test_time64_units():
             pa.time64(invalid_unit)
 
 
+def test_duration():
+    for unit in ('s', 'ms', 'us', 'ns'):
+        ty = pa.duration(unit)
+        assert ty.unit == unit
+
+    for invalid_unit in ('m', 'arbit', 'rary'):
+        with pytest.raises(ValueError, match='Invalid TimeUnit string'):
+            pa.duration(invalid_unit)
+
+
 def test_list_type():
     ty = pa.list_(pa.int64())
     assert isinstance(ty, pa.ListType)

python/pyarrow/types.pxi

@@ -42,6 +42,7 @@ cdef dict _pandas_type_map = {
     _Type_DATE32: np.dtype('datetime64[ns]'),
     _Type_DATE64: np.dtype('datetime64[ns]'),
     _Type_TIMESTAMP: np.dtype('datetime64[ns]'),
+    _Type_DURATION: np.dtype('timedelta64[ns]'),
     _Type_BINARY: np.object_,
     _Type_FIXED_SIZE_BINARY: np.object_,
     _Type_STRING: np.object_,
@@ -466,6 +467,23 @@ cdef class Time64Type(DataType):
         return timeunit_to_string(self.time_type.unit())
 
 
+cdef class DurationType(DataType):
+    """
+    Concrete class for duration data types.
+    """
+
+    cdef void init(self, const shared_ptr[CDataType]& type) except *:
+        DataType.init(self, type)
+        self.duration_type = <const CDurationType*> type.get()
+
+    @property
+    def unit(self):
+        """
+        The duration unit ('s', 'ms', 'us' or 'ns').
+        """
+        return timeunit_to_string(self.duration_type.unit())
+
+
 cdef class FixedSizeBinaryType(DataType):
     """
     Concrete class for fixed-size binary data types.
@@ -1388,6 +1406,7 @@ def int64():
 
 cdef dict _timestamp_type_cache = {}
 cdef dict _time_type_cache = {}
+cdef dict _duration_type_cache = {}
 
 
 cdef timeunit_to_string(TimeUnit unit):
@@ -1609,6 +1628,48 @@ def time64(unit):
     return out
 
 
+def duration(unit):
+    """
+    Create instance of a duration type with unit resolution.
+
+    Parameters
+    ----------
+    unit : string
+        one of 's' [second], 'ms' [millisecond], 'us' [microsecond], or 'ns'
+        [nanosecond]
+
+    Examples
+    --------
+    ::
+
+        t1 = pa.duration('us')
+        t2 = pa.duration('s')
+    """
+    cdef:
+        TimeUnit unit_code
+
+    if unit == "s":
+        unit_code = TimeUnit_SECOND
+    elif unit == 'ms':
+        unit_code = TimeUnit_MILLI
+    elif unit == 'us':
+        unit_code = TimeUnit_MICRO
+    elif unit == 'ns':
+        unit_code = TimeUnit_NANO
+    else:
+        raise ValueError('Invalid TimeUnit string')
+
+    if unit_code in _duration_type_cache:
+        return _duration_type_cache[unit_code]
+
+    cdef DurationType out = DurationType.__new__(DurationType)
+
+    out.init(cduration(unit_code))
+    _duration_type_cache[unit_code] = out
+
+    return out
+
+
 def date32():
     """
     Create instance of 32-bit date (days since UNIX epoch 1970-01-01)
@@ -1960,6 +2021,10 @@ cdef dict _type_aliases = {
     'timestamp[ms]': timestamp('ms'),
     'timestamp[us]': timestamp('us'),
     'timestamp[ns]': timestamp('ns'),
+    'duration[s]': duration('s'),
+    'duration[ms]': duration('ms'),
+    'duration[us]': duration('us'),
+    'duration[ns]': duration('ns'),
 }