Merge branch 'main' into bh/signed-commits

.github/workflows/doltgres-dependency.yml

@@ -4,7 +4,7 @@ on:
   pull_request:
     types: [opened, synchronize, reopened]
   issue_comment:
-    types: [created, edited]
+    types: [created, edited, deleted]
 
 jobs:
   test-integration:
@@ -12,35 +12,62 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - name: Check for a DoltgreSQL PR link
-        run: |
-          COMMENTS=$(curl -s -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
-            https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.pull_request.number }}/comments)
-          COMMENT_EXISTS=$(echo "$COMMENTS" | jq -r '.[] | select(.body | contains("github.com/dolthub/doltgresql/pull/"))')
-          if [ -n "$COMMENT_EXISTS" ]; then
-            exit 0
-          fi
-
       - name: Checkout Dolt
         uses: actions/checkout@v4
         with:
           ref: ${{ github.event.pull_request.head.sha }}
 
+      - name: Setup Git User
+        uses: fregante/setup-git-user@v2
+
+      - name: Merge main into PR
+        id: merge_main
+        run: |
+          git fetch --all --unshallow
+          git merge origin/main --no-commit --no-ff
+          if [ $? -ne 0 ]; then
+            echo "Skipping the remainder of the workflow due to a merge conflict."
+            echo "skip=true" >> $GITHUB_OUTPUT
+          else
+            echo "Merge performed successfully, continuing workflow."
+            echo "skip=false" >> $GITHUB_OUTPUT
+          fi
+
+      - name: Check for a DoltgreSQL PR link
+        if: steps.merge_main.outputs.skip == 'false'
+        run: |
+          PR_DESCRIPTION=$(curl -s -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
+            https://api.github.com/repos/${{ github.repository }}/pulls/${{ github.event.pull_request.number != '' && github.event.pull_request.number || github.event.issue.number }})
+          COMMENTS=$(curl -s -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
+            https://api.github.com/repos/${{ github.repository }}/issues/${{ github.event.pull_request.number != '' && github.event.pull_request.number || github.event.issue.number }}/comments)
+          echo "$PR_DESCRIPTION$COMMENTS"
+          if echo "$PR_DESCRIPTION$COMMENTS" | grep -q "github.com/dolthub/doltgresql/pull/"; then
+            echo "comment_exists=true" >> $GITHUB_OUTPUT
+            echo "DoltgreSQL PR link exists"
+          else
+            echo "comment_exists=false" >> $GITHUB_OUTPUT
+            echo "DoltgreSQL PR link does not exist"
+          fi
+
       - name: Install Go
         uses: actions/setup-go@v5
+        if: steps.merge_main.outputs.skip == 'false'
         with:
           go-version-file: go/go.mod
 
       - name: Clone DoltgreSQL repository
+        if: steps.merge_main.outputs.skip == 'false'
         run: git clone https://github.com/dolthub/doltgresql.git
 
       - name: Build DoltgreSQL's parser
+        if: steps.merge_main.outputs.skip == 'false'
         run: |
           cd doltgresql
           ./postgres/parser/build.sh
 
       - name: Test DoltgreSQL against main
         id: test_doltgresql_main
+        if: steps.merge_main.outputs.skip == 'false'
         continue-on-error: true
         run: |
           cd doltgresql
@@ -50,11 +77,11 @@ jobs:
           go test ./... --count=1 -skip Replication
 
       - name: Test DoltgreSQL against PR
-        if: steps.test_doltgresql_main.outcome == 'success'
+        if: steps.merge_main.outputs.skip == 'false' && steps.test_doltgresql_main.outcome == 'success'
         run: |
           cd doltgresql
           git reset --hard
-          go get github.com/${{ github.event.pull_request.head.repo.full_name }}/go@${{ github.event.pull_request.head.sha }}
+          go mod edit -replace github.com/dolthub/dolt/go=../go
           go mod tidy
           cd testing/go
           go test ./... --count=1 -skip Replication

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

@@ -41,7 +41,7 @@ Cherry-picking merge commits or commits with table drops/renames is not currentl
 If any data conflicts, schema conflicts, or constraint violations are detected during cherry-picking, you can use Dolt's conflict resolution features to resolve them. For more information on resolving conflicts, see: https://docs.dolthub.com/concepts/dolt/git/conflicts.
 `,
 	Synopsis: []string{
-		`{{.LessThan}}commit{{.GreaterThan}}`,
+		`[--allow-empty] {{.LessThan}}commit{{.GreaterThan}}`,
 	},
 }
 

go/cmd/dolt/commands/rebase.go

@@ -31,6 +31,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/doltcore/rebase"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dprocedures"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
 	"github.com/dolthub/dolt/go/libraries/utils/argparser"
 	"github.com/dolthub/dolt/go/libraries/utils/config"
 	"github.com/dolthub/dolt/go/libraries/utils/editor"
@@ -48,7 +49,7 @@ branch. For example, you can drop commits that contain debugging or test changes
 single commit, or reorder commits so that related changes are adjacent in the new commit history.
 `,
 	Synopsis: []string{
-		`(-i | --interactive) {{.LessThan}}upstream{{.GreaterThan}}`,
+		`(-i | --interactive) [--empty=drop|keep] {{.LessThan}}upstream{{.GreaterThan}}`,
 		`(--continue | --abort)`,
 	},
 }
@@ -95,6 +96,12 @@ func (cmd RebaseCmd) Exec(ctx context.Context, commandStr string, args []string,
 		defer closeFunc()
 	}
 
+	// Set @@dolt_allow_commit_conflicts in case there are data conflicts that need to be resolved by the caller.
+	// Without this, the conflicts can't be committed to the branch working set, and the caller can't access them.
+	if _, err = GetRowsForSql(queryist, sqlCtx, "set @@dolt_allow_commit_conflicts=1;"); err != nil {
+		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+	}
+
 	branchName, err := getActiveBranchName(sqlCtx, queryist)
 	if err != nil {
 		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
@@ -118,65 +125,84 @@ func (cmd RebaseCmd) Exec(ctx context.Context, commandStr string, args []string,
 		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(errors.New("error: "+rows[0][1].(string))), usage)
 	}
 
+	// If the rebase was successful, or if it was aborted, print out the message and
+	// ensure the branch being rebased is checked out in the CLI
 	message := rows[0][1].(string)
-	if strings.Contains(message, dprocedures.SuccessfulRebaseMessage) {
-		cli.Println(dprocedures.SuccessfulRebaseMessage + branchName)
-	} else if strings.Contains(message, dprocedures.RebaseAbortedMessage) {
-		cli.Println(dprocedures.RebaseAbortedMessage)
-	} else {
-		rebasePlan, err := getRebasePlan(cliCtx, sqlCtx, queryist, apr.Arg(0), branchName)
-		if err != nil {
-			// attempt to abort the rebase
-			_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
+	if strings.Contains(message, dprocedures.SuccessfulRebaseMessage) ||
+		strings.Contains(message, dprocedures.RebaseAbortedMessage) {
+		cli.Println(message)
+		if err = syncCliBranchToSqlSessionBranch(sqlCtx, dEnv); err != nil {
 			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
 		}
-
-		// if all uncommented lines are deleted in the editor, abort the rebase
-		if rebasePlan == nil || rebasePlan.Steps == nil || len(rebasePlan.Steps) == 0 {
-			rows, err := GetRowsForSql(queryist, sqlCtx, "CALL DOLT_REBASE('--abort');")
-			if err != nil {
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
-			}
-			status, err := getInt64ColAsInt64(rows[0][0])
-			if err != nil {
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
-			}
-			if status == 1 {
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(errors.New("error: "+rows[0][1].(string))), usage)
-			}
-
-			cli.Println(dprocedures.RebaseAbortedMessage)
-		} else {
-			err = insertRebasePlanIntoDoltRebaseTable(rebasePlan, sqlCtx, queryist)
-			if err != nil {
-				// attempt to abort the rebase
-				_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
-			}
-
-			rows, err := GetRowsForSql(queryist, sqlCtx, "CALL DOLT_REBASE('--continue');")
-			if err != nil {
-				// attempt to abort the rebase
-				_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
-			}
-			status, err := getInt64ColAsInt64(rows[0][0])
-			if err != nil {
-				// attempt to abort the rebase
-				_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
-			}
-			if status == 1 {
-				// attempt to abort the rebase
-				_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
-				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(errors.New("error: "+rows[0][1].(string))), usage)
-			}
-
-			cli.Println(dprocedures.SuccessfulRebaseMessage + branchName)
-		}
+		return 0
 	}
 
-	return HandleVErrAndExitCode(nil, usage)
+	rebasePlan, err := getRebasePlan(cliCtx, sqlCtx, queryist, apr.Arg(0), branchName)
+	if err != nil {
+		// attempt to abort the rebase
+		_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
+		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+	}
+
+	// if all uncommented lines are deleted in the editor, abort the rebase
+	if rebasePlan == nil || rebasePlan.Steps == nil || len(rebasePlan.Steps) == 0 {
+		rows, err := GetRowsForSql(queryist, sqlCtx, "CALL DOLT_REBASE('--abort');")
+		if err != nil {
+			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		}
+		status, err := getInt64ColAsInt64(rows[0][0])
+		if err != nil {
+			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		}
+		if status == 1 {
+			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(errors.New("error: "+rows[0][1].(string))), usage)
+		}
+
+		cli.Println(dprocedures.RebaseAbortedMessage)
+		return 0
+	}
+
+	err = insertRebasePlanIntoDoltRebaseTable(rebasePlan, sqlCtx, queryist)
+	if err != nil {
+		// attempt to abort the rebase
+		_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
+		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+	}
+
+	rows, err = GetRowsForSql(queryist, sqlCtx, "CALL DOLT_REBASE('--continue');")
+	if err != nil {
+		// If the error is a data conflict, don't abort the rebase, but let the caller resolve the conflicts
+		if dprocedures.ErrRebaseDataConflict.Is(err) || strings.Contains(err.Error(), dprocedures.ErrRebaseDataConflict.Message[:40]) {
+			if checkoutErr := syncCliBranchToSqlSessionBranch(sqlCtx, dEnv); checkoutErr != nil {
+				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(checkoutErr), usage)
+			}
+		} else {
+			_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
+			if checkoutErr := syncCliBranchToSqlSessionBranch(sqlCtx, dEnv); checkoutErr != nil {
+				return HandleVErrAndExitCode(errhand.VerboseErrorFromError(checkoutErr), usage)
+			}
+		}
+		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+	}
+
+	status, err = getInt64ColAsInt64(rows[0][0])
+	if err != nil {
+		_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
+		if err = syncCliBranchToSqlSessionBranch(sqlCtx, dEnv); err != nil {
+			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		}
+		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+	}
+	if status == 1 {
+		_, _, _, _ = queryist.Query(sqlCtx, "CALL DOLT_REBASE('--abort');")
+		if err = syncCliBranchToSqlSessionBranch(sqlCtx, dEnv); err != nil {
+			return HandleVErrAndExitCode(errhand.VerboseErrorFromError(err), usage)
+		}
+		return HandleVErrAndExitCode(errhand.VerboseErrorFromError(errors.New("error: "+rows[0][1].(string))), usage)
+	}
+
+	cli.Println(rows[0][1].(string))
+	return 0
 }
 
 // getRebasePlan opens an editor for users to edit the rebase plan and returns the parsed rebase plan from the editor.
@@ -313,3 +339,17 @@ func insertRebasePlanIntoDoltRebaseTable(plan *rebase.RebasePlan, sqlCtx *sql.Co
 
 	return nil
 }
+
+// syncCliBranchToSqlSessionBranch sets the current branch for the CLI (in repo_state.json) to the active branch
+// for the current session. This is needed during rebasing, since any conflicts need to be resolved while the
+// session is on the rebase working branch (e.g. dolt_rebase_t1) and after the rebase finishes, the session needs
+// to be back on the branch being rebased (e.g. t1).
+func syncCliBranchToSqlSessionBranch(ctx *sql.Context, dEnv *env.DoltEnv) error {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	currentBranch, err := doltSession.GetBranch()
+	if err != nil {
+		return err
+	}
+
+	return saveHeadBranch(dEnv.FS, currentBranch)
+}

go/cmd/dolt/doltversion/version.go

@@ -16,5 +16,5 @@
 package doltversion
 
 const (
-	Version = "1.42.16"
+	Version = "1.42.17"
 )

go/gen/fb/serial/workingset.go

@@ -555,7 +555,31 @@ func (rcv *RebaseState) MutateCommitBecomesEmptyHandling(n byte) bool {
 	return rcv._tab.MutateByteSlot(12, n)
 }
 
-const RebaseStateNumFields = 5
+func (rcv *RebaseState) LastAttemptedStep() float32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(14))
+	if o != 0 {
+		return rcv._tab.GetFloat32(o + rcv._tab.Pos)
+	}
+	return 0.0
+}
+
+func (rcv *RebaseState) MutateLastAttemptedStep(n float32) bool {
+	return rcv._tab.MutateFloat32Slot(14, n)
+}
+
+func (rcv *RebaseState) RebasingStarted() bool {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(16))
+	if o != 0 {
+		return rcv._tab.GetBool(o + rcv._tab.Pos)
+	}
+	return false
+}
+
+func (rcv *RebaseState) MutateRebasingStarted(n bool) bool {
+	return rcv._tab.MutateBoolSlot(16, n)
+}
+
+const RebaseStateNumFields = 7
 
 func RebaseStateStart(builder *flatbuffers.Builder) {
 	builder.StartObject(RebaseStateNumFields)
@@ -584,6 +608,12 @@ func RebaseStateAddEmptyCommitHandling(builder *flatbuffers.Builder, emptyCommit
 func RebaseStateAddCommitBecomesEmptyHandling(builder *flatbuffers.Builder, commitBecomesEmptyHandling byte) {
 	builder.PrependByteSlot(4, commitBecomesEmptyHandling, 0)
 }
+func RebaseStateAddLastAttemptedStep(builder *flatbuffers.Builder, lastAttemptedStep float32) {
+	builder.PrependFloat32Slot(5, lastAttemptedStep, 0.0)
+}
+func RebaseStateAddRebasingStarted(builder *flatbuffers.Builder, rebasingStarted bool) {
+	builder.PrependBoolSlot(6, rebasingStarted, false)
+}
 func RebaseStateEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
 	return builder.EndObject()
 }

go/go.mod

@@ -57,7 +57,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/creasty/defaults v1.6.0
 	github.com/dolthub/flatbuffers/v23 v23.3.3-dh.2
-	github.com/dolthub/go-mysql-server v0.18.2-0.20240827225330-65d79910212e
+	github.com/dolthub/go-mysql-server v0.18.2-0.20240829233246-ed8de8d3a4e6
 	github.com/dolthub/gozstd v0.0.0-20240423170813-23a2903bca63
 	github.com/dolthub/swiss v0.1.0
 	github.com/goccy/go-json v0.10.2

go/go.sum

@@ -183,8 +183,8 @@ 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-icu-regex v0.0.0-20230524105445-af7e7991c97e h1:kPsT4a47cw1+y/N5SSCkma7FhAPw7KeGmD6c9PBZW9Y=
 github.com/dolthub/go-icu-regex v0.0.0-20230524105445-af7e7991c97e/go.mod h1:KPUcpx070QOfJK1gNe0zx4pA5sicIK1GMikIGLKC168=
-github.com/dolthub/go-mysql-server v0.18.2-0.20240827225330-65d79910212e h1:Fl9cRq6WCRusdbXDNl+iO/1NxVuVHRF7BPh0p0OAhLc=
-github.com/dolthub/go-mysql-server v0.18.2-0.20240827225330-65d79910212e/go.mod h1:nbdOzd0ceWONE80vbfwoRBjut7z3CIj69ZgDF/cKuaA=
+github.com/dolthub/go-mysql-server v0.18.2-0.20240829233246-ed8de8d3a4e6 h1:BcCI8gy6XyhgUZwxMALsHlhyRJR/NuSJ2XGf/CIaAjo=
+github.com/dolthub/go-mysql-server v0.18.2-0.20240829233246-ed8de8d3a4e6/go.mod h1:nbdOzd0ceWONE80vbfwoRBjut7z3CIj69ZgDF/cKuaA=
 github.com/dolthub/gozstd v0.0.0-20240423170813-23a2903bca63 h1:OAsXLAPL4du6tfbBgK0xXHZkOlos63RdKYS3Sgw/dfI=
 github.com/dolthub/gozstd v0.0.0-20240423170813-23a2903bca63/go.mod h1:lV7lUeuDhH5thVGDCKXbatwKy2KW80L4rMT46n+Y2/Q=
 github.com/dolthub/ishell v0.0.0-20240701202509-2b217167d718 h1:lT7hE5k+0nkBdj/1UOSFwjWpNxf+LCApbRHgnCA17XE=

go/libraries/doltcore/cherry_pick/cherry_pick.go

@@ -108,29 +108,15 @@ func CherryPick(ctx *sql.Context, commit string, options CherryPickOptions) (str
 		return "", mergeResult, nil
 	}
 
-	commitProps := actions.CommitStagedProps{
-		Date:    ctx.QueryTime(),
-		Name:    ctx.Client().User,
-		Email:   fmt.Sprintf("%s@%s", ctx.Client().User, ctx.Client().Address),
-		Message: commitMsg,
+	commitProps, err := CreateCommitStagedPropsFromCherryPickOptions(ctx, options)
+	if err != nil {
+		return "", nil, err
 	}
 
-	if options.CommitMessage != "" {
-		commitProps.Message = options.CommitMessage
-	}
-	if options.Amend {
-		commitProps.Amend = true
-	}
-	if options.EmptyCommitHandling == doltdb.KeepEmptyCommit {
-		commitProps.AllowEmpty = true
-	}
-
-	if options.CommitBecomesEmptyHandling == doltdb.DropEmptyCommit {
-		commitProps.SkipEmpty = true
-	} else if options.CommitBecomesEmptyHandling == doltdb.KeepEmptyCommit {
-		commitProps.AllowEmpty = true
-	} else if options.CommitBecomesEmptyHandling == doltdb.StopOnEmptyCommit {
-		return "", nil, fmt.Errorf("stop on empty commit is not currently supported")
+	// If no commit message was explicitly provided in the cherry-pick options,
+	// use the commit message from the cherry-picked commit.
+	if commitProps.Message == "" {
+		commitProps.Message = commitMsg
 	}
 
 	// NOTE: roots are old here (after staging the tables) and need to be refreshed
@@ -139,7 +125,7 @@ func CherryPick(ctx *sql.Context, commit string, options CherryPickOptions) (str
 		return "", nil, fmt.Errorf("failed to get roots for current session")
 	}
 
-	pendingCommit, err := doltSession.NewPendingCommit(ctx, dbName, roots, commitProps)
+	pendingCommit, err := doltSession.NewPendingCommit(ctx, dbName, roots, *commitProps)
 	if err != nil {
 		return "", nil, err
 	}
@@ -164,6 +150,36 @@ func CherryPick(ctx *sql.Context, commit string, options CherryPickOptions) (str
 	return h.String(), nil, nil
 }
 
+// CreateCommitStagedPropsFromCherryPickOptions converts the specified cherry-pick |options| into a CommitStagedProps
+// instance that can be used to create a pending commit.
+func CreateCommitStagedPropsFromCherryPickOptions(ctx *sql.Context, options CherryPickOptions) (*actions.CommitStagedProps, error) {
+	commitProps := actions.CommitStagedProps{
+		Date:  ctx.QueryTime(),
+		Name:  ctx.Client().User,
+		Email: fmt.Sprintf("%s@%s", ctx.Client().User, ctx.Client().Address),
+	}
+
+	if options.CommitMessage != "" {
+		commitProps.Message = options.CommitMessage
+	}
+	if options.Amend {
+		commitProps.Amend = true
+	}
+	if options.EmptyCommitHandling == doltdb.KeepEmptyCommit {
+		commitProps.AllowEmpty = true
+	}
+
+	if options.CommitBecomesEmptyHandling == doltdb.DropEmptyCommit {
+		commitProps.SkipEmpty = true
+	} else if options.CommitBecomesEmptyHandling == doltdb.KeepEmptyCommit {
+		commitProps.AllowEmpty = true
+	} else if options.CommitBecomesEmptyHandling == doltdb.StopOnEmptyCommit {
+		return nil, fmt.Errorf("stop on empty commit is not currently supported")
+	}
+
+	return &commitProps, nil
+}
+
 func previousCommitMessage(ctx *sql.Context) (string, error) {
 	doltSession := dsess.DSessFromSess(ctx.Session)
 	headCommit, err := doltSession.GetHeadCommit(ctx, ctx.GetCurrentDatabase())

go/libraries/doltcore/doltdb/workingset.go

@@ -66,6 +66,15 @@ type RebaseState struct {
 
 	// emptyCommitHandling specifies how to handle empty commits that contain no changes.
 	emptyCommitHandling EmptyCommitHandling
+
+	// lastAttemptedStep records the last rebase plan step that was attempted, whether it completed successfully, or
+	// resulted in conflicts for the user to manually resolve. This field is not valid unless rebasingStarted is set
+	// to true.
+	lastAttemptedStep float32
+
+	// rebasingStarted is true once the rebase plan has been started to execute. Once rebasingStarted is true, the
+	// value in lastAttemptedStep has been initialized and is valid to read.
+	rebasingStarted bool
 }
 
 // Branch returns the name of the branch being actively rebased. This is the branch that will be updated to point
@@ -93,6 +102,24 @@ func (rs RebaseState) CommitBecomesEmptyHandling() EmptyCommitHandling {
 	return rs.commitBecomesEmptyHandling
 }
 
+func (rs RebaseState) LastAttemptedStep() float32 {
+	return rs.lastAttemptedStep
+}
+
+func (rs RebaseState) WithLastAttemptedStep(step float32) *RebaseState {
+	rs.lastAttemptedStep = step
+	return &rs
+}
+
+func (rs RebaseState) RebasingStarted() bool {
+	return rs.rebasingStarted
+}
+
+func (rs RebaseState) WithRebasingStarted(rebasingStarted bool) *RebaseState {
+	rs.rebasingStarted = rebasingStarted
+	return &rs
+}
+
 type MergeState struct {
 	// the source commit
 	commit *Commit
@@ -517,6 +544,8 @@ func newWorkingSet(ctx context.Context, name string, vrw types.ValueReadWriter,
 			branch:                     dsws.RebaseState.Branch(ctx),
 			commitBecomesEmptyHandling: EmptyCommitHandling(dsws.RebaseState.CommitBecomesEmptyHandling(ctx)),
 			emptyCommitHandling:        EmptyCommitHandling(dsws.RebaseState.EmptyCommitHandling(ctx)),
+			lastAttemptedStep:          dsws.RebaseState.LastAttemptedStep(ctx),
+			rebasingStarted:            dsws.RebaseState.RebasingStarted(ctx),
 		}
 	}
 
@@ -613,7 +642,8 @@ func (ws *WorkingSet) writeValues(ctx context.Context, db *DoltDB, meta *datas.W
 		}
 
 		rebaseState = datas.NewRebaseState(preRebaseWorking.TargetHash(), dCommit.Addr(), ws.rebaseState.branch,
-			uint8(ws.rebaseState.commitBecomesEmptyHandling), uint8(ws.rebaseState.emptyCommitHandling))
+			uint8(ws.rebaseState.commitBecomesEmptyHandling), uint8(ws.rebaseState.emptyCommitHandling),
+			ws.rebaseState.lastAttemptedStep, ws.rebaseState.rebasingStarted)
 	}
 
 	return &datas.WorkingSetSpec{

go/libraries/doltcore/env/actions/checkout.go

@@ -18,13 +18,14 @@ import (
 	"context"
 	"time"
 
-	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
-	"github.com/dolthub/dolt/go/store/datas"
+	"github.com/dolthub/go-mysql-server/sql"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/doltcore/ref"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/utils/set"
+	"github.com/dolthub/dolt/go/store/datas"
 	"github.com/dolthub/dolt/go/store/hash"
 )
 
@@ -148,7 +149,7 @@ func RootsForBranch(ctx context.Context, roots doltdb.Roots, branchRoot doltdb.R
 
 // CleanOldWorkingSet resets the source branch's working set to the branch head, leaving the source branch unchanged
 func CleanOldWorkingSet(
-	ctx context.Context,
+	ctx *sql.Context,
 	dbData env.DbData,
 	doltDb *doltdb.DoltDB,
 	username, email string,

go/libraries/doltcore/env/actions/reset.go

@@ -19,19 +19,21 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/dolthub/dolt/go/store/datas"
+	"github.com/dolthub/go-mysql-server/sql"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
 	"github.com/dolthub/dolt/go/libraries/doltcore/env"
 	"github.com/dolthub/dolt/go/libraries/doltcore/ref"
 	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/resolve"
 	"github.com/dolthub/dolt/go/libraries/utils/argparser"
+	"github.com/dolthub/dolt/go/store/datas"
 )
 
 // resetHardTables resolves a new HEAD commit from a refSpec and updates working set roots by
 // resetting the table contexts for tracked tables. New tables are ignored. Returns new HEAD
 // Commit and Roots.
-func resetHardTables(ctx context.Context, dbData env.DbData, cSpecStr string, roots doltdb.Roots) (*doltdb.Commit, doltdb.Roots, error) {
+func resetHardTables(ctx *sql.Context, dbData env.DbData, cSpecStr string, roots doltdb.Roots) (*doltdb.Commit, doltdb.Roots, error) {
 	ddb := dbData.Ddb
 	rsr := dbData.Rsr
 
@@ -100,11 +102,15 @@ func resetHardTables(ctx context.Context, dbData env.DbData, cSpecStr string, ro
 	}
 
 	for name := range untracked {
-		tbl, _, err := roots.Working.GetTable(ctx, doltdb.TableName{Name: name})
+		tname, tbl, exists, err := resolve.Table(ctx, roots.Working, name)
 		if err != nil {
 			return nil, doltdb.Roots{}, err
 		}
-		newWkRoot, err = newWkRoot.PutTable(ctx, doltdb.TableName{Name: name}, tbl)
+		if !exists {
+			return nil, doltdb.Roots{}, fmt.Errorf("untracked table %s does not exist in working set", name)
+		}
+
+		newWkRoot, err = newWkRoot.PutTable(ctx, tname, tbl)
 		if err != nil {
 			return nil, doltdb.Roots{}, fmt.Errorf("failed to write table back to database: %s", err)
 		}
@@ -144,7 +150,7 @@ func resetHardTables(ctx context.Context, dbData env.DbData, cSpecStr string, ro
 
 // ResetHardTables resets the tables in working, staged, and head based on the given parameters. Returns the new
 // head commit and resulting roots
-func ResetHardTables(ctx context.Context, dbData env.DbData, cSpecStr string, roots doltdb.Roots) (*doltdb.Commit, doltdb.Roots, error) {
+func ResetHardTables(ctx *sql.Context, dbData env.DbData, cSpecStr string, roots doltdb.Roots) (*doltdb.Commit, doltdb.Roots, error) {
 	return resetHardTables(ctx, dbData, cSpecStr, roots)
 }
 
@@ -152,7 +158,7 @@ func ResetHardTables(ctx context.Context, dbData env.DbData, cSpecStr string, ro
 // The reset can be performed on a non-current branch and working set.
 // Returns an error if the reset fails.
 func ResetHard(
-	ctx context.Context,
+	ctx *sql.Context,
 	dbData env.DbData,
 	doltDb *doltdb.DoltDB,
 	username, email string,

go/libraries/doltcore/rebase/rebase.go

@@ -62,6 +62,13 @@ type RebasePlanStep struct {
 	CommitMsg   string
 }
 
+// RebaseOrderAsFloat returns the RebaseOrder as a float32. Float32 provides enough scale and precision to hold
+// rebase order values, since they are limited to two decimal points of precision and six total digits.
+func (rps *RebasePlanStep) RebaseOrderAsFloat() float32 {
+	f64, _ := rps.RebaseOrder.Float64()
+	return float32(f64)
+}
+
 // CreateDefaultRebasePlan creates and returns the default rebase plan for the commits between
 // |startCommit| and |upstreamCommit|, equivalent to the log of startCommit..upstreamCommit. The
 // default plan includes each of those commits, in the same order they were originally applied, and

go/libraries/doltcore/sqle/binlogreplication/binlog_replica_applier.go

@@ -55,15 +55,16 @@ const (
 // This type is NOT used concurrently – there is currently only one single applier process running to process binlog
 // events, so the state in this type is NOT protected with a mutex.
 type binlogReplicaApplier struct {
-	format                *mysql.BinlogFormat
-	tableMapsById         map[uint64]*mysql.TableMap
-	stopReplicationChan   chan struct{}
-	currentGtid           mysql.GTID
-	replicationSourceUuid string
-	currentPosition       *mysql.Position // successfully executed GTIDs
-	filters               *filterConfiguration
-	running               atomic.Bool
-	engine                *gms.Engine
+	format                    *mysql.BinlogFormat
+	tableMapsById             map[uint64]*mysql.TableMap
+	stopReplicationChan       chan struct{}
+	currentGtid               mysql.GTID
+	replicationSourceUuid     string
+	currentPosition           *mysql.Position // successfully executed GTIDs
+	filters                   *filterConfiguration
+	running                   atomic.Bool
+	engine                    *gms.Engine
+	dbsWithUncommittedChanges map[string]struct{}
 }
 
 func newBinlogReplicaApplier(filters *filterConfiguration) *binlogReplicaApplier {
@@ -326,7 +327,6 @@ func (a *binlogReplicaApplier) replicaBinlogEventHandler(ctx *sql.Context) error
 func (a *binlogReplicaApplier) processBinlogEvent(ctx *sql.Context, engine *gms.Engine, event mysql.BinlogEvent) error {
 	var err error
 	createCommit := false
-	commitToAllDatabases := false
 
 	// We don't support checksum validation, so we MUST strip off any checksum bytes if present, otherwise it gets
 	// interpreted as part of the payload and corrupts the data. Future checksum sizes, are not guaranteed to be the
@@ -356,7 +356,6 @@ func (a *binlogReplicaApplier) processBinlogEvent(ctx *sql.Context, engine *gms.
 		// XA-capable storage engine. For more details, see: https://mariadb.com/kb/en/xid_event/
 		ctx.GetLogger().Trace("Received binlog event: XID")
 		createCommit = true
-		commitToAllDatabases = true
 
 	case event.IsQuery():
 		// A Query event represents a statement executed on the source server that should be executed on the
@@ -375,13 +374,6 @@ func (a *binlogReplicaApplier) processBinlogEvent(ctx *sql.Context, engine *gms.
 			"sql_mode": fmt.Sprintf("0x%x", query.SqlMode),
 		}).Trace("Received binlog event: Query")
 
-		// When executing SQL statements sent from the primary, we can't be sure what database was modified unless we
-		// look closely at the statement. For example, we could be connected to db01, but executed
-		// "create table db02.t (...);" – i.e., looking at query.Database is NOT enough to always determine the correct
-		// database that was modified, so instead, we commit to all databases when we see a Query binlog event to
-		// avoid issues with correctness, at the cost of being slightly less efficient
-		commitToAllDatabases = true
-
 		if query.Options&mysql.QFlagOptionAutoIsNull > 0 {
 			ctx.GetLogger().Tracef("Setting sql_auto_is_null ON")
 			ctx.SetSessionVariable(ctx, "sql_auto_is_null", 1)
@@ -541,13 +533,10 @@ func (a *binlogReplicaApplier) processBinlogEvent(ctx *sql.Context, engine *gms.
 	}
 
 	if createCommit {
-		var databasesToCommit []string
-		if commitToAllDatabases {
-			databasesToCommit = getAllUserDatabaseNames(ctx, engine)
-			for _, database := range databasesToCommit {
-				executeQueryWithEngine(ctx, engine, "use `"+database+"`;")
-				executeQueryWithEngine(ctx, engine, "commit;")
-			}
+		doltSession := dsess.DSessFromSess(ctx.Session)
+		databasesToCommit := doltSession.DirtyDatabases()
+		if err = doltSession.CommitTransaction(ctx, doltSession.GetTransaction()); err != nil {
+			return err
 		}
 
 		// Record the last GTID processed after the commit
@@ -562,17 +551,46 @@ func (a *binlogReplicaApplier) processBinlogEvent(ctx *sql.Context, engine *gms.
 		}
 
 		// For now, create a Dolt commit from every data update. Eventually, we'll want to make this configurable.
-		ctx.GetLogger().Trace("Creating Dolt commit(s)")
-		for _, database := range databasesToCommit {
+		// We commit to every database that we saw had a dirty session – these identify the databases where we have
+		// run DML commands through the engine. We also commit to every database that was modified through a RowEvent,
+		// which is all tracked through the applier's databasesWithUncommitedChanges property – these don't show up
+		// as dirty in our session, since we used TableWriter to update them.
+		a.addDatabasesWithUncommittedChanges(databasesToCommit...)
+		for _, database := range a.databasesWithUncommittedChanges() {
 			executeQueryWithEngine(ctx, engine, "use `"+database+"`;")
 			executeQueryWithEngine(ctx, engine,
 				fmt.Sprintf("call dolt_commit('-Am', 'Dolt binlog replica commit: GTID %s');", a.currentGtid))
 		}
+		a.dbsWithUncommittedChanges = nil
 	}
 
 	return nil
 }
 
+// addDatabasesWithUncommittedChanges marks the specifeid |dbNames| as databases with uncommitted changes so that
+// the replica applier knows which databases need to have Dolt commits created.
+func (a *binlogReplicaApplier) addDatabasesWithUncommittedChanges(dbNames ...string) {
+	if a.dbsWithUncommittedChanges == nil {
+		a.dbsWithUncommittedChanges = make(map[string]struct{})
+	}
+	for _, dbName := range dbNames {
+		a.dbsWithUncommittedChanges[dbName] = struct{}{}
+	}
+}
+
+// databasesWithUncommittedChanges returns a slice of database names indicating which databases have uncommitted
+// changes and need a Dolt commit created.
+func (a *binlogReplicaApplier) databasesWithUncommittedChanges() []string {
+	if a.dbsWithUncommittedChanges == nil {
+		return nil
+	}
+	dbNames := make([]string, 0, len(a.dbsWithUncommittedChanges))
+	for dbName, _ := range a.dbsWithUncommittedChanges {
+		dbNames = append(dbNames, dbName)
+	}
+	return dbNames
+}
+
 // processRowEvent processes a WriteRows, DeleteRows, or UpdateRows binlog event and returns an error if any problems
 // were encountered.
 func (a *binlogReplicaApplier) processRowEvent(ctx *sql.Context, event mysql.BinlogEvent, engine *gms.Engine) error {
@@ -599,6 +617,7 @@ func (a *binlogReplicaApplier) processRowEvent(ctx *sql.Context, event mysql.Bin
 		return nil
 	}
 
+	a.addDatabasesWithUncommittedChanges(tableMap.Database)
 	rows, err := event.Rows(*a.format, tableMap)
 	if err != nil {
 		return err

go/libraries/doltcore/sqle/binlogreplication/binlog_replication_test.go

@@ -35,6 +35,7 @@ import (
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/jmoiron/sqlx"
+	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 
 	"github.com/dolthub/go-mysql-server/sql/binlogreplication"
@@ -123,6 +124,47 @@ func TestBinlogReplicationSanityCheck(t *testing.T) {
 	requireReplicaResults(t, "select * from db01.tableT", [][]any{{"300"}})
 }
 
+// TestBinlogReplicationWithHundredsOfDatabases asserts that we can efficiently replicate the creation of hundreds of databases.
+func TestBinlogReplicationWithHundredsOfDatabases(t *testing.T) {
+	defer teardown(t)
+	startSqlServersWithDoltSystemVars(t, doltReplicaSystemVars)
+	startReplicationAndCreateTestDb(t, mySqlPort)
+
+	// Create a table on the primary and verify on the replica
+	primaryDatabase.MustExec("create table tableT (pk int primary key)")
+	waitForReplicaToCatchUp(t)
+	assertCreateTableStatement(t, replicaDatabase, "tableT",
+		"CREATE TABLE tableT ( pk int NOT NULL, PRIMARY KEY (pk)) "+
+			"ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_bin")
+	assertRepoStateFileExists(t, "db01")
+
+	// Create a few hundred databases on the primary and let them replicate to the replica
+	dbCount := 300
+	startTime := time.Now()
+	for i := range dbCount {
+		dbName := fmt.Sprintf("db%03d", i)
+		primaryDatabase.MustExec(fmt.Sprintf("create database %s", dbName))
+	}
+	waitForReplicaToCatchUp(t)
+	endTime := time.Now()
+	logrus.Infof("Time to replicate %d databases: %v", dbCount, endTime.Sub(startTime))
+
+	// Spot check the presence of a database on the replica
+	assertRepoStateFileExists(t, "db042")
+
+	// Insert some data in one database
+	startTime = time.Now()
+	primaryDatabase.MustExec("use db042;")
+	primaryDatabase.MustExec("create table t (pk int primary key);")
+	primaryDatabase.MustExec("insert into t values (100), (101), (102);")
+
+	// Verify the results on the replica
+	waitForReplicaToCatchUp(t)
+	requireReplicaResults(t, "select * from db042.t;", [][]any{{"100"}, {"101"}, {"102"}})
+	endTime = time.Now()
+	logrus.Infof("Time to replicate inserts to 1 database (out of %d): %v", endTime.Sub(startTime), dbCount)
+}
+
 // TestAutoRestartReplica tests that a Dolt replica automatically starts up replication if
 // replication was running when the replica was shut down.
 func TestAutoRestartReplica(t *testing.T) {

go/libraries/doltcore/sqle/dprocedures/dolt_rebase.go

@@ -81,11 +81,34 @@ var DoltRebaseSystemTableSchema = []*sql.Column{
 
 // ErrRebaseUncommittedChanges is used when a rebase is started, but there are uncommitted (and not
 // ignored) changes in the working set.
-var ErrRebaseUncommittedChanges = fmt.Errorf("cannot start a rebase with uncommitted changes")
+var ErrRebaseUncommittedChanges = goerrors.NewKind("cannot start a rebase with uncommitted changes")
 
-// ErrRebaseConflict is used when a merge conflict is detected while rebasing a commit.
-var ErrRebaseConflict = goerrors.NewKind(
-	"merge conflict detected while rebasing commit %s. " +
+// ErrRebaseUnstagedChanges is used when a rebase is continued, but there are unstaged changes
+// in the working set.
+var ErrRebaseUnstagedChanges = goerrors.NewKind("cannot continue a rebase with unstaged changes. " +
+	"Use dolt_add() to stage tables and then continue the rebase")
+
+// ErrRebaseUnresolvedConflicts is used when a rebase is continued, but there are
+// unresolved conflicts still present.
+var ErrRebaseUnresolvedConflicts = goerrors.NewKind(
+	"conflicts detected in tables %s; resolve conflicts before continuing the rebase")
+
+// ErrRebaseDataConflictsCantBeResolved is used when data conflicts are detected in a rebase, but @@autocommit has not
+// been disabled or @@dolt_allow_commit_conflicts has not been enabled, so it's not possible to resolve the conflicts
+// since they would get rolled back automatically.
+var ErrRebaseDataConflictsCantBeResolved = goerrors.NewKind(
+	"data conflicts from rebase, but session settings do not allow preserving conflicts, so they cannot be resolved. " +
+		"The rebase has been aborted. Set @@autocommit to 0 or set @@dolt_allow_commit_conflicts to 1 and " +
+		"try the rebase again to resolve the conflicts.")
+
+// ErrRebaseDataConflict is used when a data conflict is detected while rebasing a commit.
+var ErrRebaseDataConflict = goerrors.NewKind("data conflict detected while rebasing commit %s (%s). \n\n" +
+	"Resolve the conflicts and remove them from the dolt_conflicts_<table> tables, " +
+	"then continue the rebase by calling dolt_rebase('--continue')")
+
+// ErrRebaseSchemaConflict is used when a schema conflict is detected while rebasing a commit.
+var ErrRebaseSchemaConflict = goerrors.NewKind(
+	"schema conflict detected while rebasing commit %s. " +
 		"the rebase has been automatically aborted")
 
 // ErrRebaseConflictWithAbortError is used when a merge conflict is detected while rebasing a commit,
@@ -367,7 +390,7 @@ func validateWorkingSetCanStartRebase(ctx *sql.Context) error {
 		return err
 	}
 	if !wsOnlyHasIgnoredTables {
-		return ErrRebaseUncommittedChanges
+		return ErrRebaseUncommittedChanges.New()
 	}
 
 	return nil
@@ -419,46 +442,245 @@ func abortRebase(ctx *sql.Context) error {
 	return doltSession.SwitchWorkingSet(ctx, ctx.GetCurrentDatabase(), wsRef)
 }
 
-func continueRebase(ctx *sql.Context) (string, error) {
-	// TODO: Eventually, when we allow interactive-rebases to be stopped and started (e.g. with the break action,
-	//       or for conflict resolution), we'll need to track what step we're at in the rebase plan.
-
-	// Validate that we are in an interactive rebase
+func validateActiveRebase(ctx *sql.Context) error {
 	doltSession := dsess.DSessFromSess(ctx.Session)
 	workingSet, err := doltSession.WorkingSet(ctx, ctx.GetCurrentDatabase())
 	if err != nil {
-		return "", err
+		return err
 	}
 	if !workingSet.RebaseActive() {
-		return "", fmt.Errorf("no rebase in progress")
+		return fmt.Errorf("no rebase in progress")
+	}
+	return nil
+}
+
+func validateNoConflicts(ctx *sql.Context) error {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	workingSet, err := doltSession.WorkingSet(ctx, ctx.GetCurrentDatabase())
+	if err != nil {
+		return err
 	}
 
+	tablesWithDataConflicts, err := doltdb.TablesWithDataConflicts(ctx, workingSet.WorkingRoot())
+	if err != nil {
+		return err
+	}
+	if len(tablesWithDataConflicts) > 0 {
+		return ErrRebaseUnresolvedConflicts.New(
+			strings.Join(tablesWithDataConflicts, ", "))
+	}
+
+	return nil
+}
+
+// loadRebasePlan loads the rebase plan from the current database for the current session and validates it.
+func loadRebasePlan(ctx *sql.Context) (*rebase.RebasePlan, error) {
+	doltSession := dsess.DSessFromSess(ctx.Session)
 	db, err := doltSession.Provider().Database(ctx, ctx.GetCurrentDatabase())
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
 	rdb, ok := db.(rebase.RebasePlanDatabase)
 	if !ok {
-		return "", fmt.Errorf("expected a dsess.RebasePlanDatabase implementation, but received a %T", db)
+		return nil, fmt.Errorf("expected a dsess.RebasePlanDatabase implementation, but received a %T", db)
 	}
 	rebasePlan, err := rdb.LoadRebasePlan(ctx)
 	if err != nil {
+		return nil, err
+	}
+	err = rebase.ValidateRebasePlan(ctx, rebasePlan)
+	if err != nil {
+		return nil, err
+	}
+
+	return rebasePlan, nil
+}
+
+// filterIgnoredTables iterates over the specified |unstagedTableDeltas|, filters out any TableDeltas for
+// ignored tables, and returns the filtered list.
+func filterIgnoredTables(ctx *sql.Context, unstagedTableDeltas []diff.TableDelta) ([]diff.TableDelta, error) {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	roots, ok := doltSession.GetRoots(ctx, ctx.GetCurrentDatabase())
+	if !ok {
+		return nil, fmt.Errorf("unable to get roots for current session")
+	}
+
+	filteredTableDeltas := make([]diff.TableDelta, 0, len(unstagedTableDeltas))
+	ignorePatterns, err := doltdb.GetIgnoredTablePatterns(ctx, roots)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, tableDelta := range unstagedTableDeltas {
+		isIgnored, err := ignorePatterns.IsTableNameIgnored(tableDelta.ToName)
+		if err != nil {
+			return nil, err
+		}
+		if isIgnored != doltdb.Ignore {
+			filteredTableDeltas = append(filteredTableDeltas, tableDelta)
+		}
+	}
+
+	return filteredTableDeltas, nil
+}
+
+// workingSetStatus checks the workspace status and returns whether there are any staged changes and unstaged changes.
+func workingSetStatus(ctx *sql.Context) (stagedChanges, unstagedChanges bool, err error) {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	roots, ok := doltSession.GetRoots(ctx, ctx.GetCurrentDatabase())
+	if !ok {
+		return false, false, fmt.Errorf("unable to get roots for current session")
+	}
+
+	stagedTables, unstagedTables, err := diff.GetStagedUnstagedTableDeltas(ctx, roots)
+	if err != nil {
+		return false, false, err
+	}
+
+	unstagedTables, err = filterIgnoredTables(ctx, unstagedTables)
+	if err != nil {
+		return false, false, err
+	}
+
+	return len(stagedTables) > 0, len(unstagedTables) > 0, nil
+}
+
+// recordCurrentStep updates working set metadata to record the current rebase plan step number as well
+// as the rebase started flag indicating that execution of the rebase plan has been started. This
+// information is all stored in the RebaseState of the WorkingSet.
+func recordCurrentStep(ctx *sql.Context, step rebase.RebasePlanStep) error {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	if doltSession.GetTransaction() == nil {
+		_, err := doltSession.StartTransaction(ctx, sql.ReadWrite)
+		if err != nil {
+			return err
+		}
+	}
+
+	workingSet, err := doltSession.WorkingSet(ctx, ctx.GetCurrentDatabase())
+	if err != nil {
+		return err
+	}
+
+	// Update the current step in the working set, so that we can continue the merge if this hits a conflict
+	newWorkingSet := workingSet.WithRebaseState(workingSet.RebaseState().
+		WithLastAttemptedStep(step.RebaseOrderAsFloat()).
+		WithRebasingStarted(true))
+	err = doltSession.SetWorkingSet(ctx, ctx.GetCurrentDatabase(), newWorkingSet)
+	if err != nil {
+		return err
+	}
+
+	// Commit the SQL transaction with the LastAttemptedStep update to set that in the branch head's working set
+	if doltSession.GetTransaction() != nil {
+		err = doltSession.CommitTransaction(ctx, doltSession.GetTransaction())
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func continueRebase(ctx *sql.Context) (string, error) {
+	// Validate that we are in an interactive rebase
+	if err := validateActiveRebase(ctx); err != nil {
 		return "", err
 	}
 
-	err = rebase.ValidateRebasePlan(ctx, rebasePlan)
+	// If there are conflicts, stop the rebase with an error message about resolving the conflicts before continuing
+	if err := validateNoConflicts(ctx); err != nil {
+		return "", err
+	}
+
+	// After we've checked for conflicts in the working set, commit the SQL transaction to ensure
+	// our session is in sync with the latest changes on the branch
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	if doltSession.GetTransaction() == nil {
+		_, err := doltSession.StartTransaction(ctx, sql.ReadWrite)
+		if err != nil {
+			return "", err
+		}
+	}
+	if err := doltSession.CommitTransaction(ctx, doltSession.GetTransaction()); err != nil {
+		return "", err
+	}
+
+	rebasePlan, err := loadRebasePlan(ctx)
 	if err != nil {
 		return "", err
 	}
 
 	for _, step := range rebasePlan.Steps {
-		err = processRebasePlanStep(ctx, &step,
-			workingSet.RebaseState().CommitBecomesEmptyHandling(),
-			workingSet.RebaseState().EmptyCommitHandling())
+		workingSet, err := doltSession.WorkingSet(ctx, ctx.GetCurrentDatabase())
 		if err != nil {
 			return "", err
 		}
+
+		rebaseStepOrder := step.RebaseOrderAsFloat()
+		lastAttemptedStep := workingSet.RebaseState().LastAttemptedStep()
+		rebasingStarted := workingSet.RebaseState().RebasingStarted()
+
+		hasStagedChanges, hasUnstagedChanges, err := workingSetStatus(ctx)
+		if err != nil {
+			return "", err
+		}
+
+		// If the rebase is just starting but there are any uncommitted changes in the working set,
+		// tell the user they need to commit them before we can start executing the rebase plan.
+		if !rebasingStarted && (hasStagedChanges || hasUnstagedChanges) {
+			return "", ErrRebaseUncommittedChanges.New()
+		}
+
+		// If we've already executed this step, move to the next plan step
+		if rebasingStarted && rebaseStepOrder < lastAttemptedStep {
+			continue
+		}
+
+		// If the rebase is continued, but not all working set changes are staged, then tell the user
+		// they need to explicitly stage the tables before the rebase can be continued.
+		if hasUnstagedChanges {
+			return "", ErrRebaseUnstagedChanges.New()
+		}
+
+		// If we've already executed this step, but the working set has staged changes,
+		// then we need to make the commit for the manual changes made for this step.
+		if rebasingStarted && rebaseStepOrder == lastAttemptedStep && hasStagedChanges {
+			if err = commitManuallyStagedChangesForStep(ctx, step); err != nil {
+				return "", err
+			}
+			continue
+		}
+
+		// If rebasing hasn't started yet or this step is greater than the last attempted step,
+		// go ahead and execute this step.
+		if !rebasingStarted || rebaseStepOrder > lastAttemptedStep {
+			if err = recordCurrentStep(ctx, step); err != nil {
+				return "", err
+			}
+
+			doltSession := dsess.DSessFromSess(ctx.Session)
+			workingSet, err := doltSession.WorkingSet(ctx, ctx.GetCurrentDatabase())
+			if err != nil {
+				return "", err
+			}
+
+			err = processRebasePlanStep(ctx, &step,
+				workingSet.RebaseState().CommitBecomesEmptyHandling(),
+				workingSet.RebaseState().EmptyCommitHandling())
+			if err != nil {
+				return "", err
+			}
+		}
+
+		// Ensure a transaction has been started, so that the session is in sync with the latest changes
+		if doltSession.GetTransaction() == nil {
+			_, err = doltSession.StartTransaction(ctx, sql.ReadWrite)
+			if err != nil {
+				return "", err
+			}
+		}
 	}
 
 	// Update the branch being rebased to point to the same commit as our temporary working branch
@@ -501,6 +723,11 @@ func continueRebase(ctx *sql.Context) (string, error) {
 		return "", err
 	}
 
+	// Start a new transaction so the session will see the changes to the branch pointer
+	if _, err = doltSession.StartTransaction(ctx, sql.ReadWrite); err != nil {
+		return "", err
+	}
+
 	// delete the temporary working branch
 	dbData, ok = doltSession.GetDbData(ctx, ctx.GetCurrentDatabase())
 	if !ok {
@@ -511,6 +738,50 @@ func continueRebase(ctx *sql.Context) (string, error) {
 	}, doltSession.Provider(), nil)
 }
 
+// commitManuallyStagedChangesForStep handles committing staged changes after a conflict has been manually
+// resolved by the caller before rebasing has been continued. This involves building the correct commit
+// message based on the details of the rebase plan |step| and then creating the commit.
+func commitManuallyStagedChangesForStep(ctx *sql.Context, step rebase.RebasePlanStep) error {
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	workingSet, err := doltSession.WorkingSet(ctx, ctx.GetCurrentDatabase())
+	if err != nil {
+		return err
+	}
+
+	options, err := createCherryPickOptionsForRebaseStep(ctx, &step, workingSet.RebaseState().CommitBecomesEmptyHandling(),
+		workingSet.RebaseState().EmptyCommitHandling())
+
+	commitProps, err := cherry_pick.CreateCommitStagedPropsFromCherryPickOptions(ctx, *options)
+	if err != nil {
+		return err
+	}
+
+	// If the commit message wasn't set when we created the cherry-pick options, then set it to the step's commit
+	// message. For fixup commits, we don't use their commit message, so we keep it empty, and let the amend commit
+	// codepath use the previous commit's message.
+	if commitProps.Message == "" && step.Action != rebase.RebaseActionFixup {
+		commitProps.Message = step.CommitMsg
+	}
+
+	roots, ok := doltSession.GetRoots(ctx, ctx.GetCurrentDatabase())
+	if !ok {
+		return fmt.Errorf("unable to get roots for current session")
+	}
+	pendingCommit, err := doltSession.NewPendingCommit(ctx, ctx.GetCurrentDatabase(), roots, *commitProps)
+	if err != nil {
+		return err
+	}
+
+	// Ensure a SQL transaction is set in the session
+	if doltSession.GetTransaction() == nil {
+		if _, err = doltSession.StartTransaction(ctx, sql.ReadWrite); err != nil {
+			return err
+		}
+	}
+	_, err = doltSession.DoltCommit(ctx, ctx.GetCurrentDatabase(), doltSession.GetTransaction(), pendingCommit)
+	return err
+}
+
 func processRebasePlanStep(ctx *sql.Context, planStep *rebase.RebasePlanStep,
 	commitBecomesEmptyHandling doltdb.EmptyCommitHandling, emptyCommitHandling doltdb.EmptyCommitHandling) error {
 	// Make sure we have a transaction opened for the session
@@ -524,6 +795,20 @@ func processRebasePlanStep(ctx *sql.Context, planStep *rebase.RebasePlanStep,
 		}
 	}
 
+	// If the action is "drop", then we don't need to do anything
+	if planStep.Action == rebase.RebaseActionDrop {
+		return nil
+	}
+
+	options, err := createCherryPickOptionsForRebaseStep(ctx, planStep, commitBecomesEmptyHandling, emptyCommitHandling)
+	if err != nil {
+		return err
+	}
+
+	return handleRebaseCherryPick(ctx, planStep, *options)
+}
+
+func createCherryPickOptionsForRebaseStep(ctx *sql.Context, planStep *rebase.RebasePlanStep, commitBecomesEmptyHandling doltdb.EmptyCommitHandling, emptyCommitHandling doltdb.EmptyCommitHandling) (*cherry_pick.CherryPickOptions, error) {
 	// Override the default empty commit handling options for cherry-pick, since
 	// rebase has slightly different defaults
 	options := cherry_pick.NewCherryPickOptions()
@@ -531,51 +816,81 @@ func processRebasePlanStep(ctx *sql.Context, planStep *rebase.RebasePlanStep,
 	options.EmptyCommitHandling = emptyCommitHandling
 
 	switch planStep.Action {
-	case rebase.RebaseActionDrop:
-		return nil
+	case rebase.RebaseActionDrop, rebase.RebaseActionPick:
+		// Nothing to do – the drop action doesn't result in a cherry pick and the pick action
+		// doesn't require any special options (i.e. no amend, no custom commit message).
 
-	case rebase.RebaseActionPick, rebase.RebaseActionReword:
-		if planStep.Action == rebase.RebaseActionReword {
-			options.CommitMessage = planStep.CommitMsg
-		}
-		return handleRebaseCherryPick(ctx, planStep.CommitHash, options)
+	case rebase.RebaseActionReword:
+		options.CommitMessage = planStep.CommitMsg
 
-	case rebase.RebaseActionSquash, rebase.RebaseActionFixup:
+	case rebase.RebaseActionSquash:
 		options.Amend = true
-		if planStep.Action == rebase.RebaseActionSquash {
-			commitMessage, err := squashCommitMessage(ctx, planStep.CommitHash)
-			if err != nil {
-				return err
-			}
-			options.CommitMessage = commitMessage
+		commitMessage, err := squashCommitMessage(ctx, planStep.CommitHash)
+		if err != nil {
+			return nil, err
 		}
-		return handleRebaseCherryPick(ctx, planStep.CommitHash, options)
+		options.CommitMessage = commitMessage
+
+	case rebase.RebaseActionFixup:
+		options.Amend = true
 
 	default:
-		return fmt.Errorf("rebase action '%s' is not supported", planStep.Action)
+		return nil, fmt.Errorf("rebase action '%s' is not supported", planStep.Action)
 	}
+
+	return &options, nil
 }
 
 // handleRebaseCherryPick runs a cherry-pick for the specified |commitHash|, using the specified
-// cherry-pick |options| and checks the results for any errors or merge conflicts. The initial
-// version of rebase doesn't support conflict resolution, so if any conflicts are detected, the
-// rebase is aborted and an error is returned.
-func handleRebaseCherryPick(ctx *sql.Context, commitHash string, options cherry_pick.CherryPickOptions) error {
-	_, mergeResult, err := cherry_pick.CherryPick(ctx, commitHash, options)
+// cherry-pick |options| and checks the results for any errors or merge conflicts. If a data conflict
+// is detected, then the ErrRebaseDataConflict error is returned. If a schema conflict is detected,
+// then the ErrRebaseSchemaConflict error is returned.
+func handleRebaseCherryPick(ctx *sql.Context, planStep *rebase.RebasePlanStep, options cherry_pick.CherryPickOptions) error {
+	_, mergeResult, err := cherry_pick.CherryPick(ctx, planStep.CommitHash, options)
 
+	// TODO: rebase doesn't support schema conflict resolution yet. Ideally, when a schema conflict
+	//       is detected, the rebase would be paused and the user would resolve the conflict just
+	//       like any other conflict, and then call dolt_rebase --continue to keep going.
 	var schemaConflict merge.SchemaConflict
-	isSchemaConflict := errors.As(err, &schemaConflict)
-
-	if (mergeResult != nil && mergeResult.HasMergeArtifacts()) || isSchemaConflict {
-		// TODO: rebase doesn't currently support conflict resolution, but ideally, when a conflict
-		//       is detected, the rebase would be paused and the user would resolve the conflict just
-		//       like any other conflict, and then call dolt_rebase --continue to keep going.
-		abortErr := abortRebase(ctx)
-		if abortErr != nil {
-			return ErrRebaseConflictWithAbortError.New(commitHash, abortErr)
+	if errors.As(err, &schemaConflict) {
+		if abortErr := abortRebase(ctx); abortErr != nil {
+			return ErrRebaseConflictWithAbortError.New(planStep.CommitHash, abortErr)
 		}
-		return ErrRebaseConflict.New(commitHash)
+		return ErrRebaseSchemaConflict.New(planStep.CommitHash)
 	}
+
+	doltSession := dsess.DSessFromSess(ctx.Session)
+	if mergeResult != nil && mergeResult.HasMergeArtifacts() {
+		if err := validateConflictsCanBeResolved(ctx, planStep); err != nil {
+			return err
+		}
+
+		// If @@dolt_allow_commit_conflicts is enabled, then we need to make a SQL commit here, which
+		// will save the conflicts and make them visible to other sessions. This is necessary for the CLI's
+		// rebase command support. However, it's also valid to use @@autocommit and resolve the data
+		// conflicts within the same session, so in that case, we do NOT make a SQL commit.
+		allowCommitConflictsEnabled, err := isAllowCommitConflictsEnabled(ctx)
+		if err != nil {
+			return err
+		}
+		if allowCommitConflictsEnabled {
+			if doltSession.GetTransaction() == nil {
+				_, err := doltSession.StartTransaction(ctx, sql.ReadWrite)
+				if err != nil {
+					return err
+				}
+			}
+
+			err = doltSession.CommitTransaction(ctx, doltSession.GetTransaction())
+			if err != nil {
+				return err
+			}
+		}
+
+		// Otherwise, let the caller know about the conflict and how to resolve
+		return ErrRebaseDataConflict.New(planStep.CommitHash, planStep.CommitMsg)
+	}
+
 	return err
 }
 
@@ -632,3 +947,55 @@ func currentBranch(ctx *sql.Context) (string, error) {
 	}
 	return headRef.GetPath(), nil
 }
+
+// validateConflictsCanBeResolved checks to see if conflicts can be recorded in the current session, by looking to see
+// if @@autocommit is disabled, or if @@dolt_allow_commit_conflicts has been enabled. If conflicts cannot be recorded
+// then the current rebase is aborted, and an error is returned describing how to change the session settings and
+// restart the rebase.
+func validateConflictsCanBeResolved(ctx *sql.Context, planStep *rebase.RebasePlanStep) error {
+	autocommitEnabled, err := isAutocommitEnabled(ctx)
+	if err != nil {
+		return err
+	}
+
+	allowCommitConflictsEnabled, err := isAllowCommitConflictsEnabled(ctx)
+	if err != nil {
+		return err
+	}
+
+	// If @@autocommit is turned off, or if @@dolt_allow_commit_conflicts is enabled, then the user will
+	// be able to resolve conflicts in the session.
+	if !autocommitEnabled || allowCommitConflictsEnabled {
+		return nil
+	}
+
+	// Otherwise, abort the rebase and return an error message, since conflicts can't be worked with.
+	if abortErr := abortRebase(ctx); abortErr != nil {
+		return ErrRebaseConflictWithAbortError.New(planStep.CommitHash, abortErr)
+	}
+	return ErrRebaseDataConflictsCantBeResolved.New()
+}
+
+// isAutocommitEnabled returns true if @@autocommit is enabled in the current session.
+func isAutocommitEnabled(ctx *sql.Context) (bool, error) {
+	return lookupBoolSysVar(ctx, "autocommit")
+}
+
+// isAllowCommitConflictsEnabled returns true if @@dolt_allow_commit_conflicts is enabled in the current session.
+func isAllowCommitConflictsEnabled(ctx *sql.Context) (bool, error) {
+	return lookupBoolSysVar(ctx, "dolt_allow_commit_conflicts")
+}
+
+// lookupBoolSysVar returns the value of the boolean system variable named |systemVarName| for the current session.
+func lookupBoolSysVar(ctx *sql.Context, systemVarName string) (bool, error) {
+	value, err := ctx.GetSessionVariable(ctx, systemVarName)
+	if err != nil {
+		return false, err
+	}
+	boolValue, _, err := types.Boolean.Convert(value)
+	if err != nil {
+		return false, err
+	}
+
+	return boolValue == int8(1) || boolValue == true, nil
+}

go/libraries/doltcore/sqle/dprocedures/dolt_reset.go

@@ -114,6 +114,11 @@ func doDoltReset(ctx *sql.Context, args []string) (int, error) {
 		if err != nil {
 			return 1, err
 		}
+		err = dSess.ResetGlobals(ctx, dbName, roots.Working)
+		if err != nil {
+			return 1, err
+		}
+
 	} else if apr.Contains(cli.SoftResetParam) {
 		arg := ""
 		if apr.NArg() > 1 {

go/libraries/doltcore/sqle/dsess/autoincrement_tracker.go

@@ -61,41 +61,7 @@ func NewAutoIncrementTracker(ctx context.Context, dbName string, roots ...doltdb
 		sequences: &sync.Map{},
 		mm:        mutexmap.NewMutexMap(),
 	}
-
-	for _, root := range roots {
-		root, err := root.ResolveRootValue(ctx)
-		if err != nil {
-			return &AutoIncrementTracker{}, err
-		}
-
-		err = root.IterTables(ctx, func(tableName doltdb.TableName, table *doltdb.Table, sch schema.Schema) (bool, error) {
-			ok := schema.HasAutoIncrement(sch)
-			if !ok {
-				return false, nil
-			}
-
-			tableName = tableName.ToLower()
-
-			seq, err := table.GetAutoIncrementValue(ctx)
-			if err != nil {
-				return true, err
-			}
-
-			// TODO: support schema name as part of the key
-			tableNameStr := tableName.Name
-			oldValue, loaded := ait.sequences.LoadOrStore(tableNameStr, seq)
-			if loaded && seq > oldValue.(uint64) {
-				ait.sequences.Store(tableNameStr, seq)
-			}
-
-			return false, nil
-		})
-
-		if err != nil {
-			return &AutoIncrementTracker{}, err
-		}
-	}
-
+	ait.InitWithRoots(ctx, roots...)
 	return &ait, nil
 }
 
@@ -109,13 +75,13 @@ func loadAutoIncValue(sequences *sync.Map, tableName string) uint64 {
 }
 
 // Current returns the next value to be generated in the auto increment sequence for the table named
-func (a AutoIncrementTracker) Current(tableName string) uint64 {
+func (a *AutoIncrementTracker) Current(tableName string) uint64 {
 	return loadAutoIncValue(a.sequences, tableName)
 }
 
 // Next returns the next auto increment value for the table named using the provided value from an insert (which may
 // be null or 0, in which case it will be generated from the sequence).
-func (a AutoIncrementTracker) Next(tbl string, insertVal interface{}) (uint64, error) {
+func (a *AutoIncrementTracker) Next(tbl string, insertVal interface{}) (uint64, error) {
 	tbl = strings.ToLower(tbl)
 
 	given, err := CoerceAutoIncrementValue(insertVal)
@@ -145,7 +111,7 @@ func (a AutoIncrementTracker) Next(tbl string, insertVal interface{}) (uint64, e
 	return given, nil
 }
 
-func (a AutoIncrementTracker) CoerceAutoIncrementValue(val interface{}) (uint64, error) {
+func (a *AutoIncrementTracker) CoerceAutoIncrementValue(val interface{}) (uint64, error) {
 	return CoerceAutoIncrementValue(val)
 }
 
@@ -172,7 +138,7 @@ func CoerceAutoIncrementValue(val interface{}) (uint64, error) {
 // Set sets the auto increment value for the table named, if it's greater than the one already registered for this
 // table. Otherwise, the update is silently disregarded. So far this matches the MySQL behavior, but Dolt uses the
 // maximum value for this table across all branches.
-func (a AutoIncrementTracker) Set(ctx *sql.Context, tableName string, table *doltdb.Table, ws ref.WorkingSetRef, newAutoIncVal uint64) (*doltdb.Table, error) {
+func (a *AutoIncrementTracker) Set(ctx *sql.Context, tableName string, table *doltdb.Table, ws ref.WorkingSetRef, newAutoIncVal uint64) (*doltdb.Table, error) {
 	tableName = strings.ToLower(tableName)
 
 	release := a.mm.Lock(tableName)
@@ -190,7 +156,7 @@ func (a AutoIncrementTracker) Set(ctx *sql.Context, tableName string, table *dol
 
 // deepSet sets the auto increment value for the table named, if it's greater than the one on any branch head for this
 // database, ignoring the current in-memory tracker value
-func (a AutoIncrementTracker) deepSet(ctx *sql.Context, tableName string, table *doltdb.Table, ws ref.WorkingSetRef, newAutoIncVal uint64) (*doltdb.Table, error) {
+func (a *AutoIncrementTracker) deepSet(ctx *sql.Context, tableName string, table *doltdb.Table, ws ref.WorkingSetRef, newAutoIncVal uint64) (*doltdb.Table, error) {
 	sess := DSessFromSess(ctx.Session)
 	db, ok := sess.Provider().BaseDatabase(ctx, a.dbName)
 
@@ -371,7 +337,7 @@ func getMaxIndexValue(ctx context.Context, indexData durable.Index) (uint64, err
 }
 
 // AddNewTable initializes a new table with an auto increment column to the tracker, as necessary
-func (a AutoIncrementTracker) AddNewTable(tableName string) {
+func (a *AutoIncrementTracker) AddNewTable(tableName string) {
 	tableName = strings.ToLower(tableName)
 	// only initialize the sequence for this table if no other branch has such a table
 	a.sequences.LoadOrStore(tableName, uint64(1))
@@ -380,7 +346,7 @@ func (a AutoIncrementTracker) AddNewTable(tableName string) {
 // DropTable drops the table with the name given.
 // To establish the new auto increment value, callers must also pass all other working sets in scope that may include
 // a table with the same name, omitting the working set that just deleted the table named.
-func (a AutoIncrementTracker) DropTable(ctx *sql.Context, tableName string, wses ...*doltdb.WorkingSet) error {
+func (a *AutoIncrementTracker) DropTable(ctx *sql.Context, tableName string, wses ...*doltdb.WorkingSet) error {
 	tableName = strings.ToLower(tableName)
 
 	release := a.mm.Lock(tableName)
@@ -430,3 +396,36 @@ func (a *AutoIncrementTracker) AcquireTableLock(ctx *sql.Context, tableName stri
 	a.lockMode = lockMode
 	return a.mm.Lock(tableName), nil
 }
+
+func (a *AutoIncrementTracker) InitWithRoots(ctx context.Context, roots ...doltdb.Rootish) error {
+	for _, root := range roots {
+		r, err := root.ResolveRootValue(ctx)
+		if err != nil {
+			return err
+		}
+
+		err = r.IterTables(ctx, func(tableName doltdb.TableName, table *doltdb.Table, sch schema.Schema) (bool, error) {
+			if !schema.HasAutoIncrement(sch) {
+				return false, nil
+			}
+
+			seq, err := table.GetAutoIncrementValue(ctx)
+			if err != nil {
+				return true, err
+			}
+
+			tableNameStr := tableName.ToLower().Name
+			if oldValue, loaded := a.sequences.LoadOrStore(tableNameStr, seq); loaded && seq > oldValue.(uint64) {
+				a.sequences.Store(tableNameStr, seq)
+			}
+
+			return false, nil
+		})
+
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

go/libraries/doltcore/sqle/dsess/session.go

@@ -563,6 +563,21 @@ func (d *DoltSession) dirtyWorkingSets() []*branchState {
 	return dirtyStates
 }
 
+// DirtyDatabases returns the names of databases who have outstanding changes in this session and need to be committed
+// in a SQL transaction before they are visible to other sessions.
+func (d *DoltSession) DirtyDatabases() []string {
+	var dbNames []string
+	for _, dbState := range d.dbStates {
+		for _, branchState := range dbState.heads {
+			if branchState.dirty {
+				dbNames = append(dbNames, dbState.dbName)
+				break
+			}
+		}
+	}
+	return dbNames
+}
+
 // CommitWorkingSet commits the working set for the transaction given, without creating a new dolt commit.
 // Clients should typically use CommitTransaction, which performs additional checks, instead of this method.
 func (d *DoltSession) CommitWorkingSet(ctx *sql.Context, dbName string, tx sql.Transaction) error {
@@ -707,6 +722,31 @@ func (d *DoltSession) newPendingCommit(ctx *sql.Context, branchState *branchStat
 			mergeParentCommits = append(mergeParentCommits, parentCommit)
 		}
 
+		// If the commit message isn't set and we're amending the previous commit,
+		// go ahead and set the commit message from the current HEAD
+		if props.Message == "" && props.Amend {
+			cs, err := doltdb.NewCommitSpec("HEAD")
+			if err != nil {
+				return nil, err
+			}
+
+			headRef, err := branchState.dbData.Rsr.CWBHeadRef()
+			if err != nil {
+				return nil, err
+			}
+			optCmt, err := branchState.dbData.Ddb.Resolve(ctx, cs, headRef)
+			commit, ok := optCmt.ToCommit()
+			if !ok {
+				return nil, doltdb.ErrGhostCommitEncountered
+			}
+
+			meta, err := commit.GetCommitMeta(ctx)
+			if err != nil {
+				return nil, err
+			}
+			props.Message = meta.Description
+		}
+
 		// TODO: This is not the correct way to write this commit as an amend. While this commit is running
 		//  the branch head moves backwards and concurrency control here is not principled.
 		newRoots, err := actions.ResetSoftToRef(ctx, branchState.dbData, "HEAD~1")
@@ -978,7 +1018,7 @@ func (d *DoltSession) SetStagingRoot(ctx *sql.Context, dbName string, newRoot do
 // via setRoot. This method is for clients that need to update more of the session state, such as the dolt_ functions.
 // Unlike setting the working root, this method always marks the database state dirty.
 func (d *DoltSession) SetRoots(ctx *sql.Context, dbName string, roots doltdb.Roots) error {
-	sessionState, _, err := d.LookupDbState(ctx, dbName)
+	sessionState, _, err := d.lookupDbState(ctx, dbName)
 	if err != nil {
 		return err
 	}
@@ -991,6 +1031,25 @@ func (d *DoltSession) SetRoots(ctx *sql.Context, dbName string, roots doltdb.Roo
 	return d.SetWorkingSet(ctx, dbName, workingSet)
 }
 
+func (d *DoltSession) ResetGlobals(ctx *sql.Context, dbName string, root doltdb.RootValue) error {
+	sessionState, _, err := d.lookupDbState(ctx, dbName)
+	if err != nil {
+		return err
+	}
+
+	tracker, err := sessionState.dbState.globalState.AutoIncrementTracker(ctx)
+	if err != nil {
+		return err
+	}
+
+	err = tracker.InitWithRoots(ctx, root)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (d *DoltSession) SetFileSystem(fs filesys.Filesys) {
 	d.fs = fs
 }
@@ -1019,8 +1078,8 @@ func (d *DoltSession) SetWorkingSet(ctx *sql.Context, dbName string, ws *doltdb.
 		return err
 	}
 
-	if writeSess := branchState.WriteSession(); writeSess != nil {
-		err = writeSess.SetWorkingSet(ctx, ws)
+	if branchState.writeSession != nil {
+		err = branchState.writeSession.SetWorkingSet(ctx, ws)
 		if err != nil {
 			return err
 		}
@@ -1444,9 +1503,10 @@ func (d *DoltSession) dbSessionVarsStale(ctx *sql.Context, state *branchState) b
 	return d.dbCache.CacheSessionVars(state, dtx)
 }
 
-func (d DoltSession) WithGlobals(conf config.ReadWriteConfig) *DoltSession {
-	d.globalsConf = conf
-	return &d
+func (d *DoltSession) WithGlobals(conf config.ReadWriteConfig) *DoltSession {
+	nd := *d
+	nd.globalsConf = conf
+	return &nd
 }
 
 // PersistGlobal implements sql.PersistableSession

go/libraries/doltcore/sqle/enginetest/dolt_queries.go

@@ -5637,6 +5637,89 @@ var DoltAutoIncrementTests = []queries.ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "hard reset dropped table restores auto increment",
+		SetUpScript: []string{
+			"create table t (a int primary key auto_increment, b int)",
+			"insert into t (b) values (1), (2)",
+			"call dolt_commit('-Am', 'initialize table')",
+			"drop table t",
+			"call dolt_reset('--hard')",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "insert into t(b) values (3)",
+				Expected: []sql.Row{
+					{types.OkResult{RowsAffected: 1, InsertID: 3}},
+				},
+			},
+			{
+				Query: "select * from t order by a",
+				Expected: []sql.Row{
+					{1, 1},
+					{2, 2},
+					{3, 3},
+				},
+			},
+		},
+	},
+	{
+		// this behavior aligns with how we treat branches
+		Name: "hard reset inserted rows continues auto increment",
+		SetUpScript: []string{
+			"create table t (a int primary key auto_increment, b int)",
+			"insert into t (b) values (1), (2)",
+			"call dolt_commit('-Am', 'initialize table')",
+			"insert into t (b) values (3), (4)",
+			"call dolt_reset('--hard')",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "insert into t(b) values (5)",
+				Expected: []sql.Row{
+					{types.OkResult{RowsAffected: 1, InsertID: 5}},
+				},
+			},
+			{
+				Query: "select * from t order by a",
+				Expected: []sql.Row{
+					{1, 1},
+					{2, 2},
+					{5, 5},
+				},
+			},
+		},
+	},
+	{
+		Name: "hard reset dropped table with branch restores auto increment",
+		SetUpScript: []string{
+			"create table t (a int primary key auto_increment, b int)",
+			"insert into t (b) values (1), (2)",
+			"call dolt_commit('-Am', 'initialize table')",
+			"call dolt_checkout('-b', 'branch1')",
+			"insert into t values (100, 100)",
+			"call dolt_commit('-Am', 'other')",
+			"call dolt_checkout('main')",
+			"drop table t",
+			"call dolt_reset('--hard')",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "insert into t(b) values (101)",
+				Expected: []sql.Row{
+					{types.OkResult{RowsAffected: 1, InsertID: 101}},
+				},
+			},
+			{
+				Query: "select * from t order by a",
+				Expected: []sql.Row{
+					{1, 1},
+					{2, 2},
+					{101, 101},
+				},
+			},
+		},
+	},
 }
 
 var DoltCherryPickTests = []queries.ScriptTest{

go/libraries/doltcore/sqle/enginetest/dolt_queries_rebase.go

@@ -62,16 +62,16 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 		},
 		Assertions: []queries.ScriptTestAssertion{
 			{
-				Query:          "call dolt_rebase('-i', 'main');",
-				ExpectedErrStr: dprocedures.ErrRebaseUncommittedChanges.Error(),
+				Query:       "call dolt_rebase('-i', 'main');",
+				ExpectedErr: dprocedures.ErrRebaseUncommittedChanges,
 			},
 			{
 				Query:    "call dolt_add('t');",
 				Expected: []sql.Row{{0}},
 			},
 			{
-				Query:          "call dolt_rebase('-i', 'main');",
-				ExpectedErrStr: dprocedures.ErrRebaseUncommittedChanges.Error(),
+				Query:       "call dolt_rebase('-i', 'main');",
+				ExpectedErr: dprocedures.ErrRebaseUncommittedChanges,
 			},
 		},
 	},
@@ -239,6 +239,133 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 			},
 		},
 	},
+	{
+		Name: "dolt_rebase errors: unresolved conflicts",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"insert into t values (1, 'one');",
+			"call dolt_commit('-am', 'inserting row 1 on branch1');",
+			"update t set c1='uno' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1 on branch1');",
+
+			"set @@autocommit=0;",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query:    "delete from dolt_rebase where rebase_order=1;",
+				Expected: []sql.Row{{gmstypes.NewOkResult(1)}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"2", "pick", doltCommit, "updating row 1 on branch1"},
+				},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				Query:    "select * from dolt_conflicts;",
+				Expected: []sql.Row{{"t", uint64(1)}},
+			},
+			{
+				// Trying to --continue a rebase when there are conflicts results in an error.
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseUnresolvedConflicts,
+			},
+		},
+	},
+	{
+		Name: "dolt_rebase errors: uncommitted changes before first --continue",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"insert into t values (1, 'one');",
+			"call dolt_commit('-am', 'inserting row 1 on branch1');",
+			"update t set c1='uno' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1 on branch1');",
+
+			"set @@autocommit=0;",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query:    "insert into t values (100, 'hundo');",
+				Expected: []sql.Row{{gmstypes.NewOkResult(1)}},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseUncommittedChanges,
+			},
+		},
+	},
+	{
+		Name: "dolt_rebase errors: data conflicts when @@autocommit is enabled",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"insert into t values (1, 'one');",
+			"call dolt_commit('-am', 'inserting row 1 on branch1');",
+			"update t set c1='uno' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1 on branch1');",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query: "update dolt_rebase set rebase_order=3 where rebase_order=1;",
+				Expected: []sql.Row{{gmstypes.OkResult{RowsAffected: uint64(1), Info: plan.UpdateInfo{
+					Matched: 1,
+					Updated: 1,
+				}}}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"2", "pick", doltCommit, "updating row 1 on branch1"},
+					{"3.00", "pick", doltCommit, "inserting row 1 on branch1"},
+				},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflictsCantBeResolved,
+			},
+		},
+	},
 	{
 		Name: "dolt_rebase: rebased commit becomes empty; --empty not specified",
 		SetUpScript: []string{
@@ -537,6 +664,17 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 					Updated: 1,
 				}}}},
 			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"1", "pick", doltCommit, "inserting row 1"},
+					{"2", "squash", doltCommit, "inserting row 10"},
+					{"3", "squash", doltCommit, "inserting row 100"},
+					{"4", "drop", doltCommit, "inserting row 1000"},
+					{"5", "reword", doltCommit, "reworded!"},
+					{"6.10", "fixup", doltCommit, "inserting row 100000"},
+				},
+			},
 			{
 				Query:    "call dolt_rebase('--continue');",
 				Expected: []sql.Row{{0, "Successfully rebased and updated refs/heads/branch1"}},
@@ -573,22 +711,89 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 		},
 	},
 	{
-		Name: "dolt_rebase: data conflicts",
+		Name: "dolt_rebase: negative rebase order",
 		SetUpScript: []string{
 			"create table t (pk int primary key, c1 varchar(100));",
 			"call dolt_commit('-Am', 'creating table t');",
 			"call dolt_branch('branch1');",
 
 			"insert into t values (0, 'zero');",
-			"call dolt_commit('-am', 'inserting row 0');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
 
 			"call dolt_checkout('branch1');",
 			"insert into t values (1, 'one');",
-			"call dolt_commit('-am', 'inserting row 1');",
+			"call dolt_commit('-am', 'inserting row 1 on branch1');",
 			"update t set c1='uno' where pk=1;",
-			"call dolt_commit('-am', 'updating row 1');",
+			"call dolt_commit('-am', 'updating row 1 on branch1');",
 			"update t set c1='ein' where pk=1;",
-			"call dolt_commit('-am', 'updating row 1');",
+			"call dolt_commit('-am', 'updating row 1, again, on branch1');",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query: "update dolt_rebase set rebase_order=rebase_order-12.34;",
+				Expected: []sql.Row{{gmstypes.OkResult{RowsAffected: uint64(3), Info: plan.UpdateInfo{
+					Matched: 3,
+					Updated: 3,
+				}}}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"-11.34", "pick", doltCommit, "inserting row 1 on branch1"},
+					{"-10.34", "pick", doltCommit, "updating row 1 on branch1"},
+					{"-9.34", "pick", doltCommit, "updating row 1, again, on branch1"},
+				},
+			},
+			{
+				Query:    "call dolt_rebase('--continue');",
+				Expected: []sql.Row{{0, "Successfully rebased and updated refs/heads/branch1"}},
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "ein"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"updating row 1, again, on branch1"},
+					{"updating row 1 on branch1"},
+					{"inserting row 1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+		},
+	},
+	{
+		Name: "dolt_rebase: data conflicts with pick",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"insert into t values (1, 'one');",
+			"call dolt_commit('-am', 'inserting row 1 on branch1');",
+			"update t set c1='uno' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1 on branch1');",
+			"update t set c1='ein' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1, again, on branch1');",
+
+			"set @@autocommit=0;",
 		},
 		Assertions: []queries.ScriptTestAssertion{
 			{
@@ -600,9 +805,9 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 			{
 				Query: "select * from dolt_rebase order by rebase_order ASC;",
 				Expected: []sql.Row{
-					{"1", "pick", doltCommit, "inserting row 1"},
-					{"2", "pick", doltCommit, "updating row 1"},
-					{"3", "pick", doltCommit, "updating row 1"},
+					{"1", "pick", doltCommit, "inserting row 1 on branch1"},
+					{"2", "pick", doltCommit, "updating row 1 on branch1"},
+					{"3", "pick", doltCommit, "updating row 1, again, on branch1"},
 				},
 			},
 			{
@@ -613,24 +818,592 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 				}}}},
 			},
 			{
-				// Encountering a conflict during a rebase returns an error and aborts the rebase
 				Query:       "call dolt_rebase('--continue');",
-				ExpectedErr: dprocedures.ErrRebaseConflict,
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
 			},
 			{
-				// The rebase state has been cleared after hitting a conflict
-				Query:          "call dolt_rebase('--continue');",
-				ExpectedErrStr: "no rebase in progress",
+				// We remain on the rebase working branch while resolving conflicts
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query:    "select * from dolt_conflicts;",
+				Expected: []sql.Row{{"t", uint64(1)}},
+			},
+			{
+				// The base of the cherry-picked commit has (1, "one"), but ours doesn't have that record (nil, nil)
+				// since we reordered the insert. The cherry-picked commit is trying to modify the row to (1, "uno").
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{1, "one", nil, nil, "removed", 1, "uno", "modified"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Resolve conflicts, by accepting theirs, which inserts (1, "uno") into t
+				// When we continue the rebase, a Dolt commit will be created for these changes
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				// Our new commit shows up as the first commit on top of the latest commit from the upstream branch
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Table t includes the change from the tip of main (0, "zero"), as well as the conflict we just resolved
+				Query:    "SELECT * FROM t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "uno"}},
+			},
+			{
+				// If we don't stage the table, then rebase will give us an error about having unstaged changes
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseUnstagedChanges,
+			},
+			{
+				Query:    "call dolt_add('t');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "ein"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"updating row 1, again, on branch1"},
+					{"updating row 1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Now we're resolving a conflict from reordering the insert of (1, "one"). This was originally
+				// an insert, so the base has (nil, nil), ours is (1, "ein").
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{nil, nil, 1, "ein", "added", 1, "one", "added"}},
+			},
+			{
+				// Accept the new values from the cherry-picked commit (1, "ein").
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				// We can commit manually, or we can continue the rebase and let it commit for us
+				Query:    "CALL DOLT_COMMIT('-am', 'OVERRIDDEN COMMIT MESSAGE');",
+				Expected: []sql.Row{{doltCommit}},
+			},
+			{
+				Query:    "call dolt_rebase('--continue');",
+				Expected: []sql.Row{{0, "Successfully rebased and updated refs/heads/branch1"}},
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "one"}},
 			},
 			{
-				// We're back to the original branch
 				Query:    "select active_branch();",
 				Expected: []sql.Row{{"branch1"}},
 			},
 			{
-				// The conflicts table should be empty, since the rebase was aborted
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"OVERRIDDEN COMMIT MESSAGE"},
+					{"updating row 1, again, on branch1"},
+					{"updating row 1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+		},
+	},
+	{
+		Name: "dolt_rebase: data conflicts with squash",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"insert into t values (1, 'one');",
+			"call dolt_commit('-am', 'inserting row 1 on branch1');",
+			"update t set c1='uno' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1 on branch1');",
+			"update t set c1='ein' where pk=1;",
+			"call dolt_commit('-am', 'updating row 1, again, on branch1');",
+
+			"set @@autocommit=0;",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"1", "pick", doltCommit, "inserting row 1 on branch1"},
+					{"2", "pick", doltCommit, "updating row 1 on branch1"},
+					{"3", "pick", doltCommit, "updating row 1, again, on branch1"},
+				},
+			},
+			{
+				Query: "update dolt_rebase set rebase_order=3.5 where rebase_order=2;",
+				Expected: []sql.Row{{gmstypes.OkResult{RowsAffected: uint64(1), Info: plan.UpdateInfo{
+					Matched: 1,
+					Updated: 1,
+				}}}},
+			},
+			{
+				Query: "update dolt_rebase set action='squash' where rebase_order=3;",
+				Expected: []sql.Row{{gmstypes.OkResult{RowsAffected: uint64(1), Info: plan.UpdateInfo{
+					Matched: 1,
+					Updated: 1,
+				}}}},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				// We remain on the rebase working branch while resolving conflicts
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
 				Query:    "select * from dolt_conflicts;",
-				Expected: []sql.Row{},
+				Expected: []sql.Row{{"t", uint64(1)}},
+			},
+			{
+				// The base of the cherry-picked commit has (1, "uno"), but ours has (1, "one"), so this is a data
+				// conflict. The cherry-picked commit is trying to modify the row to (1, "ein").
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{1, "uno", 1, "one", "modified", 1, "ein", "modified"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Resolve conflicts, by accepting theirs, which inserts (1, "ein") into t
+				// When we continue the rebase, a Dolt commit will be created for these changes
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				// Table t includes the change from the tip of main (0, "zero"), as well as the conflict we just resolved
+				Query:    "SELECT * FROM t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "ein"}},
+			},
+			{
+				// If we don't stage the table, then rebase will give us an error about having unstaged changes
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseUnstagedChanges,
+			},
+			{
+				Query:    "call dolt_add('t');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "ein"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 1 on branch1\n\nupdating row 1, again, on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Now we're resolving a conflict from updating (1, "one") to (1, "uno"). Our side currently has
+				// (1, "ein"), so this is marked as a conflict.
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{1, "one", 1, "ein", "modified", 1, "uno", "modified"}},
+			},
+			{
+				// Accept the new values from the cherry-picked commit (1, "uno").
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "CALL DOLT_COMMIT('-am', 'OVERRIDDEN COMMIT MESSAGE');",
+				Expected: []sql.Row{{doltCommit}},
+			},
+			{
+				Query:    "call dolt_rebase('--continue');",
+				Expected: []sql.Row{{0, "Successfully rebased and updated refs/heads/branch1"}},
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {1, "uno"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"OVERRIDDEN COMMIT MESSAGE"},
+					{"inserting row 1 on branch1\n\nupdating row 1, again, on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+		},
+	},
+	{
+		Name: "dolt_rebase: data conflicts with fixup",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"insert into t values (-1, 'negative');",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"update t set c1='-1' where pk=-1;",
+			"call dolt_commit('-am', 'updating row -1 on branch1');",
+			"delete from t where c1 = '-1';",
+			"call dolt_commit('-am', 'deleting -1 on branch1');",
+			"insert into t values (999, 'nines');",
+			"call dolt_commit('-am', 'inserting row 999 on branch1');",
+
+			"set @@autocommit=0;",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"1", "pick", doltCommit, "updating row -1 on branch1"},
+					{"2", "pick", doltCommit, "deleting -1 on branch1"},
+					{"3", "pick", doltCommit, "inserting row 999 on branch1"},
+				},
+			},
+			{
+				Query: "update dolt_rebase set rebase_order=3.5, action='fixup' where rebase_order=1;",
+				Expected: []sql.Row{{gmstypes.OkResult{RowsAffected: uint64(1), Info: plan.UpdateInfo{
+					Matched: 1,
+					Updated: 1,
+				}}}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"2", "pick", doltCommit, "deleting -1 on branch1"},
+					{"3", "pick", doltCommit, "inserting row 999 on branch1"},
+					{"3.50", "fixup", doltCommit, "updating row -1 on branch1"},
+				},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				// We remain on the rebase working branch while resolving conflicts
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query:    "select * from dolt_conflicts;",
+				Expected: []sql.Row{{"t", uint64(1)}},
+			},
+			{
+				// The base of the cherry-picked commit has (-1, "-1"), but ours has (-1, "negative"), so this is a
+				// data conflict. The cherry-picked commit is trying to delete the row, but can't find an exact match.
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{-1, "-1", -1, "negative", "modified", nil, nil, "removed"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Resolve conflicts, by accepting theirs, which inserts (1, "ein") into t
+				// When we continue the rebase, a Dolt commit will be created for these changes
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "SELECT * FROM t;",
+				Expected: []sql.Row{{0, "zero"}},
+			},
+			{
+				// If we don't stage the table, then rebase will give us an error about having unstaged changes
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseUnstagedChanges,
+			},
+			{
+				Query:    "call dolt_add('t');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {999, "nines"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 999 on branch1"},
+					{"deleting -1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Now we're resolving a conflict where row -1 is updated, but it has already been deleted
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{-1, "negative", nil, nil, "removed", -1, "-1", "modified"}},
+			},
+			{
+				// Accept the new values from the cherry-picked commit (1, "uno").
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "call dolt_add('t');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "call dolt_rebase('--continue');",
+				Expected: []sql.Row{{0, "Successfully rebased and updated refs/heads/branch1"}},
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{-1, "-1"}, {0, "zero"}, {999, "nines"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 999 on branch1"},
+					{"deleting -1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+		},
+	},
+	{
+		Name: "dolt_rebase: data conflicts with reword",
+		SetUpScript: []string{
+			"create table t (pk int primary key, c1 varchar(100));",
+			"insert into t values (-1, 'negative');",
+			"call dolt_commit('-Am', 'creating table t');",
+			"call dolt_branch('branch1');",
+			"insert into t values (0, 'zero');",
+			"call dolt_commit('-am', 'inserting row 0 on main');",
+
+			"call dolt_checkout('branch1');",
+			"update t set c1='-1' where pk=-1;",
+			"call dolt_commit('-am', 'updating row -1 on branch1');",
+			"delete from t where c1 = '-1';",
+			"call dolt_commit('-am', 'deleting -1 on branch1');",
+			"insert into t values (999, 'nines');",
+			"call dolt_commit('-am', 'inserting row 999 on branch1');",
+
+			"set @@autocommit=0;",
+		},
+		Assertions: []queries.ScriptTestAssertion{
+			{
+				Query: "call dolt_rebase('-i', 'main');",
+				Expected: []sql.Row{{0, "interactive rebase started on branch dolt_rebase_branch1; " +
+					"adjust the rebase plan in the dolt_rebase table, then " +
+					"continue rebasing by calling dolt_rebase('--continue')"}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"1", "pick", doltCommit, "updating row -1 on branch1"},
+					{"2", "pick", doltCommit, "deleting -1 on branch1"},
+					{"3", "pick", doltCommit, "inserting row 999 on branch1"},
+				},
+			},
+			{
+				Query: "update dolt_rebase set rebase_order=3.5, action='reword', commit_message='reworded message!' where rebase_order=1;",
+				Expected: []sql.Row{{gmstypes.OkResult{RowsAffected: uint64(1), Info: plan.UpdateInfo{
+					Matched: 1,
+					Updated: 1,
+				}}}},
+			},
+			{
+				Query: "select * from dolt_rebase order by rebase_order ASC;",
+				Expected: []sql.Row{
+					{"2", "pick", doltCommit, "deleting -1 on branch1"},
+					{"3", "pick", doltCommit, "inserting row 999 on branch1"},
+					{"3.50", "reword", doltCommit, "reworded message!"},
+				},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				// We remain on the rebase working branch while resolving conflicts
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"dolt_rebase_branch1"}},
+			},
+			{
+				Query:    "select * from dolt_conflicts;",
+				Expected: []sql.Row{{"t", uint64(1)}},
+			},
+			{
+				// The base of the cherry-picked commit has (-1, "-1"), but ours has (-1, "negative"), so this is a
+				// data conflict. The cherry-picked commit is trying to delete the row, but can't find an exact match.
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{-1, "-1", -1, "negative", "modified", nil, nil, "removed"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Resolve conflicts, by accepting theirs, which inserts (1, "ein") into t
+				// When we continue the rebase, a Dolt commit will be created for these changes
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "SELECT * FROM t;",
+				Expected: []sql.Row{{0, "zero"}},
+			},
+			{
+				// If we don't stage the table, then rebase will give us an error about having unstaged changes
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseUnstagedChanges,
+			},
+			{
+				Query:    "call dolt_add('t');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:       "call dolt_rebase('--continue');",
+				ExpectedErr: dprocedures.ErrRebaseDataConflict,
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{0, "zero"}, {999, "nines"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"inserting row 999 on branch1"},
+					{"deleting -1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
+			},
+			{
+				// Now we're resolving a conflict where row -1 is updated, but it has already been deleted
+				Query:    "select base_pk, base_c1, our_pk, our_c1, our_diff_type, their_pk, their_c1, their_diff_type from dolt_conflicts_t;",
+				Expected: []sql.Row{{-1, "negative", nil, nil, "removed", -1, "-1", "modified"}},
+			},
+			{
+				// Accept the new values from the cherry-picked commit (-1, "-1")
+				Query:    "CALL DOLT_CONFLICTS_RESOLVE('--theirs', 't');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "call dolt_add('t');",
+				Expected: []sql.Row{{0}},
+			},
+			{
+				Query:    "call dolt_rebase('--continue');",
+				Expected: []sql.Row{{0, "Successfully rebased and updated refs/heads/branch1"}},
+			},
+			{
+				Query:    "select * from t;",
+				Expected: []sql.Row{{-1, "-1"}, {0, "zero"}, {999, "nines"}},
+			},
+			{
+				Query:    "select active_branch();",
+				Expected: []sql.Row{{"branch1"}},
+			},
+			{
+				Query: "select message from dolt_log;",
+				Expected: []sql.Row{
+					{"reworded message!"},
+					{"inserting row 999 on branch1"},
+					{"deleting -1 on branch1"},
+					{"inserting row 0 on main"},
+					{"creating table t"},
+					{"Initialize data repository"},
+				},
 			},
 		},
 	},
@@ -677,7 +1450,7 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 			{
 				// Encountering a conflict during a rebase returns an error and aborts the rebase
 				Query:       "call dolt_rebase('--continue');",
-				ExpectedErr: dprocedures.ErrRebaseConflict,
+				ExpectedErr: dprocedures.ErrRebaseSchemaConflict,
 			},
 			{
 				// The rebase state has been cleared after hitting a conflict
@@ -749,6 +1522,8 @@ var DoltRebaseScriptTests = []queries.ScriptTest{
 				Expected: []sql.Row{{gmstypes.NewOkResult(2)}},
 			},
 			{
+				// NOTE: This uses "pick", not reword, so we expect the commit message from the commit to be
+				//       used, and not the custom commit message inserted into the table.
 				Query:    "insert into dolt_rebase values (2.12, 'pick', hashof('branch2'), 'inserting row 0');",
 				Expected: []sql.Row{{gmstypes.NewOkResult(1)}},
 			},

go/libraries/doltcore/sqle/globalstate/auto_increment_tracker.go

@@ -15,6 +15,8 @@
 package globalstate
 
 import (
+	"context"
+
 	"github.com/dolthub/go-mysql-server/sql"
 
 	"github.com/dolthub/dolt/go/libraries/doltcore/doltdb"
@@ -38,9 +40,10 @@ type AutoIncrementTracker interface {
 	// below the current value for this table. The table in the provided working set is assumed to already have the value
 	// given, so the new global maximum is computed without regard for its value in that working set.
 	Set(ctx *sql.Context, tableName string, table *doltdb.Table, ws ref.WorkingSetRef, newAutoIncVal uint64) (*doltdb.Table, error)
-
 	// AcquireTableLock acquires the auto increment lock on a table, and returns a callback function to release the lock.
 	// Depending on the value of the `innodb_autoinc_lock_mode` system variable, the engine may need to acquire and hold
 	// the lock for the duration of an insert statement.
 	AcquireTableLock(ctx *sql.Context, tableName string) (func(), error)
+	// InitWithRoots fills the AutoIncrementTracker with values pulled from each root in order.
+	InitWithRoots(ctx context.Context, roots ...doltdb.Rootish) error
 }

go/libraries/doltcore/sqle/index/index_reader.go

@@ -25,6 +25,7 @@ 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/row"
+	"github.com/dolthub/dolt/go/libraries/doltcore/schema"
 	"github.com/dolthub/dolt/go/libraries/doltcore/table/typed/noms"
 	"github.com/dolthub/dolt/go/store/prolly"
 	"github.com/dolthub/dolt/go/store/prolly/tree"
@@ -249,6 +250,9 @@ type IndexScanBuilder interface {
 
 	// Key returns the table root for caching purposes
 	Key() doltdb.DataCacheKey
+
+	// OutputSchema returns the output KV tuple schema
+	OutputSchema() schema.Schema
 }
 
 func NewIndexReaderBuilder(
@@ -374,6 +378,10 @@ func (ib *baseIndexImplBuilder) Key() doltdb.DataCacheKey {
 	return ib.key
 }
 
+func (ib *baseIndexImplBuilder) OutputSchema() schema.Schema {
+	return ib.idx.IndexSchema()
+}
+
 // NewPartitionRowIter implements IndexScanBuilder
 func (ib *baseIndexImplBuilder) NewPartitionRowIter(_ *sql.Context, _ sql.Partition) (sql.RowIter, error) {
 	panic("cannot call NewRowIter on baseIndexImplBuilder")
@@ -479,6 +487,10 @@ func (i *sequenceRangeIter) Next(ctx context.Context) (val.Tuple, val.Tuple, err
 	return k, v, nil
 }
 
+func (ib *coveringIndexImplBuilder) OutputSchema() schema.Schema {
+	return ib.idx.IndexSchema()
+}
+
 // NewRangeMapIter implements IndexScanBuilder
 func (ib *coveringIndexImplBuilder) NewRangeMapIter(ctx context.Context, r prolly.Range, reverse bool) (prolly.MapIter, error) {
 	if reverse {
@@ -562,6 +574,11 @@ func (i *nonCoveringMapIter) Next(ctx context.Context) (val.Tuple, val.Tuple, er
 	return pk, value, nil
 }
 
+func (ib *nonCoveringIndexImplBuilder) OutputSchema() schema.Schema {
+	// this refs the table schema
+	return ib.baseIndexImplBuilder.idx.Schema()
+}
+
 // NewRangeMapIter implements IndexScanBuilder
 func (ib *nonCoveringIndexImplBuilder) NewRangeMapIter(ctx context.Context, r prolly.Range, reverse bool) (prolly.MapIter, error) {
 	var secIter prolly.MapIter
@@ -619,6 +636,10 @@ type keylessIndexImplBuilder struct {
 	s *durableIndexState
 }
 
+func (ib *keylessIndexImplBuilder) OutputSchema() schema.Schema {
+	return ib.idx.Schema()
+}
+
 // NewRangeMapIter implements IndexScanBuilder
 func (ib *keylessIndexImplBuilder) NewRangeMapIter(ctx context.Context, r prolly.Range, reverse bool) (prolly.MapIter, error) {
 	rows := ib.s.Primary
@@ -709,6 +730,10 @@ type nomsIndexImplBuilder struct {
 	s *durableIndexState
 }
 
+func (ib *nomsIndexImplBuilder) OutputSchema() schema.Schema {
+	return ib.baseIndexImplBuilder.idx.Schema()
+}
+
 // NewPartitionRowIter implements IndexScanBuilder
 func (ib *nomsIndexImplBuilder) NewPartitionRowIter(ctx *sql.Context, part sql.Partition) (sql.RowIter, error) {
 	p := part.(rangePartition)

go/libraries/doltcore/sqle/kvexec/builder.go

@@ -43,7 +43,7 @@ func (b Builder) Build(ctx *sql.Context, n sql.Node, r sql.Row) (sql.RowIter, er
 			if ita, ok := getIta(n.Right()); ok && len(r) == 0 && simpleLookupExpressions(ita.Expressions()) {
 				if _, _, dstIter, _, dstTags, dstFilter, err := getSourceKv(ctx, n.Right(), false); err == nil && dstIter != nil {
 					if srcMap, srcIter, _, srcSchema, srcTags, srcFilter, err := getSourceKv(ctx, n.Left(), true); err == nil && srcSchema != nil {
-						if keyLookupMapper := newLookupKeyMapping(ctx, srcSchema, srcMap, dstIter.InputKeyDesc(), ita.Expressions()); keyLookupMapper.valid() {
+						if keyLookupMapper := newLookupKeyMapping(ctx, srcSchema, dstIter.InputKeyDesc(), ita.Expressions(), srcMap.NodeStore()); keyLookupMapper.valid() {
 							// conditions:
 							// (1) lookup or left lookup join
 							// (2) left-side is something we read KVs from (table or indexscan, ex: no subqueries)
@@ -315,6 +315,8 @@ func getSourceKv(ctx *sql.Context, n sql.Node, isSrc bool) (prolly.Map, prolly.M
 		}
 		indexMap = durable.ProllyMapFromIndex(rowData)
 
+		sch = lb.OutputSchema()
+
 		if isSrc {
 			l, err := n.GetLookup(ctx, nil)
 			if err != nil {

go/libraries/doltcore/sqle/kvexec/lookup_join.go

@@ -212,7 +212,7 @@ type lookupMapping struct {
 	pool pool.BuffPool
 }
 
-func newLookupKeyMapping(ctx context.Context, sourceSch schema.Schema, src prolly.Map, tgtKeyDesc val.TupleDesc, keyExprs []sql.Expression) *lookupMapping {
+func newLookupKeyMapping(ctx context.Context, sourceSch schema.Schema, tgtKeyDesc val.TupleDesc, keyExprs []sql.Expression, ns tree.NodeStore) *lookupMapping {
 	keyless := schema.IsKeyless(sourceSch)
 	// |split| is an index into the schema separating the key and value fields
 	var split int
@@ -254,12 +254,12 @@ func newLookupKeyMapping(ctx context.Context, sourceSch schema.Schema, src proll
 	litDesc := val.NewTupleDescriptor(litTypes...)
 	litTb := val.NewTupleBuilder(litDesc)
 	for i, j := range litMappings {
-		tree.PutField(ctx, src.NodeStore(), litTb, i, keyExprs[j].(*expression.Literal).Value())
+		tree.PutField(ctx, ns, litTb, i, keyExprs[j].(*expression.Literal).Value())
 	}
 
 	var litTuple val.Tuple
 	if litDesc.Count() > 0 {
-		litTuple = litTb.Build(src.Pool())
+		litTuple = litTb.Build(ns.Pool())
 	}
 
 	return &lookupMapping{
@@ -267,11 +267,11 @@ func newLookupKeyMapping(ctx context.Context, sourceSch schema.Schema, src proll
 		srcMapping: srcMapping,
 		litTuple:   litTuple,
 		litKd:      litDesc,
-		srcKd:      src.KeyDesc(),
-		srcVd:      src.ValDesc(),
+		srcKd:      sourceSch.GetKeyDescriptor(),
+		srcVd:      sourceSch.GetValueDescriptor(),
 		targetKb:   val.NewTupleBuilder(tgtKeyDesc),
-		ns:         src.NodeStore(),
-		pool:       src.Pool(),
+		ns:         ns,
+		pool:       ns.Pool(),
 	}
 }
 

go/libraries/doltcore/sqle/system_variables.go

@@ -29,6 +29,232 @@ func init() {
 	AddDoltSystemVariables()
 }
 
+var DoltSystemVariables = []sql.SystemVariable{
+	&sql.MysqlSystemVariable{
+		Name:              "log_bin_branch",
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Persist),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType("log_bin_branch"),
+		Default:           "main",
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.DoltOverrideSchema,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.DoltOverrideSchema),
+		Default:           "",
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ReplicateToRemote,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.ReplicateToRemote),
+		Default:           "",
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ReplicationRemoteURLTemplate,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.ReplicationRemoteURLTemplate),
+		Default:           "",
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ReadReplicaRemote,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.ReadReplicaRemote),
+		Default:           "",
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ReadReplicaForcePull,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.ReadReplicaForcePull),
+		Default:           int8(1),
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.SkipReplicationErrors,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.SkipReplicationErrors),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ReplicateHeads,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.ReplicateHeads),
+		Default:           "",
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ReplicateAllHeads,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.ReplicateAllHeads),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.AsyncReplication,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.AsyncReplication),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{ // If true, causes a Dolt commit to occur when you commit a transaction.
+		Name:              dsess.DoltCommitOnTransactionCommit,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.DoltCommitOnTransactionCommit),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{ // If set, use this message for automatic Dolt commits
+		Name:              dsess.DoltCommitOnTransactionCommitMessage,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.DoltCommitOnTransactionCommitMessage),
+		Default:           nil,
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.TransactionsDisabledSysVar,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Session),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.TransactionsDisabledSysVar),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{ // If true, disables the conflict and constraint violation check when you commit a transaction.
+		Name:              dsess.ForceTransactionCommit,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.ForceTransactionCommit),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.CurrentBatchModeKey,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Session),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemIntType(dsess.CurrentBatchModeKey, -9223372036854775808, 9223372036854775807, false),
+		Default:           int64(0),
+	},
+	&sql.MysqlSystemVariable{ // If true, disables the conflict violation check when you commit a transaction.
+		Name:              dsess.AllowCommitConflicts,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Session),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.AllowCommitConflicts),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.AwsCredsFile,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Session),
+		Dynamic:           false,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.AwsCredsFile),
+		Default:           nil,
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.AwsCredsProfile,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Session),
+		Dynamic:           false,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.AwsCredsProfile),
+		Default:           nil,
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.AwsCredsRegion,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Session),
+		Dynamic:           false,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemStringType(dsess.AwsCredsRegion),
+		Default:           nil,
+	},
+	&sql.MysqlSystemVariable{
+		Name:              dsess.ShowBranchDatabases,
+		Scope:             sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Dynamic:           true,
+		SetVarHintApplies: false,
+		Type:              types.NewSystemBoolType(dsess.ShowBranchDatabases),
+		Default:           int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltClusterAckWritesTimeoutSecs,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Persist),
+		Type:    types.NewSystemIntType(dsess.DoltClusterAckWritesTimeoutSecs, 0, 60, false),
+		Default: int64(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.ShowSystemTables,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Type:    types.NewSystemBoolType(dsess.ShowSystemTables),
+		Default: int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    "dolt_dont_merge_json",
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Both),
+		Type:    types.NewSystemBoolType("dolt_dont_merge_json"),
+		Default: int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltStatsAutoRefreshEnabled,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Type:    types.NewSystemBoolType(dsess.DoltStatsAutoRefreshEnabled),
+		Default: int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltStatsBootstrapEnabled,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Type:    types.NewSystemBoolType(dsess.DoltStatsBootstrapEnabled),
+		Default: int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltStatsMemoryOnly,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Type:    types.NewSystemBoolType(dsess.DoltStatsMemoryOnly),
+		Default: int8(0),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltStatsAutoRefreshThreshold,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Type:    types.NewSystemDoubleType(dsess.DoltStatsAutoRefreshEnabled, 0, 10),
+		Default: float64(.5),
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltStatsAutoRefreshInterval,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Type:    types.NewSystemIntType(dsess.DoltStatsAutoRefreshInterval, 0, math.MaxInt, false),
+		Default: 120,
+	},
+	&sql.MysqlSystemVariable{
+		Name:    dsess.DoltStatsBranches,
+		Dynamic: true,
+		Scope:   sql.GetMysqlScope(sql.SystemVariableScope_Global),
+		Type:    types.NewSystemStringType(dsess.DoltStatsBranches),
+		Default: "",
+	},
+}
+
 func AddDoltSystemVariables() {
 	sql.SystemVariables.AddSystemVariables([]sql.SystemVariable{
 		&sql.MysqlSystemVariable{
@@ -269,6 +495,7 @@ func AddDoltSystemVariables() {
 			Default: int8(0),
 		},
 	})
+	sql.SystemVariables.AddSystemVariables(DoltSystemVariables)
 }
 
 func ReadReplicaForcePull() bool {

go/libraries/utils/argparser/parser.go

@@ -343,6 +343,22 @@ func (ap *ArgParser) Parse(args []string) (*ArgParseResults, error) {
 	return &ArgParseResults{namedArgs, positionalArgs, ap, positionalArgsSeparatorIndex}, nil
 }
 
+func (ap *ArgParser) isOptionOrFlag(s string) bool {
+	if len(s) == 0 {
+		return false
+	} else if s[0] != '-' {
+		return false
+	}
+	s = s[1:]
+
+	if len(s) >= 1 && s[0] == '-' {
+		s = s[1:]
+	}
+
+	_, ok := ap.nameOrAbbrevToOpt[s]
+	return ok
+}
+
 func (ap *ArgParser) parseToken(args []string, index int, positionalArgs []string, namedArgs map[string]string) (newIndex int, newPositionalArgs []string, newNamedArgs map[string]string, err error) {
 	arg := args[index]
 
@@ -387,22 +403,31 @@ func (ap *ArgParser) parseToken(args []string, index int, positionalArgs []strin
 	}
 
 	if value == nil {
-		index++
+		next := index + 1
 		valueStr := ""
-		if index >= len(args) {
+		if next >= len(args) {
 			if opt.OptType != OptionalEmptyValue {
 				return 0, nil, nil, errors.New("error: no value for option `" + opt.Name + "'")
 			}
 		} else {
 			if opt.AllowMultipleOptions {
-				list := getListValues(args[index:])
+				list := getListValues(args[next:])
 				valueStr = strings.Join(list, ",")
-				index += len(list) - 1
+				index += len(list)
 			} else {
-				valueStr = args[index]
+				nextArg := args[next]
+				if opt.OptType == OptionalEmptyValue {
+					if !(nextArg == "--" || ap.isOptionOrFlag(nextArg)) {
+						valueStr = args[next]
+						index = next
+					}
+				} else {
+					valueStr = args[next]
+					index = next
+				}
 			}
+			value = &valueStr
 		}
-		value = &valueStr
 	}
 
 	if opt.Validator != nil {
@@ -413,6 +438,10 @@ func (ap *ArgParser) parseToken(args []string, index int, positionalArgs []strin
 		}
 	}
 
+	if value == nil {
+		value = new(string)
+	}
+
 	namedArgs[opt.Name] = *value
 	return index, positionalArgs, namedArgs, nil
 }

go/libraries/utils/argparser/parser_test.go

@@ -15,13 +15,21 @@
 package argparser
 
 import (
-	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+func createParserWithOptionalArgs() *ArgParser {
+	ap := NewArgParserWithMaxArgs("test", 16)
+	ap.SupportsFlag("flag", "f", "flag")
+	ap.SupportsString("param", "p", "param", "")
+	ap.SupportsOptionalString("optional", "o", "optional", "")
+
+	return ap
+}
+
 func TestArgParser(t *testing.T) {
 	tests := []struct {
 		ap              *ArgParser
@@ -94,10 +102,52 @@ func TestArgParser(t *testing.T) {
 			[]string{},
 		},
 		{
-			NewArgParserWithMaxArgs("test", 1),
+			createParserWithOptionalArgs(),
 			[]string{"foo", "bar"},
-			errors.New("error: test has too many positional arguments. Expected at most 1, found 2: foo, bar"),
+			nil,
 			map[string]string{},
+			[]string{"foo", "bar"},
+		},
+		{
+			createParserWithOptionalArgs(),
+			[]string{"-o", "-f", "foo", "bar"},
+			nil,
+			map[string]string{"flag": "", "optional": ""},
+			[]string{"foo", "bar"},
+		},
+		{
+			createParserWithOptionalArgs(),
+			[]string{"-o", "optional value", "-f", "foo", "bar"},
+			nil,
+			map[string]string{"flag": "", "optional": "optional value"},
+			[]string{"foo", "bar"},
+		},
+		{
+			createParserWithOptionalArgs(),
+			[]string{"-o", "--", "foo", "bar"},
+			nil,
+			map[string]string{"optional": ""},
+			[]string{"foo", "bar"},
+		},
+		{
+			createParserWithOptionalArgs(),
+			[]string{"-p", "value", "-o"},
+			nil,
+			map[string]string{"param": "value", "optional": ""},
+			[]string{},
+		},
+		{
+			createParserWithOptionalArgs(),
+			[]string{"-p", "value", "-o", "--"},
+			nil,
+			map[string]string{"param": "value", "optional": ""},
+			[]string{},
+		},
+		{
+			createParserWithOptionalArgs(),
+			[]string{"-o", "-p", "value"},
+			nil,
+			map[string]string{"param": "value", "optional": ""},
 			[]string{},
 		},
 	}

go/serial/workingset.fbs

@@ -58,6 +58,15 @@ table RebaseState {
 
   // How to handle commits that become empty during rebasing
   commit_becomes_empty_handling:uint8;
+
+  // The last_attempted_step field indicates which step in the rebase plan was last executed. The step may have ended
+  // with a data conflict that the user has to manually resolve, or it may have been cherry-picked cleanly. This field
+  // is not valid unless the rebasing_started field is set to true.
+  last_attempted_step:float;
+
+  // The rebasing_started field indicates if execution of the rebase plan has been started or not. Once execution of the
+  // plan has been started, the last_attempted_step field holds a reference to the most recent plan step attempted.
+  rebasing_started:bool;
 }
 
 // KEEP THIS IN SYNC WITH fileidentifiers.go

go/store/datas/dataset.go

@@ -167,6 +167,8 @@ type RebaseState struct {
 	branch                     string
 	commitBecomesEmptyHandling uint8
 	emptyCommitHandling        uint8
+	lastAttemptedStep          float32
+	rebasingStarted            bool
 }
 
 func (rs *RebaseState) PreRebaseWorkingAddr() hash.Hash {
@@ -188,6 +190,14 @@ func (rs *RebaseState) OntoCommit(ctx context.Context, vr types.ValueReader) (*C
 	return nil, nil
 }
 
+func (rs *RebaseState) LastAttemptedStep(_ context.Context) float32 {
+	return rs.lastAttemptedStep
+}
+
+func (rs *RebaseState) RebasingStarted(_ context.Context) bool {
+	return rs.rebasingStarted
+}
+
 func (rs *RebaseState) CommitBecomesEmptyHandling(_ context.Context) uint8 {
 	return rs.commitBecomesEmptyHandling
 }
@@ -446,6 +456,8 @@ func (h serialWorkingSetHead) HeadWorkingSet() (*WorkingSetHead, error) {
 			string(rebaseState.BranchBytes()),
 			rebaseState.CommitBecomesEmptyHandling(),
 			rebaseState.EmptyCommitHandling(),
+			rebaseState.LastAttemptedStep(),
+			rebaseState.RebasingStarted(),
 		)
 	}
 

go/store/datas/workingset.go

@@ -194,6 +194,8 @@ func workingset_flatbuffer(working hash.Hash, staged *hash.Hash, mergeState *Mer
 		serial.RebaseStateAddOntoCommitAddr(builder, ontoAddrOffset)
 		serial.RebaseStateAddCommitBecomesEmptyHandling(builder, rebaseState.commitBecomesEmptyHandling)
 		serial.RebaseStateAddEmptyCommitHandling(builder, rebaseState.emptyCommitHandling)
+		serial.RebaseStateAddLastAttemptedStep(builder, rebaseState.lastAttemptedStep)
+		serial.RebaseStateAddRebasingStarted(builder, rebaseState.rebasingStarted)
 		rebaseStateOffset = serial.RebaseStateEnd(builder)
 	}
 
@@ -262,13 +264,15 @@ func NewMergeState(
 	}
 }
 
-func NewRebaseState(preRebaseWorkingRoot hash.Hash, commitAddr hash.Hash, branch string, commitBecomesEmptyHandling uint8, emptyCommitHandling uint8) *RebaseState {
+func NewRebaseState(preRebaseWorkingRoot hash.Hash, commitAddr hash.Hash, branch string, commitBecomesEmptyHandling uint8, emptyCommitHandling uint8, lastAttemptedStep float32, rebasingStarted bool) *RebaseState {
 	return &RebaseState{
 		preRebaseWorkingAddr:       &preRebaseWorkingRoot,
 		ontoCommitAddr:             &commitAddr,
 		branch:                     branch,
 		commitBecomesEmptyHandling: commitBecomesEmptyHandling,
 		emptyCommitHandling:        emptyCommitHandling,
+		lastAttemptedStep:          lastAttemptedStep,
+		rebasingStarted:            rebasingStarted,
 	}
 }
 

integration-tests/bats/rebase.bats

@@ -336,7 +336,7 @@ setupCustomEditorScript() {
     ! [[ "$output" =~ "b1 merge commit" ]] || false
 }
 
-@test "rebase: rebase with data conflicts aborts" {
+@test "rebase: rebase with data conflict" {
     setupCustomEditorScript
 
     dolt checkout b1
@@ -345,15 +345,96 @@ setupCustomEditorScript() {
 
     run dolt rebase -i main
     [ "$status" -eq 1 ]
-    [[ "$output" =~ "merge conflict detected while rebasing commit" ]] || false
-    [[ "$output" =~ "the rebase has been automatically aborted" ]] || false
+    [[ "$output" =~ "data conflict detected while rebasing commit" ]] || false
+    [[ "$output" =~ "b1 commit 2" ]] || false
+
+    # Assert that we are on the rebase working branch (not the branch being rebased)
+    run dolt sql -q "select active_branch();"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ " dolt_rebase_b1 " ]] || false
+
+    # Review and resolve the data conflict
+    run dolt conflicts cat .
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "+  | ours   | 1  | 1 | NULL | NULL | NULL | NULL |" ]] || false
+    [[ "$output" =~ "+  | theirs | 1  | 2 | NULL | NULL | NULL | NULL |" ]] || false
+    dolt conflicts resolve --theirs t1
+    dolt add t1
 
     run dolt rebase --continue
-    [ "$status" -eq 1 ]
-    [[ "$output" =~ "no rebase in progress" ]] || false
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "Successfully rebased and updated refs/heads/b1" ]] || false
 
+    # Assert that we are back on the branch being rebased
     run dolt branch
     [ "$status" -eq 0 ]
+    [[ "$output" =~ "* b1" ]] || false
+    ! [[ "$output" =~ "dolt_rebase_b1" ]] || false
+}
+
+@test "rebase: rebase with multiple data conflicts" {
+    setupCustomEditorScript
+    dolt sql -q "INSERT INTO t1 VALUES (2,200);"
+    dolt commit -am "main commit 3"
+
+    dolt checkout b1
+    dolt sql -q "INSERT INTO t1 VALUES (1,2);"
+    dolt commit -am "b1 commit 2"
+    dolt sql -q "INSERT INTO t1 VALUES (2,3);"
+    dolt commit -am "b1 commit 3"
+
+    # Start the rebase
+    run dolt rebase -i main
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "data conflict detected while rebasing commit" ]] || false
+    [[ "$output" =~ "b1 commit 2" ]] || false
+
+    # Assert that we are on the rebase working branch now (not the branch being rebased)
+    run dolt sql -q "select active_branch();"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ " dolt_rebase_b1 " ]] || false
+
+    # Review and resolve the first data conflict
+    run dolt conflicts cat .
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "+  | ours   | 1  | 1 | NULL | NULL | NULL | NULL |" ]] || false
+    [[ "$output" =~ "+  | theirs | 1  | 2 | NULL | NULL | NULL | NULL |" ]] || false
+    dolt conflicts resolve --theirs t1
+
+    # Without staging the changed tables, trying to continue the rebase results in an error
+    run dolt rebase --continue
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "cannot continue a rebase with unstaged changes" ]] || false
+
+    # Stage the tables and then continue the rebase and hit the second data conflict
+    dolt add t1
+    run dolt rebase --continue
+    [ "$status" -eq 1 ]
+    [[ "$output" =~ "data conflict detected while rebasing commit" ]] || false
+    [[ "$output" =~ "b1 commit 3" ]] || false
+
+    # Assert that we are still on the rebase working branch
+    run dolt sql -q "select active_branch();"
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ " dolt_rebase_b1 " ]] || false
+
+    # Review and resolve the second data conflict
+    run dolt conflicts cat .
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "+  | ours   | 2  | 200 | NULL | NULL | NULL | NULL |" ]] || false
+    [[ "$output" =~ "+  | theirs | 2  | 3   | NULL | NULL | NULL | NULL |" ]] || false
+    dolt conflicts resolve --ours t1
+
+    # Finish the rebase
+    dolt add t1
+    run dolt rebase --continue
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "Successfully rebased and updated refs/heads/b1" ]] || false
+
+    # Assert that we are back on the branch that was rebased, and that the rebase working branch is gone
+    run dolt branch
+    [ "$status" -eq 0 ]
+    [[ "$output" =~ "b1" ]] || false
     ! [[ "$output" =~ "dolt_rebase_b1" ]] || false
 }
 
@@ -366,7 +447,7 @@ setupCustomEditorScript() {
 
     run dolt rebase -i main
     [ "$status" -eq 1 ]
-    [[ "$output" =~ "merge conflict detected while rebasing commit" ]] || false
+    [[ "$output" =~ "schema conflict detected while rebasing commit" ]] || false
     [[ "$output" =~ "the rebase has been automatically aborted" ]] || false
 
     run dolt rebase --continue