diff --git a/bats/index.bats b/bats/index.bats
index c147098c6a..c51aa22742 100644
--- a/bats/index.bats
+++ b/bats/index.bats
@@ -713,7 +713,57 @@ SQL
     [[ "${#lines[@]}" = "1" ]] || false
 }
 
-@test "index: SELECT = Full Match" {
+@test "index: SELECT = Primary Key" {
+    dolt sql <<SQL
+INSERT INTO onepk VALUES (1, 99, 51), (2, 11, 55), (3, 88, 52), (4, 22, 54), (5, 77, 53);
+INSERT INTO twopk VALUES (1, 99, 51, 63), (2, 11, 55, 64), (3, 88, 52, 61), (4, 22, 54, 65), (5, 77, 53, 61);
+SQL
+    # found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 = 5" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "$output" =~ "5,77,53" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    # not found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 = 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found partial pk
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 = 2" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    # not found partial pk
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 = 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 = 5 AND pk2 = 77" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    # not found key 1
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 = 999 AND pk2 = 22" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key 2
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 = 1 AND pk2 = 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key mismatch
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 = 88 AND pk2 = 3" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "index: SELECT = Secondary Index" {
     dolt sql <<SQL
 CREATE INDEX idx_v1 ON onepk(v1);
 CREATE INDEX idx_v ON twopk(v2, v1);
@@ -742,6 +792,17 @@ SQL
     [ "$status" -eq "0" ]
     [[ "$output" =~ "pk1,v1,v2" ]] || false
     [[ "${#lines[@]}" = "1" ]] || false
+    # found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 = 64" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    # not found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 = 111" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
     # found
     run dolt sql -q "SELECT * FROM twopk WHERE v2 = 61 AND v1 = 53" -r=csv
     [ "$status" -eq "0" ]
@@ -776,7 +837,61 @@ SQL
     [[ "${#lines[@]}" = "1" ]] || false
 }
 
-@test "index: SELECT > Full Match" {
+@test "index: SELECT > Primary Key" {
+    dolt sql <<SQL
+INSERT INTO onepk VALUES (1, 99, 51), (2, 11, 55), (3, 88, 52), (4, 22, 54), (5, 77, 53);
+INSERT INTO twopk VALUES (1, 99, 51, 63), (2, 11, 55, 64), (3, 88, 52, 61), (4, 22, 54, 65), (5, 77, 53, 61);
+SQL
+    # found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 > 2" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "$output" =~ "3,88,52" ]] || false
+    [[ "$output" =~ "4,22,54" ]] || false
+    [[ "$output" =~ "5,77,53" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 > 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found partial pk
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 > 2" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "$output" =~ "4,22,54,65" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found partial pk
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 > 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 > 4 AND pk2 > 22" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    # not found key 1
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 > 999 AND pk2 > 11" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key 2
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 > 2 AND pk2 > 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key mismatch
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 > 3 AND pk2 > 99" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "index: SELECT > Secondary Index" {
     dolt sql <<SQL
 CREATE INDEX idx_v1 ON onepk(v1);
 CREATE INDEX idx_v ON twopk(v2, v1);
@@ -808,6 +923,18 @@ SQL
     [ "$status" -eq "0" ]
     [[ "$output" =~ "pk1,v1,v2" ]] || false
     [[ "${#lines[@]}" = "1" ]] || false
+    # found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 > 63" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "$output" =~ "4,22,54,65" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+    # not found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 > 111" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
     # found
     run dolt sql -q "SELECT * FROM twopk WHERE v2 > 61 AND v1 > 53" -r=csv
     [ "$status" -eq "0" ]
@@ -844,7 +971,60 @@ SQL
     [[ "${#lines[@]}" = "1" ]] || false
 }
 
-@test "index: SELECT < Full Match" {
+@test "index: SELECT < Primary Key" {
+    dolt sql <<SQL
+INSERT INTO onepk VALUES (1, 99, 51), (2, 11, 55), (3, 88, 52), (4, 22, 54), (5, 77, 53);
+INSERT INTO twopk VALUES (1, 99, 51, 63), (2, 11, 55, 64), (3, 88, 52, 61), (4, 22, 54, 65), (5, 77, 53, 61);
+SQL
+    # found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 < 3" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "$output" =~ "1,99,51" ]] || false
+    [[ "$output" =~ "2,11,55" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+    # not found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 < 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found partial key
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 < 4" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "1,99,51,63" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found partial key
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 < 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 < 3 AND pk2 < 99" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "${#lines[@]}" = "2" ]] || false
+    # not found key 1
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 < 0 AND pk2 < 77" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key 2
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 < 3 AND pk2 < 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key mismatch
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 < 2 AND pk2 < 22" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "index: SELECT < Secondary Index" {
     dolt sql <<SQL
 CREATE INDEX idx_v1 ON onepk(v1);
 CREATE INDEX idx_v ON twopk(v2, v1);
@@ -877,6 +1057,19 @@ SQL
     [ "$status" -eq "0" ]
     [[ "$output" =~ "pk1,v1,v2" ]] || false
     [[ "${#lines[@]}" = "1" ]] || false
+    # found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 < 64" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "$output" =~ "1,99,51,63" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 < 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
     # found
     run dolt sql -q "SELECT * FROM twopk WHERE v2 < 64 AND v1 < 53" -r=csv
     [ "$status" -eq "0" ]
@@ -913,7 +1106,64 @@ SQL
     [[ "${#lines[@]}" = "1" ]] || false
 }
 
-@test "index: SELECT >= Full Match" {
+@test "index: SELECT >= Primary Key" {
+    dolt sql <<SQL
+INSERT INTO onepk VALUES (1, 99, 51), (2, 11, 55), (3, 88, 52), (4, 22, 54), (5, 77, 53);
+INSERT INTO twopk VALUES (1, 99, 51, 63), (2, 11, 55, 64), (3, 88, 52, 61), (4, 22, 54, 65), (5, 77, 53, 61);
+SQL
+    # found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 >= 2" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "$output" =~ "2,11,55" ]] || false
+    [[ "$output" =~ "3,88,52" ]] || false
+    [[ "$output" =~ "4,22,54" ]] || false
+    [[ "$output" =~ "5,77,53" ]] || false
+    [[ "${#lines[@]}" = "5" ]] || false
+    # not found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 >= 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found partial pk
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 >= 2" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "$output" =~ "4,22,54,65" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "${#lines[@]}" = "5" ]] || false
+    # not found partial pk
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 >= 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 >= 4 AND pk2 >= 22" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "4,22,54,65" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "${#lines[@]}" = "3" ]] || false
+    # not found key 1
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 >= 999 AND pk2 >= 11" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key 2
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 >= 2 AND pk2 >= 999" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key mismatch
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 >= 4 AND pk2 >= 88" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "index: SELECT >= Secondary Index" {
     dolt sql <<SQL
 CREATE INDEX idx_v1 ON onepk(v1);
 CREATE INDEX idx_v ON twopk(v2, v1);
@@ -946,6 +1196,19 @@ SQL
     [ "$status" -eq "0" ]
     [[ "$output" =~ "pk1,v1,v2" ]] || false
     [[ "${#lines[@]}" = "1" ]] || false
+    # found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 >= 63" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "1,99,51,63" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "$output" =~ "4,22,54,65" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 >= 111" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
     # found
     run dolt sql -q "SELECT * FROM twopk WHERE v2 >= 61 AND v1 >= 53" -r=csv
     [ "$status" -eq "0" ]
@@ -984,7 +1247,64 @@ SQL
     [[ "${#lines[@]}" = "1" ]] || false
 }
 
-@test "index: SELECT <= Full Match" {
+@test "index: SELECT <= Primary Key" {
+    dolt sql <<SQL
+INSERT INTO onepk VALUES (1, 99, 51), (2, 11, 55), (3, 88, 52), (4, 22, 54), (5, 77, 53);
+INSERT INTO twopk VALUES (1, 99, 51, 63), (2, 11, 55, 64), (3, 88, 52, 61), (4, 22, 54, 65), (5, 77, 53, 61);
+SQL
+    # found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 <= 3" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "$output" =~ "1,99,51" ]] || false
+    [[ "$output" =~ "2,11,55" ]] || false
+    [[ "$output" =~ "3,88,52" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found
+    run dolt sql -q "SELECT * FROM onepk WHERE pk1 <= 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found partial key
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 <= 4" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "1,99,51,63" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "$output" =~ "4,22,54,65" ]] || false
+    [[ "${#lines[@]}" = "5" ]] || false
+    # not found partial key
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 <= 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # found
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 <= 3 AND pk2 <= 99" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "1,99,51,63" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "${#lines[@]}" = "4" ]] || false
+    # not found key 1
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 <= 0 AND pk2 <= 77" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key 2
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 <= 3 AND pk2 <= 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+    # not found key mismatch
+    run dolt sql -q "SELECT * FROM twopk WHERE pk1 <= 1 AND pk2 <= 88" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
+}
+
+@test "index: SELECT <= Secondary Index" {
     dolt sql <<SQL
 CREATE INDEX idx_v1 ON onepk(v1);
 CREATE INDEX idx_v ON twopk(v2, v1);
@@ -1019,6 +1339,20 @@ SQL
     [ "$status" -eq "0" ]
     [[ "$output" =~ "pk1,v1,v2" ]] || false
     [[ "${#lines[@]}" = "1" ]] || false
+    # found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 <= 64" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "5,77,53,61" ]] || false
+    [[ "$output" =~ "3,88,52,61" ]] || false
+    [[ "$output" =~ "1,99,51,63" ]] || false
+    [[ "$output" =~ "2,11,55,64" ]] || false
+    [[ "${#lines[@]}" = "5" ]] || false
+    # not found partial index
+    run dolt sql -q "SELECT * FROM twopk WHERE v2 <= 0" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,pk2,v1,v2" ]] || false
+    [[ "${#lines[@]}" = "1" ]] || false
     # found
     run dolt sql -q "SELECT * FROM twopk WHERE v2 <= 64 AND v1 <= 53" -r=csv
     [ "$status" -eq "0" ]
@@ -1057,6 +1391,38 @@ SQL
     [[ "${#lines[@]}" = "1" ]] || false
 }
 
+@test "index: EXPLAIN SELECT = IndexedJoin" {
+    dolt sql <<SQL
+CREATE INDEX idx_v1 ON onepk(v1);
+CREATE INDEX idx_v ON twopk(v2, v1);
+INSERT INTO onepk VALUES (1, 11, 111), (2, 22, 222), (3, 33, 333), (4, 44, 444), (5, 55, 555);
+INSERT INTO twopk VALUES (5, 95, 222, 11), (4, 4, 333, 55), (3, 93, 444, 33), (2, 92, 111, 22), (1, 91, 555, 44);
+SQL
+    run dolt sql -q "SELECT * FROM onepk JOIN twopk ON onepk.v1 = twopk.v2;" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk1,v1,v2,pk1,pk2,v1,v2" ]] || false
+    [[ "$output" =~ "1,11,111,5,95,222,11" ]] || false
+    [[ "$output" =~ "2,22,222,2,92,111,22" ]] || false
+    [[ "$output" =~ "3,33,333,3,93,444,33" ]] || false
+    [[ "$output" =~ "4,44,444,1,91,555,44" ]] || false
+    [[ "$output" =~ "5,55,555,4,4,333,55" ]] || false
+    [[ "${#lines[@]}" = "6" ]] || false
+    run dolt sql -q "EXPLAIN SELECT * FROM onepk JOIN twopk ON onepk.v1 = twopk.v2;"
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "IndexedJoin(onepk.v1 = twopk.v2)" ]] || false
+    run dolt sql -q "SELECT * FROM onepk JOIN twopk ON onepk.pk1 = twopk.pk1;" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "1,11,111,1,91,555,44" ]] || false
+    [[ "$output" =~ "2,22,222,2,92,111,22" ]] || false
+    [[ "$output" =~ "3,33,333,3,93,444,33" ]] || false
+    [[ "$output" =~ "4,44,444,4,4,333,55" ]] || false
+    [[ "$output" =~ "5,55,555,5,95,222,11" ]] || false
+    [[ "${#lines[@]}" = "6" ]] || false
+    run dolt sql -q "EXPLAIN SELECT * FROM onepk JOIN twopk ON onepk.pk1 = twopk.pk1;"
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "IndexedJoin(onepk.pk1 = twopk.pk1)" ]] || false
+}
+
 @test "index: ALTER TABLE ADD COLUMN" {
     dolt sql <<SQL
 CREATE INDEX idx_v1 ON onepk(v1);
diff --git a/go/libraries/doltcore/schema/index.go b/go/libraries/doltcore/schema/index.go
index ce07821672..61f51c2c69 100644
--- a/go/libraries/doltcore/schema/index.go
+++ b/go/libraries/doltcore/schema/index.go
@@ -22,6 +22,8 @@ type Index interface {
 	ColumnNames() []string
 	// Comment returns the comment that was provided upon index creation.
 	Comment() string
+	// Count returns the number of indexed columns in this index.
+	Count() int
 	// GetColumn returns the column for the given tag and whether the column was found or not.
 	GetColumn(tag uint64) (Column, bool)
 	// IndexedColumnTags returns the tags of the columns in the index.
@@ -63,6 +65,10 @@ func (ix *indexImpl) Comment() string {
 	return ix.comment
 }
 
+func (ix *indexImpl) Count() int {
+	return len(ix.tags)
+}
+
 func (ix *indexImpl) GetColumn(tag uint64) (Column, bool) {
 	return ix.indexColl.colColl.GetByTag(tag)
 }
diff --git a/go/libraries/doltcore/sqle/dolt_index.go b/go/libraries/doltcore/sqle/dolt_index.go
index 5149803305..c4c5f41328 100644
--- a/go/libraries/doltcore/sqle/dolt_index.go
+++ b/go/libraries/doltcore/sqle/dolt_index.go
@@ -16,13 +16,13 @@ package sqle
 
 import (
 	"errors"
-	"fmt"
 
 	"github.com/liquidata-inc/go-mysql-server/sql"
+	"github.com/liquidata-inc/go-mysql-server/sql/expression"
 
 	"github.com/liquidata-inc/dolt/go/libraries/doltcore/doltdb"
-	"github.com/liquidata-inc/dolt/go/libraries/doltcore/row"
 	"github.com/liquidata-inc/dolt/go/libraries/doltcore/schema"
+	"github.com/liquidata-inc/dolt/go/store/types"
 )
 
 /*
@@ -47,106 +47,17 @@ type DoltIndex interface {
 	Schema() schema.Schema
 }
 
-type doltIndexPk struct {
-	db        Database
-	sch       schema.Schema
-	tableName string
-	table     *doltdb.Table
-	driver    *DoltIndexDriver
+type doltIndex struct {
 	cols      []schema.Column
 	ctx       *sql.Context
-}
-
-var _ DoltIndex = (*doltIndexPk)(nil)
-
-func (pdi *doltIndexPk) Database() string {
-	return pdi.db.name
-}
-
-func (pdi *doltIndexPk) DoltDatabase() Database {
-	return pdi.db
-}
-
-func (pdi *doltIndexPk) Driver() string {
-	return pdi.driver.ID()
-}
-
-func (pdi *doltIndexPk) Expressions() []string {
-	strs := make([]string, len(pdi.cols))
-	for i, col := range pdi.cols {
-		strs[i] = pdi.tableName + "." + col.Name
-	}
-	return strs
-}
-
-func (pdi *doltIndexPk) Get(key ...interface{}) (sql.IndexLookup, error) {
-	//TODO: replace all of this when partial keys land
-	if len(pdi.cols) != len(key) {
-		return nil, errors.New("index does not match the given key length")
-	}
-
-	taggedVals := make(row.TaggedValues)
-	for i, col := range pdi.cols {
-		val, err := col.TypeInfo.ConvertValueToNomsValue(key[i])
-		if err != nil {
-			return nil, err
-		}
-		taggedVals[col.Tag] = val
-	}
-
-	if pdi.sch.GetPKCols().Size() == 1 {
-		return &doltIndexLookup{
-			idx: pdi,
-			keyIter: &doltIndexSinglePkKeyIter{
-				hasReturned: false,
-				val:         taggedVals,
-			},
-		}, nil
-	} else {
-		rowData, err := pdi.table.GetRowData(pdi.ctx)
-		if err != nil {
-			return nil, err
-		}
-		rowDataIter, err := rowData.Iterator(pdi.ctx)
-		if err != nil {
-			return nil, err
-		}
-		return &doltIndexLookup{
-			idx: pdi,
-			keyIter: &doltIndexMultiPkKeyIter{
-				tableName:    pdi.tableName,
-				tableMapIter: rowDataIter,
-				val:          taggedVals,
-			},
-		}, nil
-	}
-}
-
-func (*doltIndexPk) Has(partition sql.Partition, key ...interface{}) (bool, error) {
-	// appears to be unused for the moment
-	panic("not used")
-}
-
-func (pdi *doltIndexPk) ID() string {
-	return fmt.Sprintf("%s:primaryKey%v", pdi.tableName, len(pdi.cols))
-}
-
-func (pdi *doltIndexPk) Schema() schema.Schema {
-	return pdi.sch
-}
-
-func (pdi *doltIndexPk) Table() string {
-	return pdi.tableName
-}
-
-type doltIndex struct {
 	db        Database
 	driver    *DoltIndexDriver
-	tableSch  schema.Schema
-	tableName string
+	id        string
+	mapSch    schema.Schema
+	rowData   types.Map
 	table     *doltdb.Table
-	index     schema.Index
-	ctx       *sql.Context
+	tableName string
+	tableSch  schema.Schema
 }
 
 var _ DoltIndex = (*doltIndex)(nil)
@@ -164,47 +75,43 @@ func (di *doltIndex) Driver() string {
 }
 
 func (di *doltIndex) Expressions() []string {
-	tags := di.index.IndexedColumnTags()
-	strs := make([]string, len(tags))
-	for i, tag := range tags {
-		col, _ := di.index.GetColumn(tag)
+	strs := make([]string, len(di.cols))
+	for i, col := range di.cols {
 		strs[i] = di.tableName + "." + col.Name
 	}
 	return strs
 }
 
 func (di *doltIndex) Get(key ...interface{}) (sql.IndexLookup, error) {
-	if len(di.index.IndexedColumnTags()) != len(key) {
-		return nil, errors.New("key must specify all columns for inner index")
+	if len(di.cols) != len(key) {
+		return nil, errors.New("key must specify all columns for index")
 	}
 
-	taggedVals := make(row.TaggedValues)
-	for i, tag := range di.index.IndexedColumnTags() {
-		if i >= len(key) {
-			break
-		}
-		col, _ := di.index.GetColumn(tag)
-		val, err := col.TypeInfo.ConvertValueToNomsValue(key[i])
-		if err != nil {
-			return nil, err
-		}
-		taggedVals[tag] = val
+	equals := make([]*expression.Equals, len(key))
+	for i := 0; i < len(key); i++ {
+		equals[i] = expression.NewEquals(
+			expression.NewGetField(i, di.cols[i].TypeInfo.ToSqlType(), di.cols[i].Name, di.cols[i].IsNullable()),
+			expression.NewLiteral(key[i], di.cols[i].TypeInfo.ToSqlType()),
+		)
+	}
+	var lastExpr sql.Expression = equals[len(equals)-1]
+	for i := len(equals) - 2; i >= 0; i-- {
+		lastExpr = expression.NewAnd(equals[i], lastExpr)
 	}
 
-	rowData, err := di.table.GetIndexRowData(di.ctx, di.index.Name())
+	crf, err := CreateReaderFuncLimitedByExpressions(di.rowData.Format(), di.mapSch, []sql.Expression{lastExpr})
 	if err != nil {
 		return nil, err
 	}
-	rowDataIter, err := rowData.Iterator(di.ctx)
+	mapIter, err := crf(di.ctx, di.rowData)
 	if err != nil {
 		return nil, err
 	}
+
 	return &doltIndexLookup{
 		di,
 		&doltIndexKeyIter{
-			index:        di.index,
-			indexMapIter: rowDataIter,
-			val:          taggedVals,
+			indexMapIter: mapIter,
 		},
 	}, nil
 }
@@ -215,7 +122,7 @@ func (*doltIndex) Has(partition sql.Partition, key ...interface{}) (bool, error)
 }
 
 func (di *doltIndex) ID() string {
-	return fmt.Sprintf("%s:%s%v", di.tableName, di.index.Name(), len(di.index.IndexedColumnTags()))
+	return di.id
 }
 
 func (di *doltIndex) Schema() schema.Schema {
diff --git a/go/libraries/doltcore/sqle/index_driver.go b/go/libraries/doltcore/sqle/index_driver.go
index 17bedc56eb..a7f7f78ddd 100644
--- a/go/libraries/doltcore/sqle/index_driver.go
+++ b/go/libraries/doltcore/sqle/index_driver.go
@@ -15,7 +15,11 @@
 package sqle
 
 import (
+	"fmt"
+
 	"github.com/liquidata-inc/go-mysql-server/sql"
+
+	"github.com/liquidata-inc/dolt/go/libraries/doltcore/schema"
 )
 
 type DoltIndexDriver struct {
@@ -69,26 +73,46 @@ func (driver *DoltIndexDriver) LoadAll(ctx *sql.Context, db, table string) ([]sq
 		return nil, err
 	}
 
+	rowData, err := tbl.GetRowData(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	cols := sch.GetPKCols().GetColumns()
 	sqlIndexes := []sql.Index{
-		&doltIndexPk{
-			db:        database,
-			sch:       sch,
-			tableName: table,
-			table:     tbl,
-			driver:    driver,
-			cols:      sch.GetPKCols().GetColumns(),
+		&doltIndex{
+			cols:      cols,
 			ctx:       ctx,
+			db:        database,
+			driver:    driver,
+			id:        fmt.Sprintf("%s:primaryKey%v", table, len(cols)),
+			mapSch:    sch,
+			rowData:   rowData,
+			table:     tbl,
+			tableName: table,
+			tableSch:  sch,
 		},
 	}
 	for _, index := range sch.Indexes().AllIndexes() {
+		rowData, err := tbl.GetIndexRowData(ctx, index.Name())
+		if err != nil {
+			return nil, err
+		}
+		cols := make([]schema.Column, index.Count())
+		for i, tag := range index.IndexedColumnTags() {
+			cols[i], _ = index.GetColumn(tag)
+		}
 		sqlIndexes = append(sqlIndexes, &doltIndex{
+			cols:      cols,
+			ctx:       ctx,
 			db:        database,
 			driver:    driver,
-			tableSch:  sch,
-			tableName: table,
+			id:        table + index.Name(),
+			mapSch:    index.Schema(),
+			rowData:   rowData,
 			table:     tbl,
-			index:     index,
-			ctx:       ctx,
+			tableName: table,
+			tableSch:  sch,
 		})
 	}
 
diff --git a/go/libraries/doltcore/sqle/index_lookup.go b/go/libraries/doltcore/sqle/index_lookup.go
index 88af5bcc04..23b325645d 100644
--- a/go/libraries/doltcore/sqle/index_lookup.go
+++ b/go/libraries/doltcore/sqle/index_lookup.go
@@ -15,14 +15,10 @@
 package sqle
 
 import (
-	"fmt"
-	"io"
-
 	"github.com/liquidata-inc/go-mysql-server/sql"
 
 	"github.com/liquidata-inc/dolt/go/libraries/doltcore/row"
-	"github.com/liquidata-inc/dolt/go/libraries/doltcore/schema"
-	"github.com/liquidata-inc/dolt/go/store/types"
+	"github.com/liquidata-inc/dolt/go/libraries/doltcore/table"
 )
 
 type IndexLookupKeyIterator interface {
@@ -48,89 +44,16 @@ func (il *doltIndexLookup) RowIter(ctx *sql.Context) (sql.RowIter, error) {
 	return &indexLookupRowIterAdapter{indexLookup: il, ctx: ctx}, nil
 }
 
-type doltIndexSinglePkKeyIter struct {
-	hasReturned bool
-	val         row.TaggedValues
-}
-
-var _ IndexLookupKeyIterator = (*doltIndexSinglePkKeyIter)(nil)
-
-func (iter *doltIndexSinglePkKeyIter) NextKey(*sql.Context) (row.TaggedValues, error) {
-	if iter.hasReturned {
-		return nil, io.EOF
-	}
-	iter.hasReturned = true
-	return iter.val, nil
-}
-
-type doltIndexMultiPkKeyIter struct {
-	tableName    string
-	tableMapIter types.MapIterator
-	val          row.TaggedValues
-}
-
-var _ IndexLookupKeyIterator = (*doltIndexMultiPkKeyIter)(nil)
-
-func (iter *doltIndexMultiPkKeyIter) NextKey(ctx *sql.Context) (row.TaggedValues, error) {
-	var k types.Value
-	var err error
-IterateOverMap:
-	for k, _, err = iter.tableMapIter.Next(ctx); k != nil && err == nil; k, _, err = iter.tableMapIter.Next(ctx) {
-		key, err := row.ParseTaggedValues(k.(types.Tuple))
-		if err != nil {
-			return nil, err
-		}
-		for tag, val := range iter.val {
-			indexVal, ok := key[tag]
-			if !ok {
-				return nil, fmt.Errorf("on table `%s`, attempted to gather value for tag `%v`", iter.tableName, tag)
-			}
-			if !val.Equals(indexVal) {
-				continue IterateOverMap
-			}
-		}
-		return key, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-	return nil, io.EOF
-}
-
 type doltIndexKeyIter struct {
-	index        schema.Index
-	indexMapIter types.MapIterator
-	val          row.TaggedValues
+	indexMapIter table.TableReadCloser
 }
 
 var _ IndexLookupKeyIterator = (*doltIndexKeyIter)(nil)
 
 func (iter *doltIndexKeyIter) NextKey(ctx *sql.Context) (row.TaggedValues, error) {
-	var k types.Value
-	var err error
-IterateOverMap:
-	for k, _, err = iter.indexMapIter.Next(ctx); k != nil && err == nil; k, _, err = iter.indexMapIter.Next(ctx) {
-		indexKeyTaggedValues, err := row.ParseTaggedValues(k.(types.Tuple))
-		if err != nil {
-			return nil, err
-		}
-		for tag, val := range iter.val {
-			indexVal, ok := indexKeyTaggedValues[tag]
-			if !ok {
-				return nil, fmt.Errorf("on index `%s`, attempted to gather value for tag `%v`", iter.index.Name(), tag)
-			}
-			if !val.Equals(indexVal) {
-				continue IterateOverMap
-			}
-		}
-		primaryKeys := make(row.TaggedValues)
-		for _, tag := range iter.index.PrimaryKeyTags() {
-			primaryKeys[tag] = indexKeyTaggedValues[tag]
-		}
-		return primaryKeys, nil
-	}
+	indexRow, err := iter.indexMapIter.ReadRow(ctx)
 	if err != nil {
 		return nil, err
 	}
-	return nil, io.EOF
+	return row.GetTaggedVals(indexRow)
 }