Merge pull request #177 from jfontan/fix/cancel-queries

Use session and support query cancelation

Author: jfontan

Gopkg.lock

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

Gopkg.toml

@@ -25,9 +25,7 @@ var (
 	blobsMaxSize     = getIntEnv(blobsMaxSizeKey, 5) * mib
 )
 
-type blobsTable struct {
-	pool *RepositoryPool
-}
+type blobsTable struct{}
 
 var blobsSchema = sql.Schema{
 	{Name: "hash", Type: sql.Text, Nullable: false, Source: blobsTableName},
@@ -37,8 +35,8 @@ var blobsSchema = sql.Schema{
 
 var _ sql.PushdownProjectionAndFiltersTable = (*blobsTable)(nil)
 
-func newBlobsTable(pool *RepositoryPool) sql.Table {
-	return &blobsTable{pool: pool}
+func newBlobsTable() sql.Table {
+	return new(blobsTable)
 }
 
 func (blobsTable) String() string {
@@ -65,10 +63,10 @@ func (r *blobsTable) TransformExpressionsUp(f sql.TransformExprFunc) (sql.Node,
 	return r, nil
 }
 
-func (r blobsTable) RowIter(_ sql.Session) (sql.RowIter, error) {
+func (r blobsTable) RowIter(ctx *sql.Context) (sql.RowIter, error) {
 	iter := new(blobIter)
 
-	repoIter, err := NewRowRepoIter(r.pool, iter)
+	repoIter, err := NewRowRepoIter(ctx, iter)
 	if err != nil {
 		return nil, err
 	}
@@ -85,11 +83,11 @@ func (blobsTable) HandledFilters(filters []sql.Expression) []sql.Expression {
 }
 
 func (r *blobsTable) WithProjectAndFilters(
-	session sql.Session,
+	ctx *sql.Context,
 	_, filters []sql.Expression,
 ) (sql.RowIter, error) {
 	return rowIterWithSelectors(
-		session, r.pool, blobsSchema, blobsTableName, filters,
+		ctx, blobsSchema, blobsTableName, filters,
 		[]string{"hash"},
 		func(selectors selectors) (RowRepoIter, error) {
 			if len(selectors["hash"]) == 0 {

blobs.go

@@ -1,21 +1,17 @@
 package gitquery
 
 import (
-	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"gopkg.in/src-d/go-mysql-server.v0/sql"
 	"gopkg.in/src-d/go-mysql-server.v0/sql/expression"
-
-	"gopkg.in/src-d/go-git-fixtures.v3"
 )
 
 func TestBlobsTable_Name(t *testing.T) {
 	require := require.New(t)
 
-	f := fixtures.ByTag("worktree").One()
-	table := getTable(require, f, blobsTableName)
+	table := getTable(require, blobsTableName)
 	require.Equal(blobsTableName, table.Name())
 
 	// Check that each column source is the same as table name
@@ -27,18 +23,18 @@ func TestBlobsTable_Name(t *testing.T) {
 func TestBlobsTable_Children(t *testing.T) {
 	require := require.New(t)
 
-	f := fixtures.ByTag("worktree").One()
-	table := getTable(require, f, blobsTableName)
+	table := getTable(require, blobsTableName)
 	require.Equal(0, len(table.Children()))
 }
 
 func TestBlobsTable_RowIter(t *testing.T) {
 	require := require.New(t)
+	ctx, _, cleanup := setup(t)
+	defer cleanup()
 
-	f := fixtures.ByTag("worktree").One()
-	table := getTable(require, f, blobsTableName)
+	table := getTable(require, blobsTableName)
 
-	rows, err := sql.NodeToRows(sql.NewBaseSession(context.TODO()), table)
+	rows, err := sql.NodeToRows(ctx, table)
 	require.NoError(err)
 	require.Len(rows, 10)
 
@@ -60,7 +56,7 @@ func TestBlobsLimit(t *testing.T) {
 		blobsMaxSize = prev
 	}()
 
-	table := newBlobsTable(session.Pool)
+	table := newBlobsTable()
 	iter, err := table.RowIter(session)
 	require.NoError(err)
 
@@ -98,7 +94,7 @@ func TestBlobsPushdown(t *testing.T) {
 	session, _, cleanup := setup(t)
 	defer cleanup()
 
-	table := newBlobsTable(session.Pool).(sql.PushdownProjectionAndFiltersTable)
+	table := newBlobsTable().(sql.PushdownProjectionAndFiltersTable)
 
 	iter, err := table.WithProjectAndFilters(session, nil, nil)
 	require.NoError(err)

blobs_test.go

@@ -42,7 +42,7 @@ func (c *CmdServer) buildDatabase() error {
 		return err
 	}
 
-	c.engine.AddDatabase(gitquery.NewDatabase(c.name, c.pool))
+	c.engine.AddDatabase(gitquery.NewDatabase(c.name))
 	c.engine.Catalog.RegisterFunctions(function.Functions)
 	return nil
 }

cmd/gitquery/server.go

@@ -10,7 +10,6 @@ import (
 )
 
 type commitsTable struct {
-	pool *RepositoryPool
 }
 
 var commitsSchema = sql.Schema{
@@ -27,8 +26,8 @@ var commitsSchema = sql.Schema{
 
 var _ sql.PushdownProjectionAndFiltersTable = (*commitsTable)(nil)
 
-func newCommitsTable(pool *RepositoryPool) sql.Table {
-	return &commitsTable{pool: pool}
+func newCommitsTable() sql.Table {
+	return new(commitsTable)
 }
 
 func (commitsTable) String() string {
@@ -55,10 +54,10 @@ func (r *commitsTable) TransformExpressionsUp(f sql.TransformExprFunc) (sql.Node
 	return r, nil
 }
 
-func (r commitsTable) RowIter(_ sql.Session) (sql.RowIter, error) {
+func (r commitsTable) RowIter(ctx *sql.Context) (sql.RowIter, error) {
 	iter := new(commitIter)
 
-	repoIter, err := NewRowRepoIter(r.pool, iter)
+	repoIter, err := NewRowRepoIter(ctx, iter)
 	if err != nil {
 		return nil, err
 	}
@@ -75,11 +74,11 @@ func (commitsTable) HandledFilters(filters []sql.Expression) []sql.Expression {
 }
 
 func (r *commitsTable) WithProjectAndFilters(
-	session sql.Session,
+	ctx *sql.Context,
 	_, filters []sql.Expression,
 ) (sql.RowIter, error) {
 	return rowIterWithSelectors(
-		session, r.pool, commitsSchema, commitsTableName, filters,
+		ctx, commitsSchema, commitsTableName, filters,
 		[]string{"hash"},
 		func(selectors selectors) (RowRepoIter, error) {
 			if len(selectors["hash"]) == 0 {

commits.go

@@ -1,21 +1,17 @@
 package gitquery
 
 import (
-	"context"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 	"gopkg.in/src-d/go-mysql-server.v0/sql"
 	"gopkg.in/src-d/go-mysql-server.v0/sql/expression"
-
-	"gopkg.in/src-d/go-git-fixtures.v3"
 )
 
 func TestCommitsTable_Name(t *testing.T) {
 	require := require.New(t)
 
-	f := fixtures.ByTag("worktree").One()
-	table := getTable(require, f, commitsTableName)
+	table := getTable(require, commitsTableName)
 	require.Equal(commitsTableName, table.Name())
 
 	// Check that each column source is the same as table name
@@ -27,18 +23,18 @@ func TestCommitsTable_Name(t *testing.T) {
 func TestCommitsTable_Children(t *testing.T) {
 	require := require.New(t)
 
-	f := fixtures.ByTag("worktree").One()
-	table := getTable(require, f, commitsTableName)
+	table := getTable(require, commitsTableName)
 	require.Equal(0, len(table.Children()))
 }
 
 func TestCommitsTable_RowIter(t *testing.T) {
 	require := require.New(t)
+	session, _, cleanup := setup(t)
+	defer cleanup()
 
-	f := fixtures.ByTag("worktree").One()
-	table := getTable(require, f, commitsTableName)
+	table := getTable(require, commitsTableName)
 
-	rows, err := sql.NodeToRows(sql.NewBaseSession(context.TODO()), table)
+	rows, err := sql.NodeToRows(session, table)
 	require.Nil(err)
 	require.Len(rows, 9)
 
@@ -54,7 +50,7 @@ func TestCommitsPushdown(t *testing.T) {
 	session, _, cleanup := setup(t)
 	defer cleanup()
 
-	table := newCommitsTable(session.Pool).(sql.PushdownProjectionAndFiltersTable)
+	table := newCommitsTable().(sql.PushdownProjectionAndFiltersTable)
 
 	iter, err := table.WithProjectAndFilters(session, nil, nil)
 	require.NoError(err)

commits_test.go

@@ -7,11 +7,12 @@ import (
 	"github.com/stretchr/testify/require"
 	fixtures "gopkg.in/src-d/go-git-fixtures.v3"
 	sqle "gopkg.in/src-d/go-mysql-server.v0"
+	"gopkg.in/src-d/go-mysql-server.v0/sql"
 )
 
 type CleanupFunc func()
 
-func setup(t *testing.T) (sess *Session, path string, cleanup CleanupFunc) {
+func setup(t *testing.T) (ctx *sql.Context, path string, cleanup CleanupFunc) {
 	require := require.New(t)
 	t.Helper()
 
@@ -22,12 +23,15 @@ func setup(t *testing.T) (sess *Session, path string, cleanup CleanupFunc) {
 	pool.AddGit(path)
 
 	engine := sqle.New()
-	engine.AddDatabase(NewDatabase("db", &pool))
+	engine.AddDatabase(NewDatabase("db"))
 
 	cleanup = func() {
 		t.Helper()
 		require.NoError(fixtures.Clean())
 	}
 
-	return NewSession(context.TODO(), &pool), path, cleanup
+	session := NewSession(&pool)
+	ctx = sql.NewContext(context.TODO(), session)
+
+	return ctx, path, cleanup
 }

common_test.go

@@ -27,15 +27,15 @@ type Database struct {
 
 // NewDatabase creates a new Database structure and initializes its
 // tables with the given pool
-func NewDatabase(name string, pool *RepositoryPool) sql.Database {
+func NewDatabase(name string) sql.Database {
 	return &Database{
 		name:         name,
-		commits:      newCommitsTable(pool),
-		references:   newReferencesTable(pool),
-		blobs:        newBlobsTable(pool),
-		treeEntries:  newTreeEntriesTable(pool),
-		repositories: newRepositoriesTable(pool),
-		remotes:      newRemotesTable(pool),
+		commits:      newCommitsTable(),
+		references:   newReferencesTable(),
+		blobs:        newBlobsTable(),
+		treeEntries:  newTreeEntriesTable(),
+		repositories: newRepositoriesTable(),
+		remotes:      newRemotesTable(),
 	}
 }
 

database.go

@@ -21,8 +21,7 @@ const (
 func TestDatabase_Tables(t *testing.T) {
 	require := require.New(t)
 
-	f := fixtures.ByTag("worktree").One()
-	db := getDB(require, f, testDBName)
+	db := getDB(require, testDBName)
 
 	tables := db.Tables()
 	var tableNames []string
@@ -47,32 +46,19 @@ func TestDatabase_Tables(t *testing.T) {
 func TestDatabase_Name(t *testing.T) {
 	require := require.New(t)
 
-	f := fixtures.ByTag("worktree").One()
-	db := getDB(require, f, testDBName)
+	db := getDB(require, testDBName)
 	require.Equal(testDBName, db.Name())
 }
 
-func getDB(
-	require *require.Assertions,
-	fixture *fixtures.Fixture,
-	name string,
-) sql.Database {
-
-	fixtures.Init()
-
-	pool := NewRepositoryPool()
-	pool.Add("repo", fixture.Worktree().Root())
-
-	db := NewDatabase(name, &pool)
+func getDB(require *require.Assertions, name string) sql.Database {
+	db := NewDatabase(name)
 	require.NotNil(db)
 
 	return db
 }
 
-func getTable(require *require.Assertions, fixture *fixtures.Fixture,
-	name string) sql.Table {
-
-	db := getDB(require, fixture, "foo")
+func getTable(require *require.Assertions, name string) sql.Table {
+	db := getDB(require, "foo")
 	require.NotNil(db)
 	require.Equal(db.Name(), "foo")