From e69b8123990ed6edf57fb3196150fd4141330be6 Mon Sep 17 00:00:00 2001
From: Daylon Wilkins <daylon@liquidata.co>
Date: Wed, 18 Aug 2021 22:06:24 -0700
Subject: [PATCH] Changed TEXT types to use noms Blobs

---
 go/libraries/doltcore/doltdb/root_val.go      |   2 +-
 .../schema/alterschema/modifycolumn_test.go   |   3 +-
 go/libraries/doltcore/schema/typeinfo/bit.go  |   2 +
 .../doltcore/schema/typeinfo/blobstring.go    | 242 ++++++++++++++++++
 .../schema/typeinfo/blobstring_test.go        | 196 ++++++++++++++
 go/libraries/doltcore/schema/typeinfo/bool.go |   2 +
 .../doltcore/schema/typeinfo/common_test.go   |  33 ++-
 .../doltcore/schema/typeinfo/datetime.go      |   2 +
 .../doltcore/schema/typeinfo/decimal.go       |   8 +
 go/libraries/doltcore/schema/typeinfo/enum.go |   2 +
 .../doltcore/schema/typeinfo/float.go         |   2 +
 .../doltcore/schema/typeinfo/inlineblob.go    |   2 +
 go/libraries/doltcore/schema/typeinfo/int.go  |   2 +
 go/libraries/doltcore/schema/typeinfo/json.go |   2 +
 go/libraries/doltcore/schema/typeinfo/set.go  |   2 +
 go/libraries/doltcore/schema/typeinfo/time.go |   2 +
 .../doltcore/schema/typeinfo/typeconverter.go |   2 +
 .../doltcore/schema/typeinfo/typeinfo.go      |   6 +-
 .../doltcore/schema/typeinfo/typeinfo_test.go |  28 +-
 go/libraries/doltcore/schema/typeinfo/uint.go |   2 +
 go/libraries/doltcore/schema/typeinfo/uuid.go |   2 +
 .../doltcore/schema/typeinfo/varbinary.go     |  54 ++--
 .../doltcore/schema/typeinfo/varstring.go     |   8 +-
 go/libraries/doltcore/schema/typeinfo/year.go |   2 +
 .../sqle/enginetest/dolt_engine_test.go       |   3 +
 go/libraries/doltcore/sqle/schema_table.go    |  24 +-
 go/libraries/doltcore/sqle/sqlddl_test.go     |  21 +-
 go/libraries/doltcore/sqle/sqldelete_test.go  |   4 +-
 go/libraries/doltcore/sqle/sqlfmt/row_fmt.go  |   2 +
 .../doltcore/sqle/sqlfmt/schema_fmt_test.go   |   2 +-
 go/libraries/doltcore/sqle/sqlinsert_test.go  |   6 +-
 go/libraries/doltcore/sqle/sqlreplace_test.go |   6 +-
 go/libraries/doltcore/sqle/sqlselect_test.go  |  27 +-
 go/libraries/doltcore/sqle/sqlupdate_test.go  |   8 +-
 .../table/untyped/sqlexport/sqlwriter_test.go |   2 +-
 go/store/types/blob.go                        |  55 +++-
 go/store/types/map_test.go                    |  10 +-
 integration-tests/bats/back-compat.bats       |   6 +-
 integration-tests/bats/column_tags.bats       |   2 +-
 integration-tests/bats/helper/common.bash     |   2 +-
 .../bats/import-create-tables.bats            |   2 +-
 integration-tests/bats/schema-import.bats     |  26 +-
 integration-tests/bats/status.bats            |   2 +-
 integration-tests/bats/types.bats             |  28 +-
 .../backward_compatible_versions.txt          |   3 +-
 .../forward_compatible_versions.txt           |   1 -
 46 files changed, 718 insertions(+), 132 deletions(-)
 create mode 100644 go/libraries/doltcore/schema/typeinfo/blobstring.go
 create mode 100644 go/libraries/doltcore/schema/typeinfo/blobstring_test.go

diff --git a/go/libraries/doltcore/doltdb/root_val.go b/go/libraries/doltcore/doltdb/root_val.go
index bca2982540..c5fae8812e 100644
--- a/go/libraries/doltcore/doltdb/root_val.go
+++ b/go/libraries/doltcore/doltdb/root_val.go
@@ -41,7 +41,7 @@ type FeatureVersion int64
 
 // DoltFeatureVersion is described in feature_version.md.
 // only variable for testing.
-var DoltFeatureVersion FeatureVersion = 1 // last bumped for CHECK constraint storage
+var DoltFeatureVersion FeatureVersion = 2 // last bumped when changing TEXT types to use noms Blobs
 
 // RootValue defines the structure used inside all Dolthub noms dbs
 type RootValue struct {
diff --git a/go/libraries/doltcore/schema/alterschema/modifycolumn_test.go b/go/libraries/doltcore/schema/alterschema/modifycolumn_test.go
index 6725db93f9..1a0958261c 100644
--- a/go/libraries/doltcore/schema/alterschema/modifycolumn_test.go
+++ b/go/libraries/doltcore/schema/alterschema/modifycolumn_test.go
@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/vitess/go/sqltypes"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -38,7 +39,7 @@ func TestModifyColumn(t *testing.T) {
 		schema.NewColumn("is_married", dtestutils.IsMarriedTag, types.BoolKind, false, schema.NotNullConstraint{}),
 		schema.NewColumn("title", dtestutils.TitleTag, types.StringKind, false),
 	)
-	ti, err := typeinfo.FromSqlType(sql.TinyText)
+	ti, err := typeinfo.FromSqlType(sql.MustCreateStringWithDefaults(sqltypes.VarChar, 599))
 	require.NoError(t, err)
 	newNameColSameTag, err := schema.NewColumnWithTypeInfo("name", dtestutils.NameTag, ti, false, "", false, "", schema.NotNullConstraint{})
 	require.NoError(t, err)
diff --git a/go/libraries/doltcore/schema/typeinfo/bit.go b/go/libraries/doltcore/schema/typeinfo/bit.go
index bd724d35e7..5b615124be 100644
--- a/go/libraries/doltcore/schema/typeinfo/bit.go
+++ b/go/libraries/doltcore/schema/typeinfo/bit.go
@@ -208,6 +208,8 @@ func bitTypeConverter(ctx context.Context, src *bitType, destTi TypeInfo) (tc Ty
 		} else {
 			return wrapIsValid(dest.IsValid, src, dest)
 		}
+	case *blobStringType:
+		return bitTypeConverterInterpretAsString(ctx, src, destTi)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/blobstring.go b/go/libraries/doltcore/schema/typeinfo/blobstring.go
new file mode 100644
index 0000000000..e83fa5c8a1
--- /dev/null
+++ b/go/libraries/doltcore/schema/typeinfo/blobstring.go
@@ -0,0 +1,242 @@
+// Copyright 2021 Dolthub, 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 (
+	"context"
+	"fmt"
+	"strconv"
+	"strings"
+	"unicode/utf8"
+
+	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/vitess/go/sqltypes"
+
+	"github.com/dolthub/dolt/go/store/types"
+)
+
+const (
+	blobStringTypeParam_Collate = "collate"
+	blobStringTypeParam_Length  = "length"
+)
+
+type blobStringType struct {
+	sqlStringType sql.StringType
+}
+
+var _ TypeInfo = (*blobStringType)(nil)
+
+func CreateBlobStringTypeFromParams(params map[string]string) (TypeInfo, error) {
+	var length int64
+	var collation sql.Collation
+	var err error
+	if collationStr, ok := params[blobStringTypeParam_Collate]; ok {
+		collation, err = sql.ParseCollation(nil, &collationStr, false)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		return nil, fmt.Errorf(`create blobstring type info is missing param "%v"`, blobStringTypeParam_Collate)
+	}
+	if maxLengthStr, ok := params[blobStringTypeParam_Length]; ok {
+		length, err = strconv.ParseInt(maxLengthStr, 10, 64)
+		if err != nil {
+			return nil, err
+		}
+
+	} else {
+		return nil, fmt.Errorf(`create blobstring type info is missing param "%v"`, blobStringTypeParam_Length)
+	}
+	sqlType, err := sql.CreateString(sqltypes.Text, length, collation)
+	if err != nil {
+		return nil, err
+	}
+	return &blobStringType{sqlType}, nil
+}
+
+// ConvertNomsValueToValue implements TypeInfo interface.
+func (ti *blobStringType) ConvertNomsValueToValue(v types.Value) (interface{}, error) {
+	if val, ok := v.(types.Blob); ok {
+		return fromBlob(val)
+	}
+	if _, ok := v.(types.Null); ok || v == nil {
+		return nil, nil
+	}
+	return nil, fmt.Errorf(`"%v" cannot convert NomsKind "%v" to a value`, ti.String(), v.Kind())
+}
+
+// ReadFrom reads a go value from a noms types.CodecReader directly
+func (ti *blobStringType) ReadFrom(_ *types.NomsBinFormat, reader types.CodecReader) (interface{}, error) {
+	k := reader.PeekKind()
+	switch k {
+	case types.BlobKind:
+		val, err := reader.ReadBlob()
+		if err != nil {
+			return nil, err
+		}
+		return fromBlob(val)
+	case types.NullKind:
+		_ = reader.ReadKind()
+		return nil, nil
+	}
+
+	return nil, fmt.Errorf(`"%v" cannot convert NomsKind "%v" to a value`, ti.String(), k)
+}
+
+// ConvertValueToNomsValue implements TypeInfo interface.
+func (ti *blobStringType) ConvertValueToNomsValue(ctx context.Context, vrw types.ValueReadWriter, v interface{}) (types.Value, error) {
+	if v == nil {
+		return types.NullValue, nil
+	}
+	strVal, err := ti.sqlStringType.Convert(v)
+	if err != nil {
+		return nil, err
+	}
+	val, ok := strVal.(string)
+	if ok && utf8.ValidString(val) { // We need to move utf8 (collation) validation into the server
+		return types.NewBlob(ctx, vrw, strings.NewReader(val))
+	}
+	return nil, fmt.Errorf(`"%v" cannot convert value "%v" of type "%T" as it is invalid`, ti.String(), v, v)
+}
+
+// Equals implements TypeInfo interface.
+func (ti *blobStringType) Equals(other TypeInfo) bool {
+	if other == nil {
+		return false
+	}
+	if ti2, ok := other.(*blobStringType); ok {
+		return ti.sqlStringType.MaxCharacterLength() == ti2.sqlStringType.MaxCharacterLength() &&
+			ti.sqlStringType.Collation().Equals(ti2.sqlStringType.Collation())
+	}
+	return false
+}
+
+// FormatValue implements TypeInfo interface.
+func (ti *blobStringType) FormatValue(v types.Value) (*string, error) {
+	if val, ok := v.(types.Blob); ok {
+		resStr, err := fromBlob(val)
+		if err != nil {
+			return nil, err
+		}
+		return &resStr, 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())
+}
+
+// GetTypeIdentifier implements TypeInfo interface.
+func (ti *blobStringType) GetTypeIdentifier() Identifier {
+	return BlobStringTypeIdentifier
+}
+
+// GetTypeParams implements TypeInfo interface.
+func (ti *blobStringType) GetTypeParams() map[string]string {
+	return map[string]string{
+		blobStringTypeParam_Collate: ti.sqlStringType.Collation().String(),
+		blobStringTypeParam_Length:  strconv.FormatInt(ti.sqlStringType.MaxCharacterLength(), 10),
+	}
+}
+
+// IsValid implements TypeInfo interface.
+func (ti *blobStringType) IsValid(v types.Value) bool {
+	if val, ok := v.(types.Blob); ok {
+		if int64(val.Len()) <= ti.sqlStringType.MaxByteLength() {
+			return true
+		}
+	}
+	if _, ok := v.(types.Null); ok || v == nil {
+		return true
+	}
+	return false
+}
+
+// NomsKind implements TypeInfo interface.
+func (ti *blobStringType) NomsKind() types.NomsKind {
+	return types.BlobKind
+}
+
+// ParseValue implements TypeInfo interface.
+func (ti *blobStringType) ParseValue(ctx context.Context, vrw types.ValueReadWriter, str *string) (types.Value, error) {
+	if str == nil {
+		return types.NullValue, nil
+	}
+	strVal, err := ti.sqlStringType.Convert(*str)
+	if err != nil {
+		return nil, err
+	}
+	if val, ok := strVal.(string); ok {
+		return types.NewBlob(ctx, vrw, strings.NewReader(val))
+	}
+	return nil, fmt.Errorf(`"%v" cannot convert the string "%v" to a value`, ti.String(), str)
+}
+
+// Promote implements TypeInfo interface.
+func (ti *blobStringType) Promote() TypeInfo {
+	return &blobStringType{ti.sqlStringType.Promote().(sql.StringType)}
+}
+
+// String implements TypeInfo interface.
+func (ti *blobStringType) String() string {
+	return fmt.Sprintf(`BlobString(%v, %v)`, ti.sqlStringType.Collation().String(), ti.sqlStringType.MaxCharacterLength())
+}
+
+// ToSqlType implements TypeInfo interface.
+func (ti *blobStringType) ToSqlType() sql.Type {
+	return ti.sqlStringType
+}
+
+// blobStringTypeConverter is an internal function for GetTypeConverter that handles the specific type as the source TypeInfo.
+func blobStringTypeConverter(ctx context.Context, src *blobStringType, destTi TypeInfo) (tc TypeConverter, needsConversion bool, err error) {
+	switch dest := destTi.(type) {
+	case *bitType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapIsValid(dest.IsValid, src, dest)
+	case *boolType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *datetimeType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *decimalType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *enumType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *floatType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *inlineBlobType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *intType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *jsonType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *setType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *timeType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *uintType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *uuidType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *varBinaryType:
+		return wrapIsValid(dest.IsValid, src, dest)
+	case *varStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *yearType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	default:
+		return nil, false, UnhandledTypeConversion.New(src.String(), destTi.String())
+	}
+}
diff --git a/go/libraries/doltcore/schema/typeinfo/blobstring_test.go b/go/libraries/doltcore/schema/typeinfo/blobstring_test.go
new file mode 100644
index 0000000000..16e52553f7
--- /dev/null
+++ b/go/libraries/doltcore/schema/typeinfo/blobstring_test.go
@@ -0,0 +1,196 @@
+// Copyright 2021 Dolthub, 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 (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/dolthub/dolt/go/store/types"
+)
+
+func TestBlobStringConvertNomsValueToValue(t *testing.T) {
+	tests := []struct {
+		typ         *blobStringType
+		input       types.Blob
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateBlobStringType(t, 10),
+			mustBlobString(t, "0  "),
+			"0  ",
+			false,
+		},
+		{
+			generateBlobStringType(t, 80),
+			mustBlobString(t, "this is some text that will be returned"),
+			"this is some text that will be returned",
+			false,
+		},
+		{
+			&blobStringType{sql.CreateLongText(sql.Collation_Default)},
+			mustBlobString(t, "  This is a sentence.  "),
+			"  This is a sentence.  ",
+			false,
+		},
+	}
+
+	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 TestBlobStringConvertValueToNomsValue(t *testing.T) {
+	tests := []struct {
+		typ         *blobStringType
+		input       interface{}
+		output      types.Blob
+		expectedErr bool
+	}{
+		{
+			generateBlobStringType(t, 10),
+			"0  ",
+			mustBlobString(t, "0  "),
+			false,
+		},
+		{
+			generateBlobStringType(t, 80),
+			int64(28354),
+			mustBlobString(t, "28354"),
+			false,
+		},
+		{
+			&blobStringType{sql.CreateLongText(sql.Collation_Default)},
+			float32(3724.75),
+			mustBlobString(t, "3724.75"),
+			false,
+		},
+		{
+			generateBlobStringType(t, 80),
+			time.Date(2030, 1, 2, 4, 6, 3, 472382485, time.UTC),
+			mustBlobString(t, "2030-01-02 04:06:03.472382"),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			vrw := types.NewMemoryValueStore()
+			output, err := test.typ.ConvertValueToNomsValue(context.Background(), vrw, test.input)
+			if !test.expectedErr {
+				require.NoError(t, err)
+				assert.Equal(t, test.output, output)
+			} else {
+				assert.Error(t, err)
+			}
+		})
+	}
+}
+
+func TestBlobStringFormatValue(t *testing.T) {
+	tests := []struct {
+		typ         *blobStringType
+		input       types.Blob
+		output      string
+		expectedErr bool
+	}{
+		{
+			generateBlobStringType(t, 10),
+			mustBlobString(t, "0  "),
+			"0  ",
+			false,
+		},
+		{
+			generateBlobStringType(t, 80),
+			mustBlobString(t, "this is some text that will be returned"),
+			"this is some text that will be returned",
+			false,
+		},
+		{
+			&blobStringType{sql.CreateLongText(sql.Collation_Default)},
+			mustBlobString(t, "  This is a sentence.  "),
+			"  This is a sentence.  ",
+			false,
+		},
+	}
+
+	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 TestBlobStringParseValue(t *testing.T) {
+	tests := []struct {
+		typ         *blobStringType
+		input       string
+		output      types.Blob
+		expectedErr bool
+	}{
+		{
+			generateBlobStringType(t, 10),
+			"0  ",
+			mustBlobString(t, "0  "),
+			false,
+		},
+		{
+			generateBlobStringType(t, 80),
+			"this is some text that will be returned",
+			mustBlobString(t, "this is some text that will be returned"),
+			false,
+		},
+		{
+			&blobStringType{sql.CreateLongText(sql.Collation_Default)},
+			"  This is a sentence.  ",
+			mustBlobString(t, "  This is a sentence.  "),
+			false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf(`%v %v`, test.typ.String(), test.input), func(t *testing.T) {
+			vrw := types.NewMemoryValueStore()
+			output, err := test.typ.ParseValue(context.Background(), vrw, &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/bool.go b/go/libraries/doltcore/schema/typeinfo/bool.go
index d9f435e2a4..558d23cdee 100644
--- a/go/libraries/doltcore/schema/typeinfo/bool.go
+++ b/go/libraries/doltcore/schema/typeinfo/bool.go
@@ -199,6 +199,8 @@ func boolTypeConverter(ctx context.Context, src *boolType, destTi TypeInfo) (tc
 				return types.Uint(0), nil
 			}
 		}, true, nil
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return identityTypeConverter, false, nil
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/common_test.go b/go/libraries/doltcore/schema/typeinfo/common_test.go
index e3047b8bc4..fa05e1de06 100644
--- a/go/libraries/doltcore/schema/typeinfo/common_test.go
+++ b/go/libraries/doltcore/schema/typeinfo/common_test.go
@@ -15,8 +15,11 @@
 package typeinfo
 
 import (
+	"bytes"
+	"context"
 	"math"
 	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/dolthub/go-mysql-server/sql"
@@ -24,6 +27,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/dolthub/dolt/go/libraries/utils/mathutil"
+	"github.com/dolthub/dolt/go/store/types"
 )
 
 func generateBitTypes(t *testing.T, numOfTypes uint16) []TypeInfo {
@@ -112,27 +116,27 @@ func generateSetType(t *testing.T, numOfElements int) *setType {
 	return &setType{sql.MustCreateSetType(vals, sql.Collation_Default)}
 }
 
-func generateVarBinaryTypes(t *testing.T, numOfTypes uint16) []TypeInfo {
+func generateInlineBlobTypes(t *testing.T, numOfTypes uint16) []TypeInfo {
 	var res []TypeInfo
 	loop(t, 1, 500, numOfTypes, func(i int64) {
 		pad := false
 		if i%2 == 0 {
 			pad = true
 		}
-		res = append(res, generateVarBinaryType(t, i, pad))
+		res = append(res, generateInlineBlobType(t, i, pad))
 	})
 	return res
 }
 
-func generateVarBinaryType(t *testing.T, length int64, pad bool) *varBinaryType {
+func generateInlineBlobType(t *testing.T, length int64, pad bool) *inlineBlobType {
 	require.True(t, length > 0)
 	if pad {
 		t, err := sql.CreateBinary(sqltypes.Binary, length)
 		if err == nil {
-			return &varBinaryType{t}
+			return &inlineBlobType{t}
 		}
 	}
-	return &varBinaryType{sql.MustCreateBinary(sqltypes.VarBinary, length)}
+	return &inlineBlobType{sql.MustCreateBinary(sqltypes.VarBinary, length)}
 }
 
 func generateVarStringTypes(t *testing.T, numOfTypes uint16) []TypeInfo {
@@ -158,6 +162,25 @@ func generateVarStringType(t *testing.T, length int64, rts bool) *varStringType
 	return &varStringType{sql.MustCreateStringWithDefaults(sqltypes.VarChar, length)}
 }
 
+func generateBlobStringType(t *testing.T, length int64) *blobStringType {
+	require.True(t, length > 0)
+	return &blobStringType{sql.MustCreateStringWithDefaults(sqltypes.Text, length)}
+}
+
+func mustBlobString(t *testing.T, str string) types.Blob {
+	vrw := types.NewMemoryValueStore()
+	blob, err := types.NewBlob(context.Background(), vrw, strings.NewReader(str))
+	require.NoError(t, err)
+	return blob
+}
+
+func mustBlobBytes(t *testing.T, b []byte) types.Blob {
+	vrw := types.NewMemoryValueStore()
+	blob, err := types.NewBlob(context.Background(), vrw, bytes.NewReader(b))
+	require.NoError(t, err)
+	return blob
+}
+
 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/datetime.go b/go/libraries/doltcore/schema/typeinfo/datetime.go
index b1a5d55681..6edb463230 100644
--- a/go/libraries/doltcore/schema/typeinfo/datetime.go
+++ b/go/libraries/doltcore/schema/typeinfo/datetime.go
@@ -222,6 +222,8 @@ func datetimeTypeConverter(ctx context.Context, src *datetimeType, destTi TypeIn
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/decimal.go b/go/libraries/doltcore/schema/typeinfo/decimal.go
index 0a176678f6..022794ad29 100644
--- a/go/libraries/doltcore/schema/typeinfo/decimal.go
+++ b/go/libraries/doltcore/schema/typeinfo/decimal.go
@@ -209,6 +209,14 @@ func decimalTypeConverter(ctx context.Context, src *decimalType, destTi TypeInfo
 			}
 			return dest.ConvertValueToNomsValue(ctx, vrw, decimal.Decimal(val))
 		}, true, nil
+	case *blobStringType:
+		return func(ctx context.Context, vrw types.ValueReadWriter, v types.Value) (types.Value, error) {
+			s, err := src.ConvertNomsValueToValue(v)
+			if err != nil {
+				return nil, err
+			}
+			return dest.ConvertValueToNomsValue(ctx, vrw, s)
+		}, true, nil
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/enum.go b/go/libraries/doltcore/schema/typeinfo/enum.go
index 6282708c24..c317874a58 100644
--- a/go/libraries/doltcore/schema/typeinfo/enum.go
+++ b/go/libraries/doltcore/schema/typeinfo/enum.go
@@ -224,6 +224,8 @@ func enumTypeConverter(ctx context.Context, src *enumType, destTi TypeInfo) (tc
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/float.go b/go/libraries/doltcore/schema/typeinfo/float.go
index dff30a7cd4..760eae9966 100644
--- a/go/libraries/doltcore/schema/typeinfo/float.go
+++ b/go/libraries/doltcore/schema/typeinfo/float.go
@@ -233,6 +233,8 @@ func floatTypeConverter(ctx context.Context, src *floatType, destTi TypeInfo) (t
 			}
 			return dest.ConvertValueToNomsValue(ctx, vrw, uint64(intVal.(int64)))
 		}, true, nil
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/inlineblob.go b/go/libraries/doltcore/schema/typeinfo/inlineblob.go
index 727b359def..87ca3a1466 100644
--- a/go/libraries/doltcore/schema/typeinfo/inlineblob.go
+++ b/go/libraries/doltcore/schema/typeinfo/inlineblob.go
@@ -222,6 +222,8 @@ func inlineBlobTypeConverter(ctx context.Context, src *inlineBlobType, destTi Ty
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/int.go b/go/libraries/doltcore/schema/typeinfo/int.go
index d26adb6bb4..2783b57250 100644
--- a/go/libraries/doltcore/schema/typeinfo/int.go
+++ b/go/libraries/doltcore/schema/typeinfo/int.go
@@ -261,6 +261,8 @@ func intTypeConverter(ctx context.Context, src *intType, destTi TypeInfo) (tc Ty
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/json.go b/go/libraries/doltcore/schema/typeinfo/json.go
index 12307b3cb7..248c044321 100644
--- a/go/libraries/doltcore/schema/typeinfo/json.go
+++ b/go/libraries/doltcore/schema/typeinfo/json.go
@@ -163,6 +163,8 @@ func jsonTypeConverter(ctx context.Context, src *jsonType, destTi TypeInfo) (tc
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/set.go b/go/libraries/doltcore/schema/typeinfo/set.go
index fa5bccc159..c6248f5b98 100644
--- a/go/libraries/doltcore/schema/typeinfo/set.go
+++ b/go/libraries/doltcore/schema/typeinfo/set.go
@@ -217,6 +217,8 @@ func setTypeConverter(ctx context.Context, src *setType, destTi TypeInfo) (tc Ty
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/time.go b/go/libraries/doltcore/schema/typeinfo/time.go
index 8a723e672a..3758d49066 100644
--- a/go/libraries/doltcore/schema/typeinfo/time.go
+++ b/go/libraries/doltcore/schema/typeinfo/time.go
@@ -153,6 +153,8 @@ func timeTypeConverter(ctx context.Context, src *timeType, destTi TypeInfo) (tc
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/typeconverter.go b/go/libraries/doltcore/schema/typeinfo/typeconverter.go
index 648912267f..45969d80b5 100644
--- a/go/libraries/doltcore/schema/typeinfo/typeconverter.go
+++ b/go/libraries/doltcore/schema/typeinfo/typeconverter.go
@@ -51,6 +51,8 @@ func GetTypeConverter(ctx context.Context, srcTi TypeInfo, destTi TypeInfo) (tc
 	switch src := srcTi.(type) {
 	case *bitType:
 		return bitTypeConverter(ctx, src, destTi)
+	case *blobStringType:
+		return blobStringTypeConverter(ctx, src, destTi)
 	case *boolType:
 		return boolTypeConverter(ctx, src, destTi)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/typeinfo.go b/go/libraries/doltcore/schema/typeinfo/typeinfo.go
index b22a272416..cecee115d9 100644
--- a/go/libraries/doltcore/schema/typeinfo/typeinfo.go
+++ b/go/libraries/doltcore/schema/typeinfo/typeinfo.go
@@ -30,6 +30,7 @@ type Identifier string
 const (
 	UnknownTypeIdentifier    Identifier = "unknown"
 	BitTypeIdentifier        Identifier = "bit"
+	BlobStringTypeIdentifier Identifier = "blobstring"
 	BoolTypeIdentifier       Identifier = "bool"
 	DatetimeTypeIdentifier   Identifier = "datetime"
 	DecimalTypeIdentifier    Identifier = "decimal"
@@ -51,6 +52,7 @@ const (
 var Identifiers = map[Identifier]struct{}{
 	UnknownTypeIdentifier:    {},
 	BitTypeIdentifier:        {},
+	BlobStringTypeIdentifier: {},
 	BoolTypeIdentifier:       {},
 	DatetimeTypeIdentifier:   {},
 	DecimalTypeIdentifier:    {},
@@ -167,7 +169,7 @@ func FromSqlType(sqlType sql.Type) (TypeInfo, error) {
 		if !ok {
 			return nil, fmt.Errorf(`expected "StringType" from SQL basetype "Text"`)
 		}
-		return &varStringType{stringType}, nil
+		return &blobStringType{stringType}, nil
 	case sqltypes.Blob:
 		stringType, ok := sqlType.(sql.StringType)
 		if !ok {
@@ -232,6 +234,8 @@ func FromTypeParams(id Identifier, params map[string]string) (TypeInfo, error) {
 	switch id {
 	case BitTypeIdentifier:
 		return CreateBitTypeFromParams(params)
+	case BlobStringTypeIdentifier:
+		return CreateBlobStringTypeFromParams(params)
 	case BoolTypeIdentifier:
 		return BoolType, nil
 	case DatetimeTypeIdentifier:
diff --git a/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go b/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go
index 22c6797b38..b31b4a88b3 100644
--- a/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go
+++ b/go/libraries/doltcore/schema/typeinfo/typeinfo_test.go
@@ -71,8 +71,6 @@ 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 VarBinaryType
-	delete(seenTypeInfos, VarBinaryTypeIdentifier)
 	for _, tiArray := range tiArrays {
 		// no row should be empty
 		require.True(t, len(tiArray) > 0, `length of array "%v" should be greater than zero`, len(tiArray))
@@ -114,7 +112,7 @@ func testTypeInfoConvertRoundTrip(t *testing.T, tiArrays [][]TypeInfo, vaArrays
 				atLeastOneValid := false
 				t.Run(ti.String(), func(t *testing.T) {
 					for _, val := range vaArrays[rowIndex] {
-						t.Run(fmt.Sprintf(`types.%v(%v)`, val.Kind().String(), val.HumanReadableString()), func(t *testing.T) {
+						t.Run(fmt.Sprintf(`types.%v(%v)`, val.Kind().String(), humanReadableString(val)), func(t *testing.T) {
 							vInterface, err := ti.ConvertNomsValueToValue(val)
 							if ti.IsValid(val) {
 								atLeastOneValid = true
@@ -202,7 +200,7 @@ func testTypeInfoForeignKindHandling(t *testing.T, tiArrays [][]TypeInfo, vaArra
 				t.Run(ti.String(), func(t *testing.T) {
 					for _, vaArray := range vaArrays {
 						for _, val := range vaArray {
-							t.Run(fmt.Sprintf(`types.%v(%v)`, val.Kind().String(), val.HumanReadableString()), func(t *testing.T) {
+							t.Run(fmt.Sprintf(`types.%v(%v)`, val.Kind().String(), humanReadableString(val)), func(t *testing.T) {
 								if ti.NomsKind() != val.Kind() {
 									_, err := ti.ConvertNomsValueToValue(val)
 									assert.Error(t, err)
@@ -226,7 +224,7 @@ func testTypeInfoFormatParseRoundTrip(t *testing.T, tiArrays [][]TypeInfo, vaArr
 				atLeastOneValid := false
 				t.Run(ti.String(), func(t *testing.T) {
 					for _, val := range vaArrays[rowIndex] {
-						t.Run(fmt.Sprintf(`types.%v(%v)`, val.Kind().String(), val.HumanReadableString()), func(t *testing.T) {
+						t.Run(fmt.Sprintf(`types.%v(%v)`, val.Kind().String(), humanReadableString(val)), func(t *testing.T) {
 							str, err := ti.FormatValue(val)
 							if ti.IsValid(val) {
 								atLeastOneValid = true
@@ -345,6 +343,8 @@ func testTypeInfoToSqlType(t *testing.T, tiArrays [][]TypeInfo) {
 func generateTypeInfoArrays(t *testing.T) ([][]TypeInfo, [][]types.Value) {
 	return [][]TypeInfo{
 			generateBitTypes(t, 16),
+			{&blobStringType{sql.TinyText}, &blobStringType{sql.Text},
+				&blobStringType{sql.MediumText}, &blobStringType{sql.LongText}},
 			{BoolType},
 			{DateType, DatetimeType, TimestampType},
 			generateDecimalTypes(t, 16),
@@ -357,9 +357,8 @@ func generateTypeInfoArrays(t *testing.T) ([][]TypeInfo, [][]types.Value) {
 			{TimeType},
 			{Uint8Type, Uint16Type, Uint24Type, Uint32Type, Uint64Type},
 			{UuidType},
-			//append(generateVarBinaryTypes(t, 12),
-			//	&varBinaryType{sql.TinyBlob}, &varBinaryType{sql.Blob},
-			//	&varBinaryType{sql.MediumBlob}, &varBinaryType{sql.LongBlob}),
+			{&varBinaryType{sql.TinyBlob}, &varBinaryType{sql.Blob},
+				&varBinaryType{sql.MediumBlob}, &varBinaryType{sql.LongBlob}},
 			append(generateVarStringTypes(t, 12),
 				&varStringType{sql.CreateTinyText(sql.Collation_Default)}, &varStringType{sql.CreateText(sql.Collation_Default)},
 				&varStringType{sql.CreateMediumText(sql.Collation_Default)}, &varStringType{sql.CreateLongText(sql.Collation_Default)}),
@@ -367,6 +366,8 @@ func generateTypeInfoArrays(t *testing.T) ([][]TypeInfo, [][]types.Value) {
 		},
 		[][]types.Value{
 			{types.Uint(1), types.Uint(207), types.Uint(79147), types.Uint(34845728), types.Uint(9274618927)}, //Bit
+			{mustBlobString(t, ""), mustBlobString(t, "a"), mustBlobString(t, "abc"), //BlobString
+				mustBlobString(t, "abcdefghijklmnopqrstuvwxyz"), mustBlobString(t, "هذا هو بعض نماذج النص التي أستخدمها لاختبار عناصر")},
 			{types.Bool(false), types.Bool(true)}, //Bool
 			{types.Timestamp(time.Date(1000, 1, 1, 0, 0, 0, 0, time.UTC)), //Datetime
 				types.Timestamp(time.Date(1970, 1, 1, 0, 0, 1, 0, time.UTC)),
@@ -388,10 +389,17 @@ func generateTypeInfoArrays(t *testing.T) ([][]TypeInfo, [][]types.Value) {
 			{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})},
+			{mustBlobBytes(t, []byte{1}), mustBlobBytes(t, []byte{42, 52}), mustBlobBytes(t, []byte{84, 32, 13, 63, 12, 86}), //VarBinary
+				mustBlobBytes(t, []byte{1, 32, 235, 64, 32, 23, 45, 76}), mustBlobBytes(t, []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
 				types.String("abcdefghijklmnopqrstuvwxyz"), types.String("هذا هو بعض نماذج النص التي أستخدمها لاختبار عناصر")},
 			{types.Int(1901), types.Int(1950), types.Int(2000), types.Int(2080), types.Int(2155)}, //Year
 		}
 }
+
+func humanReadableString(val types.Value) string {
+	defer func() {
+		_ = recover() // HumanReadableString panics for some types so we ignore the panic
+	}()
+	return val.HumanReadableString()
+}
diff --git a/go/libraries/doltcore/schema/typeinfo/uint.go b/go/libraries/doltcore/schema/typeinfo/uint.go
index e3ab83c804..d2497121bb 100644
--- a/go/libraries/doltcore/schema/typeinfo/uint.go
+++ b/go/libraries/doltcore/schema/typeinfo/uint.go
@@ -261,6 +261,8 @@ func uintTypeConverter(ctx context.Context, src *uintType, destTi TypeInfo) (tc
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/uuid.go b/go/libraries/doltcore/schema/typeinfo/uuid.go
index 74433692d6..64e2e19290 100644
--- a/go/libraries/doltcore/schema/typeinfo/uuid.go
+++ b/go/libraries/doltcore/schema/typeinfo/uuid.go
@@ -155,6 +155,8 @@ func uuidTypeConverter(ctx context.Context, src *uuidType, destTi TypeInfo) (tc
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/varbinary.go b/go/libraries/doltcore/schema/typeinfo/varbinary.go
index d82ff968f0..e057331bee 100644
--- a/go/libraries/doltcore/schema/typeinfo/varbinary.go
+++ b/go/libraries/doltcore/schema/typeinfo/varbinary.go
@@ -15,12 +15,12 @@
 package typeinfo
 
 import (
-	"bytes"
 	"context"
 	"encoding/binary"
 	"fmt"
 	"io"
 	"strconv"
+	"strings"
 	"unsafe"
 
 	"github.com/dolthub/go-mysql-server/sql"
@@ -101,7 +101,7 @@ func (ti *varBinaryType) ConvertValueToNomsValue(ctx context.Context, vrw types.
 	}
 	val, ok := strVal.(string)
 	if ok {
-		return toBlob(ctx, vrw, val)
+		return types.NewBlob(ctx, vrw, strings.NewReader(val))
 	}
 	return nil, fmt.Errorf(`"%v" cannot convert value "%v" of type "%T" as it is invalid`, ti.String(), v, v)
 }
@@ -147,11 +147,7 @@ func (ti *varBinaryType) GetTypeParams() map[string]string {
 // IsValid implements TypeInfo interface.
 func (ti *varBinaryType) IsValid(v types.Value) bool {
 	if val, ok := v.(types.Blob); ok {
-		strLen, err := fromBlobLength(val)
-		if err != nil {
-			return false
-		}
-		if int64(strLen) <= ti.sqlBinaryType.MaxByteLength() {
+		if int64(val.Len()) <= ti.sqlBinaryType.MaxByteLength() {
 			return true
 		}
 	}
@@ -176,7 +172,7 @@ func (ti *varBinaryType) ParseValue(ctx context.Context, vrw types.ValueReadWrit
 		return nil, err
 	}
 	if val, ok := strVal.(string); ok {
-		return toBlob(ctx, vrw, val)
+		return types.NewBlob(ctx, vrw, strings.NewReader(val))
 	}
 	return nil, fmt.Errorf(`"%v" cannot convert the string "%v" to a value`, ti.String(), str)
 }
@@ -198,15 +194,12 @@ func (ti *varBinaryType) ToSqlType() sql.Type {
 
 // fromBlob returns a string from a types.Blob.
 func fromBlob(b types.Blob) (string, error) {
-	strLength, err := fromBlobLength(b)
-	if err != nil {
-		return "", err
-	}
+	strLength := b.Len()
 	if strLength == 0 {
 		return "", nil
 	}
 	str := make([]byte, strLength)
-	n, err := b.ReadAt(context.Background(), str, 8)
+	n, err := b.ReadAt(context.Background(), str, 0)
 	if err != nil && err != io.EOF {
 		return "", err
 	}
@@ -223,28 +216,27 @@ func fromBlob(b types.Blob) (string, error) {
 	return *(*string)(unsafe.Pointer(&str)), nil
 }
 
-// fromBlobLength returns a string's length from a types.Blob.
-func fromBlobLength(b types.Blob) (uint64, error) {
-	countBytes := make([]byte, 8)
-	n, err := b.ReadAt(context.Background(), countBytes, 0)
-	if err == io.EOF {
-		return 0, nil
+// hasPrefix finds out if a Blob has a prefixed integer. Initially blobs for varBinary prepended an integer indicating
+// the length, which was unnecessary (as the underlying sequence tracks the total size). It's been removed, but this
+// may be used to see if a Blob is one of those older Blobs. A false positive is possible, but EXTREMELY unlikely.
+func hasPrefix(b types.Blob, ctx context.Context) (bool, error) {
+	blobLength := b.Len()
+	if blobLength < 8 {
+		return false, nil
 	}
+	countBytes := make([]byte, 8)
+	n, err := b.ReadAt(ctx, countBytes, 0)
 	if err != nil {
-		return 0, err
+		return false, err
 	}
 	if n != 8 {
-		return 0, fmt.Errorf("wanted 8 bytes from blob for count, got %d", n)
+		return false, fmt.Errorf("wanted 8 bytes from blob for count, got %d", n)
 	}
-	return binary.LittleEndian.Uint64(countBytes), nil
-}
-
-// toBlob returns a types.Blob from a string.
-func toBlob(ctx context.Context, vrw types.ValueReadWriter, s string) (types.Blob, error) {
-	data := make([]byte, 8+len(s))
-	binary.LittleEndian.PutUint64(data[:8], uint64(len(s)))
-	copy(data[8:], s)
-	return types.NewBlob(ctx, vrw, bytes.NewReader(data))
+	prefixedLength := binary.LittleEndian.Uint64(countBytes)
+	if prefixedLength == blobLength-8 {
+		return true, nil
+	}
+	return false, nil
 }
 
 // varBinaryTypeConverter is an internal function for GetTypeConverter that handles the specific type as the source TypeInfo.
@@ -252,6 +244,8 @@ func varBinaryTypeConverter(ctx context.Context, src *varBinaryType, destTi Type
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapIsValid(dest.IsValid, src, dest)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/varstring.go b/go/libraries/doltcore/schema/typeinfo/varstring.go
index 984bbeda23..9bd2d8b3f4 100644
--- a/go/libraries/doltcore/schema/typeinfo/varstring.go
+++ b/go/libraries/doltcore/schema/typeinfo/varstring.go
@@ -41,7 +41,11 @@ type varStringType struct {
 }
 
 var _ TypeInfo = (*varStringType)(nil)
-var StringDefaultType = &varStringType{sql.CreateLongText(sql.Collation_Default)}
+
+var (
+	LegacyStringDefaultType = &varStringType{sql.CreateLongText(sql.Collation_Default)}
+	StringDefaultType       = &varStringType{sql.MustCreateStringWithDefaults(sqltypes.VarChar, 16383)}
+)
 
 func CreateVarStringTypeFromParams(params map[string]string) (TypeInfo, error) {
 	var length int64
@@ -254,6 +258,8 @@ func varStringTypeConverter(ctx context.Context, src *varStringType, destTi Type
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/schema/typeinfo/year.go b/go/libraries/doltcore/schema/typeinfo/year.go
index fe97cfcbd7..c5df18262d 100644
--- a/go/libraries/doltcore/schema/typeinfo/year.go
+++ b/go/libraries/doltcore/schema/typeinfo/year.go
@@ -169,6 +169,8 @@ func yearTypeConverter(ctx context.Context, src *yearType, destTi TypeInfo) (tc
 	switch dest := destTi.(type) {
 	case *bitType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
+	case *blobStringType:
+		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *boolType:
 		return wrapConvertValueToNomsValue(dest.ConvertValueToNomsValue)
 	case *datetimeType:
diff --git a/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go b/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
index 36c27ec7d4..bcff08d0a6 100644
--- a/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
+++ b/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
@@ -181,6 +181,9 @@ func TestTruncate(t *testing.T) {
 func TestScripts(t *testing.T) {
 	skipped := []string{
 		"create index r_c0 on r (c0);",
+		// These rely on keyless tables which orders its rows by hash rather than contents, meaning changing types causes different ordering
+		"SELECT group_concat(attribute) FROM t where o_id=2",
+		"SELECT group_concat(o_id) FROM t WHERE attribute='color'",
 	}
 	enginetest.TestScripts(t, newDoltHarness(t).WithSkippedQueries(skipped))
 }
diff --git a/go/libraries/doltcore/sqle/schema_table.go b/go/libraries/doltcore/sqle/schema_table.go
index e13a8bd245..b7104fdbe8 100644
--- a/go/libraries/doltcore/sqle/schema_table.go
+++ b/go/libraries/doltcore/sqle/schema_table.go
@@ -23,6 +23,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/row"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema/typeinfo"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/sqlutil"
 	"github.com/dolthub/dolt/go/store/types"
 )
@@ -38,12 +39,23 @@ func SchemasTableSqlSchema() sql.Schema {
 
 // The fixed dolt schema for the `dolt_schemas` table.
 func SchemasTableSchema() schema.Schema {
-	colColl := schema.NewColCollection(
-		schema.NewColumn(doltdb.SchemasTablesTypeCol, schema.DoltSchemasTypeTag, types.StringKind, false),
-		schema.NewColumn(doltdb.SchemasTablesNameCol, schema.DoltSchemasNameTag, types.StringKind, false),
-		schema.NewColumn(doltdb.SchemasTablesFragmentCol, schema.DoltSchemasFragmentTag, types.StringKind, false),
-		schema.NewColumn(doltdb.SchemasTablesIdCol, schema.DoltSchemasIdTag, types.IntKind, true, schema.NotNullConstraint{}),
-	)
+	typeCol, err := schema.NewColumnWithTypeInfo(doltdb.SchemasTablesTypeCol, schema.DoltSchemasTypeTag, typeinfo.LegacyStringDefaultType, false, "", false, "")
+	if err != nil {
+		panic(err)
+	}
+	nameCol, err := schema.NewColumnWithTypeInfo(doltdb.SchemasTablesNameCol, schema.DoltSchemasNameTag, typeinfo.LegacyStringDefaultType, false, "", false, "")
+	if err != nil {
+		panic(err)
+	}
+	fragmentCol, err := schema.NewColumnWithTypeInfo(doltdb.SchemasTablesFragmentCol, schema.DoltSchemasFragmentTag, typeinfo.LegacyStringDefaultType, false, "", false, "")
+	if err != nil {
+		panic(err)
+	}
+	idCol, err := schema.NewColumnWithTypeInfo(doltdb.SchemasTablesIdCol, schema.DoltSchemasIdTag, typeinfo.Int64Type, true, "", false, "", schema.NotNullConstraint{})
+	if err != nil {
+		panic(err)
+	}
+	colColl := schema.NewColCollection(typeCol, nameCol, fragmentCol, idCol)
 	return schema.MustSchemaFromCols(colColl)
 }
 
diff --git a/go/libraries/doltcore/sqle/sqlddl_test.go b/go/libraries/doltcore/sqle/sqlddl_test.go
index 554942d93f..746164af07 100644
--- a/go/libraries/doltcore/sqle/sqlddl_test.go
+++ b/go/libraries/doltcore/sqle/sqlddl_test.go
@@ -521,7 +521,7 @@ func TestModifyAndChangeColumn(t *testing.T) {
 	}{
 		{
 			name:  "alter modify column reorder middle",
-			query: "alter table people modify column first_name longtext not null after last_name",
+			query: "alter table people modify column first_name varchar(16383) not null after last_name",
 			expectedSchema: dtestutils.CreateSchema(
 				schema.NewColumn("id", IdTag, types.IntKind, true, schema.NotNullConstraint{}),
 				schema.NewColumn("last_name", LastNameTag, types.StringKind, false, schema.NotNullConstraint{}),
@@ -536,7 +536,7 @@ func TestModifyAndChangeColumn(t *testing.T) {
 		},
 		{
 			name:  "alter modify column reorder first",
-			query: "alter table people modify column first_name longtext not null first",
+			query: "alter table people modify column first_name varchar(16383) not null first",
 			expectedSchema: dtestutils.CreateSchema(
 				schema.NewColumn("first_name", FirstNameTag, types.StringKind, false, schema.NotNullConstraint{}),
 				schema.NewColumn("id", IdTag, types.IntKind, true, schema.NotNullConstraint{}),
@@ -551,7 +551,7 @@ func TestModifyAndChangeColumn(t *testing.T) {
 		},
 		{
 			name:  "alter modify column drop null constraint",
-			query: "alter table people modify column first_name longtext null",
+			query: "alter table people modify column first_name varchar(16383) null",
 			expectedSchema: dtestutils.CreateSchema(
 				schema.NewColumn("id", IdTag, types.IntKind, true, schema.NotNullConstraint{}),
 				schema.NewColumn("first_name", FirstNameTag, types.StringKind, false),
@@ -566,7 +566,7 @@ func TestModifyAndChangeColumn(t *testing.T) {
 		},
 		{
 			name:  "alter change column rename and reorder",
-			query: "alter table people change first_name christian_name longtext not null after last_name",
+			query: "alter table people change first_name christian_name varchar(16383) not null after last_name",
 			expectedSchema: dtestutils.CreateSchema(
 				schema.NewColumn("id", IdTag, types.IntKind, true, schema.NotNullConstraint{}),
 				schema.NewColumn("last_name", LastNameTag, types.StringKind, false, schema.NotNullConstraint{}),
@@ -581,7 +581,7 @@ func TestModifyAndChangeColumn(t *testing.T) {
 		},
 		{
 			name:  "alter change column rename and reorder first",
-			query: "alter table people change column first_name christian_name longtext not null first",
+			query: "alter table people change column first_name christian_name varchar(16383) not null first",
 			expectedSchema: dtestutils.CreateSchema(
 				schema.NewColumn("christian_name", FirstNameTag, types.StringKind, false, schema.NotNullConstraint{}),
 				schema.NewColumn("id", IdTag, types.IntKind, true, schema.NotNullConstraint{}),
@@ -596,7 +596,7 @@ func TestModifyAndChangeColumn(t *testing.T) {
 		},
 		{
 			name:  "alter change column drop null constraint",
-			query: "alter table people change column first_name first_name longtext null",
+			query: "alter table people change column first_name first_name varchar(16383) null",
 			expectedSchema: dtestutils.CreateSchema(
 				schema.NewColumn("id", IdTag, types.IntKind, true, schema.NotNullConstraint{}),
 				schema.NewColumn("first_name", FirstNameTag, types.StringKind, false),
@@ -1171,7 +1171,7 @@ func TestAlterSystemTables(t *testing.T) {
 			dtables.DoltQueryCatalogSchema,
 			NewRow(types.String("abc123"), types.Uint(1), types.String("example"), types.String("select 2+2 from dual"), types.String("description")))
 		dtestutils.CreateTestTable(t, dEnv, doltdb.SchemasTableName,
-			schemasTableDoltSchema(),
+			SchemasTableSchema(),
 			NewRowWithPks([]types.Value{types.String("view"), types.String("name")}, types.String("select 2+2 from dual")))
 	}
 
@@ -1573,13 +1573,6 @@ INSERT INTO fail_unique VALUES (1, 1, 1), (2, 2, 2), (3, 2, 3);
 	}
 }
 
-func schemasTableDoltSchema() schema.Schema {
-	// this is a dummy test environment and will not be used,
-	// dolt_schema table tags will be parsed from the comments in SchemaTableSchema()
-	testEnv := dtestutils.CreateTestEnv()
-	return mustGetDoltSchema(SchemasTableSqlSchema(), doltdb.SchemasTableName, testEnv)
-}
-
 func assertFails(t *testing.T, dEnv *env.DoltEnv, query, expectedErr string) {
 	ctx := context.Background()
 	root, _ := dEnv.WorkingRoot(ctx)
diff --git a/go/libraries/doltcore/sqle/sqldelete_test.go b/go/libraries/doltcore/sqle/sqldelete_test.go
index d283dcf914..342bf8bbc4 100644
--- a/go/libraries/doltcore/sqle/sqldelete_test.go
+++ b/go/libraries/doltcore/sqle/sqldelete_test.go
@@ -212,12 +212,12 @@ var systemTableDeleteTests = []DeleteTest{
 	{
 		Name: "delete dolt_schemas",
 		AdditionalSetup: CreateTableFn(doltdb.SchemasTableName,
-			schemasTableDoltSchema(),
+			SchemasTableSchema(),
 			NewRowWithPks([]types.Value{types.String("view"), types.String("name")}, types.String("select 2+2 from dual"))),
 		DeleteQuery:    "delete from dolt_schemas",
 		SelectQuery:    "select * from dolt_schemas",
 		ExpectedRows:   ToSqlRows(dtables.DoltQueryCatalogSchema),
-		ExpectedSchema: schemasTableDoltSchema(),
+		ExpectedSchema: SchemasTableSchema(),
 	},
 }
 
diff --git a/go/libraries/doltcore/sqle/sqlfmt/row_fmt.go b/go/libraries/doltcore/sqle/sqlfmt/row_fmt.go
index fd82fd48a9..2b20e4c80e 100644
--- a/go/libraries/doltcore/sqle/sqlfmt/row_fmt.go
+++ b/go/libraries/doltcore/sqle/sqlfmt/row_fmt.go
@@ -204,6 +204,8 @@ func valueAsSqlString(ti typeinfo.TypeInfo, value types.Value) (string, error) {
 		return singleQuote + *str + singleQuote, nil
 	case typeinfo.DatetimeTypeIdentifier:
 		return singleQuote + *str + singleQuote, nil
+	case typeinfo.BlobStringTypeIdentifier, typeinfo.VarBinaryTypeIdentifier, typeinfo.InlineBlobTypeIdentifier:
+		return quoteAndEscapeString(*str), nil
 	case typeinfo.VarStringTypeIdentifier:
 		s, ok := value.(types.String)
 		if !ok {
diff --git a/go/libraries/doltcore/sqle/sqlfmt/schema_fmt_test.go b/go/libraries/doltcore/sqle/sqlfmt/schema_fmt_test.go
index 44f969ec14..b551602afb 100644
--- a/go/libraries/doltcore/sqle/sqlfmt/schema_fmt_test.go
+++ b/go/libraries/doltcore/sqle/sqlfmt/schema_fmt_test.go
@@ -36,7 +36,7 @@ func TestFmtCol(t *testing.T) {
 			0,
 			0,
 			0,
-			"`first` LONGTEXT",
+			"`first` VARCHAR(16383)",
 		},
 		{
 			schema.NewColumn("last", 123, types.IntKind, true),
diff --git a/go/libraries/doltcore/sqle/sqlinsert_test.go b/go/libraries/doltcore/sqle/sqlinsert_test.go
index f0ea287a6f..60aa9e2ad7 100644
--- a/go/libraries/doltcore/sqle/sqlinsert_test.go
+++ b/go/libraries/doltcore/sqle/sqlinsert_test.go
@@ -426,13 +426,13 @@ var systemTableInsertTests = []InsertTest{
 	},
 	{
 		Name:            "insert into dolt_schemas",
-		AdditionalSetup: CreateTableFn(doltdb.SchemasTableName, schemasTableDoltSchema()),
+		AdditionalSetup: CreateTableFn(doltdb.SchemasTableName, SchemasTableSchema()),
 		InsertQuery:     "insert into dolt_schemas (id, type, name, fragment) values (1, 'view', 'name', 'select 2+2 from dual')",
 		SelectQuery:     "select * from dolt_schemas ORDER BY id",
-		ExpectedRows: ToSqlRows(CompressSchema(schemasTableDoltSchema()),
+		ExpectedRows: ToSqlRows(CompressSchema(SchemasTableSchema()),
 			NewRow(types.String("view"), types.String("name"), types.String("select 2+2 from dual"), types.Int(1)),
 		),
-		ExpectedSchema: CompressSchema(schemasTableDoltSchema()),
+		ExpectedSchema: CompressSchema(SchemasTableSchema()),
 	},
 }
 
diff --git a/go/libraries/doltcore/sqle/sqlreplace_test.go b/go/libraries/doltcore/sqle/sqlreplace_test.go
index c5542de492..15dfe51177 100644
--- a/go/libraries/doltcore/sqle/sqlreplace_test.go
+++ b/go/libraries/doltcore/sqle/sqlreplace_test.go
@@ -276,14 +276,14 @@ var systemTableReplaceTests = []ReplaceTest{
 	{
 		Name: "replace into dolt_schemas",
 		AdditionalSetup: CreateTableFn(doltdb.SchemasTableName,
-			schemasTableDoltSchema(),
+			SchemasTableSchema(),
 			NewRowWithPks([]types.Value{types.String("view"), types.String("name")}, types.String("select 2+2 from dual"))),
 		ReplaceQuery: "replace into dolt_schemas (type, name, fragment) values ('view', 'name', 'select 1+1 from dual')",
 		SelectQuery:  "select * from dolt_schemas",
-		ExpectedRows: ToSqlRows(schemasTableDoltSchema(),
+		ExpectedRows: ToSqlRows(SchemasTableSchema(),
 			NewRow(types.String("view"), types.String("name"), types.String("select 1+1 from dual")),
 		),
-		ExpectedSchema: CompressSchema(schemasTableDoltSchema()),
+		ExpectedSchema: CompressSchema(SchemasTableSchema()),
 	},
 }
 
diff --git a/go/libraries/doltcore/sqle/sqlselect_test.go b/go/libraries/doltcore/sqle/sqlselect_test.go
index 784861a1b4..3cdf0dd356 100644
--- a/go/libraries/doltcore/sqle/sqlselect_test.go
+++ b/go/libraries/doltcore/sqle/sqlselect_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/envtestutils"
 	"github.com/dolthub/dolt/go/libraries/doltcore/row"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema/typeinfo"
 	. "github.com/dolthub/dolt/go/libraries/doltcore/sql/sqltestutil"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dtables"
 	"github.com/dolthub/dolt/go/store/types"
@@ -446,7 +447,7 @@ var BasicSelectTests = []SelectTest{
 			{"Homer", true},
 		},
 		ExpectedSqlSchema: sql.Schema{
-			&sql.Column{Name: "first_name", Type: sql.LongText},
+			&sql.Column{Name: "first_name", Type: typeinfo.StringDefaultType.ToSqlType()},
 			&sql.Column{Name: "not_marge", Type: sql.Int8},
 		},
 	},
@@ -711,7 +712,7 @@ var BasicSelectTests = []SelectTest{
 		Query: "select * from dolt_log",
 		ExpectedRows: []sql.Row{
 			{
-				"m8lrhp8bmfesmknc6d5iatmjbcjf17al",
+				"so275enkvulb96mkckbun1kjo9seg7c9",
 				"billy bob",
 				"bigbillieb@fake.horse",
 				time.Date(1970, 1, 1, 0, 0, 0, 0, &time.Location{}),
@@ -741,7 +742,7 @@ var BasicSelectTests = []SelectTest{
 		ExpectedRows: []sql.Row{
 			{
 				"master",
-				"m8lrhp8bmfesmknc6d5iatmjbcjf17al",
+				"so275enkvulb96mkckbun1kjo9seg7c9",
 				"billy bob", "bigbillieb@fake.horse",
 				time.Date(1970, 1, 1, 0, 0, 0, 0, &time.Location{}),
 				"Initialize data repository",
@@ -760,13 +761,13 @@ var BasicSelectTests = []SelectTest{
 
 var sqlDiffSchema = sql.Schema{
 	&sql.Column{Name: "to_id", Type: sql.Int64},
-	&sql.Column{Name: "to_first_name", Type: sql.LongText},
-	&sql.Column{Name: "to_last_name", Type: sql.LongText},
-	&sql.Column{Name: "to_addr", Type: sql.LongText},
+	&sql.Column{Name: "to_first_name", Type: typeinfo.StringDefaultType.ToSqlType()},
+	&sql.Column{Name: "to_last_name", Type: typeinfo.StringDefaultType.ToSqlType()},
+	&sql.Column{Name: "to_addr", Type: typeinfo.StringDefaultType.ToSqlType()},
 	&sql.Column{Name: "from_id", Type: sql.Int64},
-	&sql.Column{Name: "from_first_name", Type: sql.LongText},
-	&sql.Column{Name: "from_last_name", Type: sql.LongText},
-	&sql.Column{Name: "from_addr", Type: sql.LongText},
+	&sql.Column{Name: "from_first_name", Type: typeinfo.StringDefaultType.ToSqlType()},
+	&sql.Column{Name: "from_last_name", Type: typeinfo.StringDefaultType.ToSqlType()},
+	&sql.Column{Name: "from_addr", Type: typeinfo.StringDefaultType.ToSqlType()},
 	&sql.Column{Name: "diff_type", Type: sql.Text},
 }
 
@@ -1494,18 +1495,18 @@ var systemTableSelectTests = []SelectTest{
 	{
 		Name: "select from dolt_schemas",
 		AdditionalSetup: CreateTableFn(doltdb.SchemasTableName,
-			schemasTableDoltSchema(),
-			NewRowWithSchema(schemasTableDoltSchema(),
+			SchemasTableSchema(),
+			NewRowWithSchema(SchemasTableSchema(),
 				types.String("view"),
 				types.String("name"),
 				types.String("select 2+2 from dual"),
 				types.Int(1),
 			)),
 		Query: "select * from dolt_schemas",
-		ExpectedRows: ToSqlRows(CompressSchema(schemasTableDoltSchema()),
+		ExpectedRows: ToSqlRows(CompressSchema(SchemasTableSchema()),
 			NewRow(types.String("view"), types.String("name"), types.String("select 2+2 from dual"), types.Int(1)),
 		),
-		ExpectedSchema: CompressSchema(schemasTableDoltSchema()),
+		ExpectedSchema: CompressSchema(SchemasTableSchema()),
 	},
 }
 
diff --git a/go/libraries/doltcore/sqle/sqlupdate_test.go b/go/libraries/doltcore/sqle/sqlupdate_test.go
index 852edf22cc..42614a1337 100644
--- a/go/libraries/doltcore/sqle/sqlupdate_test.go
+++ b/go/libraries/doltcore/sqle/sqlupdate_test.go
@@ -403,8 +403,8 @@ var systemTableUpdateTests = []UpdateTest{
 	{
 		Name: "update dolt_schemas",
 		AdditionalSetup: CreateTableFn(doltdb.SchemasTableName,
-			schemasTableDoltSchema(),
-			NewRowWithSchema(schemasTableDoltSchema(),
+			SchemasTableSchema(),
+			NewRowWithSchema(SchemasTableSchema(),
 				types.String("view"),
 				types.String("name"),
 				types.String("select 2+2 from dual"),
@@ -412,10 +412,10 @@ var systemTableUpdateTests = []UpdateTest{
 			)),
 		UpdateQuery: "update dolt_schemas set type = 'not a view'",
 		SelectQuery: "select * from dolt_schemas",
-		ExpectedRows: ToSqlRows(CompressSchema(schemasTableDoltSchema()),
+		ExpectedRows: ToSqlRows(CompressSchema(SchemasTableSchema()),
 			NewRow(types.String("not a view"), types.String("name"), types.String("select 2+2 from dual"), types.Int(1)),
 		),
-		ExpectedSchema: CompressSchema(schemasTableDoltSchema()),
+		ExpectedSchema: CompressSchema(SchemasTableSchema()),
 	},
 }
 
diff --git a/go/libraries/doltcore/table/untyped/sqlexport/sqlwriter_test.go b/go/libraries/doltcore/table/untyped/sqlexport/sqlwriter_test.go
index 22d44c8eb5..d7fe800469 100644
--- a/go/libraries/doltcore/table/untyped/sqlexport/sqlwriter_test.go
+++ b/go/libraries/doltcore/table/untyped/sqlexport/sqlwriter_test.go
@@ -44,7 +44,7 @@ func TestEndToEnd(t *testing.T) {
 	id := uuid.MustParse("00000000-0000-0000-0000-000000000000")
 	tableName := "people"
 
-	dropCreateStatement := sqlfmt.DropTableIfExistsStmt(tableName) + "\nCREATE TABLE `people` (\n  `id` char(36) character set ascii collate ascii_bin NOT NULL,\n  `name` longtext NOT NULL,\n  `age` bigint unsigned NOT NULL,\n  `is_married` bit(1) NOT NULL,\n  `title` longtext,\n  PRIMARY KEY (`id`),\n  KEY `idx_name` (`name`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;"
+	dropCreateStatement := sqlfmt.DropTableIfExistsStmt(tableName) + "\nCREATE TABLE `people` (\n  `id` char(36) character set ascii collate ascii_bin NOT NULL,\n  `name` varchar(16383) NOT NULL,\n  `age` bigint unsigned NOT NULL,\n  `is_married` bit(1) NOT NULL,\n  `title` varchar(16383),\n  PRIMARY KEY (`id`),\n  KEY `idx_name` (`name`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;"
 
 	type test struct {
 		name           string
diff --git a/go/store/types/blob.go b/go/store/types/blob.go
index 4451a10f00..9bff41a8d2 100644
--- a/go/store/types/blob.go
+++ b/go/store/types/blob.go
@@ -22,14 +22,16 @@
 package types
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"
 	"runtime"
 	"sync"
 
-	"github.com/dolthub/dolt/go/store/atomicerr"
+	"github.com/dolthub/dolt/go/libraries/utils/mathutil"
 
+	"github.com/dolthub/dolt/go/store/atomicerr"
 	"github.com/dolthub/dolt/go/store/d"
 )
 
@@ -52,6 +54,49 @@ func NewEmptyBlob(vrw ValueReadWriter) (Blob, error) {
 	return Blob{seq}, nil
 }
 
+// Less implements the LesserValuable interface.
+func (b Blob) Less(nbf *NomsBinFormat, other LesserValuable) (bool, error) {
+	if b2, ok := other.(Blob); ok {
+		// Blobs can have an arbitrary length, so we compare in chunks rather than loading it entirely
+		ctx := context.Background()
+		b1Length := b.Len()
+		b2Length := b2.Len()
+		b1Reader := b.Reader(ctx)
+		b2Reader := b2.Reader(ctx)
+		minBlobLength := b1Length
+		if b1Length > b2Length {
+			minBlobLength = b2Length
+		}
+
+		const maxSliceSize = 1024 * 64 // arbitrary size
+		var b1Array [maxSliceSize]byte
+		var b2Array [maxSliceSize]byte
+		length := uint64(0)
+		for i := uint64(0); i < minBlobLength; i += length {
+			length = mathutil.MinUint64(maxSliceSize, minBlobLength-i)
+			b1Data := b1Array[:length]
+			b2Data := b2Array[:length]
+			n1, err := b1Reader.Read(b1Data)
+			if err != nil && err != io.EOF {
+				return false, err
+			}
+			n2, err := b2Reader.Read(b2Data)
+			if err != nil && err != io.EOF {
+				return false, err
+			}
+			if n1 != n2 || uint64(n1) != length {
+				return false, errors.New("incorrect length read from blob")
+			}
+			cmp := bytes.Compare(b1Data, b2Data)
+			if cmp != 0 {
+				return cmp < 0, nil
+			}
+		}
+		return b1Length < b2Length, nil
+	}
+	return BlobKind < other.Kind(), nil
+}
+
 // ReadAt implements the ReaderAt interface. Eagerly loads requested byte-range from the blob p-tree.
 func (b Blob) ReadAt(ctx context.Context, p []byte, off int64) (n int, err error) {
 	// TODO: Support negative off?
@@ -211,6 +256,10 @@ func (b Blob) Value(ctx context.Context) (Value, error) {
 	return b, nil
 }
 
+func (b Blob) Len() uint64 {
+	return b.sequence.Len()
+}
+
 func (b Blob) isPrimitive() bool {
 	return true
 }
@@ -485,9 +534,9 @@ func (b Blob) skip(nbf *NomsBinFormat, bnr *binaryNomsReader) {
 }
 
 func (b Blob) String() string {
-	panic("unreachable")
+	return "BLOB"
 }
 
 func (b Blob) HumanReadableString() string {
-	panic("unreachable")
+	return "BLOB"
 }
diff --git a/go/store/types/map_test.go b/go/store/types/map_test.go
index 93639ea209..2e75e73b0f 100644
--- a/go/store/types/map_test.go
+++ b/go/store/types/map_test.go
@@ -1354,20 +1354,12 @@ func TestMapEquals(t *testing.T) {
 
 func TestMapNotStringKeys(t *testing.T) {
 	assert := assert.New(t)
-
 	vrw := newTestValueStore()
-
-	b1, err := NewBlob(context.Background(), vrw, bytes.NewBufferString("blob1"))
-	require.NoError(t, err)
-	b2, err := NewBlob(context.Background(), vrw, bytes.NewBufferString("blob2"))
-	require.NoError(t, err)
 	l := []Value{
 		Bool(true), String("true"),
 		Bool(false), String("false"),
 		Float(1), String("Float: 1"),
 		Float(0), String("Float: 0"),
-		b1, String("blob1"),
-		b2, String("blob2"),
 		mustList(NewList(context.Background(), vrw)), String("empty list"),
 		mustList(NewList(context.Background(), vrw, mustList(NewList(context.Background(), vrw)))), String("list of list"),
 		mustMap(NewMap(context.Background(), vrw)), String("empty map"),
@@ -1377,7 +1369,7 @@ func TestMapNotStringKeys(t *testing.T) {
 	}
 	m1, err := NewMap(context.Background(), vrw, l...)
 	require.NoError(t, err)
-	assert.Equal(uint64(12), m1.Len())
+	assert.Equal(uint64(10), m1.Len())
 	for i := 0; i < len(l); i += 2 {
 		v, ok, err := m1.MaybeGet(context.Background(), l[i])
 		require.NoError(t, err)
diff --git a/integration-tests/bats/back-compat.bats b/integration-tests/bats/back-compat.bats
index dcccc769e6..bf14073231 100644
--- a/integration-tests/bats/back-compat.bats
+++ b/integration-tests/bats/back-compat.bats
@@ -25,7 +25,7 @@ teardown() {
         run dolt schema show
         [ "$status" -eq "0" ]
         [[ "$output" =~ "\`pk\` bigint NOT NULL" ]] || false
-        [[ "$output" =~ "\`a\` longtext" ]] || false
+        [[ "$output" =~ "\`a\` longtext" || "$output" =~ "\`a\` varchar(16383)" ]] || false
         [[ "$output" =~ "\`b\` datetime" ]] || false
         run dolt sql -q "select * from abc order by pk asc"
         [ "$status" -eq "0" ]
@@ -36,7 +36,7 @@ teardown() {
         run dolt schema show
         [ "$status" -eq "0" ]
         [[ "$output" =~ "\`pk\` bigint NOT NULL" ]] || false
-        [[ "$output" =~ "\`a\` longtext" ]] || false
+        [[ "$output" =~ "\`a\` longtext" || "$output" =~ "\`a\` varchar(16383)" ]] || false
         [[ "$output" =~ "\`b\` datetime" ]] || false
         run dolt sql -q "select * from abc order by pk asc"
         [ "$status" -eq "0" ]
@@ -50,7 +50,7 @@ teardown() {
         run dolt schema show
         [ "$status" -eq "0" ]
         [[ "$output" =~ "\`pk\` bigint NOT NULL" ]] || false
-        [[ "$output" =~ "\`a\` longtext" ]] || false
+        [[ "$output" =~ "\`a\` longtext" || "$output" =~ "\`a\` varchar(16383)" ]] || false
         [[ "$output" =~ "\`b\` datetime" ]] || false
         [[ "$output" =~ "\`c\` bigint unsigned" ]] || false
         run dolt sql -q "select * from abc order by pk asc"
diff --git a/integration-tests/bats/column_tags.bats b/integration-tests/bats/column_tags.bats
index 62ba592588..9308a77ab1 100644
--- a/integration-tests/bats/column_tags.bats
+++ b/integration-tests/bats/column_tags.bats
@@ -217,7 +217,7 @@ SQL
     [[ "$output" =~ "test1,pk1,10458" ]] || false
     [[ "$output" =~ "test1,c1,5951" ]] || false
     [[ "$output" =~ "test1,c2,10358" ]] || false
-    [[ "$output" =~ "test1,c3,11314" ]] || false
+    [[ "$output" =~ "test1,c3,16293" ]] || false
 }
 
 @test "column_tags: dolt table import -c uses deterministic tag generation" {
diff --git a/integration-tests/bats/helper/common.bash b/integration-tests/bats/helper/common.bash
index 483ad96727..d7673ed109 100644
--- a/integration-tests/bats/helper/common.bash
+++ b/integration-tests/bats/helper/common.bash
@@ -43,7 +43,7 @@ setup_no_dolt_init() {
 
 assert_feature_version() {
     run dolt version --feature
-    [[ "$output" =~ "feature version: 1" ]] || exit 1
+    [[ "$output" =~ "feature version: 2" ]] || exit 1
 }
 
 setup_common() {
diff --git a/integration-tests/bats/import-create-tables.bats b/integration-tests/bats/import-create-tables.bats
index 2580ab7f20..828a853752 100755
--- a/integration-tests/bats/import-create-tables.bats
+++ b/integration-tests/bats/import-create-tables.bats
@@ -522,7 +522,7 @@ DELIM
     [ "$status" -eq 0 ]
     [[ "$output" =~ "CREATE TABLE \`test\`" ]]
     [[ "$output" =~ "\`pk\` int" ]]
-    [[ "$output" =~ "\`str\` longtext" ]]
+    [[ "$output" =~ "\`str\` varchar(16383)" ]]
     [[ "$output" =~ "\`int\` int unsigned" ]]
     [[ "$output" =~ "\`bool\` bit(1)" ]]
     [[ "$output" =~ "\`float\` float" ]]
diff --git a/integration-tests/bats/schema-import.bats b/integration-tests/bats/schema-import.bats
index dd401f9df1..4b6bdf1c9d 100755
--- a/integration-tests/bats/schema-import.bats
+++ b/integration-tests/bats/schema-import.bats
@@ -79,7 +79,7 @@ teardown() {
     [[ "${lines[0]}" =~ "test" ]] || false
     [[ "$output" =~ "\`pk\` int" ]] || false
     [[ "$output" =~ "\`int\` int" ]] || false
-    [[ "$output" =~ "\`string\` longtext" ]] || false
+    [[ "$output" =~ "\`string\` varchar(16383)" ]] || false
     [[ "$output" =~ "\`boolean\` bit(1)" ]] || false
     [[ "$output" =~ "\`float\` float" ]] || false
     [[ "$output" =~ "\`uint\` int unsigned" ]] || false
@@ -104,7 +104,7 @@ DELIM
     [[ "${lines[0]}" =~ "test" ]] || false
     [[ "$output" =~ "\`pk\` int" ]] || false
     [[ "$output" =~ "\`int\` int" ]] || false
-    [[ "$output" =~ "\`string\` longtext" ]] || false
+    [[ "$output" =~ "\`string\` varchar(16383)" ]] || false
     [[ "$output" =~ "\`boolean\` bit(1)" ]] || false
     [[ "$output" =~ "\`float\` float" ]] || false
     [[ "$output" =~ "\`uint\` int" ]] || false
@@ -167,8 +167,8 @@ DELIM
     [ "$status" -eq 0 ]
     [ "${#lines[@]}" -eq 7 ]
     [[ "${lines[0]}" =~ "test" ]] || false
-    [[ "$output" =~ "\`pk\` longtext" ]] || false
-    [[ "$output" =~ "\`headerOne\` longtext" ]] || false
+    [[ "$output" =~ "\`pk\` varchar(16383)" ]] || false
+    [[ "$output" =~ "\`headerOne\` varchar(16383)" ]] || false
     [[ "$output" =~ "\`headerTwo\` int" ]] || false
 }
 
@@ -193,7 +193,7 @@ DELIM
     [[ "$output" =~ "\`c3\` int" ]] || false
     [[ "$output" =~ "\`c4\` int" ]] || false
     [[ "$output" =~ "\`c5\` int" ]] || false
-    [[ "$output" =~ "\`c6\` longtext" ]] || false
+    [[ "$output" =~ "\`c6\` varchar(16383)" ]] || false
     [[ "$output" =~ "PRIMARY KEY (\`pk\`)" ]] || false
 }
 
@@ -209,12 +209,12 @@ DELIM
     [ "${#lines[@]}" -eq 11 ]
     [[ "${lines[0]}" =~ "test" ]] || false
     [[ "$output" =~ "\`pk\` int" ]] || false
-    [[ "$output" =~ "\`c1\` longtext" ]] || false
-    [[ "$output" =~ "\`c2\` longtext" ]] || false
-    [[ "$output" =~ "\`c3\` longtext" ]] || false
-    [[ "$output" =~ "\`c4\` longtext" ]] || false
-    [[ "$output" =~ "\`c5\` longtext" ]] || false
-    [[ "$output" =~ "\`c6\` longtext" ]] || false
+    [[ "$output" =~ "\`c1\` varchar(16383)" ]] || false
+    [[ "$output" =~ "\`c2\` varchar(16383)" ]] || false
+    [[ "$output" =~ "\`c3\` varchar(16383)" ]] || false
+    [[ "$output" =~ "\`c4\` varchar(16383)" ]] || false
+    [[ "$output" =~ "\`c5\` varchar(16383)" ]] || false
+    [[ "$output" =~ "\`c6\` varchar(16383)" ]] || false
     [[ "$output" =~ "PRIMARY KEY (\`pk\`)" ]] || false
 }
 
@@ -258,7 +258,7 @@ DELIM
     [ "$status" -eq 0 ]
     run dolt diff --schema
     [ "$status" -eq 0 ]
-    [[ "$output" =~ "+   \`x\` LONGTEXT" ]] || false
+    [[ "$output" =~ "+   \`x\` VARCHAR(16383)" ]] || false
     [[ "$output" =~ "+   \`y\` FLOAT" ]] || false
     [[ "$output" =~ "+   \`z\` INT" ]] || false
     # assert no columns were deleted/replaced
@@ -279,7 +279,7 @@ DELIM
     [ "$status" -eq 0 ]
     run dolt diff --schema
     [ "$status" -eq 0 ]
-    [[ "$output" =~ "+   \`x\` LONGTEXT" ]] || false
+    [[ "$output" =~ "+   \`x\` VARCHAR(16383)" ]] || false
     [[ "$output" =~ "+   \`y\` FLOAT" ]] || false
     [[ "$output" =~ "+   \`z\` INT" ]] || false
     # assert no columns were deleted/replaced
diff --git a/integration-tests/bats/status.bats b/integration-tests/bats/status.bats
index c16906280a..a7f19a1daf 100644
--- a/integration-tests/bats/status.bats
+++ b/integration-tests/bats/status.bats
@@ -15,7 +15,7 @@ teardown() {
     run dolt version --feature
     [ "$status" -eq 0 ]
     [[ "$output" =~ "dolt version" ]] || false
-    [[ "$output" =~ "feature version: 1" ]] || false
+    [[ "$output" =~ "feature version: 2" ]] || false
 }
 
 @test "status: no changes" {
diff --git a/integration-tests/bats/types.bats b/integration-tests/bats/types.bats
index c931b00d00..fd24798912 100644
--- a/integration-tests/bats/types.bats
+++ b/integration-tests/bats/types.bats
@@ -666,11 +666,12 @@ SQL
 }
 
 @test "types: LONGTEXT" {
-    dolt sql <<SQL
+    dolt sql <<"SQL"
 CREATE TABLE test (
   pk BIGINT NOT NULL,
   v LONGTEXT,
-  PRIMARY KEY (pk)
+  PRIMARY KEY (pk),
+  INDEX (v)
 );
 SQL
     run dolt schema show
@@ -684,6 +685,29 @@ SQL
     run dolt sql -q "SELECT * FROM test"
     [ "$status" -eq "0" ]
     [[ "${lines[3]}" =~ " 1234567890 " ]] || false
+
+    dolt sql <<"SQL"
+DELETE FROM test;
+INSERT INTO test VALUES (1, '1'), (2, '11'), (3, '1'), (4, '12'), (5, '01'), (6, '0');
+SQL
+    run dolt sql -q "SELECT * FROM test WHERE v > '0'" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk,v" ]] || false
+    [[ "$output" =~ "5,01" ]] || false
+    [[ "$output" =~ "1,1" ]] || false
+    [[ "$output" =~ "3,1" ]] || false
+    [[ "$output" =~ "2,11" ]] || false
+    [[ "$output" =~ "4,12" ]] || false
+    [[ "${#lines[@]}" = "6" ]] || false
+    run dolt sql -q "SELECT * FROM test WHERE v <= '11'" -r=csv
+    [ "$status" -eq "0" ]
+    [[ "$output" =~ "pk,v" ]] || false
+    [[ "$output" =~ "2,11" ]] || false
+    [[ "$output" =~ "3,1" ]] || false
+    [[ "$output" =~ "1,1" ]] || false
+    [[ "$output" =~ "5,01" ]] || false
+    [[ "$output" =~ "6,0" ]] || false
+    [[ "${#lines[@]}" = "6" ]] || false
 }
 
 @test "types: MEDIUMBLOB" {
diff --git a/integration-tests/compatibility/test_files/backward_compatible_versions.txt b/integration-tests/compatibility/test_files/backward_compatible_versions.txt
index d496925291..207fafab36 100644
--- a/integration-tests/compatibility/test_files/backward_compatible_versions.txt
+++ b/integration-tests/compatibility/test_files/backward_compatible_versions.txt
@@ -8,4 +8,5 @@ v0.22.0
 v0.23.0
 v0.24.0
 v0.25.0
-v0.26.0
\ No newline at end of file
+v0.26.0
+v0.27.0
\ No newline at end of file
diff --git a/integration-tests/compatibility/test_files/forward_compatible_versions.txt b/integration-tests/compatibility/test_files/forward_compatible_versions.txt
index f7689f3404..e69de29bb2 100644
--- a/integration-tests/compatibility/test_files/forward_compatible_versions.txt
+++ b/integration-tests/compatibility/test_files/forward_compatible_versions.txt
@@ -1 +0,0 @@
-v0.26.0
\ No newline at end of file