diff --git a/dynamodb/src/it/scala/zio/dynamodb/LiveSpec.scala b/dynamodb/src/it/scala/zio/dynamodb/LiveSpec.scala
index e89c386fc..4c710629a 100644
--- a/dynamodb/src/it/scala/zio/dynamodb/LiveSpec.scala
+++ b/dynamodb/src/it/scala/zio/dynamodb/LiveSpec.scala
@@ -7,10 +7,6 @@ import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider
 import software.amazon.awssdk.services.dynamodb.model.{ DynamoDbException, IdempotentParameterMismatchException }
 import zio.dynamodb.UpdateExpression.Action.SetAction
 import zio.dynamodb.UpdateExpression.SetOperand
-import zio.dynamodb.PartitionKeyExpression.PartitionKey
-import zio.dynamodb.KeyConditionExpression.partitionKey
-import zio.dynamodb.KeyConditionExpression.sortKey
-import zio.dynamodb.SortKeyExpression.SortKey
 import zio.aws.{ dynamodb, netty }
 import zio._
 import zio.dynamodb.DynamoDBQuery._
@@ -65,7 +61,12 @@ object LiveSpec extends ZIOSpecDefault {
 
   private val stringSortKeyItem = Item(id -> adam, name -> adam)
 
-  private final case class Person(id: String, firstName: String, num: Int)
+  final case class Person(id: String, firstName: String, num: Int)
+  object Person {
+    implicit lazy val schema: Schema.CaseClass3[String, String, Int, Person] = DeriveSchema.gen[Person]
+
+    val (id, firstName, num) = ProjectionExpression.accessors[Person]
+  }
   private implicit lazy val person: Schema[Person] = DeriveSchema.gen[Person]
 
   private val aviPerson  = Person(first, avi, 1)
@@ -118,7 +119,7 @@ object LiveSpec extends ZIOSpecDefault {
       AttributeDefinition.attrDefnString(name)
     )
 
-  def sortKeyStringTableWithKeywords(tableName: String) =
+  private def sortKeyStringTableWithKeywords(tableName: String) =
     createTable(tableName, KeySchema("and", "source"), BillingMode.PayPerRequest)(
       AttributeDefinition.attrDefnString("and"),
       AttributeDefinition.attrDefnString("source")
@@ -133,6 +134,17 @@ object LiveSpec extends ZIOSpecDefault {
         } yield TableName(tableName)
       )(tName => deleteTable(tName.value).execute.orDie)
 
+  private def withPkKeywordsTable(
+    f: String => ZIO[DynamoDBExecutor, Throwable, TestResult]
+  ) =
+    ZIO.scoped {
+      managedTable(sortKeyStringTableWithKeywords).flatMap { table =>
+        for {
+          result <- f(table.value)
+        } yield result
+      }
+    }
+
   // TODO: Avi - fix problem with inference of this function when splitting suites
   // "a type was inferred to be `Any`; this may indicate a programming error."
   private def withTemporaryTable[R](
@@ -158,17 +170,6 @@ object LiveSpec extends ZIOSpecDefault {
       }
     }
 
-  def withPkKeywordsTable(
-    f: String => ZIO[DynamoDBExecutor, Throwable, TestResult]
-  ) =
-    ZIO.scoped {
-      managedTable(sortKeyStringTableWithKeywords).flatMap { table =>
-        for {
-          result <- f(table.value)
-        } yield result
-      }
-    }
-
   def withDefaultAndNumberTables(
     f: (String, String) => ZIO[DynamoDBExecutor, Throwable, TestResult]
   ) =
@@ -213,7 +214,7 @@ object LiveSpec extends ZIOSpecDefault {
             val query = DynamoDBQuery
               .queryAll[ExpressionAttrNamesPkKeywords](tableName)
               .whereKey(
-                ExpressionAttrNamesPkKeywords.and === "and1" && ExpressionAttrNamesPkKeywords.source === "source1"
+                ExpressionAttrNamesPkKeywords.and.partitionKey === "and1" && ExpressionAttrNamesPkKeywords.source.sortKey === "source1"
               )
               .filter(ExpressionAttrNamesPkKeywords.ttl.notExists)
             query.execute.flatMap(_.runDrain).exit.map { result =>
@@ -225,7 +226,7 @@ object LiveSpec extends ZIOSpecDefault {
           withPkKeywordsTable { tableName =>
             val query = DynamoDBQuery
               .queryAll[ExpressionAttrNamesPkKeywords](tableName)
-              .whereKey(partitionKey("and") === "and1" && sortKey("source") === "source1")
+              .whereKey($("and").partitionKey === "and1" && $("source").sortKey === "source1")
               .filter(ExpressionAttrNamesPkKeywords.ttl.notExists)
             query.execute.flatMap(_.runDrain).exit.map { result =>
               assert(result)(succeeds(isUnit))
@@ -237,7 +238,7 @@ object LiveSpec extends ZIOSpecDefault {
             val query = DynamoDBQuery
               .querySome[ExpressionAttrNamesPkKeywords](tableName, 1)
               .whereKey(
-                ExpressionAttrNamesPkKeywords.and === "and1" && ExpressionAttrNamesPkKeywords.source === "source1"
+                ExpressionAttrNamesPkKeywords.and.partitionKey === "and1" && ExpressionAttrNamesPkKeywords.source.sortKey === "source1"
               )
               .filter(ExpressionAttrNamesPkKeywords.ttl.notExists)
             for {
@@ -249,7 +250,7 @@ object LiveSpec extends ZIOSpecDefault {
           withPkKeywordsTable { tableName =>
             val query = DynamoDBQuery
               .querySome[ExpressionAttrNames](tableName, 1)
-              .whereKey(partitionKey("and") === "and1" && sortKey("source") === "source1")
+              .whereKey($("and").partitionKey === "and1" && $("source").sortKey === "source1")
               .filter(ExpressionAttrNames.ttl.notExists)
             for {
               result <- query.execute
@@ -273,7 +274,7 @@ object LiveSpec extends ZIOSpecDefault {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
                 .queryAll[ExpressionAttrNames](tableName)
-                .whereKey(ExpressionAttrNames.id === "id")
+                .whereKey(ExpressionAttrNames.id.partitionKey === "id")
                 .filter(ExpressionAttrNames.ttl.notExists)
               query.execute.flatMap(_.runDrain).exit.map { result =>
                 assert(result)(succeeds(isUnit))
@@ -297,7 +298,7 @@ object LiveSpec extends ZIOSpecDefault {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
                 .querySome[ExpressionAttrNames](tableName, 1)
-                .whereKey(PartitionKey(id) === second && SortKey(number) > 0)
+                .whereKey($(id).partitionKey === second && $(number).sortKey > 0)
                 .filter(ExpressionAttrNames.ttl.notExists)
 
               for {
@@ -312,7 +313,9 @@ object LiveSpec extends ZIOSpecDefault {
           test("delete should handle keyword") {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
-                .delete[ExpressionAttrNames](tableName, PrimaryKey("id" -> "id", "num" -> 1))
+                .delete(tableName)(
+                  ExpressionAttrNames.id.partitionKey === "id" && ExpressionAttrNames.num.sortKey === 1
+                )
                 .where(ExpressionAttrNames.ttl.notExists)
               query.execute.exit.map { result =>
                 assert(result)(succeeds(isNone))
@@ -332,7 +335,9 @@ object LiveSpec extends ZIOSpecDefault {
           test("update should handle keyword") {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
-                .update[ExpressionAttrNames](tableName, PrimaryKey("id" -> "1", "num" -> 1))(
+                .update(tableName)(
+                  ExpressionAttrNames.id.partitionKey === "id" && ExpressionAttrNames.num.sortKey === 1
+                )(
                   ExpressionAttrNames.ttl.set(Some(42L))
                 )
                 .where(ExpressionAttrNames.ttl.notExists)
@@ -438,7 +443,7 @@ object LiveSpec extends ZIOSpecDefault {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
                 .queryAllItem(tableName)
-                .whereKey($("id") === "id")
+                .whereKey($("id").partitionKey === "id")
                 .filter($("ttl").notExists)
               query.execute.flatMap(_.runDrain).exit.map { result =>
                 assert(result)(succeeds(isUnit))
@@ -449,7 +454,7 @@ object LiveSpec extends ZIOSpecDefault {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
                 .querySomeItem(tableName, 1)
-                .whereKey($("id") === "id")
+                .whereKey($("id").partitionKey === "id")
                 .filter($("ttl").notExists)
               query.execute.exit.map { result =>
                 assert(result.isSuccess)(isTrue)
@@ -460,7 +465,7 @@ object LiveSpec extends ZIOSpecDefault {
             withDefaultTable { tableName =>
               val query = DynamoDBQuery
                 .querySomeItem(tableName, 1, $("ttl"))
-                .whereKey($("id") === "id")
+                .whereKey($("id").partitionKey === "id")
               query.execute.exit.map { result =>
                 assert(result.isSuccess)(isTrue)
               }
@@ -553,7 +558,7 @@ object LiveSpec extends ZIOSpecDefault {
         },
         test("get into case class") {
           withDefaultTable { tableName =>
-            get[Person](tableName, secondPrimaryKey).execute.map(person =>
+            get(tableName)(Person.id.partitionKey === second && Person.num.sortKey === 2).execute.map(person =>
               assert(person)(equalTo(Right(Person("second", "adam", 2))))
             )
           }
@@ -683,7 +688,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             val query =
               queryAllItem(tableName, $(name), $("ttl")).whereKey(
-                PartitionKey(id) === first && SortKey(number) > 0
+                $(id).partitionKey === first && $(number).sortKey > 0
               )
 
             query.execute.flatMap(_.runDrain).map { _ =>
@@ -695,7 +700,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 10, $(name), $("ttl"))
-                         .whereKey(PartitionKey(id) === first && SortKey(number) > 0)
+                         .whereKey($(id).partitionKey === first && $(number).sortKey > 0)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(
@@ -707,7 +712,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 10, $(name))
-                         .whereKey(PartitionKey(id) === first && SortKey(number) < 2)
+                         .whereKey($(id).partitionKey === first && $(number).sortKey < 2)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(
@@ -719,7 +724,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 10, $(name))
-                         .whereKey(PartitionKey(id) === first && SortKey(number) > 0)
+                         .whereKey($(id).partitionKey === first && $(number).sortKey > 0)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(
@@ -732,7 +737,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 10, $(name))
-                         .whereKey(PartitionKey(id) === first && SortKey(number) >= 4)
+                         .whereKey($(id).partitionKey === first && $(number).sortKey >= 4)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(
@@ -744,7 +749,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 10, $(name))
-                         .whereKey(PartitionKey(id) === first && SortKey(number) <= 4)
+                         .whereKey($(id).partitionKey === first && $(number).sortKey <= 4)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(
@@ -756,7 +761,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 10, $(name))
-                         .whereKey(PartitionKey(id) === "nowhere" && SortKey(number) > 0)
+                         .whereKey($(id).partitionKey === "nowhere" && $(number).sortKey > 0)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(isEmpty)
@@ -766,7 +771,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 1, $(name))
-                         .whereKey(PartitionKey(id) === first)
+                         .whereKey($(id).partitionKey === first)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(equalTo(Chunk(Item(name -> avi))))
@@ -776,7 +781,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 3, $(name))
-                         .whereKey(PartitionKey(id) === first)
+                         .whereKey($(id).partitionKey === first)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(equalTo(Chunk(Item(name -> avi), Item(name -> avi2), Item(name -> avi3))))
@@ -786,7 +791,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               chunk <- querySomeItem(tableName, 4, $(name))
-                         .whereKey(PartitionKey(id) === first)
+                         .whereKey($(id).partitionKey === first)
                          .execute
                          .map(_._1)
             } yield assert(chunk)(equalTo(Chunk(Item(name -> avi), Item(name -> avi2), Item(name -> avi3))))
@@ -796,11 +801,11 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               startKey <- querySomeItem(tableName, 2, $(id), $(number))
-                            .whereKey(PartitionKey(id) === first)
+                            .whereKey($(id).partitionKey === first)
                             .execute
                             .map(_._2)
               chunk    <- querySomeItem(tableName, 5, $(name))
-                            .whereKey(PartitionKey(id) === first)
+                            .whereKey($(id).partitionKey === first)
                             .startKey(startKey)
                             .execute
                             .map(_._1)
@@ -811,7 +816,7 @@ object LiveSpec extends ZIOSpecDefault {
           withDefaultTable { tableName =>
             for {
               stream <- queryAllItem(tableName)
-                          .whereKey(PartitionKey(id) === second)
+                          .whereKey($(id).partitionKey === second)
                           .execute
               chunk  <- stream.run(ZSink.collectAll[Item])
             } yield assert(chunk)(
@@ -834,7 +839,7 @@ object LiveSpec extends ZIOSpecDefault {
             withDefaultTable { tableName =>
               for {
                 chunk <- querySomeItem(tableName, 10, $(name))
-                           .whereKey(PartitionKey(id) === first && SortKey(number).between(3, 8))
+                           .whereKey($(id).partitionKey === first && $(number).sortKey.between(3, 8))
                            .execute
                            .map(_._1)
               } yield assert(chunk)(
@@ -849,7 +854,7 @@ object LiveSpec extends ZIOSpecDefault {
                 for {
                   _     <- putItem(tableName, stringSortKeyItem).execute
                   chunk <- querySomeItem(tableName, 10)
-                             .whereKey(PartitionKey(id) === adam && SortKey(name).beginsWith("ad"))
+                             .whereKey($(id).partitionKey === adam && $(name).sortKey.beginsWith("ad"))
                              .execute
                              .map(_._1)
                 } yield assert(chunk)(equalTo(Chunk(stringSortKeyItem)))
@@ -1311,9 +1316,9 @@ object LiveSpec extends ZIOSpecDefault {
           },
           test("delete item handles keyword") {
             withDefaultTable { tableName =>
-              val d = delete[ExpressionAttrNames](
-                tableName = tableName,
-                key = pk(avi3Item)
+              val d = delete(tableName = tableName)(
+                primaryKeyExpr =
+                  ExpressionAttrNames.id.partitionKey === "first" && ExpressionAttrNames.num.sortKey === 7
               ).where(ExpressionAttrNames.ttl.notExists)
               d.transaction.execute.exit.map { result =>
                 assert(result.isSuccess)(isTrue)
@@ -1334,9 +1339,9 @@ object LiveSpec extends ZIOSpecDefault {
           },
           test("transact update item should handle keyword") {
             withDefaultTable { tableName =>
-              val u = update[ExpressionAttrNames](
-                tableName = tableName,
-                key = pk(avi3Item)
+              val u = update(tableName = tableName)(
+                primaryKeyExpr =
+                  ExpressionAttrNames.id.partitionKey === "first" && ExpressionAttrNames.num.sortKey === 7
               )(ExpressionAttrNames.ttl.set(None)).where(ExpressionAttrNames.ttl.notExists)
               u.transaction.execute.exit.map { result =>
                 assert(result.isSuccess)(isTrue)
@@ -1345,14 +1350,15 @@ object LiveSpec extends ZIOSpecDefault {
           },
           test("update item") {
             withDefaultTable { tableName =>
+              val key        = Person.id.partitionKey === first && Person.num.sortKey === 7
               val updateItem = UpdateItem(
-                key = pk(avi3Item),
+                key = key.asAttrMap,
                 tableName = TableName(tableName),
                 updateExpression = UpdateExpression($(name).set(notAdam))
               )
               for {
                 _       <- updateItem.transaction.execute
-                written <- get[Person](tableName, pk(avi3Item)).execute
+                written <- get(tableName)(key).execute
               } yield assert(written)(isRight(equalTo(Person(first, notAdam, 7))))
             }
           },
@@ -1379,9 +1385,9 @@ object LiveSpec extends ZIOSpecDefault {
 
               for {
                 _       <- (putItem zip conditionCheck zip updateItem zip deleteItem).transaction.execute
-                put     <- get[Person](tableName, Item(id -> first, number -> 10)).execute
-                deleted <- get[Person](tableName, Item(id -> first, number -> 4)).execute
-                updated <- get[Person](tableName, Item(id -> first, number -> 7)).execute
+                put     <- get(tableName)(Person.id.partitionKey === first && Person.num.sortKey === 10).execute
+                deleted <- get(tableName)(Person.id.partitionKey === first && Person.num.sortKey === 4).execute
+                updated <- get(tableName)(Person.id.partitionKey === first && Person.num.sortKey === 7).execute
               } yield assert(put)(isRight(equalTo(Person(first, avi3, 10)))) &&
                 assert(deleted)(isLeft) &&
                 assert(updated)(isRight(equalTo(Person(first, notAdam, 7))))
diff --git a/dynamodb/src/main/scala/zio/dynamodb/DynamoDBExecutorImpl.scala b/dynamodb/src/main/scala/zio/dynamodb/DynamoDBExecutorImpl.scala
index cf135a325..35f9aa9e7 100644
--- a/dynamodb/src/main/scala/zio/dynamodb/DynamoDBExecutorImpl.scala
+++ b/dynamodb/src/main/scala/zio/dynamodb/DynamoDBExecutorImpl.scala
@@ -773,7 +773,7 @@ case object DynamoDBExecutorImpl {
   def awsQueryRequest(queryAll: QueryAll): QueryRequest = {
     val (aliasMap, (maybeFilterExpr, maybeKeyExpr, projections)) = (for {
       filter      <- AliasMapRender.collectAll(queryAll.filterExpression.map(_.render))
-      keyExpr     <- AliasMapRender.collectAll(queryAll.keyConditionExpression.map(_.render))
+      keyExpr     <- AliasMapRender.collectAll(queryAll.keyConditionExpr.map(_.render))
       projections <- AliasMapRender.forEach(queryAll.projections)
     } yield (filter, keyExpr, projections)).execute
 
@@ -797,7 +797,7 @@ case object DynamoDBExecutorImpl {
   private def awsQueryRequest(querySome: QuerySome): QueryRequest = {
     val (aliasMap, (maybeFilterExpr, maybeKeyExpr, projections)) = (for {
       filter      <- AliasMapRender.collectAll(querySome.filterExpression.map(_.render))
-      keyExpr     <- AliasMapRender.collectAll(querySome.keyConditionExpression.map(_.render))
+      keyExpr     <- AliasMapRender.collectAll(querySome.keyConditionExpr.map(_.render))
       projections <- AliasMapRender.forEach(querySome.projections)
     } yield (filter, keyExpr, projections)).execute
 
diff --git a/dynamodb/src/main/scala/zio/dynamodb/DynamoDBQuery.scala b/dynamodb/src/main/scala/zio/dynamodb/DynamoDBQuery.scala
index f47e7fa73..5e81e283a 100644
--- a/dynamodb/src/main/scala/zio/dynamodb/DynamoDBQuery.scala
+++ b/dynamodb/src/main/scala/zio/dynamodb/DynamoDBQuery.scala
@@ -1,7 +1,7 @@
 package zio.dynamodb
 
 import zio.dynamodb.DynamoDBError.ValueNotFound
-import zio.dynamodb.proofs.{ CanFilter, CanWhere, CanWhereKey }
+import zio.dynamodb.proofs.{ CanFilter, CanWhere }
 import zio.dynamodb.DynamoDBQuery.BatchGetItem.TableGet
 import zio.dynamodb.DynamoDBQuery.BatchWriteItem.{ Delete, Put }
 import zio.dynamodb.DynamoDBQuery.{
@@ -28,6 +28,7 @@ import zio.prelude.ForEachOps
 import zio.schema.Schema
 import zio.stream.Stream
 import zio.{ Chunk, NonEmptyChunk, Schedule, ZIO, Zippable => _, _ }
+import zio.dynamodb.proofs.IsPrimaryKey
 
 sealed trait DynamoDBQuery[-In, +Out] { self =>
 
@@ -220,7 +221,7 @@ sealed trait DynamoDBQuery[-In, +Out] { self =>
   }
 
   /**
-   * Parallel executes a DynamoDB Scan in parallel.
+   * Executes a DynamoDB Scan in parallel.
    * There are no guarantees on order of returned items.
    *
    * @param n The number of parallel requests to make to DynamoDB
@@ -317,12 +318,16 @@ sealed trait DynamoDBQuery[-In, +Out] { self =>
   def selectCount: DynamoDBQuery[In, Out]                  = select(Select.Count)
 
   /**
-   * Adds a KeyConditionExpression to a DynamoDBQuery. Example:
+   * Adds a KeyConditionExpr to a DynamoDBQuery. Example:
    * {{{
-   * val newQuery = query.whereKey(partitionKey("email") === "avi@gmail.com" && sortKey("subject") === "maths")
+   * // high level type safe API where "email" and "subject" keys are defined using ProjectionExpression.accessors[Student]
+   * val newQuery = query.whereKey(email.partitionKey === "avi@gmail.com" && subject.sortKey === "maths")
+   *
+   * // low level API
+   * val newQuery = query.whereKey($("email").partitionKey === "avi@gmail.com" && $("subject").sortKey === "maths")
    * }}}
    */
-  def whereKey(keyConditionExpression: KeyConditionExpression): DynamoDBQuery[In, Out] =
+  def whereKey[From, Pk, Sk](keyConditionExpression: KeyConditionExpr[From, Pk, Sk]): DynamoDBQuery[In, Out] =
     self match {
       case Zip(left, right, zippable) =>
         Zip(left.whereKey(keyConditionExpression), right.whereKey(keyConditionExpression), zippable)
@@ -330,46 +335,12 @@ sealed trait DynamoDBQuery[-In, +Out] { self =>
       case Absolve(query)             => Absolve(query.whereKey(keyConditionExpression))
 
       case s: QuerySome =>
-        s.copy(keyConditionExpression = Some(keyConditionExpression)).asInstanceOf[DynamoDBQuery[In, Out]]
+        s.copy(keyConditionExpr = Some(keyConditionExpression)).asInstanceOf[DynamoDBQuery[In, Out]]
       case s: QueryAll  =>
-        s.copy(keyConditionExpression = Some(keyConditionExpression)).asInstanceOf[DynamoDBQuery[In, Out]]
+        s.copy(keyConditionExpr = Some(keyConditionExpression)).asInstanceOf[DynamoDBQuery[In, Out]]
       case _            => self
     }
 
-  /**
-   * Adds a KeyConditionExpression from a ConditionExpression to a DynamoDBQuery
-   * Must be in the form of `<Condition1> && <Condition2>` where format of `<Condition1>` is:
-   * {{{<ProjectionExpressionForPartitionKey> === <value>}}}
-   * and the format of `<Condition2>` is:
-   * {{{<ProjectionExpressionForSortKey> <op> <value>}}} where op can be one of `===`, `>`, `>=`, `<`, `<=`, `between`, `beginsWith`
-   *
-   * Example using type safe API:
-   * {{{
-   * // email and subject are partition and sort keys respectively
-   * val (email, subject, enrollmentDate, payment) = ProjectionExpression.accessors[Student]
-   * // ...
-   * val newQuery = query.whereKey(email === "avi@gmail.com" && subject === "maths")
-   * }}}
-   */
-  def whereKey[B](
-    conditionExpression: ConditionExpression[B]
-  )(implicit ev: CanWhereKey[B, Out]): DynamoDBQuery[In, Out] = {
-    val _                                              = ev
-    val keyConditionExpression: KeyConditionExpression =
-      KeyConditionExpression.fromConditionExpressionUnsafe(conditionExpression)
-    self match {
-      case Zip(left, right, zippable) =>
-        Zip(left.whereKey(keyConditionExpression), right.whereKey(keyConditionExpression), zippable)
-      case Map(query, mapper)         => Map(query.whereKey(keyConditionExpression), mapper)
-      case Absolve(query)             => Absolve(query.whereKey(keyConditionExpression))
-      case s: QuerySome               =>
-        s.copy(keyConditionExpression = Some(keyConditionExpression)).asInstanceOf[DynamoDBQuery[In, Out]]
-      case s: QueryAll                =>
-        s.copy(keyConditionExpression = Some(keyConditionExpression)).asInstanceOf[DynamoDBQuery[In, Out]]
-      case _                          => self
-    }
-  }
-
   def withRetryPolicy(retryPolicy: Schedule[Any, Throwable, Any]): DynamoDBQuery[In, Out] =
     self match {
       case Zip(left, right, zippable) =>
@@ -482,7 +453,17 @@ object DynamoDBQuery {
   ): DynamoDBQuery[Any, Option[Item]] =
     GetItem(TableName(tableName), key, projections.toList)
 
-  def get[A: Schema](
+  def get[From: Schema, Pk, Sk](
+    tableName: String,
+    projections: ProjectionExpression[_, _]*
+  )(
+    partitionKeyExpr: KeyConditionExpr.PrimaryKeyExpr[From, Pk, Sk]
+  )(implicit ev: IsPrimaryKey[Pk], ev2: IsPrimaryKey[Sk]): DynamoDBQuery[From, Either[DynamoDBError, From]] = {
+    val (_, _) = (ev, ev2)
+    get(tableName, partitionKeyExpr.asAttrMap, projections: _*)
+  }
+
+  private def get[A: Schema](
     tableName: String,
     key: PrimaryKey,
     projections: ProjectionExpression[_, _]*
@@ -515,13 +496,24 @@ object DynamoDBQuery {
       UpdateExpression(action)
     )
 
-  def update[A: Schema](tableName: String, key: PrimaryKey)(action: Action[A]): DynamoDBQuery[A, Option[A]] =
+  private[dynamodb] def update[A: Schema](tableName: String, key: PrimaryKey)(
+    action: Action[A]
+  ): DynamoDBQuery[A, Option[A]] =
     updateItem(tableName, key)(action).map(_.flatMap(item => fromItem(item).toOption))
 
+  def update[From: Schema, Pk, Sk](tableName: String)(primaryKeyExpr: KeyConditionExpr.PrimaryKeyExpr[From, Pk, Sk])(
+    action: Action[From]
+  ): DynamoDBQuery[From, Option[From]] =
+    updateItem(tableName, primaryKeyExpr.asAttrMap)(action).map(_.flatMap(item => fromItem(item).toOption))
+
   def deleteItem(tableName: String, key: PrimaryKey): Write[Any, Option[Item]] = DeleteItem(TableName(tableName), key)
 
-  def delete[A: Schema](tableName: String, key: PrimaryKey): DynamoDBQuery[Any, Option[A]] =
-    deleteItem(tableName, key).map(_.flatMap(item => fromItem(item).toOption))
+  def delete[From: Schema, Pk, Sk](
+    tableName: String
+  )(
+    primaryKeyExpr: KeyConditionExpr.PrimaryKeyExpr[From, Pk, Sk]
+  ): DynamoDBQuery[Any, Option[From]] =
+    deleteItem(tableName, primaryKeyExpr.asAttrMap).map(_.flatMap(item => fromItem(item).toOption))
 
   /**
    * when executed will return a Tuple of {{{(Chunk[Item], LastEvaluatedKey)}}}
@@ -846,7 +838,7 @@ object DynamoDBQuery {
     exclusiveStartKey: LastEvaluatedKey =
       None,                                                     // allows client to control start position - eg for client managed paging
     filterExpression: Option[FilterExpression[_]] = None,
-    keyConditionExpression: Option[KeyConditionExpression] = None,
+    keyConditionExpr: Option[KeyConditionExpr[_, _, _]] = None,
     projections: List[ProjectionExpression[_, _]] = List.empty, // if empty all attributes will be returned
     capacity: ReturnConsumedCapacity = ReturnConsumedCapacity.None,
     select: Option[Select] = None,                              // if ProjectExpression supplied then only valid value is SpecificAttributes
@@ -879,7 +871,7 @@ object DynamoDBQuery {
     exclusiveStartKey: LastEvaluatedKey =
       None,                                                     // allows client to control start position - eg for client managed paging
     filterExpression: Option[FilterExpression[_]] = None,
-    keyConditionExpression: Option[KeyConditionExpression] = None,
+    keyConditionExpr: Option[KeyConditionExpr[_, _, _]] = None,
     projections: List[ProjectionExpression[_, _]] = List.empty, // if empty all attributes will be returned
     capacity: ReturnConsumedCapacity = ReturnConsumedCapacity.None,
     select: Option[Select] = None,                              // if ProjectExpression supplied then only valid value is SpecificAttributes
diff --git a/dynamodb/src/main/scala/zio/dynamodb/KeyConditionExpr.scala b/dynamodb/src/main/scala/zio/dynamodb/KeyConditionExpr.scala
new file mode 100644
index 000000000..5f5a34a93
--- /dev/null
+++ b/dynamodb/src/main/scala/zio/dynamodb/KeyConditionExpr.scala
@@ -0,0 +1,147 @@
+package zio.dynamodb
+
+/**
+ * This sum type models:
+ * 1) partition key equality expressions
+ * 2) composite primary key expressions where sort key expression is equality
+ * 3) extended composite primary key expressions where sort key is not equality eg >, <, >=, <=, between, begins_with
+ *
+ * Note 1), 2) and 3) are all valid key condition expressions used in Query DynamoDB queries
+ * BUT only 1) and 2) are valid primary key expressions that can be used in GetItem, UpdateItem and DeleteItem DynamoDB queries
+ */
+sealed trait KeyConditionExpr[-From, +Pk, +Sk] extends Renderable { self =>
+  def render: AliasMapRender[String]
+}
+
+object KeyConditionExpr {
+  type SortKeyNotUsed
+
+  sealed trait PrimaryKeyExpr[-From, +Pk, +Sk] extends KeyConditionExpr[From, Pk, Sk] {
+    def asAttrMap: AttrMap
+  }
+
+  def getOrInsert[From, To](primaryKeyName: String): AliasMapRender[String] =
+    // note primary keys must be scalar values, they can't be nested
+    AliasMapRender.getOrInsert(ProjectionExpression.MapElement[From, To](ProjectionExpression.Root, primaryKeyName))
+
+  private[dynamodb] final case class PartitionKeyEquals[-From, +Pk](pk: PartitionKey[From, Pk], value: AttributeValue)
+      extends PrimaryKeyExpr[From, Pk, SortKeyNotUsed] { self =>
+
+    def &&[From1 <: From, Sk](other: SortKeyEquals[From1, Sk]): CompositePrimaryKeyExpr[From1, Pk, Sk] =
+      CompositePrimaryKeyExpr[From1, Pk, Sk](self.asInstanceOf[PartitionKeyEquals[From1, Pk]], other)
+    def &&[From1 <: From, Sk](
+      other: ExtendedSortKeyExpr[From1, Sk]
+    ): ExtendedCompositePrimaryKeyExpr[From1, Pk, Sk]                                                  =
+      ExtendedCompositePrimaryKeyExpr[From1, Pk, Sk](self.asInstanceOf[PartitionKeyEquals[From1, Pk]], other)
+
+    def asAttrMap: AttrMap = AttrMap(pk.keyName -> value)
+
+    override def render: AliasMapRender[String] =
+      for {
+        v        <- AliasMapRender.getOrInsert(value)
+        keyAlias <- KeyConditionExpr.getOrInsert(pk.keyName)
+      } yield s"${keyAlias} = $v"
+
+  }
+
+  private[dynamodb] final case class SortKeyEquals[-From, +Sk](sortKey: SortKey[From, Sk], value: AttributeValue) {
+    self =>
+    def miniRender: AliasMapRender[String] =
+      for {
+        v        <- AliasMapRender.getOrInsert(value)
+        keyAlias <- KeyConditionExpr.getOrInsert(sortKey.keyName)
+      } yield s"${keyAlias} = $v"
+  }
+
+  private[dynamodb] final case class CompositePrimaryKeyExpr[-From, +Pk, +Sk](
+    pk: PartitionKeyEquals[From, Pk],
+    sk: SortKeyEquals[From, Sk]
+  ) extends PrimaryKeyExpr[From, Pk, Sk] {
+    self =>
+
+    def asAttrMap: AttrMap = PrimaryKey(pk.pk.keyName -> pk.value, sk.sortKey.keyName -> sk.value)
+
+    override def render: AliasMapRender[String] =
+      for {
+        pkStr <- pk.render
+        skStr <- sk.miniRender
+      } yield s"$pkStr AND $skStr"
+
+  }
+  private[dynamodb] final case class ExtendedCompositePrimaryKeyExpr[-From, +Pk, +Sk](
+    pk: PartitionKeyEquals[From, Pk],
+    sk: ExtendedSortKeyExpr[From, Sk]
+  ) extends KeyConditionExpr[From, Pk, Sk] {
+    self =>
+
+    def render: AliasMapRender[String] =
+      for {
+        pkStr <- pk.render
+        skStr <- sk.miniRender
+      } yield s"$pkStr AND $skStr"
+
+  }
+
+  sealed trait ExtendedSortKeyExpr[-From, +Sk] { self =>
+    def miniRender: AliasMapRender[String] =
+      self match {
+        case ExtendedSortKeyExpr.GreaterThan(sk, value)        =>
+          for {
+            v        <- AliasMapRender.getOrInsert(value)
+            keyAlias <- KeyConditionExpr.getOrInsert(sk.keyName)
+          } yield s"${keyAlias} > $v"
+        case ExtendedSortKeyExpr.LessThan(sk, value)           =>
+          for {
+            v        <- AliasMapRender.getOrInsert(value)
+            keyAlias <- KeyConditionExpr.getOrInsert(sk.keyName)
+          } yield s"${keyAlias} < $v"
+        case ExtendedSortKeyExpr.NotEqual(sk, value)           =>
+          for {
+            v        <- AliasMapRender.getOrInsert(value)
+            keyAlias <- KeyConditionExpr.getOrInsert(sk.keyName)
+          } yield s"${keyAlias} <> $v"
+        case ExtendedSortKeyExpr.LessThanOrEqual(sk, value)    =>
+          for {
+            v        <- AliasMapRender.getOrInsert(value)
+            keyAlias <- KeyConditionExpr.getOrInsert(sk.keyName)
+          } yield s"${keyAlias} <= $v"
+        case ExtendedSortKeyExpr.GreaterThanOrEqual(sk, value) =>
+          for {
+            v        <- AliasMapRender.getOrInsert(value)
+            keyAlias <- KeyConditionExpr.getOrInsert(sk.keyName)
+          } yield s"${keyAlias} >= $v"
+        case ExtendedSortKeyExpr.Between(left, min, max)       =>
+          for {
+            min2     <- AliasMapRender.getOrInsert(min)
+            max2     <- AliasMapRender.getOrInsert(max)
+            keyAlias <- KeyConditionExpr.getOrInsert(left.keyName)
+          } yield s"${keyAlias} BETWEEN $min2 AND $max2"
+        case ExtendedSortKeyExpr.BeginsWith(left, value)       =>
+          for {
+            v        <- AliasMapRender.getOrInsert(value)
+            keyAlias <- KeyConditionExpr.getOrInsert(left.keyName)
+          } yield s"begins_with(${keyAlias}, $v)"
+      }
+
+  }
+  object ExtendedSortKeyExpr {
+    private[dynamodb] final case class GreaterThan[From, +To](sortKey: SortKey[From, To], value: AttributeValue)
+        extends ExtendedSortKeyExpr[From, To]
+    private[dynamodb] final case class LessThan[From, +To](sortKey: SortKey[From, To], value: AttributeValue)
+        extends ExtendedSortKeyExpr[From, To]
+    private[dynamodb] final case class NotEqual[From, +To](sortKey: SortKey[From, To], value: AttributeValue)
+        extends ExtendedSortKeyExpr[From, To]
+    private[dynamodb] final case class LessThanOrEqual[From, +To](sortKey: SortKey[From, To], value: AttributeValue)
+        extends ExtendedSortKeyExpr[From, To]
+    private[dynamodb] final case class GreaterThanOrEqual[From, +To](sortKey: SortKey[From, To], value: AttributeValue)
+        extends ExtendedSortKeyExpr[From, To]
+    private[dynamodb] final case class Between[From, +To](
+      left: SortKey[From, To],
+      min: AttributeValue,
+      max: AttributeValue
+    ) extends ExtendedSortKeyExpr[From, To]
+    private[dynamodb] final case class BeginsWith[From, +To](left: SortKey[From, To], value: AttributeValue)
+        extends ExtendedSortKeyExpr[From, To]
+  }
+
+}
diff --git a/dynamodb/src/main/scala/zio/dynamodb/KeyConditionExpression.scala b/dynamodb/src/main/scala/zio/dynamodb/KeyConditionExpression.scala
deleted file mode 100644
index db74996a3..000000000
--- a/dynamodb/src/main/scala/zio/dynamodb/KeyConditionExpression.scala
+++ /dev/null
@@ -1,252 +0,0 @@
-package zio.dynamodb
-
-import zio.dynamodb.ConditionExpression.Operand.ProjectionExpressionOperand
-import zio.dynamodb.PartitionKeyExpression.PartitionKey
-import zio.dynamodb.ProjectionExpression.{ MapElement, Root }
-import zio.dynamodb.SortKeyExpression.SortKey
-
-/*
-KeyCondition expression is a restricted version of ConditionExpression where by
-- partition exprn is required and can only use "=" equals comparison
-- optionally AND can be used to add a sort key expression
-
-eg partitionKeyName = :partitionkeyval AND sortKeyName = :sortkeyval
-comparisons operators are the same as for Condition
-
- */
-
-sealed trait KeyConditionExpression extends Renderable { self =>
-  def render: AliasMapRender[String] =
-    self match {
-      case KeyConditionExpression.And(left, right) =>
-        left.render
-          .zipWith(
-            right.render
-          ) { case (l, r) => s"$l AND $r" }
-      case expression: PartitionKeyExpression      => expression.render
-    }
-
-}
-
-object KeyConditionExpression {
-
-  def getOrInsert[From, To](primaryKeyName: String): AliasMapRender[String] =
-    AliasMapRender.getOrInsert(ProjectionExpression.MapElement[From, To](Root, primaryKeyName))
-  private[dynamodb] final case class And(left: PartitionKeyExpression, right: SortKeyExpression)
-      extends KeyConditionExpression
-  def partitionKey(key: String): PartitionKey                               = PartitionKey(key)
-  def sortKey(key: String): SortKey                                         = SortKey(key)
-
-  /**
-   * Create a KeyConditionExpression from a ConditionExpression
-   * Must be in the form of `<Condition1> && <Condition2>` where format of `<Condition1>` is:
-   * {{{<ProjectionExpressionForPartitionKey> === <value>}}}
-   * and the format of `<Condition2>` is:
-   * {{{<ProjectionExpressionForSortKey> <op> <value>}}} where op can be one of `===`, `>`, `>=`, `<`, `<=`, `between`, `beginsWith`
-   *
-   * Example using type API:
-   * {{{
-   * val (email, subject, enrollmentDate, payment) = ProjectionExpression.accessors[Student]
-   * // ...
-   * val keyConditionExprn = filterKey(email === "avi@gmail.com" && subject === "maths")
-   * }}}
-   */
-  private[dynamodb] def fromConditionExpressionUnsafe(c: ConditionExpression[_]): KeyConditionExpression =
-    KeyConditionExpression(c).getOrElse(
-      throw new IllegalStateException(s"Error: invalid key condition expression $c")
-    )
-
-  private[dynamodb] def apply(c: ConditionExpression[_]): Either[String, KeyConditionExpression] =
-    c match {
-      case ConditionExpression.Equals(
-            ProjectionExpressionOperand(MapElement(Root, partitionKey)),
-            ConditionExpression.Operand.ValueOperand(av)
-          ) =>
-        Right(PartitionKeyExpression.Equals(PartitionKey(partitionKey), av))
-      case ConditionExpression.And(
-            ConditionExpression.Equals(
-              ProjectionExpressionOperand(MapElement(Root, partitionKey)),
-              ConditionExpression.Operand.ValueOperand(avL)
-            ),
-            rhs
-          ) =>
-        rhs match {
-          case ConditionExpression.Equals(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                ConditionExpression.Operand.ValueOperand(avR)
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.Equals(SortKey(sortKey), avR))
-            )
-          case ConditionExpression.NotEqual(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                ConditionExpression.Operand.ValueOperand(avR)
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.NotEqual(SortKey(sortKey), avR))
-            )
-          case ConditionExpression.GreaterThan(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                ConditionExpression.Operand.ValueOperand(avR)
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.GreaterThan(SortKey(sortKey), avR))
-            )
-          case ConditionExpression.LessThan(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                ConditionExpression.Operand.ValueOperand(avR)
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.LessThan(SortKey(sortKey), avR))
-            )
-          case ConditionExpression.GreaterThanOrEqual(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                ConditionExpression.Operand.ValueOperand(avR)
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.GreaterThanOrEqual(SortKey(sortKey), avR))
-            )
-          case ConditionExpression.LessThanOrEqual(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                ConditionExpression.Operand.ValueOperand(avR)
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.LessThanOrEqual(SortKey(sortKey), avR))
-            )
-          case ConditionExpression.Between(
-                ProjectionExpressionOperand(MapElement(Root, sortKey)),
-                avMin,
-                avMax
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.Between(SortKey(sortKey), avMin, avMax))
-            )
-          case ConditionExpression.BeginsWith(
-                MapElement(Root, sortKey),
-                av
-              ) =>
-            Right(
-              PartitionKeyExpression
-                .Equals(PartitionKey(partitionKey), avL)
-                .&&(SortKeyExpression.BeginsWith(SortKey(sortKey), av))
-            )
-          case c => Left(s"condition '$c' is not a valid sort condition expression")
-        }
-
-      case c => Left(s"condition $c is not a valid key condition expression")
-    }
-
-}
-
-sealed trait PartitionKeyExpression extends KeyConditionExpression { self =>
-  import KeyConditionExpression.And
-
-  def &&(that: SortKeyExpression): KeyConditionExpression = And(self, that)
-
-  override def render: AliasMapRender[String] =
-    self match {
-      case PartitionKeyExpression.Equals(left, right) =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} = $v"
-    }
-}
-object PartitionKeyExpression {
-  final case class PartitionKey(keyName: String) { self =>
-    def ===[A](that: A)(implicit t: ToAttributeValue[A]): PartitionKeyExpression =
-      Equals(self, t.toAttributeValue(that))
-  }
-  final case class Equals(left: PartitionKey, right: AttributeValue) extends PartitionKeyExpression
-}
-
-sealed trait SortKeyExpression { self =>
-  def render: AliasMapRender[String] =
-    self match {
-      case SortKeyExpression.Equals(left, right)             =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} = $v"
-      case SortKeyExpression.LessThan(left, right)           =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} < $v"
-      case SortKeyExpression.NotEqual(left, right)           =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} <> $v"
-      case SortKeyExpression.GreaterThan(left, right)        =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} > $v"
-      case SortKeyExpression.LessThanOrEqual(left, right)    =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} <= $v"
-      case SortKeyExpression.GreaterThanOrEqual(left, right) =>
-        for {
-          v       <- AliasMapRender.getOrInsert(right)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} >= $v"
-      case SortKeyExpression.Between(left, min, max)         =>
-        for {
-          min2    <- AliasMapRender.getOrInsert(min)
-          max2    <- AliasMapRender.getOrInsert(max)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"${keyName} BETWEEN $min2 AND $max2"
-      case SortKeyExpression.BeginsWith(left, value)         =>
-        for {
-          v       <- AliasMapRender.getOrInsert(value)
-          keyName <- KeyConditionExpression.getOrInsert(left.keyName)
-        } yield s"begins_with(${keyName}, $v)"
-    }
-
-}
-
-object SortKeyExpression {
-
-  final case class SortKey(keyName: String) { self =>
-    def ===[A](that: A)(implicit t: ToAttributeValue[A]): SortKeyExpression            = Equals(self, t.toAttributeValue(that))
-    def <>[A](that: A)(implicit t: ToAttributeValue[A]): SortKeyExpression             = NotEqual(self, t.toAttributeValue(that))
-    def <[A](that: A)(implicit t: ToAttributeValue[A]): SortKeyExpression              = LessThan(self, t.toAttributeValue(that))
-    def <=[A](that: A)(implicit t: ToAttributeValue[A]): SortKeyExpression             =
-      LessThanOrEqual(self, t.toAttributeValue(that))
-    def >[A](that: A)(implicit t: ToAttributeValue[A]): SortKeyExpression              =
-      GreaterThanOrEqual(self, t.toAttributeValue(that))
-    def >=[A](that: A)(implicit t: ToAttributeValue[A]): SortKeyExpression             =
-      GreaterThanOrEqual(self, t.toAttributeValue(that))
-    def between[A](min: A, max: A)(implicit t: ToAttributeValue[A]): SortKeyExpression =
-      Between(self, t.toAttributeValue(min), t.toAttributeValue(max))
-    def beginsWith[A](value: A)(implicit t: ToAttributeValue[A]): SortKeyExpression    =
-      BeginsWith(self, t.toAttributeValue(value))
-  }
-
-  private[dynamodb] final case class Equals(left: SortKey, right: AttributeValue)             extends SortKeyExpression
-  private[dynamodb] final case class NotEqual(left: SortKey, right: AttributeValue)           extends SortKeyExpression
-  private[dynamodb] final case class LessThan(left: SortKey, right: AttributeValue)           extends SortKeyExpression
-  private[dynamodb] final case class GreaterThan(left: SortKey, right: AttributeValue)        extends SortKeyExpression
-  private[dynamodb] final case class LessThanOrEqual(left: SortKey, right: AttributeValue)    extends SortKeyExpression
-  private[dynamodb] final case class GreaterThanOrEqual(left: SortKey, right: AttributeValue) extends SortKeyExpression
-  private[dynamodb] final case class Between(left: SortKey, min: AttributeValue, max: AttributeValue)
-      extends SortKeyExpression
-  private[dynamodb] final case class BeginsWith(left: SortKey, value: AttributeValue)         extends SortKeyExpression
-}
diff --git a/dynamodb/src/main/scala/zio/dynamodb/PartitionKey.scala b/dynamodb/src/main/scala/zio/dynamodb/PartitionKey.scala
new file mode 100644
index 000000000..e623d9a68
--- /dev/null
+++ b/dynamodb/src/main/scala/zio/dynamodb/PartitionKey.scala
@@ -0,0 +1,13 @@
+package zio.dynamodb
+
+import zio.dynamodb.KeyConditionExpr.PartitionKeyEquals
+import zio.dynamodb.proofs.RefersTo
+
+private[dynamodb] final case class PartitionKey[-From, +To](keyName: String) { self =>
+  def ===[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): PartitionKeyEquals[From, To] = {
+    val _ = ev
+    PartitionKeyEquals(self, implicitly[ToAttributeValue[To2]].toAttributeValue(value))
+  }
+}
diff --git a/dynamodb/src/main/scala/zio/dynamodb/ProjectionExpression.scala b/dynamodb/src/main/scala/zio/dynamodb/ProjectionExpression.scala
index 7e70ac122..e8437b1ec 100644
--- a/dynamodb/src/main/scala/zio/dynamodb/ProjectionExpression.scala
+++ b/dynamodb/src/main/scala/zio/dynamodb/ProjectionExpression.scala
@@ -133,6 +133,22 @@ sealed trait ProjectionExpression[-From, +To] { self =>
 
 trait ProjectionExpressionLowPriorityImplicits0 extends ProjectionExpressionLowPriorityImplicits1 {
   implicit class ProjectionExpressionSyntax0[From, To: ToAttributeValue](self: ProjectionExpression[From, To]) {
+
+    def partitionKey(implicit ev: IsPrimaryKey[To]): PartitionKey[From, To] = {
+      val _ = ev
+      self match {
+        case ProjectionExpression.MapElement(_, key) => PartitionKey[From, To](key)
+        case _                                       => throw new IllegalArgumentException("Not a partition key") // should not happen
+      }
+    }
+    def sortKey(implicit ev: IsPrimaryKey[To]): SortKey[From, To] = {
+      val _ = ev
+      self match {
+        case ProjectionExpression.MapElement(_, key) => SortKey[From, To](key)
+        case _                                       => throw new IllegalArgumentException("Not a partition key") // should not happen
+      }
+    }
+
     def set(a: To): UpdateExpression.Action.SetAction[From, To] =
       UpdateExpression.Action.SetAction(
         self,
@@ -528,6 +544,17 @@ object ProjectionExpression extends ProjectionExpressionLowPriorityImplicits0 {
 
   implicit class ProjectionExpressionSyntax[From](self: ProjectionExpression[From, Unknown]) {
 
+    def partitionKey: PartitionKey[From, Unknown] =
+      self match {
+        case ProjectionExpression.MapElement(_, key) => PartitionKey[From, Unknown](key)
+        case _                                       => throw new IllegalArgumentException("Not a partition key") // should not happen
+      }
+    def sortKey: SortKey[From, Unknown]           =
+      self match {
+        case ProjectionExpression.MapElement(_, key) => SortKey[From, Unknown](key)
+        case _                                       => throw new IllegalArgumentException("Not a partition key") // should not happen
+      }
+
     /**
      * Modify or Add an item Attribute
      */
diff --git a/dynamodb/src/main/scala/zio/dynamodb/SortKey.scala b/dynamodb/src/main/scala/zio/dynamodb/SortKey.scala
new file mode 100644
index 000000000..85bfa1d34
--- /dev/null
+++ b/dynamodb/src/main/scala/zio/dynamodb/SortKey.scala
@@ -0,0 +1,84 @@
+package zio.dynamodb
+
+import zio.dynamodb.proofs.RefersTo
+import zio.dynamodb.proofs.CanSortKeyBeginsWith
+
+import zio.dynamodb.KeyConditionExpr.SortKeyEquals
+import zio.dynamodb.KeyConditionExpr.ExtendedSortKeyExpr
+
+private[dynamodb] final case class SortKey[-From, +To](keyName: String) { self =>
+  // all comparison ops apply to: Strings, Numbers, Binary values
+  def ===[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): SortKeyEquals[From, To2] = {
+    val _ = ev
+    SortKeyEquals[From, To2](
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly[ToAttributeValue[To2]].toAttributeValue(value)
+    )
+  }
+  def >[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): ExtendedSortKeyExpr[From, To2] = {
+    val _ = ev
+    ExtendedSortKeyExpr.GreaterThan(
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly(ToAttributeValue[To2]).toAttributeValue(value)
+    )
+  }
+  def <[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): ExtendedSortKeyExpr[From, To2] = {
+    val _ = ev
+    ExtendedSortKeyExpr.LessThan(
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly[ToAttributeValue[To2]].toAttributeValue(value)
+    )
+  }
+  def <>[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): ExtendedSortKeyExpr[From, To2] = {
+    val _ = ev
+    ExtendedSortKeyExpr.NotEqual(
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly(ToAttributeValue[To2]).toAttributeValue(value)
+    )
+  }
+  def <=[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): ExtendedSortKeyExpr[From, To2] = {
+    val _ = ev
+    ExtendedSortKeyExpr.LessThanOrEqual(
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly[ToAttributeValue[To2]].toAttributeValue(value)
+    )
+  }
+  def >=[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    value: To2
+  )(implicit ev: RefersTo[To1, To2]): ExtendedSortKeyExpr[From, To2] = {
+    val _ = ev
+    ExtendedSortKeyExpr.GreaterThanOrEqual(
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly[ToAttributeValue[To2]].toAttributeValue(value)
+    )
+  }
+  // applies to all PK types
+  def between[To: ToAttributeValue, IsPrimaryKey](min: To, max: To): ExtendedSortKeyExpr[From, To] =
+    ExtendedSortKeyExpr.Between[From, To](
+      self.asInstanceOf[SortKey[From, To]],
+      implicitly[ToAttributeValue[To]].toAttributeValue(min),
+      implicitly[ToAttributeValue[To]].toAttributeValue(max)
+    )
+
+  // beginsWith applies to: Strings, Binary values
+  def beginsWith[To1 >: To, To2: ToAttributeValue, IsPrimaryKey](
+    prefix: To2
+  )(implicit ev: CanSortKeyBeginsWith[To1, To2]): ExtendedSortKeyExpr[From, To2] = {
+    val _ = ev
+    ExtendedSortKeyExpr.BeginsWith[From, To2](
+      self.asInstanceOf[SortKey[From, To2]],
+      implicitly[ToAttributeValue[To2]].toAttributeValue(prefix)
+    )
+  }
+
+}
diff --git a/dynamodb/src/main/scala/zio/dynamodb/package.scala b/dynamodb/src/main/scala/zio/dynamodb/package.scala
index d0776ff0a..3b163e2bb 100644
--- a/dynamodb/src/main/scala/zio/dynamodb/package.scala
+++ b/dynamodb/src/main/scala/zio/dynamodb/package.scala
@@ -2,6 +2,7 @@ package zio
 
 import zio.schema.Schema
 import zio.stream.{ ZSink, ZStream }
+import zio.dynamodb.proofs.IsPrimaryKey
 
 package object dynamodb {
   // Filter expression is the same as a ConditionExpression but when used with Query but does not allow key attributes
@@ -35,7 +36,9 @@ package object dynamodb {
   def batchWriteFromStream[R, A, In, B](
     stream: ZStream[R, Throwable, A],
     mPar: Int = 10
-  )(f: A => DynamoDBQuery[In, B]): ZStream[DynamoDBExecutor with R, Throwable, B] =
+  )(
+    f: A => DynamoDBQuery[In, B]
+  ): ZStream[DynamoDBExecutor with R, Throwable, B] = // TODO: Avi - can we constraint query to put or write?
     stream
       .aggregateAsync(ZSink.collectAllN[A](25))
       .mapZIOPar(mPar) { chunk =>
@@ -100,25 +103,26 @@ package object dynamodb {
    * @tparam B implicit Schema[B] where B is the type of the element in the returned stream
    * @return stream of Either[DynamoDBError.DecodingError, (A, Option[B])]
    */
-  def batchReadFromStream[R, A, B: Schema](
+  def batchReadFromStream[R, A, From: Schema, Pk: IsPrimaryKey, Sk: IsPrimaryKey](
     tableName: String,
     stream: ZStream[R, Throwable, A],
     mPar: Int = 10
   )(
-    pk: A => PrimaryKey
-  ): ZStream[R with DynamoDBExecutor, Throwable, Either[DynamoDBError.DecodingError, (A, Option[B])]] =
+    pk: A => KeyConditionExpr.PrimaryKeyExpr[From, Pk, Sk]
+  ): ZStream[R with DynamoDBExecutor, Throwable, Either[DynamoDBError.DecodingError, (A, Option[From])]] =
     stream
       .aggregateAsync(ZSink.collectAllN[A](100))
       .mapZIOPar(mPar) { chunk =>
-        val batchGetItem: DynamoDBQuery[B, Chunk[Either[DynamoDBError.DecodingError, (A, Option[B])]]] = DynamoDBQuery
-          .forEach(chunk) { a =>
-            DynamoDBQuery.get(tableName, pk(a)).map {
-              case Right(b)                                 => Right((a, Some(b)))
-              case Left(DynamoDBError.ValueNotFound(_))     => Right((a, None))
-              case Left(e @ DynamoDBError.DecodingError(_)) => Left(e)
+        val batchGetItem: DynamoDBQuery[From, Chunk[Either[DynamoDBError.DecodingError, (A, Option[From])]]] =
+          DynamoDBQuery
+            .forEach(chunk) { a =>
+              DynamoDBQuery.get(tableName)(pk(a)).map {
+                case Right(b)                                 => Right((a, Some(b)))
+                case Left(DynamoDBError.ValueNotFound(_))     => Right((a, None))
+                case Left(e @ DynamoDBError.DecodingError(_)) => Left(e)
+              }
             }
-          }
-          .map(Chunk.fromIterable)
+            .map(Chunk.fromIterable)
         for {
           r <- ZIO.environment[DynamoDBExecutor]
           list <- batchGetItem.execute.provideEnvironment(r)
diff --git a/dynamodb/src/main/scala/zio/dynamodb/proofs/CanSortKeyBeginsWith.scala b/dynamodb/src/main/scala/zio/dynamodb/proofs/CanSortKeyBeginsWith.scala
new file mode 100644
index 000000000..26f95d12c
--- /dev/null
+++ b/dynamodb/src/main/scala/zio/dynamodb/proofs/CanSortKeyBeginsWith.scala
@@ -0,0 +1,23 @@
+package zio.dynamodb.proofs
+
+import zio.dynamodb.ProjectionExpression
+import scala.annotation.implicitNotFound
+
+@implicitNotFound(
+  "Fields of type ${X} has 'beginsWith' argument of type ${A} - they must be the same type"
+)
+sealed trait CanSortKeyBeginsWith[-X, -A]
+trait CanSortKeyBeginsWith0 extends CanSortKeyBeginsWith1 {
+  implicit def unknownRight[X]: CanSortKeyBeginsWith[X, ProjectionExpression.Unknown] =
+    new CanSortKeyBeginsWith[X, ProjectionExpression.Unknown] {}
+}
+trait CanSortKeyBeginsWith1 {
+  // begins_with with only applies to keys of type string or bytes
+  implicit def bytes[A <: Iterable[Byte]]: CanSortKeyBeginsWith[A, A] =
+    new CanSortKeyBeginsWith[A, A] {}
+  implicit def string: CanSortKeyBeginsWith[String, String]           = new CanSortKeyBeginsWith[String, String] {}
+}
+object CanSortKeyBeginsWith extends CanSortKeyBeginsWith0 {
+  implicit def unknownLeft[X]: CanSortKeyBeginsWith[ProjectionExpression.Unknown, X] =
+    new CanSortKeyBeginsWith[ProjectionExpression.Unknown, X] {}
+}
diff --git a/dynamodb/src/main/scala/zio/dynamodb/proofs/IsPrimaryKey.scala b/dynamodb/src/main/scala/zio/dynamodb/proofs/IsPrimaryKey.scala
new file mode 100644
index 000000000..473c49260
--- /dev/null
+++ b/dynamodb/src/main/scala/zio/dynamodb/proofs/IsPrimaryKey.scala
@@ -0,0 +1,23 @@
+package zio.dynamodb.proofs
+
+import scala.annotation.implicitNotFound
+import zio.dynamodb.KeyConditionExpr
+
+@implicitNotFound("DynamoDB does not support primary key type ${A} - allowed types are: String, Number, Binary")
+sealed trait IsPrimaryKey[-A]
+
+object IsPrimaryKey {
+  implicit val stringIsPrimaryKey: IsPrimaryKey[String] = new IsPrimaryKey[String] {}
+
+  implicit val shortIsPrimaryKey: IsPrimaryKey[Short]           = new IsPrimaryKey[Short] {}
+  implicit val intIsPrimaryKey: IsPrimaryKey[Int]               = new IsPrimaryKey[Int] {}
+  implicit val longIsPrimaryKey: IsPrimaryKey[Long]             = new IsPrimaryKey[Long] {}
+  implicit val floatIsPrimaryKey: IsPrimaryKey[Float]           = new IsPrimaryKey[Float] {}
+  implicit val doubleIsPrimaryKey: IsPrimaryKey[Double]         = new IsPrimaryKey[Double] {}
+  implicit val bigDecimalIsPrimaryKey: IsPrimaryKey[BigDecimal] = new IsPrimaryKey[BigDecimal] {}
+
+  implicit val binaryIsPrimaryKey: IsPrimaryKey[Iterable[Byte]] = new IsPrimaryKey[Iterable[Byte]] {}
+
+  implicit val sortKeyNotUsed: IsPrimaryKey[KeyConditionExpr.SortKeyNotUsed] =
+    new IsPrimaryKey[KeyConditionExpr.SortKeyNotUsed] {}
+}
diff --git a/dynamodb/src/test/scala/zio/dynamodb/AliasMapRenderSpec.scala b/dynamodb/src/test/scala/zio/dynamodb/AliasMapRenderSpec.scala
index 0f32a793d..f49a7d28e 100644
--- a/dynamodb/src/test/scala/zio/dynamodb/AliasMapRenderSpec.scala
+++ b/dynamodb/src/test/scala/zio/dynamodb/AliasMapRenderSpec.scala
@@ -1,6 +1,7 @@
 package zio.dynamodb
 
 import zio.Chunk
+import zio.dynamodb.ProjectionExpression.$
 import zio.dynamodb.ProjectionExpression._
 import zio.test.Assertion._
 import zio.test.{ ZIOSpecDefault, _ }
@@ -318,8 +319,8 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
           }
         )
       ),
-      suite("KeyConditionExpression")(
-        suite("SortKeyExpression")(
+      suite("KeyConditionExpr")(
+        suite("Sort key expressions")(
           test("Equals") {
             val map = Map(
               avKey(one)               -> ":v0",
@@ -328,7 +329,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.Equals(SortKeyExpression.SortKey("num"), one).render.execute
+              KeyConditionExpr.SortKeyEquals($("num").sortKey, one).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 = :v0"))
@@ -341,7 +342,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.LessThan(SortKeyExpression.SortKey("num"), one).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.LessThan($("num").sortKey, one).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 < :v0"))
@@ -354,7 +355,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.NotEqual(SortKeyExpression.SortKey("num"), one).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.NotEqual($("num").sortKey, one).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 <> :v0"))
@@ -367,7 +368,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.GreaterThan(SortKeyExpression.SortKey("num"), one).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.GreaterThan($("num").sortKey, one).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 > :v0"))
@@ -380,7 +381,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.LessThanOrEqual(SortKeyExpression.SortKey("num"), one).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.LessThanOrEqual($("num").sortKey, one).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 <= :v0"))
@@ -393,7 +394,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.GreaterThanOrEqual(SortKeyExpression.SortKey("num"), one).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.GreaterThanOrEqual($("num").sortKey, one).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 >= :v0"))
@@ -407,7 +408,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.Between(SortKeyExpression.SortKey("num"), one, two).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.Between($("num").sortKey, one, two).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 3))) &&
             assert(expression)(equalTo("#n2 BETWEEN :v0 AND :v1"))
@@ -420,7 +421,7 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             )
 
             val (aliasMap, expression) =
-              SortKeyExpression.BeginsWith(SortKeyExpression.SortKey("num"), name).render.execute
+              KeyConditionExpr.ExtendedSortKeyExpr.BeginsWith($("num").sortKey, name).miniRender.execute
 
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("begins_with(#n1, :v0)"))
@@ -434,10 +435,11 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
               fullPath($("num"))       -> "#n1"
             )
 
-            val (aliasMap, expression) = PartitionKeyExpression
-              .Equals(PartitionKeyExpression.PartitionKey("num"), one)
+            val (aliasMap, expression) = KeyConditionExpr
+              .PartitionKeyEquals($("num").partitionKey, one)
               .render
               .execute
+
             assert(aliasMap)(equalTo(AliasMap(map, 2))) &&
             assert(expression)(equalTo("#n1 = :v0"))
           }
@@ -453,11 +455,10 @@ object AliasMapRenderSpec extends ZIOSpecDefault {
             fullPath($("num2"))       -> "#n4"
           )
 
-          val (aliasMap, expression) = KeyConditionExpression
-            .And(
-              PartitionKeyExpression
-                .Equals(PartitionKeyExpression.PartitionKey("num"), two),
-              SortKeyExpression.Between(SortKeyExpression.SortKey("num2"), one, three)
+          val (aliasMap, expression) = KeyConditionExpr
+            .ExtendedCompositePrimaryKeyExpr(
+              KeyConditionExpr.PartitionKeyEquals($("num").partitionKey, two),
+              KeyConditionExpr.ExtendedSortKeyExpr.Between($("num2").sortKey, one, three)
             )
             .render
             .execute
diff --git a/dynamodb/src/test/scala/zio/dynamodb/GetAndPutSpec.scala b/dynamodb/src/test/scala/zio/dynamodb/GetAndPutSpec.scala
index 2ae50087c..124130f50 100644
--- a/dynamodb/src/test/scala/zio/dynamodb/GetAndPutSpec.scala
+++ b/dynamodb/src/test/scala/zio/dynamodb/GetAndPutSpec.scala
@@ -9,9 +9,14 @@ import zio.test.{ assertTrue, ZIOSpecDefault }
 
 object GetAndPutSpec extends ZIOSpecDefault {
   final case class SimpleCaseClass2(id: Int, name: String)
+  object SimpleCaseClass2 {
+    implicit val schema: Schema.CaseClass2[Int, String, SimpleCaseClass2] = DeriveSchema.gen[SimpleCaseClass2]
+    val (id, name)                                                        = ProjectionExpression.accessors[SimpleCaseClass2]
+  }
+
   final case class SimpleCaseClass2OptionalField(id: Int, maybeName: Option[String])
-  implicit lazy val simpleCaseClass2: Schema[SimpleCaseClass2]                           = DeriveSchema.gen[SimpleCaseClass2]
-  implicit lazy val simpleCaseClass2OptionalField: Schema[SimpleCaseClass2OptionalField] =
+  implicit val simpleCaseClass2: Schema[SimpleCaseClass2]                           = DeriveSchema.gen[SimpleCaseClass2]
+  implicit val simpleCaseClass2OptionalField: Schema[SimpleCaseClass2OptionalField] =
     DeriveSchema.gen[SimpleCaseClass2OptionalField]
 
   private val primaryKey1 = PrimaryKey("id" -> 1)
@@ -24,18 +29,18 @@ object GetAndPutSpec extends ZIOSpecDefault {
     test("that exists") {
       for {
         _     <- TestDynamoDBExecutor.addItems("table1", primaryKey1 -> Item("id" -> 1, "name" -> "Avi"))
-        found <- get[SimpleCaseClass2]("table1", primaryKey1).execute
+        found <- get("table1")(SimpleCaseClass2.id.partitionKey === 1).execute
       } yield assertTrue(found == Right(SimpleCaseClass2(1, "Avi")))
     },
     test("that does not exists") {
       for {
-        found <- get[SimpleCaseClass2]("table1", primaryKey1).execute
+        found <- get("table1")(SimpleCaseClass2.id.partitionKey === 1).execute
       } yield assertTrue(found == Left(ValueNotFound("value with key AttrMap(Map(id -> Number(1))) not found")))
     },
     test("with missing attributes results in an error") {
       for {
         _     <- TestDynamoDBExecutor.addItems("table1", primaryKey1 -> Item("id" -> 1))
-        found <- get[SimpleCaseClass2]("table1", primaryKey1).execute
+        found <- get("table1")(SimpleCaseClass2.id.partitionKey === 1).execute
       } yield assertTrue(found == Left(DecodingError("field 'name' not found in Map(Map(String(id) -> Number(1)))")))
     },
     test("batched") {
@@ -45,7 +50,9 @@ object GetAndPutSpec extends ZIOSpecDefault {
                primaryKey1 -> Item("id" -> 1, "name" -> "Avi"),
                primaryKey2 -> Item("id" -> 2, "name" -> "Tarlochan")
              )
-        r <- (get[SimpleCaseClass2]("table1", primaryKey1) zip get[SimpleCaseClass2]("table1", primaryKey2)).execute
+        r <- (get("table1")(SimpleCaseClass2.id.partitionKey === 1) zip get("table1")(
+                 SimpleCaseClass2.id.partitionKey === 2
+               )).execute
       } yield assertTrue(r._1 == Right(SimpleCaseClass2(1, "Avi"))) && assertTrue(
         r._2 == Right(SimpleCaseClass2(2, "Tarlochan"))
       )
@@ -56,15 +63,15 @@ object GetAndPutSpec extends ZIOSpecDefault {
     test("""SimpleCaseClass2(1, "Avi")""") {
       for {
         _     <- put[SimpleCaseClass2]("table1", SimpleCaseClass2(1, "Avi")).execute
-        found <- get[SimpleCaseClass2]("table1", primaryKey1).execute
+        found <- get("table1")(SimpleCaseClass2.id.partitionKey === 1).execute
       } yield assertTrue(found == Right(SimpleCaseClass2(1, "Avi")))
     },
     test("""top level enum PreBilled(1, "foobar")""") {
       for {
         _     <- TestDynamoDBExecutor.addTable("table1", "id")
         _     <- put[Invoice]("table1", PreBilled(1, "foobar")).execute
-        found <- get[Invoice]("table1", primaryKey1).execute
-      } yield assertTrue(found == Right(PreBilled(1, "foobar")))
+        found <- get("table1")(PreBilled.id.partitionKey === 1).execute.absolve
+      } yield assertTrue(found == PreBilled(1, "foobar"))
     },
     test("""batched SimpleCaseClass2(1, "Avi") and SimpleCaseClass2(2, "Tarlochan")""") {
       for {
@@ -72,8 +79,8 @@ object GetAndPutSpec extends ZIOSpecDefault {
                       "table1",
                       SimpleCaseClass2(2, "Tarlochan")
                     )).execute
-        found1 <- get[SimpleCaseClass2]("table1", primaryKey1).execute
-        found2 <- get[SimpleCaseClass2]("table1", primaryKey2).execute
+        found1 <- get("table1")(SimpleCaseClass2.id.partitionKey === 1).execute
+        found2 <- get("table1")(SimpleCaseClass2.id.partitionKey === 2).execute
       } yield assertTrue(found1 == Right(SimpleCaseClass2(1, "Avi"))) && assertTrue(
         found2 == Right(SimpleCaseClass2(2, "Tarlochan"))
       )
diff --git a/dynamodb/src/test/scala/zio/dynamodb/KeyConditionExpressionSpec.scala b/dynamodb/src/test/scala/zio/dynamodb/KeyConditionExpressionSpec.scala
deleted file mode 100644
index 108bca7fa..000000000
--- a/dynamodb/src/test/scala/zio/dynamodb/KeyConditionExpressionSpec.scala
+++ /dev/null
@@ -1,168 +0,0 @@
-package zio.dynamodb
-
-import zio.dynamodb.ConditionExpression.Operand.ProjectionExpressionOperand
-import zio.dynamodb.ProjectionExpression.{ MapElement, Root }
-import zio.schema.DeriveSchema
-import zio.test.Assertion.{ isLeft, isRight }
-import zio.test._
-
-import java.time.Instant
-import zio.schema.Schema
-
-object KeyConditionExpressionSpec extends ZIOSpecDefault {
-
-  final case class Student(
-    email: String,
-    subject: String,
-    enrollmentDate: Option[Instant]
-  )
-  object Student {
-    implicit val schema: Schema.CaseClass3[String, String, Option[Instant], Student] =
-      DeriveSchema.gen[Student]
-  }
-
-  val (email, subject, enrollmentDate) = ProjectionExpression.accessors[Student]
-
-  override def spec =
-    suite("KeyConditionExpression from a ConditionExpression")(happyPathSuite, unhappyPathSuite, pbtSuite)
-
-  val happyPathSuite =
-    suite("returns a Right for")(
-      test(""" email === "avi@gmail.com" """) {
-        val actual = KeyConditionExpression(email === "avi@gmail.com")
-        zio.test.assert(actual)(isRight)
-      },
-      test(""" email === "avi@gmail.com" && subject === "maths" """) {
-        val actual = KeyConditionExpression(email === "avi@gmail.com" && subject === "maths")
-        zio.test.assert(actual)(isRight)
-      },
-      test(""" email === "avi@gmail.com" && subject.beginsWith("ma") """) {
-        val actual =
-          KeyConditionExpression(
-            email === "avi@gmail.com" && subject.beginsWith("ma")
-          )
-        zio.test.assert(actual)(isRight)
-      }
-    )
-
-  val unhappyPathSuite =
-    suite("returns a Left for")(
-      test(""" email > "avi@gmail.com" && subject === "maths" """) {
-        val actual = KeyConditionExpression(email > "avi@gmail.com" && subject === "maths")
-        zio.test.assert(actual)(isLeft)
-      },
-      test(""" email >= "avi@gmail.com" && subject.beginsWith("ma") """) {
-        val actual =
-          KeyConditionExpression(
-            email >= "avi@gmail.com" && subject.beginsWith("ma")
-          )
-        zio.test.assert(actual)(isLeft)
-      },
-      test(""" email === "avi@gmail.com" && subject.beginsWith("ma") && subject.beginsWith("ma") """) {
-        val actual =
-          KeyConditionExpression(
-            email === "avi@gmail.com" && subject.beginsWith("ma") && subject.beginsWith("ma")
-          )
-        zio.test.assert(actual)(isLeft)
-      }
-    )
-
-  import Generators._
-
-  val pbtSuite =
-    suite("property based suite")(test("conversion of ConditionExpression to KeyConditionExpression must be valid") {
-      check(genConditionExpression) {
-        case (condExprn, seedDataList) => assertConditionExpression(condExprn, seedDataList)
-      }
-    })
-
-  def assertConditionExpression(
-    condExprn: ConditionExpression[_],
-    seedDataList: List[SeedData],
-    printCondExprn: Boolean = false
-  ): TestResult = {
-    val errorOrkeyCondExprn: Either[String, KeyConditionExpression] = KeyConditionExpression(condExprn)
-
-    if (printCondExprn) println(condExprn) else ()
-
-    assert(errorOrkeyCondExprn) {
-      if (
-        seedDataList.length == 0 || seedDataList.length > maxNumOfTerms || seedDataList.head._2 != ComparisonOp.Equals
-      )
-        isLeft
-      else
-        isRight
-    }
-  }
-
-  object Generators {
-    sealed trait ComparisonOp
-    object ComparisonOp {
-      case object Equals             extends ComparisonOp
-      case object NotEquals          extends ComparisonOp
-      case object LessThan           extends ComparisonOp
-      case object GreaterThan        extends ComparisonOp
-      case object LessThanOrEqual    extends ComparisonOp
-      case object GreaterThanOrEqual extends ComparisonOp
-
-      val set = Set(Equals, NotEquals, LessThan, GreaterThan, LessThanOrEqual, GreaterThanOrEqual)
-    }
-
-    val maxNumOfTerms                                                   = 2
-    val genOP                                                           = Gen.fromIterable(ComparisonOp.set)
-    private val genFieldName: Gen[Sized, String]                        = Gen.alphaNumericStringBounded(1, 5)
-    val genFieldNameAndOpList: Gen[Sized, List[(String, ComparisonOp)]] =
-      Gen.listOfBounded(0, maxNumOfTerms + 1)( // ensure we generate more that max number of terms
-        genFieldName zip genOP
-      )
-    final case class FieldNameAndComparisonOp(fieldName: String, op: ComparisonOp)
-    type SeedData = (String, ComparisonOp)
-    val genConditionExpression: Gen[Sized, (ConditionExpression[_], List[SeedData])] =
-      genFieldNameAndOpList.filter(_.nonEmpty).map { xs =>
-        val (name, op) = xs.head
-        val first      = conditionExpression(name, op)
-        // TODO: generate joining ops
-        val condEx     = xs.tail.foldRight(first) {
-          case ((name, op), acc) =>
-            acc.asInstanceOf[ConditionExpression[Any]] && conditionExpression(name, op)
-              .asInstanceOf[ConditionExpression[Any]]
-        }
-        (condEx, xs)
-      }
-
-    def conditionExpression(name: String, op: ComparisonOp): ConditionExpression[_] =
-      op match {
-        case ComparisonOp.Equals             =>
-          ConditionExpression.Equals(
-            ProjectionExpressionOperand(MapElement(Root, name)),
-            ConditionExpression.Operand.ValueOperand(AttributeValue.String("SOME_VALUE"))
-          )
-        case ComparisonOp.NotEquals          =>
-          ConditionExpression.NotEqual(
-            ProjectionExpressionOperand(MapElement(Root, name)),
-            ConditionExpression.Operand.ValueOperand(AttributeValue.String("SOME_VALUE"))
-          )
-        case ComparisonOp.LessThan           =>
-          ConditionExpression.LessThan(
-            ProjectionExpressionOperand(MapElement(Root, name)),
-            ConditionExpression.Operand.ValueOperand(AttributeValue.String("SOME_VALUE"))
-          )
-        case ComparisonOp.GreaterThan        =>
-          ConditionExpression.GreaterThan(
-            ProjectionExpressionOperand(MapElement(Root, name)),
-            ConditionExpression.Operand.ValueOperand(AttributeValue.String("SOME_VALUE"))
-          )
-        case ComparisonOp.LessThanOrEqual    =>
-          ConditionExpression.LessThanOrEqual(
-            ProjectionExpressionOperand(MapElement(Root, name)),
-            ConditionExpression.Operand.ValueOperand(AttributeValue.String("SOME_VALUE"))
-          )
-        case ComparisonOp.GreaterThanOrEqual =>
-          ConditionExpression.GreaterThanOrEqual(
-            ProjectionExpressionOperand(MapElement(Root, name)),
-            ConditionExpression.Operand.ValueOperand(AttributeValue.String("SOME_VALUE"))
-          )
-      }
-  }
-
-}
diff --git a/dynamodb/src/test/scala/zio/dynamodb/ZStreamPipeliningSpec.scala b/dynamodb/src/test/scala/zio/dynamodb/ZStreamPipeliningSpec.scala
index f1ed1c08f..ffa1c2703 100644
--- a/dynamodb/src/test/scala/zio/dynamodb/ZStreamPipeliningSpec.scala
+++ b/dynamodb/src/test/scala/zio/dynamodb/ZStreamPipeliningSpec.scala
@@ -11,7 +11,8 @@ object ZStreamPipeliningSpec extends ZIOSpecDefault {
   final case class Person(id: Int, name: String)
 
   object Person {
-    implicit val schema: Schema[Person] = DeriveSchema.gen[Person]
+    implicit val schema: Schema.CaseClass2[Int, String, Person] = DeriveSchema.gen[Person]
+    val (id, name)                                              = ProjectionExpression.accessors[Person]
   }
 
   private val people       = (1 to 200).map(i => Person(i, s"name$i")).toList
@@ -25,9 +26,8 @@ object ZStreamPipeliningSpec extends ZIOSpecDefault {
           _           <- batchWriteFromStream(personStream) { person =>
                            put("person", person)
                          }.runDrain
-          xs          <- batchReadFromStream[Any, Person, Person]("person", personStream)(person =>
-                           PrimaryKey("id" -> person.id)
-                         ).right.runCollect
+          xs          <-
+            batchReadFromStream("person", personStream)(person => Person.id.partitionKey === person.id).right.runCollect
           actualPeople = xs.toList.map { case (_, p) => p }.collect { case Some(b) => b }
         } yield assert(actualPeople)(equalTo(people))
       },
@@ -41,8 +41,8 @@ object ZStreamPipeliningSpec extends ZIOSpecDefault {
                             PrimaryKey("id" -> 1) -> Item("id" -> 1, "name" -> "Avi"),
                             PrimaryKey("id" -> 2) -> Item("id" -> 2, "boom!" -> "de-serialisation-error-expected")
                           )
-          actualPeople <- batchReadFromStream[Any, Person, Person]("person", personStream.take(3))(person =>
-                            PrimaryKey("id" -> person.id)
+          actualPeople <- batchReadFromStream("person", personStream.take(3))(person =>
+                            Person.id.partitionKey === person.id
                           ).runCollect
         } yield assertTrue(
           actualPeople == Chunk(
diff --git a/dynamodb/src/test/scala/zio/dynamodb/codec/models.scala b/dynamodb/src/test/scala/zio/dynamodb/codec/models.scala
index 585a72ce7..b7f0f8c75 100644
--- a/dynamodb/src/test/scala/zio/dynamodb/codec/models.scala
+++ b/dynamodb/src/test/scala/zio/dynamodb/codec/models.scala
@@ -5,6 +5,7 @@ import zio.schema.annotation.{ caseName, discriminatorName, fieldName }
 import zio.schema.{ DeriveSchema, Schema }
 
 import java.time.Instant
+import zio.dynamodb.ProjectionExpression
 
 // ADT example
 sealed trait Status
@@ -95,5 +96,9 @@ sealed trait Invoice {
 object Invoice       {
   final case class Billed(id: Int, i: Int)       extends Invoice
   final case class PreBilled(id: Int, s: String) extends Invoice
+  object PreBilled {
+    implicit val schema: Schema.CaseClass2[Int, String, PreBilled] = DeriveSchema.gen[PreBilled]
+    val (id, s)                                                    = ProjectionExpression.accessors[PreBilled]
+  }
   implicit val schema: Schema[Invoice] = DeriveSchema.gen[Invoice]
 }
diff --git a/examples/src/main/scala/zio/dynamodb/examples/BatchFromStreamExamples.scala b/examples/src/main/scala/zio/dynamodb/examples/BatchFromStreamExamples.scala
index 98fa8ac7e..be703b209 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/BatchFromStreamExamples.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/BatchFromStreamExamples.scala
@@ -11,7 +11,8 @@ object BatchFromStreamExamples extends ZIOAppDefault {
 
   final case class Person(id: Int, name: String)
   object Person {
-    implicit val schema: Schema[Person] = DeriveSchema.gen[Person]
+    implicit val schema: Schema.CaseClass2[Int, String, Person] = DeriveSchema.gen[Person]
+    val (id, name)                                              = ProjectionExpression.accessors[Person]
   }
 
   private val personIdStream: UStream[Int] =
@@ -37,7 +38,7 @@ object BatchFromStreamExamples extends ZIOAppDefault {
              .runDrain
 
       // same again but use Schema derived codecs to convert an Item to a Person
-      _ <- batchReadFromStream[Any, Int, Person]("person", personIdStream)(id => PrimaryKey("id" -> id))
+      _ <- batchReadFromStream("person", personIdStream)(id => Person.id.partitionKey === id)
              .mapZIOPar(4)(person => printLine(s"person=$person"))
              .runDrain
     } yield ()).provide(DynamoDBExecutor.test)
diff --git a/examples/src/main/scala/zio/dynamodb/examples/ConditionExpressionExamples.scala b/examples/src/main/scala/zio/dynamodb/examples/ConditionExpressionExamples.scala
index 091500eb5..ae44587b1 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/ConditionExpressionExamples.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/ConditionExpressionExamples.scala
@@ -24,6 +24,7 @@ object ConditionExpressionExamples {
   val peOpticNum1: ProjectionExpression[Student, Int] = Student.count1
   val peOpticNum2: ProjectionExpression[Student, Int] = Student.count2
   val ceOptic1: ConditionExpression[Student]          = peOpticNum1 > peOpticNum2
+  val ceOptic2: ConditionExpression[Student]          = Student.subject.between("a", "b")
 
   val x: ProjectionExpression[Any, Unknown] = $("col2")
   val xx: Operand.Size[Any, Unknown]        = x.size
diff --git a/examples/src/main/scala/zio/dynamodb/examples/KeyConditionExprExample.scala b/examples/src/main/scala/zio/dynamodb/examples/KeyConditionExprExample.scala
new file mode 100644
index 000000000..e207d487a
--- /dev/null
+++ b/examples/src/main/scala/zio/dynamodb/examples/KeyConditionExprExample.scala
@@ -0,0 +1,60 @@
+package zio.dynamodb.examples
+
+import zio.dynamodb.ProjectionExpression
+
+import zio.schema.Schema
+import zio.schema.DeriveSchema
+import zio.dynamodb.DynamoDBQuery
+
+object KeyConditionExprExample extends App {
+
+  import zio.dynamodb.KeyConditionExpr._
+  import zio.dynamodb.KeyConditionExpr.SortKeyEquals
+  import zio.dynamodb.ProjectionExpression.$
+
+  val x6: CompositePrimaryKeyExpr[Any, ProjectionExpression.Unknown, String]      =
+    $("foo.bar").partitionKey === 1 && $("foo.baz").sortKey === "y"
+  val x7                                                                          = $("foo.bar").partitionKey === 1 && $("foo.baz").sortKey > 1
+  val x8: ExtendedCompositePrimaryKeyExpr[Any, ProjectionExpression.Unknown, Int] =
+    $("foo.bar").partitionKey === 1 && $("foo.baz").sortKey.between(1, 2)
+  val x9                                                                          =
+    $("foo.bar").partitionKey === 1 && $("foo.baz").sortKey.beginsWith(1L)
+
+  final case class Elephant(email: String, subject: String, age: Int)
+  object Elephant {
+    implicit val schema: Schema.CaseClass3[String, String, Int, Elephant] = DeriveSchema.gen[Elephant]
+    val (email, subject, age)                                             = ProjectionExpression.accessors[Elephant]
+  }
+  final case class Student(email: String, subject: String, age: Long, binary: List[Byte], binary2: Vector[Byte])
+  object Student  {
+    implicit val schema: Schema.CaseClass5[String, String, Long, List[Byte], Vector[Byte], Student] =
+      DeriveSchema.gen[Student]
+    val (email, subject, age, binary, binary2)                                                      = ProjectionExpression.accessors[Student]
+  }
+
+  val pk: PartitionKeyEquals[Student, String]                   = Student.email.partitionKey === "x"
+//  val pkX: PartitionKeyExpr[Student, String]     = Student.age.primaryKey === "x" // as expected does not compile
+  val sk1: SortKeyEquals[Student, String]                       = Student.subject.sortKey === "y"
+  val sk2: ExtendedSortKeyExpr[Student, String]                 = Student.subject.sortKey > "y"
+  val pkAndSk: CompositePrimaryKeyExpr[Student, String, String] =
+    Student.email.partitionKey === "x" && Student.subject.sortKey === "y"
+
+  //val three = Student.email.primaryKey === "x" && Student.subject.sortKey === "y" && Student.subject.sortKey // 3 terms not allowed
+  val pkAndSkExtended1 =
+    Student.email.partitionKey === "x" && Student.subject.sortKey > "y"
+  val pkAndSkExtended2 =
+    Student.email.partitionKey === "x" && Student.subject.sortKey < "y"
+  val pkAndSkExtended3 =
+    Student.email.partitionKey === "x" && Student.subject.sortKey.between("1", "2")
+  val pkAndSkExtended4 =
+    Student.email.partitionKey === "x" && Student.subject.sortKey.beginsWith("1")
+  val pkAndSkExtended5 =
+    Student.email.partitionKey === "x" && Student.binary.sortKey.beginsWith(List(1.toByte))
+  val pkAndSkExtended6 =
+    Student.email.partitionKey === "x" && Student.binary2.sortKey.beginsWith(List(1.toByte))
+
+  val (aliasMap, s) = pkAndSkExtended1.render.execute
+  println(s"aliasMap=$aliasMap, s=$s")
+
+  val get = DynamoDBQuery.queryAllItem("table").whereKey($("foo.bar").partitionKey === 1 && $("foo.baz").sortKey > 1)
+}
diff --git a/examples/src/main/scala/zio/dynamodb/examples/KeyConditionExpressionExamples.scala b/examples/src/main/scala/zio/dynamodb/examples/KeyConditionExpressionExamples.scala
deleted file mode 100644
index a490b6b2e..000000000
--- a/examples/src/main/scala/zio/dynamodb/examples/KeyConditionExpressionExamples.scala
+++ /dev/null
@@ -1,15 +0,0 @@
-package zio.dynamodb.examples
-
-import zio.dynamodb.PartitionKeyExpression._
-import zio.dynamodb.SortKeyExpression._
-import zio.dynamodb._
-
-object KeyConditionExpressionExamples extends App {
-
-  val exprn1: KeyConditionExpression = PartitionKey("partitionKey1") === "x"
-
-  val exprn2: KeyConditionExpression = PartitionKey("partitionKey1") === "x" && SortKey("sortKey1") > "X"
-
-  val exprn3: KeyConditionExpression = PartitionKey("partitionKey1") === "x" && SortKey("sortKey1") === "X"
-
-}
diff --git a/examples/src/main/scala/zio/dynamodb/examples/Main.scala b/examples/src/main/scala/zio/dynamodb/examples/Main.scala
index 533e6cabc..d3c1f7c2e 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/Main.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/Main.scala
@@ -3,21 +3,24 @@ package zio.dynamodb.examples
 import zio.aws.core.config
 import zio.aws.{ dynamodb, netty }
 import zio.dynamodb.DynamoDBQuery.{ get, put }
-import zio.dynamodb.{ DynamoDBExecutor, PrimaryKey }
+import zio.dynamodb.{ DynamoDBExecutor }
 import zio.schema.{ DeriveSchema, Schema }
 import zio.ZIOAppDefault
+import zio.dynamodb.ProjectionExpression
 
 object Main extends ZIOAppDefault {
 
   final case class Person(id: Int, firstName: String)
   object Person {
-    implicit lazy val schema: Schema[Person] = DeriveSchema.gen[Person]
+    implicit lazy val schema: Schema.CaseClass2[Int, String, Person] = DeriveSchema.gen[Person]
+
+    val (id, firstName) = ProjectionExpression.accessors[Person]
   }
   val examplePerson = Person(1, "avi")
 
   private val program = for {
     _      <- put("personTable", examplePerson).execute
-    person <- get[Person]("personTable", PrimaryKey("id" -> 1)).execute
+    person <- get("personTable")(Person.id.partitionKey === 1).execute
     _      <- zio.Console.printLine(s"hello $person")
   } yield ()
 
diff --git a/examples/src/main/scala/zio/dynamodb/examples/QueryAndScanExamples.scala b/examples/src/main/scala/zio/dynamodb/examples/QueryAndScanExamples.scala
index 0c70dffbf..6e09340d9 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/QueryAndScanExamples.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/QueryAndScanExamples.scala
@@ -1,9 +1,7 @@
 package zio.dynamodb.examples
 
 import zio.dynamodb.DynamoDBQuery._
-import zio.dynamodb.PartitionKeyExpression.PartitionKey
 import zio.dynamodb.ProjectionExpression.$
-import zio.dynamodb.SortKeyExpression.SortKey
 import zio.dynamodb._
 import zio.stream.Stream
 import zio.{ Chunk, ZIO }
@@ -19,15 +17,15 @@ object QueryAndScanExamples extends App {
   val queryAll: ZIO[DynamoDBExecutor, Throwable, Stream[Throwable, Item]] =
     queryAllItem("tableName1", $("A"), $("B"), $("C"))
       .whereKey(
-        PartitionKey("partitionKey1") === "x" &&
-          SortKey("sortKey1") > "X"
+        $("partitionKey1").partitionKey === "x" &&
+          $("sortKey1").sortKey > "X"
       )
       .execute
 
   val querySome: ZIO[DynamoDBExecutor, Throwable, (Chunk[Item], LastEvaluatedKey)] =
     querySomeItem("tableName1", limit = 10, $("A"), $("B"), $("C"))
       .sortOrder(ascending = false)
-      .whereKey(PartitionKey("partitionKey1") === "x" && SortKey("sortKey1") > "X")
+      .whereKey($("partitionKey1").partitionKey === "x" && $("sortKey1").sortKey > "X")
       .selectCount
       .execute
 
@@ -40,7 +38,7 @@ object QueryAndScanExamples extends App {
         $("B"),
         $("C")
       )
-        .whereKey(PartitionKey("partitionKey1") === "x" && SortKey("sortKey1") > "X")
+        .whereKey($("partitionKey1").partitionKey === "x" && $("sortKey1").sortKey > "X")
         .selectCount).sortOrder(ascending = true)
 
 }
diff --git a/examples/src/main/scala/zio/dynamodb/examples/SimpleDecodedExample.scala b/examples/src/main/scala/zio/dynamodb/examples/SimpleDecodedExample.scala
index c3b1b3be2..95830f595 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/SimpleDecodedExample.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/SimpleDecodedExample.scala
@@ -1,17 +1,22 @@
 package zio.dynamodb.examples
 
 import zio.dynamodb.DynamoDBQuery._
-import zio.dynamodb.{ DynamoDBExecutor, Item, PrimaryKey }
+import zio.dynamodb.{ DynamoDBExecutor, Item }
 import zio.schema.{ DeriveSchema, Schema }
 import zio.ZIOAppDefault
 import zio.Console.printLine
+import zio.dynamodb.ProjectionExpression
 
 object SimpleDecodedExample extends ZIOAppDefault {
   val nestedItem       = Item("id" -> 2, "name" -> "Avi", "flag" -> true)
   val parentItem: Item = Item("id" -> 1, "nested" -> nestedItem)
 
   final case class NestedCaseClass2(id: Int, nested: SimpleCaseClass3)
-  implicit lazy val nestedCaseClass2: Schema[NestedCaseClass2] = DeriveSchema.gen[NestedCaseClass2]
+  object NestedCaseClass2 {
+    implicit val nestedCaseClass2: Schema.CaseClass2[Int, SimpleCaseClass3, NestedCaseClass2] =
+      DeriveSchema.gen[NestedCaseClass2]
+    val (id, nested)                                                                          = ProjectionExpression.accessors[NestedCaseClass2]
+  }
 
   final case class SimpleCaseClass3(id: Int, name: String, flag: Boolean)
   implicit lazy val simpleCaseClass3: Schema[SimpleCaseClass3] = DeriveSchema.gen[SimpleCaseClass3]
@@ -19,7 +24,7 @@ object SimpleDecodedExample extends ZIOAppDefault {
   private val program = for {
     _         <-
       put("table1", NestedCaseClass2(id = 1, SimpleCaseClass3(2, "Avi", flag = true))).execute // Save case class to DB
-    caseClass <- get[NestedCaseClass2]("table1", PrimaryKey("id" -> 1)).execute // read case class from DB
+    caseClass <- get("table1")(NestedCaseClass2.id.partitionKey === 2).execute // read case class from DB
     _         <- printLine(s"get: found $caseClass")
     either    <- scanSome[NestedCaseClass2]("table1", 10).execute
     _         <- printLine(s"scanSome: found $either")
diff --git a/examples/src/main/scala/zio/dynamodb/examples/TypeSafeRoundTripSerialisationExample.scala b/examples/src/main/scala/zio/dynamodb/examples/TypeSafeRoundTripSerialisationExample.scala
index 0f095b0d7..16765a069 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/TypeSafeRoundTripSerialisationExample.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/TypeSafeRoundTripSerialisationExample.scala
@@ -8,6 +8,7 @@ import zio.dynamodb.examples.TypeSafeRoundTripSerialisationExample.Invoice.{
   Billed,
   LineItem,
   PaymentType,
+  PreBilled,
   Product
 }
 import zio.dynamodb.{ DynamoDBExecutor, DynamoDBQuery, PrimaryKey }
@@ -15,6 +16,9 @@ import zio.schema.annotation.{ caseName, discriminatorName }
 import zio.schema.{ DeriveSchema, Schema }
 
 import java.time.Instant
+import zio.dynamodb.ProjectionExpression
+import zio.ZIO
+import zio.dynamodb.DynamoDBError
 
 object TypeSafeRoundTripSerialisationExample extends ZIOAppDefault {
 
@@ -48,17 +52,31 @@ object TypeSafeRoundTripSerialisationExample extends ZIOAppDefault {
       lineItems: List[LineItem],
       paymentType: PaymentType
     ) extends Invoice
+    object Billed {
+      implicit val schema: Schema.CaseClass10[String, Int, Instant, BigDecimal, Boolean, Map[String, String], Set[
+        String
+      ], Option[Address], List[LineItem], PaymentType, Billed] = DeriveSchema.gen[Billed]
+
+      val (id, sequence, dueDate, total, isTest, categoryMap, accountSet, address, lineItems, paymentType) =
+        ProjectionExpression.accessors[Billed]
+    }
+
     final case class PreBilled(
       id: String,
       sequence: Int,
       dueDate: Instant,
       total: BigDecimal
     ) extends Invoice
+    object PreBilled {
+      implicit val schema: Schema.CaseClass4[String, Int, Instant, BigDecimal, PreBilled] =
+        DeriveSchema.gen[PreBilled]
+      val (id, sequence, dueDate, total)                                                  = ProjectionExpression.accessors[PreBilled]
+    }
 
     implicit val schema: Schema[Invoice] = DeriveSchema.gen[Invoice]
   }
 
-  private val invoice1 = Billed(
+  private val billedInvoice: Billed               = Billed(
     id = "1",
     sequence = 1,
     dueDate = Instant.now(),
@@ -73,13 +91,36 @@ object TypeSafeRoundTripSerialisationExample extends ZIOAppDefault {
     ),
     PaymentType.DebitCard
   )
+  private val preBilledInvoice: Invoice.PreBilled = Invoice.PreBilled(
+    id = "2",
+    sequence = 2,
+    dueDate = Instant.now(),
+    total = BigDecimal(20.0)
+  )
+
+  import zio.dynamodb.KeyConditionExpr
+
+  object Repository {
+    def genericFindById[A <: Invoice](
+      pkExpr: KeyConditionExpr.PartitionKeyEquals[A, String]
+    )(implicit ev: Schema[A]): ZIO[DynamoDBExecutor, Throwable, Either[DynamoDBError, Invoice]] =
+      DynamoDBQuery.get("table1")(pkExpr).execute
+
+    def genericSave[A <: Invoice](
+      invoice: A
+    )(implicit ev: Schema[A]): ZIO[DynamoDBExecutor, Throwable, Option[Invoice]] =
+      DynamoDBQuery.put("table1", invoice).execute
+  }
 
   private val program = for {
-    _     <- DynamoDBQuery.put[Invoice]("table1", invoice1).execute
-    found <- DynamoDBQuery.get[Invoice]("table1", PrimaryKey("id" -> "1")).execute
-    item  <- DynamoDBQuery.getItem("table1", PrimaryKey("id" -> "1")).execute
-    _     <- printLine(s"found=$found")
-    _     <- printLine(s"item=$item")
+    _      <- Repository.genericSave(billedInvoice)
+    _      <- Repository.genericSave(preBilledInvoice)
+    found  <- Repository.genericFindById(Billed.id.partitionKey === "1")
+    found2 <- Repository.genericFindById(PreBilled.id.partitionKey === "2")
+    item   <- DynamoDBQuery.getItem("table1", PrimaryKey("id" -> "1")).execute
+    _      <- printLine(s"found=$found")
+    _      <- printLine(s"found2=$found2")
+    _      <- printLine(s"item=$item")
   } yield ()
 
   override def run =
diff --git a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbExampleWithOptics.scala b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbExampleWithOptics.scala
index f5b088856..514e3fa3b 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbExampleWithOptics.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbExampleWithOptics.scala
@@ -39,7 +39,7 @@ object StudentZioDynamoDbExampleWithOptics extends ZIOAppDefault {
            .filter(
              enrollmentDate === Some(enrolDate) && payment === Payment.CreditCard
            )
-           .whereKey(email === "avi@gmail.com" && subject === "maths")
+           .whereKey(email.partitionKey === "avi@gmail.com" && subject.sortKey === "maths")
            .execute
            .map(_.runCollect)
     _ <- put[Student]("student", avi)
@@ -49,17 +49,17 @@ object StudentZioDynamoDbExampleWithOptics extends ZIOAppDefault {
              ) && email === "avi@gmail.com" && payment === Payment.CreditCard
            )
            .execute
-    _ <- update[Student]("student", primaryKey("avi@gmail.com", "maths")) {
+    _ <- update("student")(primaryKey("avi@gmail.com", "maths")) {
            enrollmentDate.set(Some(enrolDate2)) + payment.set(Payment.PayPal) + address
              .set(
                Some(Address("line1", "postcode1"))
              )
          }.execute
-    _ <- delete("student", primaryKey("adam@gmail.com", "english"))
+    _ <- delete("student")(primaryKey("adam@gmail.com", "english"))
            .where(
              enrollmentDate === Some(
                enrolDate
-             ) && payment === Payment.CreditCard // && zio.dynamodb.examples.Elephant.email === "elephant@gmail.com"
+             ) && payment === Payment.CreditCard // && zio.dynamodb.amples.Elephant.email === "elephant@gmail.com"
            )
            .execute
     _ <- scanAll[Student]("student").execute
diff --git a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbTypeSafeAPIExample.scala b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbTypeSafeAPIExample.scala
index 20426445f..3e093f03a 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbTypeSafeAPIExample.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/StudentZioDynamoDbTypeSafeAPIExample.scala
@@ -46,7 +46,7 @@ object StudentZioDynamoDbTypeSafeAPIExample extends ZIOAppDefault {
            .filter(
              enrollmentDate === Some(enrolDate) && payment === Payment.PayPal //&& elephantCe
            )
-           .whereKey(email === "avi@gmail.com" && subject === "maths" /* && elephantCe */ )
+           .whereKey(email.partitionKey === "avi@gmail.com" && subject.sortKey === "maths" /* && elephantCe */ )
            .execute
            .map(_.runCollect)
     _ <- put[Student]("student", avi)
@@ -58,27 +58,27 @@ object StudentZioDynamoDbTypeSafeAPIExample extends ZIOAppDefault {
              ) === "CreditCard" /* && elephantCe */
            )
            .execute
-    _ <- update[Student]("student", primaryKey("avi@gmail.com", "maths")) {
+    _ <- update("student")(primaryKey("avi@gmail.com", "maths")) {
            altPayment.set(Payment.PayPal) + addresses.prependList(List(Address("line0", "postcode0"))) + studentNumber
              .add(1000) + groups.addSet(Set("group3")) // + elephantAction
          }.execute
 
-    _ <- update[Student]("student", primaryKey("avi@gmail.com", "maths")) {
+    _ <- update("student")(primaryKey("avi@gmail.com", "maths")) {
            altPayment.set(Payment.PayPal) + addresses.appendList(List(Address("line3", "postcode3"))) + groups
              .deleteFromSet(Set("group1"))
          }.execute
 
-    _ <- update[Student]("student", primaryKey("avi@gmail.com", "maths")) {
+    _ <- update("student")(primaryKey("avi@gmail.com", "maths")) {
            enrollmentDate.setIfNotExists(Some(enrolDate2)) + payment.set(altPayment) + address
              .set(
                Some(Address("line1", "postcode1"))
              ) // + elephantAction
          }.execute
-    _ <- update[Student]("student", primaryKey("avi@gmail.com", "maths")) {
+    _ <- update("student")(primaryKey("avi@gmail.com", "maths")) {
            addresses.remove(1)
          }.execute
     _ <-
-      delete("student", primaryKey("adam@gmail.com", "english"))
+      delete("student")(primaryKey("adam@gmail.com", "english"))
         .where(
           (enrollmentDate === Some(enrolDate) && payment <> Payment.PayPal && studentNumber
             .between(1, 3) && groups.contains("group1") && collegeName.contains(
diff --git a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithDiscriminator.scala b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithDiscriminator.scala
index 18efd1049..1c2a8c4f2 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithDiscriminator.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithDiscriminator.scala
@@ -58,7 +58,7 @@ object TypeSafeAPIExampleWithDiscriminator extends ZIOAppDefault {
     _         <- put[Box]("box", boxOfGreen).execute
     _         <- put[Box]("box", boxOfAmber).execute
     query      = queryAll[Box]("box")
-                   .whereKey(Box.id === 1)
+                   .whereKey(Box.id.partitionKey === 1)
                    .filter(Box.trafficLightColour >>> TrafficLight.green >>> Green.rgb === 1)
     stream    <- query.execute
     list      <- stream.runCollect
diff --git a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithoutDiscriminator.scala b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithoutDiscriminator.scala
index b9fb85f35..2408a81a7 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithoutDiscriminator.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/dynamodblocal/TypeSafeAPIExampleWithoutDiscriminator.scala
@@ -49,7 +49,7 @@ object TypeSafeAPIExampleWithoutDiscriminator extends ZIOAppDefault {
     _         <- put[Box]("box", boxOfGreen).execute
     _         <- put[Box]("box", boxOfAmber).execute
     query      = queryAll[Box]("box")
-                   .whereKey(Box.id === 1)
+                   .whereKey(Box.id.partitionKey === 1)
                    .filter(Box.trafficLightColour >>> TrafficLight.green >>> Green.rgb === 1)
     stream    <- query.execute
     list      <- stream.runCollect
diff --git a/examples/src/main/scala/zio/dynamodb/examples/model/Student.scala b/examples/src/main/scala/zio/dynamodb/examples/model/Student.scala
index 8bbb418e1..99cc7c8fe 100644
--- a/examples/src/main/scala/zio/dynamodb/examples/model/Student.scala
+++ b/examples/src/main/scala/zio/dynamodb/examples/model/Student.scala
@@ -1,11 +1,12 @@
 package zio.dynamodb.examples.model
 
 import zio.dynamodb.Annotations.enumOfCaseObjects
-import zio.dynamodb.{ PrimaryKey, ProjectionExpression }
+import zio.dynamodb.ProjectionExpression
 import zio.schema.DeriveSchema
 
 import java.time.Instant
 import zio.schema.Schema
+import zio.dynamodb.KeyConditionExpr
 
 @enumOfCaseObjects
 sealed trait Payment
@@ -76,7 +77,8 @@ object Student {
   ) =
     ProjectionExpression.accessors[Student]
 
-  def primaryKey(email: String, subject: String): PrimaryKey = PrimaryKey("email" -> email, "subject" -> subject)
+  def primaryKey(email: String, subject: String): KeyConditionExpr.PrimaryKeyExpr[Student, String, String] =
+    Student.email.partitionKey === email && Student.subject.sortKey === subject
 
   val enrolDate  = Instant.parse("2021-03-20T01:39:33Z")
   val enrolDate2 = Instant.parse("2022-03-20T01:39:33Z")