Cataloger create branch by ref (#1020)

api/api_controller.go

@@ -682,8 +682,8 @@ func (c *Controller) CreateBranchHandler() branches.CreateBranchHandler {
 		}
 		deps.LogAction("create_branch")
 		cataloger := deps.Cataloger
-		sourceBranch := swag.StringValue(params.Branch.Source)
-		commitLog, err := cataloger.CreateBranch(c.Context(), repository, branch, sourceBranch)
+		sourceRef := swag.StringValue(params.Branch.Source)
+		commitLog, err := cataloger.CreateBranch(c.Context(), repository, branch, sourceRef)
 		if err != nil {
 			return branches.NewCreateBranchDefault(http.StatusInternalServerError).WithPayload(responseErrorFrom(err))
 		}

catalog/cataloger.go

@@ -77,7 +77,7 @@ type Cataloger interface {
 	// In this case pass the last repository name as 'after' on the next call to ListRepositories
 	ListRepositories(ctx context.Context, limit int, after string) ([]*Repository, bool, error)
 
-	CreateBranch(ctx context.Context, repository, branch string, sourceBranch string) (*CommitLog, error)
+	CreateBranch(ctx context.Context, repository, branch string, sourceRef string) (*CommitLog, error)
 	DeleteBranch(ctx context.Context, repository, branch string) error
 	ListBranches(ctx context.Context, repository string, prefix string, limit int, after string) ([]*Branch, bool, error)
 	BranchExists(ctx context.Context, repository string, branch string) (bool, error)

catalog/mvcc/cataloger_commit_test.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/go-test/deep"
-
 	"github.com/treeverse/lakefs/catalog"
 	"github.com/treeverse/lakefs/testutil"
 )

catalog/mvcc/cataloger_create_branch.go

@@ -10,18 +10,23 @@ import (
 )
 
 const (
-	createBranchCommitMessageFormat = "Branch '%s' created, source branch '%s'"
+	createBranchCommitMessageFormat = "Branch '%s' created, source '%s'"
 )
 
-func (c *cataloger) CreateBranch(ctx context.Context, repository, branch string, sourceBranch string) (*catalog.CommitLog, error) {
+func (c *cataloger) CreateBranch(ctx context.Context, repository, branch string, sourceRef string) (*catalog.CommitLog, error) {
 	if err := Validate(ValidateFields{
 		{Name: "repository", IsValid: ValidateRepositoryName(repository)},
 		{Name: "branch", IsValid: ValidateBranchName(branch)},
-		{Name: "sourceBranch", IsValid: ValidateBranchName(sourceBranch)},
+		{Name: "sourceRef", IsValid: ValidateReference(sourceRef)},
 	}); err != nil {
 		return nil, err
 	}
 
+	ref, err := ParseRef(sourceRef)
+	if err != nil {
+		return nil, fmt.Errorf("source ref: %w", err)
+	}
+
 	res, err := c.db.Transact(func(tx db.Tx) (interface{}, error) {
 		_, err := tx.Exec("LOCK TABLE catalog_branches IN SHARE UPDATE EXCLUSIVE MODE")
 		if err != nil {
@@ -33,13 +38,28 @@ func (c *cataloger) CreateBranch(ctx context.Context, repository, branch string,
 			return nil, err
 		}
 
-		// get source branch id and
-		var sourceBranchID int
+		// get source branch id
+		var sourceBranchID int64
 		if err := tx.GetPrimitive(&sourceBranchID, `SELECT id FROM catalog_branches WHERE repository_id=$1 AND name=$2`,
-			repoID, sourceBranch); err != nil {
+			repoID, ref.Branch); err != nil {
 			return nil, fmt.Errorf("source branch id: %w", err)
 		}
 
+		// get source branch commit id or latest commit id, in case we reference a branch
+		var sourceCommitID CommitID
+		if ref.CommitID > UncommittedID {
+			// verify that the requested commit id exists in our branch commit
+			err = tx.GetPrimitive(&sourceCommitID, `SELECT commit_id FROM catalog_commits WHERE branch_id=$1 AND commit_id=$2`,
+				sourceBranchID, ref.CommitID)
+		} else {
+			// ref is just a branch - get last commit id
+			err = tx.GetPrimitive(&sourceCommitID, `SELECT MAX(commit_id) FROM catalog_commits WHERE branch_id=$1`,
+				sourceBranchID)
+		}
+		if err != nil {
+			return nil, fmt.Errorf("get source branch commit id: %w", err)
+		}
+
 		// next id for branch
 		var branchID int64
 		if err := tx.GetPrimitive(&branchID, `SELECT nextval('catalog_branches_id_seq')`); err != nil {
@@ -58,21 +78,20 @@ func (c *cataloger) CreateBranch(ctx context.Context, repository, branch string,
 			MergeSourceCommit    CommitID  `db:"merge_source_commit"`
 			TransactionTimestamp time.Time `db:"transaction_timestamp"`
 		}{}
-		commitMsg := fmt.Sprintf(createBranchCommitMessageFormat, branch, sourceBranch)
+		commitMsg := FormatBranchCommitMessage(branch, sourceRef)
 		err = tx.Get(&insertReturns, `INSERT INTO catalog_commits (branch_id,commit_id,previous_commit_id,committer,message,
 			creation_date,merge_source_branch,merge_type,lineage_commits,merge_source_commit)
 			VALUES ($1,nextval('catalog_commit_id_seq'),0,$2,$3,transaction_timestamp(),$4,'from_parent',
-				(select (select max(commit_id) from catalog_commits where branch_id=$4) ||
-					(select distinct on (branch_id) lineage_commits from catalog_commits
-						where branch_id=$4 and merge_type='from_parent' order by branch_id,commit_id desc))
-						,(select max(commit_id) from catalog_commits where branch_id=$4 ))
+				($5::bigint || (select distinct on (branch_id) lineage_commits from catalog_commits
+					where branch_id=$4 and merge_type='from_parent' and commit_id <= $5 order by branch_id,commit_id desc)),
+					$5)
 			RETURNING commit_id,merge_source_commit,transaction_timestamp()`,
-			branchID, catalog.DefaultCommitter, commitMsg, sourceBranchID)
+			branchID, catalog.DefaultCommitter, commitMsg, sourceBranchID, sourceCommitID)
 		if err != nil {
 			return nil, fmt.Errorf("insert commit: %w", err)
 		}
 		reference := MakeReference(branch, insertReturns.CommitID)
-		parentReference := MakeReference(sourceBranch, insertReturns.MergeSourceCommit)
+		parentReference := MakeReference(ref.Branch, insertReturns.MergeSourceCommit)
 
 		commitLog := &catalog.CommitLog{
 			Committer:    catalog.DefaultCommitter,
@@ -89,3 +108,7 @@ func (c *cataloger) CreateBranch(ctx context.Context, repository, branch string,
 	commitLog := res.(*catalog.CommitLog)
 	return commitLog, nil
 }
+
+func FormatBranchCommitMessage(name, source string) string {
+	return fmt.Sprintf(createBranchCommitMessageFormat, name, source)
+}

catalog/mvcc/cataloger_create_branch_test.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/go-test/deep"
 	"github.com/treeverse/lakefs/testutil"
 )
 
@@ -16,9 +17,9 @@ func TestCataloger_CreateBranch(t *testing.T) {
 	repo := testCatalogerRepo(t, ctx, c, "repo", "master")
 
 	type args struct {
-		repository   string
-		branch       string
-		sourceBranch string
+		repository string
+		branch     string
+		sourceRef  string
 	}
 	tests := []struct {
 		name              string
@@ -29,33 +30,33 @@ func TestCataloger_CreateBranch(t *testing.T) {
 	}{
 		{
 			name:              "new",
-			args:              args{repository: repo, branch: "b1", sourceBranch: "master"},
-			wantCommitMessage: "Branch 'b1' created, source branch 'master'",
+			args:              args{repository: repo, branch: "b1", sourceRef: "master"},
+			wantCommitMessage: FormatBranchCommitMessage("b1", "master"),
 			wantBranchName:    "b1",
 			wantErr:           false,
 		},
 		{
 			name:           "self",
-			args:           args{repository: repo, branch: "master", sourceBranch: "master"},
+			args:           args{repository: repo, branch: "master", sourceRef: "master"},
 			wantBranchName: "",
 			wantErr:        true,
 		},
 		{
 			name:           "unknown source",
-			args:           args{repository: repo, branch: "b2", sourceBranch: "unknown"},
+			args:           args{repository: repo, branch: "b2", sourceRef: "unknown"},
 			wantBranchName: "",
 			wantErr:        true,
 		},
 		{
 			name:           "unknown repository",
-			args:           args{repository: "repo1", branch: "b3", sourceBranch: "unknown"},
+			args:           args{repository: "repo1", branch: "b3", sourceRef: "unknown"},
 			wantBranchName: "",
 			wantErr:        true,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			commitLog, err := c.CreateBranch(ctx, tt.args.repository, tt.args.branch, tt.args.sourceBranch)
+			commitLog, err := c.CreateBranch(ctx, tt.args.repository, tt.args.branch, tt.args.sourceRef)
 			if (err != nil) != tt.wantErr {
 				t.Fatalf("CreateBranch() error = %s, wantErr %t", err, tt.wantErr)
 			}
@@ -127,3 +128,49 @@ func TestCataloger_CreateBranch_Parallel(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestCataloger_CreateBranch_FromRef(t *testing.T) {
+	ctx := context.Background()
+	c := testCataloger(t)
+	repo := testCatalogerRepo(t, ctx, c, "repo", "master")
+
+	// create entry and commit - first
+	testCatalogerCreateEntry(t, ctx, c, repo, "master", "first", nil, "")
+	commit1, err := c.Commit(ctx, repo, "master", "first", "tester", nil)
+	testutil.MustDo(t, "first commit", err)
+
+	// create entry and commit - second
+	testCatalogerCreateEntry(t, ctx, c, repo, "master", "second", nil, "")
+	commit2, err := c.Commit(ctx, repo, "master", "second", "tester", nil)
+	testutil.MustDo(t, "second commit", err)
+
+	// branch from first commit
+	commit1Log, err := c.CreateBranch(ctx, repo, "branch1", commit1.Reference)
+	testutil.MustDo(t, "branch from first commit", err)
+	if commit1Log == nil {
+		t.Fatal("CreateBranch() no error, missing commit log")
+	}
+
+	// check that only 'first' is on our branch
+	branchEntries1, _, err := c.ListEntries(ctx, repo, "branch1", "", "", "", 100)
+	testutil.MustDo(t, "list entries on branch1", err)
+	paths1 := testExtractEntriesPath(branchEntries1)
+	if diff := deep.Equal(paths1, []string{"first"}); diff != nil {
+		t.Fatal("Found diff in expected content of branch1:", diff)
+	}
+
+	// branch from second commit
+	commit2Log, err := c.CreateBranch(ctx, repo, "branch2", commit2.Reference)
+	testutil.MustDo(t, "branch from second commit", err)
+	if commit2Log == nil {
+		t.Fatal("CreateBranch() no error, missing commit log")
+	}
+
+	// check that 'first' and 'second' on our branch
+	branchEntries2, _, err := c.ListEntries(ctx, repo, "branch2", "", "", "", 100)
+	testutil.MustDo(t, "list entries on branch2", err)
+	paths2 := testExtractEntriesPath(branchEntries2)
+	if diff := deep.Equal(paths2, []string{"first", "second"}); diff != nil {
+		t.Fatal("Found diff in expected content of branch2:", diff)
+	}
+}

catalog/mvcc/cataloger_list_commits_test.go

@@ -463,7 +463,7 @@ func TestCataloger_ListCommits_Order(t *testing.T) {
 				commits[i] = commitsLog[i].Message
 			}
 			expectedCommits := []string{
-				"Branch 'branch1' created, source branch 'master'",
+				FormatBranchCommitMessage("branch1", "master"),
 				commit1Msg,
 				createRepositoryCommitMessage,
 				createRepositoryImportCommitMessage,

catalog/mvcc/cataloger_test.go

@@ -111,3 +111,11 @@ func testVerifyEntries(t testing.TB, ctx context.Context, c catalog.Cataloger, r
 		}
 	}
 }
+
+func testExtractEntriesPath(entries []*catalog.Entry) []string {
+	paths := make([]string, len(entries))
+	for i := range entries {
+		paths[i] = entries[i].Path
+	}
+	return paths
+}