Merge pull request #670 from erizocosmico/feature/index-checksum

gitbase: implement table checksums

Author: ajnavarro

Gopkg.lock

@@ -1,6 +1,6 @@
 [[constraint]]
   name = "gopkg.in/src-d/go-mysql-server.v0"
-  revision = "ac598027ca4498f318051bcb79ca5b4231faf733"
+  revision = "7afa88202b2490bb93c08603da422a690099695e"
 
 [[constraint]]
   name = "github.com/jessevdk/go-flags"

Gopkg.toml

@@ -26,6 +26,7 @@ var (
 )
 
 type blobsTable struct {
+	checksumable
 	partitioned
 	filters    []sql.Expression
 	projection []string
@@ -40,8 +41,8 @@ var BlobsSchema = sql.Schema{
 	{Name: "blob_content", Type: sql.Blob, Nullable: false, Source: BlobsTableName},
 }
 
-func newBlobsTable() *blobsTable {
-	return new(blobsTable)
+func newBlobsTable(pool *RepositoryPool) *blobsTable {
+	return &blobsTable{checksumable: checksumable{pool}}
 }
 
 var _ Table = (*blobsTable)(nil)
@@ -155,13 +156,13 @@ func (*blobsTable) handledColumns() []string {
 }
 
 // IndexKeyValues implements the sql.IndexableTable interface.
-func (*blobsTable) IndexKeyValues(
+func (r *blobsTable) IndexKeyValues(
 	ctx *sql.Context,
 	colNames []string,
 ) (sql.PartitionIndexKeyValueIter, error) {
 	return newPartitionedIndexKeyValueIter(
 		ctx,
-		newBlobsTable(),
+		newBlobsTable(r.pool),
 		colNames,
 		newBlobsKeyValueIter,
 	)

blobs.go

@@ -13,7 +13,7 @@ func TestBlobsTable(t *testing.T) {
 	ctx, _, cleanup := setup(t)
 	defer cleanup()
 
-	table := getTable(require, BlobsTableName)
+	table := getTable(t, BlobsTableName, ctx)
 
 	rows, err := tableToRows(ctx, table)
 	require.NoError(err)
@@ -28,7 +28,7 @@ func TestBlobsTable(t *testing.T) {
 
 func TestBlobsLimit(t *testing.T) {
 	require := require.New(t)
-	session, _, cleanup := setup(t)
+	ctx, _, cleanup := setup(t)
 	defer cleanup()
 
 	prev := blobsMaxSize
@@ -37,8 +37,9 @@ func TestBlobsLimit(t *testing.T) {
 		blobsMaxSize = prev
 	}()
 
-	table := newBlobsTable().WithProjection([]string{"blob_content"})
-	rows, err := tableToRows(session, table)
+	table := newBlobsTable(poolFromCtx(t, ctx)).
+		WithProjection([]string{"blob_content"})
+	rows, err := tableToRows(ctx, table)
 	require.NoError(err)
 
 	expected := []struct {
@@ -72,7 +73,7 @@ func TestBlobsPushdown(t *testing.T) {
 	ctx, _, cleanup := setup(t)
 	defer cleanup()
 
-	table := newBlobsTable()
+	table := newBlobsTable(poolFromCtx(t, ctx))
 
 	rows, err := tableToRows(ctx, table)
 	require.NoError(err)

blobs_test.go

@@ -0,0 +1,116 @@
+package gitbase
+
+import (
+	"bytes"
+	"crypto/sha1"
+	"encoding/base64"
+	"io"
+
+	git "gopkg.in/src-d/go-git.v4"
+	"gopkg.in/src-d/go-git.v4/plumbing"
+)
+
+type checksumable struct {
+	pool *RepositoryPool
+}
+
+func (c *checksumable) Checksum() (string, error) {
+	hash := sha1.New()
+	for _, id := range c.pool.idOrder {
+		repo := c.pool.repositories[id]
+		hash.Write([]byte(id))
+
+		bytes, err := readChecksum(repo)
+		if err != nil {
+			return "", err
+		}
+
+		if _, err = hash.Write(bytes); err != nil {
+			return "", err
+		}
+
+		bytes, err = readRefs(repo)
+		if err != nil {
+			return "", err
+		}
+
+		if _, err = hash.Write(bytes); err != nil {
+			return "", err
+		}
+	}
+
+	return base64.StdEncoding.EncodeToString(hash.Sum(nil)), nil
+}
+
+func readChecksum(r repository) ([]byte, error) {
+	fs, err := r.FS()
+	if err != nil {
+		return nil, err
+	}
+
+	dot, packfiles, err := repositoryPackfiles(fs)
+	if err != nil {
+		return nil, err
+	}
+
+	var result []byte
+	for _, p := range packfiles {
+		f, err := dot.ObjectPack(p)
+		if err != nil {
+			return nil, err
+		}
+
+		if _, err = f.Seek(-20, io.SeekEnd); err != nil {
+			return nil, err
+		}
+
+		var checksum = make([]byte, 20)
+		if _, err = io.ReadFull(f, checksum); err != nil {
+			return nil, err
+		}
+
+		if err = f.Close(); err != nil {
+			return nil, err
+		}
+
+		result = append(result, checksum...)
+	}
+
+	return result, nil
+}
+
+func readRefs(r repository) ([]byte, error) {
+	repo, err := r.Repo()
+	if err != nil {
+		if err == git.ErrRepositoryNotExists {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	buf := bytes.NewBuffer(nil)
+
+	head, err := repo.Head()
+	if err != nil && err != plumbing.ErrReferenceNotFound {
+		return nil, err
+	} else {
+		buf.WriteString("HEAD")
+		buf.WriteString(head.Hash().String())
+	}
+
+	refs, err := repo.References()
+	if err != nil {
+		return nil, err
+	}
+
+	err = refs.ForEach(func(r *plumbing.Reference) error {
+		buf.WriteString(string(r.Name()))
+		buf.WriteString(r.Hash().String())
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return buf.Bytes(), nil
+}

checksum.go

@@ -0,0 +1,40 @@
+package gitbase
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	fixtures "gopkg.in/src-d/go-git-fixtures.v3"
+	"gopkg.in/src-d/go-git.v4/plumbing/cache"
+)
+
+func TestChecksum(t *testing.T) {
+	require := require.New(t)
+
+	require.NoError(fixtures.Init())
+	defer func() {
+		require.NoError(fixtures.Clean())
+	}()
+
+	pool := NewRepositoryPool(cache.DefaultMaxSize)
+
+	for i, f := range fixtures.ByTag("worktree") {
+		path := f.Worktree().Root()
+		require.NoError(pool.AddGitWithID(fmt.Sprintf("repo_%d", i), path))
+	}
+
+	c := &checksumable{pool}
+	checksum, err := c.Checksum()
+	require.NoError(err)
+	require.Equal("ogfv7HAwFigDgtuW4tbnEP+Zl40=", checksum)
+
+	pool = NewRepositoryPool(cache.DefaultMaxSize)
+	path := fixtures.ByTag("worktree").One().Worktree().Root()
+	require.NoError(pool.AddGitWithID("worktree", path))
+
+	c = &checksumable{pool}
+	checksum, err = c.Checksum()
+	require.NoError(err)
+	require.Equal("5kfLCygyBSZFMh+nFzFNk3zAUTQ=", checksum)
+}

checksum_test.go

@@ -204,7 +204,7 @@ func (c *Server) buildDatabase() error {
 		return err
 	}
 
-	c.engine.AddDatabase(gitbase.NewDatabase(c.Name))
+	c.engine.AddDatabase(gitbase.NewDatabase(c.Name, c.pool))
 	c.engine.AddDatabase(sql.NewInformationSchemaDatabase(c.engine.Catalog))
 	c.engine.Catalog.SetCurrentDatabase(c.Name)
 	logrus.WithField("db", c.Name).Debug("registered database to catalog")

cmd/gitbase/command/server.go

@@ -10,6 +10,7 @@ import (
 )
 
 type commitBlobsTable struct {
+	checksumable
 	partitioned
 	filters []sql.Expression
 	index   sql.IndexLookup
@@ -24,8 +25,8 @@ var CommitBlobsSchema = sql.Schema{
 
 var _ Table = (*commitBlobsTable)(nil)
 
-func newCommitBlobsTable() Indexable {
-	return new(commitBlobsTable)
+func newCommitBlobsTable(pool *RepositoryPool) Indexable {
+	return &commitBlobsTable{checksumable: checksumable{pool}}
 }
 
 var _ Squashable = (*blobsTable)(nil)
@@ -120,13 +121,13 @@ func (commitBlobsTable) HandledFilters(filters []sql.Expression) []sql.Expressio
 }
 
 // IndexKeyValues implements the sql.IndexableTable interface.
-func (*commitBlobsTable) IndexKeyValues(
+func (t *commitBlobsTable) IndexKeyValues(
 	ctx *sql.Context,
 	colNames []string,
 ) (sql.PartitionIndexKeyValueIter, error) {
 	return newTablePartitionIndexKeyValueIter(
 		ctx,
-		newCommitBlobsTable(),
+		newCommitBlobsTable(t.pool),
 		CommitBlobsTableName,
 		colNames,
 		new(commitBlobsRowKeyMapper),

commit_blobs.go

@@ -14,12 +14,12 @@ import (
 func TestCommitBlobsTableRowIter(t *testing.T) {
 	require := require.New(t)
 
-	table := newCommitBlobsTable()
-	require.NotNil(table)
-
 	ctx, paths, cleanup := setupRepos(t)
 	defer cleanup()
 
+	table := newCommitBlobsTable(poolFromCtx(t, ctx))
+	require.NotNil(table)
+
 	expectedRows := []sql.Row{
 		sql.NewRow(paths[0], "e8d3ffab552895c19b9fcf7aa264d277cde33881", "32858aad3c383ed1ff0a0f9bdf231d54a00c9e88"),
 		sql.NewRow(paths[0], "e8d3ffab552895c19b9fcf7aa264d277cde33881", "d3ff53e0564a9f87d8e84b6e28e5060e517008aa"),

commit_blobs_test.go

@@ -14,6 +14,7 @@ import (
 )
 
 type commitFilesTable struct {
+	checksumable
 	partitioned
 	filters []sql.Expression
 	index   sql.IndexLookup
@@ -28,8 +29,8 @@ var CommitFilesSchema = sql.Schema{
 	{Name: "tree_hash", Type: sql.Text, Source: CommitFilesTableName},
 }
 
-func newCommitFilesTable() Indexable {
-	return new(commitFilesTable)
+func newCommitFilesTable(pool *RepositoryPool) Indexable {
+	return &commitFilesTable{checksumable: checksumable{pool}}
 }
 
 var _ Table = (*commitFilesTable)(nil)
@@ -127,13 +128,13 @@ func (commitFilesTable) handledColumns() []string {
 }
 
 // IndexKeyValues implements the sql.IndexableTable interface.
-func (*commitFilesTable) IndexKeyValues(
+func (t *commitFilesTable) IndexKeyValues(
 	ctx *sql.Context,
 	colNames []string,
 ) (sql.PartitionIndexKeyValueIter, error) {
 	return newPartitionedIndexKeyValueIter(
 		ctx,
-		newCommitFilesTable(),
+		newCommitFilesTable(t.pool),
 		colNames,
 		newCommitFilesKeyValueIter,
 	)