Merge branch 'main' into james/local-user

2026-02-12 02:58:53 -06:00 · 2022-06-07 13:32:37 -07:00
parent 6b14afaee9 fe7ff993d9
commit 00b6d19dd2
38 changed files with 1210 additions and 134 deletions
--- a/go/cmd/dolt/commands/init.go
+++ b/go/cmd/dolt/commands/init.go
@@ -140,6 +140,21 @@ func (cmd InitCmd) Exec(ctx context.Context, commandStr string, args []string, d
 		return 1
 	}

+	configuration := make(map[string]string)
+	if apr.Contains(usernameParamName) {
+		configuration[env.UserNameKey] = name
+	}
+	if apr.Contains(emailParamName) {
+		configuration[env.UserEmailKey] = email
+	}
+	if len(configuration) > 0 {
+		err = dEnv.Config.WriteableConfig().SetStrings(configuration)
+		if err != nil {
+			cli.PrintErrln(color.RedString("Failed to store initial configuration. %s", err.Error()))
+			return 1
+		}
+	}
+
 	cli.Println(color.CyanString("Successfully initialized dolt data repository."))
 	return 0
 }
--- a/go/cmd/dolt/commands/init_test.go
+++ b/go/cmd/dolt/commands/init_test.go
@@ -18,16 +18,20 @@ import (
 	"context"
 	"testing"

+	"github.com/stretchr/testify/require"
+
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 )

+type initTest struct {
+	Name          string
+	Args          []string
+	GlobalConfig  map[string]string
+	ExpectSuccess bool
+}
+
 func TestInit(t *testing.T) {
-	tests := []struct {
-		Name          string
-		Args          []string
-		GlobalConfig  map[string]string
-		ExpectSuccess bool
-	}{
+	tests := []initTest{
 		{
 			"Command Line name and email",
 			[]string{"-name", "Bill Billerson", "-email", "bigbillieb@fake.horse"},
@@ -65,18 +69,15 @@ func TestInit(t *testing.T) {

 			result := InitCmd{}.Exec(context.Background(), "dolt init", test.Args, dEnv)

-			if (result == 0) != test.ExpectSuccess {
-				t.Error(test.Name, "- Expected success:", test.ExpectSuccess, "result:", result == 0)
-			} else if test.ExpectSuccess {
-				// succceeded as expected
-				if !dEnv.HasDoltDir() {
-					t.Error(test.Name, "- .dolt dir should exist after initialization")
-				}
+			require.Equalf(t, test.ExpectSuccess, result == 0, "- Expected success: %t; result: %t;", test.ExpectSuccess, result == 0)
+
+			if test.ExpectSuccess {
+				require.True(t, dEnv.HasDoltDir(), "- .dolt dir should exist after initialization")
+				testLocalConfigValue(t, dEnv, test, usernameParamName, env.UserNameKey)
+				testLocalConfigValue(t, dEnv, test, emailParamName, env.UserEmailKey)
 			} else {
-				// failed as expected
-				if dEnv.HasDoltDir() {
-					t.Error(test.Name, "- dolt directory shouldn't exist after failure to initialize")
-				}
+				require.False(t, dEnv.HasDoltDir(),
+					"- dolt directory shouldn't exist after failure to initialize")
 			}
 		})
 	}
@@ -84,15 +85,37 @@ func TestInit(t *testing.T) {

 func TestInitTwice(t *testing.T) {
 	dEnv := createUninitializedEnv()
-	result := InitCmd{}.Exec(context.Background(), "dolt init", []string{"-name", "Bill Billerson", "-email", "bigbillieb@fake.horse"}, dEnv)
+	result := InitCmd{}.Exec(context.Background(), "dolt init",
+		[]string{"-name", "Bill Billerson", "-email", "bigbillieb@fake.horse"}, dEnv)
+	require.True(t, result == 0, "First init should succeed")

-	if result != 0 {
-		t.Error("First init should succeed")
+	result = InitCmd{}.Exec(context.Background(), "dolt init",
+		[]string{"-name", "Bill Billerson", "-email", "bigbillieb@fake.horse"}, dEnv)
+	require.True(t, result != 0, "Second init should fail")
+}
+
+// testLocalConfigValue tests that local config data is set correctly when the specified argument
+// is present in the command line args, and is not set when the argument is not present.
+func testLocalConfigValue(t *testing.T, dEnv *env.DoltEnv, test initTest, argKey, envKey string) {
+	localConfig, ok := dEnv.Config.GetConfig(env.LocalConfig)
+	require.True(t, ok, "- Unable to load local configuration")
+
+	found := false
+	expectedValue := ""
+	for i := 0; i <= len(test.Args)-2; i = i + 2 {
+		if test.Args[i] == "-"+argKey {
+			found = true
+			expectedValue = test.Args[i+1]
+		}
 	}

-	result = InitCmd{}.Exec(context.Background(), "dolt init", []string{"-name", "Bill Billerson", "-email", "bigbillieb@fake.horse"}, dEnv)
-
-	if result == 0 {
-		t.Error("Second init should fail")
+	actualValue, err := localConfig.GetString(envKey)
+	if found {
+		require.NoErrorf(t, err, "- Expected '%s', but not found in local config; error: %v",
+			expectedValue, err)
+		require.Equalf(t, expectedValue, actualValue, "- Expected '%s' in local config, but found '%s'",
+			expectedValue, actualValue)
+	} else {
+		require.Errorf(t, err, "- Expected nothing in local config, but found '%s'", actualValue)
 	}
 }
--- a/go/go.mod
+++ b/go/go.mod
@@ -68,7 +68,7 @@ require (
 )

 require (
-	github.com/dolthub/go-mysql-server v0.11.1-0.20220604073908-f302c0189c8a
+	github.com/dolthub/go-mysql-server v0.11.1-0.20220607183612-59df03f03a67
 	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.11.1-0.20220604073908-f302c0189c8a h1:cwUbfuVOs5QgyiOf+V8Ms/qtqmvnABw2ZyWwp0/yirI=
-github.com/dolthub/go-mysql-server v0.11.1-0.20220604073908-f302c0189c8a/go.mod h1:gvDEMITJQDVYDLR4XtcqEZx6rawTvMh2veM1bPsJC3I=
+github.com/dolthub/go-mysql-server v0.11.1-0.20220607183612-59df03f03a67 h1:25IWOFOFq4ADJokQN6ma83j06leYE+uUU91Y7sHP8Hg=
+github.com/dolthub/go-mysql-server v0.11.1-0.20220607183612-59df03f03a67/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/diff/async_differ.go
+++ b/go/libraries/doltcore/diff/async_differ.go
@@ -32,6 +32,9 @@ type RowDiffer interface {
 	// Start starts the RowDiffer.
 	Start(ctx context.Context, from, to types.Map)

+	// StartWithRange starts the RowDiffer with the specified range
+	StartWithRange(ctx context.Context, from, to types.Map, start types.Value, inRange types.ValueInRange)
+
 	// GetDiffs returns the requested number of diff.Differences, or times out.
 	GetDiffs(numDiffs int, timeout time.Duration) ([]*diff.Difference, bool, error)

@@ -298,6 +301,10 @@ var _ RowDiffer = &EmptyRowDiffer{}
 func (e EmptyRowDiffer) Start(ctx context.Context, from, to types.Map) {
 }

+func (e EmptyRowDiffer) StartWithRange(ctx context.Context, from, to types.Map, start types.Value, inRange types.ValueInRange) {
+
+}
+
 func (e EmptyRowDiffer) GetDiffs(numDiffs int, timeout time.Duration) ([]*diff.Difference, bool, error) {
 	return nil, false, nil
 }
--- a/go/libraries/doltcore/diff/async_differ_test.go
+++ b/go/libraries/doltcore/diff/async_differ_test.go
@@ -112,8 +112,8 @@ func TestAsyncDiffer(t *testing.T) {
 			name: "iter range starting with nil",
 			createdStarted: func(ctx context.Context, m1, m2 types.Map) *AsyncDiffer {
 				ad := NewAsyncDiffer(4)
-				ad.StartWithRange(ctx, m1, m2, nil, func(value types.Value) (bool, error) {
-					return true, nil
+				ad.StartWithRange(ctx, m1, m2, nil, func(ctx context.Context, value types.Value) (bool, bool, error) {
+					return true, false, nil
 				})
 				return ad
 			},
@@ -128,8 +128,8 @@ func TestAsyncDiffer(t *testing.T) {
 			name: "iter range staring with Null Value",
 			createdStarted: func(ctx context.Context, m1, m2 types.Map) *AsyncDiffer {
 				ad := NewAsyncDiffer(4)
-				ad.StartWithRange(ctx, m1, m2, types.NullValue, func(value types.Value) (bool, error) {
-					return true, nil
+				ad.StartWithRange(ctx, m1, m2, types.NullValue, func(ctx context.Context, value types.Value) (bool, bool, error) {
+					return true, false, nil
 				})
 				return ad
 			},
@@ -145,8 +145,9 @@ func TestAsyncDiffer(t *testing.T) {
 			createdStarted: func(ctx context.Context, m1, m2 types.Map) *AsyncDiffer {
 				ad := NewAsyncDiffer(4)
 				end := types.Uint(27)
-				ad.StartWithRange(ctx, m1, m2, types.NullValue, func(value types.Value) (bool, error) {
-					return value.Less(m1.Format(), end)
+				ad.StartWithRange(ctx, m1, m2, types.NullValue, func(ctx context.Context, value types.Value) (bool, bool, error) {
+					valid, err := value.Less(m1.Format(), end)
+					return valid, false, err
 				})
 				return ad
 			},
@@ -162,8 +163,9 @@ func TestAsyncDiffer(t *testing.T) {
 			createdStarted: func(ctx context.Context, m1, m2 types.Map) *AsyncDiffer {
 				ad := NewAsyncDiffer(4)
 				end := types.Uint(15)
-				ad.StartWithRange(ctx, m1, m2, types.NullValue, func(value types.Value) (bool, error) {
-					return value.Less(m1.Format(), end)
+				ad.StartWithRange(ctx, m1, m2, types.NullValue, func(ctx context.Context, value types.Value) (bool, bool, error) {
+					valid, err := value.Less(m1.Format(), end)
+					return valid, false, err
 				})
 				return ad
 			},
@@ -180,8 +182,9 @@ func TestAsyncDiffer(t *testing.T) {
 				ad := NewAsyncDiffer(4)
 				start := types.Uint(10)
 				end := types.Uint(15)
-				ad.StartWithRange(ctx, m1, m2, start, func(value types.Value) (bool, error) {
-					return value.Less(m1.Format(), end)
+				ad.StartWithRange(ctx, m1, m2, start, func(ctx context.Context, value types.Value) (bool, bool, error) {
+					valid, err := value.Less(m1.Format(), end)
+					return valid, false, err
 				})
 				return ad
 			},
--- a/go/libraries/doltcore/doltdb/durable/index.go
+++ b/go/libraries/doltcore/doltdb/durable/index.go
@@ -147,6 +147,24 @@ type nomsIndex struct {

 var _ Index = nomsIndex{}

+func IterAllIndexes(
+	ctx context.Context,
+	sch schema.Schema,
+	set IndexSet,
+	cb func(name string, idx Index) error,
+) error {
+	for _, def := range sch.Indexes().AllIndexes() {
+		idx, err := set.GetIndex(ctx, sch, def.Name())
+		if err != nil {
+			return err
+		}
+		if err = cb(def.Name(), idx); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // NomsMapFromIndex unwraps the Index and returns the underlying types.Map.
 func NomsMapFromIndex(i Index) types.Map {
 	return i.(nomsIndex).index
--- a/go/libraries/doltcore/sqle/dfunctions/dolt_checkout.go
+++ b/go/libraries/doltcore/sqle/dfunctions/dolt_checkout.go
@@ -78,6 +78,8 @@ func DoDoltCheckout(ctx *sql.Context, args []string) (int, error) {
 	if newBranch, newBranchOk := apr.GetValue(cli.CheckoutCoBranch); newBranchOk {
 		if len(newBranch) == 0 {
 			err = errors.New("error: cannot checkout empty string")
+		} else if len(apr.Args) > 0 {
+			err = checkoutNewBranch(ctx, dbName, dbData, roots, newBranch, apr.Arg(0))
 		} else {
 			err = checkoutNewBranch(ctx, dbName, dbData, roots, newBranch, "")
 		}
--- a/go/libraries/doltcore/sqle/dolt_diff_table_function.go
+++ b/go/libraries/doltcore/sqle/dolt_diff_table_function.go
@@ -397,7 +397,7 @@ func (itr *diffTableFunctionRowIter) Next(ctx *sql.Context) (sql.Row, error) {

 		if itr.currentRowIter == nil {
 			dp := (*itr.currentPartition).(dtables.DiffPartition)
-			rowIter, err := dp.GetRowIter(ctx, itr.ddb, itr.joiner)
+			rowIter, err := dp.GetRowIter(ctx, itr.ddb, itr.joiner, nil)
 			if err != nil {
 				return nil, err
 			}
--- a/go/libraries/doltcore/sqle/dtables/commit_diff_table.go
+++ b/go/libraries/doltcore/sqle/dtables/commit_diff_table.go
@@ -288,5 +288,5 @@ func (dt *CommitDiffTable) WithFilters(_ *sql.Context, _ []sql.Expression) sql.T

 func (dt *CommitDiffTable) PartitionRows(ctx *sql.Context, part sql.Partition) (sql.RowIter, error) {
 	dp := part.(DiffPartition)
-	return dp.GetRowIter(ctx, dt.ddb, dt.joiner)
+	return dp.GetRowIter(ctx, dt.ddb, dt.joiner, nil)
 }
--- a/go/libraries/doltcore/sqle/dtables/diff_iter.go
+++ b/go/libraries/doltcore/sqle/dtables/diff_iter.go
@@ -27,6 +27,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb/durable"
 	"github.com/dolthub/dolt/go/libraries/doltcore/rowconv"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/index"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/sqlutil"
 	"github.com/dolthub/dolt/go/store/prolly"
 	"github.com/dolthub/dolt/go/store/prolly/tree"
@@ -52,7 +53,7 @@ type commitInfo struct {
 	dateTag uint64
 }

-func newNomsDiffIter(ctx *sql.Context, ddb *doltdb.DoltDB, joiner *rowconv.Joiner, dp DiffPartition) (*diffRowItr, error) {
+func newNomsDiffIter(ctx *sql.Context, ddb *doltdb.DoltDB, joiner *rowconv.Joiner, dp DiffPartition, lookup sql.IndexLookup) (*diffRowItr, error) {
 	fromData, fromSch, err := tableData(ctx, dp.from, ddb)

 	if err != nil {
@@ -88,7 +89,21 @@ func newNomsDiffIter(ctx *sql.Context, ddb *doltdb.DoltDB, joiner *rowconv.Joine

 	rd := diff.NewRowDiffer(ctx, fromSch, toSch, 1024)
 	// TODO (dhruv) don't cast to noms map
-	rd.Start(ctx, durable.NomsMapFromIndex(fromData), durable.NomsMapFromIndex(toData))
+	// Use index lookup if it exists
+	if lookup == nil {
+		rd.Start(ctx, durable.NomsMapFromIndex(fromData), durable.NomsMapFromIndex(toData))
+	} else {
+		ranges := index.NomsRangesFromIndexLookup(lookup) // TODO: this is a testing method
+		// TODO: maybe just use Check
+		rangeFunc := func(ctx context.Context, val types.Value) (bool, bool, error) {
+			v, ok := val.(types.Tuple)
+			if !ok {
+				return false, false, nil
+			}
+			return ranges[0].Check.Check(ctx, v)
+		}
+		rd.StartWithRange(ctx, durable.NomsMapFromIndex(fromData), durable.NomsMapFromIndex(toData), ranges[0].Start, rangeFunc)
+	}

 	src := diff.NewRowDiffSource(rd, joiner, ctx.Warn)
 	src.AddInputRowConversion(fromConv, toConv)
--- a/go/libraries/doltcore/sqle/dtables/diff_table.go
+++ b/go/libraries/doltcore/sqle/dtables/diff_table.go
@@ -29,6 +29,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/rowconv"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/expreval"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/index"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/sqlutil"
 	"github.com/dolthub/dolt/go/libraries/utils/set"
 	"github.com/dolthub/dolt/go/store/hash"
@@ -49,6 +50,8 @@ const (

 var _ sql.Table = (*DiffTable)(nil)
 var _ sql.FilteredTable = (*DiffTable)(nil)
+var _ sql.IndexedTable = (*DiffTable)(nil)
+var _ sql.ParallelizedIndexAddressableTable = (*DiffTable)(nil)

 type DiffTable struct {
 	name        string
@@ -67,6 +70,9 @@ type DiffTable struct {
 	partitionFilters []sql.Expression
 	rowFilters       []sql.Expression

+	table  *doltdb.Table
+	lookup sql.IndexLookup
+
 	// noms only
 	joiner *rowconv.Joiner
 }
@@ -110,6 +116,7 @@ func NewDiffTable(ctx *sql.Context, tblName string, ddb *doltdb.DoltDB, root *do
 		sqlSch:           sqlSch,
 		partitionFilters: nil,
 		rowFilters:       nil,
+		table:            table,
 		joiner:           j,
 	}, nil
 }
@@ -199,7 +206,26 @@ func (dt *DiffTable) WithFilters(_ *sql.Context, filters []sql.Expression) sql.T

 func (dt *DiffTable) PartitionRows(ctx *sql.Context, part sql.Partition) (sql.RowIter, error) {
 	dp := part.(DiffPartition)
-	return dp.GetRowIter(ctx, dt.ddb, dt.joiner)
+	return dp.GetRowIter(ctx, dt.ddb, dt.joiner, dt.lookup)
+}
+
+func (dt *DiffTable) GetIndexes(ctx *sql.Context) ([]sql.Index, error) {
+	return index.DoltDiffIndexesFromTable(ctx, "", dt.name, dt.table)
+}
+
+func (dt *DiffTable) WithIndexLookup(lookup sql.IndexLookup) sql.Table {
+	if lookup == nil {
+		return dt
+	}
+
+	nt := *dt
+	nt.lookup = lookup
+
+	return &nt
+}
+
+func (dt *DiffTable) ShouldParallelizeAccess() bool {
+	return true
 }

 // tableData returns the map of primary key to values for the specified table (or an empty map if the tbl is null)
@@ -279,11 +305,11 @@ func (dp DiffPartition) Key() []byte {
 	return []byte(dp.toName + dp.fromName)
 }

-func (dp DiffPartition) GetRowIter(ctx *sql.Context, ddb *doltdb.DoltDB, joiner *rowconv.Joiner) (sql.RowIter, error) {
+func (dp DiffPartition) GetRowIter(ctx *sql.Context, ddb *doltdb.DoltDB, joiner *rowconv.Joiner, lookup sql.IndexLookup) (sql.RowIter, error) {
 	if types.IsFormat_DOLT_1(ddb.Format()) {
 		return newProllyDiffIter(ctx, dp, ddb, *dp.fromSch, *dp.toSch)
 	} else {
-		return newNomsDiffIter(ctx, ddb, joiner, dp)
+		return newNomsDiffIter(ctx, ddb, joiner, dp, lookup)
 	}
 }

--- a/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
+++ b/go/libraries/doltcore/sqle/enginetest/dolt_engine_test.go
@@ -152,6 +152,18 @@ func TestQueryPlans(t *testing.T) {
 	enginetest.TestQueryPlans(t, newDoltHarness(t).WithParallelism(1).WithSkippedQueries(skipped))
 }

+func TestDoltDiffQueryPlans(t *testing.T) {
+	skipNewFormat(t)
+	harness := newDoltHarness(t).WithParallelism(2) // want Exchange nodes
+	harness.Setup(setup.SimpleSetup...)
+	e, err := harness.NewEngine(t)
+	require.NoError(t, err)
+	defer e.Close()
+	for _, tt := range DoltDiffPlanTests {
+		enginetest.TestQueryPlan(t, harness, e, tt.Query, tt.ExpectedPlan)
+	}
+}
+
 func TestQueryErrors(t *testing.T) {
 	enginetest.TestQueryErrors(t, newDoltHarness(t))
 }
@@ -281,7 +293,7 @@ func TestDoltUserPrivileges(t *testing.T) {
 						t.Skip()
 					}
 				}
-				enginetest.RunQueryWithContext(t, engine, ctx, statement)
+				enginetest.RunQueryWithContext(t, engine, harness, ctx, statement)
 			}
 			for _, assertion := range script.Assertions {
 				if sh, ok := interface{}(harness).(enginetest.SkippingHarness); ok {
@@ -305,11 +317,11 @@ func TestDoltUserPrivileges(t *testing.T) {

 				if assertion.ExpectedErr != nil {
 					t.Run(assertion.Query, func(t *testing.T) {
-						enginetest.AssertErrWithCtx(t, engine, ctx, assertion.Query, assertion.ExpectedErr)
+						enginetest.AssertErrWithCtx(t, engine, harness, ctx, assertion.Query, assertion.ExpectedErr)
 					})
 				} else if assertion.ExpectedErrStr != "" {
 					t.Run(assertion.Query, func(t *testing.T) {
-						enginetest.AssertErrWithCtx(t, engine, ctx, assertion.Query, nil, assertion.ExpectedErrStr)
+						enginetest.AssertErrWithCtx(t, engine, harness, ctx, assertion.Query, nil, assertion.ExpectedErrStr)
 					})
 				} else {
 					t.Run(assertion.Query, func(t *testing.T) {
--- a/go/libraries/doltcore/sqle/enginetest/dolt_harness.go
+++ b/go/libraries/doltcore/sqle/enginetest/dolt_harness.go
@@ -69,6 +69,7 @@ var _ enginetest.VersionedDBHarness = (*DoltHarness)(nil)
 var _ enginetest.ForeignKeyHarness = (*DoltHarness)(nil)
 var _ enginetest.KeylessTableHarness = (*DoltHarness)(nil)
 var _ enginetest.ReadOnlyDatabaseHarness = (*DoltHarness)(nil)
+var _ enginetest.ValidatingHarness = (*DoltHarness)(nil)

 func newDoltHarness(t *testing.T) *DoltHarness {
 	dEnv := dtestutils.CreateTestEnv()
@@ -445,6 +446,15 @@ func (d *DoltHarness) SnapshotTable(db sql.VersionedDatabase, name string, asOf
 	return nil
 }

+func (d *DoltHarness) ValidateEngine(ctx *sql.Context, e *gms.Engine) (err error) {
+	for _, db := range e.Analyzer.Catalog.AllDatabases(ctx) {
+		if err = ValidateDatabase(ctx, db); err != nil {
+			return err
+		}
+	}
+	return
+}
+
 func dsqleDBsAsSqlDBs(dbs []sqle.Database) []sql.Database {
 	sqlDbs := make([]sql.Database, 0, len(dbs))
 	for _, db := range dbs {
--- a/go/libraries/doltcore/sqle/enginetest/dolt_queries.go
+++ b/go/libraries/doltcore/sqle/enginetest/dolt_queries.go
@@ -1433,6 +1433,39 @@ var DoltBranchScripts = []queries.ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "Create branch from startpoint",
+		SetUpScript: []string{
+			"create table a (x int)",
+			"set @commit1 = (select DOLT_COMMIT('-am', 'add table a'));",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query:    "show tables",
+				Expected: []sql.Row{{"a"}, {"myview"}},
+			},
+			{
+				Query:    "CALL DOLT_CHECKOUT('-b', 'newBranch', 'head~1')",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "show tables",
+				Expected: []sql.Row{{"myview"}},
+			},
+			{
+				Query:    "CALL DOLT_CHECKOUT('-b', 'newBranch2', @commit1)",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "show tables",
+				Expected: []sql.Row{{"a"}, {"myview"}},
+			},
+			{
+				Query:          "CALL DOLT_CHECKOUT('-b', 'otherBranch', 'unknownCommit')",
+				ExpectedErrStr: "fatal: 'unknownCommit' is not a commit and a branch 'otherBranch' cannot be created from it",
+			},
+		},
+	},
 }

 var DoltReset = []queries.ScriptTest{
@@ -1504,7 +1537,7 @@ var DiffSystemTableScriptTests = []queries.ScriptTest{
 				Expected: []sql.Row{{2}},
 			},
 			{
-				Query: "SELECT to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type FROM DOLT_DIFF_t WHERE TO_COMMIT=@Commit1 ORDER BY to_pk;",
+				Query: "SELECT to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type FROM DOLT_DIFF_t WHERE TO_COMMIT=@Commit1 ORDER BY to_pk, to_c2, to_c2, from_pk, from_c1, from_c2, diff_type;",
 				Expected: []sql.Row{
 					{1, 2, 3, nil, nil, nil, "added"},
 					{4, 5, 6, nil, nil, nil, "added"},
@@ -1528,7 +1561,7 @@ var DiffSystemTableScriptTests = []queries.ScriptTest{
 				Expected: []sql.Row{{3}},
 			},
 			{
-				Query: "SELECT to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type FROM DOLT_DIFF_t WHERE TO_COMMIT=@Commit2 ORDER BY to_pk;",
+				Query: "SELECT to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type FROM DOLT_DIFF_t WHERE TO_COMMIT=@Commit2 ORDER BY to_pk, to_c2, to_c2, from_pk, from_c1, from_c2, diff_type;",
 				Expected: []sql.Row{
 					{1, 2, 0, 1, 2, 3, "modified"},
 				},
@@ -1938,6 +1971,76 @@ var DiffSystemTableScriptTests = []queries.ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "selecting to_pk columns",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 int, c2 int);",
+			"insert into t values (1, 2, 3), (4, 5, 6);",
+			"set @Commit1 = (select DOLT_COMMIT('-am', 'first commit'));",
+			"insert into t values (7, 8, 9);",
+			"set @Commit2 = (select DOLT_COMMIT('-am', 'second commit'));",
+			"update t set c1 = 0 where pk > 5;",
+			"set @Commit3 = (select DOLT_COMMIT('-am', 'third commit'));",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query:    "SELECT COUNT(*) FROM DOLT_DIFF_t;",
+				Expected: []sql.Row{{4}},
+			},
+			{
+				Query: "SELECT to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type FROM DOLT_DIFF_t WHERE to_pk = 1 ORDER BY to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type;",
+				Expected: []sql.Row{
+					{1, 2, 3, nil, nil, nil, "added"},
+				},
+			},
+			{
+				Query: "SELECT to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type FROM DOLT_DIFF_t WHERE to_pk > 1 ORDER BY to_pk, to_c1, to_c2, from_pk, from_c1, from_c2, diff_type;",
+				Expected: []sql.Row{
+					{4, 5, 6, nil, nil, nil, "added"},
+					{7, 0, 9, 7, 8, 9, "modified"},
+					{7, 8, 9, nil, nil, nil, "added"},
+				},
+			},
+		},
+	},
+	{
+		Name: "selecting to_pk1 and to_pk2 columns",
+		SetUpScript: []string{
+			"create table t (pk1 int, pk2 int, c1 int, primary key (pk1, pk2));",
+			"insert into t values (1, 2, 3), (4, 5, 6);",
+			"set @Commit1 = (select DOLT_COMMIT('-am', 'first commit'));",
+			"insert into t values (7, 8, 9);",
+			"set @Commit2 = (select DOLT_COMMIT('-am', 'second commit'));",
+			"update t set c1 = 0 where pk1 > 5;",
+			"set @Commit3 = (select DOLT_COMMIT('-am', 'third commit'));",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query:    "SELECT COUNT(*) FROM DOLT_DIFF_t;",
+				Expected: []sql.Row{{4}},
+			},
+			{
+				Query: "SELECT to_pk1, to_pk2, to_c1, from_pk1, from_pk2, from_c1, diff_type FROM DOLT_DIFF_t WHERE to_pk1 = 1 ORDER BY to_pk1, to_pk2, to_c1, from_pk1, from_pk2, from_c1, diff_type;",
+				Expected: []sql.Row{
+					{1, 2, 3, nil, nil, nil, "added"},
+				},
+			},
+			{
+				Query: "SELECT to_pk1, to_pk2, to_c1, from_pk1, from_pk2, from_c1, diff_type FROM DOLT_DIFF_t WHERE to_pk1 = 1 and to_pk2 = 2 ORDER BY to_pk1, to_pk2, to_c1, from_pk1, from_pk2, from_c1, diff_type;",
+				Expected: []sql.Row{
+					{1, 2, 3, nil, nil, nil, "added"},
+				},
+			},
+			{
+				Query: "SELECT to_pk1, to_pk2, to_c1, from_pk1, from_pk2, from_c1, diff_type FROM DOLT_DIFF_t WHERE to_pk1 > 1 and to_pk2 < 10 ORDER BY to_pk1, to_pk2, to_c1, from_pk1, from_pk2, from_c1, diff_type;",
+				Expected: []sql.Row{
+					{4, 5, 6, nil, nil, nil, "added"},
+					{7, 8, 0, 7, 8, 9, "modified"},
+					{7, 8, 9, nil, nil, nil, "added"},
+				},
+			},
+		},
+	},
 }

 var DiffTableFunctionScriptTests = []queries.ScriptTest{
@@ -2842,3 +2945,37 @@ var CommitDiffSystemTableScriptTests = []queries.ScriptTest{
 		},
 	},
 }
+
+// DoltDiffPlanTests are tests that check our query plans for various operations on the dolt diff system tables
+var DoltDiffPlanTests = []queries.QueryPlanTest{
+	{
+		Query: `select * from dolt_diff_one_pk where to_pk=1`,
+		ExpectedPlan: "Exchange(parallelism=2)\n" +
+			" └─ IndexedTableAccess(dolt_diff_one_pk on [dolt_diff_one_pk.to_pk] with ranges: [{[1, 1]}])\n" +
+			"",
+	},
+	{
+		Query: `select * from dolt_diff_one_pk where to_pk>=10 and to_pk<=100`,
+		ExpectedPlan: "Exchange(parallelism=2)\n" +
+			" └─ IndexedTableAccess(dolt_diff_one_pk on [dolt_diff_one_pk.to_pk] with ranges: [{[10, 100]}])\n" +
+			"",
+	},
+	{
+		Query: `select * from dolt_diff_two_pk where to_pk1=1`,
+		ExpectedPlan: "Exchange(parallelism=2)\n" +
+			" └─ IndexedTableAccess(dolt_diff_two_pk on [dolt_diff_two_pk.to_pk1,dolt_diff_two_pk.to_pk2] with ranges: [{[1, 1], (-∞, ∞)}])\n" +
+			"",
+	},
+	{
+		Query: `select * from dolt_diff_two_pk where to_pk1=1 and to_pk2=2`,
+		ExpectedPlan: "Exchange(parallelism=2)\n" +
+			" └─ IndexedTableAccess(dolt_diff_two_pk on [dolt_diff_two_pk.to_pk1,dolt_diff_two_pk.to_pk2] with ranges: [{[1, 1], [2, 2]}])\n" +
+			"",
+	},
+	{
+		Query: `select * from dolt_diff_two_pk where to_pk1 < 1 and to_pk2 > 10`,
+		ExpectedPlan: "Exchange(parallelism=2)\n" +
+			" └─ IndexedTableAccess(dolt_diff_two_pk on [dolt_diff_two_pk.to_pk1,dolt_diff_two_pk.to_pk2] with ranges: [{(-∞, 1), (10, ∞)}])\n" +
+			"",
+	},
+}
--- a/go/libraries/doltcore/sqle/enginetest/validation.go
+++ b/go/libraries/doltcore/sqle/enginetest/validation.go
@@ -0,0 +1,131 @@
+// Copyright 2020 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 (
+	"context"
+	"fmt"
+
+	"github.com/dolthub/go-mysql-server/sql"
+	"github.com/dolthub/go-mysql-server/sql/mysql_db"
+
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
+	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb/durable"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
+	"github.com/dolthub/dolt/go/store/types"
+)
+
+func ValidateDatabase(ctx context.Context, db sql.Database) (err error) {
+	switch tdb := db.(type) {
+	case sqle.Database:
+		return ValidateDoltDatabase(ctx, tdb)
+	case mysql_db.PrivilegedDatabase:
+		return ValidateDatabase(ctx, tdb.Unwrap())
+	default:
+		return nil
+	}
+}
+
+func ValidateDoltDatabase(ctx context.Context, db sqle.Database) (err error) {
+	if !types.IsFormat_DOLT_1(db.GetDoltDB().Format()) {
+		return nil
+	}
+	for _, stage := range validationStages {
+		if err = stage(ctx, db); err != nil {
+			return err
+		}
+	}
+	return
+}
+
+type validator func(ctx context.Context, db sqle.Database) error
+
+var validationStages = []validator{
+	validateChunkReferences,
+}
+
+// validateChunkReferences checks for dangling chunks.
+func validateChunkReferences(ctx context.Context, db sqle.Database) error {
+	validateIndex := func(ctx context.Context, idx durable.Index) error {
+		pm := durable.ProllyMapFromIndex(idx)
+		return pm.WalkNodes(ctx, func(ctx context.Context, nd tree.Node) error {
+			if nd.Size() <= 0 {
+				return fmt.Errorf("encountered nil tree.Node")
+			}
+			return nil
+		})
+	}
+
+	cb := func(n string, t *doltdb.Table, sch schema.Schema) (stop bool, err error) {
+		if sch == nil {
+			return true, fmt.Errorf("expected non-nil schema: %v", sch)
+		}
+
+		rows, err := t.GetRowData(ctx)
+		if err != nil {
+			return true, err
+		}
+		if err = validateIndex(ctx, rows); err != nil {
+			return true, err
+		}
+
+		indexes, err := t.GetIndexSet(ctx)
+		if err != nil {
+			return true, err
+		}
+		err = durable.IterAllIndexes(ctx, sch, indexes, func(_ string, idx durable.Index) error {
+			return validateIndex(ctx, idx)
+		})
+		if err != nil {
+			return true, err
+		}
+		return
+	}
+
+	return iterDatabaseTables(ctx, db, cb)
+}
+
+// iterDatabaseTables is a utility to factor out common validation access patterns.
+func iterDatabaseTables(
+	ctx context.Context,
+	db sqle.Database,
+	cb func(name string, t *doltdb.Table, sch schema.Schema) (bool, error),
+) error {
+	ddb := db.GetDoltDB()
+	branches, err := ddb.GetBranches(ctx)
+	if err != nil {
+		return err
+	}
+
+	for i := range branches {
+		var c *doltdb.Commit
+		var r *doltdb.RootValue
+
+		c, err = ddb.ResolveCommitRef(ctx, branches[i])
+		if err != nil {
+			return err
+		}
+		r, err = c.GetRootValue(ctx)
+		if err != nil {
+			return err
+		}
+		if err = r.IterTables(ctx, cb); err != nil {
+			return err
+		}
+	}
+	return nil
+}
--- a/go/libraries/doltcore/sqle/index/dolt_index.go
+++ b/go/libraries/doltcore/sqle/index/dolt_index.go
@@ -39,6 +39,53 @@ type DoltIndex interface {
 	GetDurableIndexes(*sql.Context, *doltdb.Table) (durable.Index, durable.Index, error)
 }

+func DoltDiffIndexesFromTable(ctx context.Context, db, tbl string, t *doltdb.Table) (indexes []sql.Index, err error) {
+	sch, err := t.GetSchema(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	// Currently, only support diffs on tables with primary keys, panic?
+	if schema.IsKeyless(sch) {
+		return nil, nil
+	}
+
+	// TODO: do this for other indexes?
+	tableRows, err := t.GetRowData(ctx)
+	if err != nil {
+		return nil, err
+	}
+	keyBld := maybeGetKeyBuilder(tableRows)
+
+	// TODO: two primary keys???
+	cols := sch.GetPKCols().GetColumns()
+
+	// add to_ prefix
+	toCols := make([]schema.Column, len(cols))
+	for i, col := range cols {
+		toCols[i] = col
+		toCols[i].Name = "to_" + col.Name
+	}
+
+	// to_ columns
+	toIndex := doltIndex{
+		id:       "PRIMARY",
+		tblName:  doltdb.DoltDiffTablePrefix + tbl,
+		dbName:   db,
+		columns:  toCols,
+		indexSch: sch,
+		tableSch: sch,
+		unique:   true,
+		comment:  "",
+		vrw:      t.ValueReadWriter(),
+		keyBld:   keyBld,
+	}
+
+	// TODO: need to add from_ columns
+
+	return append(indexes, toIndex), nil
+}
+
 func DoltIndexesFromTable(ctx context.Context, db, tbl string, t *doltdb.Table) (indexes []sql.Index, err error) {
 	sch, err := t.GetSchema(ctx)
 	if err != nil {
--- a/go/libraries/doltcore/sqle/index/prolly_fields.go
+++ b/go/libraries/doltcore/sqle/index/prolly_fields.go
@@ -16,7 +16,9 @@ package index

 import (
 	"encoding/json"
+	"errors"
 	"fmt"
+	"math"
 	"time"

 	"github.com/dolthub/go-mysql-server/sql"
@@ -28,6 +30,8 @@ import (
 	"github.com/dolthub/dolt/go/store/val"
 )

+var ErrValueExceededMaxFieldSize = errors.New("value exceeded max field size of 65kb")
+
 // todo(andy): this should go in GMS
 func DenormalizeRow(sch sql.Schema, row sql.Row) (sql.Row, error) {
 	var err error
@@ -201,13 +205,23 @@ func PutField(tb *val.TupleBuilder, i int, v interface{}) error {
 		tb.PutString(i, v.(string))
 	case val.ByteStringEnc:
 		if s, ok := v.(string); ok {
+			if len(s) > math.MaxUint16 {
+				return ErrValueExceededMaxFieldSize
+			}
 			v = []byte(s)
 		}
 		tb.PutByteString(i, v.([]byte))
 	case val.GeometryEnc:
+		geo := serializeGeometry(v)
+		if len(geo) > math.MaxUint16 {
+			return ErrValueExceededMaxFieldSize
+		}
 		tb.PutGeometry(i, serializeGeometry(v))
 	case val.JSONEnc:
 		buf, err := convJson(v)
+		if len(buf) > math.MaxUint16 {
+			return ErrValueExceededMaxFieldSize
+		}
 		if err != nil {
 			return err
 		}
--- a/go/store/diff/diff.go
+++ b/go/store/diff/diff.go
@@ -275,8 +275,8 @@ func (d differ) diffLists(ctx context.Context, p types.Path, v1, v2 types.List)
 }

 func (d differ) diffMaps(ctx context.Context, p types.Path, v1, v2 types.Map) error {
-	trueFunc := func(value types.Value) (bool, error) {
-		return true, nil
+	trueFunc := func(ctx context.Context, value types.Value) (bool, bool, error) {
+		return true, false, nil
 	}

 	return d.diffMapsInRange(ctx, p, v1, v2, nil, trueFunc)
--- a/go/store/prolly/map_test.go
+++ b/go/store/prolly/map_test.go
@@ -95,7 +95,7 @@ func TestNewEmptyNode(t *testing.T) {
 	assert.Equal(t, 0, empty.Level())
 	assert.Equal(t, 0, empty.Count())
 	assert.Equal(t, 0, empty.TreeCount())
-	assert.Equal(t, 72, empty.Size())
+	assert.Equal(t, 75, empty.Size())
 	assert.True(t, empty.IsLeaf())
 }

--- a/go/store/prolly/message/address_map.go
+++ b/go/store/prolly/message/address_map.go
@@ -74,26 +74,47 @@ func (s AddressMapSerializer) Serialize(keys, addrs [][]byte, subtrees []uint64,
 		serial.AddressMapAddTreeCount(b, uint64(len(keys)))
 	}
 	serial.AddressMapAddTreeLevel(b, uint8(level))
-	b.FinishWithFileIdentifier(serial.AddressMapEnd(b), addressMapFileID)
-	return b.FinishedBytes()
+
+	return finishMessage(b, serial.AddressMapEnd(b), addressMapFileID)
+}
+
+func finishMessage(b *fb.Builder, off fb.UOffsetT, fileID []byte) []byte {
+	// We finish the buffer by prefixing it with:
+	// 1) 1 byte NomsKind == TupleRowStorage.
+	// 2) big endian uint16 representing the size of the message, not
+	// including the kind or size prefix bytes.
+	//
+	// This allows chunks we serialize here to be read by types binary
+	// codec.
+	//
+	// All accessors in this package expect this prefix to be on the front
+	// of the message bytes as well. See |messagePrefixSz|.
+
+	b.Prep(1, fb.SizeInt32+4+messagePrefixSz)
+	b.FinishWithFileIdentifier(off, fileID)
+
+	bytes := b.Bytes[b.Head()-messagePrefixSz:]
+	bytes[0] = byte(MessageTypesKind)
+	binary.BigEndian.PutUint16(bytes[1:], uint16(len(b.Bytes)-int(b.Head())))
+	return bytes
 }

 func getAddressMapKeys(msg Message) (keys val.SlicedBuffer) {
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	keys.Buf = am.KeyItemsBytes()
 	keys.Offs = getAddressMapKeyOffsets(am)
 	return
 }

 func getAddressMapValues(msg Message) (values val.SlicedBuffer) {
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	values.Buf = am.AddressArrayBytes()
 	values.Offs = offsetsForAddressArray(values.Buf)
 	return
 }

 func walkAddressMapAddresses(ctx context.Context, msg Message, cb func(ctx context.Context, addr hash.Hash) error) error {
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	arr := am.AddressArrayBytes()
 	for i := 0; i < len(arr)/hash.ByteLen; i++ {
 		addr := hash.New(arr[i*addrSize : (i+1)*addrSize])
@@ -105,7 +126,7 @@ func walkAddressMapAddresses(ctx context.Context, msg Message, cb func(ctx conte
 }

 func getAddressMapCount(msg Message) uint16 {
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	if am.KeyItemsLength() == 0 {
 		return 0
 	}
@@ -114,18 +135,18 @@ func getAddressMapCount(msg Message) uint16 {
 }

 func getAddressMapTreeLevel(msg Message) int {
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	return int(am.TreeLevel())
 }

 func getAddressMapTreeCount(msg Message) int {
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	return int(am.TreeCount())
 }

 func getAddressMapSubtrees(msg Message) []uint64 {
 	counts := make([]uint64, getAddressMapCount(msg))
-	am := serial.GetRootAsAddressMap(msg, 0)
+	am := serial.GetRootAsAddressMap(msg, messagePrefixSz)
 	return decodeVarints(am.SubtreeCountsBytes(), counts)
 }

@@ -149,5 +170,6 @@ func estimateAddressMapSize(keys, addresses [][]byte, subtrees []uint64) (keySz,
 	totalSz += len(subtrees) * binary.MaxVarintLen64
 	totalSz += 8 + 1 + 1 + 1
 	totalSz += 72
+	totalSz += messagePrefixSz
 	return
 }
--- a/go/store/prolly/message/message.go
+++ b/go/store/prolly/message/message.go
@@ -23,6 +23,10 @@ import (
 	"github.com/dolthub/dolt/go/store/val"
 )

+const MessageTypesKind int = 28
+
+const messagePrefixSz = 3
+
 type Message []byte

 type Serializer interface {
@@ -30,7 +34,7 @@ type Serializer interface {
 }

 func GetKeysAndValues(msg Message) (keys, values val.SlicedBuffer, cnt uint16) {
-	id := serial.GetFileID(msg)
+	id := serial.GetFileID(msg[messagePrefixSz:])

 	if id == serial.ProllyTreeNodeFileID {
 		return getProllyMapKeysAndValues(msg)
@@ -46,7 +50,7 @@ func GetKeysAndValues(msg Message) (keys, values val.SlicedBuffer, cnt uint16) {
 }

 func WalkAddresses(ctx context.Context, msg Message, cb func(ctx context.Context, addr hash.Hash) error) error {
-	id := serial.GetFileID(msg)
+	id := serial.GetFileID(msg[messagePrefixSz:])
 	switch id {
 	case serial.ProllyTreeNodeFileID:
 		return walkProllyMapAddresses(ctx, msg, cb)
@@ -58,7 +62,7 @@ func WalkAddresses(ctx context.Context, msg Message, cb func(ctx context.Context
 }

 func GetTreeLevel(msg Message) int {
-	id := serial.GetFileID(msg)
+	id := serial.GetFileID(msg[messagePrefixSz:])
 	switch id {
 	case serial.ProllyTreeNodeFileID:
 		return getProllyMapTreeLevel(msg)
@@ -70,7 +74,7 @@ func GetTreeLevel(msg Message) int {
 }

 func GetTreeCount(msg Message) int {
-	id := serial.GetFileID(msg)
+	id := serial.GetFileID(msg[messagePrefixSz:])
 	switch id {
 	case serial.ProllyTreeNodeFileID:
 		return getProllyMapTreeCount(msg)
@@ -82,7 +86,7 @@ func GetTreeCount(msg Message) int {
 }

 func GetSubtrees(msg Message) []uint64 {
-	id := serial.GetFileID(msg)
+	id := serial.GetFileID(msg[messagePrefixSz:])
 	switch id {
 	case serial.ProllyTreeNodeFileID:
 		return getProllyMapSubtrees(msg)
--- a/go/store/prolly/message/prolly_map.go
+++ b/go/store/prolly/message/prolly_map.go
@@ -85,12 +85,12 @@ func (s ProllyMapSerializer) Serialize(keys, values [][]byte, subtrees []uint64,
 	serial.ProllyTreeNodeAddKeyType(b, serial.ItemTypeTupleFormatAlpha)
 	serial.ProllyTreeNodeAddValueType(b, serial.ItemTypeTupleFormatAlpha)
 	serial.ProllyTreeNodeAddTreeLevel(b, uint8(level))
-	b.FinishWithFileIdentifier(serial.ProllyTreeNodeEnd(b), prollyMapFileID)
-	return b.FinishedBytes()
+
+	return finishMessage(b, serial.ProllyTreeNodeEnd(b), prollyMapFileID)
 }

 func getProllyMapKeysAndValues(msg Message) (keys, values val.SlicedBuffer, cnt uint16) {
-	pm := serial.GetRootAsProllyTreeNode(msg, 0)
+	pm := serial.GetRootAsProllyTreeNode(msg, messagePrefixSz)

 	keys.Buf = pm.KeyItemsBytes()
 	keys.Offs = getProllyMapKeyOffsets(pm)
@@ -113,7 +113,7 @@ func getProllyMapKeysAndValues(msg Message) (keys, values val.SlicedBuffer, cnt
 }

 func walkProllyMapAddresses(ctx context.Context, msg Message, cb func(ctx context.Context, addr hash.Hash) error) error {
-	pm := serial.GetRootAsProllyTreeNode(msg, 0)
+	pm := serial.GetRootAsProllyTreeNode(msg, messagePrefixSz)
 	arr := pm.AddressArrayBytes()
 	for i := 0; i < len(arr)/hash.ByteLen; i++ {
 		addr := hash.New(arr[i*addrSize : (i+1)*addrSize])
@@ -136,7 +136,7 @@ func walkProllyMapAddresses(ctx context.Context, msg Message, cb func(ctx contex
 }

 func getProllyMapCount(msg Message) uint16 {
-	pm := serial.GetRootAsProllyTreeNode(msg, 0)
+	pm := serial.GetRootAsProllyTreeNode(msg, messagePrefixSz)
 	if pm.KeyItemsLength() == 0 {
 		return 0
 	}
@@ -145,18 +145,18 @@ func getProllyMapCount(msg Message) uint16 {
 }

 func getProllyMapTreeLevel(msg Message) int {
-	pm := serial.GetRootAsProllyTreeNode(msg, 0)
+	pm := serial.GetRootAsProllyTreeNode(msg, messagePrefixSz)
 	return int(pm.TreeLevel())
 }

 func getProllyMapTreeCount(msg Message) int {
-	pm := serial.GetRootAsProllyTreeNode(msg, 0)
+	pm := serial.GetRootAsProllyTreeNode(msg, messagePrefixSz)
 	return int(pm.TreeCount())
 }

 func getProllyMapSubtrees(msg Message) []uint64 {
 	counts := make([]uint64, getProllyMapCount(msg))
-	pm := serial.GetRootAsProllyTreeNode(msg, 0)
+	pm := serial.GetRootAsProllyTreeNode(msg, messagePrefixSz)
 	return decodeVarints(pm.SubtreeCountsBytes(), counts)
 }

@@ -204,6 +204,7 @@ func estimateProllyMapSize(keys, values [][]byte, subtrees []uint64) (keySz, val
 	bufSz += 8 + 1 + 1 + 1               // metadata
 	bufSz += 72                          // vtable (approx)
 	bufSz += 100                         // padding?
+	bufSz += messagePrefixSz

 	return
 }
--- a/go/store/prolly/tree/content_address_test.go
+++ b/go/store/prolly/tree/content_address_test.go
@@ -27,10 +27,11 @@ import (
 )

 var goldenHash = hash.Hash{
-	0xcb, 0x1d, 0xe1, 0xbd, 0x4a,
-	0x3c, 0x27, 0x18, 0xe, 0xba,
-	0xb7, 0xe8, 0x30, 0x44, 0x69,
-	0x3d, 0x86, 0x27, 0x26, 0x2d,
+	0x39, 0x1c, 0xcb, 0xd8,
+	0xea, 0xd2, 0xdc, 0x42,
+	0x76, 0xb7, 0x38, 0xf1,
+	0x0d, 0x4f, 0x48, 0x91,
+	0x7d, 0x1f, 0xb7, 0xb4,
 }

 // todo(andy): need and analogous test in pkg prolly
--- a/go/store/prolly/tree/immutable_tree.go
+++ b/go/store/prolly/tree/immutable_tree.go
@@ -0,0 +1,150 @@
+// 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 tree
+
+import (
+	"context"
+	"errors"
+	"io"
+
+	"github.com/dolthub/dolt/go/store/prolly/message"
+)
+
+const defaultFixedChunkLength = 4000
+
+var ErrInvalidChunkSize = errors.New("invalid chunkSize; value must be > 1")
+
+// buildImmutableTree writes the contents of |reader| as an append-only
+// tree, returning the root node or an error if applicable. |chunkSize|
+// fixes the split size of leaf and intermediate node chunks.
+func buildImmutableTree(ctx context.Context, r io.Reader, ns NodeStore, S message.Serializer, chunkSize int) (Node, error) {
+	if chunkSize <= 1 {
+		return Node{}, ErrInvalidChunkSize
+	}
+
+	var levels [][]novelNode
+	var levelCnts []int
+	var finalize bool
+
+	// We use lookahead to check whether the reader has
+	// more bytes. The reader will only EOF when reading
+	// zero bytes into the lookahead buffer, but we want
+	// to know at the beginning of a loop whether we are
+	// finished.
+	lookahead := make([]byte, chunkSize)
+	lookaheadN, err := r.Read(lookahead)
+	if err != nil {
+		return Node{}, err
+	}
+
+	buf := make([]byte, chunkSize)
+	for {
+		copy(buf, lookahead)
+		curN := lookaheadN
+		lookaheadN, err = r.Read(lookahead)
+		if err == io.EOF {
+			finalize = true
+		} else if err != nil {
+			return Node{}, err
+		}
+
+		novel, err := _newLeaf(ctx, ns, S, buf[:curN])
+		if err != nil {
+			return Node{}, err
+		}
+
+		i := 0
+		for {
+			// Three cases for building tree
+			// 1) reached new level => create new level
+			// 2) add novel node to current level
+			// 3) we didn't fill the current level => break
+			// 4) we filled current level, chunk and recurse into parent
+			//
+			// Two cases for finalizing tree
+			// 1) we haven't hit root, so we add the final chunk, finalize level, and continue upwards
+			// 2) we overshot root finalizing chunks, and we return the single root in the lower level
+			if i > len(levels)-1 {
+				levels = append(levels, make([]novelNode, chunkSize))
+				levelCnts = append(levelCnts, 0)
+			}
+
+			levels[i][levelCnts[i]] = novel
+			levelCnts[i]++
+			if levelCnts[i] < chunkSize {
+				// current level is not full
+				if !finalize {
+					// only continue and chunk this level if finalizing all in-progress nodes
+					break
+				}
+			}
+
+			nodes := levels[i][:levelCnts[i]]
+			if len(nodes) == 1 && i == len(levels)-1 {
+				// this is necessary and only possible if we're finalizing
+				// note: this is the only non-error return
+				return nodes[0].node, nil
+			}
+
+			// chunk the current level
+			novel, err = _newInternal(ctx, ns, S, nodes, i+1, chunkSize)
+			if err != nil {
+				return Node{}, err
+			}
+			levelCnts[i] = 0
+			i++
+		}
+	}
+}
+
+func _newInternal(ctx context.Context, ns NodeStore, s message.Serializer, nodes []novelNode, level int, chunkSize int) (novelNode, error) {
+	keys := make([][]byte, len(nodes))
+	vals := make([][]byte, len(nodes))
+	subtrees := make([]uint64, len(nodes))
+	treeCnt := uint64(0)
+	for i := range nodes {
+		keys[i] = []byte{0}
+		vals[i] = nodes[i].addr[:]
+		subtrees[i] = nodes[i].treeCount
+		treeCnt += nodes[i].treeCount
+	}
+	msg := s.Serialize(keys, vals, subtrees, level)
+	node := NodeFromBytes(msg)
+	addr, err := ns.Write(ctx, node)
+	if err != nil {
+		return novelNode{}, err
+	}
+	return novelNode{
+		addr:      addr,
+		node:      node,
+		lastKey:   []byte{0},
+		treeCount: treeCnt,
+	}, nil
+}
+
+func _newLeaf(ctx context.Context, ns NodeStore, s message.Serializer, buf []byte) (novelNode, error) {
+	msg := s.Serialize([][]byte{{0}}, [][]byte{buf}, []uint64{1}, 0)
+	node := NodeFromBytes(msg)
+	addr, err := ns.Write(ctx, node)
+	if err != nil {
+		return novelNode{}, err
+	}
+	return novelNode{
+		addr:      addr,
+		node:      node,
+		lastKey:   []byte{0},
+		treeCount: 1,
+	}, nil
+}
--- a/go/store/prolly/tree/immutable_tree_test.go
+++ b/go/store/prolly/tree/immutable_tree_test.go
@@ -0,0 +1,198 @@
+// 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 tree
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/dolthub/dolt/go/store/prolly/message"
+)
+
+func TestWriteImmutableTree(t *testing.T) {
+	tests := []struct {
+		inputSize int
+		chunkSize int
+		err       error
+	}{
+		{
+			inputSize: 100,
+			chunkSize: 5,
+		},
+		{
+			inputSize: 100,
+			chunkSize: 100,
+		},
+		{
+			inputSize: 100,
+			chunkSize: 101,
+		},
+		{
+			inputSize: 255,
+			chunkSize: 5,
+		},
+		{
+			inputSize: 243,
+			chunkSize: 5,
+		},
+		{
+			inputSize: 47,
+			chunkSize: 3,
+		},
+		{
+			inputSize: 200,
+			chunkSize: 7,
+		},
+		{
+			inputSize: 200,
+			chunkSize: 40,
+		},
+		{
+			inputSize: 1,
+			chunkSize: 5,
+		},
+		{
+			inputSize: 20,
+			chunkSize: 500,
+		},
+		{
+			inputSize: 10,
+			chunkSize: 1,
+			err:       ErrInvalidChunkSize,
+		},
+		{
+			inputSize: 10,
+			chunkSize: -1,
+			err:       ErrInvalidChunkSize,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("inputSize=%d; chunkSize=%d", tt.inputSize, tt.chunkSize), func(t *testing.T) {
+			buf := make([]byte, tt.inputSize)
+			for i := range buf {
+				buf[i] = byte(i)
+			}
+			ctx := context.Background()
+			r := bytes.NewReader(buf)
+			ns := NewTestNodeStore()
+			serializer := message.ProllyMapSerializer{Pool: ns.Pool()}
+			root, err := buildImmutableTree(ctx, r, ns, serializer, tt.chunkSize)
+			if tt.err != nil {
+				require.True(t, errors.Is(err, tt.err))
+				return
+			}
+			require.NoError(t, err)
+
+			expSubtrees := expectedSubtrees(tt.inputSize, tt.chunkSize)
+			expLevel := expectedLevel(tt.inputSize, tt.chunkSize)
+			expSum := expectedSum(tt.inputSize)
+			expUnfilled := expectedUnfilled(tt.inputSize, tt.chunkSize)
+
+			unfilledCnt := 0
+			sum := 0
+			byteCnt := 0
+			WalkNodes(ctx, root, ns, func(ctx context.Context, n Node) error {
+				var keyCnt int
+				if n.IsLeaf() {
+					byteCnt += len(n.values.Buf)
+					for _, i := range n.getValue(0) {
+						sum += int(i)
+					}
+					keyCnt = len(n.values.Buf)
+				} else {
+					keyCnt = n.Count()
+				}
+				if keyCnt != tt.chunkSize {
+					unfilledCnt += 1
+				}
+				return nil
+			})
+
+			require.Equal(t, expLevel, root.Level())
+			require.Equal(t, expSum, sum)
+			require.Equal(t, tt.inputSize, byteCnt)
+			require.Equal(t, expUnfilled, unfilledCnt)
+			require.Equal(t, expSubtrees, root.getSubtreeCounts())
+		})
+	}
+}
+
+func expectedLevel(size, chunk int) int {
+	l := 0
+	for size > chunk {
+		size = size / chunk
+		l += 1
+	}
+	return l
+}
+
+func expectedSubtrees(size, chunk int) SubtreeCounts {
+	if size <= chunk {
+		return SubtreeCounts{0}
+	}
+	size = int(math.Ceil(float64(size) / float64(chunk)))
+	l := chunk
+	for l < size {
+		l *= chunk
+	}
+	l /= chunk
+
+	res := make(SubtreeCounts, 0)
+	for size > l {
+		res = append(res, uint64(l))
+		size -= l
+	}
+	res = append(res, uint64(size))
+
+	return res
+}
+
+func expectedSum(size int) int {
+	return (size * (size + 1) / 2) - size
+}
+
+func expectedUnfilled(size, chunk int) int {
+	l := chunk
+	for l < size {
+		l *= chunk
+	}
+	l /= chunk
+	size -= l
+	cnt := 0
+	i := 1
+	for size > 0 {
+		if l > size {
+			if i < chunk-1 {
+				cnt += 1
+			}
+			l /= chunk
+			i = 0
+		} else {
+			size -= l
+			i++
+		}
+	}
+	if i < chunk {
+		cnt += 1
+	}
+	return cnt
+}
--- a/go/store/prolly/tree/node_test.go
+++ b/go/store/prolly/tree/node_test.go
@@ -15,6 +15,7 @@
 package tree

 import (
+	"context"
 	"math"
 	"math/rand"
 	"testing"
@@ -23,7 +24,10 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

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

@@ -66,6 +70,43 @@ func TestNodeSize(t *testing.T) {
 	assert.Equal(t, 128, int(sz))
 }

+func TestNodeHashValueCompatibility(t *testing.T) {
+	keys, values := randomNodeItemPairs(t, (rand.Int()%101)+50)
+	nd := newLeafNode(keys, values)
+	nbf := types.Format_DOLT_1
+	th, err := ValueFromNode(nd).Hash(nbf)
+	require.NoError(t, err)
+	assert.Equal(t, nd.HashOf(), th)
+
+	h1 := hash.Parse("kvup5vdur99ush7c18g0kjc6rhdkfdgo")
+	h2 := hash.Parse("7e54ill10nji9oao1ja88buh9itaj7k9")
+	msg := message.AddressMapSerializer{Pool: sharedPool}.Serialize(
+		[][]byte{[]byte("chopin"), []byte("listz")},
+		[][]byte{h1[:], h2[:]},
+		[]uint64{},
+		0)
+	nd = NodeFromBytes(msg)
+	th, err = ValueFromNode(nd).Hash(nbf)
+	require.NoError(t, err)
+	assert.Equal(t, nd.HashOf(), th)
+}
+
+func TestNodeDecodeValueCompatibility(t *testing.T) {
+	keys, values := randomNodeItemPairs(t, (rand.Int()%101)+50)
+	nd := newLeafNode(keys, values)
+
+	ts := &chunks.TestStorage{}
+	cs := ts.NewView()
+	ns := NewNodeStore(cs)
+	vs := types.NewValueStore(cs)
+	h, err := ns.Write(context.Background(), nd)
+	require.NoError(t, err)
+
+	v, err := vs.ReadValue(context.Background(), h)
+	require.NoError(t, err)
+	assert.Equal(t, nd.bytes(), []byte(v.(types.TupleRowStorage)))
+}
+
 func randomNodeItemPairs(t *testing.T, count int) (keys, values []Item) {
 	keys = make([]Item, count)
 	for i := range keys {
--- a/go/store/types/codec.go
+++ b/go/store/types/codec.go
@@ -325,6 +325,14 @@ func (b *binaryNomsReader) ReadInlineBlob() []byte {
 	return bytes
 }

+func (b *binaryNomsReader) readTupleRowStorage() []byte {
+	size := uint32(b.readUint16())
+	// start at offset-3, to include the kind byte + Uint16 for size...
+	bytes := b.buff[b.offset-3 : b.offset+size]
+	b.offset += size
+	return bytes
+}
+
 func (b *binaryNomsReader) ReadUUID() uuid.UUID {
 	id := uuid.UUID{}
 	copy(id[:uuidNumBytes], b.readBytes(uuidNumBytes))
--- a/go/store/types/map.go
+++ b/go/store/types/map.go
@@ -32,7 +32,8 @@ import (
 	"github.com/dolthub/dolt/go/store/hash"
 )

-type ValueInRange func(Value) (bool, error)
+//type ValueInRange func(Value) (bool, error)
+type ValueInRange func(context.Context, Value) (bool, bool, error)

 var ErrKeysNotOrdered = errors.New("streaming map keys not ordered")

@@ -203,8 +204,8 @@ func (m Map) Diff(ctx context.Context, last Map, changes chan<- ValueChanged) er
 // streaming approach, optimised for returning results early, but not
 // completing quickly.
 func (m Map) DiffLeftRight(ctx context.Context, last Map, changes chan<- ValueChanged) error {
-	trueFunc := func(Value) (bool, error) {
-		return true, nil
+	trueFunc := func(context.Context, Value) (bool, bool, error) {
+		return true, false, nil
 	}
 	return m.DiffLeftRightInRange(ctx, last, nil, trueFunc, changes)
 }
--- a/go/store/types/noms_kind.go
+++ b/go/store/types/noms_kind.go
@@ -21,6 +21,10 @@

 package types

+import (
+	"github.com/dolthub/dolt/go/store/prolly/message"
+)
+
 // NomsKind allows a TypeDesc to indicate what kind of type is described.
 type NomsKind uint8

@@ -125,6 +129,10 @@ func init() {
 	SupportedKinds[PolygonKind] = true
 	SupportedKinds[SerialMessageKind] = true
 	SupportedKinds[TupleRowStorageKind] = true
+
+	if message.MessageTypesKind != int(TupleRowStorageKind) {
+		panic("internal error: message.MessageTypesKind != TupleRowStorageKind")
+	}
 }

 var KindToTypeSlice []Value
--- a/go/store/types/ordered_sequences_diff.go
+++ b/go/store/types/ordered_sequences_diff.go
@@ -52,8 +52,8 @@ func sendChange(ctx context.Context, changes chan<- ValueChanged, change ValueCh
 // Streams the diff from |last| to |current| into |changes|, using a left-right approach.
 // Left-right immediately descends to the first change and starts streaming changes, but compared to top-down it's serial and much slower to calculate the full diff.
 func orderedSequenceDiffLeftRight(ctx context.Context, last orderedSequence, current orderedSequence, changes chan<- ValueChanged) error {
-	trueFunc := func(Value) (bool, error) {
-		return true, nil
+	trueFunc := func(context.Context, Value) (bool, bool, error) {
+		return true, false, nil
 	}
 	return orderedSequenceDiffLeftRightInRange(ctx, last, current, emptyKey, trueFunc, changes)
 }
@@ -99,20 +99,26 @@ VALIDRANGES:
 			if isLess, err := currentKey.Less(last.format(), lastKey); err != nil {
 				return err
 			} else if isLess {
-				isInRange, err := inRange(currentKey.v)
+				valid, skip, err := inRange(ctx, currentKey.v)
 				if err != nil {
 					return err
-				} else if !isInRange {
+				}
+
+				// Out of range
+				if !valid {
 					break VALIDRANGES
 				}

-				mv, err := getMapValue(currentCur)
-				if err != nil {
-					return err
-				}
+				// Not skipping this value (want to skip for non-inclusive lower bound range)
+				if !skip {
+					mv, err := getMapValue(currentCur)
+					if err != nil {
+						return err
+					}

-				if err := sendChange(ctx, changes, ValueChanged{DiffChangeAdded, currentKey.v, nil, mv}); err != nil {
-					return err
+					if err := sendChange(ctx, changes, ValueChanged{DiffChangeAdded, currentKey.v, nil, mv}); err != nil {
+						return err
+					}
 				}

 				_, err = currentCur.advance(ctx)
@@ -120,13 +126,25 @@ VALIDRANGES:
 					return err
 				}
 			} else {
-				isInRange, err := inRange(lastKey.v)
-				if !isInRange {
+				valid, skip, err := inRange(ctx, lastKey.v)
+				if err != nil {
 					return err
-				} else if !isInRange {
+				}
+
+				// Out of range
+				if !valid {
 					break VALIDRANGES
 				}

+				// Skip this last key
+				if skip {
+					_, err = lastCur.advance(ctx)
+					if err != nil {
+						return err
+					}
+					continue
+				}
+
 				if isLess, err := lastKey.Less(last.format(), currentKey); err != nil {
 					return err
 				} else if isLess {
@@ -178,20 +196,24 @@ VALIDRANGES:
 			return err
 		}

-		isInRange, err := inRange(lastKey.v)
+		valid, skip, err := inRange(ctx, lastKey.v)
 		if err != nil {
 			return err
-		} else if !isInRange {
+		}
+
+		if !valid {
 			break
 		}

-		mv, err := getMapValue(lastCur)
-		if err != nil {
-			return err
-		}
+		if !skip {
+			mv, err := getMapValue(lastCur)
+			if err != nil {
+				return err
+			}

-		if err := sendChange(ctx, changes, ValueChanged{DiffChangeRemoved, lastKey.v, mv, nil}); err != nil {
-			return err
+			if err := sendChange(ctx, changes, ValueChanged{DiffChangeRemoved, lastKey.v, mv, nil}); err != nil {
+				return err
+			}
 		}

 		_, err = lastCur.advance(ctx)
@@ -206,20 +228,24 @@ VALIDRANGES:
 			return err
 		}

-		isInRange, err := inRange(currKey.v)
+		valid, skip, err := inRange(ctx, currKey.v)
 		if err != nil {
 			return err
-		} else if !isInRange {
+		}
+
+		if !valid {
 			break
 		}

-		mv, err := getMapValue(currentCur)
-		if err != nil {
-			return err
-		}
+		if !skip {
+			mv, err := getMapValue(currentCur)
+			if err != nil {
+				return err
+			}

-		if err := sendChange(ctx, changes, ValueChanged{DiffChangeAdded, currKey.v, nil, mv}); err != nil {
-			return err
+			if err := sendChange(ctx, changes, ValueChanged{DiffChangeAdded, currKey.v, nil, mv}); err != nil {
+				return err
+			}
 		}

 		_, err = currentCur.advance(ctx)
--- a/go/store/types/tuplerowstorage.go
+++ b/go/store/types/tuplerowstorage.go
@@ -18,16 +18,20 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
-	"fmt"
-	"math"
 	"strings"

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

-// TupleRowStorage is a clone of InlineBlob. It only exists to be able to easily differentiate these two very different
-// use cases during the migration from the old storage format to the new one.
+// TupleRowStorage cribs its implementation from InlineBlob. It bridges
+// prolly/message byte arrays between types.Value and prolly/tree.Node.
+//
+// Unlike SerialMessage, the byte array held in TupleRowStorage includes the
+// NomsKind byte and the BigEndian uint16 size of the message. |writeTo| is
+// simply a call through to writeRaw, and |readFrom| has to pick up bytes from
+// the reader that have already been "read" to determine kind and size.
+
 type TupleRowStorage []byte

 func (v TupleRowStorage) Value(ctx context.Context) (Value, error) {
@@ -81,23 +85,12 @@ func (v TupleRowStorage) valueReadWriter() ValueReadWriter {
 }

 func (v TupleRowStorage) writeTo(w nomsWriter, nbf *NomsBinFormat) error {
-	byteLen := len(v)
-	if byteLen > math.MaxUint16 {
-		return fmt.Errorf("TupleRowStorage has length %v when max is %v", byteLen, math.MaxUint16)
-	}
-
-	err := TupleRowStorageKind.writeTo(w, nbf)
-	if err != nil {
-		return err
-	}
-
-	w.writeUint16(uint16(byteLen))
 	w.writeRaw(v)
 	return nil
 }

 func (v TupleRowStorage) readFrom(nbf *NomsBinFormat, b *binaryNomsReader) (Value, error) {
-	bytes := b.ReadInlineBlob()
+	bytes := b.readTupleRowStorage()
 	return TupleRowStorage(bytes), nil
 }

--- a/integration-tests/bats/auto_increment.bats
+++ b/integration-tests/bats/auto_increment.bats
@@ -3,7 +3,6 @@ load $BATS_TEST_DIRNAME/helper/common.bash

 setup() {
    setup_common
-    skip_nbf_dolt_1

    dolt sql <<SQL
 CREATE TABLE test (
@@ -435,6 +434,8 @@ SQL
 }

@test "auto_increment: dolt_merge() works with no auto increment overlap" {
+    skip_nbf_dolt_1
+    
    dolt sql <<SQL
 CREATE TABLE t (
    pk int PRIMARY KEY AUTO_INCREMENT,
@@ -499,6 +500,8 @@ SQL
 }

@test "auto_increment: dolt_merge() with a gap in an auto increment key" {
+    skip_nbf_dolt_1
+    
    dolt sql <<SQL
 CREATE TABLE t (
    pk int PRIMARY KEY AUTO_INCREMENT,
--- a/integration-tests/bats/branch.bats
+++ b/integration-tests/bats/branch.bats
@@ -11,7 +11,6 @@ teardown() {
 }

@test "branch: deleting a branch deletes its working set" {
-    skip_nbf_dolt_1
    dolt checkout -b to_delete

    root=$(noms root .dolt/noms)
--- a/integration-tests/bats/column_tags.bats
+++ b/integration-tests/bats/column_tags.bats
@@ -11,7 +11,6 @@ teardown() {
 }

@test "column_tags: Renaming a column should preserve the tag number" {
-    skip_nbf_dolt_1
    dolt sql <<SQL
 CREATE TABLE test (
  pk BIGINT NOT NULL,
@@ -60,7 +59,6 @@ SQL


@test "column_tags: Merging two branches that added same tag, name, type, and constraints" {
-    skip_nbf_dolt_1
    dolt sql <<SQL
 CREATE TABLE test (
  pk BIGINT NOT NULL,
@@ -85,7 +83,6 @@ SQL
 }

@test "column_tags: Merging branches that use the same tag referring to different schema fails" {
-    skip_nbf_dolt_1
    dolt sql <<SQL
 CREATE TABLE test (
  pk BIGINT NOT NULL COMMENT 'tag:1234',
@@ -111,7 +108,6 @@ SQL
 }

@test "column_tags: Merging branches that use the same tag referring to different column names fails" {
-    skip_nbf_dolt_1
    dolt sql <<SQL
 CREATE TABLE test (
  pk BIGINT NOT NULL COMMENT 'tag:1234',
@@ -138,7 +134,6 @@ SQL
 }

@test "column_tags: Merging branches that both created the same column succeeds" {
-    skip_nbf_dolt_1
    dolt sql <<SQL
 CREATE TABLE test (
  pk BIGINT NOT NULL COMMENT 'tag:0',
--- a/integration-tests/bats/default-values.bats
+++ b/integration-tests/bats/default-values.bats
@@ -3,7 +3,6 @@ load $BATS_TEST_DIRNAME/helper/common.bash

 setup() {
    setup_common
-    skip_nbf_dolt_1

    dolt sql <<SQL
 CREATE TABLE parent (
--- a/integration-tests/bats/helper/common.bash
+++ b/integration-tests/bats/helper/common.bash
@@ -24,6 +24,11 @@ set_dolt_user() {
    dolt config --global --add user.email "$2" > /dev/null 2>&1
 }

+unset_dolt_user() {
+  dolt config --global --unset user.name
+  dolt config --global --unset user.email
+}
+
 current_dolt_user_name() {
    dolt config --global --get user.name
 }
@@ -52,6 +57,9 @@ assert_feature_version() {
    # command, don't check its output in that case
    if [ "$status" -eq 0 ]; then
        [[ "$output" =~ "feature version: 3" ]] || exit 1
+    else
+      # Clear status to avoid BATS failing if this is the last run command
+      status=0
    fi
 }

--- a/integration-tests/bats/init.bats
+++ b/integration-tests/bats/init.bats
@@ -0,0 +1,159 @@
+#!/usr/bin/env bats
+load $BATS_TEST_DIRNAME/helper/common.bash
+
+setup() {
+    setup_no_dolt_init
+    stash_current_dolt_user
+}
+
+teardown() {
+    restore_stashed_dolt_user
+    assert_feature_version
+    teardown_common
+}
+
+@test "init: implicit global configuration" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init
+  [ "$status" -eq 0 ]
+
+  run dolt config --local --get user.name
+  [ "$status" -eq 1 ]
+
+  run dolt config --local --get user.email
+  [ "$status" -eq 1 ]
+
+  assert_valid_repository
+}
+
+@test "init: explicit local configuration for name" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init --name foo
+  [ "$status" -eq 0 ]
+
+  run dolt config --local --get user.name
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "foo" ]] || false
+
+  run dolt config --local --get user.email
+  [ "$status" -eq 1 ]
+
+  assert_valid_repository
+}
+
+@test "init: explicit local configuration for email" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init --email foo@bar.com
+  [ "$status" -eq 0 ]
+
+  run dolt config --local --get user.name
+  [ "$status" -eq 1 ]
+
+  run dolt config --local --get user.email
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "foo@bar.com" ]] || false
+
+  assert_valid_repository
+}
+
+@test "init: explicit local configuration for name and email" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init --name foo --email foo@bar.com
+  [ "$status" -eq 0 ]
+
+  run dolt config --local --get user.name
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "foo" ]] || false
+
+  run dolt config --local --get user.email
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "foo@bar.com" ]] || false
+
+  assert_valid_repository
+}
+
+@test "init: explicit local configuration for name and email with no global config" {
+  unset_dolt_user
+
+  run dolt init --name foo --email foo@bar.com
+  [ "$status" -eq 0 ]
+
+  run dolt config --local --get user.name
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "foo" ]] || false
+
+  run dolt config --local --get user.email
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "foo@bar.com" ]] || false
+
+  assert_valid_repository
+}
+
+@test "init: no explicit or implicit configuration for name and email" {
+  unset_dolt_user
+
+  run dolt init
+  [ "$status" -eq 1 ]
+  [[ "$output" =~ "Author identity unknown" ]] || false
+}
+
+@test "init: implicit default initial branch" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init
+  [ "$status" -eq 0 ]
+
+  run dolt branch --show-current
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "main" ]] || false
+
+  assert_valid_repository
+}
+
+@test "init: implicit global initial branch" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt config --global -add init.defaultbranch globalInitialBranch
+
+  run dolt init
+  [ "$status" -eq 0 ]
+
+  run dolt branch --show-current
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "globalInitialBranch" ]] || false
+
+  assert_valid_repository
+}
+
+@test "init: explicit initial branch" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init -b initialBranch
+  [ "$status" -eq 0 ]
+
+  run dolt branch --show-current
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "initialBranch" ]] || false
+
+  assert_valid_repository
+}
+
+@test "init: running init in existing Dolt directory fails" {
+  set_dolt_user "baz", "baz@bash.com"
+
+  run dolt init
+  [ "$status" -eq 0 ]
+
+  run dolt init
+  [ "$status" -eq 1 ]
+}
+
+assert_valid_repository () {
+  run dolt log
+  [ "$status" -eq 0 ]
+  [[ "$output" =~ "Initialize data repository" ]] || false
+}