Merge pull request #201 from erizocosmico/feature/join-pushdown

 *: implement rule to squash chainable tables into a single one

Author: erizocosmico

Gopkg.lock

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

Gopkg.toml

@@ -63,6 +63,10 @@ To make some common tasks easier for the user, there are some functions to inter
 |is_remote(reference_name)bool| check if the given reference name is from a remote one                               |
 |is_tag(reference_name)bool| check if the given reference name is a tag                                              |
 
+## Unstable features
+
+- **Table squashing:** there is an optimization that collects inner joins between tables with a set of supported conditions and converts them into a single node that retrieves the data in chained steps (getting first the commits and then the blobs of every commit instead of joinin all commits and all blobs, for example). It can be enabled with the environment variable `UNSTABLE_SQUASH_ENABLE`.
+
 ## Examples
 
 ### Get all the HEAD references from all the repositories

README.md

@@ -27,10 +27,11 @@ var (
 
 type blobsTable struct{}
 
-var blobsSchema = sql.Schema{
-	{Name: "hash", Type: sql.Text, Nullable: false, Source: blobsTableName},
-	{Name: "size", Type: sql.Int64, Nullable: false, Source: blobsTableName},
-	{Name: "content", Type: sql.Blob, Nullable: false, Source: blobsTableName},
+// BlobsSchema is the schema for the blobs table.
+var BlobsSchema = sql.Schema{
+	{Name: "hash", Type: sql.Text, Nullable: false, Source: BlobsTableName},
+	{Name: "size", Type: sql.Int64, Nullable: false, Source: BlobsTableName},
+	{Name: "content", Type: sql.Blob, Nullable: false, Source: BlobsTableName},
 }
 
 var _ sql.PushdownProjectionAndFiltersTable = (*blobsTable)(nil)
@@ -39,20 +40,24 @@ func newBlobsTable() sql.Table {
 	return new(blobsTable)
 }
 
+var _ Table = (*blobsTable)(nil)
+
+func (blobsTable) isGitbaseTable() {}
+
 func (blobsTable) String() string {
-	return printTable(blobsTableName, blobsSchema)
+	return printTable(BlobsTableName, BlobsSchema)
 }
 
 func (blobsTable) Resolved() bool {
 	return true
 }
 
 func (blobsTable) Name() string {
-	return blobsTableName
+	return BlobsTableName
 }
 
 func (blobsTable) Schema() sql.Schema {
-	return blobsSchema
+	return BlobsSchema
 }
 
 func (r *blobsTable) TransformUp(f sql.TransformNodeFunc) (sql.Node, error) {
@@ -79,15 +84,15 @@ func (blobsTable) Children() []sql.Node {
 }
 
 func (blobsTable) HandledFilters(filters []sql.Expression) []sql.Expression {
-	return handledFilters(blobsTableName, blobsSchema, filters)
+	return handledFilters(BlobsTableName, BlobsSchema, filters)
 }
 
 func (r *blobsTable) WithProjectAndFilters(
 	ctx *sql.Context,
 	_, filters []sql.Expression,
 ) (sql.RowIter, error) {
 	return rowIterWithSelectors(
-		ctx, blobsSchema, blobsTableName, filters,
+		ctx, BlobsSchema, BlobsTableName, filters,
 		[]string{"hash"},
 		func(selectors selectors) (RowRepoIter, error) {
 			if len(selectors["hash"]) == 0 {

blobs.go

@@ -11,19 +11,19 @@ import (
 func TestBlobsTable_Name(t *testing.T) {
 	require := require.New(t)
 
-	table := getTable(require, blobsTableName)
-	require.Equal(blobsTableName, table.Name())
+	table := getTable(require, BlobsTableName)
+	require.Equal(BlobsTableName, table.Name())
 
 	// Check that each column source is the same as table name
 	for _, c := range table.Schema() {
-		require.Equal(blobsTableName, c.Source)
+		require.Equal(BlobsTableName, c.Source)
 	}
 }
 
 func TestBlobsTable_Children(t *testing.T) {
 	require := require.New(t)
 
-	table := getTable(require, blobsTableName)
+	table := getTable(require, BlobsTableName)
 	require.Equal(0, len(table.Children()))
 }
 
@@ -32,7 +32,7 @@ func TestBlobsTable_RowIter(t *testing.T) {
 	ctx, _, cleanup := setup(t)
 	defer cleanup()
 
-	table := getTable(require, blobsTableName)
+	table := getTable(require, BlobsTableName)
 
 	rows, err := sql.NodeToRows(ctx, table)
 	require.NoError(err)
@@ -105,7 +105,7 @@ func TestBlobsPushdown(t *testing.T) {
 
 	iter, err = table.WithProjectAndFilters(session, nil, []sql.Expression{
 		expression.NewEquals(
-			expression.NewGetFieldWithTable(0, sql.Text, blobsTableName, "hash", false),
+			expression.NewGetFieldWithTable(0, sql.Text, BlobsTableName, "hash", false),
 			expression.NewLiteral("32858aad3c383ed1ff0a0f9bdf231d54a00c9e88", sql.Text),
 		),
 	})
@@ -117,15 +117,15 @@ func TestBlobsPushdown(t *testing.T) {
 
 	iter, err = table.WithProjectAndFilters(session, nil, []sql.Expression{
 		expression.NewLessThan(
-			expression.NewGetFieldWithTable(1, sql.Int64, blobsTableName, "size", false),
+			expression.NewGetFieldWithTable(1, sql.Int64, BlobsTableName, "size", false),
 			expression.NewLiteral(int64(10), sql.Int64),
 		),
 	})
 	require.NoError(err)
 
 	iter, err = table.WithProjectAndFilters(session, nil, []sql.Expression{
 		expression.NewEquals(
-			expression.NewGetFieldWithTable(0, sql.Text, blobsTableName, "hash", false),
+			expression.NewGetFieldWithTable(0, sql.Text, BlobsTableName, "hash", false),
 			expression.NewLiteral("not exists", sql.Text),
 		),
 	})

blobs_test.go

@@ -2,15 +2,23 @@ package main
 
 import (
 	"net"
+	"os"
 	"strconv"
 
 	"github.com/src-d/gitbase"
 	"github.com/src-d/gitbase/internal/function"
+	"github.com/src-d/gitbase/internal/rule"
+
 	sqle "gopkg.in/src-d/go-mysql-server.v0"
 	"gopkg.in/src-d/go-mysql-server.v0/server"
 	"gopkg.in/src-d/go-vitess.v0/mysql"
 )
 
+// Squashing tables and pushing down join conditions is still a work in
+// progress and unstable. To enable it, the UNSTABLE_SQUASH_ENABLE must
+// not be empty.
+var enableUnstableSquash = os.Getenv("UNSTABLE_SQUASH_ENABLE") != ""
+
 // CmdServer defines server command
 type CmdServer struct {
 	cmd
@@ -44,6 +52,11 @@ func (c *CmdServer) buildDatabase() error {
 
 	c.engine.AddDatabase(gitbase.NewDatabase(c.name))
 	c.engine.Catalog.RegisterFunctions(function.Functions)
+
+	if enableUnstableSquash {
+		c.engine.Analyzer.AddRule(rule.SquashJoinsRule, rule.SquashJoins)
+	}
+
 	return nil
 }
 
@@ -60,9 +73,11 @@ func (c *CmdServer) Execute(args []string) error {
 
 	hostString := net.JoinHostPort(c.Host, strconv.Itoa(c.Port))
 	s, err := server.NewServer(
-		"tcp",
-		hostString,
-		auth,
+		server.Config{
+			Protocol: "tcp",
+			Address:  hostString,
+			Auth:     auth,
+		},
 		c.engine,
 		gitbase.NewSessionBuilder(c.pool),
 	)

cmd/gitquery/server.go

@@ -12,16 +12,17 @@ import (
 type commitsTable struct {
 }
 
-var commitsSchema = sql.Schema{
-	{Name: "hash", Type: sql.Text, Nullable: false, Source: commitsTableName},
-	{Name: "author_name", Type: sql.Text, Nullable: false, Source: commitsTableName},
-	{Name: "author_email", Type: sql.Text, Nullable: false, Source: commitsTableName},
-	{Name: "author_when", Type: sql.Timestamp, Nullable: false, Source: commitsTableName},
-	{Name: "committer_name", Type: sql.Text, Nullable: false, Source: commitsTableName},
-	{Name: "committer_email", Type: sql.Text, Nullable: false, Source: commitsTableName},
-	{Name: "committer_when", Type: sql.Timestamp, Nullable: false, Source: commitsTableName},
-	{Name: "message", Type: sql.Text, Nullable: false, Source: commitsTableName},
-	{Name: "tree_hash", Type: sql.Text, Nullable: false, Source: commitsTableName},
+// CommitsSchema is the schema for the commits table.
+var CommitsSchema = sql.Schema{
+	{Name: "hash", Type: sql.Text, Nullable: false, Source: CommitsTableName},
+	{Name: "author_name", Type: sql.Text, Nullable: false, Source: CommitsTableName},
+	{Name: "author_email", Type: sql.Text, Nullable: false, Source: CommitsTableName},
+	{Name: "author_when", Type: sql.Timestamp, Nullable: false, Source: CommitsTableName},
+	{Name: "committer_name", Type: sql.Text, Nullable: false, Source: CommitsTableName},
+	{Name: "committer_email", Type: sql.Text, Nullable: false, Source: CommitsTableName},
+	{Name: "committer_when", Type: sql.Timestamp, Nullable: false, Source: CommitsTableName},
+	{Name: "message", Type: sql.Text, Nullable: false, Source: CommitsTableName},
+	{Name: "tree_hash", Type: sql.Text, Nullable: false, Source: CommitsTableName},
 }
 
 var _ sql.PushdownProjectionAndFiltersTable = (*commitsTable)(nil)
@@ -30,20 +31,24 @@ func newCommitsTable() sql.Table {
 	return new(commitsTable)
 }
 
+var _ Table = (*commitsTable)(nil)
+
+func (commitsTable) isGitbaseTable() {}
+
 func (commitsTable) String() string {
-	return printTable(commitsTableName, commitsSchema)
+	return printTable(CommitsTableName, CommitsSchema)
 }
 
 func (commitsTable) Resolved() bool {
 	return true
 }
 
 func (commitsTable) Name() string {
-	return commitsTableName
+	return CommitsTableName
 }
 
 func (commitsTable) Schema() sql.Schema {
-	return commitsSchema
+	return CommitsSchema
 }
 
 func (r *commitsTable) TransformUp(f sql.TransformNodeFunc) (sql.Node, error) {
@@ -70,15 +75,15 @@ func (commitsTable) Children() []sql.Node {
 }
 
 func (commitsTable) HandledFilters(filters []sql.Expression) []sql.Expression {
-	return handledFilters(commitsTableName, commitsSchema, filters)
+	return handledFilters(CommitsTableName, CommitsSchema, filters)
 }
 
 func (r *commitsTable) WithProjectAndFilters(
 	ctx *sql.Context,
 	_, filters []sql.Expression,
 ) (sql.RowIter, error) {
 	return rowIterWithSelectors(
-		ctx, commitsSchema, commitsTableName, filters,
+		ctx, CommitsSchema, CommitsTableName, filters,
 		[]string{"hash"},
 		func(selectors selectors) (RowRepoIter, error) {
 			if len(selectors["hash"]) == 0 {