New format blobs (#3588)

* [no-release-notes] New format blob read/write Use blob chunker to write blobs out of band. Tests for round tripping and address walking at the tree level. Run Blob enginetests and extra large file blob tests. * NodeStore ref refactors * fix index * run verify * fix more bats * small PR cleanup * more changes * last bats errors
2026-03-03 10:08:59 -06:00 · 2022-06-22 18:02:48 -07:00
parent dbaaa8c516
commit ac80ba7770
61 changed files with 813 additions and 381 deletions
--- a/go/gen/fb/serial/encoding.go
+++ b/go/gen/fb/serial/encoding.go
@@ -23,87 +23,90 @@ import (
 type Encoding byte

 const (
-	EncodingNull     Encoding = 0
-	EncodingInt8     Encoding = 1
-	EncodingUint8    Encoding = 2
-	EncodingInt16    Encoding = 3
-	EncodingUint16   Encoding = 4
-	EncodingInt32    Encoding = 7
-	EncodingUint32   Encoding = 8
-	EncodingInt64    Encoding = 9
-	EncodingUint64   Encoding = 10
-	EncodingFloat32  Encoding = 11
-	EncodingFloat64  Encoding = 12
-	EncodingBit64    Encoding = 13
-	EncodingHash128  Encoding = 14
-	EncodingYear     Encoding = 15
-	EncodingDate     Encoding = 16
-	EncodingTime     Encoding = 17
-	EncodingDatetime Encoding = 18
-	EncodingEnum     Encoding = 19
-	EncodingSet      Encoding = 20
-	EncodingAddress  Encoding = 21
-	EncodingString   Encoding = 128
-	EncodingBytes    Encoding = 129
-	EncodingDecimal  Encoding = 130
-	EncodingJSON     Encoding = 131
-	EncodingGeometry Encoding = 133
+	EncodingNull       Encoding = 0
+	EncodingInt8       Encoding = 1
+	EncodingUint8      Encoding = 2
+	EncodingInt16      Encoding = 3
+	EncodingUint16     Encoding = 4
+	EncodingInt32      Encoding = 7
+	EncodingUint32     Encoding = 8
+	EncodingInt64      Encoding = 9
+	EncodingUint64     Encoding = 10
+	EncodingFloat32    Encoding = 11
+	EncodingFloat64    Encoding = 12
+	EncodingBit64      Encoding = 13
+	EncodingHash128    Encoding = 14
+	EncodingYear       Encoding = 15
+	EncodingDate       Encoding = 16
+	EncodingTime       Encoding = 17
+	EncodingDatetime   Encoding = 18
+	EncodingEnum       Encoding = 19
+	EncodingSet        Encoding = 20
+	EncodingBytesAddr  Encoding = 21
+	EncodingCommitAddr Encoding = 22
+	EncodingString     Encoding = 128
+	EncodingBytes      Encoding = 129
+	EncodingDecimal    Encoding = 130
+	EncodingJSON       Encoding = 131
+	EncodingGeometry   Encoding = 133
 )

 var EnumNamesEncoding = map[Encoding]string{
-	EncodingNull:     "Null",
-	EncodingInt8:     "Int8",
-	EncodingUint8:    "Uint8",
-	EncodingInt16:    "Int16",
-	EncodingUint16:   "Uint16",
-	EncodingInt32:    "Int32",
-	EncodingUint32:   "Uint32",
-	EncodingInt64:    "Int64",
-	EncodingUint64:   "Uint64",
-	EncodingFloat32:  "Float32",
-	EncodingFloat64:  "Float64",
-	EncodingBit64:    "Bit64",
-	EncodingHash128:  "Hash128",
-	EncodingYear:     "Year",
-	EncodingDate:     "Date",
-	EncodingTime:     "Time",
-	EncodingDatetime: "Datetime",
-	EncodingEnum:     "Enum",
-	EncodingSet:      "Set",
-	EncodingAddress:  "Address",
-	EncodingString:   "String",
-	EncodingBytes:    "Bytes",
-	EncodingDecimal:  "Decimal",
-	EncodingJSON:     "JSON",
-	EncodingGeometry: "Geometry",
+	EncodingNull:       "Null",
+	EncodingInt8:       "Int8",
+	EncodingUint8:      "Uint8",
+	EncodingInt16:      "Int16",
+	EncodingUint16:     "Uint16",
+	EncodingInt32:      "Int32",
+	EncodingUint32:     "Uint32",
+	EncodingInt64:      "Int64",
+	EncodingUint64:     "Uint64",
+	EncodingFloat32:    "Float32",
+	EncodingFloat64:    "Float64",
+	EncodingBit64:      "Bit64",
+	EncodingHash128:    "Hash128",
+	EncodingYear:       "Year",
+	EncodingDate:       "Date",
+	EncodingTime:       "Time",
+	EncodingDatetime:   "Datetime",
+	EncodingEnum:       "Enum",
+	EncodingSet:        "Set",
+	EncodingBytesAddr:  "BytesAddr",
+	EncodingCommitAddr: "CommitAddr",
+	EncodingString:     "String",
+	EncodingBytes:      "Bytes",
+	EncodingDecimal:    "Decimal",
+	EncodingJSON:       "JSON",
+	EncodingGeometry:   "Geometry",
 }

 var EnumValuesEncoding = map[string]Encoding{
-	"Null":     EncodingNull,
-	"Int8":     EncodingInt8,
-	"Uint8":    EncodingUint8,
-	"Int16":    EncodingInt16,
-	"Uint16":   EncodingUint16,
-	"Int32":    EncodingInt32,
-	"Uint32":   EncodingUint32,
-	"Int64":    EncodingInt64,
-	"Uint64":   EncodingUint64,
-	"Float32":  EncodingFloat32,
-	"Float64":  EncodingFloat64,
-	"Bit64":    EncodingBit64,
-	"Hash128":  EncodingHash128,
-	"Year":     EncodingYear,
-	"Date":     EncodingDate,
-	"Time":     EncodingTime,
-	"Datetime": EncodingDatetime,
-	"Enum":     EncodingEnum,
-	"Set":      EncodingSet,
-	"Address":  EncodingAddress,
-	"String":   EncodingString,
-	"Bytes":    EncodingBytes,
-	"Decimal":  EncodingDecimal,
-	"JSON":     EncodingJSON,
-	"Geometry": EncodingGeometry,
+	"Null":       EncodingNull,
+	"Int8":       EncodingInt8,
+	"Uint8":      EncodingUint8,
+	"Int16":      EncodingInt16,
+	"Uint16":     EncodingUint16,
+	"Int32":      EncodingInt32,
+	"Uint32":     EncodingUint32,
+	"Int64":      EncodingInt64,
+	"Uint64":     EncodingUint64,
+	"Float32":    EncodingFloat32,
+	"Float64":    EncodingFloat64,
+	"Bit64":      EncodingBit64,
+	"Hash128":    EncodingHash128,
+	"Year":       EncodingYear,
+	"Date":       EncodingDate,
+	"Time":       EncodingTime,
+	"Datetime":   EncodingDatetime,
+	"Enum":       EncodingEnum,
+	"Set":        EncodingSet,
+	"BytesAddr":  EncodingBytesAddr,
+	"CommitAddr": EncodingCommitAddr,
+	"String":     EncodingString,
+	"Bytes":      EncodingBytes,
+	"Decimal":    EncodingDecimal,
+	"JSON":       EncodingJSON,
+	"Geometry":   EncodingGeometry,
 }

 func (v Encoding) String() string {
--- a/go/go.mod
+++ b/go/go.mod
@@ -68,7 +68,7 @@ require (
 )

 require (
-	github.com/dolthub/go-mysql-server v0.12.1-0.20220621185654-bbe8bee61d0a
+	github.com/dolthub/go-mysql-server v0.12.1-0.20220622173612-932627449b01
 	github.com/google/flatbuffers v2.0.6+incompatible
 	github.com/gosuri/uilive v0.0.4
 	github.com/kch42/buzhash v0.0.0-20160816060738-9bdec3dec7c6
--- a/go/go.sum
+++ b/go/go.sum
@@ -178,8 +178,8 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZm
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/dolthub/fslock v0.0.3 h1:iLMpUIvJKMKm92+N1fmHVdxJP5NdyDK5bK7z7Ba2s2U=
 github.com/dolthub/fslock v0.0.3/go.mod h1:QWql+P17oAAMLnL4HGB5tiovtDuAjdDTPbuqx7bYfa0=
-github.com/dolthub/go-mysql-server v0.12.1-0.20220621185654-bbe8bee61d0a h1:+gT2+flsaQUJ/H/Ap8X4PWvdQU3zjfDG0Q6klxpKYuM=
-github.com/dolthub/go-mysql-server v0.12.1-0.20220621185654-bbe8bee61d0a/go.mod h1:gvDEMITJQDVYDLR4XtcqEZx6rawTvMh2veM1bPsJC3I=
+github.com/dolthub/go-mysql-server v0.12.1-0.20220622173612-932627449b01 h1:K7D2ge4FwpTqS3FnnTum0Zg3MqqnYGPELeoIbHIjJgg=
+github.com/dolthub/go-mysql-server v0.12.1-0.20220622173612-932627449b01/go.mod h1:gvDEMITJQDVYDLR4XtcqEZx6rawTvMh2veM1bPsJC3I=
 github.com/dolthub/ishell v0.0.0-20220112232610-14e753f0f371 h1:oyPHJlzumKta1vnOQqUnfdz+pk3EmnHS3Nd0cCT0I2g=
 github.com/dolthub/ishell v0.0.0-20220112232610-14e753f0f371/go.mod h1:dhGBqcCEfK5kuFmeO5+WOx3hqc1k3M29c1oS/R7N4ms=
 github.com/dolthub/jsonpath v0.0.0-20210609232853-d49537a30474 h1:xTrR+l5l+1Lfq0NvhiEsctylXinUMFhhsqaEcl414p8=
--- a/go/libraries/doltcore/doltdb/table.go
+++ b/go/libraries/doltcore/doltdb/table.go
@@ -29,6 +29,8 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema/typeinfo"
 	"github.com/dolthub/dolt/go/store/hash"
+	"github.com/dolthub/dolt/go/store/prolly/shim"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/types"
 )

@@ -106,6 +108,11 @@ func (t *Table) ValueReadWriter() types.ValueReadWriter {
 	return durable.VrwFromTable(t.table)
 }

+// NodeStore returns the NodeStore for this table.
+func (t *Table) NodeStore() tree.NodeStore {
+	return tree.NewNodeStore(shim.ChunkStoreFromVRW(t.ValueReadWriter()))
+}
+
 // SetConflicts sets the merge conflicts for this table.
 func (t *Table) SetConflicts(ctx context.Context, schemas conflict.ConflictSchema, conflictData durable.ConflictIndex) (*Table, error) {
 	table, err := t.table.SetConflicts(ctx, schemas, conflictData)
--- a/go/libraries/doltcore/schema/serial_encoding.go
+++ b/go/libraries/doltcore/schema/serial_encoding.go
@@ -24,13 +24,6 @@ import (

 // EncodingFromSqlType returns a serial.Encoding for a query.Type.
 func EncodingFromSqlType(typ query.Type) serial.Encoding {
-	// todo(andy): replace temp encodings
-	switch typ {
-	case query.Type_BLOB, query.Type_TEXT:
-		// todo: temporary hack for enginetests
-		return serial.EncodingString
-	}
-
 	switch typ {
 	case query.Type_INT8:
 		return serial.EncodingInt8
@@ -86,6 +79,10 @@ func EncodingFromSqlType(typ query.Type) serial.Encoding {
 		return serial.EncodingJSON
 	case query.Type_GEOMETRY:
 		return serial.EncodingGeometry
+	case query.Type_BLOB:
+		return serial.EncodingBytesAddr
+	case query.Type_TEXT:
+		return serial.EncodingString
 	default:
 		panic(fmt.Sprintf("unknown encoding %v", typ))
 	}
--- a/go/libraries/doltcore/schema/typeinfo/blobstring.go
+++ b/go/libraries/doltcore/schema/typeinfo/blobstring.go
@@ -20,6 +20,7 @@ import (
 	"strconv"
 	"strings"
 	"unicode/utf8"
+	"unsafe"

 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/dolthub/vitess/go/sqltypes"
@@ -73,7 +74,11 @@ func CreateBlobStringTypeFromParams(params map[string]string) (TypeInfo, error)
 // ConvertNomsValueToValue implements TypeInfo interface.
 func (ti *blobStringType) ConvertNomsValueToValue(v types.Value) (interface{}, error) {
 	if val, ok := v.(types.Blob); ok {
-		return fromBlob(val)
+		b, err := fromBlob(val)
+		if sql.IsBinaryType(ti.sqlStringType) {
+			return b, err
+		}
+		return string(b), err
 	}
 	if _, ok := v.(types.Null); ok || v == nil {
 		return nil, nil
@@ -90,7 +95,11 @@ func (ti *blobStringType) ReadFrom(_ *types.NomsBinFormat, reader types.CodecRea
 		if err != nil {
 			return nil, err
 		}
-		return fromBlob(val)
+		b, err := fromBlob(val)
+		if sql.IsBinaryType(ti.sqlStringType) {
+			return b, err
+		}
+		return string(b), err
 	case types.NullKind:
 		_ = reader.ReadKind()
 		return nil, nil
@@ -134,7 +143,7 @@ func (ti *blobStringType) FormatValue(v types.Value) (*string, error) {
 		if err != nil {
 			return nil, err
 		}
-		return &resStr, nil
+		return (*string)(unsafe.Pointer(&resStr)), nil
 	}
 	if _, ok := v.(types.Null); ok || v == nil {
 		return nil, nil
@@ -204,7 +213,7 @@ func blobStringTypeConverter(ctx context.Context, src *blobStringType, destTi Ty
 			if err != nil {
 				return nil, err
 			}
-			newVal, err := strconv.ParseUint(val, 10, int(dest.sqlBitType.NumberOfBits()))
+			newVal, err := strconv.ParseUint(string(val), 10, int(dest.sqlBitType.NumberOfBits()))
 			if err != nil {
 				return nil, err
 			}
--- a/go/libraries/doltcore/schema/typeinfo/inlineblob.go
+++ b/go/libraries/doltcore/schema/typeinfo/inlineblob.go
@@ -105,9 +105,9 @@ func (ti *inlineBlobType) ConvertValueToNomsValue(ctx context.Context, vrw types
 	if err != nil {
 		return nil, err
 	}
-	val, ok := strVal.(string)
+	val, ok := strVal.([]byte)
 	if ok {
-		return *(*types.InlineBlob)(unsafe.Pointer(&val)), nil
+		return types.InlineBlob(val), nil
 	}
 	return nil, fmt.Errorf(`"%v" has unexpectedly encountered a value of type "%T" from embedded type`, ti.String(), v)
 }
--- a/go/libraries/doltcore/schema/typeinfo/typeconverter.go
+++ b/go/libraries/doltcore/schema/typeinfo/typeconverter.go
@@ -122,7 +122,7 @@ func wrapConvertValueToNomsValue(
 			if err != nil {
 				return nil, err
 			}
-			vInt = str
+			vInt = string(str)
 		case types.Bool:
 			vInt = bool(val)
 		case types.Decimal:
--- a/go/libraries/doltcore/schema/typeinfo/varbinary.go
+++ b/go/libraries/doltcore/schema/typeinfo/varbinary.go
@@ -101,9 +101,9 @@ func (ti *varBinaryType) ConvertValueToNomsValue(ctx context.Context, vrw types.
 	if err != nil {
 		return nil, err
 	}
-	val, ok := strVal.(string)
+	val, ok := strVal.([]byte)
 	if ok {
-		return types.NewBlob(ctx, vrw, strings.NewReader(val))
+		return types.NewBlob(ctx, vrw, strings.NewReader(string(val)))
 	}
 	return nil, fmt.Errorf(`"%v" cannot convert value "%v" of type "%T" as it is invalid`, ti.String(), v, v)
 }
@@ -126,7 +126,7 @@ func (ti *varBinaryType) FormatValue(v types.Value) (*string, error) {
 		if err != nil {
 			return nil, err
 		}
-		return &resStr, nil
+		return (*string)(unsafe.Pointer(&resStr)), nil
 	}
 	if _, ok := v.(types.Null); ok || v == nil {
 		return nil, nil
@@ -180,18 +180,18 @@ func (ti *varBinaryType) ToSqlType() sql.Type {
 }

 // fromBlob returns a string from a types.Blob.
-func fromBlob(b types.Blob) (string, error) {
+func fromBlob(b types.Blob) ([]byte, error) {
 	strLength := b.Len()
 	if strLength == 0 {
-		return "", nil
+		return []byte{}, nil
 	}
 	str := make([]byte, strLength)
 	n, err := b.ReadAt(context.Background(), str, 0)
 	if err != nil && err != io.EOF {
-		return "", err
+		return []byte{}, err
 	}
 	if uint64(n) != strLength {
-		return "", fmt.Errorf("wanted %d bytes from blob for data, got %d", strLength, n)
+		return []byte{}, fmt.Errorf("wanted %d bytes from blob for data, got %d", strLength, n)
 	}

 	// For very large byte slices, the standard method of converting a byte slice to a string using "string(str)" will
@@ -200,7 +200,7 @@ func fromBlob(b types.Blob) (string, error) {
 	// testing, performance improved by 40%.
 	// This is inspired by Go's own source code in strings.Builder.String(): https://golang.org/src/strings/builder.go#L48
 	// This is also marked as a valid strategy in unsafe.Pointer's own method documentation.
-	return *(*string)(unsafe.Pointer(&str)), nil
+	return str, nil
 }

 // hasPrefix finds out if a Blob has a prefixed integer. Initially blobs for varBinary prepended an integer indicating
@@ -242,7 +242,7 @@ func varBinaryTypeConverter(ctx context.Context, src *varBinaryType, destTi Type
 			if err != nil {
 				return nil, err
 			}
-			newVal, err := strconv.ParseUint(val, 10, int(dest.sqlBitType.NumberOfBits()))
+			newVal, err := strconv.ParseUint(string(val), 10, int(dest.sqlBitType.NumberOfBits()))
 			if err != nil {
 				return nil, err
 			}
--- a/go/libraries/doltcore/sqle/dtables/conflicts_tables_prolly.go
+++ b/go/libraries/doltcore/sqle/dtables/conflicts_tables_prolly.go
@@ -169,7 +169,7 @@ func (itr *prollyConflictRowIter) Next(ctx *sql.Context) (sql.Row, error) {
 	r[0] = c.h.String()

 	for i := 0; i < itr.kd.Count(); i++ {
-		f, err := index.GetField(itr.kd, i, c.k)
+		f, err := index.GetField(ctx, itr.kd, i, c.k, itr.baseRows.NodeStore())
 		if err != nil {
 			return nil, err
 		}
@@ -186,7 +186,7 @@ func (itr *prollyConflictRowIter) Next(ctx *sql.Context) (sql.Row, error) {

 	if c.bV != nil {
 		for i := 0; i < itr.baseVD.Count(); i++ {
-			f, err := index.GetField(itr.baseVD, i, c.bV)
+			f, err := index.GetField(ctx, itr.baseVD, i, c.bV, itr.baseRows.NodeStore())
 			if err != nil {
 				return nil, err
 			}
@@ -196,7 +196,7 @@ func (itr *prollyConflictRowIter) Next(ctx *sql.Context) (sql.Row, error) {

 	if c.oV != nil {
 		for i := 0; i < itr.oursVD.Count(); i++ {
-			f, err := index.GetField(itr.oursVD, i, c.oV)
+			f, err := index.GetField(ctx, itr.oursVD, i, c.oV, itr.baseRows.NodeStore())
 			if err != nil {
 				return nil, err
 			}
@@ -206,7 +206,7 @@ func (itr *prollyConflictRowIter) Next(ctx *sql.Context) (sql.Row, error) {

 	if c.tV != nil {
 		for i := 0; i < itr.theirsVD.Count(); i++ {
-			f, err := index.GetField(itr.theirsVD, i, c.tV)
+			f, err := index.GetField(ctx, itr.theirsVD, i, c.tV, itr.baseRows.NodeStore())
 			if err != nil {
 				return nil, err
 			}
@@ -353,7 +353,8 @@ func (cd *prollyConflictDeleter) Delete(ctx *sql.Context, r sql.Row) error {

 	// first part of the artifact key is the keys of the source table
 	for i := 0; i < cd.kd.Count()-2; i++ {
-		err := index.PutField(cd.kB, i, r[o+i])
+		err := index.PutField(ctx, cd.ed.Mut.NodeStore(), cd.kB, i, r[o+i])
+
 		if err != nil {
 			return err
 		}
@@ -361,7 +362,7 @@ func (cd *prollyConflictDeleter) Delete(ctx *sql.Context, r sql.Row) error {

 	// then the hash follows. It is the first column of the row and the second to last in the key
 	h := hash.Parse(r[0].(string))
-	cd.kB.PutAddress(cd.kd.Count()-2, h)
+	cd.kB.PutCommitAddr(cd.kd.Count()-2, h)

 	// Finally the artifact type which is always a conflict
 	cd.kB.PutUint8(cd.kd.Count()-1, uint8(prolly.ArtifactTypeConflict))
--- a/go/libraries/doltcore/sqle/dtables/constraint_violations_prolly.go
+++ b/go/libraries/doltcore/sqle/dtables/constraint_violations_prolly.go
@@ -29,6 +29,7 @@ import (
 	"github.com/dolthub/dolt/go/store/pool"
 	"github.com/dolthub/dolt/go/store/prolly"
 	"github.com/dolthub/dolt/go/store/prolly/shim"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )

@@ -112,7 +113,13 @@ func (cvt *prollyConstraintViolationsTable) PartitionRows(ctx *sql.Context, part
 		return nil, err
 	}
 	kd, vd := shim.MapDescriptorsFromSchema(sch)
-	return prollyCVIter{itr, sch, kd, vd}, nil
+	return prollyCVIter{
+		itr: itr,
+		sch: sch,
+		kd:  kd,
+		vd:  vd,
+		ns:  cvt.artM.NodeStore(),
+	}, nil
 }

 func (cvt *prollyConstraintViolationsTable) Deleter(context *sql.Context) sql.RowDeleter {
@@ -134,6 +141,7 @@ type prollyCVIter struct {
 	itr    prolly.ArtifactIter
 	sch    schema.Schema
 	kd, vd val.TupleDesc
+	ns     tree.NodeStore
 }

 func (itr prollyCVIter) Next(ctx *sql.Context) (sql.Row, error) {
@@ -154,7 +162,7 @@ func (itr prollyCVIter) Next(ctx *sql.Context) (sql.Row, error) {

 	o := 2
 	for i := 0; i < itr.kd.Count(); i++ {
-		r[o+i], err = index.GetField(itr.kd, i, art.Key)
+		r[o+i], err = index.GetField(ctx, itr.kd, i, art.Key, itr.ns)
 		if err != nil {
 			return nil, err
 		}
@@ -162,7 +170,7 @@ func (itr prollyCVIter) Next(ctx *sql.Context) (sql.Row, error) {
 	o += itr.kd.Count()

 	for i := 0; i < itr.vd.Count(); i++ {
-		r[o+i], err = index.GetField(itr.vd, i, meta.Value)
+		r[o+i], err = index.GetField(ctx, itr.vd, i, meta.Value, itr.ns)
 		if err != nil {
 			return nil, err
 		}
@@ -193,7 +201,7 @@ var _ sql.RowDeleter = (*prollyCVDeleter)(nil)
 func (d *prollyCVDeleter) Delete(ctx *sql.Context, r sql.Row) error {
 	// first part of the artifact key is the keys of the source table
 	for i := 0; i < d.kd.Count()-2; i++ {
-		err := index.PutField(d.kb, i, r[i+2])
+		err := index.PutField(ctx, d.cvt.artM.NodeStore(), d.kb, i, r[i+2])
 		if err != nil {
 			return err
 		}
@@ -201,7 +209,7 @@ func (d *prollyCVDeleter) Delete(ctx *sql.Context, r sql.Row) error {

 	// then the hash
 	h := hash.Parse(r[0].(string))
-	d.kb.PutAddress(d.kd.Count()-2, h)
+	d.kb.PutCommitAddr(d.kd.Count()-2, h)

 	// Finally the artifact type
 	artType := unmapCVType(merge.CvType(r[1].(uint64)))
--- a/go/libraries/doltcore/sqle/dtables/diff_iter.go
+++ b/go/libraries/doltcore/sqle/dtables/diff_iter.go
@@ -256,12 +256,12 @@ func newProllyDiffIter(ctx *sql.Context, dp DiffPartition, ddb *doltdb.DoltDB, t
 	}
 	to := durable.ProllyMapFromIndex(t)

-	fromConverter, err := NewProllyRowConverter(fSch, targetFromSchema, ctx.Warn)
+	fromConverter, err := NewProllyRowConverter(fSch, targetFromSchema, ctx.Warn, nil)
 	if err != nil {
 		return prollyDiffIter{}, err
 	}

-	toConverter, err := NewProllyRowConverter(tSch, targetToSchema, ctx.Warn)
+	toConverter, err := NewProllyRowConverter(tSch, targetToSchema, ctx.Warn, dp.to.NodeStore())
 	if err != nil {
 		return prollyDiffIter{}, err
 	}
@@ -312,7 +312,7 @@ func (itr prollyDiffIter) Close(ctx *sql.Context) error {

 func (itr prollyDiffIter) queueRows(ctx context.Context) {
 	err := prolly.DiffMaps(ctx, itr.from, itr.to, func(ctx context.Context, d tree.Diff) error {
-		r, err := itr.makeDiffRow(d)
+		r, err := itr.makeDiffRow(ctx, d)
 		if err != nil {
 			return err
 		}
@@ -335,7 +335,7 @@ func (itr prollyDiffIter) queueRows(ctx context.Context) {
 }

 // todo(andy): copy string fields
-func (itr prollyDiffIter) makeDiffRow(d tree.Diff) (r sql.Row, err error) {
+func (itr prollyDiffIter) makeDiffRow(ctx context.Context, d tree.Diff) (r sql.Row, err error) {

 	n := itr.targetToSch.GetAllCols().Size()
 	m := itr.targetFromSch.GetAllCols().Size()
@@ -345,7 +345,7 @@ func (itr prollyDiffIter) makeDiffRow(d tree.Diff) (r sql.Row, err error) {
 	// todo (dhruv): implement warnings for row column value coercions.

 	if d.Type != tree.RemovedDiff {
-		err = itr.toConverter.PutConverted(val.Tuple(d.Key), val.Tuple(d.To), r[0:n])
+		err = itr.toConverter.PutConverted(ctx, val.Tuple(d.Key), val.Tuple(d.To), r[0:n])
 		if err != nil {
 			return nil, err
 		}
@@ -356,7 +356,7 @@ func (itr prollyDiffIter) makeDiffRow(d tree.Diff) (r sql.Row, err error) {
 	r[o+1] = maybeTime(itr.toCm.ts)

 	if d.Type != tree.AddedDiff {
-		err = itr.fromConverter.PutConverted(val.Tuple(d.Key), val.Tuple(d.From), r[n+2:n+2+m])
+		err = itr.fromConverter.PutConverted(ctx, val.Tuple(d.Key), val.Tuple(d.From), r[n+2:n+2+m])
 		if err != nil {
 			return nil, err
 		}
--- a/go/libraries/doltcore/sqle/dtables/prolly_row_conv.go
+++ b/go/libraries/doltcore/sqle/dtables/prolly_row_conv.go
@@ -15,6 +15,8 @@
 package dtables

 import (
+	"context"
+
 	"github.com/dolthub/go-mysql-server/sql"

 	"github.com/dolthub/dolt/go/libraries/doltcore/diff"
@@ -22,6 +24,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/index"
 	"github.com/dolthub/dolt/go/store/prolly/shim"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )

@@ -37,9 +40,10 @@ type ProllyRowConverter struct {
 	pkTargetTypes    []sql.Type
 	nonPkTargetTypes []sql.Type
 	warnFn           rowconv.WarnFunction
+	ns               tree.NodeStore
 }

-func NewProllyRowConverter(inSch, outSch schema.Schema, warnFn rowconv.WarnFunction) (ProllyRowConverter, error) {
+func NewProllyRowConverter(inSch, outSch schema.Schema, warnFn rowconv.WarnFunction, ns tree.NodeStore) (ProllyRowConverter, error) {
 	keyProj, valProj, err := diff.MapSchemaBasedOnName(inSch, outSch)
 	if err != nil {
 		return ProllyRowConverter{}, err
@@ -89,18 +93,19 @@ func NewProllyRowConverter(inSch, outSch schema.Schema, warnFn rowconv.WarnFunct
 		pkTargetTypes:    pkTargetTypes,
 		nonPkTargetTypes: nonPkTargetTypes,
 		warnFn:           warnFn,
+		ns:               ns,
 	}, nil
 }

 // PutConverted converts the |key| and |value| val.Tuple from |inSchema| to |outSchema|
 // and places the converted row in |dstRow|.
-func (c ProllyRowConverter) PutConverted(key, value val.Tuple, dstRow []interface{}) error {
-	err := c.putFields(key, c.keyProj, c.keyDesc, c.pkTargetTypes, dstRow)
+func (c ProllyRowConverter) PutConverted(ctx context.Context, key, value val.Tuple, dstRow []interface{}) error {
+	err := c.putFields(ctx, key, c.keyProj, c.keyDesc, c.pkTargetTypes, dstRow)
 	if err != nil {
 		return err
 	}

-	err = c.putFields(value, c.valProj, c.valDesc, c.nonPkTargetTypes, dstRow)
+	err = c.putFields(ctx, value, c.valProj, c.valDesc, c.nonPkTargetTypes, dstRow)
 	if err != nil {
 		return err
 	}
@@ -108,12 +113,12 @@ func (c ProllyRowConverter) PutConverted(key, value val.Tuple, dstRow []interfac
 	return nil
 }

-func (c ProllyRowConverter) putFields(tup val.Tuple, proj val.OrdinalMapping, desc val.TupleDesc, targetTypes []sql.Type, dstRow []interface{}) error {
+func (c ProllyRowConverter) putFields(ctx context.Context, tup val.Tuple, proj val.OrdinalMapping, desc val.TupleDesc, targetTypes []sql.Type, dstRow []interface{}) error {
 	for i, j := range proj {
 		if j == -1 {
 			continue
 		}
-		f, err := index.GetField(desc, i, tup)
+		f, err := index.GetField(ctx, desc, i, tup, c.ns)
 		if err != nil {
 			return err
 		}
--- a/go/libraries/doltcore/sqle/enginetest/blob_queries.go
+++ b/go/libraries/doltcore/sqle/enginetest/blob_queries.go
@@ -0,0 +1,29 @@
+// 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 enginetest
+
+import (
+	"github.com/dolthub/go-mysql-server/enginetest/queries"
+	"github.com/dolthub/go-mysql-server/sql"
+)
+
+var BigBlobQueries = []queries.WriteQueryTest{
+	{
+		WriteQuery:          "INSERT INTO blobt VALUES(4, LOAD_FILE('testdata/test1.png'))",
+		ExpectedWriteResult: []sql.Row{{sql.NewOkResult(1)}},
+		SelectQuery:         "select sha1(b) from blobt where i = 4",
+		ExpectedSelect:      []sql.Row{{"012bcb75a319f2913614a5170fc046fb6c49ee86"}},
+	},
+}
--- a/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
+++ b/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
@@ -440,6 +440,21 @@ func TestCreateDatabase(t *testing.T) {
 	enginetest.TestCreateDatabase(t, newDoltHarness(t))
 }

+func TestBlobs(t *testing.T) {
+	skipOldFormat(t)
+	enginetest.TestBlobs(t, newDoltHarness(t))
+}
+
+func TestBigBlobs(t *testing.T) {
+	skipOldFormat(t)
+
+	h := newDoltHarness(t)
+	h.Setup(setup.MydbData, setup.BlobData)
+	for _, tt := range BigBlobQueries {
+		enginetest.RunWriteQueryTest(t, h, tt)
+	}
+}
+
 func TestDropDatabase(t *testing.T) {
 	enginetest.TestScript(t, newDoltHarness(t), queries.ScriptTest{
 		Name: "Drop database engine tests for Dolt only",
@@ -1198,6 +1213,12 @@ var newFormatSkippedScripts = []string{
 	"Multiple indexes on the same columns in a different order",
 }

+func skipOldFormat(t *testing.T) {
+	if !types.IsFormat_DOLT_1(types.Format_Default) {
+		t.Skip()
+	}
+}
+
 func skipPreparedTests(t *testing.T) {
 	if skipPrepared {
 		t.Skip("skip prepared")
--- a/go/libraries/doltcore/sqle/enginetest/testdata/test1.png
+++ b/go/libraries/doltcore/sqle/enginetest/testdata/test1.png
--- a/go/libraries/doltcore/sqle/index/dolt_index.go
+++ b/go/libraries/doltcore/sqle/index/dolt_index.go
@@ -27,6 +27,8 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/table/typed/noms"
 	"github.com/dolthub/dolt/go/store/pool"
 	"github.com/dolthub/dolt/go/store/prolly"
+	"github.com/dolthub/dolt/go/store/prolly/shim"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/types"
 	"github.com/dolthub/dolt/go/store/val"
 )
@@ -379,7 +381,7 @@ func (di *doltIndex) NewLookup(ctx *sql.Context, ranges ...sql.Range) (sql.Index
 	}

 	if types.IsFormat_DOLT_1(di.vrw.Format()) {
-		return di.newProllyLookup(ctx, ranges...)
+		return di.newProllyLookup(ctx, tree.NewNodeStore(shim.ChunkStoreFromVRW(di.vrw)), ranges...)
 	}

 	return di.newNomsLookup(ctx, ranges...)
@@ -436,7 +438,7 @@ func (di *doltIndex) getDurableState(ctx *sql.Context, ti DoltTableable) (*durab
 	return ret, nil
 }

-func (di *doltIndex) newProllyLookup(ctx *sql.Context, ranges ...sql.Range) (sql.IndexLookup, error) {
+func (di *doltIndex) newProllyLookup(ctx *sql.Context, ns tree.NodeStore, ranges ...sql.Range) (sql.IndexLookup, error) {
 	var err error
 	sqlRanges, err := pruneEmptyRanges(ranges)
 	if err != nil {
@@ -445,7 +447,7 @@ func (di *doltIndex) newProllyLookup(ctx *sql.Context, ranges ...sql.Range) (sql

 	prs := make([]prolly.Range, len(sqlRanges))
 	for i, sr := range sqlRanges {
-		prs[i], err = prollyRangeFromSqlRange(sr, di.keyBld)
+		prs[i], err = prollyRangeFromSqlRange(ctx, ns, sr, di.keyBld)
 		if err != nil {
 			return nil, err
 		}
@@ -732,7 +734,7 @@ func pruneEmptyRanges(sqlRanges []sql.Range) (pruned []sql.Range, err error) {
 	return pruned, nil
 }

-func prollyRangeFromSqlRange(rng sql.Range, tb *val.TupleBuilder) (prolly.Range, error) {
+func prollyRangeFromSqlRange(ctx context.Context, ns tree.NodeStore, rng sql.Range, tb *val.TupleBuilder) (prolly.Range, error) {
 	prollyRange := prolly.Range{
 		Start: make([]prolly.RangeCut, len(rng)),
 		Stop:  make([]prolly.RangeCut, len(rng)),
@@ -748,7 +750,7 @@ func prollyRangeFromSqlRange(rng sql.Range, tb *val.TupleBuilder) (prolly.Range,
 		if err != nil {
 			return prolly.Range{}, err
 		}
-		if err = PutField(tb, i, v); err != nil {
+		if err = PutField(ctx, ns, tb, i, v); err != nil {
 			return prolly.Range{}, err
 		}
 	}
@@ -780,7 +782,7 @@ func prollyRangeFromSqlRange(rng sql.Range, tb *val.TupleBuilder) (prolly.Range,
 		if err != nil {
 			return prolly.Range{}, err
 		}
-		if err = PutField(tb, i, v); err != nil {
+		if err = PutField(ctx, ns, tb, i, v); err != nil {
 			return prolly.Range{}, err
 		}
 	}
--- a/go/libraries/doltcore/sqle/index/index_lookup.go
+++ b/go/libraries/doltcore/sqle/index/index_lookup.go
@@ -69,9 +69,8 @@ func RowIterForProllyRange(ctx *sql.Context, idx DoltIndex, r prolly.Range, pkSc
 	covers := idx.coversColumns(durableState, columns)
 	if covers {
 		return newProllyCoveringIndexIter(ctx, idx, r, pkSch, durableState.Secondary)
-	} else {
-		return newProllyIndexIter(ctx, idx, r, pkSch, durableState.Primary, durableState.Secondary)
 	}
+	return newProllyIndexIter(ctx, idx, r, pkSch, durableState.Primary, durableState.Secondary)
 }

 func RowIterForNomsRanges(ctx *sql.Context, idx DoltIndex, ranges []*noms.ReadRange, columns []string, durableState *durableIndexState) (sql.RowIter, error) {
--- a/go/libraries/doltcore/sqle/index/prolly_fields.go
+++ b/go/libraries/doltcore/sqle/index/prolly_fields.go
@@ -15,9 +15,12 @@
 package index

 import (
+	"bytes"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"math"
 	"time"

@@ -26,13 +29,14 @@ import (

 	geo "github.com/dolthub/dolt/go/store/geometry"
 	"github.com/dolthub/dolt/go/store/hash"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )

 var ErrValueExceededMaxFieldSize = errors.New("value exceeded max field size of 65kb")

 // GetField reads the value from the ith field of the Tuple as an interface{}.
-func GetField(td val.TupleDesc, i int, tup val.Tuple) (v interface{}, err error) {
+func GetField(ctx context.Context, td val.TupleDesc, i int, tup val.Tuple, ns tree.NodeStore) (v interface{}, err error) {
 	var ok bool
 	switch td.Types[i].Enc {
 	case val.Int8Enc:
@@ -95,8 +99,14 @@ func GetField(td val.TupleDesc, i int, tup val.Tuple) (v interface{}, err error)
 		}
 	case val.Hash128Enc:
 		v, ok = td.GetHash128(i, tup)
-	case val.AddressEnc:
-		v, ok = td.GetAddress(i, tup)
+	case val.BytesAddrEnc:
+		var b val.BytesAddr
+		b, ok = td.GetBlob(i, tup)
+		if ok {
+			v, err = tree.NewByteArray(b.Addr, ns).ToBytes(ctx)
+		}
+	case val.CommitAddrEnc:
+		v, ok = td.GetCommitAddr(i, tup)
 	default:
 		panic("unknown val.encoding")
 	}
@@ -107,7 +117,7 @@ func GetField(td val.TupleDesc, i int, tup val.Tuple) (v interface{}, err error)
 }

 // PutField writes an interface{} to the ith field of the Tuple being built.
-func PutField(tb *val.TupleBuilder, i int, v interface{}) error {
+func PutField(ctx context.Context, ns tree.NodeStore, tb *val.TupleBuilder, i int, v interface{}) error {
 	if v == nil {
 		return nil // NULL
 	}
@@ -177,8 +187,14 @@ func PutField(tb *val.TupleBuilder, i int, v interface{}) error {
 		tb.PutJSON(i, buf)
 	case val.Hash128Enc:
 		tb.PutHash128(i, v.([]byte))
-	case val.AddressEnc:
-		tb.PutAddress(i, v.(hash.Hash))
+	case val.BytesAddrEnc:
+		b, err := serializeBytesToAddr(ctx, ns, bytes.NewReader(v.([]byte)))
+		if err != nil {
+			return err
+		}
+		tb.PutBytesAddr(i, b.Addr)
+	case val.CommitAddrEnc:
+		tb.PutCommitAddr(i, v.(hash.Hash))
 	default:
 		panic(fmt.Sprintf("unknown encoding %v %v", enc, v))
 	}
@@ -266,6 +282,14 @@ func serializeGeometry(v interface{}) []byte {
 	}
 }

+func serializeBytesToAddr(ctx context.Context, ns tree.NodeStore, r io.Reader) (val.BytesAddr, error) {
+	tree, err := tree.NewImmutableTreeFromReader(ctx, r, ns, tree.DefaultFixedChunkLength)
+	if err != nil {
+		return val.BytesAddr{}, err
+	}
+	return val.NewBytesAddr(tree.Addr), nil
+}
+
 func convJson(v interface{}) (buf []byte, err error) {
 	v, err = sql.JSON.Convert(v)
 	if err != nil {
--- a/go/libraries/doltcore/sqle/index/prolly_fields_test.go
+++ b/go/libraries/doltcore/sqle/index/prolly_fields_test.go
@@ -15,6 +15,7 @@
 package index

 import (
+	"context"
 	"encoding/json"
 	"math"
 	"testing"
@@ -27,6 +28,7 @@ import (
 	"github.com/stretchr/testify/require"

 	"github.com/dolthub/dolt/go/store/pool"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )

@@ -161,6 +163,11 @@ func TestRoundTripProllyFields(t *testing.T) {
 			typ:   val.Type{Enc: val.GeometryEnc},
 			value: mustParseGeometryType(t, "POLYGON((0 0,1 1,1 0,0 0))"),
 		},
+		{
+			name:  "binary",
+			typ:   val.Type{Enc: val.BytesAddrEnc},
+			value: []byte("lorem ipsum"),
+		},
 	}

 	for _, test := range tests {
@@ -175,13 +182,14 @@ var testPool = pool.NewBuffPool()
 func testRoundTripProllyFields(t *testing.T, test prollyFieldTest) {
 	desc := val.NewTupleDescriptor(test.typ)
 	builder := val.NewTupleBuilder(desc)
+	ns := tree.NewTestNodeStore()

-	err := PutField(builder, 0, test.value)
+	err := PutField(context.Background(), ns, builder, 0, test.value)
 	assert.NoError(t, err)

 	tup := builder.Build(testPool)

-	v, err := GetField(desc, 0, tup)
+	v, err := GetField(context.Background(), desc, 0, tup, ns)
 	assert.NoError(t, err)
 	assert.Equal(t, test.value, v)
 }
--- a/go/libraries/doltcore/sqle/index/prolly_index_iter.go
+++ b/go/libraries/doltcore/sqle/index/prolly_index_iter.go
@@ -23,6 +23,7 @@ import (

 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb/durable"
 	"github.com/dolthub/dolt/go/store/prolly"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )

@@ -51,13 +52,7 @@ var _ sql.RowIter = prollyIndexIter{}
 var _ sql.RowIter2 = prollyIndexIter{}

 // NewProllyIndexIter returns a new prollyIndexIter.
-func newProllyIndexIter(
-	ctx *sql.Context,
-	idx DoltIndex,
-	rng prolly.Range,
-	pkSch sql.PrimaryKeySchema,
-	dprimary, dsecondary durable.Index,
-) (prollyIndexIter, error) {
+func newProllyIndexIter(ctx *sql.Context, idx DoltIndex, rng prolly.Range, pkSch sql.PrimaryKeySchema, dprimary, dsecondary durable.Index) (prollyIndexIter, error) {
 	secondary := durable.ProllyMapFromIndex(dsecondary)
 	indexIter, err := secondary.IterRange(ctx, rng)
 	if err != nil {
@@ -128,7 +123,7 @@ func (p prollyIndexIter) queueRows(ctx context.Context) error {

 		r := make(sql.Row, len(p.keyMap)+len(p.valMap))
 		err = p.primary.Get(ctx, pk, func(key, value val.Tuple) error {
-			return p.rowFromTuples(key, value, r)
+			return p.rowFromTuples(ctx, key, value, r)
 		})
 		if err != nil {
 			return err
@@ -142,14 +137,14 @@ func (p prollyIndexIter) queueRows(ctx context.Context) error {
 	}
 }

-func (p prollyIndexIter) rowFromTuples(key, value val.Tuple, r sql.Row) (err error) {
+func (p prollyIndexIter) rowFromTuples(ctx context.Context, key, value val.Tuple, r sql.Row) (err error) {
 	keyDesc, valDesc := p.primary.Descriptors()

 	for keyIdx, rowIdx := range p.keyMap {
 		if rowIdx == -1 {
 			continue
 		}
-		r[rowIdx], err = GetField(keyDesc, keyIdx, key)
+		r[rowIdx], err = GetField(ctx, keyDesc, keyIdx, key, p.primary.NodeStore())
 		if err != nil {
 			return err
 		}
@@ -158,7 +153,7 @@ func (p prollyIndexIter) rowFromTuples(key, value val.Tuple, r sql.Row) (err err
 		if rowIdx == -1 {
 			continue
 		}
-		r[rowIdx], err = GetField(valDesc, valIdx, value)
+		r[rowIdx], err = GetField(ctx, valDesc, valIdx, value, p.primary.NodeStore())
 		if err != nil {
 			return err
 		}
@@ -208,6 +203,8 @@ type prollyCoveringIndexIter struct {
 	keyDesc   val.TupleDesc
 	valDesc   val.TupleDesc

+	ns tree.NodeStore
+
 	// keyMap transforms secondary index key tuples into SQL tuples.
 	// secondary index value tuples are assumed to be empty.

@@ -242,6 +239,7 @@ func newProllyCoveringIndexIter(ctx *sql.Context, idx DoltIndex, rng prolly.Rang
 		keyMap:    keyMap,
 		valMap:    valMap,
 		sqlSch:    pkSch.Schema,
+		ns:        secondary.NodeStore(),
 	}

 	return iter, nil
@@ -255,7 +253,7 @@ func (p prollyCoveringIndexIter) Next(ctx *sql.Context) (sql.Row, error) {
 	}

 	r := make(sql.Row, len(p.keyMap))
-	if err := p.writeRowFromTuples(k, v, r); err != nil {
+	if err := p.writeRowFromTuples(ctx, k, v, r); err != nil {
 		return nil, err
 	}

@@ -271,13 +269,13 @@ func (p prollyCoveringIndexIter) Next2(ctx *sql.Context, f *sql.RowFrame) error
 	return p.writeRow2FromTuples(k, v, f)
 }

-func (p prollyCoveringIndexIter) writeRowFromTuples(key, value val.Tuple, r sql.Row) (err error) {
+func (p prollyCoveringIndexIter) writeRowFromTuples(ctx context.Context, key, value val.Tuple, r sql.Row) (err error) {
 	for to := range p.keyMap {
 		from := p.keyMap.MapOrdinal(to)
 		if from == -1 {
 			continue
 		}
-		r[to], err = GetField(p.keyDesc, from, key)
+		r[to], err = GetField(ctx, p.keyDesc, from, key, p.ns)
 		if err != nil {
 			return err
 		}
@@ -288,7 +286,7 @@ func (p prollyCoveringIndexIter) writeRowFromTuples(key, value val.Tuple, r sql.
 		if from == -1 {
 			continue
 		}
-		r[to], err = GetField(p.valDesc, from, value)
+		r[to], err = GetField(ctx, p.valDesc, from, value, p.ns)
 		if err != nil {
 			return err
 		}
@@ -384,13 +382,7 @@ type prollyKeylessIndexIter struct {
 var _ sql.RowIter = prollyKeylessIndexIter{}
 var _ sql.RowIter2 = prollyKeylessIndexIter{}

-func newProllyKeylessIndexIter(
-	ctx *sql.Context,
-	idx DoltIndex,
-	rng prolly.Range,
-	pkSch sql.PrimaryKeySchema,
-	rows, dsecondary durable.Index,
-) (prollyKeylessIndexIter, error) {
+func newProllyKeylessIndexIter(ctx *sql.Context, idx DoltIndex, rng prolly.Range, pkSch sql.PrimaryKeySchema, rows, dsecondary durable.Index) (prollyKeylessIndexIter, error) {
 	secondary := durable.ProllyMapFromIndex(dsecondary)
 	indexIter, err := secondary.IterRange(ctx, rng)
 	if err != nil {
@@ -468,7 +460,7 @@ func (p prollyKeylessIndexIter) queueRows(ctx context.Context) error {
 			return err
 		}

-		rows, err := p.keylessRowsFromValueTuple(value)
+		rows, err := p.keylessRowsFromValueTuple(ctx, p.clustered.NodeStore(), value)
 		if err != nil {
 			return err
 		}
@@ -483,7 +475,7 @@ func (p prollyKeylessIndexIter) queueRows(ctx context.Context) error {
 	}
 }

-func (p prollyKeylessIndexIter) keylessRowsFromValueTuple(value val.Tuple) (rows []sql.Row, err error) {
+func (p prollyKeylessIndexIter) keylessRowsFromValueTuple(ctx context.Context, ns tree.NodeStore, value val.Tuple) (rows []sql.Row, err error) {
 	card := val.ReadKeylessCardinality(value)
 	rows = make([]sql.Row, card)
 	rows[0] = make(sql.Row, len(p.valueMap)-1) // omit cardinality field
@@ -492,7 +484,7 @@ func (p prollyKeylessIndexIter) keylessRowsFromValueTuple(value val.Tuple) (rows
 		if rowIdx == -1 {
 			continue
 		}
-		rows[0][rowIdx], err = GetField(p.valueDesc, valIdx, value)
+		rows[0][rowIdx], err = GetField(ctx, p.valueDesc, valIdx, value, ns)
 		if err != nil {
 			return nil, err
 		}
--- a/go/libraries/doltcore/sqle/index/prolly_row_iter.go
+++ b/go/libraries/doltcore/sqle/index/prolly_row_iter.go
@@ -22,6 +22,7 @@ import (

 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/store/prolly"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )

@@ -50,6 +51,7 @@ var encodingToType [256]query.Type

 type prollyRowIter struct {
 	iter prolly.MapIter
+	ns   tree.NodeStore

 	sqlSch  sql.Schema
 	keyDesc val.TupleDesc
@@ -62,13 +64,7 @@ type prollyRowIter struct {
 var _ sql.RowIter = prollyRowIter{}
 var _ sql.RowIter2 = prollyRowIter{}

-func NewProllyRowIter(
-	sch schema.Schema,
-	schSch sql.Schema,
-	rows prolly.Map,
-	iter prolly.MapIter,
-	projections []string,
-) (sql.RowIter, error) {
+func NewProllyRowIter(sch schema.Schema, schSch sql.Schema, rows prolly.Map, iter prolly.MapIter, projections []string) (sql.RowIter, error) {

 	// todo(andy): NomsRangeReader seemingly ignores projections
 	//if projections == nil {
@@ -86,6 +82,7 @@ func NewProllyRowIter(
 			valDesc: vd,
 			valProj: valProj,
 			rowLen:  len(projections),
+			ns:      rows.NodeStore(),
 		}, nil
 	}

@@ -97,6 +94,7 @@ func NewProllyRowIter(
 		keyProj: keyProj,
 		valProj: valProj,
 		rowLen:  len(projections),
+		ns:      rows.NodeStore(),
 	}, nil
 }

@@ -146,7 +144,7 @@ func (it prollyRowIter) Next(ctx *sql.Context) (sql.Row, error) {
 		if rowIdx == -1 {
 			continue
 		}
-		row[rowIdx], err = GetField(it.keyDesc, keyIdx, key)
+		row[rowIdx], err = GetField(ctx, it.keyDesc, keyIdx, key, it.ns)
 		if err != nil {
 			return nil, err
 		}
@@ -155,7 +153,7 @@ func (it prollyRowIter) Next(ctx *sql.Context) (sql.Row, error) {
 		if rowIdx == -1 {
 			continue
 		}
-		row[rowIdx], err = GetField(it.valDesc, valIdx, value)
+		row[rowIdx], err = GetField(ctx, it.valDesc, valIdx, value, it.ns)
 		if err != nil {
 			return nil, err
 		}
@@ -205,6 +203,7 @@ func (it prollyRowIter) Close(ctx *sql.Context) error {

 type prollyKeylessIter struct {
 	iter prolly.MapIter
+	ns   tree.NodeStore

 	valDesc val.TupleDesc
 	valProj []int
@@ -243,7 +242,7 @@ func (it *prollyKeylessIter) nextTuple(ctx *sql.Context) error {
 		if rowIdx == -1 {
 			continue
 		}
-		it.curr[rowIdx], err = GetField(it.valDesc, valIdx, value)
+		it.curr[rowIdx], err = GetField(ctx, it.valDesc, valIdx, value, it.ns)
 		if err != nil {
 			return err
 		}
--- a/go/libraries/doltcore/sqle/sqlutil/sql_row.go
+++ b/go/libraries/doltcore/sqle/sqlutil/sql_row.go
@@ -319,6 +319,7 @@ func SqlColToStr(ctx context.Context, sqlType sql.Type, col interface{}) (string
 				return "", err
 			}
 			return res.ToString(), nil
+
 		}
 	}

--- a/go/libraries/doltcore/sqle/writer/prolly_fk_indexer.go
+++ b/go/libraries/doltcore/sqle/writer/prolly_fk_indexer.go
@@ -128,13 +128,13 @@ func (iter prollyFkPkRowIter) Next(ctx *sql.Context) (sql.Row, error) {
 	err = iter.primary.mut.Get(ctx, pkTup, func(tblKey, tblVal val.Tuple) error {
 		for from := range iter.primary.keyMap {
 			to := iter.primary.keyMap.MapOrdinal(from)
-			if nextRow[to], err = index.GetField(iter.primary.keyBld.Desc, from, tblKey); err != nil {
+			if nextRow[to], err = index.GetField(ctx, iter.primary.keyBld.Desc, from, tblKey, iter.primary.mut.NodeStore()); err != nil {
 				return err
 			}
 		}
 		for from := range iter.primary.valMap {
 			to := iter.primary.valMap.MapOrdinal(from)
-			if nextRow[to], err = index.GetField(iter.primary.valBld.Desc, from, tblVal); err != nil {
+			if nextRow[to], err = index.GetField(ctx, iter.primary.valBld.Desc, from, tblVal, iter.primary.mut.NodeStore()); err != nil {
 				return err
 			}
 		}
@@ -177,7 +177,7 @@ func (iter prollyFkKeylessRowIter) Next(ctx *sql.Context) (sql.Row, error) {
 	err = iter.primary.mut.Get(ctx, primaryKey, func(tblKey, tblVal val.Tuple) error {
 		for from := range iter.primary.valMap {
 			to := iter.primary.valMap.MapOrdinal(from)
-			if nextRow[to], err = index.GetField(iter.primary.valBld.Desc, from+1, tblVal); err != nil {
+			if nextRow[to], err = index.GetField(ctx, iter.primary.valBld.Desc, from+1, tblVal, iter.primary.mut.NodeStore()); err != nil {
 				return err
 			}
 		}
--- a/go/libraries/doltcore/sqle/writer/prolly_index_writer.go
+++ b/go/libraries/doltcore/sqle/writer/prolly_index_writer.go
@@ -106,7 +106,7 @@ func (m prollyIndexWriter) Map(ctx context.Context) (prolly.Map, error) {
 func (m prollyIndexWriter) Insert(ctx context.Context, sqlRow sql.Row) error {
 	for to := range m.keyMap {
 		from := m.keyMap.MapOrdinal(to)
-		if err := index.PutField(m.keyBld, to, sqlRow[from]); err != nil {
+		if err := index.PutField(ctx, m.mut.NodeStore(), m.keyBld, to, sqlRow[from]); err != nil {
 			return err
 		}
 	}
@@ -127,7 +127,7 @@ func (m prollyIndexWriter) Insert(ctx context.Context, sqlRow sql.Row) error {

 	for to := range m.valMap {
 		from := m.valMap.MapOrdinal(to)
-		if err = index.PutField(m.valBld, to, sqlRow[from]); err != nil {
+		if err = index.PutField(ctx, m.mut.NodeStore(), m.valBld, to, sqlRow[from]); err != nil {
 			return err
 		}
 	}
@@ -139,7 +139,7 @@ func (m prollyIndexWriter) Insert(ctx context.Context, sqlRow sql.Row) error {
 func (m prollyIndexWriter) Delete(ctx context.Context, sqlRow sql.Row) error {
 	for to := range m.keyMap {
 		from := m.keyMap.MapOrdinal(to)
-		if err := index.PutField(m.keyBld, to, sqlRow[from]); err != nil {
+		if err := index.PutField(ctx, m.mut.NodeStore(), m.keyBld, to, sqlRow[from]); err != nil {
 			return err
 		}
 	}
@@ -151,7 +151,7 @@ func (m prollyIndexWriter) Delete(ctx context.Context, sqlRow sql.Row) error {
 func (m prollyIndexWriter) Update(ctx context.Context, oldRow sql.Row, newRow sql.Row) error {
 	for to := range m.keyMap {
 		from := m.keyMap.MapOrdinal(to)
-		if err := index.PutField(m.keyBld, to, oldRow[from]); err != nil {
+		if err := index.PutField(ctx, m.mut.NodeStore(), m.keyBld, to, oldRow[from]); err != nil {
 			return err
 		}
 	}
@@ -165,7 +165,7 @@ func (m prollyIndexWriter) Update(ctx context.Context, oldRow sql.Row, newRow sq

 	for to := range m.keyMap {
 		from := m.keyMap.MapOrdinal(to)
-		if err := index.PutField(m.keyBld, to, newRow[from]); err != nil {
+		if err := index.PutField(ctx, m.mut.NodeStore(), m.keyBld, to, newRow[from]); err != nil {
 			return err
 		}
 	}
@@ -186,7 +186,7 @@ func (m prollyIndexWriter) Update(ctx context.Context, oldRow sql.Row, newRow sq

 	for to := range m.valMap {
 		from := m.valMap.MapOrdinal(to)
-		if err = index.PutField(m.valBld, to, newRow[from]); err != nil {
+		if err = index.PutField(ctx, m.mut.NodeStore(), m.valBld, to, newRow[from]); err != nil {
 			return err
 		}
 	}
@@ -211,7 +211,7 @@ func (m prollyIndexWriter) HasEdits(ctx context.Context) bool {
 func (m prollyIndexWriter) UniqueKeyError(ctx context.Context, sqlRow sql.Row) error {
 	for to := range m.keyMap {
 		from := m.keyMap.MapOrdinal(to)
-		if err := index.PutField(m.keyBld, to, sqlRow[from]); err != nil {
+		if err := index.PutField(ctx, m.mut.NodeStore(), m.keyBld, to, sqlRow[from]); err != nil {
 			return err
 		}
 	}
@@ -226,7 +226,7 @@ func (m prollyIndexWriter) keyError(ctx context.Context, key val.Tuple, isPk boo
 		kd := m.keyBld.Desc
 		for from := range m.keyMap {
 			to := m.keyMap.MapOrdinal(from)
-			if dupe[to], err = index.GetField(kd, from, key); err != nil {
+			if dupe[to], err = index.GetField(ctx, kd, from, key, m.mut.NodeStore()); err != nil {
 				return err
 			}
 		}
@@ -234,7 +234,7 @@ func (m prollyIndexWriter) keyError(ctx context.Context, key val.Tuple, isPk boo
 		vd := m.valBld.Desc
 		for from := range m.valMap {
 			to := m.valMap.MapOrdinal(from)
-			if dupe[to], err = index.GetField(vd, from, value); err != nil {
+			if dupe[to], err = index.GetField(ctx, vd, from, value, m.mut.NodeStore()); err != nil {
 				return err
 			}
 		}
@@ -276,7 +276,7 @@ func (k prollyKeylessWriter) Map(ctx context.Context) (prolly.Map, error) {
 }

 func (k prollyKeylessWriter) Insert(ctx context.Context, sqlRow sql.Row) error {
-	hashId, value, err := k.tuplesFromRow(sqlRow)
+	hashId, value, err := k.tuplesFromRow(ctx, sqlRow)
 	if err != nil {
 		return err
 	}
@@ -298,7 +298,7 @@ func (k prollyKeylessWriter) Insert(ctx context.Context, sqlRow sql.Row) error {
 }

 func (k prollyKeylessWriter) Delete(ctx context.Context, sqlRow sql.Row) error {
-	hashId, _, err := k.tuplesFromRow(sqlRow)
+	hashId, _, err := k.tuplesFromRow(ctx, sqlRow)
 	if err != nil {
 		return err
 	}
@@ -355,15 +355,15 @@ func (k prollyKeylessWriter) UniqueKeyError(ctx context.Context, sqlRow sql.Row)
 	return fmt.Errorf("keyless does not yet know how to handle unique key errors")
 }

-func (k prollyKeylessWriter) tuplesFromRow(sqlRow sql.Row) (hashId, value val.Tuple, err error) {
+func (k prollyKeylessWriter) tuplesFromRow(ctx context.Context, sqlRow sql.Row) (hashId, value val.Tuple, err error) {
 	// initialize cardinality to 0
-	if err = index.PutField(k.valBld, 0, uint64(0)); err != nil {
+	if err = index.PutField(ctx, k.mut.NodeStore(), k.valBld, 0, uint64(0)); err != nil {
 		return nil, nil, err
 	}

 	for to := range k.valMap {
 		from := k.valMap.MapOrdinal(to)
-		if err = index.PutField(k.valBld, to+1, sqlRow[from]); err != nil {
+		if err = index.PutField(ctx, k.mut.NodeStore(), k.valBld, to+1, sqlRow[from]); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -402,11 +402,11 @@ func (writer prollyKeylessSecondaryWriter) Map(ctx context.Context) (prolly.Map,
 func (writer prollyKeylessSecondaryWriter) Insert(ctx context.Context, sqlRow sql.Row) error {
 	for to := range writer.keyMap {
 		from := writer.keyMap.MapOrdinal(to)
-		if err := index.PutField(writer.keyBld, to, sqlRow[from]); err != nil {
+		if err := index.PutField(ctx, writer.mut.NodeStore(), writer.keyBld, to, sqlRow[from]); err != nil {
 			return err
 		}
 	}
-	hashId, _, err := writer.primary.tuplesFromRow(sqlRow)
+	hashId, _, err := writer.primary.tuplesFromRow(ctx, sqlRow)
 	if err != nil {
 		return err
 	}
@@ -428,7 +428,7 @@ func (writer prollyKeylessSecondaryWriter) Insert(ctx context.Context, sqlRow sq

 // Delete implements the interface indexWriter.
 func (writer prollyKeylessSecondaryWriter) Delete(ctx context.Context, sqlRow sql.Row) error {
-	hashId, cardRow, err := writer.primary.tuplesFromRow(sqlRow)
+	hashId, cardRow, err := writer.primary.tuplesFromRow(ctx, sqlRow)
 	if err != nil {
 		return err
 	}
@@ -444,7 +444,7 @@ func (writer prollyKeylessSecondaryWriter) Delete(ctx context.Context, sqlRow sq

 	for to := range writer.keyMap {
 		from := writer.keyMap.MapOrdinal(to)
-		if err := index.PutField(writer.keyBld, to, sqlRow[from]); err != nil {
+		if err := index.PutField(ctx, writer.mut.NodeStore(), writer.keyBld, to, sqlRow[from]); err != nil {
 			return err
 		}
 	}
--- a/go/serial/encoding.fbs
+++ b/go/serial/encoding.fbs
@@ -16,26 +16,27 @@ namespace serial;

 enum Encoding : uint8 {
  // fixed width
-  Null     = 0,
-  Int8     = 1,
-  Uint8    = 2,
-  Int16    = 3,
-  Uint16   = 4,
-  Int32    = 7,
-  Uint32   = 8,
-  Int64    = 9,
-  Uint64   = 10,
-  Float32  = 11,
-  Float64  = 12,
-  Bit64    = 13,
-  Hash128  = 14,
-  Year     = 15,
-  Date     = 16,
-  Time     = 17,
-  Datetime = 18,
-  Enum     = 19,
-  Set      = 20,
-  Address  = 21,
+  Null       = 0,
+  Int8       = 1,
+  Uint8      = 2,
+  Int16      = 3,
+  Uint16     = 4,
+  Int32      = 7,
+  Uint32     = 8,
+  Int64      = 9,
+  Uint64     = 10,
+  Float32    = 11,
+  Float64    = 12,
+  Bit64      = 13,
+  Hash128    = 14,
+  Year       = 15,
+  Date       = 16,
+  Time       = 17,
+  Datetime   = 18,
+  Enum       = 19,
+  Set        = 20,
+  BytesAddr = 21,
+  CommitAddr  = 22,

  // variable width
  String   = 128,
--- a/go/store/datas/commit.go
+++ b/go/store/datas/commit.go
@@ -179,7 +179,7 @@ func commit_flatbuffer(vaddr hash.Hash, opts CommitOptions, heights []uint64, pa

 var commitKeyTupleDesc = val.NewTupleDescriptor(
 	val.Type{Enc: val.Uint64Enc, Nullable: false},
-	val.Type{Enc: val.AddressEnc, Nullable: false},
+	val.Type{Enc: val.CommitAddrEnc, Nullable: false},
 )
 var commitValueTupleDesc = val.NewTupleDescriptor()

--- a/go/store/prolly/artifact_map.go
+++ b/go/store/prolly/artifact_map.go
@@ -110,6 +110,10 @@ func (m ArtifactMap) Node() tree.Node {
 	return m.tuples.root
 }

+func (m ArtifactMap) NodeStore() tree.NodeStore {
+	return m.tuples.ns
+}
+
 func (m ArtifactMap) Format() *types.NomsBinFormat {
 	return m.tuples.ns.Format()
 }
@@ -142,7 +146,7 @@ func (m ArtifactMap) Editor() ArtifactsEditor {
 	artKD, artVD := m.Descriptors()
 	return ArtifactsEditor{
 		srcKeyDesc: m.srcKeyDesc,
-		mut: MutableMap{
+		Mut: MutableMap{
 			tuples:  m.tuples.mutate(),
 			keyDesc: m.keyDesc,
 			valDesc: m.valDesc,
@@ -280,7 +284,7 @@ func (m ArtifactMap) iterAllOfTypes(ctx context.Context, artTypes ...ArtifactTyp

 func MergeArtifactMaps(ctx context.Context, left, right, base ArtifactMap, cb tree.CollisionFn) (ArtifactMap, error) {
 	serializer := message.ProllyMapSerializer{Pool: left.tuples.ns.Pool()}
-	tuples, err := mergeOrderedTrees(ctx, left.tuples, right.tuples, base.tuples, cb, serializer)
+	tuples, err := mergeOrderedTrees(ctx, left.tuples, right.tuples, base.tuples, cb, serializer, base.valDesc)
 	if err != nil {
 		return ArtifactMap{}, err
 	}
@@ -293,7 +297,7 @@ func MergeArtifactMaps(ctx context.Context, left, right, base ArtifactMap, cb tr
 }

 type ArtifactsEditor struct {
-	mut          MutableMap
+	Mut          MutableMap
 	srcKeyDesc   val.TupleDesc
 	artKB, artVB *val.TupleBuilder
 	pool         pool.BuffPool
@@ -303,14 +307,14 @@ func (wr ArtifactsEditor) Add(ctx context.Context, srcKey val.Tuple, theirRootIs
 	for i := 0; i < srcKey.Count(); i++ {
 		wr.artKB.PutRaw(i, srcKey.GetField(i))
 	}
-	wr.artKB.PutAddress(srcKey.Count(), theirRootIsh)
+	wr.artKB.PutCommitAddr(srcKey.Count(), theirRootIsh)
 	wr.artKB.PutUint8(srcKey.Count()+1, uint8(artType))
 	key := wr.artKB.Build(wr.pool)

 	wr.artVB.PutJSON(0, meta)
 	value := wr.artVB.Build(wr.pool)

-	return wr.mut.Put(ctx, key, value)
+	return wr.Mut.Put(ctx, key, value)
 }

 // ReplaceFKConstraintViolation replaces foreign key constraint violations that
@@ -318,14 +322,14 @@ func (wr ArtifactsEditor) Add(ctx context.Context, srcKey val.Tuple, theirRootIs
 // exists, the given will be inserted.
 func (wr ArtifactsEditor) ReplaceFKConstraintViolation(ctx context.Context, srcKey val.Tuple, theirRootIsh hash.Hash, meta ConstraintViolationMeta) error {
 	rng := ClosedRange(srcKey, srcKey, wr.srcKeyDesc)
-	itr, err := wr.mut.IterRange(ctx, rng)
+	itr, err := wr.Mut.IterRange(ctx, rng)
 	if err != nil {
 		return err
 	}
 	aItr := artifactIterImpl{
 		itr:    itr,
-		artKD:  wr.mut.keyDesc,
-		artVD:  wr.mut.valDesc,
+		artKD:  wr.Mut.keyDesc,
+		artVD:  wr.Mut.valDesc,
 		pool:   wr.pool,
 		tb:     val.NewTupleBuilder(wr.srcKeyDesc),
 		numPks: wr.srcKeyDesc.Count(),
@@ -368,11 +372,11 @@ func (wr ArtifactsEditor) ReplaceFKConstraintViolation(ctx context.Context, srcK
 }

 func (wr ArtifactsEditor) Delete(ctx context.Context, key val.Tuple) error {
-	return wr.mut.Delete(ctx, key)
+	return wr.Mut.Delete(ctx, key)
 }

 func (wr ArtifactsEditor) Flush(ctx context.Context) (ArtifactMap, error) {
-	m, err := wr.mut.Map(ctx)
+	m, err := wr.Mut.Map(ctx)
 	if err != nil {
 		return ArtifactMap{}, err
 	}
@@ -380,8 +384,8 @@ func (wr ArtifactsEditor) Flush(ctx context.Context) (ArtifactMap, error) {
 	return ArtifactMap{
 		tuples:     m.tuples,
 		srcKeyDesc: wr.srcKeyDesc,
-		keyDesc:    wr.mut.keyDesc,
-		valDesc:    wr.mut.valDesc,
+		keyDesc:    wr.Mut.keyDesc,
+		valDesc:    wr.Mut.valDesc,
 	}, nil
 }

@@ -498,7 +502,7 @@ func (itr artifactIterImpl) Next(ctx context.Context) (Artifact, error) {
 	}

 	srcKey := itr.getSrcKeyFromArtKey(artKey)
-	cmHash, _ := itr.artKD.GetAddress(itr.numPks, artKey)
+	cmHash, _ := itr.artKD.GetCommitAddr(itr.numPks, artKey)
 	artType, _ := itr.artKD.GetUint8(itr.numPks+1, artKey)
 	metadata, _ := itr.artVD.GetJSON(0, v)

@@ -539,7 +543,7 @@ func calcArtifactsDescriptors(srcKd val.TupleDesc) (kd, vd val.TupleDesc) {
 	keyTypes := srcKd.Types

 	// target branch commit hash
-	keyTypes = append(keyTypes, val.Type{Enc: val.AddressEnc, Nullable: false})
+	keyTypes = append(keyTypes, val.Type{Enc: val.CommitAddrEnc, Nullable: false})

 	// artifact type
 	keyTypes = append(keyTypes, val.Type{Enc: val.Uint8Enc, Nullable: false})
--- a/go/store/prolly/map.go
+++ b/go/store/prolly/map.go
@@ -83,7 +83,7 @@ func DiffMaps(ctx context.Context, from, to Map, cb DiffFn) error {

 func MergeMaps(ctx context.Context, left, right, base Map, cb tree.CollisionFn) (Map, error) {
 	serializer := message.ProllyMapSerializer{Pool: left.tuples.ns.Pool()}
-	tuples, err := mergeOrderedTrees(ctx, left.tuples, right.tuples, base.tuples, cb, serializer)
+	tuples, err := mergeOrderedTrees(ctx, left.tuples, right.tuples, base.tuples, cb, serializer, base.valDesc)
 	if err != nil {
 		return Map{}, err
 	}
@@ -95,6 +95,11 @@ func MergeMaps(ctx context.Context, left, right, base Map, cb tree.CollisionFn)
 	}, nil
 }

+// NodeStore returns the map's NodeStore
+func (m Map) NodeStore() tree.NodeStore {
+	return m.tuples.ns
+}
+
 // Mutate makes a MutableMap from a Map.
 func (m Map) Mutate() MutableMap {
 	return newMutableMap(m)
--- a/go/store/prolly/map_diff_test.go
+++ b/go/store/prolly/map_diff_test.go
@@ -247,7 +247,8 @@ func makeMapWithInserts(t *testing.T, m Map, numInserts int) (Map, [][2]val.Tupl
 // generates tuple pairs not currently in |m|
 func generateInserts(t *testing.T, m Map, kd, vd val.TupleDesc, numInserts int) [][2]val.Tuple {
 	ctx := context.Background()
-	tups := tree.RandomTuplePairs(numInserts*2, kd, vd)
+	ns := tree.NewTestNodeStore()
+	tups := tree.RandomTuplePairs(numInserts*2, kd, vd, ns)
 	inserts, extra := tups[:numInserts], tups[numInserts:]

 	j := 0
@@ -293,10 +294,12 @@ func makeUpdatesToTuples(kd, vd val.TupleDesc, tuples ...[2]val.Tuple) (updates
 	updates = make([][3]val.Tuple, len(tuples))

 	valBuilder := val.NewTupleBuilder(vd)
+	ns := tree.NewTestNodeStore()
+
 	for i := range updates {
 		updates[i][0] = tuples[i][0]
 		updates[i][1] = tuples[i][1]
-		updates[i][2] = tree.RandomTuple(valBuilder)
+		updates[i][2] = tree.RandomTuple(valBuilder, ns)
 	}

 	sort.Slice(updates, func(i, j int) bool {
--- a/go/store/prolly/map_merge_test.go
+++ b/go/store/prolly/map_merge_test.go
@@ -42,6 +42,7 @@ func Test3WayMapMerge(t *testing.T) {
 		val.Type{Enc: val.Uint32Enc, Nullable: true},
 		val.Type{Enc: val.Uint32Enc, Nullable: true},
 	)
+	ns := tree.NewTestNodeStore()

 	for _, s := range scales {
 		name := fmt.Sprintf("test proCur map at scale %d", s)
@@ -51,12 +52,12 @@ func Test3WayMapMerge(t *testing.T) {
 			})
 			t.Run("3way merge inserts", func(t *testing.T) {
 				for k := 0; k < 10; k++ {
-					testThreeWayMapMerge(t, kd, vd, s)
+					testThreeWayMapMerge(t, kd, vd, s, ns)
 				}
 			})
 			t.Run("tuple merge fn", func(t *testing.T) {
 				for k := 0; k < 10; k++ {
-					testTupleMergeFn(t, kd, vd, s)
+					testTupleMergeFn(t, kd, vd, s, ns)
 				}
 			})
 		})
@@ -73,8 +74,8 @@ func testEqualMapMerge(t *testing.T, sz int) {
 	assert.Equal(t, m.HashOf(), mm.HashOf())
 }

-func testThreeWayMapMerge(t *testing.T, kd, vd val.TupleDesc, sz int) {
-	baseTuples, leftEdits, rightEdits := makeTuplesAndMutations(kd, vd, sz)
+func testThreeWayMapMerge(t *testing.T, kd, vd val.TupleDesc, sz int, ns tree.NodeStore) {
+	baseTuples, leftEdits, rightEdits := makeTuplesAndMutations(kd, vd, sz, ns)
 	om := prollyMapFromTuples(t, kd, vd, baseTuples)

 	base := om.(Map)
@@ -139,9 +140,9 @@ func testThreeWayMapMerge(t *testing.T, kd, vd val.TupleDesc, sz int) {
 	}
 }

-func testTupleMergeFn(t *testing.T, kd, vd val.TupleDesc, sz int) {
+func testTupleMergeFn(t *testing.T, kd, vd val.TupleDesc, sz int, ns tree.NodeStore) {
 	ctx := context.Background()
-	tuples := tree.RandomTuplePairs(sz, kd, vd)
+	tuples := tree.RandomTuplePairs(sz, kd, vd, ns)
 	om := prollyMapFromTuples(t, kd, vd, tuples)
 	base := om.(Map)

@@ -210,10 +211,10 @@ type mutationSet struct {
 	updates [][3]val.Tuple
 }

-func makeTuplesAndMutations(kd, vd val.TupleDesc, sz int) (base [][2]val.Tuple, left, right mutationSet) {
+func makeTuplesAndMutations(kd, vd val.TupleDesc, sz int, ns tree.NodeStore) (base [][2]val.Tuple, left, right mutationSet) {
 	mutSz := sz / 10
 	totalSz := sz + (mutSz * 2)
-	tuples := tree.RandomTuplePairs(totalSz, kd, vd)
+	tuples := tree.RandomTuplePairs(totalSz, kd, vd, ns)

 	base = tuples[:sz]

--- a/go/store/prolly/map_test.go
+++ b/go/store/prolly/map_test.go
@@ -123,8 +123,9 @@ func makeProllyMap(t *testing.T, count int) (testMap, [][2]val.Tuple) {
 		val.Type{Enc: val.Uint32Enc, Nullable: true},
 		val.Type{Enc: val.Uint32Enc, Nullable: true},
 	)
+	ns := tree.NewTestNodeStore()

-	tuples := tree.RandomTuplePairs(count, kd, vd)
+	tuples := tree.RandomTuplePairs(count, kd, vd, ns)
 	om := prollyMapFromTuples(t, kd, vd, tuples)

 	return om, tuples
@@ -136,8 +137,8 @@ func makeProllySecondaryIndex(t *testing.T, count int) (testMap, [][2]val.Tuple)
 		val.Type{Enc: val.Uint32Enc, Nullable: false},
 	)
 	vd := val.NewTupleDescriptor()
-
-	tuples := tree.RandomCompositeTuplePairs(count, kd, vd)
+	ns := tree.NewTestNodeStore()
+	tuples := tree.RandomCompositeTuplePairs(count, kd, vd, ns)
 	om := prollyMapFromTuples(t, kd, vd, tuples)

 	return om, tuples
--- a/go/store/prolly/message/message.go
+++ b/go/store/prolly/message/message.go
@@ -18,9 +18,10 @@ import (
 	"context"
 	"fmt"

+	"github.com/dolthub/dolt/go/store/val"
+
 	"github.com/dolthub/dolt/go/gen/fb/serial"
 	"github.com/dolthub/dolt/go/store/hash"
-	"github.com/dolthub/dolt/go/store/val"
 )

 const MessageTypesKind int = 28
--- a/go/store/prolly/message/prolly_map.go
+++ b/go/store/prolly/message/prolly_map.go
@@ -38,7 +38,8 @@ const (
 var prollyMapFileID = []byte(serial.ProllyTreeNodeFileID)

 type ProllyMapSerializer struct {
-	Pool pool.BuffPool
+	Pool    pool.BuffPool
+	ValDesc val.TupleDesc
 }

 var _ Serializer = ProllyMapSerializer{}
@@ -47,10 +48,11 @@ func (s ProllyMapSerializer) Serialize(keys, values [][]byte, subtrees []uint64,
 	var (
 		keyTups, keyOffs fb.UOffsetT
 		valTups, valOffs fb.UOffsetT
+		valAddrOffs      fb.UOffsetT
 		refArr, cardArr  fb.UOffsetT
 	)

-	keySz, valSz, bufSz := estimateProllyMapSize(keys, values, subtrees)
+	keySz, valSz, bufSz := estimateProllyMapSize(keys, values, subtrees, len(s.ValDesc.Addrs))
 	b := getFlatbufferBuilder(s.Pool, bufSz)

 	// serialize keys and offsets
@@ -63,6 +65,10 @@ func (s ProllyMapSerializer) Serialize(keys, values [][]byte, subtrees []uint64,
 		valTups = writeItemBytes(b, values, valSz)
 		serial.ProllyTreeNodeStartValueOffsetsVector(b, len(values)-1)
 		valOffs = writeItemOffsets(b, values, valSz)
+		if len(s.ValDesc.Addrs) > 0 {
+			serial.ProllyTreeNodeStartValueAddressOffsetsVector(b, len(values)*len(s.ValDesc.Addrs))
+			valAddrOffs = writeValAddrOffsets(b, values, valSz, s.ValDesc)
+		}
 	} else {
 		// serialize child refs and subtree counts for internal nodes
 		refArr = writeItemBytes(b, values, valSz)
@@ -77,6 +83,7 @@ func (s ProllyMapSerializer) Serialize(keys, values [][]byte, subtrees []uint64,
 		serial.ProllyTreeNodeAddValueItems(b, valTups)
 		serial.ProllyTreeNodeAddValueOffsets(b, valOffs)
 		serial.ProllyTreeNodeAddTreeCount(b, uint64(len(keys)))
+		serial.ProllyTreeNodeAddValueAddressOffsets(b, valAddrOffs)
 	} else {
 		serial.ProllyTreeNodeAddAddressArray(b, refArr)
 		serial.ProllyTreeNodeAddSubtreeCounts(b, cardArr)
@@ -126,7 +133,7 @@ func walkProllyMapAddresses(ctx context.Context, msg Message, cb func(ctx contex
 	arr2 := pm.ValueItemsBytes()
 	for i := 0; i < cnt; i++ {
 		o := pm.ValueAddressOffsets(i)
-		addr := hash.New(arr[o : o+addrSize])
+		addr := hash.New(arr2[o : o+addrSize])
 		if err := cb(ctx, addr); err != nil {
 			return err
 		}
@@ -182,7 +189,8 @@ func getProllyMapValueOffsets(pm *serial.ProllyTreeNode) []byte {

 // estimateProllyMapSize returns the exact Size of the tuple vectors for keys and values,
 // and an estimate of the overall Size of the final flatbuffer.
-func estimateProllyMapSize(keys, values [][]byte, subtrees []uint64) (keySz, valSz, bufSz int) {
+func estimateProllyMapSize(keys, values [][]byte, subtrees []uint64, valAddrsCnt int) (int, int, int) {
+	var keySz, valSz, bufSz int
 	for i := range keys {
 		keySz += len(keys[i])
 		valSz += len(values[i])
@@ -204,7 +212,8 @@ func estimateProllyMapSize(keys, values [][]byte, subtrees []uint64) (keySz, val
 	bufSz += 8 + 1 + 1 + 1               // metadata
 	bufSz += 72                          // vtable (approx)
 	bufSz += 100                         // padding?
+	bufSz += valAddrsCnt * len(values)
 	bufSz += messagePrefixSz

-	return
+	return keySz, valSz, bufSz
 }
--- a/go/store/prolly/message/serialize.go
+++ b/go/store/prolly/message/serialize.go
@@ -19,7 +19,9 @@ import (

 	fb "github.com/google/flatbuffers/go"

+	"github.com/dolthub/dolt/go/store/hash"
 	"github.com/dolthub/dolt/go/store/pool"
+	"github.com/dolthub/dolt/go/store/val"
 )

 const (
@@ -58,6 +60,29 @@ func writeItemOffsets(b *fb.Builder, items [][]byte, sumSz int) fb.UOffsetT {
 	return b.EndVector(cnt)
 }

+// writeAddrOffests returns offsets into the values array that correspond to addr root
+// hashes that themselves have subtrees
+func writeValAddrOffsets(b *fb.Builder, items [][]byte, sumSz int, valDesc val.TupleDesc) fb.UOffsetT {
+	var cnt int
+	var off = sumSz
+	for i := len(items) - 1; i >= 0; i-- {
+		tup := val.Tuple(items[i])
+		off -= len(tup) // start of tuple
+		for _, j := range valDesc.Addrs {
+			// get index into value tuple pointing at address
+			o, _ := tup.GetOffset(j)
+			a := tup.GetField(j)
+			if len(a) == 0 || hash.New(a).IsEmpty() {
+				continue
+			}
+			o += off // offset is tuple start plus field start
+			b.PrependUint16(uint16(o))
+			cnt++
+		}
+	}
+	return b.EndVector(cnt)
+}
+
 func writeCountArray(b *fb.Builder, counts []uint64) fb.UOffsetT {
 	// todo(andy): encode without alloc
 	buf := make([]byte, maxEncodedSize(len(counts)))
--- a/go/store/prolly/mutable_map.go
+++ b/go/store/prolly/mutable_map.go
@@ -51,7 +51,7 @@ func (mut MutableMap) Map(ctx context.Context) (Map, error) {
 		return Map{}, err
 	}
 	tr := mut.tuples.tree
-	serializer := message.ProllyMapSerializer{Pool: tr.ns.Pool()}
+	serializer := message.ProllyMapSerializer{Pool: tr.ns.Pool(), ValDesc: mut.valDesc}

 	root, err := tree.ApplyMutations(ctx, tr.ns, tr.root, serializer, mut.tuples.mutations(), tr.compareItems)
 	if err != nil {
@@ -69,6 +69,11 @@ func (mut MutableMap) Map(ctx context.Context) (Map, error) {
 	}, nil
 }

+// NodeStore returns the map's NodeStore
+func (mut MutableMap) NodeStore() tree.NodeStore {
+	return mut.tuples.tree.ns
+}
+
 // Put adds the Tuple pair |key|, |value| to the MutableMap.
 func (mut MutableMap) Put(ctx context.Context, key, value val.Tuple) error {
 	return mut.tuples.put(ctx, key, value)
--- a/go/store/prolly/mutable_map_read_test.go
+++ b/go/store/prolly/mutable_map_read_test.go
@@ -117,7 +117,7 @@ func makeMutableMap(t *testing.T, count int) (testMap, [][2]val.Tuple) {
 		val.Type{Enc: val.Uint32Enc, Nullable: true},
 	)

-	tuples := tree.RandomTuplePairs(count, kd, vd)
+	tuples := tree.RandomTuplePairs(count, kd, vd, ns)
 	// 2/3 of tuples in Map
 	// 1/3 of tuples in memoryMap
 	clone := tree.CloneRandomTuples(tuples)
--- a/go/store/prolly/ordered_tree.go
+++ b/go/store/prolly/ordered_tree.go
@@ -19,11 +19,11 @@ import (
 	"fmt"
 	"io"

-	"github.com/dolthub/dolt/go/store/prolly/message"
-
 	"github.com/dolthub/dolt/go/store/hash"
+	"github.com/dolthub/dolt/go/store/prolly/message"
 	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/skip"
+	"github.com/dolthub/dolt/go/store/val"
 )

 type KeyValueFn[K, V ~[]byte] func(key K, value V) error
@@ -71,9 +71,10 @@ func mergeOrderedTrees[K, V ~[]byte, O ordering[K], S message.Serializer](
 	l, r, base orderedTree[K, V, O],
 	cb tree.CollisionFn,
 	serializer S,
+	valDesc val.TupleDesc,
 ) (orderedTree[K, V, O], error) {
 	cfn := base.compareItems
-	root, err := tree.ThreeWayMerge(ctx, base.ns, l.root, r.root, base.root, cfn, cb, serializer)
+	root, err := tree.ThreeWayMerge(ctx, base.ns, l.root, r.root, base.root, cfn, cb, serializer, valDesc)
 	if err != nil {
 		return orderedTree[K, V, O]{}, err
 	}
--- a/go/store/prolly/tree/immutable_tree.go
+++ b/go/store/prolly/tree/immutable_tree.go
@@ -15,14 +15,17 @@
 package tree

 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"

+	"github.com/dolthub/dolt/go/store/hash"
 	"github.com/dolthub/dolt/go/store/prolly/message"
+	"github.com/dolthub/dolt/go/store/val"
 )

-const defaultFixedChunkLength = 4000
+const DefaultFixedChunkLength = 4000

 var ErrInvalidChunkSize = errors.New("invalid chunkSize; value must be > 1")

@@ -148,3 +151,83 @@ func _newLeaf(ctx context.Context, ns NodeStore, s message.Serializer, buf []byt
 		treeCount: 1,
 	}, nil
 }
+
+type ByteArray struct {
+	ImmutableTree
+}
+
+func NewByteArray(addr hash.Hash, ns NodeStore) *ByteArray {
+	return &ByteArray{ImmutableTree{Addr: addr, ns: ns}}
+}
+
+func (b *ByteArray) ToBytes(ctx context.Context) ([]byte, error) {
+	if b.buf == nil {
+		err := b.load(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return b.buf[:], nil
+}
+
+func (b *ByteArray) ToString(ctx context.Context) (string, error) {
+	if b.buf == nil {
+		err := b.load(ctx)
+		if err != nil {
+			return "", err
+		}
+	}
+	toShow := 128
+	if len(b.buf) < toShow {
+		toShow = len(b.buf)
+	}
+	return string(b.buf[:toShow]), nil
+}
+
+type ImmutableTree struct {
+	Addr hash.Hash
+	buf  []byte
+	ns   NodeStore
+}
+
+func NewImmutableTreeFromReader(ctx context.Context, r io.Reader, ns NodeStore, chunkSize int) (*ImmutableTree, error) {
+	s := message.ProllyMapSerializer{Pool: ns.Pool(), ValDesc: val.TupleDesc{}}
+	root, err := buildImmutableTree(ctx, r, ns, s, chunkSize)
+	if errors.Is(err, io.EOF) {
+		return &ImmutableTree{Addr: hash.Hash{}}, nil
+	} else if err != nil {
+		return nil, err
+	}
+	return &ImmutableTree{Addr: root.HashOf()}, nil
+}
+
+func (t *ImmutableTree) load(ctx context.Context) error {
+	if t.Addr.IsEmpty() {
+		t.buf = []byte{}
+		return nil
+	}
+	n, err := t.ns.Read(ctx, t.Addr)
+	if err != nil {
+		return err
+	}
+
+	WalkNodes(ctx, n, t.ns, func(ctx context.Context, n Node) error {
+		if n.IsLeaf() {
+			t.buf = append(t.buf, n.getValue(0)...)
+		}
+		return nil
+	})
+	return nil
+}
+
+func (t *ImmutableTree) next() (Node, error) {
+	panic("not implemented")
+}
+
+func (t *ImmutableTree) close() error {
+	panic("not implemented")
+}
+
+func (t *ImmutableTree) Read(buf bytes.Buffer) (int, error) {
+	panic("not implemented")
+}
--- a/go/store/prolly/tree/immutable_tree_test.go
+++ b/go/store/prolly/tree/immutable_tree_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/stretchr/testify/require"

 	"github.com/dolthub/dolt/go/store/prolly/message"
+	"github.com/dolthub/dolt/go/store/val"
 )

 func TestWriteImmutableTree(t *testing.T) {
@@ -196,3 +197,110 @@ func expectedUnfilled(size, chunk int) int {
 	}
 	return cnt
 }
+
+func TestImmutableTreeWalk(t *testing.T) {
+	tests := []struct {
+		blobLen   int
+		chunkSize int
+		keyCnt    int
+	}{
+		{
+			blobLen:   25,
+			chunkSize: 6,
+			keyCnt:    4,
+		},
+		{
+			blobLen:   25,
+			chunkSize: 5,
+			keyCnt:    4,
+		},
+		{
+			blobLen:   378,
+			chunkSize: 5,
+			keyCnt:    12,
+		},
+		{
+			blobLen:   5000,
+			chunkSize: 12,
+			keyCnt:    6,
+		},
+		{
+			blobLen:   1,
+			chunkSize: 12,
+			keyCnt:    6,
+		},
+		{
+			blobLen:   0,
+			chunkSize: 12,
+			keyCnt:    6,
+		},
+	}
+
+	ns := NewTestNodeStore()
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("inputSize=%d; chunkSize=%d; keyCnt=%d", tt.blobLen, tt.chunkSize, tt.keyCnt), func(t *testing.T) {
+			r := newTree(t, ns, tt.keyCnt, tt.blobLen, tt.chunkSize)
+			var cnt int
+			walkOpaqueNodes(context.Background(), r, ns, func(ctx context.Context, n Node) error {
+				cnt++
+				return nil
+			})
+			require.Equal(t, leafAddrCnt(tt.blobLen, tt.chunkSize)*tt.keyCnt+1, cnt)
+		})
+	}
+}
+
+func leafAddrCnt(size, chunk int) int {
+	if size == 0 {
+		return 0
+	}
+	l := 1
+	for size > chunk {
+		size = int(math.Ceil(float64(size) / float64(chunk)))
+		l += size
+	}
+	return l
+}
+
+func newTree(t *testing.T, ns NodeStore, keyCnt, blobLen, chunkSize int) Node {
+	ctx := context.Background()
+
+	keyDesc := val.NewTupleDescriptor(val.Type{Enc: val.Uint32Enc})
+	valDesc := val.NewTupleDescriptor(val.Type{Enc: val.BytesAddrEnc})
+
+	tuples := make([][2]val.Tuple, keyCnt)
+	keyBld := val.NewTupleBuilder(keyDesc)
+	valBld := val.NewTupleBuilder(valDesc)
+	for i := range tuples {
+		keyBld.PutUint32(0, uint32(i))
+		tuples[i][0] = keyBld.Build(sharedPool)
+
+		b := mustNewBlob(ctx, ns, blobLen, chunkSize)
+		valBld.PutBytesAddr(0, b.Addr)
+		tuples[i][1] = valBld.Build(sharedPool)
+	}
+
+	serializer := message.ProllyMapSerializer{Pool: ns.Pool(), ValDesc: valDesc}
+	chunker, err := newEmptyChunker(ctx, ns, serializer)
+	require.NoError(t, err)
+	for _, pair := range tuples {
+		err := chunker.AddPair(ctx, Item(pair[0]), Item(pair[1]))
+		require.NoError(t, err)
+	}
+	root, err := chunker.Done(ctx)
+	require.NoError(t, err)
+	return root
+}
+
+func mustNewBlob(ctx context.Context, ns NodeStore, len, chunkSize int) *ImmutableTree {
+	buf := make([]byte, len)
+	for i := range buf {
+		buf[i] = byte(i)
+	}
+	r := bytes.NewReader(buf)
+	root, err := NewImmutableTreeFromReader(ctx, r, ns, chunkSize)
+	if err != nil {
+		panic(err)
+	}
+	return root
+}
--- a/go/store/prolly/tree/merge.go
+++ b/go/store/prolly/tree/merge.go
@@ -22,6 +22,7 @@ import (
 	"golang.org/x/sync/errgroup"

 	"github.com/dolthub/dolt/go/store/prolly/message"
+	"github.com/dolthub/dolt/go/store/val"
 )

 const patchBufferSize = 1024
@@ -44,6 +45,7 @@ func ThreeWayMerge[S message.Serializer](
 	compare CompareFn,
 	collide CollisionFn,
 	serializer S,
+	valDesc val.TupleDesc,
 ) (final Node, err error) {

 	ld, err := DifferFromRoots(ctx, ns, base, left, compare)
--- a/go/store/prolly/tree/node.go
+++ b/go/store/prolly/tree/node.go
@@ -58,6 +58,8 @@ func WalkAddresses(ctx context.Context, nd Node, ns NodeStore, cb AddressCb) err

 type NodeCb func(ctx context.Context, nd Node) error

+// WalkNodes runs a callback function on every node found in the DFS of |nd|
+// that is of the same message type as |nd|.
 func WalkNodes(ctx context.Context, nd Node, ns NodeStore, cb NodeCb) error {
 	if err := cb(ctx, nd); err != nil {
 		return err
@@ -76,6 +78,22 @@ func WalkNodes(ctx context.Context, nd Node, ns NodeStore, cb NodeCb) error {
 	})
 }

+// walkOpaqueNodes runs a callback function on every node found in the DFS of |nd|
+// including nested trees.
+func walkOpaqueNodes(ctx context.Context, nd Node, ns NodeStore, cb NodeCb) error {
+	if err := cb(ctx, nd); err != nil {
+		return err
+	}
+
+	return walkAddresses(ctx, nd, func(ctx context.Context, addr hash.Hash) error {
+		child, err := ns.Read(ctx, addr)
+		if err != nil {
+			return err
+		}
+		return WalkNodes(ctx, child, ns, cb)
+	})
+}
+
 func NodeFromBytes(msg []byte) Node {
 	keys, values, count := message.GetKeysAndValues(msg)
 	return Node{
--- a/go/store/prolly/tree/node_cursor_test.go
+++ b/go/store/prolly/tree/node_cursor_test.go
@@ -85,7 +85,7 @@ func randomTree(t *testing.T, count int) (Node, [][2]Item, NodeStore) {
 	chkr, err := newEmptyChunker(ctx, ns, serializer)
 	require.NoError(t, err)

-	items := randomTupleItemPairs(count / 2)
+	items := randomTupleItemPairs(count/2, ns)
 	for _, item := range items {
 		err = chkr.AddPair(ctx, Item(item[0]), Item(item[1]))
 		assert.NoError(t, err)
@@ -112,8 +112,8 @@ func searchTestTree(item Item, nd Node) int {
 	})
 }

-func randomTupleItemPairs(count int) (items [][2]Item) {
-	tups := RandomTuplePairs(count, keyDesc, valDesc)
+func randomTupleItemPairs(count int, ns NodeStore) (items [][2]Item) {
+	tups := RandomTuplePairs(count, keyDesc, valDesc, ns)
 	items = make([][2]Item, count)
 	if len(tups) != len(items) {
 		panic("mismatch")
--- a/go/store/prolly/tree/testutils.go
+++ b/go/store/prolly/tree/testutils.go
@@ -15,6 +15,7 @@
 package tree

 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math"
@@ -43,14 +44,14 @@ func NewTupleLeafNode(keys, values []val.Tuple) Node {
 	return newLeafNode(ks, vs)
 }

-func RandomTuplePairs(count int, keyDesc, valDesc val.TupleDesc) (items [][2]val.Tuple) {
+func RandomTuplePairs(count int, keyDesc, valDesc val.TupleDesc, ns NodeStore) (items [][2]val.Tuple) {
 	keyBuilder := val.NewTupleBuilder(keyDesc)
 	valBuilder := val.NewTupleBuilder(valDesc)

 	items = make([][2]val.Tuple, count)
 	for i := range items {
-		items[i][0] = RandomTuple(keyBuilder)
-		items[i][1] = RandomTuple(valBuilder)
+		items[i][0] = RandomTuple(keyBuilder, ns)
+		items[i][1] = RandomTuple(valBuilder, ns)
 	}

 	dupes := make([]int, 0, count)
@@ -70,14 +71,14 @@ func RandomTuplePairs(count int, keyDesc, valDesc val.TupleDesc) (items [][2]val

 		// replace duplicates and validate again
 		for _, d := range dupes {
-			items[d][0] = RandomTuple(keyBuilder)
+			items[d][0] = RandomTuple(keyBuilder, ns)
 		}
 		dupes = dupes[:0]
 	}
 	return items
 }

-func RandomCompositeTuplePairs(count int, keyDesc, valDesc val.TupleDesc) (items [][2]val.Tuple) {
+func RandomCompositeTuplePairs(count int, keyDesc, valDesc val.TupleDesc, ns NodeStore) (items [][2]val.Tuple) {
 	// preconditions
 	if count%5 != 0 {
 		panic("expected empty divisible by 5")
@@ -86,7 +87,7 @@ func RandomCompositeTuplePairs(count int, keyDesc, valDesc val.TupleDesc) (items
 		panic("expected composite key")
 	}

-	tt := RandomTuplePairs(count, keyDesc, valDesc)
+	tt := RandomTuplePairs(count, keyDesc, valDesc, ns)

 	tuples := make([][2]val.Tuple, len(tt)*3)
 	for i := range tuples {
@@ -125,9 +126,9 @@ func AscendingUintTuples(count int) (tuples [][2]val.Tuple, desc val.TupleDesc)
 	return
 }

-func RandomTuple(tb *val.TupleBuilder) (tup val.Tuple) {
+func RandomTuple(tb *val.TupleBuilder, ns NodeStore) (tup val.Tuple) {
 	for i, typ := range tb.Desc.Types {
-		randomField(tb, i, typ)
+		randomField(tb, i, typ, ns)
 	}
 	return tb.Build(sharedPool)
 }
@@ -181,7 +182,7 @@ func deduplicateTuples(desc val.TupleDesc, tups [][2]val.Tuple) (uniq [][2]val.T
 	return
 }

-func randomField(tb *val.TupleBuilder, idx int, typ val.Type) {
+func randomField(tb *val.TupleBuilder, idx int, typ val.Type, ns NodeStore) {
 	// todo(andy): add NULLs

 	neg := -1
@@ -230,10 +231,19 @@ func randomField(tb *val.TupleBuilder, idx int, typ val.Type) {
 		buf := make([]byte, 16)
 		testRand.Read(buf)
 		tb.PutHash128(idx, buf)
-	case val.AddressEnc:
+	case val.CommitAddrEnc:
 		buf := make([]byte, 20)
 		testRand.Read(buf)
-		tb.PutAddress(idx, hash.New(buf))
+		tb.PutCommitAddr(idx, hash.New(buf))
+	case val.BytesAddrEnc:
+		buf := make([]byte, (testRand.Int63()%40)+10)
+		testRand.Read(buf)
+		tree, err := NewImmutableTreeFromReader(context.Background(), bytes.NewReader(buf), ns, DefaultFixedChunkLength)
+		if err != nil {
+			panic("failed to write blob tree")
+		}
+		tb.PutBytesAddr(idx, tree.Addr)
+
 	default:
 		panic("unknown encoding")
 	}
--- a/go/store/val/byte_array.go
+++ b/go/store/val/byte_array.go
@@ -0,0 +1,25 @@
+// Copyright 2022 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 val
+
+import "github.com/dolthub/dolt/go/store/hash"
+
+type BytesAddr struct {
+	Addr hash.Hash
+}
+
+func NewBytesAddr(addr hash.Hash) BytesAddr {
+	return BytesAddr{Addr: addr}
+}
--- a/go/store/val/codec.go
+++ b/go/store/val/codec.go
@@ -53,39 +53,40 @@ const (
 	float64Size  ByteSize = 8
 	bit64Size    ByteSize = 8
 	hash128Size  ByteSize = 16
-	addressSize  ByteSize = 20
 	yearSize     ByteSize = 1
 	dateSize     ByteSize = 4
 	timeSize     ByteSize = 8
 	datetimeSize ByteSize = 8
 	enumSize     ByteSize = 2
 	setSize      ByteSize = 8
+	addrSize     ByteSize = hash.ByteLen
 )

 type Encoding byte

 // Fixed Width Encodings
 const (
-	NullEnc     = Encoding(serial.EncodingNull)
-	Int8Enc     = Encoding(serial.EncodingInt8)
-	Uint8Enc    = Encoding(serial.EncodingUint8)
-	Int16Enc    = Encoding(serial.EncodingInt16)
-	Uint16Enc   = Encoding(serial.EncodingUint16)
-	Int32Enc    = Encoding(serial.EncodingInt32)
-	Uint32Enc   = Encoding(serial.EncodingUint32)
-	Int64Enc    = Encoding(serial.EncodingInt64)
-	Uint64Enc   = Encoding(serial.EncodingUint64)
-	Float32Enc  = Encoding(serial.EncodingFloat32)
-	Float64Enc  = Encoding(serial.EncodingFloat64)
-	Bit64Enc    = Encoding(serial.EncodingBit64)
-	Hash128Enc  = Encoding(serial.EncodingHash128)
-	AddressEnc  = Encoding(serial.EncodingAddress)
-	YearEnc     = Encoding(serial.EncodingYear)
-	DateEnc     = Encoding(serial.EncodingDate)
-	TimeEnc     = Encoding(serial.EncodingTime)
-	DatetimeEnc = Encoding(serial.EncodingDatetime)
-	EnumEnc     = Encoding(serial.EncodingEnum)
-	SetEnc      = Encoding(serial.EncodingSet)
+	NullEnc       = Encoding(serial.EncodingNull)
+	Int8Enc       = Encoding(serial.EncodingInt8)
+	Uint8Enc      = Encoding(serial.EncodingUint8)
+	Int16Enc      = Encoding(serial.EncodingInt16)
+	Uint16Enc     = Encoding(serial.EncodingUint16)
+	Int32Enc      = Encoding(serial.EncodingInt32)
+	Uint32Enc     = Encoding(serial.EncodingUint32)
+	Int64Enc      = Encoding(serial.EncodingInt64)
+	Uint64Enc     = Encoding(serial.EncodingUint64)
+	Float32Enc    = Encoding(serial.EncodingFloat32)
+	Float64Enc    = Encoding(serial.EncodingFloat64)
+	Bit64Enc      = Encoding(serial.EncodingBit64)
+	Hash128Enc    = Encoding(serial.EncodingHash128)
+	YearEnc       = Encoding(serial.EncodingYear)
+	DateEnc       = Encoding(serial.EncodingDate)
+	TimeEnc       = Encoding(serial.EncodingTime)
+	DatetimeEnc   = Encoding(serial.EncodingDatetime)
+	EnumEnc       = Encoding(serial.EncodingEnum)
+	SetEnc        = Encoding(serial.EncodingSet)
+	BytesAddrEnc  = Encoding(serial.EncodingBytesAddr)
+	CommitAddrEnc = Encoding(serial.EncodingCommitAddr)

 	sentinel Encoding = 127
 )
@@ -132,8 +133,8 @@ func sizeFromType(t Type) (ByteSize, bool) {
 		return float64Size, true
 	case Hash128Enc:
 		return hash128Size, true
-	case AddressEnc:
-		return addressSize, true
+	case BytesAddrEnc:
+		return addrSize, true
 	case YearEnc:
 		return yearSize, true
 	case DateEnc:
@@ -559,22 +560,17 @@ func writeHash128(buf, val []byte) {
 	copy(buf, val)
 }

-func readAddress(val []byte) []byte {
-	expectSize(val, addressSize)
-	return val
-}
-
-func writeAddress(buf []byte, val hash.Hash) {
-	expectSize(buf, addressSize)
-	copy(buf, val[:])
-}
-
 func compareHash128(l, r []byte) int {
 	return bytes.Compare(l, r)
 }

-func compareAddress(l, r []byte) int {
-	return bytes.Compare(l, r)
+func compareBytesAddr(l, r hash.Hash) int {
+	// TODO sort
+	return l.Compare(r)
+}
+
+func compareCommitAddr(l, r hash.Hash) int {
+	return l.Compare(r)
 }

 func writeRaw(buf, val []byte) {
@@ -582,6 +578,16 @@ func writeRaw(buf, val []byte) {
 	copy(buf, val)
 }

+func writeAddr(buf []byte, v []byte) {
+	expectSize(buf, addrSize)
+	copy(buf, v)
+}
+
+func readAddr(val []byte) hash.Hash {
+	expectSize(val, addrSize)
+	return hash.New(val)
+}
+
 func expectSize(buf []byte, sz ByteSize) {
 	if ByteSize(len(buf)) != sz {
 		panic("byte slice is not of expected size")
--- a/go/store/val/tuple.go
+++ b/go/store/val/tuple.go
@@ -127,6 +127,29 @@ func allocateTuple(pool pool.BuffPool, bufSz ByteSize, fields int) (tup Tuple, o
 	return
 }

+func (tup Tuple) GetOffset(i int) (int, bool) {
+	cnt := tup.Count()
+	if i >= cnt {
+		return 0, false
+	}
+
+	sz := ByteSize(len(tup))
+	split := sz - uint16Size*ByteSize(cnt)
+	offs := tup[split : sz-countSize]
+
+	start, stop := uint16(0), uint16(split)
+	if i*2 < len(offs) {
+		pos := i * 2
+		stop = readUint16(offs[pos : pos+2])
+	}
+	if i > 0 {
+		pos := (i - 1) * 2
+		start = readUint16(offs[pos : pos+2])
+	}
+
+	return int(start), start == stop
+}
+
 // GetField returns the value for field |i|.
 func (tup Tuple) GetField(i int) []byte {
 	cnt := tup.Count()
--- a/go/store/val/tuple_builder.go
+++ b/go/store/val/tuple_builder.go
@@ -263,11 +263,11 @@ func (tb *TupleBuilder) PutHash128(i int, v []byte) {
 	tb.pos += hash128Size
 }

-func (tb *TupleBuilder) PutAddress(i int, v hash.Hash) {
-	tb.Desc.expectEncoding(i, AddressEnc)
-	tb.fields[i] = tb.buf[tb.pos : tb.pos+addressSize]
-	writeAddress(tb.fields[i], v)
-	tb.pos += addressSize
+func (tb *TupleBuilder) PutCommitAddr(i int, v hash.Hash) {
+	tb.Desc.expectEncoding(i, CommitAddrEnc)
+	tb.fields[i] = tb.buf[tb.pos : tb.pos+addrSize]
+	writeAddr(tb.fields[i], v[:])
+	tb.pos += addrSize
 }

 // PutRaw writes a []byte to the ith field of the Tuple being built.
@@ -282,3 +282,11 @@ func (tb *TupleBuilder) PutRaw(i int, buf []byte) {
 	writeRaw(tb.fields[i], buf)
 	tb.pos += sz
 }
+
+// PutBytesAddr writes an out of band []byte to the ith field of the Tuple being built.
+func (tb *TupleBuilder) PutBytesAddr(i int, v hash.Hash) {
+	tb.Desc.expectEncoding(i, BytesAddrEnc)
+	tb.fields[i] = tb.buf[tb.pos : tb.pos+addrSize]
+	writeAddr(tb.fields[i], v[:])
+	tb.pos += addrSize
+}
--- a/go/store/val/tuple_compare.go
+++ b/go/store/val/tuple_compare.go
@@ -112,8 +112,10 @@ func compare(typ Type, left, right []byte) int {
 		return compareByteString(readByteString(left), readByteString(right))
 	case Hash128Enc:
 		return compareHash128(readHash128(left), readHash128(right))
-	case AddressEnc:
-		return compareAddress(readAddress(left), readAddress(right))
+	case BytesAddrEnc:
+		return compareBytesAddr(readAddr(left), readAddr(right))
+	case CommitAddrEnc:
+		return compareCommitAddr(readAddr(left), readAddr(right))
 	default:
 		panic("unknown encoding")
 	}
--- a/go/store/val/tuple_descriptor.go
+++ b/go/store/val/tuple_descriptor.go
@@ -20,9 +20,9 @@ import (
 	"strings"
 	"time"

-	"github.com/shopspring/decimal"
-
 	"github.com/dolthub/dolt/go/store/hash"
+
+	"github.com/shopspring/decimal"
 )

 // TupleDesc describes a Tuple set.
@@ -32,6 +32,7 @@ type TupleDesc struct {
 	Types []Type
 	cmp   TupleComparator
 	fast  fixedAccess
+	Addrs []int
 }

 // NewTupleDescriptor makes a TupleDescriptor from |types|.
@@ -51,10 +52,18 @@ func NewTupleDescriptorWithComparator(cmp TupleComparator, types ...Type) (td Tu
 		}
 	}

+	var addrIdxs []int
+	for i, t := range types {
+		if t.Enc == BytesAddrEnc {
+			addrIdxs = append(addrIdxs, i)
+		}
+	}
+
 	td = TupleDesc{
 		Types: types,
 		cmp:   cmp,
 		fast:  makeFixedAccess(types),
+		Addrs: addrIdxs,
 	}

 	return
@@ -399,8 +408,17 @@ func (td TupleDesc) GetHash128(i int, tup Tuple) (v []byte, ok bool) {
 	return
 }

-func (td TupleDesc) GetAddress(i int, tup Tuple) (v hash.Hash, ok bool) {
-	td.expectEncoding(i, AddressEnc)
+func (td TupleDesc) GetBlob(i int, tup Tuple) (BytesAddr, bool) {
+	td.expectEncoding(i, BytesAddrEnc)
+	b := td.GetField(i, tup)
+	if len(b) == 0 {
+		return BytesAddr{}, false
+	}
+	return NewBytesAddr(hash.New(b)), true
+}
+
+func (td TupleDesc) GetCommitAddr(i int, tup Tuple) (v hash.Hash, ok bool) {
+	td.expectEncoding(i, CommitAddrEnc)
 	b := td.GetField(i, tup)
 	if b != nil {
 		v = hash.New(b)
@@ -508,7 +526,9 @@ func formatValue(enc Encoding, value []byte) string {
 		return string(value)
 	case Hash128Enc:
 		return string(value)
-	case AddressEnc:
+	case BytesAddrEnc:
+		return string(value)
+	case CommitAddrEnc:
 		return string(value)
 	default:
 		return string(value)
--- a/integration-tests/bats/1pk5col-strings.bats
+++ b/integration-tests/bats/1pk5col-strings.bats
@@ -5,7 +5,7 @@ setup() {
    setup_common
    dolt sql <<SQL
 CREATE TABLE test (
-  pk LONGTEXT NOT NULL COMMENT 'tag:0',
+  pk varchar(20) NOT NULL COMMENT 'tag:0',
  c1 LONGTEXT COMMENT 'tag:1',
  c2 LONGTEXT COMMENT 'tag:2',
  c3 LONGTEXT COMMENT 'tag:3',
--- a/integration-tests/bats/blame-system-view.bats
+++ b/integration-tests/bats/blame-system-view.bats
@@ -19,7 +19,7 @@ setup_repository() {
    dolt sql <<SQL
 CREATE TABLE blame_test (
  pk1 BIGINT NOT NULL COMMENT 'tag:0',
-  pk2 TEXT,
+  pk2 varchar(40),
  name LONGTEXT COMMENT 'tag:1',
  PRIMARY KEY (pk1, pk2)
 );
@@ -206,4 +206,4 @@ SQL
    [ "$status" -eq 0 ]
    echo -e "OUTPUT:\n $output"
    false
-}
+}
--- a/integration-tests/bats/diff.bats
+++ b/integration-tests/bats/diff.bats
@@ -319,7 +319,7 @@ DELIM
    dolt sql -q "insert into test values (1, 1, 1, 1, 1, 1)"
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL,
+  \`id\` varchar(20) NOT NULL,
  \`first name\` LONGTEXT,
  \`last name\` LONGTEXT,
  \`title\` LONGTEXT,
@@ -673,4 +673,4 @@ SQL

    run dolt diff HEAD~1
    [ "${#lines[@]}" -eq 2007 ] # 2000 diffs + 6 for top rows before data + 1 for bottom row of table
-}
+}
--- a/integration-tests/bats/docs.bats
+++ b/integration-tests/bats/docs.bats
@@ -713,7 +713,7 @@ SQL
    [ "$status" -eq 0 ]
    [[ ! "$output" =~ "dolt_docs" ]] || false

-    run dolt sql -q "CREATE TABLE dolt_docs (doc_name TEXT, doc_text LONGTEXT, PRIMARY KEY(doc_name))"
+    run dolt sql -q "CREATE TABLE dolt_docs (doc_name varchar(20), doc_text varchar(20), PRIMARY KEY(doc_name))"
    [ "$status" -eq 1 ]
    [[ "$output" =~ "reserved" ]] || false
    
--- a/integration-tests/bats/export-tables.bats
+++ b/integration-tests/bats/export-tables.bats
@@ -16,7 +16,7 @@ CREATE TABLE test_int (
  PRIMARY KEY (pk)
 );
 CREATE TABLE test_string (
-  pk LONGTEXT NOT NULL,
+  pk varchar(20) NOT NULL,
  c1 LONGTEXT,
  c2 LONGTEXT,
  c3 LONGTEXT,
--- a/integration-tests/bats/helper/employees-sch.sql
+++ b/integration-tests/bats/helper/employees-sch.sql
@@ -1,9 +1,9 @@
 CREATE TABLE employees (
-    `id` LONGTEXT NOT NULL,
+    `id` varchar(20) NOT NULL,
    `first name` LONGTEXT,
    `last name` LONGTEXT,
    `title` LONGTEXT,
    `start date` LONGTEXT,
    `end date` LONGTEXT,
    PRIMARY KEY (`id`)
-);
+);
--- a/integration-tests/bats/import-create-tables.bats
+++ b/integration-tests/bats/import-create-tables.bats
@@ -485,7 +485,7 @@ SQL
@test "import-create-tables: create a table with null values from json import with json file" {
    dolt sql <<SQL
 CREATE TABLE test (
-  pk LONGTEXT NOT NULL,
+  pk varchar(20) NOT NULL,
  headerOne LONGTEXT,
  headerTwo BIGINT,
  PRIMARY KEY (pk)
@@ -512,7 +512,7 @@ SQL
@test "import-create-tables: fail to create a table with null values from json import with json file" {
    dolt sql <<SQL
 CREATE TABLE test (
-  pk LONGTEXT NOT NULL,
+  pk varchar(20) NOT NULL,
  headerOne LONGTEXT NOT NULL,
  headerTwo BIGINT NOT NULL,
  PRIMARY KEY (pk)
--- a/integration-tests/bats/import-replace-tables.bats
+++ b/integration-tests/bats/import-replace-tables.bats
@@ -103,7 +103,7 @@ SQL
@test "import-replace-tables: replace table using json" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -121,7 +121,7 @@ SQL
@test "import-replace-tables: replace table using json with wrong schema" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`idz\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`idz\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first namez\` LONGTEXT COMMENT 'tag:1',
  \`last namez\` LONGTEXT COMMENT 'tag:2',
  \`titlez\` LONGTEXT COMMENT 'tag:3',
@@ -140,7 +140,7 @@ SQL
@test "import-replace-tables: replace table using schema with json" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`idz\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`idz\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first namez\` LONGTEXT COMMENT 'tag:1',
  \`last namez\` LONGTEXT COMMENT 'tag:2',
  \`titlez\` LONGTEXT COMMENT 'tag:3',
@@ -182,7 +182,7 @@ SQL
@test "import-replace-tables: replace table with bad json" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -199,7 +199,7 @@ SQL
@test "import-replace-tables: replace table using xlsx file" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first\` LONGTEXT COMMENT 'tag:1',
  \`last\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -217,7 +217,7 @@ SQL
@test "import-replace-tables: replace table using xlsx file with wrong schema" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -253,7 +253,7 @@ SQL
@test "import-replace-tables: replace table with a json with columns in different order" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -271,7 +271,7 @@ SQL
@test "import-replace-tables: replace table with a csv with columns in different order" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
--- a/integration-tests/bats/import-update-tables.bats
+++ b/integration-tests/bats/import-update-tables.bats
@@ -37,7 +37,7 @@ DELIM

    cat <<SQL > employees-sch.sql
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -158,7 +158,7 @@ teardown() {
@test "import-update-tables: update table using csv with newlines" {
    dolt sql <<SQL
 CREATE TABLE test (
-  pk LONGTEXT NOT NULL COMMENT 'tag:0',
+  pk varchar(20) NOT NULL COMMENT 'tag:0',
  c1 LONGTEXT COMMENT 'tag:1',
  c2 LONGTEXT COMMENT 'tag:2',
  c3 LONGTEXT COMMENT 'tag:3',
@@ -182,7 +182,7 @@ SQL
@test "import-update-tables: update table using wrong json" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`idz\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`idz\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first namez\` LONGTEXT COMMENT 'tag:1',
  \`last namez\` LONGTEXT COMMENT 'tag:2',
  \`titlez\` LONGTEXT COMMENT 'tag:3',
@@ -221,7 +221,7 @@ SQL
@test "import-update-tables: update table with a json with columns in different order" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
@@ -247,7 +247,7 @@ SQL
@test "import-update-tables: update table with a csv with columns in different order" {
    dolt sql <<SQL
 CREATE TABLE employees (
-  \`id\` LONGTEXT NOT NULL COMMENT 'tag:0',
+  \`id\` varchar(20) NOT NULL COMMENT 'tag:0',
  \`first name\` LONGTEXT COMMENT 'tag:1',
  \`last name\` LONGTEXT COMMENT 'tag:2',
  \`title\` LONGTEXT COMMENT 'tag:3',
--- a/integration-tests/bats/regression-tests.bats
+++ b/integration-tests/bats/regression-tests.bats
--- a/integration-tests/bats/types.bats
+++ b/integration-tests/bats/types.bats
@@ -704,8 +704,7 @@ SQL
 CREATE TABLE test (
  pk BIGINT NOT NULL,
  v LONGTEXT,
-  PRIMARY KEY (pk),
-  INDEX (v)
+  PRIMARY KEY (pk)
 );
 SQL
    run dolt schema show