diff --git a/bats/types.bats b/bats/types.bats
index ee1900f43a3..8e2e5ca98f8 100644
--- a/bats/types.bats
+++ b/bats/types.bats
@@ -12,14 +12,14 @@ teardown() {
 @test "types: BIGINT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v BIGINT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v BIGINT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` BIGINT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` BIGINT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 4611686018427387903);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -37,14 +37,14 @@ SQL
 @test "types: BIGINT UNSIGNED" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v BIGINT UNSIGNED COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v BIGINT UNSIGNED,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` BIGINT UNSIGNED COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` BIGINT UNSIGNED" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 9223372036854775807);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -64,14 +64,14 @@ SQL
     skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v BINARY(10) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v BINARY(10),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` BINARY(10) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` BINARY(10)" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -87,14 +87,14 @@ SQL
 @test "types: BIT(10)" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v BIT(10) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v BIT(10),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` BIT(10) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` BIT(10)" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 511);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -113,14 +113,14 @@ SQL
     skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v BLOB COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v BLOB,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` BLOB COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` BLOB" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -131,17 +131,30 @@ SQL
     [[ "${lines[3]}" =~ " 1234567890 " ]] || false
 }
 
+@test "types: BOOL" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v BOOL,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` TINYINT" ]] || false
+}
+
 @test "types: BOOLEAN" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v BOOLEAN COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v BOOLEAN,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TINYINT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TINYINT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, true);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -155,14 +168,14 @@ SQL
 @test "types: CHAR(10)" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v CHAR(10) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v CHAR(10),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` CHAR(10) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` CHAR(10)" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -175,17 +188,43 @@ SQL
     [ "$status" -eq "1" ]
 }
 
+@test "types: CHARACTER(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v CHARACTER(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` CHAR(10)" ]] || false
+}
+
+@test "types: CHARACTER VARYING(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v CHARACTER VARYING(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` VARCHAR(10)" ]] || false
+}
+
 @test "types: DATE" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v DATE COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v DATE,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DATE COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DATE" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, '2020-02-10 11:12:13.456789');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -207,14 +246,14 @@ SQL
 @test "types: DATETIME" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v DATETIME COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v DATETIME,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DATETIME COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DATETIME" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, '2020-02-10 11:12:13.456789');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -233,46 +272,82 @@ SQL
     [ "$status" -eq "1" ]
 }
 
+@test "types: DEC" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v DEC,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(10,0)" ]] || false
+}
+
+@test "types: DEC(9)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v DEC(9),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(9,0)" ]] || false
+}
+
+@test "types: DEC(9,5)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v DEC(9,5),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(9,5)" ]] || false
+}
+
 @test "types: DECIMAL" {
-    skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v DECIMAL COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v DECIMAL,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DECIMAL(10, 0) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DECIMAL(10,0)" ]] || false
 }
 
 @test "types: DECIMAL(9)" {
-    skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v DECIMAL(9) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v DECIMAL(9),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DECIMAL(9, 0) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DECIMAL(9,0)" ]] || false
 }
 
-@test "types: DECIMAL(9, 5)" {
-    skip "This is not yet persisted in dolt"
+@test "types: DECIMAL(9,5)" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v DECIMAL(10, 5) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v DECIMAL(9,5),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DECIMAL(10, 5) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DECIMAL(9,5)" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 1234.56789);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -288,14 +363,14 @@ SQL
 @test "types: DOUBLE" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v DOUBLE COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v DOUBLE,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DOUBLE COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DOUBLE" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 1.25);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -315,18 +390,30 @@ SQL
     [ "$status" -eq "1" ]
 }
 
+@test "types: DOUBLE PRECISION" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v DOUBLE PRECISION,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DOUBLE" ]] || false
+}
+
 @test "types: ENUM('a','b','c')" {
-    skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v ENUM('a','b','c') COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v ENUM('a','b','c'),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` ENUM('a','b','c') COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` ENUM('a','b','c')" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'a');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -341,17 +428,56 @@ SQL
     [ "$status" -eq "1" ]
 }
 
+@test "types: FIXED" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v FIXED,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(10,0)" ]] || false
+}
+
+@test "types: FIXED(9)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v FIXED(9),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(9,0)" ]] || false
+}
+
+@test "types: FIXED(9,5)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v FIXED(9,5),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(9,5)" ]] || false
+}
+
 @test "types: FLOAT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v FLOAT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v FLOAT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` FLOAT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` FLOAT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 1.25);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -374,14 +500,14 @@ SQL
 @test "types: INT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v INT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v INT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` INT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` INT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 1073741823);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -399,14 +525,14 @@ SQL
 @test "types: INT UNSIGNED" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v INT UNSIGNED COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v INT UNSIGNED,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` INT UNSIGNED COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` INT UNSIGNED" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 2147483647);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -424,41 +550,67 @@ SQL
 @test "types: INTEGER" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v INTEGER COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v INTEGER,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` INT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` INT" ]] || false
 }
 
 @test "types: INTEGER UNSIGNED" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v INTEGER UNSIGNED COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v INTEGER UNSIGNED,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` INT UNSIGNED COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` INT UNSIGNED" ]] || false
+}
+
+@test "types: LONG" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v LONG,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` MEDIUMTEXT" ]] || false
+}
+
+@test "types: LONG VARCHAR" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v LONG VARCHAR,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` MEDIUMTEXT" ]] || false
 }
 
 @test "types: LONGBLOB" {
     skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v LONGBLOB COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v LONGBLOB,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` LONGBLOB COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` LONGBLOB" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -472,14 +624,14 @@ SQL
 @test "types: LONGTEXT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v LONGTEXT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v LONGTEXT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` LONGTEXT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` LONGTEXT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -494,14 +646,14 @@ SQL
     skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v MEDIUMBLOB COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v MEDIUMBLOB,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` MEDIUMBLOB COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` MEDIUMBLOB" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -515,14 +667,14 @@ SQL
 @test "types: MEDIUMINT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v MEDIUMINT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v MEDIUMINT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` MEDIUMINT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` MEDIUMINT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 4194303);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -540,14 +692,14 @@ SQL
 @test "types: MEDIUMINT UNSIGNED" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v MEDIUMINT UNSIGNED COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v MEDIUMINT UNSIGNED,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` MEDIUMINT UNSIGNED COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` MEDIUMINT UNSIGNED" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 8388607);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -565,14 +717,14 @@ SQL
 @test "types: MEDIUMTEXT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v MEDIUMTEXT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v MEDIUMTEXT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` MEDIUMTEXT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` MEDIUMTEXT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -583,31 +735,147 @@ SQL
     [[ "${lines[3]}" =~ " 1234567890 " ]] || false
 }
 
+@test "types: NATIONAL CHAR(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NATIONAL CHAR(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` CHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci" ]] || false
+}
+
+@test "types: NATIONAL CHARACTER(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NATIONAL CHARACTER(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` CHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci" ]] || false
+}
+
+@test "types: NATIONAL CHARACTER VARYING(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NATIONAL CHARACTER VARYING(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` VARCHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci" ]] || false
+}
+
+@test "types: NATIONAL VARCHAR(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NATIONAL VARCHAR(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` VARCHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci" ]] || false
+}
+
+@test "types: NCHAR(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NCHAR(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` CHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci" ]] || false
+}
+
+@test "types: NVARCHAR(10)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NVARCHAR(10),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` VARCHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci" ]] || false
+}
+
+@test "types: NUMERIC" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NUMERIC,
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(10,0)" ]] || false
+}
+
+@test "types: NUMERIC(9)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NUMERIC(9),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(9,0)" ]] || false
+}
+
+@test "types: NUMERIC(9,5)" {
+    dolt sql <<SQL
+CREATE TABLE test (
+  pk BIGINT NOT NULL,
+  v NUMERIC(9,5),
+  PRIMARY KEY (pk)
+);
+SQL
+    run dolt schema show
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "\`v\` DECIMAL(9,5)" ]] || false
+}
+
 @test "types: REAL" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v REAL COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v REAL,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` DOUBLE COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` DOUBLE" ]] || false
 }
 
 @test "types: SET('a','b','c')" {
-    skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v SET('a','b','c') COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v SET('a','b','c'),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` SET('a','b','c') COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` SET('a','b','c')" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'b,a');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -626,14 +894,14 @@ SQL
 @test "types: SMALLINT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v SMALLINT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v SMALLINT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` SMALLINT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` SMALLINT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 16383);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -651,14 +919,14 @@ SQL
 @test "types: SMALLINT UNSIGNED" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v SMALLINT UNSIGNED COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v SMALLINT UNSIGNED,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` SMALLINT UNSIGNED COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` SMALLINT UNSIGNED" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 32767);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -676,14 +944,14 @@ SQL
 @test "types: TEXT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TEXT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TEXT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TEXT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TEXT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -695,17 +963,16 @@ SQL
 }
 
 @test "types: TIME" {
-    skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TIME COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TIME,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TIME COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TIME" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, '11:22:33.444444');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -727,14 +994,14 @@ SQL
 @test "types: TIMESTAMP" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TIMESTAMP COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TIMESTAMP,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TIMESTAMP COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TIMESTAMP" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, '2020-02-10 11:12:13.456789');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -757,14 +1024,14 @@ SQL
     skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TINYBLOB COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TINYBLOB,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TINYBLOB COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TINYBLOB" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -778,14 +1045,14 @@ SQL
 @test "types: TINYINT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TINYINT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TINYINT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TINYINT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TINYINT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 63);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -803,14 +1070,14 @@ SQL
 @test "types: TINYINT UNSIGNED" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TINYINT UNSIGNED COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TINYINT UNSIGNED,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TINYINT UNSIGNED COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TINYINT UNSIGNED" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 127);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -828,14 +1095,14 @@ SQL
 @test "types: TINYTEXT" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v TINYTEXT COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v TINYTEXT,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` TINYTEXT COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` TINYTEXT" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -850,14 +1117,14 @@ SQL
     skip "This is not yet persisted in dolt"
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v VARBINARY(10) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v VARBINARY(10),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` VARBINARY(10) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` VARBINARY(10)" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -873,14 +1140,14 @@ SQL
 @test "types: VARCHAR(10)" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v VARCHAR(10) COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v VARCHAR(10),
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` VARCHAR(10) COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` VARCHAR(10)" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -893,17 +1160,17 @@ SQL
     [ "$status" -eq "1" ]
 }
 
-@test "types: VARCHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin" {
+@test "types: VARCHAR(10) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v VARCHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v VARCHAR(10) CHARACTER SET utf32 COLLATE utf32_general_ci,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` VARCHAR(10) CHARACTER SET utf8mb3 COLLATE utf8mb3_bin COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` VARCHAR(10) CHARACTER SET utf32 COLLATE utf32_general_ci" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 'abcdefg');"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
@@ -919,14 +1186,14 @@ SQL
 @test "types: YEAR" {
     dolt sql <<SQL
 CREATE TABLE test (
-  pk BIGINT NOT NULL COMMENT 'tag:0',
-  v YEAR COMMENT 'tag:1',
+  pk BIGINT NOT NULL,
+  v YEAR,
   PRIMARY KEY (pk)
 );
 SQL
     run dolt schema show
     [ "$status" -eq "0" ]
-    [[ "$output" =~ "\`v\` YEAR COMMENT 'tag:1'" ]] || false
+    [[ "$output" =~ "\`v\` YEAR" ]] || false
     dolt sql -q "INSERT INTO test VALUES (1, 1901);"
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
diff --git a/go/go.mod b/go/go.mod
index 9307c94824b..f85012a88a7 100644
--- a/go/go.mod
+++ b/go/go.mod
@@ -87,7 +87,7 @@ require (
 
 replace github.com/liquidata-inc/dolt/go/gen/proto/dolt/services/eventsapi => ./gen/proto/dolt/services/eventsapi
 
-replace github.com/src-d/go-mysql-server => github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414183459-0b63a0868140
+replace github.com/src-d/go-mysql-server => github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414214346-0c65dac7ca1e
 
 replace vitess.io/vitess => github.com/liquidata-inc/vitess v0.0.0-20200413233505-a88cc54bd1ee
 
diff --git a/go/go.sum b/go/go.sum
index 88a5a8fe686..3fa561762b0 100644
--- a/go/go.sum
+++ b/go/go.sum
@@ -359,14 +359,46 @@ github.com/krishicks/yaml-patch v0.0.10/go.mod h1:Sm5TchwZS6sm7RJoyg87tzxm2ZcKzd
 github.com/lib/pq v1.0.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.3.0 h1:/qkRGz8zljWiDcFvgpwUpwIAPu3r07TDvs3Rws+o/pU=
 github.com/lib/pq v1.3.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
+github.com/liquidata-inc/go-mysql-server v0.4.1-0.20200311072715-b12ae9d0cc97 h1:Zr78cjOfa0bM4X5JA692xhx3QvFPTsJiM0bD0xl/22Q=
+github.com/liquidata-inc/go-mysql-server v0.4.1-0.20200311072715-b12ae9d0cc97/go.mod h1:Lh0pg7jnO08HxFm6oj6gtcSTUeeOTu4Spt50Aeo2mes=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200330231002-2ac5a85cf8d6 h1:iKET+xfMh3NaiiIbrMBLi+MJ9hwmm++7DBtPGfarf50=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200330231002-2ac5a85cf8d6 h1:iKET+xfMh3NaiiIbrMBLi+MJ9hwmm++7DBtPGfarf50=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200330231002-2ac5a85cf8d6/go.mod h1:TCTrDbzIA05e8zV3SW+nsjc1LCR58GRSOIcF32lJ+Qc=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200330231002-2ac5a85cf8d6/go.mod h1:TCTrDbzIA05e8zV3SW+nsjc1LCR58GRSOIcF32lJ+Qc=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200403150612-34c67410dcd4 h1:UIksBT7bRENT38ErKSz+auGLc7a5tDpCHwNhuajoJbU=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200403150612-34c67410dcd4/go.mod h1:TCTrDbzIA05e8zV3SW+nsjc1LCR58GRSOIcF32lJ+Qc=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200403171307-83ac7e7158e2 h1:l4mXLvgHMoihWuEqcmcJKEvQtAccHxhsKwkVn/okoxc=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200403171307-83ac7e7158e2/go.mod h1:TCTrDbzIA05e8zV3SW+nsjc1LCR58GRSOIcF32lJ+Qc=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200407175239-21fb18d4d9fd h1:SGGh7+XPqPYw3LaIK4VUvy/81Za1Y3p29lh4WDMtXh0=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200407175239-21fb18d4d9fd/go.mod h1:xu1cUi3vfWVJZ/9mQl9f8sdfJGobnS7kIucM3lfWIPk=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200412072052-a6583959dafb/go.mod h1:n8M6jtUZ5myu3O9kfcLqRnXPBVkqbkoBPe45mnYCKd0=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200412232521-1e406e8056fb/go.mod h1:n8M6jtUZ5myu3O9kfcLqRnXPBVkqbkoBPe45mnYCKd0=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414052025-88d3eff3f7f5 h1:Liiz/stNuLoWg1j1A/yGChITW4H/IbEyFwPGZPk+B8M=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414052025-88d3eff3f7f5/go.mod h1:n8M6jtUZ5myu3O9kfcLqRnXPBVkqbkoBPe45mnYCKd0=
 github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414183459-0b63a0868140 h1:rxT0Pkt2ZLS0P4m8scQ3TATRjKYcntF6F0X5/yHcIDg=
 github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414183459-0b63a0868140/go.mod h1:tK/saWoda2x+KXyGsdVariMdfVOsjmRgQF2pbl4Mr1E=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414214346-0c65dac7ca1e h1:cYKHqocy3oNkPmfayDwIswVy14Dcp8q5FFSYLS4FvIA=
+github.com/liquidata-inc/go-mysql-server v0.5.1-0.20200414214346-0c65dac7ca1e/go.mod h1:tK/saWoda2x+KXyGsdVariMdfVOsjmRgQF2pbl4Mr1E=
 github.com/liquidata-inc/ishell v0.0.0-20190514193646-693241f1f2a0 h1:phMgajKClMUiIr+hF2LGt8KRuUa2Vd2GI1sNgHgSXoU=
 github.com/liquidata-inc/ishell v0.0.0-20190514193646-693241f1f2a0/go.mod h1:YC1rI9k5gx8D02ljlbxDfZe80s/iq8bGvaaQsvR+qxs=
 github.com/liquidata-inc/mmap-go v1.0.3 h1:2LndAeAtup9rpvUmu4wZSYCsjCQ0Zpc+NqE+6+PnT7g=
 github.com/liquidata-inc/mmap-go v1.0.3/go.mod h1:w0doE7jfkuDEZyxb/zD3VWnRaQBYx1uDTS816kH8HoY=
+github.com/liquidata-inc/sqllogictest/go v0.0.0-20200225183643-358992a611e1 h1:BDpmbvQ9I8npWe7TOzQcGkrn7EYHrW1hJtTd9h8MNZA=
+github.com/liquidata-inc/sqllogictest/go v0.0.0-20200225183643-358992a611e1/go.mod h1:kKRVtyuomkqz15YFRpS0OT8kpsU8y/F3jyiZtvALdKU=
 github.com/liquidata-inc/sqllogictest/go v0.0.0-20200320151923-b11801f10e15 h1:H3RwcYfzkdW4kFh7znTUopcX3XZqnFXm6pcmxSy0mNo=
+github.com/liquidata-inc/sqllogictest/go v0.0.0-20200320151923-b11801f10e15 h1:H3RwcYfzkdW4kFh7znTUopcX3XZqnFXm6pcmxSy0mNo=
+github.com/liquidata-inc/sqllogictest/go v0.0.0-20200320151923-b11801f10e15/go.mod h1:kKRVtyuomkqz15YFRpS0OT8kpsU8y/F3jyiZtvALdKU=
 github.com/liquidata-inc/sqllogictest/go v0.0.0-20200320151923-b11801f10e15/go.mod h1:kKRVtyuomkqz15YFRpS0OT8kpsU8y/F3jyiZtvALdKU=
+github.com/liquidata-inc/vitess v0.0.0-20200102230944-f3410911d61f h1:fqsJy7h3D3esm9tYSzU7LV6h2tfifdYTanPuDL5LJ1A=
+github.com/liquidata-inc/vitess v0.0.0-20200102230944-f3410911d61f/go.mod h1:vn/QvIl/1+N6+qjheejcLt8jmX2kQSQwFinzZuoY1VY=
+github.com/liquidata-inc/vitess v0.0.0-20200318153456-e0b079da3f54 h1:LR/OEhgIYVQuo5a/lxr8Ps76AZ1FNWUgNANfKCA0XSQ=
+github.com/liquidata-inc/vitess v0.0.0-20200318153456-e0b079da3f54 h1:LR/OEhgIYVQuo5a/lxr8Ps76AZ1FNWUgNANfKCA0XSQ=
+github.com/liquidata-inc/vitess v0.0.0-20200318153456-e0b079da3f54/go.mod h1:vn/QvIl/1+N6+qjheejcLt8jmX2kQSQwFinzZuoY1VY=
+github.com/liquidata-inc/vitess v0.0.0-20200318153456-e0b079da3f54/go.mod h1:vn/QvIl/1+N6+qjheejcLt8jmX2kQSQwFinzZuoY1VY=
+github.com/liquidata-inc/vitess v0.0.0-20200407071440-54a487aaf7d9 h1:eaE6IFxMviaDSNFaKlTbNPA/+0Vhj/XgV6lG2SaoAWM=
+github.com/liquidata-inc/vitess v0.0.0-20200407071440-54a487aaf7d9/go.mod h1:vn/QvIl/1+N6+qjheejcLt8jmX2kQSQwFinzZuoY1VY=
+github.com/liquidata-inc/vitess v0.0.0-20200410001601-55d11bea14ca h1:m09m0bRpTa3PCxMNcnRf5AiVK7ME0PVIci1vwuciZ5w=
+github.com/liquidata-inc/vitess v0.0.0-20200410001601-55d11bea14ca/go.mod h1:vn/QvIl/1+N6+qjheejcLt8jmX2kQSQwFinzZuoY1VY=
 github.com/liquidata-inc/vitess v0.0.0-20200413233505-a88cc54bd1ee h1:r8ApUMNHHEyzRhPbuIHrWbr7FOTW4Yo5Sm1HpOEzPrQ=
 github.com/liquidata-inc/vitess v0.0.0-20200413233505-a88cc54bd1ee/go.mod h1:vn/QvIl/1+N6+qjheejcLt8jmX2kQSQwFinzZuoY1VY=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
diff --git a/go/libraries/doltcore/rowconv/row_converter.go b/go/libraries/doltcore/rowconv/row_converter.go
index f85ceac3bbb..bf9f5d39b16 100644
--- a/go/libraries/doltcore/rowconv/row_converter.go
+++ b/go/libraries/doltcore/rowconv/row_converter.go
@@ -190,7 +190,7 @@ func isNecessary(srcSch, destSch schema.Schema, destToSrc map[uint64]uint64) (bo
 			return true, nil
 		}
 
-		if srcCol.Kind != destCol.Kind {
+		if !srcCol.TypeInfo.Equals(destCol.TypeInfo) {
 			return true, nil
 		}
 	}
diff --git a/go/libraries/doltcore/schema/encoding/schema_marshaling_test.go b/go/libraries/doltcore/schema/encoding/schema_marshaling_test.go
index 59d83dcc3f4..8de313f555f 100644
--- a/go/libraries/doltcore/schema/encoding/schema_marshaling_test.go
+++ b/go/libraries/doltcore/schema/encoding/schema_marshaling_test.go
@@ -118,12 +118,8 @@ func TestJSONMarshalling(t *testing.T) {
 func TestTypeInfoMarshalling(t *testing.T) {
 	//TODO: determine the storage format for BINARY
 	//TODO: determine the storage format for BLOB
-	//TODO: determine the storage format for DECIMAL
-	//TODO: determine the storage format for ENUM
 	//TODO: determine the storage format for LONGBLOB
 	//TODO: determine the storage format for MEDIUMBLOB
-	//TODO: determine the storage format for SET
-	//TODO: determine the storage format for TIME
 	//TODO: determine the storage format for TINYBLOB
 	//TODO: determine the storage format for VARBINARY
 	sqlTypes := []sql.Type{
@@ -134,11 +130,11 @@ func TestTypeInfoMarshalling(t *testing.T) {
 		//sql.Blob, //BLOB
 		sql.Boolean, //BOOLEAN
 		sql.MustCreateStringWithDefaults(sqltypes.Char, 10), //CHAR(10)
-		sql.Date,     //DATE
-		sql.Datetime, //DATETIME
-		//sql.MustCreateDecimalType(9, 5), //DECIMAL(9, 5)
-		sql.Float64, //DOUBLE
-		//sql.MustCreateEnumType([]string{"a", "b", "c"}, sql.Collation_Default), //ENUM('a','b','c')
+		sql.Date,                        //DATE
+		sql.Datetime,                    //DATETIME
+		sql.MustCreateDecimalType(9, 5), //DECIMAL(9, 5)
+		sql.Float64,                     //DOUBLE
+		sql.MustCreateEnumType([]string{"a", "b", "c"}, sql.Collation_Default), //ENUM('a','b','c')
 		sql.Float32, //FLOAT
 		sql.Int32,   //INT
 		sql.Uint32,  //INT UNSIGNED
@@ -148,11 +144,11 @@ func TestTypeInfoMarshalling(t *testing.T) {
 		sql.Int24,      //MEDIUMINT
 		sql.Uint24,     //MEDIUMINT UNSIGNED
 		sql.MediumText, //MEDIUMTEXT
-		//sql.MustCreateSetType([]string{"a", "b", "c"}, sql.Collation_Default), //SET('a','b','c')
-		sql.Int16,  //SMALLINT
-		sql.Uint16, //SMALLINT UNSIGNED
-		sql.Text,   //TEXT
-		//sql.Time, //TIME
+		sql.MustCreateSetType([]string{"a", "b", "c"}, sql.Collation_Default), //SET('a','b','c')
+		sql.Int16,     //SMALLINT
+		sql.Uint16,    //SMALLINT UNSIGNED
+		sql.Text,      //TEXT
+		sql.Time,      //TIME
 		sql.Timestamp, //TIMESTAMP
 		//sql.TinyBlob, //TINYBLOB
 		sql.Int8,     //TINYINT
diff --git a/go/libraries/doltcore/schema/typeinfo/common_test.go b/go/libraries/doltcore/schema/typeinfo/common_test.go
index 351690710e8..a973a9171d5 100644
--- a/go/libraries/doltcore/schema/typeinfo/common_test.go
+++ b/go/libraries/doltcore/schema/typeinfo/common_test.go
@@ -135,6 +135,29 @@ func generateVarBinaryType(t *testing.T, length int64, pad bool) *varBinaryType
 	return &varBinaryType{sql.MustCreateBinary(sqltypes.VarBinary, length)}
 }
 
+func generateVarStringTypes(t *testing.T, numOfTypes uint16) []TypeInfo {
+	var res []TypeInfo
+	loop(t, 1, 500, numOfTypes, func(i int64) {
+		rts := false
+		if i%2 == 0 {
+			rts = true
+		}
+		res = append(res, generateVarStringType(t, i, rts))
+	})
+	return res
+}
+
+func generateVarStringType(t *testing.T, length int64, rts bool) *varStringType {
+	require.True(t, length > 0)
+	if rts {
+		t, err := sql.CreateStringWithDefaults(sqltypes.Char, length)
+		if err == nil {
+			return &varStringType{t}
+		}
+	}
+	return &varStringType{sql.MustCreateStringWithDefaults(sqltypes.VarChar, length)}
+}
+
 func loop(t *testing.T, start int64, endInclusive int64, numOfSteps uint16, loopedFunc func(int64)) {
 	require.True(t, endInclusive > start)
 	maxNumOfSteps := endInclusive - start + 1
diff --git a/go/libraries/doltcore/schema/typeinfo/decimal.go b/go/libraries/doltcore/schema/typeinfo/decimal.go
index 1c75fb4643e..fa3841e2677 100644
--- a/go/libraries/doltcore/schema/typeinfo/decimal.go
+++ b/go/libraries/doltcore/schema/typeinfo/decimal.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"strconv"
 
+	"github.com/shopspring/decimal"
 	"github.com/src-d/go-mysql-server/sql"
 
 	"github.com/liquidata-inc/dolt/go/store/types"
@@ -66,12 +67,8 @@ func CreateDecimalTypeFromParams(params map[string]string) (TypeInfo, error) {
 
 // ConvertNomsValueToValue implements TypeInfo interface.
 func (ti *decimalType) ConvertNomsValueToValue(v types.Value) (interface{}, error) {
-	if val, ok := v.(types.String); ok {
-		res, err := ti.sqlDecimalType.Convert(string(val))
-		if err != nil {
-			return nil, fmt.Errorf(`"%v" cannot convert "%v" to value`, ti.String(), val)
-		}
-		return res, nil
+	if val, ok := v.(types.Decimal); ok {
+		return ti.sqlDecimalType.Convert(decimal.Decimal(val))
 	}
 	if _, ok := v.(types.Null); ok || v == nil {
 		return nil, nil
@@ -84,15 +81,14 @@ func (ti *decimalType) ConvertValueToNomsValue(v interface{}) (types.Value, erro
 	if v == nil {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlDecimalType.Convert(v)
+	decVal, err := ti.sqlDecimalType.ConvertToDecimal(v)
 	if err != nil {
 		return nil, err
 	}
-	val, ok := strVal.(string)
-	if ok {
-		return types.String(val), nil
+	if !decVal.Valid {
+		return nil, fmt.Errorf(`"%v" has unexpectedly encountered a null value from embedded type`, ti.String())
 	}
-	return nil, fmt.Errorf(`"%v" has unexpectedly encountered a value of type "%T" from embedded type`, ti.String(), v)
+	return types.Decimal(decVal.Decimal), nil
 }
 
 // Equals implements TypeInfo interface.
@@ -144,7 +140,7 @@ func (ti *decimalType) IsValid(v types.Value) bool {
 
 // NomsKind implements TypeInfo interface.
 func (ti *decimalType) NomsKind() types.NomsKind {
-	return types.StringKind
+	return types.DecimalKind
 }
 
 // ParseValue implements TypeInfo interface.
@@ -152,14 +148,7 @@ func (ti *decimalType) ParseValue(str *string) (types.Value, error) {
 	if str == nil || *str == "" {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlDecimalType.Convert(*str)
-	if err != nil {
-		return nil, err
-	}
-	if val, ok := strVal.(string); ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" cannot convert the string "%v" to a value`, ti.String(), str)
+	return ti.ConvertValueToNomsValue(*str)
 }
 
 // String implements TypeInfo interface.
diff --git a/go/libraries/doltcore/schema/typeinfo/decimal_test.go b/go/libraries/doltcore/schema/typeinfo/decimal_test.go
new file mode 100644
index 00000000000..7ae6179bed0
--- /dev/null
+++ b/go/libraries/doltcore/schema/typeinfo/decimal_test.go
@@ -0,0 +1,595 @@
+// Copyright 2020 Liquidata, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package typeinfo
+
+import (
+	"fmt"
+	"math/big"
+	"testing"
+	"time"
+
+	"github.com/shopspring/decimal"
+	"github.com/src-d/go-mysql-server/sql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/liquidata-inc/dolt/go/store/types"
+)
+
+func TestDecimalConvertNomsValueToValue(t *testing.T) {
+	tests := []struct {
+		typ         *decimalType
+		input       types.Decimal
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateDecimalType(t, 1, 0),
+			types.Decimal(decimal.RequireFromString("0")),
+			"0",
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			types.Decimal(decimal.RequireFromString("-1.5")),
+			"-2",
+			false,
+		},
+		{
+			generateDecimalType(t, 2, 1),
+			types.Decimal(decimal.RequireFromString("-1.5")),
+			"-1.5",
+			false,
+		},
+		{
+			generateDecimalType(t, 5, 4),
+			types.Decimal(decimal.RequireFromString("-5.7159")),
+			"-5.7159",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			types.Decimal(decimal.RequireFromString("4723245")),
+			"4723245.00",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			types.Decimal(decimal.RequireFromString("4723245.01")),
+			"4723245.01",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			types.Decimal(decimal.RequireFromString("14723245.01")),
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 5, 4),
+			types.Decimal(decimal.RequireFromString("55.7159")),
+			"",
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ConvertNomsValueToValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, output)
+			}
+		})
+	}
+}
+
+func TestDecimalConvertValueToNomsValue(t *testing.T) {
+	tests := []struct {
+		typ         *decimalType
+		input       interface{}
+		output      types.Decimal
+		expectedErr bool
+	}{
+		{
+			generateDecimalType(t, 1, 0),
+			7,
+			types.Decimal(decimal.RequireFromString("7")),
+			false,
+		},
+		{
+			generateDecimalType(t, 5, 1),
+			-4.5,
+			types.Decimal(decimal.RequireFromString("-4.5")),
+			false,
+		},
+		{
+			generateDecimalType(t, 10, 0),
+			"77",
+			types.Decimal(decimal.RequireFromString("77")),
+			false,
+		},
+		{
+			generateDecimalType(t, 5, 0),
+			"dog",
+			types.Decimal{},
+			true,
+		},
+		{
+			generateDecimalType(t, 15, 7),
+			true,
+			types.Decimal{},
+			true,
+		},
+		{
+			generateDecimalType(t, 20, 5),
+			time.Unix(137849, 0),
+			types.Decimal{},
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ConvertValueToNomsValue(test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.True(t, test.output.Equals(output))
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestDecimalFormatValue(t *testing.T) {
+	tests := []struct {
+		typ         *decimalType
+		input       types.Decimal
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateDecimalType(t, 1, 0),
+			types.Decimal(decimal.RequireFromString("0")),
+			"0",
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			types.Decimal(decimal.RequireFromString("-1.5")),
+			"-2",
+			false,
+		},
+		{
+			generateDecimalType(t, 2, 1),
+			types.Decimal(decimal.RequireFromString("-1.5")),
+			"-1.5",
+			false,
+		},
+		{
+			generateDecimalType(t, 5, 4),
+			types.Decimal(decimal.RequireFromString("-5.7159")),
+			"-5.7159",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			types.Decimal(decimal.RequireFromString("4723245")),
+			"4723245.00",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			types.Decimal(decimal.RequireFromString("4723245.01")),
+			"4723245.01",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			types.Decimal(decimal.RequireFromString("14723245.01")),
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 5, 4),
+			types.Decimal(decimal.RequireFromString("55.7159")),
+			"",
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.FormatValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, *output)
+			}
+		})
+	}
+}
+
+func TestDecimalParseValue(t *testing.T) {
+	tests := []struct {
+		typ         *decimalType
+		input       string
+		output      types.Decimal
+		expectedErr bool
+	}{
+		{
+			generateDecimalType(t, 1, 0),
+			"0",
+			types.Decimal(decimal.RequireFromString("0")),
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			"-1.5",
+			types.Decimal(decimal.RequireFromString("-2")),
+			false,
+		},
+		{
+			generateDecimalType(t, 2, 1),
+			"-1.5",
+			types.Decimal(decimal.RequireFromString("-1.5")),
+			false,
+		},
+		{
+			generateDecimalType(t, 5, 4),
+			"-5.7159",
+			types.Decimal(decimal.RequireFromString("-5.7159")),
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			"4723245.00",
+			types.Decimal(decimal.RequireFromString("4723245.00")),
+			false,
+		},
+		{
+			generateDecimalType(t, 13, 2),
+			"4723245.01",
+			types.Decimal(decimal.RequireFromString("4723245.01")),
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 2),
+			"24723245.01",
+			types.Decimal{},
+			true,
+		},
+		{
+			generateDecimalType(t, 5, 4),
+			"-44.2841",
+			types.Decimal{},
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ParseValue(&test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.True(t, test.output.Equals(output))
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestDecimalMarshal(t *testing.T) {
+	tests := []struct {
+		precision   uint8
+		scale       uint8
+		val         interface{}
+		expectedVal string
+		expectedErr bool
+	}{
+		{1, 0, byte(0), "0", false},
+		{1, 0, int8(3), "3", false},
+		{1, 0, "-3.7e0", "-4", false},
+		{1, 0, uint(4), "4", false},
+		{1, 0, int16(9), "9", false},
+		{1, 0, "0.00000000000000000003e20", "3", false},
+		{1, 0, float64(-9.4), "-9", false},
+		{1, 0, float32(9.5), "", true},
+		{1, 0, int32(-10), "", true},
+
+		{1, 1, 0, "0.0", false},
+		{1, 1, .01, "0.0", false},
+		{1, 1, .1, "0.1", false},
+		{1, 1, ".22", "0.2", false},
+		{1, 1, .55, "0.6", false},
+		{1, 1, "-.7863294659345624", "-0.8", false},
+		{1, 1, "2634193746329327479.32030573792e-19", "0.3", false},
+		{1, 1, 1, "", true},
+		{1, 1, new(big.Rat).SetInt64(2), "", true},
+
+		{5, 0, 0, "0", false},
+		{5, 0, -5, "-5", false},
+		{5, 0, -99995, "-99995", false},
+		{5, 0, 5000.2, "5000", false},
+		{5, 0, "7742", "7742", false},
+		{5, 0, new(big.Float).SetFloat64(-4723.875), "-4724", false},
+		{5, 0, 99999, "99999", false},
+		{5, 0, "0xf8e1", "63713", false},
+		{5, 0, "0b1001110101100110", "40294", false},
+		{5, 0, new(big.Rat).SetFrac64(999999, 10), "", true},
+		{5, 0, 673927, "", true},
+
+		{10, 5, 0, "0.00000", false},
+		{10, 5, "25.1", "25.10000", false},
+		{10, 5, "-25.1", "-25.10000", false},
+		{10, 5, "-99205.8572", "-99205.85720", false},
+		{10, 5, "99999.999994", "99999.99999", false},
+		{10, 5, "5.5729136e3", "5572.91360", false},
+		{10, 5, "600e-2", "6.00000", false},
+		{10, 5, new(big.Rat).SetFrac64(-22, 7), "-3.14286", false},
+		{10, 5, "-99995.1", "-99995.10000", false},
+		{10, 5, 100000, "", true},
+		{10, 5, "-99999.999995", "", true},
+
+		{65, 0, "99999999999999999999999999999999999999999999999999999999999999999",
+			"99999999999999999999999999999999999999999999999999999999999999999", false},
+		{65, 0, "99999999999999999999999999999999999999999999999999999999999999999.1",
+			"99999999999999999999999999999999999999999999999999999999999999999", false},
+		{65, 0, "99999999999999999999999999999999999999999999999999999999999999999.99", "", true},
+
+		{65, 12, "16976349273982359874209023948672021737840592720387475.2719128737543572927374503832837350563300243035038234972093785",
+			"16976349273982359874209023948672021737840592720387475.271912873754", false},
+		{65, 12, "99999999999999999999999999999999999999999999999999999.9999999999999", "", true},
+
+		{20, 10, []byte{32}, "", true},
+		{20, 10, time.Date(2019, 12, 12, 12, 12, 12, 0, time.UTC), "", true},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%v %v %v", test.precision, test.scale, test.val), func(t *testing.T) {
+			typ := &decimalType{sql.MustCreateDecimalType(test.precision, test.scale)}
+			val, err := typ.ConvertValueToNomsValue(test.val)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, test.expectedVal, typ.sqlDecimalType.MustConvert(decimal.Decimal(val.(types.Decimal))))
+				umar, err := typ.ConvertNomsValueToValue(val)
+				require.NoError(t, err)
+				testVal := typ.sqlDecimalType.MustConvert(test.val)
+				cmp, err := typ.sqlDecimalType.Compare(testVal, umar)
+				require.NoError(t, err)
+				assert.Equal(t, 0, cmp)
+			}
+		})
+	}
+}
+
+func TestDecimalRoundTrip(t *testing.T) {
+	tests := []struct {
+		typ         *decimalType
+		input       string
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateDecimalType(t, 1, 0),
+			"0",
+			"0",
+			false,
+		},
+		{
+			generateDecimalType(t, 4, 1),
+			"0",
+			"0.0",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 4),
+			"0",
+			"0.0000",
+			false,
+		},
+		{
+			generateDecimalType(t, 26, 0),
+			"0",
+			"0",
+			false,
+		},
+		{
+			generateDecimalType(t, 48, 22),
+			"0",
+			"0.0000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 65, 30),
+			"0",
+			"0.000000000000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			"-1.5",
+			"-2",
+			false,
+		},
+		{
+			generateDecimalType(t, 4, 1),
+			"-1.5",
+			"-1.5",
+			false,
+		},
+		{
+			generateDecimalType(t, 9, 4),
+			"-1.5",
+			"-1.5000",
+			false,
+		},
+		{
+			generateDecimalType(t, 26, 0),
+			"-1.5",
+			"-2",
+			false,
+		},
+		{
+			generateDecimalType(t, 48, 22),
+			"-1.5",
+			"-1.5000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 65, 30),
+			"-1.5",
+			"-1.500000000000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			"9351580",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 4, 1),
+			"9351580",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 9, 4),
+			"9351580",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 26, 0),
+			"9351580",
+			"9351580",
+			false,
+		},
+		{
+			generateDecimalType(t, 48, 22),
+			"9351580",
+			"9351580.0000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 65, 30),
+			"9351580",
+			"9351580.000000000000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			"-1076416.875",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 4, 1),
+			"-1076416.875",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 9, 4),
+			"-1076416.875",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 26, 0),
+			"-1076416.875",
+			"-1076417",
+			false,
+		},
+		{
+			generateDecimalType(t, 48, 22),
+			"-1076416.875",
+			"-1076416.8750000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 65, 30),
+			"-1076416.875",
+			"-1076416.875000000000000000000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 1, 0),
+			"198728394234798423466321.27349757",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 4, 1),
+			"198728394234798423466321.27349757",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 9, 4),
+			"198728394234798423466321.27349757",
+			"",
+			true,
+		},
+		{
+			generateDecimalType(t, 26, 0),
+			"198728394234798423466321.27349757",
+			"198728394234798423466321",
+			false,
+		},
+		{
+			generateDecimalType(t, 48, 22),
+			"198728394234798423466321.27349757",
+			"198728394234798423466321.2734975700000000000000",
+			false,
+		},
+		{
+			generateDecimalType(t, 65, 30),
+			"198728394234798423466321.27349757",
+			"198728394234798423466321.273497570000000000000000000000",
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v %v`, test.typ.String(), test.input, test.output), func(t *testing.T) {
+			parsed, err := test.typ.ConvertValueToNomsValue(test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				output, err := test.typ.ConvertNomsValueToValue(parsed)
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+				parsed2, err := test.typ.ParseValue(&test.input)
+				require.NoError(t, err)
+				assert.Equal(t, parsed, parsed2)
+				output2, err := test.typ.FormatValue(parsed2)
+				require.NoError(t, err)
+				assert.Equal(t, test.output, *output2)
+			} else {
+				assert.Error(t, err)
+				_, err = test.typ.ParseValue(&test.input)
+				assert.Error(t, err)
+			}
+		})
+	}
+}
diff --git a/go/libraries/doltcore/schema/typeinfo/enum.go b/go/libraries/doltcore/schema/typeinfo/enum.go
index 6a699f6e8cb..75c353972b7 100644
--- a/go/libraries/doltcore/schema/typeinfo/enum.go
+++ b/go/libraries/doltcore/schema/typeinfo/enum.go
@@ -67,8 +67,8 @@ func CreateEnumTypeFromParams(params map[string]string) (TypeInfo, error) {
 
 // ConvertNomsValueToValue implements TypeInfo interface.
 func (ti *enumType) ConvertNomsValueToValue(v types.Value) (interface{}, error) {
-	if val, ok := v.(types.String); ok {
-		res, err := ti.sqlEnumType.Convert(string(val))
+	if val, ok := v.(types.Uint); ok {
+		res, err := ti.sqlEnumType.Unmarshal(int64(val))
 		if err != nil {
 			return nil, fmt.Errorf(`"%v" cannot convert "%v" to value`, ti.String(), val)
 		}
@@ -85,15 +85,11 @@ func (ti *enumType) ConvertValueToNomsValue(v interface{}) (types.Value, error)
 	if v == nil {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlEnumType.Convert(v)
+	val, err := ti.sqlEnumType.Marshal(v)
 	if err != nil {
 		return nil, err
 	}
-	val, ok := strVal.(string)
-	if ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" has unexpectedly encountered a value of type "%T" from embedded type`, ti.String(), v)
+	return types.Uint(val), nil
 }
 
 // Equals implements TypeInfo interface.
@@ -158,7 +154,7 @@ func (ti *enumType) IsValid(v types.Value) bool {
 
 // NomsKind implements TypeInfo interface.
 func (ti *enumType) NomsKind() types.NomsKind {
-	return types.StringKind
+	return types.UintKind
 }
 
 // ParseValue implements TypeInfo interface.
@@ -166,14 +162,11 @@ func (ti *enumType) ParseValue(str *string) (types.Value, error) {
 	if str == nil || *str == "" {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlEnumType.Convert(*str)
+	val, err := ti.sqlEnumType.Marshal(*str)
 	if err != nil {
 		return nil, err
 	}
-	if val, ok := strVal.(string); ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" cannot convert the string "%v" to a value`, ti.String(), str)
+	return types.Uint(val), nil
 }
 
 // String implements TypeInfo interface.
diff --git a/go/libraries/doltcore/schema/typeinfo/enum_test.go b/go/libraries/doltcore/schema/typeinfo/enum_test.go
new file mode 100644
index 00000000000..264ad85cf68
--- /dev/null
+++ b/go/libraries/doltcore/schema/typeinfo/enum_test.go
@@ -0,0 +1,258 @@
+// Copyright 2020 Liquidata, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package typeinfo
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/liquidata-inc/dolt/go/store/types"
+)
+
+func TestEnumConvertNomsValueToValue(t *testing.T) {
+	tests := []struct {
+		typ         *enumType
+		input       types.Uint
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateEnumType(t, 3),
+			1,
+			"aaaa",
+			false,
+		},
+		{
+			generateEnumType(t, 5),
+			2,
+			"aaab",
+			false,
+		},
+		{
+			generateEnumType(t, 8),
+			3,
+			"aaac",
+			false,
+		},
+		{
+			generateEnumType(t, 7),
+			7,
+			"aaag",
+			false,
+		},
+		{
+			generateEnumType(t, 2),
+			0,
+			"",
+			true,
+		},
+		{
+			generateEnumType(t, 3),
+			4,
+			"",
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ConvertNomsValueToValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, output)
+			}
+		})
+	}
+}
+
+func TestEnumConvertValueToNomsValue(t *testing.T) {
+	tests := []struct {
+		typ         *enumType
+		input       interface{}
+		output      types.Uint
+		expectedErr bool
+	}{
+		{
+			generateEnumType(t, 4),
+			"aaac",
+			3,
+			false,
+		},
+		{
+			generateEnumType(t, 7),
+			uint64(3),
+			3,
+			false,
+		},
+		{
+			generateEnumType(t, 4),
+			"dog",
+			0,
+			true,
+		},
+		{
+			generateEnumType(t, 3),
+			true,
+			0,
+			true,
+		},
+		{
+			generateEnumType(t, 10),
+			time.Unix(137849, 0),
+			0,
+			true,
+		},
+		{
+			generateEnumType(t, 5),
+			complex128(14i),
+			0,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ConvertValueToNomsValue(test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestEnumFormatValue(t *testing.T) {
+	tests := []struct {
+		typ         *enumType
+		input       types.Uint
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateEnumType(t, 3),
+			1,
+			"aaaa",
+			false,
+		},
+		{
+			generateEnumType(t, 5),
+			2,
+			"aaab",
+			false,
+		},
+		{
+			generateEnumType(t, 8),
+			3,
+			"aaac",
+			false,
+		},
+		{
+			generateEnumType(t, 7),
+			7,
+			"aaag",
+			false,
+		},
+		{
+			generateEnumType(t, 2),
+			0,
+			"",
+			true,
+		},
+		{
+			generateEnumType(t, 3),
+			4,
+			"",
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.FormatValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, *output)
+			}
+		})
+	}
+}
+
+func TestEnumParseValue(t *testing.T) {
+	tests := []struct {
+		typ         *enumType
+		input       string
+		output      types.Uint
+		expectedErr bool
+	}{
+		{
+			generateEnumType(t, 3),
+			"aaaa",
+			1,
+			false,
+		},
+		{
+			generateEnumType(t, 5),
+			"aaab",
+			2,
+			false,
+		},
+		{
+			generateEnumType(t, 8),
+			"aaac",
+			3,
+			false,
+		},
+		{
+			generateEnumType(t, 7),
+			"aaag",
+			7,
+			false,
+		},
+		{
+			generateEnumType(t, 2),
+			"dog",
+			0,
+			true,
+		},
+		{
+			generateEnumType(t, 3),
+			"aaad",
+			4,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ParseValue(&test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
diff --git a/go/libraries/doltcore/schema/typeinfo/set.go b/go/libraries/doltcore/schema/typeinfo/set.go
index 4d845ac8232..2a7cb9982dc 100644
--- a/go/libraries/doltcore/schema/typeinfo/set.go
+++ b/go/libraries/doltcore/schema/typeinfo/set.go
@@ -67,8 +67,8 @@ func CreateSetTypeFromParams(params map[string]string) (TypeInfo, error) {
 
 // ConvertNomsValueToValue implements TypeInfo interface.
 func (ti *setType) ConvertNomsValueToValue(v types.Value) (interface{}, error) {
-	if val, ok := v.(types.String); ok {
-		res, err := ti.sqlSetType.Convert(string(val))
+	if val, ok := v.(types.Uint); ok {
+		res, err := ti.sqlSetType.Unmarshal(uint64(val))
 		if err != nil {
 			return nil, fmt.Errorf(`"%v" cannot convert "%v" to value`, ti.String(), val)
 		}
@@ -85,15 +85,11 @@ func (ti *setType) ConvertValueToNomsValue(v interface{}) (types.Value, error) {
 	if v == nil {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlSetType.Convert(v)
+	val, err := ti.sqlSetType.Marshal(v)
 	if err != nil {
 		return nil, err
 	}
-	val, ok := strVal.(string)
-	if ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" cannot convert value "%v" of type "%T" as it is invalid`, ti.String(), v, v)
+	return types.Uint(val), nil
 }
 
 // Equals implements TypeInfo interface.
@@ -158,22 +154,19 @@ func (ti *setType) IsValid(v types.Value) bool {
 
 // NomsKind implements TypeInfo interface.
 func (ti *setType) NomsKind() types.NomsKind {
-	return types.StringKind
+	return types.UintKind
 }
 
 // ParseValue implements TypeInfo interface.
 func (ti *setType) ParseValue(str *string) (types.Value, error) {
-	if str == nil || *str == "" {
+	if str == nil {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlSetType.Convert(*str)
+	val, err := ti.sqlSetType.Marshal(*str)
 	if err != nil {
 		return nil, err
 	}
-	if val, ok := strVal.(string); ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" cannot convert the string "%v" to a value`, ti.String(), str)
+	return types.Uint(val), nil
 }
 
 // String implements TypeInfo interface.
diff --git a/go/libraries/doltcore/schema/typeinfo/set_test.go b/go/libraries/doltcore/schema/typeinfo/set_test.go
new file mode 100644
index 00000000000..c7120675a10
--- /dev/null
+++ b/go/libraries/doltcore/schema/typeinfo/set_test.go
@@ -0,0 +1,270 @@
+// Copyright 2020 Liquidata, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package typeinfo
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/liquidata-inc/dolt/go/store/types"
+)
+
+func TestSetConvertNomsValueToValue(t *testing.T) {
+	tests := []struct {
+		typ         *setType
+		input       types.Uint
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateSetType(t, 2),
+			0,
+			"",
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			1,
+			"aa",
+			false,
+		},
+		{
+			generateSetType(t, 5),
+			2,
+			"ab",
+			false,
+		},
+		{
+			generateSetType(t, 8),
+			3,
+			"aa,ab",
+			false,
+		},
+		{
+			generateSetType(t, 7),
+			4,
+			"ac",
+			false,
+		},
+		{
+			generateSetType(t, 4),
+			7,
+			"aa,ab,ac",
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			8,
+			"",
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ConvertNomsValueToValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, output)
+			}
+		})
+	}
+}
+
+func TestSetConvertValueToNomsValue(t *testing.T) {
+	tests := []struct {
+		typ         *setType
+		input       interface{}
+		output      types.Uint
+		expectedErr bool
+	}{
+		{
+			generateSetType(t, 4),
+			"aa,ab",
+			3,
+			false,
+		},
+		{
+			generateSetType(t, 7),
+			uint64(3),
+			3,
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			true,
+			0,
+			true,
+		},
+		{
+			generateSetType(t, 10),
+			time.Unix(137849, 0),
+			0,
+			true,
+		},
+		{
+			generateSetType(t, 5),
+			complex128(14i),
+			0,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ConvertValueToNomsValue(test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestSetFormatValue(t *testing.T) {
+	tests := []struct {
+		typ         *setType
+		input       types.Uint
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateSetType(t, 2),
+			0,
+			"",
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			1,
+			"aa",
+			false,
+		},
+		{
+			generateSetType(t, 5),
+			2,
+			"ab",
+			false,
+		},
+		{
+			generateSetType(t, 8),
+			3,
+			"aa,ab",
+			false,
+		},
+		{
+			generateSetType(t, 7),
+			4,
+			"ac",
+			false,
+		},
+		{
+			generateSetType(t, 4),
+			7,
+			"aa,ab,ac",
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			8,
+			"",
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.FormatValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, *output)
+			}
+		})
+	}
+}
+
+func TestSetParseValue(t *testing.T) {
+	tests := []struct {
+		typ         *setType
+		input       string
+		output      types.Uint
+		expectedErr bool
+	}{
+		{
+			generateSetType(t, 2),
+			"",
+			0,
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			"aa",
+			1,
+			false,
+		},
+		{
+			generateSetType(t, 5),
+			"ab",
+			2,
+			false,
+		},
+		{
+			generateSetType(t, 8),
+			"aa,ab",
+			3,
+			false,
+		},
+		{
+			generateSetType(t, 7),
+			"ac",
+			4,
+			false,
+		},
+		{
+			generateSetType(t, 4),
+			"aa,ab,ac",
+			7,
+			false,
+		},
+		{
+			generateSetType(t, 3),
+			"ad",
+			0,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			output, err := test.typ.ParseValue(&test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
diff --git a/go/libraries/doltcore/schema/typeinfo/time.go b/go/libraries/doltcore/schema/typeinfo/time.go
index 4eb11508bf0..a61d184aed3 100644
--- a/go/libraries/doltcore/schema/typeinfo/time.go
+++ b/go/libraries/doltcore/schema/typeinfo/time.go
@@ -34,9 +34,8 @@ var TimeType = &timeType{sql.Time}
 
 // ConvertNomsValueToValue implements TypeInfo interface.
 func (ti *timeType) ConvertNomsValueToValue(v types.Value) (interface{}, error) {
-	//TODO: expose the MySQL type's microsecond implementation and persist that to disk? Enables sorting
-	if val, ok := v.(types.String); ok {
-		return string(val), nil
+	if val, ok := v.(types.Int); ok {
+		return ti.sqlTimeType.Unmarshal(int64(val)), nil
 	}
 	if _, ok := v.(types.Null); ok || v == nil {
 		return nil, nil
@@ -49,15 +48,11 @@ func (ti *timeType) ConvertValueToNomsValue(v interface{}) (types.Value, error)
 	if v == nil {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlTimeType.Convert(v)
+	val, err := ti.sqlTimeType.Marshal(v)
 	if err != nil {
 		return nil, err
 	}
-	val, ok := strVal.(string)
-	if ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" cannot convert value "%v" of type "%T" as it is invalid`, ti.String(), v, v)
+	return types.Int(val), nil
 }
 
 // Equals implements TypeInfo interface.
@@ -71,14 +66,18 @@ func (ti *timeType) Equals(other TypeInfo) bool {
 
 // FormatValue implements TypeInfo interface.
 func (ti *timeType) FormatValue(v types.Value) (*string, error) {
-	if val, ok := v.(types.String); ok {
-		res := string(val)
-		return &res, nil
-	}
 	if _, ok := v.(types.Null); ok || v == nil {
 		return nil, nil
 	}
-	return nil, fmt.Errorf(`"%v" cannot convert NomsKind "%v" to a string`, ti.String(), v.Kind())
+	strVal, err := ti.ConvertNomsValueToValue(v)
+	if err != nil {
+		return nil, err
+	}
+	val, ok := strVal.(string)
+	if !ok {
+		return nil, fmt.Errorf(`"%v" has unexpectedly encountered a value of type "%T" from embedded type`, ti.String(), v)
+	}
+	return &val, nil
 }
 
 // GetTypeIdentifier implements TypeInfo interface.
@@ -99,7 +98,7 @@ func (ti *timeType) IsValid(v types.Value) bool {
 
 // NomsKind implements TypeInfo interface.
 func (ti *timeType) NomsKind() types.NomsKind {
-	return types.StringKind
+	return types.IntKind
 }
 
 // ParseValue implements TypeInfo interface.
@@ -107,14 +106,11 @@ func (ti *timeType) ParseValue(str *string) (types.Value, error) {
 	if str == nil || *str == "" {
 		return types.NullValue, nil
 	}
-	strVal, err := ti.sqlTimeType.Convert(*str)
+	val, err := ti.sqlTimeType.Marshal(*str)
 	if err != nil {
 		return nil, err
 	}
-	if val, ok := strVal.(string); ok {
-		return types.String(val), nil
-	}
-	return nil, fmt.Errorf(`"%v" cannot convert the string "%v" to a value`, ti.String(), str)
+	return types.Int(val), nil
 }
 
 // String implements TypeInfo interface.
diff --git a/go/libraries/doltcore/schema/typeinfo/time_test.go b/go/libraries/doltcore/schema/typeinfo/time_test.go
new file mode 100644
index 00000000000..574fae9cd45
--- /dev/null
+++ b/go/libraries/doltcore/schema/typeinfo/time_test.go
@@ -0,0 +1,230 @@
+// Copyright 2020 Liquidata, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package typeinfo
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/liquidata-inc/dolt/go/store/types"
+)
+
+func TestTimeConvertNomsValueToValue(t *testing.T) {
+	tests := []struct {
+		input       types.Int
+		output      string
+		expectedErr bool
+	}{
+		{
+			1000000,
+			"00:00:01",
+			false,
+		},
+		{
+			113000000,
+			"00:01:53",
+			false,
+		},
+		{
+			247019000000,
+			"68:36:59",
+			false,
+		},
+		{
+			458830485214,
+			"127:27:10.485214",
+			false,
+		},
+		{
+			-3020399000000,
+			"-838:59:59",
+			false,
+		},
+		{ // no integer can cause an error, values beyond the max/min are set equal to the max/min
+			922337203685477580,
+			"838:59:59",
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v`, test.input), func(t *testing.T) {
+			output, err := TimeType.ConvertNomsValueToValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, output)
+			}
+		})
+	}
+}
+
+func TestTimeConvertValueToNomsValue(t *testing.T) {
+	tests := []struct {
+		input       interface{}
+		output      types.Int
+		expectedErr bool
+	}{
+		{
+			153,
+			113000000,
+			false,
+		},
+		{
+			1.576,
+			1576000,
+			false,
+		},
+		{
+			"68:36:59",
+			247019000000,
+			false,
+		},
+		{
+			"683659",
+			247019000000,
+			false,
+		},
+		{
+			"dog",
+			0,
+			true,
+		},
+		{
+			true,
+			0,
+			true,
+		},
+		{
+			time.Unix(137849, 0),
+			0,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v`, test.input), func(t *testing.T) {
+			output, err := TimeType.ConvertValueToNomsValue(test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestTimeFormatValue(t *testing.T) {
+	tests := []struct {
+		input       types.Int
+		output      string
+		expectedErr bool
+	}{
+		{
+			1000000,
+			"00:00:01",
+			false,
+		},
+		{
+			113000000,
+			"00:01:53",
+			false,
+		},
+		{
+			247019000000,
+			"68:36:59",
+			false,
+		},
+		{
+			458830485214,
+			"127:27:10.485214",
+			false,
+		},
+		{
+			-3020399000000,
+			"-838:59:59",
+			false,
+		},
+		{
+			922337203685477580,
+			"838:59:59",
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v`, test.input), func(t *testing.T) {
+			output, err := TimeType.FormatValue(test.input)
+			if test.expectedErr {
+				assert.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.output, *output)
+			}
+		})
+	}
+}
+
+func TestTimeParseValue(t *testing.T) {
+	tests := []struct {
+		input       string
+		output      types.Int
+		expectedErr bool
+	}{
+		{
+			"683659",
+			247019000000,
+			false,
+		},
+		{
+			"127:27:10.485214",
+			458830485214,
+			false,
+		},
+		{
+			"-838:59:59",
+			-3020399000000,
+			false,
+		},
+		{
+			"850:00:00",
+			3020399000000,
+			false,
+		},
+		{
+			"dog",
+			0,
+			true,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v`, test.input), func(t *testing.T) {
+			output, err := TimeType.ParseValue(&test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
diff --git a/go/libraries/doltcore/schema/typeinfo/typeinfo.go b/go/libraries/doltcore/schema/typeinfo/typeinfo.go
index 0932dfe1786..39b1f621d83 100644
--- a/go/libraries/doltcore/schema/typeinfo/typeinfo.go
+++ b/go/libraries/doltcore/schema/typeinfo/typeinfo.go
@@ -139,20 +139,12 @@ func FromSqlType(sqlType sql.Type) (TypeInfo, error) {
 	case sqltypes.Date:
 		return DateType, nil
 	case sqltypes.Time:
-		//TODO: determine the storage format
-		if fmt.Sprintf("a") != "" { // always evaluates to true, compiler won't complain about unreachable code
-			return nil, fmt.Errorf(`"%v" has not yet been implemented`, sqlType.String())
-		}
 		return TimeType, nil
 	case sqltypes.Datetime:
 		return DatetimeType, nil
 	case sqltypes.Year:
 		return YearType, nil
 	case sqltypes.Decimal:
-		//TODO: determine the storage format
-		if fmt.Sprintf("a") != "" { // always evaluates to true, compiler won't complain about unreachable code
-			return nil, fmt.Errorf(`"%v" has not yet been implemented`, sqlType.String())
-		}
 		decimalSQLType, ok := sqlType.(sql.DecimalType)
 		if !ok {
 			return nil, fmt.Errorf(`expected "DecimalTypeIdentifier" from SQL basetype "Decimal"`)
@@ -213,20 +205,12 @@ func FromSqlType(sqlType sql.Type) (TypeInfo, error) {
 		}
 		return &bitType{bitSQLType}, nil
 	case sqltypes.Enum:
-		//TODO: determine the storage format
-		if fmt.Sprintf("a") != "" { // always evaluates to true, compiler won't complain about unreachable code
-			return nil, fmt.Errorf(`"%v" has not yet been implemented`, sqlType.String())
-		}
 		enumSQLType, ok := sqlType.(sql.EnumType)
 		if !ok {
 			return nil, fmt.Errorf(`expected "EnumTypeIdentifier" from SQL basetype "Enum"`)
 		}
 		return &enumType{enumSQLType}, nil
 	case sqltypes.Set:
-		//TODO: determine the storage format
-		if fmt.Sprintf("a") != "" { // always evaluates to true, compiler won't complain about unreachable code
-			return nil, fmt.Errorf(`"%v" has not yet been implemented`, sqlType.String())
-		}
 		setSQLType, ok := sqlType.(sql.SetType)
 		if !ok {
 			return nil, fmt.Errorf(`expected "SetTypeIdentifier" from SQL basetype "Set"`)
diff --git a/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go b/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go
index f54cdb84b73..88abc026676 100644
--- a/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go
+++ b/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go
@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/shopspring/decimal"
 	"github.com/src-d/go-mysql-server/sql"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -68,8 +69,7 @@ func verifyTypeInfoArrays(t *testing.T, tiArrays [][]TypeInfo, vaArrays [][]type
 	// delete any types that should not be tested
 	delete(seenTypeInfos, UnknownTypeIdentifier)
 	delete(seenTypeInfos, TupleTypeIdentifier)
-	//TODO: determine the storage format for DecimalType and VarBinaryType
-	delete(seenTypeInfos, DecimalTypeIdentifier)
+	//TODO: determine the storage format for VarBinaryType
 	delete(seenTypeInfos, VarBinaryTypeIdentifier)
 	for _, tiArray := range tiArrays {
 		// no row should be empty
@@ -120,7 +120,7 @@ func testTypeInfoConvertRoundTrip(t *testing.T, tiArrays [][]TypeInfo, vaArrays
 								if ti == DateType { // Special case as DateType removes the hh:mm:ss
 									val = types.Timestamp(time.Time(val.(types.Timestamp)).Truncate(24 * time.Hour))
 									require.True(t, val.Equals(outVal), "\"%v\"\n\"%v\"", val, outVal)
-								} else {
+								} else if ti.GetTypeIdentifier() != DecimalTypeIdentifier { // Any Decimal's on-disk representation varies by precision/scale
 									require.True(t, val.Equals(outVal), "\"%v\"\n\"%v\"", val, outVal)
 								}
 							} else {
@@ -220,7 +220,7 @@ func testTypeInfoFormatParseRoundTrip(t *testing.T, tiArrays [][]TypeInfo, vaArr
 								if ti == DateType { // special case as DateType removes the hh:mm:ss
 									val = types.Timestamp(time.Time(val.(types.Timestamp)).Truncate(24 * time.Hour))
 									require.True(t, val.Equals(outVal), "\"%v\"\n\"%v\"", val, outVal)
-								} else {
+								} else if ti.GetTypeIdentifier() != DecimalTypeIdentifier { // Any Decimal's on-disk representation varies by precision/scale
 									require.True(t, val.Equals(outVal), "\"%v\"\n\"%v\"", val, outVal)
 								}
 							} else {
@@ -330,7 +330,7 @@ func generateTypeInfoArrays(t *testing.T) ([][]TypeInfo, [][]types.Value) {
 			generateBitTypes(t, 16),
 			{BoolType},
 			{DateType, DatetimeType, TimestampType},
-			//generateDecimalTypes(t, 16),
+			generateDecimalTypes(t, 16),
 			generateEnumTypes(t, 16),
 			{Float32Type, Float64Type},
 			{InlineBlobType},
@@ -355,16 +355,19 @@ func generateTypeInfoArrays(t *testing.T) ([][]TypeInfo, [][]types.Value) {
 				types.Timestamp(time.Date(2000, 2, 28, 14, 38, 43, 583395000, time.UTC)),
 				types.Timestamp(time.Date(2038, 1, 19, 3, 14, 7, 999999000, time.UTC)),
 				types.Timestamp(time.Date(9999, 12, 31, 23, 59, 59, 999999000, time.UTC))},
-			//{types.String("1"), types.String("-1.5"), types.String("4723245"), //Decimal
-			//	types.String("8923583.125"), types.String("1198728394234798423466321.27349757")},
-			{types.String("aaaa"), types.String("aaaa,aaac"), types.String("aaag"), types.String("aaab,aaad,aaaf"), types.String("aaag,aaah")},                                         //Enum
-			{types.Float(1.0), types.Float(65513.75), types.Float(4293902592), types.Float(4.58E71), types.Float(7.172E285)},                                                           //Float
-			{types.InlineBlob{0}, types.InlineBlob{21}, types.InlineBlob{1, 17}, types.InlineBlob{72, 42}, types.InlineBlob{21, 122, 236}},                                             //InlineBlob
-			{types.Int(20), types.Int(215), types.Int(237493), types.Int(2035753568), types.Int(2384384576063)},                                                                        //Int
-			{types.String("aa"), types.String("aa,ac"), types.String("ag"), types.String("ab,ad,af"), types.String("ag,ah")},                                                           //Set
-			{types.String("00:00:00"), types.String("00:00:01"), types.String("00:01:53"), types.String("68:36:59"), types.String("127:27:10.485214")},                                 //Time
-			{types.Uint(20), types.Uint(275), types.Uint(328395), types.Uint(630257298), types.Uint(93897259874)},                                                                      //Uint
-			{types.UUID{3}, types.UUID{3, 13}, types.UUID{128, 238, 82, 12}, types.UUID{31, 54, 23, 13, 63, 43}, types.UUID{83, 64, 21, 14, 42, 6, 35, 7, 54, 234, 6, 32, 1, 4, 2, 4}}, //Uuid
+			{types.Decimal(decimal.RequireFromString("0")), //Decimal
+				types.Decimal(decimal.RequireFromString("-1.5")),
+				types.Decimal(decimal.RequireFromString("4723245")),
+				types.Decimal(decimal.RequireFromString("-1076416.875")),
+				types.Decimal(decimal.RequireFromString("198728394234798423466321.27349757"))},
+			{types.Uint(1), types.Uint(3), types.Uint(5), types.Uint(7), types.Uint(8)},                                                                                                    //Enum
+			{types.Float(1.0), types.Float(65513.75), types.Float(4293902592), types.Float(4.58E71), types.Float(7.172E285)},                                                               //Float
+			{types.InlineBlob{0}, types.InlineBlob{21}, types.InlineBlob{1, 17}, types.InlineBlob{72, 42}, types.InlineBlob{21, 122, 236}},                                                 //InlineBlob
+			{types.Int(20), types.Int(215), types.Int(237493), types.Int(2035753568), types.Int(2384384576063)},                                                                            //Int
+			{types.Uint(1), types.Uint(5), types.Uint(64), types.Uint(42), types.Uint(192)},                                                                                                //Set
+			{types.Int(0), types.Int(1000000 /*"00:00:01"*/), types.Int(113000000 /*"00:01:53"*/), types.Int(247019000000 /*"68:36:59"*/), types.Int(458830485214 /*"127:27:10.485214"*/)}, //Time
+			{types.Uint(20), types.Uint(275), types.Uint(328395), types.Uint(630257298), types.Uint(93897259874)},                                                                          //Uint
+			{types.UUID{3}, types.UUID{3, 13}, types.UUID{128, 238, 82, 12}, types.UUID{31, 54, 23, 13, 63, 43}, types.UUID{83, 64, 21, 14, 42, 6, 35, 7, 54, 234, 6, 32, 1, 4, 2, 4}},     //Uuid
 			//{types.String([]byte{1}), types.String([]byte{42, 52}), types.String([]byte{84, 32, 13, 63, 12, 86}), //VarBinary
 			//	types.String([]byte{1, 32, 235, 64, 32, 23, 45, 76}), types.String([]byte{123, 234, 34, 223, 76, 35, 32, 12, 84, 26, 15, 34, 65, 86, 45, 23, 43, 12, 76, 154, 234, 76, 34})},
 			{types.String(""), types.String("a"), types.String("abc"), //VarString
diff --git a/go/libraries/doltcore/schema/typeinfo/varstring_test.go b/go/libraries/doltcore/schema/typeinfo/varstring_test.go
index 9e682afbeab..b6d3bcd9215 100644
--- a/go/libraries/doltcore/schema/typeinfo/varstring_test.go
+++ b/go/libraries/doltcore/schema/typeinfo/varstring_test.go
@@ -22,7 +22,6 @@ import (
 	"github.com/src-d/go-mysql-server/sql"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"vitess.io/vitess/go/sqltypes"
 
 	"github.com/liquidata-inc/dolt/go/store/types"
 )
@@ -264,26 +263,3 @@ func TestVarStringParseValue(t *testing.T) {
 		})
 	}
 }
-
-func generateVarStringTypes(t *testing.T, numOfTypes uint16) []TypeInfo {
-	var res []TypeInfo
-	loop(t, 1, 500, numOfTypes, func(i int64) {
-		rts := false
-		if i%2 == 0 {
-			rts = true
-		}
-		res = append(res, generateVarStringType(t, i, rts))
-	})
-	return res
-}
-
-func generateVarStringType(t *testing.T, length int64, rts bool) *varStringType {
-	require.True(t, length > 0)
-	if rts {
-		t, err := sql.CreateStringWithDefaults(sqltypes.Char, length)
-		if err == nil {
-			return &varStringType{t}
-		}
-	}
-	return &varStringType{sql.MustCreateStringWithDefaults(sqltypes.VarChar, length)}
-}
diff --git a/go/libraries/doltcore/sqle/indexes.go b/go/libraries/doltcore/sqle/indexes.go
index 95b11647fa4..48c4c19c73b 100644
--- a/go/libraries/doltcore/sqle/indexes.go
+++ b/go/libraries/doltcore/sqle/indexes.go
@@ -23,7 +23,6 @@ import (
 
 	"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"
 )
 
 // IndexDriver implementation. Not ready for prime time.
@@ -96,23 +95,18 @@ type doltIndex struct {
 }
 
 func (di *doltIndex) Get(key ...interface{}) (sql.IndexLookup, error) {
-	taggedVals, err := keyColsToTuple(di.sch, key)
-	if err != nil {
-		return nil, err
-	}
-
-	return &doltIndexLookup{di, taggedVals}, nil
-}
-
-func keyColsToTuple(sch schema.Schema, key []interface{}) (row.TaggedValues, error) {
-	if sch.GetPKCols().Size() != len(key) {
+	if di.sch.GetPKCols().Size() != len(key) {
 		return nil, errors.New("key must specify all columns")
 	}
 
 	var i int
 	taggedVals := make(row.TaggedValues)
-	err := sch.GetPKCols().Iter(func(tag uint64, col schema.Column) (stop bool, err error) {
-		taggedVals[tag] = keyColToValue(key[i], col)
+	err := di.sch.GetPKCols().Iter(func(tag uint64, col schema.Column) (stop bool, err error) {
+		val, err := col.TypeInfo.ConvertValueToNomsValue(key[i])
+		if err != nil {
+			return true, err
+		}
+		taggedVals[tag] = val
 		i++
 		return false, nil
 	})
@@ -121,63 +115,7 @@ func keyColsToTuple(sch schema.Schema, key []interface{}) (row.TaggedValues, err
 		return nil, err
 	}
 
-	return taggedVals, nil
-}
-
-func keyColToValue(v interface{}, column schema.Column) types.Value {
-	// TODO: type conversion
-	switch column.Kind {
-	case types.BoolKind:
-		return types.Bool(v.(bool))
-	case types.IntKind:
-		switch i := v.(type) {
-		case int:
-			return types.Int(i)
-		case int8:
-			return types.Int(i)
-		case int16:
-			return types.Int(i)
-		case int32:
-			return types.Int(i)
-		case int64:
-			return types.Int(i)
-		default:
-			panic(fmt.Sprintf("unhandled type %T", i))
-		}
-	case types.FloatKind:
-		return types.Float(v.(float64))
-	case types.UintKind:
-		switch i := v.(type) {
-		case int:
-			return types.Uint(i)
-		case int8:
-			return types.Uint(i)
-		case int16:
-			return types.Uint(i)
-		case int32:
-			return types.Uint(i)
-		case int64:
-			return types.Uint(i)
-		case uint:
-			return types.Uint(i)
-		case uint8:
-			return types.Uint(i)
-		case uint16:
-			return types.Uint(i)
-		case uint32:
-			return types.Uint(i)
-		case uint64:
-			return types.Uint(i)
-		default:
-			panic(fmt.Sprintf("unhandled type %T", i))
-		}
-	case types.UUIDKind:
-		panic("Implement me")
-	case types.StringKind:
-		return types.String(v.(string))
-	default:
-		panic(fmt.Sprintf("unhandled type %T", v))
-	}
+	return &doltIndexLookup{di, taggedVals}, nil
 }
 
 func (*doltIndex) Has(partition sql.Partition, key ...interface{}) (bool, error) {
diff --git a/go/store/types/decimal.go b/go/store/types/decimal.go
new file mode 100644
index 00000000000..c43b0ac0696
--- /dev/null
+++ b/go/store/types/decimal.go
@@ -0,0 +1,109 @@
+// Copyright 2020 Liquidata, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package types
+
+import (
+	"context"
+
+	"github.com/shopspring/decimal"
+
+	"github.com/liquidata-inc/dolt/go/store/hash"
+)
+
+type Decimal decimal.Decimal
+
+func (v Decimal) Value(ctx context.Context) (Value, error) {
+	return v, nil
+}
+
+func (v Decimal) Equals(other Value) bool {
+	v2, ok := other.(Decimal)
+	if !ok {
+		return false
+	}
+
+	return decimal.Decimal(v).Equal(decimal.Decimal(v2))
+}
+
+func (v Decimal) Less(nbf *NomsBinFormat, other LesserValuable) (bool, error) {
+	if v2, ok := other.(Decimal); ok {
+		return decimal.Decimal(v).LessThan(decimal.Decimal(v2)), nil
+	}
+	return DecimalKind < other.Kind(), nil
+}
+
+func (v Decimal) Hash(nbf *NomsBinFormat) (hash.Hash, error) {
+	return getHash(v, nbf)
+}
+
+func (v Decimal) isPrimitive() bool {
+	return true
+}
+
+func (v Decimal) WalkValues(ctx context.Context, cb ValueCallback) error {
+	return nil
+}
+
+func (v Decimal) WalkRefs(nbf *NomsBinFormat, cb RefCallback) error {
+	return nil
+}
+
+func (v Decimal) typeOf() (*Type, error) {
+	return PrimitiveTypeMap[DecimalKind], nil
+}
+
+func (v Decimal) Kind() NomsKind {
+	return DecimalKind
+}
+
+func (v Decimal) valueReadWriter() ValueReadWriter {
+	return nil
+}
+
+func (v Decimal) writeTo(w nomsWriter, nbf *NomsBinFormat) error {
+	encodedDecimal, err := decimal.Decimal(v).GobEncode()
+	if err != nil {
+		return err
+	}
+
+	err = DecimalKind.writeTo(w, nbf)
+	if err != nil {
+		return err
+	}
+
+	w.writeUint16(uint16(len(encodedDecimal)))
+	w.writeRaw(encodedDecimal)
+	return nil
+}
+
+func (v Decimal) readFrom(nbf *NomsBinFormat, b *binaryNomsReader) (Value, error) {
+	size := uint32(b.readUint16())
+	db := b.readBytes(size)
+	dec := decimal.Decimal{}
+	err := dec.GobDecode(db)
+	if err != nil {
+		return nil, err
+	}
+	return Decimal(dec), nil
+}
+
+func (v Decimal) skip(nbf *NomsBinFormat, b *binaryNomsReader) {
+	size := uint32(b.readUint16())
+	b.skipBytes(size)
+}
+
+func (v Decimal) HumanReadableString() string {
+	return decimal.Decimal(v).String()
+}
diff --git a/go/store/types/decimal_test.go b/go/store/types/decimal_test.go
new file mode 100644
index 00000000000..fd8b1eb5a28
--- /dev/null
+++ b/go/store/types/decimal_test.go
@@ -0,0 +1,34 @@
+// Copyright 2020 Liquidata, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package types
+
+import (
+	"testing"
+
+	"github.com/shopspring/decimal"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDecimalLibraryEncoding(t *testing.T) {
+	expectedBytes := []byte{255, 255, 255, 250, 3, 25, 222, 110, 95, 84, 132}
+	dec := decimal.RequireFromString("-28443125.175428")
+	bytes, err := dec.GobEncode()
+	require.NoError(t, err)
+	require.Equal(t, expectedBytes, bytes)
+	expectedDec := decimal.Decimal{}
+	err = expectedDec.GobDecode(expectedBytes)
+	require.NoError(t, err)
+	require.True(t, expectedDec.Equal(dec))
+}
diff --git a/go/store/types/map_test.go b/go/store/types/map_test.go
index 615bb52c684..c40ecf3877b 100644
--- a/go/store/types/map_test.go
+++ b/go/store/types/map_test.go
@@ -31,6 +31,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/shopspring/decimal"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
 
@@ -1668,6 +1669,30 @@ func TestMapOrdering(t *testing.T) {
 			Timestamp(time.Unix(9000, 0).UTC()),
 		},
 	)
+
+	testMapOrder(assert, vrw,
+		PrimitiveTypeMap[DecimalKind], PrimitiveTypeMap[StringKind],
+		[]Value{
+			Decimal(decimal.RequireFromString("-99.125434")), String("unused"),
+			Decimal(decimal.RequireFromString("482.124")), String("unused"),
+			Decimal(decimal.RequireFromString("858093.12654")), String("unused"),
+			Decimal(decimal.RequireFromString("1")), String("unused"),
+			Decimal(decimal.RequireFromString("-99.125432")), String("unused"),
+			Decimal(decimal.RequireFromString("0")), String("unused"),
+			Decimal(decimal.RequireFromString("-123845")), String("unused"),
+			Decimal(decimal.RequireFromString("-99.125433")), String("unused"),
+		},
+		[]Value{
+			Decimal(decimal.RequireFromString("-123845")),
+			Decimal(decimal.RequireFromString("-99.125434")),
+			Decimal(decimal.RequireFromString("-99.125433")),
+			Decimal(decimal.RequireFromString("-99.125432")),
+			Decimal(decimal.RequireFromString("0")),
+			Decimal(decimal.RequireFromString("1")),
+			Decimal(decimal.RequireFromString("482.124")),
+			Decimal(decimal.RequireFromString("858093.12654")),
+		},
+	)
 }
 
 func TestMapEmpty(t *testing.T) {
diff --git a/go/store/types/noms_kind.go b/go/store/types/noms_kind.go
index 5c05db2ee18..af26427f3b7 100644
--- a/go/store/types/noms_kind.go
+++ b/go/store/types/noms_kind.go
@@ -54,6 +54,7 @@ const (
 	TupleKind
 	InlineBlobKind
 	TimestampKind
+	DecimalKind
 
 	UnknownKind NomsKind = 255
 )
@@ -79,6 +80,7 @@ var KindToType = map[NomsKind]Value{
 	TupleKind:      EmptyTuple(Format_7_18),
 	InlineBlobKind: InlineBlob{},
 	TimestampKind:  Timestamp{},
+	DecimalKind:    Decimal{},
 }
 
 var KindToTypeSlice []Value
@@ -105,6 +107,7 @@ var KindToString = map[NomsKind]string{
 	TupleKind:      "Tuple",
 	InlineBlobKind: "InlineBlob",
 	TimestampKind:  "Timestamp",
+	DecimalKind:    "Decimal",
 }
 
 // String returns the name of the kind.
diff --git a/go/store/types/type_test.go b/go/store/types/type_test.go
index 5db34846592..8f1dbfbd039 100644
--- a/go/store/types/type_test.go
+++ b/go/store/types/type_test.go
@@ -73,6 +73,7 @@ func TestTypeRefDescribe(t *testing.T) {
 	assert.Equal("Int", mustString(PrimitiveTypeMap[IntKind].Describe(context.Background())))
 	assert.Equal("Uint", mustString(PrimitiveTypeMap[UintKind].Describe(context.Background())))
 	assert.Equal("InlineBlob", mustString(PrimitiveTypeMap[InlineBlobKind].Describe(context.Background())))
+	assert.Equal("Decimal", mustString(PrimitiveTypeMap[DecimalKind].Describe(context.Background())))
 	assert.Equal("Map<String, Float>", mustString(mapType.Describe(context.Background())))
 	assert.Equal("Set<String>", mustString(setType.Describe(context.Background())))
 
@@ -93,6 +94,7 @@ func TestTypeOrdered(t *testing.T) {
 	assert.True(isKindOrderedByValue(PrimitiveTypeMap[IntKind].TargetKind()))
 	assert.True(isKindOrderedByValue(PrimitiveTypeMap[UintKind].TargetKind()))
 	assert.True(isKindOrderedByValue(PrimitiveTypeMap[InlineBlobKind].TargetKind()))
+	assert.True(isKindOrderedByValue(PrimitiveTypeMap[DecimalKind].TargetKind()))
 	assert.True(isKindOrderedByValue(TupleKind))
 
 	assert.False(isKindOrderedByValue(PrimitiveTypeMap[BlobKind].TargetKind()))