Allow for three way merges to automatically resolve concurrent changes to a JSON object when the changes affect different keys.

go/cmd/dolt/cli/flags.go

@@ -48,6 +48,7 @@ const (
 	NoFFParam            = "no-ff"
 	NoPrettyFlag         = "no-pretty"
 	NoTLSFlag            = "no-tls"
+	NoJsonMergeFlag      = "dont-merge-json"
 	NotFlag              = "not"
 	NumberFlag           = "number"
 	OneLineFlag          = "oneline"

go/cmd/dolt/commands/cherry-pick.go

@@ -82,6 +82,7 @@ func (cmd CherryPickCmd) EventType() eventsapi.ClientEventType {
 // Exec executes the command.
 func (cmd CherryPickCmd) Exec(ctx context.Context, commandStr string, args []string, dEnv *env.DoltEnv, cliCtx cli.CliContext) int {
 	ap := cli.CreateCherryPickArgParser()
+	ap.SupportsFlag(cli.NoJsonMergeFlag, "", "Do not attempt to automatically resolve multiple changes to the same JSON value, report a conflict instead.")
 	help, usage := cli.HelpAndUsagePrinters(cli.CommandDocsForCommandString(commandStr, cherryPickDocs, ap))
 	apr := cli.ParseArgsOrDie(ap, args, help)
 
@@ -105,6 +106,14 @@ func (cmd CherryPickCmd) Exec(ctx context.Context, commandStr string, args []str
 		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
 	}
 
+	if apr.Contains(cli.NoJsonMergeFlag) {
+		_, _, err = queryist.Query(sqlCtx, "set @@session.dolt_dont_merge_json = 1")
+		if err != nil {
+			cli.Println(err.Error())
+			return 1
+		}
+	}
+
 	// TODO : support single commit cherry-pick only for now
 	if apr.NArg() == 0 {
 		usage()

go/cmd/dolt/commands/merge.go

@@ -91,6 +91,7 @@ func (cmd MergeCmd) RequiresRepo() bool {
 // Exec executes the command
 func (cmd MergeCmd) Exec(ctx context.Context, commandStr string, args []string, dEnv *env.DoltEnv, cliCtx cli.CliContext) int {
 	ap := cli.CreateMergeArgParser()
+	ap.SupportsFlag(cli.NoJsonMergeFlag, "", "Do not attempt to automatically resolve multiple changes to the same JSON value, report a conflict instead.")
 	help, usage := cli.HelpAndUsagePrinters(cli.CommandDocsForCommandString(commandStr, mergeDocs, ap))
 	apr := cli.ParseArgsOrDie(ap, args, help)
 
@@ -121,6 +122,14 @@ func (cmd MergeCmd) Exec(ctx context.Context, commandStr string, args []string,
 		return 1
 	}
 
+	if apr.Contains(cli.NoJsonMergeFlag) {
+		_, _, err = queryist.Query(sqlCtx, "set @@dolt_dont_merge_json = 1")
+		if err != nil {
+			cli.Println(err.Error())
+			return 1
+		}
+	}
+
 	query, err := constructInterpolatedDoltMergeQuery(apr, cliCtx)
 	if err != nil {
 		cli.Println(err.Error())

go/libraries/doltcore/doltdb/durable/index.go

@@ -52,7 +52,7 @@ type Index interface {
 	// Non-public. Used for flatbuffers Table persistence.
 	bytes() ([]byte, error)
 
-	DebugString(ctx context.Context) string
+	DebugString(ctx context.Context, ns tree.NodeStore, schema schema.Schema) string
 }
 
 // IndexSet stores a collection secondary Indexes.
@@ -221,7 +221,7 @@ func (i nomsIndex) AddColumnToRows(ctx context.Context, newCol string, newSchema
 	return i, nil
 }
 
-func (i nomsIndex) DebugString(ctx context.Context) string {
+func (i nomsIndex) DebugString(ctx context.Context, ns tree.NodeStore, schema schema.Schema) string {
 	panic("Not implemented")
 }
 
@@ -336,10 +336,10 @@ func (i prollyIndex) AddColumnToRows(ctx context.Context, newCol string, newSche
 	return IndexFromProllyMap(newMap), nil
 }
 
-func (i prollyIndex) DebugString(ctx context.Context) string {
+func (i prollyIndex) DebugString(ctx context.Context, ns tree.NodeStore, schema schema.Schema) string {
 	var b bytes.Buffer
 	i.index.WalkNodes(ctx, func(ctx context.Context, nd tree.Node) error {
-		return tree.OutputProllyNode(&b, nd)
+		return tree.OutputProllyNode(ctx, &b, nd, ns, schema)
 	})
 	return b.String()
 }

go/libraries/doltcore/doltdb/durable/table.go

@@ -106,7 +106,7 @@ type Table interface {
 	SetAutoIncrement(ctx context.Context, val uint64) (Table, error)
 
 	// DebugString returns the table contents for debugging purposes
-	DebugString(ctx context.Context) string
+	DebugString(ctx context.Context, ns tree.NodeStore) string
 }
 
 type nomsTable struct {
@@ -615,7 +615,7 @@ func (t nomsTable) SetAutoIncrement(ctx context.Context, val uint64) (Table, err
 	return nomsTable{t.vrw, t.ns, st}, nil
 }
 
-func (t nomsTable) DebugString(ctx context.Context) string {
+func (t nomsTable) DebugString(ctx context.Context, ns tree.NodeStore) string {
 	var buf bytes.Buffer
 	err := types.WriteEncodedValue(ctx, &buf, t.tableStruct)
 	if err != nil {
@@ -685,13 +685,18 @@ type doltDevTable struct {
 	msg *serial.Table
 }
 
-func (t doltDevTable) DebugString(ctx context.Context) string {
+func (t doltDevTable) DebugString(ctx context.Context, ns tree.NodeStore) string {
 	rows, err := t.GetTableRows(ctx)
 	if err != nil {
 		panic(err)
 	}
 
-	return rows.DebugString(ctx)
+	schema, err := t.GetSchema(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	return rows.DebugString(ctx, ns, schema)
 }
 
 var _ Table = doltDevTable{}

go/libraries/doltcore/doltdb/root_val.go

@@ -1147,7 +1147,7 @@ func (root *RootValue) DebugString(ctx context.Context, transitive bool) string
 			buf.WriteString(name)
 			buf.WriteString(":\n")
 
-			buf.WriteString(table.DebugString(ctx))
+			buf.WriteString(table.DebugString(ctx, root.ns))
 
 			return false, nil
 		})

go/libraries/doltcore/doltdb/table.go

@@ -668,6 +668,6 @@ func (t *Table) AddColumnToRows(ctx context.Context, newCol string, newSchema sc
 	return &Table{table: newTable}, nil
 }
 
-func (t *Table) DebugString(ctx context.Context) string {
-	return t.table.DebugString(ctx)
+func (t *Table) DebugString(ctx context.Context, ns tree.NodeStore) string {
+	return t.table.DebugString(ctx, ns)
 }

go/libraries/doltcore/merge/fulltext_table.go

@@ -25,9 +25,9 @@ import (
 	"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/index"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/sqlutil"
 	"github.com/dolthub/dolt/go/store/pool"
+	"github.com/dolthub/dolt/go/store/prolly/tree"
 	"github.com/dolthub/dolt/go/store/val"
 )
 
@@ -154,15 +154,15 @@ func (table *fulltextTable) ApplyToTable(ctx *sql.Context) (*doltdb.Table, error
 	for ; err == nil; sqlRow, err = rowIter.Next(ctx) {
 		for to := range keyMap {
 			from := keyMap.MapOrdinal(to)
-			if err = index.PutField(ctx, mut.NodeStore(), keyBld, to, sqlRow[from]); err != nil {
+			if err = tree.PutField(ctx, mut.NodeStore(), keyBld, to, sqlRow[from]); err != nil {
 				return nil, err
 			}
 		}
 		k := keyBld.Build(sharePool)
 
 		for to := range valMap {
 			from := valMap.MapOrdinal(to)
-			if err = index.PutField(ctx, mut.NodeStore(), valBld, to, sqlRow[from]); err != nil {
+			if err = tree.PutField(ctx, mut.NodeStore(), valBld, to, sqlRow[from]); err != nil {
 				return nil, err
 			}
 		}

go/libraries/doltcore/merge/merge_prolly_rows.go

@@ -25,6 +25,8 @@ import (
 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/dolthub/go-mysql-server/sql/expression"
 	"github.com/dolthub/go-mysql-server/sql/transform"
+	"github.com/dolthub/go-mysql-server/sql/types"
+	"golang.org/x/exp/maps"
 	errorkinds "gopkg.in/src-d/go-errors.v1"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
@@ -1464,7 +1466,7 @@ func remapTupleWithColumnDefaults(
 				secondPass = append(secondPass, to)
 			}
 
-			value, err = index.GetField(ctx, valDesc, from, valueTuple, tm.ns)
+			value, err = tree.GetField(ctx, valDesc, from, valueTuple, tm.ns)
 			if err != nil {
 				return nil, err
 			}
@@ -1475,7 +1477,7 @@ func remapTupleWithColumnDefaults(
 				return nil, err
 			}
 
-			err = index.PutField(ctx, tm.ns, tb, to, value)
+			err = tree.PutField(ctx, tm.ns, tb, to, value)
 			if err != nil {
 				return nil, err
 			}
@@ -1525,7 +1527,7 @@ func writeTupleExpression(
 		return err
 	}
 
-	return index.PutField(ctx, tm.ns, tb, colIdx, value)
+	return tree.PutField(ctx, tm.ns, tb, colIdx, value)
 }
 
 // convertValueToNewType handles converting a value from a previous type into a new type. |value| is the value from
@@ -1691,7 +1693,7 @@ func findNonPKColumnMappingByTagOrName(sch schema.Schema, col schema.Column) int
 // tuples. It returns the merged cell value tuple and a bool indicating if a
 // conflict occurred. tryMerge should only be called if left and right produce
 // non-identical diffs against base.
-func (m *valueMerger) tryMerge(ctx context.Context, left, right, base val.Tuple) (val.Tuple, bool, error) {
+func (m *valueMerger) tryMerge(ctx *sql.Context, left, right, base val.Tuple) (val.Tuple, bool, error) {
 	// If we're merging a keyless table and the keys match, but the values are different,
 	// that means that the row data is the same, but the cardinality has changed, and if the
 	// cardinality has changed in different ways on each merge side, we can't auto resolve.
@@ -1825,7 +1827,7 @@ func (m *valueMerger) processBaseColumn(ctx context.Context, i int, left, right,
 
 // processColumn returns the merged value of column |i| of the merged schema,
 // based on the |left|, |right|, and |base| schema.
-func (m *valueMerger) processColumn(ctx context.Context, i int, left, right, base val.Tuple) (result []byte, conflict bool, err error) {
+func (m *valueMerger) processColumn(ctx *sql.Context, i int, left, right, base val.Tuple) (result []byte, conflict bool, err error) {
 	// missing columns are coerced into NULL column values
 
 	var baseCol []byte
@@ -1840,6 +1842,8 @@ func (m *valueMerger) processColumn(ctx context.Context, i int, left, right, bas
 	resultColumn := m.resultSchema.GetNonPKCols().GetByIndex(i)
 	generatedColumn := resultColumn.Generated != ""
 
+	sqlType := m.resultSchema.GetNonPKCols().GetByIndex(i).TypeInfo.ToSqlType()
+
 	// We previously asserted that left and right are not nil.
 	// But base can be nil in the event of convergent inserts.
 	if base == nil || !baseColExists {
@@ -1947,6 +1951,19 @@ func (m *valueMerger) processColumn(ctx context.Context, i int, left, right, bas
 			return leftCol, false, nil
 		}
 		// concurrent modification
+		// if the result type is JSON, we can attempt to merge the JSON changes.
+		dontMergeJsonVar, err := ctx.Session.GetSessionVariable(ctx, "dolt_dont_merge_json")
+		if err != nil {
+			return nil, true, err
+		}
+		disallowJsonMerge, err := sql.ConvertToBool(ctx, dontMergeJsonVar)
+		if err != nil {
+			return nil, true, err
+		}
+		if _, ok := sqlType.(types.JsonType); ok && !disallowJsonMerge {
+			return m.mergeJSONAddr(ctx, baseCol, leftCol, rightCol)
+		}
+		// otherwise, this is a conflict.
 		return nil, true, nil
 	case leftModified:
 		return leftCol, false, nil
@@ -1955,6 +1972,97 @@ func (m *valueMerger) processColumn(ctx context.Context, i int, left, right, bas
 	}
 }
 
+func (m *valueMerger) mergeJSONAddr(ctx context.Context, baseAddr []byte, leftAddr []byte, rightAddr []byte) (resultAddr []byte, conflict bool, err error) {
+	baseDoc, err := tree.NewJSONDoc(hash.New(baseAddr), m.ns).ToJSONDocument(ctx)
+	if err != nil {
+		return nil, true, err
+	}
+	leftDoc, err := tree.NewJSONDoc(hash.New(leftAddr), m.ns).ToJSONDocument(ctx)
+	if err != nil {
+		return nil, true, err
+	}
+	rightDoc, err := tree.NewJSONDoc(hash.New(rightAddr), m.ns).ToJSONDocument(ctx)
+	if err != nil {
+		return nil, true, err
+	}
+
+	mergedDoc, conflict, err := mergeJSON(baseDoc, leftDoc, rightDoc)
+	if err != nil {
+		return nil, true, err
+	}
+	if conflict {
+		return nil, true, nil
+	}
+
+	mergedBytes, err := json.Marshal(mergedDoc.ToInterface())
+	if err != nil {
+		return nil, true, err
+	}
+	mergedAddr, err := tree.SerializeBytesToAddr(ctx, m.ns, bytes.NewReader(mergedBytes), len(mergedBytes))
+	if err != nil {
+		return nil, true, err
+	}
+	return mergedAddr[:], false, nil
+
+}
+
+func mergeJSON(base types.JSONDocument, left types.JSONDocument, right types.JSONDocument) (resultDoc types.JSONDocument, conflict bool, err error) {
+	// First, deserialize each value into JSON.
+	// We can only merge if the value at all three commits is a JSON object.
+
+	baseObject, baseIsObject := base.Val.(types.JsonObject)
+	leftObject, leftIsObject := left.Val.(types.JsonObject)
+	rightObject, rightIsObject := right.Val.(types.JsonObject)
+
+	if !baseIsObject || !leftIsObject || !rightIsObject {
+		// At least one of the commits does not have a JSON object.
+		// If both left and right have the same value, use that value.
+		// But if they differ, this is an unresolvable merge conflict.
+		cmp, err := left.Compare(right)
+		if err != nil {
+			return types.JSONDocument{}, true, err
+		}
+		if cmp == 0 {
+			//convergent operation.
+			return left, false, nil
+		} else {
+			return types.JSONDocument{}, true, nil
+		}
+	}
+
+	mergedObject := maps.Clone(leftObject)
+	merged := types.JSONDocument{Val: mergedObject}
+
+	threeWayDiffer := NewThreeWayJsonDiffer(baseObject, leftObject, rightObject)
+
+	// Compute the merged object by applying diffs to the left object as needed.
+	for {
+		threeWayDiff, err := threeWayDiffer.Next()
+		if err == io.EOF {
+			return merged, false, nil
+		}
+
+		switch threeWayDiff.Op {
+		case tree.DiffOpRightAdd, tree.DiffOpConvergentAdd, tree.DiffOpRightModify, tree.DiffOpConvergentModify:
+			_, _, err := merged.Set(threeWayDiff.Key, threeWayDiff.Right)
+			if err != nil {
+				return types.JSONDocument{}, true, err
+			}
+		case tree.DiffOpRightDelete, tree.DiffOpConvergentDelete:
+			_, _, err := merged.Remove(threeWayDiff.Key)
+			if err != nil {
+				return types.JSONDocument{}, true, err
+			}
+		case tree.DiffOpLeftAdd, tree.DiffOpLeftModify, tree.DiffOpLeftDelete:
+			// these changes already exist on the left, so do nothing.
+		case tree.DiffOpDivergentModifyConflict, tree.DiffOpDivergentDeleteConflict:
+			return types.JSONDocument{}, true, nil
+		default:
+			panic("unreachable")
+		}
+	}
+}
+
 func isEqual(i int, left []byte, right []byte, resultType val.Type) bool {
 	// We use a default comparator instead of the comparator in the schema.
 	// This is necessary to force a binary collation for string comparisons.
@@ -1976,7 +2084,7 @@ func convert(ctx context.Context, fromDesc, toDesc val.TupleDesc, toSchema schem
 		// No conversion is necessary here.
 		return originalValue, nil
 	}
-	parsedCell, err := index.GetField(ctx, fromDesc, fromIndex, tuple, ns)
+	parsedCell, err := tree.GetField(ctx, fromDesc, fromIndex, tuple, ns)
 	if err != nil {
 		return nil, err
 	}
@@ -1989,5 +2097,5 @@ func convert(ctx context.Context, fromDesc, toDesc val.TupleDesc, toSchema schem
 	// If a merge results in assigning NULL to a non-null column, don't panic.
 	// Instead we validate the merged tuple before merging it into the table.
 	typ.Nullable = true
-	return index.Serialize(ctx, ns, typ, convertedCell)
+	return tree.Serialize(ctx, ns, typ, convertedCell)
 }

go/libraries/doltcore/merge/row_merge_test.go

@@ -19,6 +19,7 @@ import (
 	"strconv"
 	"testing"
 
+	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
@@ -201,7 +202,7 @@ func TestRowMerge(t *testing.T) {
 		t.Skip()
 	}
 
-	ctx := context.Background()
+	ctx := sql.NewEmptyContext()
 
 	tests := make([]rowMergeTest, len(testCases))
 	for i, t := range testCases {