diff --git a/README.md b/README.md
index a96107080..d2e9a7d6b 100644
--- a/README.md
+++ b/README.md
@@ -1,4 +1,4 @@
-phantom [![Build Status](https://travis-ci.org/outworkers/phantom.svg?branch=develop)](https://travis-ci.org/outworkers/phantom) [![Coverage Status](https://coveralls.io/repos/outworkers/phantom/badge.svg)](https://coveralls.io/r/outworkers/phantom) [![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.websudos/phantom_2.10/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.websudos/phantom_2.10) [![Bintray](https://api.bintray.com/packages/websudos/oss-releases/phantom/images/download.svg) ](https://bintray.com/websudos/oss-releases/phantom/_latestVersion) [![Gitter](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/outworkers/phantom?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
+phantom [![Build Status](https://travis-ci.org/outworkers/phantom.svg?branch=develop)](https://travis-ci.org/outworkers/phantom) [![Coverage Status](https://coveralls.io/repos/outworkers/phantom/badge.svg)](https://coveralls.io/r/outworkers/phantom) [![Maven Central](https://maven-badges.herokuapp.com/maven-central/com.websudos/phantom_2.10/badge.svg)](https://maven-badges.herokuapp.com/maven-central/com.websudos/phantom_2.10) [![Bintray](https://api.bintray.com/packages/websudos/oss-releases/phantom/images/download.svg) ](https://bintray.com/websudos/oss-releases/phantom/_latestVersion) [![Codacy Rating](https://api.codacy.com/project/badge/grade/25bee222a7d142ff8151e6ceb39151b4)](https://www.codacy.com/app/flavian/phantom_2) [![Gitter](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/outworkers/phantom?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 ==================================================================================================
 
 Reactive type-safe Scala DSL for Cassandra
diff --git a/build/publish_develop.sh b/build/publish_develop.sh
index 1ce1cf338..bd4248c18 100755
--- a/build/publish_develop.sh
+++ b/build/publish_develop.sh
@@ -5,12 +5,30 @@ then
     echo "The current JDK version is ${TRAVIS_JDK_VERSION}"
     echo "The current Scala version is ${TRAVIS_SCALA_VERSION}"
 
-    CURRENT_VERSION = "$(sbt version)"
-    echo "Bumping release version with a patch increment from $CURRENT_VERSION"
+    echo "Creating credentials file"
+    if [ -e "$HOME/.bintray/.credentials" ]; then
+        echo "Bintray redentials file already exists"
+    else
+        mkdir -p "$HOME/.bintray/"
+        touch "$HOME/.bintray/.credentials"
+        echo "realm = Bintray API Realm" >> "$HOME/.bintray/.credentials"
+        echo "host = api.bintray.com" >> "$HOME/.bintray/.credentials"
+        echo "user = $bintray_user" >> "$HOME/.bintray/.credentials"
+        echo "password = $bintray_password" >> "$HOME/.bintray/.credentials"
+    fi
+
+    if [ -e "$HOME/.bintray/.credentials" ]; then
+        echo "Bintray credentials file succesfully created"
+    else
+        echo "Bintray credentials still not found"
+    fi
+
+
+    echo "Bumping release version with a patch increment from $(sbt version)"
     sbt version-bump-patch
 
-    NEW_VERSION = "$(sbt version)"
-    echo "Creating Git tag for version $NEW_VERSION"
+    SET NEW_VERSION = "$(sbt version)"
+    echo "Creating Git tag for version $(sbt version)"
 
     echo "Pushing tag to GitHub."
     git push --tags "https://${github_token}@${GH_REF}" > /dev/null 2>&1
@@ -19,8 +37,8 @@ then
 
     if [ "${TRAVIS_SCALA_VERSION}" == "2.11.7" ] && [ "${TRAVIS_JDK_VERSION}" == "oraclejdk8" ];
     then
-        "Publishing $NEW_VERSION to bintray"
-        sbt bintray:publish
+        "Publishing $(sbt version) to bintray"
+        sbt +publish
     else
         echo "Only publishing version for Scala 2.11.7 and Oracle JDK 8 to prevent multiple artifacts"
     fi
@@ -28,7 +46,7 @@ then
     git checkout master
     git merge develop
 
-    git push --all  "https://${github_token}@${GH_REF}" master > /dev/null 2>&1
+    git push --all  "https://${github_token}@${GH_REF}" develop:master > /dev/null 2>&1
 
 
 else
diff --git a/phantom-dsl/src/main/scala/com/websudos/phantom/builder/ops/QueryColumn.scala b/phantom-dsl/src/main/scala/com/websudos/phantom/builder/ops/QueryColumn.scala
index fb106cdd0..36107b261 100644
--- a/phantom-dsl/src/main/scala/com/websudos/phantom/builder/ops/QueryColumn.scala
+++ b/phantom-dsl/src/main/scala/com/websudos/phantom/builder/ops/QueryColumn.scala
@@ -145,6 +145,30 @@ sealed class QueryColumn[RR : Primitive](val col: AbstractColumn[RR]) {
   final def eqs(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = {
     new PreparedWhereClause.ParametricCondition[RR](QueryBuilder.Where.eqs(col.name, value.symbol))
   }
+
+  final def lt(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = {
+    new PreparedWhereClause.ParametricCondition[RR](QueryBuilder.Where.lt(col.name, value.symbol))
+  }
+
+  final def <(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = lt(value)
+
+  final def lte(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = {
+    new PreparedWhereClause.ParametricCondition[RR](QueryBuilder.Where.lte(col.name, value.symbol))
+  }
+
+  final def <=(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = lte(value)
+
+  final def gt(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = {
+    new PreparedWhereClause.ParametricCondition[RR](QueryBuilder.Where.gt(col.name, value.symbol))
+  }
+
+  final def >(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = gt(value)
+
+  final def gte(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = {
+    new PreparedWhereClause.ParametricCondition[RR](QueryBuilder.Where.gte(col.name, value.symbol))
+  }
+
+  final def >=(value: PrepareMark): PreparedWhereClause.ParametricCondition[RR] = gte(value)
 }
 
 /**
diff --git a/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/RelationalOperatorsTest.scala b/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/RelationalOperatorsTest.scala
new file mode 100644
index 000000000..a4fedf450
--- /dev/null
+++ b/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/RelationalOperatorsTest.scala
@@ -0,0 +1,283 @@
+/*
+ * Copyright 2013-2016 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.builder.query.db.crud
+
+import com.twitter.util.{Future => TwitterFuture}
+import com.websudos.phantom.PhantomSuite
+import com.websudos.phantom.builder.query.db.ordering.TimeSeriesTest
+import com.websudos.phantom.builder.query.prepared._
+import com.websudos.phantom.dsl._
+import com.websudos.phantom.tables._
+import com.websudos.util.testing._
+import org.slf4j.LoggerFactory
+import scala.concurrent.{Future => ScalaFuture}
+
+class RelationalOperatorsTest extends PhantomSuite {
+  val logger = LoggerFactory.getLogger(this.getClass)
+
+  val numRecords = 100
+  val records: Seq[TimeSeriesRecord] = TimeSeriesTest.genSequentialRecords(numRecords)
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+
+    TestDatabase.timeSeriesTable.insertSchema()
+
+    val chain = for {
+      truncate <- TestDatabase.timeSeriesTable.truncate.future()
+      inserts <- TimeSeriesTest.addRecordsToBatch(records).future()
+    } yield inserts
+
+    chain.successful { inserts =>
+      logger.debug(s"Initialized table with $numRecords records")
+    }
+  }
+
+  it should "fetch records using less than operator" in {
+    val maxIndex = 50
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp < maxTimestamp)
+      .allowFiltering()
+      .fetch()
+
+    val expected = records.filter(_.timestamp.isBefore(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than operator with Twitter Futures" in {
+    val maxIndex = 50
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp < maxTimestamp)
+      .allowFiltering()
+      .collect()
+
+    val expected = records.filter(_.timestamp.isBefore(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than operator with prepared statement" in {
+    val maxIndex = 50
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.timestamp < ?)
+      .allowFiltering()
+      .prepare()
+
+    val futureResults = query.bind(maxTimestamp).fetch()
+    val expected = records.filter(_.timestamp.isBefore(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than or equal operator" in {
+    val maxIndex = 40
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp <= maxTimestamp)
+      .allowFiltering()
+      .fetch()
+
+    val expected = records.filter(!_.timestamp.isAfter(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than or equal operator with Twitter Futures" in {
+    val maxIndex = 40
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp <= maxTimestamp)
+      .allowFiltering()
+      .collect()
+
+    val expected = records.filter(!_.timestamp.isAfter(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than or equal operator with prepared statement" in {
+    val maxIndex = 40
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.timestamp <= ?)
+      .allowFiltering()
+      .prepare()
+
+    val futureResults = query.bind(maxTimestamp).fetch()
+    val expected = records.filter(!_.timestamp.isAfter(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using greater than operator" in {
+    val minIndex = 60
+    val minTimestamp = records(minIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp > minTimestamp)
+      .allowFiltering()
+      .fetch()
+
+    val expected = records.filter(_.timestamp.isAfter(minTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using greater than operator with Twitter Futures" in {
+    val minIndex = 60
+    val minTimestamp = records(minIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp > minTimestamp)
+      .allowFiltering()
+      .collect()
+
+    val expected = records.filter(_.timestamp.isAfter(minTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using greater than operator with prepared statement" in {
+    val minIndex = 60
+    val minTimestamp = records(minIndex).timestamp
+
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.timestamp > ?)
+      .allowFiltering()
+      .prepare()
+
+    val futureResults = query.bind(minTimestamp).fetch()
+    val expected = records.filter(_.timestamp.isAfter(minTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using greater than or equal operator" in {
+    val minIndex = 75
+    val minTimestamp = records(minIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp >= minTimestamp)
+      .allowFiltering()
+      .fetch()
+
+    val expected = records.filter(!_.timestamp.isBefore(minTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using greater than or equal operator with Twitter Futures" in {
+    val minIndex = 75
+    val minTimestamp = records(minIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp >= minTimestamp)
+      .allowFiltering()
+      .collect()
+
+    val expected = records.filter(!_.timestamp.isBefore(minTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using greater than or equal operator with prepared statement" in {
+    val minIndex = 75
+    val minTimestamp = records(minIndex).timestamp
+
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.timestamp >= ?)
+      .allowFiltering()
+      .prepare()
+
+    val futureResults = query.bind(minTimestamp).fetch()
+    val expected = records.filter(!_.timestamp.isBefore(minTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than and greater than operators" in {
+    val minIndex = 10
+    val maxIndex = 40
+    val minTimestamp = records(minIndex).timestamp
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp > minTimestamp)
+      .and(_.timestamp < maxTimestamp)
+      .allowFiltering()
+      .fetch()
+
+    val expected =  records.filter(r => r.timestamp.isAfter(minTimestamp) && r.timestamp.isBefore(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than and greater than operators with Twitter Futures" in {
+    val minIndex = 10
+    val maxIndex = 40
+    val minTimestamp = records(minIndex).timestamp
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val futureResults = TestDatabase.timeSeriesTable.select
+      .where(_.timestamp > minTimestamp)
+      .and(_.timestamp < maxTimestamp)
+      .allowFiltering()
+      .collect()
+
+    val expected =  records.filter(r => r.timestamp.isAfter(minTimestamp) && r.timestamp.isBefore(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  it should "fetch records using less than and greater than operators with prepared statement" in {
+    val minIndex = 10
+    val maxIndex = 40
+    val minTimestamp = records(minIndex).timestamp
+    val maxTimestamp = records(maxIndex).timestamp
+
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.timestamp > ?)
+      .p_and(_.timestamp < ?)
+      .allowFiltering()
+      .prepare()
+
+    val futureResults = query.bind(minTimestamp, maxTimestamp).fetch()
+    val expected =  records.filter(r => r.timestamp.isAfter(minTimestamp) && r.timestamp.isBefore(maxTimestamp))
+    verifyResults(futureResults, expected)
+  }
+
+  def verifyResults(futureResults: ScalaFuture[Seq[TimeSeriesRecord]], expected: Seq[TimeSeriesRecord]): Unit = {
+    futureResults.successful { results =>
+      results.toSet shouldEqual expected.toSet
+    }
+  }
+
+  def verifyResults(futureResults: TwitterFuture[Seq[TimeSeriesRecord]], expected: Seq[TimeSeriesRecord]): Unit = {
+    futureResults.successful { results =>
+      results.toSet shouldEqual expected.toSet
+    }
+  }
+}
diff --git a/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/TTLTest.scala b/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/TTLTest.scala
index 119d7c41d..b638b7f89 100644
--- a/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/TTLTest.scala
+++ b/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/crud/TTLTest.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 Websudos, Limited.
+ * Copyright 2013-2016 Websudos, Limited.
  *
  * All rights reserved.
  *
@@ -30,12 +30,13 @@
 package com.websudos.phantom.builder.query.db.crud
 
 import com.websudos.phantom.PhantomSuite
+import com.websudos.phantom.builder.query.prepared._
 import com.websudos.phantom.dsl._
 import com.websudos.phantom.tables.{TestDatabase, Primitive}
 import com.websudos.util.testing._
 import org.scalatest.concurrent.Eventually
 import org.scalatest.time.{Milliseconds, Seconds, Span}
-import org.scalatest.time.SpanSugar._
+import scala.concurrent.duration._
 
 class TTLTest extends PhantomSuite with Eventually {
 
@@ -46,42 +47,99 @@ class TTLTest extends PhantomSuite with Eventually {
     TestDatabase.primitives.insertSchema()
   }
 
-  it should "expire inserted records after 2 seconds" in {
+  private val ttl = 2 seconds
+  private val granularity = 1 second
+
+  it should "expire inserted records after TTL" in {
     val row = gen[Primitive]
 
     val chain = for {
-      store <- TestDatabase.primitives.store(row).ttl(2).future()
+      store <- TestDatabase.primitives.store(row).ttl(ttl).future()
       get <- TestDatabase.primitives.select.where(_.pkey eqs row.pkey).one()
     } yield get
 
-    whenReady(chain) {
-      record => {
-        record.value shouldEqual row
+    chain.successful { record =>
+      record shouldEqual Some(row)
+    }
 
-        eventually {
-          val record = TestDatabase.primitives.select.where(_.pkey eqs row.pkey).one().block(3.seconds)
-          record shouldBe empty
-        }
+    eventually(timeout(ttl + granularity)) {
+      val futureRecord = TestDatabase.primitives.select.where(_.pkey eqs row.pkey).one()
+      futureRecord.successful { record =>
+        record shouldBe empty
       }
     }
   }
 
-  it should "expire inserted records after 2 seconds with Twitter Futures" in {
+  it should "expire inserted records after TTL with Twitter Futures" in {
     val row = gen[Primitive]
 
     val chain = for {
-      store <- TestDatabase.primitives.store(row).ttl(2).execute()
+      store <- TestDatabase.primitives.store(row).ttl(ttl).execute()
       get <- TestDatabase.primitives.select.where(_.pkey eqs row.pkey).get()
     } yield get
 
-    chain.successful {
-      record => {
-        record.value shouldEqual row
+    chain.successful { record =>
+      record shouldEqual Some(row)
+    }
+
+    eventually(timeout(ttl + granularity)) {
+      val futureRecord = TestDatabase.primitives.select.where(_.pkey eqs row.pkey).get()
+      futureRecord.successful { record =>
+        record shouldBe empty
+      }
+    }
+  }
+
+  it should "expire inserted records after TTL with prepared statement" in {
+    val row = gen[Primitive]
+
+    val fetchQuery = TestDatabase.primitives.select
+      .p_where(_.pkey eqs ?)
+      .prepare()
+
+    val insertQuery = TestDatabase.primitives.insert
+      .p_value(_.pkey, ?)
+      .p_value(_.long, ?)
+      .p_value(_.boolean, ?)
+      .p_value(_.bDecimal, ?)
+      .p_value(_.double, ?)
+      .p_value(_.float, ?)
+      .p_value(_.inet, ?)
+      .p_value(_.int, ?)
+      .p_value(_.date, ?)
+      .p_value(_.uuid, ?)
+      .p_value(_.bi, ?)
+      .ttl(ttl)
+      .prepare()
+
+    def preparedInsert(row: Primitive): ExecutablePreparedQuery = {
+      insertQuery.bind(
+        row.pkey,
+        row.long,
+        row.boolean,
+        row.bDecimal,
+        row.double,
+        row.float,
+        row.inet,
+        row.int,
+        row.date,
+        row.uuid,
+        row.bi)
+    }
+
+    val chain = for {
+      store <- preparedInsert(row).future()
+      get <- fetchQuery.bind(row.pkey).one()
+    } yield get
+
+    chain.successful { result =>
+      result shouldEqual Some(row)
+    }
 
-        eventually {
-          val record = TestDatabase.primitives.select.where(_.pkey eqs row.pkey).one().block(3.seconds)
-          record shouldBe empty
-        }
+    eventually(timeout(ttl + granularity)) {
+      val futureResults = fetchQuery.bind(row.pkey).one()
+      futureResults.successful { results =>
+        results.isEmpty shouldBe true
       }
     }
   }
diff --git a/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/ordering/TimeSeriesTest.scala b/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/ordering/TimeSeriesTest.scala
index 3faa865fd..ce3925011 100644
--- a/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/ordering/TimeSeriesTest.scala
+++ b/phantom-dsl/src/test/scala/com/websudos/phantom/builder/query/db/ordering/TimeSeriesTest.scala
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 Websudos, Limited.
+ * Copyright 2013-2016 Websudos, Limited.
  *
  * All rights reserved.
  *
@@ -29,15 +29,20 @@
  */
 package com.websudos.phantom.builder.query.db.ordering
 
+import com.datastax.driver.core.Session
+import com.twitter.util.{Future => TwitterFuture}
 import com.websudos.phantom.PhantomSuite
-
-import scala.concurrent.duration._
-
-import org.scalatest.concurrent.PatienceConfiguration
-
+import com.websudos.phantom.batch.BatchQuery
+import com.websudos.phantom.builder.Unspecified
+import com.websudos.phantom.builder.query.prepared._
+import com.websudos.phantom.connectors.KeySpace
 import com.websudos.phantom.dsl._
 import com.websudos.phantom.tables._
 import com.websudos.util.testing._
+import org.scalatest.concurrent.PatienceConfiguration
+import scala.concurrent.duration._
+import scala.concurrent.{Future => ScalaFuture}
+import TimeSeriesTest._
 
 class TimeSeriesTest extends PhantomSuite {
 
@@ -48,234 +53,192 @@ class TimeSeriesTest extends PhantomSuite {
     TestDatabase.timeSeriesTable.insertSchema()
   }
 
-  protected[this] final val durationOffset = 1000
+  it should "fetch records in natural order for a descending clustering order" in {
+    val number = 10
+    val limit = 5
 
-  it should "allow using naturally fetch the records in descending order for a descending clustering order" in {
+    val records = genSequentialRecords(number)
 
-    var i = 0
-    val number = 5
+    val chain = for {
+      truncate <- TestDatabase.timeSeriesTable.truncate.future()
+      insert <- addRecordsToBatch(records).future()
+      chunks <- TestDatabase.timeSeriesTable.select.limit(limit).fetch()
+    } yield chunks
 
-    val recordList = genList[TimeSeriesRecord](number).map(
-      item => {
-        i += 1
-        item.copy(
-          id = TestDatabase.timeSeriesTable.testUUID,
-          timestamp = item.timestamp.withDurationAdded(durationOffset, i)
-        )
-      })
+    verifyResults(chain, records.reverse.take(limit))
+  }
 
-    val ts = recordList.map(_.timestamp.getSecondOfDay)
+  it should "fetch records in natural order for a descending clustering order with Twitter Futures" in {
+    val number = 10
+    val limit = 5
 
-    val batch = recordList.foldLeft(Batch.unlogged) {
-      (b, record) => {
-        b.add(TestDatabase.timeSeriesTable.insert
-          .value(_.id, record.id)
-          .value(_.name, record.name)
-          .value(_.timestamp, record.timestamp))
-      }
-    }
+    val records = genSequentialRecords(number)
 
     val chain = for {
-      truncate <- TestDatabase.timeSeriesTable.truncate.future()
-      insert <- batch.future()
-      chunks <- TestDatabase.timeSeriesTable.select.limit(number).fetch()
+      truncate <- TestDatabase.timeSeriesTable.truncate.execute()
+      insert <- addRecordsToBatch(records).execute()
+      chunks <- TestDatabase.timeSeriesTable.select.limit(limit).collect()
     } yield chunks
 
-
-    chain.successful {
-      res =>
-        val mapped = res.map(_.timestamp.getSecondOfDay)
-        mapped.toList shouldEqual ts.reverse
-    }
+    verifyResults(chain, records.reverse.take(limit))
   }
 
-  it should "allow using naturally fetch the records in descending order for a descending clustering order with Twitter Futures" in {
-    var i = 0
-    val number = 5
-
-    val recordList = genList[TimeSeriesRecord](number).map(
-      item => {
-        i += 1
-        item.copy(
-          id = TestDatabase.timeSeriesTable.testUUID,
-          timestamp = item.timestamp.withDurationAdded(durationOffset, i)
-        )
-      })
+  it should "fetch records in natural order for a descending clustering order with prepared statements" in {
+    val number = 10
+    val limit = 5
 
-    val ts = recordList.map(_.timestamp.getSecondOfDay)
+    val records = genSequentialRecords(number)
 
-    val batch = recordList.foldLeft(Batch.unlogged) {
-      (b, record) => {
-        b.add(TestDatabase.timeSeriesTable.insert
-          .value(_.id, record.id)
-          .value(_.name, record.name)
-          .value(_.timestamp, record.timestamp))
-      }
-    }
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.id eqs ?)
+      .limit(limit)
+      .prepare()
 
     val chain = for {
-      truncate <- TestDatabase.timeSeriesTable.truncate.execute()
-      insert <- batch.execute()
-      chunks <- TestDatabase.timeSeriesTable.select.limit(number).collect()
+      truncate <- TestDatabase.timeSeriesTable.truncate.future()
+      insert <- addRecordsToBatch(records).future()
+      chunks <- query.bind(TestDatabase.timeSeriesTable.testUUID).fetch()
     } yield chunks
 
-    chain.successful {
-      res =>
-        val mapped = res.map(_.timestamp.getSecondOfDay)
-        mapped.toList shouldEqual ts.reverse
-    }
+    verifyResults(chain, records.reverse.take(limit))
   }
 
-  it should "allow fetching the records in ascending order for a descending clustering order using order by clause" in {
-    var i = 0
-    val number = 5
+  it should "fetch records in ascending order for a descending clustering order using order by clause" in {
+    val number = 10
+    val limit = 5
 
-    val recordList = genList[TimeSeriesRecord](number).map(
-      item => {
-        i += 1
-        item.copy(
-          id = TestDatabase.timeSeriesTable.testUUID,
-          timestamp = item.timestamp.withDurationAdded(durationOffset / 2, i)
-        )
-      })
+    val records = genSequentialRecords(number)
 
-    val batch = recordList.foldLeft(Batch.unlogged) {
-      (b, record) => {
-        b.add(TestDatabase.timeSeriesTable.insert
-          .value(_.id, record.id)
-          .value(_.name, record.name)
-          .value(_.timestamp, record.timestamp))
-      }
-    }
     val chain = for {
       truncate <- TestDatabase.timeSeriesTable.truncate.future()
-      insert <- batch.future()
+      insert <- addRecordsToBatch(records).future()
       chunks <- {
-        TestDatabase.timeSeriesTable.select.where(_.id eqs TestDatabase.timeSeriesTable.testUUID)
+        TestDatabase.timeSeriesTable.select
+          .where(_.id eqs TestDatabase.timeSeriesTable.testUUID)
           .orderBy(_.timestamp.asc)
-          .limit(number)
+          .limit(limit)
           .fetch()
       }
     } yield chunks
 
-    chain.successful {
-      res =>
-        val ts = recordList.map(_.timestamp.getSecondOfDay)
-
-        res.map(_.timestamp.getSecondOfDay) shouldEqual ts
-    }
+    verifyResults(chain, records.take(limit))
   }
 
-  it should "allow fetching the records in ascending order for a descending clustering order using order by clause with Twitter Futures" in {
-    var i = 0
-    val number = 5
+  it should "fetch records in ascending order for a descending clustering order using order by clause with Twitter Futures" in {
+    val number = 10
+    val limit = 5
 
-    val recordList = genList[TimeSeriesRecord](number).map(
-      item => {
-        i += 1
-        item.copy(
-          id = TestDatabase.timeSeriesTable.testUUID,
-          timestamp = item.timestamp.withDurationAdded(durationOffset, i)
-        )
-      })
+    val records = genSequentialRecords(number)
 
-    val batch = recordList.foldLeft(Batch.unlogged) {
-      (b, record) => {
-        b.add(TestDatabase.timeSeriesTable.insert
-          .value(_.id, record.id)
-          .value(_.name, record.name)
-          .value(_.timestamp, record.timestamp))
-      }
-    }
     val chain = for {
       truncate <- TestDatabase.timeSeriesTable.truncate.execute()
-      insert <- batch.execute()
-      chunks <- TestDatabase.timeSeriesTable
-        .select
+      insert <- addRecordsToBatch(records).execute()
+      chunks <- TestDatabase.timeSeriesTable.select
         .where(_.id eqs TestDatabase.timeSeriesTable.testUUID)
-        .orderBy(_.timestamp.asc).limit(number)
+        .orderBy(_.timestamp.asc)
+        .limit(limit)
         .collect()
     } yield chunks
 
-    chain.successful {
-      res =>
-        val ts = recordList.map(_.timestamp.getSecondOfDay)
-        res.map(_.timestamp.getSecondOfDay) shouldEqual ts
-    }
+    verifyResults(chain, records.take(limit))
   }
 
-  it should "allow fetching the records in descending order for a descending clustering order using order by clause" in {
-    var i = 0
-    val number = 5
-
-    val recordList = genList[TimeSeriesRecord](number).map(
-      item => {
-        i += 1
-        item.copy(
-          id = TestDatabase.timeSeriesTable.testUUID,
-          timestamp = item.timestamp.withDurationAdded(durationOffset / 2, i)
-        )
-      })
-
-    val batch = recordList.foldLeft(Batch.unlogged) {
-      (b, record) =>
-        b.add(TestDatabase.timeSeriesTable.insert
-          .value(_.id, record.id)
-          .value(_.name, record.name)
-          .value(_.timestamp, record.timestamp))
-    }
+  it should "fetch records in ascending order for a descending clustering order using prepared statements" in {
+    val number = 10
+    val limit = 5
+
+    val records = genSequentialRecords(number)
+
+    val query = TestDatabase.timeSeriesTable.select
+      .p_where(_.id eqs ?)
+      .orderBy(_.timestamp.asc)
+      .limit(limit)
+      .prepare()
+
+    val chain = for {
+      truncate <- TestDatabase.timeSeriesTable.truncate.future()
+      insert <- addRecordsToBatch(records).future()
+      chunks <- query.bind(TestDatabase.timeSeriesTable.testUUID).fetch()
+    } yield chunks
+
+    verifyResults(chain, records.take(limit))
+  }
+
+  it should "fetch records in descending order for a descending clustering order using order by clause" in {
+    val number = 10
+    val limit = 5
+
+    val records = genSequentialRecords(number)
+
     val chain = for {
       truncate <- TestDatabase.timeSeriesTable.truncate.future()
-      insert <- batch.future()
-      chunks <- TestDatabase.timeSeriesTable
-        .select
+      insert <- addRecordsToBatch(records).future()
+      chunks <- TestDatabase.timeSeriesTable.select
         .where(_.id eqs TestDatabase.timeSeriesTable.testUUID)
         .orderBy(_.timestamp.descending)
-        .limit(number)
+        .limit(limit)
         .fetch()
     } yield chunks
 
-    chain.successful {
-      res =>
-        val ts = recordList.map(_.timestamp.getSecondOfDay)
-        res.map(_.timestamp.getSecondOfDay) shouldEqual ts.reverse
-    }
+    verifyResults(chain, records.reverse.take(limit))
   }
 
-  it should "allow fetching the records in descending order for a descending clustering order using order by clause with Twitter Futures" in {
-    var i = 0
-    val number = 5
-
-    val recordList = genList[TimeSeriesRecord](number).map(
-      item => {
-        i += 1
-        item.copy(
-          id = TestDatabase.timeSeriesTable.testUUID,
-          timestamp = item.timestamp.withDurationAdded(durationOffset, i)
-        )
-      })
-
-    val batch = recordList.foldLeft(Batch.unlogged) {
-      (b, record) =>
-        b.add(TestDatabase.timeSeriesTable.insert
-          .value(_.id, record.id)
-          .value(_.name, record.name)
-          .value(_.timestamp, record.timestamp))
-    }
+  it should "fetch records in descending order for a descending clustering order using order by clause with Twitter Futures" in {
+    val number = 10
+    val limit = 5
+
+    val records = genSequentialRecords(number)
+
     val chain = for {
       truncate <- TestDatabase.timeSeriesTable.truncate.execute()
-      insert <- batch.execute()
+      insert <- addRecordsToBatch(records).execute()
       chunks <- TestDatabase.timeSeriesTable.select
         .where(_.id eqs TestDatabase.timeSeriesTable.testUUID)
-        .orderBy(_.timestamp.desc).limit(number)
+        .orderBy(_.timestamp.desc)
+        .limit(limit)
         .collect()
     } yield chunks
 
-    chain.successful {
-      res =>
-        val ts = recordList.map(_.timestamp.getSecondOfDay)
-        res.map(_.timestamp.getSecondOfDay) shouldEqual ts.reverse
+    verifyResults(chain, records.reverse.take(limit))
+  }
+
+  def verifyResults(futureResults: ScalaFuture[Seq[TimeSeriesRecord]], expected: Seq[TimeSeriesRecord]): Unit = {
+    futureResults.successful { results =>
+      results shouldEqual expected
     }
   }
 
+  def verifyResults(futureResults: TwitterFuture[Seq[TimeSeriesRecord]], expected: Seq[TimeSeriesRecord]): Unit = {
+    futureResults.successful { results =>
+      results shouldEqual expected
+    }
+  }
+}
+
+object TimeSeriesTest {
+  def genSequentialRecords(number: Int): Seq[TimeSeriesRecord] = {
+    val durationOffset = 1000
+
+    (1 to number).map { i =>
+      val record = gen[TimeSeriesRecord]
+      record.copy(
+        id = TestDatabase.timeSeriesTable.testUUID,
+        timestamp = record.timestamp.withDurationAdded(durationOffset, i))
+    }
+  }
+
+  def addRecordsToBatch(
+      records: Seq[TimeSeriesRecord])(
+      implicit space: KeySpace,
+      session: Session): BatchQuery[Unspecified] = {
+
+    records.foldLeft(Batch.unlogged) {
+      (b, record) => {
+        b.add(TestDatabase.timeSeriesTable.insert
+          .value(_.id, record.id)
+          .value(_.name, record.name)
+          .value(_.timestamp, record.timestamp))
+      }
+    }
+  }
 }
diff --git a/phantom-dsl/src/test/scala/com/websudos/phantom/db/TestDatabase.scala b/phantom-dsl/src/test/scala/com/websudos/phantom/db/TestDatabase.scala
index 50275a407..2d5dd152d 100644
--- a/phantom-dsl/src/test/scala/com/websudos/phantom/db/TestDatabase.scala
+++ b/phantom-dsl/src/test/scala/com/websudos/phantom/db/TestDatabase.scala
@@ -30,7 +30,7 @@
 package com.websudos.phantom.db
 
 import com.websudos.phantom.connectors.ContactPoint
-import com.websudos.phantom.tables.{Recipes, JsonTable, EnumTable, BasicTable}
+import com.websudos.phantom.tables._
 
 
 private[this] object DefaultKeyspace {
diff --git a/phantom-jdk8/src/main/scala/com/websudos/phantom/jdk8/DefaultJava8Primitives.scala b/phantom-jdk8/src/main/scala/com/websudos/phantom/jdk8/DefaultJava8Primitives.scala
new file mode 100644
index 000000000..791aca59c
--- /dev/null
+++ b/phantom-jdk8/src/main/scala/com/websudos/phantom/jdk8/DefaultJava8Primitives.scala
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8
+
+import java.time._
+import java.util.Date
+
+import com.datastax.driver.core.Row
+import com.websudos.phantom.builder.primitives.Primitive
+import com.websudos.phantom.builder.query.CQLQuery
+import com.websudos.phantom.builder.syntax.CQLSyntax
+import com.websudos.phantom.jdk8.dsl.{JdkLocalDate, OffsetDateTime, ZonedDateTime}
+
+import scala.util.Try
+
+trait DefaultJava8Primitives {
+
+  implicit object OffsetDateTimeIsPrimitive extends Primitive[OffsetDateTime] {
+
+    override type PrimitiveType = java.util.Date
+
+    val cassandraType = CQLSyntax.Types.Timestamp
+
+    override def asCql(value: OffsetDateTime): String = {
+      value.toInstant.toEpochMilli.toString
+    }
+
+    override def fromRow(column: String, row: Row): Try[OffsetDateTime] = nullCheck(column, row) {
+      r => OffsetDateTime.ofInstant(r.getTimestamp(column).toInstant, ZoneOffset.UTC)
+    }
+
+    override def fromString(value: String): OffsetDateTime = OffsetDateTime.parse(value)
+
+    override def clz: Class[Date] = classOf[Date]
+  }
+
+  implicit object ZonedDateTimeIsPrimitive extends Primitive[ZonedDateTime] {
+
+    override type PrimitiveType = java.util.Date
+
+    val cassandraType = CQLSyntax.Types.Timestamp
+
+    override def asCql(value: ZonedDateTime): String = {
+      value.toInstant.toEpochMilli.toString
+    }
+
+    override def fromRow(column: String, row: Row): Try[ZonedDateTime] = nullCheck(column, row) {
+      r => ZonedDateTime.ofInstant(r.getTimestamp(column).toInstant, ZoneOffset.UTC)
+    }
+
+    override def fromString(value: String): ZonedDateTime = ZonedDateTime.parse(value)
+
+    override def clz: Class[Date] = classOf[Date]
+  }
+
+  implicit object JdkLocalDateIsPrimitive extends Primitive[JdkLocalDate] {
+
+    override type PrimitiveType = com.datastax.driver.core.LocalDate
+
+    val cassandraType = CQLSyntax.Types.Date
+
+    override def asCql(value: JdkLocalDate): String = {
+      CQLQuery.empty.singleQuote(value.toString)
+    }
+
+    override def fromRow(column: String, row: Row): Try[JdkLocalDate] = nullCheck(column, row) {
+      r => LocalDate.ofEpochDay(r.getDate(column).getDaysSinceEpoch)
+    }
+
+    override def fromString(value: String): JdkLocalDate = {
+      Instant.ofEpochMilli(value.toLong).atOffset(ZoneOffset.UTC).toLocalDate
+    }
+
+    override def clz: Class[com.datastax.driver.core.LocalDate] = classOf[com.datastax.driver.core.LocalDate]
+  }
+
+}
diff --git a/phantom-jdk8/src/main/scala/com/websudos/phantom/jdk8/dsl/package.scala b/phantom-jdk8/src/main/scala/com/websudos/phantom/jdk8/dsl/package.scala
new file mode 100644
index 000000000..25af8de8b
--- /dev/null
+++ b/phantom-jdk8/src/main/scala/com/websudos/phantom/jdk8/dsl/package.scala
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8
+
+import com.websudos.phantom.dsl.CassandraTable
+
+package object dsl extends DefaultJava8Primitives {
+
+  type OffsetDateTimeColumn[Owner <: CassandraTable[Owner, Record], Record] = com.websudos.phantom.column.PrimitiveColumn[Owner, Record, OffsetDateTime]
+  type ZonedDateTimeColumn[Owner <: CassandraTable[Owner, Record], Record] = com.websudos.phantom.column.PrimitiveColumn[Owner, Record, ZonedDateTime]
+  type JdkLocalDateColumn[Owner <: CassandraTable[Owner, Record], Record] = com.websudos.phantom.column.PrimitiveColumn[Owner, Record, JdkLocalDate]
+
+  type OptionalOffsetDateTimeColumn[Owner <: CassandraTable[Owner, Record], Record] = com.websudos.phantom.column.OptionalPrimitiveColumn[Owner, Record, OffsetDateTime]
+  type OptionalZonedDateTimeColumn[Owner <: CassandraTable[Owner, Record], Record] = com.websudos.phantom.column.OptionalPrimitiveColumn[Owner, Record, ZonedDateTime]
+  type OptionalJdkLocalDateColumn[Owner <: CassandraTable[Owner, Record], Record] = com.websudos.phantom.column.OptionalPrimitiveColumn[Owner, Record, JdkLocalDate]
+
+  type OffsetDateTime = java.time.OffsetDateTime
+  type ZonedDateTime = java.time.ZonedDateTime
+  type JdkLocalDate = java.time.LocalDate
+
+}
diff --git a/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/Jdk8TimeColumnsTest.scala b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/Jdk8TimeColumnsTest.scala
new file mode 100644
index 000000000..83ecef912
--- /dev/null
+++ b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/Jdk8TimeColumnsTest.scala
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8
+
+import com.websudos.phantom.PhantomSuite
+import com.websudos.phantom.dsl._
+import com.websudos.phantom.jdk8.tables.{Jdk8Row, TestDatabase, _}
+import com.websudos.util.testing._
+
+class Jdk8TimeColumnsTest extends PhantomSuite {
+
+  override def beforeAll(): Unit = {
+    super.beforeAll()
+    if (session.v4orNewer) {
+      TestDatabase.primitivesJdk8.insertSchema()
+      TestDatabase.optionalPrimitivesJdk8.insertSchema()
+    }
+  }
+
+  if (session.v4orNewer) {
+    it should "correctly insert and extract java.time columns" in {
+      val row = gen[Jdk8Row]
+
+      val chain = for {
+        store <- TestDatabase.primitivesJdk8.store(row).future()
+        select <- TestDatabase.primitivesJdk8.select.where(_.pkey eqs row.pkey).one()
+      } yield select
+
+      chain successful {
+        res => res.value shouldEqual row
+      }
+    }
+
+    it should "correctly insert and extract optional java.time columns" in {
+      val row = gen[OptionalJdk8Row]
+
+      val chain = for {
+        store <- TestDatabase.optionalPrimitivesJdk8.store(row).future()
+        select <- TestDatabase.optionalPrimitivesJdk8.select.where(_.pkey eqs row.pkey).one()
+      } yield select
+
+      chain successful {
+        res => res.value shouldEqual row
+      }
+    }
+  }
+}
diff --git a/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/ConcretePrimitivesJdk8.scala b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/ConcretePrimitivesJdk8.scala
new file mode 100644
index 000000000..daee6821f
--- /dev/null
+++ b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/ConcretePrimitivesJdk8.scala
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8.tables
+
+import java.time.{LocalDate, OffsetDateTime}
+
+import com.websudos.phantom.CassandraTable
+import com.websudos.phantom.builder.query.InsertQuery
+import com.websudos.phantom.dsl._
+import com.websudos.phantom.jdk8.dsl._
+
+case class Jdk8Row(
+  pkey: String,
+  offsetDateTime: OffsetDateTime,
+  zonedDateTime: ZonedDateTime,
+  localDate: LocalDate
+)
+
+sealed class PrimitivesJdk8 extends CassandraTable[ConcretePrimitivesJdk8, Jdk8Row] {
+
+  object pkey extends StringColumn(this) with PartitionKey[String]
+
+  object offsetDateTime extends OffsetDateTimeColumn(this)
+
+  object zonedDateTime extends ZonedDateTimeColumn(this)
+
+  object localDate extends JdkLocalDateColumn(this)
+
+  override def fromRow(r: Row): Jdk8Row = {
+    Jdk8Row(
+      pkey = pkey(r),
+      offsetDateTime = offsetDateTime(r),
+      zonedDateTime = zonedDateTime(r),
+      localDate = localDate(r)
+    )
+  }
+}
+
+abstract class ConcretePrimitivesJdk8 extends PrimitivesJdk8 with RootConnector {
+
+  def store(primitive: Jdk8Row): InsertQuery.Default[ConcretePrimitivesJdk8, Jdk8Row] = {
+    insert.value(_.pkey, primitive.pkey)
+      .value(_.offsetDateTime, primitive.offsetDateTime)
+      .value(_.zonedDateTime, primitive.zonedDateTime)
+      .value(_.localDate, primitive.localDate)
+  }
+
+  override val tableName = "PrimitivesJdk8"
+
+}
\ No newline at end of file
diff --git a/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/OptionalPrimitivesJdk8.scala b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/OptionalPrimitivesJdk8.scala
new file mode 100644
index 000000000..b8a0c76df
--- /dev/null
+++ b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/OptionalPrimitivesJdk8.scala
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8.tables
+
+import java.time.{LocalDate, OffsetDateTime}
+
+import com.websudos.phantom.CassandraTable
+import com.websudos.phantom.builder.query.InsertQuery
+import com.websudos.phantom.dsl._
+import com.websudos.phantom.jdk8.dsl._
+
+case class OptionalJdk8Row(
+  pkey: String,
+  offsetDateTime: Option[OffsetDateTime],
+  zonedDateTime: Option[ZonedDateTime],
+  localDate: Option[LocalDate]
+)
+
+sealed class OptionalPrimitivesJdk8 extends CassandraTable[ConcreteOptionalPrimitivesJdk8, OptionalJdk8Row] {
+
+  object pkey extends StringColumn(this) with PartitionKey[String]
+
+  object offsetDateTime extends OptionalOffsetDateTimeColumn(this)
+
+  object zonedDateTime extends OptionalZonedDateTimeColumn(this)
+
+  object localDate extends OptionalJdkLocalDateColumn(this)
+
+  override def fromRow(r: Row): OptionalJdk8Row = {
+    OptionalJdk8Row(
+      pkey = pkey(r),
+      offsetDateTime = offsetDateTime(r),
+      zonedDateTime = zonedDateTime(r),
+      localDate = localDate(r)
+    )
+  }
+}
+
+abstract class ConcreteOptionalPrimitivesJdk8 extends OptionalPrimitivesJdk8 with RootConnector {
+
+  def store(primitive: OptionalJdk8Row): InsertQuery.Default[ConcreteOptionalPrimitivesJdk8, OptionalJdk8Row] = {
+    insert.value(_.pkey, primitive.pkey)
+      .value(_.offsetDateTime, primitive.offsetDateTime)
+      .value(_.zonedDateTime, primitive.zonedDateTime)
+      .value(_.localDate, primitive.localDate)
+  }
+
+  override val tableName = "OptionalPrimitivesJdk8"
+
+}
\ No newline at end of file
diff --git a/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/TestDatabase.scala b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/TestDatabase.scala
new file mode 100644
index 000000000..a34a2d427
--- /dev/null
+++ b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/TestDatabase.scala
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8.tables
+
+import com.websudos.phantom.connectors.{ContactPoint, KeySpaceDef}
+import com.websudos.phantom.db.DatabaseImpl
+
+
+class TestDatabase(override val connector: KeySpaceDef) extends DatabaseImpl(connector) {
+
+  object primitivesJdk8 extends ConcretePrimitivesJdk8 with connector.Connector
+
+  object optionalPrimitivesJdk8 extends ConcreteOptionalPrimitivesJdk8 with connector.Connector
+
+}
+
+object TestDatabase extends TestDatabase(ContactPoint.local.keySpace("phantom"))
\ No newline at end of file
diff --git a/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/package.scala b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/package.scala
new file mode 100644
index 000000000..e8e850d7f
--- /dev/null
+++ b/phantom-jdk8/src/test/scala/com/websudos/phantom/jdk8/tables/package.scala
@@ -0,0 +1,60 @@
+/*
+ * Copyright 2013-2015 Websudos, Limited.
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright notice,
+ * this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * - Explicit consent must be obtained from the copyright owner, Websudos Limited before any redistribution is made.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.websudos.phantom.jdk8
+
+import java.time.{LocalDate, OffsetDateTime, ZoneOffset, ZonedDateTime}
+
+import com.websudos.util.testing.{Sample, _}
+
+package object tables {
+
+  implicit object Jdk8RowSampler extends Sample[Jdk8Row] {
+    def sample: Jdk8Row = {
+      Jdk8Row(
+        gen[String],
+        OffsetDateTime.now(ZoneOffset.UTC).plusSeconds(gen[Long]),
+        ZonedDateTime.now(ZoneOffset.UTC).plusSeconds(gen[Long]),
+        LocalDate.now().plusDays(gen[Long])
+      )
+    }
+  }
+
+  implicit object OptionalJdk8RowSampler extends Sample[OptionalJdk8Row] {
+    def sample: OptionalJdk8Row = {
+      OptionalJdk8Row(
+        gen[String],
+        Some(OffsetDateTime.now(ZoneOffset.UTC).plusSeconds(gen[Long])),
+        Some(ZonedDateTime.now(ZoneOffset.UTC).plusSeconds(gen[Long])),
+        Some(LocalDate.now().plusDays(gen[Long]))
+      )
+    }
+  }
+
+}
diff --git a/project/Build.scala b/project/Build.scala
index 50fff0265..be702ed52 100644
--- a/project/Build.scala
+++ b/project/Build.scala
@@ -152,6 +152,24 @@ object Build extends Build {
     parallelExecution in ThisBuild := false
   ) ++ net.virtualvoid.sbt.graph.Plugin.graphSettings ++ VersionManagement.newSettings ++ PublishTasks.bintrayPublishSettings
 
+
+  private[this] def isJdk8: Boolean = sys.props("java.specification.version") == "1.8"
+
+  private[this] def addOnCondition(condition: Boolean, projectReference: ProjectReference): Seq[ProjectReference] =
+    if (condition) projectReference :: Nil else Nil
+
+  lazy val baseProjectList: Seq[ProjectReference] = Seq(
+    phantomDsl,
+    phantomExample,
+    phantomConnectors,
+    phantomReactiveStreams,
+    phantomThrift,
+    phantomUdt,
+    phantomZookeeper
+  )
+
+  lazy val fullProjectList = baseProjectList ++ addOnCondition(isJdk8, phantomJdk8)
+
   lazy val phantom = Project(
     id = "phantom",
     base = file("."),
@@ -163,13 +181,7 @@ object Build extends Build {
   ).settings(
     name := "phantom"
   ).aggregate(
-    phantomDsl,
-    phantomExample,
-    phantomConnectors,
-    phantomReactiveStreams,
-    phantomThrift,
-    phantomUdt,
-    phantomZookeeper
+    fullProjectList: _*
   )
 
   lazy val phantomDsl = Project(
@@ -209,6 +221,21 @@ object Build extends Build {
       phantomConnectors
     )
 
+  lazy val phantomJdk8 = Project(
+    id = "phantom-jdk8",
+    base = file("phantom-jdk8"),
+    settings = Defaults.coreDefaultSettings ++ sharedSettings
+  ).settings(
+    name := "phantom-jdk8",
+    testOptions in Test += Tests.Argument("-oF"),
+    logBuffered in Test := false,
+    concurrentRestrictions in Test := Seq(
+      Tags.limit(Tags.ForkedTestGroup, 4)
+    )
+  ).dependsOn(
+    phantomDsl % "compile->compile;test->test"
+  )
+
   lazy val phantomConnectors = Project(
     id = "phantom-connectors",
     base = file("phantom-connectors"),