diff --git a/examples/duration/CMakeLists.txt b/examples/duration/CMakeLists.txt
new file mode 100644
index 000000000..aacd5d7aa
--- /dev/null
+++ b/examples/duration/CMakeLists.txt
@@ -0,0 +1,14 @@
+cmake_minimum_required(VERSION 2.6.4)
+
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ".")
+set(PROJECT_EXAMPLE_NAME duration)
+
+file(GLOB EXAMPLE_SRC_FILES ${PROJECT_SOURCE_DIR}/examples/duration/*.c)
+include_directories(${INCLUDES})
+add_executable(${PROJECT_EXAMPLE_NAME} ${EXAMPLE_SRC_FILES})
+target_link_libraries(${PROJECT_EXAMPLE_NAME} ${PROJECT_LIB_NAME_TARGET} ${CASS_LIBS})
+add_dependencies(${PROJECT_EXAMPLE_NAME} ${PROJECT_LIB_NAME_TARGET})
+
+set_property(
+  TARGET ${PROJECT_EXAMPLE_NAME}
+  APPEND PROPERTY COMPILE_FLAGS ${CASS_EXAMPLE_C_FLAGS})
diff --git a/examples/duration/duration.c b/examples/duration/duration.c
new file mode 100644
index 000000000..370860e64
--- /dev/null
+++ b/examples/duration/duration.c
@@ -0,0 +1,189 @@
+/*
+  This is free and unencumbered software released into the public domain.
+
+  Anyone is free to copy, modify, publish, use, compile, sell, or
+  distribute this software, either in source code form or as a compiled
+  binary, for any purpose, commercial or non-commercial, and by any
+  means.
+
+  In jurisdictions that recognize copyright laws, the author or authors
+  of this software dedicate any and all copyright interest in the
+  software to the public domain. We make this dedication for the benefit
+  of the public at large and to the detriment of our heirs and
+  successors. We intend this dedication to be an overt act of
+  relinquishment in perpetuity of all present and future rights to this
+  software under copyright law.
+
+  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+  IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+  OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+  ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+  OTHER DEALINGS IN THE SOFTWARE.
+
+  For more information, please refer to <http://unlicense.org/>
+*/
+
+#include <stdio.h>
+#include "cassandra.h"
+
+void print_error(CassFuture* future) {
+  const char* message;
+  size_t message_length;
+  cass_future_error_message(future, &message, &message_length);
+  fprintf(stderr, "Error: %.*s\n", (int)message_length, message);
+}
+
+CassCluster* create_cluster(const char* hosts) {
+  CassCluster* cluster = cass_cluster_new();
+  cass_cluster_set_contact_points(cluster, hosts);
+  return cluster;
+}
+
+CassError connect_session(CassSession* session, const CassCluster* cluster) {
+  CassError rc = CASS_OK;
+  CassFuture* future = cass_session_connect(session, cluster);
+
+  cass_future_wait(future);
+  rc = cass_future_error_code(future);
+  if (rc != CASS_OK) {
+    print_error(future);
+  }
+  cass_future_free(future);
+
+  return rc;
+}
+
+CassError execute_query(CassSession* session, const char* query) {
+  CassError rc = CASS_OK;
+  CassFuture* future = NULL;
+  CassStatement* statement = cass_statement_new(query, 0);
+
+  future = cass_session_execute(session, statement);
+  cass_future_wait(future);
+
+  rc = cass_future_error_code(future);
+  if (rc != CASS_OK) {
+    print_error(future);
+  }
+
+  cass_future_free(future);
+  cass_statement_free(statement);
+
+  return rc;
+}
+
+CassError insert_into(CassSession* session, const char* key, cass_int64_t months, cass_int64_t days, cass_int64_t nanos) {
+  CassError rc = CASS_OK;
+  CassStatement* statement = NULL;
+  CassFuture* future = NULL;
+  const char* query = "INSERT INTO examples.duration (key, d) VALUES (?, ?);";
+  cass_byte_t* data;
+  size_t data_size;
+
+
+  statement = cass_statement_new(query, 2);
+
+  cass_statement_bind_string(statement, 0, key);
+  cass_statement_bind_duration(statement, 1, months, days, nanos);
+
+  future = cass_session_execute(session, statement);
+  cass_future_wait(future);
+
+  rc = cass_future_error_code(future);
+  if (rc != CASS_OK) {
+    print_error(future);
+  }
+
+  cass_future_free(future);
+  cass_statement_free(statement);
+
+  return rc;
+}
+
+CassError select_from(CassSession* session, const char* key) {
+  CassError rc = CASS_OK;
+  CassStatement* statement = NULL;
+  CassFuture* future = NULL;
+  const char* query = "SELECT d FROM examples.duration WHERE key = ?";
+
+  statement = cass_statement_new(query, 1);
+
+  cass_statement_bind_string(statement, 0, key);
+
+  future = cass_session_execute(session, statement);
+  cass_future_wait(future);
+
+  rc = cass_future_error_code(future);
+  if (rc != CASS_OK) {
+    print_error(future);
+  } else {
+    const CassResult* result = cass_future_get_result(future);
+    CassIterator* iterator = cass_iterator_from_result(result);
+
+    if (cass_iterator_next(iterator)) {
+      cass_int64_t months, days, nanos;
+      const CassRow* row = cass_iterator_get_row(iterator);
+
+      cass_value_get_duration(cass_row_get_column(row, 0), &months, &days, &nanos);
+      printf("months: %lld  days: %lld  nanos: %lld\n", months, days, nanos);
+    }
+
+    cass_result_free(result);
+    cass_iterator_free(iterator);
+  }
+
+  cass_future_free(future);
+  cass_statement_free(statement);
+
+  return rc;
+}
+
+int main(int argc, char* argv[]) {
+  CassCluster* cluster = NULL;
+  CassSession* session = cass_session_new();
+  CassFuture* close_future = NULL;
+  char* hosts = "127.0.0.1";
+
+  if (argc > 1) {
+    hosts = argv[1];
+  }
+  cluster = create_cluster(hosts);
+
+  if (connect_session(session, cluster) != CASS_OK) {
+    cass_cluster_free(cluster);
+    cass_session_free(session);
+    return -1;
+  }
+
+  execute_query(session,
+                "CREATE KEYSPACE examples WITH replication = { \
+                           'class': 'SimpleStrategy', 'replication_factor': '3' };");
+
+
+  execute_query(session,
+                "CREATE TABLE examples.duration (key text PRIMARY KEY, d duration)");
+
+  // Insert some rows into the table, playing with edge values; each of months, days, and nanos may be
+  // a 64-bit long.
+
+  insert_into(session, "base", 0, 0, 0);
+  insert_into(session, "simple", 1, 2, 3);
+  insert_into(session, "edge", (1LL << 63) - 1, 0xffffffffffffffff, 1LL << 63);
+  select_from(session, "base");
+  select_from(session, "simple");
+
+  // This case should print out
+  // months: 9223372036854775807  days: -1  nanos: -9223372036854775808
+  select_from(session, "edge");
+
+  close_future = cass_session_close(session);
+  cass_future_wait(close_future);
+  cass_future_free(close_future);
+
+  cass_cluster_free(cluster);
+  cass_session_free(session);
+
+  return 0;
+}
diff --git a/include/cassandra.h b/include/cassandra.h
index 2c7f11b48..90ccab8b0 100644
--- a/include/cassandra.h
+++ b/include/cassandra.h
@@ -548,6 +548,7 @@ typedef enum CassValueType_ {
   CASS_VALUE_TYPE_TIME      = 0x0012,
   CASS_VALUE_TYPE_SMALL_INT = 0x0013,
   CASS_VALUE_TYPE_TINY_INT  = 0x0014,
+  CASS_VALUE_TYPE_DURATION  = 0x0015,
   CASS_VALUE_TYPE_LIST      = 0x0020,
   CASS_VALUE_TYPE_MAP       = 0x0021,
   CASS_VALUE_TYPE_SET       = 0x0022,
@@ -4927,6 +4928,74 @@ cass_statement_bind_decimal_by_name_n(CassStatement* statement,
                                       size_t varint_size,
                                       cass_int32_t scale);
 
+/**
+ * Binds a "duration" to a query or bound statement at the specified index.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassStatement
+ *
+ * @param[in] statement
+ * @param[in] index
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_statement_bind_duration(CassStatement* statement,
+                             size_t index,
+                             cass_int64_t months,
+                             cass_int64_t days,
+                             cass_int64_t nanos);
+
+/**
+ * Binds a "duration" to all the values with the specified name.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassStatement
+ *
+ * @param[in] statement
+ * @param[in] name
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_statement_bind_duration_by_name(CassStatement* statement,
+                                     const char* name,
+                                     cass_int64_t months,
+                                     cass_int64_t days,
+                                     cass_int64_t nanos);
+
+/**
+ * Same as cass_statement_bind_duration_by_name(), but with lengths for string
+ * parameters.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassStatement
+ *
+ * @param[in] statement
+ * @param[in] name
+ * @param[in] name_length
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return same as cass_statement_bind_duration_by_name()
+ *
+ * @see cass_statement_bind_duration_by_name()
+ */
+CASS_EXPORT CassError
+cass_statement_bind_duration_by_name_n(CassStatement* statement,
+                                       const char* name,
+                                       size_t name_length,
+                                       cass_int64_t months,
+                                       cass_int64_t days,
+                                       cass_int64_t nanos);
+
 /**
  * Bind a "list", "map" or "set" to a query or bound statement at the
  * specified index.
@@ -6081,6 +6150,25 @@ cass_collection_append_decimal(CassCollection* collection,
                                size_t varint_size,
                                cass_int32_t scale);
 
+/**
+ * Appends a "duration" to the collection.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassCollection
+ *
+ * @param[in] collection
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_collection_append_duration(CassCollection* collection,
+                                cass_int64_t months,
+                                cass_int64_t days,
+                                cass_int64_t nanos);
+
 /**
  * Appends a "list", "map" or "set" to the collection.
  *
@@ -6497,6 +6585,27 @@ cass_tuple_set_decimal(CassTuple* tuple,
                        size_t varint_size,
                        cass_int32_t scale);
 
+/**
+ * Sets a "duration" in a tuple at the specified index.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassTuple
+ *
+ * @param[in] tuple
+ * @param[in] index
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_tuple_set_duration(CassTuple* tuple,
+                        size_t index,
+                        cass_int64_t months,
+                        cass_int64_t days,
+                        cass_int64_t nanos);
+
 /**
  * Sets a "list", "map" or "set" in a tuple at the specified index.
  *
@@ -7509,6 +7618,74 @@ cass_user_type_set_decimal_by_name_n(CassUserType* user_type,
                                      size_t varint_size,
                                      int scale);
 
+/**
+ * Sets a "duration" in a user defined type at the specified index.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassUserType
+ *
+ * @param[in] user_type
+ * @param[in] index
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_user_type_set_duration(CassUserType* user_type,
+                            size_t index,
+                            cass_int64_t months,
+                            cass_int64_t days,
+                            cass_int64_t nanos);
+
+/**
+ * Sets "duration" in a user defined type at the specified name.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassUserType
+ *
+ * @param[in] user_type
+ * @param[in] name
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return CASS_OK if successful, otherwise an error occurred.
+ */
+CASS_EXPORT CassError
+cass_user_type_set_duration_by_name(CassUserType* user_type,
+                                    const char* name,
+                                    cass_int64_t months,
+                                    cass_int64_t days,
+                                    cass_int64_t nanos);
+
+/**
+ * Same as cass_user_type_set_duration_by_name(), but with lengths for string
+ * parameters.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassUserType
+ *
+ * @param[in] user_type
+ * @param[in] name
+ * @param[in] name_length
+ * @param[in] months
+ * @param[in] days
+ * @param[in] nanos
+ * @return same as cass_user_type_set_duration_by_name()
+ *
+ * @see cass_user_type_set_duration_by_name()
+ */
+CASS_EXPORT CassError
+cass_user_type_set_duration_by_name_n(CassUserType* user_type,
+                                      const char* name,
+                                      size_t name_length,
+                                      cass_int64_t months,
+                                      cass_int64_t days,
+                                      cass_int64_t nanos);
+
 /**
  * Sets a "list", "map" or "set" in a user defined type at the
  * specified index.
@@ -8982,6 +9159,25 @@ cass_value_get_decimal(const CassValue* value,
                        size_t* varint_size,
                        cass_int32_t* scale);
 
+/**
+ * Gets a duration for the specified value.
+ *
+ * @cassandra{3.10+}
+ *
+ * @public @memberof CassValue
+ *
+ * @param[in] value
+ * @param[out] months
+ * @param[out] days
+ * @param[out] nanos
+ * @return CASS_OK if successful, otherwise error occurred
+ */
+CASS_EXPORT CassError
+cass_value_get_duration(const CassValue* value,
+                        cass_int64_t* months,
+                        cass_int64_t* days,
+                        cass_int64_t* nanos);
+
 /**
  * Gets the type of the specified value.
  *
@@ -9015,6 +9211,17 @@ cass_value_is_null(const CassValue* value);
 CASS_EXPORT cass_bool_t
 cass_value_is_collection(const CassValue* value);
 
+/**
+ * Returns true if a specified value is a duration.
+ *
+ * @public @memberof CassValue
+ *
+ * @param[in] value
+ * @return true if the value is a duration, otherwise false.
+ */
+CASS_EXPORT cass_bool_t
+cass_value_is_duration(const CassValue* value);
+
 /**
  * Get the number of items in a collection. Works for all collection types.
  *
diff --git a/src/abstract_data.hpp b/src/abstract_data.hpp
index e56f3c204..3e0c0ea50 100644
--- a/src/abstract_data.hpp
+++ b/src/abstract_data.hpp
@@ -117,6 +117,7 @@ class AbstractData {
   SET_TYPE(CassUuid)
   SET_TYPE(CassInet)
   SET_TYPE(CassDecimal)
+  SET_TYPE(CassDuration)
 
 #undef SET_TYPE
 
diff --git a/src/collection.cpp b/src/collection.cpp
index a5826f80a..38afb5999 100644
--- a/src/collection.cpp
+++ b/src/collection.cpp
@@ -78,6 +78,9 @@ CASS_COLLECTION_APPEND(bytes,
 CASS_COLLECTION_APPEND(decimal,
                        THREE_PARAMS_(const cass_byte_t* varint, size_t varint_size, int scale),
                        cass::CassDecimal(varint, varint_size, scale))
+CASS_COLLECTION_APPEND(duration,
+                       THREE_PARAMS_(cass_int64_t months, cass_int64_t days, cass_int64_t nanos),
+                       cass::CassDuration(months, days, nanos))
 
 #undef CASS_COLLECTION_APPEND
 
diff --git a/src/collection.hpp b/src/collection.hpp
index 42a99932c..58720bb4e 100644
--- a/src/collection.hpp
+++ b/src/collection.hpp
@@ -76,6 +76,7 @@ class Collection : public RefCounted<Collection> {
   APPEND_TYPE(CassUuid)
   APPEND_TYPE(CassInet)
   APPEND_TYPE(CassDecimal)
+  APPEND_TYPE(CassDuration)
 
 #undef APPEND_TYPE
 
diff --git a/src/data_type.cpp b/src/data_type.cpp
index 8d1edf77f..007f1dfa9 100644
--- a/src/data_type.cpp
+++ b/src/data_type.cpp
@@ -393,6 +393,7 @@ void NativeDataTypes::init_class_names() {
   by_class_names_["org.apache.cassandra.db.marshal.DateType"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_TIMESTAMP));
   by_class_names_["org.apache.cassandra.db.marshal.DecimalType"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DECIMAL));
   by_class_names_["org.apache.cassandra.db.marshal.DoubleType"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DOUBLE));
+  by_class_names_["org.apache.cassandra.db.marshal.DurationType"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DURATION));
   by_class_names_["org.apache.cassandra.db.marshal.FloatType"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_FLOAT));
   by_class_names_["org.apache.cassandra.db.marshal.InetAddressType"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_INET));
   by_class_names_["org.apache.cassandra.db.marshal.Int32Type"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_INT));
@@ -423,6 +424,7 @@ void NativeDataTypes::init_cql_names() {
   by_cql_names_["date"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DATE));
   by_cql_names_["decimal"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DECIMAL));
   by_cql_names_["double"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DOUBLE));
+  by_cql_names_["duration"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_DURATION));
   by_cql_names_["float"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_FLOAT));
   by_cql_names_["inet"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_INET));
   by_cql_names_["int"] = DataType::ConstPtr(new DataType(CASS_VALUE_TYPE_INT));
diff --git a/src/data_type.hpp b/src/data_type.hpp
index 6cd60c28d..86444e26a 100644
--- a/src/data_type.hpp
+++ b/src/data_type.hpp
@@ -118,6 +118,7 @@ class DataType : public RefCounted<DataType> {
       case CASS_VALUE_TYPE_COUNTER: return "counter";
       case CASS_VALUE_TYPE_DECIMAL: return "decimal";
       case CASS_VALUE_TYPE_DOUBLE: return "double";
+      case CASS_VALUE_TYPE_DURATION: return "duration";
       case CASS_VALUE_TYPE_FLOAT: return "float";
       case CASS_VALUE_TYPE_INT: return "int";
       case CASS_VALUE_TYPE_TEXT: return "text";
@@ -576,6 +577,13 @@ struct IsValidDataType<CassDecimal> {
   }
 };
 
+template<>
+struct IsValidDataType<CassDuration> {
+  bool operator()(CassDuration, const DataType::ConstPtr& data_type) const {
+    return data_type->value_type() == CASS_VALUE_TYPE_DURATION;
+  }
+};
+
 template<>
 struct IsValidDataType<const Collection*> {
   bool operator()(const Collection* value, const DataType::ConstPtr& data_type) const;
diff --git a/src/encode.cpp b/src/encode.cpp
new file mode 100644
index 000000000..f9e7eaa82
--- /dev/null
+++ b/src/encode.cpp
@@ -0,0 +1,90 @@
+/*
+  Copyright (c) 2014-2016 DataStax
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+#include "encode.hpp"
+#include "serialization.hpp"
+#include "utils.hpp"
+
+namespace cass {
+
+static char* encode_vint(char* output, uint64_t value, size_t value_size)
+{
+  if (value_size == 1) {
+    // This is just a one byte value; write it and get out.
+    *output = value;
+    return output + 1;
+  }
+
+  // Write the bytes of zigzag value to the output array with most significant byte
+  // in first byte of buffer, and so on.
+  for (int j = value_size - 1 ; j >= 0 ; --j) {
+    *(output + j) = value & 0xff;
+    value >>= 8;
+  }
+
+  // Now mix in the size of the vint into the first byte of the buffer,
+  // setting "value_size-1" higher order bits.
+  *output |= ~(0xff >> (value_size - 1));
+
+  // We're done with value_size bytes
+  return output + value_size;
+}
+
+static Buffer encode_internal(CassDuration value, bool with_length) {
+  // Each duration attribute needs to be converted to zigzag form. Use an array to make it
+  // easy to do the same encoding ops on all three values.
+  uint64_t zigzag_values[3];
+
+  // We need vint sizes for each attribute.
+  size_t vint_sizes[3];
+
+  zigzag_values[0] = cass::encode_zig_zag(value.months);
+  zigzag_values[1] = cass::encode_zig_zag(value.days);
+  zigzag_values[2] = cass::encode_zig_zag(value.nanos);
+
+  // We also need the total size of all three vint's.
+  size_t data_size = 0;
+  for (int i = 0; i < 3; ++i) {
+    vint_sizes[i] = cass::vint_size(zigzag_values[i]);
+    data_size += vint_sizes[i];
+  }
+
+  // Allocate our data buffer and then start populating it. If we're including the length,
+  // allocate extra space for that.
+  Buffer buf(with_length ? sizeof(int32_t) + data_size : data_size);
+
+  // Write the length if with_length is set; this means subsequent data is written afterwards
+  // in buf. Use pos to set our next write position properly regardless of with_length. Since we
+  // don't write sequentially to the buffer (we jump around a bit) the easiest thing to do is grab
+  // a pointer to the underlying data buffer and manipulate it.
+  size_t pos = with_length ? buf.encode_int32(0, data_size) : 0;
+  char* cur_byte = buf.data() + pos;
+
+  for (int i = 0; i < 3; ++i) {
+    cur_byte = encode_vint(cur_byte, zigzag_values[i], vint_sizes[i]);
+  }
+  return buf;
+}
+
+Buffer encode(CassDuration value) {
+  return encode_internal(value, false);
+}
+
+Buffer encode_with_length(CassDuration value) {
+  return encode_internal(value, true);
+}
+
+} // namespace cass
diff --git a/src/encode.hpp b/src/encode.hpp
index 729ab3349..9126ed4ae 100644
--- a/src/encode.hpp
+++ b/src/encode.hpp
@@ -218,6 +218,10 @@ inline Buffer encode(CassDecimal value) {
   return buf;
 }
 
+Buffer encode(CassDuration value);
+
+Buffer encode_with_length(CassDuration value);
+
 } // namespace cass
 
 #endif
diff --git a/src/result_response.cpp b/src/result_response.cpp
index 4e5a5fec0..071c69898 100644
--- a/src/result_response.cpp
+++ b/src/result_response.cpp
@@ -20,6 +20,8 @@
 #include "result_metadata.hpp"
 #include "serialization.hpp"
 
+#include <sparsehash/dense_hash_map>
+
 extern "C" {
 
 void cass_result_free(const CassResult* result) {
@@ -104,7 +106,10 @@ namespace cass {
 class DataTypeDecoder {
 public:
   DataTypeDecoder(char* input)
-    : buffer_(input) { }
+    : buffer_(input) {
+    data_type_cache_.set_empty_key(CASS_VALUE_TYPE_LAST_ENTRY);
+    native_types_.init_class_names();
+  }
 
   char* buffer() const { return buffer_; }
 
@@ -129,14 +134,7 @@ class DataTypeDecoder {
 
       default:
         if (value_type < CASS_VALUE_TYPE_LAST_ENTRY) {
-          if (data_type_cache_[value_type]) {
-            return data_type_cache_[value_type];
-          } else {
-            DataType::Ptr data_type(
-                  new DataType(static_cast<CassValueType>(value_type)));
-            data_type_cache_[value_type] = data_type;
-            return data_type;
-          }
+          return decode_simple_type(value_type);
         }
         break;
     }
@@ -148,7 +146,24 @@ class DataTypeDecoder {
   DataType::Ptr decode_custom() {
     StringRef class_name;
     buffer_ = decode_string(buffer_, &class_name);
-    return DataType::Ptr(new CustomType(class_name.to_string()));
+
+    const DataType::ConstPtr& type = native_types_.by_class_name(class_name.to_string());
+    if (type == DataType::NIL)
+      // If no mapping exists, return an actual custom type.
+      return DataType::Ptr(new CustomType(class_name.to_string()));
+    else
+      return type->copy();
+  }
+
+  DataType::Ptr decode_simple_type(uint16_t value_type) {
+    if (data_type_cache_[value_type]) {
+      return data_type_cache_[value_type];
+    } else {
+      DataType::Ptr data_type(
+            new DataType(static_cast<CassValueType>(value_type)));
+      data_type_cache_[value_type] = data_type;
+      return data_type;
+    }
   }
 
   DataType::Ptr decode_collection(CassValueType collection_type) {
@@ -195,7 +210,8 @@ class DataTypeDecoder {
 
 private:
   char* buffer_;
-  DataType::Ptr data_type_cache_[CASS_VALUE_TYPE_LAST_ENTRY];
+  sparsehash::dense_hash_map<uint16_t, DataType::Ptr> data_type_cache_;
+  NativeDataTypes native_types_;
 };
 
 bool ResultResponse::decode(int version, char* input, size_t size) {
diff --git a/src/serialization.hpp b/src/serialization.hpp
index 3f6ce1f7c..4f20447d9 100644
--- a/src/serialization.hpp
+++ b/src/serialization.hpp
@@ -375,6 +375,16 @@ inline char* decode_size(int protocol_version, char* input, int32_t& size) {
   return pos;
 }
 
+inline cass_int64_t decode_zig_zag(cass_uint64_t n) {
+  // n is an unsigned long because we want a logical shift right
+  // (it should 0-fill high order bits), not arithmetic shift right.
+  return (n >> 1) ^ -(n & 1);
+}
+
+inline cass_uint64_t encode_zig_zag(cass_int64_t n) {
+  return (n << 1) ^ (n >> 63);
+}
+
 } // namespace cass
 
 #endif
diff --git a/src/statement.cpp b/src/statement.cpp
index 8e525ce83..2f2083d64 100644
--- a/src/statement.cpp
+++ b/src/statement.cpp
@@ -170,6 +170,9 @@ CASS_STATEMENT_BIND(bytes,
 CASS_STATEMENT_BIND(decimal,
                     THREE_PARAMS_(const cass_byte_t* varint, size_t varint_size, int scale),
                     cass::CassDecimal(varint, varint_size, scale))
+CASS_STATEMENT_BIND(duration,
+                    THREE_PARAMS_(cass_int64_t months, cass_int64_t days, cass_int64_t nanos),
+                    cass::CassDuration(months, days, nanos))
 
 #undef CASS_STATEMENT_BIND
 
diff --git a/src/tuple.cpp b/src/tuple.cpp
index 50ce2fc00..a76ebe6c3 100644
--- a/src/tuple.cpp
+++ b/src/tuple.cpp
@@ -73,6 +73,9 @@ CASS_TUPLE_SET(bytes,
 CASS_TUPLE_SET(decimal,
                THREE_PARAMS_(const cass_byte_t* varint, size_t varint_size, int scale),
                cass::CassDecimal(varint, varint_size, scale))
+CASS_TUPLE_SET(duration,
+               THREE_PARAMS_(cass_int64_t months, cass_int64_t days, cass_int64_t nanos),
+               cass::CassDuration(months, days, nanos))
 
 #undef CASS_TUPLE_SET
 
diff --git a/src/tuple.hpp b/src/tuple.hpp
index c50a34eb4..88b0d2f7d 100644
--- a/src/tuple.hpp
+++ b/src/tuple.hpp
@@ -69,6 +69,7 @@ class Tuple {
   SET_TYPE(CassUuid)
   SET_TYPE(CassInet)
   SET_TYPE(CassDecimal)
+  SET_TYPE(CassDuration)
 
 #undef SET_TYPE
 
diff --git a/src/types.hpp b/src/types.hpp
index 7ba2e2e78..359f56b93 100644
--- a/src/types.hpp
+++ b/src/types.hpp
@@ -62,6 +62,18 @@ struct CassDecimal {
   cass_int32_t scale;
 };
 
+struct CassDuration {
+  CassDuration(cass_int64_t months,
+               cass_int64_t days,
+               cass_int64_t nanos)
+    : months(months)
+    , days(days)
+    , nanos(nanos) { }
+  cass_int64_t months;
+  cass_int64_t days;
+  cass_int64_t nanos;
+};
+
 } // namespace cass
 
 #endif
diff --git a/src/user_type_value.cpp b/src/user_type_value.cpp
index bf1bbd17f..40e47514a 100644
--- a/src/user_type_value.cpp
+++ b/src/user_type_value.cpp
@@ -78,6 +78,9 @@ CASS_USER_TYPE_SET(bytes,
 CASS_USER_TYPE_SET(decimal,
                    THREE_PARAMS_(const cass_byte_t* varint, size_t varint_size, int scale),
                    cass::CassDecimal(varint, varint_size, scale))
+CASS_USER_TYPE_SET(duration,
+                   THREE_PARAMS_(cass_int64_t months, cass_int64_t days, cass_int64_t nanos),
+                   cass::CassDuration(months, days, nanos))
 
 #undef CASS_USER_TYPE_SET
 
diff --git a/src/utils.hpp b/src/utils.hpp
index af27a1b3f..e0057b370 100644
--- a/src/utils.hpp
+++ b/src/utils.hpp
@@ -91,6 +91,42 @@ std::string& to_cql_id(std::string& str);
 
 std::string& escape_id(std::string& str);
 
+inline size_t num_leading_zeros(cass_int64_t value) {
+  if (value == 0)
+    return 64;
+
+#if defined(_MSC_VER)
+  unsigned long index;
+#  if defined(_M_AMD64)
+  _BitScanReverse64(&index, value);
+#  else
+  // On 32-bit this needs to be split into two operations
+  char isNonzero = _BitScanReverse(&index, (unsigned long)(value >> 32));
+
+  if (isNonzero)
+    // The most significant 4 bytes has a bit set, and our index is relative to that.
+    // Add 32 to account for the lower 4 bytes that make up our 64-bit number.
+    index += 32;
+  else {
+    // Scan the last 32 bits by truncating the 64-bit value
+    _BitScanReverse(&index, (unsigned long) value);
+  }
+#  endif
+  // index is the (zero based) index, counting from lsb, of the most-significant 1 bit.
+  // For example, a value of 12 (b1100) would return 3. The 4th bit is set, so there are
+  // 60 leading zeros.
+  return 64 - index - 1;
+#else
+  return __builtin_clzll(value);
+#endif
+}
+
+inline size_t vint_size(cass_int64_t value) {
+  // | with 1 to ensure magnitude <= 63, so (63 - 1) / 7 <= 8
+  size_t magnitude = num_leading_zeros(value | 1);
+  return magnitude ? (9 - ((magnitude - 1) / 7)) : 9;
+}
+
 cass_int32_t get_pid();
 
 } // namespace cass
diff --git a/src/value.cpp b/src/value.cpp
index ffe0daa84..eb784aaa9 100644
--- a/src/value.cpp
+++ b/src/value.cpp
@@ -139,6 +139,79 @@ CassError cass_value_get_bytes(const CassValue* value,
   return CASS_OK;
 }
 
+static const cass_byte_t* decode_vint(const cass_byte_t* input, const cass_byte_t* end, cass_int64_t* output)
+{
+  int num_extra_bytes;
+  int i;
+  cass_byte_t first_byte = *input++;
+  if (first_byte <= 127) {
+    // If this is a multibyte vint, at least the MSB of the first byte
+    // will be set. Since that's not the case, this is a one-byte value.
+    *output = first_byte;
+  } else {
+    // The number of consecutive most significant bits of the first-byte tell us how
+    // many additional bytes are in this vint. Count them like this:
+    // 1. Invert the firstByte so that all leading 1s become 0s.
+    // 2. Count the number of leading zeros; num_leading_zeros assumes a 64-bit long.
+    // 3. We care about leading 0s in the byte, not int, so subtract out the
+    //    appropriate number of extra bits (56 for a 64-bit int).
+
+    // We mask out high-order bits to prevent sign-extension as the value is placed in a 64-bit arg
+    // to the num_leading_zeros function.
+    num_extra_bytes = cass::num_leading_zeros(~first_byte & 0xff) - 56;
+
+    // Error out if we don't have num_extra_bytes left in our data.
+    if (input + num_extra_bytes > end) {
+      // There aren't enough bytes. This duration object is not fully defined.
+      return NULL;
+    }
+
+    // Build up the vint value one byte at a time from the data bytes.
+    // The firstByte contains size as well as the most significant bits of
+    // the value. Extract just the value.
+    *output = first_byte & (0xff >> num_extra_bytes);
+    for (i = 0; i < num_extra_bytes; ++i) {
+      cass_byte_t b = *input++;
+      *output <<= 8;
+      *output |= b & 0xff;
+    }
+  }
+  return input;
+}
+
+CassError cass_value_get_duration(const CassValue* value, cass_int64_t* months, cass_int64_t* days, cass_int64_t* nanos)
+{
+  cass_int64_t *outs[3];
+  int ctr;
+  size_t data_size = 0;
+  const cass_byte_t* cur_byte = NULL;
+  const cass_byte_t* end = NULL;
+
+  if (value == NULL || value->is_null()) return CASS_ERROR_LIB_NULL_VALUE;
+  if (!cass_value_is_duration(value)) return CASS_ERROR_LIB_INVALID_VALUE_TYPE;
+
+  // Package up the out-args in an array. Duration's always have months, then days, then nanos.
+  outs[0] = months;
+  outs[1] = days;
+  outs[2] = nanos;
+
+  cass_value_get_bytes(value, &cur_byte, &data_size);
+  end = cur_byte + data_size;
+
+  for (ctr = 0; ctr < 3 && cur_byte != end; ++ctr) {
+    cur_byte = decode_vint(cur_byte, end, outs[ctr]);
+    if (cur_byte == NULL)
+      return CASS_ERROR_LIB_BAD_PARAMS;
+    *outs[ctr] = cass::decode_zig_zag(*outs[ctr]);
+  }
+
+  if (ctr < 3) {
+    // There aren't enough bytes. This duration object is not fully defined.
+    return CASS_ERROR_LIB_BAD_PARAMS;
+  }
+  return CASS_OK;
+}
+
 CassError cass_value_get_decimal(const CassValue* value,
                                  const cass_byte_t** varint,
                                  size_t* varint_size,
@@ -165,6 +238,12 @@ cass_bool_t cass_value_is_collection(const CassValue* value) {
   return static_cast<cass_bool_t>(value->is_collection());
 }
 
+cass_bool_t cass_value_is_duration(const CassValue* value) {
+  cass::IsValidDataType<cass::CassDuration> is_valid;
+  cass::CassDuration dummy(0, 0, 0);
+  return static_cast<cass_bool_t>(is_valid(dummy, value->data_type()));
+}
+
 size_t cass_value_item_count(const CassValue* collection) {
   return collection->count();
 }
diff --git a/test/integration_tests/src/test_basics.cpp b/test/integration_tests/src/test_basics.cpp
index 20d13075e..182b220b1 100644
--- a/test/integration_tests/src/test_basics.cpp
+++ b/test/integration_tests/src/test_basics.cpp
@@ -395,6 +395,17 @@ BOOST_AUTO_TEST_CASE(basic_types)
     CassDecimal value(varint, sizeof(varint), scale);
     insert_single_value<CassDecimal>(CASS_VALUE_TYPE_DECIMAL, value);
   }
+
+  if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+    CassDuration value = CassDuration(0, 0, 0);
+    insert_single_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value);
+
+    value = CassDuration(1, 2, 3);
+    insert_single_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value);
+
+    value = CassDuration(-1, -2, -3);
+    insert_single_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value);
+  }
 }
 
 BOOST_AUTO_TEST_CASE(min_max)
@@ -438,6 +449,9 @@ BOOST_AUTO_TEST_CASE(null)
   insert_null_value<CassBytes>(CASS_VALUE_TYPE_BLOB);
   insert_null_value<cass_bool_t>(CASS_VALUE_TYPE_BOOLEAN);
   insert_null_value<CassDecimal>(CASS_VALUE_TYPE_DECIMAL);
+  if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+    insert_null_value<CassDuration>(CASS_VALUE_TYPE_DURATION);
+  }
   insert_null_value<cass_double_t>(CASS_VALUE_TYPE_DOUBLE);
   insert_null_value<cass_float_t>(CASS_VALUE_TYPE_FLOAT);
   insert_null_value<cass_int32_t>(CASS_VALUE_TYPE_INT);
diff --git a/test/integration_tests/src/test_collections.cpp b/test/integration_tests/src/test_collections.cpp
index 09a591ef6..d5cf445e0 100644
--- a/test/integration_tests/src/test_collections.cpp
+++ b/test/integration_tests/src/test_collections.cpp
@@ -67,6 +67,7 @@ struct CollectionsTests : public test_utils::MultipleNodesTest {
 
     const CassValue* output = cass_row_get_column(row, 1);
     BOOST_REQUIRE(cass_value_type(output) == type);
+
     BOOST_REQUIRE(cass_value_primary_sub_type(output) == primary_type);
 
     test_utils::CassIteratorPtr iterator(cass_iterator_from_collection(output));
@@ -191,6 +192,16 @@ struct CollectionsTests : public test_utils::MultipleNodesTest {
       insert_collection_value<CassDecimal>(session.get(), type, CASS_VALUE_TYPE_DECIMAL,  values);
     }
 
+    // C* doesn't support set<Duration>.
+    if (type != CASS_VALUE_TYPE_SET && ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4)) {
+      std::vector<CassDuration> values;
+      for (int i = 0; i < 3; ++i) {
+        CassDuration value(1, 2, 3);
+        values.push_back(value);
+      }
+      insert_collection_value<CassDuration>(session.get(), type, CASS_VALUE_TYPE_DURATION,  values);
+    }
+
     drop_keyspace(session);
   }
 
@@ -377,6 +388,15 @@ struct CollectionsTests : public test_utils::MultipleNodesTest {
       insert_map_value<CassDecimal, CassDecimal>(session.get(), CASS_VALUE_TYPE_DECIMAL, CASS_VALUE_TYPE_DECIMAL, values);
     }
 
+    if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+      // Duration keys are not supported.
+      std::map<cass_int32_t, CassDuration> values;
+      values[1] = CassDuration(0, 0, 1);
+      values[3] = CassDuration(1, 2, 5);
+      values[5] = CassDuration(-1, -2, -3);
+      insert_map_value<cass_int32_t, CassDuration>(session.get(), CASS_VALUE_TYPE_INT, CASS_VALUE_TYPE_DURATION, values);
+    }
+
     {
       std::map<CassString, cass_int32_t> values;
       values[CassString("a")] = 1;
diff --git a/test/integration_tests/src/test_datatypes.cpp b/test/integration_tests/src/test_datatypes.cpp
index 4037140cf..c6f8d6ba6 100644
--- a/test/integration_tests/src/test_datatypes.cpp
+++ b/test/integration_tests/src/test_datatypes.cpp
@@ -143,6 +143,17 @@ BOOST_AUTO_TEST_CASE(read_write_primitives) {
     insert_value<CassDecimal>(CASS_VALUE_TYPE_DECIMAL, value);
   }
 
+  if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+    CassDuration value = CassDuration(0, 0, 0);
+    insert_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value);
+
+    value = CassDuration(1, 2, 3);
+    insert_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value);
+
+    value = CassDuration((1LL << 63) - 1, -1, 1LL << 63);
+    insert_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value);
+  }
+
   insert_value<cass_double_t>(CASS_VALUE_TYPE_DOUBLE, 3.141592653589793);
   insert_value<cass_float_t>(CASS_VALUE_TYPE_FLOAT, 3.1415926f);
   insert_value<cass_int32_t>(CASS_VALUE_TYPE_INT, 123);
diff --git a/test/integration_tests/src/test_named_parameters.cpp b/test/integration_tests/src/test_named_parameters.cpp
index 69a85b08f..bfb6b3c7e 100644
--- a/test/integration_tests/src/test_named_parameters.cpp
+++ b/test/integration_tests/src/test_named_parameters.cpp
@@ -350,6 +350,10 @@ BOOST_AUTO_TEST_CASE(all_primitives) {
         tester.insert_primitive_value<CassDecimal>(CASS_VALUE_TYPE_DECIMAL, value, is_prepared);
       }
 
+      if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+        tester.insert_primitive_value<CassDuration>(CASS_VALUE_TYPE_DURATION, CassDuration(1, 2, 3), is_prepared);
+      }
+
       tester.insert_primitive_value<cass_double_t>(CASS_VALUE_TYPE_DOUBLE, 3.141592653589793, is_prepared);
       tester.insert_primitive_value<cass_float_t>(CASS_VALUE_TYPE_FLOAT, 3.1415926f, is_prepared);
       tester.insert_primitive_value<cass_int32_t>(CASS_VALUE_TYPE_INT, 123, is_prepared);
@@ -424,6 +428,10 @@ BOOST_AUTO_TEST_CASE(all_primitives_batched) {
       tester.insert_primitive_batch_value<CassDecimal>(CASS_VALUE_TYPE_DECIMAL, value, TOTAL_NUMBER_OF_BATCHES);
     }
 
+    if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+      tester.insert_primitive_batch_value<CassDuration>(CASS_VALUE_TYPE_DURATION, CassDuration(1, 2, 3), TOTAL_NUMBER_OF_BATCHES);
+    }
+
     tester.insert_primitive_batch_value<cass_double_t>(CASS_VALUE_TYPE_DOUBLE, 3.141592653589793, TOTAL_NUMBER_OF_BATCHES);
     tester.insert_primitive_batch_value<cass_float_t>(CASS_VALUE_TYPE_FLOAT, 3.1415926f, TOTAL_NUMBER_OF_BATCHES);
     tester.insert_primitive_batch_value<cass_int32_t>(CASS_VALUE_TYPE_INT, 123, TOTAL_NUMBER_OF_BATCHES);
diff --git a/test/integration_tests/src/test_prepared.cpp b/test/integration_tests/src/test_prepared.cpp
index b5385cbdd..16336011a 100644
--- a/test/integration_tests/src/test_prepared.cpp
+++ b/test/integration_tests/src/test_prepared.cpp
@@ -54,6 +54,7 @@ struct AllTypes {
   cass_int16_t smallint_sample;
   CassDate date_sample;
   CassTime time_sample;
+  CassDuration duration_sample;
 };
 
 struct PreparedTests : public test_utils::SingleSessionTest {
@@ -66,7 +67,10 @@ struct PreparedTests : public test_utils::SingleSessionTest {
     test_utils::execute_query(session, str(boost::format(test_utils::CREATE_KEYSPACE_SIMPLE_FORMAT)
                                            % test_utils::SIMPLE_KEYSPACE % "1"));
     test_utils::execute_query(session, str(boost::format("USE %s") % test_utils::SIMPLE_KEYSPACE));
-    if ((version.major_version >= 2 && version.minor_version >= 2) || version.major_version >= 3) {
+    if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+      test_utils::execute_query(session, str(boost::format(test_utils::CREATE_TABLE_ALL_TYPES_V4_1) % ALL_TYPE_TABLE_NAME));
+      column_size_ = 16;
+    } else if ((version.major_version >= 2 && version.minor_version >= 2) || version.major_version >= 3) {
       test_utils::execute_query(session, str(boost::format(test_utils::CREATE_TABLE_ALL_TYPES_V4) % ALL_TYPE_TABLE_NAME));
       column_size_ = 15;
     } else {
@@ -81,6 +85,10 @@ struct PreparedTests : public test_utils::SingleSessionTest {
       columns_ += ", tinyint_sample, smallint_sample, date_sample, time_sample";
       values_ += ", ?, ?, ?, ?";
     }
+    if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+      columns_ += ", duration_sample";
+      values_ += ", ?";
+    }
   }
 
   ~PreparedTests() {
@@ -114,6 +122,11 @@ struct PreparedTests : public test_utils::SingleSessionTest {
       cass_statement_bind_uint32(statement.get(), 13, all_types.date_sample.date);
       cass_statement_bind_int64(statement.get(), 14, all_types.time_sample.time);
     }
+    if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+      cass_statement_bind_duration(statement.get(), 15,
+                                   all_types.duration_sample.months, all_types.duration_sample.days,
+                                   all_types.duration_sample.nanos);
+    }
 
     test_utils::CassFuturePtr future(cass_session_execute(session, statement.get()));
 
@@ -168,6 +181,14 @@ struct PreparedTests : public test_utils::SingleSessionTest {
       BOOST_REQUIRE(cass_value_get_int64(cass_row_get_column(row, 14), &output.time_sample.time) == CASS_OK);
       BOOST_REQUIRE(test_utils::Value<CassTime>::equal(input.time_sample, output.time_sample));
     }
+
+    if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+      BOOST_REQUIRE(cass_value_get_duration(cass_row_get_column(row, 15),
+                                           &output.duration_sample.months,
+                                           &output.duration_sample.days,
+                                           &output.duration_sample.nanos) == CASS_OK);
+      BOOST_REQUIRE(test_utils::Value<CassDuration>::equal(input.duration_sample, output.duration_sample));
+    }
   }
 };
 
@@ -217,6 +238,9 @@ BOOST_AUTO_TEST_CASE(bound_all_types_different_values)
     all_types[0].date_sample = test_utils::Value<CassDate>::max_value();
     all_types[0].time_sample = test_utils::Value<CassTime>::max_value();
   }
+  if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+    all_types[0].duration_sample = CassDuration(1, 2, 3);
+  }
 
   all_types[1].id = test_utils::generate_time_uuid(uuid_gen);
   all_types[1].text_sample = CassString("second");
@@ -235,6 +259,9 @@ BOOST_AUTO_TEST_CASE(bound_all_types_different_values)
     all_types[1].date_sample = 0;
     all_types[1].time_sample = 0;
   }
+  if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+    all_types[1].duration_sample = CassDuration(0, 0, 0);
+  }
 
   all_types[2].id = test_utils::generate_time_uuid(uuid_gen);
   all_types[2].text_sample = CassString("third");
@@ -253,6 +280,9 @@ BOOST_AUTO_TEST_CASE(bound_all_types_different_values)
     all_types[2].date_sample = test_utils::Value<CassDate>::min_value();
     all_types[2].time_sample = 12345678;
   }
+  if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+    all_types[2].duration_sample = CassDuration((1LL << 63) - 1, -1, 1LL << 63);
+  }
 
   for (size_t i = 0; i < all_types_count; ++i) {
     insert_all_types(session, prepared.get(), all_types[i]);
diff --git a/test/integration_tests/src/test_tuples.cpp b/test/integration_tests/src/test_tuples.cpp
index 0aa0f9a10..793e1a157 100644
--- a/test/integration_tests/src/test_tuples.cpp
+++ b/test/integration_tests/src/test_tuples.cpp
@@ -483,6 +483,11 @@ BOOST_AUTO_TEST_CASE(varying_size) {
             tester.insert_varying_sized_value<CassDecimal>(CASS_VALUE_TYPE_DECIMAL, value, size, nested_collection_type);
           }
 
+          if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+            CassDuration value = CassDuration(1, 2, 3);
+            tester.insert_varying_sized_value<CassDuration>(CASS_VALUE_TYPE_DURATION, value, size, nested_collection_type);
+          }
+
           tester.insert_varying_sized_value<cass_double_t>(CASS_VALUE_TYPE_DOUBLE, 3.141592653589793, size, nested_collection_type);
           tester.insert_varying_sized_value<cass_float_t>(CASS_VALUE_TYPE_FLOAT, 3.1415926f, size, nested_collection_type);
           tester.insert_varying_sized_value<cass_int32_t>(CASS_VALUE_TYPE_INT, 123, size, nested_collection_type);
@@ -559,6 +564,9 @@ BOOST_AUTO_TEST_CASE(null) {
           tester.insert_varying_sized_null_value<CassDate>(CASS_VALUE_TYPE_DATE, size, nested_collection_type);
           tester.insert_varying_sized_null_value<CassTime>(CASS_VALUE_TYPE_TIME, size, nested_collection_type);
         }
+        if ((version.major_version >= 3 && version.minor_version >= 10) || version.major_version >= 4) {
+          tester.insert_varying_sized_null_value<CassDuration>(CASS_VALUE_TYPE_DURATION, size, nested_collection_type);
+        }
       }
     }
   } else {
diff --git a/test/integration_tests/src/test_utils.cpp b/test/integration_tests/src/test_utils.cpp
index f01d031ef..9eb377cee 100644
--- a/test/integration_tests/src/test_utils.cpp
+++ b/test/integration_tests/src/test_utils.cpp
@@ -69,6 +69,25 @@ const char* CREATE_TABLE_ALL_TYPES_V4 =
     "date_sample date,"
     "time_sample time);";
 
+const char* CREATE_TABLE_ALL_TYPES_V4_1 =
+    "CREATE TABLE %s ("
+    "id uuid PRIMARY KEY,"
+    "text_sample text,"
+    "int_sample int,"
+    "bigint_sample bigint,"
+    "float_sample float,"
+    "double_sample double,"
+    "decimal_sample decimal,"
+    "blob_sample blob,"
+    "boolean_sample boolean,"
+    "timestamp_sample timestamp,"
+    "inet_sample inet,"
+    "tinyint_sample tinyint,"
+    "smallint_sample smallint,"
+    "date_sample date,"
+    "time_sample time,"
+    "duration_sample duration);";
+
 const char* CREATE_TABLE_TIME_SERIES =
     "CREATE TABLE %s ("
     "id uuid,"
@@ -128,6 +147,7 @@ const char* get_value_type(CassValueType type) {
     case CASS_VALUE_TYPE_BOOLEAN: return "boolean";
     case CASS_VALUE_TYPE_COUNTER: return "counter";
     case CASS_VALUE_TYPE_DECIMAL: return "decimal";
+    case CASS_VALUE_TYPE_DURATION: return "duration";
     case CASS_VALUE_TYPE_DOUBLE: return "double";
     case CASS_VALUE_TYPE_FLOAT: return "float";
     case CASS_VALUE_TYPE_INT: return "int";
diff --git a/test/integration_tests/src/test_utils.hpp b/test/integration_tests/src/test_utils.hpp
index 43761363d..60a42a2b5 100644
--- a/test/integration_tests/src/test_utils.hpp
+++ b/test/integration_tests/src/test_utils.hpp
@@ -83,6 +83,18 @@ struct CassDecimal {
   cass_int32_t scale;
 };
 
+struct CassDuration {
+  CassDuration()
+    : months(0)
+    , days(0)
+    , nanos(0) {}
+  CassDuration(cass_int64_t months, cass_int64_t days, cass_int64_t nanos)
+    : months(months), days(days), nanos(nanos) {}
+  cass_int64_t months;
+  cass_int64_t days;
+  cass_int64_t nanos;
+};
+
 struct CassDate {
   CassDate(cass_uint32_t date = 0)
     : date(date) { }
@@ -1104,6 +1116,50 @@ struct Value<CassDecimal> {
   }
 };
 
+template<>
+struct Value<CassDuration> {
+  static CassError bind(CassStatement* statement, size_t index, CassDuration value) {
+    return cass_statement_bind_duration(statement, index, value.months, value.days, value.nanos);
+  }
+
+  static CassError bind_by_name(CassStatement* statement, const char* name, CassDuration value) {
+    return cass_statement_bind_duration_by_name(statement, name, value.months, value.days, value.nanos);
+  }
+
+  static CassError append(CassCollection* collection, CassDuration value) {
+    return cass_collection_append_duration(collection, value.months, value.days, value.nanos);
+  }
+
+  static CassError tuple_set(CassTuple* tuple, size_t index, CassDuration value) {
+    return cass_tuple_set_duration(tuple, index, value.months, value.days, value.nanos);
+  }
+
+  static CassError user_type_set(CassUserType* user_type, size_t index, CassDuration value) {
+    return cass_user_type_set_duration(user_type, index, value.months, value.days, value.nanos);
+  }
+
+  static CassError user_type_set_by_name(CassUserType* user_type, const char* name, CassDuration value) {
+    return cass_user_type_set_duration_by_name(user_type, name, value.months, value.days, value.nanos);
+  }
+
+  static CassError get(const CassValue* value, CassDuration* output) {
+    return cass_value_get_duration(value, &output->months, &output->days, &output->nanos);
+  }
+
+  static bool equal(CassDuration a, CassDuration b) {
+    return a.months == b.months && a.days == b.days && a.nanos == b.nanos;
+  }
+
+  static std::string to_string(CassDuration value) {
+    // String representation of Duration is wonky in C*. (-3, -2, -1) is represented by
+    // -3mo2d1ns. There is no way to represent a mix of positive and negative attributes. We tippy-toe
+    // around this in our testing...
+    std::ostringstream buf;
+    buf << value.months << "mo" << (value.days < 0 ? -value.days : value.days) << "d" << (value.nanos < 0 ? -value.nanos : value.nanos) << "ns";
+    return buf.str();
+  }
+};
+
 /*
  * TODO: Implement https://datastax-oss.atlassian.net/browse/CPP-244 to avoid
  *       current test skip implementation in batch and serial_consistency.
@@ -1275,6 +1331,7 @@ void wait_for_node_connections(const std::string& ip_prefix, std::vector<int> no
 
 extern const char* CREATE_TABLE_ALL_TYPES;
 extern const char* CREATE_TABLE_ALL_TYPES_V4;
+extern const char* CREATE_TABLE_ALL_TYPES_V4_1;
 extern const char* CREATE_TABLE_TIME_SERIES;
 extern const char* CREATE_TABLE_SIMPLE;
 
@@ -1353,3 +1410,23 @@ inline bool operator<(CassDecimal a, CassDecimal b) {
   }
   return memcmp(a.varint, b.varint, a.varint_size) < 0;
 }
+
+inline bool operator==(CassDuration a, CassDuration b) {
+  return a.months == b.months && a.days == b.days && a.nanos == b.nanos;
+}
+
+inline bool operator<(CassDuration a, CassDuration b) {
+  if (a.months > b.months) {
+    return false;
+  } else if (a.months < b.months) {
+    return true;
+  }
+
+  if (a.days > b.days) {
+    return false;
+  } else if (a.days < b.days) {
+    return true;
+  }
+
+  return a.nanos < b.nanos;
+}
\ No newline at end of file
diff --git a/test/unit_tests/src/test_data_type.cpp b/test/unit_tests/src/test_data_type.cpp
index 36b427992..b39dfe9f9 100644
--- a/test/unit_tests/src/test_data_type.cpp
+++ b/test/unit_tests/src/test_data_type.cpp
@@ -102,7 +102,7 @@ BOOST_AUTO_TEST_CASE(class_name)
 
   // Invalid type
   {
-    // Only custom data types support keyspace and type name
+    // Only custom data types support class name
     DataTypeWrapper data_type(cass_data_type_new(CASS_VALUE_TYPE_UDT));
 
     BOOST_CHECK_EQUAL(cass_data_type_set_class_name(data_type, "class_name1"),
diff --git a/test/unit_tests/src/test_encode.cpp b/test/unit_tests/src/test_encode.cpp
new file mode 100644
index 000000000..4339bab2c
--- /dev/null
+++ b/test/unit_tests/src/test_encode.cpp
@@ -0,0 +1,82 @@
+/*
+  Copyright (c) 2014-2016 DataStax
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+#ifdef STAND_ALONE
+#   define BOOST_TEST_MODULE cassandra
+#endif
+
+#include "encode.hpp"
+
+#include <boost/test/unit_test.hpp>
+#include <time.h>
+
+using namespace cass;
+
+BOOST_AUTO_TEST_SUITE(encode_duration)
+
+BOOST_AUTO_TEST_CASE(base)
+{
+  CassDuration value(0, 0, 0);
+  Buffer result = encode(value);
+  BOOST_CHECK_EQUAL(3, result.size());
+  const char* result_data = result.data();
+  BOOST_CHECK_EQUAL(result_data[0], 0);
+  BOOST_CHECK_EQUAL(result_data[1], 0);
+  BOOST_CHECK_EQUAL(result_data[2], 0);
+}
+
+BOOST_AUTO_TEST_CASE(simple)
+{
+  CassDuration value(1, 2, 3);
+  Buffer result = encode(value);
+  BOOST_CHECK_EQUAL(3, result.size());
+  const char* result_data = result.data();
+  BOOST_CHECK_EQUAL(result_data[0], 2);
+  BOOST_CHECK_EQUAL(result_data[1], 4);
+  BOOST_CHECK_EQUAL(result_data[2], 6);
+}
+
+BOOST_AUTO_TEST_CASE(edge)
+{
+  CassDuration value((1ULL << 63) - 1, -1, 1LL << 63);
+  Buffer result = encode(value);
+  BOOST_CHECK_EQUAL(19, result.size());
+  unsigned const char* result_data = reinterpret_cast<unsigned const char*>(result.data());
+
+  // The first 9 bytes represent (1LL<<63 - 1), the max 64-bit number. Byte 0
+  // has all bits set to indicate that there are 8 bytes beyond this one that
+  // define this field (each field is a vint of a zigzag encoding of the original
+  // value). Encoding places the least-significant byte at byte 8 and works backwards
+  // to record more significant bytes. Zigzag encoding just left shifts a value
+  // by one bit for positive values, so byte 8 ends in a 0.
+
+  for (int ind = 0; ind < 8; ++ind) {
+    BOOST_CHECK_EQUAL(result_data[ind], 0xff);
+  }
+  BOOST_CHECK_EQUAL(result_data[8], 0xfe);
+
+  // Next we have a 1-byte vint for -1.
+  BOOST_CHECK_EQUAL(result_data[9], 1);
+
+  // Finally, we have 9-bytes for 1LL << 63, the min 64-bit number. The zigzag
+  // representation is 8 bytes of 0xff, and the first byte is 0xff to say we have
+  // 8 bytes of value beyond these size-spec-bits.
+  for (int ind = 10; ind < 19; ++ind) {
+    BOOST_CHECK_EQUAL(result_data[ind], 0xff);
+  }
+}
+
+BOOST_AUTO_TEST_SUITE_END()
diff --git a/test/unit_tests/src/test_serialization.cpp b/test/unit_tests/src/test_serialization.cpp
new file mode 100644
index 000000000..0561a8dec
--- /dev/null
+++ b/test/unit_tests/src/test_serialization.cpp
@@ -0,0 +1,35 @@
+/*
+  Copyright (c) 2014-2016 DataStax
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+#ifdef STAND_ALONE
+#   define BOOST_TEST_MODULE cassandra
+#endif
+
+#include "serialization.hpp"
+
+#include <ctype.h>
+#include <stdio.h>
+
+#include <boost/test/unit_test.hpp>
+
+BOOST_AUTO_TEST_SUITE(serialization)
+
+BOOST_AUTO_TEST_CASE(decode_zig_zag)
+{
+  BOOST_CHECK_EQUAL(1LL << 63, cass::decode_zig_zag((long) -1));
+}
+
+BOOST_AUTO_TEST_SUITE_END()
diff --git a/test/unit_tests/src/test_utils.cpp b/test/unit_tests/src/test_utils.cpp
index 20c19a11f..ecc5667ef 100644
--- a/test/unit_tests/src/test_utils.cpp
+++ b/test/unit_tests/src/test_utils.cpp
@@ -65,4 +65,11 @@ BOOST_AUTO_TEST_CASE(escape_id)
   BOOST_CHECK_EQUAL(cass::escape_id(s), std::string("\"a\"\"Bc\""));
 }
 
+BOOST_AUTO_TEST_CASE(num_leading_zeros)
+{
+  BOOST_CHECK_EQUAL(64, cass::num_leading_zeros(0));
+  BOOST_CHECK_EQUAL(0, cass::num_leading_zeros(1LL << 63));
+  BOOST_CHECK_EQUAL(0, cass::num_leading_zeros(1LL << 63 | 1 << 5));
+}
+
 BOOST_AUTO_TEST_SUITE_END()
diff --git a/test/unit_tests/src/test_value.cpp b/test/unit_tests/src/test_value.cpp
new file mode 100644
index 000000000..85f2a03fe
--- /dev/null
+++ b/test/unit_tests/src/test_value.cpp
@@ -0,0 +1,71 @@
+/*
+  Copyright (c) 2014-2016 DataStax
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+*/
+
+#ifdef STAND_ALONE
+#   define BOOST_TEST_MODULE cassandra
+#endif
+
+#include "value.hpp"
+#include "cassandra.h"
+
+#include <boost/test/unit_test.hpp>
+#include <time.h>
+
+// The following CassValue's are used in tests as "bad data".
+
+// Create a CassValue representing a text type.
+static cass::DataType::ConstPtr s_text_type(new cass::DataType(CASS_VALUE_TYPE_TEXT));
+static cass::Value s_text_value_Value(4, s_text_type, NULL, 0);
+static CassValue* s_text_value = CassValue::to(&s_text_value_Value);
+
+BOOST_AUTO_TEST_SUITE(bad_value)
+
+// ST is simple-type name (e.g. int8 and the like) and T is the full type name (e.g. cass_int8_t, CassUuid, etc.).
+#define TEST_TYPE(ST, T) \
+BOOST_AUTO_TEST_CASE(bad_##ST) \
+{ \
+  T output; \
+  BOOST_CHECK_EQUAL(cass_value_get_##ST(s_text_value, &output), CASS_ERROR_LIB_INVALID_VALUE_TYPE); \
+}
+
+#define TEST_SIMPLE_TYPE(T) TEST_TYPE(T, cass_##T##_t)
+
+TEST_SIMPLE_TYPE(int8)
+TEST_SIMPLE_TYPE(int16)
+TEST_SIMPLE_TYPE(int32)
+TEST_SIMPLE_TYPE(uint32)
+TEST_SIMPLE_TYPE(int64)
+TEST_SIMPLE_TYPE(float)
+TEST_SIMPLE_TYPE(double)
+TEST_TYPE(uuid, CassUuid)
+TEST_TYPE(inet, CassInet)
+
+BOOST_AUTO_TEST_CASE(bad_duration)
+{
+  cass_int64_t months, days, nanos;
+  BOOST_CHECK_EQUAL(cass_value_get_duration(s_text_value, &months, &days, &nanos), CASS_ERROR_LIB_INVALID_VALUE_TYPE);
+}
+
+BOOST_AUTO_TEST_CASE(bad_decimal)
+{
+  const cass_byte_t* varint;
+  size_t varint_size;
+  cass_int32_t scale;
+  BOOST_CHECK_EQUAL(cass_value_get_decimal(s_text_value, &varint, &varint_size, &scale),
+    CASS_ERROR_LIB_INVALID_VALUE_TYPE);
+}
+
+BOOST_AUTO_TEST_SUITE_END()